diff --git a/.changeset/issue-1660-recovery-metadata-events.md b/.changeset/issue-1660-recovery-metadata-events.md new file mode 100644 index 000000000..4ad89f8b0 --- /dev/null +++ b/.changeset/issue-1660-recovery-metadata-events.md @@ -0,0 +1,6 @@ +--- +"@aoagents/ao-cli": patch +"@aoagents/ao-core": minor +--- + +Wire activity events for the recovery subsystem, metadata-corruption detection, and agent-report apply path. New event kinds: `recovery.session_failed`, `recovery.action_failed`, `metadata.corrupt_detected`, `api.agent_report.session_not_found`, `api.agent_report.transition_rejected`. Adds `"recovery"` to the `ActivityEventSource` union. Lets RCA reconstruct `ao recover` invocations, find every silent metadata overwrite, and audit rejected agent transitions. Adds `ao events list --source` and `--kind` so these forensic event queries are available from the CLI. diff --git a/packages/cli/__tests__/commands/events.test.ts b/packages/cli/__tests__/commands/events.test.ts new file mode 100644 index 000000000..e96633234 --- /dev/null +++ b/packages/cli/__tests__/commands/events.test.ts @@ -0,0 +1,87 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { Command } from "commander"; + +const { mockQueryActivityEvents, mockSearchActivityEvents, mockGetActivityEventStats } = vi.hoisted( + () => ({ + mockQueryActivityEvents: vi.fn(), + mockSearchActivityEvents: vi.fn(), + mockGetActivityEventStats: vi.fn(), + }), +); + +vi.mock("@aoagents/ao-core", () => ({ + queryActivityEvents: (...args: unknown[]) => mockQueryActivityEvents(...args), + searchActivityEvents: (...args: unknown[]) => mockSearchActivityEvents(...args), + getActivityEventStats: (...args: unknown[]) => mockGetActivityEventStats(...args), + droppedEventCount: () => 0, + isActivityEventsFtsEnabled: () => true, +})); + +import { registerEvents } from "../../src/commands/events.js"; + +describe("events command", () => { + let program: Command; + let consoleLogSpy: ReturnType; + + beforeEach(() => { + program = new Command(); + program.exitOverride(); + registerEvents(program); + + consoleLogSpy = vi.spyOn(console, "log").mockImplementation(() => {}); + mockQueryActivityEvents.mockReset(); + mockSearchActivityEvents.mockReset(); + mockGetActivityEventStats.mockReset(); + mockQueryActivityEvents.mockReturnValue([]); + }); + + afterEach(() => { + consoleLogSpy.mockRestore(); + }); + + it("filters list output by source and --kind alias", async () => { + await program.parseAsync([ + "node", + "test", + "events", + "list", + "--source", + "recovery", + "--kind", + "metadata.corrupt_detected", + "--limit", + "1", + "--json", + ]); + + expect(mockQueryActivityEvents).toHaveBeenCalledWith( + expect.objectContaining({ + source: "recovery", + kind: "metadata.corrupt_detected", + limit: 1, + }), + ); + expect(consoleLogSpy).toHaveBeenCalledWith(expect.stringContaining('"source": "recovery"')); + expect(consoleLogSpy).toHaveBeenCalledWith( + expect.stringContaining('"kind": "metadata.corrupt_detected"'), + ); + }); + + it("keeps --type as the existing event-kind filter", async () => { + await program.parseAsync([ + "node", + "test", + "events", + "list", + "--type", + "recovery.session_failed", + "--json", + ]); + + expect(mockQueryActivityEvents).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "recovery.session_failed", + }), + ); + }); +}); diff --git a/packages/cli/src/commands/events.ts b/packages/cli/src/commands/events.ts index 87430d2a3..bbb72a7bd 100644 --- a/packages/cli/src/commands/events.ts +++ b/packages/cli/src/commands/events.ts @@ -9,6 +9,7 @@ import { type ActivityEvent, type ActivityEventLevel, type ActivityEventKind, + type ActivityEventSource, } from "@aoagents/ao-core"; interface JsonEnvelope { @@ -80,7 +81,12 @@ export function registerEvents(program: Command): void { .description("List recent activity events") .option("-p, --project ", "Filter by project ID") .option("-s, --session ", "Filter by session ID") - .option("-t, --type ", "Filter by event kind (e.g. session.spawned, lifecycle.transition)") + .option( + "-t, --type ", + "Filter by event kind (e.g. session.spawned, lifecycle.transition)", + ) + .option("--kind ", "Alias for --type") + .option("--source ", "Filter by event source (e.g. lifecycle, recovery, api)") .option("--log-level ", "Filter by log level (debug, info, warn, error)") .option("--since ", "Show events from last N minutes/hours/days (e.g. 30m, 2h, 1d)") .option("-n, --limit ", "Max results", "50") @@ -91,15 +97,21 @@ export function registerEvents(program: Command): void { if (sinceRaw) { since = parseSinceDuration(sinceRaw); if (!since) { - console.error(chalk.yellow(`Warning: unrecognised --since format "${sinceRaw}" (use e.g. 30m, 2h, 1d). No time filter applied.`)); + console.error( + chalk.yellow( + `Warning: unrecognised --since format "${sinceRaw}" (use e.g. 30m, 2h, 1d). No time filter applied.`, + ), + ); } } const limit = parseInt(opts["limit"] ?? "50", 10); + const kind = opts["type"] ?? opts["kind"]; const results = queryActivityEvents({ projectId: opts["project"], sessionId: opts["session"], - kind: opts["type"] as ActivityEventKind, + kind: kind as ActivityEventKind, + source: opts["source"] as ActivityEventSource, level: opts["logLevel"] as ActivityEventLevel, since, limit, @@ -111,7 +123,8 @@ export function registerEvents(program: Command): void { query: { projectId: opts["project"] ?? null, sessionId: opts["session"] ?? null, - kind: opts["type"] ?? null, + kind: kind ?? null, + source: opts["source"] ?? null, level: opts["logLevel"] ?? null, since: sinceRaw ?? null, limit, diff --git a/packages/core/src/__tests__/agent-report.test.ts b/packages/core/src/__tests__/agent-report.test.ts index 7e3fddce1..f49483b9d 100644 --- a/packages/core/src/__tests__/agent-report.test.ts +++ b/packages/core/src/__tests__/agent-report.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; import { mkdirSync, readFileSync, rmSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; @@ -18,17 +18,25 @@ import { } from "../agent-report.js"; import { writeMetadata, writeCanonicalLifecycle, readMetadataRaw } from "../metadata.js"; import { createInitialCanonicalLifecycle } from "../lifecycle-state.js"; +import { recordActivityEvent } from "../activity-events.js"; import type { CanonicalSessionLifecycle } from "../types.js"; +vi.mock("../activity-events.js", () => ({ + recordActivityEvent: vi.fn(), +})); + let dataDir: string; +let tempRoot: string; beforeEach(() => { - dataDir = join(tmpdir(), `ao-test-agent-report-${randomUUID()}`); + tempRoot = join(tmpdir(), `ao-test-agent-report-${randomUUID()}`); + dataDir = join(tempRoot, "projects", "project-alpha", "sessions"); mkdirSync(dataDir, { recursive: true }); + vi.mocked(recordActivityEvent).mockClear(); }); afterEach(() => { - rmSync(dataDir, { recursive: true, force: true }); + rmSync(tempRoot, { recursive: true, force: true }); }); function seedWorkerSession( @@ -433,6 +441,13 @@ describe("applyAgentReport", () => { sessionState: "terminated", }, }); + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + projectId: "project-alpha", + sessionId, + kind: "api.agent_report.transition_rejected", + }), + ); }); it("throws when the session does not exist", () => { @@ -442,6 +457,13 @@ describe("applyAgentReport", () => { now: new Date(), }), ).toThrow(/not found/); + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + projectId: "project-alpha", + sessionId: "missing-session", + kind: "api.agent_report.session_not_found", + }), + ); }); // 260 atomic-write cycles are slow on Windows (rename + AV scan); bump the diff --git a/packages/core/src/__tests__/metadata.test.ts b/packages/core/src/__tests__/metadata.test.ts index 03ac1c16b..47c66b0e0 100644 --- a/packages/core/src/__tests__/metadata.test.ts +++ b/packages/core/src/__tests__/metadata.test.ts @@ -1,5 +1,6 @@ -import { describe, it, expect, beforeEach, afterEach } from "vitest"; -import { mkdirSync, rmSync, readFileSync, existsSync, writeFileSync, readdirSync } from "node:fs"; +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { mkdirSync, rmSync, readFileSync, existsSync, writeFileSync, readdirSync, renameSync } from "node:fs"; +import type * as NodeFs from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { randomUUID } from "node:crypto"; @@ -13,12 +14,29 @@ import { deleteMetadata, listMetadata, } from "../metadata.js"; +import { recordActivityEvent } from "../activity-events.js"; + +vi.mock("node:fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + renameSync: vi.fn((...args: Parameters) => + actual.renameSync(...args), + ), + }; +}); + +vi.mock("../activity-events.js", () => ({ + recordActivityEvent: vi.fn(), +})); let dataDir: string; beforeEach(() => { dataDir = join(tmpdir(), `ao-test-metadata-${randomUUID()}`); mkdirSync(dataDir, { recursive: true }); + vi.mocked(recordActivityEvent).mockClear(); + vi.mocked(renameSync).mockClear(); }); afterEach(() => { @@ -390,6 +408,123 @@ describe("mutateMetadata corrupt-file handling", () => { const corruptCopies = readdirSync(dataDir).filter((f) => f.includes(".corrupt-")); expect(corruptCopies).toHaveLength(0); }); + + it("emits metadata.corrupt_detected when JSON parse fails and file is renamed", () => { + const sessionPath = join(dataDir, "ao-3.json"); + writeFileSync(sessionPath, "{ broken json", "utf-8"); + + const result = mutateMetadata( + dataDir, + "ao-3", + (existing) => ({ ...existing, branch: "feat/x" }), + { createIfMissing: true }, + ); + + expect(result).not.toBeNull(); + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: "ao-3", + source: "session-manager", + kind: "metadata.corrupt_detected", + level: "error", + summary: expect.stringContaining("renamed to"), + data: expect.objectContaining({ + renamedTo: expect.stringContaining(`${sessionPath}.corrupt-`), + renameSucceeded: true, + contentSample: "{ broken json", + path: sessionPath, + }), + }), + ); + }); + + it("emits a rename-failed summary when corrupt metadata cannot be renamed", () => { + const sessionPath = join(dataDir, "ao-rename-failed.json"); + writeFileSync(sessionPath, "{ broken json", "utf-8"); + vi.mocked(renameSync).mockImplementationOnce(() => { + throw new Error("rename denied"); + }); + + const result = mutateMetadata( + dataDir, + "ao-rename-failed", + (existing) => ({ ...existing, branch: "feat/x" }), + { createIfMissing: true }, + ); + + expect(result).not.toBeNull(); + const call = vi + .mocked(recordActivityEvent) + .mock.calls.find((c) => c[0].kind === "metadata.corrupt_detected"); + expect(call).toBeDefined(); + expect(call![0]).toMatchObject({ + sessionId: "ao-rename-failed", + summary: expect.stringContaining("failed to rename"), + data: expect.objectContaining({ + renamedTo: null, + renameSucceeded: false, + path: sessionPath, + }), + }); + expect(call![0].summary).not.toContain("renamed to"); + }); + + it("uses the provided source for metadata.corrupt_detected", () => { + const sessionPath = join(dataDir, "ao-api-source.json"); + writeFileSync(sessionPath, "{ broken json", "utf-8"); + + mutateMetadata( + dataDir, + "ao-api-source", + (existing) => ({ ...existing, branch: "feat/api" }), + { createIfMissing: true, activityEventSource: "api" }, + ); + + expect(recordActivityEvent).toHaveBeenCalledWith( + expect.objectContaining({ + sessionId: "ao-api-source", + source: "api", + kind: "metadata.corrupt_detected", + }), + ); + }); + + it("truncates contentSample to 200 chars in metadata.corrupt_detected", () => { + const sessionPath = join(dataDir, "ao-4.json"); + // 250 char garbage payload — sanitizer cap is 16KB but invariant B11 caps + // forensic sample at 200 chars. + const huge = "x".repeat(250); + writeFileSync(sessionPath, huge, "utf-8"); + + mutateMetadata( + dataDir, + "ao-4", + (existing) => ({ ...existing, branch: "feat/y" }), + { createIfMissing: true }, + ); + + const call = vi + .mocked(recordActivityEvent) + .mock.calls.find((c) => c[0].kind === "metadata.corrupt_detected"); + expect(call).toBeDefined(); + const sample = (call![0].data as Record)["contentSample"] as string; + expect(sample.length).toBe(200); + expect((call![0].data as Record)["contentLength"]).toBe(250); + }); + + it("does not emit metadata.corrupt_detected for healthy JSON", () => { + writeMetadata(dataDir, "ao-5", { + worktree: "/tmp/w", + branch: "main", + status: "working", + }); + mutateMetadata(dataDir, "ao-5", (existing) => ({ ...existing, summary: "hi" })); + + const corruptCalls = vi + .mocked(recordActivityEvent) + .mock.calls.filter((c) => c[0].kind === "metadata.corrupt_detected"); + expect(corruptCalls).toHaveLength(0); + }); }); describe("readCanonicalLifecycle", () => { diff --git a/packages/core/src/__tests__/recovery-manager.test.ts b/packages/core/src/__tests__/recovery-manager.test.ts new file mode 100644 index 000000000..80647dc29 --- /dev/null +++ b/packages/core/src/__tests__/recovery-manager.test.ts @@ -0,0 +1,269 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { randomUUID } from "node:crypto"; +import { recoverSessionById, runRecovery } from "../recovery/manager.js"; +import { recordActivityEvent } from "../activity-events.js"; +import { getProjectDir, getProjectSessionsDir } from "../paths.js"; +import { + DEFAULT_RECOVERY_CONFIG, + type RecoveryAssessment, + type RecoveryResult, +} from "../recovery/types.js"; +import * as actionsModule from "../recovery/actions.js"; +import * as validatorModule from "../recovery/validator.js"; +import type { OrchestratorConfig, PluginRegistry } from "../types.js"; + +vi.mock("../activity-events.js", () => ({ + recordActivityEvent: vi.fn(), +})); + +const PROJECT_ID = "app"; + +function makeConfig(rootDir: string): OrchestratorConfig { + return { + configPath: join(rootDir, "agent-orchestrator.yaml"), + port: 3000, + readyThresholdMs: 300_000, + power: { preventIdleSleep: false }, + defaults: { + runtime: "tmux", + agent: "claude-code", + workspace: "worktree", + notifiers: ["desktop"], + }, + projects: { + app: { + name: "app", + repo: "org/repo", + path: join(rootDir, "project"), + defaultBranch: "main", + sessionPrefix: "app", + }, + }, + notifiers: {}, + notificationRouting: { + urgent: ["desktop"], + action: ["desktop"], + warning: ["desktop"], + info: ["desktop"], + }, + reactions: {}, + }; +} + +function makeRegistry(): PluginRegistry { + return { + register: vi.fn(), + get: vi.fn().mockReturnValue(null), + list: vi.fn().mockReturnValue([]), + loadBuiltins: vi.fn().mockResolvedValue(undefined), + loadFromConfig: vi.fn().mockResolvedValue(undefined), + }; +} + +function makeAssessment(sessionId: string): RecoveryAssessment { + return { + sessionId, + projectId: PROJECT_ID, + classification: "live", + action: "recover", + reason: "needs recovery", + runtimeProbeSucceeded: true, + processProbeSucceeded: true, + signalDisagreement: false, + recoveryRule: "auto", + runtimeAlive: true, + runtimeHandle: null, + workspaceExists: true, + workspacePath: "/tmp/worktree", + agentProcessRunning: true, + agentActivity: "active", + metadataValid: true, + metadataStatus: "working", + rawMetadata: { project: PROJECT_ID, status: "working" }, + }; +} + +describe("runRecovery activity events", () => { + let rootDir: string; + let previousHome: string | undefined; + + beforeEach(() => { + rootDir = join(tmpdir(), `ao-recovery-events-${randomUUID()}`); + mkdirSync(rootDir, { recursive: true }); + mkdirSync(join(rootDir, "project"), { recursive: true }); + writeFileSync(join(rootDir, "agent-orchestrator.yaml"), "projects: {}\n", "utf-8"); + previousHome = process.env["HOME"]; + process.env["HOME"] = rootDir; + + const sessionsDir = getProjectSessionsDir(PROJECT_ID); + mkdirSync(sessionsDir, { recursive: true }); + writeFileSync( + join(sessionsDir, "app-1.json"), + JSON.stringify({ project: PROJECT_ID, status: "working" }) + "\n", + "utf-8", + ); + writeFileSync( + join(sessionsDir, "app-2.json"), + JSON.stringify({ project: PROJECT_ID, status: "working" }) + "\n", + "utf-8", + ); + + vi.mocked(recordActivityEvent).mockClear(); + vi.restoreAllMocks(); + }); + + afterEach(() => { + if (previousHome === undefined) { + delete process.env["HOME"]; + } else { + process.env["HOME"] = previousHome; + } + if (rootDir) { + const projectBaseDir = getProjectDir(PROJECT_ID); + if (existsSync(projectBaseDir)) { + rmSync(projectBaseDir, { recursive: true, force: true }); + } + rmSync(rootDir, { recursive: true, force: true }); + } + vi.restoreAllMocks(); + }); + + it("emits recovery.session_failed for each session that recovery couldn't fix", async () => { + vi.spyOn(validatorModule, "validateSession").mockImplementation(async (scanned) => + makeAssessment(scanned.sessionId), + ); + + const successResult: RecoveryResult = { + success: true, + sessionId: "app-1", + action: "recover", + }; + const failedResult: RecoveryResult = { + success: false, + sessionId: "app-2", + action: "recover", + error: "worktree missing", + }; + + vi.spyOn(actionsModule, "executeAction") + .mockResolvedValueOnce(successResult) + .mockResolvedValueOnce(failedResult); + + const config = makeConfig(rootDir); + const registry = makeRegistry(); + + const { report } = await runRecovery({ + config, + registry, + recoveryConfig: { + ...DEFAULT_RECOVERY_CONFIG, + logPath: join(rootDir, "recovery.log"), + }, + }); + + expect(report.errors).toHaveLength(1); + expect(report.errors[0]?.sessionId).toBe("app-2"); + + const emitCalls = vi + .mocked(recordActivityEvent) + .mock.calls.map((c) => c[0]) + .filter((e) => e.kind === "recovery.session_failed"); + + expect(emitCalls).toHaveLength(1); + expect(emitCalls[0]).toEqual( + expect.objectContaining({ + sessionId: "app-2", + projectId: PROJECT_ID, + source: "recovery", + kind: "recovery.session_failed", + level: "error", + data: expect.objectContaining({ + action: "recover", + errorMessage: "worktree missing", + }), + }), + ); + }); + + it("emits recovery.session_failed when single-session recovery fails", async () => { + vi.spyOn(validatorModule, "validateSession").mockImplementation(async (scanned) => + makeAssessment(scanned.sessionId), + ); + + const failedResult: RecoveryResult = { + success: false, + sessionId: "app-2", + action: "recover", + error: "agent process missing", + }; + + vi.spyOn(actionsModule, "executeAction").mockResolvedValueOnce(failedResult); + + const config = makeConfig(rootDir); + const registry = makeRegistry(); + + const result = await recoverSessionById("app-2", { + config, + registry, + recoveryConfig: { + ...DEFAULT_RECOVERY_CONFIG, + logPath: join(rootDir, "recovery.log"), + }, + }); + + expect(result).toEqual(failedResult); + + const emitCalls = vi + .mocked(recordActivityEvent) + .mock.calls.map((c) => c[0]) + .filter((e) => e.kind === "recovery.session_failed"); + + expect(emitCalls).toHaveLength(1); + expect(emitCalls[0]).toEqual( + expect.objectContaining({ + sessionId: "app-2", + projectId: PROJECT_ID, + source: "recovery", + kind: "recovery.session_failed", + level: "error", + data: expect.objectContaining({ + action: "recover", + errorMessage: "agent process missing", + }), + }), + ); + }); + + it("does not emit recovery.session_failed when every session recovers cleanly", async () => { + vi.spyOn(validatorModule, "validateSession").mockImplementation(async (scanned) => + makeAssessment(scanned.sessionId), + ); + + vi.spyOn(actionsModule, "executeAction").mockImplementation(async (assessment) => ({ + success: true, + sessionId: assessment.sessionId, + action: "recover", + })); + + const config = makeConfig(rootDir); + const registry = makeRegistry(); + + await runRecovery({ + config, + registry, + recoveryConfig: { + ...DEFAULT_RECOVERY_CONFIG, + logPath: join(rootDir, "recovery.log"), + }, + }); + + const failedEmits = vi + .mocked(recordActivityEvent) + .mock.calls.map((c) => c[0]) + .filter((e) => e.kind === "recovery.session_failed"); + expect(failedEmits).toHaveLength(0); + }); +}); diff --git a/packages/core/src/activity-events.ts b/packages/core/src/activity-events.ts index 0c1072e88..789e2616d 100644 --- a/packages/core/src/activity-events.ts +++ b/packages/core/src/activity-events.ts @@ -24,7 +24,8 @@ export type ActivityEventSource = | "workspace" | "notifier" | "reaction" - | "report-watcher"; + | "report-watcher" + | "recovery"; export type ActivityEventKind = | "session.spawn_started" @@ -69,7 +70,13 @@ export type ActivityEventKind = | "lifecycle.poll_failed" | "detecting.escalated" // Report watcher - | "report_watcher.triggered"; + | "report_watcher.triggered" + // Recovery/forensic instrumentation + | "recovery.session_failed" + | "recovery.action_failed" + | "metadata.corrupt_detected" + | "api.agent_report.session_not_found" + | "api.agent_report.transition_rejected"; export type ActivityEventLevel = "debug" | "info" | "warn" | "error"; diff --git a/packages/core/src/agent-report.ts b/packages/core/src/agent-report.ts index 07e380e30..13b1a187a 100644 --- a/packages/core/src/agent-report.ts +++ b/packages/core/src/agent-report.ts @@ -18,7 +18,7 @@ import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; import { readFile } from "node:fs/promises"; -import { join } from "node:path"; +import { basename, dirname, join } from "node:path"; import type { CanonicalSessionLifecycle, CanonicalSessionReason, @@ -26,6 +26,7 @@ import type { SessionId, SessionStatus, } from "./types.js"; +import { recordActivityEvent } from "./activity-events.js"; import { mutateMetadata, readMetadataRaw } from "./metadata.js"; import { buildLifecycleMetadataPatch, cloneLifecycle, deriveLegacyStatus, parseCanonicalLifecycle } from "./lifecycle-state.js"; import { parsePrFromUrl } from "./utils/pr.js"; @@ -254,6 +255,11 @@ function buildAuditDir(dataDir: string): string { return join(dataDir, ".agent-report-audit"); } +function inferProjectIdFromDataDir(dataDir: string): string | undefined { + // dataDir is `.../projects/{projectId}/sessions`; recover projectId for filtering. + return basename(dirname(dataDir)) || undefined; +} + const AGENT_REPORT_AUDIT_MAX_BYTES = 256 * 1024; const AGENT_REPORT_AUDIT_MAX_ENTRIES = 200; @@ -383,8 +389,22 @@ export function applyAgentReport( sessionId: SessionId, input: ApplyAgentReportInput, ): ApplyAgentReportResult { + const projectId = inferProjectIdFromDataDir(dataDir); const raw = readMetadataRaw(dataDir, sessionId); if (!raw) { + recordActivityEvent({ + projectId, + sessionId, + source: "api", + kind: "api.agent_report.session_not_found", + level: "warn", + summary: `applyAgentReport: session not found: ${sessionId}`, + data: { + reportState: input.state, + actor: input.actor, + source: input.source, + }, + }); throw new Error(`Session not found: ${sessionId}`); } @@ -439,6 +459,7 @@ export function applyAgentReport( before = buildAuditSnapshot(current, previousLegacyStatus); const validation = validateAgentReportTransition(current, input.state); if (!validation.ok) { + const rejectionReason = validation.reason ?? "transition rejected"; appendAgentReportAuditEntry(dataDir, sessionId, { timestamp: now, actor, @@ -449,11 +470,28 @@ export function applyAgentReport( prUrl: trimmedPrUrl, prIsDraft, accepted: false, - rejectionReason: validation.reason ?? "transition rejected", + rejectionReason, before, after: before, }); - throw new Error(validation.reason ?? "transition rejected"); + recordActivityEvent({ + projectId, + sessionId, + source: "api", + kind: "api.agent_report.transition_rejected", + level: "warn", + summary: `applyAgentReport rejected ${input.state} for ${sessionId}: ${rejectionReason}`, + data: { + reportState: input.state, + rejectionReason, + actor, + reportSource: source, + fromState: current.session.state, + fromReason: current.session.reason, + legacyStatus: previousLegacyStatus, + }, + }); + throw new Error(rejectionReason); } const mapped = mapAgentReportToLifecycle(input.state); previousState = current.session.state; @@ -512,7 +550,7 @@ export function applyAgentReport( } } return next; - }); + }, { activityEventSource: "api" }); if (!nextMetadata || !before || !previousState || !nextState || !legacyStatus) { throw new Error(`Failed to apply agent report for session ${sessionId}`); diff --git a/packages/core/src/metadata.ts b/packages/core/src/metadata.ts index a4f3a19c5..451c127b5 100644 --- a/packages/core/src/metadata.ts +++ b/packages/core/src/metadata.ts @@ -22,9 +22,10 @@ import { closeSync, constants, } from "node:fs"; -import { join, dirname } from "node:path"; +import { basename, join, dirname } from "node:path"; import type { CanonicalSessionLifecycle, RuntimeHandle, SessionId, SessionMetadata } from "./types.js"; import { atomicWriteFileSync } from "./atomic-write.js"; +import { recordActivityEvent, type ActivityEventSource } from "./activity-events.js"; import { buildLifecycleMetadataPatch, cloneLifecycle, @@ -330,6 +331,11 @@ export function applyMetadataUpdates( return next; } +interface MutateMetadataOptions { + createIfMissing?: boolean; + activityEventSource?: ActivityEventSource; +} + function normalizeMetadataRecord(data: Record): Record { return Object.fromEntries( Object.entries(data).filter(([, value]) => value !== undefined && value !== ""), @@ -340,7 +346,7 @@ export function mutateMetadata( dataDir: string, sessionId: SessionId, updater: (existing: Record) => Record, - options: { createIfMissing?: boolean } = {}, + options: MutateMetadataOptions = {}, ): Record | null { const path = metadataPath(dataDir, sessionId); const lockPath = `${path}.lock`; @@ -368,8 +374,10 @@ export function mutateMetadata( // that anything was wrong — the file just becomes "not // corrupt anymore — and missing fields". const corruptPath = `${path}.corrupt-${Date.now()}`; + let renamed = false; try { renameSync(path, corruptPath); + renamed = true; // eslint-disable-next-line no-console console.warn( `[metadata] corrupt JSON at ${path}; preserved as ${corruptPath} before rewriting`, @@ -377,6 +385,31 @@ export function mutateMetadata( } catch { // best effort — proceed even if the rename fails (e.g. EACCES) } + // Forensic activity event so RCA can find every silent overwrite. + // Truncate the bad-JSON sample to 200 chars (B11 invariant — full file + // could be 16KB+ and would be dropped by the sanitizer cap). + const contentSample = + content.length > 200 ? content.slice(0, 200) : content; + // dataDir is `.../projects/{projectId}/sessions`; recover projectId for filtering. + const inferredProjectId = basename(dirname(dataDir)); + const summary = renamed + ? `Corrupt metadata for session ${sessionId} renamed to ${basename(corruptPath)}` + : `Corrupt metadata detected for session ${sessionId}; failed to rename forensic copy before rewrite`; + recordActivityEvent({ + projectId: inferredProjectId || undefined, + sessionId, + source: options.activityEventSource ?? "session-manager", + kind: "metadata.corrupt_detected", + level: "error", + summary, + data: { + path, + renamedTo: renamed ? corruptPath : null, + renameSucceeded: renamed, + contentSample, + contentLength: content.length, + }, + }); } } } else if (!options.createIfMissing) { diff --git a/packages/core/src/recovery/actions.ts b/packages/core/src/recovery/actions.ts index 43a9eec87..f7b57980a 100644 --- a/packages/core/src/recovery/actions.ts +++ b/packages/core/src/recovery/actions.ts @@ -5,6 +5,7 @@ import type { Runtime, Workspace, } from "../types.js"; +import { recordActivityEvent } from "../activity-events.js"; import { updateMetadata } from "../metadata.js"; import { getProjectSessionsDir } from "../paths.js"; import { validateStatus } from "../utils/validation.js"; @@ -146,11 +147,25 @@ export async function recoverSession( session, }; } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + recordActivityEvent({ + projectId, + sessionId, + source: "recovery", + kind: "recovery.action_failed", + level: "error", + summary: `recoverSession threw for ${sessionId}: ${errorMessage}`, + data: { + action: "recover", + recoveryCount, + errorMessage, + }, + }); return { success: false, sessionId, action: "recover", - error: error instanceof Error ? error.message : String(error), + error: errorMessage, }; } } diff --git a/packages/core/src/recovery/manager.ts b/packages/core/src/recovery/manager.ts index 1c90e007f..330922e7f 100644 --- a/packages/core/src/recovery/manager.ts +++ b/packages/core/src/recovery/manager.ts @@ -1,4 +1,5 @@ import type { OrchestratorConfig, PluginRegistry, Session } from "../types.js"; +import { recordActivityEvent } from "../activity-events.js"; import { scanAllSessions, getRecoveryLogPath } from "./scanner.js"; import { validateSession } from "./validator.js"; import { executeAction } from "./actions.js"; @@ -106,9 +107,10 @@ export async function runRecovery(options: RecoveryManagerOptions): Promise