fix: unblock replies after multi-agent room reset (#99091)

* fix: release stale reply work during reset cleanup

* fix: release archived reply runs during reset

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

src/auto-reply/reply/reply-run-registry.ts

@@ -858,6 +858,19 @@ export function forceClearReplyRunBySessionId(sessionId: string, cause?: unknown
   return true;
 }
 
+export function clearReplyRunForResetBySessionId(sessionId: string): void {
+  const operation = resolveReplyRunForCurrentSessionId(sessionId);
+  if (!operation || operation.phase === "queued") {
+    return;
+  }
+  operation.abortForRestart();
+  // Backend cancellation may synchronously retire this operation and admit a
+  // replacement. Only clear the exact archived operation resolved above.
+  if (replyRunState.activeRunsByKey.get(operation.key) === operation) {
+    operation.complete();
+  }
+}
+
 export function waitForReplyRunEndBySessionId(
   sessionId: string,
   timeoutMs: number,

src/auto-reply/reply/session-reset-cleanup.test.ts

@@ -1,13 +1,21 @@
 // Tests session reset cleanup for stale files and persisted state.
-import { afterEach, describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import {
   enqueueSystemEvent,
   peekSystemEvents,
   resetSystemEventsForTest,
 } from "../../infra/system-events.js";
+import { resetDiagnosticRunActivityForTest } from "../../logging/diagnostic-run-activity.js";
+import {
+  createReplyOperation,
+  replyRunRegistry,
+  testing as replyRunTesting,
+} from "./reply-run-registry.js";
 import { clearSessionResetRuntimeState } from "./session-reset-cleanup.js";
 
 afterEach(() => {
+  replyRunTesting.resetReplyRunRegistry();
+  resetDiagnosticRunActivityForTest();
   resetSystemEventsForTest();
 });
 
@@ -25,4 +33,92 @@ describe("clearSessionResetRuntimeState", () => {
     expect(peekSystemEvents("beta")).toStrictEqual([]);
     expect(peekSystemEvents("gamma")).toEqual(["fresh gamma"]);
   });
+
+  it("releases active reply work owned by the archived reset session id", () => {
+    const cancel = vi.fn();
+    const operation = createReplyOperation({
+      sessionKey: "agent:main:slack:room:1",
+      sessionId: "old-session",
+      resetTriggered: false,
+    });
+    operation.attachBackend({
+      kind: "embedded",
+      cancel,
+      isStreaming: () => false,
+    });
+    operation.setPhase("running");
+
+    clearSessionResetRuntimeState(["agent:main:slack:room:1", "old-session"], {
+      activeReplySessionId: "old-session",
+    });
+
+    expect(cancel).toHaveBeenCalledWith("restart");
+    expect(replyRunRegistry.isActive("agent:main:slack:room:1")).toBe(false);
+    const nextOperation = createReplyOperation({
+      sessionKey: "agent:main:slack:room:1",
+      sessionId: "new-session",
+      resetTriggered: false,
+    });
+    expect(nextOperation.sessionId).toBe("new-session");
+  });
+
+  it("does not clear a fresh active reply under the same key when only the archived id is reset", () => {
+    const operation = createReplyOperation({
+      sessionKey: "agent:main:slack:room:1",
+      sessionId: "new-session",
+      resetTriggered: false,
+    });
+    operation.setPhase("running");
+
+    clearSessionResetRuntimeState(["agent:main:slack:room:1", "old-session"], {
+      activeReplySessionId: "old-session",
+    });
+
+    expect(replyRunRegistry.get("agent:main:slack:room:1")).toBe(operation);
+  });
+
+  it("does not clear a replacement admitted while the archived run is cancelling", () => {
+    let replacement: ReturnType<typeof createReplyOperation> | undefined;
+    const operation = createReplyOperation({
+      sessionKey: "agent:main:slack:room:1",
+      sessionId: "old-session",
+      resetTriggered: false,
+    });
+    operation.attachBackend({
+      kind: "embedded",
+      cancel() {
+        operation.complete();
+        replacement = createReplyOperation({
+          sessionKey: "agent:main:slack:room:1",
+          sessionId: "old-session",
+          resetTriggered: false,
+        });
+        replacement.setPhase("running");
+      },
+      isStreaming: () => false,
+    });
+    operation.setPhase("running");
+
+    clearSessionResetRuntimeState(["agent:main:slack:room:1", "old-session"], {
+      activeReplySessionId: "old-session",
+    });
+
+    expect(replacement).toBeDefined();
+    expect(replyRunRegistry.get("agent:main:slack:room:1")).toBe(replacement);
+  });
+
+  it("leaves queued reservations for the archived id so session init can rebind them", () => {
+    const operation = createReplyOperation({
+      sessionKey: "agent:main:slack:room:1",
+      sessionId: "old-session",
+      resetTriggered: false,
+    });
+
+    clearSessionResetRuntimeState(["agent:main:slack:room:1", "old-session"], {
+      activeReplySessionId: "old-session",
+    });
+
+    expect(operation.phase).toBe("queued");
+    expect(replyRunRegistry.get("agent:main:slack:room:1")).toBe(operation);
+  });
 });

src/auto-reply/reply/session-reset-cleanup.ts

@@ -1,6 +1,7 @@
 /** Clears reset-related queues and system events for session keys. */
 import { drainSystemEventEntries } from "../../infra/system-events.js";
 import { clearSessionQueues, type ClearSessionQueueResult } from "./queue/cleanup.js";
+import { clearReplyRunForResetBySessionId } from "./reply-run-registry.js";
 
 /** Runtime cleanup result for reset-related queues and system events. */
 type ClearSessionResetRuntimeStateResult = ClearSessionQueueResult & {
@@ -10,6 +11,7 @@ type ClearSessionResetRuntimeStateResult = ClearSessionQueueResult & {
 /** Clears queued follow-ups and pending system events for reset session keys. */
 export function clearSessionResetRuntimeState(
   keys: Array<string | undefined>,
+  opts?: { activeReplySessionId?: string },
 ): ClearSessionResetRuntimeStateResult {
   const cleared = clearSessionQueues(keys);
   let systemEventsCleared = 0;
@@ -18,6 +20,10 @@ export function clearSessionResetRuntimeState(
     systemEventsCleared += drainSystemEventEntries(key).length;
   }
 
+  if (opts?.activeReplySessionId) {
+    clearReplyRunForResetBySessionId(opts.activeReplySessionId);
+  }
+
   return {
     ...cleared,
     systemEventsCleared,

src/auto-reply/reply/session.ts

@@ -685,7 +685,9 @@ async function initSessionStateAttemptLocked(
     previousSessionId: previousSessionEntry?.sessionId,
   });
   if (previousSessionEntry) {
-    clearSessionResetRuntimeState([sessionKey, previousSessionEntry.sessionId]);
+    clearSessionResetRuntimeState([sessionKey, previousSessionEntry.sessionId], {
+      activeReplySessionId: previousSessionEntry.sessionId,
+    });
   }
 
   const recoveredTerminalEntry =

src/gateway/session-reset-service.ts

@@ -401,7 +401,9 @@ async function ensureSessionRuntimeCleanup(params: {
   if (params.sessionId) {
     queueKeys.add(params.sessionId);
   }
-  clearSessionResetRuntimeState([...queueKeys]);
+  clearSessionResetRuntimeState([...queueKeys], {
+    activeReplySessionId: params.sessionId,
+  });
   stopSubagentsForRequester({ cfg: params.cfg, requesterSessionKey: params.target.canonicalKey });
   if (!params.sessionId) {
     params.assertCurrent?.();