Wait for background workers before finish notices and cancel tools on shutdown

extensions/lineage/server.ts

@@ -7,7 +7,7 @@ import type { ConversationId, ConversationRecord, Cursor, EntryId, JsonObject, M
 import type { ModelThinkingLevel } from "@earendil-works/pi-ai"
 import { onCommit } from "../messages/compat.ts"
 import { Started as BoardStarted } from "../messages/addresses.ts"
-import { Standing, Started } from "../subagents/api.ts"
+import { Standing, Started, Subagents } from "../subagents/api.ts"
 import { Branch, Branched, LineageApi, Rooted, Roots } from "./api.ts"
 import { TallyDoc, SubmissionLanded } from "./refs.ts"
 import { readMemory, applyMemoryOps, memoryBrief, memoryIndex, memoryText, memorizeTask, Memorized, MEMORY_RULES, MEMORY_SLUG, LINE_CHARS, BODY_CHARS } from "./memory.ts"
@@ -174,6 +174,24 @@ async function liveTask(tx: Tx, id: ConversationId, kind: string) {
   return false
 }
 
+/** Include background reporters: their deliveries can start another coordinator turn. */
+async function waitForBranchIdle(host: ExtensionHost, id: ConversationId, ctx: Context): Promise<void> {
+  const harness = host.harness()
+  const conversation = await harness.conversation(id, ctx)
+  if (!conversation) return
+  for (;;) {
+    await conversation.waitForIdle(ctx)
+    const state = await harness.snapshot(Subagents, id, ctx)
+    const actions = Object.values(state?.actions ?? {})
+    for (const action of actions) await harness.waitForTask(action.reporter, ctx)
+    for (const child of Object.values(state?.agents ?? {}))
+      await waitForBranchIdle(host, child.conversationId, ctx)
+    await conversation.waitForIdle(ctx)
+    const current = await harness.snapshot(Subagents, id, ctx)
+    if (Object.keys(current?.actions ?? {}).length === actions.length) return
+  }
+}
+
 function lineage(host: ExtensionHost) {
   /**
    * Delivers a new branch's prompt as its first input; created with the branch, so a crash cannot leave it without
@@ -230,7 +248,7 @@ function lineage(host: ExtensionHost) {
               written.resolve()
           })
           try {
-            await (await runtime.conversation(target, ctx))?.waitForIdle(ctx)
+            await waitForBranchIdle(host, target, ctx)
             // A fork inherits its source's entries; only an answer written in the target itself counts as its own.
             const view = await runtime.context(target, ctx)
             said = ''

host/js/bundle.js

@@ -68776,13 +68776,23 @@ function environment(cwd = "/", conversation, invocation) {
           if (!appended.ok) throw new Error(appended.error.message);
         }
       };
+      let rid;
+      const abort = () => {
+        if (rid !== void 0) {
+          Deno.core.close(rid);
+          rid = void 0;
+        }
+      };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.throwIfAborted();
+        rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.addEventListener("abort", abort, { once: true });
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result;
         while (true) {
           const chunk = await ops.op_exec_next(rid);
           if (!chunk.stream) {
+            rid = void 0;
             result = chunk;
             break;
           }
@@ -68796,6 +68806,9 @@ function environment(cwd = "/", conversation, invocation) {
         return ok({ exitCode: result.exitCode, ...spillPath ? { spillPath } : {} });
       } catch (error) {
         return { ok: false, error: Object.assign(new Error(String(error)), { code: "unknown", ...spillPath ? { spillPath } : {} }) };
+      } finally {
+        context.abortSignal?.removeEventListener("abort", abort);
+        abort();
       }
     },
     cleanup: async () => {
@@ -69812,6 +69825,11 @@ async function serve(harness2, context, afterSubmit, api2) {
   };
   while (true) {
     const input = await Deno.core.ops.op_input();
+    if (input?.type === "shutdown") {
+      await harness2.close(context);
+      await Promise.all(pending);
+      return;
+    }
     if (input === null || input.type === "stop") {
       await harness2.waitForIdle(context);
       await Promise.all(pending);

host/js/chain-test-bundle.js

@@ -21374,13 +21374,23 @@ function environment(cwd = "/", conversation, invocation) {
           if (!appended.ok) throw new Error(appended.error.message);
         }
       };
+      let rid;
+      const abort = () => {
+        if (rid !== void 0) {
+          Deno.core.close(rid);
+          rid = void 0;
+        }
+      };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.throwIfAborted();
+        rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.addEventListener("abort", abort, { once: true });
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result;
         while (true) {
           const chunk = await ops.op_exec_next(rid);
           if (!chunk.stream) {
+            rid = void 0;
             result = chunk;
             break;
           }
@@ -21394,6 +21404,9 @@ function environment(cwd = "/", conversation, invocation) {
         return ok({ exitCode: result.exitCode, ...spillPath ? { spillPath } : {} });
       } catch (error) {
         return { ok: false, error: Object.assign(new Error(String(error)), { code: "unknown", ...spillPath ? { spillPath } : {} }) };
+      } finally {
+        context.abortSignal?.removeEventListener("abort", abort);
+        abort();
       }
     },
     cleanup: async () => {

host/js/environment.ts

@@ -119,13 +119,17 @@ export function environment(cwd = '/', conversation?: string | number | string[]
           if (!appended.ok) throw new Error(appended.error.message);
         }
       };
+      let rid: number | undefined;
+      const abort = () => { if (rid !== undefined) { Deno.core.close(rid); rid = undefined; } };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, undefined, undefined, conversations, tag);
+        context.abortSignal?.throwIfAborted();
+        rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, undefined, undefined, conversations, tag);
+        context.abortSignal?.addEventListener('abort', abort, { once: true });
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result;
         while (true) {
           const chunk = await ops.op_exec_next(rid);
-          if (!chunk.stream) { result = chunk; break; }
+          if (!chunk.stream) { rid = undefined; result = chunk; break; }
           await output(decoders[chunk.stream].decode(new Uint8Array(chunk.bytes), { stream: true }), chunk.stream);
         }
         if (result.error) throw new Error(result.error);
@@ -135,6 +139,7 @@ export function environment(cwd = '/', conversation?: string | number | string[]
         }
         return ok({ exitCode: result.exitCode, ...(spillPath ? { spillPath } : {}) });
       } catch (error) { return { ok: false, error: Object.assign(new Error(String(error)), { code: 'unknown', ...(spillPath ? { spillPath } : {}) }) }; }
+      finally { context.abortSignal?.removeEventListener('abort', abort); abort(); }
     },
     cleanup: async () => {},
   };

host/js/foundation-test.ts

@@ -2,9 +2,11 @@
 import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
 import { createModels } from '@earendil-works/pi-ai/models';
 import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
-import { Harness, createRegistry, defineDoc, defineExtension, defineTask, GenerationTask, hook, InboxDoc } from '@earendil-works/pi-durable';
+import { Harness, createRegistry, defineDoc, defineExtension, defineTask, GenerationTask, hook, InboxDoc, defineTool } from '@earendil-works/pi-durable';
 import { openStorage } from './sqlite';
 import { serve } from './input';
+import { environment } from './environment';
+import { obj } from '@earendil-works/pi-schema';
 const assert = (value: any, message: string) => { if (!value) throw new Error(message); };
 const equal = (a: any, b: any) => assert(JSON.stringify(a) === JSON.stringify(b), `Expected ${JSON.stringify(b)}, got ${JSON.stringify(a)}`);
 const until = async (condition: () => any) => {
@@ -47,11 +49,24 @@ models.setProvider(faux.provider);
 const registry = createRegistry();
 registry.install(Notes);
 registry.install(defineExtension({ name: 'hold-test', tasks: [Park] }));
+const Shutdown = defineExtension({name: 'shutdown-test', tools: [defineTool({name: 'long_tool', description: 'Run a long world process', parameters: obj({}), replay: 'safe', execute: async (_args, _api, ctx) => {
+  const env = environment('/');
+  const resume = (await env.readTextFile('/resume')).ok;
+  const work = env.exec(['/bin/sleep.wasm', resume ? '0' : '600'], {}, ctx);
+  if (!resume) {
+    await until(async () => (await env.readTextFile('/proc/processes')).value?.includes('600'));
+    await env.writeFile('/sleep-ready', 'yes');
+  }
+  const result = await work;
+  assert(result.ok, `Sleep failed: ${JSON.stringify(result)}`);
+  return {output: [{type: 'text', text: 'sleep completed'}]};
+}})]});
+registry.install(Shutdown);
 const storage = await openStorage(globalThis.storagePath);
 const harness = await Harness.open(storage, { models, registry, settings: { followUpMode: 'all' } }, context);
 const scenario = globalThis.foundationCase;
 const root = await harness.root(context, { agent: { model: { provider: 'faux', modelId: 'faux-1' },
-  extensions: scenario === 'placement' ? [Notes] : [] } });
+  extensions: scenario === 'placement' ? [Notes] : scenario === 'shutdown' ? [Shutdown] : [] } });
 try {
   if (scenario === 'placement' || scenario === 'followups') {
     faux.setResponses([async () => {
@@ -113,6 +128,20 @@ try {
     await until(() => second);
     await harness.waitForIdle(context);
     equal((await harness.getTask(task, context))?.state.status, 'terminal');
+  } else if (scenario === 'shutdown') {
+    faux.setResponses([fauxAssistantMessage([fauxToolCall('long_tool', {})], {stopReason: 'toolUse'}), fauxAssistantMessage('resumed answer')]);
+    const resumed = (await environment('/').readTextFile('/resume')).ok;
+    if (!resumed) await root.submit({type: 'input', content: 'sleep now', requestId: 'sleep-request'}, context);
+    harness.resume();
+    if (resumed) {
+      faux.setResponses([fauxAssistantMessage('resumed answer')]);
+      await harness.waitForIdle(context);
+      const transcript = JSON.stringify(await root.context(context));
+      assert(transcript.includes('resumed answer'), 'Shutdown did not resume');
+      assert(transcript.includes('sleep completed'), 'Shutdown did not replay the interrupted tool');
+      const submissions = await storage.scanSubmissions({conversationId: root.id}, 100, undefined, context);
+      equal(submissions.items.length, 1);
+    } else await serve(harness, context);
   } else if (scenario === 'headless') {
     faux.setResponses([fauxAssistantMessage('channel answer')]);
     await serve(harness, context);

host/js/harness-api-test-bundle.js

@@ -31971,6 +31971,11 @@ async function serve(harness2, context, afterSubmit, api2) {
   };
   while (true) {
     const input = await Deno.core.ops.op_input();
+    if (input?.type === "shutdown") {
+      await harness2.close(context);
+      await Promise.all(pending);
+      return;
+    }
     if (input === null || input.type === "stop") {
       await harness2.waitForIdle(context);
       await Promise.all(pending);

host/js/input.ts

@@ -1,4 +1,4 @@
-// Channel submissions target any conversation. EOF and Stop close the harness cleanly.
+// EOF and Stop drain submissions. Shutdown cancels invocations and preserves durable tasks for restart.
 export async function serve(harness: any, context: any, afterSubmit?: () => Promise<void>, api?: any) {
   const pending = new Set<Promise<void>>();
   const track = (work: Promise<void>) => {
@@ -7,6 +7,12 @@ export async function serve(harness: any, context: any, afterSubmit?: () => Prom
   };
   while (true) {
     const input = await Deno.core.ops.op_input();
+    if (input?.type === 'shutdown') {
+      // Close cancels invocations without terminal outcomes, so durable tasks resume after restart.
+      await harness.close(context);
+      await Promise.all(pending);
+      return;
+    }
     if (input === null || input.type === 'stop') {
       await harness.waitForIdle(context);
       await Promise.all(pending);

host/js/messages-test-bundle.js

@@ -31706,6 +31706,13 @@ var harness = await Harness.open(storage, { models, registry, settings: { get ex
 var root = await harness.root(BACKGROUND_CONTEXT, { agent: { model: { provider: "faux", modelId: "faux-1" }, cwd: "/src/sausage" } });
 var gate = Promise.withResolvers();
 var holding = /* @__PURE__ */ new Set();
+var workerGate = Promise.withResolvers();
+var workerStarted = false;
+var worker = defineExtension({ name: "worker-hold", tools: [defineTool({ name: "worker_hold", description: "Hold a background worker", parameters: obj({}), replay: "safe", execute: async (_args, _api, ctx) => {
+  workerStarted = true;
+  await awaitWithContext(workerGate.promise, ctx);
+  return { output: [{ type: "text", text: "worker released" }] };
+} })] });
 var held2 = defineExtension({ name: "hold", tools: [defineTool({ name: "hold", description: "Wait for the test gate", parameters: obj({}), replay: "safe", execute: async (_args, api, ctx) => {
   holding.add(api.conversationId);
   await awaitWithContext(gate.promise, ctx);
@@ -31748,11 +31755,12 @@ var route = (transcript) => {
     const input = JSON.parse(text.slice(5));
     return fauxAssistantMessage([fauxToolCall(input.name, input.args)], { stopReason: "toolUse" });
   }
+  if (last.role === "user" && text === "HOLD WORKER") return fauxAssistantMessage([fauxToolCall("worker_hold", {})], { stopReason: "toolUse" });
   if (last.role === "user" && text === "HOLD") return fauxAssistantMessage([fauxToolCall("hold", {})], { stopReason: "toolUse" });
   return fauxAssistantMessage(last.role === "user" && /^(Reply|Message|Note) from/.test(text) ? "Received" : `Answer: ${text}`);
 };
 faux.setResponses([route]);
-var loader = await loadExtensions(harness, registry, storage, models, BACKGROUND_CONTEXT, [held2]);
+var loader = await loadExtensions(harness, registry, storage, models, BACKGROUND_CONTEXT, [held2, worker]);
 assert(!loader.state().error, `Extension load: ${loader.state().error}`);
 var snapshot = await fs.read("/src/sausage/.git/sausage-last-good");
 var base = `file:///run/sausage/extensions/${snapshot}/extensions`;
@@ -31885,13 +31893,23 @@ try {
     assert((await entries2(branch)).some((e) => e.kind === "lineage.note" && e.data.source === "main" && e.data.text.includes("Catch-up commit") && e.data.text.includes(`#${branch.id}`)), "Attributed commit catch-up");
     assert((await entries2(branch)).some((e) => e.kind === "lineage.note" && e.data.source === "main" && e.data.text.includes("Use memory")), "Tally catch-up must include the new branch");
     await harness.waitForIdle(BACKGROUND_CONTEXT);
+    await tool(branch, "subagent", { action: "start", agent: "coordinator", task: "HOLD WORKER", background: true });
+    await until2(() => workerStarted);
+    await branch.waitForIdle(BACKGROUND_CONTEXT);
     await tool(primary, "notify", { id: branch.id, wake: false });
+    await new Promise((resolve2) => setTimeout(resolve2, 50));
+    assert(!(await entries2(primary)).some((e) => e.kind === "lineage.note" && e.data.source === "finished"), "Idle coordinator with a live worker was called finished");
+    workerGate.resolve();
     await until2(async () => (await entries2(primary)).some((e) => e.kind === "lineage.note" && e.data.source === "finished"));
     await harness.waitForIdle(BACKGROUND_CONTEXT);
+    const finished = (await entries2(primary)).filter((e) => e.kind === "lineage.note" && e.data.source === "finished");
+    assert(finished.length === 1, "Worker completion must notify once");
+    assert(finished[0].data.text.includes("worker released"), "Notice must contain the coordinator answer after the worker report");
     assert((await harness.snapshot(Roots, root.id, BACKGROUND_CONTEXT)).roots.includes(root.id), "Listed root");
   }
 } finally {
   gate.resolve();
+  workerGate.resolve();
   await loader.close();
   await harness.close(BACKGROUND_CONTEXT);
 }

host/js/messages-test.ts

@@ -16,6 +16,9 @@ const registry = createRegistry(), storage = await openStorage(globalThis.storag
 const harness = await Harness.open(storage, {models, registry, settings: {get extensions() {return registry.snapshot().installed()}}}, context);
 const root = await harness.root(context, {agent: {model: {provider: 'faux', modelId: 'faux-1'}, cwd: '/src/sausage'}});
 const gate = Promise.withResolvers<void>(), holding = new Set<number>();
+const workerGate = Promise.withResolvers<void>();
+let workerStarted = false;
+const worker = defineExtension({name: 'worker-hold', tools: [defineTool({name: 'worker_hold', description: 'Hold a background worker', parameters: obj({}), replay: 'safe', execute: async (_args, _api, ctx) => {workerStarted = true; await awaitWithContext(workerGate.promise, ctx); return {output: [{type: 'text', text: 'worker released'}]}}})]});
 const held = defineExtension({name: 'hold', tools: [defineTool({name: 'hold', description: 'Wait for the test gate', parameters: obj({}), replay: 'safe', execute: async (_args, api, ctx) => {holding.add(api.conversationId); await awaitWithContext(gate.promise, ctx); return {output: [{type: 'text', text: 'released'}]}}})]});
 if (globalThis.messagesMode === 'kill') {
   const install = registry.install.bind(registry);
@@ -41,11 +44,12 @@ const route = (transcript: any): any => {
   const last = transcript.messages.findLast((m: any) => m.role !== 'system');
   const text = typeof last.content === 'string' ? last.content : last.content.filter((p: any) => p.type === 'text').map((p: any) => p.text).join('');
   if (last.role === 'user' && text.startsWith('CALL ')) {const input = JSON.parse(text.slice(5)); return fauxAssistantMessage([fauxToolCall(input.name, input.args)], {stopReason: 'toolUse'})}
+  if (last.role === 'user' && text === 'HOLD WORKER') return fauxAssistantMessage([fauxToolCall('worker_hold', {})], {stopReason: 'toolUse'});
   if (last.role === 'user' && text === 'HOLD') return fauxAssistantMessage([fauxToolCall('hold', {})], {stopReason: 'toolUse'});
   return fauxAssistantMessage(last.role === 'user' && /^(Reply|Message|Note) from/.test(text) ? 'Received' : `Answer: ${text}`);
 };
 faux.setResponses([route]);
-const loader = await loadExtensions(harness, registry, storage, models, context, [held]);
+const loader = await loadExtensions(harness, registry, storage, models, context, [held, worker]);
 assert(!loader.state().error, `Extension load: ${loader.state().error}`);
 const snapshot = await fs.read('/src/sausage/.git/sausage-last-good');
 const base = `file:///run/sausage/extensions/${snapshot}/extensions`;
@@ -162,9 +166,19 @@ try {
     assert((await entries(branch)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'main' && e.data.text.includes('Catch-up commit') && e.data.text.includes(`#${branch.id}`)), 'Attributed commit catch-up');
     assert((await entries(branch)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'main' && e.data.text.includes('Use memory')), 'Tally catch-up must include the new branch');
     await harness.waitForIdle(context);
+    await tool(branch, 'subagent', {action: 'start', agent: 'coordinator', task: 'HOLD WORKER', background: true});
+    await until(() => workerStarted);
+    await branch.waitForIdle(context);
     await tool(primary, 'notify', {id: branch.id, wake: false});
+    // Allow the notify invocation to reach its wait. The active worker must prevent delivery.
+    await new Promise(resolve => setTimeout(resolve, 50));
+    assert(!(await entries(primary)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'finished'), 'Idle coordinator with a live worker was called finished');
+    workerGate.resolve();
     await until(async () => (await entries(primary)).some((e: any) => e.kind === 'lineage.note' && e.data.source === 'finished'));
     await harness.waitForIdle(context);
+    const finished = (await entries(primary)).filter((e: any) => e.kind === 'lineage.note' && e.data.source === 'finished');
+    assert(finished.length === 1, 'Worker completion must notify once');
+    assert(finished[0].data.text.includes('worker released'), 'Notice must contain the coordinator answer after the worker report');
     assert((await harness.snapshot(Roots, root.id, context)).roots.includes(root.id), 'Listed root');
   }
-} finally {gate.resolve(); await loader.close(); await harness.close(context)}
+} finally {gate.resolve(); workerGate.resolve(); await loader.close(); await harness.close(context)}

host/js/p2-test-bundle.js

@@ -21353,13 +21353,23 @@ function environment(cwd = "/", conversation, invocation) {
           if (!appended.ok) throw new Error(appended.error.message);
         }
       };
+      let rid;
+      const abort = () => {
+        if (rid !== void 0) {
+          Deno.core.close(rid);
+          rid = void 0;
+        }
+      };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.throwIfAborted();
+        rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.addEventListener("abort", abort, { once: true });
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result;
         while (true) {
           const chunk = await ops.op_exec_next(rid);
           if (!chunk.stream) {
+            rid = void 0;
             result = chunk;
             break;
           }
@@ -21373,6 +21383,9 @@ function environment(cwd = "/", conversation, invocation) {
         return ok({ exitCode: result.exitCode, ...spillPath ? { spillPath } : {} });
       } catch (error) {
         return { ok: false, error: Object.assign(new Error(String(error)), { code: "unknown", ...spillPath ? { spillPath } : {} }) };
+      } finally {
+        context.abortSignal?.removeEventListener("abort", abort);
+        abort();
       }
     },
     cleanup: async () => {

host/js/test-bundle.js

@@ -21353,13 +21353,23 @@ function environment(cwd = "/", conversation, invocation) {
           if (!appended.ok) throw new Error(appended.error.message);
         }
       };
+      let rid;
+      const abort = () => {
+        if (rid !== void 0) {
+          Deno.core.close(rid);
+          rid = void 0;
+        }
+      };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.throwIfAborted();
+        rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.addEventListener("abort", abort, { once: true });
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result2;
         while (true) {
           const chunk = await ops.op_exec_next(rid);
           if (!chunk.stream) {
+            rid = void 0;
             result2 = chunk;
             break;
           }
@@ -21373,6 +21383,9 @@ function environment(cwd = "/", conversation, invocation) {
         return ok({ exitCode: result2.exitCode, ...spillPath ? { spillPath } : {} });
       } catch (error) {
         return { ok: false, error: Object.assign(new Error(String(error)), { code: "unknown", ...spillPath ? { spillPath } : {} }) };
+      } finally {
+        context.abortSignal?.removeEventListener("abort", abort);
+        abort();
       }
     },
     cleanup: async () => {

host/js/undo-harness-test-bundle.js

@@ -21353,13 +21353,23 @@ function environment(cwd = "/", conversation, invocation) {
           if (!appended.ok) throw new Error(appended.error.message);
         }
       };
+      let rid;
+      const abort = () => {
+        if (rid !== void 0) {
+          Deno.core.close(rid);
+          rid = void 0;
+        }
+      };
       try {
-        const rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.throwIfAborted();
+        rid = ops.op_exec(command, options?.cwd ?? cwd, options?.timeout, void 0, void 0, conversations, tag);
+        context.abortSignal?.addEventListener("abort", abort, { once: true });
         const decoders = { stdout: new TextDecoder(), stderr: new TextDecoder() };
         let result2;
         while (true) {
           const chunk = await ops.op_exec_next(rid);
           if (!chunk.stream) {
+            rid = void 0;
             result2 = chunk;
             break;
           }
@@ -21373,6 +21383,9 @@ function environment(cwd = "/", conversation, invocation) {
         return ok({ exitCode: result2.exitCode, ...spillPath ? { spillPath } : {} });
       } catch (error) {
         return { ok: false, error: Object.assign(new Error(String(error)), { code: "unknown", ...spillPath ? { spillPath } : {} }) };
+      } finally {
+        context.abortSignal?.removeEventListener("abort", abort);
+        abort();
       }
     },
     cleanup: async () => {

host/src/harness/input.rs

@@ -24,6 +24,7 @@ pub enum Input {
         reply: mpsc::UnboundedSender<Value>,
     },
     Stop,
+    Shutdown,
 }
 
 pub type Sender = mpsc::UnboundedSender<Input>;
@@ -78,7 +79,8 @@ pub(super) fn console(sender: Sender) {
     });
 }
 
-pub(super) fn stop_signal(
+/// Register SIGTERM and SIGINT before the harness starts.
+pub fn stop_signal(
     sender: Sender,
 ) -> std::io::Result<impl std::future::Future<Output = ()> + Send> {
     // Register before V8 starts, so a stop cannot arrive before the listener.
@@ -89,6 +91,6 @@ pub(super) fn stop_signal(
             _ = interrupt.recv() => {}
             _ = terminate.recv() => {}
         }
-        let _ = sender.send(Input::Stop);
+        let _ = sender.send(Input::Shutdown);
     })
 }

host/tests/harness_shutdown.rs

@@ -0,0 +1,135 @@
+//! SIGTERM cancels a world tool and preserves its durable invocation for restart.
+use sausage::{
+    cred::Cred,
+    harness::{input, run_module_with_input},
+    world::World,
+};
+use std::{
+    path::Path,
+    process::{Command, Stdio},
+    sync::Arc,
+    time::{Duration, Instant},
+};
+
+#[tokio::test(flavor = "current_thread")]
+async fn shutdown_child() -> anyhow::Result<()> {
+    let Ok(path) = std::env::var("SHUTDOWN_WORLD") else {
+        return Ok(());
+    };
+    let world = World::open(Path::new(&path))?;
+    let processes = world.processes.clone();
+    let cred = Arc::new(Cred::open(&world.host_dir)?);
+    let exec = sausage::process::exec(World {
+        root: world.root.clone(),
+        host_dir: world.host_dir.clone(),
+        processes: processes.clone(),
+    })?;
+    let (sender, receiver) = input::channel();
+    let signal = tokio::spawn(input::stop_signal(sender)?);
+    run_module_with_input(
+        world,
+        exec,
+        cred,
+        receiver,
+        concat!(
+            "globalThis.foundationCase = 'shutdown';\n",
+            include_str!("../js/foundation-test-bundle.js")
+        ),
+    )
+    .await?;
+    signal.abort();
+    tokio::time::timeout(Duration::from_secs(1), async {
+        while !processes.lock().unwrap().is_empty() {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await?;
+    Ok(())
+}
+
+#[test]
+fn sigterm_cancels_tool_and_resumes() -> anyhow::Result<()> {
+    let source = Path::new(env!("CARGO_MANIFEST_DIR")).join("..");
+    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap();
+    assert!(
+        Command::new("cargo")
+            .args([
+                "build",
+                "--release",
+                "--locked",
+                "-p",
+                "sausage-coreutils",
+                "--bin",
+                "sleep",
+                "--manifest-path",
+            ])
+            .arg(source.join("userland/Cargo.toml"))
+            .args(["--target", "wasm32-wasip2", "--target-dir"])
+            .arg(target)
+            .args(["-j", "2"])
+            .status()?
+            .success()
+    );
+    let dir = tempfile::tempdir()?;
+    World::open(dir.path())?;
+    std::fs::create_dir_all(dir.path().join("world/bin"))?;
+    std::fs::copy(
+        target.join("wasm32-wasip2/release/sleep.wasm"),
+        dir.path().join("world/bin/sleep.wasm"),
+    )?;
+    let spawn = || {
+        Command::new(std::env::current_exe().unwrap())
+            .args(["--exact", "shutdown_child", "--nocapture"])
+            .env("SHUTDOWN_WORLD", dir.path())
+            .stdin(Stdio::null())
+            .spawn()
+    };
+    let mut child = spawn()?;
+    let deadline = Instant::now() + Duration::from_secs(60);
+    while !dir.path().join("world/sleep-ready").exists() {
+        if let Some(status) = child.try_wait()? {
+            anyhow::bail!("Tool child exited early: {status}");
+        }
+        if Instant::now() > deadline {
+            child.kill()?;
+            child.wait()?;
+            anyhow::bail!("World sleep did not start");
+        }
+        std::thread::yield_now();
+    }
+    assert!(
+        Command::new("kill")
+            .args(["-TERM", &child.id().to_string()])
+            .status()?
+            .success()
+    );
+    let deadline = Instant::now() + Duration::from_secs(5);
+    loop {
+        if let Some(status) = child.try_wait()? {
+            assert!(status.success());
+            break;
+        }
+        if Instant::now() > deadline {
+            child.kill()?;
+            child.wait()?;
+            anyhow::bail!("SIGTERM exceeded five seconds");
+        }
+        std::thread::yield_now();
+    }
+    std::fs::write(dir.path().join("world/resume"), "yes")?;
+    let mut child = spawn()?;
+    let deadline = Instant::now() + Duration::from_secs(20);
+    loop {
+        if let Some(status) = child.try_wait()? {
+            assert!(status.success());
+            break;
+        }
+        if Instant::now() > deadline {
+            child.kill()?;
+            child.wait()?;
+            anyhow::bail!("Restart did not complete");
+        }
+        std::thread::yield_now();
+    }
+    Ok(())
+}