From 93dbd954ec2c74d0abe130ac5aa13e3e7d9a5808 Mon Sep 17 00:00:00 2001 From: Prasanth Date: Thu, 2 Jul 2026 17:10:29 +0530 Subject: [PATCH 1/4] refactor: add a run-listener observer and adopt it in the cli MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce RunListener — an Observer over the run lifecycle — so reporters, progress UIs, and telemetry become listeners and a new output format needs no engine edits. The engine keeps its single ProgressEvent stream; dispatchProgress adapts it to the typed hooks. - core/src/execute/runListener.ts: RunListener interface (all hooks optional) with payloads derived from ProgressEvent (Omit>), so they can't drift; a notifyListeners helper that isolates a throwing listener (logged and skipped) so a buggy reporter/telemetry adapter can never abort the run; dispatchProgress with a compile-time exhaustiveness guard. - core/src/execute/runAll.ts: listeners?: RunListener[] on RunAllOptions (additive); notify fans out to onProgress AND listeners; onRunStart before the loop, onRunFinish after the report, and onRunError in a catch so the lifecycle is symmetric (onRunStart always pairs with a terminal hook). - ProgressEvent moves to execute/types.ts (single source of truth) and is re-exported from runAll.ts, breaking the runListener<->runAll module cycle while keeping every existing importer working. - runners/cli: ConsoleProgressListener implements RunListener; run.ts swaps its inline onProgress switch for listeners: [new ConsoleProgressListener()] — the first adopter, proving the pattern end-to-end. Same progress output. - lib/verdictIcon.ts: shared verdict->glyph helper, replacing copy-pasted ternaries in the listener and evaluatorLoop. Backward-compatible: onProgress fires exactly as before; listeners is additive; the SDK's own ProgressEvent type is untouched. New tests pin the event->hook mapping and that a throwing listener is isolated. Co-Authored-By: Claude Opus 4.8 (1M context) --- core/src/execute/evaluatorLoop.ts | 7 +- core/src/execute/runAll.ts | 39 ++++++++-- core/src/execute/runListener.ts | 75 ++++++++++++++++++ core/src/execute/types.ts | 11 +++ core/src/lib/verdictIcon.ts | 6 ++ core/tests/runListener.test.ts | 77 +++++++++++++++++++ runners/cli/src/commands/run.ts | 10 +-- .../cli/src/lib/consoleProgressListener.ts | 18 +++++ 8 files changed, 224 insertions(+), 19 deletions(-) create mode 100644 core/src/execute/runListener.ts create mode 100644 core/src/lib/verdictIcon.ts create mode 100644 core/tests/runListener.test.ts create mode 100644 runners/cli/src/lib/consoleProgressListener.ts diff --git a/core/src/execute/evaluatorLoop.ts b/core/src/execute/evaluatorLoop.ts index 505d6e44..02f48a7a 100644 --- a/core/src/execute/evaluatorLoop.ts +++ b/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); diff --git a/core/src/execute/runAll.ts b/core/src/execute/runAll.ts index c0430ab0..d0b6c449 100644 --- a/core/src/execute/runAll.ts +++ b/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,7 +43,13 @@ export async function runAll( config: RunConfig, options?: RunAllOptions ): Promise { - const notify = options?.onProgress ?? (() => {}); + // 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)); + }; const attackModel = resolveModel(config.attackerLlm); // Single source of truth for the judge LLM: explicit judge model, else the @@ -74,6 +83,12 @@ export async function runAll( const evaluatorResults: EvaluatorResult[] = []; let stopReason: string | undefined; + // evaluatorCount is the attack-evaluator count; MCP baseline pre-flight scans + // (run below) contribute extra results not reflected here. + notifyListeners(listeners, (listener) => + listener.onRunStart?.({ evaluatorCount: ordered.length }) + ); + try { // ── MCP pre-flight scans (always run for MCP targets) ────────────── if (isMcp && mcpTarget) { @@ -104,6 +119,11 @@ export async function runAll( }); evaluatorResults.push(...loop.evaluatorResults); stopReason = loop.stopReason; + } catch (err) { + // Terminal error signal so listeners that opened state in onRunStart can + // finalize (pairs with onRunStart, symmetric with onRunFinish). + notifyListeners(listeners, (listener) => listener.onRunError?.({ error: err })); + throw err; } finally { if (mcpTarget) await mcpTarget.close().catch(() => {}); } @@ -113,6 +133,9 @@ export async function runAll( if (stopReason) { (report as UnifiedRunReport & { stopReason?: string }).stopReason = stopReason; } + + notifyListeners(listeners, (listener) => listener.onRunFinish?.(report)); + return report; } diff --git a/core/src/execute/runListener.ts b/core/src/execute/runListener.ts new file mode 100644 index 00000000..06818c70 --- /dev/null +++ b/core/src/execute/runListener.ts @@ -0,0 +1,75 @@ +// 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 = Omit, "type">; + +export interface RunListener { + /** Fired once before the first evaluator runs. */ + 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; + /** Fired when a non-retryable error halts the run gracefully (partial report). */ + onRunStopped?(info: Payload<"run_stopped">): void; + /** Fired if the run throws an unexpected error (terminal; pairs with onRunStart). */ + onRunError?(info: { error: unknown }): void; + /** Fired once after the report is assembled (terminal; pairs with onRunStart). */ + onRunFinish?(report: UnifiedRunReport): void; +} + +/** + * Invoke a hook on every listener with error isolation: a listener that throws is + * logged and skipped so a buggy reporter/telemetry adapter can never abort the + * run itself. Centralizes the fan-out so every notification site is uniform. + */ +export function notifyListeners( + listeners: readonly RunListener[], + invoke: (listener: RunListener) => void +): void { + for (const listener of listeners) { + try { + invoke(listener); + } catch (err) { + log.warn(`RunListener threw and was skipped: ${err instanceof Error ? err.message : 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; + } + } +} diff --git a/core/src/execute/types.ts b/core/src/execute/types.ts index 1465def0..e8795be0 100644 --- a/core/src/execute/types.ts +++ b/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 }; diff --git a/core/src/lib/verdictIcon.ts b/core/src/lib/verdictIcon.ts new file mode 100644 index 00000000..a6be1c5f --- /dev/null +++ b/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" ? "✗" : "⚠"; +} diff --git a/core/tests/runListener.test.ts b/core/tests/runListener.test.ts new file mode 100644 index 00000000..e5cc2672 --- /dev/null +++ b/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"]); +}); diff --git a/runners/cli/src/commands/run.ts b/runners/cli/src/commands/run.ts index 96f192a0..92edff53 100644 --- a/runners/cli/src/commands/run.ts +++ b/runners/cli/src/commands/run.ts @@ -9,6 +9,7 @@ import { parseRunConfig } from "@keyvaluesystems/agent-opfor-core/config/schema. import { normalizeEffort } from "@keyvaluesystems/agent-opfor-core/execute/effortCompat.js"; import { runSetupAndWrite } from "./setup.js"; import { ensureOpforDirs, OPFOR_DIR, OPFOR_REPORTS_DIR } from "../lib/artifacts.js"; +import { ConsoleProgressListener } from "../lib/consoleProgressListener.js"; export function registerRunCommand(program: Command): void { program @@ -96,14 +97,7 @@ export function registerRunCommand(program: Command): void { await ensureOpforDirs(); const report = await runAll(runConfig, { outputDir: path.resolve(OPFOR_DIR), - onProgress: (event) => { - if (event.type === "evaluator_start") { - log.info(`\n▶ ${event.evaluatorName}`); - } else if (event.type === "attack_done") { - const icon = event.verdict === "PASS" ? "✓" : event.verdict === "FAIL" ? "✗" : "⚠"; - process.stdout.write(` ${icon}`); - } - }, + listeners: [new ConsoleProgressListener()], }); log.info("\n\nWriting report..."); diff --git a/runners/cli/src/lib/consoleProgressListener.ts b/runners/cli/src/lib/consoleProgressListener.ts new file mode 100644 index 00000000..005b10f6 --- /dev/null +++ b/runners/cli/src/lib/consoleProgressListener.ts @@ -0,0 +1,18 @@ +import type { RunListener } from "@keyvaluesystems/agent-opfor-core/execute/runListener.js"; +import { log } from "@keyvaluesystems/agent-opfor-core/lib/logger.js"; +import { verdictIcon } from "@keyvaluesystems/agent-opfor-core/lib/verdictIcon.js"; + +/** + * Prints run progress to the terminal — a header per evaluator and a ✓/✗/⚠ mark + * per completed attack. Implemented as a RunListener so the CLI consumes engine + * lifecycle events through the SPI rather than an ad-hoc onProgress switch. + */ +export class ConsoleProgressListener implements RunListener { + onEvaluatorStart(info: { evaluatorId: string; evaluatorName: string }): void { + log.info(`\n▶ ${info.evaluatorName}`); + } + + onAttackDone(info: { attackId: string; verdict: "PASS" | "FAIL" | "ERROR" }): void { + process.stdout.write(` ${verdictIcon(info.verdict)}`); + } +} From d62623ad9f0b2a8956a5c171fe6adc290e2ec75d Mon Sep 17 00:00:00 2001 From: Prasanth Date: Thu, 2 Jul 2026 22:44:16 +0530 Subject: [PATCH 2/4] fix: make the run-listener lifecycle fully symmetric MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback on the RunListener wiring: - onRunStart now fires at the true start, before the MCP connect and trace curation, so watchers see the run begin during those (possibly slow) steps. - The MCP connect, trace curation, baseline scans, attack loop, AND report building now all run inside the try, so onRunError covers a throw from any of them — onRunStart is always paired with exactly one terminal hook (previously buildReport/onRunFinish sat outside the try and could be skipped on a throw). - As a side effect the finally now closes the MCP target even when listTools() throws after a successful connect (a latent leak). - Clarify the lifecycle contract in the interface docs: onRunStopped is a non-terminal notice (onRunFinish still follows with the partial report); onRunFinish and onRunError are the terminal hooks. Behavior-preserving for the success and graceful-stop paths (smoke/equivalence green); only the error path and event ordering change. Co-Authored-By: Claude Opus 4.8 (1M context) --- core/src/execute/runAll.ts | 52 ++++++++++++++++++--------------- core/src/execute/runListener.ts | 13 ++++++--- 2 files changed, 38 insertions(+), 27 deletions(-) diff --git a/core/src/execute/runAll.ts b/core/src/execute/runAll.ts index d0b6c449..b157fdc4 100644 --- a/core/src/execute/runAll.ts +++ b/core/src/execute/runAll.ts @@ -65,24 +65,18 @@ export async function runAll( config.selection.dependsOn ); - // For MCP targets, connect once and discover tools + // Declared before the try so the finally can always close the MCP target — + // including when listTools() throws after a successful connect. let mcpTarget: Awaited> | 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`); - } - - // 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); - const ordered = topoSortEvaluators(evaluators); const evaluatorResults: EvaluatorResult[] = []; let stopReason: string | undefined; + // Fire onRunStart at the true start — before the (possibly slow) MCP connect and + // trace curation — so watchers see the run begin during those steps. Everything + // after it runs inside the try, so onRunStart always pairs with a terminal hook. // evaluatorCount is the attack-evaluator count; MCP baseline pre-flight scans // (run below) contribute extra results not reflected here. notifyListeners(listeners, (listener) => @@ -90,6 +84,17 @@ export async function runAll( ); try { + // 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); + // ── MCP pre-flight scans (always run for MCP targets) ────────────── if (isMcp && mcpTarget) { evaluatorResults.push( @@ -119,24 +124,25 @@ export async function runAll( }); evaluatorResults.push(...loop.evaluatorResults); 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 so listeners that opened state in onRunStart can - // finalize (pairs with onRunStart, symmetric with onRunFinish). + // Terminal error signal — pairs with onRunStart on the throwing path (report + // building, MCP connect, and the loop are all covered). 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; - } - - notifyListeners(listeners, (listener) => listener.onRunFinish?.(report)); - - return report; } // --------------------------------------------------------------------------- diff --git a/core/src/execute/runListener.ts b/core/src/execute/runListener.ts index 06818c70..76b09ba3 100644 --- a/core/src/execute/runListener.ts +++ b/core/src/execute/runListener.ts @@ -12,17 +12,22 @@ import { log } from "../lib/logger.js"; type Payload = Omit, "type">; export interface RunListener { - /** Fired once before the first evaluator runs. */ + // Lifecycle: onRunStart fires once at the start and is always paired with exactly + // one terminal hook — onRunFinish (the run produced a report, complete or partial) + // or onRunError (the run threw). onRunStopped is a non-terminal notice: onRunFinish + // still follows it with the partial report. + + /** Fired once at the start, before the target connects. */ 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; - /** Fired when a non-retryable error halts the run gracefully (partial report). */ + /** Non-terminal notice that a non-retryable error cut the run short; onRunFinish still follows. */ onRunStopped?(info: Payload<"run_stopped">): void; - /** Fired if the run throws an unexpected error (terminal; pairs with onRunStart). */ + /** Terminal: the run threw. onRunFinish does NOT fire on this path. */ onRunError?(info: { error: unknown }): void; - /** Fired once after the report is assembled (terminal; pairs with onRunStart). */ + /** Terminal: fired once with the final report (complete, or partial after a stop). */ onRunFinish?(report: UnifiedRunReport): void; } From 76025a4d17b1a648b75ab1094338d1961f9695bf Mon Sep 17 00:00:00 2001 From: Prasanth Date: Thu, 2 Jul 2026 23:31:43 +0530 Subject: [PATCH 3/4] fix: cover run setup with the listener lifecycle and isolate async listeners MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback on the RunListener wiring: - runAll now runs model + evaluator resolution and the topo-sort INSIDE the try, so a failure there reaches onRunError too (previously they ran before the try and bypassed the terminal hook). onRunStart still fires once the evaluator count is known, just before the MCP connect; a setup failure throws before it, so listeners get onRunError with no preceding onRunStart — documented in the contract. - notifyListeners now isolates async listeners: a hook that rejects its returned promise (not just a synchronous throw) is caught and skipped, so a buggy reporter/telemetry adapter still can't abort the run. The warning is actionable ("fix the listener implementation"). Behavior-preserving for successful and graceful-stop runs (smoke/equivalence green). Co-Authored-By: Claude Opus 4.8 (1M context) --- core/src/execute/runAll.ts | 67 +++++++++++++++++---------------- core/src/execute/runListener.ts | 38 +++++++++++++------ 2 files changed, 60 insertions(+), 45 deletions(-) diff --git a/core/src/execute/runAll.ts b/core/src/execute/runAll.ts index b157fdc4..987ef01e 100644 --- a/core/src/execute/runAll.ts +++ b/core/src/execute/runAll.ts @@ -51,39 +51,40 @@ export async function runAll( notifyListeners(listeners, (listener) => dispatchProgress(listener, event)); }; - 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 - ); - - // Declared before the try so the finally can always close the MCP target — - // including when listTools() throws after a successful connect. + // 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> | null = null; - let tools: ToolInfo[] = []; - - const ordered = topoSortEvaluators(evaluators); - const evaluatorResults: EvaluatorResult[] = []; - let stopReason: string | undefined; - - // Fire onRunStart at the true start — before the (possibly slow) MCP connect and - // trace curation — so watchers see the run begin during those steps. Everything - // after it runs inside the try, so onRunStart always pairs with a terminal hook. - // evaluatorCount is the attack-evaluator count; MCP baseline pre-flight scans - // (run below) contribute extra results not reflected here. - notifyListeners(listeners, (listener) => - listener.onRunStart?.({ evaluatorCount: ordered.length }) - ); 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 + ); + + 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 }) + ); + // For MCP targets, connect once and discover tools. if (isMcp) { mcpTarget = await createMcpTarget(config.target as import("./types.js").McpTargetConfig); @@ -123,7 +124,7 @@ 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); @@ -136,8 +137,8 @@ export async function runAll( notifyListeners(listeners, (listener) => listener.onRunFinish?.(report)); return report; } catch (err) { - // Terminal error signal — pairs with onRunStart on the throwing path (report - // building, MCP connect, and the loop are all covered). + // 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 { diff --git a/core/src/execute/runListener.ts b/core/src/execute/runListener.ts index 76b09ba3..b8edbfec 100644 --- a/core/src/execute/runListener.ts +++ b/core/src/execute/runListener.ts @@ -12,12 +12,13 @@ import { log } from "../lib/logger.js"; type Payload = Omit, "type">; export interface RunListener { - // Lifecycle: onRunStart fires once at the start and is always paired with exactly - // one terminal hook — onRunFinish (the run produced a report, complete or partial) - // or onRunError (the run threw). onRunStopped is a non-terminal notice: onRunFinish - // still follows it with the partial report. + // 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. */ + /** 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; @@ -25,26 +26,39 @@ export interface RunListener { 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. onRunFinish does NOT fire on this path. */ + /** 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 is - * logged and skipped so a buggy reporter/telemetry adapter can never abort the - * run itself. Centralizes the fan-out so every notification site is uniform. + * 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 + invoke: (listener: RunListener) => void | Promise ): void { for (const listener of listeners) { try { - invoke(listener); + const result = invoke(listener); + if (result && typeof (result as Promise).then === "function") { + void (result as Promise).then(undefined, warnListenerFailed); + } } catch (err) { - log.warn(`RunListener threw and was skipped: ${err instanceof Error ? err.message : err}`); + warnListenerFailed(err); } } } From 272d7fcb9cd92480bda7dbd810449b45da53e451 Mon Sep 17 00:00:00 2001 From: Prasanth Date: Fri, 3 Jul 2026 08:21:47 +0530 Subject: [PATCH 4/4] refactor: adopt the run-listener in the mcp runner and expose it in the sdk Extend the RunListener adoption beyond the CLI, per review: - runners/mcp: the MCP runner had the same inline onProgress switch this PR replaced in the CLI. Convert it to a RunListener (onEvaluatorStart / onEvaluatorDone) so both runners consume engine events through the SPI. - runners/sdk: run() only forwarded onProgress (per-attack events). Add a listeners?: RunListener[] option, forward it to runAll, and re-export the RunListener type so SDK users can observe the full run lifecycle (onRunStart / onRunFinish / onRunError), not just per-attack progress. Additive and backward-compatible: onProgress is unchanged on every path. Co-Authored-By: Claude Opus 4.8 (1M context) --- runners/mcp/src/index.ts | 14 +++++++------- runners/sdk/src/index.ts | 1 + runners/sdk/src/run.ts | 1 + runners/sdk/src/types.ts | 12 ++++++++++++ 4 files changed, 21 insertions(+), 7 deletions(-) diff --git a/runners/mcp/src/index.ts b/runners/mcp/src/index.ts index 5cc1040f..e6db5f96 100644 --- a/runners/mcp/src/index.ts +++ b/runners/mcp/src/index.ts @@ -352,13 +352,13 @@ server.tool( ]; const report = await runAll(config, { - onProgress: (event) => { - if (event.type === "evaluator_start") { - lines.push(`\n▶ ${event.evaluatorName}`); - } else if (event.type === "evaluator_done") { - lines.push(` ${event.passed} passed, ${event.failed} failed, ${event.errors} errors`); - } - }, + listeners: [ + { + onEvaluatorStart: (e) => lines.push(`\n▶ ${e.evaluatorName}`), + onEvaluatorDone: (e) => + lines.push(` ${e.passed} passed, ${e.failed} failed, ${e.errors} errors`), + }, + ], }); const outputDir = path.resolve(args.output_dir ?? path.dirname(args.config_path)); diff --git a/runners/sdk/src/index.ts b/runners/sdk/src/index.ts index 7f8ebfe7..0b0bc97f 100644 --- a/runners/sdk/src/index.ts +++ b/runners/sdk/src/index.ts @@ -19,6 +19,7 @@ export type { RunOptions, RunResults, ProgressEvent, + RunListener, // Target configuration TargetConfig, diff --git a/runners/sdk/src/run.ts b/runners/sdk/src/run.ts index 2d69fca9..5573a0d1 100644 --- a/runners/sdk/src/run.ts +++ b/runners/sdk/src/run.ts @@ -29,6 +29,7 @@ export async function run(options: RunOptions): Promise { const coreReport = await runAll(runConfig, { onProgress: options.onProgress, + listeners: options.listeners, }); return transformReport(coreReport); diff --git a/runners/sdk/src/types.ts b/runners/sdk/src/types.ts index 68ea6d33..8d3ec9f3 100644 --- a/runners/sdk/src/types.ts +++ b/runners/sdk/src/types.ts @@ -1,3 +1,9 @@ +import type { RunListener } from "@keyvaluesystems/agent-opfor-core/execute/runListener.js"; + +// Re-export the run lifecycle observer so SDK users can implement onRunStart / +// onRunFinish / onRunError, not just per-attack onProgress events. +export type { RunListener }; + // --------------------------------------------------------------------------- // Target Configuration // --------------------------------------------------------------------------- @@ -82,6 +88,12 @@ export interface RunOptions { apiKey?: string; onProgress?: (event: ProgressEvent) => void; + + /** + * Run lifecycle observers. Receive the same per-attack events as onProgress plus + * run-level onRunStart / onRunFinish / onRunError hooks. + */ + listeners?: RunListener[]; } export type ProgressEvent =