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

feat(KB-326): implement real-time steering comment injection

Browse Source 
- Add real-time steering comment injection mechanism for active tasks
- Improve steering injection logging and simplify trigger conditions
- Add comprehensive tests for steering injection functionality
- Add documentation for the real-time steering mechanism
This commit is contained in:
gsxdsm
2026-03-31 18:36:59 -07:00
parent d711b2b9dd
commit cc6029ef68
3 changed files with 514 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

420
packages/engine/src/executor.test.ts
View File

@@ -4008,3 +4008,423 @@ describe("Workflow Steps Execution", () => {
});
});
describe("Real-time steering injection", () => {
beforeEach(() => {
vi.clearAllMocks();
});
it("initializes seenSteeringIds with existing comments at session start", async () => {
const store = createMockStore();
const steerFn = vi.fn().mockResolvedValue(undefined);
// Mock session with steer method
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockImplementation(async () => {
// Simulate execution running
await new Promise(resolve => setTimeout(resolve, 10));
}),
dispose: vi.fn(),
steer: steerFn,
},
} as any);
const executor = new TaskExecutor(store, "/tmp/test");
const existingComment = {
id: "1234567890-abc123",
text: "Existing comment",
createdAt: new Date().toISOString(),
author: "user" as const,
};
await executor.execute({
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [existingComment],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for execution to complete
await new Promise(resolve => setTimeout(resolve, 50));
// No steer calls should be made for existing comments
expect(steerFn).not.toHaveBeenCalled();
});
it("injects new steering comments via session.steer() on task:updated", async () => {
const store = createMockStore();
const steerFn = vi.fn().mockResolvedValue(undefined);
let promptResolve: () => void;
const promptPromise = new Promise<void>(resolve => { promptResolve = resolve; });
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockImplementation(async () => {
// Wait for signal to complete
await promptPromise;
}),
dispose: vi.fn(),
steer: steerFn,
},
} as any);
const executor = new TaskExecutor(store, "/tmp/test");
// Start execution
const executePromise = executor.execute({
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for agent to start
await new Promise(resolve => setTimeout(resolve, 20));
// Simulate adding a steering comment mid-execution
const newComment = {
id: "9876543210-def456",
text: "Please use a different approach",
createdAt: new Date().toISOString(),
author: "user" as const,
};
store._trigger("task:updated", {
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [newComment],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for steer to be called
await new Promise(resolve => setTimeout(resolve, 20));
// Verify steer was called with the formatted message
expect(steerFn).toHaveBeenCalledOnce();
expect(steerFn.mock.calls[0][0]).toContain("📣 **New steering feedback**");
expect(steerFn.mock.calls[0][0]).toContain("Please use a different approach");
// Verify log entry was created
expect(store.logEntry).toHaveBeenCalledWith(
"KB-001",
expect.stringContaining("Steering comment received mid-execution"),
"by user"
);
// Complete the execution
promptResolve!();
await executePromise;
});
it("does not re-inject already seen steering comments", async () => {
const store = createMockStore();
const steerFn = vi.fn().mockResolvedValue(undefined);
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
steer: steerFn,
},
} as any);
const executor = new TaskExecutor(store, "/tmp/test");
const commentId = "1111111111-aaa111";
// Start execution with one comment
await executor.execute({
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [{
id: commentId,
text: "Original comment",
createdAt: new Date().toISOString(),
author: "user" as const,
}],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for execution to start
await new Promise(resolve => setTimeout(resolve, 20));
// Trigger task:updated with the SAME comment (simulating a non-steering update)
store._trigger("task:updated", {
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [{
id: commentId,
text: "Original comment",
createdAt: new Date().toISOString(),
author: "user" as const,
}],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait and verify steer was not called again
await new Promise(resolve => setTimeout(resolve, 20));
expect(steerFn).not.toHaveBeenCalled();
});
it("marks comment as seen even if steer() throws", async () => {
const store = createMockStore();
const steerFn = vi.fn().mockRejectedValue(new Error("Session disconnected"));
let resolvePrompt: () => void;
const promptPromise = new Promise<void>(resolve => { resolvePrompt = resolve; });
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockImplementation(() => promptPromise),
dispose: vi.fn(),
steer: steerFn,
},
} as any);
const executor = new TaskExecutor(store, "/tmp/test");
const commentId = "2222222222-bbb222";
// Start execution (don't await yet)
const executePromise = executor.execute({
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for execution to start
await new Promise(resolve => setTimeout(resolve, 20));
// Add a new comment that will fail to inject
store._trigger("task:updated", {
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [{
id: commentId,
text: "Comment that fails",
createdAt: new Date().toISOString(),
author: "user" as const,
}],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for processing
await new Promise(resolve => setTimeout(resolve, 20));
// Verify steer was called (and failed)
expect(steerFn).toHaveBeenCalledOnce();
// Trigger task:updated again with the same comment
store._trigger("task:updated", {
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [{
id: commentId,
text: "Comment that fails",
createdAt: new Date().toISOString(),
author: "user" as const,
}],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait and verify steer was NOT called again (comment marked as seen)
await new Promise(resolve => setTimeout(resolve, 20));
expect(steerFn).toHaveBeenCalledTimes(1);
// Complete execution
resolvePrompt!();
await executePromise;
});
it("does not inject steering comments for tasks not in activeSessions", async () => {
const store = createMockStore();
const steerFn = vi.fn().mockResolvedValue(undefined);
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
steer: steerFn,
},
} as any);
new TaskExecutor(store, "/tmp/test");
// Trigger task:updated for a task that is not in activeSessions
store._trigger("task:updated", {
id: "KB-NOT-EXECUTING",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [{
id: "3333333333-ccc333",
text: "Should not be injected",
createdAt: new Date().toISOString(),
author: "user" as const,
}],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait and verify steer was not called
await new Promise(resolve => setTimeout(resolve, 20));
expect(steerFn).not.toHaveBeenCalled();
});
it("handles multiple new steering comments in a single task:updated", async () => {
const store = createMockStore();
const steerFn = vi.fn().mockResolvedValue(undefined);
let resolvePrompt: () => void;
const promptPromise = new Promise<void>(resolve => { resolvePrompt = resolve; });
// Set up getTask to return the task with existing steering comment
store.getTask.mockResolvedValue({
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
prompt: "# test\n## Steps\n### Step 0: Preflight\n- [ ] check",
steeringComments: [{
id: "existing-comment",
text: "Original",
createdAt: new Date().toISOString(),
author: "user",
}],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockImplementation(() => promptPromise),
dispose: vi.fn(),
steer: steerFn,
},
} as any);
const executor = new TaskExecutor(store, "/tmp/test");
// Start execution (don't await yet)
const executePromise = executor.execute({
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for execution to start
await new Promise(resolve => setTimeout(resolve, 20));
// Add two new comments at once
store._trigger("task:updated", {
id: "KB-001",
title: "Test",
description: "Test",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
steeringComments: [
{
id: "existing-comment",
text: "Original",
createdAt: new Date().toISOString(),
author: "user",
},
{
id: "new-comment-1",
text: "First new comment",
createdAt: new Date().toISOString(),
author: "user",
},
{
id: "new-comment-2",
text: "Second new comment",
createdAt: new Date().toISOString(),
author: "user",
},
],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
});
// Wait for processing
await new Promise(resolve => setTimeout(resolve, 20));
// Verify steer was called twice (once for each new comment)
expect(steerFn).toHaveBeenCalledTimes(2);
// Complete execution
resolvePrompt!();
await executePromise;
});
});

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

@@ -160,8 +160,11 @@ export interface TaskExecutorOptions {
export class TaskExecutor {
private activeWorktrees = new Map<string, string>();
private executing = new Set<string>();
/** Active agent sessions per task, used to terminate on pause. */
private activeSessions = new Map<string, { dispose: () => void }>();
/** Active agent sessions per task, used to terminate on pause and inject steering. */
private activeSessions = new Map<string, {
session: AgentSession;
seenSteeringIds: Set<string>;
}>();
/** Tasks that were paused mid-execution (to avoid marking them as "failed"). */
private pausedAborted = new Set<string>();
/** Tasks that had a dependency added mid-execution (abort + discard worktree). */
@@ -195,20 +198,72 @@ export class TaskExecutor {
});
// When a task is paused while executing, terminate the agent session.
store.on("task:updated", (task) => {
// When steering comments are added during execution, inject them into the running session.
//
// Real-time steering comment injection mechanism:
// 1. When execution starts, we initialize seenSteeringIds with all existing comment IDs
// 2. On each task:updated event, we check if there are new comments not in seenSteeringIds
// 3. New comments are injected via session.steer() which queues them for delivery
// after the current assistant turn completes (before the next LLM call)
// 4. Comments are marked as seen BEFORE injection to prevent retry loops on failure
// 5. Each injection is logged to the task for user visibility
store.on("task:updated", async (task) => {
// Handle pause - terminate the agent session
if (task.paused && this.activeSessions.has(task.id)) {
executorLog.log(`Pausing ${task.id} — terminating agent session`);
this.pausedAborted.add(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
const session = this.activeSessions.get(task.id);
session?.dispose();
const { session } = this.activeSessions.get(task.id)!;
session.dispose();
return;
}
// 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;
// Find new steering comments that haven't been seen yet
const newComments = task.steeringComments.filter(c => !seenSteeringIds.has(c.id));
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);
// Format and inject the steering comment
const steeringMessage = formatSteeringCommentForInjection(comment);
try {
executorLog.log(`Injecting steering comment into ${task.id}: ${summary}`);
await session.steer(steeringMessage);
executorLog.log(`Successfully injected steering comment into ${task.id}`);
// Log to the task that steering was received
await this.store.logEntry(
task.id,
`Steering comment received mid-execution: ${summary}`,
`by ${comment.author}`
);
} catch (err) {
executorLog.error(`Failed to inject steering comment for ${task.id}:`, 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.
}
}
}
}
});
// When globalPause transitions from false → true, terminate all active agent sessions.
store.on("settings:updated", ({ settings, previous }) => {
if (settings.globalPause && !previous.globalPause) {
for (const [taskId, session] of this.activeSessions) {
for (const [taskId, { session }] of this.activeSessions) {
executorLog.log(`Global pause — terminating agent session for ${taskId}`);
this.pausedAborted.add(taskId);
this.options.stuckTaskDetector?.untrackTask(taskId);
@@ -477,7 +532,14 @@ export class TaskExecutor {
sessionRef.current = session;
// Register session so the pause listener can terminate it
this.activeSessions.set(task.id, session);
// Initialize with empty set of seen steering comments
const seenSteeringIds = new Set<string>();
if (detail.steeringComments) {
for (const comment of detail.steeringComments) {
seenSteeringIds.add(comment.id);
}
}
this.activeSessions.set(task.id, { session, seenSteeringIds });
// Register with stuck task detector for heartbeat monitoring
stuckDetector?.trackTask(task.id, session);
@@ -763,8 +825,8 @@ export class TaskExecutor {
// Trigger abort flow (same pattern as pausedAborted)
this.depAborted.add(taskId);
const session = this.activeSessions.get(taskId);
session?.dispose();
const activeSession = this.activeSessions.get(taskId);
activeSession?.session.dispose();
return {
content: [{
@@ -1684,3 +1746,12 @@ Use \`task_create\` if you find out-of-scope work that needs doing.
Commit at step boundaries: \`git commit -m "feat(${task.id}): complete Step N — description"\`
When all steps are complete: call \`task_done()\``;
}
/**
* Format a steering comment for injection into a running agent session.
* Used for real-time steering during task execution.
*/
function formatSteeringCommentForInjection(comment: import("@fusion/core").SteeringComment): string {
const timestamp = formatTimestamp(comment.createdAt);
return `📣 **New steering feedback** — ${timestamp} (${comment.author}):\n\n${comment.text}\n\nPlease adjust your approach based on this feedback.`;
}