diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index b4468bd4c6d..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, @@ -843,6 +840,82 @@ 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 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"), + 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 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"), + 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..00839c455b4 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?.num_turns, + hasUsage: result?.usage !== undefined, + ...(errorMessage ? { errorMessage } : {}), }); return; }