diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 8c10de3142028ef23ea92a8e7338950148f874a1..ccd7e383593310cca3c05df9e0be7f031798c2ab 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -230,6 +230,7 @@ export const layer: Layer.Layer< ]; const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES); + // Keep subagent.updated and turn-item.updated rows: slimmed payloads reference their prompts. const COMPACTION_PAGE_SIZE = 500; /** diff --git a/apps/server/src/persistence/OrchestrationEventStore.test.ts b/apps/server/src/persistence/OrchestrationEventStore.test.ts index 6a075e19c46eed02018c4e4e3e1e3d4e150409f7..acfd95ed98475200868b775b2bbfe3fe26bd9585 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.test.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.test.ts @@ -3,7 +3,9 @@ import * as NodeV8 from "node:v8"; import { CommandId, EventId, + NodeId, ProjectId, + ProviderDriverKind, ProviderInstanceId, ThreadId, TurnItemId, @@ -103,6 +105,199 @@ layer("OrchestrationEventStore", (it) => { ), ); + it.effect("stores an unchanged subagent prompt once and replays it on every update", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const afterSequence = yield* store.latestApplicationSequence; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:subagent-prompt"); + const id = NodeId.make("node:subagent-prompt"); + const driver = ProviderDriverKind.make("claudeAgent"); + const providerInstanceId = ProviderInstanceId.make("claudeAgent"); + const prompt = "Audit the repository. ".repeat(200); + const subagent = (index: number, progress: string, nextPrompt = prompt) => + ({ + id: EventId.make(`event:subagent-prompt:subagent:${index}`), + type: "subagent.updated", + threadId, + occurredAt: now, + payload: { + id, + threadId, + runId: null, + parentNodeId: NodeId.make("node:subagent-prompt-parent"), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId: null, + childThreadId: null, + nativeTaskRef: { driver, nativeId: "subagent-prompt", strength: "strong" }, + prompt: nextPrompt, + progress, + title: "Audit", + model: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }) satisfies OrchestrationV2DomainEvent; + const item = (index: number, progress: string, nextPrompt = prompt) => + ({ + id: EventId.make(`event:subagent-prompt:item:${index}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make("item:subagent-prompt"), + type: "subagent", + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "running", + title: "Audit", + subagentId: id, + origin: "provider_native", + driver, + providerInstanceId, + childThreadId: null, + prompt: nextPrompt, + progress, + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }) satisfies OrchestrationV2DomainEvent; + const events = [ + subagent(0, "Reading files"), + item(0, "Reading files"), + subagent(1, "Running tests"), + item(1, "Running tests"), + subagent(2, "Writing report"), + item(2, "Writing report"), + subagent(3, "Retrying", "Audit again."), + item(3, "Retrying", "Audit again."), + ]; + yield* store.appendAgentEvents({ events }); + + const stored = yield* sql<{ readonly payload_json: string }>` + SELECT payload_json FROM orchestration_events + WHERE stream_id = ${threadId} ORDER BY sequence ASC + `; + assert.deepEqual( + stored.map((row) => row.payload_json.includes(prompt)), + [true, true, false, false, false, false, false, false], + ); + const replayed = yield* store + .readAgentEvents({ threadId, afterSequence }) + .pipe(Stream.runCollect); + assert.deepEqual( + replayed.map((entry) => entry.event), + events, + ); + // Replay budgets measure the payloads replay decodes, not the slimmer stored rows. + const stats = yield* store.getAgentReplayStats({ + threadId, + afterSequence, + throughSequence: yield* store.latestApplicationSequence, + maxEvents: events.length, + }); + assert.isAbove(stats.rawPayloadBytes, prompt.length * 6); + const applicationStats = yield* store.getReplayStats({ + afterSequence, + throughSequence: yield* store.latestApplicationSequence, + }); + assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6); + }), + ); + + it.effect("stores the full subagent prompt when its hint row was rolled back", () => + Effect.gen(function* () { + const store = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const afterSequence = yield* store.latestApplicationSequence; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:subagent-rollback"); + const event = { + id: EventId.make("event:subagent-rollback"), + type: "subagent.updated", + threadId, + occurredAt: now, + payload: { + id: NodeId.make("node:subagent-rollback"), + threadId, + runId: null, + parentNodeId: NodeId.make("node:subagent-rollback-parent"), + origin: "provider_native", + createdBy: "agent", + driver: ProviderDriverKind.make("claudeAgent"), + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + providerThreadId: null, + childThreadId: null, + nativeTaskRef: null, + prompt: "Audit the repository. ".repeat(200), + progress: "Reading files", + title: "Audit", + model: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + } satisfies OrchestrationV2DomainEvent; + const rolledBack = yield* sql + .withTransaction( + store + .appendAgentEvents({ events: [event] }) + .pipe(Effect.andThen(Effect.fail("rollback"))), + ) + .pipe(Effect.flip); + assert.equal(rolledBack, "rollback"); + const rowsAfterRollback = yield* sql` + SELECT sequence FROM orchestration_events WHERE stream_id = ${threadId} + `; + assert.deepEqual(rowsAfterRollback, []); + + const events = [event, { ...event, id: EventId.make("event:subagent-after-rollback") }]; + const appended = yield* store.appendAgentEvents({ events }); + assert.deepEqual( + appended.map((entry) => entry.event), + events, + ); + const stored = yield* sql<{ + readonly prompt: string | null; + readonly prompt_sequence: number | null; + }>` + SELECT json_extract(payload_json, '$.prompt') AS prompt, + json_extract(payload_json, '$.promptSequence') AS prompt_sequence + FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence + `; + assert.deepEqual(stored, [ + { prompt: event.payload.prompt, prompt_sequence: null }, + { prompt: "", prompt_sequence: appended[0]!.sequence }, + ]); + const replayed = yield* store.readAgentEvents({ threadId }).pipe(Stream.runCollect); + assert.deepEqual(replayed, appended); + const applicationEvents = yield* store + .readApplicationEvents({ + afterSequence, + throughSequence: appended[1]!.sequence, + }) + .pipe(Stream.runCollect); + assert.deepEqual(applicationEvents, [...appended]); + }), + ); + it.effect("stores json columns as strings and replays CLI-origin events", () => Effect.gen(function* () { const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore; diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 16cb8208e66ccf37d23228bcbfbc00d9a91a2bb3..45b44808b4f0d2f6dfc39398d14f3d5ccd10c0e2 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -324,6 +324,21 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st const makeEventStore = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const committedEvents = yield* PubSub.unbounded(); + // Sequence of the latest row storing each subagent's full prompt. The INSERT re-checks it. + const promptHints = new Map(); + // Puts back a subagent prompt that appendAgentEvents stored once, behind `promptSequence`. + const restoredPayloadJson = sql`CASE + WHEN orchestration_events.event_type IN ('subagent.updated', 'turn-item.updated') + AND json_type(orchestration_events.payload_json, '$.promptSequence') = 'integer' + THEN json_set( + json_remove(orchestration_events.payload_json, '$.promptSequence'), + '$.prompt', + (SELECT json_extract(anchor.payload_json, '$.prompt') + FROM orchestration_events AS anchor + WHERE anchor.sequence = json_extract(orchestration_events.payload_json, '$.promptSequence')) + ) + ELSE orchestration_events.payload_json + END`; const appendProjectEventRow = SqlSchema.findOne({ Request: AppendProjectEventRequestSchema, @@ -428,7 +443,7 @@ const makeEventStore = Effect.gen(function* () { stream_id, event_type, occurred_at, - payload_json, + ${restoredPayloadJson} AS payload_json, metadata_json, application_event_version, causation_event_id, @@ -468,7 +483,33 @@ const makeEventStore = Effect.gen(function* () { (event) => Effect.gen(function* () { const encoded = yield* encodeV2EventJson(event); - const rows = yield* sql<{ readonly sequence: number }>` + const subagent = + encoded.type === "subagent.updated" + ? encoded.payload + : encoded.type === "turn-item.updated" && encoded.payload.type === "subagent" + ? encoded.payload + : undefined; + const promptKey = subagent === undefined ? undefined : `${event.type}:${subagent.id}`; + const hint = promptKey === undefined ? undefined : promptHints.get(promptKey); + const payloadJson = yield* encodeJson(encoded.payload); + // Progress ticks repeat an unchanged prompt, so store it once and point + // later rows at it. Keep an empty prompt so older servers still decode + // the row. ProjectionMaintenance never deletes these anchor rows. The + // hint is re-checked here because its transaction may have rolled back. + const storedPayload = + hint !== undefined && subagent !== undefined + ? sql`CASE WHEN EXISTS ( + SELECT 1 FROM orchestration_events + WHERE sequence = ${hint} + AND json_type(payload_json, '$.promptSequence') IS NULL + AND json_extract(payload_json, '$.prompt') = ${subagent.prompt} + ) THEN json_set(${payloadJson}, '$.prompt', '', '$.promptSequence', CAST(${hint} AS INTEGER)) + ELSE ${payloadJson} END` + : sql`${payloadJson}`; + const rows = yield* sql<{ + readonly sequence: number; + readonly stores_full_prompt?: number; + }>` INSERT INTO orchestration_events ( event_id, aggregate_kind, @@ -502,12 +543,29 @@ const makeEventStore = Effect.gen(function* () { NULL, ${input.commandId ?? null}, ${event.rawEventId === undefined ? "server" : "provider"}, - ${yield* encodeJson(encoded.payload)}, + ${storedPayload}, ${yield* encodeJson(metadataForV2Event(event))}, 2 ) - RETURNING sequence + RETURNING sequence${ + hint === undefined + ? sql`` + : sql`, json_type(payload_json, '$.promptSequence') IS NULL AS stores_full_prompt` + } `; + const row = rows[0]; + if (promptKey !== undefined && row !== undefined) { + // Delete first so Map order tracks recent use and eviction drops idle subagents. + promptHints.delete(promptKey); + promptHints.set( + promptKey, + hint === undefined || row.stores_full_prompt === 1 ? row.sequence : hint, + ); + if (promptHints.size > 4096) { + const oldest = promptHints.keys().next().value; + if (oldest !== undefined) promptHints.delete(oldest); + } + } return yield* decodeV2StoredEvent({ sequence: rows[0]?.sequence, commandId: input.commandId ?? null, @@ -584,7 +642,7 @@ const makeEventStore = Effect.gen(function* () { COALESCE(SUM(octet_length(payload_json)), 0) AS "rawPayloadBytes", COALESCE(MAX(event_type = 'thread.created'), 0) AS "hasCreateEvent" FROM ( - SELECT payload_json, event_type + SELECT ${restoredPayloadJson} AS payload_json, event_type FROM orchestration_events WHERE aggregate_kind = 'thread' AND stream_id = ${input.threadId} @@ -607,7 +665,7 @@ const makeEventStore = Effect.gen(function* () { sql<{ readonly eventCount: number; readonly rawPayloadBytes: number }>` SELECT COUNT(*) AS "eventCount", - COALESCE(SUM(octet_length(payload_json)), 0) AS "rawPayloadBytes" + COALESCE(SUM(octet_length(${restoredPayloadJson})), 0) AS "rawPayloadBytes" FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water WHERE sequence > ${input.afterSequence} AND sequence <= ${input.throughSequence}