Merge pull request #1430 from corbitsdev/cl-10661-logging-cleanup-collapse-storms-fix-levels-add-correla

refactor(logging): collapse double-log storms, fix levels, correlate ids

Author: TheGreatAxios

src/cost/pricing-fetcher.ts

@@ -306,9 +306,7 @@ export function startPricingRefresh(
   const refreshIntervalMs =
     options.refreshIntervalMs ?? DEFAULT_REFRESH_INTERVAL_MS;
   const timer = setInterval(() => {
-    loadPricing(options).catch((err: unknown) => {
-      process.stderr.write(`pricing-fetcher: refresh error: ${err}\n`);
-    });
+    void loadPricing(options);
   }, refreshIntervalMs);
   return timer;
 }

src/cost/pricing-metadata.ts

@@ -29,14 +29,9 @@ export function schedulePricingMetadataRefresh(
 ): void {
   if (refreshScheduled) return;
   refreshScheduled = true;
-  void loadPricing(options)
-    .then((cache) => {
-      if (cache !== null) applyPricingCacheMetadata(cache);
-    })
-    .catch((err: unknown) => {
-      // Match pricing-fetcher: keep refresh best-effort, surface the failure.
-      process.stderr.write(`pricing-metadata: refresh error: ${err}\n`);
-    });
+  void loadPricing(options).then((cache) => {
+    if (cache !== null) applyPricingCacheMetadata(cache);
+  });
 }
 
 export async function bootstrapPricingMetadata(

src/exec/dispose.ts

@@ -1,6 +1,7 @@
 import { getLogger } from "@intx/log";
 
 import { LOG_NAMESPACE_ROOT } from "../branding.js";
+import { caughtDebug } from "../logging/caught.js";
 import { awaitCloseWithoutHidingLeftover } from "../subagent/dispose.js";
 import type { SubAgentSessionStore } from "../subagent/index.js";
 
@@ -59,9 +60,11 @@ async function runExecDispose(args: {
     try {
       await args.toolset.dispose();
     } catch (err: unknown) {
-      logger.debug("toolset.dispose during exec finally failed: {error}", {
-        error: (args.sanitizeDiagnostic ?? String)(formatCaughtError(err)),
-      });
+      caughtDebug(
+        logger,
+        "toolset.dispose during exec finally failed",
+        args.sanitizeDiagnostic,
+      )(err);
       failures.push(err);
     }
   }
@@ -74,9 +77,11 @@ async function runExecDispose(args: {
     try {
       await awaitCloseWithoutHidingLeftover(args.agent.close(), failures[0]);
     } catch (err: unknown) {
-      logger.debug("agent.close during exec finally failed: {error}", {
-        error: (args.sanitizeDiagnostic ?? String)(formatCaughtError(err)),
-      });
+      caughtDebug(
+        logger,
+        "agent.close during exec finally failed",
+        args.sanitizeDiagnostic,
+      )(err);
       failures.push(err);
     }
   }

src/exec/runner.ts

@@ -42,6 +42,7 @@ import {
 } from "../subagent/index.js";
 import { getProcessAdmissionQueue } from "../subagent/admission.js";
 import { disposeExecRuntime, formatCaughtError } from "./dispose.js";
+import { caughtDebug } from "../logging/caught.js";
 import {
   EXEC_MCP_CONNECT_WAIT_MS,
   EXEC_MCP_HANDSHAKE_TIMEOUT_MS,
@@ -469,7 +470,10 @@ export async function runExec(config: Config): Promise<ExecResult> {
     priorState = prior?.kind === "ok" ? prior.state : undefined;
   } catch (err) {
     const message = formatCaughtError(err);
-    logger.error("exec setup failed: {error}", { error: message });
+    logger.error("exec setup failed: {error}", {
+      sessionId,
+      error: message,
+    });
     stderr.write(`Error: ${message}\n`);
     liveTelemetry.capture(
       "session_end",
@@ -623,11 +627,13 @@ export async function runExec(config: Config): Promise<ExecResult> {
 
   try {
     // Pricing seed is optional for exec; continue without rates.
-    const inferenceDeps = await assembleInferenceBase((err: unknown) => {
-      logger.debug("seedPricingMetadataFromCache failed: {error}", {
-        error: sanitizeExecDiagnostic(formatCaughtError(err)),
-      });
-    });
+    const inferenceDeps = await assembleInferenceBase(
+      caughtDebug(
+        logger,
+        "seedPricingMetadataFromCache failed",
+        sanitizeExecDiagnostic,
+      ),
+    );
 
     // One-shot migration only when the path-trust file is absent. Headless
     // exec has no frame to corrupt, so skipped members write straight to stderr.
@@ -1246,22 +1252,22 @@ export async function runExec(config: Config): Promise<ExecResult> {
         // entry and would write the whole reply as a spurious partial.
         // Dispose is a no-op on success and a real record only if a cycle
         // died without a terminal event.
-        await activeAgent.close().catch((err: unknown) => {
-          logger.debug(
-            "agent.close during successful-send teardown failed: {error}",
-            {
-              error: sanitizeExecDiagnostic(formatCaughtError(err)),
-            },
-          );
-        });
-        await streamPromise.catch((err: unknown) => {
-          logger.debug(
-            "stream drain during successful-send teardown failed: {error}",
-            {
-              error: sanitizeExecDiagnostic(formatCaughtError(err)),
-            },
+        await activeAgent
+          .close()
+          .catch(
+            caughtDebug(
+              logger,
+              "agent.close during successful-send teardown failed",
+              sanitizeExecDiagnostic,
+            ),
           );
-        });
+        await streamPromise.catch(
+          caughtDebug(
+            logger,
+            "stream drain during successful-send teardown failed",
+            sanitizeExecDiagnostic,
+          ),
+        );
         await cycleRecorder.dispose("cancelled");
       } else {
         // Failed or aborted send: close() tears down stream consumers before
@@ -1270,22 +1276,22 @@ export async function runExec(config: Config): Promise<ExecResult> {
         await cycleRecorder.dispose(
           sendCompleted ? "cancelled" : "send-failed",
         );
-        await activeAgent.close().catch((err: unknown) => {
-          logger.debug(
-            "agent.close during failed-send teardown failed: {error}",
-            {
-              error: sanitizeExecDiagnostic(formatCaughtError(err)),
-            },
-          );
-        });
-        await streamPromise.catch((err: unknown) => {
-          logger.debug(
-            "stream drain during failed-send teardown failed: {error}",
-            {
-              error: sanitizeExecDiagnostic(formatCaughtError(err)),
-            },
+        await activeAgent
+          .close()
+          .catch(
+            caughtDebug(
+              logger,
+              "agent.close during failed-send teardown failed",
+              sanitizeExecDiagnostic,
+            ),
           );
-        });
+        await streamPromise.catch(
+          caughtDebug(
+            logger,
+            "stream drain during failed-send teardown failed",
+            sanitizeExecDiagnostic,
+          ),
+        );
       }
     }
 
@@ -1379,7 +1385,10 @@ export async function runExec(config: Config): Promise<ExecResult> {
     return result;
   } catch (err) {
     const diagnosticMessage = sanitizeExecDiagnostic(formatCaughtError(err));
-    logger.error("exec failed: {error}", { error: diagnosticMessage });
+    logger.error("exec failed: {error}", {
+      sessionId,
+      error: diagnosticMessage,
+    });
     const userMessage = sanitizeExecDiagnostic(
       execUserFailureMessage(
         config,
@@ -1420,7 +1429,10 @@ export async function runExec(config: Config): Promise<ExecResult> {
       });
     } catch (err: unknown) {
       const message = sanitizeExecDiagnostic(formatCaughtError(err));
-      logger.error("runtime dispose failed: {error}", { error: message });
+      logger.error("runtime dispose failed: {error}", {
+        sessionId,
+        error: message,
+      });
       stderr.write(`Error: runtime dispose failed: ${message}\n`);
       if (result !== undefined) {
         result.exitCode = 1;

src/logging/caught.test.ts

@@ -0,0 +1,39 @@
+import { describe, expect, test } from "bun:test";
+
+import { caughtDebug } from "./caught.js";
+
+// @intx/log renders the {error} template at sink level, so capture both the
+// message template and the properties to pin the normalization-and-append
+// contract.
+describe("caughtDebug", () => {
+  interface Capture {
+    message: string;
+    properties: Record<string, unknown> | undefined;
+  }
+  const capture = {
+    debug: (message: string, properties?: Record<string, unknown>) => {
+      capture.last = { message, properties };
+    },
+    last: undefined as Capture | undefined,
+  };
+
+  test("Error values feed {error} from .message", () => {
+    caughtDebug(capture, "teardown failed")(new Error("boom"));
+    expect(capture.last?.message).toBe("teardown failed: {error}");
+    expect(capture.last?.properties).toEqual({ error: "boom" });
+  });
+
+  test("non-Error values are String()ed", () => {
+    caughtDebug(capture, "teardown failed")("boom-string");
+    expect(capture.last?.properties).toEqual({ error: "boom-string" });
+  });
+
+  test("a sanitizer runs on the normalized error message", () => {
+    caughtDebug(capture, "teardown failed", (error) =>
+      error.replaceAll("secret", "[redacted]"),
+    )(new Error("secret leaked"));
+    expect(capture.last?.properties).toEqual({
+      error: "[redacted] leaked",
+    });
+  });
+});

src/logging/caught.ts

@@ -0,0 +1,14 @@
+interface DebugSink {
+  debug: (message: string, properties?: Record<string, unknown>) => void;
+}
+
+export function caughtDebug(
+  logger: DebugSink,
+  message: string,
+  sanitize?: (error: string) => string,
+): (err: unknown) => void {
+  return (err) => {
+    const error = err instanceof Error ? err.message : String(err);
+    logger.debug(`${message}: {error}`, { error: (sanitize ?? String)(error) });
+  };
+}

src/perf/otel-sink.ts

@@ -270,7 +270,9 @@ export async function flushPerfToOtel(
   try {
     const resolved = resolveOtelExportConfig(settings, env);
     if (!resolved.ok) {
-      log.warn("OTEL export skipped: {message}", { message: resolved.message });
+      log.debug("OTEL export skipped: {message}", {
+        message: resolved.message,
+      });
       return;
     }
     config = resolved.config;

src/permission/approval-store-migration.test.ts

@@ -9,6 +9,7 @@ import {
 } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { dirname, join } from "node:path";
+import { installFileLogSink } from "../logging/sink.js";
 import { generateSessionId, sessionDir } from "../session/index.js";
 import { loadSeededApprovals } from "../session/runtime-assembly.js";
 import { normalizeSeededApprovals } from "./authz-grants.js";
@@ -297,6 +298,56 @@ describe("migratePersistedApprovalStores", () => {
     }
   });
 
+  test("per-file migration failures surface once as a single summary warn", async () => {
+    const logDir = await mkdtemp(join(tmpdir(), "approval-migration-log-"));
+    installFileLogSink(join(logDir, "corbits.log"));
+    const sessionOriginal = {
+      approvals: [
+        { tool: "update_plan", pattern: "plan *" },
+        { tool: "run_shell", pattern: "npm *" },
+      ],
+    };
+    await writeFile(sessionStorePath(), JSON.stringify(sessionOriginal));
+    await mkdir(join(cwd, ".corbits"), { recursive: true });
+    await writeFile(
+      projectStorePath(),
+      JSON.stringify({
+        approvals: [{ tool: "update_plan", pattern: "plan *" }],
+      }),
+    );
+    // Read-only store dirs make the backup step fail, so both stores skip
+    // untouched while the global path (absent) stays a clean no-op.
+    const sessionStoreDir = dirname(sessionStorePath());
+    const projectStoreDir = join(cwd, ".corbits");
+    await chmod(sessionStoreDir, 0o555);
+    await chmod(projectStoreDir, 0o555);
+    try {
+      const result = await migratePersistedApprovalStores(cwd, sessionId, home);
+      expect(result.purged).toBe(0);
+      const failed = result.files.filter((file) => file.error !== undefined);
+      expect(failed).toHaveLength(2);
+      for (const file of failed) {
+        expect(file.error).toContain("backup failed");
+        expect(file.purged).toBe(0);
+      }
+
+      const lines = (await readFile(join(logDir, "corbits.log"), "utf8"))
+        .split("\n")
+        .filter((line) => line.includes("Skipped approval-store migration"));
+      // One summary warn for the whole migration, not one per failing store.
+      expect(lines).toHaveLength(1);
+      const summary = lines[0];
+      if (summary === undefined) throw new Error("summary warn missing");
+      expect(summary).toMatch(/"level":"warning"/);
+      expect(summary).toContain(sessionStorePath());
+      expect(summary).toContain(projectStorePath());
+    } finally {
+      await chmod(sessionStoreDir, 0o755);
+      await chmod(projectStoreDir, 0o755);
+      await rm(logDir, { recursive: true, force: true });
+    }
+  });
+
   test("existing torn .bak plus live update_plan still purges live", async () => {
     const path = sessionStorePath();
     const original = {