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

- 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 99ecab4324
commit 91f3085754
4 changed files with 259 additions and 19 deletions

View File

@@ -44,10 +44,10 @@ Heartbeat values are validated and minimum-clamped.
The New Agent dialog in the dashboard provides quick-start presets for common agent roles. Each preset includes: The New Agent dialog in the dashboard provides quick-start presets for common agent roles. Each preset includes:
- **Name and icon** Display identification - **Name and icon** - Display identification
- **Professional title** Descriptive role title - **Professional title** - Descriptive role title
- **Soul** Personality and operating principles defining how the agent thinks and communicates - **Soul** - Personality and operating principles defining how the agent thinks and communicates
- **Instructions** Actionable behavioral guidelines - **Instructions** - Actionable behavioral guidelines
### Preset Library Location ### Preset Library Location
@@ -98,10 +98,10 @@ Each `soul.md` file is a Markdown document containing:
Soul content should be: Soul content should be:
- **First-person** Written from the agent's perspective ("I am...") - **First-person** - Written from the agent's perspective ("I am...")
- **Role-specific** Defines the unique character of this role - **Role-specific** - Defines the unique character of this role
- **Actionable** Describes concrete behaviors, not abstract qualities - **Actionable** - Describes concrete behaviors, not abstract qualities
- **Paperclip-inspired** Clear ownership, decision discipline, communication standards - **Paperclip-inspired** - Clear ownership, decision discipline, communication standards
### Adding or Modifying Presets ### Adding or Modifying Presets
@@ -157,7 +157,7 @@ Behavior:
## Heartbeat Monitoring and Trigger Scheduling ## Heartbeat Monitoring and Trigger Scheduling
Fusions `HeartbeatTriggerScheduler` supports three trigger types: Fusion's `HeartbeatTriggerScheduler` supports three trigger types:
- `timer` — periodic wake based on heartbeat interval - `timer` — periodic wake based on heartbeat interval
- `assignment` — wake when task is assigned to agent - `assignment` — wake when task is assigned to agent
@@ -165,6 +165,58 @@ Fusions `HeartbeatTriggerScheduler` supports three trigger types:
All triggers respect per-agent `maxConcurrentRuns` and produce structured wake context metadata. All triggers respect per-agent `maxConcurrentRuns` and produce structured wake context metadata.
## Heartbeat Run Lifecycle
Agent runs have a defined lifecycle managed by `AgentStore`:
### Run States
A heartbeat run can be in one of these states:
- `active` — Run is currently executing
- `completed` — Run finished successfully (via `endHeartbeatRun(runId, "completed")`)
- `terminated` — Run was stopped (via `endHeartbeatRun(runId, "terminated")`)
- `failed` — Run encountered an error
### Run Lifecycle API
- `startHeartbeatRun(agentId)` — Creates a new run and persists it to structured storage
- `endHeartbeatRun(runId, status)` — Ends a run with terminal status, updates persisted state
- `getActiveHeartbeatRun(agentId)` — Returns the current active run (or null)
- `getCompletedHeartbeatRuns(agentId)` — Returns all terminal runs (newest first)
- `saveRun(run)` — Persists run to structured storage
- `getRunDetail(agentId, runId)` — Gets a specific run by ID
### Active-Run Conflict Semantics
When an agent already has an active run, attempts to start a new run return **409 Conflict**:
```
POST /api/agents/:id/runs → 409 { error: "Agent already has an active run", details: { runId } }
```
After a run is completed (or terminated), a new run can be started successfully:
```
POST /api/agents/:id/runs → 201 { id: "run-xxx", status: "active", ... }
```
### Storage Architecture
Run records are stored in structured JSON files at `.fusion/agents/{agentId}-runs/{runId}.json`.
Heartbeat events are also appended to `.fusion/agents/{agentId}-heartbeats.jsonl` for legacy compatibility. The structured storage is the source of truth; heartbeat events provide a fallback for older run data.
### Stopping Runs
Use `POST /api/agents/:id/runs/stop` to terminate an active run:
```
POST /api/agents/:id/runs/stop → 200 { ok: true, runId: "run-xxx" }
```
If there's no active run, returns `{ ok: true, message: "No active run" }`.
## Related Docs ## Related Docs
- [Workflow Steps](./workflow-steps.md) - [Workflow Steps](./workflow-steps.md)

View File

