diff --git a/changelog.d/feat-disabled-mode-typing.changed.md b/changelog.d/feat-disabled-mode-typing.changed.md new file mode 100644 index 000000000..11b65b9c3 --- /dev/null +++ b/changelog.d/feat-disabled-mode-typing.changed.md @@ -0,0 +1,21 @@ +- Agents with `proseRouting: 'disabled'` now show the typing indicator during + a turn, on the channel the trigger came from, the same as explicit mode. A + batched gate wake names no `channelId`, so its `wakeChannelId` is used + instead. Before, disabled-mode turns showed no indicator at all, so a + resident that answers only through send tools looked idle while it worked. + Never on a tuned-out channel: the agent turned its attention off there, yet + an ambient push or a gate wake can still name the channel. Silent wakes + (`suppressProse`, such as a silent heartbeat) still show none. Explicit and + locus modes pick their typing channel as before. +- Entering tune-out on a channel now stops a typing indicator already running + there. Before, its 7 s refresh continued until the turn ended. A + disabled-mode turn does not restart it; an explicit or locus turn can still + restart it at stream start or on a retry, as before. +- With conversation routing, a disabled-mode gate wake shows typing only + where the channel's messages reach the agent: a fork on its home channel, + any other agent never on a fork-bound channel. +- A silent request batched with ordinary ones no longer picks the turn's + channel or wake provenance; the ordinary requests decide, in every + prose-routing mode. A batch of only silent requests is unchanged. +- A typing `stop` now honors the server's `channels.typing` grant, like the + start: a server without the grant no longer receives an unpaired stop. diff --git a/src/framework.ts b/src/framework.ts index 013277a59..f6a3fa1d7 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -10443,9 +10443,14 @@ export class AgentFramework { // addressed message's direct wake, and queue order would hand the turn // to the older one. Ties keep the later-queued request, as before. const eventAt = (r: InferenceRequest): number => r.wakeAt ?? r.timestamp; + // A silent tick never suppresses a genuine coalesced wake. If any + // ordinary request shares this batch, the ordinary turn semantics win, + // including where the turn is: a silent request names no channel then. + const silentOnly = requests.every((r) => r.suppressProse === true); + const routable = silentOnly ? requests : requests.filter((r) => r.suppressProse !== true); let ambientReq: InferenceRequest | undefined; let addressedReq: InferenceRequest | undefined; - for (const r of requests) { + for (const r of routable) { if (!r.channelId) continue; if (!ambientReq || eventAt(r) >= eventAt(ambientReq)) ambientReq = r; if (r.addressed && (!addressedReq || eventAt(r) >= eventAt(addressedReq))) addressedReq = r; @@ -10461,14 +10466,11 @@ export class AgentFramework { let provReq: InferenceRequest | undefined = addressedReq ?? (channelReq?.counterparty || channelReq?.wakeChannelId ? channelReq : undefined); if (!provReq) { - for (const r of requests) { + for (const r of routable) { if (!r.counterparty && !r.wakeChannelId) continue; if (!provReq || (r.wakeAt ?? r.timestamp) >= (provReq.wakeAt ?? provReq.timestamp)) provReq = r; } } - // A silent tick never suppresses a genuine coalesced wake. If any - // ordinary request shares this batch, the ordinary turn semantics win. - const silentOnly = requests.every((r) => r.suppressProse === true); // RFC-006 §5: when every cause is a deferred batch, the turn's content // exists only once assembly renders it. Recorded for the whole batch // (not inherited from requests[0]), so a requeued trigger stays honest. @@ -11334,6 +11336,36 @@ export class AgentFramework { } } + /** + * Where a `proseRouting: 'disabled'` turn shows typing: the trigger + * channel, as in explicit mode, else where a batched gate wake came from + * (`wakeChannelId`; the gate deliberately names no `channelId`) when that + * channel's messages reach this agent. Never a + * tuned-out channel: its incoming traffic, addressed included, is diverted + * away from the resident, yet an ambient push or a gate wake can still name + * the channel, and typing there would show attendance the agent turned off + * (`enterTuneOut` also stops typing already running there). A silent wake + * (`suppressProse`, which also runs as disabled) shows none. + */ + private disabledTypingChannel(agentName: string, trigger: InferenceRequest | undefined): string | null { + if (!trigger || trigger.suppressProse) return null; + let channelId = trigger.channelId; + if (!channelId && trigger.wakeChannelId) { + // The gate wakes every agent, but with conversation routing a channel's + // messages reach only the fork bound to it: a fork types only on its + // home channel, and no other agent types on a fork-bound channel. + const home = this.conversationAgentHomes.get(agentName); + const reachesAgent = home !== undefined + ? trigger.wakeChannelId === home + : ![...this.conversationAgentHomes.values()].includes(trigger.wakeChannelId); + if (reachesAgent) channelId = trigger.wakeChannelId; + } + if (!channelId) return null; + const serverId = this.channelRegistry?.getChannelServerId(channelId); + if (serverId && this.channelRegistry?.getTuneOutState(serverId, channelId)) return null; + return channelId; + } + /** * Body of startAgentStream, split out so the caller's try/finally owns the * turn token unconditionally. Returns true iff the token was handed off to @@ -11572,9 +11604,11 @@ export class AgentFramework { // idempotent per channel, and owns the 7s refresh); the catch below stops // it on the no-driveStream failure paths (e.g. a compile refusal). const earlyTypingChannel = - turnProseRouting === 'explicit' - ? trigger?.channelId ?? null - : turnLocus ?? null; + turnProseRouting === 'disabled' + ? this.disabledTypingChannel(agent.name, trigger) + : turnProseRouting === 'explicit' + ? trigger?.channelId ?? null + : turnLocus ?? null; if (earlyTypingChannel) this.channelRegistry?.startTyping(earlyTypingChannel); try { @@ -11926,9 +11960,10 @@ export class AgentFramework { // this doesn't violate never-guess: it says "attending to what you // sent here", which is true regardless of where the reply goes. // Heartbeat/no-trigger explicit turns show no indicator. + // - disabled mode: see disabledTypingChannel. const typingChannel = turnProseRouting === 'disabled' - ? null + ? this.disabledTypingChannel(agent.name, trigger) : turnProseRouting === 'explicit' ? trigger?.channelId ?? null : resolveTurnLocus(); diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 082e6ed9b..a1f81fbd2 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -1291,9 +1291,7 @@ export class ChannelRegistry { // global-clear branch's semantics, and keeps defensive stopTyping(ch) // calls from spamming stops at a server that never saw a start. const entry = this.findChannelEntry(channelId); - if (entry && this.sendTypingFn) { - this.sendTypingFn(entry.serverId, channelId, this.typingMetadata.get(channelId), 'stop'); - } + if (entry) this.sendTypingStop(entry.serverId, channelId); } this.typingMetadata.delete(channelId); } else { @@ -1303,11 +1301,9 @@ export class ChannelRegistry { clearInterval(interval); } this.typingIntervals.clear(); - if (this.sendTypingFn) { - for (const id of channels) { - const entry = this.findChannelEntry(id); - if (entry) this.sendTypingFn(entry.serverId, id, this.typingMetadata.get(id), 'stop'); - } + for (const id of channels) { + const entry = this.findChannelEntry(id); + if (entry) this.sendTypingStop(entry.serverId, id); } this.typingMetadata.clear(); } @@ -1988,6 +1984,9 @@ export class ChannelRegistry { source, timestamp: new Date().toISOString(), }); + // A tuned-out channel shows no attendance: stop a typing run already + // going there, or its 7s refresh would continue until the turn ends. + this.stopTyping(channelId); } /** @@ -2271,6 +2270,16 @@ export class ChannelRegistry { // Private: Typing notification // ========================================================================== + /** + * Send a typing 'stop'. The same §14.1 grant gate as the start: a timer can + * exist for a server whose start was skipped for lack of `channels.typing`, + * and it must not receive an unpaired stop either. + */ + private sendTypingStop(serverId: string, channelId: string): void { + if (!CapabilityGrant.of(this.serverRegistry.getServer(serverId)).has('channels.typing')) return; + this.sendTypingFn?.(serverId, channelId, this.typingMetadata.get(channelId), 'stop'); + } + /** * Send a typing notification for a channel. * diff --git a/test/channel-registry-typing.test.ts b/test/channel-registry-typing.test.ts index 1d49067f0..4cd18f737 100644 --- a/test/channel-registry-typing.test.ts +++ b/test/channel-registry-typing.test.ts @@ -140,3 +140,24 @@ test('stopTyping() global branch only dispatches stops for channels with interva assert.equal(stops.length, 1); assert.equal(stops[0].channelId, 'ch1'); }); + +test('a server without the channels.typing grant gets no stop, from either stopTyping branch', () => { + const calls: TypingCall[] = []; + const registry = new ChannelRegistry( + { getServer: () => ({ grant: new CapabilityGrant(new Set(), []) }) } as unknown as McplServerRegistry, + {} as FeatureSetManager, + () => {}, + () => {}, + { sendTypingFn: (serverId, channelId, metadata, op) => { calls.push({ serverId, channelId, metadata, op }); } }, + ); + (registry as unknown as { + channels: Map; + }).channels.set('srv:ch1', { serverId: 'srv', descriptor: { id: 'ch1' }, open: true }); + + registry.startTyping('ch1'); // the timer exists; the start was skipped for lack of the grant + registry.stopTyping('ch1'); + registry.startTyping('ch1'); + registry.stopTyping(); + + assert.deepEqual(calls, [], 'an ungranted server receives neither start nor stop'); +}); diff --git a/test/disabled-prose-typing.test.ts b/test/disabled-prose-typing.test.ts new file mode 100644 index 000000000..1a35eefcf --- /dev/null +++ b/test/disabled-prose-typing.test.ts @@ -0,0 +1,230 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { AgentFramework } from '../src/index.js'; +import { ChannelRegistry } from '../src/mcpl/channel-registry.js'; +import { CapabilityGrant } from '../src/mcpl/capability-grant.js'; +import { TuneOutCoordinator } from '../src/tune-out/coordinator.js'; +import { MockMembrane, MockYieldingStream, createMockResponse } from './helpers/mock-membrane.js'; +import type { ContentBlock } from '@animalabs/membrane'; + +async function make(proseRouting: 'disabled' | 'explicit') { + const dir = mkdtempSync(join(tmpdir(), 'disabled-typing-af-')); + const membrane = new MockMembrane(); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: membrane.asMembrane(), + agents: [{ name: 'assistant', model: 'test', systemPrompt: 'test', proseRouting }], + modules: [], + }); + const typing: string[] = []; + (framework as any).channelRegistry = new Proxy({ + resolveLocus: () => 'world:commons', getDefaultPublishChannel: () => 'world:commons', + routeSpeech: async () => ({ delivered: true, channelId: 'world:commons' }), + startTyping: (channel: string) => { typing.push(channel); }, + stopTyping: () => {}, + getChannelTools: () => [], getDescriptor: () => undefined, + }, { get: (t, p: string) => p in t ? (t as any)[p] : () => undefined }); + // In explicit mode, prose with no `>>` target keeps this run from going idle + // (the mock answers every further call with prose); a turn with no prose ends. + membrane.pushResponse(createMockResponse( + (proseRouting === 'explicit' ? [] : [{ type: 'text', text: 'working on it' }]) as ContentBlock[])); + return { dir, framework, typed: () => [...new Set(typing)] }; +} + +async function wake( + proseRouting: 'disabled' | 'explicit', + request: Record | Record[] | 'silent', + homes: Record = {}, +) { + const x = await make(proseRouting); + for (const [agent, home] of Object.entries(homes)) (x.framework as any).conversationAgentHomes.set(agent, home); + try { + if (request === 'silent') { + (x.framework as any).handleMcplPushEvent({ + type: 'mcpl:push-event', serverId: 'heartbeat', featureSet: 'heartbeat', + eventId: 'hb-1', content: [], timestamp: new Date().toISOString(), + origin: { source: 'heartbeat', reason: 'schedule', silent: true, scheduledAt: '2026-10-05 12:00:00 PT' }, + inferenceId: 'inf-1', triggerInference: true, + }); + } else { + const now = Date.now(); + for (const [i, r] of (Array.isArray(request) ? request : [request]).entries()) { + (x.framework as any).pendingRequests.push({ agentName: 'assistant', timestamp: now + i, ...r }); + } + } + await x.framework.runUntilIdle(); + return x.typed(); + } finally { await x.framework.stop(); rmSync(x.dir, { recursive: true, force: true }); } +} + +describe('typing with proseRouting disabled', () => { + it('shows typing on the channel the trigger came from', async () => { + const typed = await wake('disabled', { reason: 'channel-message', source: 'zulip', channelId: 'zulip:support', addressed: true }); + assert.deepEqual(typed, ['zulip:support']); + }); + + it('shows typing where a batched gate wake came from (wakeChannelId)', async () => { + const typed = await wake('disabled', { reason: 'gate', source: 'gate', wakeChannelId: 'zulip:support' }); + assert.deepEqual(typed, ['zulip:support']); + }); + + it('keeps a silent wake private: no typing', async () => { + assert.deepEqual(await wake('disabled', 'silent'), []); + }); + + it('shows no typing for a wake that names no channel', async () => { + assert.deepEqual(await wake('disabled', { reason: 'heartbeat', source: 'heartbeat' }), []); + }); + + it('a silent request batched with ordinary ones does not pick the turn\'s channel', async () => { + const silent = { reason: 'heartbeat', source: 'heartbeat', suppressProse: true, channelId: 'zulip:private', wakeChannelId: 'zulip:private' }; + assert.deepEqual(await wake('disabled', [silent, { reason: 'heartbeat', source: 'heartbeat' }]), []); + assert.deepEqual( + await wake('disabled', [silent, { reason: 'channel-message', source: 'zulip', channelId: 'zulip:support' }]), + ['zulip:support'], + ); + }); + + it('a gate wake types only where the channel\'s messages reach this agent', async () => { + const gate = (ch: string) => ({ reason: 'gate', source: 'gate', wakeChannelId: ch }); + // This agent is the fork bound to zulip:b: only its home channel. + assert.deepEqual(await wake('disabled', gate('zulip:a'), { assistant: 'zulip:b' }), []); + assert.deepEqual(await wake('disabled', gate('zulip:b'), { assistant: 'zulip:b' }), ['zulip:b']); + // Another fork owns zulip:a: not this agent's channel. + assert.deepEqual(await wake('disabled', gate('zulip:a'), { 'fork-1': 'zulip:a' }), []); + assert.deepEqual(await wake('disabled', gate('zulip:c'), { 'fork-1': 'zulip:a' }), ['zulip:c']); + }); + + it('leaves explicit mode unchanged: a batched gate wake shows no typing', async () => { + assert.deepEqual(await wake('explicit', { reason: 'gate', source: 'gate', wakeChannelId: 'zulip:support' }), []); + assert.deepEqual( + await wake('explicit', { reason: 'channel-message', source: 'zulip', channelId: 'zulip:support', addressed: true }), + ['zulip:support'], + ); + }); +}); + +/** + * The real path: typing is recorded at the ChannelRegistry's sendTypingFn, + * with the real EventGate (debounce) and TuneOutCoordinator in place. + */ +async function makeRealPath(opts: { tunedOut?: boolean; onStream?: (registry: ChannelRegistry) => void } = {}) { + const dir = mkdtempSync(join(tmpdir(), 'disabled-typing-real-')); + let streams = 0; + let registry!: ChannelRegistry; + const membrane = { + streamYielding: () => { streams++; opts.onStream?.(registry); return new MockYieldingStream([createMockResponse([])]); }, + }; + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: membrane as any, + agents: [{ name: 'assistant', model: 'test', systemPrompt: 'test', proseRouting: 'disabled' }], + modules: [], + gate: { config: { default: 'always', policies: [{ name: 'batch', match: { scope: ['mcpl:channel-incoming'] }, behavior: { debounce: 100 } }] } }, + } as any); + const typing: { channelId: string; op: string }[] = []; + const server = { grant: new CapabilityGrant(new Set(['channels.typing']), []) }; + const servers = { getServer: () => server } as any; + const f = framework as any; + registry = new ChannelRegistry(servers, {} as any, (e: any) => f.pushEvent(e), () => {}, { + store: f.store, + shouldTriggerInference: f.eventGate.asShouldTriggerCallback(), + sendTypingFn: (_s: string, channelId: string, _m: unknown, op?: string) => { typing.push({ channelId, op: op ?? 'start' }); }, + } as any); + (registry as any).channels.set('zulip:zulip:support', { + serverId: 'zulip', open: true, + descriptor: { id: 'zulip:support', type: 'zulip-stream', address: { streamId: 1 }, label: 'support' }, + }); + f.channelRegistry = registry; + f.tuneOutCoordinator = new TuneOutCoordinator(registry, servers, { + addMessage: (p: any, c: any, m: any, a: any) => f.addMessage(p, c, m, a ? { forAgent: a } : undefined), + requestInference: (agentName: string, reason: string, source: string) => f.pendingRequests.push({ agentName, reason, source, timestamp: Date.now() }), + subconsciousName: () => 'Subconscious', primaryName: () => 'assistant', + getStoredMessages: () => f.getAgent('assistant').getContextManager().getAllMessages(), + currentSequence: () => f.store.currentSequence(), setSubconsciousAnchor() {}, + isForkBound: () => false, isPrivilegedAuthor: () => false, allowChannelSpeech: () => false, + emitTrace: () => {}, + } as any); + const tuneOut = () => registry.enterTuneOut('zulip', 'zulip:support', { + epochId: 'epoch-1', cadenceSeconds: 3600, backlogCap: 20, maxWakes: 5, startedAtSequence: f.store.currentSequence(), + } as any, 'agent-tool' as any); + if (opts.tunedOut) tuneOut(); + return { + framework, registry, typing, tuneOut, + streams: () => streams, + starts: () => typing.filter((t) => t.op === 'start').map((t) => t.channelId), + async ambient() { + registry.handleIncoming('zulip', { messages: [{ + channelId: 'zulip:support', messageId: 'm1', author: { id: 'u1', name: 'Human' }, + timestamp: new Date().toISOString(), content: [{ type: 'text', text: 'ambient traffic' }], + tags: ['chat:ambient', 'chat:from-human'], + }] } as any); + await framework.runUntilIdle(); + await new Promise((r) => setTimeout(r, 300)); // past the 100 ms debounce + await framework.runUntilIdle(); + }, + async request(r: Record) { + f.pendingRequests.push({ agentName: 'assistant', timestamp: Date.now(), ...r }); + await framework.runUntilIdle(); + }, + async close() { await framework.stop(); rmSync(dir, { recursive: true, force: true }); }, + }; +} + +describe('typing with proseRouting disabled, through the real gate and registry', () => { + it('a debounced ambient wake types where it came from, then stops', async () => { + const x = await makeRealPath(); + try { + await x.ambient(); + assert.equal(x.streams(), 1, 'the gate wake ran a turn'); + assert.deepEqual(x.typing, [{ channelId: 'zulip:support', op: 'start' }, { channelId: 'zulip:support', op: 'stop' }]); + assert.equal((x.registry as any).typingIntervals.size, 0); + } finally { await x.close(); } + }); + + it('a tuned-out channel shows no typing when its traffic still reaches the gate', async () => { + const x = await makeRealPath({ tunedOut: true }); + try { + await x.ambient(); + assert.equal(x.streams(), 1, 'the gate wake ran a turn'); + assert.deepEqual(x.starts(), []); + } finally { await x.close(); } + }); + + it('a tuned-out channel shows no typing for a push that names it, ambient or addressed', async () => { + for (const addressed of [false, true]) { + const x = await makeRealPath({ tunedOut: true }); + try { + await x.request({ reason: 'mcpl:push-event', source: 'zulip', channelId: 'zulip:support', addressed }); + assert.equal(x.streams(), 1, 'the push ran a turn'); + assert.deepEqual(x.starts(), [], `addressed=${addressed}`); + } finally { await x.close(); } + } + }); + + it('tune-out entered mid-turn, before the stream starts, stops the typing already running there', async () => { + let intervalsAfterTuneOut = -1; + const x = await makeRealPath({ + onStream: (registry) => { + x.tuneOut(); + intervalsAfterTuneOut = (registry as any).typingIntervals.size; + }, + }); + try { + await x.request({ reason: 'mcpl:push-event', source: 'zulip', channelId: 'zulip:support', addressed: true }); + assert.equal(x.streams(), 1); + assert.equal(intervalsAfterTuneOut, 0, 'no refresh keeps running after tune-out'); + assert.deepEqual(x.typing, [{ channelId: 'zulip:support', op: 'start' }, { channelId: 'zulip:support', op: 'stop' }]); + } finally { await x.close(); } + }); + + it('a silent wake that names channels still shows no typing', async () => { + const x = await makeRealPath(); + try { + await x.request({ reason: 'heartbeat', source: 'heartbeat', suppressProse: true, channelId: 'zulip:support', wakeChannelId: 'zulip:support' }); + assert.equal(x.streams(), 1); + assert.deepEqual(x.starts(), []); + } finally { await x.close(); } + }); +});