Tier 3 agent runner: streaming, compaction, steering

src/adapters/roleBindings.ts

@@ -5,6 +5,7 @@ import type { Actor, Critic } from "./agents.js";
 import { RunnerActor, RunnerCritic } from "./runnerRoles.js";
 import type { ActorPromptTemplates, CriticPromptTemplates } from "./prompts.js";
 import { instrumentRunner, type RunnerCallSink } from "../observability/instrument.js";
+import type { RunnerActivityEvent } from "../observability/events.js";
 
 /**
  * Configuration -> roles wiring (Task 13.3).
@@ -38,6 +39,8 @@ export interface CreateRolesOptions {
   log?: (line: string) => void;
   /** when set, every runner call emits an event attributed to its role (M5) */
   onRunnerCall?: RunnerCallSink;
+  /** when set, mid-call runner activity (tool calls) streams to this sink */
+  onActivity?: (e: RunnerActivityEvent) => void;
   /** cooperative cancellation, threaded into the actor/critic runner calls */
   signal?: AbortSignal;
 }
@@ -99,13 +102,15 @@ export function createRoles(
     ...(opts.cwd ? { cwd: opts.cwd } : {}),
     ...(opts.log ? { log: opts.log } : {}),
     ...(opts.signal ? { signal: opts.signal } : {}),
+    ...(opts.onActivity ? { onActivity: opts.onActivity } : {}),
   });
 
   const critic = new RunnerCritic(instrumentedCritic, {
     ...(opts.criticPrompts ? { prompts: opts.criticPrompts } : {}),
     ...(opts.cwd ? { cwd: opts.cwd } : {}),
     ...(opts.log ? { log: opts.log } : {}),
     ...(opts.signal ? { signal: opts.signal } : {}),
+    ...(opts.onActivity ? { onActivity: opts.onActivity } : {}),
   });
 
   return { actor, critic };

src/adapters/runnerRoles.ts

@@ -3,8 +3,9 @@ import { PlanSchema } from "../schemas/plan.js";
 import type { TaskSpec } from "../schemas/plan.js";
 import type { Finding } from "../schemas/critic.js";
 import type { TaskArtifactBundle } from "../schemas/artifact.js";
-import type { AgentRunner } from "../runners/agentRunner.js";
+import type { AgentRunner, RunnerActivity } from "../runners/agentRunner.js";
 import { parseLastValidJson, type JsonParseResult } from "../engine/jsonExtract.js";
