diff --git a/apps/server/src/actionResume/ActionResume.test.ts b/apps/server/src/actionResume/ActionResume.test.ts index e4ae020f2bed..91d36bf5866d 100644 --- a/apps/server/src/actionResume/ActionResume.test.ts +++ b/apps/server/src/actionResume/ActionResume.test.ts @@ -313,6 +313,17 @@ it.effect("requires an explicit resume after a running Action is found on startu delivery: "available", finishedAt: now, }; + const settledThreadId = ThreadId.make("thread-action-resume-settled"); + const settled: ActionResumeState = { + ...running, + runId: "settled-run", + threadId: settledThreadId, + terminalId: "action-settled-run", + outcome: "succeeded", + delivery: "disposed", + finishedAt: now, + exitCode: 0, + }; const dependencies = Layer.mergeAll( Layer.mock(OrchestrationEngineService)({ @@ -372,6 +383,26 @@ it.effect("requires an explicit resume after a running Action is found on startu sequence: 2, createdAt: now, }, + { + activityId: EventId.make("action-resume:settled-run:succeeded:disposed"), + threadId: settledThreadId, + turnId: null, + tone: "info", + kind: ActionResume.ACTION_RESUME_ACTIVITY_KIND, + summary: "Action completed: QA", + payload: settled, + createdAt: now, + }, + { + activityId: EventId.make("action-resume:settled-run:succeeded:pending"), + threadId: settledThreadId, + turnId: null, + tone: "info", + kind: ActionResume.ACTION_RESUME_ACTIVITY_KIND, + summary: "Action completed: QA", + payload: { ...settled, delivery: "pending" }, + createdAt: now, + }, ]), }), Layer.mock(TerminalManager.TerminalManager)({ @@ -400,6 +431,11 @@ it.effect("requires an explicit resume after a running Action is found on startu outcome: "process_lost", delivery: "available", }); + assert.deepInclude(registry.getLatest(settledThreadId), { + outcome: "succeeded", + delivery: "disposed", + }); + assert.isNull(registry.getForShell(settledThreadId)); yield* service.discardInterrupted(recoveredThreadId); assert.deepInclude(registry.getLatest(recoveredThreadId), { delivery: "disposed" }); registry.record(available); diff --git a/apps/server/src/actionResume/ActionResume.ts b/apps/server/src/actionResume/ActionResume.ts index 843109d3ad7e..291bdbfae67c 100644 --- a/apps/server/src/actionResume/ActionResume.ts +++ b/apps/server/src/actionResume/ActionResume.ts @@ -699,10 +699,12 @@ const make = Effect.gen(function* () { const hydrate = Effect.fn("ActionResume.hydrate")(function* () { const rows = yield* activities.listByKind({ kind: ACTION_RESUME_ACTIVITY_KIND }); + const states: ActionResumeState[] = []; for (const row of rows) { const decoded = yield* Effect.option(decodeState(row.payload)); - if (Option.isSome(decoded)) registry.record(decoded.value); + if (Option.isSome(decoded)) states.push(decoded.value); } + registry.hydrate(states); }); const reconcile = Effect.fn("ActionResume.reconcile")(function* () { diff --git a/apps/server/src/orchestration/ThreadActionResume.test.ts b/apps/server/src/orchestration/ThreadActionResume.test.ts new file mode 100644 index 000000000000..b96c6dd791b9 --- /dev/null +++ b/apps/server/src/orchestration/ThreadActionResume.test.ts @@ -0,0 +1,61 @@ +import { assert, it } from "@effect/vitest"; +import { type ActionResumeState, ProjectId, ThreadId } from "@t3tools/contracts"; + +import { make } from "./ThreadActionResume.ts"; + +const threadId = ThreadId.make("thread-action-resume-hydration"); +const projectId = ProjectId.make("project-action-resume-hydration"); + +const state = (input: Partial): ActionResumeState => ({ + runId: "completed-run", + threadId, + projectId, + actionId: "qa", + actionName: "QA", + terminalId: "action-completed-run", + outcome: "succeeded", + delivery: "pending", + startedAt: "2026-08-17T00:00:00.000Z", + finishedAt: "2026-08-17T00:01:00.000Z", + exitCode: 0, + exitSignal: null, + ...input, +}); + +it("does not resurrect stale pending state for a settled Action", () => { + const registry = make(); + + registry.hydrate([ + state({ outcome: "running", delivery: "armed", finishedAt: null, exitCode: null }), + state({ delivery: "available" }), + state({ delivery: "delivered" }), + state({ delivery: "disposed" }), + state({ delivery: "pending" }), + ]); + + assert.deepInclude(registry.getLatest(threadId), { + outcome: "succeeded", + delivery: "disposed", + }); + assert.isNull(registry.getForShell(threadId)); +}); + +it("keeps a genuinely pending newer Action visible after hydration", () => { + const registry = make(); + + registry.hydrate([ + state({ delivery: "delivered" }), + state({ + runId: "newer-run", + terminalId: "action-newer-run", + delivery: "pending", + startedAt: "2026-08-17T00:02:00.000Z", + finishedAt: "2026-08-17T00:03:00.000Z", + }), + ]); + + assert.deepInclude(registry.getForShell(threadId), { + runId: "newer-run", + delivery: "pending", + }); +}); diff --git a/apps/server/src/orchestration/ThreadActionResume.ts b/apps/server/src/orchestration/ThreadActionResume.ts index 4ca242bbc2ed..2ce9da3e065f 100644 --- a/apps/server/src/orchestration/ThreadActionResume.ts +++ b/apps/server/src/orchestration/ThreadActionResume.ts @@ -16,7 +16,24 @@ const isShellVisible = (state: ActionResumeState): boolean => state.delivery === "pending" || state.delivery === "available"; +const hydrationDeliveryRank = (delivery: ActionResumeState["delivery"]): number => { + switch (delivery) { + case "armed": + return 0; + case "pending": + return 1; + case "available": + return 2; + case "delivered": + return 3; + case "disposed": + return 4; + } +}; + export interface ThreadActionResumeShape { + /** Restore one latest run per thread without reviving superseded delivery states. */ + readonly hydrate: (states: ReadonlyArray) => void; readonly record: (state: ActionResumeState) => void; readonly clear: (threadId: string) => void; readonly getLatest: (threadId: string) => ActionResumeState | null; @@ -29,6 +46,19 @@ export function make(): ThreadActionResumeShape { const latestByThreadId = new Map(); return { + hydrate: (states) => { + for (const state of states) { + const current = latestByThreadId.get(state.threadId); + if ( + current === undefined || + state.startedAt > current.startedAt || + (state.runId === current.runId && + hydrationDeliveryRank(state.delivery) > hydrationDeliveryRank(current.delivery)) + ) { + latestByThreadId.set(state.threadId, state); + } + } + }, record: (state) => { latestByThreadId.set(state.threadId, state); },