From 5dcde207ed7c6193b8f4f3f610795ff65306bd35 Mon Sep 17 00:00:00 2001 From: Lanzelot1 <35410605+Lanzelot1@users.noreply.github.com> Date: Tue, 7 Jul 2026 00:05:56 -0700 Subject: [PATCH] =?UTF-8?q?feat(console):=20operator=20participant=20mode?= =?UTF-8?q?=20=E2=80=94=20agents=20can=20reply=20to=20you?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A console send was one-way: the observer is invisible (registerPresence false) with no inbox (consume false), so an agent's DM reply to the operator had nowhere to land. On the operator's first send the console now upgrades the observer in place — CotalEndpoint.startPresence() registers presence so agents see and can reply to the operator, and MeshView reveals the DM lens and captures the operator's own inbox (inst..>). Outbound DMs are recorded locally (auth path) for two-sided threads; a clean offline leave fires on exit. canWrite-gated, so a pure-watch session stays invisible; works on open meshes (or a capable --creds). Runtime upgrade, no rebuild, no provisioning or manager. --- docs/mesh-view.md | 4 +- docs/watch-a-mesh.md | 11 ++++ implementations/cli/src/console/app.tsx | 50 ++++++++++++----- implementations/cli/src/console/commands.ts | 11 ++++ implementations/cli/src/console/mesh.ts | 23 ++++++-- implementations/cli/src/view/mesh-view.ts | 59 ++++++++++++++++++++- packages/core/src/endpoint.ts | 32 ++++++++++- 7 files changed, 171 insertions(+), 19 deletions(-) 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 = {