Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 114 additions & 22 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,23 @@ const driver = ProviderDriverKind.make("codex");
const providerInstanceId = modelSelection.instanceId;
const encodeUnknownJsonString = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown));

const traceSqlStatements = () => {
const statements: Array<string> = [];
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options);
const end = span.end.bind(span);
span.end = (endTime, exit) => {
end(endTime, exit);
const query = span.attributes.get("db.query.text");
if (typeof query === "string") statements.push(query);
};
return span;
},
});
return { statements, tracer };
};

const addRolledBackRecoveryCandidate = Effect.fn("addRolledBackRecoveryCandidate")(function* (
suffix: string,
) {
Expand Down Expand Up @@ -590,19 +607,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
});
yield* sql`INSERT INTO orchestration_v2_projection_turn_items ${sql.insert(rows)}`;
}
const statements: Array<string> = [];
const tracer = Tracer.make({
span(options) {
const span = new Tracer.NativeSpan(options);
const end = span.end.bind(span);
span.end = (endTime, exit) => {
end(endTime, exit);
const query = span.attributes.get("db.query.text");
if (typeof query === "string") statements.push(query);
};
return span;
},
});
const { statements, tracer } = traceSqlStatements();
const initial = yield* Effect.acquireUseRelease(
Effect.sync(() => vi.spyOn(JSON, "parse")),
(parse) =>
Expand Down Expand Up @@ -635,10 +640,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
/SEARCH item USING INDEX orchestration_v2_projection_turn_items_user_message_idx \(thread_id=\? AND ordinal<\?\)/,
);
const topLevel = windowPlan.filter((row) => row.parent === 0).map((row) => row.detail);
assert.include(
topLevel,
"SEARCH item USING INDEX sqlite_autoindex_orchestration_v2_projection_turn_items_1 (turn_item_id=?)",
);
assert.include(topLevel, "SEARCH item USING INTEGER PRIMARY KEY (rowid=?)");
assert.notInclude(topLevel, "USE TEMP B-TREE FOR ORDER BY");
const nodeStatement = statements.find((statement) =>
statement.includes("WITH RECURSIVE retained"),
Expand Down Expand Up @@ -779,7 +781,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
nativeItemRef: null,
runtimeRequestId: null,
checkpointScopeId: null,
startedAt: now,
startedAt:
index === 998
? null
: index === 999
? DateTime.makeUnsafe("2020-01-01T00:00:00.000Z")
: now,
completedAt: index === -1 ? null : now,
},
});
Expand Down Expand Up @@ -838,7 +845,32 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
resolvedAt: null,
},
});
const snapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 75 });
const { statements, tracer } = traceSqlStatements();
const snapshot = yield* projectionStore
.getThreadSnapshotWindow(threadId, { rowLimit: 75 })
.pipe(Effect.withTracer(tracer));
const nodeStatement = statements.find(
(statement) =>
statement.includes("payload_json") &&
statement.includes("orchestration_v2_projection_nodes") &&
!statement.includes("WITH RECURSIVE"),
);
assert.isDefined(nodeStatement);
const nodePlan = yield* (yield* SqlClient.SqlClient).unsafe<{ readonly detail: string }>(
`EXPLAIN QUERY PLAN ${nodeStatement}`,
);
const nodeLookups = nodePlan
.map((row) => row.detail)
.filter((detail) => detail.startsWith("SEARCH "));
// Cost must follow the retained cohort, even when the thread has
// thousands of completed nodes outside the requested window.
assert.deepEqual(nodeLookups, [
"SEARCH node USING INDEX sqlite_autoindex_orchestration_v2_projection_nodes_1 (node_id=?)",
]);
assert.notInclude(
nodePlan.map((row) => row.detail),
"USE TEMP B-TREE FOR ORDER BY",
);
assert.lengthOf(snapshot.projection.visibleTurnItems, 75);
assert.lengthOf(snapshot.projection.nodes, 79);
const retained = new Set(snapshot.projection.nodes.map((node) => node.id));
Expand All @@ -862,6 +894,27 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
);
const full = yield* projectionStore.getThreadSnapshot(threadId);
assert.lengthOf(full.projection.nodes, 1003);
// Bounded reads keep the full read's payloads and ordering, including
// null start times and equal timestamps.
assert.deepEqual(
snapshot.projection.nodes,
full.projection.nodes.filter((node) => retained.has(node.id)),
);
const sql = yield* SqlClient.SqlClient;
const duplicateItemId = TurnItemId.make("item:bounded-node-history:duplicate");
yield* sql`
INSERT INTO orchestration_v2_projection_turn_items
SELECT ${duplicateItemId}, thread_id, run_id, node_id, provider_thread_id,
provider_turn_id, parent_item_id, ordinal + 1, type, status, updated_at,
json_set(payload_json, '$.id', ${duplicateItemId}, '$.ordinal', ordinal + 1)
FROM orchestration_v2_projection_turn_items
WHERE turn_item_id = ${TurnItemId.make("item:bounded-node-history:999")}
`;
const sharedNodeSnapshot = yield* projectionStore.getThreadSnapshotWindow(threadId, {
rowLimit: 76,
});
assert.lengthOf(sharedNodeSnapshot.projection.visibleTurnItems, 76);
assert.deepEqual(sharedNodeSnapshot.projection.nodes, snapshot.projection.nodes);
}),
);

