Skip to content

perf(server): provider-thread reads start from owned rows, not every node of the thread #31012

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
juliusmarminge merged 1 commit into from
Nov 6, 2026
Merged

perf(server): provider-thread reads start from owned rows, not every node of the thread #31012

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
56 changes: 56 additions & 0 deletions apps/server/src/orchestration-v2/ProjectionStore.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -472,6 +472,62 @@ it.effect("record reads restrict provider histories to delivery identities", ()
})).providerThreads,
[],
);
const ownerNodeId = NodeId.make("node:delivery-identities:owner");
yield* store.apply({
id: EventId.make("event:delivery-identities:owner-node"),
type: "node.updated",
threadId,
runId: RunId.make("run:delivery-identities:rolled-back"),
nodeId: ownerNodeId,
driver,
occurredAt: now,
payload: {
id: ownerNodeId,
threadId,
runId: RunId.make("run:delivery-identities:rolled-back"),
rootNodeId: NodeId.make("node:delivery-identities:rolled-back"),
parentNodeId: NodeId.make("node:delivery-identities:rolled-back"),
kind: "subagent",
status: "completed",
countsForRun: false,
providerThreadId: null,
providerTurnId: null,
nativeItemRef: null,
runtimeRequestId: null,
checkpointScopeId: null,
startedAt: now,
completedAt: now,
},
});
yield* store.apply({
id: EventId.make("event:delivery-identities:owned"),
type: "provider-thread.updated",
threadId,
occurredAt: now,
payload: {
id: ProviderThreadId.make("provider-thread:owned"),
appThreadId: ThreadId.make("thread:delivery-identities:child"),
driver,
providerInstanceId,
providerSessionId: null,
ownerNodeId,
nativeThreadRef: null,
nativeConversationHeadRef: null,
lastRunOrdinal: null,
handoffIds: [],
forkedFrom: null,
status: "idle",
firstRunOrdinal: 2,
createdAt: now,
updatedAt: now,
},
});
assert.deepEqual(
(yield* store.getThreadRecords(threadId, ["providerThreads"])).providerThreads.map(
(row) => row.id,
),
["provider-thread:history", "provider-thread:selected", "provider-thread:owned"],
);
// Unselected history must not be decoded, even if its payload is unreadable.
yield* sql`UPDATE orchestration_v2_projection_provider_threads SET payload_json = '{}'
WHERE provider_thread_id = 'provider-thread:history'`;
Expand Down
22 changes: 14 additions & 8 deletions apps/server/src/orchestration-v2/ProjectionStore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1961,6 +1961,16 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
OR item.payload_json LIKE '%"viewedImagePath"%'
OR item.payload_json LIKE '%"toolName":"%delegate_task"%')`;

// Start from the few owned provider threads; a thread can have a large settled node history.
const nodesOwningProviderThreads = (threadId: ThreadId) => sql`
SELECT owner.node_id
FROM orchestration_v2_projection_provider_threads AS owned
INDEXED BY orchestration_v2_projection_provider_threads_owner_idx
CROSS JOIN orchestration_v2_projection_nodes AS owner
ON owner.node_id = owned.owner_node_id
WHERE owned.owner_node_id IS NOT NULL AND owner.thread_id = ${threadId}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
`;

const keepRecordedRunField = (payload: Statement.Fragment, path: string) => sql`
CASE
WHEN json_type(excluded.payload_json, ${path}) IS NULL
Expand Down Expand Up @@ -3495,11 +3505,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
SELECT payload_json
FROM orchestration_v2_projection_provider_threads
WHERE (thread_id = ${threadId}
OR owner_node_id IN (
SELECT node_id
FROM orchestration_v2_projection_nodes
WHERE thread_id = ${threadId}
)
OR owner_node_id IN (${nodesOwningProviderThreads(threadId)})
OR provider_thread_id IN (
SELECT provider_thread_id
FROM orchestration_v2_projection_subagents
Expand Down Expand Up @@ -5040,7 +5046,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_provider_threads
WHERE provider_thread_id = ${target.providerThreadId} AND (
thread_id = ${threadId}
OR owner_node_id IN (SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId})
OR owner_node_id IN (${nodesOwningProviderThreads(threadId)})
OR provider_thread_id IN (SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id = ${threadId})
)`,
sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_provider_turns
Expand Down Expand Up @@ -5280,7 +5286,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_provider_threads
WHERE provider_thread_id = ${run?.providerThreadId ?? null}
AND (thread_id = ${threadId}
OR owner_node_id IN (SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId})
OR owner_node_id IN (${nodesOwningProviderThreads(threadId)})
OR provider_thread_id IN (SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id = ${threadId}))`,
sql<{ readonly ordinal: number }>`SELECT ordinal_within_scope AS ordinal
FROM orchestration_v2_projection_checkpoints
Expand Down Expand Up @@ -5319,7 +5325,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_provider_threads
WHERE provider_thread_id = ${target.providerThreadId}
AND (thread_id = ${threadId}
OR owner_node_id IN (SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId})
OR owner_node_id IN (${nodesOwningProviderThreads(threadId)})
OR provider_thread_id IN (SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id = ${threadId}))`,
sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_checkpoints
WHERE thread_id = ${threadId} AND checkpoint_id = ${target.checkpointId}`,
Expand Down
Loading