fix(orchestrator): keep stop and steer responsive

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -15,6 +15,7 @@ import { assert, describe, it } from "@effect/vitest";
 import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess";
 import { SpawnExecutableResolution } from "@t3tools/shared/shell";
 import * as DateTime from "effect/DateTime";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Ref from "effect/Ref";
 import { TestClock } from "effect/testing";
@@ -31,7 +32,9 @@ import {
   makeCodexAppServerProtocolLogger,
   makeCodexAppServerSpawnCommand,
   projectCodexDynamicToolItem,
+  registerCodexTurnTerminalWaiter,
   resolveCodexRollbackTurnCount,
+  settleCodexTurnTerminal,
 } from "./CodexAdapterV2.ts";
 
 describe("CodexAdapterV2 assistant message streaming", () => {
@@ -434,6 +437,103 @@ describe("CodexAdapterV2 native protocol logging", () => {
   });
 });
 
+describe("CodexAdapterV2 interrupt terminal tracking", () => {
+  const makeTracking = Effect.gen(function* () {
+    const activeTurns = yield* Ref.make(new Map([["turn-1", "context"]]));
+    const turnWaiters = yield* Ref.make(
+      new Map<string, ReadonlySet<Deferred.Deferred<void, never>>>(),
+    );
+    return { activeTurns, turnWaiters };
+  });
+
+  it.effect("signals a first-time terminal waiter once the turn settles", () =>
+    Effect.gen(function* () {
+      const { activeTurns, turnWaiters } = yield* makeTracking;
+
+      const waiter = yield* registerCodexTurnTerminalWaiter({
+        activeTurns,
+        turnWaiters,
+        nativeTurnId: "turn-1",
+      });
+
+      assert.isFalse(waiter.alreadyTerminal);
+      assert.equal((yield* Ref.get(turnWaiters)).get("turn-1")?.size, 1);
+
+      yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" });
+      yield* waiter.awaitTerminal;
+
+      assert.isFalse((yield* Ref.get(activeTurns)).has("turn-1"));
+      assert.isUndefined((yield* Ref.get(turnWaiters)).get("turn-1"));
+    }),
+  );
+
+  it.effect("reports already-terminal when completion lands before registration", () =>
+    Effect.gen(function* () {
+      const { activeTurns, turnWaiters } = yield* makeTracking;
+
+      // turn/completed can run on another fiber between interruptTurn's
+      // active-turn lookup and its waiter registration; the registration
+      // re-check must detect the settled turn instead of waiting forever.
+      yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" });
+      const waiter = yield* registerCodexTurnTerminalWaiter({
+        activeTurns,
+        turnWaiters,
+        nativeTurnId: "turn-1",
+      });
+
+      assert.isTrue(waiter.alreadyTerminal);
+      yield* waiter.unregister;
+      assert.isUndefined((yield* Ref.get(turnWaiters)).get("turn-1"));
+    }),
+  );
+
+  it.effect("signals every concurrently registered waiter for the same turn", () =>
+    Effect.gen(function* () {
+      const { activeTurns, turnWaiters } = yield* makeTracking;
+
+      const first = yield* registerCodexTurnTerminalWaiter({
+        activeTurns,
+        turnWaiters,
+        nativeTurnId: "turn-1",
+      });
+      const second = yield* registerCodexTurnTerminalWaiter({
+        activeTurns,
+        turnWaiters,
+        nativeTurnId: "turn-1",
+      });
+
+      assert.equal((yield* Ref.get(turnWaiters)).get("turn-1")?.size, 2);
+
+      yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" });
+      yield* first.awaitTerminal;
+      yield* second.awaitTerminal;
+    }),
+  );
+
+  it.effect("unregistering one waiter keeps the remaining waiters registered", () =>
+    Effect.gen(function* () {
+      const { activeTurns, turnWaiters } = yield* makeTracking;
+
+      const abandoned = yield* registerCodexTurnTerminalWaiter({
+        activeTurns,
+        turnWaiters,
+        nativeTurnId: "turn-1",
+      });
+      const kept = yield* registerCodexTurnTerminalWaiter({
+        activeTurns,
+        turnWaiters,
+        nativeTurnId: "turn-1",
+      });
+
+      yield* abandoned.unregister;
+      assert.equal((yield* Ref.get(turnWaiters)).get("turn-1")?.size, 1);
+
+      yield* settleCodexTurnTerminal({ activeTurns, turnWaiters, nativeTurnId: "turn-1" });
+      yield* kept.awaitTerminal;
+    }),
+  );
+});
+
 describe("CodexAdapterV2 rollback mapping", () => {
   it.effect("derives native rollback count from durable provider turns", () =>
     Effect.gen(function* () {

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -35,6 +35,7 @@ import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
 import * as Path from "effect/Path";
 import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
@@ -1230,6 +1231,87 @@ export const layer: Layer.Layer<
   }),
 );
 
+export interface CodexTurnTerminalTrackingOptions<Context> {
+  readonly activeTurns: Ref.Ref<Map<string, Context>>;
+  readonly turnWaiters: Ref.Ref<Map<string, ReadonlySet<Deferred.Deferred<void, never>>>>;
+  readonly nativeTurnId: string;
+}
+
+export interface CodexTurnTerminalWaiter {
+  /** Resolves once the turn reaches a terminal state. */
+  readonly awaitTerminal: Effect.Effect<void>;
+  readonly unregister: Effect.Effect<void>;
+  /**
+   * The turn was already removed from the active set when the waiter was
+   * registered, so no terminal signal will arrive for it.
+   */
+  readonly alreadyTerminal: boolean;
+}
+
+/**
+ * Registers a terminal waiter for a native turn, then re-checks the active
+ * set. Because {@link settleCodexTurnTerminal} removes the active turn before
+ * draining waiters, a turn still active after registration is guaranteed to
+ * signal this waiter; a missing turn already reached a terminal state.
+ */
+export const registerCodexTurnTerminalWaiter = <Context>(
+  options: CodexTurnTerminalTrackingOptions<Context>,
+): Effect.Effect<CodexTurnTerminalWaiter> =>
+  Effect.gen(function* () {
+    const terminal = yield* Deferred.make<void>();
+    yield* Ref.update(options.turnWaiters, (current) => {
+      const updated = new Map(current);
+      const waiters = new Set(updated.get(options.nativeTurnId) ?? []);
+      waiters.add(terminal);
+      updated.set(options.nativeTurnId, waiters);
+      return updated;
+    });
+    const unregister = Ref.update(options.turnWaiters, (current) => {
+      const existing = current.get(options.nativeTurnId);
+      if (existing === undefined || !existing.has(terminal)) {
+        return current;
+      }
+      const updated = new Map(current);
+      const waiters = new Set(existing);
+      waiters.delete(terminal);
+      if (waiters.size === 0) {
+        updated.delete(options.nativeTurnId);
+      } else {
+        updated.set(options.nativeTurnId, waiters);
+      }
+      return updated;
+    });
+    const alreadyTerminal = !(yield* Ref.get(options.activeTurns)).has(options.nativeTurnId);
+    return { awaitTerminal: Deferred.await(terminal), unregister, alreadyTerminal };
+  });
+
+/**
+ * Marks a native turn terminal: removes it from the active set first, then
+ * drains and signals its waiters. The ordering pairs with
+ * {@link registerCodexTurnTerminalWaiter} so a completion racing an interrupt
+ * can never strand a waiter registered after the drain.
+ */
+export const settleCodexTurnTerminal = <Context>(
+  options: CodexTurnTerminalTrackingOptions<Context>,
+): Effect.Effect<void> =>
+  Effect.gen(function* () {
+    yield* Ref.update(options.activeTurns, (current) => {
+      const updated = new Map(current);
+      updated.delete(options.nativeTurnId);
+      return updated;
+    });
+    const waiters = yield* Ref.modify(options.turnWaiters, (current) => {
+      const matching =
+        current.get(options.nativeTurnId) ?? new Set<Deferred.Deferred<void, never>>();
+      const updated = new Map(current);
+      updated.delete(options.nativeTurnId);
+      return [matching, updated] as const;
+    });
+    yield* Effect.forEach(waiters, (waiter) => Deferred.succeed(waiter, undefined), {
+      discard: true,
+    });
+  });
+
 export interface CodexAdapterV2Options {
   readonly instanceId: ProviderInstanceId;
   readonly settings: CodexSettings;
@@ -1283,7 +1365,9 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
         const events = yield* Queue.unbounded<ProviderAdapterV2Event>();
         const activeTurns = yield* Ref.make(new Map<string, ActiveCodexTurnContext>());
         const pendingRootTurns = yield* Ref.make(new Map<string, ProviderAdapterV2TurnInput>());
-        const turnWaiters = yield* Ref.make(new Map<string, Deferred.Deferred<void, never>>());
+        const turnWaiters = yield* Ref.make(
+          new Map<string, ReadonlySet<Deferred.Deferred<void, never>>>(),
+        );
         const subagentThreads = yield* Ref.make(new Map<string, CodexSubagentThreadContext>());
         const pendingSubagentTurns = yield* Ref.make(
           new Map<string, ReadonlyArray<PendingCodexSubagentTurnStarted>>(),
@@ -3514,14 +3598,10 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
                     },
               );
             }
-            const waiter = (yield* Ref.get(turnWaiters)).get(payload.turn.id);
-            if (waiter !== undefined) {
-              yield* Deferred.succeed(waiter, undefined);
-            }
-            yield* Ref.update(activeTurns, (current) => {
-              const updated = new Map(current);
-              updated.delete(payload.turn.id);
-              return updated;
+            yield* settleCodexTurnTerminal({
+              activeTurns,
+              turnWaiters,
+              nativeTurnId: payload.turn.id,
             });
           }),
         );
