History

perf(orchestration): per-thread shell deltas, visit throttling, event compaction (#4971)

Julius Marminge committed 3 months ago

Commit b76a0911 parent 413089317 files · +875 −188Browse files

Unified · Split

17 files
  1. apps/server/src/orchestration-v2/FoundationPersistence.test.ts +139 −0
  2. apps/server/src/orchestration-v2/LegacyV1ThreadImporter.ts +23 −10
  3. apps/server/src/orchestration-v2/Orchestrator.ts +15 −0
  4. apps/server/src/orchestration-v2/ProjectionMaintenance.ts +170 −0
  5. apps/server/src/orchestration-v2/ProjectionStore.ts +187 −73
  6. apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +3 −4
  7. apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +1 −0
  8. apps/server/src/orchestration-v2/ShellStream.test.ts +132 −1
  9. apps/server/src/orchestration-v2/ShellStream.ts +27 −27
  10. apps/server/src/orchestration-v2/ThreadManagementService.test.ts +19 −0
  11. apps/server/src/orchestration-v2/ThreadManagementService.ts +31 −12
  12. apps/server/src/orchestration-v2/runtimeLayer.test.ts +7 −4
  13. apps/server/src/serverRuntimeStartup.ts +24 −0
  14. apps/server/src/ws.ts +31 −50
  15. apps/web/src/components/ChatView.tsx +38 −7
  16. apps/web/src/components/Sidebar.logic.test.ts +25 −0
  17. apps/web/src/components/Sidebar.logic.ts +3 −0
apps/server/src/orchestration-v2/FoundationPersistence.test.ts+139−0
@@ -265,6 +265,145 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { · 264 unmodified lines
265 }),265 }),
266 );266 );
267 267
268 it.effect("compacts superseded state events, imported v1 events, and legacy receipts", () =>
269 Effect.gen(function* () {
270 const eventSink = yield* EventSinkV2;
271 const maintenance = yield* ProjectionMaintenanceV2;
272 const sql = yield* SqlClient.SqlClient;
273 const now = yield* DateTime.now;
274 const nowIso = DateTime.formatIso(now);
275 const threadId = ThreadId.make("thread:foundation-compact");
276 const thread = makeThread(threadId, now);
277 const messageId = MessageId.make("message:foundation-compact");
278 const threadStateEvent = (
279 suffix: string,
280 type: "thread.visited" | "thread.metadata-updated",
281 ): OrchestrationV2DomainEvent => ({
282 id: EventId.make(`event:foundation-compact:${suffix}`),
283 type,
284 threadId,
285 providerInstanceId,
286 occurredAt: now,
287 payload: { ...thread, lastVisitedAt: now },
288 });
289 const messageEvent = (suffix: string, text: string): OrchestrationV2DomainEvent => ({
290 id: EventId.make(`event:foundation-compact:${suffix}`),
291 type: "message.updated",
292 threadId,
293 occurredAt: now,
294 payload: {
295 createdBy: "user",
296 creationSource: "web",
297 id: messageId,
298 threadId,
299 runId: null,
300 nodeId: null,
301 role: "user",
302 text,
303 attachments: [],
304 streaming: false,
305 createdAt: now,
306 updatedAt: now,
307 },
308 });
309
310 yield* eventSink.write({
311 events: [
312 threadCreatedEvent({ id: "event:foundation-compact:create", thread, now }),
313 threadStateEvent("meta", "thread.metadata-updated"),
314 threadStateEvent("visit-1", "thread.visited"),
315 threadStateEvent("visit-2", "thread.visited"),
316 messageEvent("message-1", "streaming"),
317 messageEvent("message-2", "final"),
318 ],
319 });
320
321 // A fully imported legacy thread: its v1 events and pre-migration
322 // receipts are dead weight; a still-pending import keeps its rows.
323 const importedV1ThreadId = "thread:foundation-compact-v1-imported";
324 const pendingV1ThreadId = "thread:foundation-compact-v1-pending";
325 const insertV1Event = (threadIdValue: string, version: number) => sql`
326 INSERT INTO orchestration_events (
327 event_id, aggregate_kind, stream_id, stream_version, event_type,
328 occurred_at, actor_kind, payload_json, metadata_json, application_event_version
329 )
330 VALUES (
331 ${`event:v1:${threadIdValue}:${version}`}, 'thread', ${threadIdValue}, ${version},
332 'thread.message-appended', ${nowIso}, 'user', '{}', '{}', 1
333 )
334 `;
335 yield* insertV1Event(importedV1ThreadId, 1);
336 yield* insertV1Event(importedV1ThreadId, 2);
337 yield* insertV1Event(pendingV1ThreadId, 1);
338 yield* sql`
339 INSERT INTO orchestration_v2_legacy_imports (
340 thread_id, source_updated_at, shell_imported_at, transcript_imported_at,
341 imported_message_count, last_error
342 )
343 VALUES
344 (${importedV1ThreadId}, ${nowIso}, ${nowIso}, ${nowIso}, 2, NULL),
345 (${pendingV1ThreadId}, ${nowIso}, ${nowIso}, NULL, 0, NULL)
346 ON CONFLICT(thread_id) DO NOTHING
347 `;
348 yield* sql`
349 INSERT INTO orchestration_command_receipts (
350 command_id, aggregate_kind, aggregate_id, accepted_at, result_sequence, status, command_type
351 )
352 VALUES
353 ('command:foundation-compact:legacy', 'thread', ${importedV1ThreadId}, ${nowIso}, 1, 'accepted', 'legacy'),
354 ('command:foundation-compact:pending', 'thread', ${pendingV1ThreadId}, ${nowIso}, 1, 'accepted', 'legacy')
355 ON CONFLICT(command_id) DO NOTHING
356 `;
357
358 const summary = yield* maintenance.compactEventStore;
359 // meta + visit-1 (superseded thread state), message-1, and the two
360 // imported v1 events.
361 assert.isAtLeast(summary.deletedEventCount, 5);
362 assert.isAtLeast(summary.deletedReceiptCount, 1);
363
364 const remaining = yield* sql<{ readonly event_id: string }>`
365 SELECT event_id
366 FROM orchestration_events
367 WHERE aggregate_kind = 'thread'
368 AND stream_id = ${threadId}
369 ORDER BY sequence ASC
370 `;
371 assert.deepEqual(
372 remaining.map((row) => row.event_id),
373 [
374 "event:foundation-compact:create",
375 "event:foundation-compact:visit-2",
376 "event:foundation-compact:message-2",
377 ],
378 );
379
380 const remainingV1 = yield* sql<{ readonly stream_id: string }>`
381 SELECT stream_id
382 FROM orchestration_events
383 WHERE application_event_version = 1
384 AND stream_id IN (${importedV1ThreadId}, ${pendingV1ThreadId})
385 `;
386 assert.deepEqual(
387 remainingV1.map((row) => row.stream_id),
388 [pendingV1ThreadId],
389 );
390
391 const remainingReceipts = yield* sql<{ readonly command_id: string }>`
392 SELECT command_id
393 FROM orchestration_command_receipts
394 WHERE command_id IN ('command:foundation-compact:legacy', 'command:foundation-compact:pending')
395 `;
396 assert.deepEqual(
397 remainingReceipts.map((row) => row.command_id),
398 ["command:foundation-compact:pending"],
399 );
400
401 // Replay across the deletion gaps must still produce a valid projection.
402 assert.isTrue((yield* maintenance.verify).valid);
403 assert.isTrue((yield* maintenance.rebuild).valid);
404 }),
405 );
406
268 it.effect("verifies and rebuilds projections with cross-thread subagent relations", () =>407 it.effect("verifies and rebuilds projections with cross-thread subagent relations", () =>
269 Effect.gen(function* () {408 Effect.gen(function* () {
270 const eventSink = yield* EventSinkV2;409 const eventSink = yield* EventSinkV2;
apps/server/src/orchestration-v2/LegacyV1ThreadImporter.ts+23−10
@@ -488,6 +488,13 @@ const make = Effect.gen(function* () { · 487 unmodified lines
488 ),488 ),
489 );489 );
490 490
491 // Threads whose transcript import this process has already confirmed.
492 // `transcript_imported_at` is never reset to NULL, so a positive answer
493 // stays valid for the process lifetime; ensureTranscript runs on most
494 // thread reads and command dispatches, so skipping the lock + lookup here
495 // keeps that path off the database entirely after first confirmation.
496 const confirmedTranscriptThreadIds = new Set<ThreadId>();
497
491 const ensureTranscriptBase = (threadId: ThreadId) =>498 const ensureTranscriptBase = (threadId: ThreadId) =>
492 transcriptImports.withLock(499 transcriptImports.withLock(
493 threadId,500 threadId,
@@ -500,6 +507,9 @@ const make = Effect.gen(function* () { · 6 unmodified lines
500 `;507 `;
501 const imported = imports[0];508 const imported = imports[0];
502 if (imported === undefined || imported.transcript_imported_at !== null) {509 if (imported === undefined || imported.transcript_imported_at !== null) {
510 if (imported !== undefined) {
511 confirmedTranscriptThreadIds.add(threadId);
512 }
503 return { importedThreadCount: 0, importedMessageCount: 0 };513 return { importedThreadCount: 0, importedMessageCount: 0 };
504 }514 }
505 const messages = yield* listMessages(threadId);515 const messages = yield* listMessages(threadId);
@@ -546,6 +556,7 @@ const make = Effect.gen(function* () { · 40 unmodified lines
546 last_error = NULL556 last_error = NULL
547 WHERE thread_id = ${threadId}557 WHERE thread_id = ${threadId}
548 `;558 `;
559 confirmedTranscriptThreadIds.add(threadId);
549 return {560 return {
550 importedThreadCount: 1,561 importedThreadCount: 1,
551 importedMessageCount: missing.length,562 importedMessageCount: missing.length,
@@ -554,16 +565,18 @@ const make = Effect.gen(function* () { · 2 unmodified lines
554 );565 );
555 566
556 const ensureTranscript = (threadId: ThreadId) =>567 const ensureTranscript = (threadId: ThreadId) =>
557 ensureTranscriptBase(threadId).pipe(568 confirmedTranscriptThreadIds.has(threadId)
558 Effect.mapError(569 ? Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 })
559 (cause) =>570 : ensureTranscriptBase(threadId).pipe(
560 new LegacyV1ThreadImportError({571 Effect.mapError(
561 operation: "hydrate transcript for",572 (cause) =>
562 threadId,573 new LegacyV1ThreadImportError({
563 cause,574 operation: "hydrate transcript for",
564 }),575 threadId,
565 ),576 cause,
566 );577 }),
578 ),
579 );
567 580
568 const importPendingTranscripts = Effect.gen(function* () {581 const importPendingTranscripts = Effect.gen(function* () {
569 const rows = yield* sql<LegacyImportRow>`582 const rows = yield* sql<LegacyImportRow>`
apps/server/src/orchestration-v2/Orchestrator.ts+15−0
@@ -15,6 +15,7 @@ import { · 14 unmodified lines
15 type OrchestrationV2ProviderTurn,15 type OrchestrationV2ProviderTurn,
16 type OrchestrationV2Run,16 type OrchestrationV2Run,
17 type OrchestrationV2RunAttempt,17 type OrchestrationV2RunAttempt,
18 type OrchestrationV2ThreadShell,
18 type OrchestrationV2ThreadShellSnapshot,19 type OrchestrationV2ThreadShellSnapshot,
19 type OrchestrationV2StoredEvent,20 type OrchestrationV2StoredEvent,
20 type OrchestrationV2Subagent,21 type OrchestrationV2Subagent,
@@ -160,6 +161,9 @@ export interface OrchestratorV2Shape { · 139 unmodified lines
160 OrchestrationV2ThreadShellSnapshot,161 OrchestrationV2ThreadShellSnapshot,
161 OrchestratorV2Error162 OrchestratorV2Error
162 >;163 >;
164 readonly getThreadShell: (
165 threadId: ThreadId,
166 ) => Effect.Effect<OrchestrationV2ThreadShell | null, OrchestratorV2Error>;
163 readonly getThreadEventSequence: (167 readonly getThreadEventSequence: (
164 threadId: ThreadId,168 threadId: ThreadId,
165 ) => Effect.Effect<number, OrchestratorV2Error>;169 ) => Effect.Effect<number, OrchestratorV2Error>;
@@ -5933,6 +5937,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio · 5767 unmodified lines
5933 }),5937 }),
5934 ),5938 ),
5935 ),5939 ),
5940 getThreadShell: (threadId) =>
5941 projectionStore
5942 .getThreadShell(threadId)
5943 .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))),
5936 getThreadEventSequence: (threadId) =>5944 getThreadEventSequence: (threadId) =>
5937 eventSink5945 eventSink
5938 .latestSequence({ threadId })5946 .latestSequence({ threadId })
@@ -6022,6 +6030,13 @@ export const layerUnavailable: Layer.Layer<OrchestratorV2> = Layer.succeed( · 83 unmodified lines
6022 cause: "Orchestration V2 live runtime is not configured.",6030 cause: "Orchestration V2 live runtime is not configured.",
6023 }),6031 }),
6024 ),6032 ),
6033 getThreadShell: (threadId) =>
6034 Effect.fail(
6035 new OrchestratorProjectionError({
6036 threadId,
6037 cause: "Orchestration V2 live runtime is not configured.",
6038 }),
6039 ),
6025 getThreadEventSequence: (threadId) =>6040 getThreadEventSequence: (threadId) =>
6026 Effect.fail(6041 Effect.fail(
6027 new OrchestratorProjectionError({6042 new OrchestratorProjectionError({
apps/server/src/orchestration-v2/ProjectionMaintenance.ts+170−0
@@ -34,6 +34,14 @@ export class ProjectionMaintenanceError extends Schema.TaggedErrorClass<Projecti · 33 unmodified lines
34export interface ProjectionMaintenanceV2Shape {34export interface ProjectionMaintenanceV2Shape {
35 readonly verify: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;35 readonly verify: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;
36 readonly rebuild: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;36 readonly rebuild: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;
37 readonly compactEventStore: Effect.Effect<
38 {
39 readonly deletedEventCount: number;
40 readonly deletedReceiptCount: number;
41 readonly reclaimableBytes: number;
42 },
43 ProjectionMaintenanceError
44 >;
37}45}
38 46
39export class ProjectionMaintenanceV2 extends Context.Service<47export class ProjectionMaintenanceV2 extends Context.Service<
@@ -210,9 +218,171 @@ export const layer: Layer.Layer< · 170 unmodified lines
210 Effect.mapError((cause) => new ProjectionMaintenanceError({ operation, cause })),218 Effect.mapError((cause) => new ProjectionMaintenanceError({ operation, cause })),
211 );219 );
212 220
221 // Thread-state events whose payload is the complete thread: only the
222 // newest per thread can influence a replay or an afterSequence catch-up.
223 // thread.created stays out — verify derives the expected thread set from
224 // it, and it anchors replay ordering.
225 const SUPERSEDABLE_THREAD_EVENT_TYPES = [
226 "thread.archived",
227 "thread.unarchived",
228 "thread.deleted",
229 "thread.settled",
230 "thread.unsettled",
231 "thread.snoozed",
232 "thread.unsnoozed",
233 "thread.metadata-updated",
234 "thread.runtime-mode-updated",
235 "thread.interaction-mode-updated",
236 "thread.model-selection-updated",
237 "thread.provider-switched",
238 "thread.visited",
239 "thread.marked-unread",
240 ];
241
242 const chunksOf = <A>(items: ReadonlyArray<A>, size: number): Array<ReadonlyArray<A>> => {
243 const out: Array<ReadonlyArray<A>> = [];
244 for (let index = 0; index < items.length; index += size) {
245 out.push(items.slice(index, index + size));
246 }
247 return out;
248 };
249
250 // The sqlite driver is synchronous: one statement over millions of rows
251 // would pin the event loop for its whole duration. Small auto-committed
252 // batches with a yield between them keep the server responsive and bound
253 // WAL growth; each batch is independently durable, so a crash mid-way
254 // just leaves less garbage for the next run.
255 const COMPACTION_DELETE_BATCH_SIZE = 2_000;
256 const deleteRowsBatched = (input: {
257 readonly table: "orchestration_events" | "orchestration_command_receipts";
258 readonly keys: ReadonlyArray<number | string>;
259 }) =>
260 Effect.gen(function* () {
261 let deleted = 0;
262 for (const batch of chunksOf(input.keys, COMPACTION_DELETE_BATCH_SIZE)) {
263 if (input.table === "orchestration_events") {
264 yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(batch)}`;
265 } else {
266 yield* sql`DELETE FROM orchestration_command_receipts WHERE command_id IN ${sql.in(batch)}`;
267 }
268 deleted += batch.length;
269 yield* Effect.yieldNow;
270 }
271 return deleted;
272 });
273
274 /**
275 * Full event-store compaction. Projections are the working state (startup
276 * verifies rather than replays), so events only serve afterSequence
277 * catch-up and disaster-recovery rebuild — and for full-payload "state of
278 * the entity" events, anything but the newest per entity is dead weight
279 * in both. Three passes:
280 *
281 * 1. Superseded thread-state events (visits alone accumulate at multiple
282 * per minute while a thread is open).
283 * 2. Superseded message.updated / node.updated events per entity id —
284 * streaming rewrites the same message many times. turn-item.updated is
285 * deliberately NOT compacted: rebuild derives turn_item_positions from
286 * those events with first-write-wins semantics, so keep-latest would
287 * change replay outcomes for reordered items.
288 * 3. Legacy v1 thread events and their pre-migration command receipts for
289 * threads whose v2 import completed — the v1 store is only read to
290 * import from, and imports never re-run once transcript_imported_at is
291 * set.
292 *
293 * VACUUM is deliberately not run here: on the synchronous driver it would
294 * block every query for minutes on a multi-GB file. The reclaimable page
295 * count is reported so the caller can surface a hint instead; freed pages
296 * are reused, so the file stops growing either way.
297 */
298 const compactEventStore = Effect.gen(function* () {
299 const supersededThreadStateRows = yield* sql<{ readonly sequence: number }>`
300 SELECT sequence
301 FROM orchestration_events
302 WHERE application_event_version = 2
303 AND aggregate_kind = 'thread'
304 AND event_type IN ${sql.in(SUPERSEDABLE_THREAD_EVENT_TYPES)}
305 AND sequence NOT IN (
306 SELECT MAX(sequence)
307 FROM orchestration_events
308 WHERE application_event_version = 2
309 AND aggregate_kind = 'thread'
310 AND event_type IN ${sql.in(SUPERSEDABLE_THREAD_EVENT_TYPES)}
311 GROUP BY stream_id
312 )
313 `;
314
315 const supersededEntityRows = (eventType: "message.updated" | "node.updated") => sql<{
316 readonly sequence: number;
317 }>`
318 SELECT sequence
319 FROM orchestration_events
320 WHERE application_event_version = 2
321 AND event_type = ${eventType}
322 AND sequence NOT IN (
323 SELECT MAX(sequence)
324 FROM orchestration_events
325 WHERE application_event_version = 2
326 AND event_type = ${eventType}
327 GROUP BY stream_id, json_extract(payload_json, '$.id')
328 )
329 `;
330 const supersededMessageRows = yield* supersededEntityRows("message.updated");
331 const supersededNodeRows = yield* supersededEntityRows("node.updated");
332
333 const importedLegacyEventRows = yield* sql<{ readonly sequence: number }>`
334 SELECT sequence
335 FROM orchestration_events
336 WHERE application_event_version = 1
337 AND aggregate_kind = 'thread'
338 AND stream_id IN (
339 SELECT thread_id
340 FROM orchestration_v2_legacy_imports
341 WHERE transcript_imported_at IS NOT NULL
342 )
343 `;
344
345 const deletedEventCount = yield* deleteRowsBatched({
346 table: "orchestration_events",
347 keys: [
348 ...supersededThreadStateRows,
349 ...supersededMessageRows,
350 ...supersededNodeRows,
351 ...importedLegacyEventRows,
352 ].map((row) => row.sequence),
353 });
354
355 // Receipts written before the command_type column existed default to
356 // 'legacy'; for fully imported v1 threads they guard idempotency of
357 // commands that can no longer be re-sent.
358 const legacyReceiptRows = yield* sql<{ readonly command_id: string }>`
359 SELECT command_id
360 FROM orchestration_command_receipts
361 WHERE command_type = 'legacy'
362 AND aggregate_kind = 'thread'
363 AND aggregate_id IN (
364 SELECT thread_id
365 FROM orchestration_v2_legacy_imports
366 WHERE transcript_imported_at IS NOT NULL
367 )
368 `;
369 const deletedReceiptCount = yield* deleteRowsBatched({
370 table: "orchestration_command_receipts",
371 keys: legacyReceiptRows.map((row) => row.command_id),
372 });
373
374 const freelistRows = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`;
375 const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`;
376 const reclaimableBytes =
377 (freelistRows[0]?.freelist_count ?? 0) * (pageSizeRows[0]?.page_size ?? 0);
378
379 return { deletedEventCount, deletedReceiptCount, reclaimableBytes };
380 });
381
213 return ProjectionMaintenanceV2.of({382 return ProjectionMaintenanceV2.of({
214 verify: mapError("verify")(verify),383 verify: mapError("verify")(verify),
215 rebuild: mapError("rebuild")(rebuild),384 rebuild: mapError("rebuild")(rebuild),
385 compactEventStore: mapError("compact event store")(compactEventStore),
216 });386 });
217 }),387 }),
218);388);
apps/server/src/orchestration-v2/ProjectionStore.ts+187−73
@@ -104,6 +104,9 @@ export interface ProjectionStoreV2Shape { · 103 unmodified lines
104 OrchestrationV2ThreadShellSnapshot,104 OrchestrationV2ThreadShellSnapshot,
105 ProjectionStoreV2Error105 ProjectionStoreV2Error
106 >;106 >;
107 readonly getThreadShell: (
108 threadId: ThreadId,
109 ) => Effect.Effect<OrchestrationV2ThreadShell | null, ProjectionStoreV2Error>;
107 readonly getThreadProjection: (110 readonly getThreadProjection: (
108 threadId: ThreadId,111 threadId: ThreadId,
109 ) => Effect.Effect<OrchestrationV2ThreadProjection, ProjectionStoreV2Error>;112 ) => Effect.Effect<OrchestrationV2ThreadProjection, ProjectionStoreV2Error>;
@@ -391,6 +394,7 @@ type PayloadRow = { · 281 unmodified lines
391type ShellThreadRow = {394type ShellThreadRow = {
392 readonly thread_id: string;395 readonly thread_id: string;
393 readonly payload_json: string;396 readonly payload_json: string;
397 readonly forked_from_run_source_thread_id: string | null;
394 readonly latest_run_id: string | null;398 readonly latest_run_id: string | null;
395 readonly latest_run_status: string | null;399 readonly latest_run_status: string | null;
396 readonly latest_run_requested_at: string | null;400 readonly latest_run_requested_at: string | null;
@@ -2092,15 +2096,16 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> = · 1695 unmodified lines
2092 ),2096 ),
2093 );2097 );
2094 2098
2095 const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = () =>2099 const selectShellThreadRows = (threadId?: ThreadId) =>
2096 sql2100 sql<ShellThreadRow>`
2097 .withTransaction(
2098 Effect.gen(function* () {
2099 const [threadRows, runRows, itemCountRows, sequenceRows] = yield* Effect.all([
2100 sql<ShellThreadRow>`
2101 SELECT2101 SELECT
2102 t.thread_id,2102 t.thread_id,
2103 t.payload_json,2103 t.payload_json,
2104 CASE
2105 WHEN json_extract(t.payload_json, '$.forkedFrom.type') = 'run'
2106 THEN json_extract(t.payload_json, '$.forkedFrom.threadId')
2107 ELSE NULL
2108 END AS forked_from_run_source_thread_id,
2104 (2109 (
2105 SELECT r.run_id2110 SELECT r.run_id
2106 FROM orchestration_v2_projection_runs r2111 FROM orchestration_v2_projection_runs r
@@ -2199,19 +2204,118 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> = · 92 unmodified lines
2199 AND i.run_id IS NULL2204 AND i.run_id IS NULL
2200 ) AS runless_item_count2205 ) AS runless_item_count
2201 FROM orchestration_v2_projection_threads t2206 FROM orchestration_v2_projection_threads t
2202 WHERE t.deleted_at IS NULL2207 WHERE t.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND t.thread_id = ${threadId}`}
2203 ORDER BY t.updated_at ASC, t.thread_id ASC2208 ORDER BY t.updated_at ASC, t.thread_id ASC
2204 `,2209 `;
2205 sql<ShellRunRow>`2210
2211 const selectShellRunRows = (threadIds?: ReadonlyArray<ThreadId>) =>
2212 threadIds === undefined
2213 ? sql<ShellRunRow>`
2206 SELECT thread_id, run_id, ordinal2214 SELECT thread_id, run_id, ordinal
2207 FROM orchestration_v2_projection_runs2215 FROM orchestration_v2_projection_runs
2208 `,2216 `
2209 sql<ShellRunItemCountRow>`2217 : sql<ShellRunRow>`
2218 SELECT thread_id, run_id, ordinal
2219 FROM orchestration_v2_projection_runs
2220 WHERE thread_id IN ${sql.in(threadIds)}
2221 `;
2222
2223 const selectShellRunItemCounts = (threadIds?: ReadonlyArray<ThreadId>) =>
2224 threadIds === undefined
2225 ? sql<ShellRunItemCountRow>`
2210 SELECT thread_id, run_id, COUNT(*) AS item_count2226 SELECT thread_id, run_id, COUNT(*) AS item_count
2211 FROM orchestration_v2_projection_turn_items2227 FROM orchestration_v2_projection_turn_items
2212 WHERE run_id IS NOT NULL2228 WHERE run_id IS NOT NULL
2213 GROUP BY thread_id, run_id2229 GROUP BY thread_id, run_id
2214 `,2230 `
2231 : sql<ShellRunItemCountRow>`
2232 SELECT thread_id, run_id, COUNT(*) AS item_count
2233 FROM orchestration_v2_projection_turn_items
2234 WHERE run_id IS NOT NULL
2235 AND thread_id IN ${sql.in(threadIds)}
2236 GROUP BY thread_id, run_id
2237 `;
2238
2239 const runMapsByThreadId = (input: {
2240 readonly runRows: ReadonlyArray<ShellRunRow>;
2241 readonly itemCountRows: ReadonlyArray<ShellRunItemCountRow>;
2242 }) => {
2243 const runOrdinalsByThreadId = new Map<ThreadId, Map<RunId, number>>();
2244 for (const row of input.runRows) {
2245 const threadId = ThreadId.make(row.thread_id);
2246 const runId = RunId.make(row.run_id);
2247 const existing = runOrdinalsByThreadId.get(threadId) ?? new Map<RunId, number>();
2248 existing.set(runId, row.ordinal);
2249 runOrdinalsByThreadId.set(threadId, existing);
2250 }
2251 const itemCountsByThreadId = new Map<ThreadId, Map<RunId, number>>();
2252 for (const row of input.itemCountRows) {
2253 const threadId = ThreadId.make(row.thread_id);
2254 const runId = RunId.make(row.run_id);
2255 const existing = itemCountsByThreadId.get(threadId) ?? new Map<RunId, number>();
2256 existing.set(runId, row.item_count);
2257 itemCountsByThreadId.set(threadId, existing);
2258 }
2259 return { runOrdinalsByThreadId, itemCountsByThreadId };
2260 };
2261
2262 const shellThreadStateFromRow = (input: {
2263 readonly row: ShellThreadRow;
2264 readonly runOrdinalsByThreadId: ReadonlyMap<ThreadId, Map<RunId, number>>;
2265 readonly itemCountsByThreadId: ReadonlyMap<ThreadId, Map<RunId, number>>;
2266 }) =>
2267 Effect.gen(function* () {
2268 const { row, runOrdinalsByThreadId, itemCountsByThreadId } = input;
2269 const thread = yield* decodeThreadPayload(row.payload_json);
2270 const pendingRuntimeRequest =
2271 row.pending_request_payload_json === null
2272 ? null
2273 : yield* decodeRuntimeRequestPayload(row.pending_request_payload_json);
2274 const latestVisibleMessage =
2275 row.latest_message_payload_json === null
2276 ? null
2277 : yield* decodeMessagePayload(row.latest_message_payload_json);
2278 return {
2279 thread,
2280 latestRunId: row.latest_run_id === null ? null : RunId.make(row.latest_run_id),
2281 latestRunStatus: shellStatusFromStoredRunStatus(row.latest_run_status),
2282 latestRunRequestedAt:
2283 row.latest_run_requested_at === null
2284 ? null
2285 : DateTime.makeUnsafe(row.latest_run_requested_at),
2286 latestRunStartedAt:
2287 row.latest_run_started_at === null
2288 ? null
2289 : DateTime.makeUnsafe(row.latest_run_started_at),
2290 latestRunCompletedAt:
2291 row.latest_run_completed_at === null
2292 ? null
2293 : DateTime.makeUnsafe(row.latest_run_completed_at),
2294 activeRunId: row.active_run_id === null ? null : RunId.make(row.active_run_id),
2295 lastError: row.last_error,
2296 pendingRuntimeRequest,
2297 latestVisibleMessage,
2298 latestUserMessageAt:
2299 row.latest_user_message_at === null
2300 ? null
2301 : DateTime.makeUnsafe(row.latest_user_message_at),
2302 hasActionableProposedPlan: row.has_actionable_proposed_plan === 1,
2303 itemCount: row.item_count,
2304 runlessItemCount: row.runless_item_count,
2305 updatedAt: thread.updatedAt,
2306 runOrdinalById: runOrdinalsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),
2307 itemCountByRunId: itemCountsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),
2308 } satisfies ShellThreadState;
2309 });
2310
2311 const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = () =>
2312 sql
2313 .withTransaction(
2314 Effect.gen(function* () {
2315 const [threadRows, runRows, itemCountRows, sequenceRows] = yield* Effect.all([
2316 selectShellThreadRows(),
2317 selectShellRunRows(),
2318 selectShellRunItemCounts(),
2215 sql<{ readonly snapshot_sequence: number | null }>`2319 sql<{ readonly snapshot_sequence: number | null }>`
2216 SELECT MAX(sequence) AS snapshot_sequence2320 SELECT MAX(sequence) AS snapshot_sequence
2217 FROM orchestration_events2321 FROM orchestration_events
@@ -2220,69 +2324,13 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> = · 2 unmodified lines
2220 `,2324 `,
2221 ]);2325 ]);
2222 2326
2223 const runOrdinalsByThreadId = new Map<ThreadId, Map<RunId, number>>();2327 const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({
2224 for (const row of runRows) {2328 runRows,
2225 const threadId = ThreadId.make(row.thread_id);2329 itemCountRows,
2226 const runId = RunId.make(row.run_id);2330 });
2227 const existing = runOrdinalsByThreadId.get(threadId) ?? new Map<RunId, number>();
2228 existing.set(runId, row.ordinal);
2229 runOrdinalsByThreadId.set(threadId, existing);
2230 }
2231
2232 const itemCountsByThreadId = new Map<ThreadId, Map<RunId, number>>();
2233 for (const row of itemCountRows) {
2234 const threadId = ThreadId.make(row.thread_id);
2235 const runId = RunId.make(row.run_id);
2236 const existing = itemCountsByThreadId.get(threadId) ?? new Map<RunId, number>();
2237 existing.set(runId, row.item_count);
2238 itemCountsByThreadId.set(threadId, existing);
2239 }
2240 2331
2241 const states = yield* Effect.forEach(threadRows, (row) =>2332 const states = yield* Effect.forEach(threadRows, (row) =>
2242 Effect.gen(function* () {2333 shellThreadStateFromRow({ row, runOrdinalsByThreadId, itemCountsByThreadId }),
2243 const thread = yield* decodeThreadPayload(row.payload_json);
2244 const pendingRuntimeRequest =
2245 row.pending_request_payload_json === null
2246 ? null
2247 : yield* decodeRuntimeRequestPayload(row.pending_request_payload_json);
2248 const latestVisibleMessage =
2249 row.latest_message_payload_json === null
2250 ? null
2251 : yield* decodeMessagePayload(row.latest_message_payload_json);
2252 return {
2253 thread,
2254 latestRunId: row.latest_run_id === null ? null : RunId.make(row.latest_run_id),
2255 latestRunStatus: shellStatusFromStoredRunStatus(row.latest_run_status),
2256 latestRunRequestedAt:
2257 row.latest_run_requested_at === null
2258 ? null
2259 : DateTime.makeUnsafe(row.latest_run_requested_at),
2260 latestRunStartedAt:
2261 row.latest_run_started_at === null
2262 ? null
2263 : DateTime.makeUnsafe(row.latest_run_started_at),
2264 latestRunCompletedAt:
2265 row.latest_run_completed_at === null
2266 ? null
2267 : DateTime.makeUnsafe(row.latest_run_completed_at),
2268 activeRunId: row.active_run_id === null ? null : RunId.make(row.active_run_id),
2269 lastError: row.last_error,
2270 pendingRuntimeRequest,
2271 latestVisibleMessage,
2272 latestUserMessageAt:
2273 row.latest_user_message_at === null
2274 ? null
2275 : DateTime.makeUnsafe(row.latest_user_message_at),
2276 hasActionableProposedPlan: row.has_actionable_proposed_plan === 1,
2277 itemCount: row.item_count,
2278 runlessItemCount: row.runless_item_count,
2279 updatedAt: thread.updatedAt,
2280 runOrdinalById:
2281 runOrdinalsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),
2282 itemCountByRunId:
2283 itemCountsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),
2284 } satisfies ShellThreadState;
2285 }),
2286 );2334 );
2287 const statesByThreadId = new Map(states.map((state) => [state.thread.id, state]));2335 const statesByThreadId = new Map(states.map((state) => [state.thread.id, state]));
2288 2336
@@ -2314,9 +2362,64 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> = · 25 unmodified lines
2314 ),2362 ),
2315 );2363 );
2316 2364
2365 // Per-thread shell for the live shell streams: reads only the target thread
2366 // plus its fork-source chain instead of materializing every thread. Returns
2367 // null when the thread is deleted or unknown.
2368 const getThreadShell: ProjectionStoreV2Shape["getThreadShell"] = (threadId) =>
2369 sql
2370 .withTransaction(
2371 Effect.gen(function* () {
2372 const rowsByThreadId = new Map<ThreadId, ShellThreadRow>();
2373 const pending: Array<ThreadId> = [threadId];
2374 while (pending.length > 0) {
2375 const nextId = pending.pop();
2376 if (nextId === undefined || rowsByThreadId.has(nextId)) {
2377 continue;
2378 }
2379 const rows = yield* selectShellThreadRows(nextId);
2380 const row = rows[0];
2381 if (row === undefined) {
2382 continue;
2383 }
2384 rowsByThreadId.set(nextId, row);
2385 if (row.forked_from_run_source_thread_id !== null) {
2386 pending.push(ThreadId.make(row.forked_from_run_source_thread_id));
2387 }
2388 }
2389 if (!rowsByThreadId.has(threadId)) {
2390 return null;
2391 }
2392
2393 const threadIds = [...rowsByThreadId.keys()];
2394 const [runRows, itemCountRows] = yield* Effect.all([
2395 selectShellRunRows(threadIds),
2396 selectShellRunItemCounts(threadIds),
2397 ]);
2398 const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({
2399 runRows,
2400 itemCountRows,
2401 });
2402
2403 const states = yield* Effect.forEach([...rowsByThreadId.values()], (row) =>
2404 shellThreadStateFromRow({ row, runOrdinalsByThreadId, itemCountsByThreadId }),
2405 );
2406 const statesByThreadId = new Map(states.map((state) => [state.thread.id, state]));
2407 const state = statesByThreadId.get(threadId);
2408 if (state === undefined) {
2409 return null;
2410 }
2411 return shellFromState({
2412 state,
2413 visibleItemCount: visibleItemCountForShell({ threadId, statesByThreadId }),
2414 });
2415 }),
2416 )
2417 .pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })));
2418
2317 return {2419 return {
2318 apply,2420 apply,
2319 getShellSnapshot,2421 getShellSnapshot,
2422 getThreadShell,
2320 getThreadProjection,2423 getThreadProjection,
2321 getThreadSnapshot,2424 getThreadSnapshot,
2322 } satisfies ProjectionStoreV2Shape;2425 } satisfies ProjectionStoreV2Shape;
@@ -2369,6 +2472,17 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect( · 46 unmodified lines
2369 archivedThreads: visible.filter((thread) => thread.archivedAt !== null),2472 archivedThreads: visible.filter((thread) => thread.archivedAt !== null),
2370 };2473 };
2371 }),2474 }),
2475 getThreadShell: (threadId) =>
2476 Effect.gen(function* () {
2477 const existing = (yield* Ref.get(replayState)).projections;
2478 if (!existing.has(threadId)) {
2479 return null;
2480 }
2481 const shell = yield* service
2482 .getThreadProjection(threadId)
2483 .pipe(Effect.map(threadShellFromProjection));
2484 return shell.deletedAt === null ? shell : null;
2485 }),
2372 getThreadProjection: (threadId) =>2486 getThreadProjection: (threadId) =>
2373 Effect.gen(function* () {2487 Effect.gen(function* () {
2374 const existing = (yield* Ref.get(replayState)).projections;2488 const existing = (yield* Ref.get(replayState)).projections;
apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts+3−4
@@ -301,10 +301,9 @@ layer("ProviderEventIngestorV2", (it) => { · 300 unmodified lines
301 maxAttempts: 3,301 maxAttempts: 3,
302 retryDelayMs: 2_000,302 retryDelayMs: 2_000,
303 });303 });
304 assert.equal(304 const errorStartedAt = errorItem.startedAt;
305 DateTime.toEpochMillis(errorItem.startedAt),305 assert.ok(errorStartedAt);
306 DateTime.toEpochMillis(retryStartedAt),306 assert.equal(DateTime.toEpochMillis(errorStartedAt), DateTime.toEpochMillis(retryStartedAt));
307 );
308 assert.equal(errorItem.providerThreadId, providerThreadId);307 assert.equal(errorItem.providerThreadId, providerThreadId);
309 assert.equal(errorItem.providerTurnId, providerTurnId);308 assert.equal(errorItem.providerTurnId, providerTurnId);
310 }),309 }),
apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts+1−0
@@ -215,6 +215,7 @@ it.effect( · 214 unmodified lines
215 ProjectionStoreV2.of({215 ProjectionStoreV2.of({
216 apply: () => Effect.void,216 apply: () => Effect.void,
217 getShellSnapshot: () => Effect.die("unused getShellSnapshot"),217 getShellSnapshot: () => Effect.die("unused getShellSnapshot"),
218 getThreadShell: () => Effect.die("unused getThreadShell"),
218 getThreadProjection: () => Ref.get(projection),219 getThreadProjection: () => Ref.get(projection),
219 getThreadSnapshot: () => Effect.die("unused getThreadSnapshot"),220 getThreadSnapshot: () => Effect.die("unused getThreadSnapshot"),
220 }),221 }),
apps/server/src/orchestration-v2/ShellStream.test.ts+132−1
@@ -1,12 +1,20 @@
1import { describe, expect, it } from "@effect/vitest";1import { describe, expect, it } from "@effect/vitest";
2import type { ApplicationStoredEvent, OrchestrationV2ShellSnapshot } from "@t3tools/contracts";2import type {
3 ApplicationStoredEvent,
4 OrchestrationV2ShellSnapshot,
5 OrchestrationV2StoredEvent,
6 OrchestrationV2ThreadShell,
7} from "@t3tools/contracts";
3import * as Effect from "effect/Effect";8import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";9import * as Stream from "effect/Stream";
5 10
6import {11import {
12 archivedShellStreamItemFromThreadShell,
7 coalesceShellApplicationEvents,13 coalesceShellApplicationEvents,
14 coalesceStoredThreadEvents,
8 composeShellStreamWithEnrichment,15 composeShellStreamWithEnrichment,
9 shellStreamItemFromEnrichmentRefresh,16 shellStreamItemFromEnrichmentRefresh,
17 shellStreamItemFromThreadShell,
10 shellStreamItemsFromInitialSnapshot,18 shellStreamItemsFromInitialSnapshot,
11} from "./ShellStream.ts";19} from "./ShellStream.ts";
12 20
@@ -47,6 +55,129 @@ describe("coalesceShellApplicationEvents", () => { · 34 unmodified lines
47 });55 });
48});56});
49 57
58function storedThreadEvent(
59 sequence: number,
60 threadId: string,
61 event: Record<string, unknown> = {},
62): OrchestrationV2StoredEvent {
63 return { sequence, event: { threadId, ...event } } as OrchestrationV2StoredEvent;
64}
65
66function shellFixture(overrides: Partial<OrchestrationV2ThreadShell>): OrchestrationV2ThreadShell {
67 return { id: "thread-a", archivedAt: null, ...overrides } as OrchestrationV2ThreadShell;
68}
69
70describe("coalesceStoredThreadEvents", () => {
71 it("keeps the newest stored event per thread and preserves sequence order", () => {
72 expect(
73 coalesceStoredThreadEvents([
74 storedThreadEvent(2, "thread-a"),
75 storedThreadEvent(3, "thread-b"),
76 storedThreadEvent(5, "thread-a"),
77 ]).map((stored) => stored.sequence),
78 ).toEqual([3, 5]);
79 });
80});
81
82describe("shellStreamItemFromThreadShell", () => {
83 it("emits an active thread update when the shell is not archived", () => {
84 const shell = shellFixture({ archivedAt: null });
85 expect(
86 shellStreamItemFromThreadShell({ stored: storedThreadEvent(4, "thread-a"), shell }),
87 ).toEqual({
88 kind: "thread.updated",
89 sequence: 4,
90 location: "active",
91 thread: shell,
92 });
93 });
94
95 it("emits an archive update when the shell is archived", () => {
96 const shell = shellFixture({ archivedAt: "2026-07-30T00:00:00.000Z" as never });
97 expect(
98 shellStreamItemFromThreadShell({ stored: storedThreadEvent(4, "thread-a"), shell }),
99 ).toEqual({
100 kind: "thread.updated",
101 sequence: 4,
102 location: "archive",
103 thread: shell,
104 });
105 });
106
107 it("emits a removal from the archive when an archived thread is deleted", () => {
108 expect(
109 shellStreamItemFromThreadShell({
110 stored: storedThreadEvent(6, "thread-a", {
111 type: "thread.deleted",
112 payload: { archivedAt: "2026-07-30T00:00:00.000Z" },
113 }),
114 shell: null,
115 }),
116 ).toEqual({
117 kind: "thread.removed",
118 sequence: 6,
119 location: "archive",
120 threadId: "thread-a",
121 });
122 });
123
124 it("emits a removal from the active list for other missing shells", () => {
125 expect(
126 shellStreamItemFromThreadShell({
127 stored: storedThreadEvent(6, "thread-a", {
128 type: "thread.deleted",
129 payload: { archivedAt: null },
130 }),
131 shell: null,
132 }),
133 ).toEqual({
134 kind: "thread.removed",
135 sequence: 6,
136 location: "active",
137 threadId: "thread-a",
138 });
139 });
140});
141
142describe("archivedShellStreamItemFromThreadShell", () => {
143 it("emits an update for an archived shell", () => {
144 const shell = shellFixture({ archivedAt: "2026-07-30T00:00:00.000Z" as never });
145 expect(
146 archivedShellStreamItemFromThreadShell({ stored: storedThreadEvent(4, "thread-a"), shell }),
147 ).toEqual({ kind: "thread.updated", sequence: 4, thread: shell });
148 });
149
150 it("ignores active threads that never touched the archive", () => {
151 expect(
152 archivedShellStreamItemFromThreadShell({
153 stored: storedThreadEvent(4, "thread-a", { type: "thread.settled" }),
154 shell: shellFixture({ archivedAt: null }),
155 }),
156 ).toBeNull();
157 });
158
159 it("emits a removal when a thread leaves the archive", () => {
160 expect(
161 archivedShellStreamItemFromThreadShell({
162 stored: storedThreadEvent(4, "thread-a", { type: "thread.unarchived" }),
163 shell: shellFixture({ archivedAt: null }),
164 }),
165 ).toEqual({ kind: "thread.removed", sequence: 4, threadId: "thread-a" });
166 });
167
168 it("emits a removal when an archived thread is deleted", () => {
169 expect(
170 archivedShellStreamItemFromThreadShell({
171 stored: storedThreadEvent(4, "thread-a", {
172 type: "thread.deleted",
173 payload: { archivedAt: "2026-07-30T00:00:00.000Z" },
174 }),
175 shell: null,
176 }),
177 ).toEqual({ kind: "thread.removed", sequence: 4, threadId: "thread-a" });
178 });
179});
180
50describe("shellStreamItemFromEnrichmentRefresh", () => {181describe("shellStreamItemFromEnrichmentRefresh", () => {
51 it("batches nearby completion roots onto one snapshot item", () => {182 it("batches nearby completion roots onto one snapshot item", () => {
52 expect(183 expect(
apps/server/src/orchestration-v2/ShellStream.ts+27−27
@@ -1,7 +1,7 @@
1import type {1import type {
2 OrchestrationV2ArchivedShellStreamItem,2 OrchestrationV2ArchivedShellStreamItem,
3 OrchestrationV2ShellSnapshot,3 OrchestrationV2ShellSnapshot,
4 OrchestrationV2ThreadShellSnapshot,4 OrchestrationV2ThreadShell,
5 OrchestrationV2ShellStreamItem,5 OrchestrationV2ShellStreamItem,
6 OrchestrationV2StoredEvent,6 OrchestrationV2StoredEvent,
7} from "@t3tools/contracts";7} from "@t3tools/contracts";
@@ -76,30 +76,33 @@ export function shellStreamItemsFromInitialSnapshot(input: { · 68 unmodified lines
76 ];76 ];
77}77}
78 78
79/** Converts a committed event and its resulting shell snapshot into one delta. */79/** Keep only the newest stored event per thread within a coalescing window. */
80export function shellStreamItemFromSnapshot(input: {80export function coalesceStoredThreadEvents(
81 readonly stored: OrchestrationV2StoredEvent;81 events: ReadonlyArray<OrchestrationV2StoredEvent>,
82 readonly snapshot: OrchestrationV2ThreadShellSnapshot;82): ReadonlyArray<OrchestrationV2StoredEvent> {
83}): Exclude<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }> {83 const latestByThreadId = new Map<string, OrchestrationV2StoredEvent>();
84 const active = input.snapshot.threads.find((thread) => thread.id === input.stored.event.threadId);84 for (const stored of events) {
85 if (active !== undefined) {85 latestByThreadId.set(stored.event.threadId, stored);
86 return {
87 kind: "thread.updated",
88 sequence: input.stored.sequence,
89 location: "active",
90 thread: active,
91 };
92 }86 }
93 87 return Array.from(latestByThreadId.values()).sort(
94 const archived = input.snapshot.archivedThreads.find(88 (left, right) => left.sequence - right.sequence,
95 (thread) => thread.id === input.stored.event.threadId,
96 );89 );
97 if (archived !== undefined) {90}
91
92/**
93 * Converts a committed event and the affected thread's current shell into one
94 * delta. `shell` is null when the thread is deleted or unknown.
95 */
96export function shellStreamItemFromThreadShell(input: {
97 readonly stored: OrchestrationV2StoredEvent;
98 readonly shell: OrchestrationV2ThreadShell | null;
99}): Exclude<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }> {
100 if (input.shell !== null) {
98 return {101 return {
99 kind: "thread.updated",102 kind: "thread.updated",
100 sequence: input.stored.sequence,103 sequence: input.stored.sequence,
101 location: "archive",104 location: input.shell.archivedAt === null ? "active" : "archive",
102 thread: archived,105 thread: input.shell,
103 };106 };
104 }107 }
105 108
@@ -115,18 +118,15 @@ export function shellStreamItemFromSnapshot(input: { · 9 unmodified lines
115}118}
116 119
117/** Converts a committed event into an archive-only delta when it changes archive membership. */120/** Converts a committed event into an archive-only delta when it changes archive membership. */
118export function archivedShellStreamItemFromSnapshot(input: {121export function archivedShellStreamItemFromThreadShell(input: {
119 readonly stored: OrchestrationV2StoredEvent;122 readonly stored: OrchestrationV2StoredEvent;
120 readonly snapshot: OrchestrationV2ThreadShellSnapshot;123 readonly shell: OrchestrationV2ThreadShell | null;
121}): Exclude<OrchestrationV2ArchivedShellStreamItem, { readonly kind: "snapshot" }> | null {124}): Exclude<OrchestrationV2ArchivedShellStreamItem, { readonly kind: "snapshot" }> | null {
122 const archived = input.snapshot.archivedThreads.find(125 if (input.shell !== null && input.shell.archivedAt !== null) {
123 (thread) => thread.id === input.stored.event.threadId,
124 );
125 if (archived !== undefined) {
126 return {126 return {
127 kind: "thread.updated",127 kind: "thread.updated",
128 sequence: input.stored.sequence,128 sequence: input.stored.sequence,
129 thread: archived,129 thread: input.shell,
130 };130 };
131 }131 }
132 if (132 if (
apps/server/src/orchestration-v2/ThreadManagementService.test.ts+19−0
@@ -107,6 +107,25 @@ it("identifies every existing thread that must be hydrated before dispatch", () · 106 unmodified lines
107 }),107 }),
108 ).toEqual([targetThreadId]);108 ).toEqual([targetThreadId]);
109 109
110 // Read-state commands skip transcript hydration entirely: they fire on
111 // every activity bump while a thread is open and never touch messages.
112 expect(
113 existingThreadIdsForCommand({
114 type: "thread.visit",
115 commandId: CommandId.make("command:thread-management:visit"),
116 threadId: targetThreadId,
117 visitedAt: "2026-07-30T00:00:00.000Z",
118 }),
119 ).toEqual([]);
120
121 expect(
122 existingThreadIdsForCommand({
123 type: "thread.mark-unread",
124 commandId: CommandId.make("command:thread-management:mark-unread"),
125 threadId: targetThreadId,
126 }),
127 ).toEqual([]);
128
110 expect(129 expect(
111 existingThreadIdsForCommand({130 existingThreadIdsForCommand({
112 type: "thread.fork",131 type: "thread.fork",
apps/server/src/orchestration-v2/ThreadManagementService.ts+31−12
@@ -65,6 +65,13 @@ export function existingThreadIdsForCommand( · 64 unmodified lines
65 switch (command.type) {65 switch (command.type) {
66 case "thread.create":66 case "thread.create":
67 return [];67 return [];
68 // Read-state commands only rewrite the thread payload's visited/unread
69 // watermark; they never touch messages, so they do not need the imported
70 // v1 transcript hydrated first. Visits fire on every thread-activity bump
71 // while a thread is open, so keeping them off the import path matters.
72 case "thread.visit":
73 case "thread.mark-unread":
74 return [];
68 case "thread.fork":75 case "thread.fork":
69 return [command.sourceThreadId];76 return [command.sourceThreadId];
70 case "thread.merge_back":77 case "thread.merge_back":
@@ -106,7 +113,8 @@ export interface ThreadManagementSendResult { · 35 unmodified lines
106 readonly projection: OrchestrationV2ThreadProjection;113 readonly projection: OrchestrationV2ThreadProjection;
107 readonly message: OrchestrationV2ConversationMessage;114 readonly message: OrchestrationV2ConversationMessage;
108 readonly run: OrchestrationV2Run;115 readonly run: OrchestrationV2Run;
109 readonly turnItem: Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }>;116 /** Null for queued sends: the user turn item materializes when the queued turn starts. */
117 readonly turnItem: Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }> | null;
110 readonly delivery: "started" | "queued" | "steered" | "restarted";118 readonly delivery: "started" | "queued" | "steered" | "restarted";
111}119}
112 120
@@ -273,6 +281,7 @@ export interface ThreadManagementServiceShape { · 160 unmodified lines
273 OrchestrationV2ThreadShellSnapshot,281 OrchestrationV2ThreadShellSnapshot,
274 OrchestratorV2Error282 OrchestratorV2Error
275 >;283 >;
284 readonly getThreadShell: OrchestratorV2["Service"]["getThreadShell"];
276 readonly listProjectThreads: (input: {285 readonly listProjectThreads: (input: {
277 readonly projectId: ProjectId;286 readonly projectId: ProjectId;
278 readonly includeSubagents: boolean;287 readonly includeSubagents: boolean;
@@ -505,23 +514,32 @@ const make = Effect.gen(function* () { · 226 unmodified lines
505 message?.runId === null || message?.runId === undefined514 message?.runId === null || message?.runId === undefined
506 ? undefined515 ? undefined
507 : projection.runs.find((candidate) => candidate.id === message.runId);516 : projection.runs.find((candidate) => candidate.id === message.runId);
508 const turnItem = projection.turnItems.find(517 const turnItem =
509 (518 projection.turnItems.find(
510 candidate,519 (
511 ): candidate is Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }> =>520 candidate,
512 candidate.type === "user_message" && candidate.messageId === input.messageId,521 ): candidate is Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }> =>
513 );522 candidate.type === "user_message" && candidate.messageId === input.messageId,
514 if (message === undefined || run === undefined || turnItem === undefined) {523 ) ?? null;
524 // A queued message's user turn item is deliberately not emitted at
525 // dispatch time — it materializes when the queued turn actually starts,
526 // so it can map onto the provider turn. Every other dispatch mode still
527 // produces its turn item transactionally with the run.
528 if (
529 message === undefined ||
530 run === undefined ||
531 (turnItem === null && run.status !== "queued")
532 ) {
515 return yield* new ThreadManagementDurableRunProjectionError({533 return yield* new ThreadManagementDurableRunProjectionError({
516 threadId: input.threadId,534 threadId: input.threadId,
517 messageId: input.messageId,535 messageId: input.messageId,
518 });536 });
519 }537 }
520 const delivery: ThreadManagementSendResult["delivery"] =538 const delivery: ThreadManagementSendResult["delivery"] =
521 turnItem.inputIntent === "turn_start"539 turnItem === null || turnItem.inputIntent === "queued_turn"
522 ? "started"540 ? "queued"
523 : turnItem.inputIntent === "queued_turn"541 : turnItem.inputIntent === "turn_start"
524 ? "queued"542 ? "started"
525 : input.mode === "restart"543 : input.mode === "restart"
526 ? "restarted"544 ? "restarted"
527 : "steered";545 : "steered";
@@ -631,6 +649,7 @@ const make = Effect.gen(function* () { · 103 unmodified lines
631 getThreadSnapshot,649 getThreadSnapshot,
632 getProjectThread,650 getProjectThread,
633 getShellSnapshot: orchestrator.getShellSnapshot,651 getShellSnapshot: orchestrator.getShellSnapshot,
652 getThreadShell: orchestrator.getThreadShell,
634 listProjectThreads,653 listProjectThreads,
635 sendToThread,654 sendToThread,
636 waitForThread,655 waitForThread,
apps/server/src/orchestration-v2/runtimeLayer.test.ts+7−4
@@ -45,7 +45,7 @@ import { EventSinkV2 } from "./EventSink.ts"; · 44 unmodified lines
45import { ProjectionMaintenanceV2 } from "./ProjectionMaintenance.ts";45import { ProjectionMaintenanceV2 } from "./ProjectionMaintenance.ts";
46import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";46import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts";
47import { OrchestrationV2EventSinkLayerLive, OrchestrationV2LayerLive } from "./runtimeLayer.ts";47import { OrchestrationV2EventSinkLayerLive, OrchestrationV2LayerLive } from "./runtimeLayer.ts";
48import { shellStreamItemFromSnapshot } from "./ShellStream.ts";48import { shellStreamItemFromThreadShell } from "./ShellStream.ts";
49import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";49import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
50import { ThreadManagementService } from "./ThreadManagementService.ts";50import { ThreadManagementService } from "./ThreadManagementService.ts";
51 51
@@ -632,9 +632,9 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 580 unmodified lines
632 threadId,632 threadId,
633 );633 );
634 assert.deepEqual(634 assert.deepEqual(
635 shellStreamItemFromSnapshot({635 shellStreamItemFromThreadShell({
636 stored: archive.storedEvents[0]!,636 stored: archive.storedEvents[0]!,
637 snapshot: archivedShell,637 shell: yield* orchestrator.getThreadShell(threadId),
638 }),638 }),
639 {639 {
640 kind: "thread.updated",640 kind: "thread.updated",
@@ -659,7 +659,10 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { · 18 unmodified lines
659 threadId,659 threadId,
660 );660 );
661 assert.deepEqual(661 assert.deepEqual(
662 shellStreamItemFromSnapshot({ stored: remove.storedEvents[0]!, snapshot: deletedShell }),662 shellStreamItemFromThreadShell({
663 stored: remove.storedEvents[0]!,
664 shell: yield* orchestrator.getThreadShell(threadId),
665 }),
663 {666 {
664 kind: "thread.removed",667 kind: "thread.removed",
665 sequence: remove.sequence,668 sequence: remove.sequence,
apps/server/src/serverRuntimeStartup.ts+24−0
@@ -510,6 +510,30 @@ export const make = Effect.gen(function* () { · 509 unmodified lines
510 : importPendingTranscripts510 : importPendingTranscripts
511 ).pipe(Effect.forkScoped);511 ).pipe(Effect.forkScoped);
512 512
513 // Off the startup path: the first run after an upgrade deletes the whole
514 // superseded-event and legacy-v1 backlog (potentially millions of rows,
515 // paced in small batches), and nothing at boot depends on it.
516 yield* projectionMaintenance.compactEventStore.pipe(
517 Effect.tap((summary) =>
518 summary.deletedEventCount === 0 && summary.deletedReceiptCount === 0
519 ? Effect.void
520 : Effect.logInfo("Compacted orchestration event store", summary),
521 ),
522 Effect.tap((summary) =>
523 // Freed pages are reused, so the file stops growing regardless; only
524 // an offline VACUUM shrinks it, which is not safe to run on the
525 // synchronous sqlite connection while serving.
526 summary.reclaimableBytes >= 512 * 1024 * 1024
527 ? Effect.logInfo(
528 "state.sqlite has substantial reclaimable free space; an offline VACUUM would shrink the file",
529 { reclaimableBytes: summary.reclaimableBytes },
530 )
531 : Effect.void,
532 ),
533 Effect.catch((cause) => Effect.logWarning("Unable to compact the event store", { cause })),
534 Effect.forkScoped,
535 );
536
513 yield* Effect.logDebug("startup phase: publishing welcome event", {537 yield* Effect.logDebug("startup phase: publishing welcome event", {
514 environmentId: environment.environmentId,538 environmentId: environment.environmentId,
515 cwd: welcomeBase.cwd,539 cwd: welcomeBase.cwd,
apps/server/src/ws.ts+31−50
@@ -68,11 +68,12 @@ import * as ThreadManagementService from "./orchestration-v2/ThreadManagementSer · 67 unmodified lines
68import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts";68import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts";
69import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts";69import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts";
70import {70import {
71 archivedShellStreamItemFromSnapshot,71 archivedShellStreamItemFromThreadShell,
72 coalesceShellApplicationEvents,72 coalesceShellApplicationEvents,
73 coalesceStoredThreadEvents,
73 composeShellStreamWithEnrichment,74 composeShellStreamWithEnrichment,
74 shellStreamItemFromEnrichmentRefresh,75 shellStreamItemFromEnrichmentRefresh,
75 shellStreamItemFromSnapshot,76 shellStreamItemFromThreadShell,
76 shellStreamItemsFromInitialSnapshot,77 shellStreamItemsFromInitialSnapshot,
77} from "./orchestration-v2/ShellStream.ts";78} from "./orchestration-v2/ShellStream.ts";
78import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";79import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";
@@ -700,46 +701,24 @@ const makeWsRpcLayer = (currentSession: EnvironmentAuth.AuthenticatedSession) => · 621 unmodified lines
700 ),701 ),
701 );702 );
702 703
703 const live = applicationEvents704 // Coalescing makes each per-thread shell read represent every event
704 .streamApplicationEvents({ afterSequence: snapshot.snapshotSequence })705 // for that thread in the current window; reading only the affected
705 .pipe(706 // threads keeps the cost of a busy stream independent of how many
706 Stream.mapEffect((stored) =>707 // threads exist overall.
708 const projectShellItems = Effect.fn("ws.orchestrationV2.projectShellItems")(function* (
709 events: ReadonlyArray<ApplicationStoredEvent>,
710 ) {
711 return yield* Effect.forEach(
712 coalesceShellApplicationEvents(events),
713 (stored) =>
707 Effect.gen(function* () {714 Effect.gen(function* () {
708 if ("aggregateKind" in stored) {715 if ("aggregateKind" in stored) {
709 if (stored.type === "project.deleted") {716 return yield* projectItem(stored);
710 return {
711 kind: "project.removed" as const,
712 sequence: stored.sequence,
713 projectId: stored.payload.projectId,
714 };
715 }
716 const project = yield* projectionSnapshotQuery.getProjectShellById(
717 stored.payload.projectId,
718 );
719 return Option.match(project, {
720 onNone: () => ({
721 kind: "project.removed" as const,
722 sequence: stored.sequence,
723 projectId: stored.payload.projectId,
724 }),
725 onSome: (value) => ({
726 kind: "project.updated" as const,
727 sequence: stored.sequence,
728 project: value,
729 }),
730 });
731 }717 }
732 const nextSnapshot = yield* threadManagement.getShellSnapshot();718 const shell = yield* threadManagement.getThreadShell(stored.event.threadId);
733 return shellStreamItemFromSnapshot({ stored, snapshot: nextSnapshot });719 return shellStreamItemFromThreadShell({ stored, shell });
734 }),720 }),
735 ),721 { concurrency: 8 },
736 Stream.mapError(
737 (cause) =>
738 new OrchestrationV2GetShellSnapshotError({
739 message: "Failed while streaming the application shell",
740 cause,
741 }),
742 ),
743 );722 );
744 723
745 const enrichmentRefreshes = Stream.fromSubscription(enrichmentChanges).pipe(724 const enrichmentRefreshes = Stream.fromSubscription(enrichmentChanges).pipe(
@@ -882,20 +861,22 @@ const makeWsRpcLayer = (currentSession: EnvironmentAuth.AuthenticatedSession) => · 136 unmodified lines
882 const live = threadManagement861 const live = threadManagement
883 .streamStoredEventsFrom({ afterSequence: snapshot.snapshotSequence })862 .streamStoredEventsFrom({ afterSequence: snapshot.snapshotSequence })
884 .pipe(863 .pipe(
885 Stream.mapEffect((stored) =>864 Stream.groupedWithin(512, Duration.millis(50)),
886 threadManagement.getShellSnapshot().pipe(865 Stream.mapEffect((events) =>
887 Effect.map((nextSnapshot) =>866 Effect.forEach(
888 archivedShellStreamItemFromSnapshot({ stored, snapshot: nextSnapshot }),867 coalesceStoredThreadEvents(Array.from(events)),
889 ),868 (stored) =>
890 Effect.mapError(869 threadManagement
891 (cause) =>870 .getThreadShell(stored.event.threadId)
892 new OrchestrationV2GetShellSnapshotError({871 .pipe(
893 message: "Failed while streaming archived threads",872 Effect.map((shell) =>
894 cause,873 archivedShellStreamItemFromThreadShell({ stored, shell }),
895 }),874 ),
896 ),875 ),
876 { concurrency: 8 },
897 ),877 ),
898 ),878 ),
879 Stream.flatMap(Stream.fromIterable),
899 Stream.filterMap((item) => (item === null ? Result.failVoid : Result.succeed(item))),880 Stream.filterMap((item) => (item === null ? Result.failVoid : Result.succeed(item))),
900 Stream.mapError(881 Stream.mapError(
901 (cause) =>882 (cause) =>
apps/web/src/components/ChatView.tsx+38−7
@@ -310,6 +310,12 @@ import { useAssetUrls } from "../assets/assetUrls"; · 309 unmodified lines
310const IMAGE_ONLY_BOOTSTRAP_PROMPT =310const IMAGE_ONLY_BOOTSTRAP_PROMPT =
311 "[User attached one or more images without additional text. Respond using the conversation context and the attached image(s).]";311 "[User attached one or more images without additional text. Respond using the conversation context and the attached image(s).]";
312const EMPTY_PROVIDERS: ServerProvider[] = [];312const EMPTY_PROVIDERS: ServerProvider[] = [];
313// During an active turn the thread's updatedAt advances several times per
314// second, and every server-side visit is a full command dispatch plus a
315// broadcast to all shell subscribers. Mid-turn bumps carry no unread signal
316// (unread flips on run completions, which bypass the throttle), so one
317// watermark per interval is plenty.
318const VISIT_DISPATCH_THROTTLE_MS = 10_000;
313const EMPTY_PROVIDER_SKILLS: ServerProvider["skills"] = [];319const EMPTY_PROVIDER_SKILLS: ServerProvider["skills"] = [];
314const EMPTY_PROJECTION_RUNS: OrchestrationV2ThreadProjection["runs"] = [];320const EMPTY_PROJECTION_RUNS: OrchestrationV2ThreadProjection["runs"] = [];
315const EMPTY_ATTACHMENT_IDS: string[] = [];321const EMPTY_ATTACHMENT_IDS: string[] = [];
@@ -1228,6 +1234,7 @@ function ChatViewContent(props: ChatViewProps) { · 912 unmodified lines
1228 );1234 );
1229 const visitThreadMutation = useAtomCommand(threadEnvironment.visit, { reportFailure: false });1235 const visitThreadMutation = useAtomCommand(threadEnvironment.visit, { reportFailure: false });
1230 const lastDispatchedVisitRef = useRef<string | null>(null);1236 const lastDispatchedVisitRef = useRef<string | null>(null);
1237 const lastVisitDispatchAtRef = useRef(0);
1231 const settings = useEnvironmentSettings(environmentId);1238 const settings = useEnvironmentSettings(environmentId);
1232 // New-thread defaults live in the primary environment's settings.json (the1239 // New-thread defaults live in the primary environment's settings.json (the
1233 // settings UI never writes to remote environments), so read them from the1240 // settings UI never writes to remote environments), so read them from the
@@ -1918,15 +1925,38 @@ function ChatViewContent(props: ChatViewProps) { · 684 unmodified lines
1918 if (serverThread.lastVisitedAt !== undefined) {1925 if (serverThread.lastVisitedAt !== undefined) {
1919 // Server-tracked visited state: record the watermark server-side so it1926 // Server-tracked visited state: record the watermark server-side so it
1920 // syncs across every device connected to the environment. Dedupe per1927 // syncs across every device connected to the environment. Dedupe per
1921 // watermark — the effect re-runs before the command's echo lands.1928 // watermark — the effect re-runs before the command's echo lands. The
1929 // dedupe also keeps a mark-unread on the open thread sticky: the rewind
1930 // leaves updatedAt untouched, so the already-dispatched key skips a
1931 // fresh visit until new activity lands or the thread is reopened.
1922 const dispatchKey = `${routeThreadKey}:${serverThread.updatedAt}`;1932 const dispatchKey = `${routeThreadKey}:${serverThread.updatedAt}`;
1923 if (lastDispatchedVisitRef.current === dispatchKey) return;1933 if (lastDispatchedVisitRef.current === dispatchKey) return;
1924 lastDispatchedVisitRef.current = dispatchKey;1934 const dispatch = () => {
1925 void visitThreadMutation({1935 lastDispatchedVisitRef.current = dispatchKey;
1926 environmentId: serverThread.environmentId,1936 lastVisitDispatchAtRef.current = Date.now();
1927 input: { threadId: serverThread.id, visitedAt: serverThread.updatedAt },1937 void visitThreadMutation({
1928 });1938 environmentId: serverThread.environmentId,
1929 return;1939 input: { threadId: serverThread.id, visitedAt: serverThread.updatedAt },
1940 });
1941 };
1942 // Unread prominence only flips on run completions (hasUnseenCompletion),
1943 // so an unseen completion publishes immediately; mid-turn activity bumps
1944 // — several per second while a turn streams — ride a trailing throttle,
1945 // each rerun swapping the timer so the trailing dispatch carries the
1946 // newest watermark.
1947 const latestRunCompletedAtMs = serverThread.latestRun?.completedAt
1948 ? Date.parse(serverThread.latestRun.completedAt)
1949 : NaN;
1950 const hasUnseenCompletion =
1951 !Number.isNaN(latestRunCompletedAtMs) &&
1952 (Number.isNaN(lastVisitedAt) || latestRunCompletedAtMs > lastVisitedAt);
1953 const elapsed = Date.now() - lastVisitDispatchAtRef.current;
1954 if (hasUnseenCompletion || elapsed >= VISIT_DISPATCH_THROTTLE_MS) {
1955 dispatch();
1956 return;
1957 }
1958 const timer = setTimeout(dispatch, VISIT_DISPATCH_THROTTLE_MS - elapsed);
1959 return () => clearTimeout(timer);
1930 }1960 }
1931 1961
1932 markThreadVisited(1962 markThreadVisited(
@@ -1940,6 +1970,7 @@ function ChatViewContent(props: ChatViewProps) { · 7 unmodified lines
1940 serverThread?.environmentId,1970 serverThread?.environmentId,
1941 serverThread?.id,1971 serverThread?.id,
1942 serverThread?.lastVisitedAt,1972 serverThread?.lastVisitedAt,
1973 serverThread?.latestRun?.completedAt,
1943 serverThread?.updatedAt,1974 serverThread?.updatedAt,
1944 visitThreadMutation,1975 visitThreadMutation,
1945 ]);1976 ]);
apps/web/src/components/Sidebar.logic.test.ts+25−0
@@ -29,6 +29,7 @@ import { · 28 unmodified lines
29 shouldClearThreadSelectionOnMouseDown,29 shouldClearThreadSelectionOnMouseDown,
30 sortLogicalProjectsForSidebar,30 sortLogicalProjectsForSidebar,
31 sortSidebarV2ProjectGroups,31 sortSidebarV2ProjectGroups,
32 resolveThreadLastVisitedAt,
32 sortSettledThreadsForSidebarV2,33 sortSettledThreadsForSidebarV2,
33 sortThreadsForSidebarV2,34 sortThreadsForSidebarV2,
34 sortProjectsForSidebar,35 sortProjectsForSidebar,
@@ -1613,3 +1614,27 @@ describe("sortSidebarV2ProjectGroups", () => { · 1578 unmodified lines
1613 ).toEqual(["logical-newer", "logical-older"]);1614 ).toEqual(["logical-newer", "logical-older"]);
1614 });1615 });
1615});1616});
1617
1618describe("resolveThreadLastVisitedAt", () => {
1619 it("uses the local watermark when the server does not track visits", () => {
1620 expect(resolveThreadLastVisitedAt(undefined, "2026-07-30T10:00:00.000Z")).toBe(
1621 "2026-07-30T10:00:00.000Z",
1622 );
1623 expect(resolveThreadLastVisitedAt(undefined, undefined)).toBeUndefined();
1624 });
1625
1626 it("keeps the server watermark authoritative when visited tracking exists", () => {
1627 // A rewound server value (mark-unread) must win even over a newer local
1628 // watermark left behind by earlier viewing on this device.
1629 expect(resolveThreadLastVisitedAt("2026-07-30T10:00:00.000Z", "2026-07-30T10:00:05.000Z")).toBe(
1630 "2026-07-30T10:00:00.000Z",
1631 );
1632 expect(resolveThreadLastVisitedAt("2026-07-30T10:00:00.000Z", undefined)).toBe(
1633 "2026-07-30T10:00:00.000Z",
1634 );
1635 });
1636
1637 it("treats an explicit server-side null as never visited", () => {
1638 expect(resolveThreadLastVisitedAt(null, "2026-07-30T10:00:00.000Z")).toBeUndefined();
1639 });
1640});
apps/web/src/components/Sidebar.logic.ts+3−0
@@ -213,6 +213,9 @@ export function resolveThreadLastVisitedAt( · 212 unmodified lines
213 serverLastVisitedAt: string | null | undefined,213 serverLastVisitedAt: string | null | undefined,
214 localLastVisitedAt: string | undefined,214 localLastVisitedAt: string | undefined,
215): string | undefined {215): string | undefined {
216 // When the server tracks visits it is authoritative — including explicit
217 // rewinds from mark-unread, which a newer browser-local watermark must not
218 // mask. The local value only carries servers without visited tracking.
216 if (serverLastVisitedAt === undefined) return localLastVisitedAt;219 if (serverLastVisitedAt === undefined) return localLastVisitedAt;
217 return serverLastVisitedAt ?? undefined;220 return serverLastVisitedAt ?? undefined;
218}221}