refactor(HAI-093): extract AgentLogger abstraction and rewire agents

- Add AgentLogger class encapsulating log file management and structured logging
- Add comprehensive unit tests for AgentLogger
- Rewire executor, triage, and merger to use AgentLogger instead of inline logging
- Reduce duplication across agent modules
- Export AgentLogger from engine package index
This commit is contained in:
Dustin Byrne
2026-03-26 19:29:17 -04:00
parent eac1eda49a
commit 121b509686
8 changed files with 395 additions and 155 deletions

View File

@@ -0,0 +1,193 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { AgentLogger, summarizeToolArgs } from "./agent-logger.js";
import type { TaskStore } from "@hai/core";
// ── summarizeToolArgs tests ──────────────────────────────────────────
describe("summarizeToolArgs", () => {
it("returns bash command", () => {
expect(summarizeToolArgs("Bash", { command: "ls -la" })).toBe("ls -la");
expect(summarizeToolArgs("bash", { command: "echo hello" })).toBe("echo hello");
});
it("truncates long bash commands at 80 chars", () => {
const longCmd = "a".repeat(100);
const result = summarizeToolArgs("Bash", { command: longCmd });
expect(result).toBe("a".repeat(80) + "…");
});
it("returns file path for Read/Edit/Write", () => {
expect(summarizeToolArgs("Read", { path: "src/types.ts" })).toBe("src/types.ts");
expect(summarizeToolArgs("edit", { path: "src/store.ts" })).toBe("src/store.ts");
expect(summarizeToolArgs("Write", { path: "out.txt", content: "data" })).toBe("out.txt");
});
it("falls back to first short string arg for unknown tools", () => {
expect(summarizeToolArgs("task_update", { step: 1, status: "done" })).toBe("done");
});
it("returns undefined when no args or empty args", () => {
expect(summarizeToolArgs("Bash")).toBeUndefined();
expect(summarizeToolArgs("Bash", {})).toBeUndefined();
});
it("returns undefined for non-string values only", () => {
expect(summarizeToolArgs("unknown", { count: 42, flag: true })).toBeUndefined();
});
});
// ── AgentLogger tests ────────────────────────────────────────────────
function createMockStore() {
return {
appendAgentLog: vi.fn().mockResolvedValue(undefined),
} as unknown as TaskStore;
}
describe("AgentLogger", () => {
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
it("buffers text and flushes on size threshold", async () => {
const store = createMockStore();
const logger = new AgentLogger({
store,
taskId: "HAI-001",
flushSizeBytes: 10,
flushIntervalMs: 500,
});
// Under threshold — no flush yet
logger.onText("hello");
expect(store.appendAgentLog).not.toHaveBeenCalled();
// Over threshold — triggers flush
logger.onText("worldextra");
// Allow async flush
await vi.advanceTimersByTimeAsync(0);
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-001", "helloworldextra", "text");
});
it("flushes on timer when under size threshold", async () => {
const store = createMockStore();
const logger = new AgentLogger({
store,
taskId: "HAI-002",
flushSizeBytes: 1024,
flushIntervalMs: 500,
});
logger.onText("small");
expect(store.appendAgentLog).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(500);
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-002", "small", "text");
});
it("flushes text before logging tool start", async () => {
const store = createMockStore();
const logger = new AgentLogger({
store,
taskId: "HAI-003",
flushSizeBytes: 1024,
});
logger.onText("pending text");
logger.onToolStart("Bash", { command: "ls" });
await vi.advanceTimersByTimeAsync(0);
const calls = (store.appendAgentLog as ReturnType<typeof vi.fn>).mock.calls;
expect(calls.length).toBe(2);
// Text flushed first
expect(calls[0]).toEqual(["HAI-003", "pending text", "text"]);
// Tool logged second with detail
expect(calls[1]).toEqual(["HAI-003", "Bash", "tool", "ls"]);
});
it("logs tool detail using summarizeToolArgs", async () => {
const store = createMockStore();
const logger = new AgentLogger({ store, taskId: "HAI-004" });
logger.onToolStart("Read", { path: "src/index.ts" });
await vi.advanceTimersByTimeAsync(0);
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-004", "Read", "tool", "src/index.ts");
});
it("logs tool with undefined detail for unknown args", async () => {
const store = createMockStore();
const logger = new AgentLogger({ store, taskId: "HAI-005" });
logger.onToolStart("task_done", { count: 42 });
await vi.advanceTimersByTimeAsync(0);
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-005", "task_done", "tool", undefined);
});
it("flush() clears timer and writes remaining text", async () => {
const store = createMockStore();
const logger = new AgentLogger({
store,
taskId: "HAI-006",
flushSizeBytes: 1024,
flushIntervalMs: 500,
});
logger.onText("remaining");
await logger.flush();
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-006", "remaining", "text");
});
it("flush() is safe to call when buffer is empty", async () => {
const store = createMockStore();
const logger = new AgentLogger({ store, taskId: "HAI-007" });
await logger.flush();
expect(store.appendAgentLog).not.toHaveBeenCalled();
});
it("invokes external callbacks alongside logging", async () => {
const store = createMockStore();
const onAgentText = vi.fn();
const onAgentTool = vi.fn();
const logger = new AgentLogger({
store,
taskId: "HAI-008",
onAgentText,
onAgentTool,
});
logger.onText("delta");
expect(onAgentText).toHaveBeenCalledWith("HAI-008", "delta");
logger.onToolStart("Bash", { command: "echo hi" });
expect(onAgentTool).toHaveBeenCalledWith("HAI-008", "Bash");
});
it("does not schedule multiple timers for consecutive small writes", async () => {
const store = createMockStore();
const logger = new AgentLogger({
store,
taskId: "HAI-009",
flushSizeBytes: 1024,
flushIntervalMs: 500,
});
logger.onText("a");
logger.onText("b");
logger.onText("c");
await vi.advanceTimersByTimeAsync(500);
// All text should be flushed in a single call
expect(store.appendAgentLog).toHaveBeenCalledTimes(1);
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-009", "abc", "text");
});
});

