diff --git a/changelog.d/cordelia-f96-log-before-memory.fixed.md b/changelog.d/cordelia-f96-log-before-memory.fixed.md new file mode 100644 index 000000000..0ff2ca7e3 --- /dev/null +++ b/changelog.d/cordelia-f96-log-before-memory.fixed.md @@ -0,0 +1 @@ +- A channel decision whose durable record fails to write no longer takes effect in memory alone (the first part of #417). The channel registry now appends to its lifecycle log before it changes memory, so a `channel_close`, `channel_open`, `channel_decline` or tune-out whose decision fails to append, as on a full disk, fails with the decision as it was, and a retry records it. Before, memory changed first: a retried close or open took the same-state return and reported success without recording anything, and after a restart the old record came back, so subscription policy could lift a close the agent had been told succeeded. The same order now holds for a tune-out's wake count, for the record that a server's recipe policy has been migrated, and for the channel label history that lets an agent name a channel it isn't connected to, including a label first seen in a push event. diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 082e6ed9b..5742a7497 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -1171,20 +1171,34 @@ export class ChannelRegistry { (existing.descriptor.metadata as { lazyRegistered?: boolean } | undefined)?.lazyRegistered && label && existing.descriptor.label === channelId ) { - existing.descriptor.label = label; - if (extraMetadata) { - existing.descriptor.metadata = { ...existing.descriptor.metadata, ...extraMetadata }; - } + const metadata = extraMetadata + ? { ...existing.descriptor.metadata, ...extraMetadata } + : existing.descriptor.metadata; // The channel now has a real label, not just the bare id — durable. - this.appendLabelSighting( - existing.descriptor.id, - existing.descriptor.label, - extractDmMeta(existing.descriptor.metadata), - ); + // Recorded before the live descriptor takes it, since this branch's + // guard reads the descriptor: an append that throws leaves the + // placeholder, so the next push tries again. + this.appendLabelSighting(existing.descriptor.id, label, extractDmMeta(metadata)); + existing.descriptor.label = label; + existing.descriptor.metadata = metadata; } return; } + // Only persist a REAL label durably. Falling back to the bare + // channelId as a placeholder (no server-supplied label at all — e.g. a + // Discord DM push missing both channelName and an author name, + // framework.ts's derivePushEventChannel) must never overwrite a real + // label already on file from a previous boot: after a restart the live + // `channels` map starts empty, so THIS is the very branch a + // label-less post-restart event takes — recording the placeholder here + // would clobber the good label initializeLabelHistoryStore() just + // replayed, defeating the entire point of the durable log. The live + // descriptor below still carries the placeholder for in-process + // routing; only the durable write is skipped when there's no real + // label to record. A real label is recorded before the entry exists: + // an append that throws registers nothing, so the next push tries again. + if (label) this.appendLabelSighting(channelId, label, extractDmMeta(extraMetadata)); const key = `${serverId}:${channelId}`; this.channels.set(key, { serverId, @@ -1197,19 +1211,6 @@ export class ChannelRegistry { }, open: false, }); - // Only persist a REAL label durably. Falling back to the bare - // channelId as a placeholder (no server-supplied label at all — e.g. a - // Discord DM push missing both channelName and an author name, - // framework.ts's derivePushEventChannel) must never overwrite a real - // label already on file from a previous boot: after a restart the live - // `channels` map starts empty, so THIS is the very branch a - // label-less post-restart event takes — recording the placeholder here - // would clobber the good label initializeLabelHistoryStore() just - // replayed, defeating the entire point of the durable log. The live - // descriptor above still carries the placeholder for in-process - // routing; only the durable write is skipped when there's no real - // label to record. - if (label) this.appendLabelSighting(channelId, label, extractDmMeta(extraMetadata)); this.emitTraceFn({ type: 'mcpl:channel-lazy-registered', serverId, @@ -1598,6 +1599,15 @@ export class ChannelRegistry { } } + /** + * Each writer appends its record here before it changes memory for it, so + * an append that throws, as on a full disk, fails the operation with that + * change unmade, and memory and the log still agree. A retry then records + * the change rather than finding it in memory already, where + * setDesiredState's same-state return would record nothing, and a restart + * replays what memory held. appendLabelSighting keeps the same order for + * the label history. + */ private appendLifecycleEvent(event: ChannelLifecycleEvent): void { this.store?.appendToStateJson(CHANNEL_LIFECYCLE_LOG_ID, event); } @@ -1703,6 +1713,16 @@ export class ChannelRegistry { const recipientNameUnchanged = recipientName === undefined || recipientName === this.dmRecipientNames.get(channelId); const isDmUnchanged = isDm === undefined || this.dmClassified.get(channelId) === true; if (labelUnchanged && recipientIdUnchanged && recipientNameUnchanged && isDmUnchanged) return; + // The log first, as for lifecycle events: an append that throws leaves + // the history in memory as it was, so the next sighting records it. + this.store?.appendToStateJson(CHANNEL_LABEL_HISTORY_LOG_ID, { + channelId, + label, + ts: Date.now(), + ...(recipientId ? { recipientId } : {}), + ...(recipientName ? { recipientName } : {}), + ...(isDm ? { isDm } : {}), + } satisfies ChannelLabelSightingEvent); this.labelHistory.set(channelId, label); let seen = this.allLabelsSeen.get(channelId); if (!seen) { @@ -1713,14 +1733,6 @@ export class ChannelRegistry { if (recipientId) this.dmRecipientIds.set(channelId, recipientId); if (recipientName) this.dmRecipientNames.set(channelId, recipientName); if (isDm) this.dmClassified.set(channelId, true); - this.store?.appendToStateJson(CHANNEL_LABEL_HISTORY_LOG_ID, { - channelId, - label, - ts: Date.now(), - ...(recipientId ? { recipientId } : {}), - ...(recipientName ? { recipientName } : {}), - ...(isDm ? { isDm } : {}), - } satisfies ChannelLabelSightingEvent); } /** @@ -1945,7 +1957,6 @@ export class ChannelRegistry { ): void { const key = this.lifecycleKey(serverId, channelId); if (this.desiredStates.get(key)?.state === desired) return; - this.desiredStates.set(key, { state: desired, source }); this.appendLifecycleEvent({ kind: 'desired-state', serverId, @@ -1954,6 +1965,7 @@ export class ChannelRegistry { source, timestamp: new Date().toISOString(), }); + this.desiredStates.set(key, { state: desired, source }); } // ========================================================================== @@ -1973,12 +1985,6 @@ export class ChannelRegistry { source: string, ): void { const key = this.lifecycleKey(serverId, channelId); - this.desiredStates.set(key, { - state: 'tuned-out', - source, - tuneOut: params, - wakeCount: 0, - }); this.appendLifecycleEvent({ kind: 'desired-state', serverId, @@ -1988,6 +1994,12 @@ export class ChannelRegistry { source, timestamp: new Date().toISOString(), }); + this.desiredStates.set(key, { + state: 'tuned-out', + source, + tuneOut: params, + wakeCount: 0, + }); } /** @@ -2006,7 +2018,6 @@ export class ChannelRegistry { const current = this.desiredStates.get(key); if (current?.state !== 'tuned-out' || !current.tuneOut) return null; const ended = { params: current.tuneOut, wakeCount: current.wakeCount ?? 0 }; - this.desiredStates.set(key, { state: nextState, source }); this.appendLifecycleEvent({ kind: 'desired-state', serverId, @@ -2015,6 +2026,7 @@ export class ChannelRegistry { source, timestamp: new Date().toISOString(), }); + this.desiredStates.set(key, { state: nextState, source }); return ended; } @@ -2032,16 +2044,17 @@ export class ChannelRegistry { const key = this.lifecycleKey(serverId, channelId); const current = this.desiredStates.get(key); if (current?.state !== 'tuned-out' || !current.tuneOut) return null; - current.wakeCount = (current.wakeCount ?? 0) + 1; + const wakeCount = (current.wakeCount ?? 0) + 1; this.appendLifecycleEvent({ kind: 'tune-out-wake', serverId, channelId, epochId: current.tuneOut.epochId, - wakeCount: current.wakeCount, + wakeCount, timestamp: new Date().toISOString(), }); - return { params: current.tuneOut, wakeCount: current.wakeCount }; + current.wakeCount = wakeCount; + return { params: current.tuneOut, wakeCount }; } /** Active tune-out params + wake count for a channel, or null. */ @@ -2116,13 +2129,13 @@ export class ChannelRegistry { this.setDesiredState(serverId, channel.id, desired, 'legacy-recipe-migration'); } - this.migratedLegacyPolicies.add(serverId); this.appendLifecycleEvent({ kind: 'legacy-policy-migrated', serverId, source: Array.isArray(policy) ? 'allow-list' : policy, timestamp: new Date().toISOString(), }); + this.migratedLegacyPolicies.add(serverId); } /** diff --git a/test/channel-decision-log-first.test.ts b/test/channel-decision-log-first.test.ts new file mode 100644 index 000000000..c4c63645d --- /dev/null +++ b/test/channel-decision-log-first.test.ts @@ -0,0 +1,293 @@ +/** + * A durable write that fails leaves memory as it was (the first part of + * #417). Each writer of the channel registry's lifecycle and label logs + * appends to the store before it changes memory. So an operation whose + * append throws, as on a full disk, fails without its change, a retry + * records it rather than taking a same-state return, and a registry rebuilt + * from the store replays what memory held. Before, memory changed first: a retried `channel_close` + * reported success without recording anything, and after a restart the + * old record came back, so subscription policy could lift a close the + * agent had been told succeeded. + */ +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 { CapabilityGrant, ALL_CAPABILITY_PATHS } from '../src/mcpl/capability-grant.js'; +import { ChannelRegistry, type TuneOutParams } from '../src/mcpl/channel-registry.js'; +import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; +import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; + +const LIFECYCLE = 'mcpl/channel-lifecycle'; +const LABELS = 'mcpl/channel-label-history'; +const DISK_FULL = /ENOSPC: no space left on device/; + +type Write = { id: string; item: unknown }; + +/** + * The real store, made to throw on one chosen append as a full disk would. + * It carries the calls the registry makes here. + */ +function failingStore(path: string) { + const real = JsStore.openOrCreate({ path }); + let chosen: ((write: Write) => boolean) | undefined; + const throwIfChosen = (write: Write) => { + if (chosen?.(write)) { + chosen = undefined; + throw new Error('ENOSPC: no space left on device'); + } + }; + const store = { + registerState: (opts: Parameters[0]) => real.registerState(opts), + getStateJson: (id: string) => real.getStateJson(id), + appendToStateJson: (id: string, item: unknown) => { + throwIfChosen({ id, item }); + return real.appendToStateJson(id, item as never); + }, + } as unknown as JsStore; + return { + store, + failNext: (choose: (write: Write) => boolean) => { + chosen = choose; + }, + close: () => real.close(), + }; +} + +function withStore( + name: string, + body: (store: ReturnType) => Promise | void, +): () => Promise { + return async () => { + const dir = mkdtempSync(join(tmpdir(), `${name}-`)); + let opened: ReturnType | undefined; + try { + opened = failingStore(join(dir, 'store')); + await body(opened); + } finally { + // Closed before its directory goes, however the test ended. + try { + opened?.close(); + } finally { + rmSync(dir, { recursive: true, force: true }); + } + } + }; +} + +function makeRegistry(store: JsStore) { + const openCalls: string[] = []; + const closeCalls: string[] = []; + const ackCalls: string[] = []; + const server = { + grant: new CapabilityGrant(new Set(ALL_CAPABILITY_PATHS), []), + sendChannelsOpen: async (params: { channelId: string }) => { + openCalls.push(params.channelId); + return {}; + }, + sendChannelsClose: async (params: { channelId: string }) => { + closeCalls.push(params.channelId); + return {}; + }, + sendChannelsAcknowledge: async (params: { messageId: string }) => { + ackCalls.push(params.messageId); + return { acknowledged: true, representation: 'not now' }; + }, + }; + const registry = new ChannelRegistry( + { getServer: () => server } as unknown as McplServerRegistry, + {} as FeatureSetManager, + () => {}, + () => {}, + { store }, + ); + return { registry, openCalls, closeCalls, ackCalls }; +} + +/** The decision on record in memory, with its source. */ +function decision(registry: ChannelRegistry, channelId: string) { + const internals = registry as unknown as { + desiredStates: Map; + lifecycleKey(serverId: string, channelId: string): string; + }; + const record = internals.desiredStates.get(internals.lifecycleKey('discord', channelId)); + return record && { state: record.state, source: record.source }; +} + +const commons = { id: 'c1', type: 'discord', label: '#commons', direction: 'bidirectional' as const }; +const ops = { id: 'c2', type: 'discord', label: '#ops', direction: 'bidirectional' as const }; + +const PARAMS: TuneOutParams = { + epochId: 'epoch-1', + cadenceSeconds: 1800, + backlogCap: 200, + maxWakes: 5, + startedAtSequence: 42, +}; + +const lifecycleWrite = ({ id }: Write) => id === LIFECYCLE; +const decisionWrite = ({ id, item }: Write) => + id === LIFECYCLE && (item as { kind?: string }).kind === 'desired-state'; + +test('a channel_close whose record fails to append fails and leaves the channel open; its retry records the close a restart replays', withStore('log-first-close', async ({ store, failNext }) => { + const { registry, closeCalls } = makeRegistry(store); + await registry.handleRegister('discord', { channels: [commons] }); + assert.equal((await registry.handleChannelToolCall('channel_open', { channelId: 'c1' })).success, true); + assert.deepEqual(decision(registry, 'c1'), { state: 'open', source: 'agent-tool' }); + const closesBefore = closeCalls.length; + + failNext(lifecycleWrite); + await assert.rejects(registry.handleChannelToolCall('channel_close', { channelId: 'c1' }), DISK_FULL); + assert.deepEqual(decision(registry, 'c1'), { state: 'open', source: 'agent-tool' }); + assert.equal(closeCalls.length, closesBefore, 'nothing went to the server'); + + const retried = await registry.handleChannelToolCall('channel_close', { channelId: 'c1' }); + assert.deepEqual(retried.data, { channelId: 'c1', status: 'closed' }); + assert.deepEqual(decision(registry, 'c1'), { state: 'closed', source: 'agent-tool' }); + + const restarted = makeRegistry(store).registry; + assert.deepEqual(decision(restarted, 'c1'), { state: 'closed', source: 'agent-tool' }); +})); + +test('a channel_open whose decision fails to append answers the failure and opens nothing; its retry records the open a restart replays', withStore('log-first-open', async ({ store, failNext }) => { + const { registry, openCalls } = makeRegistry(store); + await registry.handleRegister('discord', { channels: [commons] }); + const before = decision(registry, 'c1'); + assert.equal(before?.state, 'closed'); + const opensBefore = openCalls.length; + + failNext(decisionWrite); + const failed = await registry.handleChannelToolCall('channel_open', { channelId: 'c1' }); + assert.equal(failed.success, false); + assert.match(failed.error ?? '', DISK_FULL); + assert.deepEqual(decision(registry, 'c1'), before); + assert.equal(openCalls.length, opensBefore, 'nothing went to the server'); + + const retried = await registry.handleChannelToolCall('channel_open', { channelId: 'c1' }); + assert.equal(retried.success, true); + assert.equal(openCalls.length, opensBefore + 1); + const restarted = makeRegistry(store).registry; + assert.deepEqual(decision(restarted, 'c1'), { state: 'open', source: 'agent-tool' }); +})); + +test('a channel_decline whose decision fails to append fails and acknowledges nothing; its retry records the decline a restart replays', withStore('log-first-decline', async ({ store, failNext }) => { + // A channel first seen in a push has no decision yet, so the decline's + // decision is a change of state and has a record to append. + const { registry, ackCalls } = makeRegistry(store); + registry.ensureChannelRegistered('discord', 'c7', '#lobby'); + assert.equal(decision(registry, 'c7'), undefined); + const decline = { channelId: 'c7', messageId: 'm1', acknowledge: 'not now' }; + + failNext(decisionWrite); + await assert.rejects(registry.handleChannelToolCall('channel_decline', decline), DISK_FULL); + assert.equal(decision(registry, 'c7'), undefined); + assert.equal(ackCalls.length, 0, 'nothing went to the server'); + + const retried = await registry.handleChannelToolCall('channel_decline', decline); + assert.equal((retried.data as { status?: string }).status, 'remained closed'); + assert.deepEqual(ackCalls, ['m1']); + const restarted = makeRegistry(store).registry; + assert.deepEqual(decision(restarted, 'c7'), { state: 'closed', source: 'invitation-declined' }); +})); + +test('a tune-out whose record fails to append fails and leaves the channel undecided; its retry is what a restart replays', withStore('log-first-tune-out', ({ store, failNext }) => { + const { registry } = makeRegistry(store); + failNext(lifecycleWrite); + assert.throws(() => registry.enterTuneOut('discord', 'c1', PARAMS, 'agent-tool'), DISK_FULL); + assert.equal(registry.getDesiredState('discord', 'c1'), undefined); + assert.equal(registry.getTuneOutState('discord', 'c1'), null); + + registry.enterTuneOut('discord', 'c1', PARAMS, 'agent-tool'); + const restarted = makeRegistry(store).registry; + assert.deepEqual(restarted.getTuneOutState('discord', 'c1'), { params: PARAMS, wakeCount: 0 }); +})); + +test('a tune-out wake whose record fails to append fails uncounted; its retry counts once, as a restart replays', withStore('log-first-wake', ({ store, failNext }) => { + const { registry } = makeRegistry(store); + registry.enterTuneOut('discord', 'c1', PARAMS, 'agent-tool'); + + failNext(lifecycleWrite); + assert.throws(() => registry.recordTuneOutWake('discord', 'c1'), DISK_FULL); + assert.deepEqual(registry.getTuneOutState('discord', 'c1'), { params: PARAMS, wakeCount: 0 }); + + assert.deepEqual(registry.recordTuneOutWake('discord', 'c1'), { params: PARAMS, wakeCount: 1 }); + const restarted = makeRegistry(store).registry; + assert.deepEqual(restarted.getTuneOutState('discord', 'c1'), { params: PARAMS, wakeCount: 1 }); +})); + +test('a tune-out cancel whose record fails to append fails and leaves the tune-out in place; its retry ends it, as a restart replays', withStore('log-first-cancel', ({ store, failNext }) => { + const { registry } = makeRegistry(store); + registry.enterTuneOut('discord', 'c1', PARAMS, 'agent-tool'); + + failNext(lifecycleWrite); + assert.throws(() => registry.cancelTuneOut('discord', 'c1', 'open', 'agent-tool'), DISK_FULL); + assert.equal(registry.getDesiredState('discord', 'c1'), 'tuned-out'); + + assert.deepEqual(registry.cancelTuneOut('discord', 'c1', 'open', 'agent-tool'), { params: PARAMS, wakeCount: 0 }); + const restarted = makeRegistry(store).registry; + assert.equal(restarted.getDesiredState('discord', 'c1'), 'open'); + assert.equal(restarted.getTuneOutState('discord', 'c1'), null); +})); + +test('a recipe migration whose record fails to append runs again at the next registration, so after a restart a new channel is one nobody decided about', withStore('log-first-migration', async ({ store, failNext }) => { + const { registry } = makeRegistry(store); + registry.setSubscriptionPolicy('discord', ['c1']); + failNext(({ id, item }) => id === LIFECYCLE && (item as { kind?: string }).kind === 'legacy-policy-migrated'); + await assert.rejects(registry.handleRegister('discord', { channels: [commons] }), DISK_FULL); + await registry.handleRegister('discord', { channels: [commons] }); + assert.deepEqual(decision(registry, 'c1'), { state: 'open', source: 'legacy-recipe-migration' }); + + // The recipe's allow-list doesn't name c2. Migrated again, it would + // record c2 closed as the recipe's decision, which policy never lifts. + const restarted = makeRegistry(store).registry; + restarted.setSubscriptionPolicy('discord', ['c1']); + await restarted.handleRegister('discord', { channels: [commons, ops] }); + assert.deepEqual(decision(restarted, 'c2'), { state: 'closed', source: 'default-closed' }); +})); + +test('a label sighting whose record fails to append is recorded at the next sighting, so a restart still finds the channel by its label', withStore('log-first-label', async ({ store, failNext }) => { + const { registry } = makeRegistry(store); + failNext(({ id }) => id === LABELS); + await assert.rejects(registry.handleRegister('discord', { channels: [commons] }), DISK_FULL); + await registry.handleRegister('discord', { channels: [commons] }); + + const restarted = makeRegistry(store).registry; + assert.deepEqual(restarted.resolveProseTargetDurable('#commons'), { channelId: 'c1', label: '#commons' }); +})); + +test('a label first seen in a push, whose sighting fails to append, is recorded by the next push, so a restart still finds the channel by its label', withStore('log-first-push-label', ({ store, failNext }) => { + const { registry } = makeRegistry(store); + // A bare lazy registration, then a push that reveals its label. + registry.ensureChannelRegistered('discord', 'c9'); + failNext(({ id }) => id === LABELS); + assert.throws(() => registry.ensureChannelRegistered('discord', 'c9', '#commons'), DISK_FULL); + assert.ok('error' in registry.resolveProseTarget('#commons'), 'the live descriptor keeps its placeholder'); + registry.ensureChannelRegistered('discord', 'c9', '#commons'); + + // A channel whose first push carries its label. + failNext(({ id }) => id === LABELS); + assert.throws(() => registry.ensureChannelRegistered('discord', 'c8', '#ops'), DISK_FULL); + assert.ok('error' in registry.resolveProseTarget('c8'), 'nothing is registered'); + registry.ensureChannelRegistered('discord', 'c8', '#ops'); + + const restarted = makeRegistry(store).registry; + assert.deepEqual(restarted.resolveProseTargetDurable('#commons'), { channelId: 'c9', label: '#commons' }); + assert.deepEqual(restarted.resolveProseTargetDurable('#ops'), { channelId: 'c8', label: '#ops' }); +})); + +test('a DM recipient whose sighting fails to append is recorded at the next sighting, so a restart still finds the DM by its mention', withStore('log-first-dm', async ({ store, failNext }) => { + const { registry } = makeRegistry(store); + const dm = (metadata: Record) => + ({ id: 'discord:dm:antra', type: 'discord', label: 'DM: antra', direction: 'bidirectional' as const, metadata }); + await registry.handleRegister('discord', { channels: [dm({ channelType: 'dm' })] }); + + // The same label, now with the recipient's id. + failNext(({ id }) => id === LABELS); + await assert.rejects(registry.handleRegister('discord', { channels: [dm({ channelType: 'dm', recipientId: '123456789' })] }), DISK_FULL); + await registry.handleRegister('discord', { channels: [dm({ channelType: 'dm', recipientId: '123456789' })] }); + + const restarted = makeRegistry(store).registry; + assert.deepEqual(restarted.resolveProseTargetDurable('<@123456789>'), { channelId: 'discord:dm:antra', label: 'DM: antra' }); +}));