diff --git a/docs/mesh-view.md b/docs/mesh-view.md index 45b42d50..b360d094 100644 --- a/docs/mesh-view.md +++ b/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`) diff --git a/docs/watch-a-mesh.md b/docs/watch-a-mesh.md index 1da569a3..7455b033 100644 --- a/docs/watch-a-mesh.md +++ b/docs/watch-a-mesh.md @@ -61,6 +61,17 @@ kills behind a confirm (`y` graceful stop, `f` force-kill), and the `:` palette `spawn [name]`, `status `, `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`. diff --git a/implementations/cli/src/console/app.tsx b/implementations/cli/src/console/app.tsx index f2e5b134..a6a4f828 100644 --- a/implementations/cli/src/console/app.tsx +++ b/implementations/cli/src/console/app.tsx @@ -57,7 +57,7 @@ export function App({ canControl?: boolean; controlCtx?: Omit; }) { - 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; + 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); }; diff --git a/implementations/cli/src/console/commands.ts b/implementations/cli/src/console/commands.ts index 15547df9..7ba00942 100644 --- a/implementations/cli/src/console/commands.ts +++ b/implementations/cli/src/console/commands.ts @@ -29,6 +29,11 @@ export interface CommandCtx { control: (op: string, args: Record, tier: ControlTier) => Promise; /** 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; + /** 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] "); + 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> "); + await ctx.ensureParticipant(); await ctx.ep.anycast(m[1], m[2]); ctx.notify(`→ @${m[1]}`); }, diff --git a/implementations/cli/src/console/mesh.ts b/implementations/cli/src/console/mesh.ts index 8573bd26..ede0975e 100644 --- a/implementations/cli/src/console/mesh.ts +++ b/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(null); if (!viewRef.current) viewRef.current = new MeshView(ep, opts ?? {}); const [state, setState] = useState(() => 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 }; } diff --git a/implementations/cli/src/view/mesh-view.ts b/implementations/cli/src/view/mesh-view.ts index a2167cad..d1b71aae 100644 --- a/implementations/cli/src/view/mesh-view.ts +++ b/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..*`) 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..*`) 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..` 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..*`), deduped + * into `dmLog` — catches a reply that landed between the send and the tap being wired. */ + private async backfillInbox(selfId: string): Promise { + 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; diff --git a/packages/core/src/endpoint.ts b/packages/core/src/endpoint.ts index abe44450..bf19e7ba 100644 --- a/packages/core/src/endpoint.ts +++ b/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..*`) — 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 { + 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..` publish). */ + async startPresence(): Promise { + 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 { if (!this.doRegister || !this.kv) return; // observers watch but never publish their own record const p: Presence = {