feat(orchestration): a wrap-up turn when a worker stops without reporting

packages/server/src/app-tools/orchestration-tools.ts

@@ -624,6 +624,7 @@ export const OrchestrationAppToolsLive = Layer.effectDiscard(
           if (target.status !== "running" && target.status !== "starting") {
             return { threadId: target.thread.id, stopped: false, status: target.status };
           }
+          if (orchestration !== undefined) yield* orchestration.noteStop(caller, target.thread.id);
           yield* runtimes.interrupt(target.thread.id);
           return { threadId: target.thread.id, stopped: true };
         }).pipe(Effect.mapError(toolError("stop_thread"))),

packages/server/src/app-tools/pr-tools-merge-launch.test.ts

@@ -22,6 +22,7 @@ import {
   publishGit,
   settle,
   spawn,
+  stopSilently,
   waitReal,
   withServer,
   type World,
@@ -217,7 +218,7 @@ describe("merge_when_ready", () => {
             const registry = yield* AppToolRegistry;
             const { session, thread, workspaceId } = yield* openOrchestrator(world);
             const worker = yield* spawn(session, { feature: "voice-desktop" });
-            yield* completeTurn(worker.session);
+            yield* stopSilently(worker.session);
             yield* atAutonomy(thread, "full");
             world.prHeads.set(12, { headRefName: worker.thread.checkout.branch! });
             const tools = registry.forTransport(worker.thread.id, "worker");

packages/server/src/orchestration/feature-cards.test.ts

@@ -12,6 +12,7 @@ import {
   publishGit,
   settle,
   spawn,
+  stopSilently,
   waitReal,
   withServer,
 } from "./test-harness.ts";
@@ -28,7 +29,7 @@ describe("feature cards", () => {
           const orchestration = yield* Orchestration;
           const { session, workspaceId } = yield* openOrchestrator(world);
           const worker = yield* spawn(session, { feature: "mail-trigger", title: "Queue worker" });
-          yield* completeTurn(worker.session);
+          yield* stopSilently(worker.session);
           yield* settle;
 
           const one = yield* orchestration.featureCards(workspaceId);

packages/server/src/orchestration/held-briefs.test.ts

@@ -24,12 +24,12 @@ import { LIMIT_CLEARED_TEXT, LIMIT_RESUME_BATCH_GAP_MS } from "../runtime/limit-
 import { HELD_RETRY_MS, refusalPasses } from "./limit-pause.ts";
 import { Orchestration, type HeldSpawn } from "./service.ts";
 import {
-  completeTurn,
   makeWorld,
   openOrchestrator,
   outboxOf,
   settle,
   spawn,
+  stopSilently,
   waitReal,
   waitSent,
   withServer,
@@ -697,7 +697,7 @@ describe.skipIf(!hasGit)("holding low-priority briefs near a limit", () => {
 
           // Once the busy worker rests archived, the next retry starts it: a turn it still ran
           // would keep its slot.
-          yield* completeTurn(busy.session);
+          yield* stopSilently(busy.session);
           yield* session.callTool("archive_thread", { threadId: busy.thread.id });
           yield* settle;
           yield* TestClock.adjust(HELD_RETRY_MS);

packages/server/src/orchestration/outbox.ts

@@ -155,6 +155,11 @@ export const WorkerMemo = Schema.Struct({
   turnEndsAtReport: Schema.optionalKey(Schema.Int),
   /** The latest unreported stop; a new turn or report supersedes it. */
   quiet: Schema.optionalKey(WorkerQuiet),
+  /**
+   * Its last stop was sent a wrap-up turn (`wrapUpDue`), which the digest waits on rather than
+   * calling the stop silent; the next turn end takes it off, and gets no wrap-up of its own.
+   */
+  wrapUp: Schema.optionalKey(Schema.Boolean),
   /** The advisory `spawn_worker.scope` claim. */
   scope: Schema.optionalKey(Schema.Array(Schema.String)),
   /** Its background work as last reported; absent when it has none. A restart stops all of it. */

packages/server/src/orchestration/rules.ts

@@ -1154,8 +1154,9 @@ const snapshotStates = (snapshot: DigestSnapshot): ReadonlyMap<string, WorkerSta
  * alone would leave the orchestrator waiting on a worker that is waiting on it. `turnEnds` is each
  * worker's count now; one that has grown since the snapshot, on a worker that is at rest and filed
  * no report, is a change. Not when the turn that ended is the one that filed its last report
- * (`turnEndsAtReport`), which the orchestrator has read already, and not for a worker the user is
- * talking to, whose turns are the user's to follow.
+ * (`turnEndsAtReport`), which the orchestrator has read already, not for a worker the user is
+ * talking to, whose turns are the user's to follow, and not while a wrap-up turn sent at that stop
+ * (`wrappingUp`) has yet to end: its report says more, and its own silent end is told instead.
  *
  * `worthWaking` is false when nothing changed and no report is unacknowledged, and for noise alone:
  * a worker arriving in, or moving between, `briefed` and `working` is the orchestrator's own spawn
@@ -1180,6 +1181,8 @@ export const digestDiff = (input: {
   readonly interfaces?: Readonly<Record<string, string>> | undefined;
   /** By thread id, `turnEnds` as of the stop that ended the last turn the worker reported in. */
   readonly turnEndsAtReport?: Readonly<Record<string, number>> | undefined;
+  /** Workers sent a wrap-up turn at their last stop, whose report is still to come (`wrapUpDue`). */
+  readonly wrappingUp?: ReadonlySet<string> | undefined;
   /** The workspace's level, which decides whose a merge is. */
   readonly autonomy?: Autonomy | undefined;
   /** Epoch ms from the service clock, for quiet ages. */
@@ -1234,6 +1237,7 @@ export const digestDiff = (input: {
       const wentQuiet =
         !worker.userIntervened &&
         !worker.reportStands &&
+        input.wrappingUp?.has(worker.threadId) !== true &&
         !(
           AT_REST.has(worker.status) &&
           (worker.waitingOnAutomation === true || (worker.wakes?.length ?? 0) > 0)
@@ -1943,6 +1947,76 @@ export const restartContinuation = (lost: ReadonlyArray<BackgroundWork>): string
       : ` The restart also stopped your background work (${workList(lost)}); start again what you still need.`
   }`;
 
+// ---------------------------------------------------------------------------------------------
+// Wrap-up turns
+// ---------------------------------------------------------------------------------------------
+
+/** What a worker's turn end says about whether it gets a wrap-up turn (`wrapUpDue`). */
+export interface WrapUpFacts {
+  readonly reason: "completed" | "interrupted" | "error" | "maxTurns";
+  /** The host ended the turn itself: a crash, a closed session, the interrupt watchdog. */
+  readonly synthesized: boolean;
+  /** Nothing was queued behind it. */
+  readonly queueEmpty: boolean;
+  /** Its orchestrator stopped it with `stop_thread`. */
+  readonly stopAsked: boolean;
+  /** It called `report` in the turn that ended. */
+  readonly reported: boolean;
+  /** Gram opened the turn to compact the idle thread. */
+  readonly idleCompaction: boolean;
+  /** It is waiting out a usage limit, which resumes it on its own. */
+  readonly limitWait: boolean;
+  /** Archived or marked completed. */
+  readonly parked: boolean;
+  /** The user paused the workspace or its feature. */
+  readonly paused: boolean;
+  /** The user typed into it since its last report and still holds it. */
+  readonly userIntervened: boolean;
+  /**
+   * Something it started is still to come back to it, so the stop is a wait, not silence: a
+   * scheduled wake, a run or PR monitor, a merge when ready, a PR dependency, a background shell,
+   * subagent or workflow, its open `ask_question`, or a request waiting on the user.
+   */
+  readonly pending: ReadonlyArray<
+    "automation" | "dependency" | "background" | "question" | "request"
+  >;
+  /** The turn that ended was itself a wrap-up, so this stop gets none. */
+  readonly afterWrapUp: boolean;
+  /** The ledger reads it as having ended its turn without reporting. */
+  readonly ledgerQuiet: boolean;
+}
+
+/**
+ * Whether a worker's stop earns one wrap-up turn, and why: it ended a turn without a `report`
+ * (`ended`), or its orchestrator stopped it (`stopped`). Never for a stop that isn't silent: one
+ * the user paused, holds or archived, one waiting on a limit or on anything else still to come
+ * back to it (`pending`), one that errored or that a restart cut off, or the stop that ends a
+ * wrap-up turn itself.
+ */
+export const wrapUpDue = (facts: WrapUpFacts): "ended" | "stopped" | undefined => {
+  if (
+    !facts.queueEmpty ||
+    facts.reported ||
+    facts.idleCompaction ||
+    facts.limitWait ||
+    facts.parked ||
+    facts.paused ||
+    facts.userIntervened ||
+    facts.pending.length > 0 ||
+    facts.afterWrapUp ||
+    !facts.ledgerQuiet ||
+    facts.reason === "error"
+  )
+    return undefined;
+  if (facts.stopAsked) return "stopped";
+  // An interrupt nobody here asked for is the user's; a synthesized end is a crash or a restart's.
+  return facts.reason === "interrupted" || facts.synthesized ? undefined : "ended";
+};
+
+/** The one turn a worker is sent when it stops without a word (`wrapUpDue`). */
+export const wrapUpText = (why: "ended" | "stopped"): string =>
+  `${why === "stopped" ? "Your orchestrator stopped you." : "Your turn ended without a `report`."} Don't start new work. Call \`report\` once, saying what is done, what you tried and ruled out, and what remains, then end your turn.`;
+
 // ---------------------------------------------------------------------------------------------
 // Scheduled wakes
 // ---------------------------------------------------------------------------------------------

packages/server/src/orchestration/service-closing.test.ts

@@ -31,6 +31,7 @@ import {
   publishGit,
   settle,
   spawn,
+  stopSilently,
   stallGit,
   waitReal,
   waitSent,
@@ -47,7 +48,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
           const orchestration = yield* Orchestration;
           const { session, workspaceId } = yield* openOrchestrator(world);
           const worker = yield* spawn(session, { feature: "mail-trigger" });
-          yield* completeTurn(worker.session);
+          yield* stopSilently(worker.session);
           const notWorker = yield* orchestration.detach(session.options.threadId).pipe(Effect.flip);
           expect(notWorker).toMatchObject({ _tag: "InvalidState" });
 
@@ -81,7 +82,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
               const { thread, session, workspaceId } = yield* openOrchestrator(world);
               const old = yield* spawn(session, { feature: "old", group: "old-release" });
               const live = yield* spawn(session, { feature: "live" });
-              yield* completeTurn(old.session);
+              yield* stopSilently(old.session);
               yield* publishGit(world, old.thread, {
                 pr: prBadge({ state: "MERGED", checks: "passing" }),
               });
@@ -99,7 +100,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
 
               let count = 2;
               if (snapshotArchived) {
-                yield* completeTurn(live.session);
+                yield* stopSilently(live.session);
                 yield* TestClock.adjust("60 seconds");
                 yield* waitSent(thread.id, ++count);
                 yield* completeTurn(session);
@@ -168,9 +169,9 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         const { thread, session, workspaceId } = yield* openOrchestrator(world);
         const worker = yield* spawn(session, { feature: "old" });
         const other = yield* spawn(session, { feature: "live" });
-        yield* completeTurn(worker.session);
+        yield* stopSilently(worker.session);
         yield* session.callTool("archive_thread", { threadId: worker.thread.id, force: true });
-        yield* completeTurn(other.session);
+        yield* stopSilently(other.session);
         yield* TestClock.adjust("60 seconds");
         yield* waitSent(thread.id, 2);
         yield* completeTurn(session);
@@ -201,12 +202,12 @@ describe.skipIf(!hasGit)("orchestration service", () => {
               const { thread, session, workspaceId } = yield* openOrchestrator(world);
               const worker = yield* spawn(session, { feature: "old" });
               const other = yield* spawn(session, { feature: "live" });
-              yield* completeTurn(worker.session);
+              yield* stopSilently(worker.session);
               yield* session.callTool("archive_thread", {
                 threadId: worker.thread.id,
                 force: true,
               });
-              yield* completeTurn(other.session);
+              yield* stopSilently(other.session);
               yield* TestClock.adjust("60 seconds");
               yield* waitSent(thread.id, 2);
               yield* completeTurn(session);
@@ -304,7 +305,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
             const runtimes = yield* ThreadRuntimes;
             const { thread, session, workspaceId } = yield* openOrchestrator(world);
             const worker = yield* spawn(session, { feature: "old" });
-            yield* completeTurn(worker.session);
+            yield* stopSilently(worker.session);
             yield* publishGit(world, worker.thread, {
               pr: prBadge({ state: "MERGED", checks: "passing" }),
             });
@@ -367,7 +368,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
           // At full, so a Codex worker needs no gateway profile to ask before a push.
           yield* atAutonomy(thread, "full");
           const worker = yield* spawn(session, { feature: "mail-trigger" });
-          yield* completeTurn(worker.session);
+          yield* stopSilently(worker.session);
           yield* TestClock.adjust("30 seconds");
           const events = yield* collectEvents(["thread.upsert", "workspaces.changed"]);
 
@@ -582,7 +583,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         const store = yield* ThreadStore;
         const { session, workspaceId } = yield* openOrchestrator(world);
         const worker = yield* spawn(session, { feature: "mail-trigger" });
-        yield* completeTurn(worker.session);
+        yield* stopSilently(worker.session);
         const newer = JSON.stringify({ version: OUTBOX_VERSION + 1, generation: 9, wakes: {} });
         yield* store.putOutbox(workspaceId, newer);
         // The window's digest surveys first, which sets the row aside before its pause lands.
@@ -832,7 +833,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         const audit = yield* store.listOrchestrationEvents(workspaceId, 0);
         expect(audit.map((event) => event.kind)).toContain("orchestratorEdits");
 
-        yield* completeTurn(worker.session);
+        yield* stopSilently(worker.session);
         yield* TestClock.adjust("60 seconds");
         yield* waitSent(thread.id, 3);
         expect(session.sent[2]?.text).toContain("You changed 1 file in shop yourself");

packages/server/src/orchestration/service-digests.test.ts

@@ -22,6 +22,7 @@ import {
   publishGit,
   settle,
   spawn,
+  stopSilently,
   waitReal,
   waitSent,
   waitStatus,
@@ -38,9 +39,9 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         const { session, workspaceId } = yield* openOrchestrator(world);
         const a = yield* spawn(session, { feature: "mail-trigger", title: "Mail trigger" });
         const b = yield* spawn(session, { feature: "billing", title: "Billing" });
-        yield* completeTurn(a.session);
+        yield* stopSilently(a.session);
         yield* TestClock.adjust("40 seconds");
-        yield* completeTurn(b.session);
+        yield* stopSilently(b.session);
         // The second reason restarted the window: a minute after the first, nothing has gone out.
         yield* TestClock.adjust("40 seconds");
         yield* settle;
@@ -283,7 +284,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         expect(session.sent).toHaveLength(2);
         // The next one ends with no word at all, which is.
         yield* session.callTool("send_message", { threadId: a.thread.id, text: "Go on." });
-        yield* completeTurn(a.session);
+        yield* stopSilently(a.session);
         yield* TestClock.adjust("60 seconds");
         yield* waitSent(orchestrator, 3);
         expect(session.sent[2]!.text).toContain("ended its turn without reporting");
@@ -346,10 +347,11 @@ describe.skipIf(!hasGit)("orchestration service", () => {
             chunk: { type: "tool-output-available", toolCallId: "long", output: "finished" },
           });
           yield* a.session.emit({ type: "backgroundTasks", tasks: [] });
-          yield* completeTurn(a.session);
+          yield* stopSilently(a.session);
           yield* TestClock.adjust("60 seconds");
           yield* waitSent(session.options.threadId, 3);
-          expect(session.sent[2]!.text).toContain("last tool Bash");
+          // The stop told is the wrap-up turn's, which ended with no tool and no word.
+          expect(session.sent[2]!.text).toContain("last tool none");
           expect(
             (session.sent[2]!.notice as OrchestrationDigest).features[0]?.workers[0],
           ).toMatchObject({ quiet: { kind: "turnEnded" } });
@@ -364,7 +366,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         const orchestrator = session.options.threadId;
         const a = yield* spawn(session, { feature: "mail-trigger", title: "Mail trigger" });
         const b = yield* spawn(session, { feature: "billing", title: "Billing" });
-        yield* completeTurn(a.session);
+        yield* stopSilently(a.session);
         yield* publishGit(world, a.thread, {
           pr: prBadge({ state: "MERGED", checks: "passing" }),
         });
@@ -387,7 +389,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         expect((yield* outboxOf(workspaceId)).tidyNoted).toEqual([`archive:${a.thread.id}`]);
 
         // Still true, and already said: the next digest, about another worker, leaves it out.
-        yield* completeTurn(b.session);
+        yield* stopSilently(b.session);
         yield* TestClock.adjust("60 seconds");
         yield* waitSent(orchestrator, 4);
         expect(session.sent[3]!.text).toContain("'Billing' finished a turn");
@@ -558,7 +560,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
           const events = yield* collectEvents(["notify"]);
           const { session, thread, workspaceId } = yield* openOrchestrator(world);
           const worker = yield* spawn(session, { feature: "mail-trigger" });
-          yield* completeTurn(worker.session);
+          yield* stopSilently(worker.session);
           yield* TestClock.adjust("60 seconds");
           yield* waitSent(thread.id, 2);
           const flight = (yield* outboxOf(workspaceId)).inFlight;
@@ -652,7 +654,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         yield* waitSent(thread.id, 2);
         yield* session.emit({ type: "turnEnd", reason: "error", error: "overloaded" });
         yield* waitStatus(thread.id, "error");
-        yield* completeTurn(worker.session);
+        yield* stopSilently(worker.session);
         yield* TestClock.adjust("60 seconds");
         yield* waitSent(thread.id, 3);
         expect(session.sent[2]?.text).toContain("finished a turn");
@@ -664,7 +666,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         // A digest waiting behind a turn that fails isn't failed with it, and goes once it settles.
         yield* runtimes.send(thread.id, { text: "again" });
         yield* waitSent(thread.id, 4);
-        yield* completeTurn(other.session);
+        yield* stopSilently(other.session);
         yield* TestClock.adjust("60 seconds");
         yield* settle;
         yield* session.emit({ type: "turnEnd", reason: "error", error: "overloaded" });
@@ -686,7 +688,7 @@ describe.skipIf(!hasGit)("orchestration service", () => {
         const orchestration = yield* Orchestration;
         const { session, thread, workspaceId } = yield* openOrchestrator(world);
         const worker = yield* spawn(session, { feature: "mail-trigger" });
-        yield* completeTurn(worker.session);
+        yield* stopSilently(worker.session);
         yield* runtimes.send(thread.id, { text: "hello" });
         yield* waitSent(thread.id, 2);
         yield* session.emit({ type: "turnEnd", reason: "error", error: "overloaded" });