feat(fx): add safe CLI updates with cancellation compatibility

apps/server/scripts/acp-mock-agent.ts

@@ -1235,6 +1235,15 @@ const program = Effect.gen(function* () {
             content: [{ type: "content", content: { type: "text", text: "Cancelled." } }],
           },
         });
+        // A later output-only update must not undo the interrupted status.
+        yield* agent.client.sessionUpdate({
+          sessionId: requestedSessionId,
+          update: {
+            sessionUpdate: "tool_call_update",
+            toolCallId: "native-cancel-tool",
+            content: [{ type: "content", content: { type: "text", text: "Cancelled. exit 130" } }],
+          },
+        });
         yield* agent.client.sessionUpdate({
           sessionId: requestedSessionId,
           update: {

apps/server/scripts/record-fx-acp-replay-fixture.ts

@@ -1,8 +1,8 @@
 /**
- * Records an fx ACP replay transcript through the real orchestrator and
- * FxAdapterV2, teeing only the protocol logger. Run from apps/server with
- * fx 0.0.13 on PATH (or T3_FX_BIN); HOME stays the user's real login:
+ * Records an fx ACP replay transcript through the real orchestrator and FxAdapterV2.
+ * Run from apps/server with fx on PATH (or T3_FX_BIN), signed in under HOME:
  *   node scripts/record-fx-acp-replay-fixture.ts --scenario simple
+ * fx 0.0.14+ saves each recorded mode as the default in ~/.fx/settings.json; restore it after.
  */
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { ProviderInstanceId, type ProviderReplayEntry } from "@t3tools/contracts";

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

@@ -4233,15 +4233,27 @@ describe("AcpAdapterV2", () => {
       );
       yield* native.current.request("_test/finish-cancel", {});
       yield* Fiber.join(interrupt);
-      const terminal = Option.getOrThrow(
+      const untilTerminal = Array.from(
         yield* runtime.events.pipe(
-          Stream.filter(
+          Stream.takeUntil(
             (event) => event.type === "turn.terminal" && event.providerTurnId === providerTurnId,
           ),
-          Stream.runHead,
+          Stream.runCollect,
         ),
       );
-      assert.equal(terminal.type === "turn.terminal" && terminal.status, "interrupted");
+      const terminal = untilTerminal.at(-1);
+      assert.equal(terminal?.type === "turn.terminal" && terminal.status, "interrupted");
+      // The agent fails the command Stop cut off, then updates its output.
+      const command = untilTerminal
+        .flatMap((event) =>
+          event.type === "turn_item.updated" &&
+          event.turnItem.nativeItemRef?.nativeId === "native-cancel-tool"
+            ? [event.turnItem]
+            : [],
+        )
+        .at(-1);
+      assert.equal(command?.status, "interrupted");
+      assert.include(command?.type === "command_execution" ? command.output : "", "exit 130");
       yield* runtime.startTurn(
         makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal: 2 }),
       );

apps/server/src/orchestration-v2/testkit/fixtures/fx_subagent_stop/fx_output.ts

@@ -0,0 +1,40 @@
+import { assert } from "@effect/vitest";
+import type { ProviderReplayTranscript } from "@t3tools/contracts";
+
+import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts";
+import {
+  assertBaseProjection,
+  assertSemanticProjectionIntegrity,
+  assertUserMessagesInclude,
+  FX_SUBAGENT_STOP_PROMPT,
+  projectionFor,
+} from "../shared.ts";
+
+/** fx 0.0.15 fails a sub-agent that Stop cut off; T3 shows it, and its child thread, as interrupted. */
+export function assertFxSubagentStopOutput(
+  result: OrchestratorV2ScenarioResult,
+  transcript: ProviderReplayTranscript,
+) {
+  assertBaseProjection({ result, transcript, runCount: 1, runStatuses: ["interrupted"] });
+
+  const projection = projectionFor(result, transcript.scenario);
+  assertSemanticProjectionIntegrity(projection);
+  assertUserMessagesInclude(projection, [FX_SUBAGENT_STOP_PROMPT]);
+  assert.notInclude(
+    projection.turnItems.map((item) => item.type),
+    "dynamic_tool",
+    "the subagent call must not also show as a tool row",
+  );
+  assert.lengthOf(projection.subagents, 1);
+  const subagent = projection.subagents[0]!;
+  assert.equal(subagent.status, "interrupted");
+  assert.include(subagent.prompt, "stopped.txt");
+  assert.equal(subagent.result, "Cancelled", "fx's error, not its JSON envelope");
+  if (subagent.childThreadId === null) throw new Error("The subagent has no child thread");
+  assert.equal(
+    result.projections.get(subagent.childThreadId)?.thread.lineage.parentThreadId,
+    projection.thread.id,
+  );
+  assert.equal(projection.providerTurns[0]?.status, "interrupted");
+  assert.equal(projection.providerThreads[0]?.status, "idle");
+}

apps/server/src/orchestration-v2/testkit/fixtures/fx_subagent_stop/input.ts

@@ -0,0 +1,15 @@
+import { FX_SUBAGENT_STOP_PROMPT, type OrchestratorFixtureInput } from "../shared.ts";
+
+/**
+ * Supervised: fx never forwards a sub-agent's permission request over ACP,
+ * so the sub-agent waits until Stop.
+ */
+export function fxSubagentStopInput(): OrchestratorFixtureInput {
+  return {
+    runtimeMode: "approval-required",
+    steps: [
+      { type: "message", text: FX_SUBAGENT_STOP_PROMPT },
+      { type: "interrupt", targetRunIndex: 1, waitForTurnItemType: "subagent" },
+    ],
+  };
+}

apps/server/src/orchestration-v2/testkit/fixtures/index.ts

@@ -196,9 +196,12 @@ import { fxEarlyStopInput } from "./fx_early_stop/input.ts";
 import { assertFxEarlyStopOutput } from "./fx_early_stop/fx_output.ts";
 import { fxSubagentInput } from "./fx_subagent/input.ts";
 import { assertFxSubagentOutput } from "./fx_subagent/fx_output.ts";
+import { fxSubagentStopInput } from "./fx_subagent_stop/input.ts";
+import { assertFxSubagentStopOutput } from "./fx_subagent_stop/fx_output.ts";
 import { assertTurnInterruptMidToolClaudeOutput } from "./turn_interrupt_mid_tool/claude_output.ts";
 import { assertTurnInterruptMidToolCodexOutput } from "./turn_interrupt_mid_tool/codex_output.ts";
 import { assertTurnInterruptMidToolCursorOutput } from "./turn_interrupt_mid_tool/cursor_output.ts";
+import { assertTurnInterruptMidToolFxOutput } from "./turn_interrupt_mid_tool/fx_output.ts";
 import { turnInterruptMidToolInput } from "./turn_interrupt_mid_tool/input.ts";
 import { assertTurnInterruptMidToolPiOutput } from "./turn_interrupt_mid_tool/pi_output.ts";
 import { assertTurnInterruptRestartClaudeOutput } from "./turn_interrupt_restart/claude_output.ts";
@@ -1644,6 +1647,18 @@ export const ORCHESTRATOR_REPLAY_FIXTURES: ReadonlyArray<OrchestratorReplayFixtu
       },
     ],
   },
+  {
+    name: "fx_subagent_stop",
+    buildInput: fxSubagentStopInput,
+    providers: [
+      {
+        driver: ProviderDriverKind.make("fx"),
+        transcriptFile: new URL("./fx_subagent_stop/fx_transcript.ndjson", import.meta.url),
+        modelSelection: FX_MODEL_SELECTION,
+        assertOutput: assertFxSubagentStopOutput,
+      },
+    ],
+  },
   {
     name: "command_output_streaming",
     buildInput: commandOutputStreamingInput,
@@ -1703,6 +1718,17 @@ export const ORCHESTRATOR_REPLAY_FIXTURES: ReadonlyArray<OrchestratorReplayFixtu
         runtimePolicyOverride: WORKSPACE_NEVER_POLICY,
         assertOutput: assertTurnInterruptMidToolCursorOutput,
       },
+      {
+        driver: ProviderDriverKind.make("fx"),
+        transcriptFile: new URL("./turn_interrupt_mid_tool/fx_transcript.ndjson", import.meta.url),
+        modelSelection: FX_MODEL_SELECTION,
+        // fx asks before running the command; a workspace sandbox would deny it.
+        runtimePolicyOverride: {
+          approvalPolicy: "never",
+          sandboxPolicy: { type: "dangerFullAccess" },
+        },
+        assertOutput: assertTurnInterruptMidToolFxOutput,
+      },
       {
         driver: ProviderDriverKind.make("pi"),
         transcriptFile: new URL("./turn_interrupt_mid_tool/pi_transcript.ndjson", import.meta.url),

apps/server/src/orchestration-v2/testkit/fixtures/shared.ts

@@ -99,6 +99,8 @@ export const OPENCODE2_COMMAND_PROMPT = "/hello WORLD";
 export const OPENCODE2_SKILL_PROMPT = "Use $greet to say hi in three words.";
 export const TURN_INTERRUPT_PROMPT =
   "Do not answer immediately. First run the local shell command `sleep 30`, then respond with exactly: interrupt fixture should not finish naturally.";
+export const FX_SUBAGENT_STOP_PROMPT =
+  "Use exactly one subagent to create a file named stopped.txt containing STOP. Do not create the file yourself. Wait for the subagent, then reply DONE.";
 export const TURN_INTERRUPT_MID_TOOL_PROMPT =
   "Run this exact local command: `node -e \"console.log('interrupt fixture tool started'); setTimeout(() => {}, 30000)\"`. Do not answer until it completes, then respond exactly: interrupt fixture should not finish naturally.";
 const COMMAND_OUTPUT_STREAMING_COMMAND =

apps/server/src/orchestration-v2/testkit/fixtures/turn_interrupt_mid_tool/fx_output.ts

@@ -0,0 +1,64 @@
+import { assert } from "@effect/vitest";
+import type { ProviderReplayTranscript } from "@t3tools/contracts";
+
+import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts";
+import {
+  assertBaseProjection,
+  assertSemanticProjectionIntegrity,
+  assertTurnItemTypes,
+  assertUserMessagesInclude,
+  assertVisibleTurnItemsMirrorLocalTurnItems,
+  projectionFor,
+  TURN_INTERRUPT_MID_TOOL_PROMPT,
+} from "../shared.ts";
+
+function acpFrame(entry: ProviderReplayTranscript["entries"][number]) {
+  return entry.type === "runtime_exit"
+    ? undefined
+    : (entry.frame as {
+        readonly method?: string;
+        readonly params?: { readonly update?: { sessionUpdate?: string; status?: string } };
+      });
+}
+
+/** fx 0.0.15 answers Stop by failing the running command itself; T3 still shows it interrupted. */
+export function assertTurnInterruptMidToolFxOutput(
+  result: OrchestratorV2ScenarioResult,
+  transcript: ProviderReplayTranscript,
+) {
+  const cancelIndex = transcript.entries.findIndex(
+    (entry) => entry.type === "expect_outbound" && acpFrame(entry)?.method === "session/cancel",
+  );
+  const failedAfterCancel = transcript.entries.findIndex(
+    (entry, index) =>
+      index > cancelIndex &&
+      entry.type === "emit_inbound" &&
+      acpFrame(entry)?.params?.update?.sessionUpdate === "tool_call_update" &&
+      acpFrame(entry)?.params?.update?.status === "failed",
+  );
+  assert.isAtLeast(cancelIndex, 0);
+  assert.isAbove(failedAfterCancel, cancelIndex, "fx fails the cut-off command after Stop");
+  assertBaseProjection({ result, transcript, runCount: 1, runStatuses: ["interrupted"] });
+
+  const projection = projectionFor(result, transcript.scenario);
+  assertSemanticProjectionIntegrity(projection);
+  assertVisibleTurnItemsMirrorLocalTurnItems(projection);
+  assertTurnItemTypes(projection, [
+    "user_message",
+    "command_execution",
+    "run_interrupt_request",
+    "run_interrupt_result",
+  ]);
+  assertUserMessagesInclude(projection, [TURN_INTERRUPT_MID_TOOL_PROMPT]);
+
+  const commandItem = projection.turnItems.find((item) => item.type === "command_execution");
+  assert.isDefined(commandItem);
+  assert.include(commandItem.input, "node -e");
+  assert.equal(commandItem.status, "interrupted");
+  assert.equal(
+    projection.turnItems.find((item) => item.type === "run_interrupt_result")?.status,
+    "interrupted",
+  );
+  assert.equal(projection.providerTurns[0]?.status, "interrupted");
+  assert.equal(projection.providerThreads[0]?.status, "idle");
+}