feat: agent→human console replies + exclusive board subscriptions

CHANGELOG.md

@@ -38,6 +38,12 @@ marketplace version bump + update.
 - **Console inbox — workspace-wide read** — `GET /inbox?all=1` aggregates unread human→agent
   messages across every scope in one call (peek-only; never advances an ack), so an agent can
   watch the whole workspace instead of polling each board scope-by-scope (#108).
+- **Agent → human replies** — `POST /reply` posts an agent message into a scope's console thread
+  (token-gated), broadcast live so the human sees what the agent is doing; replies share the inbox
+  ring but never count toward the agent's own unread (#108).
+- **Board subscriptions (takeover)** — `POST /subscribe` / `/unsubscribe` + `GET /subscriptions`
+  give one owner per scope so two watching sessions don't both act on a board; a fresh subscribe
+  takes over and the SSE `subscribe` broadcast names the evicted session (#108).
 
 ### Themes & visuals
 - **Theme system** — System / Dark / Light / Midnight / Paper, persisted; React Flow, Vega-Lite,
@@ -66,6 +72,11 @@ marketplace version bump + update.
 - **`inbox --all`** — aggregate unread console messages across every scope in one call (no
   `--project`/`--agent`); **`inbox --peek`** — read a scope without consuming it (no ack
   advance), for non-destructive scans (#108).
+- **`reply`** — post an agent→human message into a scope's console thread, so the human sees what
+  the agent is doing in real time (#108).
+- **`subscribe` / `unsubscribe` / `subscriptions`** — claim/release exclusive ownership of a board
+  (`--session <id>`) so multiple watching sessions don't overlap; takeover wins, `subscriptions`
+  lists who owns what (#108).
 - **`push`** now requires `--description` (the label shown in `list`) (#56).
 - **Default theme is `zinc-dark`** — beautiful-mermaid's light-oriented default rendered node
   labels near-black (invisible on a dark terminal); override with `--theme zinc-light` (#80).

packages/cli/src/cli.ts

@@ -40,6 +40,10 @@ Usage:
                               --follow/-f streams messages as they arrive (resilient long-poll, like tail -f; Ctrl-C to stop)
                               --peek reads without consuming (no ack advance); --all aggregates unread across every scope in one call
   termchart suggest [flags]   Push clickable suggestion chips to the human's console (--project --agent --items '[...]' / --item)
+  termchart reply [flags]     Post an agent→human message into a scope's console thread (--project --agent --message)
+  termchart subscribe [flags] Claim ownership of a board so sessions coordinate (advisory; --project --agent --session); takeover wins
+  termchart unsubscribe …     Release a board you own (--project --agent --session)
+  termchart subscriptions     List which boards are currently owned (--project optional; [--json])
   termchart template <cmd>    Reusable diagram templates: save --project --agent --name | list | get <id> | delete <id>
   termchart --version
 
@@ -376,6 +380,19 @@ if (isEntryPoint()) {
     import("./suggest.js")
       .then(({ suggest }) => suggest(argv.slice(1), { env: process.env }).then((code) => process.exit(code)))
       .catch((e) => { process.stderr.write(`suggest failed: ${e?.message ?? e}\n`); process.exit(1); });
+  } else if (argv[0] === "reply") {
+    import("./reply.js")
+      .then(({ reply }) => reply(argv.slice(1), { env: process.env }).then((code) => process.exit(code)))
+      .catch((e) => { process.stderr.write(`reply failed: ${e?.message ?? e}\n`); process.exit(1); });
+  } else if (argv[0] === "subscribe" || argv[0] === "unsubscribe") {
+    const mode = argv[0];
+    import("./subscribe.js")
+      .then(({ subscribe }) => subscribe(mode, argv.slice(1), { env: process.env }).then((code) => process.exit(code)))
+      .catch((e) => { process.stderr.write(`${mode} failed: ${e?.message ?? e}\n`); process.exit(1); });
+  } else if (argv[0] === "subscriptions") {
+    import("./subscribe.js")
+      .then(({ subscriptions }) => subscriptions(argv.slice(1), { env: process.env }).then((code) => process.exit(code)))
+      .catch((e) => { process.stderr.write(`subscriptions failed: ${e?.message ?? e}\n`); process.exit(1); });
   } else if (argv[0] === "template") {
     import("./template.js")
       .then(({ template }) => template(argv.slice(1), { env: process.env }).then((code) => process.exit(code)))

packages/cli/src/inbox.ts

@@ -34,13 +34,16 @@ function parse(argv: string[]): Args {
   return a;
 }
 
-interface InboxEvent { seq: number; ts: number; kind: "message" | "action"; text?: string; ref?: string; }
+interface InboxEvent { seq: number; ts: number; kind: "message" | "action" | "reply"; text?: string; ref?: string; }
 interface InboxResponse { events: InboxEvent[]; cursor: number; }
 interface UnreadScope { project: string; agent: string; scope: string; unread: number; cursor: number; acked: number; events: InboxEvent[]; }
 interface InboxAllResponse { scopes: UnreadScope[]; }
 
 function fmt(e: InboxEvent): string {
   if (e.kind === "action") return `[${e.seq}] action  ${e.ref ?? ""}`;
+  // `reply` is the agent's OWN message echoed back in a plain read — label it so the agent never
+  // mistakes it for new human input (the consume path returns replies; only unread/--all drop them).
+  if (e.kind === "reply") return `[${e.seq}] reply   ${e.text ?? ""}`;
   return `[${e.seq}] message ${e.text ?? ""}`;
 }
 

packages/cli/src/reply.ts

@@ -0,0 +1,56 @@
+import { EXIT_NO_VIEWER, isConnError, missingConfigMessage, unreachableMessage } from "./viewer-detect.js";
+
+export interface ReplyDeps {
+  env: Record<string, string | undefined>;
+}
+
+interface Args { project?: string; agent?: string; message?: string; error?: string; }
+
+function parse(argv: string[]): Args {
+  const a: Args = {};
+  for (let i = 0; i < argv.length; i++) {
+    const arg = argv[i];
+    if (arg === "--project") a.project = argv[++i];
+    else if (arg === "--agent") a.agent = argv[++i];
+    else if (arg === "--message" || arg === "-m") a.message = argv[++i];
+    else a.error = `Unknown flag: ${arg}`;
+  }
+  return a;
+}
+
+/**
+ * `termchart reply` — post an agent → human message into a scope's console (the agent → human side of
+ * the EXPERIMENTAL console, issue #108). Token-gated (the agent is posting, like suggest/push). The
+ * reply lands in the same console thread as the human's messages and shows up live on every open
+ * console, so the human can see what the agent is doing. Returns an exit code.
+ */
+export async function reply(argv: string[], deps: ReplyDeps): Promise<number> {
+  const a = parse(argv);
+  if (a.error) { process.stderr.write(a.error + "\n"); return 3; }
+  const base = deps.env.TERMCHART_VIEWER_URL;
+  const token = deps.env.TERMCHART_VIEWER_TOKEN;
+  if (!base || !token) { process.stderr.write(missingConfigMessage()); return EXIT_NO_VIEWER; }
+  if (!a.project || !a.agent) { process.stderr.write("--project and --agent are required.\n"); return 3; }
+  if (!a.message) { process.stderr.write("--message (-m) is required.\n"); return 3; }
+
+  try {
+    const r = await fetch(`${base}/reply`, {
+      method: "POST",
+      headers: { "content-type": "application/json", authorization: `Bearer ${token}` },
+      body: JSON.stringify({ project: a.project, agent: a.agent, text: a.message }),
+    });
+    if (!r.ok) {
+      const detail = (await r.text().catch(() => "")).trim();
+      process.stderr.write(`reply failed: HTTP ${r.status}${detail ? ` — ${detail}` : ""}\n`);
+      return 1;
+    }
+    return 0;
+  } catch (e) {
+    if (isConnError(e)) {
+      process.stderr.write(unreachableMessage(base, e));
+      return EXIT_NO_VIEWER;
+    }
+    process.stderr.write(`reply failed: ${(e as Error).message}\n`);
+    return 1;
+  }
+}

packages/cli/src/subscribe.ts

@@ -0,0 +1,112 @@
+import { EXIT_NO_VIEWER, isConnError, missingConfigMessage, unreachableMessage } from "./viewer-detect.js";
+
+export interface SubscribeDeps {
+  env: Record<string, string | undefined>;
+}
+
+interface Args { project?: string; agent?: string; session?: string; json: boolean; error?: string; }
+
+function parse(argv: string[]): Args {
+  const a: Args = { json: false };
+  for (let i = 0; i < argv.length; i++) {
+    const arg = argv[i];
+    if (arg === "--project") a.project = argv[++i];
+    else if (arg === "--agent") a.agent = argv[++i];
+    else if (arg === "--session") a.session = argv[++i];
+    else if (arg === "--json") a.json = true;
+    else a.error = `Unknown flag: ${arg}`;
+  }
+  return a;
+}
+
+interface Subscription { scope: string; owner: string; ts: number; }
+
+/**
+ * `termchart subscribe` / `unsubscribe` — claim or release ownership of a board so two watching
+ * sessions coordinate and don't both act on its console messages (issue #108). TAKEOVER model: a
+ * fresh `subscribe` always wins and evicts the prior owner; the response prints who (if anyone) was
+ * evicted. `unsubscribe` only releases if you still own it. Token-gated. Pair `--session <id>` with a
+ * stable per-session identifier. ADVISORY: ownership is cooperative — the server does not block a
+ * non-owner's writes, so a session should check `subscriptions` / honor a takeover and stand down.
+ * Returns an exit code.
+ */
+export async function subscribe(mode: "subscribe" | "unsubscribe", argv: string[], deps: SubscribeDeps): Promise<number> {
+  const a = parse(argv);
+  if (a.error) { process.stderr.write(a.error + "\n"); return 3; }
+  const base = deps.env.TERMCHART_VIEWER_URL;
+  const token = deps.env.TERMCHART_VIEWER_TOKEN;
+  if (!base || !token) { process.stderr.write(missingConfigMessage()); return EXIT_NO_VIEWER; }
+  if (!a.project || !a.agent) { process.stderr.write("--project and --agent are required.\n"); return 3; }
+  if (!a.session) { process.stderr.write("--session is required (a stable id for this watching session).\n"); return 3; }
+
+  try {
+    const r = await fetch(`${base}/${mode}`, {
+      method: "POST",
+      headers: { "content-type": "application/json", authorization: `Bearer ${token}` },
+      body: JSON.stringify({ project: a.project, agent: a.agent, session: a.session }),
+    });
+    if (!r.ok) {
+      const detail = (await r.text().catch(() => "")).trim();
+      process.stderr.write(`${mode} failed: HTTP ${r.status}${detail ? ` — ${detail}` : ""}\n`);
+      return 1;
+    }
+    if (mode === "subscribe") {
+      const data = (await r.json().catch(() => null)) as { scope: string; owner: string; tookOverFrom?: string | null } | null;
+      if (data?.tookOverFrom) process.stderr.write(`subscribed to ${data.scope} — took over from session ${data.tookOverFrom}\n`);
+      else process.stderr.write(`subscribed to ${a.project}/${a.agent}\n`);
+    } else {
+      // 204; the server flags a no-op (you weren't the owner) via a header.
+      if (r.headers.get("x-termchart-not-owner")) process.stderr.write(`not the owner of ${a.project}/${a.agent} — nothing released\n`);
+      else process.stderr.write(`unsubscribed from ${a.project}/${a.agent}\n`);
+    }
+    return 0;
+  } catch (e) {
+    if (isConnError(e)) {
+      process.stderr.write(unreachableMessage(base, e));
+      return EXIT_NO_VIEWER;
+    }
+    process.stderr.write(`${mode} failed: ${(e as Error).message}\n`);
+    return 1;
+  }
+}
+
+/**
+ * `termchart subscriptions` — list which boards are currently owned (and by which session) so a new
+ * session can pick an unclaimed board. Open read (URL-gated, like `list`). `--json` for the raw array.
+ */
+export async function subscriptions(argv: string[], deps: SubscribeDeps): Promise<number> {
+  const a = parse(argv);
+  if (a.error) { process.stderr.write(a.error + "\n"); return 3; }
+  const base = deps.env.TERMCHART_VIEWER_URL;
+  const token = deps.env.TERMCHART_VIEWER_TOKEN;
+  if (!base) { process.stderr.write(missingConfigMessage(false)); return EXIT_NO_VIEWER; }
+  const headers: Record<string, string> = token ? { authorization: `Bearer ${token}` } : {};
+
+  try {
+    const r = await fetch(`${base}/subscriptions`, { method: "GET", headers });
+    if (!r.ok) {
+      const detail = (await r.text().catch(() => "")).trim();
+      process.stderr.write(`subscriptions failed: HTTP ${r.status}${detail ? ` — ${detail}` : ""}\n`);
+      return 1;
+    }
+    const data = (await r.json().catch(() => null)) as { subscriptions: Subscription[] } | null;
+    if (!data || !Array.isArray(data.subscriptions)) {
+      process.stderr.write("subscriptions failed: unexpected response from viewer\n");
+      return 1;
+    }
+    if (a.json) {
+      process.stdout.write(JSON.stringify(data, null, 2) + "\n");
+      return 0;
+    }
+    for (const s of data.subscriptions) process.stdout.write(`${s.scope}  owned by ${s.owner}\n`);
+    process.stderr.write(`subscribed boards: ${data.subscriptions.length}\n`);
+    return 0;
+  } catch (e) {
+    if (isConnError(e)) {
+      process.stderr.write(unreachableMessage(base, e));
+      return EXIT_NO_VIEWER;
+    }
+    process.stderr.write(`subscriptions failed: ${(e as Error).message}\n`);
+    return 1;
+  }
+}

packages/cli/test/inbox.test.ts

@@ -66,6 +66,18 @@ describe("inbox", () => {
     expect((await received).path).toBe("/w/ws1/inbox?project=p&agent=a&since=5");
   });
 
+  it("labels a reply event as `reply`, not `message`, in a plain read", async () => {
+    const { url } = await stub({ events: [{ seq: 1, ts: 1, kind: "reply", text: "on it" }], cursor: 1 });
+    const out: string[] = [];
+    const so = vi.spyOn(process.stdout, "write").mockImplementation((s) => (out.push(String(s)), true));
+    const se = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
+    const code = await inbox(["--project", "p", "--agent", "a"], { env: env(url) });
+    so.mockRestore(); se.mockRestore();
+    expect(code).toBe(0);
+    expect(out.join("")).toContain("[1] reply");
+    expect(out.join("")).not.toContain("[1] message"); // an agent's own reply isn't shown as human input
+  });
+
   it("--peek forwards peek=1 (a non-consuming read)", async () => {
     const { url, received } = await stub(resp);
     const spy = vi.spyOn(process.stdout, "write").mockImplementation(() => true);

packages/cli/test/reply.test.ts

@@ -0,0 +1,73 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { createServer, type Server } from "node:http";
+import { type AddressInfo } from "node:net";
+import { reply } from "../src/reply.js";
+import { EXIT_NO_VIEWER } from "../src/viewer-detect.js";
+
+let server: Server | undefined;
+afterEach(() => server?.close());
+
+/** A POST stub that records the request (path, auth, body) and replies 200 with the event. */
+function stub(): Promise<{ url: string; received: Promise<{ path: string; auth?: string; body: string }> }> {
+  return new Promise((resolve) => {
+    let resolveReq!: (v: { path: string; auth?: string; body: string }) => void;
+    const received = new Promise<{ path: string; auth?: string; body: string }>((r) => (resolveReq = r));
+    server = createServer((req, res) => {
+      const chunks: Buffer[] = [];
+      req.on("data", (c) => chunks.push(c));
+      req.on("end", () => {
+        res.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify({ seq: 1, kind: "reply" }));
+        resolveReq({ path: req.url!, auth: req.headers.authorization, body: Buffer.concat(chunks).toString() });
+      });
+    });
+    server.listen(0, () => resolve({ url: `http://127.0.0.1:${(server!.address() as AddressInfo).port}`, received }));
+  });
+}
+
+function errStub(status: number, body: string): Promise<string> {
+  return new Promise((resolve) => {
+    server = createServer((_req, res) => res.writeHead(status, { "content-type": "text/plain" }).end(body));
+    server.listen(0, () => resolve(`http://127.0.0.1:${(server!.address() as AddressInfo).port}`));
+  });
+}
+
+const env = (url: string) => ({ TERMCHART_VIEWER_URL: `${url}/w/ws1`, TERMCHART_VIEWER_TOKEN: "tok" });
+
+describe("reply", () => {
+  it("POSTs the message to /reply with the bearer token", async () => {
+    const { url, received } = await stub();
+    const code = await reply(["--project", "p", "--agent", "a", "--message", "on it"], { env: env(url) });
+    const req = await received;
+    expect(code).toBe(0);
+    expect(req.path).toBe("/w/ws1/reply");
+    expect(req.auth).toBe("Bearer tok");
+    const body = JSON.parse(req.body);
+    expect(body).toMatchObject({ project: "p", agent: "a", text: "on it" });
+  });
+
+  it("accepts the -m alias", async () => {
+    const { url, received } = await stub();
+    const code = await reply(["--project", "p", "--agent", "a", "-m", "short"], { env: env(url) });
+    expect(code).toBe(0);
+    expect(JSON.parse((await received).body).text).toBe("short");
+  });
+
+  it("requires URL+token (no-viewer code) and project/agent/message (exit 3)", async () => {
+    expect(await reply(["--project", "p", "--agent", "a", "-m", "x"], { env: {} })).toBe(EXIT_NO_VIEWER);
+    const { url } = await stub();
+    const spy = vi.spyOn(process.stderr, "write").mockImplementation(() => true);
+    expect(await reply(["--agent", "a", "-m", "x"], { env: env(url) })).toBe(3); // no project
+    expect(await reply(["--project", "p", "--agent", "a"], { env: env(url) })).toBe(3); // no message
+    spy.mockRestore();
+  });
+
+  it("surfaces a server error (exit 1)", async () => {
+    const url = await errStub(401, "unauthorized");
+    const errs: string[] = [];
+    const spy = vi.spyOn(process.stderr, "write").mockImplementation((s) => (errs.push(String(s)), true));
+    const code = await reply(["--project", "p", "--agent", "a", "-m", "x"], { env: env(url) });
+    spy.mockRestore();
+    expect(code).toBe(1);
+    expect(errs.join("")).toContain("HTTP 401");
+  });
+});