perf(server): compact visit and pull request refresh events

apps/server/integration/transferBudgetV2.integration.test.ts

@@ -42,6 +42,10 @@ import * as OrchestrationEventStore from "../src/persistence/OrchestrationEventS
 import * as EventStore from "../src/orchestration-v2/EventStore.ts";
 import * as EventSink from "../src/orchestration-v2/EventSink.ts";
 import * as ProjectionStore from "../src/orchestration-v2/ProjectionStore.ts";
+import * as Orchestrator from "../src/orchestration-v2/Orchestrator.ts";
+import * as ProviderSessionManager from "../src/orchestration-v2/ProviderSessionManager.ts";
+import * as CommandReceiptStore from "../src/orchestration-v2/CommandReceiptStore.ts";
+import * as ServerSettings from "../src/serverSettings.ts";
 import * as ThreadManagementService from "../src/orchestration-v2/ThreadManagementService.ts";
 import * as ProjectStore from "../src/orchestration-v2/ProjectStore.ts";
 import * as ProjectService from "../src/project/ProjectService.ts";
@@ -81,18 +85,26 @@ const layerManagement = Layer.unwrap(
   Effect.gen(function* () {
     const projections = yield* ProjectionStore.ProjectionStoreV2;
     const sink = yield* EventSink.EventSinkV2;
-    return Layer.mock(ThreadManagementService.ThreadManagementService)({
-      ensureLegacyTranscript: () => Effect.void,
-      getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie),
-      getThreadSnapshotWindow: (id, options) =>
-        projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
-      getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie),
-      getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie),
-      readShellSnapshot: (options) =>
-        projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie),
-      streamStoredEventsFrom: (input) =>
-        sink.stream({ ...input, bounded: true }).pipe(Stream.orDie),
-    });
+    return ThreadManagementService.layer.pipe(
+      Layer.provide(
+        Layer.mergeAll(
+          Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}),
+          Layer.mock(CommandReceiptStore.CommandReceiptStoreV2)({}),
+          Layer.mock(ServerSettings.ServerSettingsService)({}),
+          Layer.mock(Orchestrator.OrchestratorV2)({
+            getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie),
+            getThreadSnapshotWindow: (id, options) =>
+              projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
+            getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie),
+            getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie),
+            readShellSnapshot: (options) =>
+              projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie),
+            streamStoredEventsFrom: (input) =>
+              sink.stream({ ...input, bounded: true }).pipe(Stream.orDie),
+          }),
+        ),
+      ),
+    );
   }),
 );
 const layerEnrichment = Layer.unwrap(

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -2366,11 +2366,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       events,
       command,
     )({
-      type: "thread.visited",
+      type: "thread.visit-recorded",
       threadId: command.threadId,
       providerInstanceId: thread.providerInstanceId,
       occurredAt: yield* DateTime.now,
-      payload: movesForward ? { ...thread, lastVisitedAt: visitedAt.value } : thread,
+      payload: { lastVisitedAt: movesForward ? visitedAt.value : thread.lastVisitedAt },
     });
   });
 
@@ -3282,11 +3282,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         events,
         command,
       )({
-        type: "thread.pull-request-link-updated",
+        ...(command.type === "thread.pull-request-link.sync"
+          ? {
+              type: "thread.pull-request-link-synced" as const,
+              payload: {
+                ...normalizeThreadPullRequestKey(command),
+                snapshot: command.snapshot,
+                stack: command.stack,
+              },
+            }
+          : { type: "thread.pull-request-link-updated" as const, payload: link }),
         threadId: command.threadId,
         providerInstanceId: updatedThread.providerInstanceId,
         occurredAt: now,
-        payload: link,
       });
       return;
     }

apps/server/src/orchestration-v2/ProjectionMaintenance.ts

@@ -217,11 +217,14 @@ export const layer: Layer.Layer<
       "thread.model-selection-updated",
       "thread.provider-switched",
     ];
-    // Read-state events carry the whole thread too, but client reducers apply
-    // them without bumping updatedAt. They supersede only each other: a newer
+    // Read-state events do not bump updatedAt. They supersede only each other: a newer
     // visit must not delete the rename that carries the activity timestamp,
     // while any newer activity-bearing event supersedes both groups.
-    const READ_STATE_THREAD_EVENT_TYPES = ["thread.visited", "thread.marked-unread"];
+    const READ_STATE_THREAD_EVENT_TYPES = [
+      "thread.visited",
+      "thread.marked-unread",
+      "thread.visit-recorded",
+    ];
 
     const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES);
     const readStateThreadEventTypes = new Set(READ_STATE_THREAD_EVENT_TYPES);
