feat(control-plane): protect caller-authored first delivery on CLI/MCP

apps/presentation/dashboard/smoke/checkpoint-recovery-browser-smoke.mjs

@@ -0,0 +1,141 @@
+// Actual packaged checkpoint controls + production owning HTTP/store/controller.
+// Workspace navigation is synthetic; target /loopx responses are never fabricated.
+import assert from "node:assert/strict";
+import {spawn} from "node:child_process";
+import {once} from "node:events";
+import {createHash} from "node:crypto";
+import {mkdir, readFile, writeFile} from "node:fs/promises";
+import {resolve} from "node:path";
+import {createInterface} from "node:readline";
+import {createServer} from "node:net";
+import {resolveTestPython} from "../../../../scripts/test-python.mjs";
+import {launchBrowser, loadPlaywright, waitForHttp} from "../../../../examples/dashboard-browser-smoke-support.mjs";
+
+const at=process.argv.indexOf("--output");
+assert.ok(at>=0 && process.argv[at+1], "Declare a fresh evidence output directory");
+const out=resolve(process.argv[at+1]);
+await mkdir(out,{recursive:true});
+const probe=createServer();probe.listen(0,"127.0.0.1");await once(probe,"listening");
+process.env.LOOPX_PERSONAL_WORKSPACE_PORT=String(probe.address().port);
+await new Promise((done,reject)=>probe.close(error=>error?reject(error):done()));
+process.env.LOOPX_PERSONAL_WORKSPACE_PACKAGED="1";
+const {repoRoot,port,startServer}=await import("../../../../examples/personal-workspace-browser/fixture.mjs");
+const {openWorkspacePage,fetchWorkspaceBackend}=await import("../../../../examples/personal-workspace-browser/scenario-context.mjs");
+const fixture=spawn(resolveTestPython({repoRoot}),["-u","apps/presentation/dashboard/smoke/checkpoint-recovery-http-fixture.py",resolve(out,"backend")],
+  {cwd:repoRoot,stdio:["pipe","pipe","pipe"],windowsHide:true});
+const exited=once(fixture,"exit");
+const lines=createInterface({input:fixture.stdout});const iterator=lines[Symbol.asyncIterator]();
+let diagnostic="";fixture.stderr.on("data",chunk=>{diagnostic+=String(chunk);});
+async function next(){const line=await iterator.next();assert.equal(line.done,false,diagnostic);return JSON.parse(line.value);}
+async function command(value){fixture.stdin.write(`${value}\n`);return next();}
+async function artifact(path){return {path,sha256:createHash("sha256").update(await readFile(path)).digest("hex")};}
+let server,browser,workspace;
+const requests=[];
+try {
+  const {origin,session,original_identity}=await next();
+  server=await startServer();
+  const url=`http://127.0.0.1:${port}/chat/?statusUrl=/status.json`;
+  await waitForHttp(url);
+  browser=await launchBrowser(loadPlaywright().chromium);
+  workspace=await openWorkspacePage({newPage:options=>browser.newPage({locale:"zh-CN",...options})},url,{
+    beforeGoto:async(_api,page)=>{
+      page.__loopxRuntime.sessions.set(session.session_id,session);
+      // Only the unrelated workspace/navigation uses the existing synthetic API.
+      // Every target operation, including reads across reload, uses its owning server.
+      await page.route(`**/api/chat/sessions/${session.session_id}/loopx`,async route=>{
+        const request=route.request();
+        const body=request.method()==="POST"?request.postDataJSON():null;
+        const response=await fetchWorkspaceBackend(route,origin);
+        requests.push({method:request.method(),body,status:response.status(),response:await response.json()});
+        await route.fulfill({response});
+      });
+    },
+  });
+  const {page}=workspace;
+  await page.locator(".personal-goal-link",{hasText:"Product Release"}).click();
+  await page.getByRole("navigation",{name:"Goal 视图"}).getByRole("button",{name:"对话",exact:true}).click();
+  const section=page.getByRole("region",{name:"LoopX 运行模式"});
+  const retained=page.getByText("LoopX · 结果已保留，待协调员核验",{exact:true});
+  await retained.waitFor({timeout:60000});
+  const before=await command("inspect");
+  assert.deepEqual(before.counts,{result:1,debit:0});assert.equal(before.dispatches.length,0);
+  await page.reload();await retained.waitFor({timeout:60000});
+  const reloaded=await command("inspect");
+  assert.deepEqual(reloaded.counts,before.counts);assert.equal(reloaded.dispatches.length,0);
+  const writes=()=>requests.filter(row=>row.method==="POST" && !["operations","inspect"].includes(row.body?.operation));
+  assert.equal(writes().length,0,"Reload must not dispatch or mutate execution");
+  const button=page.getByRole("button",{name:"继续核验",exact:true});
+  await button.scrollIntoViewIfNeeded();
+  const screenshots=[];
+  let path=resolve(out,"retained-desktop.png");await page.screenshot({path,animations:"disabled"});screenshots.push(await artifact(path));
+  await button.click();
+  let after;const deadline=Date.now()+300000;
+  while(Date.now()<deadline){
+    after=await command("inspect");assert.deepEqual(after.failures,[]);
+    assert.ok(after.counts.debit<=1 && after.dispatches.length<=1);
+    if(after.counts.debit===1 && after.decisions.length===1)break;
+    await new Promise(done=>setTimeout(done,500));
+  }
+  assert.deepEqual(after.counts,{result:1,debit:1});assert.equal(after.dispatches.length,1);
+  assert.equal(after.result_sha256,before.result_sha256);
+  assert.equal(after.upstream_thread_id,before.upstream_thread_id,"Continue retains the original SDK session");
+  const posted=writes();
+  assert.equal(posted.length,1);assert.equal(posted[0].body.operation,"resume");assert.equal(posted[0].status,200);
+  await page.reload();
+  await page.getByText("LoopX · 已暂停",{exact:true}).waitFor({timeout:60000});
+  assert.equal(await retained.count(),0,"Settled native readback clears the pending-result label");
+  const replayed=await command("replay");
+  assert.deepEqual(replayed.counts,after.counts);assert.equal(replayed.dispatches.length,1);
+  assert.deepEqual(replayed.replayed_identity,original_identity);
+  assert.equal(replayed.result_sha256,before.result_sha256);
+  await page.setViewportSize({width:390,height:844});await page.emulateMedia({reducedMotion:"reduce"});
+  await page.getByRole("button",{name:"恢复推进",exact:true}).scrollIntoViewIfNeeded();
+  path=resolve(out,"settled-narrow.png");await page.screenshot({path,animations:"disabled"});screenshots.push(await artifact(path));
+  const unavailable=await command("unavailable");
+  assert.equal(unavailable.snapshot.coordinator_checkpoint.stage,"observation_unavailable");
+  assert.equal(unavailable.snapshot.coordinator_checkpoint.settlement_identity,undefined);
+  await page.reload();
+  await page.getByText("LoopX · 核验信息暂不可用",{exact:true}).waitFor({timeout:60000});
+  assert.equal(await page.getByText("原执行尚未完成结算",{exact:false}).count(),0);
+  const reread=page.getByRole("button",{name:"重新读取",exact:true});
+  await reread.click();
+  assert.equal(writes().length,1,"Unavailable observation reread is not an execution request");
+  const held=await command("inspect");
+  assert.deepEqual(held.counts,{result:1,debit:1});assert.equal(held.dispatches.length,1);
+  path=resolve(out,"unavailable-narrow.png");await page.screenshot({path,animations:"disabled"});screenshots.push(await artifact(path));
+  await page.setViewportSize({width:1512,height:982});
+  path=resolve(out,"unavailable-desktop.png");await page.screenshot({path,animations:"disabled"});screenshots.push(await artifact(path));
+  await command("restore_observation");
+  await page.reload();
+  await page.getByRole("button",{name:"恢复推进",exact:true}).waitFor({timeout:60000});
+  assert.equal(writes().length,1);
+  const saved=await command("save");
+  assert.equal(writes().length,1);
+  assert.equal(workspace.errors.length,0,workspace.errors.join("\n"));
+  await writeFile(resolve(out,"http-observations.json"),JSON.stringify(requests,null,2));
+  const proof={packaged:true,target_http_mocked:false,transport:"deterministic_sdk",paid_calls:0,
+    non_owning_shell_fixture:"existing personal-workspace navigation/status API; target sessions/loopx uses production owner",
+    browser_version:browser.version(),clicked_continue_review:true,reloads:4,
+    unavailable_observation:{read_only_reread:true,restored:true,counts:held.counts},
+    original_identity,settled_identity:saved.settled_identity,replayed_identity:replayed.replayed_identity,
+    result_sha256_before:before.result_sha256,result_sha256_after:after.result_sha256,
+    business_counts_before:before.counts,business_counts_after:after.counts,business_counts_replayed:replayed.counts,
+    direction_dispatches:saved.dispatches.length,implementation_dispatches:0,
+    session_id_before:before.session_id,session_id_after:saved.session_id,
+    upstream_thread_id_before:before.upstream_thread_id,upstream_thread_id_after:saved.upstream_thread_id,
+    errors:workspace.errors,screenshots,native_evidence:[]};
+  for(const name of ["before.json","first-result.json","inspect.json","replay.json","save.json","unavailable.json","restore_observation.json"])
+    proof.native_evidence.push(await artifact(resolve(out,"backend",name)));
+  proof.native_evidence.push(await artifact(resolve(out,"http-observations.json")));
+  await writeFile(resolve(out,"proof.json"),JSON.stringify(proof,null,2));
+  console.log("checkpoint-recovery: passed (packaged click/reload, production owning HTTP/controller/native checkpoint, original identity/session/result, one debit and zero replay increments; deterministic SDK)");
+} catch(error){
+  await writeFile(resolve(out,"http-observations-failed.json"),JSON.stringify(requests,null,2));
+  await writeFile(resolve(out,"failure.txt"),`${String(error)}\n${diagnostic}`);
+  if(workspace)await writeFile(resolve(out,"failure-body.txt"),await workspace.page.locator("body").innerText());
+  throw error;
+} finally {
+  await workspace?.close();await browser?.close();server?.kill("SIGTERM");
+  fixture.stdin.end();if(fixture.exitCode===null)await exited;lines.close();
+  await writeFile(resolve(out,"backend-stderr.txt"),diagnostic);
+}

