feat(console): operator participant mode — agents can reply to you

docs/mesh-view.md

@@ -16,7 +16,9 @@ presence) lives in the [SPEC](../SPEC.md).
 Every surface is built on one **read-only observer**: a `CotalEndpoint` started with
 `consume: false, registerPresence: false, watchPresence: true`, invisible to peers, binding no
 durables, reading the space through the live tap plus history and presence-watch. No surface opens
-its own NATS connection, and none re-implements the wire semantics.
+its own NATS connection, and none re-implements the wire semantics. (The console can upgrade its
+observer into a **participant** the moment the operator sends, so agents can reply — see
+[watch a mesh](watch-a-mesh.md); a pure-watch session stays the invisible observer.)
 
 ## The model: `MeshView` (`@cotal-ai/cli`)
 

docs/watch-a-mesh.md

@@ -61,6 +61,17 @@ kills behind a confirm (`y` graceful stop, `f` force-kill), and the `:` palette
 `spawn <persona> [name]`, `status <agent>`, `ps`, and `purge` (type the space name to confirm,
 like the space delete).
 
+**Operator participant mode.** By default the operator is invisible, and a message it sends is
+one-way: an agent's DM reply has no inbox to land in. On the operator's **first send** (`:dm` /
+`:call` / `:ask` / `:msg`, or a compose) the console lazily upgrades itself: it registers
+presence so the operator joins the roster as a `role:"operator"` peer agents can reply to, and
+the DM lens starts capturing the operator's own inbox. Outbound DMs are recorded locally so a
+thread shows both sides, and the operator leaves cleanly (offline presence) on exit. This is
+`canWrite`-gated, so a pure-watch session never registers; it works where the console can send —
+an open mesh, or an auth mesh with a capable `--creds`. Under the auth-default read-only
+god-view cred the console cannot send at all, so participant mode never activates there (a
+dedicated auth `participant` cred profile is future work).
+
 The stream is line-oriented, so the signals stay out of it; it is just a timestamped log of
 presence changes and messages, ready for `grep`.
 

implementations/cli/src/console/app.tsx

