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

fix(FN-1300): capture agent run output in heartbeat and step sessions

Browse Source 
- Wire AgentLogger into heartbeat runs, including tool/text callbacks and stdout excerpts on run records
- Persist heartbeat context taskId snapshots earlier and ensure logger flushes on success and failure paths
- Pass TaskStore into StepSessionExecutor and flush per-attempt agent logs in finally cleanup
- Expand heartbeat and step-session tests to verify log persistence and flush behavior, and add a patch changeset for @gsxdsm/fusion
This commit is contained in:
gsxdsm
2026-04-08 12:46:24 -07:00
parent f465c3688c
commit b0b3ca7e4b
6 changed files with 275 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
packages/engine/src/agent-heartbeat.test.ts
View File

@@ -1,5 +1,6 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type AgentSession, type HeartbeatExecutionOptions, HEARTBEAT_SYSTEM_PROMPT } from "./agent-heartbeat.js";
import { AgentLogger } from "./agent-logger.js";
import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent, MessageStore, Message, AgentBudgetStatus } from "@fusion/core";
// Mock logger to suppress noise in test output
@@ -982,6 +983,7 @@ describe("HeartbeatMonitor", () => {
column: "triage",
}),
logEntry: vi.fn().mockResolvedValue({}),
appendAgentLog: vi.fn().mockResolvedValue(undefined),
...overrides,
} as unknown as TaskStore;
}
@@ -1275,6 +1277,39 @@ describe("HeartbeatMonitor", () => {
taskId: "FN-001",
});
});
it("records agent logs, context taskId, and stdoutExcerpt for successful runs", async () => {
const store = createStoreWithAgentForExec();
const appendAgentLog = vi.fn().mockResolvedValue(undefined);
mockTaskStore = createMockTaskStore({ appendAgentLog });
const mockSession = createMockAgentSession();
let onText: ((delta: string) => void) | undefined;
let onToolStart: ((name: string, args?: Record<string, unknown>) => void) | undefined;
let onToolEnd: ((name: string, isError: boolean, result?: unknown) => void) | undefined;
mockedCreateKbAgent.mockImplementation(async (opts: any) => {
onText = opts.onText;
onToolStart = opts.onToolStart;
onToolEnd = opts.onToolEnd;
return { session: mockSession as any };
});
mockSession.prompt = vi.fn().mockImplementation(async () => {
onText?.("Heartbeat produced visible output");
onToolStart?.("read", { path: "README.md" });
onToolEnd?.("read", false, "done");
});
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(appendAgentLog).toHaveBeenCalledWith("FN-001", "Heartbeat produced visible output", "text", undefined, "executor");
expect(appendAgentLog).toHaveBeenCalledWith("FN-001", "read", "tool", "README.md", "executor");
expect(appendAgentLog).toHaveBeenCalledWith("FN-001", "read", "tool_result", "done", "executor");
expect(result.contextSnapshot?.taskId).toBe("FN-001");
expect(result.stdoutExcerpt).toContain("Heartbeat produced visible output");
});
});
describe("heartbeat_done tool", () => {
@@ -1387,6 +1422,33 @@ describe("HeartbeatMonitor", () => {
// Agent should be untracked
expect(monitor.getTrackedAgents()).not.toContain("agent-001");
});
it("flushes AgentLogger on execution failure", async () => {
const store = createStoreWithAgentForExec();
const mockSession = createMockAgentSession();
const flushSpy = vi.spyOn(AgentLogger.prototype, "flush").mockResolvedValue(undefined);
mockedCreateKbAgent.mockResolvedValue({
session: mockSession as any,
});
mockSession.prompt = vi.fn().mockRejectedValue(new Error("Prompt failed"));
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(flushSpy).toHaveBeenCalled();
});
it("flushes AgentLogger when session creation fails", async () => {
const store = createStoreWithAgentForExec();
const flushSpy = vi.spyOn(AgentLogger.prototype, "flush").mockResolvedValue(undefined);
mockedCreateKbAgent.mockRejectedValue(new Error("Model unavailable"));
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(flushSpy).toHaveBeenCalled();
});
});
describe("concurrency", () => {

86
packages/engine/src/agent-heartbeat.ts
View File

@@ -17,10 +17,11 @@
* - onTerminated: Called when an unresponsive agent is terminated
*/
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail } from "@fusion/core";
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
import { createTaskCreateTool, createTaskLogTool, taskCreateParams } from "./agent-tools.js";
import { AgentLogger } from "./agent-logger.js";
import { heartbeatLog } from "./logger.js";
// Lazy import for pi — avoids pulling the pi SDK into the module graph
@@ -519,8 +520,37 @@ export class HeartbeatMonitor {
return this.withAgentStartLock(agentId, async () => {
heartbeatLog.log(`Executing heartbeat for ${agentId} (source=${source})`);
let preloadedAgent: Agent | null = null;
try {
preloadedAgent = await this.store.getAgent(agentId);
} catch {
// If preloading fails, resolve again in the execution path below.
}
const resolvedTaskId = explicitTaskId ?? preloadedAgent?.taskId;
const runContextSnapshot = {
...(contextSnapshot ?? {}),
...(resolvedTaskId ? { taskId: resolvedTaskId } : {}),
};
// Start run
const run = await this.startRun(agentId, { source, triggerDetail, contextSnapshot });
const run = await this.startRun(agentId, {
source,
triggerDetail,
contextSnapshot: Object.keys(runContextSnapshot).length > 0 ? runContextSnapshot : undefined,
});
let agentLogger: AgentLogger | null = null;
const flushAgentLogger = async (): Promise<void> => {
if (!agentLogger) {
return;
}
try {
await agentLogger.flush();
} catch (error) {
heartbeatLog.warn(`Failed to flush heartbeat logs for ${agentId}: ${error instanceof Error ? error.message : String(error)}`);
}
};
try {
// Budget governance: check if agent can run
@@ -550,7 +580,7 @@ export class HeartbeatMonitor {
}
// Resolve agent
const agent = await this.store.getAgent(agentId);
const agent = preloadedAgent ?? await this.store.getAgent(agentId);
if (!agent) {
heartbeatLog.warn(`Agent ${agentId} not found — completing run as failed`);
await this.completeRun(agentId, run.id, {
@@ -562,6 +592,17 @@ export class HeartbeatMonitor {
// Resolve task assignment
const taskId = explicitTaskId ?? agent.taskId;
if (taskId && run.contextSnapshot?.taskId !== taskId) {
const updatedRun: AgentHeartbeatRun = {
...run,
contextSnapshot: {
...(run.contextSnapshot ?? {}),
taskId,
},
};
await this.store.saveRun(updatedRun);
}
if (!taskId) {
heartbeatLog.log(`Agent ${agentId} has no task assignment — graceful exit`);
await this.completeRun(agentId, run.id, {
@@ -597,9 +638,19 @@ export class HeartbeatMonitor {
}
// Track usage via callbacks
const STDOUT_EXCERPT_LIMIT = 4000;
let outputLength = 0;
let toolCallCount = 0;
let heartbeatSummary: string | undefined;
let stdoutExcerpt = "";
const appendStdoutExcerpt = (delta: string): void => {
if (stdoutExcerpt.length >= STDOUT_EXCERPT_LIMIT) {
return;
}
const remaining = STDOUT_EXCERPT_LIMIT - stdoutExcerpt.length;
stdoutExcerpt += delta.slice(0, remaining);
};
// Create heartbeat_done tool
const heartbeatDoneTool: ToolDefinition = {
@@ -628,6 +679,12 @@ export class HeartbeatMonitor {
const heartbeatTools = this.createHeartbeatTools(agentId, taskStore, taskId);
heartbeatTools.push(heartbeatDoneTool);
agentLogger = new AgentLogger({
store: taskStore,
taskId,
agent: agent.role as AgentRole,
});
// Create agent session
const { session } = await createKbAgent({
cwd: rootDir,
@@ -636,8 +693,21 @@ export class HeartbeatMonitor {
customTools: heartbeatTools,
defaultProvider: agent.runtimeConfig?.modelProvider as string | undefined,
defaultModelId: agent.runtimeConfig?.modelId as string | undefined,
onText: (delta) => { outputLength += delta.length; },
onToolEnd: () => { toolCallCount++; },
onText: (delta) => {
outputLength += delta.length;
appendStdoutExcerpt(delta);
agentLogger?.onText(delta);
},
onThinking: (delta) => {
agentLogger?.onThinking(delta);
},
onToolStart: (name, args) => {
agentLogger?.onToolStart(name, args);
},
onToolEnd: (name, isError, result) => {
toolCallCount++;
agentLogger?.onToolEnd(name, isError, result);
},
});
// Track for monitoring
@@ -664,23 +734,28 @@ export class HeartbeatMonitor {
// Estimate output tokens (rough: ~4 chars per token)
const estimatedOutputTokens = Math.ceil(outputLength / 4);
await flushAgentLogger();
// Complete run successfully
await this.completeRun(agentId, run.id, {
status: "completed",
usageJson: { inputTokens: 0, outputTokens: estimatedOutputTokens, cachedTokens: 0 },
resultJson: { summary: heartbeatSummary, toolCallCount },
stdoutExcerpt: stdoutExcerpt || undefined,
});
heartbeatLog.log(`Heartbeat completed for ${agentId} (${toolCallCount} tool calls, ~${estimatedOutputTokens} output tokens)`);
} catch (err) {
const errorMessage = err instanceof Error ? err.message : String(err);
heartbeatLog.error(`Heartbeat execution failed for ${agentId}: ${errorMessage}`);
await flushAgentLogger();
await this.completeRun(agentId, run.id, {
status: "failed",
stderrExcerpt: errorMessage,
stdoutExcerpt: stdoutExcerpt || undefined,
});
} finally {
await flushAgentLogger();
this.untrackAgent(agentId);
try { session.dispose(); } catch { /* ignore */ }
}
@@ -689,6 +764,7 @@ export class HeartbeatMonitor {
} catch (err) {
const errorMessage = err instanceof Error ? err.message : String(err);
heartbeatLog.error(`Heartbeat execution error for ${agentId}: ${errorMessage}`);
await flushAgentLogger();
// Attempt to complete the run as failed if it's still active
try {

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

@@ -854,6 +854,7 @@ export class TaskExecutor {
executorLog.log(`${task.id}: using step-session mode (maxParallel=${settings.maxParallelSteps ?? 2})`);
const stepExecutor = new StepSessionExecutor({
store: this.store,
taskDetail: detail,
worktreePath,
rootDir: this.rootDir,

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

@@ -6,7 +6,8 @@ import {
buildStepPrompt,
StepSessionExecutor,
} from "./step-session-executor.js";
import type { TaskDetail, Settings } from "@fusion/core";
import { AgentLogger } from "./agent-logger.js";
import type { TaskDetail, Settings, TaskStore } from "@fusion/core";
// ── Shared test fixtures ──────────────────────────────────────────────
@@ -1603,4 +1604,76 @@ describe("StepSessionExecutor", () => {
await executor.cleanup();
});
});
describe("logging", () => {
it("appends agent logs during step execution", async () => {
const task = makeTaskDetail({
prompt: makeStepPrompt("FN-001", 1),
steps: [{ name: "Step 0", status: "pending" }],
});
const settings = makeSettings({ maxParallelSteps: 1 });
const appendAgentLog = vi.fn().mockResolvedValue(undefined);
const store = { appendAgentLog } as unknown as TaskStore;
let onText: ((delta: string) => void) | undefined;
let onToolStart: ((name: string, args?: Record<string, unknown>) => void) | undefined;
let onToolEnd: ((name: string, isError: boolean, result?: unknown) => void) | undefined;
const session = makeMockSession(async () => {
onText?.("step output");
onToolStart?.("read", { path: "src/foo.ts" });
onToolEnd?.("read", false, "ok");
});
mockedCreateKbAgent.mockImplementation(async (opts: any) => {
onText = opts.onText;
onToolStart = opts.onToolStart;
onToolEnd = opts.onToolEnd;
return { session } as any;
});
const executor = new StepSessionExecutor({
store,
taskDetail: task,
worktreePath: "/project/.worktrees/main",
rootDir: "/project",
settings,
});
await executor.executeAll();
expect(appendAgentLog).toHaveBeenCalledWith("FN-001", "step output", "text", undefined, "executor");
expect(appendAgentLog).toHaveBeenCalledWith("FN-001", "read", "tool", "src/foo.ts", "executor");
expect(appendAgentLog).toHaveBeenCalledWith("FN-001", "read", "tool_result", "ok", "executor");
});
it("flushes AgentLogger in attempt finally block", async () => {
const task = makeTaskDetail({
prompt: makeStepPrompt("FN-001", 1),
steps: [{ name: "Step 0", status: "pending" }],
});
const settings = makeSettings({ maxParallelSteps: 1 });
const store = { appendAgentLog: vi.fn().mockResolvedValue(undefined) } as unknown as TaskStore;
const flushSpy = vi.spyOn(AgentLogger.prototype, "flush").mockResolvedValue(undefined);
const session = makeMockSession(async () => {
throw new Error("step failed");
});
mockedCreateKbAgent.mockResolvedValue({ session } as any);
const executor = new StepSessionExecutor({
store,
taskDetail: task,
worktreePath: "/project/.worktrees/main",
rootDir: "/project",
settings,
});
const resultPromise = executor.executeAll();
await vi.advanceTimersByTimeAsync(30_000);
await resultPromise;
expect(flushSpy).toHaveBeenCalled();
});
});
});

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

@@ -5,16 +5,16 @@
* parallel execution for non-conflicting steps (via git worktree isolation),
* and clean lifecycle management (pause, cleanup).
*
* The class is a standalone engine subsystem. It does not depend on TaskStore
* or any integration layer — it receives TaskDetail (read-only) and emits
* results via callbacks.
* The class is a standalone engine subsystem with minimal integration surface.
* It receives TaskDetail (read-only), a TaskStore for agent logs, and emits
* execution progress via callbacks.
*/
import { execSync } from "node:child_process";
import { existsSync } from "node:fs";
import { join } from "node:path";
import type { AgentSession, ToolDefinition } from "@mariozechner/pi-coding-agent";
import type { TaskDetail, Settings, TaskStep, StepStatus } from "@fusion/core";
import type { TaskDetail, Settings, TaskStep, StepStatus, TaskStore } from "@fusion/core";
import { createKbAgent, promptWithFallback, describeModel } from "./pi.js";
import { generateWorktreeName } from "./worktree-names.js";
@@ -49,6 +49,8 @@ export interface ParallelWave {
/** Options for creating a StepSessionExecutor. */
export interface StepSessionExecutorOptions {
/** Optional task store used to persist agent logs for each step session. */
store?: TaskStore;
/** The task to execute (read-only). */
taskDetail: TaskDetail;
/** Path to the primary git worktree for this task. */
@@ -463,6 +465,11 @@ interface SessionHandle {
dispose: () => void;
}
/** Fallback store used when step logging persistence is not configured. */
const NOOP_TASK_STORE: Pick<TaskStore, "appendAgentLog"> = {
appendAgentLog: async () => undefined,
};
/**
* StepSessionExecutor — runs each task step in its own fresh agent session.
*
@@ -473,10 +480,10 @@ interface SessionHandle {
* - **Per-step retry**: failed steps retry up to 3 times with exponential backoff
* - **Clean lifecycle**: pause via `terminateAllSessions()`, cleanup via `cleanup()`
*
* The class is a standalone engine subsystem — it does not depend on `TaskStore`.
* It receives `TaskDetail` (read-only) in its options and emits results via
* callbacks (`onStepStart`, `onStepComplete`). The integration layer (FN-1040)
* is responsible for persisting step status updates and agent logs.
* The class is a standalone engine subsystem.
* It receives `TaskDetail` (read-only) and an optional `TaskStore` in its options,
* then emits results via callbacks (`onStepStart`, `onStepComplete`).
* The integration layer (FN-1040) is responsible for persisting step status updates.
*
* @example
* ```ts
@@ -499,6 +506,7 @@ interface SessionHandle {
*/
export class StepSessionExecutor {
private options: StepSessionExecutorOptions;
private store: TaskStore;
private activeSessions: Map<number, SessionHandle> = new Map();
private parallelWorktrees: Map<number, string> = new Map();
private parallelBranches: Map<number, string> = new Map();
@@ -508,6 +516,7 @@ export class StepSessionExecutor {
constructor(options: StepSessionExecutorOptions) {
this.options = options;
this.store = options.store ?? (NOOP_TASK_STORE as TaskStore);
// Clamp maxParallelSteps to 1–4 range
this.maxParallel = Math.max(1, Math.min(4, options.settings.maxParallelSteps ?? 2));
}
@@ -684,29 +693,43 @@ export class StepSessionExecutor {
await sleep(delay);
}
const agentLogger = new AgentLogger({
store: this.store,
taskId: taskDetail.id,
agent: "executor",
});
let session: AgentSession | null = null;
try {
// Create fresh agent session for this attempt
const { session } = await createKbAgent({
const createResult = await createKbAgent({
cwd: worktreePath,
systemPrompt: `You are an AI agent executing step ${stepIndex} of task ${taskDetail.id}. Follow instructions precisely.`,
defaultProvider: taskDetail.modelProvider,
defaultModelId: taskDetail.modelId,
defaultThinkingLevel: taskDetail.thinkingLevel,
onText: () => {
onText: (delta) => {
agentLogger.onText(delta);
stuckTaskDetector?.recordActivity(trackingKey);
},
onToolStart: () => {
onThinking: (delta) => {
agentLogger.onThinking(delta);
},
onToolStart: (name, args) => {
agentLogger.onToolStart(name, args);
stuckTaskDetector?.recordActivity(trackingKey);
},
onToolEnd: () => {
onToolEnd: (name, isError, result) => {
agentLogger.onToolEnd(name, isError, result);
stuckTaskDetector?.recordActivity(trackingKey);
},
});
session = createResult.session;
// Track session for termination and stuck-task detection
const handle: SessionHandle = { dispose: () => session.dispose() };
const handle: SessionHandle = { dispose: () => session?.dispose() };
this.activeSessions.set(stepIndex, handle);
stuckTaskDetector?.trackTask(trackingKey, { dispose: () => session.dispose() });
stuckTaskDetector?.trackTask(trackingKey, { dispose: () => session?.dispose() });
stepExecLog.log(
`Step ${stepIndex} attempt ${attempt + 1} session created ` +
@@ -716,19 +739,10 @@ export class StepSessionExecutor {
// Send prompt
await promptWithFallback(session, stepPrompt);
// Success — clean up session
this.activeSessions.delete(stepIndex);
stuckTaskDetector?.untrackTask(trackingKey);
try { session.dispose(); } catch { /* best-effort */ }
const result: StepResult = { stepIndex, success: true, retries };
this.options.onStepComplete?.(stepIndex, result);
return result;
} catch (err) {
// Clean up failed session
this.activeSessions.delete(stepIndex);
stuckTaskDetector?.untrackTask(trackingKey);
const errorMessage = err instanceof Error ? err.message : String(err);
stepExecLog.warn(
`Step ${stepIndex} attempt ${attempt + 1} failed: ${errorMessage}`,
@@ -745,6 +759,21 @@ export class StepSessionExecutor {
this.options.onStepComplete?.(stepIndex, result);
return result;
}
} finally {
try {
await agentLogger.flush();
} catch (err) {
const flushError = err instanceof Error ? err.message : String(err);
stepExecLog.warn(`Failed to flush agent logs for step ${stepIndex}: ${flushError}`);
}
this.activeSessions.delete(stepIndex);
stuckTaskDetector?.untrackTask(trackingKey);
try {
session?.dispose();
} catch {
/* best-effort */
}
}
}