From 3b2ebcee2f6eeb4ed3dde6c89fc59f2eb5bd996d Mon Sep 17 00:00:00 2001 From: Hessboi <101525047+Baker-Harrison@users.noreply.github.com> Date: Wed, 14 Oct 2026 02:17:50 -0500 Subject: [PATCH] perf(server): streamed events no longer rewrite the whole thread row (#18890) Co-authored-by: Claude Opus 5.5 --- .../orchestration-v2/ProjectionStore.test.ts | 93 +++++++++++++++++++ .../src/orchestration-v2/ProjectionStore.ts | 29 +++--- 2 files changed, 105 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 066ff7def15dda2529ea2a7da59291a94bbc4227..dcd08d8a2586995a95b1ddeb1c2f16848f187ed6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2186,6 +2186,99 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); + it.effect("dates thread activity by a run event and leaves the rest of the thread alone", () => + Effect.gen(function* () { + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const createdAt = yield* DateTime.now; + const runOccurredAt = DateTime.add(createdAt, { seconds: 5 }); + const threadId = ThreadId.make("thread:projection-activity"); + const runId = RunId.make("run:projection-activity"); + const rootNodeId = NodeId.make("node:projection-activity"); + const thread = { + createdBy: "user" as const, + creationSource: "web" as const, + id: threadId, + projectId: ProjectId.make("project:projection-activity"), + title: "Projection activity", + providerInstanceId, + modelSelection, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + branch: "feature/activity", + worktreePath: null, + activeProviderThreadId: null, + lineage: { + parentThreadId: null, + relationshipToParent: null, + rootThreadId: threadId, + }, + forkedFrom: null, + createdAt, + updatedAt: createdAt, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: createdAt, + deletedAt: null, + }; + + yield* projectionStore.apply({ + id: EventId.make("event:projection-activity:created"), + type: "thread.created", + threadId, + occurredAt: createdAt, + payload: thread, + }); + const created = (yield* projectionStore.getThreadProjection(threadId)).thread; + + const { statements, tracer } = traceSqlStatements(); + yield* projectionStore + .apply({ + id: EventId.make("event:projection-activity:run-created"), + type: "run.created", + threadId, + runId, + nodeId: rootNodeId, + driver, + providerInstanceId, + occurredAt: runOccurredAt, + payload: { + id: runId, + threadId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:projection-activity"), + rootNodeId, + activeAttemptId: null, + status: "running" as const, + requestedAt: runOccurredAt, + startedAt: runOccurredAt, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }) + .pipe(Effect.withTracer(tracer)); + // The thread row is patched in place, never read back to be re-encoded. + assert.isEmpty( + statements.filter( + (statement) => + /^\s*SELECT/i.test(statement) && + statement.includes("orchestration_v2_projection_threads"), + ), + ); + + const active = (yield* projectionStore.getThreadProjection(threadId)).thread; + assert.deepEqual(active, { ...created, updatedAt: runOccurredAt }); + const shell = (yield* projectionStore.getShellSnapshot()).threads.find( + (candidate) => candidate.id === threadId, + ); + assert.deepEqual(shell?.updatedAt, runOccurredAt); + }), + ); + it.effect("preserves delegated completion ownership across stale run and task updates", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b999f6376ecb3fdc539dc24de46653adaa28d6bd..605d1b01b6f5e6ec41140122d46d0af38d6587e5 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1037,6 +1037,9 @@ const encodeIdList = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema const encodeThreadPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema), ); +const encodeThreadUpdatedAt = Schema.encodeEffect( + OrchestrationV2AppThreadJsonSchema.fields.updatedAt, +); const encodeRunPayload = Schema.encodeEffect(Schema.fromJsonString(OrchestrationV2RunJsonSchema)); const encodeRestartCancelledBackgroundWork = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2RunJsonSchema.fields.restartCancelledBackgroundWork), @@ -2690,25 +2693,17 @@ export const layer: Layer.Layer = event.type !== "thread.model-selection-updated" && event.type !== "thread.provider-switched" ) { - const rows = yield* sql` - SELECT payload_json - FROM orchestration_v2_projection_threads + // Only the activity timestamp changes. This runs for every streamed + // run, node and turn-item event, so the thread row is patched in place + // instead of being read, decoded and re-encoded each time. + const updatedAt = yield* encodeThreadUpdatedAt(event.occurredAt); + yield* sql` + UPDATE orchestration_v2_projection_threads + SET + updated_at = ${updatedAt}, + payload_json = json_set(payload_json, '$.updatedAt', ${updatedAt}) WHERE thread_id = ${event.threadId} - LIMIT 1 `; - const row = rows[0]; - if (row !== undefined) { - const thread = yield* decodeThreadPayload(row.payload_json); - const updatedThread = { ...thread, updatedAt: event.occurredAt }; - const payloadJson = yield* encodeThreadPayload(updatedThread); - yield* sql` - UPDATE orchestration_v2_projection_threads - SET - updated_at = ${stringField(parseEncodedPayload(payloadJson), "updatedAt")}, - payload_json = ${payloadJson} - WHERE thread_id = ${event.threadId} - `; - } } }).pipe( Effect.mapError( -- 2.54.0