Skip to content
176 changes: 176 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
MessageId,
type ModelSelection,
type OrchestrationV2ProviderThread,
type OrchestrationV2ThreadProjection,
type OrchestrationV2ThreadShellSnapshot,
NodeId,
ProjectId,
Expand Down Expand Up @@ -40,6 +41,7 @@ import {
decodeThreadHistoryCursor,
selectHistoryPageFromCursor,
THREAD_HISTORY_PAGE_POLICY,
THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
} from "./threadHistoryPaging.ts";

const layerTest = Layer.mergeAll(
Expand Down Expand Up @@ -5660,6 +5662,180 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
}),
);

it.effect("skips parent bodies once a fork's own turns fill its first page", () =>
Effect.gen(function* () {
const store = yield* ProjectionStore.ProjectionStoreV2;
const now = DateTime.makeUnsafe("2026-10-01T00:00:00.000Z");
const parentId = ThreadId.make("thread:filled-fork:parent");
const forkId = ThreadId.make("thread:filled-fork:fork");
const parentText = "parent prompt";
const createThread = (
threadId: ThreadId,
forkedFrom: { readonly threadId: ThreadId; readonly runId: RunId } | null,
) =>
store.apply({
id: EventId.make(`event:${threadId}`),
type: "thread.created",
threadId,
occurredAt: now,
payload: {
createdBy: "user",
creationSource: "web",
id: threadId,
projectId: ProjectId.make("project:filled-fork"),
title: "Filled fork",
providerInstanceId,
modelSelection,
runtimeMode: "full-access",
interactionMode: "default",
branch: null,
worktreePath: null,
activeProviderThreadId: null,
lineage: {
parentThreadId: forkedFrom?.threadId ?? null,
relationshipToParent: forkedFrom === null ? null : "fork",
rootThreadId: parentId,
},
forkedFrom: forkedFrom === null ? null : { type: "run", ...forkedFrom },
createdAt: now,
updatedAt: now,
archivedAt: null,
settledOverride: null,
settledAt: null,
lastVisitedAt: null,
deletedAt: null,
},
});
// Each turn is one run with one user prompt.
const addTurn = Effect.fnUntraced(function* (threadId: ThreadId, turn: number, text: string) {
const runId = RunId.make(`run:${threadId}:${turn}`);
const nodeId = NodeId.make(`node:${threadId}:${turn}`);
yield* store.apply({
id: EventId.make(`event:${runId}`),
type: "run.created",
threadId,
runId,
nodeId,
driver,
occurredAt: now,
payload: {
id: runId,
threadId,
ordinal: turn,
providerInstanceId,
modelSelection,
providerThreadId: null,
userMessageId: MessageId.make(`message:${runId}`),
rootNodeId: nodeId,
activeAttemptId: null,
status: "completed",
requestedAt: now,
startedAt: now,
completedAt: now,
checkpointId: null,
contextHandoffId: null,
},
});
yield* store.apply({
id: EventId.make(`event:${runId}:prompt`),
type: "turn-item.updated",
threadId,
runId,
driver,
occurredAt: now,
payload: {
id: TurnItemId.make(`item:${runId}`),
threadId,
runId,
nodeId: null,
providerThreadId: null,
providerTurnId: null,
nativeItemRef: null,
parentItemId: null,
ordinal: turn,
status: "completed",
title: null,
startedAt: now,
completedAt: now,
updatedAt: now,
type: "user_message",
messageId: MessageId.make(`message:${runId}`),
inputIntent: "turn_start",
createdBy: "user",
creationSource: "web",
text,
attachments: [],
},
});
});
const firstPage = (projection: OrchestrationV2ThreadProjection) => {
const bounded = buildBoundedThreadProjection({ projection, snapshotSequence: 0 });
return {
items: bounded.projection.visibleTurnItems,
historyCursor: bounded.historyCursor,
hasMoreHistory: bounded.hasMoreHistory,
};
};
// The initial window must be the tail of the full timeline, with the same
// positions, and bound to the same first page and cursor.
const readFirstPage = Effect.fnUntraced(function* (threadId: ThreadId) {
const full = (yield* store.getThreadSnapshot(threadId)).projection;
const { window, parsedParent } = yield* Effect.acquireUseRelease(
Effect.sync(() => vi.spyOn(JSON, "parse")),
(parse) =>
store
.getThreadSnapshotWindow(threadId, {
rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns,
})
.pipe(
Effect.map(({ projection }) => ({
window: projection,
parsedParent: parse.mock.calls.some(([text]) => text.includes(parentText)),
})),
),
(parse) => Effect.sync(() => parse.mockRestore()),
);
const visible = window.visibleTurnItems;
assert.deepEqual(visible, full.visibleTurnItems.slice(-visible.length));
const page = firstPage(window);
assert.deepEqual(page, firstPage(full));
return { page, parsedParent };
});

yield* createThread(parentId, null);
for (const turn of [1, 2, 3]) yield* addTurn(parentId, turn, parentText);
// The parent kept going after the fork, so its third turn is not inherited.
yield* createThread(forkId, { threadId: parentId, runId: RunId.make(`run:${parentId}:2`) });
for (let turn = 1; turn <= 9; turn += 1) yield* addTurn(forkId, turn, "fork prompt");

// Nine local turns leave room for the parent's last inherited turn, so
// the window still reads parent bodies.
const partial = yield* readFirstPage(forkId);
assert.include(
partial.page.items.map((row) => row.visibility),
"inherited",
);
assert.isTrue(partial.parsedParent);

// Ten local turns fill the page: parent rows only shift positions and
// the cursor, so their bodies are not read.
yield* addTurn(forkId, 10, "fork prompt");
const filled = yield* readFirstPage(forkId);
assert.isNotNull(filled.page.historyCursor);
assert.isFalse(filled.parsedParent);

// A fork of the fork inherits the grandparent's rows through its parent,
// and its parent's own ten turns fill a page too. Counting rows at either
// level would misplace the page, so a nested fork keeps the full read.
const nestedId = ThreadId.make("thread:filled-fork:nested");
yield* createThread(nestedId, { threadId: forkId, runId: RunId.make(`run:${forkId}:10`) });
for (let turn = 1; turn <= 10; turn += 1) yield* addTurn(nestedId, turn, "nested prompt");
const nested = yield* readFirstPage(nestedId);
assert.isTrue(nested.parsedParent);
}),
);

it.effect("a pull request watch keeps a finished thread working until it ends", () =>
Effect.gen(function* () {
const store = yield* ProjectionStore.ProjectionStoreV2;
Expand Down
Loading
Loading