From 96d610b1f82255acd0fc077917dde7a944033f23 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 20:29:10 -0800 Subject: [PATCH] more progress --- .plans/16c-pr89-remediation-checklist.md | 12 ++-- .../Layers/CheckpointReactor.test.ts | 48 +++++++++++++ .../orchestration/Layers/CheckpointReactor.ts | 8 +++ .../Layers/ProviderRuntimeIngestion.test.ts | 72 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 20 ++++++ 5 files changed, 154 insertions(+), 6 deletions(-) diff --git a/.plans/16c-pr89-remediation-checklist.md b/.plans/16c-pr89-remediation-checklist.md index 59fe3b16c614864795cde72907df668f5c7f9fb1..5ebe2fe235451e5229650ef57640e690222677d2 100644 --- a/.plans/16c-pr89-remediation-checklist.md +++ b/.plans/16c-pr89-remediation-checklist.md @@ -53,8 +53,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` - Threads: PRRT_kwDORLtfbc5wkPaL - Audit note: SessionId-only routing can misassociate events under races/rebinds. -- [ ] `C020` On `message.completed`, the message ID is added to the set and `thread.message.assistant.complete` is dispatched. On `turn.completed`, the same set is iterated and `thread.message.assistant.complete` is dispatched again for each ID—including already-completed ones. Consider removing message IDs from the set after dispatching on `message.completed`, or filtering out already-completed IDs before the `turn.completed` loop. - - Status: `TODO` +- [x] `C020` On `message.completed`, the message ID is added to the set and `thread.message.assistant.complete` is dispatched. On `turn.completed`, the same set is iterated and `thread.message.assistant.complete` is dispatched again for each ID—including already-completed ones. Consider removing message IDs from the set after dispatching on `message.completed`, or filtering out already-completed IDs before the `turn.completed` loop. + - Status: `DONE` - Verdict: `partially-valid` - Severity: `Medium` - Area: `Runtime resilience and failure handling` @@ -80,8 +80,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` - Threads: PRRT_kwDORLtfbc5wyTaW, PRRT_kwDORLtfbc5wzli3 (+1 duplicate thread(s)) - Audit note: Same unhandled rejection path remains in WS message handler. -- [ ] `C042` Duplicated `resolveThreadWorkspaceCwd` across three files - - Status: `TODO` +- [x] `C042` Duplicated `resolveThreadWorkspaceCwd` across three files + - Status: `DONE` - Verdict: `partially-valid` - Severity: `Low` - Area: `Runtime resilience and failure handling` @@ -98,8 +98,8 @@ Counts: active `53` (`valid=34`, `partially-valid=19`), closed-invalid `5` - Threads: PRRT_kwDORLtfbc5wnWwM, PRRT_kwDORLtfbc5w1C3-, PRRT_kwDORLtfbc5w1HGZ (+2 duplicate thread(s)) - Audit note: Workspace CWD resolution duplication still present across modules. -- [ ] `C044` Checkpoint reactor swallows diff errors silently for `turn.completed` - - Status: `TODO` +- [x] `C044` Checkpoint reactor swallows diff errors silently for `turn.completed` + - Status: `DONE` - Verdict: `partially-valid` - Severity: `Low` - Area: `Runtime resilience and failure handling` diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index ae39d13840d11a82847fa2fbb590388c89605bca..0bdd0aae7718bb9ac6d5c3f4b2701677781b257b 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -379,6 +379,54 @@ describe("CheckpointReactor", () => { ).toBe("v2\n"); }); + it("appends capture failure activity when turn diff summary cannot be derived", async () => { + const harness = await createHarness({ seedFilesystemCheckpoints: false }); + const createdAt = new Date().toISOString(); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.makeUnsafe("cmd-session-set-missing-baseline-diff"), + threadId: ThreadId.makeUnsafe("thread-1"), + session: { + threadId: ThreadId.makeUnsafe("thread-1"), + status: "ready", + providerName: "codex", + providerSessionId: asSessionId("sess-1"), + providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + activeTurnId: null, + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }), + ); + + harness.provider.emit({ + type: "turn.completed", + eventId: EventId.makeUnsafe("evt-turn-completed-missing-baseline"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + turnId: asTurnId("turn-missing-baseline"), + status: "completed", + }); + + await waitForEvent(harness.engine, (event) => event.type === "thread.turn-diff-completed"); + const thread = await waitForThread( + harness.engine, + (entry) => + entry.checkpoints.length === 1 && + entry.activities.some((activity) => activity.kind === "checkpoint.capture.failed"), + ); + + expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); + expect( + thread.activities.some((activity) => activity.kind === "checkpoint.capture.failed"), + ).toBe(true); + }); + it("captures pre-turn baseline from project workspace root when thread worktree is unset", async () => { const harness = await createHarness({ hasSession: false, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 1a548ec7c8be409d323a236e6f42910c5d818361..2dd3ee0aafa602094b4d5f64599f467b0110a113 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -244,6 +244,14 @@ const make = Effect.gen(function* () { deletions: file.deletions, })), ), + Effect.tapError((error) => + appendCaptureFailureActivity({ + sessionId: event.sessionId, + turnId, + detail: `Checkpoint captured, but turn diff summary is unavailable: ${error.message}`, + createdAt: event.createdAt, + }), + ), Effect.catch((error) => Effect.logWarning("failed to derive checkpoint file summary", { threadId: thread.id, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index df97787a949d504a843455e945628140e038808e..f67d562218e0f94d677191e7a8027dad086ad687 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -258,6 +258,78 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + it("does not duplicate assistant completion when message.completed is followed by turn.completed", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-for-complete-dedup"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-complete-dedup"), + }); + + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && thread.session?.activeTurnId === "turn-complete-dedup", + ); + + harness.emit({ + type: "message.delta", + eventId: asEventId("evt-message-delta-for-complete-dedup"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-complete-dedup"), + itemId: asItemId("item-complete-dedup"), + delta: "done", + }); + harness.emit({ + type: "message.completed", + eventId: asEventId("evt-message-completed-for-complete-dedup"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-complete-dedup"), + itemId: asItemId("item-complete-dedup"), + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-for-complete-dedup"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-complete-dedup"), + status: "completed", + }); + + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "ready" && + thread.session?.activeTurnId === null && + thread.messages.some( + (message) => message.id === "assistant:item-complete-dedup" && !message.streaming, + ), + ); + + const events = await Effect.runPromise( + Stream.runCollect(harness.engine.readEvents(0)).pipe(Effect.map((chunk) => Array.from(chunk))), + ); + const completionEvents = events.filter((event) => { + if (event.type !== "thread.message-sent") { + return false; + } + return ( + event.payload.messageId === "assistant:item-complete-dedup" && event.payload.streaming === false + ); + }); + expect(completionEvents).toHaveLength(1); + }); + it("maps runtime.error into errored session state", async () => { const harness = await createHarness(); const now = new Date().toISOString(); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index c5af46017e819995903d13ba6d454c4302a79bd5..8476a4751ed4a8092daa0a3c0d415eeed1c2d32e 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -145,6 +145,22 @@ const make = Effect.gen(function* () { } }; + const forgetAssistantMessageId = ( + sessionId: ProviderSessionId, + turnId: TurnId, + messageId: MessageId, + ) => { + const key = providerTurnKey(sessionId, turnId); + const existingIds = turnMessageIdsByTurnKey.get(key); + if (!existingIds) { + return; + } + existingIds.delete(messageId); + if (existingIds.size === 0) { + turnMessageIdsByTurnKey.delete(key); + } + }; + const getAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => { return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set(); }; @@ -257,6 +273,10 @@ const make = Effect.gen(function* () { ...(turnId ? { turnId } : {}), createdAt: now, }); + + if (turnId) { + forgetAssistantMessageId(event.sessionId, turnId, assistantMessageId); + } } if (event.type === "turn.completed") { -- 2.54.0