Expand Down Expand Up @@ -1055,6 +1108,34 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
Array.from({ length: 1_000 }, (_, index) => `turn-item:bounded-sql-history:${index + 1}`),
);

// Physical insertion order must not change the ordinal/ID timeline order.
const tiedIds = [
TurnItemId.make("turn-item:bounded-sql-history:tie:b"),
TurnItemId.make("turn-item:bounded-sql-history:tie:a"),
];
for (const id of tiedIds) {
yield* sql`
INSERT INTO orchestration_v2_projection_turn_items (
turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
parent_item_id, ordinal, type, status, updated_at, payload_json
)
SELECT ${id}, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
parent_item_id, 2000, type, status, updated_at,
json_set(payload_json, '$.id', ${id}, '$.ordinal', 2000)
FROM orchestration_v2_projection_turn_items
WHERE turn_item_id = ${itemId}
`;
}
const tied = yield* projectionStore.getThreadSnapshotWindow(threadId, { rowLimit: 3 });
assert.deepEqual(
tied.projection.turnItems.map((item) => item.id),
[TurnItemId.make("turn-item:bounded-sql-history:1000"), tiedIds[1], tiedIds[0]],
);
yield* sql`
DELETE FROM orchestration_v2_projection_turn_items
WHERE turn_item_id IN ${sql.in(tiedIds)}
`;

const hiddenRunId = RunId.make("run:bounded-sql-history:hidden-suffix");
yield* sql`
INSERT INTO orchestration_v2_projection_runs (
Expand Down Expand Up @@ -4056,9 +4137,20 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
`;
}