@@ -57,7 +57,7 @@ export function App({
   canControl?: boolean;
   controlCtx?: Omit<ControlCtx, "space">;
 }) {
-  const mesh = useMesh(ep, { tapSubject });
+  const { state: mesh, activateParticipant, recordOutboundDm } = useMesh(ep, { tapSubject });
   const { exit } = useApp();
   const { stdout } = useStdout();
   const { focus, focusNext, focusPrevious } = useFocusManager();
@@ -230,6 +230,23 @@ export function App({
         ? "→ @" + c.toName
         : "↩ " + c.entry.from.name + (c.entry.delivery === "multicast" && c.entry.channel ? " #" + c.entry.channel : "");
 
+  // First time the operator sends anything, upgrade the read-only observer into a participant:
+  // register presence (so agents see + can reply to the operator) and reveal/capture its own
+  // inbox in the DM lens. Idempotent, canWrite-gated (a pure-watch session never calls it).
+  const activatedRef = useRef(false);
+  const ensureParticipant = useCallback(async () => {
+    if (activatedRef.current || !canWrite) return;
+    activatedRef.current = true;
+    try {
+      await ep.startPresence();
+      activateParticipant(ep.card.id, ep.card.name);
+      setNotice("you're on the roster now — replies show in the DM lens (d)");
+    } catch (e) {
+      activatedRef.current = false; // let the next send retry
+      setNotice("participant: " + (e as Error).message);
+    }
+  }, [ep, activateParticipant, canWrite]);
+
   // Send the in-progress compose over the live endpoint.
   const submitCompose = () => {
     const c = compose;
@@ -239,17 +256,24 @@ export function App({
     if (!text) return;
     const ok = (label: string) => () => setNotice(label);
     const fail = (e: unknown) => setNotice("send: " + (e as Error).message);
-    if (c.kind === "channel")
-      void ep.multicast(text, { channel: c.channel, mentions: mentionsIn(text) }).then(ok("→ #" + c.channel)).catch(fail);
-    else if (c.kind === "dm") void ep.unicast(c.toId, text).then(ok("→ " + c.toName)).catch(fail);
-    else {
-      const e = c.entry;
-      const send =
-        e.delivery === "multicast" && e.channel
-          ? ep.multicast(text, { channel: e.channel, replyTo: e.id, mentions: mentionsIn(text) })
-          : ep.unicast(e.from.id, text, { replyTo: e.id });
-      void send.then(ok("↩ " + e.from.name)).catch(fail);
-    }
+    void ensureParticipant().then(() => {
+      if (c.kind === "channel")
+        void ep.multicast(text, { channel: c.channel, mentions: mentionsIn(text) }).then(ok("→ #" + c.channel)).catch(fail);
+      else if (c.kind === "dm") {
+        recordOutboundDm(c.toId, text); // show our own side of the thread
+        void ep.unicast(c.toId, text).then(ok("→ " + c.toName)).catch(fail);
+      } else {
+        const e = c.entry;
+        let send: Promise<unknown>;
+        if (e.delivery === "multicast" && e.channel) {
+          send = ep.multicast(text, { channel: e.channel, replyTo: e.id, mentions: mentionsIn(text) });
+        } else {
+          recordOutboundDm(e.from.id, text); // show our own side of the thread
+          send = ep.unicast(e.from.id, text, { replyTo: e.id });
+        }
+        void send.then(ok("↩ " + e.from.name)).catch(fail);
+      }
+    });
   };
 
   // Run a typed palette line against the live endpoint.
@@ -268,6 +292,8 @@ export function App({
       notify: setNotice,
       control: ctl,
       confirmPurge: () => setConfirm({ kind: "purge", space: ep.space }),
+      ensureParticipant,
+      recordOutboundDm,
     };
     runCommand(line, ctx, !!canWrite, !!canControl);
   };

implementations/cli/src/console/commands.ts

@@ -29,6 +29,11 @@ export interface CommandCtx {
   control: (op: string, args: Record<string, unknown>, tier: ControlTier) => Promise<ControlReply>;
   /** Open the type-the-space-name purge confirm (the palette never purges directly). */
   confirmPurge: () => void;
+  /** Upgrade the console to a participant on the operator's first send (presence + own inbox), so
+   *  agents can reply. Idempotent, canWrite-gated. Await before the send. */
+  ensureParticipant: () => Promise<void>;
+  /** Record an outbound DM locally so the DM lens shows both sides of the thread. */
+  recordOutboundDm: (toId: string, text: string) => void;
 }
 
 export interface ConsoleCommand {
@@ -67,6 +72,7 @@ export const COMMANDS: ConsoleCommand[] = [
         text = m[2];
       }
       if (!text.trim()) return ctx.notify("usage: msg [#channel] <text>");
+      await ctx.ensureParticipant();
       await ctx.ep.multicast(text, { channel, mentions: mentionsIn(text) });
       ctx.notify(`→ #${channel}`);
     },
@@ -87,6 +93,8 @@ export const COMMANDS: ConsoleCommand[] = [
         throw e;
       }
       if (!id) return ctx.notify(`no agent "${m[1]}"`);
+      await ctx.ensureParticipant();
+      ctx.recordOutboundDm(id, m[2]);
       await ctx.ep.unicast(id, m[2]);
       ctx.notify(`→ ${m[1].replace(/^@/, "")}`);
     },
@@ -106,6 +114,8 @@ export const COMMANDS: ConsoleCommand[] = [
         throw e;
       }
       if (!id) return ctx.notify(`no agent "${name}"`);
+      await ctx.ensureParticipant();
+      ctx.recordOutboundDm(id, "👋 ping");
       await ctx.ep.unicast(id, "👋 ping");
       ctx.setMode("dm");
       ctx.notify(`called ${name}`);
@@ -119,6 +129,7 @@ export const COMMANDS: ConsoleCommand[] = [
     run: async (ctx, rest) => {
       const m = rest.match(/^@?(\S+)\s+([\s\S]+)/);
       if (!m) return ctx.notify("usage: ask <@role> <text>");
+      await ctx.ensureParticipant();
       await ctx.ep.anycast(m[1], m[2]);
       ctx.notify(`→ @${m[1]}`);
     },

implementations/cli/src/console/mesh.ts

@@ -4,7 +4,7 @@
 // them into React state. All the normalization (classification, coalescing, windowing, roster
 // sort, rates, derived signals) lives in MeshView — see docs/mesh-view.md.
 
-import { useEffect, useRef, useState } from "react";
+import { useCallback, useEffect, useRef, useState } from "react";
 import type { CotalEndpoint } from "@cotal-ai/core";
 import type { MeshSnapshot, MeshViewOptions } from "../view/mesh-view.js";
 import { MeshView } from "../view/mesh-view.js";
@@ -16,7 +16,16 @@ export type MeshState = MeshSnapshot;
 /** Focusable panes across the console (normal panels + the DM and topology lenses). */
 export type FocusId = "roster" | "feed" | "needsyou" | "dmpeers" | "dmthread" | "topo";
 
-export function useMesh(ep: CotalEndpoint, opts?: MeshViewOptions): MeshSnapshot {
+export interface UseMesh {
+  state: MeshSnapshot;
+  /** Turn the read-only model into participant mode (reveal the DM lens + capture the operator's
+   *  own inbox). Paired with the endpoint's `startPresence()` at the App layer. Idempotent. */
+  activateParticipant: (selfId: string, selfName: string) => void;
+  /** Record an outbound DM locally so a thread shows both sides. */
+  recordOutboundDm: (toId: string, text: string) => void;
+}
+
+export function useMesh(ep: CotalEndpoint, opts?: MeshViewOptions): UseMesh {
   const viewRef = useRef<MeshView | null>(null);
   if (!viewRef.current) viewRef.current = new MeshView(ep, opts ?? {});
   const [state, setState] = useState<MeshSnapshot>(() => viewRef.current!.snapshot());
@@ -31,5 +40,13 @@ export function useMesh(ep: CotalEndpoint, opts?: MeshViewOptions): MeshSnapshot
     };
     // eslint-disable-next-line react-hooks/exhaustive-deps
   }, []);
-  return state;
+  const activateParticipant = useCallback(
+    (id: string, name: string) => viewRef.current?.activateParticipant(id, name),
+    [],
+  );
+  const recordOutboundDm = useCallback(
+    (toId: string, text: string) => viewRef.current?.recordOutboundDm(toId, text),
+    [],
+  );
+  return { state, activateParticipant, recordOutboundDm };
 }

implementations/cli/src/view/mesh-view.ts

@@ -16,7 +16,7 @@ import type {
   PresenceStatus,
   ViewSpec,
 } from "@cotal-ai/core";
-import { deliveryOf, chatWildcard } from "@cotal-ai/core";
+import { deliveryOf, chatWildcard, unicastSubject } from "@cotal-ai/core";
 
 // ---- the model the surfaces render -----------------------------------------
 
@@ -179,6 +179,9 @@ export class MeshView extends EventEmitter {
   private connected = false;
   private error?: string;
   private dirty = true;
+  // Participant mode: the operator has spoken, so it's a presence-registered peer and its own
+  // inbox (`inst.<selfId>.*`) is captured — replies land in the DM lens. Off = pure observer.
+  private participant = false;
 
   private readonly window: number;
   private readonly tapSubject?: string;
@@ -249,7 +252,7 @@ export class MeshView extends EventEmitter {
       status: {
         connected: this.connected,
         space: this.ep.space,
-        dmVisible: !this.chatOnly,
+        dmVisible: !this.chatOnly || this.participant,
         error: this.error,
       },
       signals: this.signals(),
@@ -351,6 +354,58 @@ export class MeshView extends EventEmitter {
     if (this.dmLog.length > DM_LOG_CAP) this.dmLog.splice(0, this.dmLog.length - DM_LOG_CAP);
   }
 
+  /**
+   * Upgrade the model to participant mode: the operator has become a presence-registered peer, so
+   * its own inbox is now worth surfacing. Reveals the DM lens (`dmVisible`) and, on an auth mesh
+   * where the tap is narrowed to chat, opens a SECOND tap scoped to the operator's own inbox
+   * (`inst.<selfId>.*`) so agents' replies flow into `dmLog`. On an open mesh the main tap already
+   * covers the whole space, so no second tap — a duplicate would double-count (`recordDm` has no
+   * id-dedup). A one-shot own-inbox backfill catches a reply that raced activation (best-effort;
+   * skips under a cred that can't read the DM stream, like `prefillDms`).
+   */
+  activateParticipant(selfId: string, selfName: string): void {
+    if (this.participant) return;
+    this.participant = true;
+    this.byId.set(selfId, selfName); // resolve the operator's own id → name in threads
+    if (this.tapSubject) {
+      // Auth/chat-narrowed tap: the operator's inbox isn't covered, so add it.
+      this.ep.tap((subject, msg) => {
+        if (msg) this.ingest(subject, msg);
+      }, { subject: unicastSubject(this.ep.space, selfId, "*") });
+      void this.backfillInbox(selfId);
+    }
+    this.dirty = true;
+  }
+
+  /** Record an outbound DM the operator sent, so a thread shows both sides. Only needed on the
+   *  narrow-tap (auth) path: the operator's own `inst.<agent>.<selfId>` publish isn't covered by
+   *  the chat tap or the inbox tap. On an open mesh the whole-space tap already ingests it, so
+   *  recording here would double-count (`recordDm` has no id-dedup) — skip. */
+  recordOutboundDm(toId: string, text: string): void {
+    if (!this.tapSubject) return; // open mesh: the whole-space tap already captured it
+    this.recordDm({ ts: Date.now(), from: this.ep.ref(), toId, text });
+    this.dirty = true;
+  }
+
+  /** Best-effort one-shot backlog of the operator's own inbox (scoped `inst.<selfId>.*`), deduped
+   *  into `dmLog` — catches a reply that landed between the send and the tap being wired. */
+  private async backfillInbox(selfId: string): Promise<void> {
+    let msgs: CotalMessage[];
+    try {
+      msgs = await this.ep.inboxHistory(selfId, { limit: DM_LOG_CAP });
+    } catch {
+      return;
+    }
+    for (const m of msgs) {
+      if (!m.to) continue;
+      const dup = this.dmLog.some((d) => d.from.id === m.from.id && d.toId === m.to && d.ts === m.ts);
+      if (dup) continue;
+      this.recordDm({ ts: m.ts, from: m.from, toId: m.to, text: bodyText(m) });
+    }
+    this.dmLog.sort((a, b) => a.ts - b.ts);
+    this.dirty = true;
+  }
+
   /** Retain a peer-published view, deduped by message id and windowed (newest last). */
   private recordView(msg: CotalMessage, kind: DeliveryMode, spec: ViewSpec): void {
     if (this.viewSeen.has(msg.id)) return;

packages/core/src/endpoint.ts

@@ -207,7 +207,9 @@ export class CotalEndpoint extends EventEmitter {
   private readonly tls: boolean;
   private readonly heartbeatMs: number;
   private readonly ttlMs: number;
-  private readonly doRegister: boolean;
+  // Mutable: participant activation (startPresence) flips an observer into a presence-publishing
+  // peer after construction — the console upgrades a read-only observer once the operator sends.
+  private doRegister: boolean;
   private readonly doWatch: boolean;
   private readonly doWatchChannels: boolean;
   private readonly doConsume: boolean;
@@ -1186,6 +1188,17 @@ export class CotalEndpoint extends EventEmitter {
     );
   }
 
+  /** Fetch recent DMs addressed to ONE recipient (`inst.<selfId>.*`) — the operator's own inbox
+   *  backlog for participant activation. Scoped to a single target (unlike {@link dmHistory}'s
+   *  whole-space read), so it stays within a caller's own DM confidentiality. */
+  async inboxHistory(selfId: string, opts?: { limit?: number }): Promise<CotalMessage[]> {
+    return this.streamHistory(
+      dmStream(this.space),
+      unicastSubject(this.space, selfId, "*"),
+      opts?.limit ?? 100,
+    );
+  }
+
   /** Drain up to `limit` recent messages matching `subject` from a stream's backlog via a
    *  throwaway consumer. Fetches exactly the pending count (from consumer info) so it returns
    *  the moment the backlog is delivered — a plain `fetch({max_messages: limit})` would instead
@@ -2475,6 +2488,23 @@ export class CotalEndpoint extends EventEmitter {
     }
   }
 
+  /** Begin publishing presence AFTER construction — the participant upgrade a read-only observer
+   *  (e.g. the console) applies once its operator first speaks, so it becomes an addressable peer
+   *  and agents can reply. Idempotent. The presence KV is already open (any endpoint that watches
+   *  or registers opens it), and `connectAndBind` re-arms the publish+heartbeat off `doRegister` on
+   *  every reconnect, so flipping it here is all that's needed; `stop()`'s existing offline-leave
+   *  then fires for a clean deregister. A no-op when already registering (constructed with
+   *  `registerPresence`). Requires a presence-write grant — the caller's creds must allow it (open
+   *  mesh has none; an auth cred needs `$KV.<presence>.<id>` publish). */
+  async startPresence(): Promise<void> {
+    if (this.doRegister) return;
+    this.doRegister = true;
+    await this.publishPresence();
+    this.heartbeatTimer = setInterval(() => {
+      this.publishPresence().catch((e) => this.emit("error", e as Error));
+    }, this.heartbeatMs);
+  }
+
   private async publishPresence(): Promise<void> {
     if (!this.doRegister || !this.kv) return; // observers watch but never publish their own record
     const p: Presence = {