# Files changed · PR \#20521 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/20521/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(pi\): fork and revert past turns whose session entry was never captured](https://git.cafe/pingdotgg/t3code/pull/20521.md) Version: 6 Head commit: 3093a81a3c4bc05fd62dac6bb89f577663f69f67 Base commit: 86d130af6c02a126515488ed6038b68906c35bf7 Comparison base: 86d130af6c02a126515488ed6038b68906c35bf7 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## packages/provider-pi/src/server/adapter.test.ts Status: modified; +209; −0 ```diff @@ -71,7 +71,72 @@ const FAKE_SESSION_FILE = "/fake/.pi/agent/sessions/--workspace--/0001_abc.jsonl"; /** Deliberately outside the valid pid range so a group-kill can never land. */ const FAKE_PID = 999_999_999; + +const at = (seconds: number) => DateTime.makeUnsafe(seconds * 1000); +const sessionMessage = ( + id: string, + parentId: string | null, + role: "user" | "assistant", + seconds: number, +) => ({ + type: "message", + id, + parentId, + timestamp: DateTime.formatIso(at(seconds)), + message: { role, content: [] }, +}); + +const TURN_ONE_ENTRIES = [ + sessionMessage("u1", null, "user", 10), + sessionMessage("a1", "u1", "assistant", 20), + sessionMessage("steer1", "a1", "user", 30), + sessionMessage("a1b", "steer1", "assistant", 40), +]; +const UNCAPTURED_TURN_TREE = { + entries: [ + ...TURN_ONE_ENTRIES, + sessionMessage("u2", "a1b", "user", 110), + sessionMessage("a2", "u2", "assistant", 120), + sessionMessage("steer2", "a2", "user", 130), + sessionMessage("a2b", "steer2", "assistant", 140), + ], + leafId: "a2b", +}; + +function uncapturedTurns( + providerThreadId: ProviderThreadId, + window = { startedAt: 150, completedAt: 200 }, +): ReadonlyArray { + const turn = ( + ordinal: number, + nativeTurnRef: OrchestrationV2ProviderTurn["nativeTurnRef"], + status: OrchestrationV2ProviderTurn["status"], + startedAt: number, + completedAt: number, + ): OrchestrationV2ProviderTurn => ({ + id: ProviderTurnId.make(`turn-${ordinal}`), + providerThreadId, + nodeId: NodeId.make(`node-${ordinal}`), + runAttemptId: null, + nativeTurnRef, + ordinal, + status, + startedAt: at(startedAt), + completedAt: at(completedAt), + }); + return [ + turn(1, { driver: PI_PROVIDER, nativeId: "u1", strength: "strong" }, "completed", 0, 50), + turn( + 2, + { driver: PI_PROVIDER, nativeId: "synthetic-2", strength: "weak" }, + "cancelled", + window.startedAt, + window.completedAt, + ), + ]; +} + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ runtimeMode: "full-access", interactionMode: "default", @@ -1902,6 +1967,50 @@ }).pipe(Effect.scoped, Effect.provide(layerTest)), ); + it.effect("natively forks before a turn whose start entry was never captured", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const forkFake = yield* makeFakePi; + const forkFile = "/fake/forked.jsonl"; + const { runtime } = yield* openRuntime(fake, "default", THREAD_ID, SESSION_ID, forkFake); + const source = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const forkAtFirst = (turns: ReadonlyArray) => + runtime.forkThread({ + sourceProviderThread: source, + sourceProviderTurns: turns, + providerTurnId: turns[0]!.id, + targetThreadId: ThreadId.make("fork-target"), + }); + const forkedEntries = () => + forkFake + .allRequests() + .filter((request) => request.type === "fork") + .map((request) => request.entryId); + + // A later prompt cannot stand in for a cancelled turn that ended before it landed. + forkFake.queueEntries(UNCAPTURED_TURN_TREE); + const unresolved = yield* forkAtFirst( + uncapturedTurns(source.id, { startedAt: 60, completedAt: 100 }), + ).pipe(Effect.result); + assert.deepEqual(forkedEntries(), []); + assert.equal(unresolved._tag, "Failure"); + + forkFake.queueEntries(UNCAPTURED_TURN_TREE); + forkFake.queueState({ sessionFile: forkFile }); + fake.queueState({ sessionFile: forkFile }); + const forked = yield* forkAtFirst( + uncapturedTurns(source.id, { startedAt: 100, completedAt: 200 }), + ); + assert.equal(forked.nativeThreadRef?.nativeId, forkFile); + assert.deepEqual(forkedEntries(), ["u2"]); + assert.isFalse(fake.allRequests().some((request) => request.type === "fork")); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect("presents explicitly namespaced MCP extension tools", () => Effect.gen(function* () { const fake = yield* makeFakePi; @@ -2697,6 +2806,106 @@ }).pipe(Effect.scoped, Effect.provide(layerTest)), ); + it.effect("rolls back past a turn whose start entry was never captured", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const rollBackToFirst = (turns: ReadonlyArray) => + runtime.rollbackThread({ + providerThread, + target: { + type: "provider_turn", + checkpointId: CheckpointId.make("checkpoint-1"), + appRunOrdinal: 1, + providerTurn: turns[0]!, + }, + providerThreadTurns: turns, + }); + const forkedEntries = () => + fake + .allRequests() + .filter((request) => request.type === "fork") + .map((request) => request.entryId); + + fake.queueEntries(UNCAPTURED_TURN_TREE); + const unresolved = yield* rollBackToFirst( + uncapturedTurns(providerThread.id, { startedAt: 60, completedAt: 100 }), + ).pipe(Effect.result); + assert.deepEqual(forkedEntries(), []); + assert.equal(unresolved._tag, "Failure"); + + fake.queueEntries({ entries: TURN_ONE_ENTRIES, leafId: "a1b" }); + const missing = yield* rollBackToFirst(uncapturedTurns(providerThread.id)).pipe( + Effect.result, + ); + assert.deepEqual(forkedEntries(), []); + assert.equal(missing._tag, "Failure"); + + // An adopted wake precedes its T3 turn, but follows turn 1 and its steer. + fake.queueEntries(UNCAPTURED_TURN_TREE); + yield* rollBackToFirst(uncapturedTurns(providerThread.id)); + assert.deepEqual(forkedEntries(), ["u2"]); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("preserves inherited history when rolling back an uncaptured first turn", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const rollBackToStart = (startedAt: number) => + runtime.rollbackThread({ + providerThread: { ...providerThread, createdAt: at(50) }, + target: { + type: "thread_start", + checkpointId: CheckpointId.make("checkpoint-start"), + appRunOrdinal: 0, + }, + providerThreadTurns: [ + { + ...uncapturedTurns(providerThread.id)[1]!, + ordinal: 1, + startedAt: at(startedAt), + }, + ], + }); + const forkedEntries = () => + fake + .allRequests() + .filter((request) => request.type === "fork") + .map((request) => request.entryId); + + const entries = { + entries: [ + ...TURN_ONE_ENTRIES, + sessionMessage("u2", "a1b", "user", 100), + sessionMessage("a2", "u2", "assistant", 120), + sessionMessage("u3", "a2", "user", 160), + sessionMessage("a3", "u3", "assistant", 180), + ], + leafId: "a3", + }; + + // The adopted wake precedes T3's first turn; its steer must be discarded with it. + fake.queueEntries(entries); + yield* rollBackToStart(150); + assert.deepEqual(forkedEntries(), ["u2"]); + + fake.queueEntries(entries); + yield* rollBackToStart(100); + assert.deepEqual(forkedEntries(), ["u2", "u2"]); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect("settles a command-only prompt from its deferred ack and idle probe", () => Effect.gen(function* () { const fake = yield* makeFakePi; ``` ## packages/provider-pi/src/server/adapter.ts Status: modified; +118; −31 ```diff @@ -1421,13 +1421,7 @@ const listSessionEntries = (since: string | null, timeoutMs: number) => request({ type: "get_entries", ...(since === null ? {} : { since }) }, timeoutMs).pipe( - Effect.map((data) => { - const leafId = recordField(data, "leafId"); - const entries = piSessionEntries(recordField(data, "entries")); - return entries !== null && (typeof leafId === "string" || leafId === null) - ? { entries, leafId } - : null; + Effect.map(piSessionListing), - }), Effect.orElseSucceed(() => null), ); @@ -2880,13 +2874,15 @@ if (pendingWake !== null) { return yield* protocolError("Cannot roll back while native Pi work awaits a turn"); } + const boundary = piForkBoundary(rollbackInput); + if (boundary.type === "nothing") { - const forkEntryId = piRollbackForkEntry(rollbackInput); - if (forkEntryId === null) { return piThreadSnapshot(state.providerThread); } + const forkEntryId = yield* piForkEntry( + boundary, + rollbackInput.providerThread.createdAt, - if (forkEntryId === undefined) { - return yield* protocolError("Pi rollback target has no captured session-tree entry"); - } + request({ type: "get_entries" }), + ); return yield* Effect.acquireUseRelease( sessionEventPermit.withPermits(1)( Effect.gen(function* () { @@ -3005,16 +3001,13 @@ if (forkInput.providerTurnId !== undefined && target === undefined) { return yield* protocolError("Pi fork target turn is missing"); } + const boundary: PiForkBoundary = - const beforeEntry = target === undefined - ? null - : piRollbackForkEntry({ + ? { type: "nothing" } + : piForkBoundary({ target: { type: "provider_turn", providerTurn: target }, providerThreadTurns: sourceTurns, }); - if (beforeEntry === undefined) { - return yield* protocolError("Pi fork boundary has no captured session-tree entry"); - } // CLI --fork uses Pi's own session format and sets the destination cwd. RPC // switch_session/clone alone would retain the source cwd. const forkLaunch = buildPiRpcLaunch({ @@ -3038,6 +3031,14 @@ Effect.ignore, Effect.forkScoped, ); + const beforeEntry = + boundary.type === "nothing" + ? null + : yield* piForkEntry( + boundary, + source.createdAt, + forkConnection.request({ type: "get_entries" }), + ); if (beforeEntry !== null) { const result = yield* forkConnection.request({ type: "fork", @@ -3111,8 +3112,23 @@ interface PiSessionEntry { readonly parentId: string | null; readonly isUserMessage: boolean; + /** Pi's `timestamp` as epoch milliseconds; null when missing or unparsable. */ + readonly timestamp: number | null; } +interface PiSessionListing { + readonly entries: ReadonlyMap; + readonly leafId: string | null; +} + +function piSessionListing(data: unknown): PiSessionListing | null { + const leafId = recordField(data, "leafId"); + const entries = piSessionEntries(recordField(data, "entries")); + return entries !== null && (typeof leafId === "string" || leafId === null) + ? { entries, leafId } + : null; +} + function piSessionEntries(entries: unknown): ReadonlyMap | null { const byId = new Map(); if (!Array.isArray(entries)) return null; @@ -3121,11 +3137,16 @@ const parentId = recordField(entry, "parentId"); if (id === undefined || byId.has(id) || (parentId !== null && typeof parentId !== "string")) return null; + const timestamp = recordString(entry, "timestamp"); byId.set(id, { parentId, isUserMessage: recordField(entry, "type") === "message" && recordString(recordField(entry, "message"), "role") === "user", + timestamp: + timestamp === undefined + ? null + : Option.getOrNull(Option.map(DateTime.make(timestamp), DateTime.toEpochMillis)), }); } return byId; @@ -3154,28 +3175,94 @@ return { entryCount: visited.size, userEntryIds, reachedStop: current === stopAt }; } +type PiForkBoundary = + | { readonly type: "entry"; readonly entryId: string } + | { readonly type: "nothing" } + | { + readonly type: "uncaptured"; + readonly turn: OrchestrationV2ProviderTurn; + readonly target: OrchestrationV2ProviderTurn | null; + }; + /** * Turns without a ref left nothing on the active branch, so the boundary is the first discarded + * turn that has one. `uncaptured` is a boundary turn still carrying its synthetic ref (only + * `captureTurnTreeRefs` records strong ones); `piForkEntry` locates it in the session tree. - * turn that has one. Returns `null` when no such turn follows the target (nothing to discard) and - * `undefined` when the boundary turn has no captured entry ref (only turn-boundary refs recorded - * by `captureTurnTreeRefs` are strong). */ -function piRollbackForkEntry(input: { +function piForkBoundary(input: { readonly target: | { readonly type: "thread_start" } | { readonly type: "provider_turn"; readonly providerTurn: OrchestrationV2ProviderTurn }; readonly providerThreadTurns: ReadonlyArray; +}): PiForkBoundary { + const target = input.target.type === "thread_start" ? null : input.target.providerTurn; + const boundary = input.providerThreadTurns + .filter((turn) => turn.ordinal > (target?.ordinal ?? 0) && turn.nativeTurnRef !== null) + .sort((a, b) => a.ordinal - b.ordinal)[0]; -}): string | null | undefined { - const boundaryOrdinal = + const ref = boundary?.nativeTurnRef; + if (boundary === undefined || ref == null) return { type: "nothing" }; + return ref.strength === "strong" && ref.nativeId !== null + ? { type: "entry", entryId: ref.nativeId } + : { type: "uncaptured", turn: boundary, target }; - input.target.type === "thread_start" ? 0 : input.target.providerTurn.ordinal; - const ref = input.providerThreadTurns - .filter((turn) => turn.ordinal > boundaryOrdinal && turn.nativeTurnRef !== null) - .sort((a, b) => a.ordinal - b.ordinal)[0]?.nativeTurnRef; - if (ref == null) return null; - if (ref.strength !== "strong" || ref.nativeId === null) return undefined; - return ref.nativeId; } +/** Reads the session tree only for an uncaptured boundary. */ +function piForkEntry( + boundary: Exclude, + providerThreadCreatedAt: DateTime.Utc, + readEntries: Effect.Effect, +): Effect.Effect { + if (boundary.type === "entry") return Effect.succeed(boundary.entryId); + return readEntries.pipe( + Effect.flatMap((data) => { + const listing = piSessionListing(data); + const entryId = + listing === null + ? undefined + : piUncapturedForkEntry(listing, boundary, providerThreadCreatedAt); + return entryId === undefined + ? Effect.fail( + new ProviderAdapter.ProviderAdapterProtocolError({ + driver: PI_PROVIDER, + detail: `Pi turn ${boundary.turn.ordinal} has no captured session-tree entry and none was found in the session tree`, + }), + ) + : Effect.succeed(entryId); + }), + ); +} + +/** + * Pi shares the server clock. An adopted wake can precede its T3 turn, so search from the target + * turn's end through the boundary turn's end. At thread start, use the earlier of thread creation + * and turn start to include adopted wakes while preserving inherited session history. + */ +function piUncapturedForkEntry( + listing: PiSessionListing, + boundary: Extract, + providerThreadCreatedAt: DateTime.Utc, +): string | undefined { + const { startedAt, completedAt } = boundary.turn; + if (startedAt === null || completedAt === null) return undefined; + const from = + boundary.target === null + ? Math.min(DateTime.toEpochMillis(providerThreadCreatedAt), DateTime.toEpochMillis(startedAt)) + : DateTime.toEpochMillis(boundary.target.completedAt ?? startedAt); + const to = DateTime.toEpochMillis(completedAt); + const targetRef = boundary.target?.nativeTurnRef; + const stopAt = targetRef?.strength === "strong" ? targetRef.nativeId : null; + if (stopAt !== null && !listing.entries.has(stopAt)) return undefined; + const walk = walkPiBranch(listing.entries, listing.leafId, stopAt); + if (walk === null || !walk.reachedStop) return undefined; + let earliest: string | undefined; + for (const id of walk.userEntryIds) { + const timestamp = listing.entries.get(id)?.timestamp; + if (timestamp == null) return undefined; + if (timestamp >= from && timestamp <= to) earliest = id; + } + return earliest; +} + function piSubagentOutput(result: unknown): string { const stopReason = recordString(result, "stopReason"); const failed = ```