const boundedFork = yield* projectionStore.getThreadSnapshotWindow(targetThreadId, {
rowLimit: 2,
});
const { statements, tracer } = traceSqlStatements();
const boundedFork = yield* projectionStore
.getThreadSnapshotWindow(targetThreadId, { rowLimit: 2 })
.pipe(Effect.withTracer(tracer));
// Only the target's control state is hydrated. Fork ancestors contribute
// timeline rows, runs and attempts, regardless of their own live work.
assert.lengthOf(
statements.filter((statement) => statement.includes("WITH RECURSIVE retained")),
1,
);
assert.lengthOf(
statements.filter((statement) => statement.includes("SELECT node.payload_json")),
1,
);
assert.lengthOf(boundedFork.projection.turnItems, 0);
assert.deepEqual(
boundedFork.projection.visibleTurnItems.map((row) => [row.visibility, row.item.type]),
Expand Down
94 changes: 58 additions & 36 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,6 @@ import {
import {
createOrchestrationV2TurnItemVisibility,
isOrchestrationV2SupersededInterrupt,
isOrchestrationV2TurnItemVisible,
} from "@t3tools/shared/orchestrationV2Timeline";
import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
import * as Context from "effect/Context";
Expand Down Expand Up @@ -1182,22 +1181,18 @@ function sortMessagesByTurnItemOrder(
function activeLocalTurnItems(
projection: OrchestrationV2ThreadProjection,
): Array<OrchestrationV2ProjectedTurnItem> {
return projection.turnItems
.filter((item) =>
isOrchestrationV2TurnItemVisible({
item,
runs: projection.runs,
attempts: projection.attempts,
items: projection.turnItems,
}),
)
.map((item, position) => ({
position,
visibility: "local" as const,
sourceThreadId: item.threadId,
sourceItemId: item.id,
item,
}));
const isVisible = createOrchestrationV2TurnItemVisibility({
runs: projection.runs,
attempts: projection.attempts,
items: projection.turnItems,
});
return projection.turnItems.filter(isVisible).map((item, position) => ({
position,
visibility: "local" as const,
sourceThreadId: item.threadId,
sourceItemId: item.id,
item,
}));
}

function localVisibleTurnItems(
Expand Down Expand Up @@ -2722,7 +2717,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
`
: yield* sql<PayloadRow>`
WITH eligible AS NOT MATERIALIZED (
SELECT item.payload_json, item.ordinal, item.turn_item_id,
SELECT item.rowid AS item_rowid, item.payload_json, item.ordinal, item.turn_item_id,
item.run_id, item.node_id, item.type
FROM orchestration_v2_projection_turn_items AS item
LEFT JOIN orchestration_v2_projection_runs AS run
Expand Down Expand Up @@ -2811,7 +2806,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
-- Choose and sort rows by ID, then fetch payloads in that
-- order. A turn window can hold megabytes of tool output, and
-- carrying it through the union and sort cost about a second.
SELECT ordinal, turn_item_id, run_id, type
SELECT item_rowid, ordinal, turn_item_id, run_id, type
FROM eligible
WHERE ordinal >= (SELECT ordinal FROM boundary)
ORDER BY ordinal DESC, turn_item_id DESC
Expand All @@ -2821,19 +2816,19 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
ELSE ${window.rowLimit}
END
), retained AS MATERIALIZED (
SELECT ordinal, turn_item_id FROM selected
SELECT item_rowid, ordinal, turn_item_id FROM selected
UNION
SELECT request.ordinal, request.turn_item_id
SELECT request.rowid AS item_rowid, request.ordinal, request.turn_item_id
FROM orchestration_v2_projection_turn_items AS request
WHERE request.run_id IN (
SELECT run_id FROM selected
WHERE type = 'run_interrupt_result' AND run_id IS NOT NULL
)
AND request.type = 'run_interrupt_request'
UNION
SELECT latest.ordinal, latest.turn_item_id
SELECT latest.item_rowid, latest.ordinal, latest.turn_item_id
FROM (
SELECT ordinal, turn_item_id
SELECT rowid AS item_rowid, ordinal, turn_item_id
FROM orchestration_v2_projection_turn_items
WHERE thread_id = ${threadId}
AND ${window.anchorItemId ?? null} IS NULL
Expand All @@ -2844,11 +2839,12 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
ORDER BY ordinal ASC, turn_item_id ASC
)
-- CROSS JOIN keeps the sorted IDs as the outer loop, so SQLite
-- skips sorting again once the payloads are attached.
-- skips sorting again once the payloads are attached. Carry the
-- rowid through selection to avoid a second string-key index lookup.
SELECT item.payload_json
FROM retained
CROSS JOIN orchestration_v2_projection_turn_items AS item
ON item.turn_item_id = retained.turn_item_id
ON item.rowid = retained.item_rowid
ORDER BY retained.ordinal ASC, retained.turn_item_id ASC
`;
// Reuse the decoded items for cohort IDs and the resulting projection.
Expand All @@ -2865,7 +2861,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
// A run can contain thousands of completed nodes. Load the visible
// nodes and live control dependencies, then walk only their ancestors.
const cohortNodeIds =
window === undefined
window === undefined ||
(fields !== undefined &&
fields.every(
(field) => field === "runs" || field === "attempts" || field === "turnItems",
))
? cohortJson("nodeId")
: encodeIdList(
(yield* sql<{ readonly node_id: string }>`
Expand Down Expand Up @@ -2900,7 +2900,14 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
INNER JOIN retained ON parent.node_id = retained.node_id
WHERE parent.thread_id = ${threadId} AND parent.parent_node_id IS NOT NULL
)
SELECT node_id FROM retained WHERE node_id IS NOT NULL
-- Sort IDs before fetching bodies, so large payloads never enter
-- SQLite's temporary sort. The node read restores this order by key.
SELECT retained.node_id
FROM retained
LEFT JOIN orchestration_v2_projection_nodes AS node
ON node.node_id = retained.node_id AND node.thread_id = ${threadId}
WHERE retained.node_id IS NOT NULL
ORDER BY COALESCE(node.started_at, ''), retained.node_id ASC
`).map((row) => row.node_id),
);
const cohortProviderThreadIds = cohortJson("providerThreadId");
Expand Down Expand Up @@ -2970,13 +2977,20 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
FROM orchestration_v2_projection_nodes
WHERE thread_id = ${threadId}
ORDER BY COALESCE(started_at, ''), node_id ASC
`
: sql<PayloadRow>`
SELECT payload_json FROM orchestration_v2_projection_nodes
WHERE thread_id = ${threadId}
AND node_id IN (SELECT value FROM json_each(${cohortNodeIds}))
ORDER BY COALESCE(started_at, ''), node_id ASC
`,
`
: sql<PayloadRow & { readonly position: number }>`
-- Keep retained IDs outermost so SQLite looks up each node by its
-- primary key instead of scanning every historical node in the thread.
SELECT node.payload_json, retained.key AS position
FROM json_each(${cohortNodeIds}) AS retained
CROSS JOIN orchestration_v2_projection_nodes AS node
ON node.node_id = retained.value
WHERE node.thread_id = ${threadId}
`.pipe(
Effect.map((rows) =>
rows.toSorted((left, right) => left.position - right.position),
),
),
fields !== undefined && !fields.includes("subagents")
? Effect.succeed([])
: window === undefined
Expand Down Expand Up @@ -3257,7 +3271,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
visibleTurnItems: [],
updatedAt: thread.updatedAt,
} satisfies OrchestrationV2ThreadProjection;
return fields === undefined ? withLocalVisibleTurnItems(projection) : projection;
return projection;
}).pipe(
Effect.mapError((cause) =>
isProjectionStoreThreadNotFoundError(cause)
Expand Down Expand Up @@ -3289,7 +3303,15 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
(window?.historyAnchor !== undefined && window.historyAnchor.threadId !== threadId)
? { ...window, rowLimit: 0, anchorItemId: undefined }
: window;
const projection = yield* readCanonicalProjection(threadId, localWindow);
// Ancestors contribute timeline items, not their provider/control state.
// Keep their runs and attempts for fork boundaries and visibility.
const projection = yield* readCanonicalProjection(
threadId,
localWindow,
window !== undefined && seenThreadIds.size > 0
? ["runs", "attempts", "turnItems"]
: undefined,
);
const forkedFrom = projection.thread.forkedFrom;
if (forkedFrom?.type !== "run" || seenThreadIds.has(forkedFrom.threadId)) {
return withLocalVisibleTurnItems(projection);
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/orchestration-v2/WireProjection.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,7 @@ describe("orchestration V2 wire projection", () => {
deletedAt: null,
},
runs: [],
attempts: [],
runtimeRequests: [],
messages: [
{
Expand Down
3 changes: 2 additions & 1 deletion apps/server/src/orchestration-v2/threadHistoryPaging.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -634,7 +634,7 @@ describe("threadHistoryPaging", () => {
});

it("preserves oversized actionable state and reports the budget exception", () => {
const projection = makeProjection([]);
const projection = makeProjection([makeRow(0)]);
const actionable = {
...projection,
plans: [
Expand All @@ -654,6 +654,7 @@ describe("threadHistoryPaging", () => {

expect(bounded.payloadBudgetExceeded).toBe(true);
expect(bounded.projection.plans[0]).toEqual(actionable.plans[0]);
expect(bounded.projection.visibleTurnItems).toEqual(actionable.visibleTurnItems);
});

it("keeps paged historical plan detail in the turn item and only status in its artifact", () => {
Expand Down
Loading
Loading