feat(oh-my-pi): Cotal connector — headless peer + interactive extension

examples/04-oh-my-pi/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "@cotal-ai/example-04-oh-my-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/oh-my-pi": "workspace:*"
+  },
+  "devDependencies": {
+    "tsx": "^4.22.4"
+  }
+}

examples/04-oh-my-pi/src/manager.ts

@@ -0,0 +1,28 @@
+/**
+ * Composition root for example 04 (oh-my-pi coding agent). Runs a manager that
+ * spawns oh-my-pi peers into the space. Each spawn is a real oh-my-pi agent
+ * session (extensions/connector-oh-my-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 "oh-my-pi".
+ */
+import { DEFAULT_SERVER, isReachable } from "@cotal-ai/core";
+import { Manager } from "@cotal-ai/manager";
+import "@cotal-ai/oh-my-pi"; // self-registers "oh-my-pi"
+
+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-04-oh-my-pi manager up in space "${space}" — connector: oh-my-pi`);
+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/04-oh-my-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/package.json

@@ -20,6 +20,9 @@
   "scripts": {
     "typecheck": "tsc -p tsconfig.json --noEmit",
     "build": "tsc -p tsconfig.json",
+    "smoke:inbox": "tsx inbox-turn.smoke.ts",
+    "smoke:reconnect-log": "tsx smoke/reconnect-log.smoke.ts",
+    "test": "pnpm run smoke:inbox && pnpm run smoke:reconnect-log",
     "prepublishOnly": "pnpm run build"
   },
   "dependencies": {

extensions/connector-core/smoke/reconnect-log.smoke.ts

@@ -0,0 +1,80 @@
+/**
+ * Reconnect-logging smoke (no NATS) — proves a mesh drop can't flood the host or corrupt its TUI.
+ * CotalEndpoint is an EventEmitter, so we drive MeshAgent's endpoint events directly (never
+ * connecting) and assert the anti-flood + off-terminal contract:
+ *   - a drop logs exactly ONE "connection lost" line; recovery logs exactly ONE "reconnected" line;
+ *   - the repeated endpoint errors during the outage (the TIMEOUT flood) are SUPPRESSED;
+ *   - a live-connection error still surfaces (ACL denial, etc.), and an identical repeat is deduped;
+ *   - an INJECTED logger receives every line and process.stderr is NEVER touched — so the in-process
+ *     OMP extension (which passes pi.logger) can't scribble on the shared terminal.
+ * Run: pnpm smoke:reconnect-log
+ */
+import { MeshAgent, type MeshLogLevel } from "../src/agent.js";
+import type { AgentConfig } from "../src/config.js";
+
+let failures = 0;
+function check(label: string, cond: boolean, extra?: unknown): void {
+  console.log(`${cond ? "✓" : "✗"} ${label}${cond ? "" : ` — ${JSON.stringify(extra)}`}`);
+  if (!cond) failures++;
+}
+
+const cfg: AgentConfig = {
+  space: "smoke",
+  name: "log-canary",
+  servers: "nats://127.0.0.1:1",
+  subscribe: [],
+  allowSubscribe: [],
+  allowPublish: [],
+  kind: "agent",
+  tls: false,
+};
+
+const lines: { msg: string; level: MeshLogLevel }[] = [];
+const agent = new MeshAgent(cfg, (msg, level) => lines.push({ msg, level: level ?? "info" }));
+const endpointErrors = () => lines.filter((l) => l.msg.includes("endpoint error"));
+
+// Guard: with a logger injected, NOTHING may reach the shared terminal.
+let stderrWrites = 0;
+const realWrite = process.stderr.write.bind(process.stderr);
+(process.stderr as unknown as { write: (s: string) => boolean }).write = () => {
+  stderrWrites++;
+  return true;
+};
+
+try {
+  const ep = agent.ep;
+
+  // Initial connect: the observer must NOT announce a "reconnect" (connectLoop logs the first connect).
+  ep.emit("connection", { connected: true });
+  check("initial connect logs no 'reconnected'", lines.filter((l) => l.msg.includes("reconnected")).length === 0, lines);
+
+  // Drop.
+  ep.emit("connection", { connected: false });
+  const lost = lines.filter((l) => l.msg.includes("connection lost"));
+  check("drop logs exactly one 'connection lost' at warn", lost.length === 1 && lost[0].level === "warn", lost);
+
+  // The flood: repeated endpoint errors while disconnected — the exact spam that broke the TUI.
+  for (let i = 0; i < 8; i++) ep.emit("error", new Error("TIMEOUT"));
+  check("outage endpoint errors are suppressed", endpointErrors().length === 0, lines);
+
+  // Recover.
+  ep.emit("connection", { connected: true });
+  const recon = lines.filter((l) => l.msg.includes("reconnected to the mesh"));
+  check("recovery logs exactly one 'reconnected' at info", recon.length === 1 && recon[0].level === "info", recon);
+
+  // A live-connection error DOES surface (genuine, actionable).
+  ep.emit("error", new Error("NATS permission denied: cannot publish"));
+  check("live error surfaces once", endpointErrors().length === 1, lines);
+
+  // An identical consecutive error is deduped (spam guard for a connected-but-flapping error).
+  ep.emit("error", new Error("NATS permission denied: cannot publish"));
+  check("identical consecutive live error is deduped", endpointErrors().length === 1, lines);
+
+  // The whole sequence never touched the terminal.
+  check("no writes to process.stderr (no TUI corruption)", stderrWrites === 0, stderrWrites);
+} finally {
+  (process.stderr as unknown as { write: typeof realWrite }).write = realWrite;
+}
+
+console.log(`\nRECONNECT-LOG SMOKE ${failures === 0 ? "OK ✅" : "FAILED ❌"}  (${lines.length} lines)`);
+process.exit(failures === 0 ? 0 : 1);