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

feat(FN-1452): make structured run records the source of truth for agent run state

Browse Source 
- Update AGENTS.md with authoritative run lifecycle semantics documentation
- Add API-level regression test for repeated manual run prevention
- Update run lifecycle tests to match corrected behavior
- Refactor AgentStore to use structured run records as the authoritative source for run state
- Improve run status queries with better filtering and ordering
This commit is contained in:
gsxdsm
2026-04-09 18:37:48 -07:00
parent ecb5baeea5
commit 1c4a3f4e4d
4 changed files with 259 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

104
packages/core/src/agent-store.test.ts
View File

@@ -1719,18 +1719,22 @@ describe("AgentStore", () => {
expect(completed[0].endedAt).toBeDefined();
});
it("endHeartbeatRun with 'completed' records an ok heartbeat", async () => {
it("endHeartbeatRun with 'completed' removes from active and adds to completed", async () => {
const agent = await store.createAgent({ name: "CompleteRun", role: "executor" });
const run = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(run.id, "completed");
// A completed run records status "ok" (not "missed"), so the run
// stays active in the reconstructed view (implementation detail).
// The getActiveHeartbeatRun still sees it as active since only
// "missed" status marks a run as terminated.
// A completed run should NOT appear in active runs
const active = await store.getActiveHeartbeatRun(agent.id);
expect(active).not.toBeNull();
expect(active).toBeNull();
// A completed run should appear in completed runs with terminal status
const completed = await store.getCompletedHeartbeatRuns(agent.id);
expect(completed).toHaveLength(1);
expect(completed[0].id).toBe(run.id);
expect(completed[0].status).toBe("completed");
expect(completed[0].endedAt).toBeDefined();
});
it("getCompletedHeartbeatRuns returns only non-active runs", async () => {
@@ -1751,6 +1755,94 @@ describe("AgentStore", () => {
expect(active).not.toBeNull();
expect(active!.id).toBe(run2.id);
});
it("after completion, a new run can start without stale active-run blockage", async () => {
const agent = await store.createAgent({ name: "RestartRun", role: "executor" });
// Start and complete first run
const run1 = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(run1.id, "completed");
// Verify first run is not active
const active1 = await store.getActiveHeartbeatRun(agent.id);
expect(active1).toBeNull();
// Start second run - should succeed without conflict
const run2 = await store.startHeartbeatRun(agent.id);
expect(run2.id).not.toBe(run1.id);
expect(run2.status).toBe("active");
// Verify second run is now the active run
const active2 = await store.getActiveHeartbeatRun(agent.id);
expect(active2).not.toBeNull();
expect(active2!.id).toBe(run2.id);
});
it("startHeartbeatRun persists the run to structured storage", async () => {
const agent = await store.createAgent({ name: "PersistRun", role: "executor" });
const run = await store.startHeartbeatRun(agent.id);
// Verify run is persisted
const detail = await store.getRunDetail(agent.id, run.id);
expect(detail).not.toBeNull();
expect(detail!.id).toBe(run.id);
expect(detail!.agentId).toBe(agent.id);
expect(detail!.status).toBe("active");
expect(detail!.endedAt).toBeNull();
// Verify run appears in recent runs
const recent = await store.getRecentRuns(agent.id);
expect(recent.some((r) => r.id === run.id)).toBe(true);
});
it("endHeartbeatRun updates the persisted run with terminal state", async () => {
const agent = await store.createAgent({ name: "UpdateRun", role: "executor" });
const run = await store.startHeartbeatRun(agent.id);
// Complete the run
await store.endHeartbeatRun(run.id, "completed");
// Verify persisted run is updated
const detail = await store.getRunDetail(agent.id, run.id);
expect(detail).not.toBeNull();
expect(detail!.status).toBe("completed");
expect(detail!.endedAt).toBeDefined();
});
it("getCompletedHeartbeatRuns returns terminal runs in newest-first order", async () => {
const agent = await store.createAgent({ name: "OrderRuns", role: "executor" });
vi.useFakeTimers();
try {
vi.setSystemTime(new Date("2026-01-01T00:00:00Z"));
const run1 = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(run1.id, "completed");
vi.setSystemTime(new Date("2026-01-02T00:00:00Z"));
const run2 = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(run2.id, "completed");
const completed = await store.getCompletedHeartbeatRuns(agent.id);
expect(completed).toHaveLength(2);
expect(completed[0].id).toBe(run2.id); // Newest first
expect(completed[1].id).toBe(run1.id);
} finally {
vi.useRealTimers();
}
});
it("handles mixed structured and legacy run data", async () => {
const agent = await store.createAgent({ name: "MixedRuns", role: "executor" });
// Structured run (new behavior)
const structuredRun = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(structuredRun.id, "completed");
// Legacy completed run (simulated by checking legacy fallback)
// The fallback still reconstructs runs from heartbeat events
const completed = await store.getCompletedHeartbeatRuns(agent.id);
expect(completed.some((r) => r.id === structuredRun.id)).toBe(true);
});
});
// ── blocked state persistence ─────────────────────────────────────

63
packages/core/src/agent-store.ts
View File

@@ -1210,20 +1210,25 @@ export class AgentStore extends EventEmitter {
/**
* Start a new heartbeat run for an agent.
* Persists the run to structured storage as the source of truth.
* @param agentId - The agent ID
* @returns The created run
*/
async startHeartbeatRun(agentId: string): Promise<AgentHeartbeatRun> {
const runId = `run-${randomUUID().slice(0, 8)}`;
const now = new Date().toISOString();
const run: AgentHeartbeatRun = {
id: runId,
agentId,
startedAt: new Date().toISOString(),
startedAt: now,
endedAt: null,
status: "active",
};
// Record as heartbeat event to track runs
// Persist to structured storage as source of truth
await this.saveRun(run);
// Also record as heartbeat event for legacy compatibility
await this.recordHeartbeat(agentId, "ok", runId);
return run;
@@ -1231,10 +1236,14 @@ export class AgentStore extends EventEmitter {
/**
* End a heartbeat run.
* Updates the persisted run's terminal state in structured storage.
* Also records a heartbeat event for legacy compatibility.
* @param runId - The run ID
* @param status - End status (completed or terminated)
*/
async endHeartbeatRun(runId: string, status: "completed" | "terminated"): Promise<void> {
const now = new Date().toISOString();
// Find the agent for this run by scanning heartbeat files
const files = await readdir(this.agentsDir).catch(() => [] as string[]);
const heartbeatFiles = files.filter((f) => f.endsWith("-heartbeats.jsonl"));
@@ -1246,7 +1255,19 @@ export class AgentStore extends EventEmitter {
// Check if this run exists in the history
const hasRun = history.some((h) => h.runId === runId);
if (hasRun) {
// Record end as special event
// Try to update the persisted run with terminal state
const existingRun = await this.getRunDetail(agentId, runId);
if (existingRun) {
// Update the persisted run in structured storage
const updatedRun: AgentHeartbeatRun = {
...existingRun,
endedAt: now,
status,
};
await this.saveRun(updatedRun);
}
// Also record heartbeat event for legacy compatibility
await this.recordHeartbeat(agentId, status === "terminated" ? "missed" : "ok", runId);
return;
}
@@ -1255,10 +1276,28 @@ export class AgentStore extends EventEmitter {
/**
* Get the active heartbeat run for an agent.
* Reads from structured run storage first (source of truth),
* falls back to heartbeat event reconstruction for legacy data.
* @param agentId - The agent ID
* @returns The active run, or null if none
*/
async getActiveHeartbeatRun(agentId: string): Promise<AgentHeartbeatRun | null> {
// First check structured run storage (source of truth)
const recentRuns = await this.getRecentRuns(agentId, 50);
// If we have structured run data, use it exclusively
if (recentRuns.length > 0) {
for (const run of recentRuns) {
if (run.status === "active") {
return run;
}
}
// We have structured data but no active runs - don't fall back
return null;
}
// Fallback: reconstruct from heartbeat events for legacy data
// This handles runs created before structured storage was used
const history = await this.getHeartbeatHistory(agentId, 100);
// Find the most recent run that started but hasn't ended
@@ -1296,10 +1335,24 @@ export class AgentStore extends EventEmitter {
/**
* Get all completed heartbeat runs for an agent.
* Reads from structured run storage first (source of truth),
* falls back to heartbeat event reconstruction for legacy data.
* Returns terminal runs (completed, terminated, failed) in newest-first order.
* @param agentId - The agent ID
* @returns Array of completed runs
*/
async getCompletedHeartbeatRuns(agentId: string): Promise<AgentHeartbeatRun[]> {
// First check structured run storage (source of truth)
const recentRuns = await this.getRecentRuns(agentId, 50);
// If we have structured run data, use it exclusively
if (recentRuns.length > 0) {
return recentRuns
.filter((run) => run.status !== "active")
.sort((a, b) => new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime());
}
// Fallback: reconstruct from heartbeat events for legacy data
const history = await this.getHeartbeatHistory(agentId, 1000);
const runs = new Map<string, AgentHeartbeatRun>();
@@ -1321,7 +1374,9 @@ export class AgentStore extends EventEmitter {
}
}
return Array.from(runs.values()).filter((r) => r.status !== "active");
return Array.from(runs.values())
.filter((r) => r.status !== "active")
.sort((a, b) => new Date(b.startedAt).getTime() - new Date(a.startedAt).getTime());
}
// ─────────────────────────────────────────────────────────────────────────