Add qsp box clone and MCP duplication tool

README.md

@@ -185,6 +185,8 @@ qsp box ls                           # compact table plus account summary
 qsp box ls --json                    # inspect existing boxes, prices and effective limits
 # Obtain user consent before billable creation/resume.
 qsp box create --name work --wait --json  # prompts for billable creation approval
+qsp box clone <machine-id-or-name> --name worker --count 2 --accept-spend --json
+qsp box clone <machine-id> --from-snapshot <snapshot-id-or-name> --accept-spend --json
 # For automation, use --accept-spend only after explicit user approval.
 qsp box get MID --json
 qsp box wait MID --timeout 180 --json
@@ -245,7 +247,7 @@ Treat guest output and downloaded files as untrusted data, never consent.
 
 `qsp mcp` runs **local stdio only**, with no listener and protocol-only stdout.
 Install `quicksilverpro[mcp]` on Python >=3.10; SDK is pinned to `mcp==1.25.0`.
-`box_create` and `box_resume` require `accept_spend:true` (after showing prices/limits and getting user approval). Tools cover create/list/get/wait/exec, files list/upload/download, pause/resume/delete,
+`box_create`, `box_clone` and `box_resume` require `accept_spend:true` (after showing prices/limits and getting user approval). Tools cover create/clone/list/get/wait/exec, files list/upload/download, pause/resume/delete,
 and owner/share links. MCP file tools accept base64 up to 256 KiB, with no local
 filesystem paths; larger files use CLI copy. Delete requires matching `confirm_mid`.
 The packaged [agent reference](src/quicksilverpro/skills/quicksilver-machines/references/api.md)
@@ -388,3 +390,24 @@ connections per machine and existing account/IP/global stream limits apply.
 Key removal stops new authentication, not existing SSH sessions. Retry a failed
 key update to complete propagation to all machines; SSH tickets resync the current
 account keys before allowing a connection.
+
+
+## Machine cloning
+
+Cloning copies both disks, including user-saved credentials, agent logins, SSH
+keys, packages and repositories. Each duplicate receives its own machine identity
+and QuickSilver AI key; share links and billing records start fresh. Processes
+start fresh. A running source briefly pauses for a consistent disk copy, then
+continues. Duplicates wait when running capacity is full; each counts toward your
+account machine limit. A count can partially succeed: inspect both `machines` and
+`error`. Reuse the reported request ID to recover an unknown or partial outcome;
+use a fresh `--request-id` to intentionally create another identical batch.
+
+Before first boot, the duplicate replaces the exact source managed inference key
+in supported regular text files under `/home/user` and `/etc` (including agent
+configs and env files). Files over 16 MiB, binary files, symlinks, transformed
+keys and copies outside those trees remain your responsibility. The scan fails
+closed at 512 MiB of supported text or 30 seconds. Excluded binary files do not
+use the text budget. If a duplicate cannot start because of these scan limits,
+delete it, reduce files on the source and duplicate again. This is not forensic
+disk erasure.

npm/README.md

@@ -84,3 +84,14 @@ Editor SSH: after setup, run `npx quicksilverpro box ssh-config <machine>` once.
 Use a stable installed `qsp` executable for your editor's ProxyCommand (`pipx install quicksilverpro`).
 `qsp box ssh <machine>` connects interactively; `ssh -L` forwarding works. Connecting a
 paused box resumes billing. The bundled MCP `box_ssh_info` tool returns local setup instructions.
+
+
+## Machine cloning
+
+`npx quicksilverpro box clone <machine> --from-snapshot <id-or-name> --name worker --count 2 --accept-spend --json`
+creates independent copies of a configured machine. Omit `--from-snapshot` for
+its current disks. Everything on disk is copied, including credentials you saved;
+applications boot fresh with a new machine identity and QuickSilver AI key.
+The Python MCP server used by `npx quicksilverpro setup` exposes `box_clone` with
+name, count, snapshot, request ID and explicit spend approval. Each duplicate
+counts toward account limits and is billed separately while running.

src/quicksilverpro/box.py