View File

@@ -0,0 +1,157 @@
import type { TaskStore } from "@hai/core";
/** Default byte threshold before an automatic flush. */
const FLUSH_SIZE_BYTES = 1024;
/** Default timer interval (ms) for periodic flush of small writes. */
const FLUSH_INTERVAL_MS = 500;
/**
* Produce a short human-readable summary from tool arguments.
* Returns `undefined` for unknown tools or when no meaningful arg is found.
*/
export function summarizeToolArgs(name: string, args?: Record<string, unknown>): string | undefined {
if (!args) return undefined;
const lowerName = name.toLowerCase();
if (lowerName === "bash") {
const cmd = args.command;
if (typeof cmd === "string") {
return cmd.length > 80 ? cmd.slice(0, 80) + "…" : cmd;
}
}
if (lowerName === "read" || lowerName === "edit" || lowerName === "write") {
const p = args.path;
if (typeof p === "string") return p;
}
// Fallback: return first string-valued arg if short enough
for (const val of Object.values(args)) {
if (typeof val === "string" && val.length <= 80) return val;
}
return undefined;
}
/**
* Options for creating an {@link AgentLogger}.
*/
export interface AgentLoggerOptions {
/** The task store used to persist agent log entries. */
store: TaskStore;
/** The task ID this logger is associated with. */
taskId: string;
/** Optional callback invoked alongside text logging (e.g. for SSE streaming). */
onAgentText?: (taskId: string, delta: string) => void;
/** Optional callback invoked alongside tool logging (e.g. for SSE streaming). */
onAgentTool?: (taskId: string, toolName: string) => void;
/** Byte threshold for automatic flush. Defaults to 1024. */
flushSizeBytes?: number;
/** Timer interval (ms) for periodic flush. Defaults to 500. */
flushIntervalMs?: number;
}
/**
* Buffers agent text output and flushes it to the task store periodically
* or when a size threshold is reached. Also handles tool-start logging with
* detailed argument summaries via {@link summarizeToolArgs}.
*
* Produces `onText` and `onToolStart` callbacks compatible with
* `createHaiAgent`'s `AgentOptions` interface.
*
* @example
* ```ts
* const logger = new AgentLogger({ store, taskId, onAgentText, onAgentTool });
* const { session } = await createHaiAgent({
* cwd: worktreePath,
* onText: logger.onText,
* onToolStart: logger.onToolStart,
* // ...
* });
* try {
* await session.prompt(prompt);
* } finally {
* await logger.flush();
* session.dispose();
* }
* ```
*/
export class AgentLogger {
private textBuffer = "";
private flushTimer: ReturnType<typeof setTimeout> | null = null;
private readonly flushSizeBytes: number;
private readonly flushIntervalMs: number;
private readonly store: TaskStore;
private readonly taskId: string;
private readonly externalTextCb?: (taskId: string, delta: string) => void;
private readonly externalToolCb?: (taskId: string, toolName: string) => void;
constructor(options: AgentLoggerOptions) {
this.store = options.store;
this.taskId = options.taskId;
this.externalTextCb = options.onAgentText;
this.externalToolCb = options.onAgentTool;
this.flushSizeBytes = options.flushSizeBytes ?? FLUSH_SIZE_BYTES;
this.flushIntervalMs = options.flushIntervalMs ?? FLUSH_INTERVAL_MS;
// Bind callbacks so they can be passed directly as function references
this.onText = this.onText.bind(this);
this.onToolStart = this.onToolStart.bind(this);
}
/**
* Callback for agent text deltas. Buffers text and flushes on size
* threshold or after a timer interval. Compatible with `AgentOptions.onText`.
*/
onText(delta: string): void {
this.externalTextCb?.(this.taskId, delta);
this.textBuffer += delta;
if (this.textBuffer.length >= this.flushSizeBytes) {
if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; }
this.flushTextBuffer();
} else {
this.scheduleFlush();
}
}
/**
* Callback for tool invocation starts. Flushes pending text, then logs the
* tool name with a detail summary. Compatible with `AgentOptions.onToolStart`.
*/
onToolStart(name: string, args?: Record<string, unknown>): void {
this.externalToolCb?.(this.taskId, name);
// Flush any pending text before recording the tool entry
if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; }
this.flushTextBuffer();
const detail = summarizeToolArgs(name, args);
this.store.appendAgentLog(this.taskId, name, "tool", detail).catch(() => {});
}
/**
* Flush any remaining buffered text and clear the timer.
* Call this in a `finally` block before disposing the agent session.
*/
async flush(): Promise<void> {
if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; }
await this.flushTextBuffer();
}
// ── Internal helpers ───────────────────────────────────────────────
private flushTextBuffer(): Promise<void> {
if (this.textBuffer.length === 0) return Promise.resolve();
const chunk = this.textBuffer;
this.textBuffer = "";
return this.store.appendAgentLog(this.taskId, chunk, "text").catch(() => {
/* best-effort persistence */
});
}
private scheduleFlush(): void {
if (this.flushTimer) return;
this.flushTimer = setTimeout(() => {
this.flushTimer = null;
this.flushTextBuffer();
}, this.flushIntervalMs);
}
}

