refactor: add a run-listener observer and adopt it in the cli

core/src/execute/evaluatorLoop.ts

@@ -13,6 +13,7 @@ import { runAgentAttack } from "./runAgentLoop.js";
 import { runMcpAttack } from "./mcpAttackDriver.js";
 import { summarizeVerdicts, toEvaluatorResult } from "./aggregate.js";
 import { errorJudge } from "../lib/judgeTypes.js";
+import { verdictIcon } from "../lib/verdictIcon.js";
 import { TurnPlan } from "./turnPlan.js";
 import { isStopError, getStopReason } from "../lib/llmRetry.js";
 import { log } from "../lib/logger.js";
@@ -182,9 +183,9 @@ export async function runEvaluatorAttacks(
       attackResults.push(result);
       notify({ type: "attack_done", attackId: attack.id, verdict: result.judge.verdict });
 
-      const icon =
-        result.judge.verdict === "PASS" ? "✓" : result.judge.verdict === "FAIL" ? "✗" : "⚠";
-      log.info(`     ${icon} ${result.judge.verdict} (score ${result.judge.score}/10)`);
+      log.info(
+        `     ${verdictIcon(result.judge.verdict)} ${result.judge.verdict} (score ${result.judge.score}/10)`
+      );
     }
 
     const { passed, failed, errors } = summarizeVerdicts(attackResults);

core/src/execute/runAll.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from "../lib/random.js";
 import type { LanguageModel } from "ai";
-import type { RunConfig, EvaluatorResult, UnifiedRunReport } from "./types.js";
+import type { RunConfig, EvaluatorResult, UnifiedRunReport, ProgressEvent } from "./types.js";
 import type { ToolInfo } from "../generate/generateAttacks.js";
 import type { AgentTarget } from "../targets/agentTarget.js";
 import { createMcpTarget } from "../targets/mcpTarget.js";
@@ -9,6 +9,7 @@ import type { EvaluatorSpec } from "../evaluators/parseEvaluator.js";
 import { loadSkillCatalog, resolveSuiteEvaluatorIds } from "../config/loadSkillCatalog.js";
 import { loadCatalog } from "../catalog/loadCatalog.js";
 import { runBaselineScans } from "./baselineScanner.js";
+import { dispatchProgress, notifyListeners, type RunListener } from "./runListener.js";
 import { runEvaluatorAttacks } from "./evaluatorLoop.js";
 import { buildUnifiedReport, modelLabel } from "./aggregate.js";
 import { createModel } from "../providers/factory.js";
@@ -19,17 +20,19 @@ import { log } from "../lib/logger.js";
 
 export interface RunAllOptions {
   onProgress?: (event: ProgressEvent) => void;
+  /**
+   * Lifecycle observers (reporters, telemetry). Receive the same per-attack
+   * events as onProgress, plus run-level onRunStart/onRunFinish/onRunError hooks.
+   */
+  listeners?: RunListener[];
   outputDir?: string;
   /** Pre-built agent target. When omitted, createAgentTarget is called using config.target. */
   agentTarget?: AgentTarget;
 }
 
