diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 769310380cca8f73ee5c98ed2516568ea1de267f..1bd4e274da121b87a5abedc439f9d91530924a15 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -381,6 +381,81 @@ describe("CheckpointReactor", () => { ).toBe("v2\n"); }); + it("ignores auxiliary thread turn completion while primary turn is active", 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-primary-running"), + threadId: ThreadId.makeUnsafe("thread-1"), + session: { + threadId: ThreadId.makeUnsafe("thread-1"), + status: "running", + providerName: "codex", + providerSessionId: asSessionId("sess-1"), + providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + approvalPolicy: "on-request", + sandboxMode: "workspace-write", + activeTurnId: asTurnId("turn-main"), + lastError: null, + updatedAt: createdAt, + }, + createdAt, + }), + ); + + harness.provider.emit({ + type: "turn.started", + eventId: EventId.makeUnsafe("evt-turn-started-main"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + turnId: asTurnId("turn-main"), + }); + await waitForGitRefExists( + harness.cwd, + checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0), + ); + + fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8"); + + harness.provider.emit({ + type: "turn.completed", + eventId: EventId.makeUnsafe("evt-turn-completed-aux"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: ProviderThreadId.makeUnsafe("provider-thread-aux"), + turnId: asTurnId("turn-aux"), + status: "completed", + }); + + await Effect.runPromise(Effect.sleep("40 millis")); + const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); + const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(midThread?.checkpoints).toHaveLength(0); + + harness.provider.emit({ + type: "turn.completed", + eventId: EventId.makeUnsafe("evt-turn-completed-main"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + turnId: asTurnId("turn-main"), + status: "completed", + }); + + const thread = await waitForThread( + harness.engine, + (entry) => entry.latestTurnId === "turn-main" && entry.checkpoints.length === 1, + ); + expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); + }); + it("appends capture failure activity when turn diff summary cannot be derived", async () => { const harness = await createHarness({ seedFilesystemCheckpoints: false }); const createdAt = new Date().toISOString(); diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index cd4eba3d2c61b8646f936f07a94a5222ebea5106..71b51ba737be24dc3e84d075efad07e09fe6e9f8 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -3,6 +3,7 @@ import { EventId, MessageId, ProviderSessionId, + ProviderThreadId, ThreadId, TurnId, type OrchestrationEvent, @@ -36,6 +37,17 @@ function toTurnId(value: string | undefined): TurnId | null { return value === undefined ? null : TurnId.makeUnsafe(value); } +function toProviderThreadId(value: string | undefined): ProviderThreadId | null { + return value === undefined ? null : ProviderThreadId.makeUnsafe(value); +} + +function sameId(left: string | null | undefined, right: string | null | undefined): boolean { + if (left === null || left === undefined || right === null || right === undefined) { + return false; + } + return left === right; +} + function checkpointStatusFromRuntime(status: string | undefined): "ready" | "missing" | "error" { switch (status) { case "failed": @@ -173,6 +185,21 @@ const make = Effect.gen(function* () { return; } + const projectedProviderThreadId = thread.session?.providerThreadId ?? null; + const eventProviderThreadId = toProviderThreadId(event.threadId); + if ( + projectedProviderThreadId !== null && + eventProviderThreadId !== null && + !sameId(projectedProviderThreadId, eventProviderThreadId) + ) { + return; + } + + // When a primary turn is active, only that turn may produce completion checkpoints. + if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, turnId)) { + return; + } + if (thread.checkpoints.some((checkpoint) => checkpoint.turnId === turnId)) { return; } diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 7a3a029e89a0fccd7990a7601486acb5af722b06..fa254887d8f476fd660f8447a1ac1225efee5379 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -31,6 +31,8 @@ import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeInge const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); +const asProviderThreadId = (value: string): ProviderThreadId => + ProviderThreadId.makeUnsafe(value); const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); @@ -220,6 +222,112 @@ describe("ProviderRuntimeIngestion", () => { expect(thread.session?.lastError).toBe("turn failed"); }); + it("ignores auxiliary turn completions from a different provider thread", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-primary"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + threadId: asProviderThreadId("provider-thread-1"), + turnId: asProviderTurnId("turn-primary"), + }); + + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && thread.session?.activeTurnId === "turn-primary", + ); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-aux"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-aux"), + turnId: asProviderTurnId("turn-aux"), + status: "completed", + }); + + await Effect.runPromise(Effect.sleep("40 millis")); + const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); + const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(midThread?.session?.status).toBe("running"); + expect(midThread?.session?.activeTurnId).toBe("turn-primary"); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-primary"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-1"), + turnId: asProviderTurnId("turn-primary"), + 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(); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-guarded"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + turnId: asProviderTurnId("turn-guarded-main"), + }); + + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && + thread.session?.activeTurnId === "turn-guarded-main", + ); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-guarded-other"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + turnId: asProviderTurnId("turn-guarded-other"), + status: "completed", + }); + + await Effect.runPromise(Effect.sleep("40 millis")); + const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); + const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(midThread?.session?.status).toBe("running"); + expect(midThread?.session?.activeTurnId).toBe("turn-guarded-main"); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-guarded-main"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + turnId: asProviderTurnId("turn-guarded-main"), + status: "completed", + }); + + await waitForThread( + harness.engine, + (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null, + ); + }); + it("maps message delta/completed into finalized assistant messages", 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 9da54b2a26363079a13dc2092fcc450ac45ec128..8bc4007c9ab259cffa866c22584c4ca73e03b132 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -33,6 +33,7 @@ const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120); const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000; const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120); const MAX_BUFFERED_ASSISTANT_CHARS = 24_000; +const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0"; type TurnStartRequestedDomainEvent = Extract< OrchestrationEvent, @@ -53,6 +54,17 @@ function toTurnId(value: string | undefined): TurnId | undefined { return value === undefined ? undefined : TurnId.makeUnsafe(value); } +function toProviderThreadId(value: string | undefined): ProviderThreadId | null { + return value === undefined ? null : ProviderThreadId.makeUnsafe(value); +} + +function sameId(left: string | null | undefined, right: string | null | undefined): boolean { + if (left === null || left === undefined || right === null || right === undefined) { + return false; + } + return left === right; +} + function truncateDetail(value: string, limit = 180): string { return value.length > limit ? `${value.slice(0, limit - 3)}...` : value; } @@ -332,6 +344,62 @@ const make = Effect.gen(function* () { if (!thread) return; const now = event.createdAt; + const sessionProviderThreadId = thread.session?.providerThreadId ?? null; + const eventProviderThreadId = toProviderThreadId(event.threadId); + const eventTurnId = toTurnId(event.turnId); + const activeTurnId = thread.session?.activeTurnId ?? null; + + const matchesThreadScope = + eventProviderThreadId === null || + sessionProviderThreadId === null || + sameId(eventProviderThreadId, sessionProviderThreadId); + const conflictsWithActiveTurn = + activeTurnId !== null && eventTurnId !== undefined && !sameId(activeTurnId, eventTurnId); + const missingTurnForActiveTurn = activeTurnId !== null && eventTurnId === undefined; + + const shouldApplyThreadLifecycle = (() => { + if (!STRICT_PROVIDER_LIFECYCLE_GUARD) { + return true; + } + switch (event.type) { + case "session.exited": + return true; + case "session.started": + case "thread.started": + if (!matchesThreadScope) { + return false; + } + // Never let auxiliary/provider-side spawned threads replace the primary thread binding. + if ( + eventProviderThreadId !== null && + sessionProviderThreadId !== null && + !sameId(eventProviderThreadId, sessionProviderThreadId) + ) { + return false; + } + return true; + case "turn.started": + if (!matchesThreadScope) { + return false; + } + return !conflictsWithActiveTurn; + case "turn.completed": + if (!matchesThreadScope) { + return false; + } + if (conflictsWithActiveTurn || missingTurnForActiveTurn) { + return false; + } + // Only the active turn may close the lifecycle state. + if (activeTurnId !== null && eventTurnId !== undefined) { + return sameId(activeTurnId, eventTurnId); + } + // Without an active turn, only accept completion when no thread mismatch signal exists. + return eventProviderThreadId === null || sessionProviderThreadId === null; + default: + return true; + } + })(); if ( event.type === "session.started" || @@ -340,8 +408,7 @@ const make = Effect.gen(function* () { event.type === "turn.started" || event.type === "turn.completed" ) { - const activeTurnId = - event.type === "turn.started" ? (toTurnId(event.turnId) ?? null) : null; + const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) : null; const providerThreadIdFromEvent = event.type === "thread.started" ? ProviderThreadId.makeUnsafe(event.threadId) @@ -349,7 +416,7 @@ const make = Effect.gen(function* () { ? ProviderThreadId.makeUnsafe(event.threadId) : null; const providerThreadId = - providerThreadIdFromEvent ?? thread.session?.providerThreadId ?? null; + providerThreadIdFromEvent ?? sessionProviderThreadId ?? null; const status = event.type === "turn.started" ? "running" @@ -365,24 +432,26 @@ const make = Effect.gen(function* () { ? null : (thread.session?.lastError ?? null); - yield* orchestrationEngine.dispatch({ - type: "thread.session.set", - commandId: providerCommandId(event, "thread-session-set"), - threadId: thread.id, - session: { + if (shouldApplyThreadLifecycle) { + yield* orchestrationEngine.dispatch({ + type: "thread.session.set", + commandId: providerCommandId(event, "thread-session-set"), threadId: thread.id, - status, - providerName: event.provider, - providerSessionId: event.sessionId, - providerThreadId, - approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY, - sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE, - activeTurnId, - lastError, - updatedAt: now, - }, - createdAt: now, - }); + session: { + threadId: thread.id, + status, + providerName: event.provider, + providerSessionId: event.sessionId, + providerThreadId, + approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY, + sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE, + activeTurnId: nextActiveTurnId, + lastError, + updatedAt: now, + }, + createdAt: now, + }); + } } if (event.type === "message.delta" && event.delta.length > 0) { @@ -470,29 +539,38 @@ const make = Effect.gen(function* () { } if (event.type === "runtime.error") { + const shouldApplyRuntimeError = !STRICT_PROVIDER_LIFECYCLE_GUARD + ? true + : matchesThreadScope && + (activeTurnId === null || + eventTurnId === undefined || + sameId(activeTurnId, eventTurnId)); + const providerThreadId = event.threadId !== undefined ? ProviderThreadId.makeUnsafe(event.threadId) : (thread.session?.providerThreadId ?? null); - yield* orchestrationEngine.dispatch({ - type: "thread.session.set", - commandId: providerCommandId(event, "runtime-error-session-set"), - threadId: thread.id, - session: { + if (shouldApplyRuntimeError) { + yield* orchestrationEngine.dispatch({ + type: "thread.session.set", + commandId: providerCommandId(event, "runtime-error-session-set"), threadId: thread.id, - status: "error", - providerName: event.provider, - providerSessionId: event.sessionId, - providerThreadId, - approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY, - sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE, - activeTurnId: toTurnId(event.turnId) ?? null, - lastError: event.message, - updatedAt: now, - }, - createdAt: now, - }); + session: { + threadId: thread.id, + status: "error", + providerName: event.provider, + providerSessionId: event.sessionId, + providerThreadId, + approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY, + sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE, + activeTurnId: eventTurnId ?? null, + lastError: event.message, + updatedAt: now, + }, + createdAt: now, + }); + } } const activities = runtimeEventToActivities(event); diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 9e87316f7741c273dc50d80056388f819902c05f..d6976479af046e4ec32095239011c59b0bc58b53 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -47,6 +47,7 @@ import { const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); +const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); @@ -564,7 +565,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: ThreadId.makeUnsafe("thread-1"), + threadId: asProviderThreadId("thread-1"), turnId: asTurnId("turn-1"), status: "completed", }; @@ -612,7 +613,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: ThreadId.makeUnsafe("thread-1"), + threadId: asProviderThreadId("thread-1"), turnId: asTurnId("turn-1"), toolKind: "command", title: "Command run", @@ -624,7 +625,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: ThreadId.makeUnsafe("thread-1"), + threadId: asProviderThreadId("thread-1"), turnId: asTurnId("turn-1"), delta: "hello", }, @@ -634,7 +635,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => { provider: "codex", sessionId: session.sessionId, createdAt: new Date().toISOString(), - threadId: ThreadId.makeUnsafe("thread-1"), + threadId: asProviderThreadId("thread-1"), turnId: asTurnId("turn-1"), status: "completed", }, diff --git a/docs/adr/0001-provider-runtime-lifecycle-ownership.md b/docs/adr/0001-provider-runtime-lifecycle-ownership.md new file mode 100644 index 0000000000000000000000000000000000000000..1c5afa5093a0d5424f07cd171c5c1dd6744e1a4e --- /dev/null +++ b/docs/adr/0001-provider-runtime-lifecycle-ownership.md @@ -0,0 +1,40 @@ +# ADR 0001: Provider Runtime Lifecycle Ownership + +## Status +Accepted + +## Context +Provider runtime streams can include auxiliary work (for example collab/child-agent turns) under the same provider session as a user-visible primary turn. + +The previous orchestration ingestion and checkpoint flows keyed lifecycle updates by `providerSessionId` alone. This allowed auxiliary `turn.completed` events to: + +- mark thread sessions as `ready` before the primary turn completed +- clear `activeTurnId` for a still-running primary turn +- emit checkpoint turn-diff summaries for auxiliary turns in the main thread timeline + +## Decision +Define and enforce a lifecycle ownership invariant: + +- Only events in the primary thread/turn lane may mutate thread session lifecycle state (`status`, `activeTurnId`, `providerThreadId`) and checkpoint turn completion. +- Auxiliary events may still append messages and activities. + +Current enforcement is implemented with scope guards: + +- Runtime ingestion only applies lifecycle transitions when the event targets the active primary scope (provider thread and active turn checks). +- Checkpoint capture ignores `turn.completed` events from non-primary provider thread scope, and ignores non-active turn completions while a primary turn is active. + +## Consequences + +### Positive +- Prevents premature loss of "working" state from auxiliary turn completion. +- Prevents auxiliary changed-files/checkpoint cards from appearing as if they were primary turn completion. +- Keeps useful auxiliary activity/message visibility. + +### Negative +- Runtime events that omit thread/turn identifiers are treated conservatively in lifecycle transitions. +- Full explicit lane modeling is still desirable in contracts for long-term clarity. + +## Follow-up +1. Add explicit runtime lane metadata (`primary` vs `auxiliary`) to canonical provider runtime events. +2. Persist orchestration-to-provider turn binding (`TurnId` <-> `ProviderTurnId`) for stronger reconciliation and replay safety. +3. Promote lifecycle ownership checks from heuristic guards to schema-level invariants once lane metadata is available. diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index daf70550425ebf947b6e95b9fa41b6aa488182ac..c74645ca40f45fd33c2bbc1576bab1abb90f6d9d 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -9,15 +9,13 @@ import { ProviderSessionId, ProviderThreadId, ProviderTurnId, - ThreadId, - TurnId, IsoDateTime, } from "./baseSchemas"; import { ProviderApprovalDecision, ProviderKind, ProviderRequestKind } from "./orchestration"; const TrimmedNonEmptyStringSchema = TrimmedNonEmptyString; -const RuntimeThreadIdSchema = Schema.Union([ThreadId, ProviderThreadId]); -const RuntimeTurnIdSchema = Schema.Union([TurnId, ProviderTurnId]); +const RuntimeThreadIdSchema = ProviderThreadId; +const RuntimeTurnIdSchema = ProviderTurnId; export const ProviderRuntimeToolKind = Schema.Union([ProviderRequestKind, Schema.Literal("other")]); export type ProviderRuntimeToolKind = typeof ProviderRuntimeToolKind.Type;