Active turn guardrails — block trim commands during generation

apps/server/src/orchestration/commandInvariants.ts

@@ -219,6 +219,21 @@ export function requireThreadAbsent(input: {
   );
 }
 
+export function requireTurnNotActive(input: {
+  readonly thread: OrchestrationThread;
+  readonly command: OrchestrationCommand;
+}): Effect.Effect<void, OrchestrationCommandInvariantError> {
+  if (input.thread.latestTurn?.state === "running") {
+    return Effect.fail(
+      invariantError(
+        input.command.type,
+        `Thread '${input.thread.id}' has an active turn. Wait for generation to complete before trimming context or starting a new thread.`,
+      ),
+    );
+  }
+  return Effect.void;
+}
+
 export function requireNonNegativeInteger(input: {
   readonly commandType: OrchestrationCommand["type"];
   readonly field: string;

apps/server/src/orchestration/decider.archiveAndNew.test.ts

@@ -1,9 +1,12 @@
 import {
+  CheckpointRef,
   CommandId,
   DEFAULT_PROVIDER_INTERACTION_MODE,
+  DEFAULT_RUNTIME_MODE,
   ProjectId,
   ProviderInstanceId,
   ThreadId,
+  TurnId,
   type OrchestrationCommand,
   type OrchestrationReadModel,
 } from "@t3tools/contracts";
@@ -19,6 +22,7 @@ import { createEmptyReadModel, projectEvent } from "./projector.ts";
 const asCommandId = (value: string): CommandId => CommandId.make(value);
 const asProjectId = (value: string): ProjectId => ProjectId.make(value);
 const asThreadId = (value: string): ThreadId => ThreadId.make(value);
+const asTurnId = (value: string): TurnId => TurnId.make(value);
 
 function seedThreadEffect(readModel: OrchestrationReadModel, overrides?: {
   archivedAt?: string | null;
@@ -229,4 +233,145 @@ describe("thread.archive-and-new decider", () => {
 
       assert.strictEqual(result._tag, "Failure");
     }).pipe(Effect.provide(NodeCrypto.layer)));
+
+  it.effect("rejects archive-and-new when thread has an active turn", () =>
+    Effect.gen(function* () {
+      const readModel = yield* seedThreadEffect(
+        createEmptyReadModel("2026-01-01T00:00:00.000Z"),
+      );
+
+      const modelWithActiveTurn = yield* projectEvent(readModel, {
+        sequence: readModel.snapshotSequence + 1,
+        eventId: CommandId.make("evt-session-active") as unknown as never,
+        aggregateKind: "thread",
+        aggregateId: asThreadId("thread-1"),
+        type: "thread.session-set",
+        occurredAt: "2026-01-01T00:05:00.000Z",
+        commandId: asCommandId("cmd-session-active"),
+        causationEventId: null,
+        correlationId: asCommandId("cmd-session-active"),
+        metadata: {},
+        payload: {
+          threadId: asThreadId("thread-1"),
+          session: {
+            threadId: asThreadId("thread-1"),
+            status: "running" as const,
+            providerName: "codex",
+            runtimeMode: DEFAULT_RUNTIME_MODE,
+            activeTurnId: asTurnId("turn-active"),
+            lastError: null,
+            updatedAt: "2026-01-01T00:05:00.000Z",
+          },
+        },
+      } as never);
+
+      const result = yield* Effect.exit(
+        decideOrchestrationCommand({
+          command: {
+            type: "thread.archive-and-new",
+            commandId: asCommandId("cmd-archive-new-active"),
+            threadId: asThreadId("thread-1"),
+            newThreadId: asThreadId("thread-new"),
+            createdAt: "2026-01-02T00:00:00.000Z",
+          } as Extract<OrchestrationCommand, { type: "thread.archive-and-new" }>,
+          readModel: modelWithActiveTurn,
+        }),
+      );
+
+      assert.strictEqual(result._tag, "Failure");
+    }).pipe(Effect.provide(NodeCrypto.layer)));
+
+  it.effect("accepts archive-and-new when thread has a completed turn", () =>
+    Effect.gen(function* () {
+      const readModel = yield* seedThreadEffect(
+        createEmptyReadModel("2026-01-01T00:00:00.000Z"),
+      );
+
+      let model = yield* projectEvent(readModel, {
+        sequence: readModel.snapshotSequence + 1,
+        eventId: CommandId.make("evt-session-running") as unknown as never,
+        aggregateKind: "thread",
+        aggregateId: asThreadId("thread-1"),
+        type: "thread.session-set",
+        occurredAt: "2026-01-01T00:01:00.000Z",
+        commandId: asCommandId("cmd-session-running"),
+        causationEventId: null,
+        correlationId: asCommandId("cmd-session-running"),
+        metadata: {},
+        payload: {
+          threadId: asThreadId("thread-1"),
+          session: {
+            threadId: asThreadId("thread-1"),
+            status: "running" as const,
+            providerName: "codex",
+            runtimeMode: DEFAULT_RUNTIME_MODE,
+            activeTurnId: asTurnId("turn-done"),
+            lastError: null,
+            updatedAt: "2026-01-01T00:01:00.000Z",
+          },
+        },
+      } as never);
+
+      model = yield* projectEvent(model, {
+        sequence: readModel.snapshotSequence + 2,
+        eventId: CommandId.make("evt-turn-diff-completed") as unknown as never,
+        aggregateKind: "thread",
+        aggregateId: asThreadId("thread-1"),
+        type: "thread.turn-diff-completed",
+        occurredAt: "2026-01-01T00:02:00.000Z",
+        commandId: asCommandId("cmd-turn-diff-completed"),
+        causationEventId: null,
+        correlationId: asCommandId("cmd-turn-diff-completed"),
+        metadata: {},
+        payload: {
+          threadId: asThreadId("thread-1"),
+          turnId: asTurnId("turn-done"),
+          checkpointTurnCount: 1,
+          checkpointRef: CheckpointRef.make("checkpoint-1"),
+          status: "ready" as const,
+          files: [],
+          assistantMessageId: null,
+          completedAt: "2026-01-01T00:02:00.000Z",
+        },
+      } as never);
+
+      const result = yield* decideOrchestrationCommand({
+        command: {
+          type: "thread.archive-and-new",
+          commandId: asCommandId("cmd-archive-new-done"),
+          threadId: asThreadId("thread-1"),
+          newThreadId: asThreadId("thread-new"),
+          createdAt: "2026-01-02T00:00:00.000Z",
+        } as Extract<OrchestrationCommand, { type: "thread.archive-and-new" }>,
+        readModel: model,
+      });
+
+      const events = Array.isArray(result) ? result : [result];
+      assert.strictEqual(events.length, 2);
+      assert.strictEqual(events[0]?.type, "thread.archived-and-new-created");
+      assert.strictEqual(events[1]?.type, "thread.session-stop-requested");
+    }).pipe(Effect.provide(NodeCrypto.layer)));
+
+  it.effect("accepts archive-and-new when thread has no latestTurn (idle)", () =>
+    Effect.gen(function* () {
+      const readModel = yield* seedThreadEffect(
+        createEmptyReadModel("2026-01-01T00:00:00.000Z"),
+      );
+
+      const result = yield* decideOrchestrationCommand({
+        command: {
+          type: "thread.archive-and-new",
+          commandId: asCommandId("cmd-archive-new-idle"),
+          threadId: asThreadId("thread-1"),
+          newThreadId: asThreadId("thread-new"),
+          createdAt: "2026-01-02T00:00:00.000Z",
+        } as Extract<OrchestrationCommand, { type: "thread.archive-and-new" }>,
+        readModel,
+      });
+
+      const events = Array.isArray(result) ? result : [result];
+      assert.strictEqual(events.length, 2);
+      assert.strictEqual(events[0]?.type, "thread.archived-and-new-created");
+      assert.strictEqual(events[1]?.type, "thread.session-stop-requested");
+    }).pipe(Effect.provide(NodeCrypto.layer)));
 });