apps/presentation/dashboard/smoke/checkpoint-recovery-http-fixture.py

@@ -0,0 +1,164 @@
+"""Isolated production checkpoint owner with a deterministic SDK transport.
+
+Invoked from the repository root by the packaged browser smoke. No paid model,
+real Lark, active operator Goal or user browser profile is used.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+import sys
+import threading
+from types import SimpleNamespace
+
+ROOT = Path.cwd()
+sys.path[:0] = [str(ROOT), str(ROOT/'tests'), str(ROOT/'tests/control_plane')]
+import pytest
+from loopx.chat_runtime import ChatRuntimeController
+from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
+from loopx.chat_store import ChatSessionStore
+from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+from loopx.todos import add_goal_todo
+from tests.control_plane import test_default_checkpoint_continuation as native
+from tests import test_goal_mode_mcp_settlement as mcp
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value,sort_keys=True).encode()).hexdigest()
+
+
+def main():
+    out=Path(sys.argv[1]).resolve();out.mkdir(parents=True,exist_ok=False)
+    isolated=out/'native';isolated.mkdir()
+    patches=pytest.MonkeyPatch()
+    goal_id,agent_id='product-release','lead'
+    for module in (native,mcp):
+        patches.setattr(module,'GOAL_ID',goal_id)
+        patches.setattr(module,'AGENT_ID',agent_id)
+    registry,state,todo=native._fixture('file',isolated,patches)
+    payload=json.loads(registry.read_text());goal=payload['goals'][0]
+    goal['domain']='Checkpoint recovery demonstration'
+    goal['coordination']['registered_agents']=[agent_id,'reviewer','codex']
+    goal['spawn_policy']={'mode':'multi_subagent','allowed':True,'max_children':2,
+                          'execution_config':'.loopx/config/delegations.json'}
+    registry.write_text(json.dumps(payload),encoding='utf-8')
+    project=state.parent
+    config=project/'.loopx/config/delegations.json';config.parent.mkdir(parents=True,exist_ok=True)
+    config.write_text(json.dumps({'schema_version':'loopx_local_delegation_v0','bindings':[{
+        'id':'review','agent_id':'reviewer','todo_id':'review-task','requesters':[agent_id],
+        'workspace':str(project),'host_args':['--host','dsh'],'timeout_seconds':60,
+        'output_refs':['output.json']}]}),encoding='utf-8')
+    control=mcp._control(registry)
+    original_guard=json.loads(control.should_run())
+    first=json.loads(control.complete_task(todo,agent_id,'Candidate verified.',no_follow_up=True))
+    assert first['stage']=='direction_pending' and first['completed'] is True
+    original=first['settlement_identity']
+    runtime_root=isolated/'runtime'
+    index=runtime_root/'goals'/goal_id/'runs/index.jsonl'
+    store=ChatSessionStore(runtime_root)
+    sid='session-goal-product-release-codex'
+    store.create_session(goal_id=goal_id,agent_id='codex',channel_id='goal.product-release',
+                         upstream_thread_id='synthetic-original-sdk',session_id=sid,
+                         adapter_kind='codex_app_server',upstream_mode='chat')
+    settings={'agent_id':agent_id,'token_budget':10000,'execution_config_ref':'.loopx/config/delegations.json',
+              'config_digest':hashlib.sha256(config.read_bytes()).hexdigest()}
+    store.update_session(sid,loopx_tools=True,native_goal={'status':'paused','tokensUsed':0},
+                         loopx_mode={'enabled':True,'paused':True,'settings':settings})
+    patches.setenv('LOOPX_CHAT_CODEX_HOME',str(isolated/'codex-home'))
+    controller=ChatRuntimeController(store=store,codex_bin='missing-codex',registry_path=registry)
+    adapter=SimpleNamespace(goal_driver=SimpleNamespace(stopped=threading.Event()),
+                            session=SimpleNamespace(),upstream_thread_id='synthetic-original-sdk',
+                            close_session=lambda:None,healthcheck=lambda:True)
+    patches.setattr(controller,'_ensure_adapter_locked',lambda *a,**k:adapter)
+    controller.adapters[sid]=adapter
+    dispatches=[]; failures=[]; decisions=[]
+
+    def native_snapshot():
+        items=read_canonical_todos_if_promoted(runtime_root=runtime_root,goal_id=goal_id)['todos']
+        row=next(item for item in items if item['todo_id']==todo)
+        facts={key:row.get(key) for key in ('todo_id','status','evidence','completed_at','completion_turn_key')}
+        events=[json.loads(line) for line in index.read_text(encoding='utf-8').splitlines()] if index.exists() else []
+        return {'facts':facts,'result_sha256':digest(facts),
+                'counts':{'result':int(row['status']=='done' and row['evidence']=='Candidate verified.'),
+                          'debit':sum(e.get('classification')=='quota_slot_spent' for e in events)},
+                'session_id':store.load_session(sid)['session_id'],
+                'upstream_thread_id':store.load_session(sid)['upstream_thread_id'],
+                'native_events':events,'dispatches':list(dispatches),'decisions':list(decisions)}
+
+    def provider(**kwargs):
+        dispatches.append({'session_id':kwargs['session_id'],'turn_id':kwargs['turn_id'],
+                           'upstream_thread_id':adapter.upstream_thread_id,'phase':'checkpoint_direction'})
+        lock=None
+        try:
+            assert kwargs['session_id']==sid and kwargs['loopx_execution'] is True
+            lock=controller.loopx_mode.prepare(sid,kwargs['turn_id'],adapter,lambda *a:{},lambda *a:None)
+            read=adapter.session.read_tool_handler('loopx_collaboration',{
+                'action':'read','operation_id':todo,'brief':{'purpose':'first_delivery'}})
+            assert read['ok'] and read['settlement_identity']==original,read
+            decision=adapter.session.read_tool_handler('loopx_collaboration',{
+                'action':'resume','operation_id':todo,'brief':{'purpose':'first_delivery',
+                    'read_context_id':read['read_context_id'],
+                    'vision_unchanged_reason':'The retained candidate and freshly observed remaining obligations preserve the route.'}})
+            assert decision['ok'] and decision['settlement_complete'] and decision['settlement_identity']==original,decision
+            decisions.append({'read':read,'settlement':decision})
+            store.update_turn(sid,kwargs['turn_id'],status='completed',response={'text':'Original checkpoint settled.'})
+            store.release_active_turn(sid,kwargs['turn_id'],last_activity_at='2026-10-19T00:00:00Z',last_error_code=None)
+            store.update_session(sid,loopx_mode={'enabled':True,'paused':True,'settings':settings})
+        except Exception as exc:
+            failures.append(repr(exc));raise
+        finally:
+            if lock:lock.__exit__(None,None,None)
+
+    patches.setattr(controller,'_run_started_turn',provider)
+    server=ChatHTTPServer(('127.0.0.1',0),ChatRequestHandler)
+    server.verbose=False;server.registry_path=registry;server.runtime_root=runtime_root
+    server.chat_store=store;server.runtime_controller=controller
+    http=[]
+    send=ChatRequestHandler._send_json
+    def observed_send(handler,body,*,status=200):
+        if handler.path.startswith('/api/chat/sessions/') and handler.path.endswith('/loopx'):
+            http.append({'method':handler.command,'path':handler.path,'status':status,'response':body})
+        return send(handler,body,status=status)
+    patches.setattr(ChatRequestHandler,'_send_json',observed_send)
+    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
+    baseline=native_snapshot(); assert baseline['counts']=={'result':1,'debit':0}
+    (out/'before.json').write_text(json.dumps(baseline,indent=2),encoding='utf-8')
+    (out/'first-result.json').write_text(json.dumps(first,indent=2),encoding='utf-8')
+    print(json.dumps({'origin':f'http://127.0.0.1:{server.server_port}','session':store.load_session(sid),
+                      'original_identity':original}),flush=True)
+    try:
+        for line in sys.stdin:
+            command=line.strip()
+            if command=='inspect':
+                value=native_snapshot();value['failures']=list(failures)
+            elif command=='replay':
+                assert len(decisions)==1 and not failures
+                from loopx.chat_coordinator_checkpoint import CoordinatorCheckpoints
+                checkpoint=CoordinatorCheckpoints(registry,runtime_root,goal_id,agent_id)
+                # Replay the exact successful decision, without another SDK dispatch.
+                decision=decisions[0]
+                repeated=checkpoint.resume(todo,{'read_context_id':decision['read']['read_context_id'],
+                    'vision_unchanged_reason':'The retained candidate and freshly observed remaining obligations preserve the route.'})
+                assert repeated['ok'] and repeated['settlement_identity']==original,repeated
+                value={**native_snapshot(),'replayed_identity':repeated['settlement_identity'],'replay':repeated}
+            elif command=='save':
+                assert not failures,failures
+                value={**native_snapshot(),'http':http,'original_identity':original,
+                       'settled_identity':decisions[0]['settlement']['settlement_identity']}
+            elif command in {'unavailable', 'restore_observation'}:
+                damaged=runtime_root/'goals'/goal_id/'checkpoint-contexts'/('f'*64+'.json')
+                if command=='unavailable':
+                    damaged.write_text('{',encoding='utf-8')
+                else:
+                    damaged.unlink()
+                value={'snapshot':controller.loopx_mode.snapshot(sid),'native':native_snapshot()}
+            else:raise ValueError('Unknown fixture command')
+            (out/(command+'.json')).write_text(json.dumps(value,indent=2),encoding='utf-8')
+            print(json.dumps(value),flush=True)
+    finally:
+        server.shutdown();server.server_close();thread.join(timeout=5)
+        controller.close();patches.undo()
+
+
+if __name__=='__main__':main()