Keep launch recovery evidence when broker end is unconfirmed

packages/cli/src/launcher.mjs

@@ -818,7 +818,9 @@ export function createLauncher({
       const reason = haltGeneration ? 'kill-switch' : endReason(code, signal);
       try {
         const ended = await endRun(run, reason, signal ? null : code);
-        registry.remove(b.id, run);
+        // Keep the exact launch identity for the next host if the broker
+        // refused or did not confirm the end.
+        if (typeof ended?.released === "boolean") registry.remove(b.id, run);
         // Row 125 (#1619, REQ-LIMIT-4): a run a limit check stopped names the limit, from the claim row.
         record({ event: "end", instance, ...(thread ? { thread, session } : {}), run, reason, exitCode: code, signal, ...(haltGeneration ? { haltGeneration } : {}), ...(ended?.limit ? { limit: ended.limit } : {}) });
         log(`run ${run} (${instance}) ended: ${reason}`);

packages/cli/tests/case2-install.test.mjs

@@ -71,7 +71,7 @@ function world(t, { profile = "library" } = {}) {
   script(join(bin, "stack-copy"), `echo "stack-copy $*" >>'${log}'
 [ "$1" = copy ] && [ "$3" = --copies ] || exit 4
 mkdir -p "$4/$2/scripts" && : >"$4/$2/scripts/mosaic" && echo "copied: $4/$2"
-for file in scripts/case2-persona-import.mjs packages/harness/src/bundle.mjs packages/harness/src/tools.mjs packages/business/src/vocabulary.mjs; do
+for file in scripts/case2-persona-import.mjs packages/harness/src/bundle.mjs packages/harness/src/tools.mjs packages/business/src/vocabulary.mjs packages/plugins/model-provider/cliproxy/index.mjs; do
   mkdir -p "$4/$2/$(dirname "$file")" && cp '${REPO}'"$file" "$4/$2/$file" || exit 1
 done
 mkdir -p "$4/$2/node_modules" && cp -R '${REPO}node_modules/yaml' "$4/$2/node_modules/yaml"`);

packages/cli/tests/launch-limit-checks.test.mjs

@@ -77,9 +77,8 @@ test('REQ-LIMIT-4 the launcher reaches token-limit only through the bus: launche
  const src=readFileSync(fileURLToPath(new URL('../src/launcher.mjs',import.meta.url)),'utf8');
  assert.doesNotMatch(src,/token-limit/);
  assert.doesNotMatch(src,/limit-check\//);
- // Row 65: the launcher reaches its harness through the plug-in loader, and every load names only
- // the harness kind, so a limit check still can't load here.
- assert.deepEqual([...src.matchAll(/\.load\(([^)]*)\)/g)].map(m=>m[1]),['{ harness: [m.harness] }']);
+ // The launcher loads model providers and harnesses, but never a limit check.
+ assert.deepEqual([...src.matchAll(/\.load\(([^)]*)\)/g)].map(m=>m[1]),["{ 'model-provider': providerNames }",'{ harness: [m.harness] }']);
  assert.doesNotMatch(src,/modules\(\s*["'](?!harness["'])/);
  assert.doesNotMatch(src,/\bimport\s*\(/);
 });

packages/cli/tests/stack-upgrade.test.mjs

@@ -600,6 +600,9 @@ test('#385 R2-F1 controls: library args without --send and a factory args file p
 // checkout's own case2-unit.sh is A4's RENDER, both into an owned unit directory with fake getent and
 // systemctl.
 const BASE_REV='09a847eda';
+// The real renderer requires a root-owned node path; the test runner's
+// process.execPath may be an unprivileged toolchain installation.
+const RENDER_NODE='/usr/bin/node';
 let baseRenderer;
 const baseCase2=()=>{
  if(baseRenderer)return baseRenderer;
@@ -620,7 +623,7 @@ function case2Host(e,dir){
  writeFileSync(join(bin,'systemctl'),`#!/bin/sh\necho "$*" >> '${log}'\n[ "$1" != is-active ] || exit 3\n`,{mode:0o755});
  const env={CASE2_UNIT_DIR:dir,CASE2_GETENT:join(bin,'getent'),CASE2_SYSTEMCTL:join(bin,'systemctl')};
  const serve=['--port','443','--business','acme','--public-host','books.example','--auth-header','X-Auth-User','--allow-user','melanie','--proof-header','X-Proxy-Proof','--proof-file',proof];
- const args=(copy=B,extra=[])=>['--user','fixture','--home',home,'--business','acme','--copy',copy,'--node',NODE,...extra,'--',...serve];
+ const args=(copy=B,extra=[])=>['--user','fixture','--home',home,'--business','acme','--copy',copy,'--node',RENDER_NODE,...extra,'--',...serve];
  const render=(script,a)=>{const r=spawnSync('bash',[script,...a],{env:{PATH:e.env.PATH,...env},encoding:'utf8'});return {code:r.status,out:r.stdout,err:r.stderr}};
  const legacy=()=>{const a=args();a.splice(a.indexOf('melanie')+1,0,'--send');const r=render(baseCase2(),a);assert.equal(r.code,0,r.err);return r};
  const bytes=()=>Object.fromEntries(readdirSync(dir).sort().map((n)=>[n,readFileSync(join(dir,n),'utf8')]));
@@ -632,12 +635,12 @@ function case2Host(e,dir){
 
 test('#772: the real Case 2 renderer re-pins and rolls back all three factory units with the saved --proxy args',async(t)=>{
  const e=await useed(t),h=case2Host(e,join(e.root,'case2','units')),script=join(checkout(C1),'scripts','case2-unit.sh');
- const args=(copy)=>['--user','fixture','--home',h.home,'--business','acme','--copy',copy,'--node',NODE,'--profile','factory','--proxy','--','--port','443','--business','acme'];
+ const args=(copy)=>['--user','fixture','--home',h.home,'--business','acme','--copy',copy,'--node',RENDER_NODE,'--profile','factory','--proxy','--','--port','443','--business','acme'];
  const first=h.render(script,args(B));assert.equal(first.code,0,first.out+first.err);
  const before=h.bytes();
  const next=h.render(script,args(C1));assert.equal(next.code,0,next.out+next.err);
  for(const [u,text] of Object.entries(h.bytes()))assert.ok(text.includes(`WorkingDirectory=${e.hc}/${C1}\n`),u);
- assert.ok(h.bytes()[PROXY].includes(`ExecStart=${NODE} ${e.hc}/${C1}/packages/model-proxy/src/run.mjs\n`));
+ assert.ok(h.bytes()[PROXY].includes(`ExecStart=${RENDER_NODE} ${e.hc}/${C1}/packages/model-proxy/src/run.mjs\n`));
  const changed=h.bytes(),n=h.syslog().length;
  const missing=h.render(script,args(C1).filter((x)=>x!=='--proxy'));assert.equal(missing.code,1,missing.out+missing.err);
  assert.deepEqual(h.bytes(),changed);assert.equal(h.syslog().length,n);
@@ -653,7 +656,7 @@ test('#385 R3-F1: an installed pre-#385 library Console unit with --send refuses
  const e=await useed(t),h=case2Host(e,join(e.root,'case2','units'));
  h.legacy();
  const c=readFileSync(h.con,'utf8');
- assert.ok(c.includes(`\nExecStart=${NODE} ${e.hc}/${B}/packages/webui/src/cli.mjs serve --port 443 --business acme --public-host books.example --auth-header X-Auth-User --allow-user melanie --send --proof-header X-Proxy-Proof --proof-file `),c);
+ assert.ok(c.includes(`\nExecStart=${RENDER_NODE} ${e.hc}/${B}/packages/webui/src/cli.mjs serve --port 443 --business acme --public-host books.example --auth-header X-Auth-User --allow-user melanie --send --proof-header X-Proxy-Proof --proof-file `),c);
  // The operator removed the args file's --send line, as R3's section 7 said, so read_args passes.
  h.use(h.args().join('\n')+'\n');
  assert.ok(!e.argsText.includes('--send'));
@@ -661,7 +664,7 @@ test('#385 R3-F1: an installed pre-#385 library Console unit with --send refuses
  // The premise: the current renderer, as A4 runs it, refuses the installed legacy unit and writes nothing.
  const p=h.render(join(checkout(C1),'scripts','case2-unit.sh'),h.args(C1));
  assert.equal(p.code,1,p.out+p.err);
- assert.ok(p.err.startsWith(`case2-unit: installed ${h.con}: ExecStart= must be exactly once in [Service], as ExecStart=${NODE} ${e.hc}/${B}/packages/webui/src/cli.mjs serve --port 443 `),p.err);
+ assert.ok(p.err.startsWith(`case2-unit: installed ${h.con}: ExecStart= must be exactly once in [Service], as ExecStart=${RENDER_NODE} ${e.hc}/${B}/packages/webui/src/cli.mjs serve --port 443 `),p.err);
  assert.deepEqual(h.bytes(),before);assert.equal(h.syslog().length,sys);
  for(const [why,run] of [['upgrade',()=>upgrade(e,C1)],['rollback',()=>drive(e,B,'rollback','--instance','dev','--to',B,...TO)]]){
   const n=existsSync(e.slog)?slog(e).length:0;

packages/cli/tests/thread-launch.test.mjs

@@ -89,6 +89,7 @@ async function start(t, { f, adapter }, { tickMs = 100, ops: replace } = {}) {
     authorizeLaunch: (m) => runtime.authorizeLaunch(m.cap, m.instance),
     refuse: (m) => runtime.refuse(m.cap, m.code, m.subject),
     endLaunch: (m) => runtime.endLaunch(m.record),
+    recoverLaunch: (m) => runtime.recoverLaunch(m.record),
     credentialStatus: (m) => runtime.credentialStatus(m.business, m.role),
     threadsState: (m) => runtime.threadsState(m.business),
     threadIdle: (m) => runtime.threadIdle(m.record),
@@ -131,16 +132,18 @@ async function start(t, { f, adapter }, { tickMs = 100, ops: replace } = {}) {
     threadTickMs: tickMs,
     log: (l) => logs.push(l),
   });
+  let brokerClose = null;
+  const stopBroker = () => (brokerClose ??= runtime.close());
   const close = () =>
     (closing ??= (async () => {
       for (const fn of hooks) await fn(0);
-      await runtime.close();
+      await stopBroker();
     })());
   t.after(close);
   await launcher.listen();
   const human = runtime.broker.bindHuman({ business: "acme", human: f.doc.human, via: "cli", outsideAgent: true });
   const call = (verb, args = {}) => runtime.broker.request(human, { verb, args });
-  return { host, runtime, launcher, call, logs, close };
+  return { host, runtime, launcher, call, logs, close, stopBroker };
 }
 
 const launchLog = (f) =>
@@ -166,6 +169,54 @@ async function idleStop(f, host, thread, run) {
   assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === thread).holder, null);
 }
 
+for (const alreadyEnded of [false, true]) {
+  test(`a refused end retains recovery evidence when the broker ${alreadyEnded ? "already ended the run" : "exited before ending the run"}`, { skip, timeout: 60000 }, async (t) => {
+    const prepared = prepare(t, { idle: 60 });
+    const { f } = prepared;
+    let brokerGone = false;
+    let host = await start(t, prepared, {
+      tickMs: 3600000,
+      ops: (runtime) => ({
+        endLaunch: (m) => {
+          if (brokerGone) throw new BusError("broker-exited");
+          return runtime.endLaunch(m.record);
+        },
+      }),
+    });
+    const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
+    const thread = host.call("thread.create", { project, name: "recovery" }).id;
+    const first = host.call("thread.send", { thread, body: "first" });
+    await host.launcher.tick();
+    await until(() => host.call("thread.messages", { thread }).find((m) => m.in_reply_to === first.id && m.class === "RESULT"));
+    const entry = threadRuns(f, thread)[0];
+    assert.ok(entry);
+    assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === thread).holder, entry.run);
+    if (alreadyEnded) host.runtime.endLaunch({ business: "acme", run: entry.run, reason: "stopped", exitCode: 0 });
+    await host.stopBroker();
+    brokerGone = true;
+    assert.equal(startTimeOf(entry.pid), entry.startTime);
+    process.kill(entry.pid, "SIGKILL");
+    await until(() => host.logs.some((line) => line.includes(`run ${entry.run}: endLaunch refused: broker-exited`)));
+    await until(() => launchLog(f).some((e) => e.event === "end" && e.run === entry.run));
+    assert.ok(threadRuns(f, thread).some((s) => s.run === entry.run), "unconfirmed end retains the exact recovery entry");
+    await host.close();
+
+    host = await start(t, prepared, { tickMs: 3600000 });
+    assert.equal(startTimeOf(entry.pid), null);
+    assert.deepEqual(threadRuns(f, thread), []);
+    assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === thread).holder, null);
+    const ends = host.call("trail", { subject: entry.run }).filter((e) => e.kind === "session.ended");
+    assert.equal(ends.length, 1);
+    assert.equal(ends[0].body.reason, alreadyEnded ? "stopped" : "host-lost");
+    if (alreadyEnded) assert.ok(host.logs.some((line) => line.includes(`run ${entry.run} (assistant) from an earlier host was already ended`)));
+    const next = host.call("thread.send", { thread, body: "after restart" });
+    await host.launcher.tick();
+    await until(() => host.call("thread.messages", { thread }).find((m) => m.in_reply_to === next.id && m.class === "RESULT"));
+    assert.notEqual(threadRuns(f, thread)[0].run, entry.run);
+    assert.equal(host.call("trail", { subject: entry.run }).filter((e) => e.kind === "session.ended").length, 1);
+  });
+}
+
 test("REQ-THREAD-2 REQ-LAUNCH-1 REQ-MSG-1 a thread message starts the thread's session, which stops when idle; its history survives the stop and a restart and no other thread sees it", { skip, timeout: 90000 }, async (t) => {
   const prepared = prepare(t);
   const { f } = prepared;

packages/seat/tests/sandbox.test.mjs

@@ -109,7 +109,9 @@ test("row 65 a harness module's mounts bind read-only after the adapter, each wi
     const src = join(REPO, "tools", "unslop-hook", f);
     assert.deepEqual(list.find((b) => b.src === src), { kind: "ro", src, why: "the unslop check the session's second extension runs" }, f);
   }
-  assert.equal(binds(args).length, list.length - 4);
+  const stream = join(REPO, "packages", "plugins", "harness", "pi", "model-stream.mjs");
+  assert.deepEqual(list.find((b) => b.src === stream), { kind: "ro", src: stream, why: "the managed model event stream" });
+  assert.equal(binds(args).length, list.length - 5);
 });
 
 // Row 121 (#1615): a thread run gets only its own thread's two directories,