authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-27 10:01:26-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-27 14:10:34-08:00
log872b90737211f883b73879e3feae5563140a8fbe
tree7541835a1320ecfe63b181542a21316c71319701
parent0b8e098c105297bc9564dc8bb792af21ec91ba4a

Guard provider lifecycle updates to primary thread and active turn

- Ignore auxiliary/non-active turn completions for session lifecycle and checkpoints - Restrict runtime thread/turn IDs to provider IDs in contracts - Add regression tests and ADR for lifecycle ownership rules

7 files changed, 372 insertions(+), 45 deletions(-)

apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+75
...@@ -381,6 +381,81 @@ describe("CheckpointReactor", () => {...@@ -381,6 +381,81 @@ describe("CheckpointReactor", () => {
381 ).toBe("v2\n");381 ).toBe("v2\n");
382 });382 });
383383
384 it("ignores auxiliary thread turn completion while primary turn is active", async () => {
385 const harness = await createHarness({ seedFilesystemCheckpoints: false });
386 const createdAt = new Date().toISOString();
387
388 await Effect.runPromise(
389 harness.engine.dispatch({
390 type: "thread.session.set",
391 commandId: CommandId.makeUnsafe("cmd-session-set-primary-running"),
392 threadId: ThreadId.makeUnsafe("thread-1"),
393 session: {
394 threadId: ThreadId.makeUnsafe("thread-1"),
395 status: "running",
396 providerName: "codex",
397 providerSessionId: asSessionId("sess-1"),
398 providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
399 approvalPolicy: "on-request",
400 sandboxMode: "workspace-write",
401 activeTurnId: asTurnId("turn-main"),
402 lastError: null,
403 updatedAt: createdAt,
404 },
405 createdAt,
406 }),
407 );
408
409 harness.provider.emit({
410 type: "turn.started",
411 eventId: EventId.makeUnsafe("evt-turn-started-main"),
412 provider: "codex",
413 sessionId: asSessionId("sess-1"),
414 createdAt: new Date().toISOString(),
415 threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
416 turnId: asTurnId("turn-main"),
417 });
418 await waitForGitRefExists(
419 harness.cwd,
420 checkpointRefForThreadTurn(ThreadId.makeUnsafe("thread-1"), 0),
421 );
422
423 fs.writeFileSync(path.join(harness.cwd, "README.md"), "v2\n", "utf8");
424
425 harness.provider.emit({
426 type: "turn.completed",
427 eventId: EventId.makeUnsafe("evt-turn-completed-aux"),
428 provider: "codex",
429 sessionId: asSessionId("sess-1"),
430 createdAt: new Date().toISOString(),
431 threadId: ProviderThreadId.makeUnsafe("provider-thread-aux"),
432 turnId: asTurnId("turn-aux"),
433 status: "completed",
434 });
435
436 await Effect.runPromise(Effect.sleep("40 millis"));
437 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
438 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
439 expect(midThread?.checkpoints).toHaveLength(0);
440
441 harness.provider.emit({
442 type: "turn.completed",
443 eventId: EventId.makeUnsafe("evt-turn-completed-main"),
444 provider: "codex",
445 sessionId: asSessionId("sess-1"),
446 createdAt: new Date().toISOString(),
447 threadId: ProviderThreadId.makeUnsafe("provider-thread-1"),
448 turnId: asTurnId("turn-main"),
449 status: "completed",
450 });
451
452 const thread = await waitForThread(
453 harness.engine,
454 (entry) => entry.latestTurnId === "turn-main" && entry.checkpoints.length === 1,
455 );
456 expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
457 });
458
384 it("appends capture failure activity when turn diff summary cannot be derived", async () => {459 it("appends capture failure activity when turn diff summary cannot be derived", async () => {
385 const harness = await createHarness({ seedFilesystemCheckpoints: false });460 const harness = await createHarness({ seedFilesystemCheckpoints: false });
386 const createdAt = new Date().toISOString();461 const createdAt = new Date().toISOString();
apps/server/src/orchestration/Layers/CheckpointReactor.ts+27
...@@ -3,6 +3,7 @@ import {...@@ -3,6 +3,7 @@ import {
3 EventId,3 EventId,
4 MessageId,4 MessageId,
5 ProviderSessionId,5 ProviderSessionId,
6 ProviderThreadId,
6 ThreadId,7 ThreadId,
7 TurnId,8 TurnId,
8 type OrchestrationEvent,9 type OrchestrationEvent,
...@@ -36,6 +37,17 @@ function toTurnId(value: string | undefined): TurnId | null {...@@ -36,6 +37,17 @@ function toTurnId(value: string | undefined): TurnId | null {
36 return value === undefined ? null : TurnId.makeUnsafe(value);37 return value === undefined ? null : TurnId.makeUnsafe(value);
37}38}
3839
40function toProviderThreadId(value: string | undefined): ProviderThreadId | null {
41 return value === undefined ? null : ProviderThreadId.makeUnsafe(value);
42}
43
44function sameId(left: string | null | undefined, right: string | null | undefined): boolean {
45 if (left === null || left === undefined || right === null || right === undefined) {
46 return false;
47 }
48 return left === right;
49}
50
39function checkpointStatusFromRuntime(status: string | undefined): "ready" | "missing" | "error" {51function checkpointStatusFromRuntime(status: string | undefined): "ready" | "missing" | "error" {
40 switch (status) {52 switch (status) {
41 case "failed":53 case "failed":
...@@ -173,6 +185,21 @@ const make = Effect.gen(function* () {...@@ -173,6 +185,21 @@ const make = Effect.gen(function* () {
173 return;185 return;
174 }186 }
175187
188 const projectedProviderThreadId = thread.session?.providerThreadId ?? null;
189 const eventProviderThreadId = toProviderThreadId(event.threadId);
190 if (
191 projectedProviderThreadId !== null &&
192 eventProviderThreadId !== null &&
193 !sameId(projectedProviderThreadId, eventProviderThreadId)
194 ) {
195 return;
196 }
197
198 // When a primary turn is active, only that turn may produce completion checkpoints.
199 if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, turnId)) {
200 return;
201 }
202
176 if (thread.checkpoints.some((checkpoint) => checkpoint.turnId === turnId)) {203 if (thread.checkpoints.some((checkpoint) => checkpoint.turnId === turnId)) {
177 return;204 return;
178 }205 }
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+108
...@@ -31,6 +31,8 @@ import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeInge...@@ -31,6 +31,8 @@ import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeInge
3131
32const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);32const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value);
33const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);33const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
34const asProviderThreadId = (value: string): ProviderThreadId =>
35 ProviderThreadId.makeUnsafe(value);
34const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);36const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);
35const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);37const asItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value);
36const asEventId = (value: string): EventId => EventId.makeUnsafe(value);38const asEventId = (value: string): EventId => EventId.makeUnsafe(value);
...@@ -220,6 +222,112 @@ describe("ProviderRuntimeIngestion", () => {...@@ -220,6 +222,112 @@ describe("ProviderRuntimeIngestion", () => {
220 expect(thread.session?.lastError).toBe("turn failed");222 expect(thread.session?.lastError).toBe("turn failed");
221 });223 });
222224
225 it("ignores auxiliary turn completions from a different provider thread", async () => {
226 const harness = await createHarness();
227 const now = new Date().toISOString();
228
229 harness.emit({
230 type: "turn.started",
231 eventId: asEventId("evt-turn-started-primary"),
232 provider: "codex",
233 sessionId: asSessionId("sess-1"),
234 createdAt: now,
235 threadId: asProviderThreadId("provider-thread-1"),
236 turnId: asProviderTurnId("turn-primary"),
237 });
238
239 await waitForThread(
240 harness.engine,
241 (thread) =>
242 thread.session?.status === "running" && thread.session?.activeTurnId === "turn-primary",
243 );
244
245 harness.emit({
246 type: "turn.completed",
247 eventId: asEventId("evt-turn-completed-aux"),
248 provider: "codex",
249 sessionId: asSessionId("sess-1"),
250 createdAt: new Date().toISOString(),
251 threadId: asProviderThreadId("provider-thread-aux"),
252 turnId: asProviderTurnId("turn-aux"),
253 status: "completed",
254 });
255
256 await Effect.runPromise(Effect.sleep("40 millis"));
257 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
258 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
259 expect(midThread?.session?.status).toBe("running");
260 expect(midThread?.session?.activeTurnId).toBe("turn-primary");
261
262 harness.emit({
263 type: "turn.completed",
264 eventId: asEventId("evt-turn-completed-primary"),
265 provider: "codex",
266 sessionId: asSessionId("sess-1"),
267 createdAt: new Date().toISOString(),
268 threadId: asProviderThreadId("provider-thread-1"),
269 turnId: asProviderTurnId("turn-primary"),
270 status: "completed",
271 });
272
273 await waitForThread(
274 harness.engine,
275 (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null,
276 );
277 });
278
279 it("ignores non-active turn completion when runtime omits thread id", async () => {
280 const harness = await createHarness();
281 const now = new Date().toISOString();
282
283 harness.emit({
284 type: "turn.started",
285 eventId: asEventId("evt-turn-started-guarded"),
286 provider: "codex",
287 sessionId: asSessionId("sess-1"),
288 createdAt: now,
289 turnId: asProviderTurnId("turn-guarded-main"),
290 });
291
292 await waitForThread(
293 harness.engine,
294 (thread) =>
295 thread.session?.status === "running" &&
296 thread.session?.activeTurnId === "turn-guarded-main",
297 );
298
299 harness.emit({
300 type: "turn.completed",
301 eventId: asEventId("evt-turn-completed-guarded-other"),
302 provider: "codex",
303 sessionId: asSessionId("sess-1"),
304 createdAt: new Date().toISOString(),
305 turnId: asProviderTurnId("turn-guarded-other"),
306 status: "completed",
307 });
308
309 await Effect.runPromise(Effect.sleep("40 millis"));
310 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
311 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
312 expect(midThread?.session?.status).toBe("running");
313 expect(midThread?.session?.activeTurnId).toBe("turn-guarded-main");
314
315 harness.emit({
316 type: "turn.completed",
317 eventId: asEventId("evt-turn-completed-guarded-main"),
318 provider: "codex",
319 sessionId: asSessionId("sess-1"),
320 createdAt: new Date().toISOString(),
321 turnId: asProviderTurnId("turn-guarded-main"),
322 status: "completed",
323 });
324
325 await waitForThread(
326 harness.engine,
327 (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null,
328 );
329 });
330
223 it("maps message delta/completed into finalized assistant messages", async () => {331 it("maps message delta/completed into finalized assistant messages", async () => {
224 const harness = await createHarness();332 const harness = await createHarness();
225 const now = new Date().toISOString();333 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+115-37
...@@ -33,6 +33,7 @@ const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);...@@ -33,6 +33,7 @@ const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);
33const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;33const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
34const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);34const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);
35const MAX_BUFFERED_ASSISTANT_CHARS = 24_000;35const MAX_BUFFERED_ASSISTANT_CHARS = 24_000;
36const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0";
3637
37type TurnStartRequestedDomainEvent = Extract<38type TurnStartRequestedDomainEvent = Extract<
38 OrchestrationEvent,39 OrchestrationEvent,
...@@ -53,6 +54,17 @@ function toTurnId(value: string | undefined): TurnId | undefined {...@@ -53,6 +54,17 @@ function toTurnId(value: string | undefined): TurnId | undefined {
53 return value === undefined ? undefined : TurnId.makeUnsafe(value);54 return value === undefined ? undefined : TurnId.makeUnsafe(value);
54}55}
5556
57function toProviderThreadId(value: string | undefined): ProviderThreadId | null {
58 return value === undefined ? null : ProviderThreadId.makeUnsafe(value);
59}
60
61function sameId(left: string | null | undefined, right: string | null | undefined): boolean {
62 if (left === null || left === undefined || right === null || right === undefined) {
63 return false;
64 }
65 return left === right;
66}
67
56function truncateDetail(value: string, limit = 180): string {68function truncateDetail(value: string, limit = 180): string {
57 return value.length > limit ? `${value.slice(0, limit - 3)}...` : value;69 return value.length > limit ? `${value.slice(0, limit - 3)}...` : value;
58}70}
...@@ -332,6 +344,62 @@ const make = Effect.gen(function* () {...@@ -332,6 +344,62 @@ const make = Effect.gen(function* () {
332 if (!thread) return;344 if (!thread) return;
333345
334 const now = event.createdAt;346 const now = event.createdAt;
347 const sessionProviderThreadId = thread.session?.providerThreadId ?? null;
348 const eventProviderThreadId = toProviderThreadId(event.threadId);
349 const eventTurnId = toTurnId(event.turnId);
350 const activeTurnId = thread.session?.activeTurnId ?? null;
351
352 const matchesThreadScope =
353 eventProviderThreadId === null ||
354 sessionProviderThreadId === null ||
355 sameId(eventProviderThreadId, sessionProviderThreadId);
356 const conflictsWithActiveTurn =
357 activeTurnId !== null && eventTurnId !== undefined && !sameId(activeTurnId, eventTurnId);
358 const missingTurnForActiveTurn = activeTurnId !== null && eventTurnId === undefined;
359
360 const shouldApplyThreadLifecycle = (() => {
361 if (!STRICT_PROVIDER_LIFECYCLE_GUARD) {
362 return true;
363 }
364 switch (event.type) {
365 case "session.exited":
366 return true;
367 case "session.started":
368 case "thread.started":
369 if (!matchesThreadScope) {
370 return false;
371 }
372 // Never let auxiliary/provider-side spawned threads replace the primary thread binding.
373 if (
374 eventProviderThreadId !== null &&
375 sessionProviderThreadId !== null &&
376 !sameId(eventProviderThreadId, sessionProviderThreadId)
377 ) {
378 return false;
379 }
380 return true;
381 case "turn.started":
382 if (!matchesThreadScope) {
383 return false;
384 }
385 return !conflictsWithActiveTurn;
386 case "turn.completed":
387 if (!matchesThreadScope) {
388 return false;
389 }
390 if (conflictsWithActiveTurn || missingTurnForActiveTurn) {
391 return false;
392 }
393 // Only the active turn may close the lifecycle state.
394 if (activeTurnId !== null && eventTurnId !== undefined) {
395 return sameId(activeTurnId, eventTurnId);
396 }
397 // Without an active turn, only accept completion when no thread mismatch signal exists.
398 return eventProviderThreadId === null || sessionProviderThreadId === null;
399 default:
400 return true;
401 }
402 })();
335403
336 if (404 if (
337 event.type === "session.started" ||405 event.type === "session.started" ||
...@@ -340,8 +408,7 @@ const make = Effect.gen(function* () {...@@ -340,8 +408,7 @@ const make = Effect.gen(function* () {
340 event.type === "turn.started" ||408 event.type === "turn.started" ||
341 event.type === "turn.completed"409 event.type === "turn.completed"
342 ) {410 ) {
343 const activeTurnId =411 const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) : null;
344 event.type === "turn.started" ? (toTurnId(event.turnId) ?? null) : null;
345 const providerThreadIdFromEvent =412 const providerThreadIdFromEvent =
346 event.type === "thread.started"413 event.type === "thread.started"
347 ? ProviderThreadId.makeUnsafe(event.threadId)414 ? ProviderThreadId.makeUnsafe(event.threadId)
...@@ -349,7 +416,7 @@ const make = Effect.gen(function* () {...@@ -349,7 +416,7 @@ const make = Effect.gen(function* () {
349 ? ProviderThreadId.makeUnsafe(event.threadId)416 ? ProviderThreadId.makeUnsafe(event.threadId)
350 : null;417 : null;
351 const providerThreadId =418 const providerThreadId =
352 providerThreadIdFromEvent ?? thread.session?.providerThreadId ?? null;419 providerThreadIdFromEvent ?? sessionProviderThreadId ?? null;
353 const status =420 const status =
354 event.type === "turn.started"421 event.type === "turn.started"
355 ? "running"422 ? "running"
...@@ -365,24 +432,26 @@ const make = Effect.gen(function* () {...@@ -365,24 +432,26 @@ const make = Effect.gen(function* () {
365 ? null432 ? null
366 : (thread.session?.lastError ?? null);433 : (thread.session?.lastError ?? null);
367434
368 yield* orchestrationEngine.dispatch({435 if (shouldApplyThreadLifecycle) {
369 type: "thread.session.set",436 yield* orchestrationEngine.dispatch({
370 commandId: providerCommandId(event, "thread-session-set"),437 type: "thread.session.set",
371 threadId: thread.id,438 commandId: providerCommandId(event, "thread-session-set"),
372 session: {
373 threadId: thread.id,439 threadId: thread.id,
374 status,440 session: {
375 providerName: event.provider,441 threadId: thread.id,
376 providerSessionId: event.sessionId,442 status,
377 providerThreadId,443 providerName: event.provider,
378 approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY,444 providerSessionId: event.sessionId,
379 sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE,445 providerThreadId,
380 activeTurnId,446 approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY,
381 lastError,447 sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE,
382 updatedAt: now,448 activeTurnId: nextActiveTurnId,
383 },449 lastError,
384 createdAt: now,450 updatedAt: now,
385 });451 },
452 createdAt: now,
453 });
454 }
386 }455 }
387456
388 if (event.type === "message.delta" && event.delta.length > 0) {457 if (event.type === "message.delta" && event.delta.length > 0) {
...@@ -470,29 +539,38 @@ const make = Effect.gen(function* () {...@@ -470,29 +539,38 @@ const make = Effect.gen(function* () {
470 }539 }
471540
472 if (event.type === "runtime.error") {541 if (event.type === "runtime.error") {
542 const shouldApplyRuntimeError = !STRICT_PROVIDER_LIFECYCLE_GUARD
543 ? true
544 : matchesThreadScope &&
545 (activeTurnId === null ||
546 eventTurnId === undefined ||
547 sameId(activeTurnId, eventTurnId));
548
473 const providerThreadId =549 const providerThreadId =
474 event.threadId !== undefined550 event.threadId !== undefined
475 ? ProviderThreadId.makeUnsafe(event.threadId)551 ? ProviderThreadId.makeUnsafe(event.threadId)
476 : (thread.session?.providerThreadId ?? null);552 : (thread.session?.providerThreadId ?? null);
477553
478 yield* orchestrationEngine.dispatch({554 if (shouldApplyRuntimeError) {
479 type: "thread.session.set",555 yield* orchestrationEngine.dispatch({
480 commandId: providerCommandId(event, "runtime-error-session-set"),556 type: "thread.session.set",
481 threadId: thread.id,557 commandId: providerCommandId(event, "runtime-error-session-set"),
482 session: {
483 threadId: thread.id,558 threadId: thread.id,
484 status: "error",559 session: {
485 providerName: event.provider,560 threadId: thread.id,
486 providerSessionId: event.sessionId,561 status: "error",
487 providerThreadId,562 providerName: event.provider,
488 approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY,563 providerSessionId: event.sessionId,
489 sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE,564 providerThreadId,
490 activeTurnId: toTurnId(event.turnId) ?? null,565 approvalPolicy: thread.session?.approvalPolicy ?? DEFAULT_APPROVAL_POLICY,
491 lastError: event.message,566 sandboxMode: thread.session?.sandboxMode ?? DEFAULT_SANDBOX_MODE,
492 updatedAt: now,567 activeTurnId: eventTurnId ?? null,
493 },568 lastError: event.message,
494 createdAt: now,569 updatedAt: now,
495 });570 },
571 createdAt: now,
572 });
573 }
496 }574 }
497575
498 const activities = runtimeEventToActivities(event);576 const activities = runtimeEventToActivities(event);
apps/server/src/provider/Layers/ProviderService.test.ts+5-4
...@@ -47,6 +47,7 @@ import {...@@ -47,6 +47,7 @@ import {
4747
48const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);48const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value);
49const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);49const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);
50const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value);
50const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value);51const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value);
51const asEventId = (value: string): EventId => EventId.makeUnsafe(value);52const asEventId = (value: string): EventId => EventId.makeUnsafe(value);
52const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value);53const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value);
...@@ -564,7 +565,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {...@@ -564,7 +565,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
564 provider: "codex",565 provider: "codex",
565 sessionId: session.sessionId,566 sessionId: session.sessionId,
566 createdAt: new Date().toISOString(),567 createdAt: new Date().toISOString(),
567 threadId: ThreadId.makeUnsafe("thread-1"),568 threadId: asProviderThreadId("thread-1"),
568 turnId: asTurnId("turn-1"),569 turnId: asTurnId("turn-1"),
569 status: "completed",570 status: "completed",
570 };571 };
...@@ -612,7 +613,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {...@@ -612,7 +613,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
612 provider: "codex",613 provider: "codex",
613 sessionId: session.sessionId,614 sessionId: session.sessionId,
614 createdAt: new Date().toISOString(),615 createdAt: new Date().toISOString(),
615 threadId: ThreadId.makeUnsafe("thread-1"),616 threadId: asProviderThreadId("thread-1"),
616 turnId: asTurnId("turn-1"),617 turnId: asTurnId("turn-1"),
617 toolKind: "command",618 toolKind: "command",
618 title: "Command run",619 title: "Command run",
...@@ -624,7 +625,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {...@@ -624,7 +625,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
624 provider: "codex",625 provider: "codex",
625 sessionId: session.sessionId,626 sessionId: session.sessionId,
626 createdAt: new Date().toISOString(),627 createdAt: new Date().toISOString(),
627 threadId: ThreadId.makeUnsafe("thread-1"),628 threadId: asProviderThreadId("thread-1"),
628 turnId: asTurnId("turn-1"),629 turnId: asTurnId("turn-1"),
629 delta: "hello",630 delta: "hello",
630 },631 },
...@@ -634,7 +635,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {...@@ -634,7 +635,7 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
634 provider: "codex",635 provider: "codex",
635 sessionId: session.sessionId,636 sessionId: session.sessionId,
636 createdAt: new Date().toISOString(),637 createdAt: new Date().toISOString(),
637 threadId: ThreadId.makeUnsafe("thread-1"),638 threadId: asProviderThreadId("thread-1"),
638 turnId: asTurnId("turn-1"),639 turnId: asTurnId("turn-1"),
639 status: "completed",640 status: "completed",
640 },641 },
docs/adr/0001-provider-runtime-lifecycle-ownership.md created+40
...@@ -0,0 +1,40 @@
1# ADR 0001: Provider Runtime Lifecycle Ownership
2
3## Status
4Accepted
5
6## Context
7Provider runtime streams can include auxiliary work (for example collab/child-agent turns) under the same provider session as a user-visible primary turn.
8
9The previous orchestration ingestion and checkpoint flows keyed lifecycle updates by `providerSessionId` alone. This allowed auxiliary `turn.completed` events to:
10
11- mark thread sessions as `ready` before the primary turn completed
12- clear `activeTurnId` for a still-running primary turn
13- emit checkpoint turn-diff summaries for auxiliary turns in the main thread timeline
14
15## Decision
16Define and enforce a lifecycle ownership invariant:
17
18- Only events in the primary thread/turn lane may mutate thread session lifecycle state (`status`, `activeTurnId`, `providerThreadId`) and checkpoint turn completion.
19- Auxiliary events may still append messages and activities.
20
21Current enforcement is implemented with scope guards:
22
23- Runtime ingestion only applies lifecycle transitions when the event targets the active primary scope (provider thread and active turn checks).
24- Checkpoint capture ignores `turn.completed` events from non-primary provider thread scope, and ignores non-active turn completions while a primary turn is active.
25
26## Consequences
27
28### Positive
29- Prevents premature loss of "working" state from auxiliary turn completion.
30- Prevents auxiliary changed-files/checkpoint cards from appearing as if they were primary turn completion.
31- Keeps useful auxiliary activity/message visibility.
32
33### Negative
34- Runtime events that omit thread/turn identifiers are treated conservatively in lifecycle transitions.
35- Full explicit lane modeling is still desirable in contracts for long-term clarity.
36
37## Follow-up
381. Add explicit runtime lane metadata (`primary` vs `auxiliary`) to canonical provider runtime events.
392. Persist orchestration-to-provider turn binding (`TurnId` <-> `ProviderTurnId`) for stronger reconciliation and replay safety.
403. Promote lifecycle ownership checks from heuristic guards to schema-level invariants once lane metadata is available.
packages/contracts/src/providerRuntime.ts+2-4
...@@ -9,15 +9,13 @@ import {...@@ -9,15 +9,13 @@ import {
9 ProviderSessionId,9 ProviderSessionId,
10 ProviderThreadId,10 ProviderThreadId,
11 ProviderTurnId,11 ProviderTurnId,
12 ThreadId,
13 TurnId,
14 IsoDateTime,12 IsoDateTime,
15} from "./baseSchemas";13} from "./baseSchemas";
16import { ProviderApprovalDecision, ProviderKind, ProviderRequestKind } from "./orchestration";14import { ProviderApprovalDecision, ProviderKind, ProviderRequestKind } from "./orchestration";
1715
18const TrimmedNonEmptyStringSchema = TrimmedNonEmptyString;16const TrimmedNonEmptyStringSchema = TrimmedNonEmptyString;
19const RuntimeThreadIdSchema = Schema.Union([ThreadId, ProviderThreadId]);17const RuntimeThreadIdSchema = ProviderThreadId;
20const RuntimeTurnIdSchema = Schema.Union([TurnId, ProviderTurnId]);18const RuntimeTurnIdSchema = ProviderTurnId;
2119
22export const ProviderRuntimeToolKind = Schema.Union([ProviderRequestKind, Schema.Literal("other")]);20export const ProviderRuntimeToolKind = Schema.Union([ProviderRequestKind, Schema.Literal("other")]);
23export type ProviderRuntimeToolKind = typeof ProviderRuntimeToolKind.Type;21export type ProviderRuntimeToolKind = typeof ProviderRuntimeToolKind.Type;