+import type { RoleName, RunnerActivityEvent } from "../observability/events.js";
 import type {
   Actor,
   ActorBuildResult,
@@ -66,6 +67,8 @@ export interface RunnerActorOptions {
   log?: (line: string) => void;
   /** cooperative cancellation, threaded into every runner call */
   signal?: AbortSignal;
+  /** sink for mid-call runner activity (sub-step streaming), if any */
+  onActivity?: (e: RunnerActivityEvent) => void;
 }
 
 export interface RunnerCriticOptions {
@@ -74,6 +77,34 @@ export interface RunnerCriticOptions {
   log?: (line: string) => void;
   /** cooperative cancellation, threaded into every runner call */
   signal?: AbortSignal;
+  /** sink for mid-call runner activity (sub-step streaming), if any */
+  onActivity?: (e: RunnerActivityEvent) => void;
+}
+
+/**
+ * Builds the per-call `onEvent` sink that enriches a runner's vendor-neutral
+ * {@link RunnerActivity} with the role + runner identity before forwarding it
+ * to the role's activity sink. Returns undefined when no sink is configured, so
+ * runners that stream skip the work entirely.
+ */
+function activityForwarder(
+  runner: AgentRunner,
+  role: RoleName,
+  onActivity: ((e: RunnerActivityEvent) => void) | undefined,
+): ((a: RunnerActivity) => void) | undefined {
+  if (!onActivity) return undefined;
+  return (a: RunnerActivity) =>
+    onActivity({
+      role,
+      runnerId: runner.profile.id,
+      model: runner.profile.model,
+      phase: a.phase,
+      ...(a.toolName !== undefined ? { toolName: a.toolName } : {}),
+      ...(a.toolCallId !== undefined ? { toolCallId: a.toolCallId } : {}),
+      ...(a.isError !== undefined ? { isError: a.isError } : {}),
+      ...(a.turn !== undefined ? { turn: a.turn } : {}),
+      at: a.at,
+    });
 }
 
 /** Actor role backed by an AgentRunner + prompt templates. */
@@ -84,6 +115,8 @@ export class RunnerActor implements Actor {
   private readonly repairOnce: boolean;
   private readonly log: ((line: string) => void) | undefined;
   private readonly signal: AbortSignal | undefined;
+  /** per-call activity sink, enriched with role identity (undefined = off) */
+  private readonly onEvent: ((a: RunnerActivity) => void) | undefined;
 
   constructor(runner: AgentRunner, opts: RunnerActorOptions = {}) {
     this.runner = runner;
@@ -92,6 +125,7 @@ export class RunnerActor implements Actor {
     this.repairOnce = opts.repairOnce ?? true;
     this.log = opts.log;
     this.signal = opts.signal;
+    this.onEvent = activityForwarder(runner, "actor", opts.onActivity);
   }
 
   async draftPlan(goal: string, feedback?: Finding[]): Promise<PlanDraftResult> {
@@ -128,6 +162,7 @@ export class RunnerActor implements Actor {
       cwd: cwd ?? this.cwd,
       system: this.prompts.system,
       ...(this.signal ? { signal: this.signal } : {}),
+      ...(this.onEvent ? { onEvent: this.onEvent } : {}),
     });
     return { text: res.text, quotaExhausted: res.quotaExhausted };
   }
@@ -144,6 +179,7 @@ export class RunnerActor implements Actor {
       cwd,
       system: this.prompts.system,
       ...(this.signal ? { signal: this.signal } : {}),
+      ...(this.onEvent ? { onEvent: this.onEvent } : {}),
     });
     if (res.quotaExhausted) {
       throw new RunnerRoleError(`Actor ${what} failed: runner quota exhausted.`, {
@@ -159,6 +195,7 @@ export class RunnerActor implements Actor {
         cwd,
         system: this.prompts.system,
         ...(this.signal ? { signal: this.signal } : {}),
+        ...(this.onEvent ? { onEvent: this.onEvent } : {}),
       });
       if (res.quotaExhausted) {
         throw new RunnerRoleError(`Actor ${what} failed: runner quota exhausted.`, {
@@ -181,12 +218,15 @@ export class RunnerCritic implements Critic {
   private readonly prompts: CriticPromptTemplates;
   private readonly cwd: string;
   private readonly signal: AbortSignal | undefined;
+  /** per-call activity sink, enriched with role identity (undefined = off) */
+  private readonly onEvent: ((a: RunnerActivity) => void) | undefined;
 
   constructor(runner: AgentRunner, opts: RunnerCriticOptions = {}) {
     this.runner = runner;
     this.prompts = opts.prompts ?? DEFAULT_CRITIC_PROMPTS;
     this.cwd = opts.cwd ?? ".";
     this.signal = opts.signal;
+    this.onEvent = activityForwarder(runner, "critic", opts.onActivity);
   }
 
   /**
@@ -205,6 +245,7 @@ export class RunnerCritic implements Critic {
       cwd: cwd ?? this.cwd,
       system: this.prompts.system,
       ...(this.signal ? { signal: this.signal } : {}),
+      ...(this.onEvent ? { onEvent: this.onEvent } : {}),
     });
     return { text: res.text, quotaExhausted: res.quotaExhausted };
   }

src/observability/events.ts

@@ -32,10 +32,29 @@ export interface RunnerCallEvent {
   at: string;
 }
 
+/**
+ * A mid-call activity signal from a runner's inner agentic loop (sub-step
+ * streaming), attributed to the role and runner that produced it. Emitted as
+ * the model calls tools so a monitor can show live progress within a single
+ * runner call, rather than waiting for the final result.
+ */
+export interface RunnerActivityEvent {
+  role: RoleName;
+  runnerId: string;
+  model: string;
+  phase: "turn_start" | "tool_start" | "tool_end";
+  toolName?: string;
+  toolCallId?: string;
+  isError?: boolean;
+  turn?: number;
+  at: string;
+}
+
 export const EVENT_TYPES = {
   sessionStarted: "session_started",
   planReviewed: "plan_reviewed",
   runnerCall: "runner_call",
+  runnerActivity: "runner_activity",
   sessionFinished: "session_finished",
   integration: "integration",
   publish: "publish",

src/runners/agentRunner.ts

@@ -56,6 +56,40 @@ export interface RunRequest {
    * long model call returns promptly instead of waiting for the runner timeout.
    */
   signal?: AbortSignal;
+  /**
+   * Optional sink for mid-call activity (sub-step streaming). Backends that run
+   * an inner agentic loop (e.g. the native agent runner) emit a {@link RunnerActivity}
+   * as the model calls tools, so the engine can surface live progress instead of
+   * a single opaque result. Single-shot backends (CLI/HTTP) simply ignore it.
+   */
+  onEvent?: (activity: RunnerActivity) => void;
+  /**
+   * Optional steering registration. Called once at the start of a run with a
+   * `steer(text)` function bound to the live inner loop; the caller (a human
+   * "nudge", a supervisor) may invoke it at any time while the run is in flight
+   * to inject guidance that takes effect after the current turn. Backends with
+   * no inner loop ignore it.
+   */
+  steering?: (steer: (text: string) => void) => void;
+}
+
+/**
+ * A mid-call progress signal from a runner's inner loop. Vendor-neutral and
+ * role-agnostic: the runner reports WHAT happened (a tool started/finished, a
+ * turn began); the role layer enriches it with the role/runner identity before
+ * it reaches the event stream.
+ */
+export interface RunnerActivity {
+  phase: "turn_start" | "tool_start" | "tool_end";
+  /** present for tool_start/tool_end */
+  toolName?: string;
+  /** correlates a tool_start with its tool_end */
+  toolCallId?: string;
+  /** present on tool_end: whether the tool reported an error */
+  isError?: boolean;
+  /** 1-based turn number, present on turn_start */
+  turn?: number;
+  at: string;
 }
 
 export interface RunResult {