diff --git a/docs/agents.md b/docs/agents.md
index 2bc4771a0f..81556d2a79 100644
--- a/docs/agents.md
+++ b/docs/agents.md
@@ -174,7 +174,7 @@ records. Codex uses an inherited thinking choice only when the effective model
 supports it, otherwise that model's default applies. Transient process/session
 state may appear in the Activity tab, but is not durable history.
 
-A type with `sessionControls` (Claude, Codex and goose) offers three immediate
+A type with `sessionControls` (Claude, Codex, goose and Pi) offers three immediate
 session controls in the composer's gauge popover: **Compact session**,
 **Start a new session**, and **Interrupt**. A side rail keeps all three visible.
 Compact and New are enabled while that harness session is idle; Interrupt is
diff --git a/src/bundled/claude-code/transcript.ts b/src/bundled/claude-code/transcript.ts
index 43c63dbe5d..9f7eb5f2b2 100644
--- a/src/bundled/claude-code/transcript.ts
+++ b/src/bundled/claude-code/transcript.ts
@@ -64,7 +64,7 @@ const KINDS: Record<string, string> = {
 };
 
 /** A Buzz prompt, or a steer wrapping one, as the turn's opening item. */
-export function prompt(id: string, at: number, raw: string): TranscriptItem {
+function prompt(id: string, at: number, raw: string): TranscriptItem {
   let sections = promptSections(raw);
   const steer = sections.find((section) => section.tag === STEER);
   if (steer) sections = promptSections(steer.body);
diff --git a/src/bundled/pi/README.md b/src/bundled/pi/README.md
index 46bcc07375..05bab5d201 100644
--- a/src/bundled/pi/README.md
+++ b/src/bundled/pi/README.md
@@ -12,156 +12,45 @@ Create agents that run Pi on your computer, with any provider Pi supports, and a
 
 # Pi for Agents
 
-A desktop plugin that ships with Buzz on macOS, Linux and Windows, off until
-turned on in **Settings → Plugins**. Its agents run on
-[Pi durable](https://earendil.com/posts/pi-1-0/) (`@earendil-works/pi-durable`),
-not the Pi coding agent, with any provider Pi supports.
-Agents owns the identity, admission, configuration, and signed Buzz actions. This
-plugin owns Pi work after delivery handover. One Pi host process ([`host.mjs`](host.mjs))
-runs every Pi agent here. Each agent has its own SQLite storage, and each Buzz
-conversation the agent is in is one durable Pi conversation.
-
-## Try it
-
-From the feature worktree:
-
-```sh
-source bin/activate-hermit
-BUZZODZ_PROFILE=pi-agents BUZZ_DEV_VIEWER= bin/just desktop
-```
-
-Pi needs Node.js 22.19 or later on this computer (`node:sqlite`). On macOS and
-Linux the app's PATH is searched first, then an interactive login shell's, so
-Homebrew, nvm, fnm and similar installs work. On Windows, `node` must be on the PATH Buzz
-starts with.
-
-1. Turn on **Pi** in **Settings → Plugins**. Then open **Agents → New agent**
-   and choose **Pi** as the **Harness**. Its **Model** starts on **Pi default**:
-   the `defaultProvider` and `defaultModel` in `~/.pi/agent/settings.json`, as
-   the `pi` command sets them. The first time Buzz asks Pi anything, it
-   installs Pi's packages (a few seconds).
-2. Click the model (marked with Pi, and with the provider when Buzz has its
-   mark) and choose a provider. A provider Pi has no key for shows an
-   **API key** field; **Save to Pi** stores it in
-   `~/.pi/agent/auth.json`, shared with the `pi` command. A provider that signs
-   in some other way (OAuth, cloud credentials) says to run `pi`, then `/login`.
-   Choose a model, click **Test connection**, then **Use**. Name the agent and
-   click **Create agent**. Its library card shows the same control.
-3. Open the agent. Its **Identity** tab has the **Provider**, **Model**,
-   **Thinking** level (Pi's default, or one the model has) and
-   **Instructions**. Its **Settings** tab has the provider's key with **Test
-   connection**, the **Workspace** (default `~/.buzz`), and thread or channel
-   **Conversations**.
-4. In a channel, send: `@Pi create hello.txt containing hello in the workspace,
-   read it back, and report what you found` (use the agent's actual name).
-   Expect steps under your message, one reply in the thread, and the file in the
-   workspace. Then reply in the same thread: `@Pi what did you write?`. It
-   remembers, because the thread is the same Pi conversation.
-5. While it works, send another mention in the thread. It steers the running
-   turn: Pi reads it at its next step. The activity moves to the newer message.
-6. Quit Buzz during a long turn, start it again, and mention the agent in that
-   thread. Pi continues the interrupted run from its storage, and the new
-   mention steers it.
-7. Choose **View activity** on one of its messages in that thread, or open the
-   agent's **Activity** tab from there. **This thread** is its Pi conversation,
-   with the agent's other recent ones below. **View transcript** shows each
-   turn's prompt, thinking, tool calls with their output, and what it sent,
-   read from Pi's storage.
-
-The composer's model picker offers the agent's provider's models, and each
-model's thinking levels, for one turn. Without a pick, a turn runs the
-agent's saved model and Thinking level.
-
-## Checks
+Agents run on [Pi durable](https://github.com/earendil-works/pi/blob/main/packages/durable/README.md) in one Node
+host ([`host.mjs`](host.mjs)). Each agent has its own SQLite storage, and each
+Buzz thread or channel is one durable Pi conversation. Agents owns identity,
+admission and signing; this plugin owns the turn after handover.
 
 ```sh
 bin/pnpm exec vitest run src/bundled/pi
-PI_DEPS="$HOME/Library/Application Support/Buzz/pi/deps/<sum>" bin/pnpm exec vitest run src/bundled/pi/host.live.test.ts
+PI_DEPS=<installed deps folder> bin/pnpm exec vitest run src/bundled/pi/host.live.test.ts
 ```
 
-The unit tests cover the stdio protocol (`pi.ts`), the runtime against a fake
-host, the install script in a throwaway home, and the New agent and Settings
-fields. The opt-in live test runs the real `host.mjs` on Pi durable, using Pi's
-faux provider as the model. It checks that a final `send` ends the run, that the
-model reads what a Buzz tool answered, that a plain answer comes back as text,
-that a request steers a working run, that a run cut off by a stop continues in
-the next host, that aborting a run leaves it unanswered, and that a deferred
-Buzz tool is offered once `tool_search` finds it. `PI_DEPS` is any folder
-the launcher installed Pi's packages in.
+The live test runs the real host on Pi's faux provider.
 
-## Buzz's copy of Pi's packages
+## Packages
 
-[`launch.mjs`](launch.mjs) runs the host on this computer's Node. On macOS and
-Linux, [`pi.sh`](pi.sh) finds that Node and starts the launcher on it; Windows,
-which has no bash, starts `node` itself. The launcher's first line on stdin
-carries the host source and the packages to install, since together they pass
-Windows' 32 KB command line. On first use it runs `npm ci` with
-[`deps/package-lock.json`](deps/package-lock.json), which pins every package and
-its integrity hash. Install scripts are off, and npm uses the registry it is
-configured for. Each lock gets its own folder under
-`~/Library/Application Support/Buzz/pi/deps/`
-(`${XDG_DATA_HOME:-~/.local/share}/buzz/pi/deps/` on Linux,
-`%LOCALAPPDATA%\Buzz\pi\deps\` on Windows). The folder appears
-whole or not at all, and is shared by every Buzz build. An install removes the
-folders no build has started for 30 days. Agents' storage is in `agents/<pubkey>.sqlite`
-beside it. A lock on each agent's storage keeps a second Buzz build from opening
-it; that build's turns fail and say so. It also keeps Pi's `auth.json` writes
-from colliding with the `pi` command's.
-
-To update Pi, change the versions in [`deps/package.json`](deps/package.json),
-then regenerate the lock without registry URLs:
+[`launch.mjs`](launch.mjs) installs [`deps/package-lock.json`](deps/package-lock.json)
+with `npm ci` (no install scripts) into a per-lock folder under Buzz's app data
+`pi/deps/`, shared by every Buzz build. To update Pi, change
+[`deps/package.json`](deps/package.json), then:
 
 ```sh
 cd src/bundled/pi/deps
 npm install --package-lock-only --ignore-scripts --omit-lockfile-registry-resolved
 ```
 
-## Limits and boundaries
-
-- Only the owner's events, or ones the owner delegated, start work. Watches,
-  timers and catch-ups run as standalone turns, each a one-off Pi
-  conversation, in the lanes and with the 5-minute deadline Buzz's agent
-  runner gives every harness.
-- Pi has the coding tools `read`, `write`, `edit` and `bash`, in the workspace,
-  with no sandbox or approvals, like Claude's `bypassPermissions`. Buzz's tools
-  are offered as `buzz_<name>` (`buzz_send`, `buzz_read`, …), since `read` and
-  `edit` would collide. They run inside Buzz through the agent's native-held
-  handle, so the host holds no Buzz key. A final `buzz_send` ends the run.
-  The Buzz tools Claude and Codex load on search (the attention tools, and
-  those plugins add) are deferred: Pi's `tool_search` loads them when the
-  model looks for them, and they stay loaded in that conversation.
-- **Stop** on a turn's activity aborts the Pi run it is in, and posts
-  nothing.
-- A turn that answered in text and never called `buzz_send` has its answer
-  posted in the thread, as goose does. A turn that fails posts `⚠️ I couldn't
-  finish that: …` and tags the owner.
-- A Buzz tool call that a stop cuts off runs again when the run continues, so a
-  `send` may be posted twice.
-- Pi's storage keeps every conversation, including each one-off watch or timer
-  turn; nothing prunes it. The Activity tab lists thread and channel
-  conversations only, and rereads an open transcript when its turn ends, not
-  as it streams.
-- A host that stops while a turn waits is started again once and waits for it
-  there. A run resumed that way shows no steps.
-- Runs left unfinished when the app quit continue at the first turn after it
-  starts again, in whatever thread. A mention's run shows as working and its
-  answer or failure is posted where it would have been, but it shows no steps
-  and Stop does not reach it. A watch's or timer's run is ended instead, as
-  quitting ends other harnesses' turns: continued, it would have no deadline.
-- Pi durable records some requests that have no answer, such as compaction's,
-  only in its totals. Usage counts each model's share of those as one more
-  request, at the newest compaction.
-- A provider that needs more than a key, such as Bedrock or Vertex, is set up
-  with the `pi` command. Signing in from Buzz is not supported.
-- A `!command` or `$NAME` key in `auth.json`, which the `pi` command runs or
-  reads from its environment to get the key, is not used: one host serves
-  every agent, and a command would hold them all, and the desktop app often
-  lacks the shell's environment. Its turns fail and say so; enter the key in
-  Buzz instead.
-- Only Pi's built-in providers run. One that a Pi package adds (`packages` in
-  Pi's settings), such as databricks, is not loaded: the provider field, the
-  New agent Model field and a turn's error say so, and Use and Save wait for another.
-- On Windows, Pi's `bash` tool needs a bash, such as Git for Windows'; the
-  other tools work without one.
-- Windows, Linux, packaged-app behavior, and human acceptance need separate
-  validation.
+## Limits
+
+- Pi's `read`, `write`, `edit` and `bash` run in the workspace with no sandbox
+  or approvals. Buzz tools are `buzz_<name>`.
+- Keys in Pi's `auth.json` must be literal: Buzz runs no `!command` key and
+  reads no `$NAME` key. Set up a provider that needs more than a key with
+  `pi`. Providers from Pi packages don't run.
+- Runs left unfinished when the app quit continue at the next turn. A
+  mention's run posts its answer or failure but shows no steps; its thread's
+  session controls interrupt it. A watch's or timer's run is ended.
+- When the app quits or the host stops during a Buzz tool call, Pi tells the
+  model the call may have run, so a `send` it makes again may be posted
+  twice. A Stop or Interrupt ends the run instead.
+- Pi durable records compaction's usage only in each conversation's totals,
+  not per compaction. Usage dates all of it at the newest compaction, so each
+  compaction moves the conversation's earlier compaction spend to its day.
+- On Windows, Pi's `bash` tool needs Git Bash. Linux and Windows are not yet
+  validated.
diff --git a/src/bundled/pi/host.live.test.ts b/src/bundled/pi/host.live.test.ts
index 0e8b68d3b4..e1427adbb9 100644
--- a/src/bundled/pi/host.live.test.ts
+++ b/src/bundled/pi/host.live.test.ts
@@ -129,6 +129,29 @@ const g1 = await pi.rpc("deliver", turn("G", "ninth"));
 await asking.gate;
 await pi.rpc("abort", { agent, conversation: g1.conversation });
 out.aborted = await pi.rpc("wait", { agent, submission: g1.submission });
+// Session controls act on a conversation key's Pi conversation.
+const s = held(), running = held();
+faux.setResponses([async (_context, options) => { running.release(); await abortable(s.gate, options?.signal); return fauxAssistantMessage("never"); }]);
+const s1 = await pi.rpc("deliver", turn("S", "session"));
+await running.gate;
+await pi.rpc("session", { agent, key: "chan/root", action: "interrupt" });
+out.interrupted = await pi.rpc("wait", { agent, submission: s1.submission });
+// Interrupted in a tool call, the run leaves Pi's aborted result for it.
+faux.setResponses([() => fauxAssistantMessage(fauxToolCall("bash", { command: "sleep 30" }), { stopReason: "toolUse" })]);
+const t1 = await pi.rpc("deliver", turn("T", "tool"));
+await new Promise((resolve) => { const check = () => pi.events.some((event) => event.type === "step" && event.tool === "bash") ? resolve() : setTimeout(check, 5); check(); });
+await pi.rpc("session", { agent, key: "chan/root", action: "interrupt" });
+out.toolStop = { wait: await pi.rpc("wait", { agent, submission: t1.submission }), last: (await pi.rpc("entries", { agent, conversation: t1.conversation })).entries[0] };
+// Past what Pi keeps verbatim, so compacting summarizes the older turn.
+faux.setResponses([() => fauxAssistantMessage("Read it"), () => fauxAssistantMessage("Read it too"), () => fauxAssistantMessage("Summary of the chat")]);
+for (const request of ["S2", "S3"]) {
+  const long = await pi.rpc("deliver", turn(request, "word ".repeat(25_000)));
+  await pi.rpc("wait", { agent, submission: long.submission });
+}
+await pi.rpc("session", { agent, key: "chan/root", action: "compact" });
+await pi.rpc("session", { agent, key: "chan/root", action: "new" });
+out.renewed = { place: await pi.rpc("place", { agent, key: "chan/root" }), kinds: (await pi.rpc("entries", { agent, conversation: s1.conversation })).entries.map((entry) => entry.kind).slice(0, 2) };
+out.unplaced = await pi.rpc("session", { agent, key: "chan/none", action: "compact" });
 // A tool loaded on search is offered once tool_search finds it.
 const offered = [];
 /** The tools a request offers: its own, and those its system messages add. */
@@ -352,6 +375,24 @@ describe.skipIf(!DEPS)("the Pi host on Pi durable", () => {
       status: "unanswered",
       reason: "aborted",
     });
+    expect(out.interrupted).toMatchObject({
+      status: "unanswered",
+      reason: "aborted",
+    });
+    expect(out.toolStop.wait).toMatchObject({
+      status: "unanswered",
+      reason: "aborted",
+    });
+    expect(out.toolStop.last).toMatchObject({
+      kind: "pi.tool-result",
+      aborted: true,
+    });
+    // A new session is shown the whole conversation at its next turn.
+    expect(out.renewed).toEqual({
+      place: { seen: 0 },
+      kinds: ["pi.reset", "pi.compaction"],
+    });
+    expect(out.unplaced).toEqual({});
     const [before, after] = out.search.offered;
     expect(before).toEqual(
       expect.arrayContaining(["buzz_send", "tool_search"]),
diff --git a/src/bundled/pi/host.mjs b/src/bundled/pi/host.mjs
index 3f95c7b0ec..a5003de108 100644
--- a/src/bundled/pi/host.mjs
+++ b/src/bundled/pi/host.mjs
@@ -642,9 +642,13 @@ export function serve({
       if (!found) return { entries: [] };
       const page = await found.entries({}, PAGE, cursor, context);
       return {
-        entries: page.items.map(({ id, kind, model }) => ({
+        entries: page.items.map(({ id, kind, model, data }) => ({
           id,
           kind,
+          // Pi writes this result itself for a call an abort cut off.
+          ...(data?.diagnostics?.some?.((d) => d.code === "aborted") && {
+            aborted: true,
+          }),
           // System entries hold every tool's schema; a transcript shows none.
           ...(model && kind !== "pi.system" ? { model: model.map(slim) } : {}),
         })),
@@ -790,6 +794,37 @@ export function serve({
       const { harness } = await agentOf(agent);
       await (await harness.conversation(conversation, context))?.abort(context);
     },
+    /** A session control on a conversation key's Pi conversation, if it has
+     * one: `interrupt` aborts its run and any compaction, `compact` summarizes
+     * it and settles once the summary is placed, and `new` starts a new
+     * context, shown the whole Buzz conversation at its next turn. */
+    async session({ agent, key, action }) {
+      const { harness } = await agentOf(agent);
+      const place = (await harness.snapshot(Places, context))?.[key];
+      const conversation =
+        place && (await harness.conversation(place.conversation, context));
+      if (!conversation) return;
+      if (action === "interrupt") return conversation.abort(context);
+      if (action === "new") {
+        await conversation.reset(undefined, context);
+        await harness.commit(async (tx) => {
+          const places = await tx.doc(Places);
+          if (places[key]) places[key] = { ...places[key], seen: 0 };
+        }, context);
+        return;
+      }
+      const { outcome } = (
+        await harness.waitForTask(
+          await conversation.compact(undefined, context),
+          context,
+        )
+      ).state;
+      if (outcome.status === "failed") throw new Error(outcome.error.message);
+      if (outcome.status !== "completed") return;
+      const { submissionId } = outcome.result;
+      if (submissionId !== undefined)
+        await (await harness.submission(submissionId, context))?.wait(context);
+    },
     /** Closes the agent's storage. Unfinished runs stay, and continue when it
      * opens again. */
     async close({ agent }) {
diff --git a/src/bundled/pi/index.ts b/src/bundled/pi/index.ts
index c44270f9ab..1990c39986 100644
--- a/src/bundled/pi/index.ts
+++ b/src/bundled/pi/index.ts
@@ -27,6 +27,7 @@ export const apply: PluginModule["apply"] = (ctx) => {
     (pubkey, signal) => agents.core(pubkey, signal),
     () => agents.tools(),
     repositoriesContext,
+    (pubkey) => agents.handle(pubkey),
   );
   const { ActivityTab, Create, CreateModel, SettingsTab } = createTabs({
     setup: runtime.setup,
@@ -94,6 +95,12 @@ export const apply: PluginModule["apply"] = (ctx) => {
         component: ActivityTab,
       },
     ],
+    sessionControls: {
+      scope: (agent) => config(agent.config).scope,
+      state: (agent, place) => runtime.sessionState(agent, place),
+      run: (agent, place, action) => runtime.session(agent, place, action),
+      subscribe: (listener) => runtime.subscribe(listener),
+    },
     models: (agent) => runtime.models(agent),
     run: (delivery) => runtime.run(delivery),
     runOnce: (turn) => runtime.runOnce(turn),
diff --git a/src/bundled/pi/runtime.test.ts b/src/bundled/pi/runtime.test.ts
index 1ed30b0ee1..684442ece3 100644
--- a/src/bundled/pi/runtime.test.ts
+++ b/src/bundled/pi/runtime.test.ts
@@ -4,6 +4,7 @@ import { ownerReads } from "../../features/agents/reads";
 import type { MemoryEntry } from "../../features/agents/memory";
 import type {
   Agent,
+  AgentHandle,
   Delivery,
   StandaloneTurn,
 } from "../../features/agents/service";
@@ -116,12 +117,18 @@ function setup(
     }),
   } as unknown as RelayData;
   const core = vi.fn(async () => null as string | null | undefined);
+  /** Whether `Agents.handle` gives the agent outside a run. */
+  const connection = { live: true };
   const runtime = new PiRuntime(
     { spawn: host.spawn },
     (pubkey) => ownerReads(relay, pubkey),
     core,
     () => added,
     repositories,
+    (pubkey) =>
+      connection.live && pubkey === self
+        ? (agent as unknown as AgentHandle)
+        : undefined,
   );
   const published: { kind: number; content: string; tags: string[][] }[] = [];
   const agent = {
@@ -196,6 +203,7 @@ function setup(
     agent,
     session,
     solo,
+    connection,
   };
 }
 
@@ -667,6 +675,156 @@ it("stops a mention's run when the owner stops it, and posts nothing", async ()
   expect(published).toEqual([]);
 });
 
+it("interrupts a run an earlier app left from its thread's session controls", async () => {
+  const left = { submission: 41, conversation: 5, request: "d".repeat(64) };
+  const stopped = deferred<unknown>();
+  const { host, deliver, published, runtime } = setup({
+    unfinished: () => ({
+      submissions: [{ ...left, context: { channel, root: root.id } }],
+    }),
+    wait: ({ submission }) =>
+      submission === left.submission ? stopped.promise : new Promise(() => {}),
+    session: ({ action }) => {
+      if (action === "interrupt")
+        stopped.resolve({ status: "unanswered", reason: "aborted" });
+    },
+  });
+  const agent = { pubkey: self, config: DEFAULT_CONFIG } as unknown as Agent;
+  const thread = { channelId: channel, root: root.id };
+  expect(runtime.sessionState(agent, thread)).toBe("idle");
+  await deliver({
+    type: "mention",
+    event: message("5", "@Pi elsewhere", {
+      tags: [
+        ["h", channel],
+        ["p", self],
+      ],
+    }),
+  }).run;
+  expect(runtime.sessionState(agent, thread)).toBe("active");
+  await runtime.session(agent, thread, "interrupt");
+  expect(host.requests("session")[0]?.params).toEqual({
+    agent: self,
+    key: `${channel}/${root.id}`,
+    action: "interrupt",
+  });
+  await vi.waitFor(() =>
+    expect(runtime.sessionState(agent, thread)).toBe("idle"),
+  );
+  expect(published).toEqual([]);
+});
+
+it("compacts or starts a new session only while idle, and hands a mention over once compaction ends", async () => {
+  const compacted = deferred();
+  const { host, deliver, settled, runtime, connection } = setup({
+    wait: () => ({ status: "done", answer: 5, text: "" }),
+    session: ({ action }) => (action === "compact" ? compacted.promise : {}),
+  });
+  const agent = { pubkey: self, config: DEFAULT_CONFIG } as unknown as Agent;
+  const thread = { channelId: channel, root: root.id };
+  // Without the agent's handle, its left runs could not be taken on first.
+  connection.live = false;
+  await expect(runtime.session(agent, thread, "new")).rejects.toThrow(
+    "isn't connected",
+  );
+  expect(host.requests("session")).toEqual([]);
+  connection.live = true;
+  // No mention is needed first.
+  await runtime.session(agent, thread, "new");
+  const first = deliver({ type: "mention", event: reply("2", "@Pi look") });
+  await first.run;
+  await settled(first);
+  await vi.waitFor(() =>
+    expect(runtime.sessionState(agent, thread)).toBe("idle"),
+  );
+  const compacting = runtime.session(agent, thread, "compact");
+  await vi.waitFor(() => expect(host.requests("session")).toHaveLength(1));
+  expect(runtime.sessionState(agent, thread)).toBe("compacting");
+  await expect(runtime.session(agent, thread, "new")).rejects.toThrow(
+    "Wait for the active turn",
+  );
+  const later = deliver({ type: "mention", event: reply("3", "@Pi again") });
+  // Released at once, and handed over after compaction.
+  await later.run;
+  expect(host.requests("deliver")).toHaveLength(1);
+  compacted.resolve();
+  await compacting;
+  await settled(later);
+  expect(host.requests("deliver")).toHaveLength(2);
+  await vi.waitFor(() =>
+    expect(runtime.sessionState(agent, thread)).toBe("idle"),
+  );
+  await runtime.session(agent, thread, "new");
+  expect(host.requests("session").map((item) => item.params)).toEqual([
+    { agent: self, key: `${channel}/${root.id}`, action: "new" },
+    { agent: self, key: `${channel}/${root.id}`, action: "compact" },
+    { agent: self, key: `${channel}/${root.id}`, action: "new" },
+  ]);
+});
+
+it("refuses Compact or New when a run taken on first is in the same conversation", async () => {
+  const left = { submission: 41, conversation: 5, request: "d".repeat(64) };
+  const { host, runtime } = setup({
+    unfinished: () => ({
+      submissions: [{ ...left, context: { channel, root: root.id } }],
+    }),
+  });
+  const agent = { pubkey: self, config: DEFAULT_CONFIG } as unknown as Agent;
+  const thread = { channelId: channel, root: root.id };
+  expect(runtime.sessionState(agent, thread)).toBe("idle");
+  await expect(runtime.session(agent, thread, "new")).rejects.toThrow(
+    "Wait for the active turn",
+  );
+  expect(host.requests("session")).toEqual([]);
+  expect(runtime.sessionState(agent, thread)).toBe("active");
+});
+
+it("sends no Compact that an Interrupt cut off while it was preparing", async () => {
+  const listed = deferred<unknown>();
+  const { host, runtime } = setup({ unfinished: () => listed.promise });
+  const agent = { pubkey: self, config: DEFAULT_CONFIG } as unknown as Agent;
+  const thread = { channelId: channel, root: root.id };
+  const compacting = runtime.session(agent, thread, "compact");
+  await vi.waitFor(() => expect(host.requests("unfinished")).toHaveLength(1));
+  expect(runtime.sessionState(agent, thread)).toBe("compacting");
+  const interrupting = runtime.session(agent, thread, "interrupt");
+  listed.resolve({ submissions: [] });
+  await expect(compacting).rejects.toThrow("interrupted");
+  await interrupting;
+  expect(host.requests("session").map((item) => item.params)).toEqual([
+    { agent: self, key: `${channel}/${root.id}`, action: "interrupt" },
+  ]);
+});
+
+it("takes on a run an earlier app left before compacting, with no mention since", async () => {
+  const left = { submission: 41, conversation: 5, request: "d".repeat(64) };
+  const order: string[] = [];
+  const { published, runtime } = setup({
+    unfinished: () => {
+      order.push("unfinished");
+      return {
+        submissions: [{ ...left, context: { channel, root: "other-root" } }],
+      };
+    },
+    wait: () => ({ status: "done", answer: 6, text: "Left answer" }),
+    session: () => {
+      order.push("session");
+    },
+  });
+  const agent = { pubkey: self, config: DEFAULT_CONFIG } as unknown as Agent;
+  await runtime.session(
+    agent,
+    { channelId: channel, root: root.id },
+    "compact",
+  );
+  // Taken on before Pi is asked anything that continues it.
+  expect(order).toEqual(["unfinished", "session"]);
+  await vi.waitFor(() =>
+    expect(published.map((event) => event.content)).toEqual(["Left answer"]),
+  );
+  expect(published[0]?.tags).toContainEqual(["e", "other-root", "", "reply"]);
+});
+
 it("lists the agent's sessions from Pi's storage, and whether each works", async () => {
   const answered = deferred<unknown>();
   const key = `${channel}/${root.id}`;
@@ -882,27 +1040,130 @@ it("ends the watch and timer runs an earlier app left once every left mention's
 });
 
 it("keeps a standalone turn of this app's when it lists left runs again", async () => {
-  let listings = 0;
   let own: string | undefined;
-  const { host, deliver, runOnce } = setup({
+  let waits = 0;
+  const { host, runOnce } = setup({
     once: ({ request }) => {
       own = String(request);
       return { submission: 51, conversation: 9 };
     },
-    unfinished: () => {
-      if (++listings === 1) throw new Error("busy");
-      return {
-        submissions: [{ submission: 51, conversation: 9, request: own }],
-      };
-    },
+    unfinished: () => ({
+      submissions: own
+        ? [{ submission: 51, conversation: 9, request: own }]
+        : [],
+    }),
+    // Its storage fails once, so it is opened again and left runs listed again.
+    wait: () =>
+      ++waits === 1
+        ? { status: "reopen" }
+        : { status: "done", answer: 5, text: "Done." },
   });
-  void runOnce("watch", "Check the build");
-  await vi.waitFor(() => expect(own).toBeDefined());
-  await deliver({ type: "mention", event: reply("2", "@Pi look") }).run;
-  expect(listings).toBe(2);
+  await expect(runOnce("watch", "Check the build")).resolves.toEqual({
+    ok: true,
+  });
+  expect(host.requests("unfinished")).toHaveLength(2);
   expect(host.requests("drop")).toEqual([]);
 });
 
+it("hands over no turn and waits for no left run while Pi could not end a left standalone run, and ends it at the next turn first", async () => {
+  // Pi continues every run once asked anything: one left for no one is
+  // ended before any wait, or not continued at all.
+  let drops = 0;
+  const left = {
+    submission: 42,
+    conversation: 6,
+    request: "d".repeat(64),
+    context: { channel, root: root.id },
+  };
+  const { host, deliver, settled, published, runtime } = setup({
+    unfinished: () => ({
+      submissions: [
+        { submission: 51, conversation: 9, request: "timer/left" },
+        left,
+      ],
+    }),
+    drop: () => {
+      if (++drops === 1) throw new Error("database is locked");
+    },
+  });
+  const first = deliver({ type: "mention", event: reply("2", "@Pi look") });
+  await first.run.catch(() => undefined);
+  await expect(settled(first)).resolves.toEqual({
+    ok: false,
+    error: "database is locked",
+  });
+  expect(host.requests("deliver")).toEqual([]);
+  expect(host.requests("wait")).toEqual([]);
+  expect(runtime.working(self)).not.toContain(left.conversation);
+  await vi.waitFor(() => expect(published).toHaveLength(1));
+  expect(published[0]?.content).toContain("database is locked");
+
+  await deliver({ type: "mention", event: reply("3", "@Pi again") }).run;
+  expect(host.requests("drop")).toHaveLength(2);
+  const methods = host.processes[0]?.requests.map((item) => item.method) ?? [];
+  expect(methods.lastIndexOf("drop")).toBeLessThan(methods.indexOf("wait"));
+  expect(methods.lastIndexOf("drop")).toBeLessThan(methods.indexOf("deliver"));
+  expect(host.requests("wait").map((item) => item.params.submission)).toContain(
+    left.submission,
+  );
+});
+
+it("ends a standalone run left for no one before it waits again in an agent opened again", async () => {
+  // Its storage failed while a watch's run was abandoned: ended before the
+  // wait that would continue it.
+  let waits = 0;
+  let reopened = false;
+  const { host, deliver, settled } = setup({
+    open: () => {
+      if (host.requests("open").length > 1) reopened = true;
+    },
+    unfinished: () => ({
+      submissions: reopened
+        ? [{ submission: 51, conversation: 9, request: "watch/abandoned" }]
+        : [],
+    }),
+    wait: () =>
+      ++waits === 1
+        ? { status: "reopen" }
+        : { status: "done", answer: 5, text: "Resumed." },
+  });
+  const turn = deliver({ type: "mention", event: reply("2", "@Pi look") });
+  await turn.run;
+  await expect(settled(turn)).resolves.toEqual({ ok: true });
+  const methods = host.processes[0]?.requests.map((item) => item.method) ?? [];
+  expect(methods.slice(methods.indexOf("wait") + 1)).toEqual([
+    "open",
+    "unfinished",
+    "drop",
+    "wait",
+  ]);
+});
+
+it("waits for no run again in an agent opened again while Pi could not end a standalone run left for no one", async () => {
+  let reopened = false;
+  const { host, deliver, settled } = setup({
+    open: () => {
+      if (host.requests("open").length > 1) reopened = true;
+    },
+    unfinished: () => ({
+      submissions: reopened
+        ? [{ submission: 51, conversation: 9, request: "watch/abandoned" }]
+        : [],
+    }),
+    drop: () => {
+      throw new Error("database is locked");
+    },
+    wait: () => ({ status: "reopen" }),
+  });
+  const turn = deliver({ type: "mention", event: reply("2", "@Pi look") });
+  await turn.run;
+  await expect(settled(turn)).resolves.toEqual({
+    ok: false,
+    error: "database is locked",
+  });
+  expect(host.requests("wait")).toHaveLength(1);
+});
+
 it("posts no answer for a left run that sent once another left run's wait continued it", async () => {
   // The second run's thread is looked up; the first's wait continues every
   // run, and the second sends its answer itself.
@@ -993,11 +1254,15 @@ it("answers and reports left runs of a conversation threads share in their menti
       (filters?.[0]?.ids ?? [root.id]).includes(event.id),
     ),
   );
-  await deliver({ type: "mention", event: reply("2", "@Pi look") }).run;
-  expect(published).toHaveLength(0);
+  // A turn that could not list them hands nothing over, and says so.
+  await deliver({ type: "mention", event: reply("2", "@Pi look") }).run.catch(
+    () => undefined,
+  );
+  await vi.waitFor(() => expect(published).toHaveLength(1));
+  expect(published[0]?.content).toContain("busy");
   await deliver({ type: "mention", event: reply("3", "@Pi again") }).run;
-  await vi.waitFor(() => expect(published).toHaveLength(2));
-  expect(published.map(({ content }) => content)).toEqual([
+  await vi.waitFor(() => expect(published).toHaveLength(3));
+  expect(published.slice(1).map(({ content }) => content)).toEqual([
     "Finished the old one.",
     expect.stringContaining("400 Bad model"),
   ]);
@@ -1025,8 +1290,10 @@ it("waits for a turn again in a new host when the host stops", async () => {
   await turn.run;
   await expect(settled(turn)).resolves.toEqual({ ok: true });
   expect(host.processes).toHaveLength(2);
+  // What the stopped host left for no one is ended before the wait continues it.
   expect(host.processes[1]?.requests.map((request) => request.method)).toEqual([
     "open",
+    "unfinished",
     "wait",
   ]);
   expect(published.map((item) => item.content)).toEqual(["Resumed."]);
@@ -1054,6 +1321,94 @@ it("hands a turn to the next host when the host stops before it answers", async
   expect(host.processes[1]?.requests[0]?.method).toBe("open");
 });
 
+it("ends what the stopped host left before the retried hand-over", async () => {
+  const left = { submission: 41, conversation: 7, request: "chan/one-off" };
+  const { host, deliver, settled } = setup({
+    deliver: (_params, process) => {
+      if (process === host.processes[0]) {
+        queueMicrotask(() => process.exit(1, "Killed\n"));
+        return new Promise(() => {});
+      }
+      return { submission: 1, conversation: CONVERSATION };
+    },
+    // Listed only once the next host opens.
+    unfinished: (_params, process) => ({
+      submissions: process === host.processes[0] ? [] : [left],
+    }),
+    wait: () => ({ status: "done", answer: 5, text: "Done." }),
+  });
+  const turn = deliver({ type: "mention", event: reply("2", "@Pi look") });
+  await turn.run;
+  await expect(settled(turn)).resolves.toEqual({ ok: true });
+  expect(
+    host.processes[1]?.requests.map((request) => request.method).slice(0, 4),
+  ).toEqual(["open", "unfinished", "drop", "deliver"]);
+  expect(host.requests("drop")[0]?.params).toMatchObject({
+    submissions: [{ submission: 41, conversation: 7 }],
+  });
+});
+
+it.each(["mention", "standalone"] as const)(
+  "takes on left runs again after a %s turn gave up on its run",
+  async (kind) => {
+    let waits = 0;
+    const { host, deliver, settled, runOnce } = setup({
+      // The first two hosts stop while it waits, and the turn gives up.
+      wait: (_params, process) => {
+        if (++waits <= 2) {
+          queueMicrotask(() => process.exit(1, "Killed\n"));
+          return new Promise(() => {});
+        }
+        return { status: "done", answer: 5, text: "Done." };
+      },
+    });
+    if (kind === "mention") {
+      const first = deliver({ type: "mention", event: reply("2", "@Pi look") });
+      await first.run;
+      await expect(settled(first)).resolves.toMatchObject({ ok: false });
+    } else
+      await expect(runOnce("chan", "check")).resolves.toMatchObject({
+        ok: false,
+      });
+    expect(host.processes).toHaveLength(2);
+    const next = deliver({ type: "mention", event: reply("3", "@Pi again") });
+    await next.run;
+    await expect(settled(next)).resolves.toEqual({ ok: true });
+    const methods =
+      host.processes[2]?.requests.map((item) => item.method) ?? [];
+    expect(methods.indexOf("unfinished")).toBeGreaterThan(-1);
+    expect(methods.indexOf("unfinished")).toBeLessThan(
+      methods.indexOf("deliver"),
+    );
+  },
+);
+
+it("cancels a mention still being handed over when Interrupt is pressed", async () => {
+  const placing = deferred<unknown>();
+  let places = 0;
+  const { host, deliver, settled, published, runtime } = setup({
+    place: () => (++places === 1 ? placing.promise : { seen: 0 }),
+    wait: () => ({ status: "done", answer: 5, text: "Done." }),
+  });
+  const agent = { pubkey: self, config: DEFAULT_CONFIG } as unknown as Agent;
+  const thread = { channelId: channel, root: root.id };
+  const first = deliver({ type: "mention", event: reply("2", "@Pi look") });
+  await vi.waitFor(() => expect(host.requests("place")).toHaveLength(1));
+  expect(runtime.sessionState(agent, thread)).toBe("active");
+  const interrupting = runtime.session(agent, thread, "interrupt");
+  placing.resolve({ seen: 0 });
+  await first.run;
+  await interrupting;
+  await expect(settled(first)).resolves.toEqual({ ok: true });
+  expect(host.requests("deliver")).toEqual([]);
+  expect(published).toEqual([]);
+  // A later mention is handed over as usual.
+  const later = deliver({ type: "mention", event: reply("3", "@Pi again") });
+  await later.run;
+  await expect(settled(later)).resolves.toEqual({ ok: true });
+  expect(host.requests("deliver")).toHaveLength(1);
+});
+
 it("hands over no queued mention of an agent removed while it waited", async () => {
   const first = deferred<object>();
   let delivers = 0;
diff --git a/src/bundled/pi/runtime.ts b/src/bundled/pi/runtime.ts
index b69be4d1bd..c9217f6bfe 100644
--- a/src/bundled/pi/runtime.ts
+++ b/src/bundled/pi/runtime.ts
@@ -17,6 +17,7 @@ import {
   sessionTools,
   turnNote,
 } from "../../buzz-mcp/session-tools";
+import { SessionControls } from "../../features/agent-harness/session-control";
 import {
   MemoryReads,
   READ_TIMEOUT_MS,
@@ -31,6 +32,9 @@ import type {
   AgentSession,
   Delivery,
   Models,
+  SessionControlAction,
+  SessionPlace,
+  SessionState,
   StandaloneTurn,
   TurnResult,
 } from "../../features/agents/service";
@@ -110,6 +114,8 @@ type Waited =
   | { status: "unanswered"; reason?: string; detail?: string }
   | { status: "closed" };
 type Receipt = Pick<Delivery, "step" | "acknowledged" | "superseded"> & {
+  /** Its conversation key, which session controls act on. */
+  key: string;
   /** Where its activity shows: a thread, or a Solo chat's channel. */
   scope: string;
   /** Where a run's answer for it is posted. */
@@ -127,10 +133,16 @@ type Entry = {
   /** The agent's handle from its latest delivery, for the tools to act with. */
   handle?: AgentHandle;
   /** Each conversation's latest posting policy, as in Claude: when its turns
-   * were handed over, and when all of them settled. */
+   * were handed over, and when all of them settled. Interrupt cancels the
+   * handovers still pending. */
   policies: Map<
     string,
-    { topLevel: boolean; handed: Promise<unknown>; settled: Promise<unknown> }
+    {
+      topLevel: boolean;
+      handed: Promise<unknown>;
+      settled: Promise<unknown>;
+      cancel(): void;
+    }
   >;
   /** Mentions handed to Pi and not settled, by request. */
   receipts: Map<string, Receipt>;
@@ -142,8 +154,12 @@ type Entry = {
   sends: Map<number, number>;
   /** The answer each conversation last posted for a turn without `send`. */
   posted: Map<number, number>;
-  /** Runs a stopped app left in Pi's storage, waited for once. */
+  /** The latest taking-on of the runs Pi's storage holds that nothing here
+   * waits for; absent until the agent opens here, or after one failed. */
   adopted?: Promise<void>;
+  /** A turn stopped waiting for a run still unfinished in Pi's storage, so
+   * the next one takes on its left runs again. */
+  abandoned?: boolean;
   /** The agent was removed: its turns wait in its storage. */
   closed: boolean;
 };
@@ -151,6 +167,7 @@ type Entry = {
 export class PiRuntime {
   private readonly agents = new Map<string, Entry>();
   private readonly memoryReads = new MemoryReads();
+  private readonly controls = new SessionControls(() => this.notify());
   private readonly listeners = new Set<() => void>();
   /** Each agent's working conversations, as last read, for a stable view. */
   private readonly views = new Map<string, readonly number[]>();
@@ -178,6 +195,9 @@ export class PiRuntime {
     /** Desktop-local repository instructions; standalone hosts keep their own layout. */
     private readonly repositories: (scope: string | undefined) => string = () =>
       "",
+    /** The agent as a run holds it, outside any run (`Agents.handle`). */
+    private readonly handle: (pubkey: string) => AgentHandle | undefined = () =>
+      undefined,
   ) {
     this.pi = new Pi(() => this.spawn, {
       tool: (call) => this.tool(call),
@@ -318,6 +338,58 @@ export class PiRuntime {
     for (const listener of this.listeners) listener();
   }
 
+  /** A conversation is active from a delivery's handover, or a left run's
+   * taking-on, until its run settles: Interrupt reaches both. */
+  sessionState(agent: Agent, place: SessionPlace): SessionState {
+    const key = controlKey(agent, place);
+    const control = `${agent.pubkey}/${key}`;
+    if (this.controls.compacting(control)) return "compacting";
+    const receipts = this.agents.get(agent.pubkey)?.receipts.values() ?? [];
+    return this.controls.active(control) ||
+      [...receipts].some((receipt) => receipt.key === key)
+      ? "active"
+      : "idle";
+  }
+  async session(
+    agent: Agent,
+    place: SessionPlace,
+    action: SessionControlAction,
+  ) {
+    const key = controlKey(agent, place);
+    const control = `${agent.pubkey}/${key}`;
+    // Pi continues every run once asked anything: its left runs are taken on
+    // first, and posting their answers needs the agent's handle.
+    const act = async (signal?: AbortSignal) => {
+      const handle =
+        this.agents.get(agent.pubkey)?.handle ?? this.handle(agent.pubkey);
+      if (!handle)
+        throw new Error("The agent isn't connected yet. Try again shortly.");
+      const entry = this.entry(handle, config(agent.config));
+      entry.handle ??= handle;
+      await this.open(agent.pubkey);
+      await this.adopted(entry, agent.pubkey);
+      // An Interrupt while this prepared, or a left run just taken on here,
+      // stops a Compact or New: both act only on an idle conversation.
+      if (signal) {
+        signal.throwIfAborted();
+        if ([...entry.receipts.values()].some((taken) => taken.key === key))
+          throw new Error("Wait for the active turn to finish first.");
+      }
+      await this.pi.request("session", { agent: agent.pubkey, key, action });
+    };
+    if (action === "interrupt") {
+      this.controls.interrupt(control);
+      this.agents.get(agent.pubkey)?.policies.get(key)?.cancel();
+      return act();
+    }
+    if (this.sessionState(agent, place) !== "idle")
+      throw new Error("Wait for the active turn to finish first.");
+    return this.controls.run(control, action, async (signal) => {
+      signal.throwIfAborted();
+      await act(signal);
+    });
+  }
+
   dispose() {
     for (const entry of this.agents.values()) entry.closed = true;
     this.agents.clear();
@@ -378,12 +450,15 @@ export class PiRuntime {
     const request = event.id;
     let submitted!: { submission: number; conversation: number };
     let sends = 0;
-    const handed = (async () => {
+    // A delivery during a session action waits for it, past its handover.
+    const control = `${agent.pubkey}/${key}`;
+    let interrupted = false;
+    const handoff = this.controls.deliver(control, async () => {
       await wait;
+      if (interrupted) return;
       this.live(entry, agent.pubkey);
       await this.open(agent.pubkey);
-      entry.adopted ??= this.adopt(entry, agent.pubkey, request);
-      await entry.adopted;
+      await this.adopted(entry, agent.pubkey, { current: request });
       const { seen } = await this.pi.request<{ seen: number }>("place", {
         agent: agent.pubkey,
         key,
@@ -412,6 +487,7 @@ export class PiRuntime {
           : { channel, root };
       // Known before Pi has it, as it can say it joined a run first.
       entry.receipts.set(request, {
+        key,
         scope: topLevel ? channel : root,
         where,
         ...(step ? { step } : {}),
@@ -420,6 +496,7 @@ export class PiRuntime {
       });
       const picked = await this.turnOf(entry, agent.pubkey, model, effort);
       this.live(entry, agent.pubkey);
+      if (interrupted) return;
       submitted = await this.deliver(entry, agent.pubkey, {
         agent: agent.pubkey,
         key,
@@ -440,9 +517,12 @@ export class PiRuntime {
         )
       )
         this.notify();
-    })();
+    });
+    const handed = handoff.done;
     const settled = handed
       .then(async (): Promise<TurnResult> => {
+        // Interrupted before Pi had it, it says nothing.
+        if (!submitted) return { ok: true };
         const waited = await this.settle(
           entry,
           agent.pubkey,
@@ -476,6 +556,10 @@ export class PiRuntime {
       });
     const policy = {
       topLevel,
+      cancel() {
+        interrupted = true;
+        prior?.cancel();
+      },
       handed: handed.catch(() => undefined),
       settled: Promise.all([same ? prior?.settled : undefined, settled]).then(
         () => undefined,
@@ -487,12 +571,13 @@ export class PiRuntime {
     });
     // Deliveries keep their order: a later one waits on this handover, and
     // one with nothing to wait for is handed over before `run` returns.
-    if (!wait) await policy.handed;
+    if (!wait && !handoff.deferred) await policy.handed;
     // The turn outlives `run`, which the host is told so it shows it as working.
     workingUntil(settled);
     // Stop on its activity stops the run the request is in.
     interruptible?.(async () => {
       await handed;
+      if (!submitted) return;
       await this.pi.request("abort", {
         agent: agent.pubkey,
         conversation: submitted.conversation,
@@ -519,8 +604,7 @@ export class PiRuntime {
     let request: string | undefined;
     try {
       await this.open(agent.pubkey);
-      entry.adopted ??= this.adopt(entry, agent.pubkey);
-      await entry.adopted;
+      await this.adopted(entry, agent.pubkey);
       signal.throwIfAborted();
       const text = await prompt();
       if (text === undefined) return { ok: true };
@@ -551,6 +635,8 @@ export class PiRuntime {
         ? { ok: false, error: unanswered(waited) }
         : { ok: true };
     } catch (error) {
+      // Pi may hold its run, which nothing waits for now.
+      if (request) entry.abandoned = true;
       return { ok: false, error: errorText(error) };
     } finally {
       if (stop) signal.removeEventListener("abort", stop);
@@ -681,34 +767,56 @@ export class PiRuntime {
       return await this.pi.request("deliver", params);
     } catch (error) {
       if (!(error instanceof PiStopped) || entry.closed) throw error;
-      await this.open(pubkey);
-      this.live(entry, pubkey);
-      return this.pi.request("deliver", params);
+      try {
+        await this.reopen(entry, pubkey);
+        return await this.pi.request("deliver", params);
+      } catch (again) {
+        // The stopped host may have taken it, and nothing waits for it now.
+        entry.abandoned = true;
+        throw again;
+      }
     }
   }
 
-  /** Takes on the runs an earlier app left unfinished in Pi's storage, which
-   * any delivery continues, in whatever thread: they show as working, and
-   * their answers are posted and failures reported as a delivery's are, but
-   * Stop is not restored. Standalone turns' runs are ended instead. `current`
-   * is the request being delivered, which its own run waits for. */
+  /** Takes on the runs Pi's storage holds that nothing here waits for before
+   * anything continues them: once the agent opens here, and `again` once it is
+   * opened again, after any taking-on under way. One that fails stops the
+   * turn that needed it, and the next turn does it again. */
+  private adopted(
+    entry: Entry,
+    pubkey: string,
+    { current, again }: { current?: string; again?: boolean } = {},
+  ) {
+    if (entry.adopted && !again && !entry.abandoned) return entry.adopted;
+    entry.abandoned = false;
+    const prior = entry.adopted?.catch(() => undefined);
+    const adopted = prior
+      ? prior.then(() => this.adopt(entry, pubkey, current))
+      : this.adopt(entry, pubkey, current);
+    entry.adopted = adopted;
+    adopted.catch(() => {
+      if (entry.adopted === adopted) delete entry.adopted;
+    });
+    return adopted;
+  }
+
+  /** Takes on the runs an earlier app, or a host that stopped, left
+   * unfinished in Pi's storage, which any delivery continues, in whatever
+   * thread: they show as working, and their answers are posted and failures
+   * reported as a delivery's are, but Stop is not restored. Standalone turns'
+   * runs are ended instead, and rejects when they could not be, as Pi would
+   * continue them. `current` is the request being delivered, which its own
+   * run waits for. */
   private async adopt(entry: Entry, pubkey: string, current?: string) {
-    let unfinished: readonly {
-      submission: number;
-      conversation: number;
-      request: string;
-      context?: { channel: string; root?: string; topLevel?: boolean };
-    }[];
-    try {
-      ({ submissions: unfinished } = await this.pi.request<{
-        submissions: typeof unfinished;
-      }>("unfinished", { agent: pubkey }));
-    } catch (error) {
-      // Listed again at the next delivery.
-      delete entry.adopted;
-      console.warn("Pi could not list an agent's unfinished runs", error);
-      return;
-    }
+    const { submissions: unfinished } = await this.pi.request<{
+      submissions: readonly {
+        submission: number;
+        conversation: number;
+        request: string;
+        context?: { channel: string; root?: string; topLevel?: boolean };
+      }[];
+    }>("unfinished", { agent: pubkey });
+    const taken: Receipt[] = [];
     const waits: (() => Promise<void>)[] = [];
     const dropped: { submission: number; conversation: number }[] = [];
     for (const { submission, conversation, request, context } of unfinished) {
@@ -737,12 +845,17 @@ export class PiRuntime {
         topLevel: !!context.topLevel,
       };
       const receipt: Receipt = {
+        // Only a thread's own conversation has a root.
+        key: context.root
+          ? `${context.channel}/${context.root}`
+          : context.channel,
         scope: context.topLevel ? context.channel : root,
         where,
       };
       // A delivery of the same request takes it over, and says how it ended.
       const ours = () => entry.receipts.get(request) === receipt;
       entry.receipts.set(request, receipt);
+      taken.push(receipt);
       const active = entry.active.get(conversation) ?? new Map();
       active.set(receipt.scope, receipt);
       entry.active.set(conversation, active);
@@ -776,13 +889,15 @@ export class PiRuntime {
     }
     // Only once every run's sends are counted: a wait continues all of them,
     // and one could send while a later run's thread is still looked up.
-    // Ending the others continues none.
+    // Ending the others continues none; until they are ended, nothing waits.
     if (dropped.length)
-      await this.pi
-        .request("drop", { agent: pubkey, submissions: dropped })
-        .catch((error: unknown) =>
-          console.warn("Pi could not end left standalone turns", error),
-        );
+      try {
+        await this.pi.request("drop", { agent: pubkey, submissions: dropped });
+      } catch (error) {
+        for (const [request, receipt] of entry.receipts)
+          if (taken.includes(receipt)) this.forget(entry, request);
+        throw error;
+      }
     for (const wait of waits) void wait();
     this.notify();
   }
@@ -801,6 +916,15 @@ export class PiRuntime {
     }
   }
 
+  /** Opens the agent again after its host stopped or its storage failed.
+   * Pi continues every run once asked anything, so what was left for no one
+   * is ended first. */
+  private async reopen(entry: Entry, pubkey: string) {
+    await this.open(pubkey);
+    this.live(entry, pubkey);
+    await this.adopted(entry, pubkey, { again: true });
+  }
+
   private async open(pubkey: string) {
     await this.pi.request("open", {
       agent: pubkey,
@@ -817,19 +941,24 @@ export class PiRuntime {
     pubkey: string,
     submission: number,
   ): Promise<Waited> {
+    // Given up, the run is still unfinished, and nothing waits for it now.
+    const abandon = (error: unknown) => {
+      entry.abandoned = true;
+      return error;
+    };
     for (let attempt = 0; ; attempt++) {
       let waited: Waited | { status: "reopen" };
       try {
-        if (attempt) await this.open(pubkey);
+        if (attempt) await this.reopen(entry, pubkey);
         waited = await this.pi.request("wait", { agent: pubkey, submission });
       } catch (error) {
         if (attempt || !(error instanceof PiStopped) || entry.closed)
-          throw error;
+          throw abandon(error);
         continue;
       }
       if (waited.status !== "reopen") return waited;
       if (attempt || entry.closed)
-        throw new Error("Pi's storage for this agent keeps failing");
+        throw abandon(new Error("Pi's storage for this agent keeps failing"));
     }
   }
 
@@ -1069,6 +1198,13 @@ export class PiRuntime {
   }
 }
 
+/** The conversation key session controls in `place` act on, as `key` makes
+ * it: a DM or Solo chat has no root here. */
+const controlKey = (agent: Agent, place: SessionPlace) =>
+  config(agent.config).scope === "channel" || !place.root
+    ? place.channelId
+    : `${place.channelId}/${place.root}`;
+
 /** Why Pi left a request unanswered, in words. */
 function unanswered({
   reason,
diff --git a/src/bundled/pi/transcript.test.ts b/src/bundled/pi/transcript.test.ts
index 63708fea8f..e04fc4e555 100644
--- a/src/bundled/pi/transcript.test.ts
+++ b/src/bundled/pi/transcript.test.ts
@@ -201,6 +201,53 @@ it("replaces an interrupted run's partial answer with the one that continues it,
   expect(next?.stopReason).toBeUndefined();
 });
 
+it("shows a Stop during a tool call as stopped, unless the run went on", () => {
+  const call = (id: number, name: string) =>
+    assistant(
+      id,
+      [{ type: "toolCall", id: `c${id}`, name, arguments: { command: name } }],
+      { stopReason: "toolUse" },
+    );
+  const cut = (id: number, call: string) => ({
+    ...result(id, call, "[error] Tool bash was aborted", true),
+    aborted: true,
+  });
+  const tool = (turn?: TranscriptTurn) =>
+    turn?.items.flatMap((item) => (item.type === "tool" ? [item.status] : []));
+  const [stopped, failed] = piTurns(
+    [
+      user(1, event("first")),
+      call(2, "bash"),
+      cut(3, "c2"),
+      user(4, event("second")),
+      call(5, "ls"),
+      result(6, "c5", "No such file", true),
+      assistant(7, [{ type: "text", text: "It failed" }], {
+        stopReason: "stop",
+      }),
+    ],
+    PLACE,
+    false,
+  );
+  expect(stopped?.stopReason).toBe("stopped");
+  expect(tool(stopped)).toEqual(["terminated"]);
+  expect(failed?.stopReason).toBeUndefined();
+  expect(tool(failed)).toEqual(["failed"]);
+  // Aborting one call alone lets the run answer after it.
+  const [continued] = piTurns(
+    [
+      user(1, event("first")),
+      call(2, "bash"),
+      cut(3, "c2"),
+      assistant(4, [{ type: "text", text: "Done" }], { stopReason: "stop" }),
+    ],
+    PLACE,
+    false,
+  );
+  expect(continued?.stopReason).toBeUndefined();
+  expect(tool(continued)).toEqual(["terminated"]);
+});
+
 it("reads pages, newest first, until one turn more than it shows", async () => {
   // Two entries a turn, a page of three entries at a time.
   const entries = Array.from({ length: (TURN_LIMIT + 5) * 2 }, (_, index) =>
diff --git a/src/bundled/pi/transcript.ts b/src/bundled/pi/transcript.ts
index 89cf451cc3..0dd856db0f 100644
--- a/src/bundled/pi/transcript.ts
+++ b/src/bundled/pi/transcript.ts
@@ -1,18 +1,20 @@
 // A Buzz conversation's Pi conversation, read from the agent's Pi durable
 // storage through the host and shown in the host's turn-transcript model. Pi
 // owns the entries; nothing here writes or keeps them.
-import type {
-  ToolItem,
-  Transcript,
-  TranscriptItem,
-  TranscriptTurn,
+import {
+  promptMessage,
+  promptSections,
+  type ToolItem,
+  type Transcript,
+  type TranscriptItem,
+  type TranscriptTurn,
 } from "../../features/agents/activity-transcript";
-import { prompt } from "../claude-code/transcript";
 import type { Pi } from "./pi";
 
 /** The latest turns shown; Pi's storage keeps the rest. */
 export const TURN_LIMIT = 50;
 const PREFIX = "buzz_";
+const STEER = "new-message-arrived-while-you-were-working";
 const KINDS: Record<string, string> = {
   read: "read",
   write: "edit",
@@ -46,6 +48,8 @@ type Message = Readonly<{
 export type PiEntry = Readonly<{
   id: number;
   kind: string;
+  /** A tool result Pi wrote for a call an abort cut off. */
+  aborted?: boolean;
   model?: readonly Message[];
 }>;
 /** The Pi conversation of a Buzz thread or channel. */
@@ -70,6 +74,22 @@ const text = (content: Message["content"]) =>
         .filter(Boolean)
         .join("\n");
 const str = (value: unknown) => (typeof value === "string" ? value : "");
+/** A Buzz prompt, or a steer wrapping one, as the turn's opening item. */
+function prompt(id: string, at: number, raw: string): TranscriptItem {
+  let sections = promptSections(raw);
+  const steer = sections.find((section) => section.tag === STEER);
+  if (steer) sections = promptSections(steer.body);
+  const { text, author } = promptMessage(sections);
+  return {
+    id,
+    at,
+    type: "prompt",
+    text: text || raw,
+    ...(author ? { author } : {}),
+    sections,
+    steer: !!steer,
+  };
+}
 /** Whether an entry is a prompt that opens a turn, not a steer into one. */
 const opens = ({ id, kind, model }: PiEntry) => {
   if (kind !== "pi.user") return false;
@@ -137,10 +157,17 @@ export function piTurns(
             type: "status",
             text: `The message above was not posted: ${text(message.content)}`,
           });
+        // A Stop in the tools phase leaves no aborted answer, only these
+        // results; an answer after them means the run went on.
+        if (entry.aborted) current.stopReason = "stopped";
         const tool = tools.get(message.toolCallId ?? "");
         if (!tool) return;
         tool.output = text(message.content);
-        tool.status = message.isError ? "failed" : "completed";
+        tool.status = entry.aborted
+          ? "terminated"
+          : message.isError
+            ? "failed"
+            : "completed";
         tool.completedAt = at;
         return;
       }