Integrate upstream remote-agent reliability improvements

AGENTS.md

@@ -33,12 +33,6 @@ If a tradeoff is required, choose correctness and robustness over short-term con
 
 Long term maintainability is a core priority. If you add new functionality, first check if there is shared logic that can be extracted to a separate module. Duplicate logic across multiple files is a code smell and should be avoided. Don't be afraid to change existing code. Don't take shortcuts by just adding local logic to solve a problem.
 
-- After frontend feature development or any user-visible frontend behavior change, the primary agent must run one integrated verification pass for each affected client surface after integrating the work:
-  - Web: use the `test-t3-app` skill. Launch one isolated environment, authenticate through the printed pairing URL, and verify the affected flow in the controlled browser.
-  - Mobile: use the `test-t3-mobile` skill. Connect one representative iOS Simulator or Android Emulator available on the host to one isolated environment and verify the affected flow. On compatible macOS hosts, prefer iOS for cross-platform changes and stream it through serve-sim in the T3 Code in-app browser or another available agent browser; use Android when it is the affected or viable platform.
-  - Subagents must not independently launch dev servers or repeat integrated client verification unless their delegated task explicitly requires it.
-  - Stop dev servers, watchers, and other long-running verification processes when the focused verification is complete.
-
 ## Package Roles
 
 - `apps/server`: Node.js WebSocket server. Wraps Codex app-server (JSON-RPC over stdio), serves the React web app, and manages provider sessions.

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts

