Reorganize worker observability layer into worker/observability/ (M8 part 3/3)

Collects the five post-run diagnostic + escalation modules into src/main/worker/observability/: run_events persistence, stdout/stderr noise filter, credential-shape log redactor, self-wake circuit breaker, and retry-aware failure escalation pipeline. circuit-breaker.ts and failure-escalation.ts remain as standalone files inside the directory — each owns a distinct state machine that deserves its own boundary — rather than collapsing into a single index.ts. handlers/*.ts and tests/unit/*.test.ts updated to point at the new paths.

src/main/worker/observability/circuit-breaker.ts

@@ -0,0 +1,49 @@
+/**
+ * Self-wake circuit breaker.
+ *
+ * Prevents infinite self-wake loops when an agent is stuck on a task.
+ * After MAX_CONSECUTIVE_WAKES consecutive wakes without progress (i.e.,
+ * without a successful run), self-wake is suppressed so the manager can
+ * be notified instead.
+ */
+
+export const MAX_CONSECUTIVE_WAKES = 5;
+
+export interface CircuitBreaker {
+  /** Attempt a self-wake. Returns true if allowed, false if circuit breaker trips. */
+  tryWake(agentId: string, companyId: string): boolean;
+  /** Reset the counter for an agent (called on task progress / successful run). */
+  reset(agentId: string, companyId: string): void;
+  /** Get the current consecutive-wake count for an agent. */
+  getCount(agentId: string, companyId: string): number;
+}
+
+export function createCircuitBreaker(
+  consecutiveWakeMap: Map<string, number>,
+  maxConsecutive: number = MAX_CONSECUTIVE_WAKES,
+): CircuitBreaker {
+  function wakeKey(agentId: string, companyId: string): string {
+    return `${agentId}:${companyId}`;
+  }
+
+  return {
+    tryWake(agentId: string, companyId: string): boolean {
+      const key = wakeKey(agentId, companyId);
+      const count = consecutiveWakeMap.get(key) ?? 0;
+      if (count >= maxConsecutive) {
+        consecutiveWakeMap.delete(key);
+        return false;
+      }
+      consecutiveWakeMap.set(key, count + 1);
+      return true;
+    },
+
+    reset(agentId: string, companyId: string): void {
+      consecutiveWakeMap.delete(wakeKey(agentId, companyId));
+    },
+
+    getCount(agentId: string, companyId: string): number {
+      return consecutiveWakeMap.get(wakeKey(agentId, companyId)) ?? 0;
+    },
+  };
+}

src/main/worker/observability/failure-escalation.ts

