diff --git a/src/agents/main-session-recovery-state.test.ts b/src/agents/main-session-recovery-state.test.ts index 4fdef80631d..1fc0d11c869 100644 --- a/src/agents/main-session-recovery-state.test.ts +++ b/src/agents/main-session-recovery-state.test.ts @@ -204,6 +204,53 @@ describe("main session recovery state", () => { expect(entry).toEqual(before); }); + it("clears orphaned lifecycle fences before healthy foreground admission", () => { + const entry = interruptedEntry({ + abortedLastRun: false, + mainRestartRecovery: undefined, + restartRecoveryRuns: [{ runId: "stale-run", lifecycleGeneration: "stale-generation" }], + }); + + expect( + transitionMainSessionRecovery(entry, { + kind: "claim_foreground", + cycleId: "unused", + lifecycleGeneration: "generation-1", + sessionId: "session-1", + sessionKey, + claimId: "foreground-1", + }), + ).toEqual({ kind: "applied" }); + expect(entry).toMatchObject({ + status: "running", + abortedLastRun: false, + }); + expect(entry.restartRecoveryRuns).toBeUndefined(); + expect(entry.mainRestartRecovery).toBeUndefined(); + }); + + it("keeps lifecycle fences while a recovery delivery owner remains", () => { + const entry = interruptedEntry({ + abortedLastRun: false, + mainRestartRecovery: undefined, + restartRecoveryDeliveryRunId: "active-recovery", + restartRecoveryRuns: [{ runId: "active-recovery", lifecycleGeneration: "generation-1" }], + }); + const before = structuredClone(entry); + + expect( + transitionMainSessionRecovery(entry, { + kind: "claim_foreground", + cycleId: "unused", + lifecycleGeneration: "generation-1", + sessionId: "session-1", + sessionKey, + claimId: "foreground-1", + }), + ).toEqual({ kind: "no_change" }); + expect(entry).toEqual(before); + }); + it("retires a stale reservation before granting foreground ownership", () => { const entry = interruptedEntry({ mainRestartRecovery: recoveryState({ diff --git a/src/agents/main-session-recovery-state.ts b/src/agents/main-session-recovery-state.ts index 3ca9f4d9295..dc0ce103989 100644 --- a/src/agents/main-session-recovery-state.ts +++ b/src/agents/main-session-recovery-state.ts @@ -155,6 +155,19 @@ export function isMainRestartRecoveryCandidate(entry: SessionEntry, sessionKey: ); } +// A healthy session can retain lifecycle fences after its final recovery owner +// clears. With no active delivery or aggregate, those fences no longer own work. +function hasOrphanedMainRestartRecoveryFences(entry: SessionEntry, sessionKey: string): boolean { + return ( + entry.status === "running" && + entry.abortedLastRun !== true && + entry.restartRecoveryRuns !== undefined && + entry.mainRestartRecovery === undefined && + entry.restartRecoveryDeliveryRunId === undefined && + isMainRestartRecoveryCandidate(entry, sessionKey) + ); +} + function inspectMainSessionRecovery(params: { entry: SessionEntry; lifecycleGeneration: string; @@ -444,6 +457,13 @@ export function transitionMainSessionRecovery( return { kind: "applied" }; } case "claim_foreground": { + if ( + entry.sessionId === command.sessionId && + hasOrphanedMainRestartRecoveryFences(entry, command.sessionKey) + ) { + Object.assign(entry, buildMainSessionRecoveryClearPatch(entry)); + return { kind: "applied" }; + } if ( entry.sessionId !== command.sessionId || entry.status !== "running" || diff --git a/src/agents/main-session-recovery-store.test.ts b/src/agents/main-session-recovery-store.test.ts index 9e3c1ce499b..704aafb8720 100644 --- a/src/agents/main-session-recovery-store.test.ts +++ b/src/agents/main-session-recovery-store.test.ts @@ -366,6 +366,31 @@ describe("main session recovery store", () => { expect(accessorSpy.mock.calls[0]?.[0]).toMatchObject({ sessionKeys: [sessionKey] }); }); + it("atomically clears orphaned lifecycle fences from a healthy row", async () => { + await write( + interruptedEntry({ + abortedLastRun: false, + mainRestartRecovery: undefined, + restartRecoveryRuns: [{ runId: "stale-run", lifecycleGeneration: "stale-generation" }], + }), + ); + + await expect( + claimMainSessionRecoveryOwner({ + lifecycleGeneration, + sessionId: "session-1", + target: { sessionKey, storePath }, + }), + ).resolves.toEqual({ kind: "not_required" }); + expect(read()).toMatchObject({ + sessionId: "session-1", + status: "running", + abortedLastRun: false, + }); + expect(read().restartRecoveryRuns).toBeUndefined(); + expect(read().mainRestartRecovery).toBeUndefined(); + }); + it("binds a foreground claim to its lifecycle run", async () => { await write(interruptedEntry()); diff --git a/src/auto-reply/reply/reply-turn-admission.test.ts b/src/auto-reply/reply/reply-turn-admission.test.ts index 8d6b6f200b5..7416a5bb3db 100644 --- a/src/auto-reply/reply/reply-turn-admission.test.ts +++ b/src/auto-reply/reply/reply-turn-admission.test.ts @@ -425,6 +425,36 @@ describe("reply turn admission", () => { }, ); + it("admits a visible turn after clearing orphaned restart-recovery fences", async () => { + const sessionKey = "agent:main:telegram:topic:orphaned-recovery-fence"; + const sessionId = "healthy-session"; + const storePath = createSessionStore({ + [sessionKey]: { + sessionId, + updatedAt: 100, + status: "running", + abortedLastRun: false, + restartRecoveryRuns: [{ runId: "stale-run", lifecycleGeneration: "stale-generation" }], + }, + }); + + const admission = await admitReplyTurn({ + sessionKey, + sessionId, + expectedSessionId: sessionId, + storePath, + kind: "visible", + resetTriggered: false, + }); + expect(admission.status).toBe("owned"); + const persisted = await readSessionEntry(storePath, sessionKey); + expect(persisted?.restartRecoveryRuns).toBeUndefined(); + expect(persisted?.mainRestartRecovery).toBeUndefined(); + if (admission.status === "owned") { + admission.operation.complete(); + } + }); + it("drops a queued followup for an admitted recovery fence", async () => { const sessionKey = "agent:main:telegram:topic:admitted-recovery"; const sessionId = "admitted-recovery-session";