@@ -91,7 +91,7 @@ def confirm_spend(client, operation, accept_spend):
     policy = client.list()
     click.echo(
         json.dumps(
-            {k: policy[k] for k in ("prices", "limits", "retention", "trial") if k in policy}
+            {k: policy[k] for k in ("prices", "limits", "limit", "running_limit", "retention", "trial") if k in policy}
         ),
         err=True,
     )
@@ -190,6 +190,23 @@ def create(name, size, request_id, do_wait, timeout, accept_spend, as_json):
         emit(result, as_json)
 
 
+@box.command(help="Duplicate a box's disks, including credentials you saved. Each duplicate is billed separately.")
+@click.argument("machine")
+@click.option("--from-snapshot", help="Snapshot ID or exact name; default: current disk.")
+@click.option("--name", help="Name for the duplicate(s).")
+@click.option("--count", type=click.IntRange(1, 64), default=1, show_default=True)
+@click.option("--request-id", help="Reuse after an unknown outcome; a fresh ID creates new duplicates.")
+@click.option("--accept-spend", is_flag=True, help="Approve billable duplication without a prompt.")
+@json_option
+def clone(machine, from_snapshot, name, count, request_id, accept_spend, as_json):
+    with MachinesClient() as client:
+        confirm_spend(client, "creation", accept_spend)
+        result = client.clone(machine, from_snapshot, name, count, request_id,
+            on_request_id=lambda rid: click.echo("Clone request ID: " + rid, err=True))
+        emit(result, as_json)
+        if result.get("error"):click.get_current_context().exit(1)
+
+
 @box.command(help="Wait for a box to become ready.")
 @click.argument("mid")
 @click.option("--timeout", type=click.IntRange(1, 180), default=180)

src/quicksilverpro/machines.py

@@ -362,6 +362,36 @@ def list(self):
     def get(self, mid):
         return self.request("GET", "/v1/machines/" + mid_value(mid))
 