@@ -280,7 +283,7 @@ export const layer: Layer.Layer<
                 AND event.event_type IN ('message.updated', 'node.updated')
               THEN json_extract(event.payload_json, '$.id')
               WHEN event.application_event_version = 2
-                AND event.event_type = 'thread.pull-request-link-updated'
+                AND event.event_type IN ('thread.pull-request-link-updated', 'thread.pull-request-link-synced')
               THEN json_array(json_extract(event.payload_json, '$.host'),
                 json_extract(event.payload_json, '$.repository'),
                 json_extract(event.payload_json, '$.number'))
@@ -322,13 +325,15 @@ export const layer: Layer.Layer<
                 obsolete.push(row.sequence);
               } else retainedReadStateThreadIds.add(row.stream_id);
             } else if (
-              row.event_type === "thread.pull-request-link-updated" &&
+              (row.event_type === "thread.pull-request-link-updated" ||
+                row.event_type === "thread.pull-request-link-synced") &&
               retainedThreadIds.has(row.stream_id)
             ) {
               // A newer whole-thread payload already carries this link's state.
               obsolete.push(row.sequence);
             } else if (
               row.event_type === "thread.pull-request-link-updated" ||
+              row.event_type === "thread.pull-request-link-synced" ||
               row.event_type === "message.updated" ||
               row.event_type === "node.updated"
             ) {

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -13,7 +13,10 @@ import {
   threadErrorSummary,
   usageLimitRunPresentedAsLatest,
 } from "@t3tools/shared/orchestrationV2ThreadError";
-import { threadPullRequestsOf } from "@t3tools/shared/threadPullRequests";
+import {
+  applyThreadPullRequestLinkSync,
+  threadPullRequestsOf,
+} from "@t3tools/shared/threadPullRequests";
 import type {
   OrchestrationV2AppThread,
   OrchestrationV2CheckpointScope,
@@ -758,12 +761,23 @@ export function applyToProjection(
         ...base,
         thread: event.payload,
       };
+    // A host refresh is not activity, matching the stored projection.
+    case "thread.pull-request-link-synced":
+      return {
+        ...projection,
+        thread: applyThreadPullRequestLinkSync(projection.thread, event.payload),
+      };
     case "thread.visited":
     case "thread.marked-unread":
       return {
         ...projection,
         thread: event.payload,
       };
+    case "thread.visit-recorded":
+      return {
+        ...projection,
+        thread: { ...projection.thread, lastVisitedAt: event.payload.lastVisitedAt },
+      };
     case "run.created":
     case "run.updated":
       return withLocalVisibleTurnItems({
@@ -1930,6 +1944,28 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
             `;
             break;
           }
+          case "thread.visit-recorded": {
+            const visitedAt = event.payload.lastVisitedAt;
+            yield* sql`
+              UPDATE orchestration_v2_projection_threads
+              SET payload_json = json_set(payload_json, '$.lastVisitedAt',
+                ${visitedAt === null ? null : DateTime.formatIso(visitedAt)})
+              WHERE thread_id = ${event.threadId}
+            `;
+            break;
+          }
+          case "thread.pull-request-link-synced": {
+            const thread = yield* getThread(event.threadId);
+            const updated = applyThreadPullRequestLinkSync(thread, event.payload);
+            if (updated === thread) break;
+            const payloadJson = yield* encodeThreadPayload(updated);
+            yield* sql`
+              UPDATE orchestration_v2_projection_threads
+              SET payload_json = ${payloadJson}
+              WHERE thread_id = ${event.threadId}
+            `;
+            break;
+          }
           case "run.created":
           case "run.updated": {
             const payloadJson = yield* encodeRunPayload(event.payload);
@@ -2704,6 +2740,9 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           event.type !== "thread.unpinned" &&
           event.type !== "thread.pin-reordered" &&
           event.type !== "thread.visited" &&
+          event.type !== "thread.visit-recorded" &&
+          event.type !== "thread.pull-request-link-synced" &&
+          event.type !== "thread.pull-request-synced" &&
           event.type !== "thread.marked-unread" &&
           event.type !== "thread.metadata-updated" &&
           event.type !== "thread.runtime-mode-updated" &&

apps/server/src/orchestration-v2/PullRequestSyncReactor.ts

@@ -433,6 +433,8 @@ export const make = Effect.gen(function* () {
         switch (event.type) {
           case "thread.pull-request-link-updated":
             return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void;
+          case "thread.pull-request-link-synced":
+            return Effect.void;
           case "thread.pull-request-synced":
             return Effect.forEach(
               visibleThreadPullRequests(event.payload.pullRequests ?? []).filter(