root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

FN-6368: route task chat steering to active sessions

Browse Source 
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
This commit is contained in:
gsxdsm
2026-06-13 09:41:20 -07:00
parent 00282fbf20
commit e0ec3d1fbd
8 changed files with 312 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/fn-6368-steering-running-session.md Normal file
View File

@@ -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.

12
packages/dashboard/app/components/TaskChatTab.tsx
View File

@@ -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}
</div>
{sessionHint ? (
<div className="task-chat-session-hint" role="status">
{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}

18
packages/dashboard/app/components/__tests__/TaskChatTab.test.tsx
View File

@@ -117,8 +117,10 @@ function expectComposerSendableAfterDraft(message = "Please continue") {
expect(sendButton).not.toBeDisabled();
}
function expectQueuedSessionCopy() {
expect(screen.getByText(/picked up by the next session/i)).toBeInTheDocument();
function expectNoInactiveSessionHint() {
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();
},
);

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

@@ -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);

3
packages/engine/src/__tests__/executor-test-helpers.ts
View File

@@ -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

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

@@ -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);

186
packages/engine/src/executor.ts
View File

@@ -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, {
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;
}>();
private activeSessions = new Map<string, ActiveExecutorSessionState>();
/** 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: {
session: AgentSession;
seenSteeringIds: Set<string>;
lastResolvedModelProvider?: string;
lastResolvedModelId?: string;
lastTaskModelProvider?: string | null;
lastTaskModelId?: string | null;
lastAssignedAgentId?: string | null;
lastEffectiveColumnAgentId?: string | null;
}, worktreePath: string): void {
private setActiveSession(taskId: string, sessionState: ActiveExecutorSessionState, 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
// Only process if session is active (activeSessions check is sufficient
// since entries are only added when a task is in-progress)
if (this.activeSessions.has(task.id) && task.steeringComments) {
const activeSession = this.activeSessions.get(task.id)!;
const { session, seenSteeringIds } = activeSession;
// Handle steering comments - inject new ones into whichever execution
// surface currently owns the task: legacy single-session, step-session
// executor (including graph-pinned/workflow stepwise runs), or an
// individual workflow step AgentSession.
if (task.steeringComments) {
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 newComments = task.steeringComments.filter(c => !seenSteeringIds.has(c.id));
const activeSession = this.activeSessions.get(task.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
seenSteeringIds.add(comment.id);
// Mark as seen BEFORE attempting injection to prevent retry loops on failure.
target.seenSteeringIds.add(comment.id);
// Format and inject the comment
const commentMessage = formatCommentForInjection(comment);
try {
executorLog.log(`Injecting comment into ${task.id}: ${summary}`);
await session.steer(commentMessage);
executorLog.log(`Successfully injected comment into ${task.id}`);
executorLog.log(`Injecting comment into ${task.id} (${target.kind}): ${summary}`);
await target.inject(commentMessage);
executorLog.log(`Successfully injected comment into ${task.id} (${target.kind})`);
// Log to the task that comment was received
await this.store.logEntry(
task.id,
`Comment received mid-execution: ${summary}`,
`by ${comment.author}`
);
// Log to the task once per comment/tick even if multiple active surfaces exist.
if (!loggedCommentIds.has(comment.id)) {
await this.store.logEntry(
task.id,
`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
// Initialize with empty set of seen comments
const seenSteeringIds = new Set<string>();
if (detail.comments) {
for (const comment of detail.comments) {
seenSteeringIds.add(comment.id);
}
}
// Register session so the pause listener can terminate it.
// Initialize with all existing steering comments so only mid-flight
// comments are injected into the running session.
const seenSteeringIds = this.createSeenSteeringIds(detail);
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();

16
packages/engine/src/step-session-executor.ts
View File

@@ -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);