Make steering the default during active turns

README.md

@@ -66,7 +66,7 @@ cd your_project
 fx
 ```
 
-The current directory becomes the primary workspace. Enter a prompt, or run `/help` to browse interactive commands. While fx is working, press Enter to queue a follow-up or Ctrl+Enter to steer the active turn at its next model boundary. If the turn has already closed, fx safely queues the steering prompt as the next turn.
+The current directory becomes the primary workspace. Enter a prompt, or run `/help` to browse interactive commands. While fx is working, Enter steers the active turn at its next safe model boundary. If a tool is running, fx waits for it to finish; press Escape to interrupt the active work and apply the update as soon as the turn settles.
 
 Tool calls are expanded by default. Enable `Collapse tool calls` in `/settings`, or set `"collapse_tool_calls": true` in `~/.fx/settings.json`, to show one summary per tool-call group in the main transcript. Individual calls remain available in the full transcript with Ctrl+O.
 

sdk/tests/test-term.mjs

@@ -16,10 +16,11 @@ if (!supportsJspi()) {
 const output = [];
 const streamedDecoder = new TextDecoder();
 let streamedText = "";
-const liveDraft = "queued draft";
-const queuedAnswer = "§";
+const liveDraft = "steering draft";
+const steeringAnswer = "§";
 let draftVisibleAt;
-let queuedVisibleAt;
+let steeringSubmittedAt;
+let postSubmitText = "";
 const originalSetTimeout = globalThis.setTimeout;
 let observeZeroTimeouts = false;
 let zeroTimeoutCount = 0;
@@ -37,9 +38,10 @@ const terminal = {
   write(bytes) {
     const chunk = bytes instanceof Uint8Array ? bytes : new TextEncoder().encode(bytes);
     output.push(chunk);
-    streamedText += streamedDecoder.decode(chunk, { stream: true });
+    const decoded = streamedDecoder.decode(chunk, { stream: true });
+    streamedText += decoded;
+    if (steeringSubmittedAt !== undefined) postSubmitText += decoded;
     if (draftVisibleAt === undefined && streamedText.includes(liveDraft)) draftVisibleAt = performance.now();
-    if (queuedVisibleAt === undefined && streamedText.includes("queued 1")) queuedVisibleAt = performance.now();
     process.stdout.write(chunk);
   },
   async drain() {
@@ -81,7 +83,7 @@ const mockFetch = async (_url, init) => {
     secondRequestBody = JSON.parse(new TextDecoder().decode(init.body));
     return new Response(new ReadableStream({
       start(controller) {
-        controller.enqueue(encoded.encode(`data: {"type":"text-delta","delta":"${queuedAnswer}"}\n`));
+        controller.enqueue(encoded.encode(`data: {"type":"text-delta","delta":"${steeringAnswer}"}\n`));
         controller.enqueue(encoded.encode('data: {"type":"finish","finishReason":{"unified":"stop"},"usage":{"inputTokens":{"total":1},"outputTokens":{"total":2}}}\n'));
         controller.enqueue(encoded.encode("data: [DONE]\n"));
         controller.close();
@@ -146,22 +148,20 @@ while (draftVisibleAt === undefined) {
   if (performance.now() >= deadline) throw new Error("timed out waiting for live follow-up input");
   await new Promise((resolve) => setTimeout(resolve, 10));
 }
+steeringSubmittedAt = performance.now();
 runtime.write("\r");
-while (queuedVisibleAt === undefined) {
-  if (streamFinishedAt !== undefined) throw new Error("terminal did not queue follow-up input while the response was active");
-  if (performance.now() >= deadline) throw new Error("timed out waiting for queued follow-up input");
+observeZeroTimeouts = false;
+const steeringDeadline = performance.now() + 5000;
+while (
+  secondRequestAt === undefined ||
+  !streamedText.includes(steeringAnswer)
+) {
+  if (performance.now() >= steeringDeadline) throw new Error("timed out waiting for steered fx-term response");
   await new Promise((resolve) => setTimeout(resolve, 10));
 }
-observeZeroTimeouts = false;
-if (secondRequestAt !== undefined) throw new Error("queued follow-up started before the active response finished");
 releaseFirstStream();
 while (streamFinishedAt === undefined) {
-  if (performance.now() >= deadline) throw new Error("timed out waiting for streamed fx-term response");
-  await new Promise((resolve) => setTimeout(resolve, 10));
-}
-const queuedDeadline = performance.now() + 5000;
-while (secondRequestAt === undefined || !streamedText.includes(queuedAnswer)) {
-  if (performance.now() >= queuedDeadline) throw new Error("timed out waiting for queued fx-term response");
+  if (performance.now() >= steeringDeadline) throw new Error("timed out releasing held fx-term response");
   await new Promise((resolve) => setTimeout(resolve, 10));
 }
 runtime.write("/exit\r");
@@ -176,16 +176,29 @@ if (exitCode !== 0) throw new Error(`fx-term exited with code ${exitCode}`);
 if (!text.includes("𝒇x")) throw new Error("shared fx welcome frame was not observed");
 if (!text.includes("Run /help for commands")) throw new Error("shared fx welcome guidance was not observed");
 if (requestedModel !== "sdk/term-model") throw new Error(`terminal prompt did not use the host-restored model: ${requestedModel}`);
-if (!(streamStartedAt < streamFinishedAt)) throw new Error("terminal fetch did not remain active for continuous streaming");
-if (!(draftVisibleAt < streamFinishedAt)) throw new Error("terminal rendered follow-up input only after continuous streaming finished");
-if (!(queuedVisibleAt < streamFinishedAt)) throw new Error("terminal queued follow-up input only after continuous streaming finished");
-if (!(secondRequestAt >= streamFinishedAt)) throw new Error("terminal started queued follow-up before continuous streaming finished");
-const queuedUser = secondRequestBody.prompt?.filter((message) => message.role === "user").at(-1);
-const queuedText = queuedUser?.content?.filter((part) => part.type === "text").map((part) => part.text);
-if (queuedText?.length !== 1 || queuedText[0] !== liveDraft) {
-  throw new Error(`queued follow-up request changed the submitted draft: ${JSON.stringify(queuedText)}`);
+if (!(streamStartedAt < secondRequestAt)) throw new Error("terminal started steering before the active response");
+if (!(draftVisibleAt < steeringSubmittedAt)) throw new Error("terminal did not render the steering draft before submission");
+if (!(steeringSubmittedAt <= secondRequestAt)) throw new Error("terminal started steering before submission");
+if (!(secondRequestAt < streamFinishedAt)) throw new Error("terminal waited for the active response before steering");
+if (postSubmitText.includes(`${liveDraft} · Esc to steer now`)) throw new Error("terminal exposed tool-only pending UI during immediate steering");
+if (!postSubmitText.includes(liveDraft)) throw new Error("terminal did not commit the steering user row after cutoff");
+if (!postSubmitText.includes("Thinking")) throw new Error("terminal hid activity during immediate steering");
+const steeringUser = secondRequestBody.prompt?.filter((message) => message.role === "user").at(-1);
+const steeringText = steeringUser?.content?.filter((part) => part.type === "text").map((part) => part.text);
+const steeringRequest = JSON.stringify(secondRequestBody.prompt);
+if (
+  steeringText?.length !== 1 ||
+  !steeringText[0].includes("<user_steering>") ||
+  !steeringText[0].includes("Apply this live user update to the current task.") ||
+  !steeringText[0].includes(liveDraft)
+) {
+  throw new Error(`steering request changed the submitted draft or directive: ${JSON.stringify(steeringText)}`);
+}
+if (!steeringRequest.includes("hello")) throw new Error("steering request omitted the visible assistant prefix");
+if (steeringRequest.includes("<turn_aborted>") || steeringRequest.includes("The previous response ended before completion.")) {
+  throw new Error("steering request included an interruption marker");
 }
-if (requestCount !== 2) throw new Error(`terminal sent ${requestCount} requests instead of the active and queued turns`);
+if (requestCount !== 2) throw new Error(`terminal sent ${requestCount} requests instead of the active and steered steps`);
 if (zeroTimeoutCount !== 0) throw new Error(`terminal allocated ${zeroTimeoutCount} zero-timeout poll timer(s)`);
 if (!events.some((event) => event.type === "config.restore" && event.configId === "model")) throw new Error("terminal model restore event was not emitted");
 if (!events.some((event) => event.type === "config.restore" && event.configId === "mode")) throw new Error("terminal mode restore event was not emitted");

src/core/agent/runtime/deps.zig

@@ -180,6 +180,12 @@ pub const AgentRuntimeDeps = struct {
     /// Drains user guidance admitted to the active turn. Returned text is owned
     /// by `arena` and is non-authoritative context, never permission evidence.
     take_steering: ?*const fn (ctx: *anyopaque, arena: Allocator, turn_id: u64) anyerror![]const []const u8 = null,
+    /// Drains guidance whose admission cancelled the current provider wait.
+    /// A non-empty result also clears that host-owned cancellation request.
+    take_immediate_steering: ?*const fn (ctx: *anyopaque, arena: Allocator, turn_id: u64) anyerror![]const []const u8 = null,
+    /// True when pending interactive guidance cannot be transferred into the
+    /// current request and must start after the active turn settles.
+    steering_handoff_required: ?*const fn (ctx: *anyopaque, turn_id: u64) bool = null,
     release_agent_terminal_lease: *const fn (ctx: *anyopaque, session_id: []const u8) anyerror!void = terminalLeaseCleanupUnavailable,
     prepare_parent_turn_context: ?*const fn (ctx: *anyopaque, arena: Allocator) anyerror!?PreparedParentTurnContext = null,
     acknowledge_parent_turn_context: ?*const fn (ctx: *anyopaque, arena: Allocator, acknowledgements: []const ParentTurnDeliveryAck) void = null,

src/core/agent/runtime/execution_memory.zig

@@ -24,13 +24,24 @@ const ToolCall = types.ToolCall;
 const Config = runtime_config.Config;
 const ToolExecutionStatus = runtime_tool_contracts.ToolExecutionStatus;
 
-const steering_open = "<user_steering>\n";
+const steering_open =
+    "<user_steering>\n" ++
+    "Apply this live user update to the current task. Continue working unless the user asks you to stop, the task is complete, or a genuine blocker prevents progress.\n\n";
 const steering_close = "\n</user_steering>";
 
 pub fn steeringMessage(alloc: Allocator, text: []const u8) ![]u8 {
     return std.fmt.allocPrint(alloc, steering_open ++ "{s}" ++ steering_close, .{text});
 }
 
+test "steering message tells the model to apply the update and continue" {
+    const message = try steeringMessage(std.testing.allocator, "focus on rendering");
+    defer std.testing.allocator.free(message);
+
+    try std.testing.expect(std.mem.find(u8, message, "live user update") != null);
+    try std.testing.expect(std.mem.find(u8, message, "Continue working") != null);
+    try std.testing.expectEqualStrings("focus on rendering", steeringText(message).?);
+}
+
 pub fn persistedStatusForCurrentFxLocalResult(
     status: ToolExecutionStatus,
     output: []const u8,
@@ -1033,11 +1044,15 @@ test "large result storage redacts secret-bearing output before preview and disk
 
 test "execution memory persists consumed steering without protocol wrappers" {
     const alloc = std.testing.allocator;
+    const first = try steeringMessage(alloc, "focus on rendering");
+    defer alloc.free(first);
+    const second = try steeringMessage(alloc, "run the focused test");
+    defer alloc.free(second);
     const messages = [_]ChatMessage{
         .{ .role = .user, .content = "ordinary user context" },
-        .{ .role = .user, .content = "<user_steering>\nfocus on rendering\n</user_steering>" },
+        .{ .role = .user, .content = first },
         .{ .role = .assistant, .content = "continuing" },
-        .{ .role = .user, .content = "<user_steering>\nrun the focused test\n</user_steering>" },
+        .{ .role = .user, .content = second },
     };
 
     const execution = try buildExecutionMemory(alloc, &messages);