View File

@@ -10,37 +10,13 @@ import { reviewStep } from "./reviewer.js";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import type { AgentSemaphore } from "./concurrency.js";
import type { WorktreePool } from "./worktree-pool.js";
import { AgentLogger } from "./agent-logger.js";
// Re-export for backward compatibility (tests import from executor.ts)
export { summarizeToolArgs } from "./agent-logger.js";
const STEP_STATUSES: StepStatus[] = ["pending", "in-progress", "done", "skipped"];
/**
* Produce a short human-readable summary from tool arguments.
* Returns `undefined` for unknown tools or when no meaningful arg is found.
*/
export function summarizeToolArgs(name: string, args?: Record<string, unknown>): string | undefined {
if (!args) return undefined;
const lowerName = name.toLowerCase();
if (lowerName === "bash") {
const cmd = args.command;
if (typeof cmd === "string") {
return cmd.length > 80 ? cmd.slice(0, 80) + "…" : cmd;
}
}
if (lowerName === "read" || lowerName === "edit" || lowerName === "write") {
const p = args.path;
if (typeof p === "string") return p;
}
// Fallback: return first string-valued arg if short enough
for (const val of Object.values(args)) {
if (typeof val === "string" && val.length <= 80) return val;
}
return undefined;
}
// ── Tool parameter schemas (module-level for reuse in ToolDefinition generics) ──
const taskUpdateParams = Type.Object({
@@ -344,30 +320,12 @@ export class TaskExecutor {
this.createReviewStepTool(task.id, worktreePath, detail.prompt),
];
// ── Agent log buffering ──────────────────────────────────────────
// Buffer text deltas and flush to disk periodically to avoid
// excessive I/O from many small writes.
let textBuffer = "";
let flushTimer: ReturnType<typeof setTimeout> | null = null;
const FLUSH_INTERVAL_MS = 500;
const FLUSH_SIZE_BYTES = 1024;
const flushTextBuffer = async () => {
if (textBuffer.length === 0) return;
const chunk = textBuffer;
textBuffer = "";
try {
await this.store.appendAgentLog(task.id, chunk, "text");
} catch { /* best-effort persistence */ }
};
const scheduleFlush = () => {
if (flushTimer) return;
flushTimer = setTimeout(async () => {
flushTimer = null;
await flushTextBuffer();
}, FLUSH_INTERVAL_MS);
};
const agentLogger = new AgentLogger({
store: this.store,
taskId: task.id,
onAgentText: this.options.onAgentText,
onAgentTool: this.options.onAgentTool,
});
const agentWork = async () => {
const { session } = await createHaiAgent({
@@ -375,24 +333,8 @@ export class TaskExecutor {
systemPrompt: EXECUTOR_SYSTEM_PROMPT,
tools: "coding",
customTools,
onText: (delta) => {
this.options.onAgentText?.(task.id, delta);
textBuffer += delta;
if (textBuffer.length >= FLUSH_SIZE_BYTES) {
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
flushTextBuffer();
} else {
scheduleFlush();
}
},
onToolStart: (name, args) => {
this.options.onAgentTool?.(task.id, name);
// Flush any pending text before recording the tool entry
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
flushTextBuffer();
const detail = summarizeToolArgs(name, args);
this.store.appendAgentLog(task.id, name, "tool", detail).catch(() => {});
},
onText: agentLogger.onText,
onToolStart: agentLogger.onToolStart,
});
try {
@@ -410,9 +352,7 @@ export class TaskExecutor {
this.options.onComplete?.(task);
}
} finally {
// Flush remaining buffered text before disposing the session
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
await flushTextBuffer();
await agentLogger.flush();
session.dispose();
}
};

View File

@@ -1,3 +1,4 @@
export { AgentLogger, type AgentLoggerOptions, summarizeToolArgs } from "./agent-logger.js";
export { AgentSemaphore } from "./concurrency.js";
export { TriageProcessor, type TriageProcessorOptions } from "./triage.js";
export { TaskExecutor, type TaskExecutorOptions } from "./executor.js";

View File

@@ -242,7 +242,7 @@ describe("aiMergeTask — agent log persistence", () => {
await aiMergeTask(store, "/tmp/root", "HAI-050");
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-050", "Bash", "tool");
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-050", "Bash", "tool", "git status");
});
it("still fires onAgentText callback alongside logging", async () => {

View File

@@ -3,6 +3,7 @@ import { existsSync } from "node:fs";
import type { TaskStore, Task, MergeResult } from "@hai/core";
import { createHaiAgent } from "./pi.js";
import type { WorktreePool } from "./worktree-pool.js";
import { AgentLogger } from "./agent-logger.js";
const MERGE_SYSTEM_PROMPT = `You are a merge agent for "hai", an AI-orchestrated task board.
@@ -185,49 +186,24 @@ export async function aiMergeTask(
`[merger] ${taskId}: ${hasConflicts ? "resolving conflicts + " : ""}writing commit message`,
);
// ── Agent log buffering ──────────────────────────────────────────
let textBuffer = "";
let flushTimer: ReturnType<typeof setTimeout> | null = null;
const FLUSH_INTERVAL_MS = 500;
const FLUSH_SIZE_BYTES = 1024;
const flushTextBuffer = async () => {
if (textBuffer.length === 0) return;
const chunk = textBuffer;
textBuffer = "";
try {
await store.appendAgentLog(taskId, chunk, "text");
} catch { /* best-effort persistence */ }
};
const scheduleFlush = () => {
if (flushTimer) return;
flushTimer = setTimeout(async () => {
flushTimer = null;
await flushTextBuffer();
}, FLUSH_INTERVAL_MS);
};
const agentLogger = new AgentLogger({
store,
taskId,
// Merger callbacks don't include taskId — wrap to match AgentLogger signature
onAgentText: options.onAgentText
? (_id, delta) => options.onAgentText!(delta)
: undefined,
onAgentTool: options.onAgentTool
? (_id, name) => options.onAgentTool!(name)
: undefined,
});
const { session } = await createHaiAgent({
cwd: rootDir,
systemPrompt: MERGE_SYSTEM_PROMPT,
tools: "coding",
onText: (delta) => {
options.onAgentText?.(delta);
textBuffer += delta;
if (textBuffer.length >= FLUSH_SIZE_BYTES) {
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
flushTextBuffer();
} else {
scheduleFlush();
}
},
onToolStart: (name, _args) => {
options.onAgentTool?.(name);
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
flushTextBuffer();
store.appendAgentLog(taskId, name, "tool").catch(() => {});
},
onText: agentLogger.onText,
onToolStart: agentLogger.onToolStart,
});
try {
@@ -258,8 +234,7 @@ export async function aiMergeTask(
} catch { /* */ }
throw new Error(`AI merge failed for ${taskId}: ${err.message}`);
} finally {
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
await flushTextBuffer();
await agentLogger.flush();
session.dispose();
}

View File

@@ -493,7 +493,7 @@ describe("TriageProcessor agent log persistence", () => {
updatedAt: new Date().toISOString(),
});
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-001", "Read", "tool");
expect(store.appendAgentLog).toHaveBeenCalledWith("HAI-001", "Read", "tool", "foo.ts");
});
it("still fires onAgentText callback alongside logging", async () => {

View File

@@ -4,6 +4,7 @@ import { Type, type Static } from "@mariozechner/pi-ai";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { createHaiAgent } from "./pi.js";
import type { AgentSemaphore } from "./concurrency.js";
import { AgentLogger } from "./agent-logger.js";
const TRIAGE_SYSTEM_PROMPT = `You are a task specification agent for "hai", an AI-orchestrated task board.
@@ -251,50 +252,24 @@ export class TriageProcessor {
const promptPath = `.hai/tasks/${task.id}/PROMPT.md`;
const agentWork = async () => {
// ── Agent log buffering ──────────────────────────────────────────
let textBuffer = "";
let flushTimer: ReturnType<typeof setTimeout> | null = null;
const FLUSH_INTERVAL_MS = 500;
const FLUSH_SIZE_BYTES = 1024;
const flushTextBuffer = async () => {
if (textBuffer.length === 0) return;
const chunk = textBuffer;
textBuffer = "";
try {
await this.store.appendAgentLog(task.id, chunk, "text");
} catch { /* best-effort persistence */ }
};
const scheduleFlush = () => {
if (flushTimer) return;
flushTimer = setTimeout(async () => {
flushTimer = null;
await flushTextBuffer();
}, FLUSH_INTERVAL_MS);
};
const agentLogger = new AgentLogger({
store: this.store,
taskId: task.id,
onAgentText: this.options.onAgentText
? (id, delta) => this.options.onAgentText!(id, delta)
: undefined,
onAgentTool: (_id, name) => {
console.log(`[triage] ${task.id} tool: ${name}`);
},
});
const { session } = await createHaiAgent({
cwd: this.rootDir,
systemPrompt: TRIAGE_SYSTEM_PROMPT,
tools: "coding",
customTools: this.createTriageTools(),
onText: (delta) => {
this.options.onAgentText?.(task.id, delta);
textBuffer += delta;
if (textBuffer.length >= FLUSH_SIZE_BYTES) {
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
flushTextBuffer();
} else {
scheduleFlush();
}
},
onToolStart: (name, _args) => {
console.log(`[triage] ${task.id} tool: ${name}`);
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
flushTextBuffer();
this.store.appendAgentLog(task.id, name, "tool").catch(() => {});
},
onText: agentLogger.onText,
onToolStart: agentLogger.onToolStart,
});
try {
@@ -326,8 +301,7 @@ export class TriageProcessor {
this.options.onSpecifyComplete?.(task);
}
} finally {
if (flushTimer) { clearTimeout(flushTimer); flushTimer = null; }
await flushTextBuffer();
await agentLogger.flush();
session.dispose();
}
};