@@ -0,0 +1,129 @@
+/**
+ * Retry-aware failure escalation pipeline.
+ *
+ * State machine:
+ *   1. Task fails with retry_count < MAX_RETRY_ATTEMPTS  -> increment count, re-queue
+ *   2. Task fails with retry_count >= MAX_RETRY_ATTEMPTS -> escalate to manager chain
+ *   3. No manager in chain                               -> fall through to CEO
+ *   4. No CEO available                                  -> notify user
+ *   5. Task succeeds                                     -> reset retry count
+ */
+
+export interface EscalationDeps {
+  db: {
+    getTaskRetryCount: (taskId: string) => number;
+    incrementTaskRetryCount: (taskId: string) => void;
+    resetTaskRetryCount: (taskId: string) => void;
+    getChainOfCommand: (agentId: string) => Array<{ id: string; name: string }>;
+    getCeoForCompany: (companyId: string) => { id: string; name: string } | null;
+    advanceTaskStatus: (taskId: string, status: string, note: string) => unknown;
+  };
+  wakeAgentIfPossible: (agentId: string, companyId: string, trigger: string) => unknown;
+  notify: (title: string, body: string) => void;
+  logger: {
+    info: (...args: unknown[]) => void;
+    warn: (...args: unknown[]) => void;
+    error: (...args: unknown[]) => void;
+  };
+}
+
+export interface EscalationResult {
+  action: "retry" | "escalated" | "user_notified" | "none";
+  retryScheduled?: boolean;
+}
+
+export function handleFailureEscalation(
+  deps: EscalationDeps,
+  run: { taskId: string; agentId: string; companyId: string },
+  task: { title: string } | null,
+  eventStatus: "failed" | "timed_out",
+  maxRetryAttempts: number,
+  retryCooldownMs: number,
+  scheduleRetry: (fn: () => void, delayMs: number) => void,
+): EscalationResult {
+  const { db, wakeAgentIfPossible, notify, logger } = deps;
+  const retryCount = db.getTaskRetryCount(run.taskId);
+
+  if (retryCount < maxRetryAttempts) {
+    db.incrementTaskRetryCount(run.taskId);
+    db.advanceTaskStatus(
+      run.taskId,
+      "todo",
+      `Run ${eventStatus} (attempt ${retryCount + 1}/${maxRetryAttempts}). Auto-retrying after ${retryCooldownMs / 1000}s cooldown.`,
+    );
+    logger.info(
+      `[escalation] Retrying task ${run.taskId} (attempt ${retryCount + 1}/${maxRetryAttempts}) for agent ${run.agentId}`,
+    );
+    scheduleRetry(() => {
+      wakeAgentIfPossible(run.agentId, run.companyId, "retry_failed");
+    }, retryCooldownMs);
+    return { action: "retry", retryScheduled: true };
+  }
+
+  // Retry limit reached — escalate up the chain
+  db.resetTaskRetryCount(run.taskId);
+  logger.warn(
+    `[escalation] Task ${run.taskId} failed ${maxRetryAttempts} times. Escalating to manager.`,
+  );
+
+  let escalated = false;
+  try {
+    const chain = db.getChainOfCommand(run.agentId);
+    if (chain.length > 0) {
+      wakeAgentIfPossible(chain[0].id, run.companyId, "report_failed");
+      escalated = true;
+      logger.info(
+        `[escalation] Escalated task ${run.taskId} to manager ${chain[0].name} (${chain[0].id})`,
+      );
+    }
+  } catch (error) {
+    logger.warn(
+      `[escalation] Chain-of-command lookup failed for agent ${run.agentId}: ${error instanceof Error ? error.message : String(error)}`,
+    );
+  }
+
+  if (!escalated) {
+    try {
+      const ceo = db.getCeoForCompany(run.companyId);
+      if (ceo) {
+        wakeAgentIfPossible(ceo.id, run.companyId, "report_failed");
+        escalated = true;
+        logger.info(
+          `[escalation] Escalated task ${run.taskId} to CEO ${ceo.name} (${ceo.id})`,
+        );
+      }
+    } catch (error) {
+      logger.warn(
+        `[escalation] CEO lookup failed for company ${run.companyId}: ${error instanceof Error ? error.message : String(error)}`,
+      );
+    }
+  }
+
+  if (!escalated) {
+    const taskTitle = task?.title ?? run.taskId;
+    notify(
+      "Task failed -- manual intervention needed",
+      `"${taskTitle}" failed ${maxRetryAttempts} times and no manager is available to handle it.`,
+    );
+    logger.error(
+      `[escalation] Task ${run.taskId} failed with no escalation target. User notified.`,
+    );
+    return { action: "user_notified" };
+  }
+
+  return { action: "escalated" };
+}
+
+export function handleSuccessReset(
+  db: EscalationDeps["db"],
+  taskId: string,
+  logger?: EscalationDeps["logger"],
+): void {
+  try {
+    db.resetTaskRetryCount(taskId);
+  } catch (error) {
+    logger?.warn(
+      `[escalation] resetTaskRetryCount failed for task ${taskId}: ${error instanceof Error ? error.message : String(error)}`,
+    );
+  }
+}

src/main/worker/observability/log-redactor.ts