@@ -3694,10 +3774,31 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
                   `Provider turn ${turnInput.providerTurnId} is not active and cannot be interrupted.`,
                 );
               }
-              yield* client.request("turn/interrupt", {
-                threadId,
-                turnId: activeTurn.nativeTurnId,
+              const waiter = yield* registerCodexTurnTerminalWaiter({
+                activeTurns,
+                turnWaiters,
+                nativeTurnId: activeTurn.nativeTurnId,
               });
+              if (waiter.alreadyTerminal) {
+                // The turn settled between the active-turn lookup and waiter
+                // registration; there is nothing left to interrupt.
+                return yield* waiter.unregister;
+              }
+              const stopped = yield* client
+                .request("turn/interrupt", {
+                  threadId,
+                  turnId: activeTurn.nativeTurnId,
+                })
+                .pipe(
+                  Effect.andThen(waiter.awaitTerminal),
+                  Effect.timeoutOption("10 seconds"),
+                  Effect.ensuring(waiter.unregister),
+                );
+              if (Option.isNone(stopped)) {
+                return yield* toProtocolError(
+                  `Provider turn ${turnInput.providerTurnId} did not reach a terminal state before the interrupt timeout.`,
+                );
+              }
             }).pipe(
               Effect.mapError(
                 (cause) =>

apps/web/src/components/ChatView.tsx

@@ -4393,6 +4393,7 @@ function ChatViewContent(props: ChatViewProps) {
       environmentId,
       input: {
         threadId: activeThread.id,
+        ...(activeRuntime?.activeRunId ? { runId: activeRuntime.activeRunId } : {}),
       },
     });
     if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) {

apps/web/src/components/chat/ComposerPrimaryActions.test.ts

@@ -123,7 +123,7 @@ describe("active-turn primary action", () => {
     expect(markup).not.toContain("steer active turn");
   });
 
-  it("replaces stop with send while the active composer has content", () => {
+  it("keeps stop available alongside send while the active composer has content", () => {
     const markup = renderToStaticMarkup(
       createElement(ComposerPrimaryActions, {
         ...activeTurnProps,
@@ -133,6 +133,6 @@ describe("active-turn primary action", () => {
     );
 
     expect(markup).toContain('aria-label="Send message to steer active turn"');
-    expect(markup).not.toContain('aria-label="Stop generation"');
+    expect(markup).toContain('aria-label="Stop generation"');
   });
 });

apps/web/src/components/chat/ComposerPrimaryActions.tsx

@@ -124,21 +124,21 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({
     );
   }
 
-  if (isRunning && !hasSendableContent) {
-    return (
-      <button
-        type="button"
-        className="flex size-8 cursor-pointer items-center justify-center rounded-full bg-destructive/90 text-white shadow-xs shadow-destructive/24 inset-shadow-[0_1px_--theme(--color-white/16%)] transition-all duration-150 hover:bg-destructive hover:scale-105 active:inset-shadow-[0_1px_--theme(--color-black/8%)] active:shadow-none sm:h-8 sm:w-8"
-        {...pointerFocusProps}
-        onClick={onInterrupt}
-        aria-label="Stop generation"
-      >
-        <svg width="12" height="12" viewBox="0 0 12 12" fill="currentColor" aria-hidden="true">
-          <rect x="2" y="2" width="8" height="8" rx="1.5" />
-        </svg>
-      </button>
-    );
-  }
+  const stopButton = (
+    <button
+      type="button"
+      className="flex size-8 cursor-pointer items-center justify-center rounded-full bg-destructive/90 text-white shadow-xs shadow-destructive/24 inset-shadow-[0_1px_--theme(--color-white/16%)] transition-all duration-150 hover:bg-destructive hover:scale-105 active:inset-shadow-[0_1px_--theme(--color-black/8%)] active:shadow-none sm:h-8 sm:w-8"
+      {...pointerFocusProps}
+      onClick={onInterrupt}
+      aria-label="Stop generation"
+    >
+      <svg width="12" height="12" viewBox="0 0 12 12" fill="currentColor" aria-hidden="true">
+        <rect x="2" y="2" width="8" height="8" rx="1.5" />
+      </svg>
+    </button>
+  );
+
+  if (isRunning && !hasSendableContent) return stopButton;
 
   if (showPlanFollowUpPrompt) {
     if (promptHasText) {
@@ -233,9 +233,12 @@ export const ComposerPrimaryActions = memo(function ComposerPrimaryActions({
   if (!isRunning) return sendButton;
 
   return (
-    <Tooltip>
-      <TooltipTrigger render={sendButton} />
-      <TooltipPopup side="top">Send now to steer the active turn</TooltipPopup>
-    </Tooltip>
+    <div className="flex items-center gap-2">
+      {stopButton}
+      <Tooltip>
+        <TooltipTrigger render={sendButton} />
+        <TooltipPopup side="top">Send now to steer the active turn</TooltipPopup>
+      </Tooltip>
+    </div>
   );
 });