From 92f1a4aa3ca9cb11f10e3ab97d08970f443d22aa Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 27 Feb 2026 17:10:50 -0800 Subject: [PATCH] Ignore stale resume cursors and Claude placeholder thread IDs - Do not pass `resumeCursor` when restarting a session after changing providers - Treat synthetic Claude thread IDs (`claude-thread-*`) as unscoped during runtime ingestion - Add tests covering provider-switch restart behavior and Claude turn lifecycle acceptance --- .../Layers/ProviderCommandReactor.test.ts | 2 +- .../Layers/ProviderCommandReactor.ts | 4 +- .../Layers/ProviderRuntimeIngestion.test.ts | 59 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 26 ++++++-- 4 files changed, 83 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 7d6183b9635d2039a3eef0b964ffaeb24fc9efc4..75b8e397ecc1d1bc9cfa4cd4cc8613840721c261 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -442,10 +442,10 @@ describe("ProviderCommandReactor", () => { expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") }); expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({ provider: "claudeCode", - resumeCursor: { opaque: "cursor-1" }, approvalPolicy: "on-request", sandboxMode: "workspace-write", }); + expect(harness.startSession.mock.calls[1]?.[1]).not.toHaveProperty("resumeCursor"); const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 9fc90384184e7fd5eb0a32524f52447ec7d813cb..ca075215e62b93f5f060fbd319813b1ca8e2b04b 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -245,7 +245,9 @@ const make = Effect.gen(function* () { return existingSessionId; } - const resumeCursor = yield* resolveResumeCursorForSession(existingSessionId); + const resumeCursor = providerChanged + ? undefined + : yield* resolveResumeCursorForSession(existingSessionId); const restartedSession = yield* startProviderSession({ ...(resumeCursor !== undefined ? { resumeCursor } : {}), ...(options?.provider !== undefined ? { provider: options.provider } : {}), diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 55a8452be89120bb81bbe1462b91bd8ba04dbb33..caeb1d95182cac967b5a32311de004ff66274b8e 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -281,6 +281,65 @@ describe("ProviderRuntimeIngestion", () => { ); }); + it("accepts claude turn lifecycle when seeded thread id is a synthetic placeholder", async () => { + const harness = await createHarness(); + const seededAt = new Date().toISOString(); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.makeUnsafe("cmd-session-seed-claude-placeholder"), + threadId: ThreadId.makeUnsafe("thread-1"), + session: { + threadId: ThreadId.makeUnsafe("thread-1"), + status: "ready", + providerName: "claudeCode", + providerSessionId: asSessionId("sess-1"), + providerThreadId: asProviderThreadId("claude-thread-placeholder"), + approvalPolicy: "on-request", + sandboxMode: "workspace-write", + activeTurnId: null, + updatedAt: seededAt, + lastError: null, + }, + createdAt: seededAt, + }), + ); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-claude-placeholder"), + provider: "claudeCode", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-real"), + turnId: asProviderTurnId("turn-claude-placeholder"), + }); + + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && + thread.session?.activeTurnId === "turn-claude-placeholder", + ); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-claude-placeholder"), + provider: "claudeCode", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-real"), + turnId: asProviderTurnId("turn-claude-placeholder"), + status: "completed", + }); + + await waitForThread( + harness.engine, + (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null, + ); + }); + it("ignores non-active turn completion when runtime omits thread id", 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 6668f5e5b38790217fec1b45ce1bd5fb15cd1714..b0611e38bdea17baa5db730fa336b50edf89d0c6 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -65,6 +65,13 @@ function sameId(left: string | null | undefined, right: string | null | undefine return left === right; } +function isSyntheticClaudeThreadId( + provider: ProviderRuntimeEvent["provider"], + threadId: ProviderThreadId | null, +): boolean { + return provider === "claudeCode" && threadId !== null && threadId.startsWith("claude-thread-"); +} + function truncateDetail(value: string, limit = 180): string { return value.length > limit ? `${value.slice(0, limit - 3)}...` : value; } @@ -345,14 +352,20 @@ const make = Effect.gen(function* () { const now = event.createdAt; const sessionProviderThreadId = thread.session?.providerThreadId ?? null; + const scopedSessionProviderThreadId = isSyntheticClaudeThreadId( + event.provider, + sessionProviderThreadId, + ) + ? null + : sessionProviderThreadId; const eventProviderThreadId = toProviderThreadId(event.threadId); const eventTurnId = toTurnId("turnId" in event ? event.turnId : undefined); const activeTurnId = thread.session?.activeTurnId ?? null; const matchesThreadScope = eventProviderThreadId === null || - sessionProviderThreadId === null || - sameId(eventProviderThreadId, sessionProviderThreadId); + scopedSessionProviderThreadId === null || + sameId(eventProviderThreadId, scopedSessionProviderThreadId); const conflictsWithActiveTurn = activeTurnId !== null && eventTurnId !== undefined && !sameId(activeTurnId, eventTurnId); const missingTurnForActiveTurn = activeTurnId !== null && eventTurnId === undefined; @@ -372,8 +385,8 @@ const make = Effect.gen(function* () { // Never let auxiliary/provider-side spawned threads replace the primary thread binding. if ( eventProviderThreadId !== null && - sessionProviderThreadId !== null && - !sameId(eventProviderThreadId, sessionProviderThreadId) + scopedSessionProviderThreadId !== null && + !sameId(eventProviderThreadId, scopedSessionProviderThreadId) ) { return false; } @@ -395,7 +408,7 @@ const make = Effect.gen(function* () { return sameId(activeTurnId, eventTurnId); } // Without an active turn, only accept completion when no thread mismatch signal exists. - return eventProviderThreadId === null || sessionProviderThreadId === null; + return eventProviderThreadId === null || scopedSessionProviderThreadId === null; default: return true; } @@ -415,7 +428,8 @@ const make = Effect.gen(function* () { : event.threadId !== undefined ? ProviderThreadId.makeUnsafe(event.threadId) : null; - const providerThreadId = providerThreadIdFromEvent ?? sessionProviderThreadId ?? null; + const providerThreadId = + providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null; const status = event.type === "turn.started" ? "running" -- 2.54.0