FN-6368: route task chat steering to active sessions

Ensure task chat messages reach the live execution surface instead of waiting for a future session.

- Track seen steering comments for legacy, step-session, and workflow-step execution paths.
- Forward new task chat steering to active step sessions and workflow step sessions, including parallel step handles.
- Remove misleading inactive-session composer copy and cover the steering paths with regression tests.
- Add a patch changeset for the published Fusion package.

Files changed:
 .changeset/fn-6368-steering-running-session.md     |   5 +
 packages/dashboard/app/components/TaskChatTab.tsx  |  12 +-
 .../app/components/__tests__/TaskChatTab.test.tsx  |  18 +-
 .../src/__tests__/executor-step-session.test.ts    | 108 ++++++++++++
 .../engine/src/__tests__/executor-test-helpers.ts  |   3 +
 .../src/__tests__/step-session-executor.test.ts    |  40 +++++
 packages/engine/src/executor.ts                    | 186 ++++++++++++++-------
 packages/engine/src/step-session-executor.ts       |  16 ++
 8 files changed, 312 insertions(+), 76 deletions(-)

Fusion-Task-Id: FN-6368

Fusion-Task-Lineage: 610a6185-b136-4fea-a4bd-ea78ab5aab47

.changeset/fn-6368-steering-running-session.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Steering messages sent from task chat now reach active step-session and workflow runs, including parallel step sessions, and the misleading inactive-session "next session" composer copy was removed.

packages/dashboard/app/components/TaskChatTab.tsx