@@ -77,6 +77,8 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getThreadActivitiesPage: () =>
+              Effect.die("CheckpointDiffQuery should not request thread activities"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -185,6 +187,8 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getThreadActivitiesPage: () =>
+              Effect.die("CheckpointDiffQuery should not request thread activities"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -268,6 +272,8 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getThreadActivitiesPage: () =>
+              Effect.die("CheckpointDiffQuery should not request thread activities"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -336,6 +342,8 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getThreadActivitiesPage: () =>
+              Effect.die("CheckpointDiffQuery should not request thread activities"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>
@@ -389,6 +397,8 @@ describe("CheckpointDiffQuery.layer", () => {
           Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, {
             getCommandReadModel: () =>
               Effect.die("CheckpointDiffQuery should not request the command read model"),
+            getThreadActivitiesPage: () =>
+              Effect.die("CheckpointDiffQuery should not request thread activities"),
             getSnapshot: () =>
               Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
             getShellSnapshot: () =>

apps/server/src/cli/project.ts

@@ -337,7 +337,9 @@ const dispatchLiveOrchestrationCommand = (
 
 const getOfflineSnapshot = Effect.fn("getOfflineSnapshot")(function* () {
   const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
-  return yield* projectionSnapshotQuery.getSnapshot();
+  // Project resolution only reads `.projects`; the command read model returns the
+  // same shape without loading the heavy per-thread activity/message tables.
+  return yield* projectionSnapshotQuery.getCommandReadModel();
 });
 
 const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecutionMode")(

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -201,6 +201,7 @@ describe("OrchestrationEngine", () => {
           getThreadShellById: () => Effect.succeed(Option.none()),
           getThreadDetailById: () => Effect.succeed(Option.none()),
           getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
+          getThreadActivitiesPage: () => Effect.die("unused"),
         }),
       ),
       Layer.provide(

apps/server/src/orchestration/Layers/OrchestrationEngine.ts

@@ -1,9 +1,4 @@
-import type {
-  OrchestrationEvent,
-  OrchestrationReadModel,
-  ProjectId,
-  ThreadId,
-} from "@t3tools/contracts";
+import type { OrchestrationEvent, ProjectId, ThreadId } from "@t3tools/contracts";
 import { OrchestrationCommand } from "@t3tools/contracts";
 import * as Cause from "effect/Cause";
 import * as Clock from "effect/Clock";
@@ -13,6 +8,7 @@ import * as Deferred from "effect/Deferred";
 import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
+import * as HashMap from "effect/HashMap";
 import * as Layer from "effect/Layer";
 import * as Metric from "effect/Metric";
 import * as Option from "effect/Option";
@@ -37,8 +33,13 @@ import {
   type OrchestrationDispatchError,
   type OrchestrationProjectorDecodeError,
 } from "../Errors.ts";
+import {
+  createEmptyCommandReadModel,
+  fromWireReadModel,
+  type CommandReadModel,
+} from "../commandReadModel.ts";
 import { decideOrchestrationCommand } from "../decider.ts";
-import { createEmptyReadModel, projectEvent } from "../projector.ts";
+import { projectEvent } from "../projector.ts";
 import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
 import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
 import {
@@ -85,15 +86,15 @@ const makeOrchestrationEngine = Effect.gen(function* () {
   const crypto = yield* Crypto.Crypto;
 
   const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
-  let commandReadModel = createEmptyReadModel(yield* nowIso);
+  let commandReadModel: CommandReadModel = createEmptyCommandReadModel(yield* nowIso);
 
   const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
   const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
 
   const projectEventsOntoReadModel = (
-    baseReadModel: OrchestrationReadModel,
+    baseReadModel: CommandReadModel,
     events: ReadonlyArray<OrchestrationEvent>,
-  ): Effect.Effect<OrchestrationReadModel, OrchestrationProjectorDecodeError, never> =>
+  ): Effect.Effect<CommandReadModel, OrchestrationProjectorDecodeError, never> =>
     Effect.gen(function* () {
       let nextReadModel = baseReadModel;
       for (const event of events) {
@@ -298,12 +299,21 @@ const makeOrchestrationEngine = Effect.gen(function* () {
   };
 
   yield* projectionPipeline.bootstrap;
-  commandReadModel = yield* projectionSnapshotQuery.getCommandReadModel();
+  // Seed the in-memory command model from the DB projection. Deleted threads
+  // are dropped so the model starts consistent with the projector's eviction
+  // policy (see commandReadModel.ts / the `thread.deleted` projector branch).
+  commandReadModel = fromWireReadModel(yield* projectionSnapshotQuery.getCommandReadModel(), {
+    dropDeletedThreads: true,
+  });
 
   const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
   yield* Effect.forkScoped(worker);
   yield* Effect.logDebug("orchestration engine started").pipe(
-    Effect.annotateLogs({ sequence: commandReadModel.snapshotSequence }),
+    Effect.annotateLogs({
+      sequence: commandReadModel.snapshotSequence,
+      threadCount: HashMap.size(commandReadModel.threads),
+      projectCount: HashMap.size(commandReadModel.projects),
+    }),
   );
 
   const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -175,6 +175,138 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
   );
 });
 
+it.layer(
+  Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-streaming-shell-summary-")),
+)("OrchestrationProjectionPipeline", (it) => {
+  it.effect("does not rescan thread shell history for streaming assistant messages", () =>
+    Effect.gen(function* () {
+      const projectionPipeline = yield* OrchestrationProjectionPipeline;
+      const eventStore = yield* OrchestrationEventStore;
+      const sql = yield* SqlClient.SqlClient;
+      const threadId = ThreadId.make("thread-streaming-shell-summary");
+      const messageId = MessageId.make("message-streaming-shell-summary");
+      const createdAt = "2026-01-01T00:00:00.000Z";
+      const deltaAt = "2026-01-01T00:00:01.000Z";
+      const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
+        eventStore
+          .append(event)
+          .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
+
+      yield* appendAndProject({
+        type: "project.created",
+        eventId: EventId.make("evt-streaming-shell-summary-1"),
+        aggregateKind: "project",
+        aggregateId: ProjectId.make("project-streaming-shell-summary"),
+        occurredAt: createdAt,
+        commandId: CommandId.make("cmd-streaming-shell-summary-1"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-streaming-shell-summary-1"),
+        metadata: {},
+        payload: {
+          projectId: ProjectId.make("project-streaming-shell-summary"),
+          title: "Streaming shell summary",
+          workspaceRoot: "/tmp/project-streaming-shell-summary",
+          defaultModelSelection: null,
+          scripts: [],
+          createdAt,
+          updatedAt: createdAt,
+        },
+      });
+
+      yield* appendAndProject({
+        type: "thread.created",
+        eventId: EventId.make("evt-streaming-shell-summary-2"),
+        aggregateKind: "thread",
+        aggregateId: threadId,
+        occurredAt: createdAt,
+        commandId: CommandId.make("cmd-streaming-shell-summary-2"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-streaming-shell-summary-2"),
+        metadata: {},
+        payload: {
+          threadId,
+          projectId: ProjectId.make("project-streaming-shell-summary"),
+          title: "Streaming shell summary",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          runtimeMode: "full-access",
+          branch: null,
+          worktreePath: null,
+          createdAt,
+          updatedAt: createdAt,
+        },
+      });
+
+      // A streaming assistant delta must not read activity history. This invalid
+      // sentinel makes any accidental list/decode deterministic and immediately red.
+      yield* sql`
+        INSERT INTO projection_thread_activities (
+          activity_id,
+          thread_id,
+          turn_id,
+          tone,
+          kind,
+          summary,
+          payload_json,
+          sequence,
+          created_at
+        ) VALUES (
+          'activity-streaming-shell-summary',
+          ${threadId},
+          NULL,
+          'info',
+          'test.sentinel',
+          'must not be decoded',
+          '{',
+          NULL,
+          ${createdAt}
+        )
+      `;
+
+      yield* appendAndProject({
+        type: "thread.message-sent",
+        eventId: EventId.make("evt-streaming-shell-summary-3"),
+        aggregateKind: "thread",
+        aggregateId: threadId,
+        occurredAt: deltaAt,
+        commandId: CommandId.make("cmd-streaming-shell-summary-3"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-streaming-shell-summary-3"),
+        metadata: {},
+        payload: {
+          threadId,
+          messageId,
+          role: "assistant",
+          text: "delta",
+          turnId: null,
+          streaming: true,
+          createdAt,
+          updatedAt: deltaAt,
+        },
+      });
+
+      const messageRows = yield* sql<{
+        readonly text: string;
+        readonly isStreaming: number;
+      }>`
+        SELECT text, is_streaming AS "isStreaming"
+        FROM projection_thread_messages
+        WHERE message_id = ${messageId}
+      `;
+      assert.deepEqual(messageRows, [{ text: "delta", isStreaming: 1 }]);
+
+      const threadRows = yield* sql<{ readonly updatedAt: string }>`
+        SELECT updated_at AS "updatedAt"
+        FROM projection_threads
+        WHERE thread_id = ${threadId}
+      `;
+      assert.deepEqual(threadRows, [{ updatedAt: deltaAt }]);
+    }),
+  );
+});
+
 it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-base-")))(
   "OrchestrationProjectionPipeline",
   (it) => {

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -729,7 +729,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
             ...existingRow.value,
             updatedAt: event.occurredAt,
           });
-          yield* refreshThreadShellSummary(event.payload.threadId);
+          if (
+            event.type !== "thread.message-sent" ||
+            event.payload.role !== "assistant" ||
+            !event.payload.streaming
+          ) {
+            yield* refreshThreadShellSummary(event.payload.threadId);
+          }
           return;
         }