diff --git a/changelog.d/feat-focus-mode.added.md b/changelog.d/feat-focus-mode.added.md new file mode 100644 index 000000000..45ec2e0f7 --- /dev/null +++ b/changelog.d/feat-focus-mode.added.md @@ -0,0 +1,38 @@ +- `focus` tool (opt-in via `FrameworkConfig.focus.enabled`): the primary + resident narrows attention to one channel for a bounded time. Every other + channel and DM is held — stored with `metadata.focusHeld` and excluded from + the resident's view, never woken on, and dropped by the wake gate before + any debounce policy can batch it. Addressed messages elsewhere get one + host-authored automatic reply per (channel, author) per epoch. Focus ends + on `mode: "end"`, at its deadline (durable across restarts), or when an + expired epoch is found at boot; ending delivers a per-channel + `` of the newest `backlogCap` held messages and points at + the resident's actual history/backscroll tools for the rest. `mode: + "check"` reports held counts (and peeks one channel) without ending; + `enter` while focused re-targets, delivering the new channel's backlog. +- While focused, plain speech lands in the focus channel: it outranks the + trigger channel and the global most-recent-inbound fallback for every + turn, a successful `enter` moves the current turn's prose there too + (announced in the tool result), and held inbound neither retargets the + fallback locus nor qualifies for the mid-turn addressed re-pin. +- Review hardening: channel identity is `(serverId, channelId)` throughout + (hold, gate, `enter` with an optional `serverId`, server-qualified + autoreply publish); the gate keeps each queued event's metadata so the + focus-entry purge, delivery-time re-check and live evaluation derive a + push event's channel identically; a held message never reaches a live + turn through the mid-turn deferred injection; held messages still deferred + behind the current turn are part of the unfocus backlog, and a deferred + focus dump makes the deferred queue durable so a crash cannot lose it; the + autoreply speaks only in channels the resident is open in and in DMs, and + a failed publish releases the per-author allowance; channels under + tune-out are left to the subconscious (and cannot be focused); the tool is + not offered under per-channel conversation routing; a persisted epoch + still ends at boot after `focus.enabled` is turned off; configured maxima + are floored/capped integers; media blocks are named in the dump. + Speech under focus is routed server-qualified (a same-id channel on + another server cannot receive it); automatic replies that published are + recorded in the lifecycle log, so once-per-author survives a restart and + a failed publish is retried, and a publish settling after its epoch ended + touches nothing; a held message awaiting its dump counts as unread for + event coalescing (an edit replaces it in place); wakes already queued for + a channel are dropped when that channel becomes held. diff --git a/src/focus/coordinator.ts b/src/focus/coordinator.ts new file mode 100644 index 000000000..dc00ea54d --- /dev/null +++ b/src/focus/coordinator.ts @@ -0,0 +1,825 @@ +/** + * FocusCoordinator — a bounded, agent-chosen attention narrowing. + * + * While focused, the resident receives traffic from exactly ONE channel + * (identity = (serverId, channelId), as in the registry). Everything else + * with a channel identity (other channels, DMs) is HELD: stored in the + * shared slot with an ingestion-time stamp (`metadata.focusHeld = { epochId }`) + * that the residents' viewFilter excludes permanently, never woken on, and + * never injected into a live turn. Addressed messages (mentions, replies, + * DMs) in channels the resident is open in get one deterministic + * host-authored autoreply per (channel, author) per epoch so the mention-er + * knows the resident saw nothing. Focus ends by agent tool, by operator, + * or when its deadline elapses; ending delivers the held backlog per + * channel — the newest `backlogCap` messages raw, the rest pointed at the + * resident's history/backscroll tools. + * + * Scope, stated plainly: the hold is host-wide — it narrows the shared + * window every resident on this host reads (the subconscious reads + * unfiltered). Events with no channel identity (heartbeats, non-chat + * pushes) pass. Channels under tune-out (#77) stay with the subconscious + * and are neither held nor auto-replied here. + * + * Sibling of the tune-out coordinator: same stamp-and-filter exclusion, + * same lifecycle-log durability, same deadline pattern — but per-resident + * rather than per-channel, no subconscious, and an inverted predicate. + * + * Voice doctrine: host-authored framing is bracket-styled system text + * (`[Focus …]`, ``), never voiced as the resident. The + * in-channel autoreply is explicitly marked as automatic. + */ + +import { randomUUID } from 'node:crypto'; +import type { ContentBlock } from '@animalabs/membrane'; +import type { StoredMessage } from '@animalabs/context-manager'; +import type { ChannelRegistry, FocusParams } from '../mcpl/channel-registry.js'; +import type { ToolDefinition } from '../types/index.js'; +import { formatZonedTime } from '../timezone.js'; + +/** Configuration for focus mode (FrameworkConfig.focus). */ +export interface FocusConfig { + /** + * Master switch: exposes the `focus` tool to the primary resident. The + * coordinator itself runs whenever a channel subsystem exists, so a focus + * epoch persisted before the switch was turned off still ends (and + * delivers its backlog) at the next boot instead of stranding its held + * messages behind the view filter. + */ + enabled: boolean; + /** Duration when the tool call names none (default 1800 s). */ + defaultDurationSeconds?: number; + /** Hard ceiling on a single focus epoch (default 14400 s = 4 h; floored + * at 60 s, capped at 7 days; integers). */ + maxDurationSeconds?: number; + /** Per-channel raw backlog delivered at unfocus (default 20). */ + defaultBacklogCap?: number; + /** Ceiling on backlogCap (default 200; integer ≥ 0). */ + maxBacklogCap?: number; + /** Post an automatic reply to addressed messages while focused (default + * true). Replies go only into channels the resident is open in and into + * DMs — never into a channel the resident chose to stay out of. */ + autoReply?: boolean; + /** + * Autoreply text. Placeholders: `{name}` (resident), `{until}` (local + * wall-clock deadline), `{remaining}` (e.g. "25m"), `{channel}` (focus + * channel label). Default: see DEFAULT_AUTOREPLY. + */ + autoReplyTemplate?: string; +} + +export const FOCUS_DEFAULTS = { + defaultDurationSeconds: 1800, + maxDurationSeconds: 14_400, + defaultBacklogCap: 20, + maxBacklogCap: 200, + minDurationSeconds: 60, + /** Absolute ceiling on any configured maximum: a week. */ + hardMaxDurationSeconds: 7 * 24 * 3600, +} as const; + +export const DEFAULT_AUTOREPLY = + '[Automatic reply] {name} is in focus mode for about {remaining} more ' + + '(until ~{until}). Your message has been held and will be read when focus ends.'; + +export const FOCUS_TOOL_NAME = 'focus'; + +/** + * Effective limits from a config: every maximum is an integer, the duration + * maximum is floored at the 60 s minimum and capped at a week, and each + * default is clamped into its range — so a recipe cannot configure an epoch + * the tool description says cannot exist, or a fractional cap that + * `slice(-0.5)` would turn into "deliver everything". + */ +export function resolveFocusLimits(cfg: FocusConfig): { + minDuration: number; maxDuration: number; defaultDuration: number; + maxCap: number; defaultCap: number; +} { + const int = (v: number | undefined, fallback: number): number => + typeof v === 'number' && Number.isFinite(v) ? Math.floor(v) : fallback; + const minDuration = FOCUS_DEFAULTS.minDurationSeconds; + const maxDuration = Math.min( + FOCUS_DEFAULTS.hardMaxDurationSeconds, + Math.max(minDuration, int(cfg.maxDurationSeconds, FOCUS_DEFAULTS.maxDurationSeconds)), + ); + const defaultDuration = Math.min(maxDuration, Math.max(minDuration, int(cfg.defaultDurationSeconds, FOCUS_DEFAULTS.defaultDurationSeconds))); + const maxCap = Math.max(0, int(cfg.maxBacklogCap, FOCUS_DEFAULTS.maxBacklogCap)); + const defaultCap = Math.min(maxCap, Math.max(0, int(cfg.defaultBacklogCap, FOCUS_DEFAULTS.defaultBacklogCap))); + return { minDuration, maxDuration, defaultDuration, maxCap, defaultCap }; +} + +export function buildFocusToolDefinition(cfg: FocusConfig): ToolDefinition { + const limits = resolveFocusLimits(cfg); + return { + name: FOCUS_TOOL_NAME, + description: + 'Narrow your attention to ONE channel for a bounded time. While focused, ' + + 'only that channel reaches you; every other channel and DM is held (stored, ' + + 'not shown, no wake). People who address you in a channel you are open in ' + + 'get an automatic reply saying you are in focus mode. Focus ends by itself ' + + 'at the deadline, or on mode "end"; either way each held channel\'s newest ' + + 'messages are delivered (capped per channel) and the rest stay reachable ' + + 'through your history/backscroll tools. Mode "check" shows what is being ' + + 'held without ending focus. Calling "enter" while already focused ' + + 're-targets: the new channel\'s held backlog is delivered and the deadline ' + + 'resets. Not held: events without a channel (heartbeats, non-chat ' + + 'notifications) and channels you have tuned out (those stay with your ' + + 'subconscious).', + inputSchema: { + type: 'object', + properties: { + mode: { + type: 'string', + enum: ['enter', 'check', 'end'], + description: 'Default: enter.', + }, + channelId: { + type: 'string', + description: + 'enter: the channel to focus on (required). check: limit the ' + + 'report to one held channel and show its held messages.', + }, + serverId: { + type: 'string', + description: 'enter: disambiguates a channelId registered by more than one server.', + }, + durationSeconds: { + type: 'number', + description: + `enter: how long to stay focused (default ${limits.defaultDuration}, ` + + `min ${limits.minDuration}, max ${limits.maxDuration}).`, + }, + backlogCap: { + type: 'number', + description: + `enter: newest messages delivered raw PER CHANNEL when focus ends ` + + `(default ${limits.defaultCap}, max ${limits.maxCap}). Older ones are counted and left to history tools.`, + }, + limit: { + type: 'number', + description: 'check with channelId: how many newest held messages to show (default: backlogCap).', + }, + }, + }, + }; +} + +/** A message bound for the primary's window that has not landed in the store yet. */ +export interface DeferredMessageView { + participant: string; + content: ContentBlock[]; + metadata?: Record; +} + +export interface FocusFrameworkHooks { + /** Deliver a message into the residents' shared window (turn-alive safe). + * `deferred` = parked until the next turn boundary. */ + addMessage: ( + participant: string, + content: ContentBlock[], + metadata?: Record, + ) => { id: string; deferred: boolean }; + /** Queue an inference request for an agent. */ + requestInference: (agentName: string, reason: string, source: string) => void; + /** The primary resident's registry name. */ + primaryName: () => string | null; + /** Read the primary's stored messages (shared slot, unfiltered). */ + getStoredMessages: () => StoredMessage[]; + /** Messages for the primary still parked in the deferred queue (a held + * message that arrived mid-turn lives here until the turn boundary). */ + getDeferredMessages: () => DeferredMessageView[]; + /** Current chronicle head sequence. */ + currentSequence: () => number; + /** Human label for a channel, when known. */ + channelLabel: (serverId: string, channelId: string) => string | undefined; + /** Post host-authored text into a channel, server-qualified. */ + publish: (serverId: string, channelId: string, text: string) => Promise<{ success: boolean; error?: string }>; + /** May the host speak into this channel on the resident's behalf? Open + * channels and DMs yes; a channel the resident stays out of, no. */ + canAutoReplyInto: (serverId: string, channelId: string) => boolean; + /** Is the channel under an active tune-out (#77)? */ + isTunedOut: (serverId: string, channelId: string) => boolean; + /** The resident's IANA zone for wall-clock rendering. */ + timeZone: () => string; + /** Tool names the resident can currently call (for the backscroll hint). */ + availableToolNames: () => string[]; + /** True when per-channel conversation routing is active (focus unsupported). */ + isForkRouted: () => boolean; + /** Focus state changed — the framework refreshes locus + gate hold. */ + onFocusChanged: (state: FocusParams | null) => void; + emitTrace: (event: { type: string; [key: string]: unknown }) => void; +} + +interface HeldStamp { + epochId: string; +} + +/** A held message as the dump sees it: stored, or still deferred. */ +interface HeldMessage { + sequence: number; + participant: string; + content: ContentBlock[]; + metadata?: Record; + timestamp: Date; + deferred: boolean; +} + +const DUMP_KINDS = new Set(['focus-end', 'focus-retarget']); + +export class FocusCoordinator { + private expiryTimer: ReturnType | null = null; + /** (server, channel, author) autoreplies currently being published — the + * in-process half of once-per-author; the durable half (replies that DID + * publish) lives in the lifecycle log via the registry. */ + private inFlightReplies = new Set(); + /** Held-message tallies for the active epoch (rebuilt from the store at resume). */ + private heldCount = 0; + private addressedCount = 0; + + constructor( + private readonly channelRegistry: ChannelRegistry, + private readonly config: FocusConfig, + private readonly hooks: FocusFrameworkHooks, + ) {} + + // ========================================================================== + // State + // ========================================================================== + + getState(): FocusParams | null { + return this.channelRegistry.getFocus(); + } + + isFocused(): boolean { + return this.getState() !== null; + } + + /** + * After a process restart: re-arm a live epoch, end one whose deadline + * passed or that the configuration can no longer serve, and re-wake the + * resident if an end-of-focus dump was restored into the deferred queue + * (its wake was in-memory and died with the old process). + */ + resume(): void { + const state = this.getState(); + if (!state) { + if (this.hasDeferredDump()) { + const primary = this.hooks.primaryName(); + if (primary) this.hooks.requestInference(primary, 'focus backlog restored after restart', 'focus'); + } + return; + } + if (!this.config.enabled) { + this.end('config', 'focus is disabled in the configuration'); + return; + } + if (this.hooks.isForkRouted()) { + this.end('config', 'per-channel conversation routing is active'); + return; + } + if (state.expiresAtMs <= Date.now()) { + // The deadline passed while the host was down: the resident gets the + // backlog on the schedule they committed to, not whenever we restarted. + this.end('duration', 'deadline elapsed while the host was down'); + return; + } + this.rebuildTallies(state); + this.armExpiry(state); + this.hooks.onFocusChanged(state); + } + + private hasDeferredDump(): boolean { + return this.hooks.getDeferredMessages().some((m) => { + const kind = m.metadata?.kind; + return typeof kind === 'string' && DUMP_KINDS.has(kind); + }); + } + + private rebuildTallies(state: FocusParams): void { + this.inFlightReplies.clear(); + this.heldCount = 0; + this.addressedCount = 0; + for (const m of this.heldMessages(state.epochId)) { + this.heldCount++; + const tags = m.metadata?.tags; + if (Array.isArray(tags) && tags.includes('chat:addressed')) this.addressedCount++; + } + } + + /** + * Is this stored message held for a delivery that has not happened yet — + * stamped with the ACTIVE epoch and not released by a re-target? Until the + * unfocus dump renders it, nobody has read it, so an edit or retraction + * must replace it in place (the dump then shows the final text). Once its + * epoch ended the dump has consumed it, and a later edit is new traffic. + */ + isAwaitingDelivery(message: { sequence: number; metadata?: unknown }): boolean { + const state = this.getState(); + if (!state) return false; + const md = message.metadata as Record | undefined; + if ((md?.focusHeld as HeldStamp | undefined)?.epochId !== state.epochId) return false; + return !this.isReleased(state, { + sequence: message.sequence, participant: '', content: [], metadata: md, timestamp: new Date(0), deferred: false, + }); + } + + private armExpiry(state: FocusParams): void { + this.clearExpiry(); + // Node timers overflow past 2^31-1 ms; durations are capped far below + // that, but a corrupted record must not wrap into an immediate fire. + const remaining = Math.min(Math.max(0, state.expiresAtMs - Date.now()), 0x7fff_ffff); + this.expiryTimer = setTimeout(() => { + this.expiryTimer = null; + this.end('duration', 'deadline reached'); + }, remaining); + } + + private clearExpiry(): void { + if (this.expiryTimer) clearTimeout(this.expiryTimer); + this.expiryTimer = null; + } + + // ========================================================================== + // Enter / re-target / end + // ========================================================================== + + enter( + channelId: string, + opts: { durationSeconds?: number; backlogCap?: number; serverId?: string }, + source: string, + ): { ok: true; params: FocusParams; retargeted: boolean } | { ok: false; error: string } { + if (!this.config.enabled) { + return { ok: false, error: 'focus is disabled in the configuration' }; + } + if (this.hooks.isForkRouted()) { + return { ok: false, error: 'focus is not available under per-channel conversation routing' }; + } + const resolved = this.channelRegistry.resolveChannel(channelId, opts.serverId); + if (!resolved.entry) { + return { ok: false, error: resolved.error ?? `unknown channel: ${channelId}` }; + } + const target = resolved.entry; + if (this.hooks.isTunedOut(target.serverId, target.channelId)) { + return { + ok: false, + error: + `channel ${channelId} is tuned out — its traffic goes to your subconscious, so focusing on it ` + + `would deliver nothing. Cancel the tune-out first (tune_out with mode: 'cancel').`, + }; + } + const now = Date.now(); + const limits = resolveFocusLimits(this.config); + const requested = opts.durationSeconds ?? limits.defaultDuration; + const durationSeconds = Math.min(limits.maxDuration, Math.max(limits.minDuration, Math.floor(requested))); + + const current = this.getState(); + if (current) { + // Re-target: the new channel's held backlog is what the resident is + // walking into — deliver it now; the old focus channel becomes held + // from here on. Same epoch: stamps stay valid, tallies carry over. + // Only STORED messages release early: one still deferred behind the + // current turn has no sequence to watermark against, so it arrives at + // unfocus with the rest. + const requestedCap = opts.backlogCap ?? current.backlogCap; + const backlogCap = Math.min(limits.maxCap, Math.max(0, Math.floor(requestedCap))); + const released = { ...(current.released ?? {}) }; + const pending = this.heldMessages(current.epochId) + .filter((m) => !m.deferred && serverIdOf(m) === target.serverId && channelIdOf(m) === target.channelId + && !this.isReleased(current, m)); + let dumpText = ''; + if (pending.length > 0) { + released[this.releaseKey(target.serverId, target.channelId)] = pending[pending.length - 1]!.sequence; + dumpText = '\n\n' + this.renderChannelDump(target.serverId, target.channelId, pending, backlogCap); + if (pending.length > backlogCap) { + const hint = this.backscrollHint(backlogCap); + if (hint) dumpText += '\n\n' + hint; + } + } + const params: FocusParams = { + ...current, + serverId: target.serverId, + channelId: target.channelId, + backlogCap, + expiresAtMs: now + durationSeconds * 1000, + released, + }; + this.channelRegistry.setFocus(params, source); + this.armExpiry(params); + this.hooks.onFocusChanged(params); + const oldLabel = this.labelFor(current.serverId, current.channelId); + this.hooks.addMessage( + 'user', + [{ type: 'text', text: + `[Focus re-targeted: ${oldLabel} → ${this.labelFor(target.serverId, target.channelId)}, ` + + `until ~${this.untilText(params.expiresAtMs)}. ${oldLabel} is now held.` + + `${pending.length > 0 ? ` ${pending.length} held message${pending.length === 1 ? '' : 's'} from the new channel:` : ''}]` + + dumpText }], + { system: true, kind: 'focus-retarget', channelId: target.channelId, epochId: params.epochId }, + ); + this.hooks.emitTrace({ type: 'focus:retargeted', from: current.channelId, to: target.channelId, epochId: params.epochId }); + return { ok: true, params, retargeted: true }; + } + + const requestedCap = opts.backlogCap ?? limits.defaultCap; + const params: FocusParams = { + epochId: randomUUID(), + serverId: target.serverId, + channelId: target.channelId, + startedAtMs: now, + startedAtSequence: this.hooks.currentSequence(), + expiresAtMs: now + durationSeconds * 1000, + backlogCap: Math.min(limits.maxCap, Math.max(0, Math.floor(requestedCap))), + }; + this.channelRegistry.setFocus(params, source); + this.inFlightReplies.clear(); + this.heldCount = 0; + this.addressedCount = 0; + this.armExpiry(params); + this.hooks.onFocusChanged(params); + this.hooks.emitTrace({ type: 'focus:entered', serverId: target.serverId, channelId: target.channelId, epochId: params.epochId, expiresAtMs: params.expiresAtMs }); + return { ok: true, params, retargeted: false }; + } + + /** + * End focus: durable flip, per-channel backlog dumps (newest `backlogCap` + * raw, remainder counted + pointed at history tools), lifecycle notice, + * one wake for the resident. Held messages still deferred behind the + * current turn are included — they are held traffic too. + */ + end(source: string, reason: string): { ok: true; held: number } | { ok: false; error: string } { + const state = this.getState(); + if (!state) return { ok: false, error: 'not in focus mode' }; + // Read before the flip: the registry's reply record is per active epoch. + const repliedCount = this.channelRegistry.focusAutoReplyCount(); + this.channelRegistry.setFocus(null, source); + this.clearExpiry(); + this.hooks.onFocusChanged(null); + + const byChannel = new Map(); + let held = 0; + for (const m of this.heldMessages(state.epochId)) { + if (this.isReleased(state, m)) continue; + const serverId = serverIdOf(m) ?? state.serverId; + const channelId = channelIdOf(m) ?? '(unknown channel)'; + const key = this.releaseKey(serverId, channelId); + const entry = byChannel.get(key) ?? { serverId, channelId, messages: [] }; + entry.messages.push(m); + byChannel.set(key, entry); + held++; + } + + const minutes = Math.max(1, Math.round((Date.now() - state.startedAtMs) / 60_000)); + const focusLabel = this.labelFor(state.serverId, state.channelId); + const addressedNote = this.addressedCount === 0 + ? '' + : `; ${this.addressedCount} addressed you` + + (repliedCount > 0 + ? ` (${repliedCount} automatic repl${repliedCount === 1 ? 'y was' : 'ies were'} sent)` + : ' (no automatic reply was sent)'); + const header = + `[Focus ended — ${reason}. You were focused on ${focusLabel} for ~${minutes}m. ` + + (held === 0 + ? 'Nothing was held.]' + : `${held} message${held === 1 ? '' : 's'} held across ${byChannel.size} channel${byChannel.size === 1 ? '' : 's'}${addressedNote}.]`); + + const dumps: string[] = []; + // Most recently active channel last, so the freshest context is nearest + // the resident's next turn. + const ordered = [...byChannel.values()].sort( + (a, b) => a.messages[a.messages.length - 1]!.sequence - b.messages[b.messages.length - 1]!.sequence, + ); + for (const { serverId, channelId, messages } of ordered) { + dumps.push(this.renderChannelDump(serverId, channelId, messages, state.backlogCap)); + } + const truncatedAny = ordered.some((c) => c.messages.length > state.backlogCap); + const hint = truncatedAny ? this.backscrollHint(state.backlogCap) : null; + + this.hooks.addMessage( + 'user', + [{ type: 'text', text: [header, ...dumps, ...(hint ? [hint] : [])].join('\n\n') }], + { system: true, kind: 'focus-end', epochId: state.epochId, channelId: state.channelId }, + ); + const primary = this.hooks.primaryName(); + if (primary) { + this.hooks.requestInference(primary, `focus ended (${reason})`, 'focus'); + } + this.hooks.emitTrace({ + type: 'focus:ended', epochId: state.epochId, channelId: state.channelId, reason, held, + addressed: this.addressedCount, replied: repliedCount, + }); + this.inFlightReplies.clear(); + this.heldCount = 0; + this.addressedCount = 0; + return { ok: true, held }; + } + + // ========================================================================== + // Ingestion: hold, stamp, autoreply + // ========================================================================== + + /** + * Consulted by the framework for every incoming channel/DM message BEFORE + * storage. Null = not focused, or this IS the focus channel (same server, + * same channel — proceed normally). Otherwise the caller stamps + * `metadata.focusHeld` with the returned object, stores the message, and + * wakes nobody. + */ + onIncoming( + serverId: string, + channelId: string, + messageId: string, + tags: string[] | undefined, + author: { id?: string; name?: string } | undefined, + ): HeldStamp | null { + const state = this.getState(); + if (!state) return null; + if (serverId === state.serverId && channelId === state.channelId) return null; + + this.heldCount++; + const addressed = Array.isArray(tags) && tags.includes('chat:addressed'); + const stamp: HeldStamp = { epochId: state.epochId }; + if (!addressed) return stamp; + + this.addressedCount++; + if (this.config.autoReply === false) return stamp; + // Where the host may speak: channels the resident is open in, and DMs. + // A mention in a channel the resident stays out of is held and counted, + // reply-less — before focus it produced an invitation the resident + // decided on; the host must not answer in a room they chose not to join. + if (!this.hooks.canAutoReplyInto(serverId, channelId)) return stamp; + // Once per (server, channel, author) per epoch: the durable record says a + // reply PUBLISHED; the in-flight set covers the publish still settling. + if (this.channelRegistry.hasFocusAutoReply(serverId, channelId, author?.id)) return stamp; + const key = `${serverId}\u0000${channelId}\u0000${author?.id ?? ''}`; + if (this.inFlightReplies.has(key)) return stamp; + this.inFlightReplies.add(key); + void this.sendAutoReply(state, serverId, channelId, messageId, author?.id, key); + return stamp; + } + + private async sendAutoReply( + state: FocusParams, serverId: string, channelId: string, messageId: string, + authorId: string | undefined, key: string, + ): Promise { + const template = this.config.autoReplyTemplate ?? DEFAULT_AUTOREPLY; + const remainingMin = Math.max(1, Math.round((state.expiresAtMs - Date.now()) / 60_000)); + const text = template + .replaceAll('{name}', this.hooks.primaryName() ?? 'The agent') + .replaceAll('{until}', this.untilText(state.expiresAtMs)) + .replaceAll('{remaining}', `${remainingMin}m`) + .replaceAll('{channel}', this.labelFor(state.serverId, state.channelId)); + let failure: string | undefined; + try { + const result = await this.hooks.publish(serverId, channelId, text); + if (!result.success) failure = result.error ?? 'publish failed'; + } catch (err) { + failure = (err as Error).message; + } + // A publish that settles after its epoch ended (or was replaced by a new + // one) belongs to that epoch: it must neither release an allowance nor + // count a reply in the epoch that is active now. + const sameEpoch = this.getState()?.epochId === state.epochId; + if (sameEpoch) this.inFlightReplies.delete(key); + if (failure !== undefined) { + // Nothing recorded: the next addressed message from this author retries + // (also after a restart — only replies that published are durable). + this.hooks.emitTrace({ type: 'focus:autoreply-failed', serverId, channelId, messageId, error: failure }); + return; + } + if (sameEpoch) this.channelRegistry.recordFocusAutoReply(state.epochId, serverId, channelId, authorId); + this.hooks.emitTrace({ type: 'focus:autoreply', serverId, channelId, messageId }); + } + + // ========================================================================== + // Check (status + peek) + // ========================================================================== + + status(): { + focused: boolean; + channelId?: string; + serverId?: string; + channel?: string; + expiresAt?: string; + remainingSeconds?: number; + backlogCap?: number; + held?: Array<{ channelId: string; serverId: string; channel: string; messages: number; addressed: number; latest: string }>; + } { + const state = this.getState(); + if (!state) return { focused: false }; + const byChannel = new Map(); + for (const m of this.heldMessages(state.epochId)) { + if (this.isReleased(state, m)) continue; + const serverId = serverIdOf(m) ?? state.serverId; + const channelId = channelIdOf(m) ?? '(unknown channel)'; + const key = this.releaseKey(serverId, channelId); + const entry = byChannel.get(key) ?? { serverId, channelId, messages: 0, addressed: 0, latest: '' }; + entry.messages++; + const tags = m.metadata?.tags; + if (Array.isArray(tags) && tags.includes('chat:addressed')) entry.addressed++; + entry.latest = this.clock(m.timestamp.getTime()); + byChannel.set(key, entry); + } + return { + focused: true, + channelId: state.channelId, + serverId: state.serverId, + channel: this.labelFor(state.serverId, state.channelId), + expiresAt: this.untilText(state.expiresAtMs), + remainingSeconds: Math.max(0, Math.round((state.expiresAtMs - Date.now()) / 1000)), + backlogCap: state.backlogCap, + held: [...byChannel.values()].map((e) => ({ + channelId: e.channelId, + serverId: e.serverId, + channel: this.labelFor(e.serverId, e.channelId), + messages: e.messages, + addressed: e.addressed, + latest: e.latest, + })), + }; + } + + /** Newest `limit` held messages of one channel, rendered — without releasing them. */ + peek(channelId: string, limit?: number, serverId?: string): { ok: true; text: string; total: number } | { ok: false; error: string } { + const state = this.getState(); + if (!state) return { ok: false, error: 'not in focus mode' }; + const messages = this.heldMessages(state.epochId) + .filter((m) => channelIdOf(m) === channelId && (!serverId || serverIdOf(m) === serverId) && !this.isReleased(state, m)); + const sid = serverId ?? serverIdOf(messages[0]) ?? state.serverId; + const cap = Math.max(0, Math.floor(limit ?? state.backlogCap)); + return { ok: true, total: messages.length, text: this.renderChannelDump(sid, channelId, messages, cap) }; + } + + // ========================================================================== + // Tool dispatch (resident) + // ========================================================================== + + handleTool(input: Record): { success: boolean; data?: unknown; error?: string; isError?: boolean } { + const mode = typeof input.mode === 'string' ? input.mode : 'enter'; + const serverId = typeof input.serverId === 'string' && input.serverId ? input.serverId : undefined; + switch (mode) { + case 'enter': { + const channelId = String(input.channelId ?? ''); + if (!channelId) return { success: false, error: 'channelId is required to enter focus', isError: false }; + const r = this.enter(channelId, { + durationSeconds: numberOrUndefined(input.durationSeconds), + backlogCap: numberOrUndefined(input.backlogCap), + serverId, + }, 'agent-tool'); + if (!r.ok) return { success: false, error: r.error, isError: false }; + return { + success: true, + data: { + focused: r.params.channelId, + serverId: r.params.serverId, + channel: this.labelFor(r.params.serverId, r.params.channelId), + until: this.untilText(r.params.expiresAtMs), + backlogCap: r.params.backlogCap, + ...(r.retargeted ? { retargeted: true } : {}), + }, + }; + } + case 'check': { + const status = this.status(); + if (!status.focused) return { success: true, data: status }; + const channelId = typeof input.channelId === 'string' && input.channelId ? input.channelId : undefined; + if (!channelId) return { success: true, data: status }; + const p = this.peek(channelId, numberOrUndefined(input.limit), serverId); + if (!p.ok) return { success: false, error: p.error, isError: false }; + return { + success: true, + data: { + ...status, + held: status.held?.filter((h) => h.channelId === channelId && (!serverId || h.serverId === serverId)), + peek: p.text, + }, + }; + } + case 'end': { + const r = this.end('agent-tool', 'ended by you'); + return r.ok + ? { success: true, data: { ended: true, held: r.held } } + : { success: false, error: r.error, isError: false }; + } + default: + return { success: false, error: `unknown focus mode: ${mode}`, isError: false }; + } + } + + // ========================================================================== + // Helpers + // ========================================================================== + + /** Every message stamped with this epoch: stored ones in store order, then + * the ones still deferred behind the current turn (in queue order). */ + private heldMessages(epochId: string): HeldMessage[] { + const stampedWith = (md: Record | undefined): boolean => + (md?.focusHeld as HeldStamp | undefined)?.epochId === epochId; + const stored: HeldMessage[] = this.hooks.getStoredMessages() + .filter((m) => stampedWith(m.metadata as Record | undefined)) + .map((m) => ({ + sequence: m.sequence, participant: m.participant, content: m.content, + metadata: m.metadata as Record | undefined, timestamp: m.timestamp, deferred: false, + })); + const base = stored.length > 0 ? stored[stored.length - 1]!.sequence : this.hooks.currentSequence(); + const deferred: HeldMessage[] = this.hooks.getDeferredMessages() + .filter((m) => stampedWith(m.metadata)) + .map((m, i) => ({ + sequence: base + 1 + i, participant: m.participant, content: m.content, + metadata: m.metadata, timestamp: new Date(), deferred: true, + })); + return [...stored, ...deferred]; + } + + private releaseKey(serverId: string, channelId: string): string { + return `${serverId}\u0000${channelId}`; + } + + private isReleased(state: FocusParams, m: HeldMessage): boolean { + if (m.deferred) return false; + const ch = channelIdOf(m); + if (!ch) return false; + const released = state.released ?? {}; + // Keys are server-qualified; a record from before qualification used the + // bare channel id — honour both so an in-flight epoch keeps its watermarks. + const through = released[this.releaseKey(serverIdOf(m) ?? state.serverId, ch)] ?? released[ch]; + return through !== undefined && m.sequence <= through; + } + + /** "18:40 America/Los_Angeles" — for deadlines the reader must act on. */ + private untilText(ms: number): string { + const tz = this.hooks.timeZone(); + return `${formatZonedTime(ms, tz)} ${tz}`; + } + + /** "18:40" — per-line stamps inside a dump (zone given once on the block). */ + private clock(ms: number): string { + return formatZonedTime(ms, this.hooks.timeZone()); + } + + private labelFor(serverId: string, channelId: string): string { + const label = this.hooks.channelLabel(serverId, channelId); + return label ? `#${label} (${channelId})` : channelId; + } + + private renderChannelDump(serverId: string, channelId: string, messages: HeldMessage[], cap: number): string { + const shown = cap > 0 ? messages.slice(-cap) : []; + const truncated = messages.length - shown.length; + const lines = shown.map((m) => { + const md = m.metadata; + const author = (md?.author as { name?: string } | undefined)?.name + ?? (typeof md?.authorName === 'string' && md.authorName ? md.authorName : undefined) + ?? m.participant; + const text = m.content + .filter((b): b is Extract => b.type === 'text') + .map((b) => b.text) + .join('\n'); + // Non-text blocks (images, audio, documents) are not re-delivered raw; + // name them so a media-only message is not an empty line and a mixed + // one does not silently lose its attachment. + const extras = new Map(); + for (const b of m.content) { + if (b.type !== 'text') extras.set(b.type, (extras.get(b.type) ?? 0) + 1); + } + const extraNote = [...extras.entries()].map(([t, n]) => `[+${n} ${t}${n === 1 ? '' : 's'}]`).join(' '); + const stamp = m.deferred ? 'arrived mid-turn' : this.clock(m.timestamp.getTime()); + return `[${stamp}] ${author}: ${[text, extraNote].filter(Boolean).join(' ')}`; + }); + const attrs = [ + `channel="${this.labelFor(serverId, channelId)}"`, + `messages=${messages.length}`, + `tz="${this.hooks.timeZone()}"`, + ...(truncated > 0 ? [`truncated=${truncated} (oldest, not shown)`] : []), + ]; + return `\n${lines.join('\n')}\n`; + } + + /** Name only tools the resident actually has (tune-out's dump pointed at a + * fetch_history most residents lack). */ + private backscrollHint(cap: number): string | null { + const tools = new Set(this.hooks.availableToolNames()); + const options: string[] = []; + if (tools.has('history--extract')) options.push(`history--extract {channelId, limit}`); + if (tools.has('channel_open')) options.push(`channel_open {channelId, backscroll}`); + for (const t of tools) if (t.endsWith('fetch_history')) options.push(`${t} {channelId, limit}`); + if (options.length === 0) return null; + return `[Held messages above the per-channel cap of ${cap} are in your history: ${options.join(' · ')}]`; + } + + /** Stop timers (framework shutdown). */ + stop(): void { + this.clearExpiry(); + } +} + +function channelIdOf(m: HeldMessage): string | undefined { + const v = m.metadata?.channelId; + return typeof v === 'string' && v ? v : undefined; +} + +function serverIdOf(m: HeldMessage | undefined): string | undefined { + const v = m?.metadata?.serverId; + return typeof v === 'string' && v ? v : undefined; +} + +function numberOrUndefined(v: unknown): number | undefined { + return typeof v === 'number' && Number.isFinite(v) ? v : undefined; +} diff --git a/src/framework.ts b/src/framework.ts index 39a907a98..e116f6a74 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -9,6 +9,7 @@ import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverB import type { CacheWireReceipt } from './kv-unified-wire.js'; import { SUBCONSCIOUS_TOOLS, SUBCONSCIOUS_TOOL_NAMES, type SubconsciousConfig } from './tune-out/tools.js'; import { TuneOutCoordinator, TUNE_OUT_DEFAULTS } from './tune-out/coordinator.js'; +import { FocusCoordinator, FOCUS_TOOL_NAME, buildFocusToolDefinition, type FocusConfig } from './focus/coordinator.js'; import type { MessageId, MessageMetadata, @@ -472,6 +473,18 @@ const bySeq = (a: { seq: number }, b: { seq: number }): number => a.seq - b.seq; /** Stamp a deferred write's durable id into the metadata it is stored with * (boot recovery dedups replays by it). Idempotent for an already-stamped * message. */ +/** + * Stored-but-not-for-the-resident: tune-out-diverted (#77) and focus-held + * messages. ONE predicate for every surface that puts stored messages in + * front of the resident — the compile viewFilter AND the mid-turn live + * injection of deferred messages — so a held message cannot reach a live + * turn through a path the filter does not cover. + */ +function isHeldFromResidentView(metadata: MessageMetadata | Record | undefined): boolean { + const md = metadata as { tuneOut?: unknown; focusHeld?: unknown } | undefined; + return !!(md?.tuneOut || md?.focusHeld); +} + function withDeferredWriteId(metadata: MessageMetadata | undefined, id: string): MessageMetadata { return { ...(metadata ?? {}), deferredWriteId: id } as MessageMetadata; } @@ -1159,6 +1172,9 @@ export class AgentFramework { private subconsciousConfig: SubconsciousConfig | null = null; /** Tune-out coordinator (issue #77); non-null iff subconscious + channels. */ private tuneOutCoordinator: TuneOutCoordinator | null = null; + /** Focus coordinator; non-null iff `config.focus.enabled` + channels. */ + private focusCoordinator: FocusCoordinator | null = null; + private focusConfig: FocusConfig | null = null; /** Agents an abandon could not cancel (token held, no stream) — surfaced * in HostModeStatus until the next quiesce/resume. */ @@ -1804,6 +1820,131 @@ export class AgentFramework { framework.tuneOutCoordinator.resumeActiveEpochs(); } + // Focus coordinator: per-resident single-channel attention. Needs only + // the channel subsystem (no subconscious). Built whenever channels exist + // — not only when enabled — so a persisted epoch from before the switch + // was turned off (or before conversation routing was configured) still + // ends and delivers its backlog at boot. The tool is offered only when + // enabled and no router is present (`enter` could never succeed under + // per-channel routing). + if (framework.channelRegistry) { + const registry = framework.channelRegistry; + const focusConfig: FocusConfig = config.focus ?? { enabled: false }; + framework.focusConfig = focusConfig; + framework.focusCoordinator = new FocusCoordinator(registry, focusConfig, { + addMessage: (participant, content, metadata) => { + const placement: { agent?: string; messageId?: MessageId; deferredId?: string } = {}; + const id = framework.addMessage(participant, content, metadata as MessageMetadata, { placement }); + if (placement.deferredId) { + // A deferred focus dump is the ONLY delivery of the messages it + // carries (their originals stay behind the view filter). The + // deferred queue is persisted only during/after quiesce by + // default — turn persistence on now so a crash before the flush + // replays the dump instead of losing the backlog. + framework.deferredWritesPersisted = true; + framework.persistDeferredWrites(); + return { id: placement.deferredId, deferred: true }; + } + return { id, deferred: false }; + }, + requestInference: (agentName, reason, source) => { + framework.pendingRequests.push({ agentName, reason, source, timestamp: Date.now() }); + }, + primaryName: () => framework.primaryAgentName, + getStoredMessages: () => { + const primary = framework.primaryAgentName + ? framework.agents.get(framework.primaryAgentName) : undefined; + return primary ? primary.getContextManager().getAllMessages() : []; + }, + getDeferredMessages: () => { + const primary = framework.primaryAgentName; + return framework.deferredMessages + .filter((m) => (m.forAgent ?? primary) === primary) + .map((m) => ({ + participant: m.participant, content: m.content, + ...(m.metadata ? { metadata: m.metadata as Record } : {}), + })); + }, + currentSequence: () => framework.store.currentSequence(), + channelLabel: (serverId, channelId) => registry.channelLabel(serverId, channelId), + publish: async (serverId, channelId, text) => { + const r = await registry.publishForAgent( + channelId, text, `${framework.primaryAgentName ?? 'agent'}/focus-autoreply`, serverId); + // A server answering `delivered: false` did not post the reply. + const delivered = (r.data as { delivered?: unknown } | undefined)?.delivered; + if (r.success && delivered === false) return { success: false, error: 'server reported delivered: false' }; + return { success: r.success, ...(r.error ? { error: r.error } : {}) }; + }, + canAutoReplyInto: (serverId, channelId) => registry.canAutoReplyInto(serverId, channelId), + isTunedOut: (serverId, channelId) => registry.getTuneOutState(serverId, channelId) !== null, + timeZone: () => framework.timeZone, + availableToolNames: () => { + const primary = framework.primaryAgentName + ? framework.agents.get(framework.primaryAgentName) : undefined; + if (!primary) return []; + return framework.getToolsForAgent(primary.name) + .filter((t) => primary.canUseTool(t.name)) + .map((t) => t.name); + }, + isForkRouted: () => framework.conversationRouter !== null, + onFocusChanged: (state) => { + // Routing first (no gate dependency): while focused, plain speech + // lands in the focus channel and held inbound stops moving the + // fallback locus. + registry.setFocusLocus(state ? { serverId: state.serverId, channelId: state.channelId } : null); + if (state) { + // Wakes already queued for channels that are held from now on + // (a channel message that arrived a moment before `enter`, a + // coalesced push batch) must not run a turn inside the focus — + // the gate purge below covers its own queues, this covers the + // framework's. Their messages predate the epoch and stay visible. + const primary = framework.primaryAgentName; + const staleWake = (r: InferenceRequest): boolean => + r.agentName === primary && !!r.channelId + && (r.reason === 'mcpl:channel-incoming' || r.reason === 'mcpl:push-event') + && !(r.source === state.serverId && r.channelId === state.channelId) + && !registry.getTuneOutState(r.source, r.channelId); + const before = framework.pendingRequests.length; + framework.pendingRequests = framework.pendingRequests.filter((r) => !staleWake(r)); + for (const cooldown of framework.providerAccelerationCooldowns.values()) { + cooldown.heldRequests = cooldown.heldRequests.filter((r) => !staleWake(r)); + } + const dropped = before - framework.pendingRequests.length; + if (dropped > 0) framework.emitTrace({ type: 'focus:purged-pending-wakes', count: dropped } as never); + } + const gate = framework.eventGate; + if (!gate) return; + if (!state) { + gate.setHoldPredicate(null); + return; + } + const purged = gate.setHoldPredicate((info) => { + if (info.eventType !== 'mcpl:channel-incoming' && info.eventType !== 'mcpl:push-event') return false; + const serverId = info.serverId + || (typeof info.metadata?.serverId === 'string' ? info.metadata.serverId : ''); + const channelId = info.eventType === 'mcpl:push-event' + ? framework.derivePushEventChannel(info.metadata)?.channelId ?? info.channelId + : info.channelId; + // Events with no channel (heartbeats, non-chat pushes) are not held. + if (!channelId) return false; + // Identity is (serverId, channelId): a same-named channel on + // another server is not the focus channel. + if (serverId === state.serverId && channelId === state.channelId) return false; + // Tuned-out channels stay with the subconscious: tune-out's own + // classification (suppressed-mention ack, subconscious wake) + // composes with the gate verdict and must keep seeing it. + if (registry.getTuneOutState(serverId, channelId)) return false; + return true; + }); + if (purged > 0) { + framework.emitTrace({ type: 'focus:purged-queued-wakes', count: purged } as never); + } + }, + emitTrace: (e) => framework.emitTrace(e as never), + }); + framework.focusCoordinator.resume(); + } + // Diagnostics: `kill -USR2 ` dumps live wake/inference state to stderr // (journal) without a restart — for catching the wake-wedge on the running // process. Shows the gate's `inferring` set + buffered-event count (the @@ -1879,6 +2020,7 @@ export class AgentFramework { this.providerAdmissionClosed = true; this.queue.close(); this.tuneOutCoordinator?.stop(); + this.focusCoordinator?.stop(); // Kill running code_execution scripts before cancelling streams: a // zombie script must not keep firing side-effectful tool calls into a @@ -2379,6 +2521,9 @@ export class AgentFramework { if (this.tuneOutCoordinator) { channelTools.push(AgentFramework.TUNE_OUT_TOOL); } + if (this.focusCoordinator && this.focusConfig?.enabled && !this.conversationRouter) { + channelTools.push(buildFocusToolDefinition(this.focusConfig)); + } const gateTools = this.eventGate ? [ this.eventGate.getToolDefinition(), @@ -6258,7 +6403,9 @@ export class AgentFramework { // the timeline (KV-prefix-stable delivery). The stored originals // remain in the shared slot for the subconscious's merged view, // fetch_history, and audit. - viewFilter: (message) => !(message.metadata as { tuneOut?: unknown } | undefined)?.tuneOut, + // Focus-held messages (`metadata.focusHeld`) are excluded the same way; + // the dump at unfocus is the delivery. + viewFilter: (message) => !isHeldFromResidentView(message.metadata), }); const agent = new Agent(config, contextManager, this.membrane); @@ -6503,7 +6650,9 @@ export class AgentFramework { const hasToolBlocks = msg.content.some( (b) => b.type === 'tool_use' || b.type === 'tool_result' ); - if (!hasToolBlocks && msg.participant !== agent.name) { + // A tune-out-diverted or focus-held message is stored for + // the backlog only — it must not reach the live turn either. + if (!hasToolBlocks && msg.participant !== agent.name && !isHeldFromResidentView(msg.metadata)) { midTurnInjections.push({ participant: msg.participant, content: msg.content, @@ -6606,11 +6755,19 @@ export class AgentFramework { // Reactions and system markers are already excluded by // isConversationalInjection, and the engaged-this-turn scope // keeps unrelated channels from moving the pin. + // + // Focus: a held message (stored, excluded from every view) never + // qualifies, and while focused only the focus channel can take + // the pin — the resident chose where their words go for this + // window; an injection from elsewhere can't overrule that. const engaged = this.turnEngagedChannels.get(agent.name); + const focusChannel = this.channelRegistry?.getFocusLocus()?.channelId; const lastQualifying = [...midTurnInjections].reverse().find((inj) => { const m = inj.metadata as Record | undefined; if (!isConversationalInjection(inj.metadata)) return false; if (typeof m?.channelId !== 'string') return false; + if (m.focusHeld) return false; + if (focusChannel !== undefined && m.channelId !== focusChannel) return false; const tags = m.tags as string[] | undefined; if (isAddressedMessage(tags, m)) return true; return engaged?.has(m.channelId as string) === true; @@ -6934,10 +7091,43 @@ export class AgentFramework { return this.routeConversationIncoming(event, metadata); } + // Tune-out divert (issue #77): stamped, stored, no resident wake. The + // coordinator has already handled wake bookkeeping (coalesced + // subconscious invocation, durable count, suppression ack) before we + // stamp; the subconscious reads the message through its merged view. + const divert = this.tuneOutCoordinator?.onIncoming( + event.serverId, + event.channelId, + event.messageId, + event.tags, + event.author?.id, + // The resident's gate verdict rides the event; it preconditions + // subconscious wakes (gate composes, it is not replaced). + event.triggerInference !== false, + ) ?? null; + if (divert) { + metadata.tuneOut = { epochId: divert.epochId }; + } + // Focus hold: every channel but the focus channel is stored-not-shown + // and wakes nobody; addressed messages get the automatic reply. + // Host-wide by design: residents share one message slot, so an event + // targeted at a side agent still lands in the window the primary reads — + // exempting it from the hold would show the primary traffic it asked not + // to see. Held regardless of `targetAgents`. + const held = !divert + ? this.focusCoordinator?.onIncoming( + event.serverId, event.channelId, event.messageId, event.tags, event.author) ?? null + : null; + if (held) { + metadata.focusHeld = held; + } + // Addressed-while-closed invitation — parity with the push-event path // (channels.publish surfaces like portal-mcpl deliver mentions here). + // Not for held messages: the invitation is an at-arrival affordance + // ("reply this turn…") that would be stale noise in the unfocus dump. const incomingContent = [...event.content]; - { + if (!held) { const invitation = this.buildClosedChannelInvitation({ serverId: event.serverId, channelId: event.channelId, @@ -6960,29 +7150,11 @@ export class AgentFramework { } } - // Tune-out divert (issue #77): stamped, stored, no resident wake. The - // coordinator has already handled wake bookkeeping (coalesced - // subconscious invocation, durable count, suppression ack) before we - // stamp; the subconscious reads the message through its merged view. - const divert = this.tuneOutCoordinator?.onIncoming( - event.serverId, - event.channelId, - event.messageId, - event.tags, - event.author?.id, - // The resident's gate verdict rides the event; it preconditions - // subconscious wakes (gate composes, it is not replaced). - event.triggerInference !== false, - ) ?? null; - if (divert) { - metadata.tuneOut = { epochId: divert.epochId }; - } - const placement: CoalescingPlacement = { agent: '' }; const id = this.addMessage('user', incomingContent, metadata, { placement, bypassDeferralFor: event.assemblingFor }); this.emitTrace({ type: 'message:added', messageId: id, source: 'mcpl:channel-incoming' }); - if (event.triggerInference && !divert) { + if (event.triggerInference && !divert && !held) { const addressed = isAddressedMessage(event.tags, event.metadata); // Honor explicit targeting, mirroring the push-event path: an event // that names its agents wakes exactly those; an untargeted event @@ -7784,6 +7956,13 @@ export class AgentFramework { // not the agent that happened to be asked). A tune-out-diverted message // never enters a resident's compiled view; only the subconscious reads // it through its merged view — no subconscious, nobody reads it. + // A focus-held message awaiting its unfocus dump has been read by nobody + // (the view filter hides it; the dump is its delivery): unread, whatever + // the watermark says — an edit or retraction must replace it in place so + // the dump renders the final text. After its epoch ended the dump has + // consumed it and the ordinary watermark rule applies (a later edit is + // new traffic, appended where the resident will see it). + if (this.focusCoordinator?.isAwaitingDelivery(message)) return true; let readers: Agent[]; if ((message.metadata as { tuneOut?: unknown } | undefined)?.tuneOut) { const subconscious = this.subconsciousAgentName ? this.agents.get(this.subconsciousAgentName) : undefined; @@ -7869,8 +8048,28 @@ export class AgentFramework { metadata.channelId = triggerChannel.channelId; } + // Focus hold — DMs and addressed-while-closed messages arrive here, so + // this path must hold too or they leak around the focus. + let held: { epochId: string } | null = null; + if (triggerChannel && this.focusCoordinator) { + const origin = (event.origin ?? {}) as Record; + held = this.focusCoordinator.onIncoming( + event.serverId, + triggerChannel.channelId, + typeof origin.messageId === 'string' ? origin.messageId : event.eventId, + event.tags, + { + ...(typeof origin.authorId === 'string' ? { id: origin.authorId } : {}), + ...(typeof origin.authorName === 'string' ? { name: origin.authorName } : {}), + }, + ); + if (held) metadata.focusHeld = held; + } + + // Invitation only for messages the resident will actually see now; a + // held message's affordances would be stale in the unfocus dump. const content = [...event.content]; - if (triggerChannel) { + if (triggerChannel && !held) { const origin = (event.origin ?? {}) as Record; const invitation = this.buildClosedChannelInvitation({ serverId: event.serverId, @@ -7909,7 +8108,7 @@ export class AgentFramework { console.error(`[heartbeat] ${event.serverId}: accepted silent scheduled wake ${event.eventId}`); } - if (event.triggerInference) { + if (event.triggerInference && !held) { // Default broadcast excludes conversation forks (channel-driven). const targetAgents = event.targetAgents ?? [...this.agents.keys()].filter( @@ -11795,6 +11994,46 @@ export class AgentFramework { this.dispatchTuneOutToolCall(agentName, enrichedCall); return; } + if (enrichedCall.name === FOCUS_TOOL_NAME && this.focusCoordinator) { + // Focus is a property of the primary resident's attention; a fork or + // side-agent calling it would narrow someone else's window. + let result: { success: boolean; data?: unknown; error?: string; isError?: boolean } = + agentName === this.primaryAgentName + ? this.focusCoordinator.handleTool(enrichedCall.input as Record) + : { success: false, isError: false, error: 'focus is available to the primary resident only' }; + // A successful enter (or re-target) moves THIS turn's prose pin to the + // focus channel, same as channel_open: choosing the channel is the + // strongest "my next words go here" signal there is, and prose written + // after the call must not land in the stale pre-focus channel. The + // announcement rides the tool result. Later turns get the focus + // channel from resolveLocus (the registry's focus locus). + const focused = result.success + ? (result.data as { focused?: unknown } | undefined)?.focused + : undefined; + if (typeof focused === 'string') { + const focusAgent = this.agents.get(agentName); + if (focusAgent && (focusAgent.proseRouting === 'locus' || focusAgent.proseRouting === 'hybrid')) { + this.turnLocusPins.set(agentName, focused); + this.lastAnnouncedLocus.set(agentName, focused); + result = { + ...result, + data: { + ...(result.data as Record), + routing: 'Your plain speech lands in the focus channel until focus ends. Other channels need an explicit send tool.', + }, + }; + console.error(`[routing] ${agentName}: focus -> pin moved to ${focused} (announced in tool result)`); + } + } + this.queue.push({ + type: 'tool-result', + callId: enrichedCall.id, + agentName, + moduleName: 'focus', + result, + }); + return; + } if ( this.tuneOutCoordinator && agentName === this.subconsciousAgentName && diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 6db0a737c..dbc6f98fa 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -88,6 +88,12 @@ interface PendingEvent { * newer ambient chatter when a batched wake picks its provenance — the same * rule the framework applies to the turn's speech locus. */ addressed: boolean; + /** The gate metadata the event was evaluated with, kept so a hold predicate + * installed LATER (focus entry) can re-derive the event's channel exactly + * as the live evaluation does — a push event's channel rides in its origin + * fields, not in `channelId`. */ + metadata?: Record; + tags?: string[]; } /** @@ -534,6 +540,8 @@ export class EventGate { private inferenceBuffer: PendingEvent[] = []; /** Host quiesce suppression — see setQuiesced. */ private quiesced = false; + /** Focus hold — see setHoldPredicate. */ + private holdPredicate: ((info: GateEventInfo) => boolean) | null = null; /** Lifetime count of buffer-cap evictions (see bufferForInference). */ private droppedBufferedEvents = 0; @@ -997,6 +1005,25 @@ export class EventGate { evaluate(info: GateEventInfo): GateDecision { this.reloadIfChanged(); + // Focus hold (strict — nothing breaks through): a held event must not + // reach ANY policy. In particular a debounce policy would otherwise queue + // it here and later deliver a "[Gate: N events]" wake for traffic the + // framework is holding out of the window. + if (this.holdPredicate?.(info)) { + this.totalEvaluations++; + this.emitTrace({ + type: 'gate:decision', + eventType: info.eventType, + serverId: info.serverId || undefined, + channelId: info.channelId || undefined, + matchedPolicy: 'focus-held', + trigger: false, + behavior: 'skip', + timestamp: this.now(), + }); + return { trigger: false, policyName: 'focus-held', behavior: 'skip' }; + } + // Sleep suppression: while asleep, drop EXTERNAL wakes. Two things still // rouse the agent — its own heartbeat (an internal clock, not an outside // interruption: each tick lets the agent reconsider and optionally `wake`) @@ -1451,6 +1478,8 @@ export class EventGate { authorId: this.extractAuthorId(info.metadata) ?? undefined, serverId: info.serverId || undefined, addressed: Array.isArray(info.tags) && info.tags.includes('chat:addressed'), + ...(info.metadata ? { metadata: info.metadata } : {}), + ...(info.tags ? { tags: info.tags } : {}), }; const existing = this.debounceTimers.get(policy.name); @@ -1498,6 +1527,10 @@ export class EventGate { // ========================================================================= private deliverEvents(events: PendingEvent[]): void { + // Delivery-time re-check of the hold: a queued event the predicate now + // holds (focus entered after it was queued, purge notwithstanding) must + // not wake the agent inside the window it is holding. + if (this.holdPredicate) events = events.filter((e) => !this.isHeldPending(e)); if (events.length === 0) return; const policyNames = [...new Set(events.map(e => e.policyName))]; @@ -1588,6 +1621,48 @@ export class EventGate { if (!quiesced) this.flushInferenceBufferIfClear(); } + /** + * Install (or clear with null) the focus hold: events the predicate + * accepts are dropped before policy evaluation (no trigger, no debounce + * queueing). Installing also purges already-queued debounce/buffered + * events the predicate would hold, so a wake queued seconds before focus + * began cannot land inside it. + */ + setHoldPredicate(predicate: ((info: GateEventInfo) => boolean) | null): number { + this.holdPredicate = predicate; + if (!predicate) return 0; + const held = (e: PendingEvent): boolean => this.isHeldPending(e); + let purged = 0; + for (const [name, state] of this.debounceTimers) { + const before = state.events.length; + state.events = state.events.filter((e) => !held(e)); + purged += before - state.events.length; + if (state.events.length === 0) { + clearTimeout(state.timer); + this.debounceTimers.delete(name); + } + } + const beforeBuffer = this.inferenceBuffer.length; + this.inferenceBuffer = this.inferenceBuffer.filter((e) => !held(e)); + purged += beforeBuffer - this.inferenceBuffer.length; + return purged; + } + + /** The hold predicate applied to a queued event, with the SAME inputs the + * live evaluation had (metadata + tags), so purge, delivery re-check and + * live evaluation cannot disagree about one event. */ + private isHeldPending(e: PendingEvent): boolean { + if (!this.holdPredicate) return false; + return this.holdPredicate({ + content: e.content, + eventType: e.eventType, + serverId: e.serverId ?? '', + channelId: e.channelId ?? '', + metadata: e.metadata ?? (e.channelId ? { channelId: e.channelId } : undefined), + ...(e.tags ? { tags: e.tags } : {}), + }); + } + private flushInferenceBufferIfClear(): void { if (this.quiesced) return; if (this.inferring.size === 0 && this.inferenceBuffer.length > 0) { diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 9e31ef7e1..0add5e04b 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -89,18 +89,52 @@ export interface TuneOutParams { expiresAtMs?: number; } +/** + * Parameters of the active focus epoch (per resident, not per channel): + * only `channelId` reaches the resident; everything else is held with + * `metadata.focusHeld = { epochId }` (permanent main-view exclusion) until + * the epoch ends. Durable in the lifecycle log as a `focus` record + * (null = ended), last-record-wins on replay. + */ +export interface FocusParams { + epochId: string; + /** The one channel that stays live. */ + serverId: string; + channelId: string; + startedAtMs: number; + /** Chronicle sequence when focus began (audit bound). */ + startedAtSequence: number; + /** Absolute deadline; the epoch ends itself here. */ + expiresAtMs: number; + /** Newest raw messages delivered PER CHANNEL at unfocus. */ + backlogCap: number; + /** Re-target release points: held messages of `channelId` with + * sequence <= value were already delivered (a re-target dumps the new + * focus channel's backlog early). Stamps are never mutated. */ + released?: Record; +} + interface ChannelLifecycleEvent { kind: | 'desired-state' | 'legacy-policy-migrated' | 'invitation-declined' - | 'tune-out-wake'; + | 'tune-out-wake' + | 'focus' + | 'focus-autoreply'; serverId: string; timestamp: string; channelId?: string; desired?: DesiredChannelState; /** Present when desired === 'tuned-out'. */ tuneOut?: TuneOutParams; + /** kind 'focus': the active epoch, or null when focus ended. */ + focus?: FocusParams | null; + /** kind 'focus-autoreply': an automatic reply that actually published, + * for (serverId, channelId, authorId) under `epochId`. Durable so the + * once-per-author rule survives a restart without trusting an optimistic + * stamp written before the publish settled. */ + authorId?: string; /** kind 'tune-out-wake': durable running wake count for an epoch. * Lives in the lifecycle log (not gate stats) because gate runtime * state dies with the process and max-wakes must not reset on restart. */ @@ -633,6 +667,16 @@ export class ChannelRegistry { private defaultPublishMessageId: string | null = null; private defaultPublishThreadId: string | undefined = undefined; + /** + * The focus channel while the host is in focus mode, installed by the + * framework's focus coordinator (NOT read from the durable focus record: + * a replayed epoch with no live coordinator must not pin anything). While + * set it outranks the trigger channel and the global default in locus + * resolution, and inbound from any other channel stops retargeting the + * global default — a held message must not move outbound speech. + */ + private focusLocus: { serverId: string; channelId: string } | null = null; + /** Per-channel typing indicator timers. */ private typingIntervals = new Map>(); @@ -653,6 +697,12 @@ export class ChannelRegistry { wakeCount?: number; }>(); + /** Chronicle-projected focus epoch (one per resident), or null. */ + private focusState: FocusParams | null = null; + /** Autoreplies published under the ACTIVE focus epoch + * (`serverId\0channelId\0authorId`); cleared when the epoch changes. */ + private focusAutoReplies = new Set(); + /** One-time migration inputs from the retired recipe auto-open policy. */ private legacyPolicies = new Map(); private migratedLegacyPolicies = new Set(); @@ -932,12 +982,19 @@ export class ChannelRegistry { // deliberately after §14.5 validation: a rejected message from an // unregistered channel must not retarget outbound speech (the locus is // exactly the authority a self-attested channel would be stealing). + // Under focus, only the focus channel's traffic counts: everything + // else is held, and a message the resident can't see must not decide + // where the resident's next words go (2026-09-30: an ambient channel + // message arriving one second after a DM pulled the DM reply into + // that channel). // A coalesced item is "accepted" only once the coalescer admits it, so // for those this runs after the hook (below). const markAccepted = () => { - this.defaultPublishChannel = message.channelId; - this.defaultPublishMessageId = message.messageId; - this.defaultPublishThreadId = message.threadId; + if (this.admitsAsDefault(serverId, message.channelId)) { + this.defaultPublishChannel = message.channelId; + this.defaultPublishMessageId = message.messageId; + this.defaultPublishThreadId = message.threadId; + } // A server sending channels/incoming is authoritative evidence that // the transport is actually open. This repairs transient status only; // durable desired state still changes exclusively through lifecycle @@ -1346,7 +1403,7 @@ export class ChannelRegistry { // agent was told the wrong channel under concurrency. const home = agentName ? this.homeChannelResolver?.(agentName) : undefined; const active = agentName ? this.activeChannelResolver?.(agentName) : undefined; - const outgoing = home ?? active ?? this.defaultPublishChannel; + const outgoing = home ?? this.focusLocus?.channelId ?? active ?? this.defaultPublishChannel; if (openChannels.length === 0 && !outgoing) { return undefined; @@ -1478,6 +1535,38 @@ export class ChannelRegistry { } } else if (event.kind === 'legacy-policy-migrated') { this.migratedLegacyPolicies.add(event.serverId); + } else if (event.kind === 'focus') { + // Last record wins; a malformed epoch (no id / channel) reads as + // ended rather than crashing boot on a corrupt log. + const f = event.focus; + if ( + f && typeof f === 'object' && typeof f.epochId === 'string' && + typeof f.channelId === 'string' && typeof f.serverId === 'string' && + typeof f.expiresAtMs === 'number' && Number.isFinite(f.expiresAtMs) + ) { + // Optional fields fall back to sane values rather than failing the + // epoch: a record missing backlogCap must still dump a backlog. + if (this.focusState?.epochId !== f.epochId) this.focusAutoReplies.clear(); + this.focusState = { + ...f, + backlogCap: typeof f.backlogCap === 'number' && Number.isFinite(f.backlogCap) + ? Math.max(0, Math.floor(f.backlogCap)) : 20, + startedAtMs: typeof f.startedAtMs === 'number' && Number.isFinite(f.startedAtMs) + ? f.startedAtMs : f.expiresAtMs, + startedAtSequence: typeof f.startedAtSequence === 'number' ? f.startedAtSequence : 0, + }; + } else { + this.focusState = null; + this.focusAutoReplies.clear(); + } + } else if ( + event.kind === 'focus-autoreply' && + typeof event.channelId === 'string' && typeof event.epochId === 'string' + ) { + // Only while the epoch that recorded it is still the active one. + if (this.focusState?.epochId === event.epochId) { + this.focusAutoReplies.add(this.autoReplyKey(event.serverId, event.channelId, event.authorId)); + } } } } @@ -1938,6 +2027,122 @@ export class ChannelRegistry { return { params: current.tuneOut, wakeCount: current.wakeCount ?? 0 }; } + /** + * Durably set (params) or end (null) the resident's focus epoch. The + * focus coordinator owns the hold/dump flow; this is only the state flip. + */ + setFocus(params: FocusParams | null, source: string): void { + const serverId = params?.serverId ?? this.focusState?.serverId ?? 'host'; + if (this.focusState?.epochId !== params?.epochId) this.focusAutoReplies.clear(); + this.focusState = params; + this.appendLifecycleEvent({ + kind: 'focus', + serverId, + ...(params ? { channelId: params.channelId } : {}), + focus: params, + source, + timestamp: new Date().toISOString(), + }); + } + + /** The active focus epoch, or null. */ + getFocus(): FocusParams | null { + return this.focusState; + } + + private autoReplyKey(serverId: string, channelId: string, authorId: string | undefined): string { + return `${serverId}\u0000${channelId}\u0000${authorId ?? ''}`; + } + + /** Durably record that an automatic reply PUBLISHED under the active + * epoch. No-op (false) if `epochId` is no longer the active epoch — a + * publish that settles after its epoch ended must not mark the next one. */ + recordFocusAutoReply(epochId: string, serverId: string, channelId: string, authorId: string | undefined): boolean { + if (this.focusState?.epochId !== epochId) return false; + this.focusAutoReplies.add(this.autoReplyKey(serverId, channelId, authorId)); + this.appendLifecycleEvent({ + kind: 'focus-autoreply', + serverId, + channelId, + epochId, + ...(authorId !== undefined ? { authorId } : {}), + timestamp: new Date().toISOString(), + }); + return true; + } + + /** Has an automatic reply already published for this (channel, author) + * under the active epoch? */ + hasFocusAutoReply(serverId: string, channelId: string, authorId: string | undefined): boolean { + return this.focusAutoReplies.has(this.autoReplyKey(serverId, channelId, authorId)); + } + + /** Number of automatic replies published under the active epoch. */ + focusAutoReplyCount(): number { + return this.focusAutoReplies.size; + } + + /** + * Install (target) or clear (null) the focus routing locus. Called by the + * framework whenever the focus coordinator's state changes; see + * `focusLocus` for what it governs. + */ + setFocusLocus(target: { serverId: string; channelId: string } | null): void { + this.focusLocus = target ? { serverId: target.serverId, channelId: target.channelId } : null; + } + + /** The installed focus routing locus, or null when not focused. */ + getFocusLocus(): { serverId: string; channelId: string } | null { + return this.focusLocus; + } + + /** May inbound on (serverId, channelId) retarget the global default locus? */ + private admitsAsDefault(serverId: string, channelId: string): boolean { + const focus = this.focusLocus; + return focus === null || (focus.serverId === serverId && focus.channelId === channelId); + } + + /** + * Resolve a channel the way the channel tools do: by id, optionally + * server-qualified, with an explicit error when the bare id is registered + * by more than one server (identity is (serverId, channelId)). + */ + resolveChannel( + channelId: string, + serverId?: string, + ): { entry?: { serverId: string; channelId: string; label?: string }; error?: string } { + const r = this.resolveToolChannelEntry(channelId, serverId); + if (!r.entry) return { error: r.error }; + return { + entry: { + serverId: r.entry.serverId, + channelId: r.entry.descriptor.id, + ...(typeof r.entry.descriptor.label === 'string' && r.entry.descriptor.label + ? { label: r.entry.descriptor.label } : {}), + }, + }; + } + + /** + * May the host post into this channel on the resident's behalf (focus + * autoreply)? Channels the resident is open in, and DMs (which carry no + * open/closed decision of the resident's). A channel the resident declined + * or never joined is off limits: the host must not speak in a room the + * resident chose to stay out of. + */ + canAutoReplyInto(serverId: string, channelId: string): boolean { + const entry = this.channels.get(`${serverId}:${channelId}`); + if (!entry) return false; + if (extractDmMeta(entry.descriptor.metadata).isDm) return true; + return this.getDesiredState(serverId, channelId) === 'open'; + } + + /** Human label of a registered channel, when known. */ + channelLabel(serverId: string, channelId: string): string | undefined { + const label = this.channels.get(`${serverId}:${channelId}`)?.descriptor.label; + return typeof label === 'string' && label ? label : undefined; + } + /** Registered channel entries (read-only iteration for the coordinator). */ listChannelsRaw(): Array<{ serverId: string; descriptor: ChannelDescriptor }> { return [...this.channels.values()].map((e) => ({ @@ -1976,9 +2181,10 @@ export class ChannelRegistry { channelId: string, text: string, agentName: string, + serverId?: string, ): Promise<{ success: boolean; data?: unknown; error?: string; isError?: boolean }> { this.emitTraceFn({ type: 'mcpl:speech-routed', conversationId: agentName, channelId, text }); - return this.handleToolPublish({ channelId, content: text }); + return this.handleToolPublish({ channelId, content: text, ...(serverId ? { serverId } : {}) }); } /** @@ -2937,7 +3143,11 @@ export class ChannelRegistry { resolveLocus(conversationId: string): string | null { const home = this.homeChannelResolver?.(conversationId); - return home ?? this.activeChannelResolver?.(conversationId) ?? this.defaultPublishChannel ?? null; + return home + ?? this.focusLocus?.channelId + ?? this.activeChannelResolver?.(conversationId) + ?? this.defaultPublishChannel + ?? null; } async routeSpeech( @@ -2982,7 +3192,12 @@ export class ChannelRegistry { return fail(null, 'turn has no locus (no home/trigger channel; nothing to deliver into)'); } - const entry = this.findChannelEntry(channelId); + // Under focus the locus IS the focus channel, whose identity is + // (serverId, channelId): resolve it server-qualified so a same-id channel + // registered by another server cannot receive the resident's words. + const entry = (this.focusLocus && this.focusLocus.channelId === channelId + ? this.channels.get(`${this.focusLocus.serverId}:${channelId}`) + : undefined) ?? this.findChannelEntry(channelId); if (!entry) { return fail(channelId, `no registered channel for locus "${channelId}"`); } @@ -3065,7 +3280,12 @@ export class ChannelRegistry { return { delivered, channelId, ...(messageId !== undefined ? { messageId } : {}) }; } - private async handleToolPublish(input: { channelId?: string; content?: string; text?: string }): Promise { + private async handleToolPublish(input: { + channelId?: string; content?: string; text?: string; + /** Server-qualify the channel (identity is (serverId, channelId)); an + * unqualified id that two servers registered is first-match. */ + serverId?: string; + }): Promise { // Resolve content: accept both `content` and `text` (backward compat) const messageText = input.content ?? input.text; if (!messageText) { @@ -3086,11 +3306,15 @@ export class ChannelRegistry { }; } - const entry = this.findChannelEntry(channelId); + const entry = input.serverId + ? this.resolveToolChannelEntry(channelId, input.serverId).entry + : this.findChannelEntry(channelId); if (!entry) { return { success: false, - error: `Channel not found: ${channelId}`, + error: input.serverId + ? `Channel not found: ${channelId} on server ${input.serverId}` + : `Channel not found: ${channelId}`, isError: true, }; } diff --git a/src/types/framework.ts b/src/types/framework.ts index a0312561a..e49cce379 100644 --- a/src/types/framework.ts +++ b/src/types/framework.ts @@ -99,6 +99,13 @@ export interface FrameworkConfig { */ subconscious?: import('../tune-out/tools.js').SubconsciousConfig; + /** + * Focus mode: the primary resident can narrow attention to one channel for + * a bounded time; everything else is held and delivered at unfocus. Exposes + * the `focus` tool when enabled. See src/focus/coordinator.ts. + */ + focus?: import('../focus/coordinator.js').FocusConfig; + /** * IANA zone used only when rendering wall-clock times for the agent. * Stored/protocol timestamps remain epoch/UTC. Defaults to AGENT_TIMEZONE, diff --git a/test/channel-registry-routing.test.ts b/test/channel-registry-routing.test.ts index 80a825658..e664029a7 100644 --- a/test/channel-registry-routing.test.ts +++ b/test/channel-registry-routing.test.ts @@ -528,6 +528,63 @@ test('DM prose targets resolve people-first: @name, prefix-lenient names, and <@ assert.ok('error' in missing && (missing.candidates?.length ?? 0) === 2, 'lists known DMs by name'); }); +test('focus locus: held inbound does not retarget the default; the focus channel outranks trigger and default', async () => { + // 2026-09-30: an ambient channel message arriving one second after a DM + // retargeted the process-global default, and the DM's batched (channel- + // less) wake delivered the DM reply into that channel. Under focus, only + // the focus channel's traffic may move the default, and the focus channel + // is the locus regardless of trigger/default. + let active: string | undefined; + const { registry, publishCalls } = makeRegistry({ delivered: true }, undefined, () => active); + seedRegistered(registry, 'discord', 'chanA', 'chanB'); + + registry.handleIncoming('discord', incoming('chanA', 'hi from A')); + assert.equal(registry.getDefaultPublishChannel(), 'chanA'); + + registry.setFocusLocus({ serverId: 'discord', channelId: 'chanA' }); + registry.handleIncoming('discord', incoming('chanB', 'held chatter from B')); + assert.equal(registry.getDefaultPublishChannel(), 'chanA', 'held inbound must not retarget the default'); + assert.equal(registry.buildChannelContext('trunk')?.incoming?.channelId, 'chanA', + 'refusal reactions / typing target the admitted inbound, not the held one'); + + active = 'chanB'; // even a trigger that slipped past the hold + assert.equal(registry.resolveLocus('trunk'), 'chanA', 'focus channel outranks the trigger channel'); + assert.equal(registry.buildChannelContext('trunk')?.defaultOutgoing?.channelId, 'chanA', + 'advertised outgoing matches where speech lands'); + const res = await registry.routeSpeech('trunk', 'reply', registry.resolveLocus('trunk')); + assert.deepEqual(res, { delivered: true, channelId: 'chanA' }); + assert.equal(publishCalls.at(-1)?.channelId, 'chanA'); + + // Focus inbound on the focus channel still counts. + registry.handleIncoming('discord', incoming('chanA', 'more from A')); + assert.equal(registry.getDefaultPublishChannel(), 'chanA'); + + // Unfocused: normal resolution resumes; the held message never became the default. + registry.setFocusLocus(null); + assert.equal(registry.resolveLocus('trunk'), 'chanB', 'trigger channel wins again once unfocused'); + active = undefined; + assert.equal(registry.resolveLocus('trunk'), 'chanA', 'default is the last ADMITTED inbound'); + registry.handleIncoming('discord', incoming('chanB', 'after focus')); + assert.equal(registry.getDefaultPublishChannel(), 'chanB', 'inbound retargets normally after focus'); +}); + +test('focus locus: same channel id on another server does not retarget the default', () => { + const { registry } = makeRegistry({ delivered: true }); + seedRegistered(registry, 'discord', 'chanA'); + seedRegistered(registry, 'slack', 'chanA', 'chanZ'); + registry.handleIncoming('discord', incoming('chanA', 'hi')); + registry.setFocusLocus({ serverId: 'discord', channelId: 'chanA' }); + registry.handleIncoming('slack', { + messages: [{ ...incoming('chanZ', 'z').messages[0], messageId: 'mz' }], + }); + assert.equal(registry.getDefaultPublishChannel(), 'chanA'); + registry.handleIncoming('slack', { + messages: [{ ...incoming('chanA', 'same id, other server').messages[0], messageId: 'm-slack' }], + }); + assert.equal(registry.buildChannelContext('trunk')?.incoming?.messageId, 'm1', + 'the other server\'s chanA is not the focus channel'); +}); + test('proseTargetFor: one whitespace-free token per channel that resolves back to it', async () => { const { registry } = makeRegistry({ delivered: true }); await registry.handleChanged('discord', { @@ -590,3 +647,17 @@ test('proseTargetFor: no token when none is safe (shared id across servers, whit assert.equal(registry.proseTargetFor('chan:8', 'discord-a'), '#only-a'); assert.equal(registry.proseTargetFor('chan:8', 'discord-b'), undefined); }); + +test('focus locus: speech goes to the focused SERVER\'s channel when another server registered the same id first', async () => { + const { registry, traces } = makeRegistry({ delivered: true }); + seedRegistered(registry, 'discord', 'chanA'); // registered first → an id-only lookup finds this one + seedRegistered(registry, 'slack', 'chanA'); + registry.setFocusLocus({ serverId: 'slack', channelId: 'chanA' }); + const res = await registry.routeSpeech('trunk', 'for the slack room', registry.resolveLocus('trunk')); + assert.deepEqual(res, { delivered: true, channelId: 'chanA' }); + const routed = traces.filter((t) => t.type === 'mcpl:speech-routed' || t.type === 'mcpl:channel-opened-by-send'); + assert.ok(routed.length > 0, 'speech was routed'); + assert.ok(routed.every((t) => t.serverId === undefined || t.serverId === 'slack'), + `routed through the focused server, not the first-registered one: ${JSON.stringify(routed)}`); + assert.ok(routed.some((t) => t.serverId === 'slack')); +}); diff --git a/test/focus-coordinator.test.ts b/test/focus-coordinator.test.ts new file mode 100644 index 000000000..de5d2ccab --- /dev/null +++ b/test/focus-coordinator.test.ts @@ -0,0 +1,547 @@ +/** + * FocusCoordinator — unit tests against stub registry/hooks. + * + * Covers: hold predicate on (serverId, channelId), autoreply once per + * (channel, author) per epoch and only where the host may speak, end dump + * with per-channel cap + backscroll hint naming only tools the resident + * has, deferred (mid-turn) held messages in the backlog, re-target + * (release + no double delivery), duration/cap clamps incl. fractional + * config maxima, tune-out refusal, disabled-config teardown, expiry at + * resume, media placeholders. + */ +import { describe, it, after } from 'node:test'; +import assert from 'node:assert/strict'; +import type { StoredMessage } from '@animalabs/context-manager'; +import type { ContentBlock } from '@animalabs/membrane'; +import { + FocusCoordinator, DEFAULT_AUTOREPLY, resolveFocusLimits, buildFocusToolDefinition, + type FocusFrameworkHooks, type FocusConfig, type DeferredMessageView, +} from '../src/focus/coordinator.js'; +import type { ChannelRegistry, FocusParams } from '../src/mcpl/channel-registry.js'; + +/** Every coordinator built here; armed expiry timers are cleared at the end so + * no test leaves a live timer keeping the runner alive. */ +const built: FocusCoordinator[] = []; +after(() => { for (const c of built) c.stop(); }); + +const CHANNELS: Array<{ serverId: string; id: string; label: string; open: boolean; dm?: boolean }> = [ + { serverId: 'discord', id: 'discord:g:work', label: 'work', open: true }, + { serverId: 'discord', id: 'discord:g:chat', label: 'chat', open: true }, + { serverId: 'discord', id: 'discord:g:lurk', label: 'lurk', open: false }, + { serverId: 'discord', id: 'discord:dm:antra', label: 'DM: antra', open: false, dm: true }, + // Same id string as `work`, on another server. + { serverId: 'slack', id: 'discord:g:work', label: 'imposter', open: true }, + { serverId: 'slack', id: 'slack:general', label: 'general', open: true }, +]; + +function harness(opts?: { + tools?: string[]; config?: Partial; tunedOut?: string[]; publishFails?: boolean; +}) { + let focus: FocusParams | null = null; + const stored: StoredMessage[] = []; + const deferredQueue: DeferredMessageView[] = []; + const delivered: Array<{ participant: string; text: string; metadata?: Record }> = []; + const published: Array<{ serverId: string; channelId: string; text: string }> = []; + const wakes: string[] = []; + const gateStates: Array = []; + let seq = 0; + let publishFails = opts?.publishFails ?? false; + /** When set, publishes wait on it — lets a test settle one late. */ + let publishGate: Promise | null = null; + + const replies = new Set(); + const rk = (s: string, c: string, a?: string) => `${s}|${c}|${a ?? ''}`; + const registry = { + getFocus: () => focus, + setFocus: (p: FocusParams | null) => { if (focus?.epochId !== p?.epochId) replies.clear(); focus = p; }, + hasFocusAutoReply: (s: string, c: string, a?: string) => replies.has(rk(s, c, a)), + recordFocusAutoReply: (epochId: string, s: string, c: string, a?: string) => { + if (focus?.epochId !== epochId) return false; + replies.add(rk(s, c, a)); + return true; + }, + focusAutoReplyCount: () => replies.size, + resolveChannel: (channelId: string, serverId?: string) => { + const matches = CHANNELS.filter((c) => c.id === channelId && (!serverId || c.serverId === serverId)); + if (matches.length === 0) return { error: `Channel not found: ${channelId}` }; + if (matches.length > 1) return { error: `Channel id is ambiguous across MCPL servers: ${channelId}. Include serverId.` }; + return { entry: { serverId: matches[0]!.serverId, channelId: matches[0]!.id, label: matches[0]!.label } }; + }, + listChannelsRaw: () => CHANNELS.map((c) => ({ serverId: c.serverId, descriptor: { id: c.id, label: c.label } })), + } as unknown as ChannelRegistry; + + const hooks: FocusFrameworkHooks = { + addMessage: (participant, content, metadata) => { + const text = content.map((b) => (b.type === 'text' ? b.text : '')).join(''); + delivered.push({ participant, text, metadata }); + return { id: `d-${delivered.length}`, deferred: false }; + }, + requestInference: (agentName) => { wakes.push(agentName); }, + primaryName: () => 'scout', + getStoredMessages: () => stored, + getDeferredMessages: () => deferredQueue, + currentSequence: () => seq, + channelLabel: (s, c) => CHANNELS.find((x) => x.serverId === s && x.id === c)?.label, + publish: async (serverId, channelId, text) => { + if (publishGate) await publishGate; + if (publishFails) return { success: false, error: 'boom' }; + published.push({ serverId, channelId, text }); + return { success: true }; + }, + canAutoReplyInto: (s, c) => { + const ch = CHANNELS.find((x) => x.serverId === s && x.id === c); + return !!ch && (ch.open || !!ch.dm); + }, + isTunedOut: (s, c) => (opts?.tunedOut ?? []).includes(`${s}:${c}`), + timeZone: () => 'UTC', + availableToolNames: () => opts?.tools ?? ['history--extract', 'channel_open'], + isForkRouted: () => false, + onFocusChanged: (s) => { gateStates.push(s); }, + emitTrace: () => {}, + }; + const coordinator = new FocusCoordinator(registry, { + enabled: true, + defaultBacklogCap: 3, + ...(opts?.config ?? {}), + }, hooks); + built.push(coordinator); + + const mkMeta = (serverId: string, channelId: string, o?: { addressed?: boolean; authorId?: string }, stamp?: unknown) => ({ + channelId, serverId, author: { id: o?.authorId ?? 'U1', name: 'someone' }, + ...(o?.addressed ? { tags: ['chat:mention', 'chat:addressed'] } : {}), + ...(stamp ? { focusHeld: stamp } : {}), + }) as StoredMessage['metadata']; + + /** Simulate the framework's ingestion: consult, stamp, store. */ + const incoming = (channelId: string, text: string, o?: { addressed?: boolean; authorId?: string; serverId?: string; content?: ContentBlock[] }) => { + const serverId = o?.serverId ?? 'discord'; + const stamp = coordinator.onIncoming(serverId, channelId, `m${++seq}`, + o?.addressed ? ['chat:mention', 'chat:addressed'] : [], { id: o?.authorId ?? 'U1', name: 'someone' }); + stored.push({ + id: `m${seq}`, sequence: seq, participant: 'user', timestamp: new Date(seq * 1000), + content: o?.content ?? [{ type: 'text', text }], + metadata: mkMeta(serverId, channelId, o, stamp), + }); + return stamp; + }; + /** A held message that arrived mid-turn: stamped, but parked in the deferred queue. */ + const incomingDeferred = (channelId: string, text: string, o?: { addressed?: boolean; authorId?: string }) => { + const stamp = coordinator.onIncoming('discord', channelId, `m${++seq}`, + o?.addressed ? ['chat:mention', 'chat:addressed'] : [], { id: o?.authorId ?? 'U1', name: 'someone' }); + deferredQueue.push({ participant: 'user', content: [{ type: 'text', text }], metadata: mkMeta('discord', channelId, o, stamp) as Record }); + return stamp; + }; + + return { + coordinator, incoming, incomingDeferred, delivered, published, wakes, gateStates, stored, deferredQueue, + getFocus: () => focus, setFocus: (p: FocusParams | null) => { if (focus?.epochId !== p?.epochId) replies.clear(); focus = p; }, + setPublishFails: (v: boolean) => { publishFails = v; }, + setPublishGate: (g: Promise | null) => { publishGate = g; }, + recordReply: (s: string, c: string, a?: string) => { replies.add(rk(s, c, a)); }, + replyCount: () => replies.size, + }; +} + +const settle = () => new Promise((r) => setImmediate(r)); + +describe('focus: hold predicate', () => { + it('not focused → nothing is held', () => { + const h = harness(); + assert.equal(h.incoming('discord:g:chat', 'hi'), null); + }); + + it('focus channel passes; every other channel and DM is held and stamped with the epoch', () => { + const h = harness(); + const r = h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + assert.ok(r.ok); + assert.equal(h.incoming('discord:g:work', 'on topic'), null); + const held = h.incoming('discord:g:chat', 'off topic'); + assert.equal(held?.epochId, h.getFocus()!.epochId); + const dm = h.incoming('discord:dm:antra', 'psst'); + assert.equal(dm?.epochId, h.getFocus()!.epochId); + assert.equal(h.gateStates.length, 1); + assert.equal(h.gateStates[0]?.channelId, 'discord:g:work'); + }); + + it('identity is (serverId, channelId): the same id string on another server is held', () => { + const h = harness(); + assert.ok(h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool').ok); + const imposter = h.incoming('discord:g:work', 'same id, other server', { serverId: 'slack' }); + assert.ok(imposter, 'held'); + assert.equal(h.incoming('discord:g:work', 'the real one', { serverId: 'discord' }), null); + }); + + it('an ambiguous channel id is refused until serverId disambiguates; unknown channel refused', () => { + const h = harness(); + const ambiguous = h.coordinator.enter('discord:g:work', {}, 'agent-tool'); + assert.equal(ambiguous.ok, false); + assert.match((ambiguous as { error: string }).error, /ambiguous/); + assert.ok(h.coordinator.enter('discord:g:work', { serverId: 'slack' }, 'agent-tool').ok); + assert.equal(h.getFocus()!.serverId, 'slack'); + const h2 = harness(); + assert.equal(h2.coordinator.enter('discord:g:nope', {}, 'agent-tool').ok, false); + }); + + it('a tuned-out channel cannot be focused (its traffic goes to the subconscious)', () => { + const h = harness({ tunedOut: ['discord:discord:g:chat'] }); + const r = h.coordinator.enter('discord:g:chat', {}, 'agent-tool'); + assert.equal(r.ok, false); + assert.match((r as { error: string }).error, /tuned out/); + }); + + it('disabled configuration: enter refused; a stale persisted epoch ends at resume with its dump', () => { + const h = harness({ config: { enabled: false } }); + assert.match((h.coordinator.enter('discord:g:chat', {}, 'x') as { error: string }).error, /disabled/); + h.setFocus({ + epochId: 'stale', serverId: 'discord', channelId: 'discord:g:work', + startedAtMs: Date.now() - 1000, startedAtSequence: 0, expiresAtMs: Date.now() + 60_000, backlogCap: 3, + }); + h.stored.push({ + id: 'x1', sequence: 1, participant: 'user', timestamp: new Date(), + content: [{ type: 'text', text: 'held before the switch flipped' }], + metadata: { channelId: 'discord:g:chat', serverId: 'discord', focusHeld: { epochId: 'stale' } } as StoredMessage['metadata'], + }); + h.coordinator.resume(); + assert.equal(h.getFocus(), null); + assert.match(h.delivered[0]!.text, /disabled in the configuration/); + assert.match(h.delivered[0]!.text, /held before the switch flipped/); + assert.deepEqual(h.wakes, ['scout']); + }); +}); + +describe('focus: autoreply', () => { + it('replies once per (channel, author) per epoch, to addressed messages only, server-qualified', async () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', 'ambient', { authorId: 'A' }); + h.incoming('discord:g:chat', '@scout?', { addressed: true, authorId: 'A' }); + h.incoming('discord:g:chat', '@scout??', { addressed: true, authorId: 'A' }); // same tick: in-flight dedupe + h.incoming('discord:g:chat', '@scout', { addressed: true, authorId: 'B' }); + h.incoming('discord:dm:antra', 'dm', { addressed: true, authorId: 'A' }); + await settle(); + h.incoming('discord:g:chat', '@scout???', { addressed: true, authorId: 'A' }); // later: durable dedupe + await settle(); + assert.deepEqual(h.published.map((p) => [p.serverId, p.channelId]), + [['discord', 'discord:g:chat'], ['discord', 'discord:g:chat'], ['discord', 'discord:dm:antra']]); + assert.equal(h.replyCount(), 3, 'each published reply is recorded durably'); + assert.match(h.published[0]!.text, /scout is in focus mode/); + assert.match(h.published[0]!.text, /^\[Automatic reply\]/); + }); + + it('never speaks in a channel the resident is not open in; DMs are fine', async () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + const lurk = h.incoming('discord:g:lurk', '@scout from a room you never joined', { addressed: true }); + const dm = h.incoming('discord:dm:antra', 'dm', { addressed: true }); + await settle(); + assert.ok(lurk && dm, 'both held'); + assert.deepEqual(h.published.map((p) => p.channelId), ['discord:dm:antra'], 'held, counted, reply-less in the never-joined room'); + h.coordinator.end('t', 'x'); + assert.match(h.delivered[0]!.text, /2 addressed you \(1 automatic reply was sent\)/); + }); + + it('a failed publish releases the (channel, author) allowance so the next mention retries', async () => { + const h = harness({ publishFails: true }); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', '@scout 1', { addressed: true, authorId: 'A' }); + await settle(); + assert.equal(h.published.length, 0); + assert.equal(h.replyCount(), 0, 'a failed publish records nothing durable — a restart retries too'); + h.setPublishFails(false); + h.incoming('discord:g:chat', '@scout 2', { addressed: true, authorId: 'A' }); + await settle(); + assert.equal(h.published.length, 1, 'retried after the failure'); + h.incoming('discord:g:chat', '@scout 3', { addressed: true, authorId: 'A' }); + await settle(); + assert.equal(h.published.length, 1, 'and then once is once'); + }); + + it('a publish that settles after its epoch ended does not touch the next epoch', async () => { + const h = harness(); + let release!: () => void; + h.setPublishGate(new Promise((r) => { release = r; })); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', '@scout', { addressed: true, authorId: 'A' }); // publish parked + h.coordinator.end('t', 'x'); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); // new epoch + h.setPublishGate(null); + release(); + await settle(); await settle(); + assert.equal(h.replyCount(), 0, 'the late success is not counted in the new epoch'); + // …and did not consume the new epoch's allowance for that author. + h.incoming('discord:g:chat', '@scout again', { addressed: true, authorId: 'A' }); + await settle(); + assert.equal(h.replyCount(), 1); + assert.equal(h.published.length, 2); + }); + + it('autoReply: false suppresses the reply but still holds; the end notice says so', async () => { + const h = harness({ config: { autoReply: false } }); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + const held = h.incoming('discord:g:chat', '@scout', { addressed: true }); + await settle(); + assert.ok(held); + assert.equal(h.published.length, 0); + h.coordinator.end('t', 'x'); + assert.match(h.delivered[0]!.text, /1 addressed you \(no automatic reply was sent\)/); + }); + + it('the default template carries no placeholder residue', async () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', '@scout', { addressed: true }); + await settle(); + assert.doesNotMatch(h.published[0]!.text, /\{[a-z]+\}/); + assert.ok(DEFAULT_AUTOREPLY.includes('{remaining}')); + }); +}); + +describe('focus: end + backlog', () => { + it('delivers per-channel dumps capped to backlogCap (newest kept), counts the rest, hints only real tools, wakes the resident', () => { + const h = harness({ tools: ['channel_open', 'think'] }); + h.coordinator.enter('discord:g:work', { backlogCap: 2, serverId: 'discord' }, 'agent-tool'); + for (let i = 1; i <= 5; i++) h.incoming('discord:g:chat', `chat ${i}`); + h.incoming('discord:dm:antra', 'dm 1', { addressed: true }); + h.incoming('discord:g:work', 'live'); + const r = h.coordinator.end('agent-tool', 'ended by you'); + assert.ok(r.ok && r.held === 6); + assert.equal(h.getFocus(), null); + assert.deepEqual(h.gateStates.at(-1), null); + assert.deepEqual(h.wakes, ['scout']); + + assert.equal(h.delivered.length, 1); + const { text, metadata } = h.delivered[0]!; + assert.equal(metadata?.kind, 'focus-end'); + assert.equal(metadata?.system, true); + assert.match(text, /^\[Focus ended — ended by you\./); + assert.match(text, /6 messages held across 2 channels; 1 addressed you/); + assert.match(text, //); + assert.doesNotMatch(text, /chat 3\n/); + assert.match(text, /someone: chat 4\n.*someone: chat 5\n<\/focus-backlog>/); + assert.match(text, /\n.*dm 1\n<\/focus-backlog>/); + assert.doesNotMatch(text, /live/); + assert.match(text, /channel_open \{channelId, backscroll\}/); + assert.doesNotMatch(text, /history--extract/); + }); + + it('held messages still deferred behind the current turn are in the dump', () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', 'stored one'); + h.incomingDeferred('discord:g:chat', 'arrived mid-turn'); + const s = h.coordinator.status(); + assert.equal(s.held?.[0]?.messages, 2, 'check counts it too'); + const r = h.coordinator.end('t', 'x'); + assert.ok(r.ok && r.held === 2); + assert.match(h.delivered[0]!.text, /messages=2 tz="UTC">\n\[\d\d:\d\d\] someone: stored one\n\[arrived mid-turn\] someone: arrived mid-turn\n<\/focus-backlog>/); + }); + + it('isAwaitingDelivery: true only for the active epoch\'s unreleased held messages', () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', 'held'); + h.incoming('discord:g:work', 'live'); + const [held, live] = h.stored; + assert.equal(h.coordinator.isAwaitingDelivery(held!), true); + assert.equal(h.coordinator.isAwaitingDelivery(live!), false, 'not stamped'); + h.coordinator.enter('discord:g:chat', {}, 'agent-tool'); // re-target releases it + assert.equal(h.coordinator.isAwaitingDelivery(held!), false, 'released by the re-target dump'); + h.incoming('discord:g:work', 'now held'); + const nowHeld = h.stored.at(-1)!; + assert.equal(h.coordinator.isAwaitingDelivery(nowHeld), true); + h.coordinator.end('t', 'x'); + assert.equal(h.coordinator.isAwaitingDelivery(nowHeld), false, 'the dump consumed it'); + }); + + it('media blocks are named, not dropped', () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', '', { content: [{ type: 'image', source: { type: 'base64', media_type: 'image/png', data: '' } } as unknown as ContentBlock] }); + h.incoming('discord:g:chat', 'look', { content: [{ type: 'text', text: 'look' }, { type: 'image', source: { type: 'base64', media_type: 'image/png', data: '' } } as unknown as ContentBlock, { type: 'image', source: { type: 'base64', media_type: 'image/png', data: '' } } as unknown as ContentBlock] }); + h.coordinator.end('t', 'x'); + assert.match(h.delivered[0]!.text, /someone: \[\+1 image\]\n/); + assert.match(h.delivered[0]!.text, /someone: look \[\+2 images\]\n/); + }); + + it('no hint when nothing was truncated; "Nothing was held" when empty', () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', 'one'); + h.coordinator.end('agent-tool', 'x'); + assert.doesNotMatch(h.delivered[0]!.text, /Held messages above/); + + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.coordinator.end('agent-tool', 'x'); + assert.match(h.delivered[1]!.text, /Nothing was held\.\]$/); + }); + + it('ending when not focused is an error, not a crash', () => { + const h = harness(); + assert.equal(h.coordinator.end('agent-tool', 'x').ok, false); + }); +}); + +describe('focus: check', () => { + it('status tallies held per channel; peek renders without releasing', () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', 'a'); + h.incoming('discord:g:chat', '@scout b', { addressed: true }); + h.incoming('discord:dm:antra', 'c'); + const s = h.coordinator.status(); + assert.equal(s.focused, true); + assert.deepEqual(s.held?.map((x) => [x.channelId, x.messages, x.addressed]), [ + ['discord:g:chat', 2, 1], ['discord:dm:antra', 1, 0], + ]); + const p = h.coordinator.peek('discord:g:chat', 1); + assert.ok(p.ok && p.total === 2); + assert.match(p.text, /truncated=1/); + assert.match(p.text, /@scout b/); + h.coordinator.end('agent-tool', 'x'); + assert.match(h.delivered[0]!.text, /messages=2 tz="UTC">/); + }); + + it('tool surface: check while unfocused reports focused:false; serverId threads through enter', () => { + const h = harness(); + assert.deepEqual(h.coordinator.handleTool({ mode: 'check' }), { success: true, data: { focused: false } }); + assert.equal(h.coordinator.handleTool({ mode: 'enter' }).success, false); + assert.equal(h.coordinator.handleTool({ mode: 'bogus' }).success, false); + const r = h.coordinator.handleTool({ mode: 'enter', channelId: 'discord:g:work', serverId: 'slack' }); + assert.equal(r.success, true); + assert.equal((r.data as { serverId: string }).serverId, 'slack'); + }); +}); + +describe('focus: re-target', () => { + it('delivers the new channel\'s stored backlog now, keeps the epoch, and does not deliver it again at end', () => { + const h = harness(); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + const epoch = h.getFocus()!.epochId; + h.incoming('discord:g:chat', 'chat 1'); + h.incoming('discord:g:chat', 'chat 2'); + h.incoming('discord:dm:antra', 'dm 1'); + const r = h.coordinator.enter('discord:g:chat', {}, 'agent-tool'); + assert.ok(r.ok && r.retargeted); + assert.equal(h.getFocus()!.epochId, epoch); + assert.equal(h.getFocus()!.channelId, 'discord:g:chat'); + assert.equal(h.delivered.length, 1); + assert.match(h.delivered[0]!.text, /^\[Focus re-targeted: #work \(discord:g:work\) → #chat \(discord:g:chat\)/); + assert.match(h.delivered[0]!.text, /messages=2 tz="UTC">/); + assert.equal(h.incoming('discord:g:chat', 'chat 3'), null); + assert.ok(h.incoming('discord:g:work', 'work 1')); + h.coordinator.end('agent-tool', 'x'); + const endText = h.delivered[1]!.text; + assert.doesNotMatch(endText, /chat 1|chat 2|chat 3/); + assert.match(endText, /dm 1/); + assert.match(endText, /work 1/); + assert.match(endText, /2 messages held across 2 channels/); + }); + + it('a truncated re-target dump carries the history hint; deferred held messages wait for the end', () => { + const h = harness({ config: { defaultBacklogCap: 1 } }); + h.coordinator.enter('discord:g:work', { serverId: 'discord' }, 'agent-tool'); + h.incoming('discord:g:chat', 'chat 1'); + h.incoming('discord:g:chat', 'chat 2'); + h.incomingDeferred('discord:g:chat', 'chat 3 mid-turn'); + const r = h.coordinator.enter('discord:g:chat', {}, 'agent-tool'); + assert.ok(r.ok); + const t = h.delivered[0]!.text; + assert.match(t, /messages=2 tz="UTC" truncated=1/); + assert.match(t, /Held messages above the per-channel cap of 1/); + assert.doesNotMatch(t, /mid-turn/); + h.coordinator.end('t', 'x'); + assert.match(h.delivered[1]!.text, /chat 3 mid-turn/, 'the deferred one arrives at unfocus'); + assert.doesNotMatch(h.delivered[1]!.text, /chat 1|chat 2/); + }); +}); + +describe('focus: limits', () => { + it('config maxima are floored/capped and integers; defaults clamp into range', () => { + assert.deepEqual(resolveFocusLimits({ enabled: true }), { + minDuration: 60, maxDuration: 14_400, defaultDuration: 1800, maxCap: 200, defaultCap: 20, + }); + const odd = resolveFocusLimits({ enabled: true, maxDurationSeconds: 0.5, maxBacklogCap: 0.5, defaultDurationSeconds: 1e308, defaultBacklogCap: 7 }); + assert.deepEqual(odd, { minDuration: 60, maxDuration: 60, defaultDuration: 60, maxCap: 0, defaultCap: 0 }); + const huge = resolveFocusLimits({ enabled: true, maxDurationSeconds: 1e308 }); + assert.equal(huge.maxDuration, 7 * 24 * 3600); + const desc = buildFocusToolDefinition({ enabled: true, maxDurationSeconds: 10 }).inputSchema.properties!.durationSeconds as { description: string }; + assert.match(desc.description, /min 60, max 60/); + }); + + it('a fractional configured cap cannot deliver the whole backlog', () => { + const h = harness({ config: { maxBacklogCap: 0.5 } }); + h.coordinator.enter('discord:g:work', { backlogCap: 5, serverId: 'discord' }, 'agent-tool'); + assert.equal(h.getFocus()!.backlogCap, 0); + h.incoming('discord:g:chat', 'a'); + h.incoming('discord:g:chat', 'b'); + h.coordinator.end('t', 'x'); + assert.match(h.delivered[0]!.text, /messages=2 tz="UTC" truncated=2 \(oldest, not shown\)>\n\n<\/focus-backlog>/); + }); + + it('duration is clamped to [60, max] and defaults from config', () => { + const h = harness({ config: { maxDurationSeconds: 600 } }); + const before = Date.now(); + const r1 = h.coordinator.enter('discord:g:work', { durationSeconds: 5, serverId: 'discord' }, 't'); + assert.ok(r1.ok); + assert.ok(r1.params.expiresAtMs - before >= 60_000 - 5 && r1.params.expiresAtMs - before <= 61_000); + h.coordinator.end('t', 'x'); + const r2 = h.coordinator.enter('discord:g:work', { durationSeconds: 99_999, serverId: 'discord' }, 't'); + assert.ok(r2.ok); + assert.ok(r2.params.expiresAtMs - Date.now() <= 600_000); + }); +}); + +describe('focus: resume', () => { + it('a deadline that passed while the host was down ends focus at resume, with the dump', () => { + const h = harness(); + const params: FocusParams = { + epochId: 'old', serverId: 'discord', channelId: 'discord:g:work', + startedAtMs: Date.now() - 3_600_000, startedAtSequence: 0, + expiresAtMs: Date.now() - 1000, backlogCap: 3, + }; + h.stored.push({ + id: 'x1', sequence: 1, participant: 'user', timestamp: new Date(), + content: [{ type: 'text', text: 'while you were away' }], + metadata: { channelId: 'discord:g:chat', serverId: 'discord', focusHeld: { epochId: 'old' } } as StoredMessage['metadata'], + }); + h.setFocus(params); + h.coordinator.resume(); + assert.equal(h.getFocus(), null); + assert.match(h.delivered[0]!.text, /deadline elapsed while the host was down/); + assert.match(h.delivered[0]!.text, /while you were away/); + assert.deepEqual(h.wakes, ['scout']); + }); + + it('a live deadline re-arms; the once-per-author rule holds across the restart via the durable record', async () => { + const h = harness(); + const params: FocusParams = { + epochId: 'live', serverId: 'discord', channelId: 'discord:g:work', + startedAtMs: Date.now(), startedAtSequence: 0, + expiresAtMs: Date.now() + 60_000, backlogCap: 3, + }; + h.stored.push({ + id: 'x1', sequence: 1, participant: 'user', timestamp: new Date(), + content: [{ type: 'text', text: '@scout' }], + metadata: { + channelId: 'discord:g:chat', serverId: 'discord', author: { id: 'A', name: 'a' }, + tags: ['chat:addressed'], focusHeld: { epochId: 'live' }, + } as StoredMessage['metadata'], + }); + h.setFocus(params); + h.recordReply('discord', 'discord:g:chat', 'A'); // replayed from the lifecycle log + h.coordinator.resume(); + assert.equal(h.getFocus()?.epochId, 'live'); + assert.deepEqual(h.gateStates.at(-1)?.epochId, 'live'); + h.incoming('discord:g:chat', '@scout again', { addressed: true, authorId: 'A' }); + await settle(); + assert.equal(h.published.length, 0); + }); + + it('a focus dump restored into the deferred queue re-wakes the resident (its wake died with the old process)', () => { + const h = harness(); + h.deferredQueue.push({ participant: 'user', content: [{ type: 'text', text: '[Focus ended — …]' }], metadata: { system: true, kind: 'focus-end', epochId: 'gone' } }); + h.coordinator.resume(); + assert.deepEqual(h.wakes, ['scout']); + assert.equal(h.getFocus(), null); + }); +}); diff --git a/test/focus-framework.test.ts b/test/focus-framework.test.ts new file mode 100644 index 000000000..b07f5113f --- /dev/null +++ b/test/focus-framework.test.ts @@ -0,0 +1,289 @@ +/** + * Focus mode through the framework seams: ingestion on BOTH inbound paths + * (channels/incoming and push/event — DMs live on the latter), the wake + * gate's hold (no debounce leak, pre-queued wakes purged), the resident's + * view exclusion (stored but not compiled), the in-channel autoreply over a + * real MCPL child, and the end-of-focus dump + wake. + */ +import { describe, it, before, after } from 'node:test'; +import assert from 'node:assert/strict'; +import { tmpdir } from 'node:os'; +import { mkdtempSync, rmSync, readFileSync, existsSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +import type { FocusCoordinator } from '../src/focus/coordinator.js'; +import type { EventGate } from '../src/gate/event-gate.js'; +import type { ChannelRegistry } from '../src/mcpl/channel-registry.js'; + +const FIXTURE = join(import.meta.dirname, 'fixtures/tune-out-mcpl-server.mjs'); +const FOCUS = 'disc:guild:noisy'; // registered by the fixture +const OTHER = 'disc:guild:other'; // registered here, same server, opened +const LURK = 'disc:guild:lurk'; // registered here, never opened + +type Stored = { id?: string; participant: string; content: Array<{ type: string; text?: string }>; metadata?: Record }; + +function internals(framework: AgentFramework) { + return framework as unknown as { + focusCoordinator: FocusCoordinator | null; + eventGate: EventGate | null; + channelRegistry: ChannelRegistry | null; + pendingRequests: Array<{ agentName: string; reason: string; source?: string; channelId?: string; timestamp?: number }>; + isUnreadStoredMessage(agent: unknown, messageId: string): boolean; + handleMcplChannelIncoming(event: Record): Promise; + handleMcplPushEvent(event: Record): void; + agents: Map; + } }>; + }; +} + +async function waitFor(cond: () => boolean | Promise, what: string, timeoutMs = 60_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await cond()) return; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error(`timed out waiting for: ${what}`); +} + +const textOf = (m: Stored): string => m.content.filter((b) => b.type === 'text').map((b) => b.text ?? '').join('\n'); + +describe('focus through the framework', () => { + let tempDir: string; + let membrane: MockMembrane; + let framework: AgentFramework; + let statusPath: string; + let n = 0; + + before(async () => { + tempDir = mkdtempSync(join(tmpdir(), 'focus-fw-')); + statusPath = join(tempDir, 'status.jsonl'); + const commandPath = join(tempDir, 'commands.txt'); + writeFileSync(commandPath, ''); + membrane = new MockMembrane(); + framework = await AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'scout', model: 'test-model', systemPrompt: 'You are scout.' }], + focus: { enabled: true, defaultBacklogCap: 2 }, + gate: { + config: { + // Debounce windows far beyond any run of this file: a queued wake + // that SURVIVES the purge (the focus channel's own, asserted below) + // must never fire mid-test — it would wake scout with no mock + // response queued, and the harness then spins silently (a wake + // with no MockMembrane response restart-loops on microtasks and + // starves the TAP reporter — cf. the harness note atop + // channel-incoming-targeting.test.ts). A 60 s window hung a slow + // CI runner exactly that way. + policies: [ + { name: 'batch-other', match: { channel: OTHER }, behavior: { debounce: 3_600_000 } }, + { name: 'batch-push', match: { scope: ['mcpl:push-event'] }, behavior: { debounce: 3_600_000 } }, + ], + default: 'always', + }, + }, + mcplServers: [{ + id: 'disc', + command: process.execPath, + args: [FIXTURE], + env: { STATUS_PATH: statusPath, COMMAND_PATH: commandPath }, + }], + modules: [], + }); + await framework.start(); + await waitFor( + () => (internals(framework).channelRegistry?.listChannelsRaw().length ?? 0) > 0, + 'channel registration', + ); + const registry = internals(framework).channelRegistry!; + registry.ensureChannelRegistered('disc', OTHER, 'other'); + registry.ensureChannelRegistered('disc', LURK, 'lurk'); + // A second server whose channel id STRING equals the focus channel's id. + registry.ensureChannelRegistered('disc2', FOCUS, 'imposter'); + // The resident is open in OTHER (autoreply may speak there) and not in LURK. + const opened = await registry.handleChannelToolCall('channel_open', { channelId: OTHER, serverId: 'disc' }); + assert.equal(opened.success, true, JSON.stringify(opened)); + }); + + after(async () => { + await framework.stop(); + rmSync(tempDir, { recursive: true, force: true }); + }); + + function incoming(channelId: string, text: string, o?: { addressed?: boolean; authorId?: string }): Record { + return { + type: 'mcpl:channel-incoming', + serverId: 'disc', + channelId, + messageId: `m-${++n}`, + author: { id: o?.authorId ?? 'U1', name: 'antra' }, + content: [{ type: 'text', text }], + timestamp: new Date().toISOString(), + metadata: {}, + ...(o?.addressed ? { tags: ['chat:mention', 'chat:addressed'] } : {}), + triggerInference: true, + }; + } + + function publishes(): string[] { + if (!existsSync(statusPath)) return []; + return readFileSync(statusPath, 'utf8').split('\n').filter(Boolean) + .map((l) => JSON.parse(l) as { event: string; channelId?: string }) + .filter((e) => e.event === 'publish') + .map((e) => e.channelId!); + } + + it('exposes the focus tool only when enabled', () => { + assert.ok(framework.getAllTools().some((t) => t.name === 'focus')); + }); + + it('holds every non-focus channel and DM; gate drops them before debounce; view excludes them; autoreplies once', async () => { + const i = internals(framework); + const gate = i.eventGate!; + const coordinator = i.focusCoordinator!; + assert.ok(coordinator, 'coordinator exists when focus is enabled + mcpl configured'); + + // A wake for OTHER queued in the gate BEFORE focus begins… + const pre = gate.evaluate({ content: 'pre', eventType: 'mcpl:channel-incoming', serverId: 'disc', channelId: OTHER, metadata: {} }); + assert.equal(pre.policyName, 'batch-other'); + assert.equal(gate.getStatus().policies.find((p) => p.name === 'batch-other')?.debounceState?.pendingCount, 1); + + // ---- enter --------------------------------------------------------- + // …and a queued mcpl-native PUSH wake for a held channel (channel rides in + // origin.mcplChannelId, no top-level channelId — the shape the purge used + // to miss; slimepriestess's probe 1). + const pushMeta = { serverId: 'disc', featureSet: 'chat', eventId: 'e-pre-1', eventType: 'mcpl:push-event', mcplChannelId: OTHER, tags: ['chat:message'] }; + assert.equal(gate.evaluate({ content: 'queued before focus', eventType: 'mcpl:push-event', serverId: 'disc', channelId: '', metadata: pushMeta, tags: ['chat:message'] }).policyName, 'batch-push'); + // …and a queued push wake for the FOCUS channel itself, which must survive. + const focusPushMeta = { ...pushMeta, eventId: 'e-pre-2', mcplChannelId: FOCUS }; + gate.evaluate({ content: 'focus channel, queued before focus', eventType: 'mcpl:push-event', serverId: 'disc', channelId: '', metadata: focusPushMeta, tags: ['chat:message'] }); + assert.equal(gate.getStatus().policies.find((p) => p.name === 'batch-push')?.debounceState?.pendingCount, 2); + + // …and a framework-level wake already queued for a channel that is about + // to be held (a message that arrived a moment before `enter`). Pushed and + // purged within one tick, so the loop never sees it. + i.pendingRequests.push({ agentName: 'scout', reason: 'mcpl:channel-incoming', source: 'disc', channelId: OTHER, timestamp: Date.now() }); + + const entered = coordinator.handleTool({ mode: 'enter', channelId: FOCUS, serverId: 'disc', durationSeconds: 120 }); + assert.equal(entered.success, true, JSON.stringify(entered)); + assert.ok(!i.pendingRequests.some((r) => r.channelId === OTHER), 'a queued wake for a now-held channel does not run inside focus'); + + // …is purged on enter, and new OTHER events are held at the gate. + assert.equal(gate.getStatus().policies.find((p) => p.name === 'batch-other')?.debounceState?.pendingCount ?? 0, 0); + assert.equal(gate.getStatus().policies.find((p) => p.name === 'batch-push')?.debounceState?.pendingCount ?? 0, 1, + 'the held channel\'s push wake is purged; the focus channel\'s survives'); + assert.equal(gate.evaluate({ content: 'same shape after focus', eventType: 'mcpl:push-event', serverId: 'disc', channelId: '', metadata: { ...pushMeta, eventId: 'e-pre-3' }, tags: ['chat:message'] }).policyName, + 'focus-held', 'live evaluation and purge share one derivation'); + // Same id string on another server is NOT the focus channel (probe 2). + assert.equal(gate.evaluate({ content: 'x', eventType: 'mcpl:channel-incoming', serverId: 'disc2', channelId: FOCUS, metadata: {} }).policyName, 'focus-held'); + const heldDecision = gate.evaluate({ content: 'x', eventType: 'mcpl:channel-incoming', serverId: 'disc', channelId: OTHER, metadata: {} }); + assert.deepEqual([heldDecision.trigger, heldDecision.policyName], [false, 'focus-held']); + const liveDecision = gate.evaluate({ content: 'x', eventType: 'mcpl:channel-incoming', serverId: 'disc', channelId: FOCUS, metadata: {} }); + assert.equal(liveDecision.trigger, true); + // A push event whose origin derives to a non-focus channel is held; one with no channel is not. + const dmDecision = gate.evaluate({ + content: 'dm', eventType: 'mcpl:push-event', serverId: 'disc', channelId: '', + metadata: { source: 'discord', channelId: '999', guildId: null, isDM: true, authorName: 'bob' }, + }); + assert.equal(dmDecision.policyName, 'focus-held'); + const beat = gate.evaluate({ content: 'tick', eventType: 'mcpl:push-event', serverId: 'heartbeat', channelId: '', metadata: { source: 'heartbeat' } }); + assert.notEqual(beat.policyName, 'focus-held', 'a channel-less push is not held (it falls to the ordinary policies)'); + + // ---- ingestion: channels/incoming -------------------------------- + i.pendingRequests.length = 0; + await i.handleMcplChannelIncoming(incoming(OTHER, 'other 1')); + await i.handleMcplChannelIncoming(incoming(OTHER, '@scout other 2', { addressed: true, authorId: 'A' })); + await i.handleMcplChannelIncoming(incoming(OTHER, '@scout other 3', { addressed: true, authorId: 'A' })); + // A mention in a channel the resident never joined: held, counted, no reply. + await i.handleMcplChannelIncoming(incoming(LURK, '@scout from lurk', { addressed: true, authorId: 'C' })); + // Cross-server collision: same id string as the focus channel, other server. + await i.handleMcplChannelIncoming({ ...incoming(FOCUS, 'hello from the imposter channel', { authorId: 'U9' }), serverId: 'disc2' }); + assert.equal(i.pendingRequests.length, 0, 'held traffic wakes nobody'); + + // ---- ingestion: push/event (DM) ---------------------------------- + i.handleMcplPushEvent({ + type: 'mcpl:push-event', serverId: 'disc', featureSet: 'chat', eventId: 'e-dm-1', + origin: { source: 'discord', channelId: '999', guildId: null, isDM: true, authorId: 'B', authorName: 'bob', messageId: 'pm-1' }, + content: [{ type: 'text', text: 'psst, dm' }], + tags: ['chat:dm', 'chat:addressed'], + triggerInference: true, + }); + assert.equal(i.pendingRequests.length, 0, 'held DM wakes nobody'); + + const cm = i.agents.get('scout')!.getContextManager(); + const stored = cm.getAllMessages(); + const heldStored = stored.filter((m) => (m.metadata as { focusHeld?: unknown })?.focusHeld); + assert.equal(heldStored.length, 6, 'all six held messages are stored (incl. lurk + imposter)'); + assert.ok(heldStored.some((m) => textOf(m).includes('imposter channel')), 'the other server\'s same-id channel is held'); + assert.ok(heldStored.every((m) => (m.metadata as { focusHeld: { epochId: string } }).focusHeld.epochId)); + const compiled = await cm.compile(); + assert.ok(!compiled.messages.some((m) => /other \d|psst|lurk|imposter/.test(textOf(m))), 'held messages never compile'); + + // ---- autoreply: once per (channel, author), over the real child -- + await waitFor(() => publishes().length >= 2, 'autoreply publishes'); + await new Promise((r) => setTimeout(r, 200)); + assert.deepEqual(publishes().sort(), [OTHER, 'discord:dm:999'].sort(), + 'one reply per channel+author despite two addressed messages from A; none into the never-joined channel'); + + // ---- the focus channel still wakes -------------------------------- + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'on it' }])); + await i.handleMcplChannelIncoming(incoming(FOCUS, 'live one')); + assert.deepEqual(i.pendingRequests.map((r) => r.agentName), ['scout']); + await waitFor(() => membrane.calls.length >= 1, 'scout turn on the focus channel'); + await waitFor(() => i.agents.get('scout')!.state.status === 'idle', 'scout idle'); + i.pendingRequests.length = 0; + + // The focus-channel turn advanced the watermark past the held messages, + // but nobody has read them: the coalescer must still treat them as + // unread (an edit replaces in place, so the dump shows the final text). + const heldOne = cm.getAllMessages().find((m) => textOf(m) === 'other 1')!; + assert.equal(i.isUnreadStoredMessage(i.agents.get('scout'), heldOne.id!), true, 'held = unread while awaiting the dump'); + + // ---- check --------------------------------------------------------- + const check = coordinator.handleTool({ mode: 'check' }); + const data = check.data as { focused: boolean; held: Array<{ serverId: string; channelId: string; messages: number; addressed: number }> }; + assert.equal(data.focused, true); + assert.deepEqual(data.held.map((h) => [h.serverId, h.channelId, h.messages, h.addressed]).sort(), + [['disc', OTHER, 3, 2], ['disc', LURK, 1, 1], ['disc2', FOCUS, 1, 0], ['disc', 'discord:dm:999', 1, 1]].sort()); + // ---- end ----------------------------------------------------------- + // The end-of-focus wake runs a real scout turn; MockMembrane needs a + // response queued or the harness restart-loops the turn. + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'back' }])); + const ended = coordinator.handleTool({ mode: 'end' }); + // Diagnostic guard: if an assertion below fails, `after()` stops the + // framework mid-turn and the harness microtask-spins, starving the TAP + // reporter — so the failure would be invisible. Log it synchronously. + try { + assert.deepEqual(ended, { success: true, data: { ended: true, held: 6 } }); + assert.deepEqual(i.pendingRequests.map((r) => [r.agentName, r.reason]), [['scout', 'focus ended (ended by you)']]); + // The dump goes through addMessage, which defers under the turn-alive + // guard and flushes at the next turn boundary — the wake queued above. + let dump: Stored | undefined; + await waitFor(() => Boolean(dump = cm.getAllMessages().find((m) => m.metadata?.kind === 'focus-end')), 'dump delivered'); + const text = textOf(dump!); + assert.match(text, /6 messages held across 4 channels; 4 addressed you \(2 automatic replies were sent\)/); + assert.match(text, //); + assert.doesNotMatch(text, /other 1\n/); + assert.match(text, /other 2\n.*other 3\n<\/focus-backlog>/); + assert.match(text, /\n\[\d\d:\d\d\] bob: psst, dm\n<\/focus-backlog>/); + assert.doesNotMatch(text, /Channel invitation/, 'held messages carry no at-arrival invitation'); + assert.match(text, /channel_open \{channelId, backscroll\}/, 'hint names a tool the resident has'); + // The dump compiles; the held originals still do not. + const after = await cm.compile(); + assert.ok(after.messages.some((m) => textOf(m).includes(' /^other 1$|psst, dm$/m.test(textOf(m)) && !textOf(m).includes(' m.metadata?.kind === 'focus-end'); + console.error('[t] post-end assertion failed: ' + (err as Error).message + '\n--- dump ---\n' + (dumpMsg ? textOf(dumpMsg) : '(none)') + '\n--- pending ---\n' + JSON.stringify(i.pendingRequests)); + throw err; + } + await waitFor(() => membrane.calls.length >= 2, 'scout turn after focus ended'); + await waitFor(() => i.agents.get('scout')!.state.status === 'idle', 'scout idle after end'); + assert.equal(i.isUnreadStoredMessage(i.agents.get('scout'), heldOne.id!), false, 'once dumped, the ordinary watermark rule applies'); + }); +}); diff --git a/test/focus-lifecycle.test.ts b/test/focus-lifecycle.test.ts new file mode 100644 index 000000000..9384c8c0f --- /dev/null +++ b/test/focus-lifecycle.test.ts @@ -0,0 +1,122 @@ +/** + * Focus epoch durability — the `focus` record in mcpl/channel-lifecycle, + * replayed last-record-wins so a restart lands in the same epoch (same + * stamps, same deadline) and a corrupt record reads as "not focused". + */ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { JsStore } from '@animalabs/chronicle'; +import { ChannelRegistry, type FocusParams } from '../src/mcpl/channel-registry.js'; +import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; +import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; + +function makeRegistry(store?: JsStore) { + const serverRegistry = { getServer: (_id: string) => null } as unknown as McplServerRegistry; + return new ChannelRegistry(serverRegistry, {} as FeatureSetManager, () => {}, () => {}, store ? { store } : undefined); +} + +const PARAMS: FocusParams = { + epochId: 'epoch-1', + serverId: 'discord', + channelId: 'discord:g:work', + startedAtMs: 1_000, + startedAtSequence: 42, + expiresAtMs: 2_000_000_000_000, + backlogCap: 20, +}; + +test('set/get/end round-trip in the projection', () => { + const registry = makeRegistry(); + assert.equal(registry.getFocus(), null); + registry.setFocus(PARAMS, 'agent-tool'); + assert.deepEqual(registry.getFocus(), PARAMS); + registry.setFocus(null, 'agent-tool'); + assert.equal(registry.getFocus(), null); +}); + +test('the epoch replays across restart, including re-target release points; end replays as null', () => { + const dir = mkdtempSync(join(tmpdir(), 'focus-lifecycle-')); + try { + const store = JsStore.openOrCreate({ path: join(dir, 'store') }); + const first = makeRegistry(store); + first.setFocus(PARAMS, 'agent-tool'); + const retargeted = { ...PARAMS, channelId: 'discord:g:chat', released: { 'discord:g:chat': 7 } }; + first.setFocus(retargeted, 'agent-tool'); + + const second = makeRegistry(store); + assert.deepEqual(second.getFocus(), retargeted); + + second.setFocus(null, 'duration'); + const third = makeRegistry(store); + assert.equal(third.getFocus(), null); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a malformed focus record replays as not focused rather than failing boot', () => { + const dir = mkdtempSync(join(tmpdir(), 'focus-lifecycle-bad-')); + try { + const store = JsStore.openOrCreate({ path: join(dir, 'store') }); + const first = makeRegistry(store); + first.setFocus(PARAMS, 'agent-tool'); + store.appendToStateJson('mcpl/channel-lifecycle', { + kind: 'focus', serverId: 'discord', timestamp: 'now', focus: { epochId: 'x' }, + }); + const second = makeRegistry(store); + assert.equal(second.getFocus(), null); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('focus records do not disturb per-channel desired state', () => { + const registry = makeRegistry(); + registry.setFocus(PARAMS, 'agent-tool'); + assert.equal(registry.getDesiredState('discord', 'discord:g:work'), undefined); +}); + +test('replay tolerates a record missing optional fields (backlogCap → 20) but needs serverId', () => { + const dir = mkdtempSync(join(tmpdir(), 'focus-lifecycle-partial-')); + try { + const store = JsStore.openOrCreate({ path: join(dir, 'store') }); + makeRegistry(store); + store.appendToStateJson('mcpl/channel-lifecycle', { + kind: 'focus', serverId: 'discord', timestamp: 'now', + focus: { epochId: 'p', serverId: 'discord', channelId: 'discord:g:work', expiresAtMs: 2_000_000_000_000 }, + }); + const partial = makeRegistry(store); + assert.equal(partial.getFocus()?.backlogCap, 20); + assert.equal(partial.getFocus()?.startedAtMs, 2_000_000_000_000); + + store.appendToStateJson('mcpl/channel-lifecycle', { + kind: 'focus', serverId: 'discord', timestamp: 'now', + focus: { epochId: 'q', channelId: 'discord:g:work', expiresAtMs: 2_000_000_000_000, backlogCap: 5 }, + }); + assert.equal(makeRegistry(store).getFocus(), null, 'no serverId → identity unknown → not focused'); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('resolveChannel is server-qualified and refuses ambiguity; canAutoReplyInto admits open channels and DMs only', async () => { + const registry = makeRegistry(); + // handleChanged, not ensureChannelRegistered: the latter is id-only and + // treats a second server's same-id channel as already registered. + await registry.handleChanged('a', { added: [ + { id: 'shared', type: 'x', label: 'a-shared', direction: 'bidirectional' }, + { id: 'dm-1', type: 'x', label: 'DM: x', direction: 'bidirectional', metadata: { channelType: 'dm' } }, + ] } as never); + await registry.handleChanged('b', { added: [ + { id: 'shared', type: 'x', label: 'b-shared', direction: 'bidirectional' }, + ] } as never); + assert.match(registry.resolveChannel('shared').error ?? '', /ambiguous/); + assert.deepEqual(registry.resolveChannel('shared', 'b').entry, { serverId: 'b', channelId: 'shared', label: 'b-shared' }); + assert.match(registry.resolveChannel('nope').error ?? '', /not found/); + assert.equal(registry.canAutoReplyInto('a', 'shared'), false, 'registered but not open'); + assert.equal(registry.canAutoReplyInto('a', 'dm-1'), true, 'DM'); + assert.equal(registry.canAutoReplyInto('zzz', 'shared'), false, 'unknown server'); +}); diff --git a/test/focus-routing.test.ts b/test/focus-routing.test.ts new file mode 100644 index 000000000..5d605eb84 --- /dev/null +++ b/test/focus-routing.test.ts @@ -0,0 +1,168 @@ +/** + * Focus mode and outbound routing, end to end over a real MCPL child. + * + * The 2026-09-30 misroute: a batched (channel-less) wake for a DM fell back to + * the process-global default locus, which an ambient message in another + * channel had retargeted one second earlier — the DM reply was published + * into that channel. Focus must close this: while focused, held inbound does + * not move the default, the focus channel is the locus for every turn, and + * entering focus mid-turn moves the current turn's prose there. + */ +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { tmpdir } from 'node:os'; +import { mkdtempSync, rmSync, readFileSync, existsSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import type { ContentBlock } from '@animalabs/membrane'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +import type { FocusCoordinator } from '../src/focus/coordinator.js'; +import type { ChannelRegistry } from '../src/mcpl/channel-registry.js'; + +const FIXTURE = join(import.meta.dirname, 'fixtures/tune-out-mcpl-server.mjs'); +const FOCUS = 'disc:guild:noisy'; // registered by the fixture +const OTHER = 'disc:guild:other'; // registered here, same server + +type Stored = { participant: string; content: Array<{ type: string; text?: string }>; metadata?: Record }; + +function internals(framework: AgentFramework) { + return framework as unknown as { + focusCoordinator: FocusCoordinator | null; + channelRegistry: ChannelRegistry | null; + pendingRequests: Array<{ agentName: string; reason: string; source: string; timestamp: number }>; + agents: Map; + }; +} + +async function waitFor(cond: () => boolean | Promise, what: string, timeoutMs = 15_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await cond()) return; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error(`timed out waiting for: ${what}`); +} + +describe('focus routing through the framework', () => { + let tempDir: string; + let statusPath: string; + let membrane: MockMembrane; + let framework: AgentFramework; + let n = 0; + + beforeEach(async () => { + tempDir = mkdtempSync(join(tmpdir(), 'focus-route-')); + statusPath = join(tempDir, 'status.jsonl'); + const commandPath = join(tempDir, 'commands.txt'); + writeFileSync(commandPath, ''); + membrane = new MockMembrane(); + framework = await AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'scout', model: 'test-model', systemPrompt: 'You are scout.' }], + focus: { enabled: true }, + gate: { config: { policies: [], default: 'always' } }, + mcplServers: [{ + id: 'disc', + command: process.execPath, + args: [FIXTURE], + env: { STATUS_PATH: statusPath, COMMAND_PATH: commandPath }, + }], + modules: [], + }); + await framework.start(); + await waitFor( + () => (internals(framework).channelRegistry?.listChannelsRaw().length ?? 0) > 0, + 'channel registration', + ); + internals(framework).channelRegistry!.ensureChannelRegistered('disc', OTHER, 'other'); + }); + + afterEach(async () => { + await framework.stop(); + rmSync(tempDir, { recursive: true, force: true }); + }); + + /** channels/incoming through the REAL registry path (it owns the default locus). */ + function inbound(channelId: string, text: string): void { + internals(framework).channelRegistry!.handleIncoming('disc', { + messages: [{ + channelId, + messageId: `m-${++n}`, + author: { id: 'U1', name: 'antra' }, + timestamp: new Date().toISOString(), + content: [{ type: 'text', text }], + tags: ['chat:ambient', 'chat:from-human'], + }], + }); + } + + function publishes(): string[] { + if (!existsSync(statusPath)) return []; + return readFileSync(statusPath, 'utf8').split('\n').filter(Boolean) + .map((l) => JSON.parse(l) as { event: string; channelId?: string }) + .filter((e) => e.event === 'publish') + .map((e) => e.channelId!); + } + + async function idle(): Promise { + await waitFor(() => internals(framework).agents.get('scout')!.state.status === 'idle', 'scout idle'); + } + + it('held inbound cannot steer a channel-less wake; the focus channel is the locus, during and after', async () => { + const i = internals(framework); + + // Establish the focus channel as the last inbound (a normal wake + reply). + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello focus' }] as ContentBlock[])); + inbound(FOCUS, 'hi'); + await waitFor(() => publishes().length >= 1, 'first reply'); + await idle(); + assert.deepEqual(publishes(), [FOCUS]); + + assert.equal(i.focusCoordinator!.handleTool({ mode: 'enter', channelId: FOCUS, durationSeconds: 600 }).success, true); + + // An ambient message elsewhere arrives (held — stored, no wake). + inbound(OTHER, 'chatter in another room'); + await waitFor( + () => i.agents.get('scout')!.getContextManager().getAllMessages() + .some((m) => (m.metadata as { focusHeld?: unknown } | undefined)?.focusHeld), + 'held message stored', + ); + assert.equal(i.channelRegistry!.getDefaultPublishChannel(), FOCUS, 'held inbound did not retarget the default'); + + // A batched gate wake carries no channel (the incident's wake shape). + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'reply to the focus room' }] as ContentBlock[])); + i.pendingRequests.push({ agentName: 'scout', reason: 'gate:debounce', source: 'gate', timestamp: Date.now() }); + await waitFor(() => publishes().length >= 2, 'channel-less wake reply'); + await idle(); + assert.deepEqual(publishes(), [FOCUS, FOCUS], 'the channel-less wake replied into the focus channel'); + + // Ending focus wakes the resident with no channel; the held message never + // became the fallback, so the reply still lands where they were. + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'back' }] as ContentBlock[])); + assert.equal(i.focusCoordinator!.handleTool({ mode: 'end' }).success, true); + await waitFor(() => publishes().length >= 3, 'post-focus reply'); + await idle(); + assert.deepEqual(publishes(), [FOCUS, FOCUS, FOCUS]); + assert.equal(i.channelRegistry!.getFocusLocus(), null, 'focus locus cleared at end'); + }); + + it('entering focus mid-turn moves that turn\'s prose to the focus channel and says so in the tool result', async () => { + // Turn triggered from OTHER (not focused yet); the resident enters focus + // on FOCUS and keeps talking — that prose belongs in FOCUS. + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'f1', name: 'focus', input: { mode: 'enter', channelId: FOCUS, durationSeconds: 600 } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'now only here' }] as ContentBlock[])); + inbound(OTHER, 'come focus with me'); + await waitFor(() => publishes().length >= 1, 'post-enter prose'); + await idle(); + + assert.deepEqual(publishes(), [FOCUS], 'prose after enter landed in the focus channel, not the trigger channel'); + const results = membrane.lastStream!.receivedToolResults.flat(); + assert.ok( + JSON.stringify(results).includes('plain speech lands in the focus channel'), + 'routing note rides the focus tool result', + ); + }); +}); diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index c5c3157bc..71c527a9c 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -391,6 +391,67 @@ describe('present while acting', () => { await framework.stop(); }); + it('under focus, an addressed mid-turn injection from another channel cannot take the pin', async () => { + // Focus: the resident chose the channel their words go to. An addressed + // message from elsewhere (normally a re-pin) must not overrule that. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'In the focus room.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Still in the focus room.' }, + ] as ContentBlock[])); + + const framework = await createFramework(); + const routed = stubChannelRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + (registry as { getFocusLocus?: unknown }).getFocusLocus = () => ({ serverId: 'disc', channelId: 'chan-live-1' }); + module.toolDelayMs = 25; + module.interjection = 'over here!'; + module.interjectionMetadata = { channelId: 'discord:guild:general', tags: ['chat:addressed'] }; + + trigger(framework); + await framework.runUntilIdle(); + + assert.deepEqual(routed, [ + { text: 'In the focus room.', locus: 'chan-live-1' }, + { text: 'Still in the focus room.', locus: 'chan-live-1' }, + ]); + await framework.stop(); + }); + + it('a focus-held injection never re-pins, even when focus is no longer active', async () => { + // A held message is stored but excluded from every view; it cannot be + // the reason prose moves (focus may have ended mid-turn, so the + // focus-channel restriction alone doesn't cover it). + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Round one.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Round two.' }, + ] as ContentBlock[])); + + const framework = await createFramework(); + const routed = stubChannelRegistry(framework); + module.toolDelayMs = 25; + module.interjection = 'held while focused'; + module.interjectionMetadata = { + channelId: 'discord:dm:someone', + tags: ['chat:addressed'], + focusHeld: { epochId: 'e1' }, + }; + + trigger(framework); + await framework.runUntilIdle(); + + assert.deepEqual(routed, [ + { text: 'Round one.', locus: 'chan-live-1' }, + { text: 'Round two.', locus: 'chan-live-1' }, + ]); + await framework.stop(); + }); + it('a follow-up in a channel the agent explicitly engaged this turn re-pins', async () => { // 2026-07-31 n=7 (q's #portables reply): the agent explicitly sent into // a channel this turn; someone replies there WITHOUT a mention (ambient