@@ -424,7 +424,7 @@ export function TaskChatTab({ task, projectId, active, addToast, sessionLive, on
   const activeSession = isActiveAgentSession(task, { sessionLive });
   const sessionHint = activeSession
     ? "Message the active agent session. Guidance is delivered to the running session in real time."
-    : "Message saved here will be picked up by the next session when work resumes.";
+    : null;
   const canSend = draft.trim().length > 0 && !sending;
 
   const resizeComposer = useCallback(() => {
@@ -623,15 +623,17 @@ export function TaskChatTab({ task, projectId, active, addToast, sessionLive, on
       </div>
 
       <form className="task-chat-composer card" onSubmit={handleSubmit}>
-        <div className="task-chat-session-hint" role="status">
+        {sessionHint ? (
-          {sessionHint}
+          <div className="task-chat-session-hint" role="status">
-        </div>
+            {sessionHint}
+          </div>
+        ) : null}
         <div className="task-chat-composer-row">
           <textarea
             ref={textareaRef}
             className="input task-chat-input"
             value={draft}
-            placeholder={activeSession ? "Message the active agent session…" : "Message now; it will be picked up by the next session…"}
+            placeholder={activeSession ? "Message the active agent session…" : "Message the agent…"}
             onChange={(event) => setDraft(event.target.value)}
             onKeyDown={handleKeyDown}
             disabled={sending}

packages/dashboard/app/components/__tests__/TaskChatTab.test.tsx

@@ -117,8 +117,10 @@ function expectComposerSendableAfterDraft(message = "Please continue") {
   expect(sendButton).not.toBeDisabled();
 }
 
-function expectQueuedSessionCopy() {
+function expectNoInactiveSessionHint() {
-  expect(screen.getByText(/picked up by the next session/i)).toBeInTheDocument();
+  expect(screen.queryByText(/picked up by the next session/i)).not.toBeInTheDocument();
+  expect(document.querySelector(".task-chat-session-hint")).not.toBeInTheDocument();
+  expect(screen.getByPlaceholderText("Message the agent…")).toBeInTheDocument();
 }
 
 function expectActiveSessionCopy() {
@@ -868,7 +870,7 @@ describe("TaskChatTab", () => {
     );
 
     expect(screen.queryByText(/No active steerable agent session/)).not.toBeInTheDocument();
-    expect(screen.getByText(/picked up by the next session/i)).toBeInTheDocument();
+    expectNoInactiveSessionHint();
     const input = screen.getByLabelText("Message active agent session");
     expect(input).not.toBeDisabled();
     const sendButton = screen.getByRole("button", { name: "Send" });
@@ -938,7 +940,7 @@ describe("TaskChatTab", () => {
       />,
     );
 
-    expectQueuedSessionCopy();
+    expectNoInactiveSessionHint();
     expectComposerSendableAfterDraft();
   });
 
@@ -1071,7 +1073,7 @@ describe("TaskChatTab", () => {
     if (showsActiveCopy) {
       expectActiveSessionCopy();
     } else {
-      expectQueuedSessionCopy();
+      expectNoInactiveSessionHint();
     }
     expectComposerSendableAfterDraft();
   });
@@ -1086,7 +1088,7 @@ describe("TaskChatTab", () => {
   ])("keeps the composer sendable with queued copy for %s", (_label, task) => {
     render(<TaskChatTab task={task} active addToast={vi.fn()} />);
 
-    expectQueuedSessionCopy();
+    expectNoInactiveSessionHint();
     expectComposerSendableAfterDraft();
   });
 
@@ -1098,7 +1100,7 @@ describe("TaskChatTab", () => {
   ])("keeps the composer sendable with queued copy for %s", (_label, task) => {
     render(<TaskChatTab task={task} active addToast={vi.fn()} sessionLive={true} />);
 
-    expectQueuedSessionCopy();
+    expectNoInactiveSessionHint();
     expectComposerSendableAfterDraft();
   });
 
@@ -1107,7 +1109,7 @@ describe("TaskChatTab", () => {
     (status) => {
       render(<TaskChatTab task={makeTask({ column: "in-progress", assignedAgentId: "agent-1", status })} active addToast={vi.fn()} />);
 
-      expectQueuedSessionCopy();
+      expectNoInactiveSessionHint();
       expectComposerSendableAfterDraft();
     },
   );

packages/engine/src/__tests__/executor-step-session.test.ts

@@ -30,6 +30,7 @@ import {
   mockExecuteAll,
   mockTerminateAllSessions,
   mockCleanup,
+  mockSteerActiveSessions,
   resetExecutorMocks,
 } from "./executor-test-helpers.js";
 
@@ -3130,6 +3131,113 @@ describe("Real-time steering injection", () => {
     await executePromise;
   });
 
+  it("injects new steering comments via active StepSessionExecutor on task:updated", async () => {
+    const store = createMockStore();
+    const executor = new TaskExecutor(store, "/tmp/test");
+    const steerActiveSessions = vi.fn().mockResolvedValue(undefined);
+    const newComment = {
+      id: "step-session-comment",
+      text: "Please adjust the active step",
+      createdAt: new Date().toISOString(),
+      author: "user" as const,
+    };
+
+    (executor as any).activeStepExecutors.set("FN-001", { steerActiveSessions });
+    (executor as any).activeStepExecutorSeenSteeringIds.set("FN-001", new Set());
+
+    await (store as any)._triggerAsync("task:updated", {
+      id: "FN-001",
+      title: "Test",
+      description: "Test",
+      column: "in-progress",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      steeringComments: [newComment],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    expect(steerActiveSessions).toHaveBeenCalledOnce();
+    expect(steerActiveSessions.mock.calls[0][0]).toContain("📣 **New feedback**");
+    expect(steerActiveSessions.mock.calls[0][0]).toContain("Please adjust the active step");
+    expect(store.logEntry).toHaveBeenCalledWith(
+      "FN-001",
+      expect.stringContaining("Comment received mid-execution"),
+      "by user",
+    );
+  });
+
+  it("injects new steering comments via active workflow step session on task:updated", async () => {
+    const store = createMockStore();
+    const executor = new TaskExecutor(store, "/tmp/test");
+    const steer = vi.fn().mockResolvedValue(undefined);
+    const newComment = {
+      id: "workflow-step-comment",
+      text: "Please adjust the workflow step",
+      createdAt: new Date().toISOString(),
+      author: "user" as const,
+    };
+
+    (executor as any).activeWorkflowStepSessions.set("FN-001", { steer });
+    (executor as any).activeWorkflowStepSessionSeenSteeringIds.set("FN-001", new Set());
+
+    await (store as any)._triggerAsync("task:updated", {
+      id: "FN-001",
+      title: "Test",
+      description: "Test",
+      column: "in-progress",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      steeringComments: [newComment],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    expect(steer).toHaveBeenCalledOnce();
+    expect(steer.mock.calls[0][0]).toContain("📣 **New feedback**");
+    expect(steer.mock.calls[0][0]).toContain("Please adjust the workflow step");
+    expect(store.logEntry).toHaveBeenCalledWith(
+      "FN-001",
+      expect.stringContaining("Comment received mid-execution"),
+      "by user",
+    );
+  });
+
+  it("does not re-inject an already seen active StepSessionExecutor steering comment", async () => {
+    const store = createMockStore();
+    const executor = new TaskExecutor(store, "/tmp/test");
+    const steerActiveSessions = vi.fn().mockResolvedValue(undefined);
+    const comment = {
+      id: "step-session-seen-comment",
+      text: "Already delivered",
+      createdAt: new Date().toISOString(),
+      author: "user" as const,
+    };
+
+    (executor as any).activeStepExecutors.set("FN-001", { steerActiveSessions });
+    (executor as any).activeStepExecutorSeenSteeringIds.set("FN-001", new Set([comment.id]));
+
+    await (store as any)._triggerAsync("task:updated", {
+      id: "FN-001",
+      title: "Test",
+      description: "Test",
+      column: "in-progress",
+      dependencies: [],
+      steps: [],
+      currentStep: 0,
+      log: [],
+      steeringComments: [comment],
+      createdAt: new Date().toISOString(),
+      updatedAt: new Date().toISOString(),
+    });
+
+    expect(steerActiveSessions).not.toHaveBeenCalled();
+  });
+
   it("does not re-inject already seen steering comments", async () => {
     const store = createMockStore();
     const steerFn = vi.fn().mockResolvedValue(undefined);

packages/engine/src/__tests__/executor-test-helpers.ts

@@ -224,6 +224,7 @@ vi.mock("node:fs", () => ({
 export const mockExecuteAll: Mock<() => Promise<unknown[]>> = vi.fn().mockResolvedValue([]);
 export const mockTerminateAllSessions: Mock<() => Promise<void>> = vi.fn().mockResolvedValue(undefined);
 export const mockCleanup: Mock<() => Promise<void>> = vi.fn().mockResolvedValue(undefined);
+export const mockSteerActiveSessions: Mock<(message: string) => Promise<void>> = vi.fn().mockResolvedValue(undefined);
 
 vi.mock("../step-session-executor.js", () => ({
   StepSessionExecutor: vi.fn().mockImplementation(function () {
@@ -231,6 +232,7 @@ vi.mock("../step-session-executor.js", () => ({
       executeAll: mockExecuteAll,
       terminateAllSessions: mockTerminateAllSessions,
       cleanup: mockCleanup,
+      steerActiveSessions: mockSteerActiveSessions,
     };
   }),
 }));
@@ -416,6 +418,7 @@ export function resetExecutorMocks() {
   mockExecuteAll.mockResolvedValue([]);
   mockTerminateAllSessions.mockResolvedValue(undefined);
   mockCleanup.mockResolvedValue(undefined);
+  mockSteerActiveSessions.mockResolvedValue(undefined);
   // FN-4811 follow-up: the executingTaskLock is process-wide module state, so it must
   // be cleared between tests or earlier tests' claims will block later tests' execute()
   // calls ("expected at least 2 createFnAgent calls but got 0" / "expected not called

packages/engine/src/__tests__/step-session-executor.test.ts

@@ -986,6 +986,7 @@ function makeMockSession(promptFn?: () => Promise<void>) {
     prompt: promptFn ?? vi.fn().mockResolvedValue(undefined),
     dispose: vi.fn(),
     subscribe: vi.fn(),
+    steer: vi.fn().mockResolvedValue(undefined),
     model: { provider: "mock", id: "mock-model" },
   };
 }
@@ -1021,6 +1022,45 @@ describe("StepSessionExecutor", () => {
     vi.useRealTimers();
   });
 
+  describe("steering", () => {
+    it("steers every active step session and continues after per-session failures", async () => {
+      const task = makeTaskDetail();
+      const executor = new StepSessionExecutor({
+        taskDetail: task,
+        worktreePath: "/project/.worktrees/main",
+        rootDir: "/project",
+        settings: makeSettings(),
+        pluginRunner: undefined,
+      } as any);
+      const steerOne = vi.fn().mockResolvedValue(undefined);
+      const steerTwo = vi.fn().mockRejectedValue(new Error("disconnected"));
+      const steerThree = vi.fn().mockResolvedValue(undefined);
+
+      (executor as any).activeSessions.set(0, {
+        dispose: vi.fn(),
+        abortBash: vi.fn(),
+        steer: steerOne,
+      });
+      (executor as any).activeSessions.set(1, {
+        dispose: vi.fn(),
+        abortBash: vi.fn(),
+        steer: steerTwo,
+      });
+      (executor as any).activeSessions.set(2, {
+        dispose: vi.fn(),
+        abortBash: vi.fn(),
+        steer: steerThree,
+      });
+
+      await executor.steerActiveSessions("new guidance");
+
+      expect(steerOne).toHaveBeenCalledWith("new guidance");
+      expect(steerTwo).toHaveBeenCalledWith("new guidance");
+      expect(steerThree).toHaveBeenCalledWith("new guidance");
+      expect(getStepSessionLogger().warn).toHaveBeenCalledWith(expect.stringContaining("Failed to steer active session for step 1"));
+    });
+  });
+
   describe("sequential execution", () => {
     it("forwards taskEnv into step session creation", async () => {
       const prompt = makeStepPrompt("FN-001", 1);

packages/engine/src/executor.ts

@@ -1357,6 +1357,17 @@ export interface CliAgentRuntime {
   hookDirRoot?: string;
 }
 
+interface ActiveExecutorSessionState {
+  session: AgentSession;
+  seenSteeringIds: Set<string>;
+  lastResolvedModelProvider?: string;
+  lastResolvedModelId?: string;
+  lastTaskModelProvider?: string | null;
+  lastTaskModelId?: string | null;
+  lastAssignedAgentId?: string | null;
+  lastEffectiveColumnAgentId?: string | null;
+}
+
 export class TaskExecutor {
   private activeWorktrees = new Map<string, string>();
   private executing = new Set<string>();
@@ -1374,24 +1385,11 @@ export class TaskExecutor {
    *  session being fully reaped before creating/acquiring a new worktree. */
   private pendingTaskDisposals = new Map<string, Promise<void>>();
   /** Active agent sessions per task, used to terminate on pause and inject steering. */
-  private activeSessions = new Map<string, {
+  private activeSessions = new Map<string, ActiveExecutorSessionState>();
-    session: AgentSession;
-    seenSteeringIds: Set<string>;
-    lastResolvedModelProvider?: string;
-    lastResolvedModelId?: string;
-    lastTaskModelProvider?: string | null;
-    lastTaskModelId?: string | null;
-    lastAssignedAgentId?: string | null;
-    // Column-agent restart-invalidation (plan U5, R7/KTD-4). The effective
-    // column-agent id governing this session's seam (undefined when no binding
-    // governs — the legacy path). Tracked so the watcher can detect a workflow-
-    // definition edit or agent runtimeConfig change that re-keys the column-
-    // effective agent/model mid-flight and trigger the same restart path a
-    // task.modelProvider change does today.
-    lastEffectiveColumnAgentId?: string | null;
-  }>();
   /** Active step-session executors per task (mutually exclusive with activeSessions). */
   private activeStepExecutors = new Map<string, StepSessionExecutor>();
+  /** Steering comments already observed for active step-session executor runs. */
+  private activeStepExecutorSeenSteeringIds = new Map<string, Set<string>>();
   /** Column-agent principal alignment (plan U5, R6): the EFFECTIVE column-agent id
    *  currently running each executing task's coding/step session, when an
    *  override/defer binding governs the in-flight seam. Keyed by task id, populated
@@ -1404,6 +1402,8 @@ export class TaskExecutor {
   private effectiveColumnAgentByTask = new Map<string, string>();
   /** Active pre-merge workflow step sessions per task. */
   private activeWorkflowStepSessions = new Map<string, AgentSession>();
+  /** Steering comments already observed for active workflow step sessions. */
+  private activeWorkflowStepSessionSeenSteeringIds = new Map<string, Set<string>>();
   /** Active configured-command abort controllers keyed by task. */
   private activeConfiguredCommandControllers = new Map<string, Set<AbortController>>();
   /**
@@ -1448,16 +1448,7 @@ export class TaskExecutor {
   /** Set of ephemeral spawned agent IDs with in-flight cleanup (prevents duplicate deletion attempts). */
   private pendingEphemeralDeletions = new Set<string>();
 
-  private setActiveSession(taskId: string, sessionState: {
+  private setActiveSession(taskId: string, sessionState: ActiveExecutorSessionState, worktreePath: string): void {
-    session: AgentSession;
-    seenSteeringIds: Set<string>;
-    lastResolvedModelProvider?: string;
-    lastResolvedModelId?: string;
-    lastTaskModelProvider?: string | null;
-    lastTaskModelId?: string | null;
-    lastAssignedAgentId?: string | null;
-    lastEffectiveColumnAgentId?: string | null;
-  }, worktreePath: string): void {
     this.activeSessions.set(taskId, sessionState);
     activeSessionRegistry.registerPath(worktreePath, { taskId, kind: "executor", ownerKey: taskId });
   }
@@ -1472,13 +1463,15 @@ export class TaskExecutor {
     }
   }
 
-  private setActiveStepExecutor(taskId: string, stepExecutor: StepSessionExecutor, worktreePath: string): void {
+  private setActiveStepExecutor(taskId: string, stepExecutor: StepSessionExecutor, worktreePath: string, seenSteeringIds = new Set<string>()): void {
     this.activeStepExecutors.set(taskId, stepExecutor);
+    this.activeStepExecutorSeenSteeringIds.set(taskId, seenSteeringIds);
     activeSessionRegistry.registerPath(worktreePath, { taskId, kind: "step-session", ownerKey: `${taskId}#step-session` });
   }
 
   private deleteActiveStepExecutor(taskId: string, worktreePath?: string): void {
     this.activeStepExecutors.delete(taskId);
+    this.activeStepExecutorSeenSteeringIds.delete(taskId);
     // U5: drop the effective column-agent principal for this task's step session.
     this.effectiveColumnAgentByTask.delete(taskId);
     const resolvedWorktreePath = worktreePath ?? this.activeWorktrees.get(taskId);
@@ -1487,19 +1480,29 @@ export class TaskExecutor {
     }
   }
 
-  private setActiveWorkflowStepSession(taskId: string, session: AgentSession, worktreePath: string): void {
+  private setActiveWorkflowStepSession(taskId: string, session: AgentSession, worktreePath: string, seenSteeringIds = new Set<string>()): void {
     this.activeWorkflowStepSessions.set(taskId, session);
+    this.activeWorkflowStepSessionSeenSteeringIds.set(taskId, seenSteeringIds);
     activeSessionRegistry.registerPath(worktreePath, { taskId, kind: "workflow-step", ownerKey: `${taskId}#workflow-step` });
   }
 
   private deleteActiveWorkflowStepSession(taskId: string, worktreePath?: string): void {
     this.activeWorkflowStepSessions.delete(taskId);
+    this.activeWorkflowStepSessionSeenSteeringIds.delete(taskId);
     const resolvedWorktreePath = worktreePath ?? this.activeWorktrees.get(taskId);
     if (resolvedWorktreePath) {
       activeSessionRegistry.unregisterPath(resolvedWorktreePath);
     }
   }
 
+  private createSeenSteeringIds(task: { comments?: Array<{ id: string }>; steeringComments?: Array<{ id: string }> }): Set<string> {
+    const seenSteeringIds = new Set<string>();
+    for (const comment of task.steeringComments ?? task.comments ?? []) {
+      seenSteeringIds.add(comment.id);
+    }
+    return seenSteeringIds;
+  }
+
   private registerConfiguredCommandController(taskId: string, controller: AbortController): void {
     const controllers = this.activeConfiguredCommandControllers.get(taskId) ?? new Set<AbortController>();
     controllers.add(controller);
@@ -2544,56 +2547,117 @@ export class TaskExecutor {
           }
         }
 
-        // Handle steering comments - inject new ones into the running session
+        // Handle steering comments - inject new ones into whichever execution
-        // Only process if session is active (activeSessions check is sufficient
+        // surface currently owns the task: legacy single-session, step-session
-        // since entries are only added when a task is in-progress)
+        // executor (including graph-pinned/workflow stepwise runs), or an
-        if (this.activeSessions.has(task.id) && task.steeringComments) {
+        // individual workflow step AgentSession.
-          const activeSession = this.activeSessions.get(task.id)!;
+        if (task.steeringComments) {
-          const { session, seenSteeringIds } = activeSession;
+          const injectionTargets: Array<{
+            kind: "legacy" | "step-session" | "workflow-step";
+            seenSteeringIds: Set<string>;
+            inject: (message: string) => Promise<void>;
+            legacySession?: AgentSession;
+            legacyState?: ActiveExecutorSessionState;
+          }> = [];
 
-          // Find new steering comments that haven't been seen yet
+          const activeSession = this.activeSessions.get(task.id);
-          const newComments = task.steeringComments.filter(c => !seenSteeringIds.has(c.id));
+          if (activeSession) {
+            injectionTargets.push({
+              kind: "legacy",
+              seenSteeringIds: activeSession.seenSteeringIds,
+              inject: (message) => activeSession.session.steer(message),
+              legacySession: activeSession.session,
+              legacyState: activeSession,
+            });
+          }
+
+          const stepExecutor = this.activeStepExecutors.get(task.id);
+          if (stepExecutor) {
+            const seenSteeringIds = this.activeStepExecutorSeenSteeringIds.get(task.id) ?? this.createSeenSteeringIds(task);
+            this.activeStepExecutorSeenSteeringIds.set(task.id, seenSteeringIds);
+            injectionTargets.push({
+              kind: "step-session",
+              seenSteeringIds,
+              inject: (message) => stepExecutor.steerActiveSessions(message),
+            });
+          }
+
+          const workflowSession = this.activeWorkflowStepSessions.get(task.id);
+          if (workflowSession) {
+            const seenSteeringIds = this.activeWorkflowStepSessionSeenSteeringIds.get(task.id) ?? this.createSeenSteeringIds(task);
+            this.activeWorkflowStepSessionSeenSteeringIds.set(task.id, seenSteeringIds);
+            injectionTargets.push({
+              kind: "workflow-step",
+              seenSteeringIds,
+              inject: (message) => workflowSession.steer(message),
+            });
+          }
+
+          const loggedCommentIds = new Set<string>();
+          let legacyReviewHandoff: {
+            comments: import("@fusion/core").SteeringComment[];
+            session: AgentSession;
+            state: ActiveExecutorSessionState;
+          } | undefined;
+
+          for (const target of injectionTargets) {
+            // Find new steering comments that haven't been seen by this running surface yet.
+            const newComments = task.steeringComments.filter(c => !target.seenSteeringIds.has(c.id));
+            if (newComments.length === 0) continue;
 
-          if (newComments.length > 0) {
             for (const comment of newComments) {
               const summary = comment.text.length > 80
                 ? comment.text.slice(0, 80) + "..."
                 : comment.text;
 
-              // Mark as seen BEFORE attempting injection to prevent retry loops on failure
+              // Mark as seen BEFORE attempting injection to prevent retry loops on failure.
-              seenSteeringIds.add(comment.id);
+              target.seenSteeringIds.add(comment.id);
 
-              // Format and inject the comment
               const commentMessage = formatCommentForInjection(comment);
               try {
-                executorLog.log(`Injecting comment into ${task.id}: ${summary}`);
+                executorLog.log(`Injecting comment into ${task.id} (${target.kind}): ${summary}`);
-                await session.steer(commentMessage);
+                await target.inject(commentMessage);
-                executorLog.log(`Successfully injected comment into ${task.id}`);
+                executorLog.log(`Successfully injected comment into ${task.id} (${target.kind})`);
 
-                // Log to the task that comment was received
+                // Log to the task once per comment/tick even if multiple active surfaces exist.
-                await this.store.logEntry(
+                if (!loggedCommentIds.has(comment.id)) {
-                  task.id,
+                  await this.store.logEntry(
-                  `Comment received mid-execution: ${summary}`,
+                    task.id,
-                  `by ${comment.author}`
+                    `Comment received mid-execution: ${summary}`,
-                );
+                    `by ${comment.author}`
+                  );
+                  loggedCommentIds.add(comment.id);
+                }
               } catch (err) {
-                executorLog.error(`Failed to inject comment for ${task.id}:`, err);
+                executorLog.error(`Failed to inject comment for ${task.id} (${target.kind}):`, err);
                 // Comment is already marked as seen - we won't retry to avoid spamming
                 // the agent with failed injections. The error is logged for debugging.
               }
             }
 
-            // After injecting comments, check for review handoff intent
+            if (target.kind === "legacy" && target.legacySession && target.legacyState) {
+              legacyReviewHandoff = {
+                comments: newComments,
+                session: target.legacySession,
+                state: target.legacyState,
+              };
+            }
+          }
+
+          // After injecting comments, check for review handoff intent on the legacy
+          // session path. Step-session/workflow-step runs do not have the legacy
+          // review handoff state required by executeReviewHandoff.
+          if (legacyReviewHandoff) {
             // Only detect handoff in agent-authored comments when policy is enabled.
             // Merge per-task effective workflow settings (U3, KTD-3) so
             // reviewHandoffPolicy resolves from the workflow. Behavior-inert by default.
             const settings = await mergeEffectiveSettings(this.store, task, await this.store.getSettings());
             if (settings.reviewHandoffPolicy === "comment-triggered") {
-              const agentComments = newComments.filter(c => c.author !== "user");
+              const agentComments = legacyReviewHandoff.comments.filter(c => c.author !== "user");
               for (const comment of agentComments) {
                 if (detectReviewHandoffIntent(comment.text)) {
                   executorLog.log(`Review handoff detected in ${task.id}: ${comment.text.slice(0, 50)}...`);
-                  await this.executeReviewHandoff(task, session, activeSession);
+                  await this.executeReviewHandoff(task, legacyReviewHandoff.session, legacyReviewHandoff.state);
                   return; // Exit early - handoff handles session disposal
                 }
               }
@@ -6938,7 +7002,7 @@ export class TaskExecutor {
             });
           },
         });
-        this.setActiveStepExecutor(task.id, stepExecutor, worktreePath);
+        this.setActiveStepExecutor(task.id, stepExecutor, worktreePath, this.createSeenSteeringIds(detail));
 
         const stepWork = async () => {
           const results = await stepExecutor.executeAll();
@@ -7662,14 +7726,10 @@ export class TaskExecutor {
         // Make session available to custom tools (fn_task_update checkpoint capture, fn_review_step rewind)
         sessionRef.current = session;
 
-        // Register session so the pause listener can terminate it
+        // Register session so the pause listener can terminate it.
-        // Initialize with empty set of seen comments
+        // Initialize with all existing steering comments so only mid-flight
-        const seenSteeringIds = new Set<string>();
+        // comments are injected into the running session.
-        if (detail.comments) {
+        const seenSteeringIds = this.createSeenSteeringIds(detail);
-          for (const comment of detail.comments) {
-            seenSteeringIds.add(comment.id);
-          }
-        }
         this.setActiveSession(task.id, {
           session,
           seenSteeringIds,
@@ -11921,7 +11981,7 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit
         task.id,
         `Workflow step '${workflowStep.name}' using model: ${describeModel(session)}${useOverride && attemptLabel === "primary" ? " (workflow step override)" : ""}${attemptLabel === "fallback" ? " (fallback after timeout)" : ""}`,
       );
-      this.setActiveWorkflowStepSession(task.id, session, worktreePath);
+      this.setActiveWorkflowStepSession(task.id, session, worktreePath, this.createSeenSteeringIds(task));
 
       let output = "";
       const deltaNormalizer = createStreamingDeltaNormalizer();

packages/engine/src/step-session-executor.ts

@@ -602,6 +602,8 @@ interface SessionHandle {
    *  is killed via pi-coding-agent's killProcessTree. dispose() alone only
    *  disconnects listeners and leaves bash subtrees orphaned. */
   abortBash: () => void;
+  /** Inject mid-flight steering into the live step session. */
+  steer: (message: string) => Promise<void>;
 }
 
 
@@ -768,6 +770,16 @@ export class StepSessionExecutor {
     }
   }
 
+  async steerActiveSessions(message: string): Promise<void> {
+    for (const [stepIdx, handle] of this.activeSessions) {
+      try {
+        await handle.steer(message);
+      } catch (err) {
+        stepExecLog.warn(`Failed to steer active session for step ${stepIdx}: ${err}`);
+      }
+    }
+  }
+
   async terminateAllSessions(): Promise<void> {
     this.aborted = true;
     stepExecLog.log(
@@ -1098,6 +1110,10 @@ Follow instructions precisely and avoid unrelated changes.`,
           const handle: SessionHandle = {
             dispose: () => session?.dispose(),
             abortBash: () => session?.abortBash(),
+            steer: async (message) => {
+              if (!session) return;
+              await session.steer(message);
+            },
           };
           this.registerActiveStepSession(stepIndex, handle, worktreePath);
           stuckTaskDetector?.trackTask(trackingKey, { dispose: () => session?.dispose() }, taskDetail.id);