feat(FN-1405): add run-audit and timeline API endpoints with typed client
- Add run-audit routes in routes.ts for starting/canceling/stream/querying agent run audits - Add timeline routes for streaming agent run timeline events - Create typed API client wrappers in api.ts for frontend consumption - Add comprehensive route-level verification tests for run-audit and timeline endpoints - Fix TypeScript errors in run-audit routes - Document run-audit endpoints in README
This commit is contained in:
gsxdsm
@@ -681,6 +681,45 @@ When `FUSION_BADGE_PUBSUB_REDIS_URL` is not set, the dashboard uses an in-memory
|
||||
- `GET /api/messages/conversation/:participantType/:participantId` - Get conversation thread
|
||||
- `GET /api/agents/:id/mailbox` - View agent mailbox (admin read-only)
|
||||
|
||||
### Agent Run Audit APIs
|
||||
|
||||
The dashboard exposes run-audit retrieval and correlation endpoints for inspecting agent run mutations and timelines:
|
||||
|
||||
#### Run Audit Events
|
||||
|
||||
- `GET /api/agents/:id/runs/:runId/audit` - Get normalized audit events for a specific run
|
||||
- **Query parameters:**
|
||||
- `taskId` (optional): Filter by task ID
|
||||
- `domain` (optional): Filter by domain (`database`, `git`, `filesystem`)
|
||||
- `startTime` (optional): ISO-8601 start of time range (inclusive)
|
||||
- `endTime` (optional): ISO-8601 end of time range (inclusive)
|
||||
- `limit` (optional): Maximum events to return (1-1000, default 100)
|
||||
- **Response:** Array of normalized audit events with stable UI-friendly field names
|
||||
- **Error codes:** `400` for invalid filters, `404` for unknown run
|
||||
|
||||
#### Run Timeline
|
||||
|
||||
- `GET /api/agents/:id/runs/:runId/timeline` - Get correlated timeline combining audit events and agent logs
|
||||
- **Query parameters:**
|
||||
- `taskId` (optional): Filter by task ID (defaults to run's contextSnapshot.taskId)
|
||||
- `domain` (optional): Filter by domain (`database`, `git`, `filesystem`)
|
||||
- `startTime` (optional): ISO-8601 start of time range (inclusive)
|
||||
- `endTime` (optional): ISO-8601 end of time range (inclusive)
|
||||
- `includeLogs` (optional): Include agent logs (default `true`)
|
||||
- **Response:**
|
||||
- `run`: Run metadata (id, agentId, startedAt, endedAt, status, taskId)
|
||||
- `auditByDomain`: Audit events grouped by domain (database, git, filesystem)
|
||||
- `counts`: Metadata counts (auditEvents, logEntries)
|
||||
- `timeline`: Merged and deterministically sorted timeline entries
|
||||
- **Error codes:** `400` for invalid filters, `404` for unknown run
|
||||
|
||||
**Timeline Sorting:** Entries are sorted by timestamp with a stable tie-breaker (entry type + domain) to ensure deterministic ordering when timestamps collide.
|
||||
|
||||
**Audit Event Domains:**
|
||||
- `database`: Database mutations (task updates, status changes)
|
||||
- `git`: Git mutations (commits, branch operations)
|
||||
- `filesystem`: Filesystem mutations (file reads, writes, deletes)
|
||||
|
||||
### Configuration
|
||||
- `GET /api/config` - Server configuration
|
||||
- `GET /api/settings` - Merged settings (project overrides global)
|
||||
|
||||
@@ -37,6 +37,7 @@ import type {
|
||||
AgentRatingInput,
|
||||
ChatSession,
|
||||
ChatMessage,
|
||||
RunAuditEvent,
|
||||
} from "@fusion/core";
|
||||
import type { PlanningQuestion, PlanningSummary, PlanningResponse } from "@fusion/core";
|
||||
import type { ScheduledTask, ScheduledTaskCreateInput, ScheduledTaskUpdateInput, AutomationRunResult, AutomationStep, Routine, RoutineCreateInput, RoutineUpdateInput, RoutineExecutionResult } from "@fusion/core";
|
||||
@@ -2222,6 +2223,150 @@ export function stopAgentRun(
|
||||
);
|
||||
}
|
||||
|
||||
// ── Run-Audit & Timeline API ────────────────────────────────────────────────
|
||||
|
||||
/** Valid domain filters for run-audit queries. */
|
||||
export type RunAuditDomainFilter = "database" | "git" | "filesystem";
|
||||
|
||||
/** Filter options for run-audit queries. */
|
||||
export interface RunAuditFilters {
|
||||
/** Filter by task ID */
|
||||
taskId?: string;
|
||||
/** Filter by domain category */
|
||||
domain?: RunAuditDomainFilter;
|
||||
/** Start of time range (inclusive, ISO-8601) */
|
||||
startTime?: string;
|
||||
/** End of time range (inclusive, ISO-8601) */
|
||||
endTime?: string;
|
||||
/** Maximum number of events to return */
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
/** Normalized run-audit event for UI consumption. */
|
||||
export interface NormalizedRunAuditEvent {
|
||||
id: string;
|
||||
timestamp: string;
|
||||
taskId?: string;
|
||||
domain: "database" | "git" | "filesystem";
|
||||
mutationType: string;
|
||||
target: string;
|
||||
summary: string;
|
||||
metadata?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** Response shape for run-audit endpoint. */
|
||||
export interface RunAuditResponse {
|
||||
runId: string;
|
||||
events: NormalizedRunAuditEvent[];
|
||||
filters: {
|
||||
taskId?: string;
|
||||
domain?: RunAuditDomainFilter;
|
||||
startTime?: string;
|
||||
endTime?: string;
|
||||
};
|
||||
totalCount: number;
|
||||
hasMore: boolean;
|
||||
}
|
||||
|
||||
/** Unified timeline entry that can represent either an audit event or an agent log entry. */
|
||||
export interface TimelineEntry {
|
||||
timestamp: string;
|
||||
type: "audit" | "log";
|
||||
sortKey: string;
|
||||
audit?: NormalizedRunAuditEvent;
|
||||
log?: AgentLogEntry;
|
||||
}
|
||||
|
||||
/** Response shape for run-timeline endpoint. */
|
||||
export interface RunTimelineResponse {
|
||||
run: {
|
||||
id: string;
|
||||
agentId: string;
|
||||
startedAt: string;
|
||||
endedAt?: string;
|
||||
status: string;
|
||||
taskId?: string;
|
||||
};
|
||||
auditByDomain: {
|
||||
database: NormalizedRunAuditEvent[];
|
||||
git: NormalizedRunAuditEvent[];
|
||||
filesystem: NormalizedRunAuditEvent[];
|
||||
};
|
||||
counts: {
|
||||
auditEvents: number;
|
||||
logEntries: number;
|
||||
};
|
||||
timeline: TimelineEntry[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch normalized run-audit events for a specific agent run.
|
||||
*
|
||||
* @param agentId - The agent ID
|
||||
* @param runId - The run ID
|
||||
* @param filters - Optional filter parameters
|
||||
* @param projectId - Optional project ID for multi-project workspaces
|
||||
* @returns Promise resolving to RunAuditResponse with normalized events
|
||||
*/
|
||||
export function fetchAgentRunAudit(
|
||||
agentId: string,
|
||||
runId: string,
|
||||
filters?: RunAuditFilters,
|
||||
projectId?: string,
|
||||
): Promise<RunAuditResponse> {
|
||||
const params = new URLSearchParams();
|
||||
if (filters?.taskId) params.set("taskId", filters.taskId);
|
||||
if (filters?.domain) params.set("domain", filters.domain);
|
||||
if (filters?.startTime) params.set("startTime", filters.startTime);
|
||||
if (filters?.endTime) params.set("endTime", filters.endTime);
|
||||
if (filters?.limit !== undefined) params.set("limit", String(filters.limit));
|
||||
const query = params.size > 0 ? `?${params.toString()}` : "";
|
||||
return api<RunAuditResponse>(
|
||||
withProjectId(`/agents/${encodeURIComponent(agentId)}/runs/${encodeURIComponent(runId)}/audit${query}`, projectId),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch a correlated timeline combining run-audit events and agent logs for a specific run.
|
||||
*
|
||||
* @param agentId - The agent ID
|
||||
* @param runId - The run ID
|
||||
* @param options - Optional parameters
|
||||
* @param options.taskId - Override task ID for audit filtering (defaults to run's contextSnapshot.taskId)
|
||||
* @param options.domain - Filter audit events by domain
|
||||
* @param options.startTime - Start of time range (ISO-8601)
|
||||
* @param options.endTime - End of time range (ISO-8601)
|
||||
* @param options.includeLogs - Whether to include agent logs (default true)
|
||||
* @param options.limit - Maximum audit events to return
|
||||
* @param projectId - Optional project ID for multi-project workspaces
|
||||
* @returns Promise resolving to RunTimelineResponse with merged timeline
|
||||
*/
|
||||
export function fetchAgentRunTimeline(
|
||||
agentId: string,
|
||||
runId: string,
|
||||
options?: {
|
||||
taskId?: string;
|
||||
domain?: RunAuditDomainFilter;
|
||||
startTime?: string;
|
||||
endTime?: string;
|
||||
includeLogs?: boolean;
|
||||
limit?: number;
|
||||
},
|
||||
projectId?: string,
|
||||
): Promise<RunTimelineResponse> {
|
||||
const params = new URLSearchParams();
|
||||
if (options?.taskId) params.set("taskId", options.taskId);
|
||||
if (options?.domain) params.set("domain", options.domain);
|
||||
if (options?.startTime) params.set("startTime", options.startTime);
|
||||
if (options?.endTime) params.set("endTime", options.endTime);
|
||||
if (options?.includeLogs !== undefined) params.set("includeLogs", String(options.includeLogs));
|
||||
if (options?.limit !== undefined) params.set("limit", String(options.limit));
|
||||
const query = params.size > 0 ? `?${params.toString()}` : "";
|
||||
return api<RunTimelineResponse>(
|
||||
withProjectId(`/agents/${encodeURIComponent(agentId)}/runs/${encodeURIComponent(runId)}/timeline${query}`, projectId),
|
||||
);
|
||||
}
|
||||
|
||||
/** Fetch aggregate agent stats */
|
||||
export function fetchAgentStats(projectId?: string): Promise<AgentStats> {
|
||||
return api<AgentStats>(withProjectId("/agents/stats", projectId));
|
||||
|
||||
@@ -16,6 +16,12 @@ const mockEndHeartbeatRun = vi.fn();
|
||||
const mockListAgents = vi.fn().mockResolvedValue([]);
|
||||
const mockGetActiveHeartbeatRun = vi.fn().mockResolvedValue(null);
|
||||
|
||||
// Mock ChatStore methods
|
||||
const mockChatStoreInit = vi.fn().mockResolvedValue(undefined);
|
||||
|
||||
// Mock getRunAuditEvents
|
||||
const mockGetRunAuditEvents = vi.fn().mockReturnValue([]);
|
||||
|
||||
vi.mock("@fusion/core", () => {
|
||||
return {
|
||||
AgentStore: class MockAgentStore {
|
||||
@@ -31,12 +37,31 @@ vi.mock("@fusion/core", () => {
|
||||
listAgents = mockListAgents;
|
||||
getActiveHeartbeatRun = mockGetActiveHeartbeatRun;
|
||||
},
|
||||
ChatStore: class MockChatStore {
|
||||
init = mockChatStoreInit;
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
// ── Mock project-store-resolver ─────────────────────────────────────
|
||||
|
||||
const mockGetOrCreateProjectStore = vi.fn();
|
||||
|
||||
vi.mock("../project-store-resolver.js", () => ({
|
||||
getOrCreateProjectStore: mockGetOrCreateProjectStore,
|
||||
}));
|
||||
|
||||
// ── Mock Store ────────────────────────────────────────────────────────
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
type TaskStore = any;
|
||||
|
||||
class MockStore extends EventEmitter {
|
||||
// Mock methods for run-audit and mutations
|
||||
getRunAuditEvents = mockGetRunAuditEvents;
|
||||
getMutationsForRun = vi.fn().mockResolvedValue([]);
|
||||
getAgentLogsByTimeRange = vi.fn().mockResolvedValue([]);
|
||||
|
||||
getRootDir(): string {
|
||||
return "/tmp/fn-1059-test";
|
||||
}
|
||||
@@ -483,22 +508,16 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
|
||||
});
|
||||
|
||||
describe("GET /api/agents/:id/runs/:runId/mutations", () => {
|
||||
beforeEach(() => {
|
||||
// Reset scoped store mock
|
||||
mockScopedStore = createMockScopedStore();
|
||||
(createScopedStore as any).mockReturnValue(mockScopedStore);
|
||||
});
|
||||
|
||||
it("returns mutation trail for a valid run", async () => {
|
||||
const mockRun = createMockRun();
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
|
||||
// Mock getMutationsForRun on the scoped store
|
||||
// Mock getMutationsForRun on the store
|
||||
const mockMutations = [
|
||||
{ timestamp: "2026-01-01T00:01:00.000Z", action: "Action 1", runContext: { runId: "run-123", agentId: "agent-001" } },
|
||||
{ timestamp: "2026-01-01T00:02:00.000Z", action: "Action 2", runContext: { runId: "run-123", agentId: "agent-001" } },
|
||||
];
|
||||
mockScopedStore.getMutationsForRun = vi.fn().mockResolvedValue(mockMutations);
|
||||
store.getMutationsForRun = vi.fn().mockResolvedValue(mockMutations);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-123/mutations");
|
||||
|
||||
@@ -507,7 +526,7 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
|
||||
runId: "run-123",
|
||||
mutations: mockMutations,
|
||||
});
|
||||
expect(mockScopedStore.getMutationsForRun).toHaveBeenCalledWith("run-123");
|
||||
expect(store.getMutationsForRun).toHaveBeenCalledWith("run-123");
|
||||
});
|
||||
|
||||
it("returns 404 for unknown run", async () => {
|
||||
@@ -524,7 +543,7 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
|
||||
// Mock getMutationsForRun returning empty array
|
||||
mockScopedStore.getMutationsForRun = vi.fn().mockResolvedValue([]);
|
||||
store.getMutationsForRun = vi.fn().mockResolvedValue([]);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-empty/mutations");
|
||||
|
||||
@@ -535,4 +554,240 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("GET /api/agents/:id/runs/:runId/audit", () => {
|
||||
it("returns normalized audit events for a valid run", async () => {
|
||||
const mockRun = createMockRun();
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
|
||||
// Mock getRunAuditEvents
|
||||
const mockAuditEvents = [
|
||||
{
|
||||
id: "audit-1",
|
||||
timestamp: "2026-01-01T00:01:00.000Z",
|
||||
agentId: "agent-001",
|
||||
runId: "run-001",
|
||||
domain: "database",
|
||||
mutationType: "task:update",
|
||||
target: "FN-001",
|
||||
taskId: "FN-001",
|
||||
},
|
||||
{
|
||||
id: "audit-2",
|
||||
timestamp: "2026-01-01T00:02:00.000Z",
|
||||
agentId: "agent-001",
|
||||
runId: "run-001",
|
||||
domain: "git",
|
||||
mutationType: "git:commit",
|
||||
target: "fusion/FN-001",
|
||||
},
|
||||
];
|
||||
mockGetRunAuditEvents.mockReturnValue(mockAuditEvents);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/audit");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.runId).toBe("run-001");
|
||||
expect(Array.isArray(response.body.events)).toBe(true);
|
||||
expect(response.body.events.length).toBe(2);
|
||||
expect(response.body.totalCount).toBe(2);
|
||||
expect(response.body.hasMore).toBe(false);
|
||||
// Check normalized fields
|
||||
expect(response.body.events[0].summary).toBe("DB update (FN-001)");
|
||||
expect(response.body.events[1].summary).toBe("Git commit (fusion/FN-001)");
|
||||
});
|
||||
|
||||
it("returns 404 for unknown run", async () => {
|
||||
mockGetRunDetail.mockResolvedValue(null);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-unknown/audit");
|
||||
|
||||
expect(response.status).toBe(404);
|
||||
expect(response.body).toHaveProperty("error");
|
||||
});
|
||||
|
||||
it("returns empty events array when no audit events exist", async () => {
|
||||
const mockRun = createMockRun();
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
mockGetRunAuditEvents.mockReturnValue([]);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/audit");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.events).toEqual([]);
|
||||
expect(response.body.totalCount).toBe(0);
|
||||
});
|
||||
|
||||
it("applies domain filter correctly", async () => {
|
||||
const mockRun = createMockRun();
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
mockGetRunAuditEvents.mockReturnValue([]);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/audit?domain=git");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(mockGetRunAuditEvents).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ domain: "git" }),
|
||||
);
|
||||
});
|
||||
|
||||
it("returns 400 for invalid domain filter", async () => {
|
||||
const mockRun = createMockRun();
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/audit?domain=invalid");
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
expect(response.body.error).toContain("domain must be one of");
|
||||
});
|
||||
|
||||
it("returns 400 for invalid limit", async () => {
|
||||
const mockRun = createMockRun();
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/audit?limit=-1");
|
||||
|
||||
expect(response.status).toBe(400);
|
||||
expect(response.body.error).toContain("limit must be a positive integer");
|
||||
});
|
||||
});
|
||||
|
||||
describe("GET /api/agents/:id/runs/:runId/timeline", () => {
|
||||
it("returns correlated timeline with audit events and logs", async () => {
|
||||
const mockRun = createMockRun({
|
||||
status: "completed",
|
||||
endedAt: "2026-01-01T00:10:00.000Z",
|
||||
contextSnapshot: { taskId: "FN-001" },
|
||||
});
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
|
||||
// Mock audit events
|
||||
const mockAuditEvents = [
|
||||
{
|
||||
id: "audit-1",
|
||||
timestamp: "2026-01-01T00:01:00.000Z",
|
||||
agentId: "agent-001",
|
||||
runId: "run-001",
|
||||
domain: "database",
|
||||
mutationType: "task:update",
|
||||
target: "FN-001",
|
||||
taskId: "FN-001",
|
||||
},
|
||||
];
|
||||
mockGetRunAuditEvents.mockReturnValue(mockAuditEvents);
|
||||
|
||||
// Mock logs
|
||||
const mockLogs = [
|
||||
{ id: "log-1", timestamp: "2026-01-01T00:00:30.000Z", type: "info", message: "Starting task" },
|
||||
{ id: "log-2", timestamp: "2026-01-01T00:01:30.000Z", type: "info", message: "Task completed" },
|
||||
];
|
||||
store.getAgentLogsByTimeRange = vi.fn().mockResolvedValue(mockLogs);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/timeline");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.run.id).toBe("run-001");
|
||||
expect(response.body.run.taskId).toBe("FN-001");
|
||||
expect(Array.isArray(response.body.auditByDomain.database)).toBe(true);
|
||||
expect(Array.isArray(response.body.auditByDomain.git)).toBe(true);
|
||||
expect(Array.isArray(response.body.auditByDomain.filesystem)).toBe(true);
|
||||
expect(response.body.auditByDomain.database.length).toBe(1);
|
||||
expect(response.body.counts.auditEvents).toBe(1);
|
||||
expect(response.body.counts.logEntries).toBe(2);
|
||||
expect(Array.isArray(response.body.timeline)).toBe(true);
|
||||
expect(response.body.timeline.length).toBe(3); // 1 audit + 2 logs
|
||||
// Timeline should be sorted by timestamp
|
||||
expect(response.body.timeline[0].type).toBe("log"); // Earlier log
|
||||
expect(response.body.timeline[1].type).toBe("audit"); // Audit event
|
||||
expect(response.body.timeline[2].type).toBe("log"); // Later log
|
||||
});
|
||||
|
||||
it("returns 404 for unknown run", async () => {
|
||||
mockGetRunDetail.mockResolvedValue(null);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-unknown/timeline");
|
||||
|
||||
expect(response.status).toBe(404);
|
||||
expect(response.body).toHaveProperty("error");
|
||||
});
|
||||
|
||||
it("respects includeLogs=false parameter", async () => {
|
||||
const mockRun = createMockRun({
|
||||
contextSnapshot: { taskId: "FN-001" },
|
||||
});
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
mockGetRunAuditEvents.mockReturnValue([]);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/timeline?includeLogs=false");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.counts.logEntries).toBe(0);
|
||||
});
|
||||
|
||||
it("handles empty audit and log results gracefully", async () => {
|
||||
const mockRun = createMockRun({
|
||||
status: "completed",
|
||||
endedAt: "2026-01-01T00:10:00.000Z",
|
||||
contextSnapshot: { taskId: "FN-001" },
|
||||
});
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
mockGetRunAuditEvents.mockReturnValue([]);
|
||||
store.getAgentLogsByTimeRange = vi.fn().mockResolvedValue([]);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/timeline");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.auditByDomain.database).toEqual([]);
|
||||
expect(response.body.auditByDomain.git).toEqual([]);
|
||||
expect(response.body.auditByDomain.filesystem).toEqual([]);
|
||||
expect(response.body.counts.auditEvents).toBe(0);
|
||||
expect(response.body.counts.logEntries).toBe(0);
|
||||
expect(response.body.timeline).toEqual([]);
|
||||
});
|
||||
|
||||
it("groups audit events by domain correctly", async () => {
|
||||
const mockRun = createMockRun();
|
||||
mockGetRunDetail.mockResolvedValue(mockRun);
|
||||
|
||||
const mockAuditEvents = [
|
||||
{
|
||||
id: "audit-db",
|
||||
timestamp: "2026-01-01T00:01:00.000Z",
|
||||
agentId: "agent-001",
|
||||
runId: "run-001",
|
||||
domain: "database",
|
||||
mutationType: "task:update",
|
||||
target: "FN-001",
|
||||
},
|
||||
{
|
||||
id: "audit-git",
|
||||
timestamp: "2026-01-01T00:02:00.000Z",
|
||||
agentId: "agent-001",
|
||||
runId: "run-001",
|
||||
domain: "git",
|
||||
mutationType: "git:commit",
|
||||
target: "branch",
|
||||
},
|
||||
{
|
||||
id: "audit-fs",
|
||||
timestamp: "2026-01-01T00:03:00.000Z",
|
||||
agentId: "agent-001",
|
||||
runId: "run-001",
|
||||
domain: "filesystem",
|
||||
mutationType: "file:write",
|
||||
target: "src/main.ts",
|
||||
},
|
||||
];
|
||||
mockGetRunAuditEvents.mockReturnValue(mockAuditEvents);
|
||||
store.getAgentLogsByTimeRange = vi.fn().mockResolvedValue([]);
|
||||
|
||||
const response = await request(app, "GET", "/api/agents/agent-001/runs/run-001/timeline");
|
||||
|
||||
expect(response.status).toBe(200);
|
||||
expect(response.body.auditByDomain.database.length).toBe(1);
|
||||
expect(response.body.auditByDomain.git.length).toBe(1);
|
||||
expect(response.body.auditByDomain.filesystem.length).toBe(1);
|
||||
expect(response.body.counts.auditEvents).toBe(3);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -247,6 +247,309 @@ function validateModelPresets(value: unknown): ModelPreset[] | undefined {
|
||||
});
|
||||
}
|
||||
|
||||
// ── Run-Audit Timeline Types & Helpers ─────────────────────────────────────
|
||||
|
||||
/** Valid domain filters for run-audit queries. */
|
||||
export type RunAuditDomainFilter = "database" | "git" | "filesystem";
|
||||
|
||||
/** Filter options for run-audit queries. */
|
||||
export interface RunAuditQueryFilters {
|
||||
/** Filter by task ID */
|
||||
taskId?: string;
|
||||
/** Filter by domain category */
|
||||
domain?: RunAuditDomainFilter;
|
||||
/** Start of time range (inclusive, ISO-8601) */
|
||||
startTime?: string;
|
||||
/** End of time range (inclusive, ISO-8601) */
|
||||
endTime?: string;
|
||||
/** Maximum number of events to return */
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalized run-audit event for UI consumption.
|
||||
* Provides stable, user-friendly field names.
|
||||
*/
|
||||
export interface NormalizedRunAuditEvent {
|
||||
/** Unique event identifier */
|
||||
id: string;
|
||||
/** ISO-8601 timestamp when the event occurred */
|
||||
timestamp: string;
|
||||
/** Task ID associated with this event (if applicable) */
|
||||
taskId?: string;
|
||||
/** Domain category: database, git, or filesystem */
|
||||
domain: "database" | "git" | "filesystem";
|
||||
/** Type of mutation (e.g., "task:update", "git:commit", "file:write") */
|
||||
mutationType: string;
|
||||
/** Target of the mutation (e.g., task ID, file path, branch name) */
|
||||
target: string;
|
||||
/** Human-readable summary of the mutation */
|
||||
summary: string;
|
||||
/** Structured metadata about the mutation */
|
||||
metadata?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Unified timeline entry that can represent either an audit event or an agent log entry.
|
||||
* Used for correlated timeline views.
|
||||
*/
|
||||
export interface TimelineEntry {
|
||||
/** ISO-8601 timestamp when the entry occurred */
|
||||
timestamp: string;
|
||||
/** Entry type discriminator */
|
||||
type: "audit" | "log";
|
||||
/** Stable sort key to ensure deterministic ordering for identical timestamps */
|
||||
sortKey: string;
|
||||
/** Normalized audit event (when type is "audit") */
|
||||
audit?: NormalizedRunAuditEvent;
|
||||
/** Agent log entry (when type is "log") */
|
||||
log?: import("@fusion/core").AgentLogEntry;
|
||||
}
|
||||
|
||||
/**
|
||||
* Response shape for GET /api/agents/:id/runs/:runId/audit
|
||||
*/
|
||||
export interface RunAuditResponse {
|
||||
/** The run ID these events belong to */
|
||||
runId: string;
|
||||
/** Normalized audit events */
|
||||
events: NormalizedRunAuditEvent[];
|
||||
/** Filter metadata */
|
||||
filters: {
|
||||
taskId?: string;
|
||||
domain?: RunAuditDomainFilter;
|
||||
startTime?: string;
|
||||
endTime?: string;
|
||||
};
|
||||
/** Total count of events matching filters */
|
||||
totalCount: number;
|
||||
/** Whether there are more events (when limit was applied) */
|
||||
hasMore: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Response shape for GET /api/agents/:id/runs/:runId/timeline
|
||||
*/
|
||||
export interface RunTimelineResponse {
|
||||
/** Run metadata */
|
||||
run: {
|
||||
id: string;
|
||||
agentId: string;
|
||||
startedAt: string;
|
||||
endedAt?: string;
|
||||
status: string;
|
||||
taskId?: string;
|
||||
};
|
||||
/** Grouped audit events by domain */
|
||||
auditByDomain: {
|
||||
database: NormalizedRunAuditEvent[];
|
||||
git: NormalizedRunAuditEvent[];
|
||||
filesystem: NormalizedRunAuditEvent[];
|
||||
};
|
||||
/** Count metadata */
|
||||
counts: {
|
||||
auditEvents: number;
|
||||
logEntries: number;
|
||||
};
|
||||
/** Merged and deterministically sorted timeline */
|
||||
timeline: TimelineEntry[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse and validate run-audit query filters from request query params.
|
||||
* Throws ApiError with 400 for invalid values.
|
||||
*/
|
||||
function parseRunAuditFilters(query: Record<string, unknown>): RunAuditQueryFilters {
|
||||
const filters: RunAuditQueryFilters = {};
|
||||
|
||||
// Parse taskId
|
||||
if (query.taskId !== undefined) {
|
||||
if (typeof query.taskId !== "string" || !query.taskId.trim()) {
|
||||
throw new ApiError(400, "taskId must be a non-empty string");
|
||||
}
|
||||
filters.taskId = query.taskId.trim();
|
||||
}
|
||||
|
||||
// Parse domain
|
||||
if (query.domain !== undefined) {
|
||||
if (typeof query.domain !== "string") {
|
||||
throw new ApiError(400, "domain must be a string");
|
||||
}
|
||||
const domain = query.domain.toLowerCase();
|
||||
if (domain !== "database" && domain !== "git" && domain !== "filesystem") {
|
||||
throw new ApiError(400, "domain must be one of: database, git, filesystem");
|
||||
}
|
||||
filters.domain = domain as RunAuditDomainFilter;
|
||||
}
|
||||
|
||||
// Parse startTime
|
||||
if (query.startTime !== undefined) {
|
||||
if (typeof query.startTime !== "string" || !query.startTime.trim()) {
|
||||
throw new ApiError(400, "startTime must be a non-empty ISO-8601 string");
|
||||
}
|
||||
const date = new Date(query.startTime);
|
||||
if (isNaN(date.getTime())) {
|
||||
throw new ApiError(400, "startTime must be a valid ISO-8601 date string");
|
||||
}
|
||||
filters.startTime = query.startTime.trim();
|
||||
}
|
||||
|
||||
// Parse endTime
|
||||
if (query.endTime !== undefined) {
|
||||
if (typeof query.endTime !== "string" || !query.endTime.trim()) {
|
||||
throw new ApiError(400, "endTime must be a non-empty ISO-8601 string");
|
||||
}
|
||||
const date = new Date(query.endTime);
|
||||
if (isNaN(date.getTime())) {
|
||||
throw new ApiError(400, "endTime must be a valid ISO-8601 date string");
|
||||
}
|
||||
filters.endTime = query.endTime.trim();
|
||||
}
|
||||
|
||||
// Validate time range consistency
|
||||
if (filters.startTime && filters.endTime) {
|
||||
const start = new Date(filters.startTime);
|
||||
const end = new Date(filters.endTime);
|
||||
if (start > end) {
|
||||
throw new ApiError(400, "startTime must be before or equal to endTime");
|
||||
}
|
||||
}
|
||||
|
||||
// Parse limit
|
||||
if (query.limit !== undefined) {
|
||||
const limitStr = typeof query.limit === "string" ? query.limit : String(query.limit);
|
||||
const limit = parseInt(limitStr, 10);
|
||||
if (!Number.isFinite(limit) || limit < 1) {
|
||||
throw new ApiError(400, "limit must be a positive integer");
|
||||
}
|
||||
filters.limit = Math.min(limit, 1000); // Cap at 1000
|
||||
}
|
||||
|
||||
return filters;
|
||||
}
|
||||
|
||||
/**
|
||||
* Normalize a raw RunAuditEvent to a NormalizedRunAuditEvent for UI consumption.
|
||||
*/
|
||||
function normalizeRunAuditEvent(event: import("@fusion/core").RunAuditEvent): NormalizedRunAuditEvent {
|
||||
// Generate a human-readable summary based on domain and mutation type
|
||||
let summary = generateAuditSummary(event.domain, event.mutationType, event.target, event.metadata);
|
||||
|
||||
return {
|
||||
id: event.id,
|
||||
timestamp: event.timestamp,
|
||||
taskId: event.taskId,
|
||||
domain: event.domain,
|
||||
mutationType: event.mutationType,
|
||||
target: event.target,
|
||||
summary,
|
||||
metadata: event.metadata,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Generate a human-readable summary for an audit event.
|
||||
*/
|
||||
function generateAuditSummary(
|
||||
domain: string,
|
||||
mutationType: string,
|
||||
target: string,
|
||||
metadata?: Record<string, unknown>,
|
||||
): string {
|
||||
const parts: string[] = [];
|
||||
|
||||
// Add domain prefix
|
||||
switch (domain) {
|
||||
case "database":
|
||||
parts.push("DB");
|
||||
break;
|
||||
case "git":
|
||||
parts.push("Git");
|
||||
break;
|
||||
case "filesystem":
|
||||
parts.push("FS");
|
||||
break;
|
||||
default:
|
||||
parts.push(domain);
|
||||
}
|
||||
|
||||
// Add mutation action
|
||||
const action = mutationType.split(":").pop() ?? mutationType;
|
||||
parts.push(action);
|
||||
|
||||
// Add target context
|
||||
if (target) {
|
||||
// Truncate long targets for readability
|
||||
const displayTarget = target.length > 50 ? `${target.slice(0, 47)}...` : target;
|
||||
parts.push(`(${displayTarget})`);
|
||||
}
|
||||
|
||||
return parts.join(" ");
|
||||
}
|
||||
|
||||
/**
|
||||
* Sort comparator for timeline entries with deterministic tie-breaking.
|
||||
* Primary sort: timestamp ascending
|
||||
* Tie-breaker: sortKey ascending (which incorporates type and event ID)
|
||||
*/
|
||||
function compareTimelineEntries(a: TimelineEntry, b: TimelineEntry): number {
|
||||
const timeA = new Date(a.timestamp).getTime();
|
||||
const timeB = new Date(b.timestamp).getTime();
|
||||
|
||||
if (timeA !== timeB) {
|
||||
return timeA - timeB;
|
||||
}
|
||||
|
||||
// Deterministic tie-breaker: sortKey ascending
|
||||
// This ensures consistent ordering when timestamps are identical
|
||||
return a.sortKey.localeCompare(b.sortKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a stable sort key for a timeline entry.
|
||||
* Format: "{type_prefix}_{timestamp_ms}_{entry_id}"
|
||||
* The type prefix ensures audit events and log entries don't conflict.
|
||||
* The timestamp in ms ensures microsecond precision.
|
||||
* The entry ID provides final tie-breaking.
|
||||
*/
|
||||
function createTimelineSortKey(
|
||||
type: "audit" | "log",
|
||||
timestamp: string,
|
||||
id: string,
|
||||
): string {
|
||||
const ms = new Date(timestamp).getTime();
|
||||
const typePrefix = type === "audit" ? "A" : "L";
|
||||
// Use a sanitized ID that won't interfere with sorting
|
||||
const sanitizedId = id.replace(/[^a-zA-Z0-9_-]/g, "_");
|
||||
return `${typePrefix}_${String(ms).padStart(16, "0")}_${sanitizedId}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert an audit event to a timeline entry.
|
||||
*/
|
||||
function auditEventToTimelineEntry(event: import("@fusion/core").RunAuditEvent): TimelineEntry {
|
||||
const normalized = normalizeRunAuditEvent(event);
|
||||
return {
|
||||
timestamp: event.timestamp,
|
||||
type: "audit",
|
||||
sortKey: createTimelineSortKey("audit", event.timestamp, event.id),
|
||||
audit: normalized,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert an agent log entry to a timeline entry.
|
||||
*/
|
||||
function logEntryToTimelineEntry(entry: import("@fusion/core").AgentLogEntry): TimelineEntry {
|
||||
// Use timestamp as the unique sort key for log entries (AgentLogEntry has no id field)
|
||||
return {
|
||||
timestamp: entry.timestamp,
|
||||
type: "log",
|
||||
sortKey: createTimelineSortKey("log", entry.timestamp, entry.timestamp),
|
||||
log: entry,
|
||||
};
|
||||
}
|
||||
|
||||
// ── Git Remote Detection ──────────────────────────────────────────
|
||||
|
||||
/** Git remote info returned by the remotes endpoint */
|
||||
@@ -10099,6 +10402,221 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* GET /api/agents/:id/runs/:runId/audit
|
||||
* Get normalized run-audit events for a specific agent run.
|
||||
*
|
||||
* Query params:
|
||||
* - taskId: Filter by task ID
|
||||
* - domain: Filter by domain (database, git, filesystem)
|
||||
* - startTime: Start of time range (ISO-8601)
|
||||
* - endTime: End of time range (ISO-8601)
|
||||
* - limit: Maximum events to return (default 100, max 1000)
|
||||
*
|
||||
* Response: RunAuditResponse with normalized events and filter metadata
|
||||
*/
|
||||
router.get("/agents/:id/runs/:runId/audit", async (req, res) => {
|
||||
try {
|
||||
const scopedStore = await getScopedStore(req);
|
||||
const { AgentStore } = await import("@fusion/core");
|
||||
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
|
||||
await agentStore.init();
|
||||
|
||||
// Verify the run exists
|
||||
const run = await agentStore.getRunDetail(req.params.id, req.params.runId);
|
||||
if (!run) {
|
||||
throw notFound("Run not found");
|
||||
}
|
||||
|
||||
// Parse and validate query filters
|
||||
const filters = parseRunAuditFilters(req.query as Record<string, unknown>);
|
||||
|
||||
// Query run-audit events with runId as the primary filter
|
||||
const auditEvents = scopedStore.getRunAuditEvents({
|
||||
runId: req.params.runId,
|
||||
taskId: filters.taskId,
|
||||
domain: filters.domain,
|
||||
startTime: filters.startTime,
|
||||
endTime: filters.endTime,
|
||||
limit: filters.limit,
|
||||
});
|
||||
|
||||
// Normalize events for UI consumption
|
||||
const normalizedEvents = auditEvents.map(normalizeRunAuditEvent);
|
||||
|
||||
// Get total count (without limit) for pagination metadata
|
||||
const totalEvents = scopedStore.getRunAuditEvents({
|
||||
runId: req.params.runId,
|
||||
taskId: filters.taskId,
|
||||
domain: filters.domain,
|
||||
startTime: filters.startTime,
|
||||
endTime: filters.endTime,
|
||||
});
|
||||
|
||||
const response: RunAuditResponse = {
|
||||
runId: req.params.runId,
|
||||
events: normalizedEvents,
|
||||
filters: {
|
||||
taskId: filters.taskId,
|
||||
domain: filters.domain,
|
||||
startTime: filters.startTime,
|
||||
endTime: filters.endTime,
|
||||
},
|
||||
totalCount: totalEvents.length,
|
||||
hasMore: filters.limit !== undefined && totalEvents.length > filters.limit,
|
||||
};
|
||||
|
||||
res.json(response);
|
||||
} catch (err: any) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
}
|
||||
if (err.message?.includes("not found")) {
|
||||
throw notFound(err.message);
|
||||
} else {
|
||||
rethrowAsApiError(err);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* GET /api/agents/:id/runs/:runId/timeline
|
||||
* Get a correlated timeline combining run-audit events and agent logs for a specific run.
|
||||
*
|
||||
* Query params:
|
||||
* - taskId: Override task ID for audit filtering (defaults to run's contextSnapshot.taskId)
|
||||
* - domain: Filter audit events by domain (database, git, filesystem)
|
||||
* - startTime: Start of time range (ISO-8601)
|
||||
* - endTime: End of time range (ISO-8601)
|
||||
* - includeLogs: Whether to include agent logs (default true)
|
||||
* - limit: Maximum audit events to return (default 100, max 1000)
|
||||
*
|
||||
* Response: RunTimelineResponse with run metadata, grouped audit events, and merged timeline
|
||||
*/
|
||||
router.get("/agents/:id/runs/:runId/timeline", async (req, res) => {
|
||||
try {
|
||||
const scopedStore = await getScopedStore(req);
|
||||
const { AgentStore } = await import("@fusion/core");
|
||||
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
|
||||
await agentStore.init();
|
||||
|
||||
// Verify the run exists
|
||||
const run = await agentStore.getRunDetail(req.params.id, req.params.runId);
|
||||
if (!run) {
|
||||
throw notFound("Run not found");
|
||||
}
|
||||
|
||||
// Parse and validate query filters
|
||||
const filters = parseRunAuditFilters(req.query as Record<string, unknown>);
|
||||
|
||||
// Check includeLogs flag (default true)
|
||||
const includeLogs = (() => {
|
||||
if (req.query.includeLogs === undefined) return true;
|
||||
if (typeof req.query.includeLogs === "string") {
|
||||
const val = req.query.includeLogs.toLowerCase();
|
||||
return val === "true" || val === "1";
|
||||
}
|
||||
if (typeof req.query.includeLogs === "boolean") {
|
||||
return req.query.includeLogs;
|
||||
}
|
||||
return true;
|
||||
})();
|
||||
|
||||
// Determine the task ID for audit filtering
|
||||
// Use explicit taskId filter if provided, otherwise fall back to run's contextSnapshot.taskId
|
||||
const auditTaskId = (filters.taskId ?? run.contextSnapshot?.taskId ?? undefined) as string | undefined;
|
||||
|
||||
// Query run-audit events
|
||||
const auditEvents = scopedStore.getRunAuditEvents({
|
||||
runId: req.params.runId,
|
||||
taskId: auditTaskId,
|
||||
domain: filters.domain,
|
||||
startTime: filters.startTime,
|
||||
endTime: filters.endTime,
|
||||
limit: filters.limit,
|
||||
});
|
||||
|
||||
// Normalize events
|
||||
const normalizedAuditEvents = auditEvents.map(normalizeRunAuditEvent);
|
||||
|
||||
// Group audit events by domain
|
||||
const auditByDomain: RunTimelineResponse["auditByDomain"] = {
|
||||
database: [],
|
||||
git: [],
|
||||
filesystem: [],
|
||||
};
|
||||
|
||||
for (const event of normalizedAuditEvents) {
|
||||
if (event.domain === "database") {
|
||||
auditByDomain.database.push(event);
|
||||
} else if (event.domain === "git") {
|
||||
auditByDomain.git.push(event);
|
||||
} else if (event.domain === "filesystem") {
|
||||
auditByDomain.filesystem.push(event);
|
||||
}
|
||||
}
|
||||
|
||||
// Build timeline entries
|
||||
const timelineEntries: TimelineEntry[] = [];
|
||||
|
||||
// Add audit events to timeline
|
||||
for (const event of auditEvents) {
|
||||
timelineEntries.push(auditEventToTimelineEntry(event));
|
||||
}
|
||||
|
||||
// Add agent logs to timeline if requested and we have a task ID
|
||||
if (includeLogs && run.startedAt) {
|
||||
const taskId = auditTaskId;
|
||||
if (taskId) {
|
||||
const logs = await scopedStore.getAgentLogsByTimeRange(
|
||||
taskId,
|
||||
run.startedAt,
|
||||
run.endedAt,
|
||||
);
|
||||
|
||||
for (const log of logs) {
|
||||
timelineEntries.push(logEntryToTimelineEntry(log));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Sort timeline deterministically
|
||||
timelineEntries.sort(compareTimelineEntries);
|
||||
|
||||
const response: RunTimelineResponse = {
|
||||
run: {
|
||||
id: run.id,
|
||||
agentId: run.agentId,
|
||||
startedAt: run.startedAt,
|
||||
endedAt: run.endedAt ?? undefined,
|
||||
status: run.status,
|
||||
taskId: (auditTaskId ?? undefined) as string | undefined,
|
||||
},
|
||||
auditByDomain,
|
||||
counts: {
|
||||
auditEvents: normalizedAuditEvents.length,
|
||||
logEntries: includeLogs && auditTaskId ? (await scopedStore.getAgentLogsByTimeRange(
|
||||
auditTaskId,
|
||||
run.startedAt,
|
||||
run.endedAt,
|
||||
)).length : 0,
|
||||
},
|
||||
timeline: timelineEntries,
|
||||
};
|
||||
|
||||
res.json(response);
|
||||
} catch (err: any) {
|
||||
if (err instanceof ApiError) {
|
||||
throw err;
|
||||
}
|
||||
if (err.message?.includes("not found")) {
|
||||
throw notFound(err.message);
|
||||
} else {
|
||||
rethrowAsApiError(err);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* GET /api/agents/:id/chain-of-command
|
||||
* Fetch agent reporting chain from self to top-most manager.
|
||||
|
||||
Reference in New Issue
Block a user