@@ -1719,18 +1719,22 @@ describe("AgentStore", () => {
expect(completed[0].endedAt).toBeDefined(); 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 agent = await store.createAgent({ name: "CompleteRun", role: "executor" });
const run = await store.startHeartbeatRun(agent.id); const run = await store.startHeartbeatRun(agent.id);
await store.endHeartbeatRun(run.id, "completed"); await store.endHeartbeatRun(run.id, "completed");
// A completed run records status "ok" (not "missed"), so the run // A completed run should NOT appear in active runs
// stays active in the reconstructed view (implementation detail).
// The getActiveHeartbeatRun still sees it as active since only
// "missed" status marks a run as terminated.
const active = await store.getActiveHeartbeatRun(agent.id); 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 () => { it("getCompletedHeartbeatRuns returns only non-active runs", async () => {
@@ -1751,6 +1755,94 @@ describe("AgentStore", () => {
expect(active).not.toBeNull(); expect(active).not.toBeNull();
expect(active!.id).toBe(run2.id); 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 ───────────────────────────────────── // ── blocked state persistence ─────────────────────────────────────

View File

@@ -1210,20 +1210,25 @@ export class AgentStore extends EventEmitter {
/** /**
* Start a new heartbeat run for an agent. * Start a new heartbeat run for an agent.
* Persists the run to structured storage as the source of truth.
* @param agentId - The agent ID * @param agentId - The agent ID
* @returns The created run * @returns The created run
*/ */
async startHeartbeatRun(agentId: string): Promise<AgentHeartbeatRun> { async startHeartbeatRun(agentId: string): Promise<AgentHeartbeatRun> {
const runId = `run-${randomUUID().slice(0, 8)}`; const runId = `run-${randomUUID().slice(0, 8)}`;
const now = new Date().toISOString();
const run: AgentHeartbeatRun = { const run: AgentHeartbeatRun = {
id: runId, id: runId,
agentId, agentId,
startedAt: new Date().toISOString(), startedAt: now,
endedAt: null, endedAt: null,
status: "active", 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); await this.recordHeartbeat(agentId, "ok", runId);
return run; return run;
@@ -1231,10 +1236,14 @@ export class AgentStore extends EventEmitter {
/** /**
* End a heartbeat run. * 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 runId - The run ID
* @param status - End status (completed or terminated) * @param status - End status (completed or terminated)
*/ */
async endHeartbeatRun(runId: string, status: "completed" | "terminated"): Promise<void> { async endHeartbeatRun(runId: string, status: "completed" | "terminated"): Promise<void> {
const now = new Date().toISOString();
// Find the agent for this run by scanning heartbeat files // Find the agent for this run by scanning heartbeat files
const files = await readdir(this.agentsDir).catch(() => [] as string[]); const files = await readdir(this.agentsDir).catch(() => [] as string[]);
const heartbeatFiles = files.filter((f) => f.endsWith("-heartbeats.jsonl")); 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 // Check if this run exists in the history
const hasRun = history.some((h) => h.runId === runId); const hasRun = history.some((h) => h.runId === runId);
if (hasRun) { 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); await this.recordHeartbeat(agentId, status === "terminated" ? "missed" : "ok", runId);
return; return;
} }
@@ -1255,10 +1276,28 @@ export class AgentStore extends EventEmitter {
/** /**
* Get the active heartbeat run for an agent. * 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 * @param agentId - The agent ID
* @returns The active run, or null if none * @returns The active run, or null if none
*/ */
async getActiveHeartbeatRun(agentId: string): Promise<AgentHeartbeatRun | null> { 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); const history = await this.getHeartbeatHistory(agentId, 100);
// Find the most recent run that started but hasn't ended // 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. * 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 * @param agentId - The agent ID
* @returns Array of completed runs * @returns Array of completed runs
*/ */
async getCompletedHeartbeatRuns(agentId: string): Promise<AgentHeartbeatRun[]> { 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 history = await this.getHeartbeatHistory(agentId, 1000);
const runs = new Map<string, AgentHeartbeatRun>(); 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());
} }
// ───────────────────────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────────────────────

View File

@@ -9712,6 +9712,47 @@ describe("POST /api/agents/:id/runs", () => {
expect(res2.body.error).toContain("active run"); expect(res2.body.error).toContain("active run");
expect(res2.body.details?.runId).toBeTruthy(); expect(res2.body.details?.runId).toBeTruthy();
}); });
it("returns 201 again after a prior run is completed via stop", async () => {
// Create first run
const res1 = await REQUEST(buildApp(), "POST", `/api/agents/${agentId}/runs`);
expect(res1.status).toBe(201);
const runId1 = res1.body.id;
// Stop the run
const stopRes = await REQUEST(buildApp(), "POST", `/api/agents/${agentId}/runs/stop`);
expect(stopRes.status).toBe(200);
expect(stopRes.body.runId).toBe(runId1);
// Create second run — should succeed now that first is complete
const res2 = await REQUEST(buildApp(), "POST", `/api/agents/${agentId}/runs`);
expect(res2.status).toBe(201);
expect(res2.body.id).not.toBe(runId1);
expect(res2.body.status).toBe("active");
});
it("returns 201 again after a prior run is completed via AgentStore.endHeartbeatRun", async () => {
// Create first run
const res1 = await REQUEST(buildApp(), "POST", `/api/agents/${agentId}/runs`);
expect(res1.status).toBe(201);
const runId1 = res1.body.id;
// Complete the run directly via AgentStore
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: fusionDir });
await agentStore.init();
await agentStore.endHeartbeatRun(runId1, "completed");
// Verify run is completed
const activeRun = await agentStore.getActiveHeartbeatRun(agentId);
expect(activeRun).toBeNull();
// Create second run — should succeed now that first is complete
const res2 = await REQUEST(buildApp(), "POST", `/api/agents/${agentId}/runs`);
expect(res2.status).toBe(201);
expect(res2.body.id).not.toBe(runId1);
expect(res2.body.status).toBe("active");
});
}); });
describe("GET /api/agents/:id/runs/:runId/logs", () => { describe("GET /api/agents/:id/runs/:runId/logs", () => {