feat(FN-2979): add droid CLI probe module for node diagnostics
Added a new `droid-cli-probe` module to the dashboard package with test coverage, implementing a CLI probe capability for the droid system. Fusion-Task-Id: FN-2979
This commit is contained in:
Fusion
@@ -55,4 +55,5 @@ These tools are **not** part of the pi extension's user-invokable `extension.ts`
|
||||
|
||||
| Tool | Purpose | Parameters |
|
||||
|---|---|---|
|
||||
| `fn_identity` | Return loaded soul/instructions/memory summary for this heartbeat tick (must be called first) | none |
|
||||
| `fn_heartbeat_done` | Signal end of heartbeat run with optional summary | `summary?` (string) |
|
||||
|
||||
@@ -54,6 +54,11 @@ describe("TaskStore", () => {
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
// Ensure teardown always runs on real timers. Some tests in this file use
|
||||
// timeout/retry-based fs cleanup paths that can stall indefinitely if fake
|
||||
// timers were left enabled by a preceding test.
|
||||
vi.useRealTimers();
|
||||
|
||||
// Some watcher/polling tests can leave an in-flight poll tick queued right
|
||||
// before teardown. Stop watching first and yield once so pending callbacks
|
||||
// settle before removing temp dirs.
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
* edited as normal project files.
|
||||
*/
|
||||
|
||||
import { mkdir, readFile, writeFile, readdir, unlink, rename, access } from "node:fs/promises";
|
||||
import { mkdir, readFile, writeFile, readdir, unlink, rename, access, appendFile } from "node:fs/promises";
|
||||
import { constants as fsConstants } from "node:fs";
|
||||
import { basename, dirname, join, resolve } from "node:path";
|
||||
import { randomUUID, randomBytes, createHash } from "node:crypto";
|
||||
@@ -36,6 +36,7 @@ import type {
|
||||
AgentRatingSummary,
|
||||
AgentRatingInput,
|
||||
Task,
|
||||
AgentLogEntry,
|
||||
} from "./types.js";
|
||||
import { AGENT_VALID_TRANSITIONS, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath } from "./types.js";
|
||||
import type { RunMutationContext } from "./types.js";
|
||||
@@ -1790,6 +1791,70 @@ export class AgentStore extends EventEmitter {
|
||||
.filter((run): run is AgentHeartbeatRun => run !== null);
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
// Run-scoped log storage (JSONL files alongside run JSON in agentsDir)
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/** Maximum byte size for any single log entry field (64 KB) to bound disk growth. */
|
||||
private static readonly RUN_LOG_ENTRY_MAX_BYTES = 64 * 1024;
|
||||
|
||||
/** Return the path to the JSONL run-log file for a given agent/run pair. */
|
||||
private runLogPath(agentId: string, runId: string): string {
|
||||
return join(this.agentsDir, `${agentId}-runlogs-${runId}.jsonl`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Append a single {@link AgentLogEntry} to the JSONL run log for the given run.
|
||||
* Individual `text` and `detail` fields are capped at 64 KB so one large tool
|
||||
* result cannot grow the file unboundedly.
|
||||
* @param agentId - The agent ID
|
||||
* @param runId - The run ID
|
||||
* @param entry - The log entry to append
|
||||
*/
|
||||
async appendRunLog(agentId: string, runId: string, entry: AgentLogEntry): Promise<void> {
|
||||
const cap = AgentStore.RUN_LOG_ENTRY_MAX_BYTES;
|
||||
const safeEntry: AgentLogEntry = {
|
||||
...entry,
|
||||
text: entry.text.length > cap ? `${entry.text.slice(0, cap)}\n\n... (truncated, ${entry.text.length} chars)` : entry.text,
|
||||
...(entry.detail !== undefined && {
|
||||
detail: entry.detail.length > cap ? `${entry.detail.slice(0, cap)}\n\n... (truncated, ${entry.detail.length} chars)` : entry.detail,
|
||||
}),
|
||||
};
|
||||
const line = JSON.stringify(safeEntry) + "\n";
|
||||
await appendFile(this.runLogPath(agentId, runId), line, "utf-8");
|
||||
}
|
||||
|
||||
/**
|
||||
* Read all log entries for a given run from its JSONL file.
|
||||
* Returns an empty array when the file does not exist (e.g., the run had no
|
||||
* logs or was recorded before this feature was added).
|
||||
* @param agentId - The agent ID
|
||||
* @param runId - The run ID
|
||||
* @param opts.limit - Optional maximum number of entries to return (newest-first capped)
|
||||
*/
|
||||
async getRunLogs(agentId: string, runId: string, opts?: { limit?: number }): Promise<AgentLogEntry[]> {
|
||||
const filePath = this.runLogPath(agentId, runId);
|
||||
let raw: string;
|
||||
try {
|
||||
raw = await readFile(filePath, "utf-8");
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
const lines = raw.split("\n").filter((l) => l.trim().length > 0);
|
||||
const entries: AgentLogEntry[] = [];
|
||||
for (const line of lines) {
|
||||
try {
|
||||
entries.push(JSON.parse(line) as AgentLogEntry);
|
||||
} catch {
|
||||
// Skip malformed lines — append-only means partial writes can occur on crash
|
||||
}
|
||||
}
|
||||
if (opts?.limit !== undefined && entries.length > opts.limit) {
|
||||
return entries.slice(entries.length - opts.limit);
|
||||
}
|
||||
return entries;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the most recently persisted blocked-task dedup state for an agent.
|
||||
*/
|
||||
|
||||
@@ -2769,6 +2769,12 @@ export interface AgentHeartbeatRun {
|
||||
stdoutExcerpt?: string;
|
||||
/** Excerpt of stderr output */
|
||||
stderrExcerpt?: string;
|
||||
/** Full assembled system prompt sent to the LLM for this run (truncated to 100,000 chars). */
|
||||
systemPrompt?: string;
|
||||
/** Full per-tick execution prompt sent to the LLM for this run (truncated to 100,000 chars). */
|
||||
executionPrompt?: string;
|
||||
/** Whether a custom heartbeat procedure was loaded ("custom") or the built-in default was used ("default"). */
|
||||
heartbeatProcedureSource?: "default" | "custom";
|
||||
}
|
||||
|
||||
/** Capabilities/roles an agent can have */
|
||||
|
||||
@@ -1288,6 +1288,11 @@ function RunsTab({
|
||||
<StatusIcon size={14} className={statusInfo.color} style={run.status === "active" ? { color: statusInfo.color } : undefined} />
|
||||
{run.status}
|
||||
</span>
|
||||
{run.heartbeatProcedureSource === "custom" && (
|
||||
<span className="badge" style={{ fontSize: "10px", padding: "1px 6px" }}>
|
||||
Heartbeat: custom
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
</div>
|
||||
<div className="run-details">
|
||||
@@ -1312,6 +1317,30 @@ function RunsTab({
|
||||
</div>
|
||||
) : detailRun && (
|
||||
<div className="run-output-sections">
|
||||
{/* System Prompt */}
|
||||
<div className="run-output-section">
|
||||
<details>
|
||||
<summary className="run-output-label" style={{ cursor: "pointer", userSelect: "none" }}>System Prompt</summary>
|
||||
{detailRun.systemPrompt ? (
|
||||
<pre className="run-output-panel">{detailRun.systemPrompt}</pre>
|
||||
) : (
|
||||
<div className="text-muted run-output-empty">System prompt not captured for this run</div>
|
||||
)}
|
||||
</details>
|
||||
</div>
|
||||
|
||||
{/* Execution Prompt */}
|
||||
<div className="run-output-section">
|
||||
<details>
|
||||
<summary className="run-output-label" style={{ cursor: "pointer", userSelect: "none" }}>Execution Prompt</summary>
|
||||
{detailRun.executionPrompt ? (
|
||||
<pre className="run-output-panel">{detailRun.executionPrompt}</pre>
|
||||
) : (
|
||||
<div className="text-muted run-output-empty">Execution prompt not captured for this run</div>
|
||||
)}
|
||||
</details>
|
||||
</div>
|
||||
|
||||
{/* Token Usage */}
|
||||
{detailRun.usageJson && (
|
||||
<div className="run-output-section">
|
||||
|
||||
43
packages/dashboard/src/__tests__/droid-cli-probe.test.ts
Normal file
43
packages/dashboard/src/__tests__/droid-cli-probe.test.ts
Normal file
@@ -0,0 +1,43 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { probeDroidCli } from "../droid-cli-probe.js";
|
||||
|
||||
/**
|
||||
* These tests exercise the real probe — we deliberately do NOT mock
|
||||
* `spawn`. The probe's job is to tell the truth about the local system,
|
||||
* and mocking defeats that. Instead we:
|
||||
*
|
||||
* 1. Assert the shape is sound regardless of binary availability
|
||||
* 2. Assert timeouts fire when the binary hangs (we don't have a
|
||||
* hanging binary fixture, so we cover the structural case only)
|
||||
* 3. Let the suite pass whether or not `droid` is installed on the
|
||||
* test runner — both outcomes are legitimate.
|
||||
*
|
||||
* If you need mocked probe behavior for a higher-level route test, spy
|
||||
* on `probeDroidCli` at the import boundary rather than shimming spawn.
|
||||
*/
|
||||
describe("probeDroidCli", () => {
|
||||
it("returns a well-formed result whether or not droid is installed", async () => {
|
||||
const result = await probeDroidCli();
|
||||
expect(typeof result.available).toBe("boolean");
|
||||
expect(typeof result.probeDurationMs).toBe("number");
|
||||
if (result.available) {
|
||||
// When available: version string should come back populated.
|
||||
expect(typeof result.version).toBe("string");
|
||||
} else {
|
||||
// When unavailable: reason must be populated so the UI can render.
|
||||
expect(typeof result.reason).toBe("string");
|
||||
}
|
||||
});
|
||||
|
||||
it("respects a short timeout", async () => {
|
||||
// Not a true hang test (we don't have a hanging binary), but
|
||||
// confirms the timeoutMs option wires through and probe completes
|
||||
// in a bounded window when using a very small timeout.
|
||||
const result = await probeDroidCli({ timeoutMs: 50 });
|
||||
expect(result.probeDurationMs).toBeLessThan(5000);
|
||||
// Either it completed fast enough or hit the timeout — both fine.
|
||||
if (!result.available && result.reason?.includes("timed out")) {
|
||||
expect(result.reason).toContain("50ms");
|
||||
}
|
||||
});
|
||||
});
|
||||
148
packages/dashboard/src/droid-cli-probe.ts
Normal file
148
packages/dashboard/src/droid-cli-probe.ts
Normal file
@@ -0,0 +1,148 @@
|
||||
/**
|
||||
* Probe for the locally-installed Droid CLI binary.
|
||||
*
|
||||
* Used by GET /api/providers/droid-cli/status to power the "Factory AI —
|
||||
* via Droid CLI" provider card. The card shows `authenticated=true` only
|
||||
* when the binary is on PATH *and* the user has flipped on `useDroidCli`.
|
||||
*
|
||||
* Intentional design choices:
|
||||
*
|
||||
* - No caching. The user's PATH can change between requests (nvm switches,
|
||||
* fresh terminal, etc.) — we'd rather pay one `spawn()` per poll than
|
||||
* serve a stale "droid not installed" response. Droid's `--version`
|
||||
* flag exits in ~40ms, so cost is negligible.
|
||||
*
|
||||
* - Short timeout. A misbehaving `droid` shim could hang indefinitely;
|
||||
* we cap the probe at 2s and report `available: false` with a timeout
|
||||
* reason rather than blocking the HTTP request.
|
||||
*
|
||||
* - No authorization. We never shell-interpolate PATH or user input.
|
||||
* We spawn `droid --version` directly with argv, no shell.
|
||||
*/
|
||||
|
||||
import { spawn } from "node:child_process";
|
||||
|
||||
/** Result shape returned to the dashboard status endpoint. */
|
||||
export interface DroidCliBinaryStatus {
|
||||
/** True if the `droid` binary was found on PATH and ran to completion. */
|
||||
available: boolean;
|
||||
/** Trimmed stdout from `droid --version`, if available. */
|
||||
version?: string;
|
||||
/** Absolute path, if we could resolve it via `which`. */
|
||||
binaryPath?: string;
|
||||
/** Human-readable failure reason when `available === false`. */
|
||||
reason?: string;
|
||||
/** Wall-clock duration of the probe, useful for debugging slow paths. */
|
||||
probeDurationMs: number;
|
||||
}
|
||||
|
||||
/** Default probe timeout. Droid's --version is fast; 2s is generous. */
|
||||
const PROBE_TIMEOUT_MS = 2000;
|
||||
|
||||
/**
|
||||
* Spawn `droid --version` and return a structured status result.
|
||||
*
|
||||
* Never throws — any failure is captured as `available: false` with a reason
|
||||
* so the caller (an HTTP handler) can render the provider card without
|
||||
* try/catch.
|
||||
*/
|
||||
export async function probeDroidCli(
|
||||
options: { timeoutMs?: number } = {},
|
||||
): Promise<DroidCliBinaryStatus> {
|
||||
const startedAt = Date.now();
|
||||
const timeoutMs = options.timeoutMs ?? PROBE_TIMEOUT_MS;
|
||||
|
||||
const binaryPath = await tryResolveBinaryPath("droid");
|
||||
|
||||
return new Promise<DroidCliBinaryStatus>((resolvePromise) => {
|
||||
const finish = (result: Omit<DroidCliBinaryStatus, "probeDurationMs">): void => {
|
||||
resolvePromise({ ...result, probeDurationMs: Date.now() - startedAt });
|
||||
};
|
||||
|
||||
let settled = false;
|
||||
const child = spawn(binaryPath ?? "droid", ["--version"], {
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
});
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
try {
|
||||
child.kill("SIGKILL");
|
||||
} catch {
|
||||
// Process already gone — nothing to do.
|
||||
}
|
||||
finish({
|
||||
available: false,
|
||||
binaryPath,
|
||||
reason: `Probe timed out after ${timeoutMs}ms`,
|
||||
});
|
||||
}, timeoutMs);
|
||||
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
child.stdout?.on("data", (chunk) => {
|
||||
stdout += chunk.toString("utf-8");
|
||||
});
|
||||
child.stderr?.on("data", (chunk) => {
|
||||
stderr += chunk.toString("utf-8");
|
||||
});
|
||||
|
||||
child.on("error", (err) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
const isNotFound = (err as NodeJS.ErrnoException).code === "ENOENT";
|
||||
finish({
|
||||
available: false,
|
||||
binaryPath,
|
||||
reason: isNotFound ? "`droid` not found on PATH" : err.message,
|
||||
});
|
||||
});
|
||||
|
||||
child.on("close", (code) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
if (code === 0) {
|
||||
finish({
|
||||
available: true,
|
||||
version: stdout.trim() || undefined,
|
||||
binaryPath,
|
||||
});
|
||||
} else {
|
||||
finish({
|
||||
available: false,
|
||||
binaryPath,
|
||||
reason:
|
||||
stderr.trim() || `droid --version exited with code ${String(code)}`,
|
||||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Best-effort `which droid`. We don't fail probe on inability to resolve
|
||||
* the path — the spawn above is the actual authority. This is just for
|
||||
* surfacing a friendly "found at /opt/homebrew/bin/droid" in the UI.
|
||||
*/
|
||||
async function tryResolveBinaryPath(binary: string): Promise<string | undefined> {
|
||||
return new Promise((resolvePromise) => {
|
||||
const which = process.platform === "win32" ? "where" : "which";
|
||||
const child = spawn(which, [binary], { stdio: ["ignore", "pipe", "ignore"] });
|
||||
let out = "";
|
||||
child.stdout?.on("data", (chunk) => {
|
||||
out += chunk.toString("utf-8");
|
||||
});
|
||||
child.on("error", () => resolvePromise(undefined));
|
||||
child.on("close", (code) => {
|
||||
if (code === 0) {
|
||||
const first = out.trim().split(/\r?\n/)[0];
|
||||
resolvePromise(first?.length ? first : undefined);
|
||||
} else {
|
||||
resolvePromise(undefined);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -1114,6 +1114,15 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
|
||||
throw notFound("Run not found");
|
||||
}
|
||||
|
||||
// Prefer run-scoped JSONL logs (written by the always-on AgentLogger).
|
||||
// These exist for both no-task and task-scoped runs from this version onward.
|
||||
const runLogs = await agentStore.getRunLogs(req.params.id, req.params.runId);
|
||||
if (runLogs.length > 0) {
|
||||
res.json(runLogs);
|
||||
return;
|
||||
}
|
||||
|
||||
// Legacy fallback: use the run's context snapshot task ID to query task-scoped logs.
|
||||
// Only use the run's context snapshot for task ID — do not fall back
|
||||
// to agent.taskId since that represents the agent's *current* task,
|
||||
// not the task active during a historical run.
|
||||
|
||||
@@ -1329,6 +1329,7 @@ describe("HeartbeatMonitor", () => {
|
||||
getLastBlockedState: vi.fn().mockResolvedValue(null),
|
||||
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
appendRunLog: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as AgentStore;
|
||||
}
|
||||
|
||||
@@ -2831,8 +2832,8 @@ describe("HeartbeatMonitor", () => {
|
||||
expect(callArgs.systemPrompt).toContain("fn_task_document_write");
|
||||
expect(callArgs.tools).toBe("readonly");
|
||||
// Tools: fn_task_create, fn_task_log, fn_task_document_write, fn_task_document_read, fn_list_agents, fn_delegate_task,
|
||||
// fn_memory_search, fn_memory_get, fn_memory_append, fn_heartbeat_done
|
||||
expect(callArgs.customTools).toHaveLength(10);
|
||||
// fn_memory_search, fn_memory_get, fn_memory_append, fn_identity, fn_heartbeat_done
|
||||
expect(callArgs.customTools).toHaveLength(11);
|
||||
expect(callArgs.customTools![0]!.name).toBe("fn_task_create");
|
||||
expect(callArgs.customTools![1]!.name).toBe("fn_task_log");
|
||||
expect(callArgs.customTools![2]!.name).toBe("fn_task_document_write");
|
||||
@@ -2842,8 +2843,10 @@ describe("HeartbeatMonitor", () => {
|
||||
expect(callArgs.customTools![6]!.name).toBe("fn_memory_search");
|
||||
expect(callArgs.customTools![7]!.name).toBe("fn_memory_get");
|
||||
expect(callArgs.customTools![8]!.name).toBe("fn_memory_append");
|
||||
// fn_identity appears before fn_heartbeat_done
|
||||
expect(callArgs.customTools![9]!.name).toBe("fn_identity");
|
||||
// fn_heartbeat_done is last (terminal tool)
|
||||
expect(callArgs.customTools![9]!.name).toBe("fn_heartbeat_done");
|
||||
expect(callArgs.customTools![10]!.name).toBe("fn_heartbeat_done");
|
||||
});
|
||||
|
||||
it("includes memory instructions even when agent has no custom instructions", async () => {
|
||||
@@ -5612,6 +5615,7 @@ describe("executeHeartbeat — skill selection resolver contract (FN-1510/FN-151
|
||||
getLastBlockedState: vi.fn().mockResolvedValue(null),
|
||||
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
appendRunLog: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as AgentStore;
|
||||
}
|
||||
|
||||
@@ -5799,6 +5803,7 @@ describe("executeHeartbeat — skill selection non-fatal (FN-1510/FN-1511)", ()
|
||||
getLastBlockedState: vi.fn().mockResolvedValue(null),
|
||||
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
appendRunLog: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as AgentStore;
|
||||
}
|
||||
|
||||
@@ -5860,3 +5865,228 @@ describe("executeHeartbeat — skill selection non-fatal (FN-1510/FN-1511)", ()
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// New observability tests (FN-3xxx sweep)
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
describe("HeartbeatMonitor observability — prompt persistence + run-scoped logs", () => {
|
||||
// These tests use the same mock infrastructure as the main executeHeartbeat suite.
|
||||
let mockTaskStore: TaskStore;
|
||||
let mockAgent: Agent;
|
||||
|
||||
function createMockAgentSession() {
|
||||
return {
|
||||
prompt: vi.fn().mockResolvedValue(undefined),
|
||||
dispose: vi.fn(),
|
||||
subscribe: vi.fn(),
|
||||
model: { provider: "mock", id: "mock-model" },
|
||||
};
|
||||
}
|
||||
|
||||
function createMockTaskStore(overrides: Partial<TaskStore> = {}): TaskStore {
|
||||
return {
|
||||
getTask: vi.fn().mockResolvedValue({
|
||||
id: "FN-001",
|
||||
title: "Test Task",
|
||||
description: "Test task description",
|
||||
prompt: "# Test PROMPT.md\nSome content",
|
||||
steps: [],
|
||||
column: "todo",
|
||||
dependencies: [],
|
||||
log: [],
|
||||
attachments: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
} as unknown as TaskDetail),
|
||||
selectNextTaskForAgent: vi.fn().mockResolvedValue(null),
|
||||
createTask: vi.fn().mockResolvedValue({ id: "FN-002", description: "Created task", dependencies: [], column: "triage" }),
|
||||
logEntry: vi.fn().mockResolvedValue({}),
|
||||
addComment: vi.fn().mockResolvedValue({}),
|
||||
appendAgentLog: vi.fn().mockResolvedValue(undefined),
|
||||
upsertTaskDocument: vi.fn().mockResolvedValue({}),
|
||||
getTaskDocument: vi.fn().mockResolvedValue(null),
|
||||
getTaskDocuments: vi.fn().mockResolvedValue([]),
|
||||
...overrides,
|
||||
} as unknown as TaskStore;
|
||||
}
|
||||
|
||||
function createStoreWithAgent(agentData: Partial<Agent> = {}): AgentStore {
|
||||
mockAgent = {
|
||||
id: "agent-001",
|
||||
name: "Test Agent",
|
||||
role: "executor",
|
||||
state: "active",
|
||||
taskId: "FN-001",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
metadata: {},
|
||||
...agentData,
|
||||
} as Agent;
|
||||
|
||||
const savedRuns: Map<string, AgentHeartbeatRun> = new Map();
|
||||
|
||||
return {
|
||||
recordHeartbeat: vi.fn().mockResolvedValue(undefined),
|
||||
updateAgentState: vi.fn().mockResolvedValue(undefined),
|
||||
updateAgent: vi.fn().mockResolvedValue(undefined),
|
||||
getAgent: vi.fn().mockResolvedValue(mockAgent),
|
||||
assignTask: vi.fn().mockImplementation(async (_agentId: string, taskId: string | undefined) => {
|
||||
mockAgent.taskId = taskId;
|
||||
return mockAgent;
|
||||
}),
|
||||
startHeartbeatRun: vi.fn().mockResolvedValue({
|
||||
id: "run-obs-001",
|
||||
agentId: "agent-001",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: null,
|
||||
status: "active",
|
||||
} as AgentHeartbeatRun),
|
||||
saveRun: vi.fn().mockImplementation(async (run: AgentHeartbeatRun) => {
|
||||
savedRuns.set(run.id, run);
|
||||
}),
|
||||
getRunDetail: vi.fn().mockImplementation(async (_agentId: string, runId: string) => {
|
||||
return savedRuns.get(runId) ?? {
|
||||
id: runId,
|
||||
agentId: "agent-001",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "completed" as const,
|
||||
};
|
||||
}),
|
||||
getRatingSummary: vi.fn().mockResolvedValue(undefined),
|
||||
endHeartbeatRun: vi.fn().mockResolvedValue(undefined),
|
||||
getBudgetStatus: vi.fn().mockResolvedValue(createBudgetStatus()),
|
||||
getCachedAgent: vi.fn().mockReturnValue(null),
|
||||
getLastBlockedState: vi.fn().mockResolvedValue(null),
|
||||
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||
appendRunLog: vi.fn().mockResolvedValue(undefined),
|
||||
} as unknown as AgentStore;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
mockTaskStore = createMockTaskStore();
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("no-task heartbeat run persists systemPrompt and executionPrompt on the run record", async () => {
|
||||
// Identity agent (has soul) so a no-task run is triggered
|
||||
const store = createStoreWithAgent({ taskId: undefined, soul: "I am the ambient coordinator." });
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: mockSession as any });
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||
|
||||
// saveRun should have been called with both prompt fields populated
|
||||
const saveRunCalls = (store.saveRun as ReturnType<typeof vi.fn>).mock.calls;
|
||||
// Find the call that includes systemPrompt (the prompt-persistence saveRun)
|
||||
const promptRunCall = saveRunCalls.find(
|
||||
(args: unknown[]) => typeof (args[0] as AgentHeartbeatRun).systemPrompt === "string" && ((args[0] as AgentHeartbeatRun).systemPrompt?.length ?? 0) > 0
|
||||
);
|
||||
expect(promptRunCall).toBeDefined();
|
||||
const savedRun = promptRunCall![0] as AgentHeartbeatRun;
|
||||
expect(savedRun.systemPrompt).toBeDefined();
|
||||
expect(typeof savedRun.systemPrompt).toBe("string");
|
||||
expect(savedRun.executionPrompt).toBeDefined();
|
||||
expect(typeof savedRun.executionPrompt).toBe("string");
|
||||
// heartbeatProcedureSource should be "default" (no custom procedure file)
|
||||
expect(savedRun.heartbeatProcedureSource).toBe("default");
|
||||
|
||||
// The execution prompt should contain the procedure text before the no-task action menu
|
||||
expect(savedRun.executionPrompt).toContain("fn_identity");
|
||||
expect(savedRun.executionPrompt).toContain("Heartbeat Procedure");
|
||||
// The wake delta header should appear before the action menu items
|
||||
const procedureIdx = savedRun.executionPrompt!.indexOf("Heartbeat Procedure");
|
||||
const actionMenuIdx = savedRun.executionPrompt!.indexOf("No assigned task");
|
||||
expect(procedureIdx).toBeLessThan(actionMenuIdx);
|
||||
|
||||
expect(result.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("no-task heartbeat run-scoped logs receive at least one entry after a simulated tick", async () => {
|
||||
const store = createStoreWithAgent({ taskId: undefined, soul: "I observe the project." });
|
||||
const mockSession = createMockAgentSession();
|
||||
let capturedOnText: ((delta: string) => void) | undefined;
|
||||
|
||||
mockedCreateFnAgent.mockImplementation(async (opts: any) => {
|
||||
capturedOnText = opts.onText;
|
||||
return { session: mockSession as any };
|
||||
});
|
||||
|
||||
// Simulate the session emitting a text delta during prompt
|
||||
mockSession.prompt = vi.fn().mockImplementation(async () => {
|
||||
capturedOnText?.("I am reviewing the project state.");
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||
|
||||
// appendRunLog should have been called on the AgentStore at least once
|
||||
const appendRunLogCalls = (store.appendRunLog as ReturnType<typeof vi.fn>).mock.calls;
|
||||
expect(appendRunLogCalls.length).toBeGreaterThanOrEqual(1);
|
||||
|
||||
// Verify the entry shape: agentId, runId, entry
|
||||
const [callAgentId, callRunId, callEntry] = appendRunLogCalls[0] as [string, string, unknown];
|
||||
expect(callAgentId).toBe("agent-001");
|
||||
expect(callRunId).toBe("run-obs-001");
|
||||
expect(callEntry).toMatchObject({ type: expect.stringMatching(/^(text|thinking|tool|tool_result|tool_error)$/) });
|
||||
});
|
||||
|
||||
it("task-scoped heartbeat persists systemPrompt and executionPrompt with procedure before task content", async () => {
|
||||
const store = createStoreWithAgent({ taskId: "FN-001" });
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: mockSession as any });
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||
|
||||
const saveRunCalls = (store.saveRun as ReturnType<typeof vi.fn>).mock.calls;
|
||||
const promptRunCall = saveRunCalls.find(
|
||||
(args: unknown[]) => typeof (args[0] as AgentHeartbeatRun).systemPrompt === "string" && ((args[0] as AgentHeartbeatRun).systemPrompt?.length ?? 0) > 0
|
||||
);
|
||||
expect(promptRunCall).toBeDefined();
|
||||
const savedRun = promptRunCall![0] as AgentHeartbeatRun;
|
||||
|
||||
// The execution prompt should have procedure before task description
|
||||
const procedureIdx = savedRun.executionPrompt!.indexOf("Heartbeat Procedure");
|
||||
const taskDescIdx = savedRun.executionPrompt!.indexOf("Task description:");
|
||||
expect(procedureIdx).toBeGreaterThanOrEqual(0);
|
||||
expect(taskDescIdx).toBeGreaterThanOrEqual(0);
|
||||
expect(procedureIdx).toBeLessThan(taskDescIdx);
|
||||
|
||||
// fn_identity instruction should appear in the execution prompt
|
||||
expect(savedRun.executionPrompt).toContain("fn_identity");
|
||||
|
||||
expect(result.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("fn_identity tool returns correct agent identity information", async () => {
|
||||
const store = createStoreWithAgent({ soul: "I am a senior executor.", memory: "Always log blockers." });
|
||||
let capturedIdentityTool: any;
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateFnAgent.mockImplementation(async (opts: any) => {
|
||||
capturedIdentityTool = opts.customTools?.find((t: any) => t.name === "fn_identity");
|
||||
return { session: mockSession as any };
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(capturedIdentityTool).toBeDefined();
|
||||
expect(capturedIdentityTool.name).toBe("fn_identity");
|
||||
|
||||
// Call the tool and verify output structure
|
||||
const toolResult = await capturedIdentityTool.execute("call-1", {});
|
||||
expect(toolResult.content[0].text).toContain("agentId: agent-001");
|
||||
expect(toolResult.content[0].text).toContain("name: Test Agent");
|
||||
expect(toolResult.details.soulPresent).toBe(true);
|
||||
expect(toolResult.details.memoryPresent).toBe(true);
|
||||
expect(toolResult.details.soulPreview).toContain("I am a senior executor.");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -21,7 +21,7 @@ import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHea
|
||||
import { buildExecutionMemoryInstructions, isEphemeralAgent, hasAgentIdentity } from "@fusion/core";
|
||||
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
|
||||
import { Type, type Static } from "@mariozechner/pi-ai";
|
||||
import { createTaskCreateTool, createTaskLogToolWithContext, createTaskDocumentWriteTool, createTaskDocumentReadTool, createListAgentsTool, createDelegateTaskTool, createSendMessageTool, createReadMessagesTool, createMemoryTools, taskCreateParams } from "./agent-tools.js";
|
||||
import { createTaskCreateTool, createTaskLogToolWithContext, createTaskDocumentWriteTool, createTaskDocumentReadTool, createListAgentsTool, createDelegateTaskTool, createSendMessageTool, createReadMessagesTool, createMemoryTools, createIdentityTool, taskCreateParams } from "./agent-tools.js";
|
||||
import { AgentLogger } from "./agent-logger.js";
|
||||
import { resolveAgentInstructionsWithRatings, buildSystemPromptWithInstructions, resolveAgentHeartbeatProcedure } from "./agent-instructions.js";
|
||||
import { heartbeatLog, formatError } from "./logger.js";
|
||||
@@ -292,9 +292,9 @@ export const HEARTBEAT_SYSTEM_PROMPT_NO_TASK = HEARTBEAT_NO_TASK_SYSTEM_PROMPT;
|
||||
*/
|
||||
export const HEARTBEAT_PROCEDURE = `## Heartbeat Procedure (run every tick, in order)
|
||||
|
||||
1. **Identity & context** — review your soul, instructions, and memory (already
|
||||
loaded in the system prompt). Confirm who you are and what you're responsible
|
||||
for before continuing prior work.
|
||||
1. **Identity & context** — call fn_identity FIRST to confirm which soul,
|
||||
instructions, and memory loaded for this tick. Echo your role and any
|
||||
anomalies in your first text output before doing anything else.
|
||||
2. **Inbox** — when fn_read_messages is available, call it. Process any pending
|
||||
messages first; reply with reply_to_message_id when answering.
|
||||
3. **Wake delta** — read the Wake Delta block above. The wake reason is the
|
||||
@@ -321,6 +321,15 @@ const heartbeatDoneParams = Type.Object({
|
||||
summary: Type.Optional(Type.String({ description: "Summary of what was accomplished this heartbeat" })),
|
||||
});
|
||||
|
||||
/**
|
||||
* Truncate a string to `maxChars`, appending a marker so callers can see
|
||||
* content was clipped. Returns the original string unchanged when it fits.
|
||||
*/
|
||||
function truncatePrompt(text: string, maxChars: number): string {
|
||||
if (text.length <= maxChars) return text;
|
||||
return `${text.slice(0, maxChars)}\n\n... (truncated, ${text.length} chars)`;
|
||||
}
|
||||
|
||||
async function getHeartbeatMemorySettings(taskStore: TaskStore): Promise<Settings | undefined> {
|
||||
const maybeGetSettings = (taskStore as { getSettings?: () => Promise<Settings> }).getSettings;
|
||||
if (!maybeGetSettings) {
|
||||
@@ -1297,17 +1306,6 @@ export class HeartbeatMonitor {
|
||||
const message = memorySettingsError instanceof Error ? memorySettingsError.message : String(memorySettingsError);
|
||||
heartbeatLog.warn(`Failed to configure heartbeat memory tools for ${agentId}: ${message}`);
|
||||
}
|
||||
heartbeatTools.push(heartbeatDoneTool);
|
||||
|
||||
// AgentLogger requires a taskId — only create for task-scoped runs
|
||||
if (!isNoTaskRun && taskId) {
|
||||
agentLogger = new AgentLogger({
|
||||
store: taskStore,
|
||||
taskId,
|
||||
agent: agent.role as AgentRole,
|
||||
});
|
||||
}
|
||||
|
||||
// Build skill selection context for heartbeat session (uses waking agent's skills, no role fallback)
|
||||
const skillContext = buildSessionSkillContextSync(agent, "heartbeat", rootDir);
|
||||
|
||||
@@ -1315,8 +1313,10 @@ export class HeartbeatMonitor {
|
||||
? HEARTBEAT_NO_TASK_SYSTEM_PROMPT
|
||||
: HEARTBEAT_SYSTEM_PROMPT;
|
||||
const baseHeartbeatSystemPrompt = systemPrompt;
|
||||
let resolvedInstructionsForIdentity = "";
|
||||
try {
|
||||
const agentInstructions = await resolveAgentInstructionsWithRatings(agent, rootDir, this.store);
|
||||
resolvedInstructionsForIdentity = agentInstructions;
|
||||
const memoryInstructions = memorySettings?.memoryEnabled === false
|
||||
? ""
|
||||
: buildExecutionMemoryInstructions(rootDir, memorySettings);
|
||||
@@ -1330,6 +1330,28 @@ export class HeartbeatMonitor {
|
||||
heartbeatLog.warn(`Failed to enrich heartbeat system prompt for ${agentId}: ${message}`);
|
||||
}
|
||||
|
||||
// Register fn_identity tool before fn_heartbeat_done (which must stay last)
|
||||
heartbeatTools.push(createIdentityTool({ agent, resolvedInstructions: resolvedInstructionsForIdentity }));
|
||||
|
||||
// fn_heartbeat_done must be the last tool in the array (stable terminal signal)
|
||||
heartbeatTools.push(heartbeatDoneTool);
|
||||
|
||||
// Always-on AgentLogger: no-task runs use the callback sink wired to run-scoped JSONL;
|
||||
// task-scoped runs write to both the task store AND the run-scoped JSONL.
|
||||
if (isNoTaskRun) {
|
||||
agentLogger = new AgentLogger({
|
||||
appendLog: (entry) => this.store.appendRunLog(agentId, run.id, entry),
|
||||
agent: agent.role as AgentRole,
|
||||
});
|
||||
} else if (taskId) {
|
||||
agentLogger = new AgentLogger({
|
||||
store: taskStore,
|
||||
taskId,
|
||||
agent: agent.role as AgentRole,
|
||||
appendLog: (entry) => this.store.appendRunLog(agentId, run.id, entry),
|
||||
});
|
||||
}
|
||||
|
||||
// Create agent session
|
||||
const { session } = await createResolvedAgentSession({
|
||||
sessionPurpose: "heartbeat",
|
||||
@@ -1430,6 +1452,10 @@ export class HeartbeatMonitor {
|
||||
"Run the Heartbeat Procedure (below) before doing anything else — even a",
|
||||
"timer-only wake should re-check messages, memory, and project state.",
|
||||
"",
|
||||
"You MUST call fn_identity as your first tool action this tick before reading any task content or calling any other tool.",
|
||||
"",
|
||||
heartbeatProcedureText,
|
||||
"",
|
||||
"**No assigned task** — This heartbeat run has no task assignment.",
|
||||
"",
|
||||
"You have identity (soul, instructions, and/or memory) loaded, which means you can perform",
|
||||
@@ -1454,8 +1480,6 @@ export class HeartbeatMonitor {
|
||||
"Your soul, instructions, and memory are already loaded in the system prompt.",
|
||||
"Focus on work that benefits the project without requiring a specific task context.",
|
||||
"",
|
||||
heartbeatProcedureText,
|
||||
"",
|
||||
"Call fn_heartbeat_done when finished.",
|
||||
].join("\n");
|
||||
} else {
|
||||
@@ -1530,6 +1554,10 @@ export class HeartbeatMonitor {
|
||||
"decide what action this delta requires. Your assigned task is one input",
|
||||
"to the procedure — not the only thing to consider.",
|
||||
"",
|
||||
"You MUST call fn_identity as your first tool action this tick before reading any task content or calling any other tool.",
|
||||
"",
|
||||
heartbeatProcedureText,
|
||||
"",
|
||||
"Task description:",
|
||||
taskDetail!.description,
|
||||
"",
|
||||
@@ -1537,12 +1565,26 @@ export class HeartbeatMonitor {
|
||||
...triggeringCommentLines,
|
||||
...pendingMessagesLines,
|
||||
"",
|
||||
heartbeatProcedureText,
|
||||
"",
|
||||
"Run the Heartbeat Procedure above. Call fn_heartbeat_done when finished.",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
// Persist prompts on the run record before executing so they are
|
||||
// observable in the dashboard even if execution fails partway through.
|
||||
try {
|
||||
const runWithPrompts: AgentHeartbeatRun = {
|
||||
...run,
|
||||
systemPrompt: truncatePrompt(systemPrompt, 100_000),
|
||||
executionPrompt: truncatePrompt(executionPrompt, 100_000),
|
||||
heartbeatProcedureSource: customProcedure ? "custom" : "default",
|
||||
};
|
||||
await this.store.saveRun(runWithPrompts);
|
||||
// Update local run reference so completeRun merges correctly
|
||||
Object.assign(run, { systemPrompt: runWithPrompts.systemPrompt, executionPrompt: runWithPrompts.executionPrompt, heartbeatProcedureSource: runWithPrompts.heartbeatProcedureSource });
|
||||
} catch (promptPersistErr) {
|
||||
heartbeatLog.warn(`Failed to persist prompts for ${agentId}/${run.id}: ${promptPersistErr instanceof Error ? promptPersistErr.message : String(promptPersistErr)}`);
|
||||
}
|
||||
|
||||
// Execute
|
||||
await promptWithFallback(session, executionPrompt);
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { TaskStore, AgentRole } from "@fusion/core";
|
||||
import type { TaskStore, AgentLogEntry, AgentRole } from "@fusion/core";
|
||||
import { createLogger } from "./logger.js";
|
||||
|
||||
/** Default byte threshold before an automatic flush. */
|
||||
@@ -35,12 +35,24 @@ export function summarizeToolArgs(name: string, args?: Record<string, unknown>):
|
||||
|
||||
/**
|
||||
* Options for creating an {@link AgentLogger}.
|
||||
*
|
||||
* Two sink modes are supported:
|
||||
* 1. **Task-store mode** (original): provide `store` + `taskId`. Writes go to
|
||||
* `store.appendAgentLog(taskId, ...)`.
|
||||
* 2. **Callback mode**: provide `appendLog`. Writes go to the callback instead.
|
||||
* When both are provided, both sinks receive every entry.
|
||||
*/
|
||||
export interface AgentLoggerOptions {
|
||||
/** The task store used to persist agent log entries. */
|
||||
store: TaskStore;
|
||||
/** The task ID this logger is associated with. */
|
||||
taskId: string;
|
||||
/** The task store used to persist agent log entries (task-store mode). */
|
||||
store?: TaskStore;
|
||||
/** The task ID this logger is associated with (task-store mode). */
|
||||
taskId?: string;
|
||||
/**
|
||||
* Optional alternative sink callback. When provided, every flushed entry is
|
||||
* forwarded here in addition to (or instead of) `store.appendAgentLog`.
|
||||
* Use this for run-scoped logging where there is no task.
|
||||
*/
|
||||
appendLog?: (entry: AgentLogEntry) => Promise<void>;
|
||||
/** Which agent role is producing log entries (persisted on every entry). */
|
||||
agent?: AgentRole;
|
||||
/** Optional callback invoked alongside text logging (e.g. for SSE streaming). */
|
||||
@@ -85,8 +97,9 @@ export class AgentLogger {
|
||||
private thinkingFlushTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
private readonly flushSizeBytes: number;
|
||||
private readonly flushIntervalMs: number;
|
||||
private readonly store: TaskStore;
|
||||
private readonly store?: TaskStore;
|
||||
private readonly taskId: string;
|
||||
private readonly appendLogCb?: (entry: AgentLogEntry) => Promise<void>;
|
||||
private readonly agent?: AgentRole;
|
||||
private readonly externalTextCb?: (taskId: string, delta: string) => void;
|
||||
private readonly externalToolCb?: (taskId: string, toolName: string) => void;
|
||||
@@ -94,7 +107,8 @@ export class AgentLogger {
|
||||
|
||||
constructor(options: AgentLoggerOptions) {
|
||||
this.store = options.store;
|
||||
this.taskId = options.taskId;
|
||||
this.taskId = options.taskId ?? "";
|
||||
this.appendLogCb = options.appendLog;
|
||||
this.agent = options.agent;
|
||||
this.externalTextCb = options.onAgentText;
|
||||
this.externalToolCb = options.onAgentTool;
|
||||
@@ -149,9 +163,7 @@ export class AgentLogger {
|
||||
if (this.thinkingFlushTimer) { clearTimeout(this.thinkingFlushTimer); this.thinkingFlushTimer = null; }
|
||||
this.flushThinkingBuffer();
|
||||
const detail = summarizeToolArgs(name, args);
|
||||
this.store.appendAgentLog(this.taskId, name, "tool", detail, this.agent).catch((err) => {
|
||||
this.log.warn(`Failed to log tool start "${name}" for ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
this.writeEntry(name, "tool", detail, `Failed to log tool start "${name}" for ${this.taskId}`);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -168,9 +180,7 @@ export class AgentLogger {
|
||||
if (result !== undefined && result !== null) {
|
||||
detail = typeof result === "string" ? result : JSON.stringify(result);
|
||||
}
|
||||
this.store.appendAgentLog(this.taskId, name, type, detail, this.agent).catch((err) => {
|
||||
this.log.warn(`Failed to log tool end "${name}" (${type}) for ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
this.writeEntry(name, type, detail, `Failed to log tool end "${name}" (${type}) for ${this.taskId}`);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -186,22 +196,99 @@ export class AgentLogger {
|
||||
|
||||
// ── Internal helpers ───────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Write a single structured entry through whichever sink(s) are configured.
|
||||
* When both `store`+`taskId` and `appendLogCb` are set, both receive the entry.
|
||||
* When only `appendLogCb` is set (no store/taskId), only the callback is used.
|
||||
* @param storeWarnMsg - Warning message prefix used when the task-store write fails.
|
||||
*/
|
||||
private writeEntry(text: string, type: AgentLogEntry["type"], detail: string | undefined, storeWarnMsg: string): void {
|
||||
const entry: AgentLogEntry = {
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId: this.taskId,
|
||||
text,
|
||||
type,
|
||||
...(detail !== undefined && { detail }),
|
||||
...(this.agent !== undefined && { agent: this.agent }),
|
||||
};
|
||||
|
||||
if (this.store && this.taskId) {
|
||||
this.store.appendAgentLog(this.taskId, text, type, detail, this.agent).catch((err) => {
|
||||
this.log.warn(`${storeWarnMsg}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
}
|
||||
|
||||
if (this.appendLogCb) {
|
||||
this.appendLogCb(entry).catch((err) => {
|
||||
this.log.warn(`appendLog callback failed for entry (${type}): ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
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", undefined, this.agent).catch((err) => {
|
||||
this.log.warn(`Failed to flush text buffer for ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
const entry: AgentLogEntry = {
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId: this.taskId,
|
||||
text: chunk,
|
||||
type: "text",
|
||||
...(this.agent !== undefined && { agent: this.agent }),
|
||||
};
|
||||
|
||||
const promises: Promise<void>[] = [];
|
||||
|
||||
if (this.store && this.taskId) {
|
||||
promises.push(
|
||||
this.store.appendAgentLog(this.taskId, chunk, "text", undefined, this.agent).catch((err) => {
|
||||
this.log.warn(`Failed to flush text buffer for ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
if (this.appendLogCb) {
|
||||
promises.push(
|
||||
this.appendLogCb(entry).catch((err) => {
|
||||
this.log.warn(`appendLog callback failed for text flush: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
return Promise.all(promises).then(() => undefined);
|
||||
}
|
||||
|
||||
private flushThinkingBuffer(): Promise<void> {
|
||||
if (this.thinkingBuffer.length === 0) return Promise.resolve();
|
||||
const chunk = this.thinkingBuffer;
|
||||
this.thinkingBuffer = "";
|
||||
return this.store.appendAgentLog(this.taskId, chunk, "thinking", undefined, this.agent).catch((err) => {
|
||||
this.log.warn(`Failed to flush thinking buffer for ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
const entry: AgentLogEntry = {
|
||||
timestamp: new Date().toISOString(),
|
||||
taskId: this.taskId,
|
||||
text: chunk,
|
||||
type: "thinking",
|
||||
...(this.agent !== undefined && { agent: this.agent }),
|
||||
};
|
||||
|
||||
const promises: Promise<void>[] = [];
|
||||
|
||||
if (this.store && this.taskId) {
|
||||
promises.push(
|
||||
this.store.appendAgentLog(this.taskId, chunk, "thinking", undefined, this.agent).catch((err) => {
|
||||
this.log.warn(`Failed to flush thinking buffer for ${this.taskId}: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
if (this.appendLogCb) {
|
||||
promises.push(
|
||||
this.appendLogCb(entry).catch((err) => {
|
||||
this.log.warn(`appendLog callback failed for thinking flush: ${err instanceof Error ? err.message : String(err)}`);
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
return Promise.all(promises).then(() => undefined);
|
||||
}
|
||||
|
||||
private scheduleFlush(): void {
|
||||
|
||||
@@ -11,7 +11,7 @@ import { appendFile, mkdir, readFile, readdir, stat, writeFile } from "node:fs/p
|
||||
import { existsSync } from "node:fs";
|
||||
import { createHash } from "node:crypto";
|
||||
import { join } from "node:path";
|
||||
import type { AgentStore, AgentState, AgentCapability, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message, SourceType, Settings, ResearchRun, ResearchRunStatus } from "@fusion/core";
|
||||
import type { AgentStore, AgentState, AgentCapability, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message, SourceType, Settings, ResearchRun, ResearchRunStatus, Agent } from "@fusion/core";
|
||||
import { dailyMemoryPath, ensureOpenClawMemoryFiles, getMemoryBackendCapabilities, getProjectMemory, isEphemeralAgent, memoryLongTermPath, resolveMemoryBackend, resolveResearchSettings, scheduleQmdProjectMemoryRefresh, searchProjectMemory, shouldSkipBackgroundQmdRefresh } from "@fusion/core";
|
||||
import { ResearchOrchestrator } from "./research-orchestrator.js";
|
||||
import { ResearchProviderRegistry } from "./research/provider-registry.js";
|
||||
@@ -1328,3 +1328,79 @@ export function createReadMessagesTool(messageStore: MessageStore, agentId: stri
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** Arguments for {@link createIdentityTool}. */
|
||||
export interface CreateIdentityToolArgs {
|
||||
/** The agent record for this heartbeat run. */
|
||||
agent: Agent;
|
||||
/** The resolved instructions string (from resolveAgentInstructionsWithRatings). */
|
||||
resolvedInstructions: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create the `fn_identity` tool for heartbeat sessions.
|
||||
*
|
||||
* When called, it returns a structured summary of which soul, instructions, and
|
||||
* memory are currently loaded for this tick. The agent is expected to call this
|
||||
* as its FIRST tool action so operators (via dashboard run logs) can verify
|
||||
* correct identity was applied.
|
||||
*/
|
||||
export function createIdentityTool({ agent, resolvedInstructions }: CreateIdentityToolArgs): ToolDefinition {
|
||||
const identityParams = Type.Object({});
|
||||
return {
|
||||
name: "fn_identity",
|
||||
label: "Identity Check",
|
||||
description: "Return a structured summary of which soul, instructions, and memory are loaded for this heartbeat tick. Call this FIRST before any other tool.",
|
||||
parameters: identityParams,
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
execute: async (_id: string, _params: Static<typeof identityParams>, _signal?: any, _onUpdate?: any, _ctx?: any) => {
|
||||
const PREVIEW_CHARS = 500;
|
||||
const INSTRUCTIONS_PREVIEW_CHARS = 1000;
|
||||
const MEMORY_PREVIEW_CHARS = 1000;
|
||||
|
||||
const soulPresent = typeof agent.soul === "string" && agent.soul.trim().length > 0;
|
||||
const instructionsPresent = resolvedInstructions.trim().length > 0;
|
||||
const memoryPresent = typeof agent.memory === "string" && agent.memory.trim().length > 0;
|
||||
|
||||
const soulPreview = soulPresent ? (agent.soul as string).slice(0, PREVIEW_CHARS) : "";
|
||||
const instructionsPreview = instructionsPresent ? resolvedInstructions.slice(0, INSTRUCTIONS_PREVIEW_CHARS) : "";
|
||||
const memoryPreview = memoryPresent ? (agent.memory as string).slice(0, MEMORY_PREVIEW_CHARS) : "";
|
||||
|
||||
const result = {
|
||||
agentId: agent.id,
|
||||
name: agent.name,
|
||||
role: agent.role,
|
||||
soulPresent,
|
||||
instructionsPresent,
|
||||
memoryPresent,
|
||||
soulPreview,
|
||||
instructionsPreview,
|
||||
memoryPreview,
|
||||
};
|
||||
|
||||
const lines = [
|
||||
`agentId: ${result.agentId}`,
|
||||
`name: ${result.name}`,
|
||||
`role: ${result.role}`,
|
||||
`soul: ${result.soulPresent ? "loaded" : "absent"}`,
|
||||
`instructions: ${result.instructionsPresent ? "loaded" : "absent"}`,
|
||||
`memory: ${result.memoryPresent ? "loaded" : "absent"}`,
|
||||
];
|
||||
|
||||
if (result.soulPresent && result.soulPreview) {
|
||||
lines.push(`\nSoul preview (first ${PREVIEW_CHARS} chars):\n${result.soulPreview}`);
|
||||
}
|
||||
if (result.instructionsPresent && result.instructionsPreview) {
|
||||
lines.push(`\nInstructions preview (first ${INSTRUCTIONS_PREVIEW_CHARS} chars):\n${result.instructionsPreview}`);
|
||||
}
|
||||
if (result.memoryPresent && result.memoryPreview) {
|
||||
lines.push(`\nMemory preview (first ${MEMORY_PREVIEW_CHARS} chars):\n${result.memoryPreview}`);
|
||||
}
|
||||
|
||||
return {
|
||||
content: [{ type: "text" as const, text: lines.join("\n") }],
|
||||
details: result,
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user