authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-10-09 08:01:48+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-09 01:01:48-05:00
log7a2b1c72beed4f560aebb41aa7b26e0792eeb938
treeac35311d5778bb9d0d4a9c32fda2b648752a249f
parentc908cea52510ed29199c3cc477204da425601301
signature Signed by PGP key B5690EEEBB952194

fix(server): reconcile Pi native session rewinds (#13839)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

5 files changed, 861 insertions(+), 58 deletions(-)

apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts+253
...@@ -1,5 +1,8 @@...@@ -1,5 +1,8 @@
1import { assert, it } from "@effect/vitest";1import { assert, it } from "@effect/vitest";
2import {2import {
3 CheckpointId,
4 CheckpointRef,
5 CheckpointScopeId,
3 MessageId,6 MessageId,
4 type ModelSelection,7 type ModelSelection,
5 NodeId,8 NodeId,
...@@ -14,6 +17,8 @@ import {...@@ -14,6 +17,8 @@ import {
14 ProviderDriverKind,17 ProviderDriverKind,
15 ProviderInstanceId,18 ProviderInstanceId,
16 PlanId,19 PlanId,
20 type ProviderThreadId,
21 ProviderTurnId,
17 RunAttemptId,22 RunAttemptId,
18 RunId,23 RunId,
19 RuntimeRequestId,24 RuntimeRequestId,
...@@ -318,6 +323,254 @@ layer("ProviderEventIngestorV2", (it) => {...@@ -318,6 +323,254 @@ layer("ProviderEventIngestorV2", (it) => {
318 }),323 }),
319 );324 );
320325
326 it.effect("rolls back runs a provider rewound off its active branch", () =>
327 Effect.gen(function* () {
328 const now = yield* DateTime.now;
329 const eventSink = yield* EventSink.EventSinkV2;
330 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
331 const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
332 const idAllocator = yield* IdAllocator.IdAllocatorV2;
333 const threadEvent = yield* threadCreatedEvent(now);
334 const threadId = threadEvent.threadId;
335 const providerSessionId = yield* idAllocator.allocate.providerSession({
336 providerInstanceId: modelSelection.instanceId,
337 threadId,
338 });
339 const rewoundThreadId = idAllocator.derive.providerThread({
340 driver: CODEX_DRIVER,
341 nativeThreadId: "rewound-thread",
342 });
343 const otherThreadId = idAllocator.derive.providerThread({
344 driver: CODEX_DRIVER,
345 nativeThreadId: "other-thread",
346 });
347 const scopeId = CheckpointScopeId.make("checkpoint-scope:rewound");
348 // A run as execution leaves it: attempt, provider turn, root node, and a
349 // checkpoint once completed. A null native turn id is a weak ref.
350 const seedRun = Effect.fnUntraced(function* (input: {
351 readonly ordinal: number;
352 readonly providerThreadId: ProviderThreadId;
353 readonly status: "completed" | "interrupted" | "running";
354 readonly nativeTurnId: string | null;
355 }) {
356 const runId = RunId.make(`run:rewound:${input.ordinal}`);
357 const attemptId = RunAttemptId.make(`attempt:rewound:${input.ordinal}`);
358 const nodeId = NodeId.make(`node:rewound:${input.ordinal}`);
359 const providerTurnId = ProviderTurnId.make(`provider-turn:rewound:${input.ordinal}`);
360 const completedAt = input.status === "running" ? null : now;
361 const eventId = () => idAllocator.allocate.event({ threadId });
362 const events: Array<OrchestrationV2DomainEvent> = [
363 {
364 id: yield* eventId(),
365 type: "run.created",
366 threadId,
367 occurredAt: now,
368 payload: {
369 id: runId,
370 threadId,
371 ordinal: input.ordinal,
372 providerInstanceId: modelSelection.instanceId,
373 modelSelection,
374 providerThreadId: input.providerThreadId,
375 userMessageId: MessageId.make(`message:rewound:${input.ordinal}`),
376 rootNodeId: nodeId,
377 activeAttemptId: attemptId,
378 status: input.status,
379 requestedAt: now,
380 startedAt: now,
381 completedAt,
382 checkpointId: null,
383 contextHandoffId: null,
384 },
385 },
386 {
387 id: yield* eventId(),
388 type: "run-attempt.created",
389 threadId,
390 occurredAt: now,
391 payload: {
392 id: attemptId,
393 runId,
394 attemptOrdinal: 1,
395 rootNodeId: nodeId,
396 providerInstanceId: modelSelection.instanceId,
397 providerThreadId: input.providerThreadId,
398 providerTurnId,
399 reason: "initial",
400 status: input.status,
401 startedAt: now,
402 completedAt,
403 },
404 },
405 {
406 id: yield* eventId(),
407 type: "provider-turn.updated",
408 threadId,
409 occurredAt: now,
410 payload: {
411 id: providerTurnId,
412 providerThreadId: input.providerThreadId,
413 nodeId,
414 runAttemptId: attemptId,
415 nativeTurnRef:
416 input.nativeTurnId === null
417 ? {
418 driver: CODEX_DRIVER,
419 nativeId: `synthetic:${input.ordinal}`,
420 strength: "weak",
421 }
422 : { driver: CODEX_DRIVER, nativeId: input.nativeTurnId, strength: "strong" },
423 ordinal: input.ordinal,
424 status: input.status,
425 startedAt: now,
426 completedAt,
427 },
428 },
429 {
430 id: yield* eventId(),
431 type: "node.updated",
432 threadId,
433 occurredAt: now,
434 payload: {
435 id: nodeId,
436 threadId,
437 runId,
438 parentNodeId: null,
439 rootNodeId: nodeId,
440 kind: "root_turn",
441 status: input.status,
442 countsForRun: true,
443 providerThreadId: input.providerThreadId,
444 providerTurnId,
445 nativeItemRef: null,
446 runtimeRequestId: null,
447 checkpointScopeId: scopeId,
448 startedAt: now,
449 completedAt,
450 },
451 },
452 ...(input.status === "completed"
453 ? [
454 {
455 id: yield* eventId(),
456 type: "checkpoint.captured" as const,
457 threadId,
458 occurredAt: now,
459 payload: {
460 id: CheckpointId.make(`checkpoint:rewound:${input.ordinal}`),
461 threadId,
462 scopeId,
463 runId,
464 nodeId,
465 parentCheckpointId: null,
466 ordinalWithinScope: input.ordinal,
467 appRunOrdinal: input.ordinal,
468 ref: CheckpointRef.make(`refs/t3/checkpoints/rewound/${input.ordinal}`),
469 status: "ready" as const,
470 files: [],
471 capturedAt: now,
472 },
473 },
474 ]
475 : []),
476 ];
477 yield* eventSink.write({ events });
478 return runId;
479 });
480
481 yield* eventSink.write({ events: [threadEvent] });
482 const otherProviderRun = yield* seedRun({
483 ordinal: 1,
484 providerThreadId: otherThreadId,
485 status: "completed",
486 nativeTurnId: "other-turn",
487 });
488 const keptRun = yield* seedRun({
489 ordinal: 2,
490 providerThreadId: rewoundThreadId,
491 status: "completed",
492 nativeTurnId: "kept-turn",
493 });
494 const abandonedRun = yield* seedRun({
495 ordinal: 3,
496 providerThreadId: rewoundThreadId,
497 status: "completed",
498 nativeTurnId: "abandoned-turn",
499 });
500 const unlocatedRun = yield* seedRun({
501 ordinal: 4,
502 providerThreadId: rewoundThreadId,
503 status: "interrupted",
504 nativeTurnId: null,
505 });
506 const rewindingRun = yield* seedRun({
507 ordinal: 5,
508 providerThreadId: rewoundThreadId,
509 status: "running",
510 nativeTurnId: null,
511 });
512
513 yield* ingestor.ingestNormalized({
514 providerSessionId,
515 providerInstanceId: modelSelection.instanceId,
516 threadId,
517 runId: rewindingRun,
518 event: {
519 type: "provider_thread.updated",
520 driver: CODEX_DRIVER,
521 providerThread: {
522 id: rewoundThreadId,
523 driver: CODEX_DRIVER,
524 providerInstanceId: modelSelection.instanceId,
525 providerSessionId,
526 appThreadId: threadId,
527 ownerNodeId: null,
528 nativeThreadRef: {
529 driver: CODEX_DRIVER,
530 nativeId: "rewound-thread",
531 strength: "strong",
532 },
533 nativeConversationHeadRef: {
534 driver: CODEX_DRIVER,
535 nativeId: "kept-turn-reply",
536 strength: "strong",
537 },
538 status: "idle",
539 firstRunOrdinal: 2,
540 lastRunOrdinal: 5,
541 handoffIds: [],
542 forkedFrom: null,
543 createdAt: now,
544 updatedAt: now,
545 },
546 retainedNativeTurnIds: ["turn-started-outside-t3", "kept-turn"],
547 },
548 });
549
550 const projection = yield* projectionStore.getThreadProjection(threadId);
551 const runIds = [otherProviderRun, keptRun, abandonedRun, unlocatedRun, rewindingRun];
552 assert.deepEqual(
553 runIds.map((runId) => projection.runs.find((run) => run.id === runId)?.status),
554 ["completed", "completed", "rolled_back", "rolled_back", "running"],
555 );
556 assert.deepEqual(
557 runIds.map((runId) => projection.nodes.find((node) => node.runId === runId)?.status),
558 ["completed", "completed", "rolled_back", "rolled_back", "running"],
559 );
560 assert.deepEqual(
561 [otherProviderRun, keptRun, abandonedRun].map(
562 (runId) =>
563 projection.checkpoints.find((checkpoint) => checkpoint.runId === runId)?.status,
564 ),
565 ["ready", "ready", "stale"],
566 );
567 assert.equal(
568 projection.providerThreads.find((thread) => thread.id === rewoundThreadId)?.lastRunOrdinal,
569 5,
570 );
571 }),
572 );
573
321 it.effect("carries plan-step durations through consecutive and restarted ingestion", () =>574 it.effect("carries plan-step durations through consecutive and restarted ingestion", () =>
322 Effect.gen(function* () {575 Effect.gen(function* () {
323 yield* TestClock.setTime(Date.parse("2026-09-07T00:00:00.000Z"));576 yield* TestClock.setTime(Date.parse("2026-09-07T00:00:00.000Z"));
apps/server/src/orchestration-v2/ProviderEventIngestor.ts+148-8
...@@ -7,6 +7,7 @@ import {...@@ -7,6 +7,7 @@ import {
7 type OrchestrationV2Run,7 type OrchestrationV2Run,
8 type OrchestrationV2ProviderTurn,8 type OrchestrationV2ProviderTurn,
9 type OrchestrationV2Subagent,9 type OrchestrationV2Subagent,
10 type OrchestrationV2ThreadProjection,
10 type ModelSelection,11 type ModelSelection,
11 type RuntimeMode,12 type RuntimeMode,
12 type ProviderInteractionMode,13 type ProviderInteractionMode,
...@@ -197,6 +198,57 @@ function withPlanStepDurations(...@@ -197,6 +198,57 @@ function withPlanStepDurations(
197 };198 };
198}199}
199200
201/** Settled runs that carried conversation to the provider. */
202const CONVERSATION_RUN_STATUSES = new Set<OrchestrationV2Run["status"]>([
203 "completed",
204 "interrupted",
205 "failed",
206]);
207
208/**
209 * Settled runs on a provider thread that a native rewind left off the active
210 * branch. A run whose provider turn has a strong native ref is judged by that
211 * ref. A run without one cannot be located, so it falls with everything after
212 * the last run still on the branch.
213 */
214function runsOffNativeBranch(input: {
215 readonly projection: Pick<OrchestrationV2ThreadProjection, "runs" | "attempts" | "providerTurns">;
216 readonly providerThreadId: ProviderThreadId;
217 readonly retainedNativeTurnIds: ReadonlySet<string>;
218 readonly reportingRunId: RunId | undefined;
219}): ReadonlyArray<OrchestrationV2Run> {
220 const { attempts, providerTurns, runs } = input.projection;
221 const nativeTurnId = (run: OrchestrationV2Run) => {
222 const attempt = attempts.find((candidate) => candidate.id === run.activeAttemptId);
223 if (attempt === undefined) return null;
224 const ref = providerTurns.find(
225 (turn) => turn.id === attempt.providerTurnId || turn.runAttemptId === attempt.id,
226 )?.nativeTurnRef;
227 return ref?.strength === "strong" ? ref.nativeId : null;
228 };
229 const settled = runs.flatMap((run) =>
230 run.providerThreadId === input.providerThreadId &&
231 run.id !== input.reportingRunId &&
232 CONVERSATION_RUN_STATUSES.has(run.status)
233 ? [{ run, nativeTurnId: nativeTurnId(run) }]
234 : [],
235 );
236 const lastRetainedOrdinal = settled.reduce(
237 (last, { run, nativeTurnId }) =>
238 nativeTurnId !== null && input.retainedNativeTurnIds.has(nativeTurnId)
239 ? Math.max(last, run.ordinal)
240 : last,
241 0,
242 );
243 return settled
244 .filter(({ run, nativeTurnId }) =>
245 nativeTurnId === null
246 ? run.ordinal > lastRetainedOrdinal
247 : !input.retainedNativeTurnIds.has(nativeTurnId),
248 )
249 .map(({ run }) => run);
250}
251
200export interface ProviderEventIngestInput {252export interface ProviderEventIngestInput {
201 readonly providerSessionId: ProviderSessionId;253 readonly providerSessionId: ProviderSessionId;
202 readonly providerInstanceId: ProviderInstanceId;254 readonly providerInstanceId: ProviderInstanceId;
...@@ -387,6 +439,84 @@ export const layer: Layer.Layer<...@@ -387,6 +439,84 @@ export const layer: Layer.Layer<
387 },439 },
388 );440 );
389441
442 /**
443 * The provider rewound its conversation outside T3. Retire the runs that
444 * fell off its active branch with the same run, root-node, and checkpoint
445 * updates CheckpointRollbackService writes for a T3 revert. The run
446 * reporting the rewind performed it and stays, so it keeps lastRunOrdinal.
447 */
448 const rollBackRewoundRuns = Effect.fn("ProviderEventIngestor.rollBackRewoundRuns")(function* (
449 input: ProviderEventIngestInput,
450 rewind: {
451 readonly threadId: ThreadId;
452 readonly providerThreadId: ProviderThreadId;
453 readonly retainedNativeTurnIds: ReadonlySet<string>;
454 },
455 ) {
456 const projection = yield* projections.getThreadRecords(rewind.threadId, [
457 "runs",
458 "attempts",
459 "providerTurns",
460 "nodes",
461 "checkpoints",
462 ]);
463 const runs = runsOffNativeBranch({
464 projection,
465 providerThreadId: rewind.providerThreadId,
466 retainedNativeTurnIds: rewind.retainedNativeTurnIds,
467 reportingRunId: input.runId,
468 });
469 if (runs.length === 0) return [];
470 const now = yield* DateTime.now;
471 const eventId = () =>
472 idAllocator.allocate.event({
473 threadId: rewind.threadId,
474 providerSessionId: input.providerSessionId,
475 });
476 const runIds = new Set(runs.map((run) => run.id));
477 const events: Array<OrchestrationV2DomainEvent> = [];
478 for (const checkpoint of projection.checkpoints) {
479 if (checkpoint.status !== "ready" || checkpoint.runId === null) continue;
480 if (!runIds.has(checkpoint.runId)) continue;
481 events.push({
482 id: yield* eventId(),
483 type: "checkpoint.captured",
484 threadId: rewind.threadId,
485 runId: checkpoint.runId,
486 nodeId: checkpoint.nodeId,
487 providerInstanceId: input.providerInstanceId,
488 occurredAt: now,
489 payload: { ...checkpoint, status: "stale" },
490 });
491 }
492 for (const run of runs) {
493 const rootNode = projection.nodes.find((node) => node.id === run.rootNodeId);
494 events.push({
495 id: yield* eventId(),
496 type: "run.updated",
497 threadId: rewind.threadId,
498 runId: run.id,
499 ...(rootNode === undefined ? {} : { nodeId: rootNode.id }),
500 providerInstanceId: run.providerInstanceId,
501 occurredAt: now,
502 payload: { ...run, status: "rolled_back", completedAt: now },
503 });
504 if (rootNode !== undefined) {
505 events.push({
506 id: yield* eventId(),
507 type: "node.updated",
508 threadId: rewind.threadId,
509 runId: run.id,
510 nodeId: rootNode.id,
511 providerInstanceId: run.providerInstanceId,
512 occurredAt: now,
513 payload: { ...rootNode, status: "rolled_back", completedAt: now },
514 });
515 }
516 }
517 return events;
518 });
519
390 const normalize: ProviderEventIngestorV2Shape["normalize"] = (input) =>520 const normalize: ProviderEventIngestorV2Shape["normalize"] = (input) =>
391 Effect.gen(function* () {521 Effect.gen(function* () {
392 switch (input.event.type) {522 switch (input.event.type) {
...@@ -405,14 +535,24 @@ export const layer: Layer.Layer<...@@ -405,14 +535,24 @@ export const layer: Layer.Layer<
405 payload: input.event.providerSession,535 payload: input.event.providerSession,
406 }),536 }),
407 ];537 ];
408 case "provider_thread.updated":538 case "provider_thread.updated": {
409 return [539 const threadId = input.event.providerThread.appThreadId ?? input.threadId;
410 yield* makeDomainEvent(input, {540 const updated = yield* makeDomainEvent(input, {
411 type: "provider-thread.updated",541 type: "provider-thread.updated",
412 threadId: input.event.providerThread.appThreadId ?? input.threadId,542 threadId,
413 payload: input.event.providerThread,543 payload: input.event.providerThread,
414 }),544 });
415 ];545 return input.event.retainedNativeTurnIds === undefined
546 ? [updated]
547 : [
548 updated,
549 ...(yield* rollBackRewoundRuns(input, {
550 threadId,
551 providerThreadId: input.event.providerThread.id,
552 retainedNativeTurnIds: new Set(input.event.retainedNativeTurnIds),
553 })),
554 ];
555 }
416 case "provider_turn.updated":556 case "provider_turn.updated":
417 return [557 return [
418 ...(["completed", "interrupted", "failed", "cancelled"].includes(558 ...(["completed", "interrupted", "failed", "cancelled"].includes(
packages/provider-core/src/server/ProviderAdapter.ts+7
...@@ -90,6 +90,13 @@ export const ProviderAdapterV2Event = Schema.Union([...@@ -90,6 +90,13 @@ export const ProviderAdapterV2Event = Schema.Union([
90 type: Schema.Literal("provider_thread.updated"),90 type: Schema.Literal("provider_thread.updated"),
91 driver: ProviderDriverKind,91 driver: ProviderDriverKind,
92 providerThread: OrchestrationV2ProviderThread,92 providerThread: OrchestrationV2ProviderThread,
93 /**
94 * Set when the provider rewound its own conversation outside T3, for
95 * example a Pi extension moving the session-tree leaf back. Lists every
96 * native turn id still on the active branch; the ingestor rolls back
97 * earlier runs whose turns fell off it.
98 */
99 retainedNativeTurnIds: Schema.optional(Schema.Array(Schema.String)),
93 }),100 }),
94 Schema.Struct({101 Schema.Struct({
95 type: Schema.Literal("provider_turn.updated"),102 type: Schema.Literal("provider_turn.updated"),
packages/provider-pi/src/server/adapter.test.ts+311
...@@ -2019,6 +2019,317 @@ describe("PiAdapterV2", () => {...@@ -2019,6 +2019,317 @@ describe("PiAdapterV2", () => {
2019 }).pipe(Effect.scoped, Effect.provide(layerTest)),2019 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2020 );2020 );
20212021
2022 it.effect.each([false, true])("follows an extension rewind with summarize=%s", (summarize) =>
2023 Effect.gen(function* () {
2024 const fake = yield* makeFakePi;
2025 const { runtime, takeEvent } = yield* openRuntime(fake);
2026 const root = { type: "model_change", id: "root", parentId: null };
2027 const user = (id: string, parentId: string) => ({
2028 type: "message",
2029 id,
2030 parentId,
2031 message: { role: "user", content: [{ type: "text", text: id }] },
2032 });
2033 const reply = (id: string, parentId: string) => ({
2034 type: "message",
2035 id,
2036 parentId,
2037 message: { role: "assistant", content: [] },
2038 });
2039 fake.queueEntries({ entries: [root], leafId: "root" });
2040 const providerThread = yield* runtime.ensureThread({
2041 threadId: THREAD_ID,
2042 modelSelection: modelSelection("default"),
2043 runtimePolicy,
2044 });
2045 // Runs one turn to its terminal. `listings` answer the turn's
2046 // get_entries requests in order.
2047 const settle = Effect.fnUntraced(function* (
2048 text: string,
2049 runOrdinal: number,
2050 ...listings: ReadonlyArray<unknown>
2051 ) {
2052 yield* startTurn(runtime, providerThread, "default", [], text, undefined, runOrdinal);
2053 yield* fake.takeRequest("prompt");
2054 for (const listing of listings) fake.queueEntries(listing);
2055 // Pi acks every prompt; a command-only prompt then settles from
2056 // an idle probe instead of agent events.
2057 yield* fake.emit({ type: "response", command: "prompt", success: true });
2058 if (!text.startsWith("/")) {
2059 yield* fake.emit({ type: "agent_start" });
2060 yield* fake.emit({ type: "agent_settled" });
2061 }
2062 const turn = yield* takeEvent(
2063 (event) =>
2064 event.type === "provider_turn.updated" && event.providerTurn.status === "completed",
2065 );
2066 const thread = yield* takeEvent(
2067 (event) =>
2068 event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2069 );
2070 yield* takeEvent((event) => event.type === "turn.terminal");
2071 return {
2072 turnRef:
2073 turn.type === "provider_turn.updated" ? turn.providerTurn.nativeTurnRef : undefined,
2074 head:
2075 thread.type === "provider_thread.updated"
2076 ? thread.providerThread.nativeConversationHeadRef?.nativeId
2077 : undefined,
2078 retained:
2079 thread.type === "provider_thread.updated" ? thread.retainedNativeTurnIds : undefined,
2080 };
2081 });
2082
2083 yield* settle("prompt A", 1, {
2084 entries: [user("uA", "root"), reply("aA", "uA")],
2085 leafId: "aA",
2086 });
2087 const promptB = yield* settle("prompt B", 2, {
2088 entries: [user("uB", "aA"), reply("aB", "uB")],
2089 leafId: "aB",
2090 });
2091 assert.equal(promptB.turnRef?.nativeId, "uB");
2092 assert.isUndefined(promptB.retained);
2093
2094 // The extension moves the leaf back to A's reply. With a summary,
2095 // pi appends a branch_summary child of that reply as the new leaf.
2096 const rewoundLeaf = summarize ? "summary" : "aA";
2097 const summary = summarize ? [{ type: "branch_summary", id: "summary", parentId: "aA" }] : [];
2098 const rewind = yield* settle(
2099 "/rewind",
2100 3,
2101 { entries: summary, leafId: rewoundLeaf },
2102 {
2103 entries: [
2104 root,
2105 user("uA", "root"),
2106 reply("aA", "uA"),
2107 user("uB", "aA"),
2108 reply("aB", "uB"),
2109 ...summary,
2110 ],
2111 leafId: rewoundLeaf,
2112 },
2113 );
2114 assert.deepEqual(rewind.retained, ["uA"]);
2115 assert.equal(rewind.head, rewoundLeaf);
2116 if (summarize) {
2117 // The summary is this turn's only entry, and fork cannot re-root
2118 // before a non-user entry, so rollback may not pass it.
2119 assert.equal(rewind.turnRef?.strength, "weak");
2120 } else {
2121 // Nothing of this turn is on the branch; rollback can pass it.
2122 assert.isNull(rewind.turnRef);
2123 }
2124
2125 // get_entries is append-ordered: after a rewind to an older leaf,
2126 // the abandoned branch still comes first in the next window.
2127 const promptC = yield* settle("prompt C", 4, {
2128 entries: [
2129 ...(summarize ? [] : [user("uB", "aA"), reply("aB", "uB")]),
2130 user("uC", rewoundLeaf),
2131 reply("aC", "uC"),
2132 ],
2133 leafId: "aC",
2134 });
2135 assert.deepEqual(promptC.turnRef, {
2136 driver: PI_PROVIDER,
2137 nativeId: "uC",
2138 strength: "strong",
2139 });
2140 assert.isUndefined(promptC.retained);
2141 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2142 );
2143
2144 // Corrupted session data can loop its parentId chain. Finalizing the turn
2145 // must still finish, and must not report a branch to roll runs back to.
2146 it.effect.each(["turn window", "full tree"] as const)(
2147 "finishes a turn whose %s has a parentId cycle",
2148 (cycleIn) =>
2149 Effect.gen(function* () {
2150 const fake = yield* makeFakePi;
2151 const { runtime, takeEvent } = yield* openRuntime(fake);
2152 const root = { type: "model_change", id: "root", parentId: null };
2153 const cycle = [
2154 { type: "message", id: "x", parentId: "y", message: { role: "user", content: [] } },
2155 { type: "message", id: "y", parentId: "x", message: { role: "assistant", content: [] } },
2156 ];
2157 fake.queueEntries({ entries: [root], leafId: "root" });
2158 const providerThread = yield* runtime.ensureThread({
2159 threadId: THREAD_ID,
2160 modelSelection: modelSelection("default"),
2161 runtimePolicy,
2162 });
2163 yield* startTurn(runtime, providerThread, "default", [], "prompt", undefined, 1);
2164 yield* fake.takeRequest("prompt");
2165 if (cycleIn === "turn window") {
2166 fake.queueEntries({ entries: cycle, leafId: "x" });
2167 } else {
2168 // The leaf is outside the `since` window, so this reads as a rewind
2169 // and the full tree is walked for the surviving branch.
2170 fake.queueEntries({ entries: [], leafId: "x" });
2171 fake.queueEntries({ entries: [root, ...cycle], leafId: "x" });
2172 }
2173 yield* fake.emit({ type: "response", command: "prompt", success: true });
2174 yield* fake.emit({ type: "agent_start" });
2175 yield* fake.emit({ type: "agent_settled" });
2176 yield* takeEvent(
2177 (event) =>
2178 event.type === "provider_turn.updated" && event.providerTurn.status === "completed",
2179 );
2180 const thread = yield* takeEvent(
2181 (event) =>
2182 event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2183 );
2184 assert.isUndefined(
2185 thread.type === "provider_thread.updated" ? thread.retainedNativeTurnIds : null,
2186 );
2187 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2188 );
2189
2190 it.effect.each(["missing parent", "invalid parent", "duplicate id"] as const)(
2191 "does not reconcile malformed session entries: %s",
2192 (malformed) =>
2193 Effect.gen(function* () {
2194 const fake = yield* makeFakePi;
2195 const { runtime, takeEvent } = yield* openRuntime(fake);
2196 const root = { type: "model_change", id: "root", parentId: null };
2197 fake.queueEntries({ entries: [root], leafId: "root" });
2198 const providerThread = yield* runtime.ensureThread({
2199 threadId: THREAD_ID,
2200 modelSelection: modelSelection("default"),
2201 runtimePolicy,
2202 });
2203 yield* startTurn(runtime, providerThread);
2204 yield* fake.takeRequest("prompt");
2205 const entry = {
2206 type: "message",
2207 id: "x",
2208 message: { role: "user", content: [] },
2209 ...(malformed === "missing parent"
2210 ? {}
2211 : { parentId: malformed === "invalid parent" ? 42 : null }),
2212 };
2213 fake.queueEntries({ entries: [], leafId: "x" });
2214 fake.queueEntries({
2215 entries: [
2216 root,
2217 entry,
2218 ...(malformed === "duplicate id" ? [{ ...entry, parentId: "root" }] : []),
2219 ],
2220 leafId: "x",
2221 });
2222 yield* fake.emit({ type: "response", command: "prompt", success: true });
2223 yield* fake.emit({ type: "agent_start" });
2224 yield* fake.emit({ type: "agent_settled" });
2225 yield* takeEvent(
2226 (event) =>
2227 event.type === "provider_turn.updated" && event.providerTurn.status === "completed",
2228 );
2229 const thread = yield* takeEvent(
2230 (event) =>
2231 event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2232 );
2233 assert.isUndefined(
2234 thread.type === "provider_thread.updated" ? thread.retainedNativeTurnIds : null,
2235 );
2236 yield* takeEvent((event) => event.type === "turn.terminal");
2237 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2238 );
2239
2240 it.effect("does not reconcile a replacement session tree as a native rewind", () =>
2241 Effect.gen(function* () {
2242 const fake = yield* makeFakePi;
2243 fake.queueEntries({
2244 entries: [{ type: "model_change", id: "old-leaf", parentId: null }],
2245 leafId: "old-leaf",
2246 });
2247 const { runtime, takeEvent } = yield* openRuntime(fake);
2248 const providerThread = yield* runtime.ensureThread({
2249 threadId: THREAD_ID,
2250 modelSelection: modelSelection("default"),
2251 runtimePolicy,
2252 });
2253 const finish = Effect.fnUntraced(function* () {
2254 yield* startTurn(runtime, providerThread);
2255 yield* fake.takeRequest("prompt");
2256 yield* fake.emit({ type: "response", command: "prompt", success: true });
2257 yield* fake.emit({ type: "agent_start" });
2258 yield* fake.emit({ type: "agent_settled" });
2259 const update = yield* takeEvent(
2260 (event) =>
2261 event.type === "provider_thread.updated" && event.providerThread.status === "idle",
2262 );
2263 yield* takeEvent((event) => event.type === "turn.terminal");
2264 assert.isUndefined(
2265 update.type === "provider_thread.updated" ? update.retainedNativeTurnIds : null,
2266 );
2267 });
2268 // A failed cursor read is followed by a full listing from a new session.
2269 fake.queueEntries({ entries: [] });
2270 yield* finish();
2271 fake.queueEntries({
2272 entries: [
2273 {
2274 type: "message",
2275 id: "new-user",
2276 parentId: null,
2277 message: { role: "user", content: [] },
2278 },
2279 ],
2280 leafId: "new-user",
2281 });
2282 yield* finish();
2283 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2284 );
2285
2286 it.effect("rolls back past turns that left nothing in the session tree", () =>
2287 Effect.gen(function* () {
2288 const fake = yield* makeFakePi;
2289 const { runtime } = yield* openRuntime(fake);
2290 const providerThread = yield* runtime.ensureThread({
2291 threadId: THREAD_ID,
2292 modelSelection: modelSelection("default"),
2293 runtimePolicy,
2294 });
2295 const turn = (ordinal: number, entryId: string | null): OrchestrationV2ProviderTurn => ({
2296 id: ProviderTurnId.make(`turn-${ordinal}`),
2297 providerThreadId: providerThread.id,
2298 nodeId: NodeId.make(`node-${ordinal}`),
2299 runAttemptId: null,
2300 nativeTurnRef:
2301 entryId === null ? null : { driver: PI_PROVIDER, nativeId: entryId, strength: "strong" },
2302 ordinal,
2303 status: "completed",
2304 startedAt: null,
2305 completedAt: null,
2306 });
2307 const rollBackToFirst = (turns: ReadonlyArray<OrchestrationV2ProviderTurn>) =>
2308 runtime.rollbackThread({
2309 providerThread,
2310 target: {
2311 type: "provider_turn",
2312 checkpointId: CheckpointId.make("checkpoint-1"),
2313 appRunOrdinal: 1,
2314 providerTurn: turns[0]!,
2315 },
2316 providerThreadTurns: turns,
2317 });
2318 const forkedEntries = () =>
2319 fake
2320 .allRequests()
2321 .filter((request) => request.type === "fork")
2322 .map((request) => request.entryId);
2323
2324 // Only a turn that left nothing follows the target: already there.
2325 yield* rollBackToFirst([turn(1, "u1"), turn(2, null)]);
2326 assert.deepEqual(forkedEntries(), []);
2327 // Pi re-roots before the first later turn that left entries.
2328 yield* rollBackToFirst([turn(1, "u1"), turn(2, null), turn(3, "u3")]);
2329 assert.deepEqual(forkedEntries(), ["u3"]);
2330 }).pipe(Effect.scoped, Effect.provide(layerTest)),
2331 );
2332
2022 it.effect("settles a command-only prompt from its deferred ack and idle probe", () =>2333 it.effect("settles a command-only prompt from its deferred ack and idle probe", () =>
2023 Effect.gen(function* () {2334 Effect.gen(function* () {
2024 const fake = yield* makeFakePi;2335 const fake = yield* makeFakePi;
packages/provider-pi/src/server/adapter.ts+142-50
...@@ -341,8 +341,11 @@ interface ActivePiTurn {...@@ -341,8 +341,11 @@ interface ActivePiTurn {
341}341}
342342
343interface PiTurnTreeRefs {343interface PiTurnTreeRefs {
344 readonly turnStartEntryId: string | null;344 /** Omitted keeps the synthetic ref; null means the turn left nothing on the active branch. */
345 readonly nativeTurnRef: OrchestrationV2ProviderRef | null | undefined;
345 readonly leafId: string | null;346 readonly leafId: string | null;
347 /** Set after a native rewind: every user entry still on the active branch. */
348 readonly retainedNativeTurnIds: ReadonlyArray<string> | undefined;
346}349}
347350
348interface PendingPiPrompt {351interface PendingPiPrompt {
...@@ -567,6 +570,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -567,6 +570,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
567 const updateProviderThread = (570 const updateProviderThread = (
568 state: PiThreadState,571 state: PiThreadState,
569 patch: Partial<OrchestrationV2ProviderThread>,572 patch: Partial<OrchestrationV2ProviderThread>,
573 retainedNativeTurnIds?: ReadonlyArray<string>,
570 ) =>574 ) =>
571 Effect.gen(function* () {575 Effect.gen(function* () {
572 const updatedAt = yield* DateTime.now;576 const updatedAt = yield* DateTime.now;
...@@ -575,6 +579,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -575,6 +579,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
575 type: "provider_thread.updated",579 type: "provider_thread.updated",
576 driver: PI_PROVIDER,580 driver: PI_PROVIDER,
577 providerThread: state.providerThread,581 providerThread: state.providerThread,
582 ...(retainedNativeTurnIds === undefined ? {} : { retainedNativeTurnIds }),
578 });583 });
579 });584 });
580585
...@@ -1380,53 +1385,78 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -1380,53 +1385,78 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
13801385
1381 // ── turn lifecycle ────────────────────────────────────1386 // ── turn lifecycle ────────────────────────────────────
13821387
1388 /** One `get_entries` listing, or null when the request failed or named no leaf. */
1389 const listSessionEntries = (since: string | null, timeoutMs: number) =>
1390 request({ type: "get_entries", ...(since === null ? {} : { since }) }, timeoutMs).pipe(
1391 Effect.map((data) => {
1392 const leafId = recordField(data, "leafId");
1393 const entries = piSessionEntries(recordField(data, "entries"));
1394 return entries !== null && (typeof leafId === "string" || leafId === null)
1395 ? { entries, leafId }
1396 : null;
1397 }),
1398 Effect.orElseSucceed(() => null),
1399 );
1400
1383 /**1401 /**
1384 * Locate this turn's first user entry and the new leaf in pi's session1402 * Locate this turn's first user entry and the new leaf in pi's session
1385 * tree. The user-entry id becomes the provider turn's native ref (the1403 * tree. The user-entry id becomes the provider turn's native ref (the
1386 * point `fork` rolls back to); the leaf becomes the conversation head.1404 * point `fork` rolls back to); the leaf becomes the conversation head.
1405 * `get_entries` is append-ordered and keeps abandoned branches, so both
1406 * come from walking the leaf's `parentId` chain, never from list order.
1407 * A leaf that no longer descends from the previous one means something
1408 * outside T3 rewound the session, such as an extension calling
1409 * `navigateTree`. The user entries left on the active branch are then
1410 * returned so orchestration can roll back runs that fell off it.
1387 * Pure bookkeeping: failures degrade to the synthetic refs.1411 * Pure bookkeeping: failures degrade to the synthetic refs.
1388 */1412 */
1389 const captureTurnTreeRefs = Effect.fnUntraced(function* (1413 const captureTurnTreeRefs = Effect.fnUntraced(function* (
1390 timeoutMs = PI_REQUEST_TIMEOUT_MS,1414 timeoutMs = PI_REQUEST_TIMEOUT_MS,
1391 ): Effect.fn.Return<PiTurnTreeRefs | null> {1415 ): Effect.fn.Return<PiTurnTreeRefs | null> {
1392 const cursorWasStale = leafCursorStale;1416 const cursorWasStale = leafCursorStale;
1393 const cursor = cursorWasStale ? null : lastKnownLeaf;1417 const previousLeaf = lastKnownLeaf;
1394 const data = yield* request(1418 const fullListing = cursorWasStale || previousLeaf === null;
1395 {1419 const listing = yield* listSessionEntries(fullListing ? null : previousLeaf, timeoutMs);
1396 type: "get_entries",1420 const turnWalk =
1397 ...(cursor === null ? {} : { since: cursor }),1421 listing === null ? null : walkPiBranch(listing.entries, listing.leafId, previousLeaf);
1398 },1422 const rewound = turnWalk !== null && previousLeaf !== null && !turnWalk.reachedStop;
1399 timeoutMs,1423 // A `since` window ends at the previous leaf; the rewound branch
1400 ).pipe(Effect.orElseSucceed(() => undefined));1424 // continues past it, so its ancestry needs the whole tree.
1401 if (data === undefined) {1425 const tree = rewound && !fullListing ? yield* listSessionEntries(null, timeoutMs) : listing;
1402 // Pi may have advanced past `lastKnownLeaf` while this failed, so the1426 if (listing === null || turnWalk === null || tree === null) {
1403 // cursor can no longer be trusted to bound a single turn.1427 // The read failed or its ancestry loops. Pi may have advanced past
1428 // `lastKnownLeaf` meanwhile, so the cursor can no longer be trusted
1429 // to bound a single turn.
1404 leafCursorStale = true;1430 leafCursorStale = true;
1405 return null;1431 return null;
1406 }1432 }
1407 const entries = recordField(data, "entries");1433 lastKnownLeaf = listing.leafId;
1408 const leafId = recordString(data, "leafId");
1409 if (leafId !== undefined) lastKnownLeaf = leafId;
1410 // Without a trustworthy cursor this window spans more than one turn, so
1411 // its first user entry belongs to an earlier turn. Re-sync the cursor
1412 // and skip the turn-start ref rather than pointing rollback too far
1413 // back; the next turn gets an accurate ref again.
1414 leafCursorStale = false;1434 leafCursorStale = false;
1415 const firstUserEntryId = cursorWasStale1435 const turnStartEntryId = turnWalk.userEntryIds.at(-1);
1416 ? undefined1436 // Only a branch walked to its root lists every survivor; a partial or
1417 : Array.isArray(entries)1437 // cyclic walk would roll back runs that are still in the conversation.
1418 ? entries1438 // A replaced session has a different tree. Only an old leaf still in
1419 .filter(1439 // this tree proves an in-session rewind rather than a session switch.
1420 (entry) =>1440 const branch =
1421 recordField(entry, "type") === "message" &&1441 rewound && previousLeaf !== null && tree.entries.has(previousLeaf)
1422 recordString(recordField(entry, "message"), "role") === "user",1442 ? walkPiBranch(tree.entries, listing.leafId, null)
1423 )
1424 .map((entry) => recordString(entry, "id"))
1425 .find((id) => id !== undefined)
1426 : undefined;1443 : undefined;
1427 return {1444 return {
1428 turnStartEntryId: firstUserEntryId ?? null,1445 // Without a trustworthy cursor the walk spans more than one turn, so
1429 leafId: leafId ?? null,1446 // its earliest user entry belongs to an earlier turn. Keep the
1447 // synthetic ref rather than pointing rollback too far back; the next
1448 // turn gets an accurate ref again. A turn that left nothing on the
1449 // active branch (a command, or a rewind) gets no ref: a rollback
1450 // past it needs no fork of its own.
1451 nativeTurnRef: cursorWasStale
1452 ? undefined
1453 : turnStartEntryId !== undefined
1454 ? providerRef(turnStartEntryId)
1455 : turnWalk.entryCount === 0
1456 ? null
1457 : undefined,
1458 leafId: listing.leafId,
1459 retainedNativeTurnIds: branch?.reachedStop === true ? branch.userEntryIds : undefined,
1430 };1460 };
1431 });1461 });
14321462
...@@ -1467,20 +1497,27 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -1467,20 +1497,27 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
1467 threadId: turn.turnInput.threadId,1497 threadId: turn.turnInput.threadId,
1468 providerTurn: {1498 providerTurn: {
1469 ...turn.providerTurn,1499 ...turn.providerTurn,
1470 ...(treeRefs?.turnStartEntryId == null1500 ...(treeRefs === null || treeRefs.nativeTurnRef === undefined
1471 ? {}1501 ? {}
1472 : { nativeTurnRef: providerRef(treeRefs.turnStartEntryId) }),1502 : { nativeTurnRef: treeRefs.nativeTurnRef }),
1473 status: turn.interrupted ? "interrupted" : failure !== null ? "failed" : "completed",1503 status: turn.interrupted ? "interrupted" : failure !== null ? "failed" : "completed",
1474 completedAt,1504 completedAt,
1475 ...(tokenUsage === undefined ? {} : { tokenUsage }),1505 ...(tokenUsage === undefined ? {} : { tokenUsage }),
1476 },1506 },
1477 });1507 });
1478 yield* updateProviderThread(state, {1508 yield* updateProviderThread(
1479 status: "idle",1509 state,
1480 ...(treeRefs?.leafId == null1510 {
1481 ? {}1511 status: "idle",
1482 : { nativeConversationHeadRef: providerRef(treeRefs.leafId) }),1512 ...(treeRefs === null
1483 });1513 ? {}
1514 : {
1515 nativeConversationHeadRef:
1516 treeRefs.leafId === null ? null : providerRef(treeRefs.leafId),
1517 }),
1518 },
1519 treeRefs?.retainedNativeTurnIds,
1520 );
1484 yield* updateProviderSession(1521 yield* updateProviderSession(
1485 failure !== null ? "error" : "ready",1522 failure !== null ? "error" : "ready",
1486 failure?.message ?? null,1523 failure?.message ?? null,
...@@ -3053,11 +3090,68 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -3053,11 +3090,68 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
3053 });3090 });
3054});3091});
30553092
3093interface PiSessionEntry {
3094 readonly parentId: string | null;
3095 readonly isUserMessage: boolean;
3096}
3097
3098/** Index a `get_entries` listing, rejecting ambiguous ancestry. */
3099function piSessionEntries(entries: unknown): ReadonlyMap<string, PiSessionEntry> | null {
3100 const byId = new Map<string, PiSessionEntry>();
3101 if (!Array.isArray(entries)) return null;
3102 for (const entry of entries) {
3103 const id = recordString(entry, "id");
3104 const parentId = recordField(entry, "parentId");
3105 if (id === undefined || byId.has(id) || (parentId !== null && typeof parentId !== "string"))
3106 return null;
3107 byId.set(id, {
3108 parentId,
3109 isUserMessage:
3110 recordField(entry, "type") === "message" &&
3111 recordString(recordField(entry, "message"), "role") === "user",
3112 });
3113 }
3114 return byId;
3115}
3116
3117/**
3118 * Walk pi's active branch from `leafId` towards the root through `entries`,
3119 * stopping at `stopAt`. Returns how many entries it passed, the user entries
3120 * among them (newest first), and whether the walk reached `stopAt`. Leaving a
3121 * partial listing anywhere else means the branch does not descend from `stopAt`.
3122 * Returns null when the `parentId` chain loops, which only corrupted session
3123 * data can produce; callers treat that like a failed listing.
3124 */
3125function walkPiBranch(
3126 entries: ReadonlyMap<string, PiSessionEntry>,
3127 leafId: string | null,
3128 stopAt: string | null,
3129): {
3130 readonly entryCount: number;
3131 readonly userEntryIds: ReadonlyArray<string>;
3132 readonly reachedStop: boolean;
3133} | null {
3134 const visited = new Set<string>();
3135 const userEntryIds: Array<string> = [];
3136 let current = leafId;
3137 while (current !== null && current !== stopAt) {
3138 const entry = entries.get(current);
3139 if (entry === undefined) break;
3140 if (visited.has(current)) return null;
3141 visited.add(current);
3142 if (entry.isUserMessage) userEntryIds.push(current);
3143 current = entry.parentId;
3144 }
3145 return { entryCount: visited.size, userEntryIds, reachedStop: current === stopAt };
3146}
3147
3056/**3148/**
3057 * Resolve the pi session-tree entry `fork` should re-root at for a rollback.3149 * Resolve the pi session-tree entry `fork` should re-root at for a rollback.
3058 * Returns `null` when no turns follow the target (nothing to discard) and3150 * Turns without a ref left nothing on the active branch, so the boundary is
3059 * `undefined` when the boundary turn has no captured entry ref (only3151 * the first discarded turn that has one. Returns `null` when no such turn
3060 * turn-boundary refs recorded by `captureTurnTreeRefs` are strong).3152 * follows the target (nothing to discard) and `undefined` when the boundary
3153 * turn has no captured entry ref (only turn-boundary refs recorded by
3154 * `captureTurnTreeRefs` are strong).
3061 */3155 */
3062function piRollbackForkEntry(input: {3156function piRollbackForkEntry(input: {
3063 readonly target:3157 readonly target:
...@@ -3067,13 +3161,11 @@ function piRollbackForkEntry(input: {...@@ -3067,13 +3161,11 @@ function piRollbackForkEntry(input: {
3067}): string | null | undefined {3161}): string | null | undefined {
3068 const boundaryOrdinal =3162 const boundaryOrdinal =
3069 input.target.type === "thread_start" ? 0 : input.target.providerTurn.ordinal;3163 input.target.type === "thread_start" ? 0 : input.target.providerTurn.ordinal;
3070 const discarded = input.providerThreadTurns3164 const ref = input.providerThreadTurns
3071 .filter((turn) => turn.ordinal > boundaryOrdinal)3165 .filter((turn) => turn.ordinal > boundaryOrdinal && turn.nativeTurnRef !== null)
3072 .sort((a, b) => a.ordinal - b.ordinal);3166 .sort((a, b) => a.ordinal - b.ordinal)[0]?.nativeTurnRef;
3073 const boundary = discarded[0];3167 if (ref == null) return null;
3074 if (boundary === undefined) return null;3168 if (ref.strength !== "strong" || ref.nativeId === null) return undefined;
3075 const ref = boundary.nativeTurnRef;
3076 if (ref === null || ref.strength !== "strong" || ref.nativeId === null) return undefined;
3077 return ref.nativeId;3169 return ref.nativeId;
3078}3170}
30793171