From 73d27c2e444d5a5c9c49de7ddda0da0f83c6f4df Mon Sep 17 00:00:00 2001 From: t3-turbo-simulation Date: Sat, 8 Aug 2026 09:40:22 -0400 Subject: [PATCH 1/3] fix(server): stop resume handshakes from completing turns that never ran MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Claude result arriving with no local turnState emitted an untargeted turn.completed (no turnId). Real turns get turnState in sendTurn and assistant messages outside a turn auto-start a synthetic one, so the only things that land in that branch are the resume handshake (system/init + result(num_turns: 0)), a late result for a turn already completed locally, or a stream failure with no turn in flight — none of which is a turn. Ingestion's strict lifecycle guard rejected the phantom while an active turn was tracked, but fell through to accept it when activeTurnId was null — exactly the state while a turn start is pending — flipping the session back to "ready" for a turn that never existed (observed 2026-08-07 at 19:03:51 and 19:20). Two-sided fix: - ClaudeAdapter: a result with no turnState keeps its token-usage emission but no longer emits turn.completed; a structured log (claude.turn.result-without-active-turn) stays as the field tripwire. - ProviderRuntimeIngestion: with no active turn tracked, only completions that name their turn are applied. Targeted completions with no tracked turn still land (turn.started loss stays recoverable), and the other adapters were audited: Grok/Cursor/OpenCode always attach turnId. Both suppression tests fail against the unfixed code (verified by reverting the sources and re-running); the targeted-completion control passes on both, proving no legitimate completion is suppressed. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016BStzvrcLQ5gYmWscVk26C --- .../Layers/ProviderRuntimeIngestion.test.ts | 80 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 10 ++- .../src/provider/Layers/ClaudeAdapter.test.ts | 69 ++++++++++++++++ .../src/provider/Layers/ClaudeAdapter.ts | 34 ++++---- 4 files changed, 174 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index b4468bd4c6d..f0233a39d88 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -843,6 +843,86 @@ describe("ProviderRuntimeIngestion", () => { ); }); + it("rejects an untargeted turn.completed when no turn is active", async () => { + const harness = await createHarness(); + const seededAt = "2026-01-01T00:00:00.000Z"; + + // A turn start is pending: the session reads "starting" with no active + // turn tracked yet. This is the window the Claude resume handshake's + // phantom (turn.completed with no turnId) used to slip through, stomping + // "starting" back to "ready" for a turn that never existed. + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-seed-untargeted-completion"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "starting", + providerName: "claudeAgent", + runtimeMode: "approval-required", + activeTurnId: null, + updatedAt: seededAt, + lastError: null, + }, + createdAt: seededAt, + }), + ); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-untargeted"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: seededAt, + threadId: asThreadId("thread-1"), + status: "completed", + }); + + await harness.drain(); + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + expect(thread?.session?.status).toBe("starting"); + expect(thread?.session?.activeTurnId).toBeNull(); + }); + + it("accepts a targeted turn.completed when no turn is active", async () => { + const harness = await createHarness(); + const seededAt = "2026-01-01T00:00:00.000Z"; + + // A completion that names its turn still lands even when no active turn + // is tracked (e.g. its turn.started was lost). Only untargeted + // completions are rejected. + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-seed-targeted-completion"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "starting", + providerName: "claudeAgent", + runtimeMode: "approval-required", + activeTurnId: null, + updatedAt: seededAt, + lastError: null, + }, + createdAt: seededAt, + }), + ); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-targeted-late"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: seededAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-late"), + status: "completed", + }); + + await waitForThread(harness.readModel, (thread) => thread.session?.status === "ready"); + }); + it("ignores non-active turn completion when runtime omits thread id", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 40307cd9f25..03253797242 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -1532,8 +1532,14 @@ const make = Effect.gen(function* () { if (activeTurnId !== null && eventTurnId !== undefined) { return sameId(activeTurnId, eventTurnId); } - // If no active turn is tracked, accept completion scoped to this thread. - return true; + // No active turn tracked: accept only completions that name their + // turn (covers a real completion whose turn.started was lost). An + // untargeted completion cannot prove it belongs to any turn this + // thread ran — the known emitter was the Claude resume handshake + // (system/init + result(num_turns: 0)), which is not a turn at + // all — and applying it here stomps the "starting" lifecycle + // state while a turn start is pending. + return eventTurnId !== undefined; default: return true; } diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index d3d768b5384..711b0f6f6aa 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -978,6 +978,75 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("does not emit turn.completed for a result with no active turn", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + // Collect through session.exited so the window after the second result + // is deterministically inside the collection: both results are queued + // after sendTurn returns and drain in order on the one stream consumer. + const runtimeEventsFiber = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "session.exited"), + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "hello", + attachments: [], + }); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + num_turns: 1, + session_id: "sdk-session-1", + uuid: "result-real", + } as unknown as SDKMessage); + + // Second result with no turn in flight — the shape the resume + // handshake (system/init + result(num_turns: 0)) delivers, and the + // same completeTurn branch every no-turnState result lands in. This + // used to emit an untargeted turn.completed; it must emit nothing. + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + num_turns: 0, + usage: { input_tokens: 0, output_tokens: 0 }, + session_id: "sdk-session-1", + uuid: "result-handshake", + } as unknown as SDKMessage); + + harness.query.finish(); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + const completions = runtimeEvents.filter((event) => event.type === "turn.completed"); + // Exactly one completion — the real turn's, targeted at its turn id. + // The buggy branch produced a second, untargeted one here. + assert.equal(completions.length, 1); + const completed = completions[0]; + if (completed?.type === "turn.completed") { + assert.equal(String(completed.turnId), String(turn.turnId)); + assert.equal(completed.payload.state, "completed"); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 92445522cc4..5e284696fc8 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -2248,24 +2248,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( rawPayload: result ?? { status }, }); - const stamp = yield* makeEventStamp(); - yield* offerRuntimeEvent({ - type: "turn.completed", - eventId: stamp.eventId, - provider: PROVIDER, - createdAt: stamp.createdAt, + // A result with no local turn is never a turn this adapter started: + // real turns get turnState in sendTurn, and assistant messages that + // arrive outside a turn auto-start a synthetic one. What lands here is + // the resume handshake (system/init + result(num_turns: 0)), a late + // result for a turn already completed locally (steer auto-close, + // stream teardown), or a stream failure with no turn in flight. The + // untargeted turn.completed this branch used to emit carried no turnId, + // so ingestion could not attribute it — and whenever the projection had + // no active turn (a pending turn start included) it flipped the session + // lifecycle for a turn that never existed. Keep the usage emission, + // drop the lifecycle event, and leave a tripwire so the upstream + // trigger stays measurable in the field. + yield* Effect.logInfo("claude.turn.result-without-active-turn", { threadId: context.session.threadId, - payload: { - state: status, - ...(result?.stop_reason !== undefined ? { stopReason: result.stop_reason } : {}), - ...(result?.usage ? { usage: result.usage } : {}), - ...(result?.modelUsage ? { modelUsage: result.modelUsage } : {}), - ...(typeof result?.total_cost_usd === "number" - ? { totalCostUsd: result.total_cost_usd } - : {}), - ...(errorMessage ? { errorMessage } : {}), - }, - providerRefs: {}, + status, + numTurns: (result as { readonly num_turns?: number } | undefined)?.num_turns, + hasUsage: result?.usage !== undefined, + ...(errorMessage ? { errorMessage } : {}), }); return; } From 2fb42117562cb342eb9c80383fc7b5bdddce8c5e Mon Sep 17 00:00:00 2001 From: t3-turbo-simulation Date: Sat, 8 Aug 2026 10:06:50 -0400 Subject: [PATCH 2/3] refactor(server): drop unnecessary num_turns cast in the result tripwire SDKResultSuccess and SDKResultError both declare num_turns as required, so the widening cast was noise. Review finding (INFO). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_016BStzvrcLQ5gYmWscVk26C --- apps/server/src/provider/Layers/ClaudeAdapter.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 5e284696fc8..00839c455b4 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -2263,7 +2263,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( yield* Effect.logInfo("claude.turn.result-without-active-turn", { threadId: context.session.threadId, status, - numTurns: (result as { readonly num_turns?: number } | undefined)?.num_turns, + numTurns: result?.num_turns, hasUsage: result?.usage !== undefined, ...(errorMessage ? { errorMessage } : {}), }); From e670eb01817ffcbc0dc1774edc04ec2434a405f1 Mon Sep 17 00:00:00 2001 From: t3-turbo-simulation Date: Sat, 8 Aug 2026 19:58:03 -0400 Subject: [PATCH 3/3] fix(server): keep the new ingestion tests under the manual-runtime ratchet MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two tests added here each seeded their session with an inline `Effect.runPromise(harness.engine.dispatch(...))`, pushing this file from 31 to 33 manual Effect runtime calls — one past its `no-manual-effect-runtime-in-tests` baseline — so `vp check` failed with two errors. The harness already wraps its other runtime entry points (`readModel`, `drain`), so dispatch is wrapped the same way: one `dispatch` helper next to the engine, used for the three seed commands inside `createHarness` and exposed on the harness for the new tests. That takes the file to 29 occurrences, under the baseline, rather than raising the baseline. Test-only; no behavior change. ProviderRuntimeIngestion 47/47, ClaudeAdapter 67/67, `vp check` clean, server typecheck clean. Co-Authored-By: Claude Opus 5 (1M context) --- .../Layers/ProviderRuntimeIngestion.test.ts | 151 +++++++++--------- 1 file changed, 72 insertions(+), 79 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index f0233a39d88..258aa010e3e 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -16,6 +16,7 @@ import { DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, + type OrchestrationCommand, ProjectId, ProviderItemId, type ServerSettings, @@ -256,57 +257,52 @@ describe("ProviderRuntimeIngestion", () => { scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope))); const drain = () => Effect.runPromise(ingestion.drain); + const dispatch = (command: OrchestrationCommand) => Effect.runPromise(engine.dispatch(command)); const createdAt = "2026-01-01T00:00:00.000Z"; - await Effect.runPromise( - engine.dispatch({ - type: "project.create", - commandId: CommandId.make("cmd-provider-project-create"), - projectId: asProjectId("project-1"), - title: "Provider Project", - workspaceRoot, - defaultModelSelection: { - instanceId: ProviderInstanceId.make("codex"), - model: "gpt-5-codex", - }, - createdAt, - }), - ); - await Effect.runPromise( - engine.dispatch({ - type: "thread.create", - commandId: CommandId.make("cmd-thread-create"), + await dispatch({ + type: "project.create", + commandId: CommandId.make("cmd-provider-project-create"), + projectId: asProjectId("project-1"), + title: "Provider Project", + workspaceRoot, + defaultModelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + createdAt, + }); + await dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-thread-create"), + threadId: ThreadId.make("thread-1"), + projectId: asProjectId("project-1"), + title: "Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt, + }); + await dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-seed"), + threadId: ThreadId.make("thread-1"), + session: { threadId: ThreadId.make("thread-1"), - projectId: asProjectId("project-1"), - title: "Thread", - modelSelection: { - instanceId: ProviderInstanceId.make("codex"), - model: "gpt-5-codex", - }, - interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + status: "ready", + providerName: "codex", runtimeMode: "approval-required", - branch: null, - worktreePath: null, - createdAt, - }), - ); - await Effect.runPromise( - engine.dispatch({ - type: "thread.session.set", - commandId: CommandId.make("cmd-session-seed"), - threadId: ThreadId.make("thread-1"), - session: { - threadId: ThreadId.make("thread-1"), - status: "ready", - providerName: "codex", - runtimeMode: "approval-required", - activeTurnId: null, - updatedAt: createdAt, - lastError: null, - }, - createdAt, - }), - ); + activeTurnId: null, + updatedAt: createdAt, + lastError: null, + }, + createdAt, + }); provider.setSession({ provider: ProviderDriverKind.make("codex"), status: "ready", @@ -318,6 +314,7 @@ describe("ProviderRuntimeIngestion", () => { return { engine, + dispatch, readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()), emit: provider.emit, setProviderSession: provider.setSession, @@ -851,23 +848,21 @@ describe("ProviderRuntimeIngestion", () => { // turn tracked yet. This is the window the Claude resume handshake's // phantom (turn.completed with no turnId) used to slip through, stomping // "starting" back to "ready" for a turn that never existed. - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.session.set", - commandId: CommandId.make("cmd-session-seed-untargeted-completion"), + await harness.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-seed-untargeted-completion"), + threadId: ThreadId.make("thread-1"), + session: { threadId: ThreadId.make("thread-1"), - session: { - threadId: ThreadId.make("thread-1"), - status: "starting", - providerName: "claudeAgent", - runtimeMode: "approval-required", - activeTurnId: null, - updatedAt: seededAt, - lastError: null, - }, - createdAt: seededAt, - }), - ); + status: "starting", + providerName: "claudeAgent", + runtimeMode: "approval-required", + activeTurnId: null, + updatedAt: seededAt, + lastError: null, + }, + createdAt: seededAt, + }); harness.emit({ type: "turn.completed", @@ -892,23 +887,21 @@ describe("ProviderRuntimeIngestion", () => { // A completion that names its turn still lands even when no active turn // is tracked (e.g. its turn.started was lost). Only untargeted // completions are rejected. - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.session.set", - commandId: CommandId.make("cmd-session-seed-targeted-completion"), + await harness.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-seed-targeted-completion"), + threadId: ThreadId.make("thread-1"), + session: { threadId: ThreadId.make("thread-1"), - session: { - threadId: ThreadId.make("thread-1"), - status: "starting", - providerName: "claudeAgent", - runtimeMode: "approval-required", - activeTurnId: null, - updatedAt: seededAt, - lastError: null, - }, - createdAt: seededAt, - }), - ); + status: "starting", + providerName: "claudeAgent", + runtimeMode: "approval-required", + activeTurnId: null, + updatedAt: seededAt, + lastError: null, + }, + createdAt: seededAt, + }); harness.emit({ type: "turn.completed",