authorgravatar for 75276669+Gigioxx@users.noreply.github.comGuillermo Casanova <75276669+Gigioxx@users.noreply.github.com> 2026-10-14 04:07:21-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-14 00:07:21-07:00
log30d1010443d92d7c17f9d773639375a4c6de77bc
tree0176f1e10644cb0e5b7a59871ea5a1d0435fb3a4
parent8b460a3c851d629df896f7420aa32eb32df1daca
signature Signed by PGP key B5690EEEBB952194

fix(server): Claude subagent progress no longer re-stores the unchanged prompt (#18375)

3 files changed, 260 insertions(+), 6 deletions(-)

apps/server/src/orchestration-v2/ProjectionMaintenance.ts+1
...@@ -230,6 +230,7 @@ export const layer: Layer.Layer<...@@ -230,6 +230,7 @@ export const layer: Layer.Layer<
230 ];230 ];
231231
232 const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES);232 const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES);
233 // Keep subagent.updated and turn-item.updated rows: slimmed payloads reference their prompts.
233 const COMPACTION_PAGE_SIZE = 500;234 const COMPACTION_PAGE_SIZE = 500;
234235
235 /**236 /**
apps/server/src/persistence/OrchestrationEventStore.test.ts+195
...@@ -3,7 +3,9 @@ import * as NodeV8 from "node:v8";...@@ -3,7 +3,9 @@ import * as NodeV8 from "node:v8";
3import {3import {
4 CommandId,4 CommandId,
5 EventId,5 EventId,
6 NodeId,
6 ProjectId,7 ProjectId,
8 ProviderDriverKind,
7 ProviderInstanceId,9 ProviderInstanceId,
8 ThreadId,10 ThreadId,
9 TurnItemId,11 TurnItemId,
...@@ -103,6 +105,199 @@ layer("OrchestrationEventStore", (it) => {...@@ -103,6 +105,199 @@ layer("OrchestrationEventStore", (it) => {
103 ),105 ),
104 );106 );
105107
108 it.effect("stores an unchanged subagent prompt once and replays it on every update", () =>
109 Effect.gen(function* () {
110 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
111 const sql = yield* SqlClient.SqlClient;
112 const afterSequence = yield* store.latestApplicationSequence;
113 const now = yield* DateTime.now;
114 const threadId = ThreadId.make("thread:subagent-prompt");
115 const id = NodeId.make("node:subagent-prompt");
116 const driver = ProviderDriverKind.make("claudeAgent");
117 const providerInstanceId = ProviderInstanceId.make("claudeAgent");
118 const prompt = "Audit the repository. ".repeat(200);
119 const subagent = (index: number, progress: string, nextPrompt = prompt) =>
120 ({
121 id: EventId.make(`event:subagent-prompt:subagent:${index}`),
122 type: "subagent.updated",
123 threadId,
124 occurredAt: now,
125 payload: {
126 id,
127 threadId,
128 runId: null,
129 parentNodeId: NodeId.make("node:subagent-prompt-parent"),
130 origin: "provider_native",
131 createdBy: "agent",
132 driver,
133 providerInstanceId,
134 providerThreadId: null,
135 childThreadId: null,
136 nativeTaskRef: { driver, nativeId: "subagent-prompt", strength: "strong" },
137 prompt: nextPrompt,
138 progress,
139 title: "Audit",
140 model: null,
141 status: "running",
142 result: null,
143 startedAt: now,
144 completedAt: null,
145 updatedAt: now,
146 },
147 }) satisfies OrchestrationV2DomainEvent;
148 const item = (index: number, progress: string, nextPrompt = prompt) =>
149 ({
150 id: EventId.make(`event:subagent-prompt:item:${index}`),
151 type: "turn-item.updated",
152 threadId,
153 occurredAt: now,
154 payload: {
155 id: TurnItemId.make("item:subagent-prompt"),
156 type: "subagent",
157 threadId,
158 runId: null,
159 nodeId: null,
160 providerThreadId: null,
161 providerTurnId: null,
162 nativeItemRef: null,
163 parentItemId: null,
164 ordinal: 1,
165 status: "running",
166 title: "Audit",
167 subagentId: id,
168 origin: "provider_native",
169 driver,
170 providerInstanceId,
171 childThreadId: null,
172 prompt: nextPrompt,
173 progress,
174 result: null,
175 startedAt: now,
176 completedAt: null,
177 updatedAt: now,
178 },
179 }) satisfies OrchestrationV2DomainEvent;
180 const events = [
181 subagent(0, "Reading files"),
182 item(0, "Reading files"),
183 subagent(1, "Running tests"),
184 item(1, "Running tests"),
185 subagent(2, "Writing report"),
186 item(2, "Writing report"),
187 subagent(3, "Retrying", "Audit again."),
188 item(3, "Retrying", "Audit again."),
189 ];
190 yield* store.appendAgentEvents({ events });
191
192 const stored = yield* sql<{ readonly payload_json: string }>`
193 SELECT payload_json FROM orchestration_events
194 WHERE stream_id = ${threadId} ORDER BY sequence ASC
195 `;
196 assert.deepEqual(
197 stored.map((row) => row.payload_json.includes(prompt)),
198 [true, true, false, false, false, false, false, false],
199 );
200 const replayed = yield* store
201 .readAgentEvents({ threadId, afterSequence })
202 .pipe(Stream.runCollect);
203 assert.deepEqual(
204 replayed.map((entry) => entry.event),
205 events,
206 );
207 // Replay budgets measure the payloads replay decodes, not the slimmer stored rows.
208 const stats = yield* store.getAgentReplayStats({
209 threadId,
210 afterSequence,
211 throughSequence: yield* store.latestApplicationSequence,
212 maxEvents: events.length,
213 });
214 assert.isAbove(stats.rawPayloadBytes, prompt.length * 6);
215 const applicationStats = yield* store.getReplayStats({
216 afterSequence,
217 throughSequence: yield* store.latestApplicationSequence,
218 });
219 assert.isAbove(applicationStats.rawPayloadBytes, prompt.length * 6);
220 }),
221 );
222
223 it.effect("stores the full subagent prompt when its hint row was rolled back", () =>
224 Effect.gen(function* () {
225 const store = yield* OrchestrationEventStore.OrchestrationEventStore;
226 const sql = yield* SqlClient.SqlClient;
227 const afterSequence = yield* store.latestApplicationSequence;
228 const now = yield* DateTime.now;
229 const threadId = ThreadId.make("thread:subagent-rollback");
230 const event = {
231 id: EventId.make("event:subagent-rollback"),
232 type: "subagent.updated",
233 threadId,
234 occurredAt: now,
235 payload: {
236 id: NodeId.make("node:subagent-rollback"),
237 threadId,
238 runId: null,
239 parentNodeId: NodeId.make("node:subagent-rollback-parent"),
240 origin: "provider_native",
241 createdBy: "agent",
242 driver: ProviderDriverKind.make("claudeAgent"),
243 providerInstanceId: ProviderInstanceId.make("claudeAgent"),
244 providerThreadId: null,
245 childThreadId: null,
246 nativeTaskRef: null,
247 prompt: "Audit the repository. ".repeat(200),
248 progress: "Reading files",
249 title: "Audit",
250 model: null,
251 status: "running",
252 result: null,
253 startedAt: now,
254 completedAt: null,
255 updatedAt: now,
256 },
257 } satisfies OrchestrationV2DomainEvent;
258 const rolledBack = yield* sql
259 .withTransaction(
260 store
261 .appendAgentEvents({ events: [event] })
262 .pipe(Effect.andThen(Effect.fail("rollback"))),
263 )
264 .pipe(Effect.flip);
265 assert.equal(rolledBack, "rollback");
266 const rowsAfterRollback = yield* sql`
267 SELECT sequence FROM orchestration_events WHERE stream_id = ${threadId}
268 `;
269 assert.deepEqual(rowsAfterRollback, []);
270
271 const events = [event, { ...event, id: EventId.make("event:subagent-after-rollback") }];
272 const appended = yield* store.appendAgentEvents({ events });
273 assert.deepEqual(
274 appended.map((entry) => entry.event),
275 events,
276 );
277 const stored = yield* sql<{
278 readonly prompt: string | null;
279 readonly prompt_sequence: number | null;
280 }>`
281 SELECT json_extract(payload_json, '$.prompt') AS prompt,
282 json_extract(payload_json, '$.promptSequence') AS prompt_sequence
283 FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence
284 `;
285 assert.deepEqual(stored, [
286 { prompt: event.payload.prompt, prompt_sequence: null },
287 { prompt: "", prompt_sequence: appended[0]!.sequence },
288 ]);
289 const replayed = yield* store.readAgentEvents({ threadId }).pipe(Stream.runCollect);
290 assert.deepEqual(replayed, appended);
291 const applicationEvents = yield* store
292 .readApplicationEvents({
293 afterSequence,
294 throughSequence: appended[1]!.sequence,
295 })
296 .pipe(Stream.runCollect);
297 assert.deepEqual(applicationEvents, [...appended]);
298 }),
299 );
300
106 it.effect("stores json columns as strings and replays CLI-origin events", () =>301 it.effect("stores json columns as strings and replays CLI-origin events", () =>
107 Effect.gen(function* () {302 Effect.gen(function* () {
108 const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;303 const eventStore = yield* OrchestrationEventStore.OrchestrationEventStore;
apps/server/src/persistence/OrchestrationEventStore.ts+64-6
...@@ -324,6 +324,21 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st...@@ -324,6 +324,21 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st
324const makeEventStore = Effect.gen(function* () {324const makeEventStore = Effect.gen(function* () {
325 const sql = yield* SqlClient.SqlClient;325 const sql = yield* SqlClient.SqlClient;
326 const committedEvents = yield* PubSub.unbounded<ApplicationStoredEvent>();326 const committedEvents = yield* PubSub.unbounded<ApplicationStoredEvent>();
327 // Sequence of the latest row storing each subagent's full prompt. The INSERT re-checks it.
328 const promptHints = new Map<string, number>();
329 // Puts back a subagent prompt that appendAgentEvents stored once, behind `promptSequence`.
330 const restoredPayloadJson = sql`CASE
331 WHEN orchestration_events.event_type IN ('subagent.updated', 'turn-item.updated')
332 AND json_type(orchestration_events.payload_json, '$.promptSequence') = 'integer'
333 THEN json_set(
334 json_remove(orchestration_events.payload_json, '$.promptSequence'),
335 '$.prompt',
336 (SELECT json_extract(anchor.payload_json, '$.prompt')
337 FROM orchestration_events AS anchor
338 WHERE anchor.sequence = json_extract(orchestration_events.payload_json, '$.promptSequence'))
339 )
340 ELSE orchestration_events.payload_json
341 END`;
327342
328 const appendProjectEventRow = SqlSchema.findOne({343 const appendProjectEventRow = SqlSchema.findOne({
329 Request: AppendProjectEventRequestSchema,344 Request: AppendProjectEventRequestSchema,
...@@ -428,7 +443,7 @@ const makeEventStore = Effect.gen(function* () {...@@ -428,7 +443,7 @@ const makeEventStore = Effect.gen(function* () {
428 stream_id,443 stream_id,
429 event_type,444 event_type,
430 occurred_at,445 occurred_at,
431 payload_json,446 ${restoredPayloadJson} AS payload_json,
432 metadata_json,447 metadata_json,
433 application_event_version,448 application_event_version,
434 causation_event_id,449 causation_event_id,
...@@ -468,7 +483,33 @@ const makeEventStore = Effect.gen(function* () {...@@ -468,7 +483,33 @@ const makeEventStore = Effect.gen(function* () {
468 (event) =>483 (event) =>
469 Effect.gen(function* () {484 Effect.gen(function* () {
470 const encoded = yield* encodeV2EventJson(event);485 const encoded = yield* encodeV2EventJson(event);
471 const rows = yield* sql<{ readonly sequence: number }>`486 const subagent =
487 encoded.type === "subagent.updated"
488 ? encoded.payload
489 : encoded.type === "turn-item.updated" && encoded.payload.type === "subagent"
490 ? encoded.payload
491 : undefined;
492 const promptKey = subagent === undefined ? undefined : `${event.type}:${subagent.id}`;
493 const hint = promptKey === undefined ? undefined : promptHints.get(promptKey);
494 const payloadJson = yield* encodeJson(encoded.payload);
495 // Progress ticks repeat an unchanged prompt, so store it once and point
496 // later rows at it. Keep an empty prompt so older servers still decode
497 // the row. ProjectionMaintenance never deletes these anchor rows. The
498 // hint is re-checked here because its transaction may have rolled back.
499 const storedPayload =
500 hint !== undefined && subagent !== undefined
501 ? sql`CASE WHEN EXISTS (
502 SELECT 1 FROM orchestration_events
503 WHERE sequence = ${hint}
504 AND json_type(payload_json, '$.promptSequence') IS NULL
505 AND json_extract(payload_json, '$.prompt') = ${subagent.prompt}
506 ) THEN json_set(${payloadJson}, '$.prompt', '', '$.promptSequence', CAST(${hint} AS INTEGER))
507 ELSE ${payloadJson} END`
508 : sql`${payloadJson}`;
509 const rows = yield* sql<{
510 readonly sequence: number;
511 readonly stores_full_prompt?: number;
512 }>`
472 INSERT INTO orchestration_events (513 INSERT INTO orchestration_events (
473 event_id,514 event_id,
474 aggregate_kind,515 aggregate_kind,
...@@ -502,12 +543,29 @@ const makeEventStore = Effect.gen(function* () {...@@ -502,12 +543,29 @@ const makeEventStore = Effect.gen(function* () {
502 NULL,543 NULL,
503 ${input.commandId ?? null},544 ${input.commandId ?? null},
504 ${event.rawEventId === undefined ? "server" : "provider"},545 ${event.rawEventId === undefined ? "server" : "provider"},
505 ${yield* encodeJson(encoded.payload)},546 ${storedPayload},
506 ${yield* encodeJson(metadataForV2Event(event))},547 ${yield* encodeJson(metadataForV2Event(event))},
507 2548 2
508 )549 )
509 RETURNING sequence550 RETURNING sequence${
551 hint === undefined
552 ? sql``
553 : sql`, json_type(payload_json, '$.promptSequence') IS NULL AS stores_full_prompt`
554 }
510 `;555 `;
556 const row = rows[0];
557 if (promptKey !== undefined && row !== undefined) {
558 // Delete first so Map order tracks recent use and eviction drops idle subagents.
559 promptHints.delete(promptKey);
560 promptHints.set(
561 promptKey,
562 hint === undefined || row.stores_full_prompt === 1 ? row.sequence : hint,
563 );
564 if (promptHints.size > 4096) {
565 const oldest = promptHints.keys().next().value;
566 if (oldest !== undefined) promptHints.delete(oldest);
567 }
568 }
511 return yield* decodeV2StoredEvent({569 return yield* decodeV2StoredEvent({
512 sequence: rows[0]?.sequence,570 sequence: rows[0]?.sequence,
513 commandId: input.commandId ?? null,571 commandId: input.commandId ?? null,
...@@ -584,7 +642,7 @@ const makeEventStore = Effect.gen(function* () {...@@ -584,7 +642,7 @@ const makeEventStore = Effect.gen(function* () {
584 COALESCE(SUM(octet_length(payload_json)), 0) AS "rawPayloadBytes",642 COALESCE(SUM(octet_length(payload_json)), 0) AS "rawPayloadBytes",
585 COALESCE(MAX(event_type = 'thread.created'), 0) AS "hasCreateEvent"643 COALESCE(MAX(event_type = 'thread.created'), 0) AS "hasCreateEvent"
586 FROM (644 FROM (
587 SELECT payload_json, event_type645 SELECT ${restoredPayloadJson} AS payload_json, event_type
588 FROM orchestration_events646 FROM orchestration_events
589 WHERE aggregate_kind = 'thread'647 WHERE aggregate_kind = 'thread'
590 AND stream_id = ${input.threadId}648 AND stream_id = ${input.threadId}
...@@ -607,7 +665,7 @@ const makeEventStore = Effect.gen(function* () {...@@ -607,7 +665,7 @@ const makeEventStore = Effect.gen(function* () {
607 sql<{ readonly eventCount: number; readonly rawPayloadBytes: number }>`665 sql<{ readonly eventCount: number; readonly rawPayloadBytes: number }>`
608 SELECT666 SELECT
609 COUNT(*) AS "eventCount",667 COUNT(*) AS "eventCount",
610 COALESCE(SUM(octet_length(payload_json)), 0) AS "rawPayloadBytes"668 COALESCE(SUM(octet_length(${restoredPayloadJson})), 0) AS "rawPayloadBytes"
611 FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water669 FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water
612 WHERE sequence > ${input.afterSequence}670 WHERE sequence > ${input.afterSequence}
613 AND sequence <= ${input.throughSequence}671 AND sequence <= ${input.throughSequence}