diff --git a/README.md b/README.md index 92fe15c82..bf09bbfb7 100644 --- a/README.md +++ b/README.md @@ -239,6 +239,49 @@ Optional host-side implementation of the MCP Live protocol. External servers (ga } ``` +[RFC-006](https://github.com/anima-research/mcpl/pull/5) coalescing works across +`push/event` and `channels/incoming`. The host advertises `pushEvents`, `deferred`, +`channelsIncoming`, and `channelScopedPush` under `eventCoalescing`. + +An optional `coalesce: { key }` keeps the newest unread occurrence. Channel messages +require an occurrence `eventId` while retaining their stable platform `messageId`. +A push with `coalesce.channelId` shares that registered channel's subject namespace, so a +create arriving through one method can be edited or deleted through the other. Scope is +never inferred from `origin`. Both lanes check `channels.incoming`, current registration, +and the server config's optional `allowedIncomingChannels` allow-list (`*` globs; an empty +list denies all). A channel-scoped push also needs ordinary push authorization. + +Add `deferred: true` to a push to render a batch through `push/render` at activation +assembly; deferred mode is not valid on `channels/incoming`. Use `retract: true` to remove +unread work. Channel retractions require a deletion notice, which is appended when earlier +versions were consumed or their history is uncertain. Reply targets, conversation-fork +routing, and channel provenance survive mixed-lane updates. + +Pending events stay outside context managers until activation assembly. Shared-context +publication waits for the other residents' live turns to finish; isolated conversation +forks receive their own events. Consumed history is never rewritten. Pending work, +subject history, and occurrence receipts are reconstructed from the journal; recovery waits for +a fresh grant and channel registration, and does not repeat old source-backed renders. +Changing a configured endpoint/command creates a separate server-binding namespace. Use a +different server ID if credentials switch to another principal at the same endpoint. + +`mcpl/coalescing-journal` is the authoritative append-only operation log. Acceptance, +replacement, retraction, render decisions, and delivery receipts are durable appends. +A pure reducer rebuilds pending work and the occurrence index; replay itself performs no +RPCs, context writes, or wakes. Assembly appends a publication intent before writing +context, then appends completion. The intent seals that occurrence against replacement; +a journal-owned delivery ID makes an interrupted publication safe to retry. + +Older experimental pending/receipt stores are imported once into the journal. Their +uncertain pending deliveries are sealed conservatively and checked against legacy context +markers before publication; subsequent recovery never consults those old stores. Startup +currently replays the full journal and keeps receipt positions in memory, so recovery cost +and index memory grow with journal history. Rendering times out after five seconds. The +host retains 64 notices per batch, caps notice data at 4 KiB and content at 1 MiB, and +tracks up to 128 active subjects. +Excess pending subjects are rejected rather than silently dropped. Coalesced debounce +uses its own fixed deadline and does not change unrelated events' quiet periods. + ### Streaming Lifecycle 1. **Start**: Framework calls `agent.startStreamWithInjections()` → `YieldingStream` diff --git a/changelog.d/mcpl-channel-registration.breaking.md b/changelog.d/mcpl-channel-registration.breaking.md new file mode 100644 index 000000000..d03005b67 --- /dev/null +++ b/changelog.d/mcpl-channel-registration.breaking.md @@ -0,0 +1 @@ +- **MCPL server authors:** incoming channel content now requires an authoritative registration, including when coalescing is absent. Servers that relied on an outbound placeholder inferred from push `origin` must send `channels/register`/`channels/changed` or complete a host-authorized `channels/open` with a valid descriptor first. Outbound reply routing and conforming registered-channel delivery are unchanged. diff --git a/changelog.d/mcpl-event-coalescing.added.md b/changelog.d/mcpl-event-coalescing.added.md new file mode 100644 index 000000000..581c752e4 --- /dev/null +++ b/changelog.d/mcpl-event-coalescing.added.md @@ -0,0 +1,4 @@ +- Support RFC-006 coalescing across `push/event` and `channels/incoming`, including stable message identity, channel-scoped edits/deletions, deferred rendering, conversation routing, durable recovery/receipts, and current-grant checks. Channel retractions preserve consumed history and require a deletion notice. +- Add optional MCPL server `allowedIncomingChannels` patterns, enforced on both incoming messages and channel-scoped pushes. Host-derived outbound routing placeholders do not authorize inbound channel content. +- Keep coalesced wake deadlines separate from ordinary debounce and cancel obsolete queued wakes. Shared-context materialization waits for safe turn boundaries across residents. +- Make an append-only Chronicle operation journal authoritative for coalescing. Rebuild pending work and occurrence receipts through pure replay; use durable publication intents and stable context delivery IDs to recover interrupted publication without duplicate messages. Import older experimental state once. diff --git a/src/framework.ts b/src/framework.ts index 976d39ac4..0a4889b4a 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -67,7 +67,10 @@ import { FeatureSetManager } from './mcpl/feature-set-manager.js'; import { computeGrant, CapabilityGrant, expandAdvertisementShorthand } from './mcpl/capability-grant.js'; import { maskNegotiatedCapabilities } from './mcpl/capability-mask.js'; import { HookOrchestrator } from './mcpl/hook-orchestrator.js'; -import { PushHandler, type McplPushEvent } from './mcpl/push-handler.js'; +import { PushHandler, convertBlock as convertPushBlock, type McplPushEvent } from './mcpl/push-handler.js'; +import { PushCoalescer, CoalesceError, PUSH_COALESCING_SUPPORT, coalescingChannel, coalescingSubject, type CoalescedPush } from './mcpl/push-coalescer.js'; +import { ChronicleCoalescingJournal } from './mcpl/chronicle-coalescing-journal.js'; +import { COALESCING_SUBJECT } from './gate/event-gate.js'; import { parseProsePrefix, parseHybridProsePrefix } from './mcpl/prose-grammar.js'; import { ProseStreamRouter } from './mcpl/prose-stream-router.js'; import { detectKnownToolWrapperProse } from './tool-wrapper-prose-guard.js'; @@ -90,7 +93,9 @@ import { splitPreservingImageSlots, type ParsedImagePlaceholder, } from './tool-image-ledger.js'; -import { randomUUID } from 'node:crypto'; +import { randomUUID, createHash } from 'node:crypto'; +import { expandCoreTags } from './mcpl/tags.js'; +import type { ChannelIncomingMessage, PushEventResult } from './mcpl/types.js'; import { PyRunner, buildInjectedTools } from './code-execution/py-runner.js'; import { buildCodeExecutionToolDefinition, @@ -1213,6 +1218,13 @@ export class AgentFramework { private featureSetManager: FeatureSetManager | null = null; private hookOrchestrator: HookOrchestrator | null = null; private pushHandler: PushHandler | null = null; + private pushCoalescer: PushCoalescer | null = null; + private coalescingEpochs = new Map(); + private coalescingAdmission: Promise = Promise.resolve(); + private coalescingAssemblyOwner?: { agent: Agent; token: number }; + private coalescingSharedReaders = new WeakSet(); + private coalescingBranch: string | null = null; + private coalescingTriggerFilter?: (content: string, metadata: Record) => boolean; private inferenceRouter: InferenceRouter | null = null; private channelRegistry: ChannelRegistry | null = null; private checkpointManager: CheckpointManager | null = null; @@ -1551,6 +1563,7 @@ export class AgentFramework { emitTrace: (e) => framework.emitTrace(e as { type: TraceEvent['type']; [key: string]: unknown }), addMessage: (p, c, m, forAgent) => framework.addMessage(p, c, m as MessageMetadata, forAgent ? { forAgent } : undefined), requestInference: (agentName, reason, source, provenance) => { + if (provenance?.coalescingSubject) { framework.queueCoalescedWake(provenance.coalescingSubject); return; } framework.pendingRequests.push({ agentName, reason, source, timestamp: Date.now(), // gate-requested wakes carry where/who (EventGate wakeProvenance) @@ -1781,6 +1794,7 @@ export class AgentFramework { * Stop the event loop. */ async stop(): Promise { + this.pushCoalescer?.suspend(); this.running = false; // Flushed-but-unsynced deferred writes: sync and ack now, while the // store is still open, rather than leaving them to a reboot replay. @@ -6153,6 +6167,7 @@ export class AgentFramework { }); const agent = new Agent(config, contextManager, this.membrane); + this.coalescingSharedReaders.add(agent); const restoredSettings = this.readAgentRuntimeSettings(config.name); if (restoredSettings) { agent.restoreRuntimeSettings( @@ -6227,6 +6242,7 @@ export class AgentFramework { allowedTools: [...SUBCONSCIOUS_TOOL_NAMES, 'think', 'skip_reply', 'end_turn'], }; const agent = new Agent(agentConfig, contextManager, this.membrane); + this.coalescingSharedReaders.add(agent); // merges the residents' shared message slot this.agents.set(name, agent); this.agentConfigs.set(name, agentConfig); this.subconsciousAgentName = name; @@ -7213,9 +7229,266 @@ export class AgentFramework { }; } - /** - * Convert an MCPL push event to a context message. - */ + /** Coalescing addresses a delivery context, never an untrusted origin claim. */ + private coalescingAudience(agentName?: string): string { + return JSON.stringify(agentName === undefined + ? [this.store.currentBranch().name, 'shared-messages'] + : [this.store.currentBranch().name, 'agent', agentName]); + } + + private coalescingBinding(serverId: string): string { + const config = this.mcplServerConfigs.get(serverId); + // Transport credentials may rotate without changing this configured binding. + // Reassigning an ID to another endpoint or command gives it a new namespace. + return createHash('sha256').update(JSON.stringify([serverId, config?.url, + config?.command, config?.args])).digest('hex'); + } + + private assertIncomingChannel(serverId: string, channelId: string): void { + const server = this.mcplServerRegistry?.getServer(serverId); + if (!server?.policyEstablished || !server.grant.has('channels.incoming')) { + throw new CoalesceError('channelId', 'Channel not permitted', -32017); + } + if (!this.channelRegistry?.registeredDescriptor(serverId, channelId)) { + throw new CoalesceError('channelId', 'Unknown channel', -32023); + } + const allowed = this.mcplServerConfigs.get(serverId)?.allowedIncomingChannels; + if (allowed && !allowed.some(pattern => new RegExp('^' + pattern + .replace(/[.+?^${}()|[\]\\]/g, '\\$&').replace(/\*/g, '.*') + '$').test(channelId))) { + throw new CoalesceError('channelId', 'Channel not permitted', -32017); + } + } + + private authorizeCoalesced(push: CoalescedPush): void { + const server = this.mcplServerRegistry?.getServer(push.serverId); + if (!server?.isConnected || !server.policyEstablished || + push.binding !== this.coalescingBinding(push.serverId) || + (!push.recovered && this.coalescingEpochs.get(push.serverId) !== push.epoch)) { + throw new CoalesceError('serverId', 'Connection is not authorized', -32002); + } + if (!push.channelMessage) { + if (!server.grant.has('pushEvents')) throw new CoalesceError('pushEvents', 'Capability denied', -32002); + this.featureSetManager!.validateInbound(push.serverId, push.params.featureSet); + } + const channelId = coalescingChannel(push); + if (channelId !== undefined) this.assertIncomingChannel(push.serverId, channelId); + } + + private cancelCoalescedRequest(subject: string): void { + this.pendingRequests = this.pendingRequests.filter(r => r.coalescingSubject !== subject); + for (const cooldown of this.providerAccelerationCooldowns.values()) { + cooldown.heldRequests = cooldown.heldRequests.filter(r => r.coalescingSubject !== subject); + } + } + + private queueCoalescedWake(subject: string): boolean { + const push = this.pushCoalescer?.pending(subject); + if (!push || push.routing?.triggerAllowed === false) return false; + try { this.authorizeCoalesced(push); } catch { return false; } + this.cancelCoalescedRequest(subject); + const routing = push.routing; + const targets = routing?.targetAgents ?? [...this.agents.values()] + .filter(a => this.coalescingSharedReaders.has(a) && a.name !== this.subconsciousAgentName).map(a => a.name); + for (const agentName of targets) { + if (!this.agents.has(agentName)) continue; + this.pendingRequests.push({ + agentName, reason: push.channelMessage ? 'mcpl:channel-incoming' : 'mcpl:push-event', + source: push.serverId, timestamp: Date.now(), coalescingSubject: subject, coalescingEventId: push.params.eventId, + channelId: routing?.channelId, + counterparty: routing?.author?.id ? `${push.serverId}:user:${routing.author.id}` : undefined, + addressed: isAddressedMessage(push.params.tags, routing?.metadata ?? push.params.origin), + }); + } + return targets.some(name => this.agents.has(name)); + } + + private async resolveCoalescedRoute(push: CoalescedPush): Promise { + const message = push.channelMessage; + const explicitChannel = coalescingChannel(push); + const derived = !message && explicitChannel === undefined ? this.derivePushEventChannel(push.params.origin) : null; + const channelId = explicitChannel ?? derived?.channelId; + // Legacy feature-scoped pushes keep their ordinary provenance/routing behavior. + // This registration NEVER authorizes a channel-scoped coalescing operation. + if (derived) this.channelRegistry?.ensureChannelRegistered(push.serverId, derived.channelId, derived.label, derived.metadata); + const previousIdentity = explicitChannel ? this.pushCoalescer?.identity(coalescingSubject(push)) : undefined; + const metadata = { ...(message?.metadata ?? push.params.origin) }; + const origin = push.params.origin; + const author = message?.author ?? (typeof origin?.authorId === 'string' + ? { id: origin.authorId, name: typeof origin.authorName === 'string' ? origin.authorName : origin.authorId } : previousIdentity?.author); + const routing: NonNullable = { + channelId, messageId: message?.messageId ?? (typeof origin?.messageId === 'string' ? origin.messageId : previousIdentity?.messageId ?? push.params.eventId), + threadId: message?.threadId ?? previousIdentity?.threadId, author, metadata, + targetAgents: [...this.agents.values()].filter(a => this.coalescingSharedReaders.has(a) && a.name !== this.subconsciousAgentName).map(a => a.name), + }; + push.audience = this.coalescingAudience(); + if (explicitChannel && this.conversationRouter) { + const router = this.conversationRouter; + const prior = this.pushCoalescer?.pending(coalescingSubject(push)); + const oldTarget = prior?.routing?.targetAgents[0]; + if (oldTarget && prior?.routing?.generation && !router.getBinding(explicitChannel) && + router.exportGenerations()[explicitChannel] === prior.routing.generation) { + if (!this.agents.has(oldTarget)) await this.createConversationAgent(oldTarget, explicitChannel); + router.bind(explicitChannel, oldTarget, prior.routing.generation); + } + const decision = router.route({ channelId: explicitChannel, + mentioned: metadata.mentioned === true, // same bind/trigger predicate as ordinary channel input + kind: ConversationRouter.classifyChannel(this.channelRegistry?.registeredDescriptor(push.serverId, explicitChannel), metadata) }); + if (decision.kind === 'unbound') { push.audience = ''; routing.targetAgents = []; } + else { + if (decision.kind === 'spawn') { + await this.createConversationAgent(decision.agentName, explicitChannel); + router.bind(explicitChannel, decision.agentName, decision.generation); + this.persistConversationRouterState(); + } + push.audience = this.coalescingAudience(decision.agentName); + routing.targetAgents = [decision.agentName]; + routing.generation = router.getBinding(explicitChannel)?.generation; + routing.triggerAllowed = decision.trigger; + } + } + if (channelId && !this.conversationRouter) { + const divert = this.channelRegistry?.getTuneOutState(push.serverId, channelId); + if (divert) { + metadata.tuneOut = { epochId: divert.params.epochId }; + routing.targetAgents = this.subconsciousAgentName ? [this.subconsciousAgentName] : []; + routing.triggerAllowed = isAddressedMessage(push.params.tags, metadata) || (this.eventGate?.isPrivilegedUser(author?.id) ?? false); + } + } + push.routing = routing; + } + + private admitCoalesced(push: CoalescedPush): Promise { + // Routing may spawn a conversation agent. Serialize both delivery lanes so a + // create, edit, and delete cannot overtake one another while routing awaits. + const work = this.coalescingAdmission.then(async () => { + this.ensureCoalescingBranch(); + this.authorizeCoalesced(push); + this.pushCoalescer!.validate(push); + const duplicate = this.pushCoalescer!.receipt(push); + if (duplicate) return duplicate; + push.params.tags = push.params.tags ? expandCoreTags(push.params.tags) : undefined; + await this.resolveCoalescedRoute(push); + this.authorizeCoalesced(push); // routing awaited; authority may have changed + const result = this.pushCoalescer!.accept(push); + if (result.accepted && push.routing?.metadata?.tuneOut && push.routing.channelId) { + // Record diversion for the existing cadence path. Immediate/gated wakes + // remain subject-keyed and cancellable in the coalescing wake path. + this.tuneOutCoordinator?.onIncoming(push.serverId, push.routing.channelId, + push.routing.messageId!, push.params.tags, push.routing.author?.id, false); + } + return result; + }); + this.coalescingAdmission = work.then(() => {}, () => {}); + return work; + } + + private async resumeCoalescedChannels(serverId?: string): Promise { + if (!this.pushCoalescer) return; + const work = this.coalescingAdmission.then(async () => { + this.ensureCoalescingBranch(); + for (const push of this.pushCoalescer!.pendingPushes()) { + if (!push.recovered || serverId && push.serverId !== serverId) continue; + const channel = coalescingChannel(push); + if (channel && !this.channelRegistry?.registeredDescriptor(push.serverId, channel)) continue; + try { this.authorizeCoalesced(push); } catch { continue; } + const router = this.conversationRouter; + const target = push.routing?.targetAgents[0]; + const generation = push.routing?.generation; + if (channel && router && target && generation && !router.getBinding(channel) && + router.exportGenerations()[channel] === generation) { + if (!this.agents.has(target)) await this.createConversationAgent(target, channel); + router.bind(channel, target, generation); + } else if (!channel) { + const derived = this.derivePushEventChannel(push.params.origin); + if (derived) this.channelRegistry?.ensureChannelRegistered(push.serverId, derived.channelId, derived.label, derived.metadata); + } + } + this.pushCoalescer!.wakeRecovered(); + }); + this.coalescingAdmission = work.then(() => {}, () => {}); + await work; + } + + private materializeCoalescedPush(push: CoalescedPush, agent: Agent, consumer = agent.name): boolean { + const shared = push.audience === this.coalescingAudience(); + if ((!shared && push.audience !== this.coalescingAudience(agent.name)) || this.agents.get(agent.name) !== agent) return false; + const deliveryId = push.publicationId ?? JSON.stringify([push.serverId, push.binding ?? '', push.params.eventId, push.audience]); + const prototypeId = JSON.stringify([push.serverId, push.binding ?? '', push.params.eventId, push.audience]); + const legacyId = JSON.stringify([push.serverId, push.epoch, push.params.eventId, push.audience]); + if (push.recovered && agent.getContextManager().getAllMessages().some(m => + m.metadata?.coalescingDeliveryId === deliveryId || m.metadata?.coalescingDeliveryId === legacyId || m.metadata?.coalescingDeliveryId === prototypeId)) return true; + const routing = push.routing; + const metadata: Record = { + ...push.params.origin, ...routing?.metadata, + serverId: push.serverId, featureSet: push.params.featureSet, eventId: push.params.eventId, + channelId: routing?.channelId, messageId: routing?.messageId, author: routing?.author, + threadId: routing?.threadId, tags: push.params.tags, coalescingDeliveryId: deliveryId, + }; + const content = push.params.payload.content.map(convertPushBlock); + if (routing?.channelId && !this.conversationRouter && !metadata.tuneOut) { + const invitation = this.buildClosedChannelInvitation({ serverId: push.serverId, + channelId: routing.channelId, messageId: routing.messageId!, + authorName: routing.author?.name ?? 'Someone', tags: push.params.tags }); + if (invitation) { content.push(invitation.block); Object.assign(metadata, invitation.metadataPatch); } + } + agent.getContextManager().addMessage('user', content, metadata); + const consumedWake = (r: InferenceRequest) => r.agentName === consumer && + r.coalescingSubject === coalescingSubject(push) && r.coalescingEventId === push.params.eventId; + this.pendingRequests = this.pendingRequests.filter(r => !consumedWake(r)); + for (const cooldown of this.providerAccelerationCooldowns.values()) { + cooldown.heldRequests = cooldown.heldRequests.filter(r => !consumedWake(r)); + } + return true; + } + + private initializePushCoalescer(triggerFilter?: (content: string, metadata: Record) => boolean): void { + const branch = this.store.currentBranch().name; + this.coalescingBranch = branch; + this.coalescingTriggerFilter = triggerFilter; + const journal = new ChronicleCoalescingJournal(this.store, branch); + this.pushCoalescer = new PushCoalescer({ + journal, + available: push => { + const server = this.mcplServerRegistry?.getServer(push.serverId); + const channel = coalescingChannel(push); + return !!server?.isConnected && server.policyEstablished && (!channel || !!this.channelRegistry?.registeredDescriptor(push.serverId, channel)); + }, + authorized: push => { + if (this.store.currentBranch().name !== branch) return false; + try { this.authorizeCoalesced(push); return true; } catch { return false; } + }, + render: (push, params) => this.mcplServerRegistry!.getServer(push.serverId)!.sendPushRender(params), + wake: (push, subject) => { + this.cancelCoalescedRequest(subject); + if (push.routing?.triggerAllowed === false) { this.eventGate?.cancelCoalesced(subject); return false; } + const text = push.params.payload.content.map(convertPushBlock) + .filter((b): b is ContentBlock & { type: 'text' } => b.type === 'text').map(b => b.text).join('\n'); + const trigger = triggerFilter?.(text, { + ...push.params.origin, ...push.routing?.metadata, + serverId: push.serverId, featureSet: push.params.featureSet, eventId: push.params.eventId, + eventType: push.channelMessage ? 'mcpl:channel-incoming' : 'mcpl:push-event', + channelId: push.routing?.channelId, author: push.routing?.author, + tags: push.params.tags, [COALESCING_SUBJECT]: subject, + }) ?? true; + if (!trigger) return false; + this.eventGate?.cancelCoalesced(subject); + return this.queueCoalescedWake(subject); + }, + cancelWake: (subject, consumed) => { this.eventGate?.cancelCoalesced(subject); if (!consumed) this.cancelCoalescedRequest(subject); }, + observeLate: detail => this.pushCoalescer?.observe(detail), + }); + this.pushCoalescer.recover(branch); + } + + private ensureCoalescingBranch(): void { + if (this.pushCoalescer && this.coalescingBranch !== this.store.currentBranch().name) { + this.pushCoalescer.suspend(); + this.initializePushCoalescer(this.coalescingTriggerFilter); + this.pushCoalescer?.wakeRecovered(); + } + } + + /** Convert an ordinary MCPL push event to a context message. */ private handleMcplPushEvent(event: McplPushEvent): void { const triggerChannel = this.derivePushEventChannel(event.origin); if (triggerChannel && this.channelRegistry) { @@ -7370,7 +7643,8 @@ export class AgentFramework { // Check each agent for (const [agentName, groupedRequests] of requestsByAgent) { - let requests = groupedRequests; + let requests = groupedRequests.filter(r => !r.coalescingSubject || this.pushCoalescer?.occurrence(r.coalescingSubject) === r.coalescingEventId); + if (!requests.length) continue; const agent = this.agents.get(agentName); if (!agent) { // Agent not found — request is orphaned. Emit warning and drop. @@ -8113,13 +8387,25 @@ export class AgentFramework { // clears our own token. Token-matched, so when a retry frame has replaced // the token, or the terminal-failure branch already deleted it, this // no-ops instead of clobbering a successor's marker. + const sharedReader = this.coalescingSharedReaders?.has(agent); + this.ensureCoalescingBranch(); + const hasPending = this.pushCoalescer?.canAssemble(this.coalescingAudience()); + if (sharedReader && (this.coalescingAssemblyOwner && this.coalescingAssemblyOwner.agent !== agent || + hasPending && [...this.agents.values()].some(a => a !== agent && this.coalescingSharedReaders.has(a) && + (this.activeTurnTokens.has(a.name) || this.activeStreams.has(a.name))))) { + this.pendingRequests.push(trigger ?? { agentName: agent.name, reason: 'mcpl:coalescing-boundary', source: 'scheduler', timestamp: Date.now() }); + if (ownsProviderGate) this.releasePrimaryProviderGate(agent.name); + return; + } const turnToken = this.nextTurnToken++; + if (sharedReader) this.coalescingAssemblyOwner = { agent, token: turnToken }; this.activeTurnTokens.set(agent.name, turnToken); this.activeTurnTriggers.set(agent.name, trigger); let tokenHandedOff = false; try { tokenHandedOff = await this.beginAgentTurn(agent, trigger, attempt, turnToken, ownsProviderGate); } finally { + if (this.coalescingAssemblyOwner?.agent === agent && this.coalescingAssemblyOwner.token === turnToken) this.coalescingAssemblyOwner = undefined; if (!tokenHandedOff && this.activeTurnTokens.get(agent.name) === turnToken) { this.activeTurnTokens.delete(agent.name); this.activeTurnTriggers.delete(agent.name); @@ -8355,6 +8641,24 @@ export class AgentFramework { } } + // RFC-006: pending content is absent from all context managers until this + // assembly boundary, so background compression cannot consume it early. + this.ensureCoalescingBranch(); + if (this.coalescingSharedReaders?.has(agent)) { + const otherBusy = [...this.agents.values()].some(a => a !== agent && this.coalescingSharedReaders.has(a) && + (this.activeTurnTokens.has(a.name) || this.activeStreams.has(a.name))); + const primary = this.primaryAgentName ? this.agents.get(this.primaryAgentName) : undefined; + if (primary && !otherBusy) await this.pushCoalescer?.assemble(this.coalescingAudience(), push => this.materializeCoalescedPush(push, primary, agent.name)); + } else { + await this.pushCoalescer?.assemble(this.coalescingAudience(agent.name), push => this.materializeCoalescedPush(push, agent)); + } + if (this.providerAdmissionClosed || this.agents.get(agent.name) !== agent || + this.activeTurnTokens.get(agent.name) !== turnToken) { + this.channelRegistry?.stopTyping(); + this.eventGate?.onInferenceEnded(agent.name); + return false; + } + const { stream, request: compiledRequest, @@ -11798,11 +12102,20 @@ export class AgentFramework { ?? (this.eventGate ? this.eventGate.asShouldTriggerCallback() : undefined); // Push events handler (Step 6) + this.initializePushCoalescer(triggerFilter); this.pushHandler = new PushHandler( this.featureSetManager, (event) => this.pushEvent(event as unknown as ProcessEvent), (event) => this.emitTrace(event as { type: TraceEvent['type']; [key: string]: unknown }), triggerFilter, + (serverId, params) => { + this.ensureCoalescingBranch(); + const agentName = this.primaryAgentName; + if (!agentName) throw new CoalesceError('audience', 'no primary agent is configured', -32000); + const epoch = this.coalescingEpochs.get(serverId); + if (!epoch) throw new CoalesceError('serverId', 'no initialized connection', -32002); + return this.admitCoalesced({ serverId, binding: this.coalescingBinding(serverId), epoch, params, audience: this.coalescingAudience() }); + }, ); // Server-initiated inference router (Step 6) @@ -11832,6 +12145,15 @@ export class AgentFramework { (event) => this.emitTrace(event as { type: TraceEvent['type']; [key: string]: unknown }), { store: this.store, + authorizeIncoming: (serverId, channelId) => this.assertIncomingChannel(serverId, channelId), + handleCoalescedIncoming: async (serverId, message) => { + const params: PushEventParams = { featureSet: '', eventId: message.eventId ?? '', + timestamp: message.timestamp, tags: message.tags, coalesce: message.coalesce, + payload: { content: message.content } }; + const result = await this.admitCoalesced({ serverId, binding: this.coalescingBinding(serverId), + epoch: this.coalescingEpochs.get(serverId) ?? '', audience: this.coalescingAudience(), params, channelMessage: message }); + return { messageId: message.messageId, accepted: result.accepted, coalesce: result.coalesce, reason: result.reason }; + }, sendTypingFn: (serverId, channelId, metadata, op) => { const server = this.mcplServerRegistry!.getServer(serverId); if (server) { @@ -11906,6 +12228,7 @@ export class AgentFramework { // implemented for years — are declared (AUDIT-001/issue #76 item 12). this.mcplHostCapabilities = { version: '0.5', + eventCoalescing: PUSH_COALESCING_SUPPORT, pushEvents: true, contextHooks: { beforeInference: true, @@ -11974,6 +12297,7 @@ export class AgentFramework { // Discover tools from all connected servers await this.refreshMcplTools(); + await this.resumeCoalescedChannels(); } /** Reconcile the durable ledger with Chronicle, then deliver every server's work. */ @@ -12847,6 +13171,10 @@ export class AgentFramework { * Push events and inference requests are deferred to Steps 6/7. */ private wireMcplEvents(connection: McplServerConnection): void { + (this.coalescingEpochs ??= new Map()).set(connection.id, randomUUID()); + connection.on('orphaned-render-response', (response: Record) => { + this.pushCoalescer?.observe({ kind: 'late-render', serverId: connection.id, ...response }); + }); // Forward subprocess stderr lines as trace events so consumers (conhost, // log sinks, TUI badges) can persist and surface them. connection.on('stderr', (params: { line: string }) => { @@ -12870,7 +13198,7 @@ export class AgentFramework { ) => { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; - this.pushHandler?.handlePushEvent(connection.id, params, responder as never); + await this.pushHandler?.handlePushEvent(connection.id, params, responder as never); }); // Handle server-initiated inference requests (Step 6) @@ -12878,6 +13206,10 @@ export class AgentFramework { params: McplInferenceRequestParams, responder?: { id: string | number; respond: (result: unknown) => void; respondError: (code: number, message: string) => void }, ) => { + if (this.pushCoalescer?.isRendering(connection.id)) { + responder?.respondError(-32600, 'Cannot request inference while push/render is in progress'); + return; + } if (this.inferenceRouter && responder) { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; @@ -12895,6 +13227,7 @@ export class AgentFramework { responder?: { respond: (result: unknown) => void }, ) => { await this.channelRegistry?.handleRegister(connection.id, params, responder as never); + await this.resumeCoalescedChannels(connection.id); }); // Handle channel changes (Step 7) — §14.5 dual-mode: the responder is @@ -12951,7 +13284,7 @@ export class AgentFramework { ) => { const barrier = this.discordAwarenessBarrier; if (barrier) await barrier.promise; - this.channelRegistry?.handleIncoming(connection.id, params, responder as never); + await this.channelRegistry?.handleIncoming(connection.id, params, responder as never); }); // Handle host-level admin commands from a surface (e.g. Discord /undo) @@ -12996,6 +13329,7 @@ export class AgentFramework { // preserved across the transient close are resumed by idempotent // registration. Then refresh tools (server may have different tools). connection.on('reconnect', (info?: { attempts?: number }) => { + this.coalescingEpochs.set(connection.id, randomUUID()); // Install the barrier synchronously so any inbound event emitted after // reconnect observes it before doing work that could wake an agent. The // connection paused its data plane before wiring the fresh transport. @@ -13004,7 +13338,7 @@ export class AgentFramework { // grant) before the data-plane barrier releases — otherwise the first // post-reconnect events land while the grant is still empty and are // rejected fail-closed instead of delivered. - void (async () => { + const policyReady = (async () => { try { const config = this.mcplServerConfigs.get(connection.id); if (config) { @@ -13019,11 +13353,12 @@ export class AgentFramework { this.releaseMcplDataPlaneGate(awarenessBarrier); this.handleToolsListChanged(connection.id); })(); - void awarenessBarrier.promise.then(() => { + void Promise.all([policyReady, awarenessBarrier.promise]).then(async () => { if ( awarenessBarrier.requiresBarrier && !this.completeMcplDataPlaneGate(awarenessBarrier) ) return; + await this.resumeCoalescedChannels(connection.id); this.emitTrace({ type: 'mcpl:server-reconnected', serverId: connection.id, @@ -13113,6 +13448,9 @@ export class AgentFramework { // a SIGKILL preserves them. Permanent removal is owned solely by // disconnectMcplServer, which deletes the trees explicitly. connection.on('close', (code?: number | null, signal?: string | null) => { + this.pushCoalescer?.disconnect(connection.id); + this.channelRegistry?.invalidateServerRegistration?.(connection.id); + this.coalescingEpochs.delete(connection.id); this.featureSetManager?.removeServer(connection.id); this.emitTrace({ type: 'mcpl:server-closed', diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 6db0a737c..4418fc74c 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -48,6 +48,8 @@ const RELOAD_THROTTLE_MS = 1000; const MIN_DEBOUNCE_MS = 100; /** Maximum debounce value (ms). */ const MAX_DEBOUNCE_MS = 300_000; +/** Symbols cannot be supplied by a server's JSON origin/metadata. */ +export const COALESCING_SUBJECT = Symbol('mcpl.coalescingSubject'); // --------------------------------------------------------------------------- // Default config @@ -63,6 +65,8 @@ const DEFAULT_CONFIG: GateConfig = { // --------------------------------------------------------------------------- interface PendingEvent { + /** Host-minted RFC-006 subject; updates keep the original debounce deadline. */ + coalescingSubject?: string; policyName: string; content: string; eventType: string; @@ -96,6 +100,7 @@ interface PendingEvent { * "who woke the agent" from it). Ids only, never content or display names. */ export interface WakeProvenance { + coalescingSubject?: string; /** Composite channel id of the chosen event — telemetry only, never a * speech locus. Omitted for `mcpl:push-event` events, whose channel ids * are raw server ids (unroutable and a second spelling of the same @@ -519,6 +524,7 @@ export class EventGate { private configErrors: string[] = []; // Debounce state + private coalescedDebounce = new Map; event: PendingEvent }>(); private debounceTimers = new Map(); // Rate-limit state — policy name → key → bucket. @@ -1234,6 +1240,9 @@ export class EventGate { name: string, options: { deliverPendingDebounce: boolean }, ): void { + for (const [subject, state] of this.coalescedDebounce) { + if (state.event.policyName === name) this.cancelCoalesced(subject); + } const debounce = this.debounceTimers.get(name); if (debounce) { clearTimeout(debounce.timer); @@ -1424,6 +1433,10 @@ export class EventGate { metadata, tags: Array.isArray(metadata.tags) ? (metadata.tags as string[]) : undefined, }); + const subject = (metadata as Record)[COALESCING_SUBJECT]; + if (typeof subject === 'string' && !(typeof decision.behavior === 'object' && 'debounce' in decision.behavior)) { + this.cancelCoalesced(subject); + } return decision.trigger; }; } @@ -1436,6 +1449,7 @@ export class EventGate { const debounceMs = (policy.behavior as { debounce: number }).debounce; const event: PendingEvent = { + coalescingSubject: (info.metadata as Record | undefined)?.[COALESCING_SUBJECT] as string | undefined, policyName: policy.name, content: info.content.length > MAX_CONTENT_SNIPPET ? info.content.slice(0, MAX_CONTENT_SNIPPET) + '...' @@ -1453,6 +1467,22 @@ export class EventGate { addressed: Array.isArray(info.tags) && info.tags.includes('chat:addressed'), }; + if (event.coalescingSubject) { + const key = event.coalescingSubject; + const existing = this.coalescedDebounce.get(key); + if (existing) { existing.event = event; return; } + const timer = setTimeout(() => { + const current = this.coalescedDebounce.get(key); + this.coalescedDebounce.delete(key); + if (!current) return; + if (this.inferring.size || this.quiesced) this.bufferForInference([current.event]); + else this.deliverEvents([current.event]); + }, debounceMs); + this.coalescedDebounce.set(key, { timer, event }); + return; + } + // Ordinary events retain their own quiet-period debounce. A subject's fixed + // deadline must neither postpone nor accelerate an unrelated event. const existing = this.debounceTimers.get(policy.name); if (existing) { clearTimeout(existing.timer); @@ -1493,11 +1523,31 @@ export class EventGate { }); } + /** Withdraw a coalesced wake when its subject is consumed, removed, or revoked. */ + cancelCoalesced(subject: string): void { + const pending = this.coalescedDebounce.get(subject); + if (pending) clearTimeout(pending.timer); + this.coalescedDebounce.delete(subject); + for (const [name, state] of this.debounceTimers) { + state.events = state.events.filter(e => e.coalescingSubject !== subject); + if (!state.events.length) { clearTimeout(state.timer); this.debounceTimers.delete(name); } + } + this.inferenceBuffer = this.inferenceBuffer.filter(e => e.coalescingSubject !== subject); + } + // ========================================================================= // Event delivery // ========================================================================= private deliverEvents(events: PendingEvent[]): void { + for (const event of events.filter(e => e.coalescingSubject)) { + this.requestInferenceFn(this.getAgentNamesFn()[0] ?? '', 'gate:debounce', 'gate', { + ...wakeProvenance([event]), coalescingSubject: event.coalescingSubject, + }); + } + events = events.filter(e => !e.coalescingSubject); + if (!events.length) return; + if (events.length === 0) return; const policyNames = [...new Set(events.map(e => e.policyName))]; @@ -1507,8 +1557,11 @@ export class EventGate { // them here would inject a duplicate copy of the message minutes after // the fact, reading as fresh input the agent may have already handled. // Only events with no context entry of their own carry their content. - const quoted = events.filter(e => !e.inContext); - const referenced = events.filter(e => e.inContext); + // Coalesced pushes have no arrival-time context row. Their current content is + // materialized at assembly; a gate marker would leave traces of withdrawn work. + const contextEvents = events.filter(e => !e.coalescingSubject); + const quoted = contextEvents.filter(e => !e.inContext); + const referenced = contextEvents.filter(e => e.inContext); const lines: string[] = quoted.map(e => `- [${e.policyName}] (${e.eventType}): ${e.content}`); if (referenced.length > 0) { @@ -1534,9 +1587,9 @@ export class EventGate { } } - const text = `[Gate: ${events.length} event${events.length > 1 ? 's' : ''} matched]\n\n${lines.join('\n')}`; + const text = `[Gate: ${contextEvents.length} event${contextEvents.length > 1 ? 's' : ''} matched]\n\n${lines.join('\n')}`; - this.addMessageFn('user', [{ type: 'text', text }], { + if (contextEvents.length) this.addMessageFn('user', [{ type: 'text', text }], { source: 'gate:debounce', policies: policyNames, }); @@ -1637,7 +1690,9 @@ export class EventGate { ); } } - this.inferenceBuffer.push(event); + const prior = event.coalescingSubject ? this.inferenceBuffer.findIndex(e => e.coalescingSubject === event.coalescingSubject) : -1; + if (prior >= 0) this.inferenceBuffer[prior] = event; + else this.inferenceBuffer.push(event); } } @@ -1859,7 +1914,7 @@ export class EventGate { }; if (typeof p.behavior === 'object' && 'debounce' in p.behavior) { result.debounceState = { - pendingCount: debounce?.events.length ?? 0, + pendingCount: (debounce?.events.length ?? 0) + [...this.coalescedDebounce.values()].filter(state => state.event.policyName === p.name).length, nextDeliveryMs: null, // Timer internals not easily inspectable }; } @@ -1922,6 +1977,8 @@ export class EventGate { clearTimeout(state.timer); } this.debounceTimers.clear(); + for (const state of this.coalescedDebounce.values()) clearTimeout(state.timer); + this.coalescedDebounce.clear(); for (const timer of this.selfWakeTimers.values()) { clearTimeout(timer); } diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index ee1f9b2d3..28ac7a592 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -26,6 +26,7 @@ import type { ChannelsIncomingParams, ChannelsIncomingResult, ChannelIncomingMessageResult, + ChannelIncomingMessage, ChannelsPublishParams, ChannelsOpenResult, ChannelHistoryRequest, @@ -452,6 +453,8 @@ const CHANNEL_TOOL_DEFINITIONS: ToolDefinition[] = [ // ============================================================================ interface ChannelRegistryOptions { + authorizeIncoming?: (serverId: string, channelId: string) => void; + handleCoalescedIncoming?: (serverId: string, message: ChannelIncomingMessage) => Promise; /** Chronicle store used for durable desired channel lifecycle state. */ store?: JsStore; /** Callback to determine whether an incoming message should trigger inference. */ @@ -644,6 +647,19 @@ export class ChannelRegistry { private legacyPolicies = new Map(); private migratedLegacyPolicies = new Set(); + private incomingRegistrations = new Set(); + private authorizeIncoming?: ChannelRegistryOptions['authorizeIncoming']; + private handleCoalescedIncoming?: ChannelRegistryOptions['handleCoalescedIncoming']; + + invalidateServerRegistration(serverId: string): void { + this.removeServer(serverId); + } + + registeredDescriptor(serverId: string, channelId: string): ChannelDescriptor | undefined { + const key = `${serverId}:${channelId}`; + return this.incomingRegistrations.has(key) ? this.channels.get(key)?.descriptor : undefined; + } + constructor( serverRegistry: McplServerRegistry, featureSetManager: FeatureSetManager, @@ -658,6 +674,8 @@ export class ChannelRegistry { ) => void; }, ) { + this.authorizeIncoming = options?.authorizeIncoming; + this.handleCoalescedIncoming = options?.handleCoalescedIncoming; this.serverRegistry = serverRegistry; this.featureSetManager = featureSetManager; this.pushEventFn = pushEventFn; @@ -711,6 +729,7 @@ export class ChannelRegistry { continue; } const key = `${serverId}:${channel.id}`; + this.incomingRegistrations.add(key); this.channels.set(key, { serverId, descriptor: channel, @@ -762,6 +781,7 @@ export class ChannelRegistry { for (const channelId of params.removed) { const key = `${serverId}:${channelId}`; const existed = this.channels.delete(key); + this.incomingRegistrations.delete(key); this.stopTyping(channelId); addedResults.push({ id: channelId, accepted: existed, reason: existed ? undefined : 'not registered' }); } @@ -786,6 +806,7 @@ export class ChannelRegistry { continue; } existing.descriptor = channel; + this.incomingRegistrations.add(key); this.appendLabelSighting(channel.id, channel.label, extractDmMeta(channel.metadata)); addedResults.push({ id: channel.id, accepted: true }); } @@ -802,6 +823,7 @@ export class ChannelRegistry { continue; } const key = `${serverId}:${channel.id}`; + this.incomingRegistrations.add(key); this.channels.set(key, { serverId, descriptor: channel, @@ -838,14 +860,20 @@ export class ChannelRegistry { * Converts each message's content, pushes McplChannelIncomingEvent to the * queue, and responds with per-message results. */ - handleIncoming( + async handleIncoming( serverId: string, params: ChannelsIncomingParams, responder?: Responder, - ): void { + ): Promise { const results: ChannelIncomingMessageResult[] = []; for (const message of params.messages) { + if (!message || typeof message.channelId !== 'string' || !message.channelId || + typeof message.messageId !== 'string' || !message.messageId) { + results.push({ messageId: typeof message?.messageId === 'string' ? message.messageId : '', + accepted: false, reason: message?.coalesce !== undefined ? 'coalesce_invalid' : 'invalid channel/message identity' }); + continue; + } // §14.5 FIRST, before ANY semantic processing: admission against the // actually-registered channel precedes tag expansion and content // conversion — decoding an unregistered sender's payload (including @@ -888,6 +916,31 @@ export class ChannelRegistry { continue; } + try { + this.authorizeIncoming?.(serverId, message.channelId); + if (message.coalesce !== undefined) { + if (!this.handleCoalescedIncoming) { + results.push({ messageId: message.messageId, accepted: false, reason: 'coalesce_invalid' }); + continue; + } + const result = await this.handleCoalescedIncoming(serverId, message); + results.push(result); + if (result.accepted) { + this.defaultPublishChannel = message.channelId; + this.defaultPublishMessageId = message.messageId; + this.defaultPublishThreadId = message.threadId; + const entry = this.channels.get(incomingKey); + if (entry) entry.open = true; + } + continue; + } + } catch (error) { + const err = error as Error & { code?: number }; + results.push({ messageId: message.messageId, accepted: false, + reason: err.code === -32602 ? 'coalesce_invalid' : err.message }); + continue; + } + // ACCEPTED from here down: semantic processing only for admitted // messages. §16.3 core-tag closure, then content conversion. if (message.tags) message.tags = expandCoreTags(message.tags); @@ -1323,6 +1376,7 @@ export class ChannelRegistry { for (const [key, entry] of this.channels) { if (entry.serverId !== serverId) continue; this.channels.delete(key); + this.incomingRegistrations.delete(key); this.stopTyping(entry.descriptor.id); if (this.defaultPublishChannel === entry.descriptor.id) { this.defaultPublishChannel = null; @@ -2229,6 +2283,13 @@ export class ChannelRegistry { address: entry.descriptor.address, ...(history ? { history } : {}), }); + // A typed response to a host-authorized open is a server declaration too; + // an origin-derived outbound placeholder by itself is not. + const key = `${entry.serverId}:${entry.descriptor.id}`; + if (result?.channel?.id === entry.descriptor.id && this.channels.get(key) === entry && + server.policyEstablished && CapabilityGrant.of(server).has('channels.lifecycle')) { + this.incomingRegistrations.add(key); + } entry.open = true; return result; } diff --git a/src/mcpl/chronicle-coalescing-journal.ts b/src/mcpl/chronicle-coalescing-journal.ts new file mode 100644 index 000000000..733542549 --- /dev/null +++ b/src/mcpl/chronicle-coalescing-journal.ts @@ -0,0 +1,63 @@ +import { createHash, randomUUID } from 'node:crypto'; +import type { JsStore } from '@animalabs/chronicle'; +import type { CoalescingJournal, CoalescingRecord, LegacyCoalescingImport } from './coalescing-journal.js'; +import type { PushEventResult } from './types.js'; + +export const COALESCING_JOURNAL_ID = 'mcpl/coalescing-journal'; +const LEGACY_PENDING = 'mcpl/coalescing-pending'; +const LEGACY_RECEIPTS = 'mcpl/coalescing-receipts'; +const LEGACY_AUDIT = 'mcpl/coalescing-audit'; + +/** One append-only Chronicle state owns coalescing. No pending snapshot or receipt + * table is written. Chronicle's own append-log snapshots are storage checkpoints, + * not a second coalescing authority. Reads replay immutable operations by position. */ +export class ChronicleCoalescingJournal implements CoalescingJournal { + constructor(private readonly store: JsStore, private readonly branch = store.currentBranch().name) { + const states = new Set(store.listStates().map(s => s.id)); + if (!states.has(COALESCING_JOURNAL_ID)) store.registerState({ id: COALESCING_JOURNAL_ID, strategy: 'append_log' }); + if (this.length() === 0) this.importLegacy(states); + } + private assertBranch(): void { + if (this.store.currentBranch().name !== this.branch) throw new Error('Coalescing journal belongs to a different branch'); + } + length(): number { + this.assertBranch(); + return this.store.getStateLen(COALESCING_JOURNAL_ID) ?? 0; + } + read(index: number): CoalescingRecord { + this.assertBranch(); + const record = this.store.getStateItemJson(COALESCING_JOURNAL_ID, index); + if (!record || record.version !== 1 || !record.operation) throw new Error(`Invalid coalescing journal entry ${index}`); + return record as CoalescingRecord; + } + append(record: CoalescingRecord): void { + this.assertBranch(); + this.store.appendToStateJson(COALESCING_JOURNAL_ID, record); + this.store.sync(); // acknowledge/effect only after the operation is durable + } + + private importLegacy(states: Set): void { + const snapshot = states.has(LEGACY_PENDING) ? this.store.getStateJson(LEGACY_PENDING) : null; + const receipts = new Map(); + if (states.has(LEGACY_RECEIPTS)) { + for (const entry of this.store.treeList(LEGACY_RECEIPTS)) { + const bytes = this.store.getBlob(entry.blobHash); + if (!bytes) throw new Error('Missing legacy coalescing receipt blob'); + receipts.set(entry.path, JSON.parse(bytes.toString())); + } + } + if (snapshot?.receipt) { + receipts.set(createHash('sha256').update(snapshot.receipt[0]).digest('hex'), snapshot.receipt[1]); + } + const hadAudit = states.has(LEGACY_AUDIT) && (this.store.getStateLen(LEGACY_AUDIT) ?? 0) > 0; + if (!snapshot?.pending?.length && !snapshot?.history?.length && !receipts.size && !hadAudit) return; + if (snapshot && snapshot.version !== 1 && snapshot.version !== 2) throw new Error('Unsupported legacy coalescing state'); + const initial: LegacyCoalescingImport = { + pending: snapshot?.pending ?? [], history: snapshot?.history ?? [], + receipts: [...receipts].map(([key, result]) => ({ key, result })), + }; + this.append({ version: 1, id: randomUUID(), at: Date.now(), operation: { kind: 'legacy-import', initial } }); + // The old stores remain untouched for inspection. Once the import is appended, + // neither their continued presence nor their contents affect replay. + } +} diff --git a/src/mcpl/coalescing-journal.ts b/src/mcpl/coalescing-journal.ts new file mode 100644 index 000000000..e75d2e767 --- /dev/null +++ b/src/mcpl/coalescing-journal.ts @@ -0,0 +1,287 @@ +/** Canonical coalescing operations and their deterministic projection. + * Only this reducer changes logical coalescing state. Replaying it performs no I/O. */ +import { createHash } from 'node:crypto'; +import type { McplContentBlock, PushEventParams, PushEventResult, ChannelIncomingMessage } from './types.js'; + +export interface CoalescedPush { + /** Host-only recovery marker: never read from wire params or origin. */ + recovered?: boolean; + /** Journal-owned publication identity; never taken from a wire message. */ + publicationId?: string; + serverId: string; + /** Host-owned configured binding; changes when the endpoint/command is reassigned. */ + binding?: string; + epoch: string; + audience: string; + params: PushEventParams; + channelMessage?: ChannelIncomingMessage; + /** Host-resolved delivery, independent of the untrusted coalescing scope. */ + routing?: { + channelId?: string; messageId?: string; author?: { id: string; name: string }; + threadId?: string; metadata?: Record; + targetAgents: string[]; triggerAllowed?: boolean; generation?: number; + }; +} +export interface PushRenderParams { + channelId?: string; + featureSet: string; + key: string; + eventId: string; + notices: Array<{ eventId: string; timestamp: string; data?: unknown }>; + dropped: number; +} +export interface PushRenderResult { content: McplContentBlock[]; timestamp?: string } +/** A configured server binding survives transport reconnects. Epoch gates authority, + * not subject or occurrence identity. Channel scope is NEVER inferred from origin. */ +export function coalescingChannel(push: CoalescedPush): string | undefined { + return push.channelMessage?.channelId ?? push.params.coalesce?.channelId; +} +export function coalescingSubject(push: CoalescedPush): string { + const channel = coalescingChannel(push); + return JSON.stringify([push.serverId, push.binding ?? '', channel === undefined ? 'featureSet' : 'channel', + channel ?? push.params.featureSet, push.params.coalesce!.key]); +} + +export function coalescingReceiptKey(push: CoalescedPush): string { + return createHash('sha256').update(JSON.stringify([push.serverId, push.binding ?? '', push.params.eventId])).digest('hex'); +} + +type History = 'none' | 'some' | 'unknown'; +type MessageIdentity = Pick, 'messageId' | 'author' | 'threadId'>; +export interface CoalescingSlot { + push: CoalescedPush; + notices?: PushRenderParams['notices']; + dropped: number; +} +export interface CoalescingSubjectState { + history: History; + identity?: MessageIdentity; + consumedEventId?: string; + slot?: CoalescingSlot; + rendering?: { id: string; slot: CoalescingSlot }; +} +export type CoalescingTarget = { slot: string } | { render: string }; +export interface CoalescingPublication { + id: string; + subject: string; + push: CoalescedPush; +} +/** One-time import of the unmerged snapshot-based prototype. Once appended, the + * imported facts belong to this journal; the old stores are never read again. */ +export interface LegacyCoalescingImport { + pending: CoalescedPush[]; + history?: Array<{ key: string; history: History; consumedEventId?: string; identity?: MessageIdentity }>; + receipts: Array<{ key: string; result: PushEventResult }>; +} +export type CoalescingOperation = + | { kind: 'accepted'; push: CoalescedPush; result: PushEventResult } + | { kind: 'forgotten'; subject: string } + | { kind: 'render-started'; subject: string } + | { kind: 'prepared'; subject: string; target: CoalescingTarget; push: CoalescedPush; + source: 'plain' | 'render' | 'fallback'; observation?: unknown } + | { kind: 'delivered'; publication: string; disposition: 'published' | 'suppressed' } + | { kind: 'discarded'; subject: string; target: CoalescingTarget; reason: 'empty-render' | 'authority'; observation?: unknown } + | { kind: 'disconnected'; serverId: string } + | { kind: 'recovered'; branch?: string } + | { kind: 'history-unknown' } + | { kind: 'observed'; detail: Record } + | { kind: 'legacy-import'; initial: LegacyCoalescingImport }; +export interface CoalescingRecord { + version: 1; + id: string; + at: number; + operation: CoalescingOperation; +} +/** append must commit durably before returning. Derived indexes are disposable. */ +export interface CoalescingJournal { + length(): number; + read(index: number): CoalescingRecord; + append(record: CoalescingRecord): void; +} +export interface ReceiptReference { record: number; imported?: number } + +function sameAudience(prior: CoalescedPush | undefined, push: CoalescedPush): boolean { + return !prior || prior.audience === push.audience && + JSON.stringify(prior.routing?.metadata?.tuneOut ?? null) === JSON.stringify(push.routing?.metadata?.tuneOut ?? null); +} +function audienceBranch(push: CoalescedPush): string | undefined { + try { return JSON.parse(push.audience)[0]; } catch { return undefined; } +} + +export class CoalescingProjection { + readonly subjects = new Map(); + readonly publications = new Map(); + readonly receipts = new Map(); + uncertain = false; + cursor = 0; + + resultFor(push: CoalescedPush): PushEventResult { + const state = this.subjects.get(coalescingSubject(push)); + const c = push.params.coalesce!; + const history = state?.history ?? (this.uncertain || c.retract ? 'unknown' : 'none'); + const prior = state?.slot ?? state?.rendering?.slot; + const priorEventId = prior?.push.params.eventId ?? state?.consumedEventId; + const outcome = c.retract + ? history === 'none' ? 'retracted' : push.params.payload.content.length ? 'noted' : 'consumed' + : c.deferred ? state?.slot ? 'replaced' : 'first' + : prior ? 'replaced' : state?.consumedEventId || history === 'unknown' ? 'appended' : 'first'; + return { accepted: true, coalesce: { outcome, ...(priorEventId ? { priorEventId } : {}) } }; + } + + /** Validate and apply a committed fact, independent of the current grant, clock, + * routing policy, sockets, or process. No effects belong in this method. */ + apply(record: CoalescingRecord, index: number): void { + if (record?.version !== 1 || typeof record.id !== 'string' || index !== this.cursor) { + throw new Error(`Invalid coalescing journal record at ${index}`); + } + const op = record.operation; + switch (op.kind) { + case 'accepted': { + const push = structuredClone(op.push); + const key = coalescingSubject(push); + const expected = this.resultFor(push); + if (op.result.accepted !== true || expected.coalesce?.outcome !== op.result.coalesce?.outcome || + expected.coalesce?.priorEventId !== op.result.coalesce?.priorEventId || this.receipts.has(coalescingReceiptKey(push))) { + throw new Error(`Inconsistent coalescing acceptance at ${index}`); + } + const c = push.params.coalesce!; + const state = this.subjects.get(key) ?? { history: this.uncertain || c.retract ? 'unknown' : 'none' }; + const prior = state.slot ?? state.rendering?.slot; + const deliver = !!push.audience && sameAudience(prior?.push, push); + if (c.retract || !c.deferred) { + state.rendering = undefined; + state.slot = undefined; + if (c.retract) state.consumedEventId = undefined; + if (deliver && (!c.retract || op.result.coalesce?.outcome === 'noted')) state.slot = { push, dropped: 0 }; + } else if (deliver) { + const notices = state.slot?.notices ?? []; + let dropped = state.slot?.dropped ?? 0; + notices.push({ eventId: push.params.eventId, timestamp: push.params.timestamp, + ...(c.data !== undefined ? { data: structuredClone(c.data) } : {}) }); + if (notices.length > 64) { notices.shift(); dropped++; } + state.slot = { push, notices, dropped }; + } else state.slot = undefined; + if (push.routing && coalescingChannel(push)) { + state.identity = { messageId: push.routing.messageId, author: push.routing.author, threadId: push.routing.threadId }; + } + this.subjects.set(key, state); + this.receipts.set(coalescingReceiptKey(push), { record: index }); + break; + } + case 'forgotten': { + const state = this.subjects.get(op.subject); + if (!state || state.slot || state.rendering || [...this.publications.values()].some(p => p.subject === op.subject)) { + throw new Error('Cannot forget pending coalescing work'); + } + this.subjects.delete(op.subject); + this.uncertain = true; + break; + } + case 'render-started': { + const state = this.requireSubject(op.subject); + if (!state.slot?.notices || state.rendering) throw new Error('Invalid render start'); + state.rendering = { id: record.id, slot: state.slot }; + state.slot = undefined; + break; + } + case 'prepared': { + const state = this.requireTarget(op.subject, op.target); + if (coalescingSubject(op.push) !== op.subject) throw new Error('Publication subject mismatch'); + this.clearTarget(state, op.target); + // The publication may have escaped before a crash. The append is the + // irreversible assembly fence; pending outbox work cannot be replaced. + if (state.history === 'none') state.history = 'unknown'; + state.consumedEventId = op.push.params.eventId; + this.publications.set(record.id, { id: record.id, subject: op.subject, push: structuredClone(op.push) }); + break; + } + case 'delivered': { + const publication = this.publications.get(op.publication); + if (!publication) throw new Error('Unknown coalescing publication'); + if (op.disposition === 'published') this.requireSubject(publication.subject).history = 'some'; + this.publications.delete(op.publication); + break; + } + case 'discarded': this.clearTarget(this.requireTarget(op.subject, op.target), op.target); break; + case 'disconnected': { + for (const state of this.subjects.values()) { + const slot = state.slot ?? state.rendering?.slot; + if (slot?.push.serverId === op.serverId) this.recoverSubject(state); + } + for (const publication of this.publications.values()) { + if (publication.push.serverId === op.serverId) publication.push.recovered = true; + } + break; + } + case 'recovered': { + for (const state of this.subjects.values()) { + const slot = state.slot ?? state.rendering?.slot; + if (slot && op.branch !== undefined && audienceBranch(slot.push) !== op.branch) { + state.slot = undefined; + state.rendering = undefined; + } else this.recoverSubject(state); + } + for (const [id, publication] of this.publications) { + if (op.branch !== undefined && audienceBranch(publication.push) !== op.branch) this.publications.delete(id); + else publication.push.recovered = true; + } + break; + } + case 'history-unknown': { + this.uncertain = true; + for (const state of this.subjects.values()) state.history = 'unknown'; + break; + } + case 'legacy-import': { + if (index !== 0) throw new Error('Legacy import must be the first journal entry'); + this.uncertain = true; + for (const { key, ...state } of op.initial.history ?? []) this.subjects.set(key, structuredClone(state)); + op.initial.pending.forEach((push, pendingIndex) => { + const key = coalescingSubject(push); + const state = this.subjects.get(key) ?? { history: 'unknown' as const }; + // The old snapshot and context write were independent. We cannot prove + // an imported item was unread, so seal it conservatively as an outbox + // intent. Legacy context markers make its publication idempotent. + if (state.history !== 'some') state.history = 'unknown'; + state.consumedEventId = push.params.eventId; + this.subjects.set(key, state); + const id = `${record.id}/legacy/${pendingIndex}`; + this.publications.set(id, { id, subject: key, push: { ...structuredClone(push), recovered: true } }); + }); + op.initial.receipts.forEach((receipt, imported) => this.receipts.set(receipt.key, { record: index, imported })); + break; + } + case 'observed': break; + default: throw new Error(`Unknown coalescing journal operation at ${index}`); + } + this.cursor++; + } + + private requireSubject(key: string): CoalescingSubjectState { + const state = this.subjects.get(key); + if (!state) throw new Error(`Unknown coalescing subject ${key}`); + return state; + } + private requireTarget(key: string, target: CoalescingTarget): CoalescingSubjectState { + const state = this.requireSubject(key); + if ('slot' in target ? state.slot?.push.params.eventId !== target.slot : state.rendering?.id !== target.render) { + throw new Error('Coalescing journal target no longer pending'); + } + return state; + } + private clearTarget(state: CoalescingSubjectState, target: CoalescingTarget): void { + if ('slot' in target) state.slot = undefined; + else state.rendering = undefined; + } + private recoverSubject(state: CoalescingSubjectState): void { + const slot = state.slot ?? state.rendering?.slot; + state.rendering = undefined; + if (slot) state.slot = { push: { ...slot.push, recovered: true }, dropped: 0 }; + } +} + +export function replayCoalescingJournal(journal: CoalescingJournal): CoalescingProjection { + const projection = new CoalescingProjection(); + for (let index = 0; index < journal.length(); index++) projection.apply(journal.read(index), index); + return projection; +} diff --git a/src/mcpl/push-coalescer.ts b/src/mcpl/push-coalescer.ts new file mode 100644 index 000000000..a4f84ce81 --- /dev/null +++ b/src/mcpl/push-coalescer.ts @@ -0,0 +1,355 @@ +/** Runtime orchestration for RFC-006. The journal is authoritative; the projection + * is rebuilt by the same reducer used after every live append. Timers and RPCs are + * process-local effects, never facts reconstructed from a snapshot. */ +import { randomUUID } from 'node:crypto'; +import type { McplContentBlock, PushEventResult } from './types.js'; +import { + CoalescingProjection, replayCoalescingJournal, coalescingSubject, coalescingChannel, coalescingReceiptKey, + type CoalescedPush, type CoalescingJournal, type CoalescingOperation, type CoalescingRecord, + type PushRenderParams, type PushRenderResult, +} from './coalescing-journal.js'; +export { coalescingSubject, coalescingChannel, coalescingReceiptKey } from './coalescing-journal.js'; +export type { CoalescedPush, CoalescingJournal, CoalescingRecord, PushRenderParams, PushRenderResult } from './coalescing-journal.js'; + +export const PUSH_COALESCING_SUPPORT = { + pushEvents: true, deferred: true, channelsIncoming: true, channelScopedPush: true, +} as const; +export interface CoalescerOptions { + journal: CoalescingJournal; + authorized(push: CoalescedPush): boolean; + available?(push: CoalescedPush): boolean; + render(push: CoalescedPush, params: PushRenderParams): Promise; + wake(push: CoalescedPush, subject: string): boolean; + cancelWake(subject: string, consumed?: boolean): void; + /** Route late observations to the currently active branch's journal. */ + observeLate?(detail: Record): void; + timeoutMs?: number; + maxSubjects?: number; + maxContentBytes?: number; +} +type PublishCoalesced = (push: CoalescedPush) => boolean | void; +interface RunningRender { + id: string; + subject: string; + serverId: string; + cancel(): void; + done: Promise; + settled: boolean; +} +export class CoalesceError extends Error { + constructor(readonly field: string, message: string, readonly code = -32602) { super(message); } +} + +/** Validate wire content before it is stored, rendered, or converted. */ +export function validateCoalescedContent(content: unknown, maxBytes = 1024 * 1024): asserts content is McplContentBlock[] { + if (!Array.isArray(content)) throw new CoalesceError('payload.content', 'content must be an array'); + for (const b of content) { + if (!b || typeof b !== 'object' || Array.isArray(b)) throw new CoalesceError('payload.content', 'invalid content block'); + if (b.type === 'text' && typeof b.text === 'string') continue; + if (b.type === 'resource' && typeof b.uri === 'string') continue; + if ((b.type === 'image' || b.type === 'audio') && + (typeof b.uri === 'string' || (typeof b.data === 'string' && typeof b.mimeType === 'string'))) continue; + throw new CoalesceError('payload.content', 'invalid content block'); + } + if (Buffer.byteLength(JSON.stringify(content)) > maxBytes) throw new CoalesceError('payload.content', 'content exceeds host byte limit'); +} + +export class PushCoalescer { + private readonly projection: CoalescingProjection; + private readonly renders = new Map(); + private readonly queuedWakes = new Set(); + private readonly publicationAttempts = new Set(); + private suspended = false; + private faulted = false; + + constructor(private readonly options: CoalescerOptions) { + this.projection = replayCoalescingJournal(options.journal); + // A prepared effect from a former process may already exist in context. + for (const id of this.projection.publications.keys()) this.publicationAttempts.add(id); + } + + /** Observation only: disposable, deterministic projection data for inspection. */ + inspect() { + return structuredClone({ cursor: this.projection.cursor, uncertain: this.projection.uncertain, + subjects: [...this.projection.subjects], receipts: [...this.projection.receipts], + publications: [...this.projection.publications] }); + } + + /** Record the new process/branch boundary explicitly. Replaying the journal alone + * performs no recovery writes, network requests, context writes, or wakes. */ + recover(branch?: string): void { this.commit({ kind: 'recovered', ...(branch === undefined ? {} : { branch }) }); } + forgetHistory(): void { this.commit({ kind: 'history-unknown' }); } + + private authorized(push: CoalescedPush): boolean { return this.options.authorized(structuredClone(push)); } + private available(push: CoalescedPush): boolean { return this.options.available?.(structuredClone(push)) ?? true; } + + private assertActive(): void { + if (this.suspended) throw new CoalesceError('serverId', 'host is stopping', -32000); + if (this.faulted) throw new Error('Coalescing journal append failed; rebuild the controller from the journal before continuing'); + } + + private commit(operation: CoalescingOperation): CoalescingRecord { + this.assertActive(); + const index = this.projection.cursor; + const record: CoalescingRecord = { version: 1, id: randomUUID(), at: Date.now(), operation: structuredClone(operation) }; + try { + if (this.options.journal.length() !== index) throw new Error('Coalescing journal changed outside this controller'); + this.options.journal.append(record); // write-ahead and durable BEFORE any effect + const committed = this.options.journal.read(index); + if (committed.id !== record.id) throw new Error('Coalescing journal append identity mismatch'); + this.projection.apply(committed, index); + // Cancellation follows the committed projection, never an independent flag. + for (const render of this.renders.values()) { + if (!this.isCurrentRender(render.subject, render.id)) render.cancel(); + } + return committed; + } catch (error) { + // An append may have committed before the storage error. Never retry from a + // guessed in-memory state; a fresh replay resolves that uncertainty. + this.faulted = true; + for (const render of this.renders.values()) render.cancel(); + for (const subject of this.projection.subjects.keys()) this.cancelWake(subject); + throw error; + } + } + + observe(detail: Record): void { + if (!this.suspended) this.commit({ kind: 'observed', detail }); + } + + isRendering(serverId: string): boolean { + return [...this.renders.values()].some(r => r.serverId === serverId && !r.settled); + } + private isCurrentRender(subject: string, id: string): boolean { + return this.projection.subjects.get(subject)?.rendering?.id === id; + } + private pendingValue(subject: string): CoalescedPush | undefined { + const state = this.projection.subjects.get(subject); + return state?.slot?.push ?? state?.rendering?.slot.push ?? + [...this.projection.publications.values()].find(p => p.subject === subject)?.push; + } + pending(subject: string): CoalescedPush | undefined { + const push = this.pendingValue(subject); + return push && structuredClone(push); + } + pendingPushes(): CoalescedPush[] { + return [...this.projection.subjects.keys()].flatMap(key => { const push = this.pending(key); return push ? [push] : []; }); + } + canAssemble(audience: string): boolean { + return this.pendingPushes().some(push => push.audience === audience && + this.available(push) && this.authorized(push)); + } + identity(subject: string) { + const state = this.projection.subjects.get(subject); + const routing = (state?.slot ?? state?.rendering?.slot)?.push.routing; + return structuredClone(routing + ? { messageId: routing.messageId, author: routing.author, threadId: routing.threadId } + : state?.identity); + } + occurrence(subject: string): string | undefined { + const state = this.projection.subjects.get(subject); + return (state?.slot ?? state?.rendering?.slot)?.push.params.eventId ?? state?.consumedEventId; + } + receipt(push: CoalescedPush): PushEventResult | undefined { + this.assertActive(); + const key = coalescingReceiptKey(push); + const ref = this.projection.receipts.get(key); + if (!ref) return undefined; + // The index stores only a journal position. The receipt itself is read from + // the authoritative operation, not from a separately persisted lookup table. + const op = this.options.journal.read(ref.record).operation; + if (op.kind === 'accepted' && coalescingReceiptKey(op.push) === key) return structuredClone(op.result); + if (op.kind === 'legacy-import' && ref.imported !== undefined && op.initial.receipts[ref.imported]?.key === key) { + return structuredClone(op.initial.receipts[ref.imported].result); + } + throw new Error('Coalescing receipt projection does not match its journal record'); + } + wakeRecovered(): void { + if (this.suspended || this.faulted) return; + for (const key of this.projection.subjects.keys()) { + const push = this.pendingValue(key); + if (push?.recovered && !this.queuedWakes.has(key) && this.authorized(push)) this.wake(key); + } + } + private wake(subject: string): void { + const push = this.pending(subject); + if (push && this.options.wake(push, subject)) this.queuedWakes.add(subject); + else this.queuedWakes.delete(subject); + } + private cancelWake(subject: string, consumed = false): void { + this.queuedWakes.delete(subject); + this.options.cancelWake(subject, consumed); + } + validate(push: CoalescedPush): void { + this.assertActive(); + const { params } = push; + const c = params.coalesce; + if (!c || typeof c !== 'object' || Array.isArray(c)) throw new CoalesceError('coalesce', 'coalesce must be an object'); + if (typeof c.key !== 'string' || !Buffer.byteLength(c.key) || Buffer.byteLength(c.key) > 256) { + throw new CoalesceError('coalesce.key', 'key must be 1..256 UTF-8 bytes'); + } + if (c.channelId !== undefined && (typeof c.channelId !== 'string' || !c.channelId)) + throw new CoalesceError('coalesce.channelId', 'channelId must be non-empty'); + if (push.channelMessage && ('channelId' in c || 'deferred' in c)) + throw new CoalesceError('coalesce', 'channel messages cannot select scope or deferred mode'); + for (const field of ['deferred', 'retract'] as const) { + if (c[field] !== undefined && typeof c[field] !== 'boolean') throw new CoalesceError(`coalesce.${field}`, 'expected boolean'); + } + if (c.deferred && c.retract) throw new CoalesceError('coalesce', 'deferred and retract are exclusive'); + if (c.data !== undefined && !c.deferred) throw new CoalesceError('coalesce.data', 'data requires deferred'); + if (c.data !== undefined && Buffer.byteLength(JSON.stringify(c.data)) > 4096) throw new CoalesceError('coalesce.data', 'data exceeds 4 KiB'); + if (typeof params.eventId !== 'string' || !params.eventId) throw new CoalesceError('eventId', 'eventId is required'); + if (typeof params.timestamp !== 'string') throw new CoalesceError('timestamp', 'timestamp is required'); + validateCoalescedContent(params.payload?.content, this.options.maxContentBytes); + if (coalescingChannel(push) !== undefined && c.retract && !params.payload.content.length) + throw new CoalesceError('payload.content', 'channel retractions require a deletion notice'); + if (!this.authorized(push)) throw new CoalesceError('featureSet', 'push is no longer authorized', -32002); + } + + accept(push: CoalescedPush): PushEventResult { + this.assertActive(); + this.validate(push); + const duplicate = this.receipt(push); + if (duplicate) return duplicate; + const subject = coalescingSubject(push); + if (!this.projection.subjects.has(subject)) { + while (this.projection.subjects.size >= (this.options.maxSubjects ?? 128)) { + const idle = [...this.projection.subjects].find(([key, state]) => !state.slot && !state.rendering && + ![...this.projection.publications.values()].some(p => p.subject === key)); + if (!idle) throw new CoalesceError('coalesce.key', 'host pending-subject limit reached', -32000); + this.commit({ kind: 'forgotten', subject: idle[0] }); + } + } + const result = this.projection.resultFor(push); + this.commit({ kind: 'accepted', push, result }); + if (this.projection.subjects.get(subject)?.slot) this.wake(subject); + else if (!this.pendingValue(subject)) this.cancelWake(subject); + return structuredClone(result); + } + + /** The prepared operation is an irreversible assembly fence and durable outbox + * intent. Context publication is idempotent; completion is another append. */ + private deliver(audience: string, publish: PublishCoalesced): void { + for (const publication of [...this.projection.publications.values()]) { + if (publication.push.audience !== audience || !this.available(publication.push)) continue; + if (!this.authorized(publication.push)) { + this.commit({ kind: 'delivered', publication: publication.id, disposition: 'suppressed' }); + this.publicationAttempts.delete(publication.id); + continue; + } + const recovered = publication.push.recovered || this.publicationAttempts.has(publication.id); + this.publicationAttempts.add(publication.id); + const published = publish({ ...structuredClone(publication.push), publicationId: publication.id, recovered }); + this.commit({ kind: 'delivered', publication: publication.id, disposition: published === false ? 'suppressed' : 'published' }); + this.publicationAttempts.delete(publication.id); + } + } + + async assemble(audience: string, publish: PublishCoalesced): Promise { + if (this.suspended) return; + this.assertActive(); + this.deliver(audience, publish); + const selected = [...this.projection.subjects].filter(([, state]) => { + const push = (state.slot ?? state.rendering?.slot)?.push; + return push?.audience === audience && this.available(push); + }).map(([key]) => key); + await Promise.all(selected.map(async subject => { + const state = this.projection.subjects.get(subject)!; + if (state.rendering) { + const running = this.renders.get(state.rendering.id); + if (!running) throw new Error('Unfinished journal render requires recovery before assembly'); + await running.done; + return; + } + if (!state.slot?.notices) return; + if (!this.authorized(state.slot.push)) { + this.commit({ kind: 'discarded', subject, target: { slot: state.slot.push.params.eventId }, reason: 'authority' }); + this.cancelWake(subject); + return; + } + const record = this.commit({ kind: 'render-started', subject }); + this.cancelWake(subject, true); + let cancel!: () => void; + const cancelled = new Promise<'cancelled'>(resolve => { cancel = () => resolve('cancelled'); }); + const render: RunningRender = { id: record.id, subject, serverId: state.rendering!.slot.push.serverId, + cancel, done: Promise.resolve(), settled: false }; + this.renders.set(render.id, render); + render.done = this.runRender(render, cancelled); + await render.done; + })); + if (this.suspended) return; + this.assertActive(); + for (const subject of selected) { + const slot = this.projection.subjects.get(subject)?.slot; + if (!slot || slot.notices || slot.push.audience !== audience || !this.available(slot.push)) continue; + const target = { slot: slot.push.params.eventId }; + if (this.authorized(slot.push)) this.commit({ kind: 'prepared', subject, target, push: slot.push, source: slot.push.params.coalesce?.deferred ? 'fallback' : 'plain' }); + else this.commit({ kind: 'discarded', subject, target, reason: 'authority' }); + this.cancelWake(subject, true); + } + this.deliver(audience, publish); + } + + private async runRender(render: RunningRender, cancelled: Promise<'cancelled'>): Promise { + const { subject, id } = render; + const slot = structuredClone(this.projection.subjects.get(subject)!.rendering!.slot); + const push = slot.push; + let timer: ReturnType | undefined; + try { + const params: PushRenderParams = { featureSet: push.params.featureSet, + ...(coalescingChannel(push) ? { channelId: coalescingChannel(push) } : {}), key: push.params.coalesce!.key, + eventId: push.params.eventId, notices: slot.notices!, dropped: slot.dropped }; + const request = Promise.resolve().then(() => this.options.render(structuredClone(push), params)).then(result => { + const detail = { kind: render.settled || !this.isCurrentRender(subject, id) ? 'late-render' : 'render-response', + subject, renderId: id, result }; + if (this.options.observeLate) this.options.observeLate(detail); + else this.observe(detail); + return { result }; + }, error => ({ error: String(error) })); + const response = await Promise.race([request, cancelled, + new Promise<'timeout'>(resolve => { timer = setTimeout(() => resolve('timeout'), this.options.timeoutMs ?? 5000); })]); + render.settled = true; + if (this.suspended || response === 'cancelled' || !this.isCurrentRender(subject, id)) return; + this.assertActive(); + const target = { render: id }; + if (!this.authorized(push)) { + this.commit({ kind: 'discarded', subject, target, reason: 'authority', observation: response }); + return; + } + let content = push.params.payload.content; + let timestamp = push.params.timestamp; + let source: 'render' | 'fallback' = 'fallback'; + if (typeof response === 'object' && 'result' in response) { + try { + validateCoalescedContent(response.result.content, this.options.maxContentBytes); + content = response.result.content; + timestamp = typeof response.result.timestamp === 'string' ? response.result.timestamp : new Date().toISOString(); + source = 'render'; + } catch { /* The admitted fallback remains the bounded fail-open content. */ } + } + if (!content.length) this.commit({ kind: 'discarded', subject, target, reason: 'empty-render', observation: response }); + else this.commit({ kind: 'prepared', subject, target, + push: { ...push, params: { ...push.params, timestamp, payload: { content } } }, source, observation: response }); + } finally { + render.settled = true; + clearTimeout(timer); + this.renders.delete(id); + } + } + + disconnect(serverId: string): void { + if (this.suspended) return; + this.commit({ kind: 'disconnected', serverId }); + for (const subject of this.projection.subjects.keys()) { + if (this.pendingValue(subject)?.serverId === serverId) this.cancelWake(subject); + } + } + + /** Stopping only cancels process-local effects. The next controller replays the + * unchanged log and appends a recovery decision for unfinished work. */ + suspend(): void { + this.suspended = true; + for (const render of this.renders.values()) render.cancel(); + for (const subject of this.projection.subjects.keys()) this.cancelWake(subject); + } +} diff --git a/src/mcpl/push-handler.ts b/src/mcpl/push-handler.ts index a9cde38b5..08faba461 100644 --- a/src/mcpl/push-handler.ts +++ b/src/mcpl/push-handler.ts @@ -171,6 +171,7 @@ export class PushHandler { pushEventFn: (event: McplPushEvent) => void, emitTraceFn: (event: { type: string; [key: string]: unknown }) => void, shouldTriggerInference?: (content: string, metadata: Record) => boolean, + private readonly handleCoalesced?: (serverId: string, params: PushEventParams) => PushEventResult | Promise, ) { this.featureSetManager = featureSetManager; this.pushEventFn = pushEventFn; @@ -189,11 +190,11 @@ export class PushHandler { * 6. Emit trace * 7. Respond with accepted + inferenceId */ - handlePushEvent( + async handlePushEvent( serverId: string, params: PushEventParams, responder?: Responder, - ): void { + ): Promise { // §16.3: expand the normative chat:* core closure once, at entry, so // every downstream consumer (wake matching, metadata, the queued event) // sees the closed set. Producer `implies` edges are NOT consumed — @@ -223,6 +224,18 @@ export class PushHandler { return; } + if (params.coalesce !== undefined && this.handleCoalesced) { + try { + const result = await this.handleCoalesced(serverId, params); + responder?.respond(result); + } catch (error) { + const err = error as Error & { code?: number; field?: string }; + if (responder?.respondError) responder.respondError(typeof err.code === 'number' ? err.code : -32603, err.message, { field: err.field }); + else responder?.respond({ accepted: false, reason: err.message }); + } + return; + } + // 2. Deduplicate by eventId if (this.dedup.checkAndAdd(params.eventId)) { console.error(`[push-event-rejected] server=${serverId} eventId=${params.eventId} reason=duplicate`); diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 6a999545a..dd17bf3d4 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -186,7 +186,7 @@ export class McplServerConnection extends EventEmitter { private pendingRequests = new Map(); /** Metadata retained after selected request deadlines so a late response is * observable even though its original promise has already been rejected. */ - private orphanedRequests = new Map(); + private orphanedRequests = new Map }>(); private closed = false; /** Per-request timeout in ms (0 disables). See McplServerConfig.requestTimeoutMs. */ @@ -629,6 +629,11 @@ export class McplServerConnection extends EventEmitter { return this.sendRequest(McplMethod.BeforeInference, params as unknown as Record) as Promise; } + /** RFC-006 deferred push rendering; its own timeout bounds context assembly. */ + sendPushRender(params: import('./push-coalescer.js').PushRenderParams): Promise { + return this.sendRequest('push/render', params as unknown as Record, { timeoutMs: 5000, surfaceOrphanedResponse: true }) as Promise; + } + /** Send `featureSets/update` as a Notification. §6.7: valid ONLY for * purely descriptive metadata that does not alter the grant. Any grant @@ -952,7 +957,8 @@ export class McplServerConnection extends EventEmitter { pending.timer = setTimeout(() => { this.pendingRequests.delete(id); if (options.surfaceOrphanedResponse) { - this.orphanedRequests.set(id, { method }); + this.orphanedRequests.set(id, { method, ...(method === 'push/render' ? { renderParams: { featureSet: params.featureSet, key: params.key, eventId: params.eventId } } : {}) }); + if (this.orphanedRequests.size > 4096) this.orphanedRequests.delete(this.orphanedRequests.keys().next().value!); } reject(new Error( `MCPL server "${this.id}" did not respond to ${method} (id=${id}) ` + @@ -1142,6 +1148,10 @@ export class McplServerConnection extends EventEmitter { // re-inject either result because the original dispatch context is gone. const orphaned = this.orphanedRequests.get(response.id); this.orphanedRequests.delete(response.id); + if (orphaned?.renderParams) { + this.emit('orphaned-render-response', { params: orphaned.renderParams, result: response.result, error: response.error }); + return; // audit only: a timeout fallback can never be replaced + } const result = response.result; if (result && typeof result === 'object') { const r = result as { state?: unknown; checkpoint?: unknown }; diff --git a/src/mcpl/types.ts b/src/mcpl/types.ts index f0aec7251..d7469fba5 100644 --- a/src/mcpl/types.ts +++ b/src/mcpl/types.ts @@ -161,6 +161,13 @@ export interface McplCapabilities { */ export interface McplHostCapabilities { version: string; + /** RFC-006 support, independent of capability grants. */ + eventCoalescing?: boolean | { + pushEvents?: boolean; + channelsIncoming?: boolean; + deferred?: boolean; + channelScopedPush?: boolean; + }; pushEvents?: boolean; contextHooks?: { beforeInference?: boolean | { @@ -196,6 +203,9 @@ export interface McplChannelCapabilities { * Provided in FrameworkConfig.mcplServers[]. */ export interface McplServerConfig { + /** Optional host allow-list for inbound channel IDs (`*` glob). Empty denies all. + * Applies to channels/incoming AND explicitly channel-scoped push/event. */ + allowedIncomingChannels?: string[]; /** Unique server identifier */ id: string; @@ -634,6 +644,13 @@ export interface StateRollbackResult { * Spec Section 9.1. */ export interface PushEventParams { + coalesce?: { + key: string; + channelId?: string; + deferred?: boolean; + retract?: boolean; + data?: unknown; + }; /** Declaring feature set */ featureSet: string; @@ -661,6 +678,10 @@ export interface PushEventParams { * Spec Section 9.3. */ export interface PushEventResult { + coalesce?: { + outcome: 'first' | 'replaced' | 'appended' | 'retracted' | 'noted' | 'consumed'; + priorEventId?: string; + }; /** Whether the event was accepted */ accepted: boolean; @@ -1097,6 +1118,8 @@ export interface ChannelsPublishResult { * Spec Section 14.3. */ export interface ChannelIncomingMessage { + eventId?: string; + coalesce?: Omit, 'channelId' | 'deferred'>; /** Channel this message came from */ channelId: string; @@ -1144,6 +1167,7 @@ export interface ChannelsIncomingResult { /** Result for a single incoming message. */ export interface ChannelIncomingMessageResult { + coalesce?: PushEventResult['coalesce']; messageId: string; accepted: boolean; conversationId?: string; diff --git a/src/types/agent.ts b/src/types/agent.ts index b5cd7471f..0b2c7c03c 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -296,6 +296,9 @@ export interface CompletedToolCall { * Inference request for an agent. */ export interface InferenceRequest { + /** Host-owned identity for withdrawal of an unconsumed coalesced wake. */ + coalescingSubject?: string; + coalescingEventId?: string; agentName: string; reason: string; source: string; diff --git a/test/event-gate-coalescing.test.ts b/test/event-gate-coalescing.test.ts new file mode 100644 index 000000000..6d39d5487 --- /dev/null +++ b/test/event-gate-coalescing.test.ts @@ -0,0 +1,59 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, writeFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { EventGate, COALESCING_SUBJECT } from '../src/gate/event-gate.js'; + +for (const cancel of [false,true]) test(`coalesced debounce preserves deadline and ${cancel?'cancels without residue':'wakes without a context marker'}`, t => { + t.mock.timers.enable({apis:['setTimeout']}); + const dir=mkdtempSync(join(tmpdir(),'coalesce-gate-')); + const configPath=join(dir,'gate.json'); + writeFileSync(configPath,JSON.stringify({policies:[{name:'doc',match:{source:'editor'},behavior:{debounce:1000}}],default:'skip'})); + let wakes=0; const messages:unknown[]=[]; + const gate=new EventGate({configPath,emitTrace:()=>{},addMessage:(_p,c)=>{messages.push(c);return '';},requestInference:()=>{wakes++;},getAgentNames:()=>['agent']}); + t.after(()=>{gate.dispose();rmSync(dir,{recursive:true,force:true});}); + const cb=gate.asShouldTriggerCallback(); + cb('first',{serverId:'editor',eventType:'mcpl:push-event',[COALESCING_SUBJECT]:'subject'}); + for(let i=0;i<4;i++) { + t.mock.timers.tick(200); + cb('latest',{serverId:'editor',eventType:'mcpl:push-event',[COALESCING_SUBJECT]:'subject'}); + } + if(cancel) gate.cancelCoalesced('subject'); + t.mock.timers.tick(200); + assert.equal(wakes,cancel?0:1); assert.deepEqual(messages,[]); +}); + +test('wire metadata cannot counterfeit the host-only coalescing marker',t=>{ + t.mock.timers.enable({apis:['setTimeout']}); + const dir=mkdtempSync(join(tmpdir(),'coalesce-gate-')); const configPath=join(dir,'gate.json'); + writeFileSync(configPath,JSON.stringify({policies:[{name:'doc',match:{source:'editor'},behavior:{debounce:1000}}],default:'skip'})); + let wakes=0; + const gate=new EventGate({configPath,emitTrace:()=>{},addMessage:()=>'',requestInference:()=>{wakes++;},getAgentNames:()=>['agent']}); + t.after(()=>{gate.dispose();rmSync(dir,{recursive:true,force:true});}); + const cb=gate.asShouldTriggerCallback(); + cb('first',{serverId:'editor',eventType:'mcpl:push-event',coalescingSubject:'forged'}); + t.mock.timers.tick(500); cb('second',{serverId:'editor',eventType:'mcpl:push-event',coalescingSubject:'forged'}); + t.mock.timers.tick(500); assert.equal(wakes,0); t.mock.timers.tick(500); assert.equal(wakes,1); +}); + +for (const changePolicy of [false,true]) test(`coalesced wake survives ${changePolicy?'policy changes':'unrelated traffic'} without postponement or duplication`,t=>{ + t.mock.timers.enable({apis:['setTimeout']}); + const dir=mkdtempSync(join(tmpdir(),'coalesce-gate-')); const configPath=join(dir,'gate.json'); + writeFileSync(configPath,JSON.stringify({policies:[ + {name:'special',match:{tagsAny:['doc:special']},behavior:{debounce:2000}}, + {name:'doc',match:{source:'editor'},behavior:{debounce:1000}}, + ],default:'skip'})); + let wakes=0; + const gate=new EventGate({configPath,emitTrace:()=>{},addMessage:()=>'',requestInference:()=>{wakes++;},getAgentNames:()=>['agent']}); + t.after(()=>{gate.dispose();rmSync(dir,{recursive:true,force:true});}); + const cb=gate.asShouldTriggerCallback(); + cb('first',{serverId:'editor',eventType:'mcpl:push-event',[COALESCING_SUBJECT]:'subject'}); + for(let i=0;i<4;i++) { + t.mock.timers.tick(200); + cb('update',{serverId:'editor',eventType:'mcpl:push-event',...(changePolicy?{tags:['doc:special'],[COALESCING_SUBJECT]:'subject'}:{})}); + } + t.mock.timers.tick(200);assert.equal(wakes,1); + t.mock.timers.tick(799);assert.equal(wakes,1); + t.mock.timers.tick(1);assert.equal(wakes,changePolicy?1:2); +}); diff --git a/test/helpers/coalescing-journal.ts b/test/helpers/coalescing-journal.ts new file mode 100644 index 000000000..217083256 --- /dev/null +++ b/test/helpers/coalescing-journal.ts @@ -0,0 +1,19 @@ +import type { CoalescingJournal, CoalescingRecord } from '../../src/mcpl/coalescing-journal.js'; +const copy = (value: T): T => JSON.parse(JSON.stringify(value)) as T; + +export class MemoryCoalescingJournal implements CoalescingJournal { + readonly records: CoalescingRecord[]; + beforeAppend?: (record: CoalescingRecord) => void; + afterAppend?: (record: CoalescingRecord) => void; + constructor(records: CoalescingRecord[] = []) { this.records = copy(records); } + length(): number { return this.records.length; } + read(index: number): CoalescingRecord { return copy(this.records[index]); } + append(record: CoalescingRecord): void { + this.beforeAppend?.(record); + this.records.push(copy(record)); + this.afterAppend?.(record); + } +} +export function journalRecords(journal: CoalescingJournal): CoalescingRecord[] { + return Array.from({length: journal.length()}, (_, index) => journal.read(index)); +} diff --git a/test/mcpl-coalescing-integration.test.ts b/test/mcpl-coalescing-integration.test.ts new file mode 100644 index 000000000..569a499df --- /dev/null +++ b/test/mcpl-coalescing-integration.test.ts @@ -0,0 +1,418 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { once } from 'node:events'; +import { WebSocketServer, type WebSocket } from 'ws'; +import type { AddressInfo } from 'node:net'; +import type { FrameworkConfig } from '../src/types/framework.js'; +import { CapabilityGrant } from '../src/mcpl/capability-grant.js'; +import type { McplServerConnection } from '../src/mcpl/server-connection.js'; +import { AgentFramework } from '../src/framework.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; + +async function fixture(options: { server?: Record; framework?: Partial } = {}) { + const wss=new WebSocketServer({port:0,host:'127.0.0.1'}); await once(wss,'listening'); + let online=true; + let socket:WebSocket; let next=1000; const receipts=new Mapvoid>(); + const renders:any[]=[]; const hostCaps:any[]=[]; const published:any[]=[]; + let renderer: (params:any)=>Promise = async()=>({content:[{type:'text',text:'document_diff'}]}); + wss.on('connection',ws=>{ + if(!online){ws.close();return;} + socket=ws; + ws.on('message',async bytes=>{ + const m=JSON.parse(String(bytes)); + if(!m.method) {receipts.get(m.id)?.(m);receipts.delete(m.id);return;} + const reply=(result:unknown)=>ws.send(JSON.stringify({jsonrpc:'2.0',id:m.id,result})); + if(m.method==='initialize') { + hostCaps.push(m.params.capabilities.experimental.mcpl); + reply({protocolVersion:'2024-11-05',capabilities:{tools:{},experimental:{mcpl:{version:'0.5',pushEvents:true,inferenceRequest:true,channels:{incoming:true,register:true,lifecycle:true,publish:true},featureSets:{doc:{description:'doc',uses:['pushEvents']}}}}},serverInfo:{name:'editor',version:'1'}}); + } else if(m.method==='featureSets/update') reply({accepted:true}); + else if(m.method==='tools/list') reply({tools:[]}); + else if(m.method==='channels/publish') {published.push(m.params); if(m.id!==undefined)reply({delivered:true});} + else if(m.method==='channels/close')reply({closed:true}); + else if(m.method==='channels/open')reply({channel:{id:m.params.channelId,type:'discord',label:m.params.channelId}}); + else if(m.method==='push/render') {renders.push(m.params);reply(await renderer(m.params));} + else if(m.id!==undefined) reply({}); + }); + }); + const dir=mkdtempSync(join(tmpdir(),'coalescing-host-')); + let framework:AgentFramework; const membrane=new MockMembrane(); membrane.pushResponse(createMockResponse([{type:'text',text:'ok'}])); + const create=async(grantPush=true)=>{ + framework=await AgentFramework.create({storePath:join(dir,'store'),membrane:membrane.asMembrane(),agents:[{name:'agent',model:'test',systemPrompt:'test'}],modules:[],...options.framework,mcplServers:[{id:'editor',url:`ws://127.0.0.1:${(wss.address() as AddressInfo).port}`,enabledFeatureSets:['doc'],...options.server,...(grantPush?{}:{disabledCapabilities:['pushEvents']})}]}); + return framework; + }; + const send=(method:string,params:unknown):Promise=>new Promise((resolve,reject)=>{ + const id=next++; const timer=setTimeout(()=>reject(Error(`no reply to ${method}`)),2000); + receipts.set(id,value=>{clearTimeout(timer);resolve(value);}); + socket.send(JSON.stringify({jsonrpc:'2.0',id,method,params})); + }); + await create(); + return {get framework(){return framework;},membrane,renders,hostCaps,published,create,send, + connection:()=> (framework as unknown as {mcplServerRegistry:{getServer(id:string):McplServerConnection}}).mcplServerRegistry.getServer('editor'), + disconnect:()=>socket.terminate(), + online:(value:boolean)=>{online=value;}, + register:(id='chat')=>send('channels/register',{channels:[{id,type:'discord',label:id,metadata:{channelType:'guild_text'}}]}), + channel:(id:string,text:string,flags:Record={},channelId='chat')=>({channelId,messageId:'m',eventId:id,timestamp:'2026-09-30T00:00:00Z',author:{id:'u',name:'User'},tags:['chat:mention'],content:text?[{type:'text',text}]:[],coalesce:{key:'message:m',...flags}}), + renderer:(fn:typeof renderer)=>{renderer=fn;}, + close:async()=>{await framework.stop(); for(const client of wss.clients)client.terminate();await new Promise(r=>wss.close(()=>r()));rmSync(dir,{recursive:true,force:true});}, + params:(id:string,text:string,flags:Record={})=>({featureSet:'doc',eventId:id,timestamp:'2026-09-30T00:00:00Z',coalesce:{key:'document',...flags},payload:{content:text?[{type:'text',text}]:[]}}), + context:()=>JSON.stringify(framework.getAgent('agent')!.getContextManager().getAllMessages()), + }; +} + +test('wire push receipts coalesce, content stays outside context until the real activation',async t=>{ + const f=await fixture();t.after(f.close); + assert.deepEqual(f.hostCaps[0].eventCoalescing,{pushEvents:true,deferred:true,channelsIncoming:true,channelScopedPush:true}); + const first=await f.send('push/event',f.params('1','edit_original')); + const last=await f.send('push/event',f.params('2','edit_latest')); + assert.equal(first.result.coalesce.outcome,'first'); assert.equal(last.result.coalesce.outcome,'replaced'); + assert(!f.context().includes('edit_original'));assert(!f.context().includes('edit_latest')); + await f.framework.runUntilIdle(); + assert.equal(f.membrane.calls.length,1); + const request=JSON.stringify(f.membrane.calls[0]); + assert(request.includes('edit_latest')); assert(!request.includes('edit_original')); + assert.deepEqual((await f.send('push/event',f.params('2','edit_latest'))).result,last.result); + await f.framework.runUntilIdle(); assert.equal(f.membrane.calls.length,1); +}); + +test('deferred wire pushes render once at assembly and never expose notice data',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',f.params('1','fallback_unavailable',{deferred:true,data:{private:'hidden_notice'}})); + await f.send('push/event',f.params('2','fallback_unavailable',{deferred:true,data:{private:'hidden_notice_2'}})); + assert.equal(f.renders.length,0);await f.framework.runUntilIdle(); + assert.equal(f.renders.length,1);assert.equal(f.renders[0].notices.length,2); + const request=JSON.stringify(f.membrane.calls[0]); + assert(request.includes('document_diff'));assert(!request.includes('fallback_unavailable'));assert(!request.includes('hidden_notice')); +}); + +test('retraction before assembly cancels the wake and leaves no model-visible trace',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',f.params('1','unread_original')); + const response=await f.send('push/event',f.params('2','deletion_notice',{retract:true})); + assert.equal(response.result.coalesce.outcome,'retracted');await f.framework.runUntilIdle(); + assert.equal(f.membrane.calls.length,0);assert(!f.context().includes('unread_original'));assert(!f.context().includes('deletion_notice')); +}); + +test('unregistered channel scope is refused on both delivery lanes',async t=>{ + const f=await fixture();t.after(f.close); + const bad=await f.send('push/event',f.params('1','text',{channelId:'chat'}));assert.equal(bad.error.code,-32023); + const channel=await f.send('channels/incoming',{messages:[{channelId:'chat',messageId:'m',eventId:'e',timestamp:'now',author:{id:'u',name:'User'},content:[],coalesce:{key:'message:m'}}]}); + assert.match(channel.result.results[0].reason,/unknown channel/); +}); + +test('clean restart recovers the latest durable fallback without replaying a render',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',f.params('1','old_fallback',{deferred:true})); + await f.send('push/event',f.params('2','recovered_fallback',{deferred:true})); + await f.framework.stop();await f.create(); + assert(!f.context().includes('recovered_fallback')); + await f.framework.runUntilIdle(); + assert(!f.context().includes('old_fallback'));assert(f.context().includes('recovered_fallback'));assert.equal(f.renders.length,0); + const count=f.framework.getAgent('agent')!.getContextManager().getAllMessages().length; + await f.framework.stop();await f.create(); + assert.equal(f.framework.getAgent('agent')!.getContextManager().getAllMessages().length,count); +}); + +test('inference/request during push/render is rejected without starting recursive inference',async t=>{ + const f=await fixture();t.after(f.close); + f.renderer(async()=>{ + const response=await f.send('inference/request',{featureSet:'doc',messages:[]}); + assert.equal(response.error.code,-32600);return {content:[{type:'text',text:'rendered_after_refusal'}]}; + }); + await f.send('push/event',f.params('1','fallback',{deferred:true}));await f.framework.runUntilIdle(); + assert.equal(f.membrane.calls.length,1);assert(JSON.stringify(f.membrane.calls[0]).includes('rendered_after_refusal')); +}); + +test('plain replacement during wire render supersedes both frozen and newer pending work',async t=>{ + const f=await fixture();t.after(f.close); + let release!: (value:unknown)=>void; let began!: ()=>void; + const started=new Promise(resolve=>{began=resolve;}); + f.renderer(()=>{began();return new Promise(resolve=>{release=resolve;});}); + await f.send('push/event',f.params('1','fallback_one',{deferred:true})); + const run=f.framework.runUntilIdle(); await started; + await f.send('push/event',f.params('2','fallback_two',{deferred:true})); + const receipt=await f.send('push/event',f.params('3','complete_snapshot')); + assert.equal(receipt.result.coalesce.outcome,'replaced');release({content:[{type:'text',text:'late_old_render'}]}); + await run; + assert.equal(f.membrane.calls.length,1);assert.equal(f.renders.length,1); + const request=JSON.stringify(f.membrane.calls[0]); + assert(request.includes('complete_snapshot'));assert(!request.includes('fallback_one'));assert(!request.includes('fallback_two'));assert(!request.includes('late_old_render')); +}); + +test('reference disposition remains binding on materialized render content',async t=>{ + const f=await fixture();t.after(f.close); + f.renderer(async()=>({content:[{type:'resource',uri:'https://example.test/secret-capability',name:'render.wav',mimeType:'audio/wav',disposition:'never'}]})); + await f.send('push/event',f.params('1','fallback',{deferred:true}));await f.framework.runUntilIdle(); + const request=JSON.stringify(f.membrane.calls[0]); + assert(request.includes('render.wav'));assert(!request.includes('secret-capability')); +}); + +test('operator branch switches do not move pending payloads or overwrite another branch recovery state',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',f.params('1','main_pending')); + const cm=f.framework.getAgent('agent')!.getContextManager(); + await cm.fork('alternate'); + await f.send('push/event',f.params('2','alternate_pending')); + await f.framework.runUntilIdle(); + assert(!JSON.stringify(f.membrane.calls[0]).includes('main_pending')); + assert(JSON.stringify(f.membrane.calls[0]).includes('alternate_pending')); + await cm.switchBranch('main'); + await f.send('push/event',f.params('3','main_new')); + f.membrane.pushResponse(createMockResponse([{type:'text',text:'ok'}])); + await f.framework.runUntilIdle(); + assert(!f.context().includes('main_pending'));assert(f.context().includes('main_new')); + assert(!f.context().includes('alternate_pending')); +}); + +test('restart cannot publish a saved fallback before the newly denied grant',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',f.params('1','previously_permitted',{deferred:true})); + await f.framework.stop();await f.create(false); + await f.framework.runUntilIdle(); + assert(!f.context().includes('previously_permitted'));assert.equal(f.membrane.calls.length,0);assert.equal(f.renders.length,0); +}); + +test('a render arriving after the RPC deadline is audit-only and cannot replace fallback',async t=>{ + const f=await fixture();t.after(f.close); + f.renderer(async()=>{await new Promise(resolve=>setTimeout(resolve,5200));return {content:[{type:'text',text:'wire_late_payload'}]};}); + await f.send('push/event',f.params('1','bounded_fallback',{deferred:true}));await f.framework.runUntilIdle(); + const request=JSON.stringify(f.membrane.calls[0]); + assert(request.includes('bounded_fallback'));assert(!request.includes('wire_late_payload')); + await new Promise(resolve=>setTimeout(resolve,300)); + const store=(f.framework as unknown as {store:{getStateJson(id:string):unknown}}).store; + const audit=JSON.stringify(store.getStateJson('mcpl/coalescing-journal')); + assert(audit.includes('late-render'));assert(audit.includes('wire_late_payload')); + assert(!f.context().includes('wire_late_payload')); +}); + +async function eventually(predicate:()=>boolean) { + const deadline=Date.now()+3000; + while(!predicate()) {assert(Date.now()setTimeout(resolve,10));} +} + +test('channel create, push edit, push delete share one unread subject',async t=>{ + const f=await fixture();t.after(f.close);await f.register(); + const first=await f.send('channels/incoming',{messages:[f.channel('c','chat_original')]}); + const edit=await f.send('push/event',f.params('e','chat_edit',{channelId:'chat',key:'message:m'})); + const deletion=await f.send('push/event',f.params('d','chat_deleted',{channelId:'chat',key:'message:m',retract:true})); + assert.equal(first.result.results[0].coalesce.outcome,'first'); + assert.equal(edit.result.coalesce.outcome,'replaced');assert.equal(deletion.result.coalesce.outcome,'retracted'); + await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,0); + assert(!f.context().includes('chat_original'));assert(!f.context().includes('chat_edit'));assert(!f.context().includes('chat_deleted')); +}); + +test('read original, unread edit, deletion preserves original and appends only the notice',async t=>{ + const f=await fixture();t.after(f.close);await f.register(); + await f.send('channels/incoming',{messages:[f.channel('c','read_original')]});await f.framework.runUntilIdle(); + await f.send('push/event',f.params('e','unread_edit',{channelId:'chat',key:'message:m'})); + const deletion=await f.send('push/event',f.params('d','deletion_notice',{channelId:'chat',key:'message:m',retract:true})); + assert.equal(deletion.result.coalesce.outcome,'noted'); + f.membrane.pushResponse(createMockResponse([{type:'text',text:'ok'}]));await f.framework.runUntilIdle(); + const request=JSON.stringify(f.membrane.calls[1]); + assert(request.includes('read_original'));assert(request.includes('deletion_notice'));assert(!request.includes('unread_edit')); + assert.equal(f.published.at(-1)?.channelId,'chat'); +}); + +test('repeated channel edits keep message identity and deduplicate occurrence retries',async t=>{ + const f=await fixture();t.after(f.close);await f.register(); + await f.send('channels/incoming',{messages:[f.channel('c','original')]}); + const edit=await f.send('channels/incoming',{messages:[f.channel('e','newest')]}); + const retry=await f.send('channels/incoming',{messages:[f.channel('e','newest')]}); + assert.deepEqual(retry.result,edit.result);assert.equal(edit.result.results[0].coalesce.outcome,'replaced'); + await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,1); + assert(JSON.stringify(f.membrane.calls[0]).includes('newest'));assert(!JSON.stringify(f.membrane.calls[0]).includes('original')); +}); + +test('both delivery lanes enforce the same inbound channel allow-list',async t=>{ + const f=await fixture({server:{allowedIncomingChannels:['chat:allowed*']}});t.after(f.close); + await f.register('chat:allowed');await f.register('chat:blocked'); + const incoming=await f.send('channels/incoming',{messages:[f.channel('blocked','blocked',{},'chat:blocked'),f.channel('allowed','allowed',{},'chat:allowed')]}); + assert.equal(incoming.result.results[0].accepted,false);assert.equal(incoming.result.results[1].accepted,true); + const push=await f.send('push/event',f.params('blocked-push','blocked',{channelId:'chat:blocked',key:'message:m'})); + assert.equal(push.error.code,-32017);await f.framework.runUntilIdle(); + assert(!JSON.stringify(f.membrane.calls[0]).includes('blocked')); +}); + +test('revoked channel authority rejects a mixed-lane edit without touching the pending create',async t=>{ + const f=await fixture();t.after(f.close);await f.register(); + await f.send('channels/incoming',{messages:[f.channel('c','untouched_create')]}); + const oldGrant=f.connection().grant;f.connection().establishGrant(new CapabilityGrant(new Set(['pushEvents']),[])); + const edit=await f.send('push/event',f.params('e','forbidden_edit',{channelId:'chat',key:'message:m'})); + assert.equal(edit.error.code,-32017);f.connection().establishGrant(oldGrant); + await f.framework.runUntilIdle();assert(JSON.stringify(f.membrane.calls[0]).includes('untouched_create'));assert(!JSON.stringify(f.membrane.calls[0]).includes('forbidden_edit')); +}); + +test('malformed coalescing fails per channel message and preserves valid siblings',async t=>{ + const f=await fixture();t.after(f.close);await f.register(); + const missing={...f.channel('bad','bad'),eventId:undefined}; + const deferred=f.channel('bad2','bad2',{deferred:true}); + const response=await f.send('channels/incoming',{messages:[missing,f.channel('good','good'),deferred]}); + assert.deepEqual(response.result.results.map((r:any)=>r.accepted),[false,true,false]); + assert.equal(response.result.results[0].reason,'coalesce_invalid'); + const retract=await f.send('push/event',f.params('empty','',{channelId:'chat',key:'message:m',retract:true})); + assert.equal(retract.error.code,-32602);await f.framework.runUntilIdle();assert(JSON.stringify(f.membrane.calls[0]).includes('good')); +}); + +test('channel scope is explicit and isolated from origin claims and other channels',async t=>{ + const f=await fixture();t.after(f.close);await f.register('chat:one');await f.register('chat:two'); + await f.send('channels/incoming',{messages:[f.channel('a','channel_one',{},'chat:one'),f.channel('b','channel_two',{},'chat:two')]}); + const misleading={...f.params('feature','feature_scope',{key:'message:m'}),origin:{channelId:'chat:one'}}; + await f.send('push/event',misleading);await f.framework.runUntilIdle(); + const request=JSON.stringify(f.membrane.calls[0]);for(const text of ['channel_one','channel_two','feature_scope'])assert(request.includes(text)); +}); + +test('a transient reconnect retains pending channel content and receipt identity',async t=>{ + const f=await fixture({server:{reconnect:true,reconnectIntervalMs:20}});t.after(f.close);await f.register(); + const params={messages:[f.channel('c','survives_disconnect')]}; + const accepted=await f.send('channels/incoming',params);f.disconnect(); + await eventually(()=>f.hostCaps.length===2 && f.connection().policyEstablished); + await f.register();const retry=await f.send('channels/incoming',params); + assert.deepEqual(retry.result,accepted.result);await f.framework.runUntilIdle(); + assert.equal(f.membrane.calls.length,1);assert(JSON.stringify(f.membrane.calls[0]).includes('survives_disconnect')); +}); + +test('recovered unread content can be retracted before assembly',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',f.params('c','withdraw_after_restart')); + await f.framework.stop();await f.create(); + const deletion=await f.send('push/event',f.params('d','deleted',{retract:true}));assert.equal(deletion.result.coalesce.outcome,'retracted'); + await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,0);assert(!f.context().includes('withdraw_after_restart')); +}); + +test('durable occurrence receipts prevent duplicates after restart and consumption',async t=>{ + const f=await fixture();t.after(f.close);const params=f.params('once','exactly_once'); + const original=await f.send('push/event',params);await f.framework.runUntilIdle(); + await f.framework.stop();await f.create();const retry=await f.send('push/event',params); + assert.deepEqual(retry.result,original.result);await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,1); +}); + +test('conversation-routed channel coalescing stays in its fork and pins the reply channel',async t=>{ + const f=await fixture({framework:{conversations:{templateAgent:'agent',bind:{channel:'always'},trigger:{channel:'always'}}}});t.after(f.close);await f.register(); + await f.send('channels/incoming',{messages:[f.channel('c','fork_original')]}); + await f.send('push/event',f.params('e','fork_latest',{channelId:'chat',key:'message:m'})); + await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,1); + const request=JSON.stringify(f.membrane.calls[0]);assert(request.includes('fork_latest'));assert(!request.includes('fork_original')); + assert(!f.context().includes('fork_latest'));assert.equal(f.published.at(-1)?.channelId,'chat'); +}); + +for (const debounce of [false,true]) test(`replacement with skip cancels ${debounce?'an already-fired debounce':'an immediate'} wake`,async t=>{ + const f=await fixture({framework:{gate:{config:{policies:[{name:'attention',match:{tagsAny:['doc:wake']},behavior:debounce?{debounce:100}:'always'}],default:'skip'}}}});t.after(f.close); + await f.send('push/event',{...f.params('wake','obsolete_wake'),tags:['doc:wake']}); + if(debounce)await eventually(()=>(f.framework as unknown as {pendingRequests:unknown[]}).pendingRequests.length>0); + await f.send('push/event',{...f.params('skip','quiet_update'),tags:['doc:quiet']}); + await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,0); +}); + +test('retract after a debounce fires cancels its queued inference request',async t=>{ + const f=await fixture({framework:{gate:{config:{policies:[{name:'attention',match:{source:'editor'},behavior:{debounce:100}}],default:'skip'}}}});t.after(f.close); + await f.send('push/event',f.params('wake','withdrawn')); + await eventually(()=>(f.framework as unknown as {pendingRequests:unknown[]}).pendingRequests.length>0); + await f.send('push/event',f.params('retract','gone',{retract:true}));await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,0); +}); + +test('an idle resident cannot materialize coalesced content into a different resident live turn',async t=>{ + const f=await fixture({framework:{agents:[{name:'agent',model:'test',systemPrompt:'test'},{name:'reader',model:'test',systemPrompt:'test'}]}});t.after(f.close); + const internals=f.framework as unknown as {activeTurnTokens:Map;processNextEvent():Promise}; + internals.activeTurnTokens.set('agent',100); + await f.send('push/event',f.params('one','wait_for_boundary')); + await internals.processNextEvent();assert.equal(f.membrane.calls.length,0);assert(!f.context().includes('wait_for_boundary')); + internals.activeTurnTokens.delete('agent'); + // The provider mock needs a complete response for each independent resident. + const { MockYieldingStream }=await import('./helpers/mock-membrane.js'); + f.membrane.streamYielding=request=>{f.membrane.calls.push(request);return new MockYieldingStream([createMockResponse([{type:'text',text:'ok'}])]);}; + await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,2); + assert(f.membrane.calls.every(request=>JSON.stringify(request).includes('wait_for_boundary'))); +}); + +test('opening or closing a channel does not change a subject coalescing identity',async t=>{ + const f=await fixture();t.after(f.close);await f.register(); + await f.send('channels/incoming',{messages:[f.channel('create','before_close')]}); + const closed=await f.framework.channels!.handleChannelToolCall('channel_close',{channelId:'chat',serverId:'editor'}); + assert.equal(closed.success,true);assert.equal(f.framework.channels!.isChannelOpen('chat'),false); + const edit=await f.send('push/event',f.params('edit','after_close',{channelId:'chat',key:'message:m'})); + assert.equal(edit.result.coalesce.outcome,'replaced'); + await f.framework.runUntilIdle();assert(JSON.stringify(f.membrane.calls[0]).includes('after_close'));assert(!JSON.stringify(f.membrane.calls[0]).includes('before_close')); +}); + +test('host-derived outbound routing entries cannot authorize channel coalescing',async t=>{ + const f=await fixture();t.after(f.close); + f.framework.channels!.ensureChannelRegistered('editor','forged'); + const response=await f.send('push/event',f.params('forged','forged',{channelId:'forged'})); + assert.equal(response.error.code,-32023); +}); + +test('a host restarting while its producer is offline wakes recovered work when reconnection succeeds',async t=>{ + const f=await fixture({server:{reconnect:true,reconnectIntervalMs:20}});t.after(f.close); + await f.send('push/event',f.params('pending','offline_recovery')); + await f.framework.stop();f.online(false);await f.create();assert.equal(f.membrane.calls.length,0); + f.online(true);await eventually(()=>f.hostCaps.length>=2 && f.connection().policyEstablished); + await eventually(()=>(f.framework as unknown as {pendingRequests:unknown[]}).pendingRequests.length>0); + await f.framework.runUntilIdle();assert(JSON.stringify(f.membrane.calls[0]).includes('offline_recovery')); +}); + +test('a feature-scoped coalesced DM retains ordinary channel provenance and reply routing',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',{...f.params('dm','direct_message'),tags:['chat:dm'],origin:{source:'discord',channelId:'42',guildId:null,authorId:'u',authorName:'User'}}); + await f.framework.runUntilIdle();assert.equal(f.published.at(-1)?.channelId,'discord:dm:42'); +}); + +test('mixed-lane edits retain the stable platform reply target after consumption',async t=>{ + const f=await fixture();t.after(f.close);await f.register(); + await f.send('channels/incoming',{messages:[{...f.channel('c','first_version'),threadId:'thread'}]});await f.framework.runUntilIdle(); + await f.send('push/event',f.params('e','second_version',{channelId:'chat',key:'message:m'})); + f.membrane.pushResponse(createMockResponse([{type:'text',text:'ok'}]));await f.framework.runUntilIdle(); + const message=f.framework.getAgent('agent')!.getContextManager().getAllMessages().find(m=>m.metadata?.eventId==='e'); + assert.equal(message?.metadata?.messageId,'m');assert.equal(message?.metadata?.threadId,'thread'); + assert.deepEqual(message?.metadata?.author,{id:'u',name:'User'}); +}); + +test('coalescing preserves the ordinary conversation bind predicate',async t=>{ + const f=await fixture({framework:{conversations:{templateAgent:'agent'}}});t.after(f.close);await f.register(); + const unbound=f.channel('unbound','not_a_personal_mention');unbound.tags=['chat:reply']; + await f.send('channels/incoming',{messages:[unbound]});await f.framework.runUntilIdle();assert.equal(f.membrane.calls.length,0); + const mentioned={...f.channel('mentioned','personal_mention'),metadata:{mentioned:true}}; + await f.send('channels/incoming',{messages:[mentioned]});await f.framework.runUntilIdle(); + assert.equal(f.membrane.calls.length,1);assert(!JSON.stringify(f.membrane.calls[0]).includes('not_a_personal_mention')); +}); + +test('an isolated agent name cannot alias the shared coalescing audience',async t=>{ + const f=await fixture();t.after(f.close); + const host=f.framework as unknown as {coalescingAudience(agentName?:string):string}; + assert.notEqual(host.coalescingAudience(),host.coalescingAudience('shared-messages')); +}); + +test('journal alone remains authoritative after restart despite contradictory prototype state',async t=>{ + const f=await fixture();t.after(f.close); + const receipt=await f.send('push/event',f.params('journal-original','journal_original')); + const store=(f.framework as any).store as import('@animalabs/chronicle').JsStore; + assert.deepEqual(store.listStates().filter(s=>s.id.startsWith('mcpl/coalescing')).map(s=>s.id),['mcpl/coalescing-journal']); + store.registerState({id:'mcpl/coalescing-pending',strategy:'snapshot'}); + store.setStateJson('mcpl/coalescing-pending',{version:999,pending:[{invalid:true}]}); + await f.framework.stop();await f.create(); + assert.deepEqual((await f.send('push/event',f.params('journal-original','journal_original'))).result,receipt.result); + const edit=await f.send('push/event',f.params('journal-edit','journal_latest')); + assert.equal(edit.result.coalesce.outcome,'replaced');await f.framework.runUntilIdle(); + assert(f.context().includes('journal_latest'));assert(!f.context().includes('journal_original')); +}); + +test('context append without completion receipt retries the same publication once after restart',async t=>{ + const f=await fixture();t.after(f.close); + await f.send('push/event',f.params('crash-event','crash_window_content')); + const host=f.framework as any; + const coalescer=host.pushCoalescer as import('../src/mcpl/push-coalescer.js').PushCoalescer; + const audience=coalescer.pendingPushes()[0].audience; + await assert.rejects(coalescer.assemble(audience,p=>{ + assert.equal(host.materializeCoalescedPush(p,f.framework.getAgent('agent')!),true); + // Persist exactly the crash window: intent and context, no completion record. + host.store.sync();throw Error('crash after context append'); + }),/crash after context append/); + const messages=()=>f.framework.getAgent('agent')!.getContextManager().getAllMessages().filter(m=>m.metadata?.eventId==='crash-event'); + assert.equal(messages().length,1);const deliveryId=messages()[0].metadata!.coalescingDeliveryId; + await f.framework.stop();await f.create();await f.framework.runUntilIdle(); + assert.equal(messages().length,1);assert.equal(messages()[0].metadata!.coalescingDeliveryId,deliveryId); + const restored=(f.framework as any).pushCoalescer as import('../src/mcpl/push-coalescer.js').PushCoalescer; + assert.equal(restored.inspect().publications.length,0); + assert.equal(f.membrane.calls.length,1); +}); diff --git a/test/mcpl-coalescing-journal.test.ts b/test/mcpl-coalescing-journal.test.ts new file mode 100644 index 000000000..6e8493ae7 --- /dev/null +++ b/test/mcpl-coalescing-journal.test.ts @@ -0,0 +1,133 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { JsStore } from '@animalabs/chronicle'; +import { PushCoalescer, coalescingSubject, coalescingReceiptKey, type CoalescedPush, type CoalescerOptions } from '../src/mcpl/push-coalescer.js'; +import { ChronicleCoalescingJournal, COALESCING_JOURNAL_ID } from '../src/mcpl/chronicle-coalescing-journal.js'; +import { MemoryCoalescingJournal, journalRecords } from './helpers/coalescing-journal.js'; + +function push(id: string, flags = {}): CoalescedPush { + return {serverId:'s',binding:'b',epoch:'e',audience:'a',params:{featureSet:'f',eventId:id, + timestamp:'now',coalesce:{key:'k',...flags},payload:{content:[{type:'text',text:id}]}}}; +} +function controller(journal: CoalescerOptions['journal'], overrides: Partial = {}) { + return new PushCoalescer({journal,authorized:()=>true,wake:()=>true,cancelWake:()=>{}, + render:async()=>({content:[{type:'text',text:'rendered'}]}),...overrides}); +} + +test('every live journal prefix replays identically without policy, RPC, publication, or wake effects', async()=>{ + const journal=new MemoryCoalescingJournal();const c=controller(journal); + const prefixes:Array<{records:ReturnType;state:ReturnType}>=[]; + const capture=()=>prefixes.push({records:journalRecords(journal),state:c.inspect()}); + journal.beforeAppend=capture; // Every committed prefix, including render/outbox boundaries. + c.accept(push('1'));capture();c.accept(push('2'));capture(); + c.accept(push('3',{retract:true}));capture(); + c.accept(push('4',{deferred:true}));capture();await c.assemble('a',()=>{});capture(); + c.accept(push('5'));capture();c.disconnect('s');capture();c.recover();capture(); + for(const {records,state} of prefixes){ + const fail=()=>{throw Error('replay performed an effect');}; + const replayed=controller(new MemoryCoalescingJournal(records),{authorized:fail,available:fail,render:fail,wake:fail,cancelWake:fail}); + assert.deepEqual(replayed.inspect(),state); + } + +}); + +test('replacement and retraction append facts without modifying their predecessors',()=>{ + const journal=new MemoryCoalescingJournal();const c=controller(journal); + c.accept(push('1'));const original=JSON.stringify(journal.records); + c.accept(push('2'));assert.equal(c.pending(coalescingSubject(push('1')))!.params.eventId,'2'); + c.accept(push('3',{retract:true}));assert.deepEqual(c.pendingPushes(),[]); + assert.equal(JSON.stringify(journal.records.slice(0,1)),original); + assert.deepEqual(journal.records.map(r=>r.operation.kind),['accepted','accepted','accepted']); + assert.deepEqual(controller(journal).pendingPushes(),[]); +}); + +for(const after of [false,true]) test(`acceptance crash ${after?'after':'before'} durable append is resolved only by replay`,()=>{ + const journal=new MemoryCoalescingJournal();let wakes=0; + const c=controller(journal,{wake:()=>{wakes++;return true;}}); + journal[after?'afterAppend':'beforeAppend']=()=>{throw Error('disk failure');}; + assert.throws(()=>c.accept(push('1')),/disk failure/);assert.equal(wakes,0); + assert.throws(()=>c.accept(push('2')),/rebuild/); + const restored=controller(journal); + assert.equal(!!restored.receipt(push('1')),after);assert.equal(restored.pendingPushes().length,after?1:0); +}); + +test('prepared append survives a crash before context publication',async()=>{ + const journal=new MemoryCoalescingJournal();const c=controller(journal);c.accept(push('1')); + journal.afterAppend=r=>{if(r.operation.kind==='prepared')throw Error('crash');}; + let writes=0;await assert.rejects(c.assemble('a',()=>{writes++;}),/crash/);assert.equal(writes,0); + journal.afterAppend=undefined;const restored=controller(journal);restored.recover(); + assert.equal(restored.accept(push('2')).coalesce!.outcome,'appended'); + const ids:string[]=[];await restored.assemble('a',p=>{ids.push(p.params.eventId);}); + assert.deepEqual(ids,['1','2']);assert.deepEqual(restored.pendingPushes(),[]); +}); + +for(const after of [false,true]) test(`completion crash ${after?'after':'before'} append preserves one context publication`,async()=>{ + const journal=new MemoryCoalescingJournal();const c=controller(journal);c.accept(push('1')); + const context=new Map();let calls=0; + const publish=(p:CoalescedPush)=>{calls++;assert(p.publicationId);context.set(p.publicationId,p.params.eventId);}; + journal[after?'afterAppend':'beforeAppend']=r=>{if(r.operation.kind==='delivered')throw Error('crash');}; + await assert.rejects(c.assemble('a',publish),/crash/);assert.equal(context.size,1); + journal.beforeAppend=undefined;journal.afterAppend=undefined; + const restored=controller(journal);restored.recover();await restored.assemble('a',publish); + assert.equal(context.size,1);assert.equal(calls,after?1:2);assert.deepEqual(restored.pendingPushes(),[]); +}); + +test('unfinished render recovers admitted fallback without repeating the request',async()=>{ + const journal=new MemoryCoalescingJournal();let calls=0;let began!:()=>void; + const started=new Promise(r=>{began=r;}); + const c=controller(journal,{render:()=>{calls++;began();return new Promise(()=>{});}}); + c.accept(push('fallback',{deferred:true}));const assembly=c.assemble('a',()=>assert.fail('published before crash')); + await started;c.suspend();await assembly; + const restored=controller(journal,{render:async()=>{calls++;throw Error('render repeated');}}); + assert.equal(restored.inspect().subjects[0][1].rendering!==undefined,true); + restored.recover();const texts:string[]=[];await restored.assemble('a',p=>{texts.push(p.params.eventId);}); + assert.equal(calls,1);assert.deepEqual(texts,['fallback']); +}); + +test('current policy changes publication decisions only when new facts are appended',async()=>{ + const journal=new MemoryCoalescingJournal();const c=controller(journal);c.accept(push('1')); + await assert.rejects(c.assemble('a',()=>{throw Error('publication failed');}),/publication failed/); + const state=c.inspect();const restored=controller(journal,{authorized:()=>false}); + assert.deepEqual(restored.inspect(),state);await restored.assemble('a',()=>assert.fail('revoked content published')); + assert.deepEqual(restored.pendingPushes(),[]); + assert.equal(journal.records.at(-1)!.operation.kind,'delivered'); + assert.deepEqual(controller(journal).inspect(),restored.inspect()); +}); + +test('callers cannot mutate projected state through callbacks or diagnostic reads',()=>{ + const journal=new MemoryCoalescingJournal();const c=controller(journal,{authorized:p=>{p.params.eventId='tampered';return true;}}); + const input=push('1');c.accept(input);const state=c.inspect(); + input.params.eventId='changed';c.pendingPushes()[0].params.eventId='changed';c.inspect().subjects[0][1].slot!.push.params.eventId='changed'; + c.canAssemble('a');assert.deepEqual(c.inspect(),state);assert.deepEqual(controller(journal).inspect(),state); +}); + +test('Chronicle compaction and reopen rebuild receipts and pending work from the journal alone',async()=>{ + const dir=mkdtempSync(join(tmpdir(),'coalescing-journal-'));const path=join(dir,'store');let store=JsStore.openOrCreate({path}); + try { + const journal=new ChronicleCoalescingJournal(store);const c=controller(journal);const receipt=c.accept(push('1'));c.accept(push('2')); + const records=journalRecords(journal);store.compactAllStates();store.close();store=JsStore.openOrCreate({path}); + const reopened=new ChronicleCoalescingJournal(store);assert.deepEqual(journalRecords(reopened),records); + const restored=controller(reopened);assert.deepEqual(restored.receipt(push('1')),receipt); + assert.equal(restored.pendingPushes()[0].params.eventId,'2'); + assert.deepEqual(store.listStates().map(s=>s.id),[COALESCING_JOURNAL_ID]); + } finally {store.close();rmSync(dir,{recursive:true,force:true});} +}); + +test('prototype migration imports uncertain publication and receipts once; old stores lose authority',()=>{ + const dir=mkdtempSync(join(tmpdir(),'coalescing-migration-'));const store=JsStore.openOrCreate({path:join(dir,'store')}); + try { + const p=push('1');const result={accepted:true as const,coalesce:{outcome:'first' as const}}; + store.registerState({id:'mcpl/coalescing-pending',strategy:'snapshot'}); + store.setStateJson('mcpl/coalescing-pending',{version:2,pending:[p],history:[],receipt:[JSON.stringify(['s','b','1']),result]}); + const journal=new ChronicleCoalescingJournal(store);const c=controller(journal); + assert.equal(journal.length(),1);assert.deepEqual(c.receipt(p),result); + assert.equal(c.inspect().publications.length,1);assert.equal(c.inspect().subjects[0][1].history,'unknown'); + const key=coalescingReceiptKey(p);assert(c.inspect().receipts.some(([k])=>k===key)); + store.setStateJson('mcpl/coalescing-pending',{version:999,pending:[{invalid:true}]}); + assert.deepEqual(controller(new ChronicleCoalescingJournal(store)).inspect(),c.inspect()); + assert.equal(journal.length(),1); + } finally {store.close();rmSync(dir,{recursive:true,force:true});} +}); diff --git a/test/mcpl-push-coalescer.test.ts b/test/mcpl-push-coalescer.test.ts new file mode 100644 index 000000000..20951f0c4 --- /dev/null +++ b/test/mcpl-push-coalescer.test.ts @@ -0,0 +1,251 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { PushCoalescer, CoalesceError, PUSH_COALESCING_SUPPORT, type CoalescedPush, type CoalescerOptions, type PushRenderResult } from '../src/mcpl/push-coalescer.js'; + +import { MemoryCoalescingJournal, journalRecords } from './helpers/coalescing-journal.js'; + +function push(id: string, text = id, flags: Record = {}, override: Partial = {}): CoalescedPush { + return { serverId: 'server', epoch: 'epoch', audience: 'agent', params: { + featureSet: 'doc', eventId: id, timestamp: '2026-09-30T00:00:00Z', + coalesce: { key: 'K', ...flags }, payload: { content: text ? [{ type: 'text', text }] : [] }, + }, ...override }; +} +function pending() { + let resolve!: (v: T) => void; + let reject!: (v: unknown) => void; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + return { promise, resolve, reject }; +} +function setup(overrides: Partial = {}) { + const deliveries: CoalescedPush[] = [], wakes: string[] = [], calls: unknown[] = []; + const journal = overrides.journal ?? new MemoryCoalescingJournal(); + const c = new PushCoalescer({ + journal, authorized: () => true, + render: async (_push, params) => { calls.push(params); return { content: [{ type: 'text', text: 'rendered' }] }; }, + wake: (_push, key) => { wakes.push(key); return true; }, cancelWake: () => {}, + ...overrides, + }); + return { c, journal, deliveries, wakes, calls, + get operations() { return journalRecords(journal).map(record => record.operation); }, + assemble: () => c.assemble('agent', p => { deliveries.push(structuredClone(p)); }), + texts: () => deliveries.map(p => p.params.payload.content.map(b => b.type === 'text' ? b.text : '').join('')) }; +} + +const result = (s: ReturnType, p: CoalescedPush) => s.c.accept(p).coalesce!.outcome; +const rendered = (text: string): PushRenderResult => ({ content: [{ type: 'text', text }] }); + +test('vectors 1–6: replace until assembly, then append forever (including compression/failure)', async () => { + const s = setup(); + assert.equal(result(s,push('create')), 'first'); + for (let i=1;i<=3;i++) assert.equal(result(s,push(`edit${i}`)), 'replaced'); + assert.deepEqual(s.texts(), []); + assert.equal(s.wakes.length, 4); // policy is re-evaluated for every replacement + await s.assemble(); + assert.deepEqual(s.texts(), ['edit3']); + // Publication at assembly is the only boundary; no provider response is needed. + assert.equal(result(s,push('later')), 'appended'); + await s.assemble(); + assert.deepEqual(s.texts(), ['edit3','later']); +}); + +test('vectors 7/31/32: uncertain history never suppresses a deletion', async () => { + const s = setup(); s.c.forgetHistory(); + assert.equal(result(s,push('edit')), 'appended'); + assert.equal(result(s,push('delete','deleted',{retract:true})), 'noted'); + await s.assemble(); + assert.deepEqual(s.texts(),['deleted']); + const fresh=setup(); + assert.equal(result(fresh,push('delete','deleted',{retract:true})), 'noted'); +}); + +test('vectors 9/35: namespace isolation and retries return the original receipt', async () => { + const s=setup(); + const one=push('same'); + const receipt=s.c.accept(one); + s.c.accept(push('same','other',{},{serverId:'other'})); + assert.deepEqual(s.c.accept(one),receipt); + assert.equal(s.operations.filter(r=>r.kind==='accepted').length,2); + await s.assemble(); + assert.deepEqual(s.texts(),['same','other']); +}); + +test('feature-set and key isolate subjects; transport epochs do not', async () => { + const s=setup(); + const cases=[push('1'),push('2','2',{key:'J'}),push('3','3',{},{epoch:'new'})]; + const fourth=push('4'); fourth.params.featureSet='other'; cases.push(fourth); + assert.deepEqual(cases.map(p=>result(s,p)),['first','first','replaced','first']); + await s.assemble(); assert.deepEqual(s.texts(),['3','2','4']); +}); + +test('vector 12: replacing a subject cannot widen its audience', async () => { + const s=setup(); s.c.accept(push('1')); s.c.accept(push('2','2',{},{audience:'other'})); + await s.assemble(); await s.c.assemble('other',p=>{s.deliveries.push(p);}); + assert.deepEqual(s.texts(),[]); +}); + +test('vectors 16–20: notices render once, carry private data and never expose fallback', async () => { + const s=setup(); + for(let i=0;i<100;i++) s.c.accept(push(String(i),'fallback',{deferred:true,data:{i}})); + assert.equal(s.calls.length,0); await s.assemble(); + const params=s.calls[0] as {notices:Array<{data:{i:number}}>;dropped:number}; + assert.equal(params.notices.length,64); assert.equal(params.dropped,36); + assert.equal(params.notices[0].data.i,36); assert.equal(params.notices.at(-1)!.data.i,99); + assert.deepEqual(s.texts(),['rendered']); await s.assemble(); assert.equal(s.calls.length,1); + assert.equal(result(s,push('next','fallback',{deferred:true})),'first'); + await s.assemble(); assert.equal(s.calls.length,2); + const empty=setup({render:async()=>({content:[]})}); + empty.c.accept(push('1','fallback',{deferred:true})); await empty.assemble(); + assert.deepEqual(empty.texts(),[]); assert.deepEqual(empty.c.pendingPushes(),[]); +}); + +test('vectors 21/23: timeout consumes fallback; late render cannot rewrite it', async () => { + const r=pending(); const s=setup({timeoutMs:5,render:()=>r.promise}); + s.c.accept(push('1','fallback',{deferred:true})); await s.assemble(); + assert.deepEqual(s.texts(),['fallback']); r.resolve(rendered('late')); await Promise.resolve(); await Promise.resolve(); + assert.deepEqual(s.texts(),['fallback']); assert(s.operations.some(r=>r.kind==='observed' && r.detail.kind==='late-render')); +}); + +for(const kind of ['error','malformed','oversized'] as const) test(`vector 21: ${kind} render uses bounded admitted fallback`,async()=>{ + const s=setup({maxContentBytes:100,render:async()=>{ + if(kind==='error') throw Error('offline'); + if(kind==='malformed') return {content:[null]} as never; + return rendered('x'.repeat(101)); + }}); + s.c.accept(push('1','fallback',{deferred:true})); await s.assemble(); assert.deepEqual(s.texts(),['fallback']); +}); + +test('vectors 22/25: concurrent assemblies share a frozen render; later notices wait', async () => { + const r=pending(); let calls=0; + const s=setup({render:()=>{calls++; return r.promise;}}); + s.c.accept(push('1','f1',{deferred:true})); + const a=s.assemble(), b=s.assemble(); + assert.equal(result(s,push('2','f2',{deferred:true})),'first'); + await Promise.resolve(); assert.equal(calls,1); + r.resolve(rendered('r1')); await Promise.all([a,b]); + assert.deepEqual(s.texts(),['r1']); assert.equal(s.c.pendingPushes()[0].params.eventId,'2'); + await s.assemble(); assert.equal(calls,2); assert.deepEqual(s.texts(),['r1','r1']); +}); + +test('vector 24: authority revocation before/during render suppresses all content', async () => { + let allowed=true; const r=pending(); let calls=0; + const s=setup({authorized:()=>allowed,render:()=>{calls++;return r.promise;}}); + s.c.accept(push('1','fallback',{deferred:true})); const work=s.assemble(); + await Promise.resolve(); allowed=false; r.resolve(rendered('secret')); await work; + assert.deepEqual(s.texts(),[]); assert.equal(calls,1); + allowed=true; s.c.accept(push('2','fallback',{deferred:true})); allowed=false; + await s.assemble(); assert.equal(calls,1); assert.deepEqual(s.texts(),[]); +}); + +for(const operation of ['retract','plain'] as const) for(const completion of ['result','timeout'] as const) + test(`vectors 27a–h: ${operation} cancels frozen AND newer pending batch (${completion})`,async()=>{ + const r=pending(); let calls=0; + const s=setup({timeoutMs:5,render:()=>{calls++;return r.promise;}}); + s.c.accept(push('1','f1',{deferred:true})); const work=s.assemble(); await Promise.resolve(); + s.c.accept(push('2','f2',{deferred:true})); + assert.equal(result(s,push('3',operation==='plain'?'snapshot':'deleted',operation==='retract'?{retract:true}:{})),operation==='plain'?'replaced':'retracted'); + if(completion==='result') r.resolve(rendered('old')); + await work; await s.assemble(); + assert.deepEqual(s.texts(),operation==='plain'?['snapshot']:[]); + assert.equal(calls,1); assert.deepEqual(s.c.pendingPushes(),[]); + }); + +test('vector 27i: deferred notice replaces unread plain content; consumed plain survives', async()=>{ + const s=setup(); s.c.accept(push('plain')); + assert.equal(result(s,push('notice','fallback',{deferred:true})),'replaced'); + await s.assemble(); assert.deepEqual(s.texts(),['rendered']); + s.c.accept(push('plain2')); await s.assemble(); + s.c.accept(push('notice2','fallback',{deferred:true})); await s.assemble(); + assert.deepEqual(s.texts(),['rendered','plain2','rendered']); +}); + +test('vector 27j: cancellation does not discard a notice admitted AFTER the plain replacement', async()=>{ + const r=pending(); const notices:string[][]=[]; + const s=setup({render:(_push,params)=>{notices.push(params.notices.map(n=>n.eventId));return notices.length===1?r.promise:Promise.resolve(rendered('new'));}}); + s.c.accept(push('1','f1',{deferred:true})); const work=s.assemble(); await Promise.resolve(); + s.c.accept(push('2','f2',{deferred:true})); s.c.accept(push('plain')); + s.c.accept(push('3','f3',{deferred:true})); r.resolve(rendered('old')); await work; + await s.assemble(); assert.deepEqual(notices,[['1'],['3']]); assert.deepEqual(s.texts(),['new']); +}); + +test('vectors 28–30/34: retraction removes unread content but retains consumed history', async()=>{ + const unread=setup(); unread.c.accept(push('create')); unread.c.accept(push('edit')); + assert.equal(result(unread,push('delete','deleted',{retract:true})),'retracted'); + await unread.assemble(); assert.deepEqual(unread.texts(),[]); + const s=setup(); s.c.accept(push('create')); await s.assemble(); s.c.accept(push('edit')); + assert.equal(result(s,push('delete','deleted',{retract:true})),'noted'); + await s.assemble(); assert.deepEqual(s.texts(),['create','deleted']); + s.c.accept(push('new')); assert.equal(result(s,push('delete2','',{retract:true})),'consumed'); + await s.assemble(); assert.deepEqual(s.texts(),['create','deleted']); +}); + +for(const flags of [{key:''},{key:'é'.repeat(129)},{deferred:true,retract:true},{data:{}},{deferred:true,data:'x'.repeat(4097)},{channelId:''}, {deferred:'yes'}]) + test(`vector 38: malformed ${JSON.stringify(flags).slice(0,70)} leaves subject untouched`,async()=>{ + const s=setup(); s.c.accept(push('good')); + assert.throws(()=>s.c.accept(push('bad','bad',flags)),CoalesceError); + await s.assemble(); assert.deepEqual(s.texts(),['good']); + }); + +test('receipts and projection queries cannot be mutated by callers; journal keeps private notice data',async()=>{ + const s=setup(); const p=push('one','fallback',{deferred:true,data:{private:'notice'}}); + const receipt=s.c.accept(p); receipt.coalesce!.outcome='consumed'; + p.params.payload.content[0]={type:'text',text:'mutated'}; + assert.equal(s.c.accept(p).coalesce!.outcome,'first'); + assert.equal(s.c.pendingPushes()[0].params.payload.content[0].type,'text'); + assert(JSON.stringify(s.operations).includes('notice')); + await s.assemble(); assert(!JSON.stringify(s.deliveries.map(d=>d.params.payload)).includes('notice')); +}); + +test('disconnect preserves acknowledged fallback; reconnect retry has no second effect',async()=>{ + const r=pending(); let available=true; + const s=setup({available:()=>available,render:()=>r.promise}); + const first=s.c.accept(push('one','fallback',{deferred:true})); const work=s.assemble(); + available=false; s.c.disconnect('server'); r.resolve(rendered('old')); await work; + assert.deepEqual(s.texts(),[]); assert.equal(s.c.pendingPushes().length,1); + available=true; + assert.deepEqual(s.c.accept(push('one','fallback',{deferred:true},{epoch:'next'})),first); + await s.assemble(); assert.deepEqual(s.texts(),['fallback']); + assert.deepEqual(PUSH_COALESCING_SUPPORT,{pushEvents:true,deferred:true,channelsIncoming:true,channelScopedPush:true}); +}); + +test('resource limit rejects new pending subject without evicting admitted content',async()=>{ + const s=setup({maxSubjects:1}); s.c.accept(push('one')); + assert.throws(()=>s.c.accept(push('two','two',{key:'another'})),/limit/); + await s.assemble(); assert.deepEqual(s.texts(),['one']); + assert.equal(result(s,push('two','two',{key:'another'})),'appended'); +}); + +test('restored pending subject keeps identity and unread history across transport epochs',async()=>{ + const before=setup();before.c.accept(push('c','pending')); + const after=setup({journal:before.journal});after.c.recover(); + assert.equal(result(after,push('d','deleted',{retract:true},{epoch:'new'})),'retracted'); + await after.assemble();assert.deepEqual(after.texts(),[]); +}); + +test('old receipts are rebuilt from journal positions without a persisted receipt index',()=>{ + const s=setup();const first=s.c.accept(push('original')); + for(let i=0;i<4100;i++)s.c.accept(push(`update-${i}`)); + const rebuilt=setup({journal:s.journal});const length=s.journal.length(); + assert.deepEqual(rebuilt.c.accept(push('original')),first);assert.equal(s.journal.length(),length); +}); + +test('an acceptance appended before a crash recovers with its receipt and pending work',async()=>{ + const journal=new MemoryCoalescingJournal(); + journal.afterAppend=record=>{if(record.operation.kind==='accepted')throw Error('crash after append');}; + const before=setup({journal});assert.throws(()=>before.c.accept(push('once')),/crash/); + journal.afterAppend=undefined; + const after=setup({journal});after.c.recover(); + assert.equal(result(after,push('once','once',{},{epoch:'reconnected'})),'first'); + await after.assemble();assert.deepEqual(after.texts(),['once']); +}); + +test('reassigning a server binding cannot replace or deduplicate the former principal',async()=>{ + const s=setup();assert.equal(result(s,push('same','old',{},{binding:'old'})),'first'); + assert.equal(result(s,push('same','new',{},{binding:'new'})),'first'); + await s.assemble();assert.deepEqual(s.texts(),['old','new']); +}); + +test('replacement cannot remove tune-out visibility restrictions',async()=>{ + const s=setup();s.c.accept(push('old','private',{},{routing:{targetAgents:['subconscious'],metadata:{tuneOut:{epochId:'muted'}}}})); + s.c.accept(push('new','public',{},{routing:{targetAgents:['resident'],metadata:{}}})); + await s.assemble();assert.deepEqual(s.texts(),[]); +});