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

apps/server/src/orchestration-v2/FoundationPersistence.test.ts

@@ -265,6 +265,54 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => {
     }),
   );
 
+  it.effect("compacts superseded thread.visited events, keeping the newest per thread", () =>
+    Effect.gen(function* () {
+      const eventSink = yield* EventSinkV2;
+      const maintenance = yield* ProjectionMaintenanceV2;
+      const sql = yield* SqlClient.SqlClient;
+      const now = yield* DateTime.now;
+      const threadId = ThreadId.make("thread:foundation-compact-visits");
+      const thread = makeThread(threadId, now);
+      const visitedEvent = (suffix: string): OrchestrationV2DomainEvent => ({
+        id: EventId.make(`event:foundation-compact-visits:${suffix}`),
+        type: "thread.visited",
+        threadId,
+        providerInstanceId,
+        occurredAt: now,
+        payload: { ...thread, lastVisitedAt: now },
+      });
+
+      yield* eventSink.write({
+        events: [
+          threadCreatedEvent({ id: "event:foundation-compact-visits:create", thread, now }),
+          visitedEvent("first"),
+          visitedEvent("second"),
+          visitedEvent("third"),
+        ],
+      });
+
+      const summary = yield* maintenance.compactReadStateEvents;
+      assert.isAtLeast(summary.deletedEventCount, 2);
+
+      const remaining = yield* sql<{ readonly event_id: string }>`
+        SELECT event_id
+        FROM orchestration_events
+        WHERE aggregate_kind = 'thread'
+          AND stream_id = ${threadId}
+          AND event_type = 'thread.visited'
+        ORDER BY sequence ASC
+      `;
+      assert.deepEqual(
+        remaining.map((row) => row.event_id),
+        ["event:foundation-compact-visits:third"],
+      );
+
+      // Replay across the deletion gap must still produce a valid projection.
+      assert.isTrue((yield* maintenance.verify).valid);
+      assert.isTrue((yield* maintenance.rebuild).valid);
+    }),
+  );
+
   it.effect("verifies and rebuilds projections with cross-thread subagent relations", () =>
     Effect.gen(function* () {
       const eventSink = yield* EventSinkV2;

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

@@ -488,6 +488,13 @@ const make = Effect.gen(function* () {
     ),
   );
 
+  // Threads whose transcript import this process has already confirmed.
+  // `transcript_imported_at` is never reset to NULL, so a positive answer
+  // stays valid for the process lifetime; ensureTranscript runs on most
+  // thread reads and command dispatches, so skipping the lock + lookup here
+  // keeps that path off the database entirely after first confirmation.
+  const confirmedTranscriptThreadIds = new Set<ThreadId>();
+
   const ensureTranscriptBase = (threadId: ThreadId) =>
     transcriptImports.withLock(
       threadId,
@@ -500,6 +507,9 @@ const make = Effect.gen(function* () {
         `;
         const imported = imports[0];
         if (imported === undefined || imported.transcript_imported_at !== null) {
+          if (imported !== undefined) {
+            confirmedTranscriptThreadIds.add(threadId);
+          }
           return { importedThreadCount: 0, importedMessageCount: 0 };
         }
         const messages = yield* listMessages(threadId);
@@ -546,6 +556,7 @@ const make = Effect.gen(function* () {
             last_error = NULL
           WHERE thread_id = ${threadId}
         `;
+        confirmedTranscriptThreadIds.add(threadId);
         return {
           importedThreadCount: 1,
           importedMessageCount: missing.length,
@@ -554,16 +565,18 @@ const make = Effect.gen(function* () {
     );
 
   const ensureTranscript = (threadId: ThreadId) =>
-    ensureTranscriptBase(threadId).pipe(
-      Effect.mapError(
-        (cause) =>
-          new LegacyV1ThreadImportError({
-            operation: "hydrate transcript for",
-            threadId,
-            cause,
-          }),
-      ),
-    );
+    confirmedTranscriptThreadIds.has(threadId)
+      ? Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 })
+      : ensureTranscriptBase(threadId).pipe(
+          Effect.mapError(
+            (cause) =>
+              new LegacyV1ThreadImportError({
+                operation: "hydrate transcript for",
+                threadId,
+                cause,
+              }),
+          ),
+        );
 
   const importPendingTranscripts = Effect.gen(function* () {
     const rows = yield* sql<LegacyImportRow>`

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

@@ -15,6 +15,7 @@ import {
   type OrchestrationV2ProviderTurn,
   type OrchestrationV2Run,
   type OrchestrationV2RunAttempt,
+  type OrchestrationV2ThreadShell,
   type OrchestrationV2ThreadShellSnapshot,
   type OrchestrationV2StoredEvent,
   type OrchestrationV2Subagent,
@@ -160,6 +161,9 @@ export interface OrchestratorV2Shape {
     OrchestrationV2ThreadShellSnapshot,
     OrchestratorV2Error
   >;
+  readonly getThreadShell: (
+    threadId: ThreadId,
+  ) => Effect.Effect<OrchestrationV2ThreadShell | null, OrchestratorV2Error>;
   readonly getThreadEventSequence: (
     threadId: ThreadId,
   ) => Effect.Effect<number, OrchestratorV2Error>;
@@ -5999,6 +6003,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
             }),
         ),
       ),
+    getThreadShell: (threadId) =>
+      projectionStore
+        .getThreadShell(threadId)
+        .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))),
     getThreadEventSequence: (threadId) =>
       eventSink
         .latestSequence({ threadId })
@@ -6088,6 +6096,13 @@ export const layerUnavailable: Layer.Layer<OrchestratorV2> = Layer.succeed(
           cause: "Orchestration V2 live runtime is not configured.",
         }),
       ),
+    getThreadShell: (threadId) =>
+      Effect.fail(
+        new OrchestratorProjectionError({
+          threadId,
+          cause: "Orchestration V2 live runtime is not configured.",
+        }),
+      ),
     getThreadEventSequence: (threadId) =>
       Effect.fail(
         new OrchestratorProjectionError({

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

@@ -34,6 +34,10 @@ export class ProjectionMaintenanceError extends Schema.TaggedErrorClass<Projecti
 export interface ProjectionMaintenanceV2Shape {
   readonly verify: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;
   readonly rebuild: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;
+  readonly compactReadStateEvents: Effect.Effect<
+    { readonly deletedEventCount: number },
+    ProjectionMaintenanceError
+  >;
 }
 
 export class ProjectionMaintenanceV2 extends Context.Service<
@@ -210,9 +214,37 @@ export const layer: Layer.Layer<
           Effect.mapError((cause) => new ProjectionMaintenanceError({ operation, cause })),
         );
 
+    /**
+     * thread.visited events carry the full thread payload, so only the newest
+     * one per thread can influence a replay or an afterSequence catch-up; the
+     * rest are dead weight. Visits fire on every activity bump while a thread
+     * is open, so without compaction they dominate the event store.
+     */
+    const compactReadStateEvents = sql.withTransaction(
+      Effect.gen(function* () {
+        yield* sql`
+          DELETE FROM orchestration_events
+          WHERE application_event_version = 2
+            AND aggregate_kind = 'thread'
+            AND event_type = 'thread.visited'
+            AND sequence NOT IN (
+              SELECT MAX(sequence)
+              FROM orchestration_events
+              WHERE application_event_version = 2
+                AND aggregate_kind = 'thread'
+                AND event_type = 'thread.visited'
+              GROUP BY stream_id
+            )
+        `;
+        const rows = yield* sql<{ readonly deleted: number }>`SELECT changes() AS deleted`;
+        return { deletedEventCount: rows[0]?.deleted ?? 0 };
+      }),
+    );
+
     return ProjectionMaintenanceV2.of({
       verify: mapError("verify")(verify),
       rebuild: mapError("rebuild")(rebuild),
+      compactReadStateEvents: mapError("compact read-state events")(compactReadStateEvents),
     });
   }),
 );