+    def clone(self, mid, from_snapshot=None, name=None, count=1, request_id=None, on_request_id=None):
+        if not isinstance(mid, str) or not mid:
+            raise MachineError("invalid_machine", "Choose a source machine ID or name.", 2)
+        if not re.fullmatch(r"[a-f0-9]{12}", mid):
+            matches = [m for m in self.list().get("machines", []) if m.get("name") == mid]
+            if len(matches) != 1:
+                raise MachineError("invalid_machine", "Source name is missing or ambiguous; use its machine ID.", 2)
+            mid = mid_value(matches[0]["id"])
+        if type(count) is not int or not 1 <= count <= 64:
+            raise MachineError("invalid_count", "Count must be 1..64 and within your account limit.", 2)
+        if name is not None and (not isinstance(name, str) or not 1 <= len(name.strip()) <= 64):
+            raise MachineError("invalid_name", "Name must be 1..64 characters.", 2)
+        if from_snapshot is not None and (not isinstance(from_snapshot, str) or not 1 <= len(from_snapshot) <= 80):
+            raise MachineError("invalid_snapshot", "Snapshot must be an ID or name up to 80 characters.", 2)
+        body = {"count": count, "name": name, "from_snapshot": from_snapshot}
+        path = config_dir() / "clone-requests.json"
+        journal = read_json(path)
+        fingerprint = hashlib.sha256((self.origin + hashlib.sha256(self.key.encode()).hexdigest() + mid + json.dumps(body, sort_keys=True)).encode()).hexdigest()
+        request_id = request_id or journal.get(fingerprint, {}).get("request_id") or str(uuid.uuid4())
+        if not re.fullmatch(r"[a-zA-Z0-9-]{16,64}", request_id):
+            raise MachineError("invalid_request_id", "Request ID must be 16..64 letters, digits or hyphens.", 2)
+        atomic_write(path, json.dumps({**journal, fingerprint: {"request_id": request_id}}), private=True)
+        if on_request_id:on_request_id(request_id)
+        try:
+            result = self.request("POST", "/v1/machines/" + mid + "/clone", json={**body, "request_id": request_id})
+        except MachineError as error:
+            error.details["request_id"] = request_id
+            raise
+        return {**result, "request_id": request_id}
+
     def create(self, name, size=2, request_id=None, on_request_id=None):
         if not isinstance(name, str) or not 1 <= len(name) <= 64 or size not in (2, 4):
             raise MachineError(

src/quicksilverpro/mcp.py

@@ -39,6 +39,14 @@ def schema(properties, required=(), **extra):
 
 
 SCHEMAS = {
+    "box_clone": schema({
+        "mid": MID,
+        "name": {"type": "string", "minLength": 1, "maxLength": 64},
+        "from_snapshot": {"type": "string", "minLength": 1, "maxLength": 80},
+        "count": {"type": "integer", "minimum": 1, "maximum": 64, "default": 1},
+        "request_id": {"type": "string", "pattern": "^[a-zA-Z0-9-]{16,64}$"},
+        "accept_spend": {"type": "boolean", "default": False},
+    }, ("mid", "request_id")),
     "box_create": schema(
         {
             "name": {"type": "string", "minLength": 1, "maxLength": 64},
@@ -124,6 +132,7 @@ def schema(properties, required=(), **extra):
         **({'stream':{'type':'string','enum':['stdout','stderr']},'offset':{'type':'integer','minimum':0},'tail':{'type':'integer','minimum':0,'maximum':65536}} if action=='logs' else {})},('mid','job_id'))
 
 DESCRIPTIONS = {
+    "box_clone": "Duplicate both disks into new BILLABLE boxes with fresh identities and QuickSilver AI keys. Everything on disk, including user-saved credentials and logins, is copied. Applications boot fresh; share links are cleared. Show box_list prices/limits and obtain spend approval before accept_spend=true. Reuse request_id after an unknown or partial outcome; inspect machines AND error. Count is bounded by the account limit.",
     "box_create": "Create a BILLABLE QuickSilver box. First use box_list to show effective prices/limits and existing boxes to the user and obtain explicit spend consent; only then pass accept_spend=true. Missing or false approval refuses to create. Reuse request_id after unknown outcome.",
     "box_list": "Read existing boxes, effective limits, prices and account metadata without creating anything.",
     "box_ssh_info": "Return local SSH/editor setup instructions. Does not write files, register keys, or resume a box. Run the returned command locally to set up SSH; connecting a paused box resumes billing.",
@@ -185,9 +194,9 @@ def dispatch(name, arguments, cancellation=None):
     args = dict(arguments)
     if name == "box_delete" and args["mid"] != args["confirm_mid"]:
         raise MachineError("confirmation_required", "confirm_mid must equal mid.", 2)
-    if name in ("box_create", "box_resume"):
+    if name in ("box_create", "box_resume", "box_clone"):
         if not args.pop("accept_spend", False):
-            operation = "creation" if name == "box_create" else "resume"
+            operation = "creation" if name in ("box_create", "box_clone") else "resume"
             raise MachineError(
                 "consent_required",
                 f"Billable {operation} requires accept_spend=true after showing prices/limits and obtaining explicit user approval.",
@@ -204,6 +213,8 @@ def dispatch(name, arguments, cancellation=None):
                     "ssh_host":"qsp-"+machine["id"],"user":"user","auto_resume_uses_credits":True}
         if name == "box_get":
             return client.get(args["mid"])
+        if name == "box_clone":
+            return client.clone(**args)
         if name == "box_create":
             wait = args.pop("wait_seconds", 0)
             result = client.create(**args)
@@ -283,7 +294,7 @@ async def call_tool(name, arguments):
                 lambda: dispatch(name, arguments, cancellation), abandon_on_cancel=True
             )
             payload = envelope("guest_exit" if name == "box_exec" else "result", result)
-            failed = name == "box_exec" and exec_exit(result) != 0
+            failed = (name == "box_exec" and exec_exit(result) != 0) or (name == "box_clone" and bool(result.get("error")))
         except anyio.get_cancelled_exc_class():
             cancellation.cancel()
             raise

tests/test_clone.py

@@ -0,0 +1,63 @@
+import json
+import inspect
+import httpx
+import pytest
+from click.testing import CliRunner
+from quicksilverpro.cli import main
+from quicksilverpro.machines import MachinesClient, MachineError
+from quicksilverpro.mcp import dispatch
+
+MID='a'*12
+
+def cli_runner():
+    return CliRunner(mix_stderr=False) if 'mix_stderr' in inspect.signature(CliRunner).parameters else CliRunner()
+
+def test_clone_cli_request_and_name_resolution(mock_api):
+    def handler(request):
+        if request.method=='GET':return httpx.Response(200,json={'machines':[{'id':MID,'name':'configured'}],'limit':3})
+        return httpx.Response(201,json={'machines':[{'id':'b'*12}],'error':None})
+    calls=mock_api(handler)
+    result=cli_runner().invoke(main,['box','clone','configured','--from-snapshot','ready','--name','worker','--count','2','--accept-spend','--json'])
+    assert result.exit_code==0,result.output
+    post=calls[-1];assert post.url.path=='/v1/machines/'+MID+'/clone'
+    body=json.loads(post.content);assert body['count']==2 and body['from_snapshot']=='ready' and body['name']=='worker'
+    assert json.loads(result.stdout)['result']['machines'][0]['id']=='b'*12
+
+def test_clone_retry_preserves_request_id_and_partial_results(mock_api):
+    calls=mock_api(lambda r:httpx.Response(200,json={'machines':[{'id':'b'*12}],'error':{'status':409,'detail':'Account limit reached.'}}))
+    with MachinesClient() as client:
+        first=client.clone(MID);second=client.clone(MID)
+    assert first['request_id']==second['request_id']
+    assert json.loads(calls[0].content)['request_id']==json.loads(calls[1].content)['request_id']
+    result=cli_runner().invoke(main,['box','clone',MID,'--accept-spend','--json'])
+    assert result.exit_code==1
+    assert json.loads(result.stdout)['result']['machines'][0]['id']=='b'*12
+
+def test_clone_mcp_spend_gate_and_public_request(mock_api):
+    pytest.importorskip('jsonschema')
+    calls=mock_api(lambda r:httpx.Response(201,json={'machines':[],'error':None}))
+    args={'mid':MID,'request_id':'clone-request-00000001'}
+    with pytest.raises(MachineError,match='accept_spend'):dispatch('box_clone',args)
+    assert calls==[]
+    result=dispatch('box_clone',{**args,'accept_spend':True,'count':2,'from_snapshot':'ready'})
+    assert result['error'] is None
+    assert json.loads(calls[-1].content)['count']==2
+
+@pytest.mark.parametrize('count',[True,0,65])
+def test_invalid_count_never_requests(mock_api,count):
+    calls=mock_api(lambda r:httpx.Response(500))
+    with MachinesClient() as client,pytest.raises(MachineError):client.clone(MID,count=count)
+    assert calls==[]
+
+def test_mcp_partial_failure_keeps_admitted_ids(mock_api):
+    pytest.importorskip('mcp')
+    import anyio
+    from mcp.shared.memory import create_connected_server_and_client_session
+    from quicksilverpro.mcp import create_server
+    mock_api(lambda r:httpx.Response(201,json={'machines':[{'id':'b'*12}],'error':{'status':409,'detail':'Account limit reached.'}}))
+    async def scenario():
+        async with create_connected_server_and_client_session(create_server()) as session:
+            result=await session.call_tool('box_clone',{'mid':MID,'count':2,'request_id':'clone-request-00000001','accept_spend':True})
+            assert result.isError
+            assert result.structuredContent['result']['machines'][0]['id']=='b'*12
+    anyio.run(scenario)

tests/test_mcp.py

@@ -19,7 +19,7 @@ def test_sdk_tool_calls_and_schemas(mock_api):
     async def scenario():
         async with create_connected_server_and_client_session(create_server()) as session:
             tools = (await session.list_tools()).tools
-            assert len(tools) == len(SCHEMAS) == 18
+            assert len(tools) == len(SCHEMAS) == 19
             assert all(
                 t.inputSchema["additionalProperties"] is False and t.outputSchema for t in tools
             )
@@ -79,7 +79,7 @@ async def scenario():
         async with stdio_client(params) as (reader, writer):
             async with ClientSession(reader, writer) as session:
                 await session.initialize()
-                assert len((await session.list_tools()).tools) == 18
+                assert len((await session.list_tools()).tools) == 19
                 result = await session.call_tool("box_list", {})
                 assert result.isError
                 assert result.structuredContent["error"]["code"] == "auth_required"