diff --git a/src/agent.ts b/src/agent.ts index d16cca2..d371992 100644 --- a/src/agent.ts +++ b/src/agent.ts @@ -673,7 +673,12 @@ export class Agent { const stream = this.membrane.streamYielding(request, { emitTokens: true, - emitBlocks: false, + // Block boundaries MUST be emitted: driveStream translates them into + // inference:content_block traces, and voice clients key their + // utterance state machines on the resulting block_start / + // block_complete wire messages (melodeus can neither track nor + // interrupt an utterance without them). + emitBlocks: true, emitUsage: true, }); diff --git a/src/framework.ts b/src/framework.ts index 3e478c8..2bde499 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -64,7 +64,7 @@ import { ConversationRouter } from './mcpl/conversation-router.js'; import { safeSlice } from './safe-slice.js'; import type { WorkspaceModule } from './modules/workspace/index.js'; import { toolResultDataToHistoryString } from './tool-result-history.js'; -import { splitProseSegments } from './prose-segments.js'; +import { splitProseSegments, undeliveredSuffix } from './prose-segments.js'; /** Detect a supported image media type from magic bytes (the model API * rejects mislabeled media types, so trust bytes over extensions). @@ -486,6 +486,20 @@ export class AgentFramework { * separate from ephemeralRuns deliberately: endTurn/budget cancels happen * for resident agents too, and the key is per-stream, not per-agent. */ private frameworkCancelledStreams: Map = new Map(); + /** Prose already committed to context by this turn's round flushes + * (pendingAssistantBlocks → addAssistantResponse), '\n'-joined segments. + * A keepText abort commits only keepText's suffix past this, so a voice + * client that reports the whole activation's speech cannot duplicate + * earlier rounds' prose in the chronicle. Cleared per stream in + * driveStream's finally. */ + private turnCommittedProse: Map = new Map(); + /** Tracks the portion of an aborted message that was spoken/delivered + * (supplied via abortInference({keepText})) and should be persisted + * as a partial assistant turn. Keyed `${agentName}:${streamId}` so an + * abort that races turn completion (stored, but no 'aborted' stream + * event ever consumes it) goes inert instead of attaching an old turn's + * spoken text to a later abort of the same agent. */ + private abortKeepTexts: Map = new Map(); /** Active runEphemeralToCompletion runs, keyed by agent name. */ private ephemeralRuns: Map = new Map(); /** Per-agent count of consecutive exhausted inferences (reset on any success). @@ -1188,15 +1202,41 @@ export class AgentFramework { /** * Abort an in-flight inference for an agent. + * + * Any partial text that was already delivered before the interrupt and + * should be persisted may be passed via `keepText`. The abort handler + * will persist it as the assistant's turn and route it to the turn's + * locus instead of discarding everything. */ - abortInference(agentName: string, reason?: string): boolean { + abortInference( + agentName: string, + reasonOrOpts?: string | { reason?: string; keepText?: string }, + ): boolean { + const opts = typeof reasonOrOpts === 'string' ? { reason: reasonOrOpts } : reasonOrOpts ?? {}; const agent = this.agents.get(agentName); if (!agent) { return false; } - const result = agent.abortInference(reason); + const result = agent.abortInference(opts.reason); + // keepText is stored only for a SUCCESSFUL abort, after the fact — safe + // because nothing can consume the stream's 'aborted' event within this + // synchronous frame (and cancelStream does not bump streamId). Storing + // before the call and deleting on failure looked equivalent, but a + // second abort of an already-idle agent computes the SAME key (streamId + // unchanged) and its failure cleanup deleted the first abort's + // still-pending keepText — a duplicate interruption report would + // silently discard the spoken words. + if (result && opts.keepText) { + this.abortKeepTexts.set(`${agentName}:${agent.streamId}`, opts.keepText); + } if (result) { - this.emitTrace({ type: 'inference:aborted', agentName, reason, durationMs: result.durationMs }); + this.emitTrace({ + type: 'inference:aborted', + agentName, + reason: opts.reason, + durationMs: result.durationMs, + channelId: this.inferenceTraceChannel(agentName), + }); } return !!result; } @@ -3261,6 +3301,18 @@ export class AgentFramework { if (pendingBlocks) { agent.addAssistantResponse(pendingBlocks); this.pendingAssistantBlocks.delete(agent.name); + // Track this turn's context-committed prose so a keepText abort + // later in the turn does not commit the same words twice: a + // voice client that accumulates spokenText across the whole + // activation replays these rounds' prose inside keepText. + const flushedSegments = splitProseSegments(pendingBlocks); + if (flushedSegments.length > 0) { + const prev = this.turnCommittedProse.get(agent.name); + this.turnCommittedProse.set( + agent.name, + prev ? `${prev}\n${flushedSegments.join('\n')}` : flushedSegments.join('\n'), + ); + } } // Compute truncation limit from agent's strategy (maxMessageTokens * 4 chars) @@ -3362,7 +3414,11 @@ export class AgentFramework { // driveStream's finally is the backstop.) this.eventGate?.onInferenceEnded(agent.name); this.settleAgent(agent.name, { stopReason: 'turn_ended', speech: '' }); - this.emitTrace({ type: 'inference:turn_ended', agentName: agent.name }); + this.emitTrace({ + type: 'inference:turn_ended', + agentName: agent.name, + channelId: this.inferenceTraceChannel(agent.name), + }); } else if (overBudget) { // Context budget exceeded: break the stream, let compile() compress. // Mark the cancel as framework-initiated BEFORE cancelling: the @@ -3377,6 +3433,10 @@ export class AgentFramework { agent.cancelStream(); this.emitTrace({ type: 'inference:stream_restarted', + // The turn's locus pin survives the restart, so channel-scoped + // consumers (voice) can close the abandoned activation before + // the replacement stream opens a fresh one. + channelId: this.inferenceTraceChannel(agent.name), agentName: agent.name, reason: 'context_budget', inputTokens: agent.lastStreamInputTokens, @@ -4252,7 +4312,11 @@ export class AgentFramework { } this.touchEphemeralRun(agent.name, true); - this.emitTrace({ type: 'inference:started', agentName: agent.name }); + this.emitTrace({ + type: 'inference:started', + agentName: agent.name, + channelId: this.inferenceTraceChannel(agent.name), + }); this.eventGate?.onInferenceStarted(agent.name); this.lastInferenceAt.set(agent.name, { ...this.lastInferenceAt.get(agent.name), startedAt: Date.now() }); @@ -4316,6 +4380,7 @@ export class AgentFramework { agentName: agent.name, error: err.message, stack: err.stack, + channelId: this.inferenceTraceChannel(agent.name), }); agent.reset(); @@ -4405,6 +4470,18 @@ export class AgentFramework { // delivered exactly once, at turn end. let liveProseRouting = false; + // Prose the live path has enqueued for posting to the locus during this + // driveStream, in emission order (enqueued, not confirmed: a failed post + // is swallowed by its chain link and never retried, same as at + // completion). The abort path compares an interruption's keepText + // against this so it never re-posts text a speak-while-acting round + // already delivered. A budget-restart continuation starts a fresh + // driveStream with this empty — which matches the voice side: a restart + // emits a fresh inference:started, the client's spoken-text accumulator + // resets with the new activation, so a post-restart keepText never spans + // pre-restart prose. + let liveRoutedProse = ''; + // Typing indicator: show " is typing…" in the channel her plain // prose will actually land in — the turn-frozen locus — for the whole // duration of this turn. Started here (paired with the finally below, so @@ -4435,6 +4512,7 @@ export class AgentFramework { content: event.content, blockType: event.meta.type, blockIndex: event.meta.blockIndex, + channelId: this.inferenceTraceChannel(agent.name), }); break; @@ -4446,6 +4524,7 @@ export class AgentFramework { phase, blockType: block.type, blockIndex: index, + channelId: this.inferenceTraceChannel(agent.name), }); break; } @@ -4458,6 +4537,7 @@ export class AgentFramework { type: 'inference:tool_calls_yielded', agentName: agent.name, calls: event.calls.map((c) => ({ id: c.id, name: c.name, input: c.input })), + channelId: this.inferenceTraceChannel(agent.name), }); // Build assistant content blocks for this round. Prefer the @@ -4534,6 +4614,7 @@ export class AgentFramework { ); for (const seg of roundSegments) { enqueueSpeech(seg, locus); + liveRoutedProse += (liveRoutedProse ? '\n' : '') + seg; } } } @@ -4688,6 +4769,7 @@ export class AgentFramework { agentName: agent.name, durationMs, tokenUsage, + channelId: this.inferenceTraceChannel(agent.name), }); if (du) { @@ -4931,6 +5013,7 @@ export class AgentFramework { agentName: agent.name, error: err.message, stack: err.stack, + channelId: this.inferenceTraceChannel(agent.name), }); this.logInference({ @@ -4985,6 +5068,9 @@ export class AgentFramework { // reject an ephemeral's promise mid-run and bump the failure // streak). Gate release + stream teardown happen in `finally`. if (this.frameworkCancelledStreams.delete(`${agent.name}:${myStreamId}`)) { + // A keepText racing a framework cancel on the same stream must + // not survive to a later abort of this agent. + this.abortKeepTexts.delete(`${agent.name}:${myStreamId}`); this.eventGate?.onInferenceEnded(agent.name); return; } @@ -4993,16 +5079,99 @@ export class AgentFramework { // may have already started a new stream, bumping streamId) if (agent.streamId === myStreamId) { const durationMs = Date.now() - startTime; + + // If an external source interrupts generation (e.g. a voice + // client reporting a user talking over the agent), it passes + // the portion already delivered — the words spoken aloud — as + // keepText. The full keepText is persisted as the assistant's + // turn; the channel post routes only what the live path has not + // already posted this turn (speak-while-acting rounds were + // delivered as they happened — see undeliveredSuffix for how + // spoken text is matched against them). turnSilenced is honored + // as at completion. Without keepText the whole partial turn is + // discarded. Framework-internal cancels never reach this path + // (early return above). + const keepText = this.abortKeepTexts.get(`${agent.name}:${myStreamId}`); + this.abortKeepTexts.delete(`${agent.name}:${myStreamId}`); + if (keepText) { + // Commit only the part of keepText this turn's earlier round + // flushes have not already committed (whole-activation- + // accumulating clients replay those rounds' prose inside + // keepText; recommitting would duplicate it in the + // chronicle). Divergence means keepText is the current + // round's own fragment — committed in full. + const committedProse = this.turnCommittedProse.get(agent.name) ?? ''; + const contextKeep = committedProse + ? undeliveredSuffix(keepText, committedProse) + : keepText; + if (contextKeep) { + agent.addAssistantResponse([{ type: 'text', text: contextKeep }]); + } + if (this.channelRegistry && !turnSilenced) { + const undelivered = liveRoutedProse + ? undeliveredSuffix(keepText, liveRoutedProse) + : keepText; + if (undelivered) { + // Snapshot the locus BEFORE waiting: a successor turn + // starting during the drain re-pins turnLocusPins, and + // this tail belongs to the interrupted turn, not to + // wherever the agent speaks next. + const abortLocus = resolveTurnLocus(); + // Preserve in-channel ordering: live-routed segments were + // enqueued without await, so let the chain drain first + // (mirrors the 'complete' case). + await turnSpeechChain; + try { + await this.channelRegistry.routeSpeech( + agent.name, + undelivered, + abortLocus, + ); + } catch (err) { + console.error('keepText speech routing failed:', err); + } + } else if (liveRoutedProse) { + console.error( + `[routing] ${agent.name}: abort keepText fully covered by live-routed prose -> nothing further posted`, + ); + } + } + } + + // The awaits above (speech chain drain + routing) opened a + // window in which the scheduler can start this agent's next + // stream — cancelStream already set the agent idle. If that + // happened, the reset/settle below belong to the NEW stream's + // driveStream now; running them here would clobber its state + // and settle its requests against the wrong turn. The keepText + // context commit above already happened and stands. + if (agent.streamId !== myStreamId) { + this.logInference({ + timestamp: startTime, + agentName: agent.name, + requestId, + success: false, + error: `Stream aborted: ${reason}`, + request: compiledRequest ?? { note: 'streaming request aborted' }, + durationMs, + }); + this.eventGate?.onInferenceEnded(agent.name); + break; + } + agent.reset(); this.settleAgent(agent.name, { stopReason: 'exhausted', - speech: '', + speech: keepText ?? '', error: `Stream aborted: ${reason}`, }); this.emitTrace({ type: 'inference:exhausted', agentName: agent.name, error: `Stream aborted: ${reason}`, + // Deliberate cancel: routes around the inference-health + // machinery in emitTrace (see the funnel there). + errorType: 'abort', }); // Postmortem 2026-05-28 P2 #7: persist the abort to the // inference log so future investigations can attribute the @@ -5073,6 +5242,7 @@ export class AgentFramework { agentName: agent.name, error: err.message, stack: err.stack, + channelId: this.inferenceTraceChannel(agent.name), }); this.settleAgent(agent.name, { stopReason: 'exhausted', @@ -5114,8 +5284,21 @@ export class AgentFramework { // exhausted, abort) so it never sticks after the turn ends. this.channelRegistry?.stopTyping(); this.frameworkCancelledStreams.delete(`${agent.name}:${myStreamId}`); - this.activeStreams.delete(agent.name); - this.pendingAssistantBlocks.delete(agent.name); + // Backstop: a keepText stored for this stream but never consumed (its + // abort raced completion, so no 'aborted' event fired) must not outlive + // the stream. + this.abortKeepTexts.delete(`${agent.name}:${myStreamId}`); + // The name-keyed per-turn maps belong to whichever stream is CURRENT. + // If a successor stream started while this one's teardown awaited + // (abort keepText drain, error-path retries), these entries are the + // successor's — deleting them here would strand its tool round + // (tool_result without tool_use) and its committed-prose bookkeeping. + // The successor's own finally cleans them up. + if (agent.streamId === myStreamId) { + this.activeStreams.delete(agent.name); + this.pendingAssistantBlocks.delete(agent.name); + this.turnCommittedProse.delete(agent.name); + } // A conversation fork whose TTL closure turn just finished is done for // good — dispose it so the agent map doesn't grow monotonically. @@ -5586,6 +5769,23 @@ export class AgentFramework { }; } + /** + * Exposes the speech locus of a turn for channel-scoped voice / streaming. + * This locus is set eagerly and pinned in startAgentStream before the turn's + * first trace, so the value here is stable for the whole turn. With a + * channelRegistry the pin is the ONLY source read: the freeze already + * incorporated the triggering channel, and a mid-turn channel_open mutates + * the trigger map without moving actual routing — falling back to it would + * stamp a channel the turn's speech never lands in. Without a + * channelRegistry there is no pin, so the triggering channel is the only + * identity available. Undefined for heartbeats/timers without a locus. + * Read-only (traces are observability-only and must never touch routing + * state). */ + private inferenceTraceChannel(agentName: string): string | undefined { + if (this.channelRegistry) return this.turnLocusPins.get(agentName); + return this.activeTriggerChannels.get(agentName); + } + private emitTrace(event: { type: TraceEvent['type']; [key: string]: unknown }): void { // Centralized inference-health observability. Every terminal failure path // funnels through an `inference:exhausted` trace and every successful model @@ -5594,7 +5794,18 @@ export class AgentFramework { // produced them. In headless/daemon mode no trace client is attached, so // without this the only durable record of a failed inference is a field in // llm-calls.jsonl — invisible to operator, agent, and monitoring. - if (event.type === 'inference:exhausted') { + // `errorType: 'abort'` marks a DELIBERATE cancel (user abort, voice + // interruption) — not a model failure. Those must bypass the + // failure machinery: no consecutive-failure streak (three voice + // barge-ins in a row must not page an operator as hard-down), no + // failures.log entry, and no "[inference-failed] ... nothing was sent" + // chronicle marker — which would be false when the abort persisted + // keepText as the assistant's turn. Scope note: classifyInferenceError + // can also produce 'abort' (membrane classifies AbortError-shaped + // failures that way), so a thrown abort-classified error skips the + // booking too — intended, since those are cancels however they + // surfaced; the inference log still records them. + if (event.type === 'inference:exhausted' && event.errorType !== 'abort') { this.noteInferenceExhausted( (event.agentName as string) ?? 'unknown', (event.error as string) ?? 'unknown error', diff --git a/src/modules/index.ts b/src/modules/index.ts index 53ef3b2..e504158 100644 --- a/src/modules/index.ts +++ b/src/modules/index.ts @@ -8,6 +8,9 @@ export type { ApiEvent } from './api/index.js'; export { HealthModule } from './health/index.js'; export type { HealthModuleConfig } from './health/index.js'; +export { RelayClientModule } from './voice-relay/index.js'; +export type { RelayClientModuleConfig, RelayLogger } from './voice-relay/index.js'; + export { WorkspaceModule } from './workspace/index.js'; export type { WorkspaceConfig, diff --git a/src/modules/voice-relay/index.ts b/src/modules/voice-relay/index.ts new file mode 100644 index 0000000..5392a12 --- /dev/null +++ b/src/modules/voice-relay/index.ts @@ -0,0 +1,20 @@ +/** + * Voice relay integration: connects framework agents to an external TTS + * relay (melodeus-tts-relay) so their turns stream to voice clients as they + * are written and can be interrupted mid-utterance — the same connection the + * relay's existing bots (ChapterX) hold. + */ + +export { RelayClientModule } from './relay-client-module.js'; +export type { RelayClientModuleConfig } from './relay-client-module.js'; + +export { InferenceTraceBridge } from './trace-bridge.js'; +export type { TraceBridge, ChannelBroadcastFn, AgentIdentityResolver } from './trace-bridge.js'; + +export type { + RelayLogger, + BotStreamMessage, + RelayToVoiceClientMessage, + BlockType, + ActivationEndReason, +} from './types.js'; diff --git a/src/modules/voice-relay/relay-client-module.ts b/src/modules/voice-relay/relay-client-module.ts new file mode 100644 index 0000000..4b53bfd --- /dev/null +++ b/src/modules/voice-relay/relay-client-module.ts @@ -0,0 +1,669 @@ +/** + * RelayClientModule — the outbound half of voice integration. + * + * Connects framework agents TO an external melodeus TTS relay, holding the + * same kind of connection the relay's existing bots do (the "ChapterX" + * bots the relay was originally built to serve): dials `{url}/bot` as a + * WebSocket client, authenticates as one bot identity, and streams the + * agent's turn up the socket as relay messages (activation_start / + * block_start / chunk / block_complete / activation_end, translated from + * inference:* traces by InferenceTraceBridge). Interruptions arriving from + * the relay are mapped to `framework.abortInference(agentName, { reason, + * keepText })`, so an interrupted agent's context and posted message keep + * the words the voice client reported spoken (reports are as precise as the + * client's own audio accounting — some clients count text dispatched to + * TTS, not audio actually played). + * + * Usage: + * const relay = new RelayClientModule({ url, botId, token }); + * relay.bind(framework); // REQUIRED for interruptions: without it, + * // outbound streaming still works but relay + * // interruptions are dropped with a warning + * await framework.addModule(relay); + * + * One module instance = one bot identity on the relay (matching the relay's + * one-bot-per-connection model). By default an instance streams EVERY + * channel-bearing agent in the process under its one identity; a host + * voicing several agents under distinct relay identities registers several + * instances, each scoped to its own agents with `agents`. If a second + * connection authenticates with the same botId, the relay closes this one + * with "Replaced by new connection" — treated as fatal (no reconnect), + * because two instances sharing an identity would otherwise evict each + * other forever. + * + * Interruption addressing: an interruption names a channel; it aborts the + * agent last seen streaming there — unless that agent has meanwhile moved + * on to a turn somewhere else (another channel, or a channel-less run), in + * which case the report describes the earlier, finished turn and is + * dropped. A channel we never streamed to is dropped. An interruption with + * no channel at all is accepted only when exactly one channel is tracked + * (nothing else it could mean). Before aborting, the reported spokenText + * must prefix-match the current utterance's streamed text — voice lags + * text, so a report that does not match describes an EARLIER utterance and + * must not cut off the new turn. Voice clients differ in what they report + * (melodeus resets its spoken-text accumulator at every block_start; the + * iOS client accumulates the whole activation), so a match against either + * window is accepted, and characters clients normalize away (whitespace, + * narrator `*` markup) are ignored. A non-empty report while the current + * utterance has voiced nothing yet is stale by the same logic. Reports + * without spokenText are allowed through: absence of evidence is not + * staleness. Deployment note: iOS prefixes " says:" to speech from + * bots with no configured voice — that prefix fails verification, so give + * this bot a voice entry in the relay config (or disable announcements) + * for interruptions to land. + * + * Delivery semantics mirror the relay's: no queueing — messages produced + * while the socket is down are dropped (the relay drops for disconnected + * consumers too). Reconnects use exponential backoff and re-authenticate; + * the backoff resets only after a connection has stayed authenticated for a + * stability window, so an auth-then-drop loop cannot hammer the relay. The + * relay heartbeats every connection every ~2s; when nothing at all arrives + * for heartbeatTimeoutMs the link is presumed half-open and torn down so + * the reconnect path can recover it. + */ + +import WebSocket from 'ws'; + +import type { AgentFramework } from '../../framework.js'; +import type { TraceEvent } from '../../types/trace.js'; +import type { ModuleContext, Module, ProcessState, EventResponse } from '../../types/module.js'; +import type { ProcessEvent, ToolCall, ToolResult, ToolDefinition } from '../../types/events.js'; +import type { BotStreamMessage, RelayLogger } from './types.js'; +import { InferenceTraceBridge, type TraceBridge } from './trace-bridge.js'; +import { isWhitespaceInsensitivePrefix } from '../../prose-segments.js'; + +export interface RelayClientModuleConfig { + /** Relay base URL, e.g. "ws://localhost:8800" — "/bot" is appended. */ + url: string; + /** Bot identity on the relay (must be authorized by the relay's BOT_TOKENS). */ + botId: string; + token: string; + /** Display identity stamped on outgoing relay messages (defaults to botId). + * Supply a real Discord user id here if Discord-side features (mention + * resolution) should work; a framework agent has none of its own. */ + userId?: string; + username?: string; + /** Only stream (and address interruptions for) these agents. Default: all + * agents in the process. Set this when several instances with distinct + * bot identities run in one process, each voicing its own agents. */ + agents?: string[]; + /** First reconnect delay; doubles per attempt. Default 1000ms. */ + reconnectInitialMs?: number; + /** Backoff cap. Default 30000ms. */ + reconnectMaxMs?: number; + /** How long a connection must stay authenticated before the backoff + * resets to its initial value. Default 10000ms. */ + backoffResetAfterMs?: number; + /** Tear the connection down (and reconnect) when NOTHING has arrived for + * this long — the relay heartbeats every ~2s, so silence means a + * half-open link that would otherwise swallow turns until the OS + * timeout. Default 8000ms; 0 disables. */ + heartbeatTimeoutMs?: number; + logger?: RelayLogger; +} + +/** Default logger: console-backed for info and above (RelayLogger's + * documented default), so the permanent failure modes — auth rejection, + * the replaced-connection stop — are visible without configuration. + * debug is dropped (per-message noise); inject a logger to capture it. */ +const consoleLogger: RelayLogger = { + debug() {}, + info(msg, data) { + console.log(`[relay-client] ${msg}`, data ?? ''); + }, + warn(msg, data) { + console.warn(`[relay-client] ${msg}`, data ?? ''); + }, + error(msg, data) { + console.error(`[relay-client] ${msg}`, data ?? ''); + }, +}; + +/** Upper bound on remembered channels (interruption addressing + spoken-text + * matching). Insertion-ordered; the oldest entry is evicted first. 256 is + * far above any real deployment's simultaneous voice channels — the bound + * exists so a long-lived process touching many channels cannot grow the + * maps without limit. */ +const MAX_TRACKED_CHANNELS = 256; + +/** Inbound frame cap. Everything the relay legitimately sends a bot — + * auth_ok, heartbeats, interruptions — is tiny; ws's 100 MiB default would + * let a misbehaving relay force huge single allocations. */ +const MAX_INBOUND_PAYLOAD_BYTES = 1024 * 1024; + +/** Outbound send-buffer cap. Messages are best-effort already (dropped when + * the socket is down), so a relay that stops reading must not grow the + * buffer without bound either — past this, drop instead of enqueueing. */ +const MAX_BUFFERED_BYTES = 4 * 1024 * 1024; + +export class RelayClientModule implements Module { + readonly name: string; + + private readonly config: RelayClientModuleConfig; + private readonly logger: RelayLogger; + + private ctx: ModuleContext | null = null; + private framework: AgentFramework | null = null; + + private ws: WebSocket | null = null; + private authed = false; + private shouldRun = false; + private reconnectTimer: ReturnType | null = null; + private reconnectDelayMs: number; + private stableTimer: ReturnType | null = null; + private watchdogTimer: ReturnType | null = null; + private lastInboundAt = 0; + + private bridge: TraceBridge | null = null; + private unsubscribeTracker: (() => void) | null = null; + + /** Agents this instance streams; null = all agents in the process. */ + private readonly agentSet: Set | null; + + /** + * channelId → agentName of the last agent seen streaming there. Entries + * deliberately survive the turn's end: a voice interruption can arrive + * moments after our terminal trace (speech lags text), and abortInference + * no-ops safely on an idle agent while the name still resolves. Bounded at + * MAX_TRACKED_CHANNELS, oldest evicted first. + */ + private readonly activeAgentByChannel = new Map(); + + /** + * agentName → channel of the turn the agent is streaming RIGHT NOW (null + * for a channel-less run; set on inference:started, cleared on the + * terminal traces). Guards interruption addressing: a late report for a + * channel the agent already left must not abort the unrelated turn it is + * running elsewhere. + */ + private readonly activeChannelByAgent = new Map(); + + /** + * channelId → visible text sent to the relay for the current utterance, + * tracked over two windows: `block` (since the channel's last + * block_start) and `activation` (since its last activation_start). + * Interruptions must prefix-match one of them or be dropped as stale — + * melodeus reports spokenText per block, the iOS client per activation. + * Only messages actually sent count — text dropped while the socket was + * down never reached a client. Pruned together with activeAgentByChannel. + */ + private readonly streamedText = new Map(); + + constructor(config: RelayClientModuleConfig) { + this.config = config; + this.logger = config.logger ?? consoleLogger; + this.name = `relay-client:${config.botId}`; + this.reconnectDelayMs = config.reconnectInitialMs ?? 1000; + this.agentSet = config.agents ? new Set(config.agents) : null; + } + + /** Wire the framework handle (HealthModule pattern; store not needed). */ + bind(framework: AgentFramework): void { + this.framework = framework; + } + + // ── Module lifecycle ───────────────────────────────────────────────────── + + async start(ctx: ModuleContext): Promise { + this.ctx = ctx; + this.shouldRun = true; + // A restarted module begins at the backoff floor, not wherever a + // previous incarnation's failures left it. + this.reconnectDelayMs = this.config.reconnectInitialMs ?? 1000; + + // Interruption addressing: remember which agent last streamed per + // channel, and which channel each agent is streaming right now. + // Subscribed BEFORE the bridge so tracking is in place by the time the + // bridge's messages for the same trace reach sendToRelay (whose + // spoken-text accumulation is gated on the channel being tracked). + this.unsubscribeTracker = ctx.onTrace((event: TraceEvent) => { + const agentName = (event as { agentName?: string }).agentName; + if (!agentName || (this.agentSet && !this.agentSet.has(agentName))) return; + if (event.type === 'inference:started') { + if (event.channelId) this.trackChannel(event.channelId, agentName); + this.activeChannelByAgent.set(agentName, event.channelId ?? null); + } else if ( + event.type === 'inference:completed' || + event.type === 'inference:turn_ended' || + event.type === 'inference:aborted' || + event.type === 'inference:failed' || + event.type === 'inference:exhausted' + ) { + this.activeChannelByAgent.delete(agentName); + } + }); + + // Identity is overridden to THIS connection's bot: the relay rejects any + // message whose botId differs from the identity the connection + // authenticated as. + this.bridge = new InferenceTraceBridge( + (msg) => this.sendToRelay(msg), + () => ({ + botId: this.config.botId, + userId: this.config.userId ?? this.config.botId, + username: this.config.username ?? this.config.botId, + }), + this.logger, + this.agentSet ? (name) => this.agentSet!.has(name) : undefined, + ); + this.bridge.start(ctx); + + this.connect(); + } + + async stop(): Promise { + this.shouldRun = false; + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = null; + } + if (this.stableTimer) { + clearTimeout(this.stableTimer); + this.stableTimer = null; + } + this.stopWatchdog(); + this.teardownSubscriptions(); + const ws = this.ws; + this.ws = null; + this.authed = false; + if (ws && ws.readyState === WebSocket.OPEN) ws.close(1000, 'module stopped'); + else ws?.terminate(); + } + + /** Unsubscribe from the trace bus and drop the tracking maps. Shared by + * stop() and the fatal replaced-connection path: a permanently-down + * module must not keep translating every turn into sends that go + * nowhere. */ + private teardownSubscriptions(): void { + this.bridge?.stop(); + this.bridge = null; + this.unsubscribeTracker?.(); + this.unsubscribeTracker = null; + this.activeAgentByChannel.clear(); + this.activeChannelByAgent.clear(); + this.streamedText.clear(); + } + + getTools(): ToolDefinition[] { + return []; + } + + async handleToolCall(_call: ToolCall): Promise { + return { success: false, error: 'relay-client has no tools', isError: true }; + } + + async onProcess(_event: ProcessEvent, _state: ProcessState): Promise { + return {}; + } + + /** @internal exposed for tests */ + get isConnected(): boolean { + return this.authed && this.ws?.readyState === WebSocket.OPEN; + } + + /** @internal exposed for tests */ + get trackedChannelCount(): number { + return this.activeAgentByChannel.size; + } + + /** @internal exposed for tests */ + get currentReconnectDelayMs(): number { + return this.reconnectDelayMs; + } + + // ── Channel tracking ───────────────────────────────────────────────────── + + private trackChannel(channelId: string, agentName: string): void { + // Delete-then-set moves a re-seen channel to the back of the insertion + // order, so eviction removes the genuinely least-recently-streamed one. + this.activeAgentByChannel.delete(channelId); + this.activeAgentByChannel.set(channelId, agentName); + while (this.activeAgentByChannel.size > MAX_TRACKED_CHANNELS) { + const oldest = this.activeAgentByChannel.keys().next().value as string; + this.activeAgentByChannel.delete(oldest); + this.streamedText.delete(oldest); + } + } + + // ── Connection ─────────────────────────────────────────────────────────── + + private connect(): void { + if (!this.shouldRun) return; + + const url = `${this.config.url.replace(/\/$/, '')}/bot`; + this.logger.info('Relay client connecting', { url, botId: this.config.botId }); + // handshakeTimeout: the watchdog only arms on 'open', so a host that + // accepts TCP but stalls the WebSocket upgrade would otherwise hang the + // module in CONNECTING forever (ws has no default handshake deadline). + const ws = new WebSocket(url, { + maxPayload: MAX_INBOUND_PAYLOAD_BYTES, + handshakeTimeout: 10_000, + }); + this.ws = ws; + this.authed = false; + + ws.on('open', () => { + if (this.ws !== ws) return; + const auth: Record = { + type: 'auth', + botId: this.config.botId, + token: this.config.token, + }; + if (this.config.userId) auth.userId = this.config.userId; + if (this.config.username) auth.username = this.config.username; + ws.send(JSON.stringify(auth)); + this.lastInboundAt = Date.now(); + this.startWatchdog(ws); + }); + + ws.on('message', (data) => { + if (this.ws !== ws) return; + // Any inbound frame proves the link is alive, malformed or not. + this.lastInboundAt = Date.now(); + let msg: Record; + try { + const parsed: unknown = JSON.parse(data.toString()); + if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) { + this.logger.warn('Relay client: non-object message dropped'); + return; + } + msg = parsed as Record; + } catch { + this.logger.warn('Relay client: non-JSON message dropped'); + return; + } + // Contained: handleMessage reaches framework code (abortInference); + // a throw there must not escape the socket listener and kill the + // process. + try { + this.handleMessage(msg); + } catch (error) { + this.logger.error('Relay client: message handling failed', { + type: msg.type, + error: String(error), + }); + } + }); + + ws.on('close', (code, reason) => { + if (this.ws !== ws) return; + this.ws = null; + this.authed = false; + if (this.stableTimer) { + clearTimeout(this.stableTimer); + this.stableTimer = null; + } + this.stopWatchdog(); + if (!this.shouldRun) return; + const reasonText = reason.toString(); + if (code === 1000 && /replaced/i.test(reasonText)) { + // The relay replaced this connection: another client authenticated + // with the same botId. Reconnecting would evict THAT client, which + // would reconnect and evict us — forever. Stay down; a duplicate + // identity is a configuration error, not an outage. + this.logger.error( + 'Relay replaced this connection — another client is using the same bot identity; not reconnecting', + { botId: this.config.botId, reason: reasonText }, + ); + this.shouldRun = false; + this.teardownSubscriptions(); + return; + } + this.logger.warn('Relay client disconnected', { code, reason: reasonText }); + this.scheduleReconnect(); + }); + + ws.on('error', (error) => { + if (this.ws !== ws) return; + this.logger.warn('Relay client socket error', { error: String(error) }); + // 'close' follows and drives the reconnect. + }); + } + + private scheduleReconnect(): void { + if (!this.shouldRun || this.reconnectTimer) return; + // ±20% jitter: several instances dropped by one relay restart must not + // re-dial in lockstep. + const delay = Math.round(this.reconnectDelayMs * (0.8 + Math.random() * 0.4)); + this.reconnectDelayMs = Math.min( + this.reconnectDelayMs * 2, + this.config.reconnectMaxMs ?? 30_000, + ); + this.logger.info('Relay client reconnecting', { inMs: delay }); + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = null; + this.connect(); + }, delay); + } + + /** Presume the link dead when nothing has arrived for heartbeatTimeoutMs + * (the relay heartbeats every connection every ~2s) and terminate it, so + * the normal 'close' → reconnect path recovers a half-open TCP link — + * or a server that accepted the socket but never answers — instead of + * streaming into the void until the OS timeout. */ + private startWatchdog(ws: WebSocket): void { + this.stopWatchdog(); + const timeoutMs = this.config.heartbeatTimeoutMs ?? 8_000; + if (timeoutMs <= 0) return; + this.watchdogTimer = setInterval(() => { + if (this.ws !== ws) { + this.stopWatchdog(); + return; + } + if (Date.now() - this.lastInboundAt > timeoutMs) { + this.logger.warn('Relay link presumed dead (nothing received within the heartbeat timeout)', { + timeoutMs, + }); + this.stopWatchdog(); + ws.terminate(); // 'close' follows and schedules the reconnect + } + }, Math.max(250, Math.min(timeoutMs / 2, 2_000))); + } + + private stopWatchdog(): void { + if (this.watchdogTimer) { + clearInterval(this.watchdogTimer); + this.watchdogTimer = null; + } + } + + private handleMessage(msg: Record): void { + switch (msg.type) { + case 'auth_ok': { + if (this.authed) return; // duplicate; must not re-arm the stability timer + this.authed = true; + this.logger.info('Relay client authenticated', { botId: this.config.botId }); + // Reset the backoff only after the connection proves stable. An + // immediate reset would let an auth-then-drop loop retry at the + // floor forever, defeating the backoff. + if (this.stableTimer) clearTimeout(this.stableTimer); + this.stableTimer = setTimeout(() => { + this.stableTimer = null; + if (this.isConnected) { + this.reconnectDelayMs = this.config.reconnectInitialMs ?? 1000; + } + }, this.config.backoffResetAfterMs ?? 10_000); + return; + } + + case 'auth_error': + // Bad credentials won't heal by retrying fast; keep the backoff + // growing (the relay closes the socket after this, driving 'close'). + // authed is dropped defensively in case a nonconforming relay keeps + // the socket open — streaming into an unauthenticated connection + // would be silently discarded server-side. + this.authed = false; + this.logger.error('Relay client auth rejected', { error: msg.error }); + return; + + case 'heartbeat': + return; + + case 'interruption': { + const channelId = typeof msg.channelId === 'string' ? msg.channelId : undefined; + const spokenText = typeof msg.spokenText === 'string' ? msg.spokenText : ''; + // Protocol enum; anything else is coerced so an arbitrary string + // cannot flow into traces and logs verbatim. + const reason = + msg.reason === 'manual' || msg.reason === 'timeout' ? msg.reason : 'user_speech'; + this.handleInterruption(channelId, spokenText, reason); + return; + } + + default: + this.logger.debug('Relay client: unhandled message type', { type: msg.type }); + } + } + + private handleInterruption(channelId: string | undefined, spokenText: string, reason: string): void { + if (!this.framework) { + this.logger.warn('Interruption received but framework not bound — dropped'); + return; + } + + // Resolve the target agent. A named channel must be one we streamed to; + // guessing on an unknown channel could abort an unrelated agent. Only a + // channel-LESS interruption may fall back to the single tracked agent — + // with one candidate there is nothing else it could mean. + let agentName: string | undefined; + let matchChannel = channelId; + if (channelId) { + agentName = this.activeAgentByChannel.get(channelId); + if (!agentName) { + this.logger.warn('Interruption for unknown channel dropped', { channelId }); + return; + } + } else if (this.activeAgentByChannel.size === 1) { + const [only] = this.activeAgentByChannel.entries(); + matchChannel = only[0]; + agentName = only[1]; + } + if (!agentName) { + this.logger.warn('Interruption without channel and no single candidate — dropped', { + trackedChannels: this.activeAgentByChannel.size, + }); + return; + } + + // A late report can outlive its turn (speech lags text): the channel + // still resolves the agent, but if that agent is now mid-turn somewhere + // ELSE — another channel, or a channel-less run — the report describes + // the earlier, finished turn, and aborting would kill the unrelated one. + const activeChannel = this.activeChannelByAgent.get(agentName); + if (activeChannel !== undefined && matchChannel && activeChannel !== matchChannel) { + this.logger.warn('Interruption for a finished turn dropped (agent is now active elsewhere)', { + channelId: matchChannel, + agentName, + activeChannel: activeChannel ?? '(channel-less)', + }); + return; + } + + // Staleness guard: the client voices a prefix of what we streamed, so a + // spokenText that does not prefix-match the current utterance describes + // an EARLIER utterance (its report raced the next turn) and must not + // abort the new one. Clients differ in their report window — melodeus + // resets per block_start, the iOS client accumulates the whole + // activation — so a match against either window is accepted. Empty + // spokenText carries no staleness evidence and is allowed. An empty + // accumulator (nothing streamed yet on this connection — module + // restart, or the turn is still inside a thinking/tool block) cannot + // convict either, but then the report is UNVERIFIABLE: the abort goes + // through WITHOUT keepText, because text we cannot match against what + // we actually streamed must never be committed as words the agent said. + let verifiedSpokenText: string | undefined; + if (spokenText) { + const streamed = matchChannel ? this.streamedText.get(matchChannel) : undefined; + if (streamed && (streamed.activation.length > 0 || streamed.block.length > 0)) { + if ( + !isWhitespaceInsensitivePrefix(spokenText, streamed.block) && + !isWhitespaceInsensitivePrefix(spokenText, streamed.activation) + ) { + this.logger.warn('Stale interruption dropped (spokenText does not match the current utterance)', { + channelId: matchChannel, + agentName, + spokenChars: spokenText.length, + streamedChars: streamed.activation.length, + }); + return; + } + verifiedSpokenText = spokenText; + } else if (streamed) { + // Entry present but both windows empty: we SAW this utterance's + // activation_start on this connection and streamed no visible text + // yet (thinking / pre-first-token). Real clients never report + // non-empty spokenText for an utterance that voiced nothing, so + // this report describes the PREVIOUS utterance (voice lag) and must + // not cut off the new turn. + this.logger.warn('Stale interruption dropped (current utterance has voiced nothing yet)', { + channelId: matchChannel, + agentName, + spokenChars: spokenText.length, + }); + return; + } else { + // No entry at all: we connected mid-turn (or the channel was + // evicted), so the report is genuinely unverifiable. The abort goes + // through, but text we cannot match against what we streamed must + // never be committed as words the agent said. + this.logger.warn('Interruption spokenText unverifiable (nothing streamed on this connection) — aborting without keepText', { + channelId: matchChannel, + agentName, + spokenChars: spokenText.length, + }); + } + } + + const aborted = this.framework.abortInference(agentName, { + reason, + keepText: verifiedSpokenText, + }); + this.logger.info('Interruption relayed to framework', { + agentName, + channelId, + aborted, + spokenChars: spokenText.length, + }); + } + + /** Send one message to the relay; drop it if the socket is down or its + * send buffer is full (delivery is best-effort by design). */ + private sendToRelay(msg: BotStreamMessage): void { + const ws = this.ws; + if (!this.authed || !ws || ws.readyState !== WebSocket.OPEN) { + this.logger.debug('Relay message dropped (relay not connected)', { type: msg.type }); + return; + } + if (ws.bufferedAmount > MAX_BUFFERED_BYTES) { + this.logger.debug('Relay message dropped (send buffer full)', { + type: msg.type, + buffered: ws.bufferedAmount, + }); + return; + } + ws.send(JSON.stringify(msg)); + + // Track what a voice client could have voiced from the current + // utterance (the staleness guard's reference), over both report windows + // — see streamedText. Only visible chunk text counts: thinking and tool + // content is never voiced. Gated on the channel still being tracked so + // a channel evicted mid-flight cannot re-enter this map and escape the + // shared MAX_TRACKED_CHANNELS bound (the tracker subscribes before the + // bridge, so on a fresh turn tracking is already in place here). + if (!this.activeAgentByChannel.has(msg.channelId)) return; + if (msg.type === 'activation_start') { + this.streamedText.set(msg.channelId, { activation: '', block: '' }); + } else if (msg.type === 'block_start') { + const entry = this.streamedText.get(msg.channelId); + if (entry) entry.block = ''; + else this.streamedText.set(msg.channelId, { activation: '', block: '' }); + } else if (msg.type === 'chunk' && msg.visible) { + // Missing entry = we connected mid-turn; both windows start at the + // first chunk that actually traversed this connection. + const entry = this.streamedText.get(msg.channelId) ?? { activation: '', block: '' }; + entry.activation += msg.text; + entry.block += msg.text; + this.streamedText.set(msg.channelId, entry); + } + } +} diff --git a/src/modules/voice-relay/trace-bridge.ts b/src/modules/voice-relay/trace-bridge.ts new file mode 100644 index 0000000..e42586b --- /dev/null +++ b/src/modules/voice-relay/trace-bridge.ts @@ -0,0 +1,233 @@ +/** + * Trace bridge: framework inference traces → relay wire messages. + * + * Translation (keyed by the `channelId` carried on inference:* traces): + * inference:started → activation_start + * inference:tokens → chunk + * inference:content_block → block_start / block_complete + * inference:completed/turn_ended → activation_end (complete) + * inference:aborted → activation_end (abort) + * inference:failed → activation_end (error) + * inference:stream_restarted → activation_end (abort) — the restart + * pairs its own fresh activation_start + * + * Traces without a channelId (channel-less turns: heartbeats, timers) have + * no channel to route to and are dropped with a debug log. Registry-less + * hosts still carry a channelId on message-triggered turns (the framework + * falls back to the triggering channel), so they stream normally. + * + * The wire protocol's `visible` flag is derived as `blockType === 'text'`: + * the tokens trace does not carry membrane's per-chunk visible bit, and for + * every current blockType the two agree (thinking / tool content is never + * voiced). block_complete's `content` is accumulated from the block's chunk + * traces, matching the reference relay where bots send the full block text. + */ + +import type { TraceEvent } from '../../types/trace.js'; +import type { ModuleContext } from '../../types/module.js'; +import type { + ActivationEndReason, + BlockType, + BotStreamMessage, + RelayLogger, +} from './types.js'; + +/** Send callback the module hands the bridge: deliver one translated bot + * message to the relay connection (each message carries its channelId). */ +export type ChannelBroadcastFn = (msg: BotStreamMessage) => void; + +/** + * Resolve the relay identity for a framework agent. `userId`/`username` are + * display fields (e.g. from voice config). `botId` overrides the outgoing + * botId (default: agentName) — the outbound relay client uses this, because + * the relay rejects any message whose botId differs from the identity the + * connection authenticated as. + */ +export type AgentIdentityResolver = ( + agentName: string, +) => { botId?: string; userId?: string; username?: string } | undefined; + +export interface TraceBridge { + /** Subscribe to the framework trace bus. Called from the module's start(). */ + start(ctx: ModuleContext): void; + /** Unsubscribe. Called from the module's stop(). */ + stop(): void; +} + +/** + * Real translator: framework-hosted agents produce the same relay wire + * messages as relay-connected bots, with agentName as the default botId. + */ +export class InferenceTraceBridge implements TraceBridge { + private unsubscribe: (() => void) | null = null; + /** Per-agent, per-blockIndex text accumulation for block_complete.content. */ + private blockText: Map> = new Map(); + + constructor( + private readonly broadcast: ChannelBroadcastFn, + private readonly resolveIdentity: AgentIdentityResolver = () => undefined, + private readonly logger?: RelayLogger, + /** Only translate traces from agents passing this filter (default: all). */ + private readonly agentFilter?: (agentName: string) => boolean, + ) {} + + start(ctx: ModuleContext): void { + this.unsubscribe = ctx.onTrace((event: TraceEvent) => { + try { + this.translate(event); + } catch (error) { + this.logger?.error('Trace bridge translation failed', { + error: String(error), + traceType: event.type, + }); + } + }); + } + + stop(): void { + this.unsubscribe?.(); + this.unsubscribe = null; + this.blockText.clear(); + } + + private identity(agentName: string): { botId: string; userId: string; username: string } { + const resolved = this.resolveIdentity(agentName); + return { + botId: resolved?.botId ?? agentName, + userId: resolved?.userId ?? agentName, + username: resolved?.username ?? agentName, + }; + } + + private drop(event: TraceEvent): void { + this.logger?.debug('Trace without channelId dropped (channel-less turn)', { + traceType: event.type, + }); + } + + private endActivation( + agentName: string, + channelId: string, + reason: ActivationEndReason, + timestamp: number, + ): void { + this.blockText.delete(agentName); + this.broadcast({ + type: 'activation_end', + ...this.identity(agentName), + channelId, + reason, + timestamp, + }); + } + + private translate(event: TraceEvent): void { + if (this.agentFilter) { + const agentName = (event as { agentName?: string }).agentName; + if (agentName !== undefined && !this.agentFilter(agentName)) return; + } + switch (event.type) { + case 'inference:started': { + if (!event.channelId) return this.drop(event); + this.blockText.delete(event.agentName); + this.broadcast({ + type: 'activation_start', + ...this.identity(event.agentName), + channelId: event.channelId, + timestamp: event.timestamp, + }); + return; + } + + case 'inference:tokens': { + if (!event.channelId) return this.drop(event); + let blocks = this.blockText.get(event.agentName); + if (!blocks) { + blocks = new Map(); + this.blockText.set(event.agentName, blocks); + } + blocks.set(event.blockIndex, (blocks.get(event.blockIndex) ?? '') + event.content); + this.broadcast({ + type: 'chunk', + ...this.identity(event.agentName), + channelId: event.channelId, + text: event.content, + blockIndex: event.blockIndex, + blockType: event.blockType as BlockType, + visible: event.blockType === 'text', + timestamp: event.timestamp, + }); + return; + } + + case 'inference:content_block': { + if (!event.channelId) return this.drop(event); + if (event.phase === 'block_start') { + this.blockText.get(event.agentName)?.delete(event.blockIndex); + this.broadcast({ + type: 'block_start', + ...this.identity(event.agentName), + channelId: event.channelId, + blockIndex: event.blockIndex, + blockType: event.blockType as BlockType, + timestamp: event.timestamp, + }); + } else { + const content = this.blockText.get(event.agentName)?.get(event.blockIndex) ?? ''; + this.broadcast({ + type: 'block_complete', + ...this.identity(event.agentName), + channelId: event.channelId, + blockIndex: event.blockIndex, + blockType: event.blockType as BlockType, + content, + timestamp: event.timestamp, + }); + } + return; + } + + // Terminal signals are mutually exclusive per COMPLETED turn: a plain + // turn emits `completed`, an endTurn-tool turn emits `turn_ended`, a + // user abort emits `aborted` (from abortInference — the stream's + // follow-up `exhausted` is deliberately NOT bridged to avoid a double + // activation_end), and a provider error emits `failed` (per attempt: + // a retried stream error produces activation_end(error) followed by a + // fresh activation_start, which clients treat as a new utterance). A + // context-budget restart emits `stream_restarted` instead of any of + // these — bridged below so its activation_start is paired too. + case 'inference:completed': + case 'inference:turn_ended': { + if (!event.channelId) return this.drop(event); + this.endActivation(event.agentName, event.channelId, 'complete', event.timestamp); + return; + } + + case 'inference:aborted': { + if (!event.channelId) return this.drop(event); + this.endActivation(event.agentName, event.channelId, 'abort', event.timestamp); + return; + } + + case 'inference:stream_restarted': { + if (!event.channelId) return this.drop(event); + // The framework abandoned the in-flight stream (context budget) and + // will re-stream the turn as a fresh activation. Close the current + // one so every activation_start on the wire is paired; 'abort' tells + // voice clients the partial utterance was cut off rather than + // finished (the replacement re-delivers). + this.endActivation(event.agentName, event.channelId, 'abort', event.timestamp); + return; + } + + case 'inference:failed': { + if (!event.channelId) return this.drop(event); + this.endActivation(event.agentName, event.channelId, 'error', event.timestamp); + return; + } + + default: + return; + } + } +} diff --git a/src/modules/voice-relay/types.ts b/src/modules/voice-relay/types.ts new file mode 100644 index 0000000..17931dd --- /dev/null +++ b/src/modules/voice-relay/types.ts @@ -0,0 +1,681 @@ +/** + * Voice relay wire protocol types. + * + * Ported from melodeus-tts-relay/src/types.ts — wire shapes must stay + * byte-compatible with the standalone relay so existing voice clients + * (melodeus, the iOS app) and legacy ChapterX bots (the Discord bot stack + * the relay was originally built to serve) work unchanged. Mirrors + * melodeus-tts-relay commit ec8f0f1 (2026-07-21); nothing on the wire + * carries a version, so on any divergence that repo's types.ts is + * canonical. "v2" is the relay repo's name for its current protocol — + * there is no negotiation and no v1 on the wire. + * + * Kept as the complete v2 protocol even though RelayClientModule uses only + * the bot-side subset (the streamed relay messages + interruption + + * heartbeat): one shared definition for any future relay-facing code, and a + * typed reference for the protocol as deployed. + */ + +export type BlockType = 'text' | 'thinking' | 'tool_call' | 'tool_result'; +export type InterruptionReason = 'user_speech' | 'manual' | 'timeout'; +export type ActivationEndReason = 'complete' | 'abort' | 'error'; + +/** Attachment on a Discord message */ +export interface MessageAttachment { + id: string; + filename: string; + url: string; + contentType?: string; + size: number; + width?: number; + height?: number; +} + +// ============================================================================ +// Authentication Messages +// ============================================================================ + +export interface BotAuthMessage { + type: 'auth'; + botId: string; + token: string; + userId?: string; // Discord user ID (for @mentions) + username?: string; // Display name +} + +export interface VoiceClientAuthMessage { + type: 'auth'; + clientId: string; + token: string; + username?: string; // User account username (new auth) +} + +export interface AuthOkMessage { + type: 'auth_ok'; + user?: { + username: string; + discordUserId?: string; + discordUsername?: string; + discordAvatarUrl?: string; + /** True iff the user is a member of the configured admin guild. */ + isAdmin?: boolean; + }; +} + +export interface AuthErrorMessage { + type: 'auth_error'; + error: string; +} + +// ============================================================================ +// Subscription Messages +// ============================================================================ + +export interface SubscribeMessage { + type: 'subscribe'; + channels: string[]; +} + +export interface SubscribedMessage { + type: 'subscribed'; + channels: string[]; +} + +// ============================================================================ +// Bot → Relay Messages (streamed turn content, forwarded verbatim to clients) +// ============================================================================ + +export interface ChunkMessage { + type: 'chunk'; + botId: string; + channelId: string; + userId: string; + username: string; + text: string; + blockIndex: number; + blockType: BlockType; + visible: boolean; + timestamp: number; +} + +export interface BlockStartMessage { + type: 'block_start'; + botId: string; + channelId: string; + userId: string; + username: string; + blockIndex: number; + blockType: BlockType; + timestamp: number; +} + +export interface BlockCompleteMessage { + type: 'block_complete'; + botId: string; + channelId: string; + userId: string; + username: string; + blockIndex: number; + blockType: BlockType; + content: string; + timestamp: number; +} + +export interface ActivationStartMessage { + type: 'activation_start'; + botId: string; + channelId: string; + userId: string; + username: string; + timestamp: number; +} + +export interface ActivationEndMessage { + type: 'activation_end'; + botId: string; + channelId: string; + userId: string; + username: string; + reason: ActivationEndReason; + timestamp: number; +} + +// ============================================================================ +// Voice Client → Relay Messages +// ============================================================================ + +export interface InterruptionMessage { + type: 'interruption'; + botId: string; + channelId: string; + spokenText: string; // The text that was actually voiced + reason: InterruptionReason; + timestamp: number; +} + +export interface TranscriptMessage { + type: 'transcript'; + channelId: string; + text: string; + speakerName?: string; // For webhook display name/avatar + targetBot?: string; // Explicit bot to @mention + attachmentUrls?: string[]; // URLs of images/files to attach + timestamp: number; +} + +export interface EditMessageMessage { + type: 'edit_message'; + channelId: string; + messageId: string; + text: string; + timestamp: number; +} + +export interface ReplaceMessageMessage { + type: 'replace_message'; + channelId: string; + messageId: string; + text: string; + speakerName?: string; // New speaker identity (if changing impersonation) + targetBot?: string; // Re-resolve @mention + timestamp: number; +} + +export interface DeleteMessageMessage { + type: 'delete_message'; + channelId: string; + messageId: string; + timestamp: number; +} + +// ============================================================================ +// Relay → Voice Client Messages (Discord feedback) +// ============================================================================ + +export interface MessagePostedMessage { + type: 'message_posted'; + channelId: string; + messageId: string; + text: string; + timestamp: number; +} + +export interface MessageEditedMessage { + type: 'message_edited'; + channelId: string; + messageId: string; + text: string; + timestamp: number; +} + +export interface MessageDeletedMessage { + type: 'message_deleted'; + channelId: string; + messageId: string; + timestamp: number; +} + +export interface WebhookErrorMessage { + type: 'webhook_error'; + channelId: string; + error: string; + originalText?: string; + timestamp: number; +} + +/** Discord channel message events (via gateway) */ +export interface ChannelMessageEventMessage { + type: 'channel_message'; + event: 'created' | 'updated' | 'deleted'; + channelId: string; + guildId: string; + messageId: string; + author?: { + id: string; + username: string; + displayName?: string; + bot: boolean; + }; + content?: string; + attachments?: MessageAttachment[]; + timestamp: number; +} + +/** Discord message ID for a streamed bot message (sent instead of suppressed channel_message) */ +export interface BotMessagePostedMessage { + type: 'bot_message_posted'; + botId?: string; + channelId: string; + messageId: string; + author: { + id: string; + username: string; + displayName?: string; + bot: boolean; + }; + content?: string; + attachments?: MessageAttachment[]; + timestamp: number; +} + +/** Reaction added or removed on a message */ +export interface MessageReactionEventMessage { + type: 'message_reaction'; + event: 'added' | 'removed'; + channelId: string; + messageId: string; + userId: string; + username?: string; + emoji: { + name: string; + id?: string; // Custom emoji snowflake + animated?: boolean; + }; + timestamp: number; +} + +/** Bot joined/left a guild */ +export interface BotRosterEventMessage { + type: 'bot_roster'; + event: 'joined' | 'left'; + guildId: string; + bot: { + userId: string; + username: string; + displayName?: string; + }; + timestamp: number; +} + +/** Periodic keepalive sent to all connections (clients may ignore). */ +export interface HeartbeatMessage { + type: 'heartbeat'; + timestamp: number; +} + +// ============================================================================ +// Relay → Bot Messages +// ============================================================================ + +export interface BotInterruptionMessage { + type: 'interruption'; + channelId: string; + spokenText: string; // Bot matches this to find the message to edit + reason: InterruptionReason; + timestamp: number; +} + +// ============================================================================ +// Channel Config Messages +// ============================================================================ + +/** Client requests available channels */ +export interface GetChannelsMessage { + type: 'get_channels'; +} + +/** Relay sends available channels organized by guild/category */ +export interface ChannelsMessage { + type: 'channels'; + guilds: GuildChannelList[]; +} + +export interface GuildChannelList { + guildId: string; + guildName: string; + categories: CategoryInfo[]; + /** Channels not in any category */ + uncategorized: ChannelInfo[]; +} + +export interface CategoryInfo { + id: string; + name: string; + position: number; + channels: ChannelInfo[]; +} + +export interface ChannelInfo { + id: string; + name: string; + position: number; + /** Whether this channel is configured for voice in the relay */ + voice: boolean; + /** True if this entry is a thread (Discord ChannelType 10/11/12). */ + isThread?: boolean; + /** For threads: the parent text channel id. Absent for regular channels. */ + parentId?: string; +} + +/** Client requests the aggregated list of bots known across all guilds. */ +export interface GetBotsMessage { + type: 'get_bots'; +} + +/** Server response: flattened, deduplicated bot list. */ +export interface BotsMessage { + type: 'bots'; + bots: Array<{ + userId: string; + username: string; + displayName?: string; + avatarUrl?: string; + guildName?: string; // first guild we found this bot in + }>; +} + +export interface GetMembersMessage { + type: 'get_members'; + channelId: string; // Relay resolves to guildId +} + +export interface GuildMemberInfo { + userId: string; + username: string; + displayName?: string; + bot: boolean; + avatarUrl?: string; +} + +/** Relay sends the full member roster */ +export interface MembersMessage { + type: 'members'; + guildId: string; + members: GuildMemberInfo[]; +} + +/** A member joined or left */ +export interface MemberRosterEventMessage { + type: 'member_roster'; + event: 'joined' | 'left'; + guildId: string; + member: GuildMemberInfo; + timestamp: number; +} + +/** Client requests channel history (backscroll) */ +export interface GetHistoryMessage { + type: 'get_history'; + channelId: string; + before?: string; // Message ID — fetch messages before this (pagination) + limit?: number; // Max messages to return (default 50, max 100) +} + +/** Relay returns channel history */ +export interface HistoryMessage { + type: 'history'; + channelId: string; + messages: HistoryEntry[]; + hasMore: boolean; // True if there are older messages to fetch +} + +export interface MessageReaction { + emoji: { name: string; id?: string; animated?: boolean }; + count: number; +} + +export interface HistoryEntry { + messageId: string; + author: { + id: string; + username: string; + displayName?: string; + bot: boolean; + }; + content: string; + attachments?: MessageAttachment[]; + reactions?: MessageReaction[]; + timestamp: number; + editedTimestamp?: number; +} + +/** Client links their account to a Discord user */ +export interface LinkDiscordMessage { + type: 'link_discord'; + discordUserId: string; +} + +/** Client sets their speaking persona */ +export interface SetPersonaMessage { + type: 'set_persona'; + username?: string; // Display name for webhook (null = use Discord identity) + avatarUrl?: string; // Avatar URL (null = use Discord avatar) +} + +/** Client clears persona, reverting to their Discord identity */ +export interface ClearPersonaMessage { + type: 'clear_persona'; +} + +/** Client adds a reaction to a message */ +export interface AddReactionMessage { + type: 'add_reaction'; + channelId: string; + messageId: string; + emoji: string; // Unicode emoji or custom emoji format "name:id" +} + +/** Client removes a reaction from a message */ +export interface RemoveReactionMessage { + type: 'remove_reaction'; + channelId: string; + messageId: string; + emoji: string; +} + +/** Client toggles voice on/off for a channel */ +export interface SetVoiceChannelMessage { + type: 'set_voice_channel'; + channelId: string; + enabled: boolean; +} + +/** Client sets overrides for a specific channel */ +export interface SetChannelOverridesMessage { + type: 'set_channel_overrides'; + channelId: string; + overrides: ChannelOverrides; +} + +/** Client requests the current config */ +export interface GetConfigMessage { + type: 'get_config'; +} + +/** Relay sends full config to client (on subscribe or on request) */ +export interface ConfigMessage { + type: 'config'; + config: ChannelConfig; +} + +/** Client requests a config update (partial) */ +export interface UpdateConfigMessage { + type: 'update_config'; + update: Partial; +} + +/** Relay confirms config was updated and broadcasts new config */ +export interface ConfigUpdatedMessage { + type: 'config_updated'; + config: ChannelConfig; + updatedBy: string; // clientId that made the change +} + +// ============================================================================ +// Channel Config (served to clients) +// ============================================================================ + +export interface VoiceSettings { + speed: number; + stability: number; + similarityBoost: number; +} + +export interface VoiceConfig { + voiceId: string; + voiceSettings: VoiceSettings; + discordName: string; + enabled: boolean; + /** Narrator/emotive voice for text in asterisks. Falls back to system default if not set. */ + narratorVoiceId?: string; + narratorVoiceSettings?: VoiceSettings; +} + +/** Config that the relay serves to voice clients */ +export interface ChannelConfig { + /** ElevenLabs API key (passthrough to clients for now) */ + elevenLabsKey: string; + + /** TTS model to use */ + ttsModel: string; + + /** Bot name → voice config */ + voices: Record; + + /** Default voice for bots without a specific voice config */ + defaultBotVoice: VoiceConfig; + + /** Default voice for human messages */ + defaultHumanVoice: VoiceConfig; + + /** System-wide default narrator voice (for emotive text in asterisks) */ + defaultNarratorVoiceId: string; + defaultNarratorVoiceSettings: VoiceSettings; + + /** Speaker name → display config for webhook */ + speakers: Record; + + /** Mention routing */ + mentionMode: MentionMode; + defaultBot?: string; + + /** Director config */ + director: DirectorConfig; + + /** Per-channel overrides (channelId → overrides) */ + channelOverrides: Record; + + /** Channels explicitly configured for voice (shown as voice-enabled to clients) */ + voiceChannels?: string[]; + + /** User accounts (username → account). Redacted for non-admin clients. */ + users: Record; +} + +export interface DirectorConfig { + mode: 'off' | 'same_model' | 'director'; + defaultCharacter?: string; +} + +/** Per-channel overrides — any field set here takes precedence over the global config */ +export interface ChannelOverrides { + voices?: Record>; + defaultBotVoice?: Partial; + defaultHumanVoice?: Partial; + defaultNarratorVoiceId?: string; + defaultNarratorVoiceSettings?: Partial; + mentionMode?: MentionMode; + defaultBot?: string; + director?: Partial; +} + +export type MentionMode = 'default' | 'explicit' | 'last_speaker' | 'round_robin'; + +export interface SpeakerConfig { + discordUsername: string; + discordUserId?: string; + avatarUrl?: string; + aliases?: string[]; +} + +export interface UserAccount { + token: string; + tokenHashed?: boolean; + discordUserId?: string; +} + +// ============================================================================ +// Union Types +// ============================================================================ + +export type BotToRelayMessage = + | BotAuthMessage + | ChunkMessage + | BlockStartMessage + | BlockCompleteMessage + | ActivationStartMessage + | ActivationEndMessage; + +/** What a connected bot streams after authenticating — everything a bot may + * put on the wire except the auth handshake itself. The outbound type of + * RelayClientModule/InferenceTraceBridge, so the compiler rejects any + * message a /bot client must never send. */ +export type BotStreamMessage = Exclude; + +export type VoiceClientToRelayMessage = + | VoiceClientAuthMessage + | SubscribeMessage + | InterruptionMessage + | TranscriptMessage + | EditMessageMessage + | ReplaceMessageMessage + | DeleteMessageMessage + | GetConfigMessage + | UpdateConfigMessage + | GetHistoryMessage + | GetChannelsMessage + | GetMembersMessage + | GetBotsMessage + | AddReactionMessage + | RemoveReactionMessage + | SetVoiceChannelMessage + | SetChannelOverridesMessage + | SetPersonaMessage + | ClearPersonaMessage + | LinkDiscordMessage; + +export type RelayToBotMessage = + | AuthOkMessage + | AuthErrorMessage + | BotInterruptionMessage + | HeartbeatMessage; + +export type RelayToVoiceClientMessage = + | AuthOkMessage + | AuthErrorMessage + | SubscribedMessage + | ChunkMessage + | BlockStartMessage + | BlockCompleteMessage + | ActivationStartMessage + | ActivationEndMessage + | MessagePostedMessage + | MessageEditedMessage + | MessageDeletedMessage + | WebhookErrorMessage + | ConfigMessage + | ConfigUpdatedMessage + | ChannelMessageEventMessage + | BotMessagePostedMessage + | MessageReactionEventMessage + | BotRosterEventMessage + | HistoryMessage + | ChannelsMessage + | MembersMessage + | BotsMessage + | MemberRosterEventMessage + | HeartbeatMessage; + +// ============================================================================ +// Logging +// ============================================================================ + +/** Minimal logger surface so hosts can capture or silence output. + * Console-backed by default for info and above; debug is dropped unless a + * logger is injected. */ +export interface RelayLogger { + debug(msg: string, data?: unknown): void; + info(msg: string, data?: unknown): void; + warn(msg: string, data?: unknown): void; + error(msg: string, data?: unknown): void; +} diff --git a/src/prose-segments.ts b/src/prose-segments.ts index 6d8f3f0..ee00fae 100644 --- a/src/prose-segments.ts +++ b/src/prose-segments.ts @@ -40,3 +40,83 @@ export function splitProseSegments(content: readonly ContentBlock[]): string[] { return segments; } + +/** + * The part of `spoken` not already covered by `delivered`, ignoring the + * characters voice clients normalize away before reporting: whitespace + * (live-routed segments carry their own separators, while spoken text keeps + * the stream's spacing) and narrator-markup asterisks (the iOS client + * splits `*action*` spans to a narrator voice and reports them WITHOUT the + * asterisks, so `*grins* Hi` comes back as "grins Hi"). + * + * Three outcomes, keyed to what voice clients actually send as spokenText: + * - `delivered` covers all of `spoken` → null (everything the user heard is + * already in the channel; posting again would duplicate it). + * - `spoken` starts with `delivered` and extends past it → the suffix, + * edge-trimmed (a client that accumulates the whole turn's speech: post + * only the tail). + * - the two do not align → `spoken` unchanged. Non-alignment means this is a + * NEW utterance's text, not a re-send: the relay protocol tracks spoken + * text per message, and the reference client (melodeus) resets its + * accumulator at every block_start — so an interruption mid round 2 sends + * only round 2's fragment, which the live path has never posted. + */ +export function undeliveredSuffix(spoken: string, delivered: string): string | null { + const isWs = isNormalizedAway; + let i = 0; // index into spoken + let j = 0; // index into delivered + while (j < delivered.length) { + if (isWs(delivered[j])) { + j++; + continue; + } + while (i < spoken.length && isWs(spoken[i])) i++; + if (i >= spoken.length) return null; // spoken fully covered by delivered + if (spoken[i] !== delivered[j]) { + // Divergence: `spoken` is a different utterance than the posted prose + // (per-block client), not a prefix re-send. All of it is undelivered. + return spoken; + } + i++; + j++; + } + const rest = spoken.slice(i).trim(); + return rest.length > 0 ? rest : null; +} + +/** Characters voice clients normalize away before reporting spoken text: + * whitespace (clients join/trim segments) and narrator-markup asterisks + * (the iOS client voices `*action*` spans via a narrator voice and strips + * the asterisks from its report). Both comparison walks skip them on both + * sides so a report can never fail to match over markup the client was + * never going to echo. */ +const isNormalizedAway = (c: string): boolean => + c === ' ' || c === '\n' || c === '\t' || c === '\r' || c === '*'; + +/** + * Whether `prefix` matches the start of `text`, ignoring characters voice + * clients normalize away (same walk as undeliveredSuffix, answering only + * yes/no). + * + * Used to judge whether a voice client's reported spoken text belongs to the + * utterance currently streaming in a channel: the client voices a prefix of + * what was streamed, so a report that does not prefix-match the current + * utterance is stale — it describes an earlier utterance and must not + * interrupt the new one. An empty `prefix` trivially matches. + */ +export function isWhitespaceInsensitivePrefix(prefix: string, text: string): boolean { + const isWs = isNormalizedAway; + let i = 0; // index into text + let j = 0; // index into prefix + while (j < prefix.length) { + if (isWs(prefix[j])) { + j++; + continue; + } + while (i < text.length && isWs(text[i])) i++; + if (i >= text.length || text[i] !== prefix[j]) return false; + i++; + j++; + } + return true; +} diff --git a/src/types/trace.ts b/src/types/trace.ts index a349d97..622ae5b 100644 --- a/src/types/trace.ts +++ b/src/types/trace.ts @@ -37,24 +37,42 @@ export type TraceEvent = }) // Inference lifecycle - | (TraceEventBase & { type: 'inference:started'; agentName: string }) + | (TraceEventBase & { + type: 'inference:started'; + agentName: string; + /** + * Channel identity of this turn (the locus its speech routes to): + * the turn's pinned locus, else the triggering channel. Absent for + * channel-less turns (heartbeats, timers). Carried on the turn-scoped + * traces — started, tokens, content_block, tool_calls_yielded, + * completed, turn_ended, aborted, failed — so channel-scoped consumers + * (voice/streaming) can key per-channel streams without reaching into + * framework state. Also on stream_restarted (so the abandoned + * activation can be closed). NOT carried on exhausted, usage, + * stream_resumed, or the request_* traces. + */ + channelId?: string; + }) | (TraceEventBase & { type: 'inference:completed'; agentName: string; durationMs: number; tokenUsage?: { input: number; output: number; cacheCreation?: number; cacheRead?: number }; + channelId?: string; }) | (TraceEventBase & { type: 'inference:aborted'; agentName: string; durationMs: number; reason?: string; + channelId?: string; }) | (TraceEventBase & { type: 'inference:failed'; agentName: string; error: string; stack?: string; + channelId?: string; }) | (TraceEventBase & { type: 'inference:exhausted'; @@ -82,6 +100,7 @@ export type TraceEvent = blockType: 'text' | 'thinking' | 'tool_call' | 'tool_result'; /** 0-indexed block position in the current assistant turn. */ blockIndex: number; + channelId?: string; }) | (TraceEventBase & { /** @@ -95,11 +114,13 @@ export type TraceEvent = phase: 'block_start' | 'block_complete'; blockType: 'text' | 'thinking' | 'tool_call' | 'tool_result'; blockIndex: number; + channelId?: string; }) | (TraceEventBase & { type: 'inference:tool_calls_yielded'; agentName: string; calls: Array<{ id: string; name: string; input?: unknown }>; + channelId?: string; }) | (TraceEventBase & { type: 'inference:usage'; @@ -115,6 +136,9 @@ export type TraceEvent = | (TraceEventBase & { type: 'inference:stream_restarted'; agentName: string; + /** Present so channel-scoped consumers can close the abandoned + * activation before the replacement stream starts a fresh one. */ + channelId?: string; reason: string; inputTokens: number; budget: number; @@ -122,6 +146,7 @@ export type TraceEvent = | (TraceEventBase & { type: 'inference:turn_ended'; agentName: string; + channelId?: string; }) // Tool lifecycle diff --git a/test/helpers/mock-membrane.ts b/test/helpers/mock-membrane.ts index 21774f3..61db5d7 100644 --- a/test/helpers/mock-membrane.ts +++ b/test/helpers/mock-membrane.ts @@ -56,7 +56,14 @@ export class MockYieldingStream implements YieldingStream { { injectedMessages?: Array<{ participant?: string; content: unknown[] }> } | undefined > = []; - constructor(private responses: NormalizedResponse[]) { + constructor( + private responses: NormalizedResponse[], + /** Mirrors membrane's emitBlocks stream option: block boundary events + * are emitted only when the caller asked for them. Respecting the flag + * here means every block-dependent test (bridge, e2e) doubles as a + * regression pin on the agent actually requesting blocks. */ + private readonly emitBlocks: boolean = true, + ) { this.processResponse(0); } @@ -69,19 +76,23 @@ export class MockYieldingStream implements YieldingStream { const text = response.rawAssistantText; if (text) { - this.events.push({ - type: 'block', - event: { event: 'block_start', index: 0, block: { type: 'text' } }, - } as StreamEvent); + if (this.emitBlocks) { + this.events.push({ + type: 'block', + event: { event: 'block_start', index: 0, block: { type: 'text' } }, + } as StreamEvent); + } this.events.push({ type: 'tokens', content: text, meta: { type: 'text', visible: true, blockIndex: 0 }, } as StreamEvent); - this.events.push({ - type: 'block', - event: { event: 'block_complete', index: 0, block: { type: 'text', content: text } }, - } as StreamEvent); + if (this.emitBlocks) { + this.events.push({ + type: 'block', + event: { event: 'block_complete', index: 0, block: { type: 'text', content: text } }, + } as StreamEvent); + } } if (response.usage) { @@ -181,11 +192,14 @@ export class MockMembrane { return this.responses[this.responseIndex++]; } - streamYielding(request: NormalizedRequest, _options?: unknown): YieldingStream { + streamYielding(request: NormalizedRequest, options?: unknown): YieldingStream { this.calls.push(request); const remaining = this.responses.slice(this.responseIndex); this.responseIndex = this.responses.length; - const stream = new MockYieldingStream(remaining); + // Honor emitBlocks like the real membrane (default true) — see the + // MockYieldingStream constructor note. + const emitBlocks = (options as { emitBlocks?: boolean } | undefined)?.emitBlocks ?? true; + const stream = new MockYieldingStream(remaining, emitBlocks); this.lastStream = stream; return stream; } diff --git a/test/helpers/voice-relay-sims.ts b/test/helpers/voice-relay-sims.ts new file mode 100644 index 0000000..3135d91 --- /dev/null +++ b/test/helpers/voice-relay-sims.ts @@ -0,0 +1,261 @@ +/** + * Voice-relay test helpers: a recording WebSocket wrapper, a voice-client + * simulator speaking the v2 /tts protocol, shared fixture config, and a + * spawner that runs the reference melodeus-tts-relay for end-to-end tests. + * + * (Trimmed from the full WebSocket conformance harness: the bot simulator, + * fixture runner, and transcript comparator live on the voice-relay-module + * branch alongside the server-side module they exercise.) + */ + +import { spawn, type ChildProcess } from 'node:child_process'; +import { existsSync, writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +import { WebSocket } from 'ws'; + +// ── Wire-level socket with transcript recording ──────────────────────────── + +export class SimSocket { + readonly ws: WebSocket; + readonly transcript: Array> = []; + heartbeats = 0; + private queue: Array> = []; + private waiters: Array<(msg: Record) => void> = []; + readonly closeInfo: Promise<{ code: number; reason: string }>; + + constructor(url: string) { + this.ws = new WebSocket(url); + this.ws.on('message', (data) => { + const msg = JSON.parse(data.toString()) as Record; + if (msg.type === 'heartbeat') { + this.heartbeats++; + return; + } + this.transcript.push(msg); + const waiter = this.waiters.shift(); + if (waiter) waiter(msg); + else this.queue.push(msg); + }); + this.closeInfo = new Promise((resolve) => { + this.ws.on('close', (code, reason) => resolve({ code, reason: reason.toString() })); + }); + } + + async opened(): Promise { + if (this.ws.readyState === WebSocket.OPEN) return; + await new Promise((resolve, reject) => { + this.ws.once('open', () => resolve()); + this.ws.once('error', reject); + }); + } + + send(msg: Record): void { + this.ws.send(JSON.stringify(msg)); + } + + next(timeoutMs = 3000): Promise> { + const queued = this.queue.shift(); + if (queued) return Promise.resolve(queued); + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error(`Timed out waiting for message (transcript so far: ${JSON.stringify(this.transcript)})`)), + timeoutMs, + ); + this.waiters.push((msg) => { + clearTimeout(timer); + resolve(msg); + }); + }); + } + + /** Wait a quiet window; returns the number of unexpected queued messages. */ + async settle(ms = 200): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); + return this.queue.length; + } + + /** Drain anything queued (used before diffing full transcripts). */ + drain(): void { + this.queue.length = 0; + } + + close(): void { + this.ws.close(); + } +} + +// ── Voice client simulator (v2 client protocol) ──────────────────────────── + +export interface ClientAuthOpts { + clientId: string; + token: string; + username?: string; +} + +export class VoiceClientSim { + sock!: SimSocket; + constructor(private baseUrl: string) {} + + async connect(): Promise { + this.sock = new SimSocket(`${this.baseUrl}/tts`); + await this.sock.opened(); + } + + async auth(opts: ClientAuthOpts): Promise> { + const msg: Record = { type: 'auth', clientId: opts.clientId, token: opts.token }; + if (opts.username !== undefined) msg.username = opts.username; + this.sock.send(msg); + return this.sock.next(); + } + + /** + * Subscribe and collect the reply burst: `subscribed`, then any of + * `members`/`config` in server order, until `config` arrives (the relay + * ends the burst with config). + */ + async subscribe(channels: string[]): Promise[]> { + this.sock.send({ type: 'subscribe', channels }); + const burst: Record[] = []; + for (;;) { + const msg = await this.sock.next(); + burst.push(msg); + if (msg.type === 'config') break; + if (burst.length > 8) throw new Error(`subscribe burst never ended: ${JSON.stringify(burst)}`); + } + return burst; + } + + send(msg: Record): void { + this.sock.send(msg); + } + + next(timeoutMs?: number): Promise> { + return this.sock.next(timeoutMs); + } + + close(): void { + this.sock.close(); + } +} + +// ── Shared fixtures: tokens, user account, relay config ──────────────────── +export const TOKENS = { + bot: 'bot-secret', + client: 'client-secret', +}; + +/** + * Write the reference relay's fixture config. Returns the file path. + * (Client auth uses the BOT_TOKENS/TTS_CLIENT_TOKENS env pools, not this + * file; the config only feeds the relay's voice routing.) + */ +export function writeRelayConfig(dir: string): string { + const path = join(dir, 'config.json'); + writeFileSync( + path, + JSON.stringify({ + relay_config: { + elevenLabsKey: 'sk_conformance', + ttsModel: 'eleven_multilingual_v2', + voices: { + Opus45: { voiceId: 'V-opus', discordName: 'Opus 4.5', enabled: true }, + }, + mentionMode: 'default', + defaultBot: 'Opus45', + }, + }), + ); + return path; +} + +// ── Reference relay spawner ───────────────────────────────────────────────── + +export interface ReferenceRelay { + url: string; + httpUrl: string; + stop: () => Promise; +} + +export function referenceRelayDir(): string { + const home = process.env.HOME; + return ( + process.env.VOICE_RELAY_REFERENCE_DIR ?? + (home ? join(home, 'hot', 'melodeus-tts-relay') : '') + ); +} + +/** Null if the reference repo or its node_modules are absent. */ +export function referenceRelayAvailable(): string | null { + const dir = referenceRelayDir(); + if (!dir) return null; + if (!existsSync(join(dir, 'src', 'index.ts'))) return null; + if (!existsSync(join(dir, 'node_modules', 'ws'))) return null; + if (!existsSync(join(dir, 'node_modules', '.bin', 'tsx'))) return null; + return dir; +} + +export async function spawnReferenceRelay(configFile: string): Promise { + const dir = referenceRelayAvailable(); + if (!dir) throw new Error('reference relay unavailable'); + + let lastError: Error | null = null; + for (let attempt = 0; attempt < 3; attempt++) { + const port = 21000 + Math.floor(Math.random() * 8000); + const child: ChildProcess = spawn( + join(dir, 'node_modules', '.bin', 'tsx'), + ['src/index.ts'], + { + cwd: dir, + env: { + ...process.env, + PORT: String(port), + HOST: '127.0.0.1', + BOT_TOKENS: TOKENS.bot, + TTS_CLIENT_TOKENS: TOKENS.client, + CONFIG_FILE: configFile, + LOG_LEVEL: 'error', + // No DISCORD_BOT_TOKEN: the relay runs gateway/webhook-less — + // pure streaming fan-out, all these tests need. + DISCORD_BOT_TOKEN: '', + }, + stdio: ['ignore', 'pipe', 'pipe'], + }, + ); + let stderr = ''; + child.stderr?.on('data', (d) => (stderr += d.toString())); + + const httpUrl = `http://127.0.0.1:${port}`; + const deadline = Date.now() + 20_000; + let up = false; + while (Date.now() < deadline && child.exitCode === null) { + try { + const res = await fetch(`${httpUrl}/health`, { signal: AbortSignal.timeout(500) }); + if (res.ok) { + up = true; + break; + } + } catch { + await new Promise((r) => setTimeout(r, 200)); + } + } + if (up) { + return { + url: `ws://127.0.0.1:${port}`, + httpUrl, + stop: async () => { + child.kill('SIGTERM'); + await Promise.race([ + new Promise((r) => child.once('exit', r)), + new Promise((r) => setTimeout(r, 2500)), + ]); + if (child.exitCode === null) child.kill('SIGKILL'); + }, + }; + } + child.kill('SIGKILL'); + lastError = new Error( + `reference relay failed to start on port ${port} (exit=${child.exitCode}): ${stderr.slice(0, 500)}`, + ); + } + throw lastError ?? new Error('reference relay failed to start'); +} diff --git a/test/inference-trace-channel.test.ts b/test/inference-trace-channel.test.ts new file mode 100644 index 0000000..4ac4b64 --- /dev/null +++ b/test/inference-trace-channel.test.ts @@ -0,0 +1,515 @@ +/** + * Voice relay connection tests. + * + * Channel identity on inference:* traces: a channel-triggered turn + * stamps its channelId on the whole trace family; a channel-less wake + * leaves the field undefined; a stale locus pin from a previous turn does + * not leak into the next turn's traces. + * + * abortInference(agentName, { keepText }): a user abort mid-turn + * persists the spoken prefix as the assistant's turn (and routes it to the + * turn's locus); an abort without keepText preserves the historical + * discard; the string overload still works. + */ + +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { tmpdir } from 'node:os'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import type { + Module, + ModuleContext, + ProcessState, + ProcessEvent, + EventResponse, + ToolDefinition, + ToolCall, + ToolResult, + TraceEvent, +} from '../src/index.js'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +import type { ContentBlock } from '@animalabs/membrane'; + +/** Module with a `hold` tool that parks the stream in waiting_for_tools until released. */ +class HoldModule implements Module { + readonly name = 'hold'; + toolStarted: Promise; + private signalToolStarted!: () => void; + private pendingReleases: Array<(r: ToolResult) => void> = []; + private nextStartWaiters: Array<() => void> = []; + + constructor() { + this.toolStarted = new Promise((resolve) => (this.signalToolStarted = resolve)); + } + + /** Resolves when the next tool call AFTER this point starts (re-armable, + * for multi-round scenarios; `toolStarted` stays first-call-only). */ + nextToolStart(): Promise { + return new Promise((resolve) => this.nextStartWaiters.push(resolve)); + } + + releaseTool(): void { + const release = this.pendingReleases.shift(); + release?.({ success: true, data: { ok: true } }); + } + + async start(_ctx: ModuleContext): Promise {} + async stop(): Promise {} + + getTools(): ToolDefinition[] { + return [ + { + name: 'hold', + description: 'Blocks until released', + inputSchema: { type: 'object', properties: {} }, + }, + ]; + } + + async handleToolCall(_call: ToolCall): Promise { + this.signalToolStarted(); + for (const waiter of this.nextStartWaiters.splice(0)) waiter(); + return new Promise((resolve) => this.pendingReleases.push(resolve)); + } + + async onProcess(event: ProcessEvent, _state: ProcessState): Promise { + if (event.type === 'external-message') { + return { + addMessages: [ + { + participant: 'Nick', + content: [{ type: 'text', text: String((event as { content?: unknown }).content) }], + }, + ], + requestInference: true, + }; + } + return {}; + } +} + +function channelIncoming(channelId: string, text: string): ProcessEvent { + return { + type: 'mcpl:channel-incoming', + serverId: 'test-server', + channelId, + messageId: `m-${Math.floor(Math.random() * 1e9)}`, + author: { id: 'u1', name: 'Nick' }, + content: [{ type: 'text', text }], + timestamp: new Date().toISOString(), + triggerInference: true, + } as unknown as ProcessEvent; +} + +/** Minimal ChannelRegistry stub capturing routeSpeech calls (Proxy no-ops the rest). */ +function stubChannelRegistry(framework: AgentFramework) { + const routed: Array<{ text: string; locus: string | null }> = []; + const explicit: Record = { + // The real registry resolves home → active trigger channel → default; + // these scenarios trigger via chan-live, so a faithful stub pins it. + resolveLocus: () => 'chan-live', + routeSpeech: async (_agent: string, text: string, locus?: string | null) => { + routed.push({ text, locus: locus ?? null }); + }, + getDefaultPublishChannel: () => null, + isChannelOpen: () => true, + getDescriptor: () => undefined, + getChannelTools: () => [], + }; + (framework as unknown as { channelRegistry: unknown }).channelRegistry = new Proxy(explicit, { + get: (target, prop: string) => (prop in target ? target[prop] : () => undefined), + }); + return routed; +} + +describe('channel identity on inference traces', () => { + let tempDir: string; + let membrane: MockMembrane; + + beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), 'channel-identity-test-')); + membrane = new MockMembrane(); + }); + + afterEach(() => { + rmSync(tempDir, { recursive: true, force: true }); + }); + + async function createFramework(modules: Module[] = []): Promise { + return AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'Test.' }], + modules, + }); + } + + it('stamps the triggering channel on the whole inference trace family', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Hello there' }] as ContentBlock[])); + const framework = await createFramework(); + const traces: TraceEvent[] = []; + framework.onTrace((e) => traces.push(e)); + + framework.pushEvent(channelIncoming('chan-voice-1', 'hey assistant')); + await framework.runUntilIdle(); + + const byType = (t: string) => traces.filter((e) => e.type === t); + for (const type of [ + 'inference:started', + 'inference:tokens', + 'inference:content_block', + 'inference:completed', + ]) { + const events = byType(type); + assert.ok(events.length > 0, `expected at least one ${type} trace`); + for (const e of events) { + assert.equal( + (e as { channelId?: string }).channelId, + 'chan-voice-1', + `${type} carries the triggering channelId`, + ); + } + } + + await framework.stop(); + }); + + it('leaves channelId undefined on channel-less wakes and clears stale pins between turns', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'First (channel) turn' }] as ContentBlock[])); + const holdModule = new HoldModule(); + const framework = await createFramework([holdModule]); + const traces: TraceEvent[] = []; + framework.onTrace((e) => traces.push(e)); + + // Turn 1: channel-triggered. + framework.pushEvent(channelIncoming('chan-old', 'hello')); + await framework.runUntilIdle(); + + // Turn 2: module-triggered external message, no channel anywhere. + // (Pushed only now: the mock membrane hands ALL queued responses to the + // first stream, so turn 2's response must be queued after turn 1 ran.) + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Second (no channel) turn' }] as ContentBlock[])); + const turn1Count = traces.length; + framework.pushEvent({ + type: 'external-message', + source: 'test', + content: 'wake up', + metadata: {}, + } as unknown as ProcessEvent); + await framework.runUntilIdle(); + + const turn2 = traces.slice(turn1Count).filter((e) => e.type.startsWith('inference:')); + assert.ok(turn2.length > 0, 'second turn produced inference traces'); + for (const e of turn2) { + assert.equal( + (e as { channelId?: string }).channelId, + undefined, + `${e.type} on a channel-less turn must not inherit chan-old`, + ); + } + + await framework.stop(); + }); +}); + +describe('abortInference keepText', () => { + let tempDir: string; + let membrane: MockMembrane; + let holdModule: HoldModule; + + beforeEach(() => { + tempDir = mkdtempSync(join(tmpdir(), 'abort-keeptext-test-')); + membrane = new MockMembrane(); + holdModule = new HoldModule(); + }); + + afterEach(() => { + rmSync(tempDir, { recursive: true, force: true }); + }); + + async function createFramework(): Promise { + return AgentFramework.create({ + storePath: join(tempDir, 'test.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'Test.' }], + modules: [holdModule], + }); + } + + function assistantTexts(framework: AgentFramework): string[] { + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (q: { participant?: string }) => { + messages: Array<{ participant: string; content: Array<{ type: string; text?: string }> }>; + }; + }; + return cm + .queryMessages({ participant: 'assistant' }) + .messages.flatMap((m) => m.content) + .filter((b) => b.type === 'text') + .map((b) => b.text ?? ''); + } + + /** Drive a turn into waiting_for_tools, abort it, settle, and return routed speech. */ + async function runAbortScenario( + abortArg: string | { reason?: string; keepText?: string } | undefined, + ): Promise<{ + framework: AgentFramework; + routed: Array<{ text: string; locus: string | null }>; + aborted: boolean; + traces: Array<{ type: string; channelId?: string }>; + }> { + membrane.pushResponse( + createMockResponse( + [ + { type: 'text', text: 'The full sentence the model intended to say' }, + { type: 'tool_use', id: 'c1', name: 'hold--hold', input: {} }, + ] as ContentBlock[], + 'tool_use', + ), + ); + const framework = await createFramework(); + const routed = stubChannelRegistry(framework); + const traces: Array<{ type: string; channelId?: string }> = []; + framework.onTrace((t) => traces.push(t as never)); + + framework.pushEvent(channelIncoming('chan-live', 'talk to me')); + const idle = framework.runUntilIdle(); + await holdModule.toolStarted; + + const aborted = framework.abortInference('assistant', abortArg as never); + holdModule.releaseTool(); + await idle; + return { framework, routed, aborted, traces }; + } + + it('persists the spoken prefix to context without re-posting live-routed prose', async () => { + // The round's prose was live-routed when the round yielded its tool call, + // so the channel already holds the full sentence. keepText is a spoken + // prefix of that same prose: the abort path must persist it to context + // but must NOT post it again (the double-post this guard exists for). + const keepText = 'The full sentence the mo'; + const { framework, routed, aborted, traces } = await runAbortScenario({ + reason: 'user_speech', + keepText, + }); + + assert.equal(aborted, true, 'abort delivered'); + const texts = assistantTexts(framework); + assert.ok( + texts.includes(keepText), + `context contains the spoken prefix (got: ${JSON.stringify(texts)})`, + ); + assert.ok( + !texts.includes('The full sentence the model intended to say'), + 'the full undelivered sentence is NOT in context', + ); + const livePosts = routed.filter( + (r) => r.text === 'The full sentence the model intended to say', + ); + assert.equal(livePosts.length, 1, 'live-routed prose posted exactly once'); + assert.ok( + !routed.some((r) => r.text === keepText), + 'the abort path does not re-post a prefix the live path already covered', + ); + + // The aborted trace carries the channel (as does the tool-call trace). + const abortedTrace = traces.find((t) => t.type === 'inference:aborted'); + assert.equal(abortedTrace?.channelId, 'chan-live', 'aborted trace carries channelId'); + const yielded = traces.find((t) => t.type === 'inference:tool_calls_yielded'); + assert.equal(yielded?.channelId, 'chan-live', 'tool_calls_yielded trace carries channelId'); + + await framework.stop(); + }); + + it('routes only the suffix of keepText beyond the live-routed prose', async () => { + // A whole-turn-accumulating voice client: keepText spans the live-routed + // round AND continues past it. Only the continuation may be posted. + const keepText = 'The full sentence the model intended to say and then a bit more'; + const { framework, routed } = await runAbortScenario({ reason: 'user_speech', keepText }); + + assert.ok( + routed.some((r) => r.text === 'and then a bit more'), + `only the undelivered suffix is posted (routed: ${JSON.stringify(routed.map((r) => r.text))})`, + ); + assert.ok( + !routed.some((r) => r.text === keepText), + 'the full keepText (overlapping delivered prose) is never posted verbatim', + ); + assert.ok( + assistantTexts(framework).includes(keepText), + 'context still records the full spoken text', + ); + await framework.stop(); + }); + + it('posts diverging keepText whole (per-block voice clients)', async () => { + // The reference client (melodeus) resets its spoken-text accumulator at + // every block_start, so an interruption sends only the CURRENT + // utterance's fragment — text the live path has never posted. It must be + // posted whole, not dropped as a failed prefix match. + const keepText = 'A different fragment from the next block'; + const { framework, routed } = await runAbortScenario({ reason: 'user_speech', keepText }); + + assert.ok( + routed.some((r) => r.text === keepText), + `diverging keepText posted whole (routed: ${JSON.stringify(routed.map((r) => r.text))})`, + ); + await framework.stop(); + }); + + it('without keepText the partial turn is discarded (historical behavior)', async () => { + const { framework, routed, aborted } = await runAbortScenario({ reason: 'user_speech' }); + + assert.equal(aborted, true); + assert.deepEqual(assistantTexts(framework), [], 'no assistant turn persisted'); + // Live prose routing may have delivered round prose before the abort — + // but nothing may be routed BY the abort path itself. The only routed + // text can be the live-routed round prose (full sentence), never a prefix. + for (const r of routed) { + assert.notEqual(r.text, '', 'no empty keepText routing'); + } + + await framework.stop(); + }); + + it('string overload still works (backward compatible)', async () => { + const { framework, aborted } = await runAbortScenario('manual'); + assert.equal(aborted, true, 'string reason accepted'); + assert.deepEqual(assistantTexts(framework), [], 'string form implies no keepText'); + await framework.stop(); + }); + + it('books a user abort as a deliberate cancel, not an inference failure', async () => { + const { framework, traces } = await runAbortScenario({ + reason: 'user_speech', + keepText: 'The full', + }); + + // The stream's follow-up exhausted trace is still emitted (wire compat) + // but marked as a deliberate cancel, which must route AROUND the + // inference-health machinery: no consecutive-failure streak (voice + // barge-ins are routine), and no "[inference-failed] ... nothing was + // sent" chronicle marker — false when keepText was just persisted. + const exhausted = traces.find((t) => t.type === 'inference:exhausted') as + | { errorType?: string } + | undefined; + assert.ok(exhausted, 'stream abort still emits inference:exhausted'); + assert.equal(exhausted?.errorType, 'abort', 'marked as a deliberate cancel'); + + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (q: object) => { + messages: Array<{ content: Array<{ type: string; text?: string }> }>; + }; + }; + const allTexts = cm + .queryMessages({}) + .messages.flatMap((m) => m.content) + .filter((b) => b.type === 'text') + .map((b) => b.text ?? ''); + assert.ok( + allTexts.every((text) => !text.includes('[inference-failed]')), + `no failure marker after an interruption (got: ${JSON.stringify(allTexts)})`, + ); + + await framework.stop(); + }); + + it('a duplicate abort cannot wipe the pending keepText', async () => { + const keepText = 'The full sentence the mo'; + const { framework, aborted } = await (async () => { + membrane.pushResponse( + createMockResponse( + [ + { type: 'text', text: 'The full sentence the model intended to say' }, + { type: 'tool_use', id: 'c1', name: 'hold--hold', input: {} }, + ] as ContentBlock[], + 'tool_use', + ), + ); + const fw = await createFramework(); + stubChannelRegistry(fw); + fw.pushEvent(channelIncoming('chan-live', 'talk to me')); + const idle = fw.runUntilIdle(); + await holdModule.toolStarted; + + const first = fw.abortInference('assistant', { reason: 'user_speech', keepText }); + // A voice client can deliver the same report twice in quick + // succession; the duplicate finds the agent already idle and must not + // disturb the first abort's still-pending keepText. + const second = fw.abortInference('assistant', { reason: 'user_speech', keepText }); + assert.equal(second, false, 'duplicate abort no-ops'); + + holdModule.releaseTool(); + await idle; + return { framework: fw, aborted: first }; + })(); + + assert.equal(aborted, true); + assert.ok( + assistantTexts(framework).includes(keepText), + `keepText survives the duplicate abort (got ${JSON.stringify(assistantTexts(framework))})`, + ); + await framework.stop(); + }); + + it('keepText spanning already-flushed rounds commits only the new suffix', async () => { + membrane.pushResponse( + createMockResponse( + [ + { type: 'text', text: 'Round one prose' }, + { type: 'tool_use', id: 'c1', name: 'hold--hold', input: {} }, + ] as ContentBlock[], + 'tool_use', + ), + ); + membrane.pushResponse( + createMockResponse( + [ + { type: 'text', text: 'and round two continues' }, + { type: 'tool_use', id: 'c2', name: 'hold--hold', input: {} }, + ] as ContentBlock[], + 'tool_use', + ), + ); + const framework = await createFramework(); + const routed = stubChannelRegistry(framework); + framework.pushEvent(channelIncoming('chan-live', 'talk')); + const idle = framework.runUntilIdle(); + await holdModule.toolStarted; // round 1's tool held + const round2Started = holdModule.nextToolStart(); + holdModule.releaseTool(); // round 1 completes → its blocks flush to context + await round2Started; // round 2 streamed; its tool held + + // A whole-activation client reports speech spanning both rounds. The + // context must not receive round 1's prose a second time — only the + // part past what the round flush already committed. + const aborted = framework.abortInference('assistant', { + reason: 'user_speech', + keepText: 'Round one prose and round', + }); + assert.equal(aborted, true); + holdModule.releaseTool(); + await idle; + + const texts = assistantTexts(framework); + assert.equal( + texts.filter((text) => text.includes('Round one prose')).length, + 1, + `round-1 prose committed exactly once (got ${JSON.stringify(texts)})`, + ); + assert.ok(texts.includes('and round'), 'the unflushed suffix is committed for the abort'); + assert.ok( + !texts.includes('Round one prose and round'), + 'the spanning keepText is not committed verbatim', + ); + // The channel post side is deduped against live-routed prose, which + // already covers the whole report — nothing further posted by the abort. + assert.ok( + !routed.some((r) => r.text === 'and round'), + 'abort path does not post a suffix the live path already covered', + ); + await framework.stop(); + }); +}); diff --git a/test/prose-segments.test.ts b/test/prose-segments.test.ts index 015b159..6a1dba7 100644 --- a/test/prose-segments.test.ts +++ b/test/prose-segments.test.ts @@ -1,7 +1,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import type { ContentBlock } from '@animalabs/membrane'; -import { splitProseSegments } from '../src/prose-segments.js'; +import { splitProseSegments, undeliveredSuffix, isWhitespaceInsensitivePrefix } from '../src/prose-segments.js'; const textBlock = (text: string): ContentBlock => ({ type: 'text', text } as ContentBlock); const toolUse = (id: string): ContentBlock => @@ -60,3 +60,61 @@ test('a plain no-tool turn yields a single segment (unchanged behaviour)', () => test('a tool-only turn (no prose) yields no segments', () => { assert.deepEqual(splitProseSegments([toolUse('t1'), toolResult('t1')]), []); }); + +// ── undeliveredSuffix: matching a voice client's spokenText against +// live-routed prose (see the abort keepText path in framework.ts) ── + +test('undeliveredSuffix: spoken fully covered by delivered → null (nothing further to post)', () => { + assert.equal(undeliveredSuffix('The full sentence the mo', 'The full sentence the model intended'), null); + assert.equal(undeliveredSuffix('Same text.', 'Same text.'), null); +}); + +test('undeliveredSuffix: spoken extends past delivered → the raw suffix', () => { + assert.equal(undeliveredSuffix('Hello there general', 'Hello there'), 'general'); +}); + +test('undeliveredSuffix: whitespace differences never break alignment', () => { + // Live-routed segments are joined with '\n'; the voice stream keeps its own + // spacing. Only the non-whitespace character sequence matters. + assert.equal(undeliveredSuffix('One two three four', 'One\ntwo\nthree'), 'four'); + assert.equal(undeliveredSuffix('One two', 'One two'), null); +}); + +test('undeliveredSuffix: diverging spoken text is returned whole (per-block clients)', () => { + // A client that resets its accumulator per block sends only the current + // utterance's fragment — never posted, so all of it is undelivered. + assert.equal( + undeliveredSuffix('A new fragment entirely', 'The prose the live path posted'), + 'A new fragment entirely', + ); +}); + +test('undeliveredSuffix: suffix that is only whitespace → null', () => { + assert.equal(undeliveredSuffix('Hello there ', 'Hello there'), null); +}); + +test('undeliveredSuffix: astral characters compare safely at the boundary', () => { + assert.equal(undeliveredSuffix('Great 🎉 and onward', 'Great 🎉'), 'and onward'); + assert.equal(undeliveredSuffix('Great 🎉', 'Great 🎉'), null); +}); + +// ── isWhitespaceInsensitivePrefix: the staleness-guard predicate ── + +test('isWhitespaceInsensitivePrefix: matches across whitespace differences', () => { + assert.equal(isWhitespaceInsensitivePrefix('Hello there', 'Hello\nthere general'), true); + assert.equal(isWhitespaceInsensitivePrefix('Hello there ', 'Hello there'), true); +}); + +test('isWhitespaceInsensitivePrefix: rejects diverging and over-long prefixes', () => { + assert.equal(isWhitespaceInsensitivePrefix('Something else', 'Hello there'), false); + assert.equal( + isWhitespaceInsensitivePrefix('Hello there general', 'Hello there'), + false, + 'a report longer than what was streamed cannot be a prefix of it', + ); +}); + +test('isWhitespaceInsensitivePrefix: empty prefix trivially matches', () => { + assert.equal(isWhitespaceInsensitivePrefix('', 'anything'), true); + assert.equal(isWhitespaceInsensitivePrefix('', ''), true); +}); diff --git a/test/voice-relay-client.test.ts b/test/voice-relay-client.test.ts new file mode 100644 index 0000000..ee0e003 --- /dev/null +++ b/test/voice-relay-client.test.ts @@ -0,0 +1,1113 @@ +/** + * RelayClientModule — the outbound relay client (framework agents → external + * TTS relay, ChapterX-style). + * + * Unit half: an in-process mock /bot server covers auth, message forwarding + * with the connection's bot identity, heartbeat tolerance, interruption → + * abortInference(keepText) with the staleness guard and channel-addressing + * rules, reconnect with backoff (including the replaced-connection stop and + * the stability-gated backoff reset), channel-tracking bounds, and + * drop-when-down. + * + * E2E half (skipped when the reference repo is absent): spawns the REAL + * melodeus-tts-relay, runs a REAL AgentFramework with the module installed, + * and asserts a VoiceClientSim on the relay's /tts side hears a framework + * agent's streamed turn — then interrupts it back through the same path. + */ + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createServer, type Server } from 'node:http'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { WebSocketServer, type WebSocket as WsSocket } from 'ws'; + +import { AgentFramework } from '../src/framework.js'; +import { RelayClientModule } from '../src/modules/voice-relay/index.js'; +import type { RelayLogger } from '../src/modules/voice-relay/types.js'; +import type { TraceEvent } from '../src/types/trace.js'; +import type { ModuleContext, Module, ProcessState, EventResponse } from '../src/types/module.js'; +import type { ProcessEvent, ToolCall, ToolResult, ToolDefinition } from '../src/types/events.js'; +import type { ContentBlock } from '@animalabs/membrane'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +import { + VoiceClientSim, + TOKENS, + writeRelayConfig, + referenceRelayAvailable, + spawnReferenceRelay, +} from './helpers/voice-relay-sims.js'; + +const silentLogger: RelayLogger = { debug() {}, info() {}, warn() {}, error() {} }; + +// --------------------------------------------------------------------------- +// Mock /bot relay server +// --------------------------------------------------------------------------- + +interface MockBotServer { + url: string; + connections: number; + received: Array>; + authedSockets: WsSocket[]; + rejectAuth: boolean; + close(): Promise; +} + +async function startMockBotServer(): Promise { + const http: Server = createServer(); + const wss = new WebSocketServer({ noServer: true }); + const state: MockBotServer = { + url: '', + connections: 0, + received: [], + authedSockets: [], + rejectAuth: false, + close: async () => { + for (const ws of state.authedSockets) ws.terminate(); + wss.close(); + await new Promise((r) => http.close(r)); + }, + }; + + http.on('upgrade', (req, socket, head) => { + if (req.url !== '/bot') return socket.destroy(); + wss.handleUpgrade(req, socket, head, (ws) => { + state.connections++; + ws.on('message', (data) => { + const msg = JSON.parse(data.toString()) as Record; + if (msg.type === 'auth') { + if (state.rejectAuth) { + ws.send(JSON.stringify({ type: 'auth_error', error: 'Invalid token' })); + ws.close(4003, 'Invalid token'); + return; + } + state.authedSockets.push(ws); + ws.send(JSON.stringify({ type: 'auth_ok' })); + return; + } + state.received.push(msg); + }); + }); + }); + + await new Promise((r) => http.listen(0, '127.0.0.1', r)); + const addr = http.address(); + if (addr === null || typeof addr === 'string') throw new Error('no port'); + state.url = `ws://127.0.0.1:${addr.port}`; + return state; +} + +/** Minimal ModuleContext: the client module only uses onTrace. */ +function stubCtx(): { ctx: ModuleContext; emit: (e: TraceEvent) => void } { + const listeners: Array<(e: TraceEvent) => void> = []; + const ctx = { + onTrace: (l: (e: TraceEvent) => void) => { + listeners.push(l); + return () => { + const i = listeners.indexOf(l); + if (i >= 0) listeners.splice(i, 1); + }; + }, + } as unknown as ModuleContext; + return { ctx, emit: (e) => [...listeners].forEach((l) => l(e)) }; +} + +function fakeFramework(): { + framework: AgentFramework; + aborts: Array<{ agentName: string; reason?: string; keepText?: string }>; +} { + const aborts: Array<{ agentName: string; reason?: string; keepText?: string }> = []; + const framework = { + abortInference: (agentName: string, opts?: string | { reason?: string; keepText?: string }) => { + const o = typeof opts === 'string' ? { reason: opts } : opts ?? {}; + aborts.push({ agentName, reason: o.reason, keepText: o.keepText }); + return true; + }, + } as unknown as AgentFramework; + return { framework, aborts }; +} + +async function waitFor(cond: () => boolean, ms = 3000, what = 'condition'): Promise { + const deadline = Date.now() + ms; + while (Date.now() < deadline) { + if (cond()) return; + await new Promise((r) => setTimeout(r, 20)); + } + throw new Error(`Timed out waiting for ${what}`); +} + +const T0 = 1_710_900_000_000; + +function startedTrace(agentName: string, channelId?: string): TraceEvent { + return { type: 'inference:started', agentName, channelId, timestamp: T0 } as TraceEvent; +} + +// --------------------------------------------------------------------------- +// Unit: mock server +// --------------------------------------------------------------------------- + +test('relay client auths and forwards messages under its own bot identity', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + username: 'Opus 4.5', + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + emit(startedTrace('assistant', 'chan-1')); + emit({ + type: 'inference:content_block', + agentName: 'assistant', + channelId: 'chan-1', + phase: 'block_start', + blockType: 'text', + blockIndex: 0, + timestamp: T0, + } as TraceEvent); + emit({ + type: 'inference:tokens', + agentName: 'assistant', + channelId: 'chan-1', + content: 'Hello ', + blockType: 'text', + blockIndex: 0, + timestamp: T0, + } as TraceEvent); + emit({ + type: 'inference:tokens', + agentName: 'assistant', + channelId: 'chan-1', + content: 'world', + blockType: 'text', + blockIndex: 0, + timestamp: T0, + } as TraceEvent); + emit({ + type: 'inference:content_block', + agentName: 'assistant', + channelId: 'chan-1', + phase: 'block_complete', + blockType: 'text', + blockIndex: 0, + timestamp: T0, + } as TraceEvent); + emit({ type: 'inference:completed', agentName: 'assistant', channelId: 'chan-1', durationMs: 5, timestamp: T0 } as TraceEvent); + + await waitFor(() => server.received.length >= 6, 3000, 'relay messages'); + const types = server.received.map((m) => m.type); + assert.deepEqual(types, [ + 'activation_start', + 'block_start', + 'chunk', + 'chunk', + 'block_complete', + 'activation_end', + ]); + for (const m of server.received) { + assert.equal(m.botId, 'opus45', 'botId on the wire is the connection identity, not agentName'); + assert.equal(m.username, 'Opus 4.5'); + assert.equal(m.channelId, 'chan-1'); + } + const complete = server.received.find((m) => m.type === 'block_complete'); + assert.equal(complete?.content, 'Hello world'); + const end = server.received.find((m) => m.type === 'activation_end'); + assert.equal(end?.reason, 'complete'); + +}); + +test('relay client ignores heartbeats and drops relay traffic while down', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + reconnectInitialMs: 50_000, // long: keep it down after close + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + server.authedSockets[0].send(JSON.stringify({ type: 'heartbeat', timestamp: Date.now() })); + await new Promise((r) => setTimeout(r, 100)); + assert.equal(module.isConnected, true, 'heartbeat does not disturb the connection'); + + // Sever, then emit traces while down: dropped without error, none delivered. + server.authedSockets[0].terminate(); + await waitFor(() => !module.isConnected, 3000, 'disconnect noticed'); + emit(startedTrace('assistant', 'chan-1')); + await new Promise((r) => setTimeout(r, 100)); + assert.equal(server.received.length, 0); + +}); + +test('relay client reconnects with backoff and re-authenticates', async (t) => { + const server = await startMockBotServer(); + const { ctx } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + reconnectInitialMs: 30, + reconnectMaxMs: 200, + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'first auth'); + assert.equal(server.connections, 1); + + server.authedSockets[0].terminate(); + await waitFor(() => server.connections >= 2 && module.isConnected, 3000, 'reconnect + re-auth'); + +}); + +test('relay client keeps retrying after auth rejection', async (t) => { + const server = await startMockBotServer(); + server.rejectAuth = true; + const { ctx } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'bad', + reconnectInitialMs: 20, + reconnectMaxMs: 100, + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => server.connections >= 3, 3000, 'repeated attempts'); + assert.equal(module.isConnected, false); + + server.rejectAuth = false; + await waitFor(() => module.isConnected, 3000, 'recovers once auth allowed'); + +}); + +test('interruption from the relay maps to abortInference with keepText', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + logger: silentLogger, + }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + emit(startedTrace('assistant', 'chan-9')); + emit({ + type: 'inference:tokens', + agentName: 'assistant', + channelId: 'chan-9', + content: 'Hey! Yes, I can hear you loud and clear', + blockType: 'text', + blockIndex: 0, + timestamp: T0, + } as TraceEvent); + await waitFor(() => server.received.length >= 2, 3000, 'turn streamed'); + server.authedSockets[0].send( + JSON.stringify({ + type: 'interruption', + channelId: 'chan-9', + spokenText: 'Hey! Yes, I can hear', + reason: 'user_speech', + timestamp: T0, + }), + ); + + await waitFor(() => aborts.length === 1, 3000, 'abortInference call'); + assert.deepEqual(aborts[0], { + agentName: 'assistant', + reason: 'user_speech', + keepText: 'Hey! Yes, I can hear', + }); + + // Unknown channel with several candidates → dropped, no spurious abort. + emit(startedTrace('other-agent', 'chan-other')); + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-unknown', spokenText: 'x', reason: 'manual', timestamp: T0 }), + ); + await new Promise((r) => setTimeout(r, 150)); + assert.equal(aborts.length, 1); + +}); + +test('interruption naming an unknown channel is dropped even with a single candidate', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + // One tracked agent — but the interruption names a channel we never + // streamed to. Guessing here could abort an unrelated turn. + emit(startedTrace('assistant', 'chan-a')); + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-b', spokenText: 'x', reason: 'manual', timestamp: T0 }), + ); + await new Promise((r) => setTimeout(r, 150)); + assert.equal(aborts.length, 0, 'unknown channel never falls back to the single candidate'); + +}); + +test('interruption without a channel falls back to the single tracked agent', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + emit(startedTrace('assistant', 'chan-a')); + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', spokenText: '', reason: 'manual', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'fallback abort'); + assert.equal(aborts[0].agentName, 'assistant'); + assert.equal(aborts[0].keepText, undefined, 'empty spokenText carries no keepText'); + +}); + +test('stale interruption (spokenText from a previous utterance) is dropped', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + const chunk = (content: string, blockIndex: number, blockType = 'text'): TraceEvent => + ({ type: 'inference:tokens', agentName: 'assistant', channelId: 'chan-7', content, blockType, blockIndex, timestamp: T0 } as TraceEvent); + const blockStart = (blockIndex: number): TraceEvent => + ({ type: 'inference:content_block', agentName: 'assistant', channelId: 'chan-7', phase: 'block_start', blockType: 'text', blockIndex, timestamp: T0 } as TraceEvent); + + // Utterance 1 streams; a matching report interrupts it. + emit(startedTrace('assistant', 'chan-7')); + emit(blockStart(0)); + emit(chunk('Hello there ', 0)); + emit(chunk('general Kenobi', 0)); + await waitFor(() => server.received.length >= 4, 3000, 'utterance 1 streamed'); + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-7', spokenText: 'Hello there gen', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'matching report aborts'); + assert.equal(aborts[0].keepText, 'Hello there gen'); + + // Utterance 2 = a NEW activation; the old report races in again — stale, + // dropped. (A new block within the SAME activation is not a new utterance: + // whole-activation clients legitimately report earlier blocks' text — see + // the iOS-style test below.) + emit(startedTrace('assistant', 'chan-7')); + emit(blockStart(1)); + emit(chunk('Fresh words now', 1)); + await waitFor(() => server.received.length >= 7, 3000, 'utterance 2 streamed'); + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-7', spokenText: 'Hello there gen', reason: 'user_speech', timestamp: T0 }), + ); + await new Promise((r) => setTimeout(r, 150)); + assert.equal(aborts.length, 1, 'stale report does not abort the new utterance'); + + // A report matching the CURRENT utterance still works. + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-7', spokenText: 'Fresh words', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 2, 3000, 'current-utterance report aborts'); + assert.equal(aborts[1].keepText, 'Fresh words'); + +}); + +test('channel tracking is bounded; evicted channels no longer address interruptions', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + for (let i = 0; i < 300; i++) emit(startedTrace(`agent-${i % 3}`, `chan-${i}`)); + assert.equal(module.trackedChannelCount, 256, 'tracking bounded at 256 channels'); + + // chan-0 was evicted (oldest); chan-299 is still tracked. + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-0', spokenText: '', reason: 'manual', timestamp: T0 }), + ); + await new Promise((r) => setTimeout(r, 100)); + assert.equal(aborts.length, 0, 'evicted channel dropped'); + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-299', spokenText: '', reason: 'manual', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'tracked channel aborts'); + assert.equal(aborts[0].agentName, 'agent-2'); + +}); + +test('relay client does not reconnect after the relay replaces its connection', async (t) => { + const server = await startMockBotServer(); + const { ctx } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + reconnectInitialMs: 20, // fast enough that a reconnect WOULD show up below + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + server.authedSockets[0].close(1000, 'Replaced by new connection'); + await waitFor(() => !module.isConnected, 3000, 'replacement noticed'); + await new Promise((r) => setTimeout(r, 200)); + assert.equal(server.connections, 1, 'no reconnect after being replaced'); + +}); + +test('backoff resets only after the connection stays authenticated', async (t) => { + const server = await startMockBotServer(); + const { ctx } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + reconnectInitialMs: 30, + reconnectMaxMs: 500, + // Wide enough that a CPU-starved runner cannot let the stability timer + // fire between the second re-auth and the growth assertion below. + backoffResetAfterMs: 800, + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'first auth'); + assert.equal(module.currentReconnectDelayMs, 30); + + // Two quick auth-then-drop cycles: the backoff must keep growing because + // no connection survived the stability window. + server.authedSockets[0].terminate(); + await waitFor(() => module.isConnected && server.connections >= 2, 3000, 'reconnect 1'); + server.authedSockets[1].terminate(); + await waitFor(() => module.isConnected && server.connections >= 3, 3000, 'reconnect 2'); + assert.ok(module.currentReconnectDelayMs >= 120, `backoff grew (got ${module.currentReconnectDelayMs})`); + + // Stay connected past the stability window: backoff returns to the floor. + await waitFor(() => module.currentReconnectDelayMs === 30, 3000, 'stability reset'); + +}); + +test('watchdog: a silent link is presumed dead and re-dialed', async (t) => { + const server = await startMockBotServer(); + const { ctx } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + reconnectInitialMs: 30, + heartbeatTimeoutMs: 400, + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + assert.equal(server.connections, 1); + + // The mock server never heartbeats (the real relay does, every ~2s), so + // the watchdog must tear the link down and the client must re-dial. + await waitFor(() => server.connections >= 2, 5000, 'watchdog re-dial'); +}); + +test('watchdog: heartbeats keep the link alive', async (t) => { + const server = await startMockBotServer(); + const { ctx } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + heartbeatTimeoutMs: 400, + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + const hb = setInterval(() => { + for (const ws of server.authedSockets) { + try { + ws.send(JSON.stringify({ type: 'heartbeat', timestamp: Date.now() })); + } catch { + // socket may be mid-teardown; the assertion below still judges + } + } + }, 100); + t.after(() => clearInterval(hb)); + + await new Promise((r) => setTimeout(r, 900)); + assert.equal(server.connections, 1, 'no re-dial while heartbeats flow'); + assert.equal(module.isConnected, true); +}); + +test('malformed frames (null, primitives, bad JSON) do not crash the client', async (t) => { + const server = await startMockBotServer(); + const { ctx } = stubCtx(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + for (const frame of ['null', '123', '"x"', '[1,2]', 'not json at all']) { + server.authedSockets[0].send(frame); + } + await new Promise((r) => setTimeout(r, 150)); + assert.equal(module.isConnected, true, 'client survives hostile frames'); +}); + +test('unverifiable spokenText aborts the turn but never dictates keepText', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + // The turn starts BEFORE the socket authenticates: the tracker (a trace + // listener) records the channel, but the activation_start is dropped at + // the socket check, so no accumulator entry exists — the report below is + // genuinely unverifiable. It may stop the turn but its text must not be + // committed as words the agent said. + emit(startedTrace('assistant', 'chan-2')); + await waitFor(() => module.isConnected, 3000, 'auth'); + + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-2', spokenText: 'words we never streamed', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'abort goes through'); + assert.equal(aborts[0].agentName, 'assistant'); + assert.equal(aborts[0].keepText, undefined, 'unverifiable text is not kept'); +}); + +test('a non-empty report while the current utterance has voiced nothing is dropped as stale', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + // Turn N streamed and finished; its late report arrives only after turn + // N+1's activation_start has been SENT (accumulator present but empty — + // N+1 is still thinking). Real clients never report non-empty spokenText + // for an utterance that voiced nothing, so this can only describe turn N + // and must not cut off N+1. + const ev = (e: Record): TraceEvent => + ({ agentName: 'assistant', channelId: 'chan-3', timestamp: T0, ...e } as unknown as TraceEvent); + emit(startedTrace('assistant', 'chan-3')); + emit(ev({ type: 'inference:tokens', content: 'Turn one words', blockType: 'text', blockIndex: 0 })); + emit(ev({ type: 'inference:completed', durationMs: 5 })); + emit(startedTrace('assistant', 'chan-3')); // N+1: activation_start sent, no text yet + await waitFor(() => server.received.filter((m) => m.type === 'activation_start').length >= 2, 3000, 'both activations sent'); + + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-3', spokenText: 'Turn one words', reason: 'user_speech', timestamp: T0 }), + ); + await new Promise((r) => setTimeout(r, 150)); + assert.equal(aborts.length, 0, 'the stale report must not abort the fresh turn'); + + // Once N+1 voices text, a matching report lands normally. + emit(ev({ type: 'inference:tokens', content: 'Turn two words', blockType: 'text', blockIndex: 0 })); + await waitFor(() => server.received.filter((m) => m.type === 'chunk').length >= 2, 3000, 'turn two streamed'); + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-3', spokenText: 'Turn two', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'the live turn is interruptible'); + assert.equal(aborts[0].keepText, 'Turn two'); +}); + +test("narrator-markup asterisks are ignored when matching a client's report", async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + emit(startedTrace('assistant', 'chan-4')); + emit({ + type: 'inference:tokens', + agentName: 'assistant', + channelId: 'chan-4', + content: '*looks up* Hello there, how are you', + blockType: 'text', + blockIndex: 0, + timestamp: T0, + } as TraceEvent); + await waitFor(() => server.received.some((m) => m.type === 'chunk'), 3000, 'turn streamed'); + + // The iOS client voices `*action*` spans via its narrator voice and + // reports them WITHOUT the asterisks (segments trimmed and concatenated), + // so the report for the text above arrives as "looks upHello there". + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-4', spokenText: 'looks upHello there', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'narrated turn is interruptible'); + assert.equal(aborts[0].keepText, 'looks upHello there'); +}); + +test('the fatal replaced-connection stop also tears down the trace subscriptions', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + reconnectInitialMs: 20, + logger: silentLogger, + }); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + emit(startedTrace('assistant', 'chan-1')); + assert.equal(module.trackedChannelCount, 1); + + server.authedSockets[0].close(1000, 'Replaced by new connection'); + await waitFor(() => !module.isConnected, 3000, 'replacement noticed'); + + // A permanently-down module must not keep tracking or translating: the + // maps are cleared and later traces are ignored entirely. + assert.equal(module.trackedChannelCount, 0, 'tracking cleared on the fatal stop'); + const before = server.received.length; + emit(startedTrace('assistant', 'chan-2')); + await new Promise((r) => setTimeout(r, 100)); + assert.equal(module.trackedChannelCount, 0, 'tracker unsubscribed'); + assert.equal(server.received.length, before); +}); + +test('whole-activation report spanning blocks is accepted (iOS-style client)', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + const ev = (e: Record): TraceEvent => + ({ agentName: 'assistant', channelId: 'chan-5', timestamp: T0, ...e } as unknown as TraceEvent); + emit(startedTrace('assistant', 'chan-5')); + emit(ev({ type: 'inference:content_block', phase: 'block_start', blockType: 'text', blockIndex: 0 })); + emit(ev({ type: 'inference:tokens', content: 'First sentence. ', blockType: 'text', blockIndex: 0 })); + emit(ev({ type: 'inference:content_block', phase: 'block_start', blockType: 'text', blockIndex: 1 })); + emit(ev({ type: 'inference:tokens', content: 'Second thought', blockType: 'text', blockIndex: 1 })); + await waitFor(() => server.received.length >= 5, 3000, 'both blocks streamed'); + + // The iOS client accumulates spokenText across the WHOLE activation, so a + // legitimate mid-turn report starts with block 0's text even though the + // stream is already in block 1. It must abort, and keep what was heard. + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-5', spokenText: 'First sentence. Sec', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'whole-activation report accepted'); + assert.equal(aborts[0].keepText, 'First sentence. Sec'); +}); + +test('a late report for a channel the agent has left cannot abort its new turn', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ url: server.url, botId: 'opus45', token: 'tok', logger: silentLogger }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + const ev = (channelId: string, e: Record): TraceEvent => + ({ agentName: 'assistant', channelId, timestamp: T0, ...e } as unknown as TraceEvent); + + // Turn 1 on chan-a streams and completes. + emit(startedTrace('assistant', 'chan-a')); + emit(ev('chan-a', { type: 'inference:tokens', content: 'Old turn words', blockType: 'text', blockIndex: 0 })); + emit(ev('chan-a', { type: 'inference:completed', durationMs: 5 })); + // Turn 2 on chan-b is now streaming. + emit(startedTrace('assistant', 'chan-b')); + emit(ev('chan-b', { type: 'inference:tokens', content: 'New turn words', blockType: 'text', blockIndex: 0 })); + // Wire: activation_start, chunk, activation_end (turn 1) + activation_start, chunk (turn 2). + await waitFor(() => server.received.length >= 5, 3000, 'both turns streamed'); + + // A late chan-a report (voice audio outlives the turn) still matches + // chan-a's retained accumulator — but the agent is mid-turn on chan-b now, + // so aborting would kill the wrong turn. Must be dropped. + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-a', spokenText: 'Old turn', reason: 'user_speech', timestamp: T0 }), + ); + await new Promise((r) => setTimeout(r, 150)); + assert.equal(aborts.length, 0, 'late report for the finished turn is dropped'); + + // A report for the channel the agent is ACTUALLY on still works. + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-b', spokenText: 'New turn', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, 'active-channel report aborts'); + assert.equal(aborts[0].keepText, 'New turn'); +}); + +test('agents filter scopes streaming and interruption addressing', async (t) => { + const server = await startMockBotServer(); + const { ctx, emit } = stubCtx(); + const { framework, aborts } = fakeFramework(); + const module = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + agents: ['mine'], + logger: silentLogger, + }); + module.bind(framework); + await module.start(ctx); + t.after(async () => { + await module.stop(); + await server.close(); + }); + await waitFor(() => module.isConnected, 3000, 'auth'); + + const ev = (agentName: string, channelId: string, e: Record): TraceEvent => + ({ agentName, channelId, timestamp: T0, ...e } as unknown as TraceEvent); + emit(startedTrace('mine', 'chan-m')); + emit(ev('mine', 'chan-m', { type: 'inference:tokens', content: 'Mine speaking', blockType: 'text', blockIndex: 0 })); + emit(startedTrace('other', 'chan-o')); + emit(ev('other', 'chan-o', { type: 'inference:tokens', content: 'Other speaking', blockType: 'text', blockIndex: 0 })); + await waitFor(() => server.received.length >= 2, 3000, "the filtered-in agent's turn streamed"); + await new Promise((r) => setTimeout(r, 100)); + + assert.ok( + server.received.every((m) => m.channelId === 'chan-m'), + `only the listed agent streams (got ${JSON.stringify(server.received.map((m) => m.channelId))})`, + ); + + // The unlisted agent's channel was never tracked → its interruption drops. + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-o', spokenText: '', reason: 'manual', timestamp: T0 }), + ); + await new Promise((r) => setTimeout(r, 100)); + assert.equal(aborts.length, 0); + + server.authedSockets[0].send( + JSON.stringify({ type: 'interruption', channelId: 'chan-m', spokenText: 'Mine speak', reason: 'user_speech', timestamp: T0 }), + ); + await waitFor(() => aborts.length === 1, 3000, "the listed agent's interruption lands"); + assert.equal(aborts[0].agentName, 'mine'); +}); + +// --------------------------------------------------------------------------- +// Closed loop on CI: real framework + mock /bot server (no relay checkout) +// --------------------------------------------------------------------------- + +/** One-shot tool that blocks until released, pinning the turn mid-flight. */ +class HoldToolModule implements Module { + readonly name = 'hold'; + toolStarted: Promise; + private signalStart!: () => void; + private release!: (r: ToolResult) => void; + private held: Promise; + + constructor() { + this.toolStarted = new Promise((r) => (this.signalStart = r)); + this.held = new Promise((r) => (this.release = r)); + } + + releaseTool(): void { + this.release({ success: true, data: { ok: true } }); + } + + async start(_ctx: ModuleContext): Promise {} + async stop(): Promise {} + getTools(): ToolDefinition[] { + return [ + { name: 'hold', description: 'Blocks until released', inputSchema: { type: 'object', properties: {} } }, + ]; + } + async handleToolCall(_call: ToolCall): Promise { + this.signalStart(); + return this.held; + } + async onProcess(_e: ProcessEvent, _s: ProcessState): Promise { + return {}; + } +} + +test('closed loop: a relay interruption aborts the live turn; the wire sees activation_end(abort) and context keeps the prefix', async (t) => { + const server = await startMockBotServer(); + const tempDir = mkdtempSync(join(tmpdir(), 'relay-loop-')); + const membrane = new MockMembrane(); + membrane.pushResponse( + createMockResponse( + [ + { type: 'text', text: 'The weather is sunny today' }, + { type: 'tool_use', id: 'c1', name: 'hold--hold', input: {} }, + ] as ContentBlock[], + 'tool_use', + ), + ); + const hold = new HoldToolModule(); + const clientModule = new RelayClientModule({ + url: server.url, + botId: 'opus45', + token: 'tok', + logger: silentLogger, + }); + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'loop.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'Test.' }], + modules: [hold as unknown as Module, clientModule as unknown as Module], + }); + clientModule.bind(framework); + t.after(async () => { + await framework.stop(); + await server.close(); + rmSync(tempDir, { recursive: true, force: true }); + }); + await waitFor(() => clientModule.isConnected, 5000, 'module authed against mock relay'); + + framework.pushEvent({ + type: 'mcpl:channel-incoming', + serverId: 'test-server', + channelId: 'chan-loop', + messageId: 'm-loop-1', + author: { id: 'u1', name: 'Nick' }, + content: [{ type: 'text', text: 'what is the weather' }], + timestamp: new Date().toISOString(), + triggerInference: true, + } as unknown as ProcessEvent); + const idle = framework.runUntilIdle(); + await hold.toolStarted; + await waitFor(() => server.received.some((m) => m.type === 'chunk'), 5000, 'turn streamed to the relay'); + + server.authedSockets[0].send( + JSON.stringify({ + type: 'interruption', + channelId: 'chan-loop', + spokenText: 'The weather', + reason: 'user_speech', + timestamp: Date.now(), + }), + ); + await waitFor( + () => server.received.some((m) => m.type === 'activation_end' && m.reason === 'abort'), + 5000, + 'abort reaches the wire as activation_end', + ); + hold.releaseTool(); + await idle; + + const cm = framework.getAgent('assistant')!.getContextManager() as unknown as { + queryMessages: (q: { participant?: string }) => { + messages: Array<{ content: Array<{ type: string; text?: string }> }>; + }; + }; + const texts = cm + .queryMessages({ participant: 'assistant' }) + .messages.flatMap((m) => m.content) + .filter((b) => b.type === 'text') + .map((b) => b.text ?? ''); + assert.ok(texts.includes('The weather'), `context keeps the spoken prefix (got ${JSON.stringify(texts)})`); + assert.ok( + !texts.includes('The weather is sunny today'), + 'the full unspoken sentence is not committed', + ); +}); + +// --------------------------------------------------------------------------- +// E2E: real relay + real framework +// --------------------------------------------------------------------------- + +test( + 'e2e: a voice client hears a framework agent through the real relay, and interrupts it', + { skip: referenceRelayAvailable() === null ? 'reference relay repo unavailable' : false }, + async (t) => { + const tempDir = mkdtempSync(join(tmpdir(), 'relay-client-e2e-')); + // Cleanup via t.after so a failure at ANY point — including framework + // creation — cannot orphan the spawned relay child or the temp dir. + let relay: Awaited> | null = null; + let frameworkRef: AgentFramework | null = null; + t.after(async () => { + await frameworkRef?.stop(); + await relay?.stop(); + rmSync(tempDir, { recursive: true, force: true }); + }); + relay = await spawnReferenceRelay(writeRelayConfig(tempDir)); + const membrane = new MockMembrane(); + const clientModule = new RelayClientModule({ + url: relay.url, + botId: 'Opus45', + token: TOKENS.bot, + username: 'Opus 4.5', + reconnectInitialMs: 100, + logger: silentLogger, + }); + + const framework = await AgentFramework.create({ + storePath: join(tempDir, 'e2e.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test-model', systemPrompt: 'Test.' }], + modules: [clientModule as unknown as Module], + }); + frameworkRef = framework; + clientModule.bind(framework); + + // Spy on abortInference to observe the interruption round-trip without + // needing a stream to still be in flight when the (async) cut arrives. + const aborts: Array<{ agentName: string; keepText?: string }> = []; + const realAbort = framework.abortInference.bind(framework); + framework.abortInference = ((agentName: string, opts?: string | { reason?: string; keepText?: string }) => { + const o = typeof opts === 'string' ? { reason: opts } : opts ?? {}; + aborts.push({ agentName, keepText: o.keepText }); + return realAbort(agentName, opts as never); + }) as typeof framework.abortInference; + + await waitFor(() => clientModule.isConnected, 10_000, 'module authed against real relay'); + + { + const sim = new VoiceClientSim(relay.url); + await sim.connect(); + const authReply = await sim.auth({ clientId: 'e2e-voice', token: TOKENS.client }); + assert.equal(authReply.type, 'auth_ok'); + await sim.subscribe(['chan-e2e']); + + membrane.pushResponse( + createMockResponse([{ type: 'text', text: 'Hello from connectome' }] as ContentBlock[]), + ); + framework.pushEvent({ + type: 'mcpl:channel-incoming', + serverId: 'test-server', + channelId: 'chan-e2e', + messageId: 'm-e2e-1', + author: { id: 'u1', name: 'Nick' }, + content: [{ type: 'text', text: 'hey assistant' }], + timestamp: new Date().toISOString(), + triggerInference: true, + } as unknown as ProcessEvent); + await framework.runUntilIdle(); + + // Collect the streamed relay messages at the voice client until activation_end. + const relayMessages: Array> = []; + while (relayMessages.length === 0 || relayMessages[relayMessages.length - 1].type !== 'activation_end') { + relayMessages.push(await sim.next(5000)); + if (relayMessages.length > 50) throw new Error(`stream never ended: ${JSON.stringify(relayMessages)}`); + } + + assert.equal(relayMessages[0].type, 'activation_start'); + assert.equal(relayMessages[0].botId, 'Opus45'); + assert.equal(relayMessages[0].channelId, 'chan-e2e'); + const chunkText = relayMessages + .filter((m) => m.type === 'chunk' && m.blockType === 'text') + .map((m) => m.text) + .join(''); + assert.equal(chunkText, 'Hello from connectome'); + const blockComplete = relayMessages.find((m) => m.type === 'block_complete'); + assert.equal(blockComplete?.content, 'Hello from connectome'); + const end = relayMessages[relayMessages.length - 1]; + assert.equal(end.reason, 'complete'); + assert.equal(end.username, 'Opus 4.5'); + + // Interruption: voice client reports a cut through the real relay. + sim.send({ + type: 'interruption', + botId: 'Opus45', + channelId: 'chan-e2e', + spokenText: 'Hello from', + reason: 'user_speech', + timestamp: Date.now(), + }); + await waitFor(() => aborts.length >= 1, 5000, 'interruption reached abortInference'); + assert.equal(aborts[0].agentName, 'assistant'); + assert.equal(aborts[0].keepText, 'Hello from'); + + sim.close(); + } + }, +); diff --git a/test/voice-relay-trace-bridge.test.ts b/test/voice-relay-trace-bridge.test.ts new file mode 100644 index 0000000..95544aa --- /dev/null +++ b/test/voice-relay-trace-bridge.test.ts @@ -0,0 +1,171 @@ +/** + * InferenceTraceBridge — the trace-to-relay translation table. + * + * Synthetic framework traces in, relay wire messages out: identity + * resolution, visible derivation (text chunks are voiced, thinking is not), + * block-content accumulation for block_complete, terminal-trace mapping to + * activation_end reasons, channel-less drops, and unsubscribe on stop(). + * + * The module-level behavior around the bridge (socket delivery, interruption + * addressing, staleness guard) is covered in voice-relay-client.test.ts. + */ + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { InferenceTraceBridge } from '../src/modules/voice-relay/index.js'; +import type { BotStreamMessage, RelayLogger } from '../src/modules/voice-relay/types.js'; +import type { TraceEvent } from '../src/types/trace.js'; +import type { ModuleContext } from '../src/types/module.js'; + +const silentLogger: RelayLogger = { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: () => {}, +}; + +function makeBridge( + identity?: { userId?: string; username?: string }, + agentFilter?: (agentName: string) => boolean, +) { + const sent: Array<{ msg: BotStreamMessage }> = []; + let listener: ((e: TraceEvent) => void) | null = null; + const ctx = { + onTrace: (l: (e: TraceEvent) => void) => { + listener = l; + return () => { + listener = null; + }; + }, + } as unknown as ModuleContext; + + const bridge = new InferenceTraceBridge( + (msg) => sent.push({ msg }), + () => identity, + silentLogger, + agentFilter, + ); + bridge.start(ctx); + const emit = (e: Partial & { type: string }) => + listener!({ timestamp: 1710900000000, ...e } as TraceEvent); + return { bridge, sent, emit }; +} + +test('bridge: started → activation_start with resolved identity', () => { + const { sent, emit } = makeBridge({ username: 'Opus 4.5' }); + emit({ type: 'inference:started', agentName: 'opus45', channelId: 'chan-1' } as never); + + assert.equal(sent.length, 1); + assert.deepEqual(sent[0], { + msg: { + type: 'activation_start', + botId: 'opus45', + userId: 'opus45', + username: 'Opus 4.5', + channelId: 'chan-1', + timestamp: 1710900000000, + }, + }); +}); + +test('bridge: agent filter drops other agents\' traces entirely', () => { + const { sent, emit } = makeBridge(undefined, (name) => name === 'mine'); + emit({ type: 'inference:started', agentName: 'mine', channelId: 'c' } as never); + emit({ type: 'inference:started', agentName: 'other', channelId: 'c' } as never); + emit({ type: 'inference:tokens', agentName: 'other', channelId: 'c', content: 'x', blockType: 'text', blockIndex: 0 } as never); + + assert.equal(sent.length, 1, 'only the filtered-in agent is translated'); + assert.equal((sent[0].msg as { botId: string }).botId, 'mine'); +}); + +test('bridge: tokens → chunk with visible derived from blockType', () => { + const { sent, emit } = makeBridge(); + emit({ + type: 'inference:tokens', + agentName: 'a', + channelId: 'c', + content: 'Hello ', + blockType: 'text', + blockIndex: 0, + } as never); + emit({ + type: 'inference:tokens', + agentName: 'a', + channelId: 'c', + content: 'hmm...', + blockType: 'thinking', + blockIndex: 1, + } as never); + + const [text, thinking] = sent.map((s) => s.msg as { visible: boolean; text: string; type: string }); + assert.equal(text.type, 'chunk'); + assert.equal(text.visible, true, 'text chunks are voiced'); + assert.equal(thinking.visible, false, 'thinking chunks are not voiced'); +}); + +test('bridge: block_complete carries content accumulated from chunk traces', () => { + const { sent, emit } = makeBridge(); + emit({ type: 'inference:content_block', agentName: 'a', channelId: 'c', phase: 'block_start', blockType: 'text', blockIndex: 0 } as never); + emit({ type: 'inference:tokens', agentName: 'a', channelId: 'c', content: 'Hello ', blockType: 'text', blockIndex: 0 } as never); + emit({ type: 'inference:tokens', agentName: 'a', channelId: 'c', content: 'world', blockType: 'text', blockIndex: 0 } as never); + emit({ type: 'inference:content_block', agentName: 'a', channelId: 'c', phase: 'block_complete', blockType: 'text', blockIndex: 0 } as never); + + const complete = sent.at(-1)!.msg as { type: string; content: string }; + assert.equal(complete.type, 'block_complete'); + assert.equal(complete.content, 'Hello world'); +}); + +test('bridge: terminal traces map to activation_end reasons; accumulator clears', () => { + for (const [type, reason] of [ + ['inference:completed', 'complete'], + ['inference:turn_ended', 'complete'], + ['inference:aborted', 'abort'], + ['inference:failed', 'error'], + ] as const) { + const { sent, emit } = makeBridge(); + emit({ type: 'inference:tokens', agentName: 'a', channelId: 'c', content: 'x', blockType: 'text', blockIndex: 0 } as never); + emit({ type, agentName: 'a', channelId: 'c', durationMs: 5, error: 'boom' } as never); + + const end = sent.at(-1)!.msg as { type: string; reason: string }; + assert.equal(end.type, 'activation_end', `${type} ends the activation`); + assert.equal(end.reason, reason, `${type} → ${reason}`); + + // Accumulator cleared: a new block 0 must not inherit old text. + emit({ type: 'inference:tokens', agentName: 'a', channelId: 'c', content: 'fresh', blockType: 'text', blockIndex: 0 } as never); + emit({ type: 'inference:content_block', agentName: 'a', channelId: 'c', phase: 'block_complete', blockType: 'text', blockIndex: 0 } as never); + const complete = sent.at(-1)!.msg as { content: string }; + assert.equal(complete.content, 'fresh', `accumulator reset after ${type}`); + } +}); + +test('bridge: stream_restarted closes the abandoned activation with abort', () => { + const { sent, emit } = makeBridge(); + emit({ type: 'inference:tokens', agentName: 'a', channelId: 'c', content: 'x', blockType: 'text', blockIndex: 0 } as never); + emit({ type: 'inference:stream_restarted', agentName: 'a', channelId: 'c', reason: 'context_budget_restart', inputTokens: 1, budget: 1 } as never); + + const end = sent.at(-1)!.msg as { type: string; reason: string }; + assert.equal(end.type, 'activation_end', 'restart pairs the dangling activation_start'); + assert.equal(end.reason, 'abort', 'the partial utterance was cut off, not finished'); + + // A channel-less restart has nothing to close on the wire. + const before = sent.length; + emit({ type: 'inference:stream_restarted', agentName: 'a', reason: 'r', inputTokens: 1, budget: 1 } as never); + assert.equal(sent.length, before); +}); + +test('bridge: traces without channelId are dropped; unrelated traces ignored', () => { + const { sent, emit } = makeBridge(); + emit({ type: 'inference:started', agentName: 'a' } as never); + emit({ type: 'inference:tokens', agentName: 'a', content: 'x', blockType: 'text', blockIndex: 0 } as never); + emit({ type: 'message:added', messageId: 'm1', source: 's' } as never); + emit({ type: 'inference:usage', agentName: 'a', tokenUsage: { input: 1, output: 1 } } as never); + + assert.equal(sent.length, 0, 'nothing fans out without a channel'); +}); + +test('bridge: stop() unsubscribes', () => { + const { bridge, sent, emit } = makeBridge(); + bridge.stop(); + assert.throws(() => emit({ type: 'inference:started', agentName: 'a', channelId: 'c' } as never)); + assert.equal(sent.length, 0); +});