authorgravatar for gabe@editmypodcast.agencyGabe Fletcher <gabe@editmypodcast.agency> 2026-08-08 20:24:42-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-08 17:24:42-07:00
loge70cdb478d34342d13ba4f433992394bf7303c1d
treecb320205b51d9c92a457b8c07c402e416d2bfd52
parentbe01b287b92b4686023a5e213078a2f51b3c1880
signature Signed by PGP key B5690EEEBB952194

fix(server): stop Claude resume handshakes from completing turns that never ran (#5710)

Co-authored-by: t3-turbo-simulation <simulation@t3turbo.local> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

4 files changed, 214 insertions(+), 66 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+120-47
...@@ -16,6 +16,7 @@ import {...@@ -16,6 +16,7 @@ import {
16 DEFAULT_PROVIDER_INTERACTION_MODE,16 DEFAULT_PROVIDER_INTERACTION_MODE,
17 EventId,17 EventId,
18 MessageId,18 MessageId,
19 type OrchestrationCommand,
19 ProjectId,20 ProjectId,
20 ProviderItemId,21 ProviderItemId,
21 type ServerSettings,22 type ServerSettings,
...@@ -256,57 +257,52 @@ describe("ProviderRuntimeIngestion", () => {...@@ -256,57 +257,52 @@ describe("ProviderRuntimeIngestion", () => {
256 scope = await Effect.runPromise(Scope.make("sequential"));257 scope = await Effect.runPromise(Scope.make("sequential"));
257 await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope)));258 await Effect.runPromise(ingestion.start().pipe(Scope.provide(scope)));
258 const drain = () => Effect.runPromise(ingestion.drain);259 const drain = () => Effect.runPromise(ingestion.drain);
260 const dispatch = (command: OrchestrationCommand) => Effect.runPromise(engine.dispatch(command));
259261
260 const createdAt = "2026-01-01T00:00:00.000Z";262 const createdAt = "2026-01-01T00:00:00.000Z";
261 await Effect.runPromise(263 await dispatch({
262 engine.dispatch({264 type: "project.create",
263 type: "project.create",265 commandId: CommandId.make("cmd-provider-project-create"),
264 commandId: CommandId.make("cmd-provider-project-create"),266 projectId: asProjectId("project-1"),
265 projectId: asProjectId("project-1"),267 title: "Provider Project",
266 title: "Provider Project",268 workspaceRoot,
267 workspaceRoot,269 defaultModelSelection: {
268 defaultModelSelection: {270 instanceId: ProviderInstanceId.make("codex"),
269 instanceId: ProviderInstanceId.make("codex"),271 model: "gpt-5-codex",
270 model: "gpt-5-codex",272 },
271 },273 createdAt,
272 createdAt,274 });
273 }),275 await dispatch({
274 );276 type: "thread.create",
275 await Effect.runPromise(277 commandId: CommandId.make("cmd-thread-create"),
276 engine.dispatch({278 threadId: ThreadId.make("thread-1"),
277 type: "thread.create",279 projectId: asProjectId("project-1"),
278 commandId: CommandId.make("cmd-thread-create"),280 title: "Thread",
281 modelSelection: {
282 instanceId: ProviderInstanceId.make("codex"),
283 model: "gpt-5-codex",
284 },
285 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
286 runtimeMode: "approval-required",
287 branch: null,
288 worktreePath: null,
289 createdAt,
290 });
291 await dispatch({
292 type: "thread.session.set",
293 commandId: CommandId.make("cmd-session-seed"),
294 threadId: ThreadId.make("thread-1"),
295 session: {
279 threadId: ThreadId.make("thread-1"),296 threadId: ThreadId.make("thread-1"),
280 projectId: asProjectId("project-1"),297 status: "ready",
281 title: "Thread",298 providerName: "codex",
282 modelSelection: {
283 instanceId: ProviderInstanceId.make("codex"),
284 model: "gpt-5-codex",
285 },
286 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
287 runtimeMode: "approval-required",299 runtimeMode: "approval-required",
288 branch: null,300 activeTurnId: null,
289 worktreePath: null,301 updatedAt: createdAt,
290 createdAt,302 lastError: null,
291 }),303 },
292 );304 createdAt,
293 await Effect.runPromise(305 });
294 engine.dispatch({
295 type: "thread.session.set",
296 commandId: CommandId.make("cmd-session-seed"),
297 threadId: ThreadId.make("thread-1"),
298 session: {
299 threadId: ThreadId.make("thread-1"),
300 status: "ready",
301 providerName: "codex",
302 runtimeMode: "approval-required",
303 activeTurnId: null,
304 updatedAt: createdAt,
305 lastError: null,
306 },
307 createdAt,
308 }),
309 );
310 provider.setSession({306 provider.setSession({
311 provider: ProviderDriverKind.make("codex"),307 provider: ProviderDriverKind.make("codex"),
312 status: "ready",308 status: "ready",
...@@ -318,6 +314,7 @@ describe("ProviderRuntimeIngestion", () => {...@@ -318,6 +314,7 @@ describe("ProviderRuntimeIngestion", () => {
318314
319 return {315 return {
320 engine,316 engine,
317 dispatch,
321 readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),318 readModel: () => Effect.runPromise(snapshotQuery.getSnapshot()),
322 emit: provider.emit,319 emit: provider.emit,
323 setProviderSession: provider.setSession,320 setProviderSession: provider.setSession,
...@@ -843,6 +840,82 @@ describe("ProviderRuntimeIngestion", () => {...@@ -843,6 +840,82 @@ describe("ProviderRuntimeIngestion", () => {
843 );840 );
844 });841 });
845842
843 it("rejects an untargeted turn.completed when no turn is active", async () => {
844 const harness = await createHarness();
845 const seededAt = "2026-01-01T00:00:00.000Z";
846
847 // A turn start is pending: the session reads "starting" with no active
848 // turn tracked yet. This is the window the Claude resume handshake's
849 // phantom (turn.completed with no turnId) used to slip through, stomping
850 // "starting" back to "ready" for a turn that never existed.
851 await harness.dispatch({
852 type: "thread.session.set",
853 commandId: CommandId.make("cmd-session-seed-untargeted-completion"),
854 threadId: ThreadId.make("thread-1"),
855 session: {
856 threadId: ThreadId.make("thread-1"),
857 status: "starting",
858 providerName: "claudeAgent",
859 runtimeMode: "approval-required",
860 activeTurnId: null,
861 updatedAt: seededAt,
862 lastError: null,
863 },
864 createdAt: seededAt,
865 });
866
867 harness.emit({
868 type: "turn.completed",
869 eventId: asEventId("evt-turn-completed-untargeted"),
870 provider: ProviderDriverKind.make("claudeAgent"),
871 createdAt: seededAt,
872 threadId: asThreadId("thread-1"),
873 status: "completed",
874 });
875
876 await harness.drain();
877 const readModel = await harness.readModel();
878 const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
879 expect(thread?.session?.status).toBe("starting");
880 expect(thread?.session?.activeTurnId).toBeNull();
881 });
882
883 it("accepts a targeted turn.completed when no turn is active", async () => {
884 const harness = await createHarness();
885 const seededAt = "2026-01-01T00:00:00.000Z";
886
887 // A completion that names its turn still lands even when no active turn
888 // is tracked (e.g. its turn.started was lost). Only untargeted
889 // completions are rejected.
890 await harness.dispatch({
891 type: "thread.session.set",
892 commandId: CommandId.make("cmd-session-seed-targeted-completion"),
893 threadId: ThreadId.make("thread-1"),
894 session: {
895 threadId: ThreadId.make("thread-1"),
896 status: "starting",
897 providerName: "claudeAgent",
898 runtimeMode: "approval-required",
899 activeTurnId: null,
900 updatedAt: seededAt,
901 lastError: null,
902 },
903 createdAt: seededAt,
904 });
905
906 harness.emit({
907 type: "turn.completed",
908 eventId: asEventId("evt-turn-completed-targeted-late"),
909 provider: ProviderDriverKind.make("claudeAgent"),
910 createdAt: seededAt,
911 threadId: asThreadId("thread-1"),
912 turnId: asTurnId("turn-late"),
913 status: "completed",
914 });
915
916 await waitForThread(harness.readModel, (thread) => thread.session?.status === "ready");
917 });
918
846 it("ignores non-active turn completion when runtime omits thread id", async () => {919 it("ignores non-active turn completion when runtime omits thread id", async () => {
847 const harness = await createHarness();920 const harness = await createHarness();
848 const now = "2026-01-01T00:00:00.000Z";921 const now = "2026-01-01T00:00:00.000Z";
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+8-2
...@@ -1532,8 +1532,14 @@ const make = Effect.gen(function* () {...@@ -1532,8 +1532,14 @@ const make = Effect.gen(function* () {
1532 if (activeTurnId !== null && eventTurnId !== undefined) {1532 if (activeTurnId !== null && eventTurnId !== undefined) {
1533 return sameId(activeTurnId, eventTurnId);1533 return sameId(activeTurnId, eventTurnId);
1534 }1534 }
1535 // If no active turn is tracked, accept completion scoped to this thread.1535 // No active turn tracked: accept only completions that name their
1536 return true;1536 // turn (covers a real completion whose turn.started was lost). An
1537 // untargeted completion cannot prove it belongs to any turn this
1538 // thread ran — the known emitter was the Claude resume handshake
1539 // (system/init + result(num_turns: 0)), which is not a turn at
1540 // all — and applying it here stomps the "starting" lifecycle
1541 // state while a turn start is pending.
1542 return eventTurnId !== undefined;
1537 default:1543 default:
1538 return true;1544 return true;
1539 }1545 }
apps/server/src/provider/Layers/ClaudeAdapter.test.ts+69
...@@ -978,6 +978,75 @@ describe("ClaudeAdapterLive", () => {...@@ -978,6 +978,75 @@ describe("ClaudeAdapterLive", () => {
978 );978 );
979 });979 });
980980
981 it.effect("does not emit turn.completed for a result with no active turn", () => {
982 const harness = makeHarness();
983 return Effect.gen(function* () {
984 const adapter = yield* ClaudeAdapter;
985
986 // Collect through session.exited so the window after the second result
987 // is deterministically inside the collection: both results are queued
988 // after sendTurn returns and drain in order on the one stream consumer.
989 const runtimeEventsFiber = yield* adapter.streamEvents.pipe(
990 Stream.takeUntil((event) => event.type === "session.exited"),
991 Stream.runCollect,
992 Effect.forkChild,
993 );
994
995 const session = yield* adapter.startSession({
996 threadId: THREAD_ID,
997 provider: ProviderDriverKind.make("claudeAgent"),
998 runtimeMode: "full-access",
999 });
1000
1001 const turn = yield* adapter.sendTurn({
1002 threadId: session.threadId,
1003 input: "hello",
1004 attachments: [],
1005 });
1006
1007 harness.query.emit({
1008 type: "result",
1009 subtype: "success",
1010 is_error: false,
1011 errors: [],
1012 num_turns: 1,
1013 session_id: "sdk-session-1",
1014 uuid: "result-real",
1015 } as unknown as SDKMessage);
1016
1017 // Second result with no turn in flight — the shape the resume
1018 // handshake (system/init + result(num_turns: 0)) delivers, and the
1019 // same completeTurn branch every no-turnState result lands in. This
1020 // used to emit an untargeted turn.completed; it must emit nothing.
1021 harness.query.emit({
1022 type: "result",
1023 subtype: "success",
1024 is_error: false,
1025 errors: [],
1026 num_turns: 0,
1027 usage: { input_tokens: 0, output_tokens: 0 },
1028 session_id: "sdk-session-1",
1029 uuid: "result-handshake",
1030 } as unknown as SDKMessage);
1031
1032 harness.query.finish();
1033
1034 const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber));
1035 const completions = runtimeEvents.filter((event) => event.type === "turn.completed");
1036 // Exactly one completion — the real turn's, targeted at its turn id.
1037 // The buggy branch produced a second, untargeted one here.
1038 assert.equal(completions.length, 1);
1039 const completed = completions[0];
1040 if (completed?.type === "turn.completed") {
1041 assert.equal(String(completed.turnId), String(turn.turnId));
1042 assert.equal(completed.payload.state, "completed");
1043 }
1044 }).pipe(
1045 Effect.provideService(Random.Random, makeDeterministicRandomService()),
1046 Effect.provide(harness.layer),
1047 );
1048 });
1049
981 it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => {1050 it.effect("steers a running turn instead of opening a new one on mid-turn sendTurn", () => {
982 const harness = makeHarness();1051 const harness = makeHarness();
983 return Effect.gen(function* () {1052 return Effect.gen(function* () {
apps/server/src/provider/Layers/ClaudeAdapter.ts+17-17
...@@ -2248,24 +2248,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (...@@ -2248,24 +2248,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
2248 rawPayload: result ?? { status },2248 rawPayload: result ?? { status },
2249 });2249 });
22502250
2251 const stamp = yield* makeEventStamp();2251 // A result with no local turn is never a turn this adapter started:
2252 yield* offerRuntimeEvent({2252 // real turns get turnState in sendTurn, and assistant messages that
2253 type: "turn.completed",2253 // arrive outside a turn auto-start a synthetic one. What lands here is
2254 eventId: stamp.eventId,2254 // the resume handshake (system/init + result(num_turns: 0)), a late
2255 provider: PROVIDER,2255 // result for a turn already completed locally (steer auto-close,
2256 createdAt: stamp.createdAt,2256 // stream teardown), or a stream failure with no turn in flight. The
2257 // untargeted turn.completed this branch used to emit carried no turnId,
2258 // so ingestion could not attribute it — and whenever the projection had
2259 // no active turn (a pending turn start included) it flipped the session
2260 // lifecycle for a turn that never existed. Keep the usage emission,
2261 // drop the lifecycle event, and leave a tripwire so the upstream
2262 // trigger stays measurable in the field.
2263 yield* Effect.logInfo("claude.turn.result-without-active-turn", {
2257 threadId: context.session.threadId,2264 threadId: context.session.threadId,
2258 payload: {2265 status,
2259 state: status,2266 numTurns: result?.num_turns,
2260 ...(result?.stop_reason !== undefined ? { stopReason: result.stop_reason } : {}),2267 hasUsage: result?.usage !== undefined,
2261 ...(result?.usage ? { usage: result.usage } : {}),2268 ...(errorMessage ? { errorMessage } : {}),
2262 ...(result?.modelUsage ? { modelUsage: result.modelUsage } : {}),
2263 ...(typeof result?.total_cost_usd === "number"
2264 ? { totalCostUsd: result.total_cost_usd }
2265 : {}),
2266 ...(errorMessage ? { errorMessage } : {}),
2267 },
2268 providerRefs: {},
2269 });2269 });
2270 return;2270 return;
2271 }2271 }