@@ -0,0 +1,27 @@
+/**
+ * Pattern-based token redactor.
+ *
+ * Complements `buildRedactor` (env-value redaction) by scrubbing common
+ * credential shapes out of arbitrary strings before they reach the parent
+ * process — Bearer tokens, provider API keys, OAuth query params.
+ */
+
+const PATTERNS: Array<{ pattern: RegExp; replacement: string }> = [
+  { pattern: /Bearer\s+[A-Za-z0-9._\-+/=]{8,}/g, replacement: "Bearer [REDACTED]" },
+  { pattern: /(authorization\s*[:=]\s*)([^\s,;]{8,})/gi, replacement: "$1[REDACTED]" },
+  { pattern: /(access_token=)([^&\s"']+)/gi, replacement: "$1[REDACTED]" },
+  { pattern: /(refresh_token=)([^&\s"']+)/gi, replacement: "$1[REDACTED]" },
+  { pattern: /(api[_-]?key\s*[:=]\s*['"]?)([A-Za-z0-9._-]{12,})/gi, replacement: "$1[REDACTED]" },
+  { pattern: /sk-(?:ant-)?[A-Za-z0-9_-]{16,}/g, replacement: "sk-[REDACTED]" },
+  { pattern: /ghp_[A-Za-z0-9]{16,}/g, replacement: "ghp_[REDACTED]" },
+  { pattern: /gho_[A-Za-z0-9]{16,}/g, replacement: "gho_[REDACTED]" },
+  { pattern: /AKIA[A-Z0-9]{16}/g, replacement: "AKIA[REDACTED]" },
+];
+
+export function redactTokens(text: string): string {
+  let result = text;
+  for (const { pattern, replacement } of PATTERNS) {
+    result = result.replace(pattern, replacement);
+  }
+  return result;
+}

src/main/worker/observability/run-event-recorder.ts

@@ -0,0 +1,33 @@
+import type { AppDatabase } from "../../database";
+
+interface PersistInput {
+  companyId: string;
+  runId: string;
+  agentId: string;
+  eventType: string;
+  stream?: "stdout" | "stderr" | null;
+  level?: string | null;
+  message?: string | null;
+  payload?: Record<string, unknown> | null;
+}
+
+/**
+ * Best-effort `run_events` row writer used by the worker message handler.
+ * Failures are logged and swallowed because the run lifecycle must not be
+ * gated on audit-log writes.
+ */
+export function persistRunEvent(
+  db: AppDatabase,
+  logger: { warn: (...args: unknown[]) => void },
+  input: PersistInput,
+) {
+  try {
+    db.appendRunEvent(input);
+  } catch (error) {
+    logger.warn(
+      `[run-events] Failed to append ${input.eventType} for ${input.runId}: ${
+        error instanceof Error ? error.message : String(error)
+      }`,
+    );
+  }
+}

src/main/worker/observability/run-log-filter.ts

@@ -0,0 +1,58 @@
+/**
+ * Run-log noise filter.
+ *
+ * CLI adapters emit stdout/stderr containing system-level JSON messages,
+ * hook lifecycle events, skill injection notices, and assistant internals
+ * (thinking/tool_use). This module strips those lines before they reach
+ * the user-facing run log on disk.
+ */
+
+const RE_SKILL_INJECTION = /^\[agent-company\] Injected managed skill /;
+const RE_OMC_HOOK = /^\[omc:hooks\]/;
+const RE_AGENT_COMPANY_LOG = /^\[agent-company\] (?:heartbeat|hook_|system_)/;
+
+/** Matches system JSON except `init` subtype (which carries session metadata). */
+const RE_SYSTEM_JSON = /^\{"type"\s*:\s*"system"\s*,\s*"subtype"\s*:\s*"(?!init")/;
+
+/** Matches assistant thinking/tool_use JSON — internal reasoning, not user-visible output. */
+const RE_ASSISTANT_INTERNAL =
+  /^\{"type"\s*:\s*"assistant"\s*,\s*"subtype"\s*:\s*"(?:thinking|tool_use)"/;
+
+const RE_COMPACTING = /^Compacting conversation\.\.\./;
+
+function isNoisyLine(line: string): boolean {
+  const trimmed = line.trimStart();
+  if (trimmed.length === 0) return false;
+
+  return (
+    RE_SKILL_INJECTION.test(trimmed) ||
+    RE_OMC_HOOK.test(trimmed) ||
+    RE_AGENT_COMPANY_LOG.test(trimmed) ||
+    RE_SYSTEM_JSON.test(trimmed) ||
+    RE_ASSISTANT_INTERNAL.test(trimmed) ||
+    RE_COMPACTING.test(trimmed)
+  );
+}
+
+/**
+ * Filters a raw log chunk by removing noisy lines.
+ * @returns The cleaned chunk, or `null` if the entire chunk is noise.
+ */
+export function filterRunLogChunk(
+  _stream: "stdout" | "stderr",
+  chunk: string,
+): string | null {
+  const lines = chunk.split("\n");
+  const kept: string[] = [];
+
+  for (const line of lines) {
+    if (!isNoisyLine(line)) {
+      kept.push(line);
+    }
+  }
+
+  if (kept.length === 0) return null;
+
+  const result = kept.join("\n");
+  return result.length === 0 ? null : result;
+}