feat(pi): TUI launch mode + persona/model forwarding

AGENTS.md

@@ -76,6 +76,9 @@ with no silent fallback.
   (native in-process plugin injected via `OPENCODE_CONFIG_CONTENT`).
 - **`@cotal-ai/connector-hermes`** (`extensions/connector-hermes`): the Hermes (Nous Research)
   adapter; includes a Python sidecar.
+- **`@cotal-ai/pi`** (`extensions/pi`): native-embed adapter — embeds `MeshAgent` in the pi
+  agent's own process and drives its loop straight off the inbox, with true mid-turn `steer()`.
+  See [docs/agent-frameworks.md](docs/agent-frameworks.md).
 - **`@cotal-ai/cmux`** (`extensions/cmux`): the cmux integration: a driver over the cmux CLI
   plus a self-registering `cmux` Runtime and `TerminalLayout` provider.
 - **`@cotal-ai/tmux`** (`extensions/tmux`): the tmux integration: a driver over the tmux CLI

README.md

@@ -135,14 +135,16 @@ Full index: [docs/examples.md](docs/examples.md).
 
 <table>
 <tr>
-<td align="center" width="33%"><a href="extensions/connector-claude-code"><img src="assets/agents/claude-code.svg" height="44" alt=""><br><strong>Claude Code</strong></a><br><sub>installed plugin + hooks</sub></td>
-<td align="center" width="33%"><a href="extensions/connector-opencode"><img src="assets/agents/opencode.svg" height="44" alt=""><br><strong>OpenCode</strong></a><br><sub>native in-process plugin</sub></td>
-<td align="center" width="33%"><a href="extensions/connector-hermes"><img src="assets/agents/hermes.png" height="44" alt=""><br><strong>Hermes</strong></a><br><sub>gateway daemon + plugin</sub></td>
+<td align="center" width="25%"><a href="extensions/connector-claude-code"><img src="assets/agents/claude-code.svg" height="44" alt=""><br><strong>Claude Code</strong></a><br><sub>installed plugin + hooks</sub></td>
+<td align="center" width="25%"><a href="extensions/connector-opencode"><img src="assets/agents/opencode.svg" height="44" alt=""><br><strong>OpenCode</strong></a><br><sub>native in-process plugin</sub></td>
+<td align="center" width="25%"><a href="extensions/connector-hermes"><img src="assets/agents/hermes.png" height="44" alt=""><br><strong>Hermes</strong></a><br><sub>gateway daemon + plugin</sub></td>
+<td align="center" width="25%"><a href="extensions/pi"><img src="assets/agents/pi.svg" height="44" alt=""><br><strong>pi</strong></a><br><sub>native embed + live steer</sub></td>
 </tr>
 </table>
 
