authorgravatar for 101525047+Baker-Harrison@users.noreply.github.comHessboi <101525047+Baker-Harrison@users.noreply.github.com> 2026-10-14 02:17:50-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 00:17:50-07:00
log3b2ebcee2f6eeb4ed3dde6c89fc59f2eb5bd996d
treecb6a337e994765cf32d3c7587952de54fd4acc22
parent8659fcf145c3372161b360b74b094b8e0ab80bec
signature Signed by PGP key B5690EEEBB952194

perf(server): streamed events no longer rewrite the whole thread row (#18890)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 105 insertions(+), 17 deletions(-)

apps/server/src/orchestration-v2/ProjectionStore.test.ts+93
......@@ -2186,6 +2186,99 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => {
21862186 }),
21872187 );
21882188
2189 it.effect("dates thread activity by a run event and leaves the rest of the thread alone", () =>
2190 Effect.gen(function* () {
2191 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2192 const createdAt = yield* DateTime.now;
2193 const runOccurredAt = DateTime.add(createdAt, { seconds: 5 });
2194 const threadId = ThreadId.make("thread:projection-activity");
2195 const runId = RunId.make("run:projection-activity");
2196 const rootNodeId = NodeId.make("node:projection-activity");
2197 const thread = {
2198 createdBy: "user" as const,
2199 creationSource: "web" as const,
2200 id: threadId,
2201 projectId: ProjectId.make("project:projection-activity"),
2202 title: "Projection activity",
2203 providerInstanceId,
2204 modelSelection,
2205 runtimeMode: "full-access" as const,
2206 interactionMode: "default" as const,
2207 branch: "feature/activity",
2208 worktreePath: null,
2209 activeProviderThreadId: null,
2210 lineage: {
2211 parentThreadId: null,
2212 relationshipToParent: null,
2213 rootThreadId: threadId,
2214 },
2215 forkedFrom: null,
2216 createdAt,
2217 updatedAt: createdAt,
2218 archivedAt: null,
2219 settledOverride: null,
2220 settledAt: null,
2221 lastVisitedAt: createdAt,
2222 deletedAt: null,
2223 };
2224
2225 yield* projectionStore.apply({
2226 id: EventId.make("event:projection-activity:created"),
2227 type: "thread.created",
2228 threadId,
2229 occurredAt: createdAt,
2230 payload: thread,
2231 });
2232 const created = (yield* projectionStore.getThreadProjection(threadId)).thread;
2233
2234 const { statements, tracer } = traceSqlStatements();
2235 yield* projectionStore
2236 .apply({
2237 id: EventId.make("event:projection-activity:run-created"),
2238 type: "run.created",
2239 threadId,
2240 runId,
2241 nodeId: rootNodeId,
2242 driver,
2243 providerInstanceId,
2244 occurredAt: runOccurredAt,
2245 payload: {
2246 id: runId,
2247 threadId,
2248 ordinal: 1,
2249 providerInstanceId,
2250 modelSelection,
2251 providerThreadId: null,
2252 userMessageId: MessageId.make("message:projection-activity"),
2253 rootNodeId,
2254 activeAttemptId: null,
2255 status: "running" as const,
2256 requestedAt: runOccurredAt,
2257 startedAt: runOccurredAt,
2258 completedAt: null,
2259 checkpointId: null,
2260 contextHandoffId: null,
2261 },
2262 })
2263 .pipe(Effect.withTracer(tracer));
2264 // The thread row is patched in place, never read back to be re-encoded.
2265 assert.isEmpty(
2266 statements.filter(
2267 (statement) =>
2268 /^\s*SELECT/i.test(statement) &&
2269 statement.includes("orchestration_v2_projection_threads"),
2270 ),
2271 );
2272
2273 const active = (yield* projectionStore.getThreadProjection(threadId)).thread;
2274 assert.deepEqual(active, { ...created, updatedAt: runOccurredAt });
2275 const shell = (yield* projectionStore.getShellSnapshot()).threads.find(
2276 (candidate) => candidate.id === threadId,
2277 );
2278 assert.deepEqual(shell?.updatedAt, runOccurredAt);
2279 }),
2280 );
2281
21892282 it.effect("preserves delegated completion ownership across stale run and task updates", () =>
21902283 Effect.gen(function* () {
21912284 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
apps/server/src/orchestration-v2/ProjectionStore.ts+12-17
......@@ -1037,6 +1037,9 @@ const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema
10371037const encodeThreadPayload = Schema.encodeEffect(
10381038 Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema),
10391039);
1040const encodeThreadUpdatedAt = Schema.encodeEffect(
1041 OrchestrationV2AppThreadJsonSchema.fields.updatedAt,
1042);
10401043const encodeRunPayload = Schema.encodeEffect(Schema.fromJsonString(OrchestrationV2RunJsonSchema));
10411044const encodeRestartCancelledBackgroundWork = Schema.encodeEffect(
10421045 Schema.fromJsonString(OrchestrationV2RunJsonSchema.fields.restartCancelledBackgroundWork),
......@@ -2690,25 +2693,17 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
26902693 event.type !== "thread.model-selection-updated" &&
26912694 event.type !== "thread.provider-switched"
26922695 ) {
2693 const rows = yield* sql<PayloadRow>`
2694 SELECT payload_json
2695 FROM orchestration_v2_projection_threads
2696 // Only the activity timestamp changes. This runs for every streamed
2697 // run, node and turn-item event, so the thread row is patched in place
2698 // instead of being read, decoded and re-encoded each time.
2699 const updatedAt = yield* encodeThreadUpdatedAt(event.occurredAt);
2700 yield* sql`
2701 UPDATE orchestration_v2_projection_threads
2702 SET
2703 updated_at = ${updatedAt},
2704 payload_json = json_set(payload_json, '$.updatedAt', ${updatedAt})
26962705 WHERE thread_id = ${event.threadId}
2697 LIMIT 1
26982706 `;
2699 const row = rows[0];
2700 if (row !== undefined) {
2701 const thread = yield* decodeThreadPayload(row.payload_json);
2702 const updatedThread = { ...thread, updatedAt: event.occurredAt };
2703 const payloadJson = yield* encodeThreadPayload(updatedThread);
2704 yield* sql`
2705 UPDATE orchestration_v2_projection_threads
2706 SET
2707 updated_at = ${stringField(parseEncodedPayload(payloadJson), "updatedAt")},
2708 payload_json = ${payloadJson}
2709 WHERE thread_id = ${event.threadId}
2710 `;
2711 }
27122707 }
27132708 }).pipe(
27142709 Effect.mapError(