bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(core): activity events for recovery, metadata corruption, agent-report (#1692) 

* feat(core): activity events for recovery, metadata corruption, agent-report

Wires activity events into three forensic-critical paths so RCA can
reconstruct what happened after the fact:

1. recovery subsystem
   - recovery.session_failed (MUST) per failed session in runRecovery
   - recovery.action_failed (SHOULD) on recoverSession outer catch
   Adds "recovery" to ActivityEventSource so `ao events list --source
   recovery` reconstructs an `ao recover` invocation timeline.

2. metadata corruption
   - metadata.corrupt_detected (MUST) when mutateMetadata renames a
     corrupt session-metadata file to .corrupt-{ts}. Includes
     data.renamedTo and a 200-char data.contentSample (B11) for forensic
     recovery. Previously only console.warn — silent overwrites had no
     queryable signal.

3. agent-report apply path
   - api.agent_report.transition_rejected (SHOULD)
   - api.agent_report.session_not_found (COULD)

Per B22, recovery events fire per session, not per probe step. Per B11,
metadata.corrupt_detected truncates contentSample to 200 chars (full
file would exceed the 16KB sanitizer cap).

Closes #1660

* fix(core): align forensic activity event metadata

* fix(core): cover single-session recovery failures

* fix(core): improve corrupt metadata event attribution

* fix(cli): expose activity event source filter

---------

Co-authored-by: whoisasx <adil.business4064@gmail.com>
This commit is contained in:
Dhruv Sharma 2026-05-18 17:27:36 +05:30 committed by GitHub
parent befd910eeb
commit fcedb25031
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 676 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
.changeset/issue-1660-recovery-metadata-events.md Normal file
View File

@ -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.

87
packages/cli/__tests__/commands/events.test.ts Normal file
View File

@ -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<typeof vi.spyOn>;
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",
}),
);
});
});

21
packages/cli/src/commands/events.ts
View File

@ -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 <id>", "Filter by project ID")
.option("-s, --session <id>", "Filter by session ID")
.option("-t, --type <kind>", "Filter by event kind (e.g. session.spawned, lifecycle.transition)")
.option(
"-t, --type <kind>",
"Filter by event kind (e.g. session.spawned, lifecycle.transition)",
)
.option("--kind <kind>", "Alias for --type")
.option("--source <source>", "Filter by event source (e.g. lifecycle, recovery, api)")
.option("--log-level <level>", "Filter by log level (debug, info, warn, error)")
.option("--since <duration>", "Show events from last N minutes/hours/days (e.g. 30m, 2h, 1d)")
.option("-n, --limit <n>", "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,

28
packages/core/src/__tests__/agent-report.test.ts
View File

@ -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

139
packages/core/src/__tests__/metadata.test.ts
View File

@ -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<typeof NodeFs>();
return {
...actual,
renameSync: vi.fn((...args: Parameters<typeof actual.renameSync>) =>
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<string, unknown>)["contentSample"] as string;
expect(sample.length).toBe(200);
expect((call![0].data as Record<string, unknown>)["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", () => {

269
packages/core/src/__tests__/recovery-manager.test.ts Normal file
View File

@ -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);
});
});

11
packages/core/src/activity-events.ts
View File

@ -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";

46
packages/core/src/agent-report.ts
View File

@ -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}`);

37
packages/core/src/metadata.ts
View File

@ -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<string, string>): Record<string, string> {
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<string, string>) => Record<string, string>,
options: { createIfMissing?: boolean } = {},
options: MutateMetadataOptions = {},
): Record<string, string> | 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) {

17
packages/core/src/recovery/actions.ts
View File

@ -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,
};
}
}

34
packages/core/src/recovery/manager.ts
View File

@ -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<Reco
break;
}
} else {
const errorMessage = recordRecoverySessionFailed(assessment, result);
report.errors.push({
sessionId: assessment.sessionId,
error: result.error || "Unknown error",
error: errorMessage,
});
}
@ -133,6 +135,32 @@ export async function runRecovery(options: RecoveryManagerOptions): Promise<Reco
};
}
function recordRecoverySessionFailed(
assessment: RecoveryAssessment,
result: RecoveryResult,
): string {
const errorMessage = result.error || "Unknown error";
// Forensic event so RCA can answer: did `ao recover` actually fix
// anything? Which sessions did it skip? B22: one event per failed
// session, not per probe step.
recordActivityEvent({
projectId: assessment.projectId,
sessionId: assessment.sessionId,
source: "recovery",
kind: "recovery.session_failed",
level: "error",
summary: `Recovery failed for session ${assessment.sessionId}: ${errorMessage}`,
data: {
action: result.action,
classification: assessment.classification,
recoveryRule: assessment.recoveryRule,
metadataStatus: assessment.metadataStatus,
errorMessage,
},
});
return errorMessage;
}
function mapActionToLogAction(
action: string,
success: boolean,
@ -184,6 +212,10 @@ export async function recoverSessionById(
return result;
}
if (!result.success) {
recordRecoverySessionFailed(assessment, result);
}
const logAction = mapActionToLogAction(result.action, result.success);
writeRecoveryLog(
recoveryConfig.logPath,