-They attach differently but expose the same `cotal_*` tools, and all three push, so a
-peer message wakes an idle agent the instant it arrives. Any agent that implements the
+They attach differently but expose the same `cotal_*` tools, and all four push, so a
+peer message wakes an idle agent the instant it arrives; pi additionally drives a live
+turn, folding a same-scope message into an in-flight one with `steer()`. Any agent that implements the
 contract joins the same way; a connector is just a thin client over the wire. Want one
 for an agent that isn't here yet?
 [Vote for the next connector](https://github.com/Cotal-AI/Cotal/discussions/80).

docs/agent-frameworks.md

@@ -0,0 +1,74 @@
+# Agent frameworks
+
+Besides Claude Code (see [claude-code-integration](claude-code-integration.md)), an agent
+built on an embeddable agent library can join a Cotal space as a native lateral peer. The
+first such adapter:
+
+| Extension | Framework | Language |
+|---|---|---|
+| `@cotal-ai/pi` | [pi coding agent](https://github.com/earendil-works/pi) | TypeScript |
+
+It joins the same mesh as the host-bound connectors and interoperates over the same
+subjects, presence, and delivery modes.
+
+## The pattern: a native embedded peer
+
+The adapter embeds a Cotal endpoint in the framework's own process — not a separate
+bridge. The shared piece is `MeshAgent` (in `@cotal-ai/connector-core`, the same runtime
+behind the Claude Code and Codex connectors): it owns the NATS connection, presence, and a
+stream-backed inbox, and emits `"incoming"` for each message. The adapter wires two things
+around it:
+
+1. **Inbound drives the loop.** The peer drives straight off the inbox via an `InboxTurn` —
+   the inbox is the single source of truth, no parallel buffer. On `"incoming"` it surfaces
+   the front message, flips presence to `working`, runs the agent, and delivers the reply on
+   the same delivery mode (DM/anycast → DM the sender by id; channel → multicast back to that
+   channel), then flips to `idle`. Delivery is **ack-on-surface**: the surfaced run is
+   `drainInbox`-acked only once the turn completes, so a crash or restart redelivers and
+   nothing is lost. The loop owns delivery (always routed right, sent once); runs are
+   serialized; the peer answers DMs and anycasts but only replies on a channel when named
+   (never its own echoes — ambient chatter is ack-dropped).
+2. **Mesh awareness as tools.** The model also gets read/presence tools via the framework's
+   tool mechanism (`defineTool()` for pi): `cotal_roster` (who's present) and `cotal_status`
+   (set its own status). Sending is left to the loop, so the model can't mis-route or
+   duplicate a reply.
+
+This makes the agent a real peer that wakes on traffic, like Claude Code — not a pull-only
+tool caller.
+
+**pi** is a clean fit, since it ships as an embeddable library: `@cotal-ai/pi` embeds
+`MeshAgent` alongside a pi `createAgentSession()` in one process, reads presence straight off
+the session's own event stream (`agent_start` → `working`, `tool_execution_start` → the
+running tool, `agent_end` → `idle`), and drives the loop with the session's own verbs —
+`prompt()` to wake an idle session into a turn and `steer()` to fold a *same-scope* message
+into a *live* one (true mid-turn drive, before the next LLM call; a different-scope message
+waits for its own turn, so a private DM is never folded into a channel reply), with `abort()`
+to interrupt. No external
+channel, host process, or keystrokes — inbound `"incoming"` calls a method on the embedded
+session. pi has no built-in permission gate, but in TUI mode (`PI_PEER_MODE=tui`, or the
+agent file's `peerMode: tui` frontmatter hint) any pi extension that calls `ctx.ui.*`
+(approval gates, prompts, selects, editors) is rendered live in the peer's tmux pane and
+becomes operator-answerable per-pane (e.g. an edit-approval gate); headless peers
+(unset/`headless`) run unattended (sandbox/containerize per pi's own guidance). It needs
+Node ≥22.19 and a provider key in the env.
+
+A `Connector` extension (`buildLaunch`) lets the manager spawn it: it launches the peer via
+`tsx` (dev) or `node` (built `dist/`) and forwards the launcher's identity (`COTAL_ID`), minted
+creds (`COTAL_CREDS`), and the agent file (`COTAL_AGENT_FILE`). At launch it parses the agent
+file and forwards the resolved model (`COTAL_MODEL` — the `cotal start --model` flag takes
+precedence over the file's `model:`); each runtime path reads the file's persona body and
+injects it as the agent's system prompt, so a spawned peer runs as its declared persona.
+Under auth the peer authenticates as the id the manager provisioned. The connector
+self-registers on import (`pi`); a composition root just imports it.
+
+## Running
+
+The pi connector has an example composition root under `examples/03-pi` (a manager that
+imports `@cotal-ai/pi`). See its README to run it; in short:
+
+```bash
+export ANTHROPIC_API_KEY=sk-...
+pnpm cotal up
+pnpm --filter @cotal-ai/example-03-pi manager
+pnpm cotal start --name pi1 --role research --agent pi
+```

docs/architecture.md

@@ -170,6 +170,10 @@ Implementations stay self-contained and never import each other: the `cli` drive
 purely over the mesh (`start`/`stop`/`ps` control requests), so neither imports the other. Only
 the example wires them together.
 
+Beyond the host-bound connectors, `@cotal-ai/pi` is a **native-embed** adapter: it embeds a
+Cotal endpoint directly in the pi agent's own process and drives its run loop off the inbox —
+see [agent-frameworks](agent-frameworks.md).
+
 ## Integration surfaces (Claude Code + OpenCode)
 
 Each target agent exposes the same four surfaces. The adapters share one runtime

examples/03-pi/README.md

@@ -0,0 +1,27 @@
+# Example 03 — pi coding-agent peers
+
+An agent built with the [pi coding agent](https://github.com/earendil-works/pi) joining a
+Cotal space as a native lateral peer. The adapter (`@cotal-ai/pi`) embeds a Cotal endpoint
+alongside a pi `createAgentSession()` in one process, reads presence off the session's own
+event stream, exposes the mesh to the model as `cotal_*` tools, and drives the session on
+inbound messages — `prompt()` wakes an idle session, `steer()` folds a same-scope message
+into a live turn (true mid-turn drive) — so it answers DMs and anycasts, and replies on a
+channel when mentioned by name.
+
+## Run
+
+```bash
+pnpm cotal up                                       # local NATS/JetStream (auth on; --open for a dev mesh)
+export ANTHROPIC_API_KEY=sk-...                     # the peer's pi session calls the model (any pi-supported provider key works)
+pnpm --filter @cotal-ai/example-03-pi manager       # start the manager
+
+# spawn a peer (either form works)
+pnpm cotal start --name pi1 --role research --agent pi
+pnpm cotal watch                                    # watch it join and reply
+pnpm cotal join --name me --role human              # then DM it: /dm pi1 hello
+```
+
+pi resolves its model and credentials via its own `AuthStorage` (falling back to provider
+keys in the environment, e.g. `ANTHROPIC_API_KEY`/`OPENAI_API_KEY`); pi has no permission
+gate, so sandbox/containerize a spawned peer per pi's own guidance. Requires Node ≥22.19. See
+[docs/agent-frameworks.md](../../docs/agent-frameworks.md) for how the adapter works.

examples/03-pi/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "@cotal-ai/example-03-pi",
+  "version": "0.0.0",
+  "private": true,
+  "license": "Apache-2.0",
+  "type": "module",
+  "scripts": {
+    "manager": "tsx src/manager.ts",
+    "typecheck": "tsc -p tsconfig.json --noEmit",
+    "build": "tsc -p tsconfig.json"
+  },
+  "dependencies": {
+    "@cotal-ai/core": "workspace:*",
+    "@cotal-ai/manager": "workspace:*",
+    "@cotal-ai/pi": "workspace:*"
+  },
+  "devDependencies": {
+    "tsx": "^4.22.4"
+  }
+}

examples/03-pi/src/manager.ts

@@ -0,0 +1,36 @@
+/**
+ * Composition root for example 03 (pi coding agent). Runs a manager that spawns
+ * pi peers into the space. Each spawn is a real pi agent session
+ * (extensions/pi) that embeds a Cotal endpoint and answers DMs, anycasts, and
+ * @-mentions on channels — waking an idle session with prompt() and folding
+ * same-scope traffic into a live turn with steer(). Importing the connector
+ * self-registers it as "pi"; we also alias it under "cotal" so a bare
+ * `cotal start --name x` (the default agent type) spawns one too.
+ */
+import { DEFAULT_SERVER, isReachable, registry, type Connector } from "@cotal-ai/core";
+import { Manager } from "@cotal-ai/manager";
+import { piConnector } from "@cotal-ai/pi"; // self-registers "pi"
+
+const cotalAlias: Connector = {
+  kind: "connector",
+  name: "cotal",
+  buildLaunch: piConnector.buildLaunch,
+};
+registry.register(cotalAlias);
+
+const space = process.env.COTAL_SPACE?.trim() || "demo";
+const server = process.env.COTAL_SERVERS?.trim() || DEFAULT_SERVER;
+
+if (!(await isReachable(server))) {
+  console.error(`Can't reach NATS at ${server}. Run: pnpm cotal up`);
+  process.exit(1);
+}
+
+const mgr = new Manager({ space, servers: server });
+await mgr.start();
+console.log(`example-03-pi manager up in space "${space}" — connectors: pi, cotal`);
+console.log(`console: ${mgr.consoleUrl}`);
+
+process.on("SIGINT", () => void mgr.stop().then(() => process.exit(0)));
+process.on("SIGTERM", () => void mgr.stop().then(() => process.exit(0)));
+await new Promise<void>(() => {});

examples/03-pi/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "extends": "../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "./src",
+    "outDir": "./dist"
+  },
+  "include": ["src"]
+}

extensions/connector-core/inbox-turn.smoke.ts

@@ -0,0 +1,120 @@
+/**
+ * Smoke test for the InboxTurn ack-on-surface helper (no NATS/LLM needed): drives it against
+ * a fake inbox — including the MAX_INBOX front-eviction the real MeshAgent does — and asserts
+ * the surface/ack invariants the embed adapters rely on.
+ *
+ *   pnpm smoke:inbox
+ */
+import { InboxTurn, type InboxSource } from "./src/inbox-turn.js";
+import type { InboxItem } from "./src/agent.js";
+
+function item(id: string, fromId: string, kind: InboxItem["kind"] = "dm"): InboxItem {
+  return { id, ts: 0, fromId, fromName: fromId, kind, mentionsMe: false, text: id };
+}
+
+/** A fake inbox that mirrors MeshAgent: ingest force-acks + evicts from the front past `cap`,
+ *  drainInbox acks by position, ackInbox acks by id (no-op for an absent id). */
+class FakeInbox implements InboxSource {
+  items: InboxItem[] = [];
+  acked: InboxItem[] = [];
+  constructor(private cap = Infinity) {}
+  ingest(it: InboxItem): void {
+    this.items.push(it);
+    if (this.items.length > this.cap) {
+      for (const ev of this.items.splice(0, this.items.length - this.cap)) this.acked.push(ev);
+    }
+  }
+  peekInbox(): InboxItem[] {
+    return [...this.items];
+  }
+  drainInbox(limit?: number): InboxItem[] {
+    const n = limit && limit > 0 ? Math.min(limit, this.items.length) : this.items.length;
+    const taken = this.items.splice(0, n);
+    this.acked.push(...taken);
+    return taken;
+  }
+  ackInbox(ids: string[]): InboxItem[] {
+    const wanted = new Set(ids);
+    const taken: InboxItem[] = [];
+    this.items = this.items.filter((p) => {
+      if (!wanted.has(p.id)) return true;
+      this.acked.push(p);
+      taken.push(p);
+      return false;
+    });
+    return taken;
+  }
+}
+
+function assert(cond: boolean, msg: string): void {
+  if (!cond) throw new Error(`FAIL: ${msg}`);
+}
+
+const ids = (xs: InboxItem[]): string => xs.map((x) => x.id).join(",");
+const sameScope = (a: InboxItem, b: InboxItem): boolean =>
+  a.fromId === b.fromId && a.kind === b.kind;
+
+// 1) drop leading non-actionable, start on the front, commit acks exactly the origin
+{
+  const fake = new FakeInbox();
+  fake.items = [item("echo", "self"), item("b", "alice")];
+  const turn = new InboxTurn(fake);
+  turn.drop((i) => i.fromId === "self");
+  assert(ids(fake.acked) === "echo", "drop ack-drops the self echo");
+  assert(turn.start()?.id === "b", "start surfaces the front actionable");
+  assert(turn.count === 1, "surfaced exactly the origin");
+  turn.commit();
+  assert(ids(fake.acked) === "echo,b", "commit acks the origin");
+  assert(fake.items.length === 0 && !turn.inFlight, "inbox drained, turn idle");
+}
+
+// 2) extend folds the front-contiguous same-scope run, stops at a different-scope gap
+{
+  const fake = new FakeInbox();
+  fake.items = [item("1", "alice"), item("2", "alice"), item("3", "bob"), item("4", "alice")];
+  const turn = new InboxTurn(fake);
+  assert(turn.start()?.id === "1", "origin = 1");
+  assert(ids(turn.extend(sameScope)) === "2", "folds only contiguous same-scope #2, stops at #3");
+  assert(turn.count === 2, "surfaced the 2-message run");
+  turn.commit();
+  assert(ids(fake.acked) === "1,2", "commit acks exactly the surfaced run [1,2]");
+  assert(ids(fake.items) === "3,4", "cross-scope #3 and gapped #4 stay on the stream");
+}
+
+// 3) abandon acks nothing — the surfaced run redelivers
+{
+  const fake = new FakeInbox();
+  fake.items = [item("x", "alice")];
+  const turn = new InboxTurn(fake);
+  turn.start();
+  turn.abandon();
+  assert(fake.acked.length === 0, "abandon acks nothing");
+  assert(ids(fake.items) === "x" && !turn.inFlight, "item stays on the stream; turn idle");
+}
+
+// 4) 200+ ambient burst mid-turn: the overflow evicts the in-flight prefix from the front;
+//    ack-by-id no-ops the evicted origin, acks the surviving folded peer, and never touches
+//    the newer messages that took the prefix's place
+{
+  const fake = new FakeInbox(200);
+  fake.ingest(item("origin", "alice"));
+  const turn = new InboxTurn(fake);
+  assert(turn.start()?.id === "origin", "origin surfaced");
+  fake.ingest(item("peer", "alice"));
+  assert(ids(turn.extend(sameScope)) === "peer", "folds the same-scope peer");
+  for (let i = 0; i < 199; i++) fake.ingest(item(`amb${i}`, "bob", "channel")); // 201 → evict 1
+  assert(
+    fake.acked.some((x) => x.id === "origin") && fake.items.some((x) => x.id === "peer"),
+    "overflow evicted+acked the origin; the folded peer survived",
+  );
+  const before = fake.acked.length;
+  turn.commit(); // ackInbox(["origin","peer"])
+  assert(fake.acked.length === before + 1, "commit acks only the survivor — evicted origin no-ops");
+  assert(!fake.items.some((x) => x.id === "peer"), "the survivor was acked by id");
+  assert(
+    fake.items.length === 199 && fake.items.every((x) => x.id.startsWith("amb")),
+    "all 199 newer ambient messages left untouched — none mis-acked",
+  );
+}
+
+console.log("INBOX-TURN SMOKE OK ✅");

extensions/connector-core/src/agent.ts

@@ -304,6 +304,24 @@ export class MeshAgent extends EventEmitter {
     return taken.map((p) => p.item);
   }
 
+  /** Ack + remove the buffered messages with these ids, wherever they sit in the inbox. An id
+   *  that's no longer buffered — already acked, e.g. force-evicted by the MAX_INBOX overflow — is
+   *  a harmless no-op. This lets a consumer ack exactly the messages it surfaced, immune to the
+   *  front-eviction that shifts positions out from under {@link drainInbox}. */
+  ackInbox(ids: string[]): InboxItem[] {
+    if (!ids.length) return [];
+    const wanted = new Set(ids);
+    const taken: InboxItem[] = [];
+    this.inbox = this.inbox.filter((p) => {
+      if (!wanted.has(p.item.id)) return true;
+      p.ack();
+      this.markHandled(p.item.id);
+      taken.push(p.item);
+      return false;
+    });
+    return taken;
+  }
+
   /** Record an id as surfaced/handled, for {@link ingest}'s commit-aware cross-path dedup. Bounded via
    *  two rotating windows: when the live set fills, it becomes the previous window and a fresh one
    *  starts — so memory stays ~2× the cap while the lookup horizon never shrinks below it. */