-export type ProgressEvent =
-  | { type: "evaluator_start"; evaluatorId: string; evaluatorName: string }
-  | { type: "attack_start"; attackId: string; patternName: string }
-  | { type: "attack_done"; attackId: string; verdict: "PASS" | "FAIL" | "ERROR" }
-  | { type: "evaluator_done"; evaluatorId: string; passed: number; failed: number; errors: number }
-  | { type: "run_stopped"; reason: string };
+// Re-exported for callers importing it from this module; defined in ./types.js
+// (the single source of truth) so the listener SPI doesn't cycle back here.
+export type { ProgressEvent } from "./types.js";
 
 /**
  * Core execute loop: resolves evaluators, generates attacks per effort level,
@@ -40,41 +43,59 @@ export async function runAll(
   config: RunConfig,
   options?: RunAllOptions
 ): Promise<UnifiedRunReport> {
-  const notify = options?.onProgress ?? (() => {});
-
-  const attackModel = resolveModel(config.attackerLlm);
-  // Single source of truth for the judge LLM: explicit judge model, else the
-  // attacker model. Reused by the baseline scans and the MCP dispatch below so
-  // the fallback can't drift between paths.
-  const judgeLlmConfig = config.judgeLlm ?? config.attackerLlm;
-  const judgeModel = resolveModel(judgeLlmConfig);
-
-  const isMcp = config.target.kind === "mcp";
-  const evaluators = await resolveEvaluators(
-    config.selection,
-    isMcp ? "mcp" : "agent",
-    config.selection.dependsOn
-  );
-
-  // For MCP targets, connect once and discover tools
+  // Fan every progress event to the legacy onProgress callback (backward-compat)
+  // and to any registered lifecycle listeners.
+  const listeners = options?.listeners ?? [];
+  const notify = (event: ProgressEvent) => {
+    options?.onProgress?.(event);
+    notifyListeners(listeners, (listener) => dispatchProgress(listener, event));
+  };
+
+  // Declared before the try so the finally can always close the MCP target (even
+  // if listTools() throws after connect). Everything else — model + evaluator
+  // resolution included — runs inside the try so any failure reaches onRunError.
   let mcpTarget: Awaited<ReturnType<typeof createMcpTarget>> | null = null;
-  let tools: ToolInfo[] = [];
 
-  if (isMcp) {
-    mcpTarget = await createMcpTarget(config.target as import("./types.js").McpTargetConfig);
-    tools = await mcpTarget.listTools();
-    log.info(`MCP target connected — ${tools.length} tool(s) available`);
-  }
+  try {
+    const attackModel = resolveModel(config.attackerLlm);
+    // Single source of truth for the judge LLM: explicit judge model, else the
+    // attacker model. Reused by the baseline scans and the MCP dispatch below so
+    // the fallback can't drift between paths.
+    const judgeLlmConfig = config.judgeLlm ?? config.attackerLlm;
+    const judgeModel = resolveModel(judgeLlmConfig);
+
+    const isMcp = config.target.kind === "mcp";
+    const evaluators = await resolveEvaluators(
+      config.selection,
+      isMcp ? "mcp" : "agent",
+      config.selection.dependsOn
+    );
 
-  // Optional: pull real production traces and summarise them so attack
-  // generation can be grounded in actual usage patterns.
-  const traceContext = await curateTracesIfConfigured(config, attackModel, options?.outputDir);
+    let tools: ToolInfo[] = [];
+    const ordered = topoSortEvaluators(evaluators);
+    const evaluatorResults: EvaluatorResult[] = [];
+
+    // Fire onRunStart once the evaluator count is known but before the (possibly
+    // slow) MCP connect and trace curation, so watchers see the run begin during
+    // those steps. onRunStart always pairs with a terminal hook (onRunFinish or
+    // onRunError). A failure in the setup above throws before onRunStart, so
+    // listeners get onRunError with no preceding onRunStart. evaluatorCount is the
+    // attack-evaluator count; MCP baseline pre-flight scans add extra results.
+    notifyListeners(listeners, (listener) =>
+      listener.onRunStart?.({ evaluatorCount: ordered.length })
+    );
 
-  const ordered = topoSortEvaluators(evaluators);
-  const evaluatorResults: EvaluatorResult[] = [];
-  let stopReason: string | undefined;
+    // For MCP targets, connect once and discover tools.
+    if (isMcp) {
+      mcpTarget = await createMcpTarget(config.target as import("./types.js").McpTargetConfig);
+      tools = await mcpTarget.listTools();
+      log.info(`MCP target connected — ${tools.length} tool(s) available`);
+    }
+
+    // Optional: pull real production traces and summarise them so attack
+    // generation can be grounded in actual usage patterns.
+    const traceContext = await curateTracesIfConfigured(config, attackModel, options?.outputDir);
 
-  try {
     // ── MCP pre-flight scans (always run for MCP targets) ──────────────
     if (isMcp && mcpTarget) {
       evaluatorResults.push(
@@ -103,17 +124,26 @@ export async function runAll(
       notify,
     });
     evaluatorResults.push(...loop.evaluatorResults);
-    stopReason = loop.stopReason;
+    const stopReason = loop.stopReason;
+
+    // Build report (partial or complete) with stop reason if applicable.
+    const report = buildReport(config, evaluatorResults);
+    if (stopReason) {
+      (report as UnifiedRunReport & { stopReason?: string }).stopReason = stopReason;
+    }
+
+    // Terminal success signal — always the last hook (report may be partial if the
+    // run stopped early; onRunStopped fired earlier is a non-terminal notice).
+    notifyListeners(listeners, (listener) => listener.onRunFinish?.(report));
+    return report;
+  } catch (err) {
+    // Terminal error signal — reached by any failure in the run, including model /
+    // evaluator resolution, MCP connect, the loop, and report building.
+    notifyListeners(listeners, (listener) => listener.onRunError?.({ error: err }));
+    throw err;
   } finally {
     if (mcpTarget) await mcpTarget.close().catch(() => {});
   }
-
-  // Build report (partial or complete) with stop reason if applicable
-  const report = buildReport(config, evaluatorResults);
-  if (stopReason) {
-    (report as UnifiedRunReport & { stopReason?: string }).stopReason = stopReason;
-  }
-  return report;
 }
 
 // ---------------------------------------------------------------------------

core/src/execute/runListener.ts

@@ -0,0 +1,94 @@
+// RunListener — an observer over a run's lifecycle (Observer / SPI).
+//
+// The engine emits lifecycle events; reporters, progress UIs, and telemetry
+// adapters consume them by implementing this interface, so a new output format
+// becomes a new listener with no engine edits. Every hook is optional — a
+// listener implements only what it needs.
+
+import type { UnifiedRunReport, ProgressEvent } from "./types.js";
+import { log } from "../lib/logger.js";
+
+/** The payload of a ProgressEvent variant, minus its `type` discriminant. */
+type Payload<T extends ProgressEvent["type"]> = Omit<Extract<ProgressEvent, { type: T }>, "type">;
+
+export interface RunListener {
+  // Lifecycle: when onRunStart fires it is always paired with exactly one terminal
+  // hook — onRunFinish (the run produced a report, complete or partial) or
+  // onRunError (the run threw). A failure during run setup throws before onRunStart,
+  // so onRunError can also fire on its own. onRunStopped is a non-terminal notice:
+  // onRunFinish still follows it with the partial report.
+
+  /** Fired once at the start, before the target connects. Skipped if run setup fails first. */
+  onRunStart?(info: { evaluatorCount: number }): void;
+  onEvaluatorStart?(info: Payload<"evaluator_start">): void;
+  onAttackStart?(info: Payload<"attack_start">): void;
+  onAttackDone?(info: Payload<"attack_done">): void;
+  onEvaluatorDone?(info: Payload<"evaluator_done">): void;
+  /** Non-terminal notice that a non-retryable error cut the run short; onRunFinish still follows. */
+  onRunStopped?(info: Payload<"run_stopped">): void;
+  /** Terminal: the run threw (may fire without a preceding onRunStart on a setup failure). */
+  onRunError?(info: { error: unknown }): void;
+  /** Terminal: fired once with the final report (complete, or partial after a stop). */
+  onRunFinish?(report: UnifiedRunReport): void;
+}
+
+function warnListenerFailed(err: unknown): void {
+  log.warn(
+    `A RunListener hook failed and was skipped — fix the listener implementation: ${
+      err instanceof Error ? err.message : String(err)
+    }`
+  );
+}
+
+/**
+ * Invoke a hook on every listener with error isolation: a listener that throws —
+ * synchronously OR by rejecting an async hook — is logged and skipped, so a buggy
+ * reporter/telemetry adapter can never abort the run. Hooks are declared `void`
+ * (fire-and-forget); a returned promise is not awaited, but its rejection is caught.
+ * Centralizes the fan-out so every notification site is uniform.
+ */
+export function notifyListeners(
+  listeners: readonly RunListener[],
+  invoke: (listener: RunListener) => void | Promise<void>
+): void {
+  for (const listener of listeners) {
+    try {
+      const result = invoke(listener);
+      if (result && typeof (result as Promise<void>).then === "function") {
+        void (result as Promise<void>).then(undefined, warnListenerFailed);
+      }
+    } catch (err) {
+      warnListenerFailed(err);
+    }
+  }
+}
+
+/**
+ * Route one ProgressEvent to the matching RunListener hook. The engine keeps a
+ * single ProgressEvent stream as the source of per-attack events; this adapts it
+ * to the typed listener interface so callers don't switch on `event.type`.
+ */
+export function dispatchProgress(listener: RunListener, event: ProgressEvent): void {
+  switch (event.type) {
+    case "evaluator_start":
+      listener.onEvaluatorStart?.(event);
+      break;
+    case "attack_start":
+      listener.onAttackStart?.(event);
+      break;
+    case "attack_done":
+      listener.onAttackDone?.(event);
+      break;
+    case "evaluator_done":
+      listener.onEvaluatorDone?.(event);
+      break;
+    case "run_stopped":
+      listener.onRunStopped?.(event);
+      break;
+    default: {
+      // Compile-time exhaustiveness: a new ProgressEvent variant forces a case.
+      const _exhaustive: never = event;
+      void _exhaustive;
+    }
+  }
+}

