Merge pull request #1066 from corbitsdev/cl-7989-deliver-stashed-steer-when-interrupt-loses-the-race-with-run

Deliver a stashed steer as a follow-up when the run wins the race

Author: TheGreatAxios

src/subagent/lifecycle-tools.test.ts

@@ -950,7 +950,7 @@ describe("send_input", () => {
     expect(sessions.get(missing.id)?.lifecycleStatus).toBe("running");
   });
 
-  test("completion during interrupt keeps the original terminal report", async () => {
+  test("completion during interrupt delivers the stashed steer as a follow-up", async () => {
     const sessions = createSubAgentSessionStore();
     const fleetRecords = createFleetMailbox(sessions);
     const worker = sessions.start({
@@ -978,7 +978,7 @@ describe("send_input", () => {
       message: "late interrupt",
       interrupt: true,
     });
-    expect(result).toEqual({ agent_id: worker.id, status: "completed" });
+    expect(result).toEqual({ agent_id: worker.id, status: "interrupted" });
 
     const collected = await callTool(wait, {
       targets: [worker.id],
@@ -989,10 +989,16 @@ describe("send_input", () => {
       expect.objectContaining({
         agent_id: worker.id,
         status: "done",
-        report: "original report",
+        report: "follow-up report",
       }),
     ]);
-    expect(followupStarted).toBe(false);
+    expect(followupStarted).toBe(true);
+    expect(sessions.get(worker.id)?.entries).toContainEqual(
+      expect.objectContaining({
+        kind: "report",
+        content: expect.stringContaining("original report"),
+      }),
+    );
   });
 
   test("CL-7344: interrupt:true stashes until attachReport; resume stays fail-closed", async () => {

src/subagent/session-store.test.ts

@@ -1836,32 +1836,30 @@ describe("CL-7344 follow-up stash", () => {
     expect(store.isRunInFlight(session.id)).toBe(true);
   });
 
-  test("complete drops a stashed follow-up and keeps the original report", async () => {
+  test("complete on a resumable session delivers the stash and preserves the original report", async () => {
     const store = createSubAgentSessionStore();
     const started: string[] = [];
     const failures: unknown[] = [];
     const session = runningRetained(store, async (message) => {
       started.push(message);
-      return "should not run";
+      return `reply to ${message}`;
     });
     const onFail = (err: unknown): void => {
       failures.push(err);
     };
     store.sendInputOne(session.id, "steer one", { interrupt: true, onFail });
     store.sendInputOne(session.id, "steer two", { interrupt: true, onFail });
     store.complete(session.id, "## Summary\nOriginal done.");
-    await Promise.resolve();
-    expect(started).toEqual([]);
-    expect(store.get(session.id)?.report).toBe("## Summary\nOriginal done.");
-    expect(store.get(session.id)?.lifecycleStatus).toBe("completed");
-    expect(store.isRunInFlight(session.id)).toBe(false);
-    expect(failures).toHaveLength(2);
-    expect(String(defined(failures[0]))).toContain("steer one");
-    expect(String(defined(failures[1]))).toContain("steer two");
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(started).toEqual(["steer one", "steer two"]);
+    expect(failures).toEqual([]);
+    expect(store.get(session.id)?.lifecycle.state).toBe("completed");
+    expect(store.get(session.id)?.report).toBe("reply to steer two");
     expect(store.get(session.id)?.entries).toContainEqual(
       expect.objectContaining({
         kind: "report",
-        content: expect.stringContaining("steer two"),
+        content: expect.stringContaining("Original done"),
       }),
     );
   });
@@ -1921,7 +1919,17 @@ describe("CL-7344 follow-up stash", () => {
   test("dropped steers report which message was lost and why", async () => {
     const store = createSubAgentSessionStore();
     const failures: unknown[] = [];
-    const session = runningRetained(store, async () => "x");
+    // No retained:true: the session cannot resume, so a run-completion win
+    // surfaces the queued steer as a loss instead of delivering it.
+    const session = store.start({
+      description: "d",
+      agentId: "a",
+      brief: "b",
+    });
+    store.markRunning(session.id);
+    store.markRunInFlight(session.id);
+    store.registerInterrupt(session.id, () => undefined);
+    store.registerFollowup(session.id, async () => "x");
     store.sendInputOne(session.id, "steer now", {
       interrupt: true,
       onFail: (err) => {
@@ -2052,4 +2060,175 @@ describe("CL-7344 follow-up stash", () => {
     await new Promise((resolve) => setTimeout(resolve, 0));
     expect(store.get(session.id)?.lifecycle.state).toBe("shutdown");
   });
+
+  test("CL-7989 interrupt wins the race: stashed steer launches from attachReport", async () => {
+    const store = createSubAgentSessionStore();
+    const started: string[] = [];
+    const failures: unknown[] = [];
+    const replies: string[] = [];
+    const session = runningRetained(store, async (message) => {
+      started.push(message);
+      return "followup reply";
+    });
+    store.sendInputOne(session.id, "steer now", {
+      interrupt: true,
+      onFail: (err: unknown) => {
+        failures.push(err);
+      },
+      onFollowupReply: (reply: string) => {
+        replies.push(reply);
+      },
+    });
+    store.attachReport(session.id, "salvage", { stopReason: "interrupted" });
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(started).toEqual(["steer now"]);
+    expect(failures).toEqual([]);
+    expect(replies).toEqual(["followup reply"]);
+    expect(store.get(session.id)?.lifecycle.state).toBe("completed");
+    expect(store.get(session.id)?.report).toBe("followup reply");
+  });
+
+  test("CL-7989 run-completion wins the race: stashed steer delivers as a fresh follow-up", async () => {
+    const store = createSubAgentSessionStore();
+    const started: string[] = [];
+    const failures: unknown[] = [];
+    const replies: string[] = [];
+    const session = runningRetained(store, async (message) => {
+      started.push(message);
+      return "followup reply";
+    });
+    store.sendInputOne(session.id, "steer now", {
+      interrupt: true,
+      onFail: (err: unknown) => {
+        failures.push(err);
+      },
+      onFollowupReply: (reply: string) => {
+        replies.push(reply);
+      },
+    });
+    store.complete(session.id, "## Summary\nOriginal done.");
+    expect(store.get(session.id)?.lifecycleStatus).toBe("running");
+    expect(store.isRunInFlight(session.id)).toBe(true);
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(started).toEqual(["steer now"]);
+    expect(failures).toEqual([]);
+    expect(replies).toEqual(["followup reply"]);
+    expect(store.get(session.id)?.lifecycle.state).toBe("completed");
+    expect(store.get(session.id)?.report).toBe("followup reply");
+    expect(store.get(session.id)?.entries).toContainEqual(
+      expect.objectContaining({
+        kind: "report",
+        content: expect.stringContaining("Original done"),
+      }),
+    );
+  });
+
+  test("CL-7989 run-completion wins with queued steers: all deliver in order", async () => {
+    const store = createSubAgentSessionStore();
+    const started: string[] = [];
+    const failures: unknown[] = [];
+    const session = runningRetained(store, async (message) => {
+      started.push(message);
+      return `reply to ${message}`;
+    });
+    const onFail = (err: unknown): void => {
+      failures.push(err);
+    };
+    store.sendInputOne(session.id, "steer one", { interrupt: true, onFail });
+    store.sendInputOne(session.id, "steer two", { interrupt: true, onFail });
+    store.complete(session.id, "## Summary\nOriginal done.");
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(started).toEqual(["steer one", "steer two"]);
+    expect(failures).toEqual([]);
+    expect(store.get(session.id)?.lifecycle.state).toBe("completed");
+    expect(store.get(session.id)?.report).toBe("reply to steer two");
+  });
+
+  test("CL-7989 run-completion wins on a non-retained session: loss is surfaced", async () => {
+    const store = createSubAgentSessionStore();
+    const started: string[] = [];
+    const failures: unknown[] = [];
+    const session = store.start({
+      description: "d",
+      agentId: "a",
+      brief: "b",
+    });
+    store.markRunning(session.id);
+    store.markRunInFlight(session.id);
+    store.registerInterrupt(session.id, () => undefined);
+    store.registerFollowup(session.id, async (message) => {
+      started.push(message);
+      return "should not run";
+    });
+    store.sendInputOne(session.id, "steer now", {
+      interrupt: true,
+      onFail: (err: unknown) => {
+        failures.push(err);
+      },
+    });
+    store.complete(session.id, "## Summary\nOriginal done.");
+    await Promise.resolve();
+    expect(started).toEqual([]);
+    expect(store.get(session.id)?.report).toBe("## Summary\nOriginal done.");
+    expect(store.get(session.id)?.lifecycleStatus).toBe("completed");
+    expect(failures).toHaveLength(1);
+    expect(String(defined(failures[0]))).toContain("steer now");
+    expect(store.get(session.id)?.entries).toContainEqual(
+      expect.objectContaining({
+        kind: "report",
+        content: expect.stringContaining("steer now"),
+      }),
+    );
+  });
+
+  test("a throwing handoff onReply does not stall the queued steers behind it", async () => {
+    const store = createSubAgentSessionStore();
+    const started: string[] = [];
+    const session = runningRetained(store, async (message) => {
+      started.push(message);
+      return `reply to ${message}`;
+    });
+    store.sendInputOne(session.id, "steer one", {
+      interrupt: true,
+      onFollowupReply: () => {
+        throw new Error("observer blew up");
+      },
+    });
+    store.sendInputOne(session.id, "steer two", { interrupt: true });
+    store.complete(session.id, "## Summary\nOriginal done.");
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    await new Promise((resolve) => setTimeout(resolve, 0));
+    expect(started).toEqual(["steer one", "steer two"]);
+    expect(store.get(session.id)?.lifecycle.state).toBe("completed");
+    expect(store.get(session.id)?.report).toBe("reply to steer two");
+  });
+
+  test("CL-7989 run-failure wins the race: loss is surfaced, never silent", async () => {
+    const store = createSubAgentSessionStore();
+    const started: string[] = [];
+    const failures: unknown[] = [];
+    const session = runningRetained(store, async (message) => {
+      started.push(message);
+      return "should not run";
+    });
+    store.sendInputOne(session.id, "steer now", {
+      interrupt: true,
+      onFail: (err: unknown) => {
+        failures.push(err);
+      },
+    });
+    store.fail(session.id, "provider 500");
+    await Promise.resolve();
+    expect(started).toEqual([]);
+    expect(store.get(session.id)?.lifecycle.state).toBe("failed");
+    expect(failures).toHaveLength(1);
+    expect(String(defined(failures[0]))).toContain("steer now");
+    expect(store.get(session.id)?.entries).toContainEqual(
+      expect.objectContaining({
+        kind: "report",
+        content: expect.stringContaining("steer now"),
+      }),
+    );
+  });
 });