perf(orchestration): bound in-memory read model and client per-thread state

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/commandInvariants.test.ts

@@ -11,6 +11,7 @@ import {
 } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 
+import { fromWireReadModel } from "./commandReadModel.ts";
 import {
   findThreadById,
   listThreadsByProjectId,
@@ -21,7 +22,7 @@ import {
 
 const now = "2026-01-01T00:00:00.000Z";
 
-const readModel: OrchestrationReadModel = {
+const wireReadModel: OrchestrationReadModel = {
   snapshotSequence: 2,
   updatedAt: now,
   projects: [
@@ -102,6 +103,8 @@ const readModel: OrchestrationReadModel = {
   ],
 };
 
+const readModel = fromWireReadModel(wireReadModel, { dropDeletedThreads: false });
+
 const messageSendCommand: OrchestrationCommand = {
   type: "thread.turn.start",
   commandId: CommandId.make("cmd-1"),

apps/server/src/orchestration/commandInvariants.ts

@@ -1,46 +1,34 @@
 import type {
   OrchestrationCommand,
   OrchestrationProject,
-  OrchestrationReadModel,
   OrchestrationThread,
   ProjectId,
   ThreadId,
 } from "@t3tools/contracts";
 import { normalizeProjectPathForComparison } from "@t3tools/shared/path";
 import * as Effect from "effect/Effect";
+import * as HashMap from "effect/HashMap";
 
+import {
+  findProjectById,
+  findThreadById,
+  isThreadDeleted,
+  listThreadsByProjectId,
+  type CommandReadModel,
+} from "./commandReadModel.ts";
 import { OrchestrationCommandInvariantError } from "./Errors.ts";
 
+export { findProjectById, findThreadById, listThreadsByProjectId };
+
 function invariantError(commandType: string, detail: string): OrchestrationCommandInvariantError {
   return new OrchestrationCommandInvariantError({
     commandType,
     detail,
   });
 }
 
-export function findThreadById(
-  readModel: OrchestrationReadModel,
-  threadId: ThreadId,
-): OrchestrationThread | undefined {
-  return readModel.threads.find((thread) => thread.id === threadId);
-}
-
-export function findProjectById(
-  readModel: OrchestrationReadModel,
-  projectId: ProjectId,
-): OrchestrationProject | undefined {
-  return readModel.projects.find((project) => project.id === projectId);
-}
-
-export function listThreadsByProjectId(
-  readModel: OrchestrationReadModel,
-  projectId: ProjectId,
-): ReadonlyArray<OrchestrationThread> {
-  return readModel.threads.filter((thread) => thread.projectId === projectId);
-}
-
 export function requireProject(input: {
-  readonly readModel: OrchestrationReadModel;
+  readonly readModel: CommandReadModel;
   readonly command: OrchestrationCommand;
   readonly projectId: ProjectId;
 }): Effect.Effect<OrchestrationProject, OrchestrationCommandInvariantError> {
@@ -57,7 +45,7 @@ export function requireProject(input: {
 }
 
 export function requireProjectAbsent(input: {
-  readonly readModel: OrchestrationReadModel;
+  readonly readModel: CommandReadModel;
   readonly command: OrchestrationCommand;
   readonly projectId: ProjectId;
 }): Effect.Effect<void, OrchestrationCommandInvariantError> {
@@ -73,18 +61,23 @@ export function requireProjectAbsent(input: {
 }
 
 export function requireActiveProjectWorkspaceRootAbsent(input: {
-  readonly readModel: OrchestrationReadModel;
+  readonly readModel: CommandReadModel;
   readonly command: OrchestrationCommand;
   readonly workspaceRoot: string;
   readonly exceptProjectId?: ProjectId;
 }): Effect.Effect<void, OrchestrationCommandInvariantError> {
   const normalizedWorkspaceRoot = normalizeProjectPathForComparison(input.workspaceRoot);
-  const existingProject = input.readModel.projects.find(
-    (project) =>
+  let existingProject: OrchestrationProject | undefined;
+  for (const project of HashMap.values(input.readModel.projects)) {
+    if (
       project.deletedAt === null &&
       normalizeProjectPathForComparison(project.workspaceRoot) === normalizedWorkspaceRoot &&
-      project.id !== input.exceptProjectId,
-  );
+      project.id !== input.exceptProjectId
+    ) {
+      existingProject = project;
+      break;
+    }
+  }
   if (existingProject === undefined) {
     return Effect.void;
   }
@@ -97,7 +90,7 @@ export function requireActiveProjectWorkspaceRootAbsent(input: {
 }
 
 export function requireThread(input: {
-  readonly readModel: OrchestrationReadModel;
+  readonly readModel: CommandReadModel;
   readonly command: OrchestrationCommand;
   readonly threadId: ThreadId;
 }): Effect.Effect<OrchestrationThread, OrchestrationCommandInvariantError> {
@@ -114,7 +107,7 @@ export function requireThread(input: {
 }
 
 export function requireThreadArchived(input: {
-  readonly readModel: OrchestrationReadModel;
+  readonly readModel: CommandReadModel;
   readonly command: OrchestrationCommand;
   readonly threadId: ThreadId;
 }): Effect.Effect<OrchestrationThread, OrchestrationCommandInvariantError> {
@@ -133,7 +126,7 @@ export function requireThreadArchived(input: {
 }
 
 export function requireThreadNotArchived(input: {
-  readonly readModel: OrchestrationReadModel;
+  readonly readModel: CommandReadModel;
   readonly command: OrchestrationCommand;
   readonly threadId: ThreadId;
 }): Effect.Effect<OrchestrationThread, OrchestrationCommandInvariantError> {
@@ -152,11 +145,16 @@ export function requireThreadNotArchived(input: {
 }
 
 export function requireThreadAbsent(input: {
-  readonly readModel: OrchestrationReadModel;
+  readonly readModel: CommandReadModel;
   readonly command: OrchestrationCommand;
   readonly threadId: ThreadId;
 }): Effect.Effect<void, OrchestrationCommandInvariantError> {
-  if (!findThreadById(input.readModel, input.threadId)) {
+  // A deleted thread is evicted from `threads` but its id is retained in
+  // `deletedThreadIds`, so reject re-using a live OR previously-deleted id.
+  if (
+    !findThreadById(input.readModel, input.threadId) &&
+    !isThreadDeleted(input.readModel, input.threadId)
+  ) {
     return Effect.void;
   }
   return Effect.fail(

apps/server/src/orchestration/commandReadModel.test.ts

@@ -0,0 +1,130 @@
+import {
+  DEFAULT_PROVIDER_INTERACTION_MODE,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  type OrchestrationReadModel,
+  type OrchestrationThread,
+} from "@t3tools/contracts";
+import * as HashMap from "effect/HashMap";
+import { describe, expect, it } from "vite-plus/test";
+
+import {
+  createEmptyCommandReadModel,
+  findProjectById,
+  findThreadById,
+  fromWireReadModel,
+  isThreadDeleted,
+  listThreadsByProjectId,
+} from "./commandReadModel.ts";
+
+const now = "2026-01-01T00:00:00.000Z";
+
+function makeThread(
+  id: string,
+  projectId: string,
+  overrides?: Partial<OrchestrationThread>,
+): OrchestrationThread {
+  return {
+    id: ThreadId.make(id),
+    projectId: ProjectId.make(projectId),
+    title: id,
+    modelSelection: {
+      instanceId: ProviderInstanceId.make("codex"),
+      model: "gpt-5-codex",
+    },
+    interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
+    runtimeMode: "full-access",
+    branch: null,
+    worktreePath: null,
+    createdAt: now,
+    updatedAt: now,
+    archivedAt: null,
+    latestTurn: null,
+    messages: [],
+    session: null,
+    activities: [],
+    proposedPlans: [],
+    checkpoints: [],
+    deletedAt: null,
+    ...overrides,
+  };
+}
+
+const wireReadModel: OrchestrationReadModel = {
+  snapshotSequence: 5,
+  updatedAt: now,
+  projects: [
+    {
+      id: ProjectId.make("project-a"),
+      title: "Project A",
+      workspaceRoot: "/tmp/project-a",
+      defaultModelSelection: {
+        instanceId: ProviderInstanceId.make("codex"),
+        model: "gpt-5-codex",
+      },
+      scripts: [],
+      createdAt: now,
+      updatedAt: now,
+      deletedAt: null,
+    },
+  ],
+  threads: [
+    makeThread("thread-live", "project-a"),
+    makeThread("thread-archived", "project-a", { archivedAt: now }),
+    makeThread("thread-deleted", "project-a", { deletedAt: now }),
+    makeThread("thread-other", "project-b"),
+  ],
+};
+
+describe("commandReadModel", () => {
+  it("creates an empty model", () => {
+    const model = createEmptyCommandReadModel(now);
+    expect(model.snapshotSequence).toBe(0);
+    expect(HashMap.size(model.threads)).toBe(0);
+    expect(HashMap.size(model.projects)).toBe(0);
+    expect(model.updatedAt).toBe(now);
+  });
+
+  it("seeds from the wire model and drops deleted threads by default", () => {
+    const model = fromWireReadModel(wireReadModel);
+    expect(model.snapshotSequence).toBe(5);
+    // deleted thread is evicted; archived + live + other retained
+    expect(HashMap.size(model.threads)).toBe(3);
+    expect(HashMap.has(model.threads, ThreadId.make("thread-deleted"))).toBe(false);
+    expect(HashMap.has(model.threads, ThreadId.make("thread-archived"))).toBe(true);
+    expect(HashMap.has(model.threads, ThreadId.make("thread-live"))).toBe(true);
+    expect(HashMap.size(model.projects)).toBe(1);
+    // The evicted deleted thread's id is retained so the create-twice invariant
+    // survives a restart.
+    expect(isThreadDeleted(model, ThreadId.make("thread-deleted"))).toBe(true);
+    expect(isThreadDeleted(model, ThreadId.make("thread-live"))).toBe(false);
+    expect(isThreadDeleted(model, ThreadId.make("thread-archived"))).toBe(false);
+  });
+
+  it("retains deleted threads when dropDeletedThreads is false", () => {
+    const model = fromWireReadModel(wireReadModel, { dropDeletedThreads: false });
+    expect(HashMap.size(model.threads)).toBe(4);
+    expect(HashMap.has(model.threads, ThreadId.make("thread-deleted"))).toBe(true);
+  });
+
+  it("finds threads and projects by id with O(1) lookups", () => {
+    const model = fromWireReadModel(wireReadModel);
+    expect(findThreadById(model, ThreadId.make("thread-live"))?.projectId).toBe("project-a");
+    expect(findThreadById(model, ThreadId.make("thread-deleted"))).toBeUndefined();
+    expect(findThreadById(model, ThreadId.make("missing"))).toBeUndefined();
+    expect(findProjectById(model, ProjectId.make("project-a"))?.title).toBe("Project A");
+    expect(findProjectById(model, ProjectId.make("missing"))).toBeUndefined();
+  });
+
+  it("lists threads by project id", () => {
+    const model = fromWireReadModel(wireReadModel);
+    const ids = listThreadsByProjectId(model, ProjectId.make("project-a"))
+      .map((thread) => thread.id)
+      .toSorted();
+    expect(ids).toEqual([ThreadId.make("thread-archived"), ThreadId.make("thread-live")]);
+    expect(listThreadsByProjectId(model, ProjectId.make("project-b")).map((t) => t.id)).toEqual([
+      ThreadId.make("thread-other"),
+    ]);
+  });
+});