core/src/execute/types.ts

@@ -228,3 +228,14 @@ export interface UnifiedRunReport {
   /** Set when the run was stopped early due to a non-retryable LLM error. */
   stopReason?: string;
 }
+
+/**
+ * Per-attack progress events emitted by the run engine — the single source of
+ * truth for run lifecycle payloads. {@link RunListener} hooks derive from it.
+ */
+export type ProgressEvent =
+  | { type: "evaluator_start"; evaluatorId: string; evaluatorName: string }
+  | { type: "attack_start"; attackId: string; patternName: string }
+  | { type: "attack_done"; attackId: string; verdict: "PASS" | "FAIL" | "ERROR" }
+  | { type: "evaluator_done"; evaluatorId: string; passed: number; failed: number; errors: number }
+  | { type: "run_stopped"; reason: string };

core/src/lib/verdictIcon.ts

@@ -0,0 +1,6 @@
+import type { Verdict } from "./judgeTypes.js";
+
+/** Terminal glyph for a verdict: ✓ pass, ✗ fail, ⚠ error. */
+export function verdictIcon(verdict: Verdict): string {
+  return verdict === "PASS" ? "✓" : verdict === "FAIL" ? "✗" : "⚠";
+}

core/tests/runListener.test.ts

@@ -0,0 +1,77 @@
+/**
+ * PR11 — RunListener SPI.
+ *
+ * Pins the ProgressEvent → RunListener adapter: each event type routes to exactly
+ * one hook with the event payload, and a listener that implements no hooks (or only
+ * some) is never called with a method it lacks — dispatchProgress must not throw.
+ */
+import { test } from "node:test";
+import assert from "node:assert/strict";
+import { dispatchProgress, notifyListeners, type RunListener } from "../src/execute/runListener.js";
+import type { ProgressEvent } from "../src/execute/types.js";
+
+function recorder() {
+  const calls: Array<{ hook: string; arg: unknown }> = [];
+  const listener: RunListener = {
+    onEvaluatorStart: (a) => calls.push({ hook: "onEvaluatorStart", arg: a }),
+    onAttackStart: (a) => calls.push({ hook: "onAttackStart", arg: a }),
+    onAttackDone: (a) => calls.push({ hook: "onAttackDone", arg: a }),
+    onEvaluatorDone: (a) => calls.push({ hook: "onEvaluatorDone", arg: a }),
+    onRunStopped: (a) => calls.push({ hook: "onRunStopped", arg: a }),
+  };
+  return { listener, calls };
+}
+
+test("each ProgressEvent routes to its matching hook with the payload", () => {
+  const { listener, calls } = recorder();
+  const events: ProgressEvent[] = [
+    { type: "evaluator_start", evaluatorId: "e1", evaluatorName: "Eval One" },
+    { type: "attack_start", attackId: "a1", patternName: "pat" },
+    { type: "attack_done", attackId: "a1", verdict: "FAIL" },
+    { type: "evaluator_done", evaluatorId: "e1", passed: 1, failed: 2, errors: 0 },
+    { type: "run_stopped", reason: "rate limited" },
+  ];
+  for (const e of events) dispatchProgress(listener, e);
+
+  assert.deepStrictEqual(
+    calls.map((c) => c.hook),
+    ["onEvaluatorStart", "onAttackStart", "onAttackDone", "onEvaluatorDone", "onRunStopped"]
+  );
+  // Payload is forwarded verbatim (the discriminant `type` may ride along).
+  assert.deepStrictEqual(calls[0].arg, {
+    type: "evaluator_start",
+    evaluatorId: "e1",
+    evaluatorName: "Eval One",
+  });
+  assert.deepStrictEqual(calls[2].arg, { type: "attack_done", attackId: "a1", verdict: "FAIL" });
+});
+
+test("a listener implementing no hooks is a safe no-op", () => {
+  const empty: RunListener = {};
+  const events: ProgressEvent[] = [
+    { type: "evaluator_start", evaluatorId: "e", evaluatorName: "n" },
+    { type: "attack_done", attackId: "a", verdict: "PASS" },
+    { type: "run_stopped", reason: "x" },
+  ];
+  for (const e of events) {
+    assert.doesNotThrow(() => dispatchProgress(empty, e));
+  }
+});
+
+test("notifyListeners isolates a throwing listener and still invokes the rest", () => {
+  const seen: string[] = [];
+  const bomb: RunListener = {
+    onRunStart: () => {
+      throw new Error("boom");
+    },
+  };
+  const good: RunListener = {
+    onRunStart: (info) => seen.push(`ok:${info.evaluatorCount}`),
+  };
+
+  // The thrower must not propagate, and the later listener must still run.
+  assert.doesNotThrow(() =>
+    notifyListeners([bomb, good], (l) => l.onRunStart?.({ evaluatorCount: 3 }))
+  );
+  assert.deepStrictEqual(seen, ["ok:3"]);
+});