diff --git a/packages/core/src/__tests__/lifecycle-manager.test.ts b/packages/core/src/__tests__/lifecycle-manager.test.ts index d852fbc07..5869af993 100644 --- a/packages/core/src/__tests__/lifecycle-manager.test.ts +++ b/packages/core/src/__tests__/lifecycle-manager.test.ts @@ -145,15 +145,27 @@ describe("check (single session)", () => { }); it("does not mirror lifecycle transition observability logs to stderr during polling", async () => { + const originalAoObservabilityStderr = process.env["AO_OBSERVABILITY_STDERR"]; + delete process.env["AO_OBSERVABILITY_STDERR"]; + const stderrSpy = vi.spyOn(process.stderr, "write").mockReturnValue(true); - const lm = setupCheck("app-1", { - session: makeSession({ status: "spawning" }), - }); - await lm.check("app-1"); + try { + const lm = setupCheck("app-1", { + session: makeSession({ status: "spawning" }), + }); - expect(stderrSpy).not.toHaveBeenCalled(); - stderrSpy.mockRestore(); + await lm.check("app-1"); + + expect(stderrSpy).not.toHaveBeenCalled(); + } finally { + stderrSpy.mockRestore(); + if (originalAoObservabilityStderr === undefined) { + delete process.env["AO_OBSERVABILITY_STDERR"]; + } else { + process.env["AO_OBSERVABILITY_STDERR"] = originalAoObservabilityStderr; + } + } }); it("clears stale lifecycle compatibility metadata in memory and on disk", async () => { diff --git a/packages/core/src/__tests__/observability.test.ts b/packages/core/src/__tests__/observability.test.ts index 30e8e4687..71b67b3c9 100644 --- a/packages/core/src/__tests__/observability.test.ts +++ b/packages/core/src/__tests__/observability.test.ts @@ -104,46 +104,58 @@ describe("observability snapshot", () => { }); it("writes observability diagnostics to audit files without mirroring to stderr by default", () => { + const originalObservabilityStderr = process.env["AO_OBSERVABILITY_STDERR"]; + delete process.env["AO_OBSERVABILITY_STDERR"]; + const stderrSpy = vi.spyOn(process.stderr, "write").mockReturnValue(true); - const observer = createProjectObserver(config, "session-manager"); - observer.recordOperation({ - metric: "spawn", - operation: "session.spawn", - outcome: "success", - correlationId: "corr-1", - projectId: "my-app", - sessionId: "app-1", - level: "info", - }); + try { + const observer = createProjectObserver(config, "session-manager"); - observer.recordDiagnostic({ - operation: "batch_enrichment.log", - correlationId: "corr-2", - projectId: "my-app", - message: "GraphQL batch returned cached result", - level: "info", - data: { plugin: "github" }, - }); + observer.recordOperation({ + metric: "spawn", + operation: "session.spawn", + outcome: "success", + correlationId: "corr-1", + projectId: "my-app", + sessionId: "app-1", + level: "warn", + }); - observer.setHealth({ - surface: "lifecycle.worker", - status: "ok", - projectId: "my-app", - correlationId: "corr-3", - details: { projectId: "my-app" }, - }); + observer.recordDiagnostic?.({ + operation: "batch_enrichment.log", + correlationId: "corr-2", + projectId: "my-app", + message: "GraphQL batch returned cached result", + level: "warn", + data: { plugin: "github" }, + }); - const auditDir = join(getObservabilityBaseDir(config.configPath), "processes"); - const auditFiles = readdirSync(auditDir).filter((fileName) => fileName.endsWith(".ndjson")); - expect(auditFiles.length).toBeGreaterThan(0); + observer.setHealth({ + surface: "lifecycle.worker", + status: "warn", + projectId: "my-app", + correlationId: "corr-3", + details: { projectId: "my-app" }, + }); - const auditLog = readFileSync(join(auditDir, auditFiles[0]!), "utf-8"); - expect(auditLog).toContain('"operation":"session.spawn"'); - expect(auditLog).toContain('"operation":"batch_enrichment.log"'); - expect(auditLog).toContain('"message":"GraphQL batch returned cached result"'); - expect(stderrSpy).not.toHaveBeenCalled(); + const auditDir = join(getObservabilityBaseDir(config.configPath), "processes"); + const auditFiles = readdirSync(auditDir).filter((fileName) => fileName.endsWith(".ndjson")); + expect(auditFiles.length).toBeGreaterThan(0); - stderrSpy.mockRestore(); + const auditLog = readFileSync(join(auditDir, auditFiles[0]!), "utf-8"); + expect(auditLog).toContain('"operation":"session.spawn"'); + expect(auditLog).toContain('"operation":"batch_enrichment.log"'); + expect(auditLog).toContain('"message":"GraphQL batch returned cached result"'); + expect(auditLog).toContain('"timestamp"'); + expect(stderrSpy).not.toHaveBeenCalled(); + } finally { + stderrSpy.mockRestore(); + if (originalObservabilityStderr === undefined) { + delete process.env["AO_OBSERVABILITY_STDERR"]; + } else { + process.env["AO_OBSERVABILITY_STDERR"] = originalObservabilityStderr; + } + } }); }); diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index 7f1339337..7245684e6 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -386,7 +386,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan }); }, log(level, message) { - observer?.recordDiagnostic({ + observer?.recordDiagnostic?.({ operation: "batch_enrichment.log", correlationId: createCorrelationId("graphql-batch"), projectId: scopedProjectId, diff --git a/packages/core/src/observability.ts b/packages/core/src/observability.ts index 54e45c577..41a181de2 100644 --- a/packages/core/src/observability.ts +++ b/packages/core/src/observability.ts @@ -1,10 +1,12 @@ import { appendFileSync, + statSync, mkdirSync, existsSync, readFileSync, readdirSync, renameSync, + unlinkSync, writeFileSync, } from "node:fs"; import { join } from "node:path"; @@ -130,7 +132,7 @@ export interface SetHealthInput { export interface ProjectObserver { readonly component: string; recordOperation(input: RecordOperationInput): void; - recordDiagnostic(input: { + recordDiagnostic?(input: { operation: string; correlationId: string; projectId?: string; @@ -145,6 +147,7 @@ export interface ProjectObserver { const TRACE_LIMIT = 80; const SESSION_LIMIT = 200; +const AUDIT_LOG_MAX_BYTES = 5 * 1024 * 1024; const LEVEL_ORDER: Record = { debug: 10, info: 20, @@ -209,9 +212,32 @@ function appendAuditLog( level: ObservabilityLevel, ): void { const filePath = getAuditLogPath(config, component); + const rotatedPath = `${filePath}.1`; + if (existsSync(filePath)) { + const currentSize = statSync(filePath).size; + if (currentSize >= AUDIT_LOG_MAX_BYTES) { + if (existsSync(rotatedPath)) { + unlinkSync(rotatedPath); + } + renameSync(filePath, rotatedPath); + } + } appendFileSync(filePath, `${JSON.stringify({ ...payload, level })}\n`, "utf-8"); } +function appendObservabilityFailure( + config: OrchestratorConfig, + component: string, + payload: Record, +): void { + try { + const filePath = join(getObservabilityDir(config), "observability-errors.ndjson"); + appendFileSync(filePath, `${JSON.stringify(payload)}\n`, "utf-8"); + } catch { + // Best effort only — avoid recursive observability failures. + } +} + function readSnapshot(filePath: string, component: string): ProcessObservabilitySnapshot { if (!existsSync(filePath)) { return { @@ -344,21 +370,21 @@ export function createProjectObserver( const snapshot = readSnapshot(filePath, normalizedComponent); updater(snapshot); writeSnapshot(config, snapshot); - if (logEntry) { + if (logEntry && shouldLog(logEntry.level)) { appendAuditLog(config, normalizedComponent, logEntry.payload, logEntry.level); emitStructuredLog(logEntry.payload, logEntry.level); } } catch (error) { - emitStructuredLog( - { - source: "ao-observability", - component: normalizedComponent, - outcome: "failure", - operation: "observability.write", - reason: error instanceof Error ? error.message : String(error), - }, - "error", - ); + const payload = { + source: "ao-observability", + timestamp: nowIso(), + component: normalizedComponent, + outcome: "failure", + operation: "observability.write", + reason: error instanceof Error ? error.message : String(error), + }; + appendObservabilityFailure(config, normalizedComponent, payload); + emitStructuredLog(payload, "error"); } } @@ -428,6 +454,7 @@ export function createProjectObserver( level, payload: { source: "ao-observability", + timestamp, component: normalizedComponent, metric: input.metric, operation, @@ -489,6 +516,7 @@ export function createProjectObserver( level, payload: { source: "ao-observability", + timestamp, component: normalizedComponent, operation: input.operation, correlationId: input.correlationId, @@ -523,6 +551,7 @@ export function createProjectObserver( level: input.status === "error" ? "error" : input.status === "warn" ? "warn" : "info", payload: { source: "ao-observability", + timestamp: updatedAt, component: normalizedComponent, surface: input.surface, status: input.status,