diff --git a/src/adapters/roleBindings.ts b/src/adapters/roleBindings.ts
index 7f932d4..08740a4 100644
--- a/src/adapters/roleBindings.ts
+++ b/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,6 +102,7 @@ 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, {
@@ -106,6 +110,7 @@ export function createRoles(
     ...(opts.cwd ? { cwd: opts.cwd } : {}),
     ...(opts.log ? { log: opts.log } : {}),
     ...(opts.signal ? { signal: opts.signal } : {}),
+    ...(opts.onActivity ? { onActivity: opts.onActivity } : {}),
   });
 
   return { actor, critic };
diff --git a/src/adapters/runnerRoles.ts b/src/adapters/runnerRoles.ts
index f9d57ea..8389a06 100644
--- a/src/adapters/runnerRoles.ts
+++ b/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 };
   }
diff --git a/src/observability/events.ts b/src/observability/events.ts
index dcb6bc9..c248cee 100644
--- a/src/observability/events.ts
+++ b/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",
diff --git a/src/runners/agentRunner.ts b/src/runners/agentRunner.ts
index 9543849..c80b0ea 100644
--- a/src/runners/agentRunner.ts
+++ b/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 {
diff --git a/src/runners/piAgentRunner.ts b/src/runners/piAgentRunner.ts
index 5484101..2b8e92d 100644
--- a/src/runners/piAgentRunner.ts
+++ b/src/runners/piAgentRunner.ts
@@ -9,7 +9,15 @@ import type {
 import { redactAndTruncate } from "../engine/redaction.js";
 
 import { Agent } from "@earendil-works/pi-agent-core";
-import type { AgentEvent, AgentTool, AgentToolResult } from "@earendil-works/pi-agent-core";
+import type { AgentEvent, AgentMessage, AgentTool, AgentToolResult } from "@earendil-works/pi-agent-core";
+import {
+  convertToLlm,
+  createCompactionSummaryMessage,
+  estimateContextTokens,
+  estimateTokens,
+  generateSummary,
+  shouldCompact,
+} from "@earendil-works/pi-agent-core";
 import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
 import type { ExecutionEnv } from "@earendil-works/pi-agent-core/node";
 import {
@@ -119,6 +127,107 @@ const SafetySchema = z
   .strict()
   .default({ confineToCwd: true });
 
+/**
+ * Context-compaction policy for the inner loop. Long agent runs accumulate big
+ * tool transcripts that can exceed the model's context window; when triggered,
+ * the older middle of the transcript is summarized (via pi's `generateSummary`)
+ * and replaced with a single summary message, keeping the initial prompt and a
+ * recent tail. Off by default per call only in the sense that it never fires
+ * until the context is actually large.
+ */
+const CompactionOptionsSchema = z
+  .object({
+    /** enable automatic compaction decisions */
+    enabled: z.boolean().default(true),
+    /**
+     * Force compaction once the estimated context exceeds this many tokens.
+     * When unset, pi's `shouldCompact` decides against the model's context
+     * window (this knob mainly exists for deterministic testing/tuning).
+     */
+    thresholdTokens: z.number().int().positive().optional(),
+    /** approximate recent-context tokens to keep after compaction */
+    keepRecentTokens: z.number().int().positive().default(4_000),
+    /** tokens reserved for the summary prompt + output */
+    reserveTokens: z.number().int().positive().default(2_000),
+  })
+  .strict()
+  .default({ enabled: true });
+
+export type CompactionOptions = z.infer<typeof CompactionOptionsSchema>;
+
+/**
+ * Chooses where to cut the transcript for compaction: walk back from the end
+ * accumulating recent tokens until `keepRecentTokens` is met, then snap BACK to
+ * the assistant message that starts that turn, so the retained tail begins at a
+ * turn boundary (never a dangling toolResult whose tool call was summarized
+ * away). Returns the index where the retained tail begins (>= 1, keeping the
+ * initial prompt at messages[0]).
+ */
+export function findCompactionCut(messages: AgentMessage[], keepRecentTokens: number): number {
+  let acc = 0;
+  let cut = messages.length;
+  for (let i = messages.length - 1; i >= 1; i--) {
+    acc += estimateTokens(messages[i] as AgentMessage);
+    cut = i;
+    if (acc >= keepRecentTokens) break;
+  }
+  while (cut > 1 && (messages[cut] as AgentMessage).role !== "assistant") cut--;
+  return cut;
+}
+
+/**
+ * Builds an Agent `transformContext` that compacts the transcript when it grows
+ * past the configured budget, using pi's message-level summarization. Best
+ * effort: any failure (or nothing safe to summarize) returns the messages
+ * unchanged, so compaction can never break a run. Exported for direct testing.
+ */
+export function createCompactionTransform(
+  opts: CompactionOptions,
+  model: Model<any>,
+  apiKey: string | undefined,
+  thinkingLevel: PiAgentRunnerOptions["thinkingLevel"] = "off",
+): ((messages: AgentMessage[], signal?: AbortSignal) => Promise<AgentMessage[]>) | undefined {
+  if (!opts.enabled) return undefined;
+  const settings = {
+    enabled: true,
+    reserveTokens: opts.reserveTokens,
+    keepRecentTokens: opts.keepRecentTokens,
+  };
+  return async (messages, signal) => {
+    // Need at least: prompt + one full assistant/toolResult turn + something to keep.
+    if (messages.length < 4) return messages;
+
+    const est = estimateContextTokens(messages);
+    const trigger =
+      opts.thresholdTokens !== undefined
+        ? est.tokens > opts.thresholdTokens
+        : shouldCompact(est.tokens, model.contextWindow, settings);
+    if (!trigger) return messages;
+
+    const cut = findCompactionCut(messages, opts.keepRecentTokens);
+    // Keep messages[0] (the task prompt); summarize the middle; keep the tail.
+    if (cut <= 1 || cut >= messages.length) return messages; // nothing safe to drop
+
+    const toSummarize = messages.slice(1, cut);
+    const tail = messages.slice(cut);
+    const result = await generateSummary(
+      toSummarize,
+      model,
+      opts.reserveTokens,
+      apiKey ?? "",
+      undefined,
+      signal,
+      undefined,
+      undefined,
+      thinkingLevel,
+    );
+    if (!result.ok) return messages; // best effort: keep full context on failure
+
+    const summary = createCompactionSummaryMessage(result.value, est.tokens, new Date().toISOString());
+    return [messages[0] as AgentMessage, summary, ...tail];
+  };
+}
+
 export const PiAgentRunnerOptionsSchema = z
   .object({
     /** provider id for pi-ai model resolution, e.g. "anthropic", "openai", "google" */
@@ -135,6 +244,8 @@ export const PiAgentRunnerOptionsSchema = z
     tools: z.array(z.enum(PI_AGENT_TOOL_NAMES)).optional(),
     /** tool-call permission policy (path confinement + bash denylist) */
     safety: SafetySchema,
+    /** context compaction policy for long inner loops */
+    compaction: CompactionOptionsSchema,
     /** rolling cap on the captured final answer text */
     maxOutputChars: z.number().int().positive().default(2_000_000),
   })
@@ -338,6 +449,16 @@ export class PiAgentRunner implements AgentRunner {
 
     const apiKey = this.apiKey();
 
+    // Context compaction (pi transformContext): summarize the older transcript
+    // when the inner loop grows past budget. When enabled we also adopt pi's
+    // convertToLlm so the inserted compactionSummary message renders correctly.
+    const transformContext = createCompactionTransform(
+      this.opts.compaction,
+      model,
+      apiKey,
+      this.opts.thinkingLevel,
+    );
+
     // Permission gate (pi's beforeToolCall hook): confine file tools to the
     // worktree and apply the optional bash denylist. A blocked call becomes an
     // error tool result the model sees and can recover from.
@@ -354,6 +475,7 @@ export class PiAgentRunner implements AgentRunner {
         tools: this.buildTools(env),
       },
       ...(this.deps.streamFn ? { streamFn: this.deps.streamFn } : {}),
+      ...(transformContext ? { transformContext, convertToLlm } : {}),
       getApiKey: () => apiKey,
       beforeToolCall: async ({ toolCall, args }) => {
         const name = (toolCall as { name: string }).name as PiAgentToolName;
@@ -382,6 +504,7 @@ export class PiAgentRunner implements AgentRunner {
     let abortedForLimit = false;
     const usage = { input: 0, output: 0, total: 0 };
     let lastQuotaText = "";
+    const emit = req.onEvent;
     const unsubscribe = agent.subscribe((event: AgentEvent) => {
       if (event.type === "turn_start") {
         if (turns >= this.opts.maxTurns) {
@@ -390,6 +513,22 @@ export class PiAgentRunner implements AgentRunner {
           return;
         }
         turns++;
+        emit?.({ phase: "turn_start", turn: turns, at: new Date().toISOString() });
+      } else if (event.type === "tool_execution_start") {
+        emit?.({
+          phase: "tool_start",
+          toolName: event.toolName,
+          toolCallId: event.toolCallId,
+          at: new Date().toISOString(),
+        });
+      } else if (event.type === "tool_execution_end") {
+        emit?.({
+          phase: "tool_end",
+          toolName: event.toolName,
+          toolCallId: event.toolCallId,
+          isError: event.isError,
+          at: new Date().toISOString(),
+        });
       } else if (event.type === "message_end" && event.message.role === "assistant") {
         const m = event.message as AssistantMessage;
         usage.input += m.usage?.input ?? 0;
@@ -401,6 +540,15 @@ export class PiAgentRunner implements AgentRunner {
 
     let cancelled = false;
     let timedOut = false;
+    let steerCount = 0;
+    // Steering: hand the caller a `steer(text)` bound to this live loop. A
+    // nudge is queued and injected after the current turn (pi's steering
+    // semantics), so a human/supervisor can redirect a long run without
+    // cancelling it.
+    req.steering?.((text: string) => {
+      steerCount++;
+      agent.steer({ role: "user", content: text, timestamp: Date.now() });
+    });
     const onAbort = (): void => {
       cancelled = true;
       agent.abort();
@@ -448,6 +596,7 @@ export class PiAgentRunner implements AgentRunner {
         turns,
         abortedForLimit,
         blockedToolCalls,
+        steerCount,
         cancelled,
         timedOut,
         // shape understood by observability/events.ts normalizeUsage()
diff --git a/src/session.ts b/src/session.ts
index a61656d..d84eb8a 100644
--- a/src/session.ts
+++ b/src/session.ts
@@ -19,6 +19,7 @@ import type { Store, SessionStatus } from "./storage/store.js";
 import { storeObserver, combineObservers } from "./storage/checkpoint.js";
 import type { RunnerCallSink } from "./observability/instrument.js";
 import { EVENT_TYPES } from "./observability/events.js";
+import type { RunnerActivityEvent } from "./observability/events.js";
 
 /**
  * End-to-end run (Task 15): goal -> reviewed plan -> per-task actor-critic loop
@@ -140,17 +141,47 @@ export async function runGoal(
     // (Task 22). When persisting, calls are recorded to the store's event stream.
     const onRunnerCall: RunnerCallSink | undefined =
       store && sessionId
-        ? (e) =>
-            store.recordEvent({
-              sessionId,
-              at: e.at,
-              type: EVENT_TYPES.runnerCall,
-              data: e as unknown as Record<string, unknown>,
-            })
+        ? (e) => {
+            void store
+              .recordEvent({
+                sessionId,
+                at: e.at,
+                type: EVENT_TYPES.runnerCall,
+                data: e as unknown as Record<string, unknown>,
+              })
+              .catch((err) => {
+                opts.log?.(
+                  `failed to persist runner_call: ${String((err as Error)?.message ?? err)}`,
+                );
+              });
+          }
         : roleOpts.onRunnerCall;
+    // Mid-call runner activity (sub-step streaming): tool calls from a native
+    // agent runner's inner loop flow to the same event stream, so the monitor
+    // (and the durable trace) see live progress within a single runner call.
+    const onActivity: ((e: RunnerActivityEvent) => void) | undefined =
+      store && sessionId
+        ? (e) => {
+            // Fire-and-forget persistence: never let a rejected recordEvent
+            // surface as an unhandled rejection and destabilize the run.
+            void store
+              .recordEvent({
+                sessionId,
+                at: e.at,
+                type: EVENT_TYPES.runnerActivity,
+                data: e as unknown as Record<string, unknown>,
+              })
+              .catch((err) => {
+                opts.log?.(
+                  `failed to persist runner_activity: ${String((err as Error)?.message ?? err)}`,
+                );
+              });
+          }
+        : roleOpts.onActivity;
     const { actor, critic } = createRoles(config, {
       ...roleOpts,
       ...(onRunnerCall ? { onRunnerCall } : {}),
+      ...(onActivity ? { onActivity } : {}),
       ...(signal ? { signal } : {}),
     });
 
diff --git a/test/piAgentCompaction.test.ts b/test/piAgentCompaction.test.ts
new file mode 100644
index 0000000..be611f7
--- /dev/null
+++ b/test/piAgentCompaction.test.ts
@@ -0,0 +1,117 @@
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import {
+  registerFauxProvider,
+  fauxAssistantMessage,
+  fauxToolCall,
+  type FauxProviderRegistration,
+} from "@earendil-works/pi-ai";
+import type { AgentMessage } from "@earendil-works/pi-agent-core";
+import {
+  createCompactionTransform,
+  findCompactionCut,
+  type CompactionOptions,
+} from "../src/runners/piAgentRunner.js";
+
+/**
+ * Context compaction is the pi `transformContext` seam: when the transcript
+ * grows past budget, the older middle is summarized (via pi's generateSummary)
+ * and replaced with a single summary message, keeping the prompt + recent tail.
+ * Tested directly against the faux provider so the real summarization path runs
+ * offline and deterministically.
+ */
+
+const ts = () => Date.now();
+const user = (text: string): AgentMessage => ({ role: "user", content: text, timestamp: ts() });
+const toolResult = (id: string, text: string): AgentMessage => ({
+  role: "toolResult",
+  toolCallId: id,
+  toolName: "read_file",
+  content: [{ type: "text", text }],
+  isError: false,
+  timestamp: ts(),
+});
+const assistantToolCall = (id: string): AgentMessage =>
+  fauxAssistantMessage(fauxToolCall("read_file", { path: "f.txt" }, { id }), { stopReason: "toolUse" });
+const assistantText = (text: string): AgentMessage =>
+  fauxAssistantMessage(text, { stopReason: "stop" });
+
+/** A transcript as transformContext sees it: prompt + read turns, last message
+ *  a toolResult (the state right before the next assistant turn). */
+function transcript(): AgentMessage[] {
+  const big = "x".repeat(4_000);
+  return [
+    user("build the thing"),
+    assistantToolCall("c1"),
+    toolResult("c1", big),
+    assistantToolCall("c2"),
+    toolResult("c2", big),
+    assistantToolCall("c3"),
+    toolResult("c3", big),
+  ];
+}
+
+const opts = (over: Partial<CompactionOptions> = {}): CompactionOptions => ({
+  enabled: true,
+  keepRecentTokens: 50,
+  reserveTokens: 200,
+  ...over,
+});
+
+describe("findCompactionCut", () => {
+  it("snaps the retained-tail start to an assistant turn boundary", () => {
+    const msgs = transcript();
+    const cut = findCompactionCut(msgs, 50);
+    expect(cut).toBeGreaterThan(0);
+    expect(cut).toBeLessThan(msgs.length);
+    expect(msgs[cut]?.role).toBe("assistant"); // never starts the tail on a dangling toolResult
+  });
+
+  it("keeps everything (cut at 1) when the budget covers the whole tail", () => {
+    const msgs = transcript();
+    expect(findCompactionCut(msgs, 10_000_000)).toBe(1);
+  });
+});
+
+describe("createCompactionTransform", () => {
+  let faux: FauxProviderRegistration;
+  beforeEach(() => {
+    faux = registerFauxProvider({
+      provider: "faux",
+      models: [{ id: "faux-coder", contextWindow: 200_000, maxTokens: 8_192 }],
+    });
+  });
+  afterEach(() => faux.unregister());
+
+  it("returns undefined when compaction is disabled", () => {
+    expect(createCompactionTransform(opts({ enabled: false }), faux.getModel(), "k")).toBeUndefined();
+  });
+
+  it("summarizes the middle and keeps prompt + recent tail when over threshold", async () => {
+    faux.setResponses([fauxAssistantMessage("COMPACTED SUMMARY", { stopReason: "stop" })]);
+    const transform = createCompactionTransform(opts({ thresholdTokens: 1 }), faux.getModel(), "k");
+    expect(transform).toBeDefined();
+
+    const msgs = transcript();
+    const out = await transform!(msgs);
+
+    expect(out.length).toBeLessThan(msgs.length);
+    expect(out[0]).toBe(msgs[0]); // original prompt retained verbatim
+    expect(out[1]?.role).toBe("compactionSummary");
+    expect((out[1] as { summary: string }).summary).toContain("COMPACTED SUMMARY");
+    expect(out[2]?.role).toBe("assistant"); // retained tail starts on a turn boundary
+  });
+
+  it("leaves the transcript untouched when under threshold", async () => {
+    const transform = createCompactionTransform(opts({ thresholdTokens: 10_000_000 }), faux.getModel(), "k");
+    const msgs = transcript();
+    const out = await transform!(msgs);
+    expect(out).toBe(msgs); // same reference: no compaction performed
+  });
+
+  it("does not compact a transcript that is too short to have a safe cut", async () => {
+    const transform = createCompactionTransform(opts({ thresholdTokens: 1 }), faux.getModel(), "k");
+    const msgs: AgentMessage[] = [user("hi"), assistantText("done")];
+    const out = await transform!(msgs);
+    expect(out).toBe(msgs);
+  });
+});
diff --git a/test/piAgentRunner.test.ts b/test/piAgentRunner.test.ts
index a00160f..8cdb1a6 100644
--- a/test/piAgentRunner.test.ts
+++ b/test/piAgentRunner.test.ts
@@ -10,6 +10,7 @@ import {
 } from "@earendil-works/pi-ai";
 import { PiAgentRunner } from "../src/runners/piAgentRunner.js";
 import { createRunner } from "../src/runners/runnerFactory.js";
+import type { RunnerActivity } from "../src/runners/agentRunner.js";
 
 /**
  * These exercise the REAL pi Agent loop and the REAL file-editing tools, with
@@ -132,6 +133,31 @@ describe("PiAgentRunner (native agentic runner)", () => {
     expect(r).toBeInstanceOf(PiAgentRunner);
   });
 
+  it("streams tool activity (turn_start/tool_start/tool_end) via onEvent", async () => {
+    await writeFile(join(dir, "greeting.txt"), "hello world\n");
+    faux.setResponses([
+      fauxAssistantMessage(
+        fauxToolCall("edit_file", { path: "greeting.txt", old_str: "world", new_str: "pi" }),
+        { stopReason: "toolUse" },
+      ),
+      fauxAssistantMessage("done", { stopReason: "stop" }),
+    ]);
+
+    const activity: RunnerActivity[] = [];
+    await runner().run({ prompt: "edit", cwd: dir, onEvent: (a) => activity.push(a) });
+
+    const phases = activity.map((a) => a.phase);
+    expect(phases).toContain("turn_start");
+    expect(phases).toContain("tool_start");
+    expect(phases).toContain("tool_end");
+
+    const start = activity.find((a) => a.phase === "tool_start");
+    const end = activity.find((a) => a.phase === "tool_end");
+    expect(start?.toolName).toBe("edit_file");
+    expect(end?.toolCallId).toBe(start?.toolCallId); // start/end correlate
+    expect(end?.isError).toBe(false);
+  });
+
   describe("permission gating (beforeToolCall)", () => {
     it("confines file tools to the worktree by default (blocks a parent-escape write)", async () => {
       faux.setResponses([
@@ -221,4 +247,44 @@ describe("PiAgentRunner (native agentic runner)", () => {
       expect(await readFile(join(dir, "allowed.txt"), "utf8")).toBe("ok\n");
     });
   });
+
+  describe("steering (nudge a running agent)", () => {
+    it("injects a mid-run steering message that reaches the next turn", async () => {
+      await writeFile(join(dir, "f.txt"), "data\n");
+      // Turn 1 calls a tool; turn 2 is a factory that reports whether the
+      // steered text reached the model's context.
+      faux.setResponses([
+        fauxAssistantMessage(fauxToolCall("read_file", { path: "f.txt" }), { stopReason: "toolUse" }),
+        (context) => {
+          const saw = JSON.stringify(context.messages).includes("STEER!");
+          return fauxAssistantMessage(saw ? "got-steer" : "no-steer", { stopReason: "stop" });
+        },
+      ]);
+
+      let steer: ((t: string) => void) | undefined;
+      let steered = false;
+      const res = await runner().run({
+        prompt: "read the file",
+        cwd: dir,
+        steering: (s) => {
+          steer = s;
+        },
+        onEvent: (a) => {
+          if (a.phase === "tool_start" && !steered) {
+            steered = true;
+            steer?.("STEER!");
+          }
+        },
+      });
+
+      expect(res.meta?.steerCount).toBe(1);
+      expect(res.text).toBe("got-steer"); // the nudge influenced the next turn
+    });
+
+    it("reports zero steers when no steering source is provided", async () => {
+      faux.setResponses([fauxAssistantMessage("done", { stopReason: "stop" })]);
+      const res = await runner().run({ prompt: "noop", cwd: dir });
+      expect(res.meta?.steerCount).toBe(0);
+    });
+  });
 });
diff --git a/test/runnerRoles.test.ts b/test/runnerRoles.test.ts
index fd09bc2..788f220 100644
--- a/test/runnerRoles.test.ts
+++ b/test/runnerRoles.test.ts
@@ -1,7 +1,8 @@
 import { describe, it, expect } from "vitest";
 import { RunnerActor, RunnerCritic, RunnerRoleError } from "../src/adapters/runnerRoles.js";
 import { MockRunner } from "../src/runners/mockRunner.js";
-import type { RunnerProfile, RunResult } from "../src/runners/agentRunner.js";
+import type { AgentRunner, RunnerProfile, RunRequest, RunResult } from "../src/runners/agentRunner.js";
+import type { RunnerActivityEvent } from "../src/observability/events.js";
 import type { TaskSpec } from "../src/schemas/plan.js";
 import type { TaskArtifactBundle } from "../src/schemas/artifact.js";
 import { loadConfig } from "../src/config.js";
@@ -219,3 +220,64 @@ describe("runner-backed roles drive the real loop", () => {
     expect(outcome.buildAttempts).toBe(2);
   });
 });
+
+
+describe("runner activity streaming (sub-step events)", () => {
+  const ts = () => new Date().toISOString();
+
+  it("forwards actor runner activity enriched with the role + runner identity", async () => {
+    const runner: AgentRunner = {
+      profile,
+      async run(req: RunRequest): Promise<RunResult> {
+        req.onEvent?.({ phase: "turn_start", turn: 1, at: ts() });
+        req.onEvent?.({ phase: "tool_start", toolName: "edit_file", toolCallId: "c1", at: ts() });
+        req.onEvent?.({ phase: "tool_end", toolName: "edit_file", toolCallId: "c1", isError: false, at: ts() });
+        return { text: buildJson("t1") };
+      },
+    };
+    const events: RunnerActivityEvent[] = [];
+    const actor = new RunnerActor(runner, { onActivity: (e) => events.push(e) });
+
+    await actor.build(sampleTask, undefined, ".");
+
+    expect(events.map((e) => e.phase)).toEqual(["turn_start", "tool_start", "tool_end"]);
+    expect(events.every((e) => e.role === "actor")).toBe(true);
+    expect(events.every((e) => e.runnerId === "p" && e.model === "test-model")).toBe(true);
+    const end = events.find((e) => e.phase === "tool_end");
+    expect(end?.toolName).toBe("edit_file");
+    expect(end?.toolCallId).toBe("c1");
+    expect(end?.isError).toBe(false);
+  });
+
+  it("attributes critic runner activity to the critic role", async () => {
+    const runner: AgentRunner = {
+      profile,
+      async run(req: RunRequest): Promise<RunResult> {
+        req.onEvent?.({ phase: "tool_start", toolName: "read_file", toolCallId: "x", at: ts() });
+        return { text: criticGreen().text };
+      },
+    };
+    const events: RunnerActivityEvent[] = [];
+    const critic = new RunnerCritic(runner, { onActivity: (e) => events.push(e) });
+
+    await critic.review({ kind: "task", bundle: sampleBundle });
+
+    expect(events).toHaveLength(1);
+    expect(events[0]?.role).toBe("critic");
+    expect(events[0]?.toolName).toBe("read_file");
+  });
+
+  it("does no work and forwards nothing when no activity sink is configured", async () => {
+    let called = false;
+    const runner: AgentRunner = {
+      profile,
+      async run(req: RunRequest): Promise<RunResult> {
+        if (req.onEvent) called = true; // role must not pass an onEvent when sink is off
+        return { text: buildJson("t1") };
+      },
+    };
+    const actor = new RunnerActor(runner);
+    await actor.build(sampleTask, undefined, ".");
+    expect(called).toBe(false);
+  });
+});