diff --git a/apps/server/src/orchestration-v2/ContextHandoffService.test.ts b/apps/server/src/orchestration-v2/ContextHandoffService.test.ts index 78958c2e317d5..2f7ddf54f2f66 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffService.test.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffService.test.ts @@ -293,6 +293,7 @@ it.effect("inherited history stops at a missing fork ancestor", () => }; const items = yield* ContextHandoffService.readInheritedHandoffItems( { + getForkSourceRunIds: () => Effect.succeed([]), getThreadRecords: (missing) => Effect.fail(new ProjectionStoreThreadNotFoundError({ threadId: missing })), }, diff --git a/apps/server/src/orchestration-v2/ContextHandoffService.ts b/apps/server/src/orchestration-v2/ContextHandoffService.ts index c9022ee76a807..5f7cb11b67bd2 100644 --- a/apps/server/src/orchestration-v2/ContextHandoffService.ts +++ b/apps/server/src/orchestration-v2/ContextHandoffService.ts @@ -150,17 +150,34 @@ type ForkHistorySource = Pick, + projectionStore: Pick, source: ForkHistorySource, throughRun: Pick, + forkThreadId: ThreadId, ) { + const inheritedRunIds = new Set(yield* projectionStore.getForkSourceRunIds(forkThreadId)); const { turnItems } = yield* projectionStore.getThreadRecords(source.thread.id, ["turnItems"], { turnItemRunIds: [ - ...source.runs.filter((run) => run.ordinal <= throughRun.ordinal).map((run) => run.id), + ...source.runs + .filter((run) => run.ordinal <= throughRun.ordinal && inheritedRunIds.has(run.id)) + .map((run) => run.id), ...(source.thread.historyOrigin === "v1_import" ? [null] : []), ], }); - return visibleConversationItems(source, turnItems); + return visibleConversationItems( + { + ...source, + runs: source.runs.map((run) => + run.status === "rolled_back" && inheritedRunIds.has(run.id) + ? { + ...run, + status: run.startedAt === null ? ("cancelled" as const) : ("completed" as const), + } + : run, + ), + }, + turnItems, + ); }, ); @@ -168,7 +185,7 @@ const readOwnHistoryItems = Effect.fn("orchestrationV2.contextHandoff.readOwnHis export const readInheritedHandoffItems = Effect.fn( "orchestrationV2.contextHandoff.readInheritedHandoffItems", )(function* ( - projectionStore: Pick, + projectionStore: Pick, thread: OrchestrationV2AppThread, ) { const segments: Array> = []; @@ -186,7 +203,7 @@ export const readInheritedHandoffItems = Effect.fn( .pipe(Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) })); const forkRun = source?.runs.find((run) => run.id === runId); if (source === null || forkRun === undefined) break; - segments.unshift(yield* readOwnHistoryItems(projectionStore, source, forkRun)); + segments.unshift(yield* readOwnHistoryItems(projectionStore, source, forkRun, current.id)); current = source.thread; } return segments.flat(); @@ -196,13 +213,14 @@ export const readInheritedHandoffItems = Effect.fn( export const readForkHistoryItems = Effect.fn( "orchestrationV2.contextHandoff.readForkHistoryItems", )(function* ( - projectionStore: Pick, + projectionStore: Pick, source: ForkHistorySource, throughRun: Pick, + forkThreadId: ThreadId, ) { return [ ...(yield* readInheritedHandoffItems(projectionStore, source.thread)), - ...(yield* readOwnHistoryItems(projectionStore, source, throughRun)), + ...(yield* readOwnHistoryItems(projectionStore, source, throughRun, forkThreadId)), ]; }); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a3ad41ec9eb4a..21328521dd9eb 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -8163,7 +8163,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const portableForkItems = !requiresPortableFork || sourceProjection === null || sourceRun === null ? [] - : yield* readForkHistoryItems(projectionStore, sourceProjection, sourceRun).pipe( + : yield* readForkHistoryItems( + projectionStore, + sourceProjection, + sourceRun, + command.threadId, + ).pipe( Effect.mapError( (cause) => new OrchestratorProjectionError({ threadId: sourceProjection.thread.id, cause }), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 3b34434017e9d..0e3704422a2b7 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -48,6 +48,7 @@ import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ContextHandoffService from "./ContextHandoffService.ts"; import { buildBoundedThreadProjection, decodeThreadHistoryCursor, @@ -294,6 +295,8 @@ it.effect("routes full projection reads to readers and retains uncommitted write forkedFrom: { type: "run", threadId: parentId, runId: parent.runs[0]!.id }, }, }); + // Historical forks use the timestamp fallback even when their source run is now rolled back. + yield* sql`UPDATE orchestration_v2_projection_threads SET fork_run_ids_json = NULL WHERE thread_id = ${threadId}`; const messageId = MessageId.make("message:reader-routing"); yield* store.apply({ id: EventId.make("event:reader-routing:message"), @@ -9076,6 +9079,33 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { deletedAt: null, }, }); + yield* projectionStore.apply({ + id: EventId.make("event:projection-imported-fork-shell:source-run"), + type: "run.created", + threadId: sourceThreadId, + runId: sourceRunId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: sourceRunId, + threadId: sourceThreadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:projection-imported-fork-shell:run"), + rootNodeId, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }); + for (const [threadId, sideChat] of [ [targetThreadId, false], [sideChatThreadId, true], @@ -9119,32 +9149,6 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }, }); } - yield* projectionStore.apply({ - id: EventId.make("event:projection-imported-fork-shell:source-run"), - type: "run.created", - threadId: sourceThreadId, - runId: sourceRunId, - nodeId: rootNodeId, - driver, - occurredAt: now, - payload: { - id: sourceRunId, - threadId: sourceThreadId, - ordinal: 1, - providerInstanceId, - modelSelection, - providerThreadId: null, - userMessageId: MessageId.make("message:projection-imported-fork-shell:run"), - rootNodeId, - activeAttemptId: null, - status: "completed", - requestedAt: now, - startedAt: now, - completedAt: now, - checkpointId: null, - contextHandoffId: null, - }, - }); const applyAssistantItem = (suffix: string, runId: RunId | null, ordinal: number) => projectionStore.apply({ @@ -9664,43 +9668,6 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { deletedAt: null, }, }); - yield* projectionStore.apply({ - id: EventId.make("event:projection-fork-source-rollback:target-thread"), - type: "thread.created", - threadId: targetThreadId, - occurredAt: now, - payload: { - createdBy: "user", - creationSource: "web", - id: targetThreadId, - projectId, - title: "Projection fork source rollback target", - providerInstanceId, - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - activeProviderThreadId: targetProviderThreadId, - lineage: { - parentThreadId: sourceThreadId, - relationshipToParent: "fork", - rootThreadId: sourceThreadId, - }, - forkedFrom: { - type: "run", - threadId: sourceThreadId, - runId: sourceRun2Id, - }, - createdAt: now, - updatedAt: now, - archivedAt: null, - settledOverride: null, - settledAt: null, - lastVisitedAt: null, - deletedAt: null, - }, - }); for (const [ordinal, runId, nodeId, promptText, responseText] of [ [1, sourceRun1Id, sourceRun1NodeId, "source one", "one"], @@ -9799,6 +9766,44 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }); } + yield* projectionStore.apply({ + id: EventId.make("event:projection-fork-source-rollback:target-thread"), + type: "thread.created", + threadId: targetThreadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: targetThreadId, + projectId, + title: "Projection fork source rollback target", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: targetProviderThreadId, + lineage: { + parentThreadId: sourceThreadId, + relationshipToParent: "fork", + rootThreadId: sourceThreadId, + }, + forkedFrom: { + type: "run", + threadId: sourceThreadId, + runId: sourceRun2Id, + }, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + const targetBeforeRollback = yield* projectionStore.getThreadProjection(targetThreadId); assert.deepEqual( targetBeforeRollback.visibleTurnItems.map((row) => row.item.type), @@ -10554,10 +10559,13 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { const heldAttemptId = RunAttemptId.make(`attempt:${prefix}:held`); const answeredRunId = RunId.make(`run:${prefix}:answered`); - for (const [threadId, forkedFrom] of [ - [sourceThreadId, null], - [targetThreadId, { type: "run", threadId: sourceThreadId, runId: answeredRunId }], - ] as const) { + const createThread = Effect.fn("test.createForkHeldQueueThread")(function* ( + threadId: ThreadId, + ) { + const forkedFrom = + threadId === sourceThreadId + ? null + : { type: "run" as const, threadId: sourceThreadId, runId: answeredRunId }; yield* projectionStore.apply({ id: EventId.make(`event:${threadId}:created`), type: "thread.created", @@ -10591,7 +10599,8 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { deletedAt: null, }, }); - } + }); + yield* createThread(sourceThreadId); const applyRun = ( runId: RunId, @@ -10707,6 +10716,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { text: "delivered steer", inputIntent: "queued_turn", }); + yield* createThread(targetThreadId); // Enough local turns for a turn-limited window to count inherited rows in SQL. for (let index = 0; index < THREAD_HISTORY_PAGE_POLICY.maxUserTurns; index += 1) { yield* applyUserItem({ @@ -10917,13 +10927,14 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { } const forkedFrom = { type: "run" as const, threadId: sourceThreadId, runId: runIds[2]! }; - yield* projectionStore.apply({ + const forkEvent = { id: EventId.make(`event:${prefix}:target-thread`), - type: "thread.created", + type: "thread.created" as const, threadId: targetThreadId, occurredAt: forkedAt, - payload: threadPayload({ id: targetThreadId, forkedFrom, createdAt: forkedAt }), - }); + payload: threadPayload({ id: targetThreadId, forkedFrom, createdAt: now }), + }; + yield* projectionStore.apply(forkEvent); const visibleTexts = (rows: ReadonlyArray<{ readonly item: OrchestrationV2TurnItem }>) => rows.map((row) => @@ -10939,21 +10950,43 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { "reply three", ]); const expected = ["prompt one", "reply one", "prompt three", "reply three", "fork"]; - const target = yield* projectionStore.getThreadProjection(targetThreadId); - assert.deepEqual(visibleTexts(target.visibleTurnItems), expected); - const page = yield* projectionStore.getTimelinePage(targetThreadId, { - view: "activity", - limit: 10, - }); - assert.deepEqual(visibleTexts(page.items), expected); - const window = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, { - rowLimit: 10, + const assertHistory = Effect.fn("test.assertForkHistory")(function* () { + const handoff = yield* ContextHandoffService.readInheritedHandoffItems( + projectionStore, + (yield* projectionStore.getThreadRecords(targetThreadId, [])).thread, + ); + assert.deepEqual(visibleTexts(handoff.map((item) => ({ item }))), expected.slice(0, -1)); + const target = yield* projectionStore.getThreadProjection(targetThreadId); + assert.deepEqual(visibleTexts(target.visibleTurnItems), expected); + const page = yield* projectionStore.getTimelinePage(targetThreadId, { + view: "activity", + limit: 10, + }); + assert.deepEqual(visibleTexts(page.items), expected); + const window = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, { + rowLimit: 10, + }); + assert.deepEqual(visibleTexts(window.projection.visibleTurnItems), expected); + assert.equal( + (yield* projectionStore.getThreadShell(targetThreadId))?.visibleItemCount, + expected.length, + ); }); - assert.deepEqual(visibleTexts(window.projection.visibleTurnItems), expected); - assert.equal( - (yield* projectionStore.getThreadShell(targetThreadId))?.visibleItemCount, - expected.length, - ); + yield* assertHistory(); + for (const index of [2, 0]) { + yield* projectionStore.apply({ + id: EventId.make(`event:${prefix}:later-rewind-${index}`), + type: "run.updated", + threadId: sourceThreadId, + runId: runIds[index]!, + nodeId: nodeIds[index]!, + driver, + occurredAt: forkedAt, + payload: { ...runPayload(index, "rolled_back"), completedAt: now }, + }); + yield* projectionStore.apply(forkEvent); + yield* assertHistory(); + } }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 841cd0a0b5b7d..f4a9dd4b84e1d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -534,6 +534,10 @@ export interface ProjectionStoreV2Shape { filter?: ProjectionRecordFilter, ) => Effect.Effect, ProjectionStoreV2Error>; + readonly getForkSourceRunIds: ( + threadId: ThreadId, + ) => Effect.Effect, ProjectionStoreV2Error>; + readonly apply: ( event: OrchestrationV2DomainEvent, ) => Effect.Effect; @@ -776,7 +780,7 @@ type ProjectionWindow = Omit< "anchorThreadId" > & { readonly suppressLocal?: boolean | undefined; - readonly forkCreatedAt?: DateTime.Utc | undefined; + readonly forkRunIdsJson?: string | undefined; readonly historyAnchor?: { readonly threadId: ThreadId; readonly itemId: TurnItemId } | undefined; }; @@ -1072,6 +1076,7 @@ export type AgentAwarenessThread = ProjectThreadAwarenessV2Input["thread"] & >; type ShellThreadRow = { + readonly fork_run_ids_json: string | null; readonly thread_id: string; readonly payload_json: string; readonly forked_from_run_source_thread_id: string | null; @@ -1239,6 +1244,10 @@ const decodePullRequestThread = Schema.decodeUnknownEffect( Schema.fromJsonString(PullRequestThread), ); +const forkRunIdsJson = Schema.fromJsonString(Schema.Array(RunId)); +const decodeForkRunIds = Schema.decodeUnknownEffect(forkRunIdsJson); +const encodeForkRunIds = Schema.encodeEffect(forkRunIdsJson); + const decodeThreadPayload = Schema.decodeUnknownEffect( Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema), ); @@ -1479,11 +1488,6 @@ function createInheritedTurnItemVisibility( }); } -// Rollback stamps completedAt; a fork made after the rewind never received that run. -function runRolledBackAt(run: Pick) { - return run.status === "rolled_back" ? run.completedAt : null; -} - function isRolledBackBeforeFork( rolledBackAt: DateTime.Utc | null, forkCreatedAt: DateTime.Utc, @@ -1493,26 +1497,26 @@ function isRolledBackBeforeFork( function forkInheritedRunOrdinals( runs: ReadonlyArray, - forkCreatedAt: DateTime.Utc, + inheritedRunIds: ReadonlyArray, ): Map { - return new Map( - runs - .filter((run) => !isRolledBackBeforeFork(runRolledBackAt(run), forkCreatedAt)) - .map((run) => [run.id, run.ordinal]), - ); + const membership = new Set(inheritedRunIds); + return new Map(runs.filter((run) => membership.has(run.id)).map((run) => [run.id, run.ordinal])); } function visibleTurnItemsThroughRun(input: { readonly sourceProjection: OrchestrationV2ThreadProjection; readonly sourceRunId: NonNullable; - readonly forkCreatedAt: DateTime.Utc; + readonly inheritedRunIds: ReadonlyArray; }): Array> { const sourceRun = input.sourceProjection.runs.find((run) => run.id === input.sourceRunId); if (sourceRun === undefined) { return []; } - const runOrdinalById = forkInheritedRunOrdinals(input.sourceProjection.runs, input.forkCreatedAt); + const runOrdinalById = forkInheritedRunOrdinals( + input.sourceProjection.runs, + input.inheritedRunIds, + ); const isVisible = createInheritedTurnItemVisibility({ runs: input.sourceProjection.runs, attempts: input.sourceProjection.attempts, @@ -1555,6 +1559,7 @@ function buildVisibleTurnItems(input: { readonly projection: OrchestrationV2ThreadProjection; readonly sourceProjection: OrchestrationV2ThreadProjection | null; readonly inheritedRowCount?: number; + readonly inheritedRunIds?: ReadonlyArray; }): Array { const forkedFrom = inheritedRunFork(input.projection.thread); if ( @@ -1570,7 +1575,7 @@ function buildVisibleTurnItems(input: { : visibleTurnItemsThroughRun({ sourceProjection: input.sourceProjection, sourceRunId: forkedFrom.runId, - forkCreatedAt: input.projection.thread.createdAt, + inheritedRunIds: input.inheritedRunIds ?? [], }); const markerItem = makeForkMarkerTurnItem({ targetProjection: input.projection, @@ -1922,6 +1927,7 @@ type ShellThreadState = { readonly runlessItemCount: number; readonly updatedAt: OrchestrationV2ThreadProjection["updatedAt"]; readonly runsById: ReadonlyMap; + readonly inheritedRunIds: ReadonlySet | null; readonly itemCountByRunId: ReadonlyMap; }; @@ -1949,6 +1955,7 @@ function itemCountThroughRun(input: { readonly state: ShellThreadState; readonly runId: RunId; readonly forkCreatedAt: DateTime.Utc; + readonly inheritedRunIds: ReadonlySet | null; }): number { const runOrdinal = input.state.runsById.get(input.runId)?.ordinal; if (runOrdinal === undefined) { @@ -1961,7 +1968,8 @@ function itemCountThroughRun(input: { if ( itemRun !== undefined && itemRun.ordinal <= runOrdinal && - !isRolledBackBeforeFork(itemRun.rolledBackAt, input.forkCreatedAt) + (input.inheritedRunIds?.has(runId) ?? + !isRolledBackBeforeFork(itemRun.rolledBackAt, input.forkCreatedAt)) ) { count += itemCount; } @@ -2023,6 +2031,7 @@ function visibleItemCountForShell(input: { state: sourceState, runId: forkedFrom.runId, forkCreatedAt: state.thread.createdAt, + inheritedRunIds: state.inheritedRunIds, }) + 1 + state.itemCount @@ -2313,6 +2322,24 @@ export const layer: Layer.Layer = deleted_at = excluded.deleted_at, payload_json = excluded.payload_json `; + // EventSink applies committed events serially; retries retain the original membership. + if (event.type === "thread.created" && event.payload.forkedFrom?.type === "run") { + const fork = event.payload.forkedFrom; + yield* sql` + UPDATE orchestration_v2_projection_threads + SET fork_run_ids_json = ( + SELECT json_group_array(run_id) + FROM orchestration_v2_projection_runs + WHERE thread_id = ${fork.threadId} + AND ordinal <= ( + SELECT ordinal FROM orchestration_v2_projection_runs + WHERE thread_id = ${fork.threadId} AND run_id = ${fork.runId} + ) + AND status <> 'rolled_back' + ) + WHERE thread_id = ${event.threadId} AND fork_run_ids_json IS NULL + `; + } break; } case "run.created": @@ -3179,15 +3206,7 @@ export const layer: Layer.Layer = AND required_run.run_id = ${window.requiredRunId ?? null} LIMIT 1 ) - AND NOT COALESCE( - run.status = 'rolled_back' - AND julianday(run.completed_at) < julianday(${ - window.forkCreatedAt === undefined - ? null - : DateTime.formatIso(window.forkCreatedAt) - }), - 0 - ) + AND run.run_id IN (SELECT value FROM json_each(${window.forkRunIdsJson ?? "[]"})) ) ) AND NOT ( @@ -4373,6 +4392,7 @@ export const layer: Layer.Layer = return withLocalVisibleTurnItems(projection); } + const inheritedRunIds = yield* getForkSourceRunIds(threadId); const sourceWindow = window === undefined ? undefined @@ -4451,7 +4471,7 @@ export const layer: Layer.Layer = userTurnLimit: window.userTurnLimit, deferWork: window.deferWork, requiredRunId: forkedFrom.runId, - forkCreatedAt: projection.thread.createdAt, + forkRunIdsJson: yield* encodeForkRunIds(inheritedRunIds), suppressLocal: anchor === undefined || anchorIsInDescendant, ...(anchor === undefined || anchorIsInDescendant ? {} @@ -4511,6 +4531,7 @@ export const layer: Layer.Layer = visibleTurnItems: buildVisibleTurnItems({ projection, sourceProjection, + inheritedRunIds, }), }; }, @@ -6275,6 +6296,28 @@ export const layer: Layer.Layer = (effect, threadId) => effect.pipe(Effect.mapError(controlReadError(threadId))), ); + const getForkSourceRunIds = Effect.fn("ProjectionStore.getForkSourceRunIds")( + function* (threadId: ThreadId) { + const rows = yield* sql<{ readonly run_ids: string }>` + SELECT COALESCE(fork.fork_run_ids_json, ( + SELECT json_group_array(run.run_id) + FROM orchestration_v2_projection_runs AS run + WHERE run.thread_id = json_extract(fork.payload_json, '$.forkedFrom.threadId') + AND run.ordinal <= ( + SELECT ordinal FROM orchestration_v2_projection_runs + WHERE run_id = json_extract(fork.payload_json, '$.forkedFrom.runId') + ) + AND NOT COALESCE( + run.status = 'rolled_back' AND julianday(run.completed_at) < julianday(fork.created_at), 0 + ) + )) AS run_ids + FROM orchestration_v2_projection_threads AS fork WHERE fork.thread_id = ${threadId} + `; + return yield* decodeForkRunIds(rows[0]?.run_ids ?? "[]"); + }, + (effect, threadId) => effect.pipe(Effect.mapError(controlReadError(threadId))), + ); + const getThreadRecords: ProjectionStoreV2Shape["getThreadRecords"] = ( threadId, fields, @@ -6399,7 +6442,10 @@ export const layer: Layer.Layer = const sourceRun = source.records.runs.find((run) => run.id === fork.runId); let inherited: Array = []; if (sourceRun !== undefined) { - const ordinals = forkInheritedRunOrdinals(source.records.runs, records.thread.createdAt); + const ordinals = forkInheritedRunOrdinals( + source.records.runs, + yield* getForkSourceRunIds(threadId), + ); const isInheritedVisible = createInheritedTurnItemVisibility({ runs: source.records.runs, attempts: source.records.attempts, @@ -7606,6 +7652,7 @@ export const layer: Layer.Layer = SELECT t.thread_id, t.payload_json, + t.fork_run_ids_json, CASE WHEN json_extract(t.payload_json, '$.forkedFrom.type') = 'run' AND json_extract(t.payload_json, '$.lineage.sideChat') IS NOT 1 @@ -8408,6 +8455,10 @@ export const layer: Layer.Layer = runlessItemCount: shellItemCountsByThreadId.get(thread.id)?.runlessItemCount ?? 0, updatedAt: thread.updatedAt, runsById: runsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(), + inheritedRunIds: + row.fork_run_ids_json === null + ? null + : new Set(yield* decodeForkRunIds(row.fork_run_ids_json)), itemCountByRunId: itemCountsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(), } satisfies ShellThreadState; }); @@ -8995,6 +9046,7 @@ export const layer: Layer.Layer = getNextTurnItemOrdinal: readOnReader(getNextTurnItemOrdinal), getTurnItem: readOnReader(getTurnItem), getMessagesById: readOnReader(getMessagesById), + getForkSourceRunIds: readOnReader(getForkSourceRunIds), getThreadRecords: (threadId, fields, filter) => getThreadRecords(threadId, fields, filter).pipe(NodeSqliteClient.readOnly), getRuntimeRequest: readOnReader(getRuntimeRequest), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 3ede421ad72ab..ce451456e4b9f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -247,6 +247,7 @@ it.effect( getNextTurnItemOrdinal: () => Effect.die("unused ordinal read"), getTurnItem: () => Effect.die("unused turn item read"), getMessagesById: () => Effect.die("unused getMessagesById"), + getForkSourceRunIds: () => Effect.die("unused fork membership read"), getThreadRecords: () => Effect.die("unused record read"), getRuntimeRequest: () => Effect.die("unused getRuntimeRequest"), getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"), diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 62f095b6fd913..5adb9ecbe96e3 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -167,6 +167,7 @@ function startWithMissingWorktree( Effect.succeedSome({ workspaceRoot: "/tmp/provider-turn-start-project" } as never), }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getForkSourceRunIds: () => Effect.succeed(projection.runs.map((run) => run.id)), getThreadRecords: () => { projectionReadCount += 1; return Effect.succeed(projection); @@ -995,6 +996,7 @@ function makeLocalCommandHarness(input: { } as never), }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getForkSourceRunIds: () => Effect.succeed(parentRecords.runs.map((run) => run.id)), getThreadRecords, getTurnStartContext, getRuntimeRecoveryProjection, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 8f01d117e1739..0d2333137c306 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -1107,6 +1107,7 @@ export const layer: Layer.Layer< projectionStore, sourceProjection, sourceRun, + projection.thread.id, ); const createdAt = yield* DateTime.now; const handoff = yield* contextHandoffService.prepareProviderHandoff({ diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 6399ee7b85950..a9145918c2345 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -96,6 +96,7 @@ import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; import Migration0086 from "./Migrations/086_QueuedTurnInputIndex.ts"; import Migration0087 from "./Migrations/087_ProjectionProjectsCustomTitle.ts"; import Migration0088 from "./Migrations/088_TurnItemTypeStatusIndex.ts"; +import Migration0089 from "./Migrations/089_ForkHistoryMembership.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -187,6 +188,7 @@ const migrationEntries = [ [86, "QueuedTurnInputIndex", Migration0086], [87, "ProjectionProjectsCustomTitle", Migration0087], [88, "TurnItemTypeStatusIndex", Migration0088], + [89, "ForkHistoryMembership", Migration0089], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/089_ForkHistoryMembership.test.ts b/apps/server/src/persistence/Migrations/089_ForkHistoryMembership.test.ts new file mode 100644 index 0000000000000..4371bd65c8e07 --- /dev/null +++ b/apps/server/src/persistence/Migrations/089_ForkHistoryMembership.test.ts @@ -0,0 +1,63 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; + +it.layer(NodeSqliteClient.layer({ filename: ":memory:" }))("089_ForkHistoryMembership", (it) => { + it.effect("preserves historical boundary runs when earlier lifecycle evidence was pruned", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 88 }); + const forkedAt = "2026-11-10T12:02:00.000Z"; + const rewoundAt = "2026-11-10T12:03:00.000Z"; + for (const [id, status, ordinal] of [ + ["kept", "completed", 1], + ["boundary", "rolled_back", 2], + ] as const) { + yield* sql` + INSERT INTO orchestration_v2_projection_runs ( + run_id, thread_id, ordinal, provider, status, requested_at, completed_at, payload_json + ) VALUES (${id}, 'source', ${ordinal}, 'codex', ${status}, ${forkedAt}, ${rewoundAt}, '{}') + `; + } + yield* sql` + INSERT INTO orchestration_v2_projection_threads ( + thread_id, project_id, title, default_provider, runtime_mode, interaction_mode, + created_at, updated_at, payload_json + ) VALUES ('fork', 'project', 'fork', 'codex', 'full-access', 'default', ${forkedAt}, ${forkedAt}, + '{"forkedFrom":{"type":"run","threadId":"source","runId":"boundary"}}') + `; + for (const [sequence, streamId, type, payload] of [ + [100, "fork", "thread.created", {}], + [200, "source", "run.updated", { id: "boundary", status: "rolled_back" }], + ] as const) { + yield* sql` + INSERT INTO orchestration_events ( + sequence, event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES (${sequence}, ${`event-${sequence}`}, 'thread', ${streamId}, ${sequence}, ${type}, + ${rewoundAt}, 'server', ${JSON.stringify(payload)}, '{}', 2) + `; + } + yield* sql` + INSERT INTO orchestration_v2_projection_metadata (projection_name, schema_version, last_sequence, updated_at) + VALUES ('event-log-pruned-through', 1, 150, ${rewoundAt}) + `; + const fallback = () => sql<{ run_id: string }>` + SELECT run_id FROM orchestration_v2_projection_runs + WHERE ordinal <= 2 + AND NOT COALESCE(status = 'rolled_back' AND julianday(completed_at) < julianday(${forkedAt}), 0) + ORDER BY ordinal + `; + assert.deepEqual(yield* fallback(), [{ run_id: "kept" }, { run_id: "boundary" }]); + yield* runMigrations(); + assert.deepEqual( + yield* sql`SELECT fork_run_ids_json FROM orchestration_v2_projection_threads`, + [{ fork_run_ids_json: null }], + ); + assert.deepEqual(yield* fallback(), [{ run_id: "kept" }, { run_id: "boundary" }]); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/089_ForkHistoryMembership.ts b/apps/server/src/persistence/Migrations/089_ForkHistoryMembership.ts new file mode 100644 index 0000000000000..b982de1530b3c --- /dev/null +++ b/apps/server/src/persistence/Migrations/089_ForkHistoryMembership.ts @@ -0,0 +1,8 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + // Pruned lifecycle events cannot establish complete membership; historical forks keep the timestamp fallback. + yield* sql`ALTER TABLE orchestration_v2_projection_threads ADD COLUMN fork_run_ids_json TEXT`; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index a2bba7d790eaf..78f49e4c417bd 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -68,6 +68,7 @@ describe("V2 preview upgrade", () => { [86, "QueuedTurnInputIndex"], [87, "ProjectionProjectsCustomTitle"], [88, "TurnItemTypeStatusIndex"], + [89, "ForkHistoryMembership"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -179,6 +180,7 @@ describe("V2 preview upgrade", () => { [86, "QueuedTurnInputIndex"], [87, "ProjectionProjectsCustomTitle"], [88, "TurnItemTypeStatusIndex"], + [89, "ForkHistoryMembership"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), );