From 144f12faa3dbebabbdeb0c4d743b05a87253537c Mon Sep 17 00:00:00 2001 From: Cuarroc <47299143+Cuarroc@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:38:00 +0200 Subject: [PATCH 1/2] feat(denkraum): split pending, inbox and delivery onto the store (DR-04b) Move pending(), inbox() and flushNotifications()/#deliver from the frozen P1 DeskStore (lines 515-569) into tools/denkraum/store/delivery.mjs as DeliveryStore extends AnswerStore. The code equals the split proof's delivery.mjs lines 108-162 after two renames: this[CHANGE]( to this.change( and this.io.rootAgentId to the injected this.rootAgentId (D2). The delivery id stays the event id until a queued delivery is older than seven days; a failed attempt keeps a fixed error code and no transport detail. On V2 every receipt lookup fails closed with 503 while no root agent id is configured. No new product logic, no HTTP, ENV or UI. Test-First: tools/denkraum/store/delivery.test.mjs::V2 pending inbox and delivery fail closed without a configured root agent id Co-Authored-By: Claude Opus 5.5 (1M context) --- tools/denkraum/store/delivery.mjs | 65 +++++++++++++++ tools/denkraum/store/delivery.test.mjs | 111 +++++++++++++++++++++++++ 2 files changed, 176 insertions(+) create mode 100644 tools/denkraum/store/delivery.mjs create mode 100644 tools/denkraum/store/delivery.test.mjs diff --git a/tools/denkraum/store/delivery.mjs b/tools/denkraum/store/delivery.mjs new file mode 100644 index 00000000..beab52e9 --- /dev/null +++ b/tools/denkraum/store/delivery.mjs @@ -0,0 +1,65 @@ +// Pending answers, Root inbox and webhook delivery with a stable delivery id (DR-04b) on DR-04a. +// Split from the P1 DeskStore (pending/inbox/flushNotifications); the root agent id stays injected (D2), +// so on V2 every receipt lookup here fails closed with 503 while it is missing. +import { randomUUID } from 'node:crypto'; +import { AnswerStore } from './answers.mjs'; +import { activeAnswers, activeEvents, ensure, isoTime, receiptFor, validId } from './model.mjs'; + +export class DeliveryStore extends AnswerStore { + #notificationQueue = Promise.resolve(); + async pending() { + const state = await this.read(); + return activeAnswers(state).filter(a => state.schemaVersion === 2 + ? !receiptFor(state, a, this.rootAgentId) : !a.ack); + } + async inbox() { + const state = await this.read(); const pending = []; const received = []; + for (const event of activeEvents(state)) { + const receipt = receiptFor(state, event, this.rootAgentId) || null; + const progress = state.progress?.find(p => p.eventId === event.eventRef.eventId) ?? null; + const legacyAck = event.eventRef.kind === 'answer' ? event.content.ack ?? null : null; + const envelope = { ...event, receipt, progress, legacyAck }; + (receipt || (state.schemaVersion === 1 && legacyAck?.status === 'received') ? received : pending).push(envelope); + } + return { pending, received }; + } + flushNotifications(eventId) { + const work = this.#notificationQueue.then(() => this.#deliver(eventId)); + this.#notificationQueue = work.catch(() => {}); return work; + } + async #deliver(eventId) { + if (typeof this.io.notifyEvent !== 'function') return { status: 'not-configured' }; + const event = await this.change(state => { + if (state.schemaVersion !== 2) return null; + const at = this.io.clock(); + const eligible = activeEvents(state).filter(e => (eventId === undefined || e.eventRef.eventId === eventId) && !receiptFor(state, e, this.rootAgentId)); + const a = eligible.map(e => e.content).filter(a => + (a.webhookDelivery?.status !== 'queued' || isoTime(a.webhookDelivery.queuedAt) && at - Date.parse(a.webhookDelivery.queuedAt) >= 7 * 86400000)) + .sort((x, y) => (Date.parse(x.webhookDelivery?.lastAttemptAt) || 0) - (Date.parse(y.webhookDelivery?.lastAttemptAt) || 0))[0]; + if (!a) return null; + const ref = eligible.find(e => e.content === a).eventRef; + if (!Object.hasOwn(a, 'webhookDelivery')) a.webhookDelivery = { status: 'pending', attempts: 0, lastAttemptAt: null, queuedAt: null, lastError: null }; + if (a.webhookDelivery?.status === 'queued') { + a.webhookDelivery.status = 'pending'; a.webhookDelivery.deliveryId = randomUUID(); a.webhookDelivery.queuedAt = null; + } + ensure(a.webhookDelivery && a.webhookDelivery.status === 'pending' && Number.isSafeInteger(a.webhookDelivery.attempts) && a.webhookDelivery.attempts >= 0 && a.webhookDelivery.attempts < Number.MAX_SAFE_INTEGER, 'Outbox beschädigt.', 503); + a.webhookDelivery.deliveryId ??= ref.eventId; + ensure(validId(a.webhookDelivery.deliveryId), 'Outbox-Zustellungs-ID beschädigt.', 503); + a.webhookDelivery.attempts++; a.webhookDelivery.lastAttemptAt = new Date(at).toISOString(); + return { type: 'decision-desk.event', eventId: ref.eventId, contentRevision: ref.kind === 'answer' ? ref.questionRevision : ref.ideaRevision, deliveryId: a.webhookDelivery.deliveryId }; + }); + if (!event) return { status: 'empty' }; + let queued = false; + try { await this.io.notifyEvent(event, event.deliveryId); queued = true; } catch { /* Persist only a fixed error code, never transport secrets. */ } + return this.change(state => { + const current = activeEvents(state).find(e => e.eventRef.eventId === event.eventId); + if (!current) return { status: 'superseded' }; + if (receiptFor(state, current, this.rootAgentId)) return { status: 'received' }; + const a = current.content; + ensure(a.webhookDelivery, 'Outboxereignis fehlt.', 409); + a.webhookDelivery.status = queued ? 'queued' : 'pending'; + a.webhookDelivery.queuedAt = queued ? new Date(this.io.clock()).toISOString() : null; a.webhookDelivery.lastError = queued ? null : 'transport-failed'; + return a.webhookDelivery; + }); + } +} diff --git a/tools/denkraum/store/delivery.test.mjs b/tools/denkraum/store/delivery.test.mjs new file mode 100644 index 00000000..65d013d6 --- /dev/null +++ b/tools/denkraum/store/delivery.test.mjs @@ -0,0 +1,111 @@ +import assert from 'node:assert/strict'; +import { mkdtemp, readFile, rename } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { DeliveryStore } from './delivery.mjs'; +import { answerRef, DeskError, validateQuestion } from './model.mjs'; + +const ROOT = 'test-root-agent'; +const DAY = 86400000; +const opt = id => ({ id, label: id, rationale: 'r', impact: 'i', tradeoff: 't', effort: 'e', reversible: 'ja' }); +const question = () => ({ ...validateQuestion({ id: 'E-1', title: 'T', context: 'C', owner: 'O', category: 'K', scope: 'S', source: 'Q', + uncertainty: 'U', recommendation: { optionIds: ['a'], rationale: 'R' }, options: [opt('a'), opt('b')] }), revision: 1 }); +const verify = async request => ({ ...request, rootAgentId: ROOT, rootAcknowledgedAt: '2026-10-08T12:00:00.000Z', observedProof: 'synthetic proof' }); +const store = (file, config) => new DeliveryStore(file, { rootAgentId: ROOT, verifyReceipt: verify, ...config }); +const input = (extra = {}) => ({ questionId: 'E-1', questionRevision: 1, expectedAnswerId: null, requestId: 'req-1', action: 'answer', selected: ['a'], note: 'x', ...extra }); +async function fresh({ v2 = true, ...config } = {}) { + const file = join(await mkdtemp(join(tmpdir(), 'denkraum-delivery-')), 'ledger.json'); + const base = store(file); await base.change(state => { state.questions.push(question()); }); + if (v2) await base.migrate((await base.read()).revision); + const s = store(file, config); return { s, file, a: await s.answer(input()) }; +} +const receiptInput = a => ({ receiptId: 'rcpt-1', eventRef: answerRef(a), transportMessageId: 'mail-1', rootReplyId: 'reply-1' }); +const status = code => e => e instanceof DeskError && e.status === code; + +test('pending and inbox keep an answer open until Root acknowledges it (V2 receipt, V1 legacy ack)', async () => { + const { s, a } = await fresh(); + assert.deepEqual((await s.pending()).map(x => x.id), [a.id]); + let box = await s.inbox(); + assert.deepEqual(box.pending.map(e => e.eventRef), [answerRef(a)]); assert.equal(box.pending[0].progress.currentStatus, 'incoming'); + const r = await s.receipt(receiptInput(a)); + assert.deepEqual(await s.pending(), []); box = await s.inbox(); + assert.equal(box.pending.length, 0); assert.deepEqual(box.received[0].receipt, r); + const v1 = await fresh({ v2: false }); + assert.equal((await v1.s.pending()).length, 1); + await v1.s.ack({ answerId: v1.a.id, status: 'received', actor: 'Root', note: 'Beleg', deliveryReceipt: 'weitergegeben' }); + assert.deepEqual(await v1.s.pending(), []); box = await v1.s.inbox(); + assert.equal(box.received[0].legacyAck.status, 'received'); assert.equal(box.received[0].receipt, null); +}); + +test('flushing without a transport or on a V1 ledger writes nothing and never calls out', async () => { + const { s, file } = await fresh(); const bytes = await readFile(file, 'utf8'); + assert.equal((await s.flushNotifications()).status, 'not-configured'); + assert.equal(await readFile(file, 'utf8'), bytes); + const v1 = await fresh({ v2: false }); const v1bytes = await readFile(v1.file, 'utf8'); const calls = []; + assert.equal((await store(v1.file, { notifyEvent: async event => calls.push(event) }).flushNotifications()).status, 'empty'); + assert.deepEqual(calls, []); assert.equal(await readFile(v1.file, 'utf8'), v1bytes); +}); + +test('a failed delivery stays pending with a fixed error code and the retry reuses the event id as delivery id', async () => { + const detail = 'test-only-transport-detail'; + const { s, a, file } = await fresh({ notifyEvent: async () => { throw new Error(detail); } }); + const failed = await s.flushNotifications(); + assert.equal(failed.status, 'pending'); assert.equal(failed.lastError, 'transport-failed'); assert.equal(failed.attempts, 1); + assert.ok(!(await readFile(file, 'utf8')).includes(detail)); + const calls = []; const restarted = store(file, { notifyEvent: async (event, id) => calls.push({ event, id }) }); + const queued = await restarted.flushNotifications(); + assert.equal(queued.status, 'queued'); assert.equal(queued.attempts, 2); assert.equal(queued.lastError, null); + assert.deepEqual(calls, [{ event: { type: 'decision-desk.event', eventId: a.id, contentRevision: 1, deliveryId: a.id }, id: a.id }]); + assert.equal((await restarted.flushNotifications()).status, 'empty', 'a queued event is not sent again'); + assert.equal(calls.length, 1); assert.equal((await restarted.read()).receipts.length, 0); + assert.equal((await restarted.pending()).length, 1, 'queued is not a Root receipt'); +}); + +test('an interrupted queue commit retries the same delivery id and concurrent flushes do not duplicate it', async () => { + const { file, a } = await fresh(); let commits = 0; const ids = []; + const interrupted = store(file, { notifyEvent: async (event, id) => ids.push(id), rename: async (from, to) => { + if (to === file && ++commits === 2) throw new Error('metadata interrupted'); return rename(from, to); + } }); + await assert.rejects(interrupted.flushNotifications(), /metadata interrupted/); + assert.equal((await store(file).read()).answers[0].webhookDelivery.status, 'pending'); + const restarted = store(file, { notifyEvent: async (event, id) => ids.push(id) }); + await Promise.all([restarted.flushNotifications(), restarted.flushNotifications()]); + assert.deepEqual(ids, [a.id, a.id]); +}); + +test('a receipt or a newer answer arriving during delivery wins over the queue update', async () => { + for (const race of ['receipt', 'supersession']) { + const { s, a, file } = await fresh(); let before; + s.io.notifyEvent = async () => { + if (race === 'receipt') await s.receipt(receiptInput(a)); + else await s.answer(input({ requestId: 'race', expectedAnswerId: a.id, action: 'defer', selected: [] })); + before = await readFile(file, 'utf8'); + }; + assert.equal((await s.flushNotifications()).status, race === 'receipt' ? 'received' : 'superseded'); + assert.equal(await readFile(file, 'utf8'), before); + } +}); + +test('a queued delivery older than seven days gets a new delivery id for the same event and stops after the Root receipt', async () => { + let clock = Date.parse('2026-10-08T12:00:00.000Z'); const calls = []; + const { s, a } = await fresh({ clock: () => clock, + notifyEvent: async (event, id) => { calls.push({ event, id }); if (calls.length === 2) throw new Error('offline'); } }); + await s.flushNotifications(); clock += 7 * DAY - 1; + assert.equal((await s.flushNotifications()).status, 'empty'); clock++; + assert.equal((await s.flushNotifications()).status, 'pending'); + assert.equal(calls[1].event.eventId, a.id); assert.notEqual(calls[1].id, a.id); + await s.flushNotifications(); assert.equal(calls[2].id, calls[1].id, 'a failed renewal keeps its new id'); + assert.equal((await s.read()).answers[0].webhookDelivery.attempts, 3); + await s.receipt(receiptInput(a)); clock += 31 * DAY; + assert.equal((await s.flushNotifications()).status, 'empty'); assert.equal(calls.length, 3); +}); + +test('V2 pending inbox and delivery fail closed without a configured root agent id', async () => { + const { file } = await fresh(); const bytes = await readFile(file, 'utf8'); const calls = []; + const rootless = new DeliveryStore(file, { notifyEvent: async event => calls.push(event) }); + for (const call of [() => rootless.pending(), () => rootless.inbox(), () => rootless.flushNotifications()]) await assert.rejects(call(), status(503)); + assert.deepEqual(calls, []); assert.equal(await readFile(file, 'utf8'), bytes); + const v1 = await fresh({ v2: false }); + assert.equal((await new DeliveryStore(v1.file).pending()).length, 1, 'V1 has no receipts; the DR-02 model binds D2 to V2 only'); +}); From dc34cae381e0e113a7ed406d81b018f0e26d1cee Mon Sep 17 00:00:00 2001 From: Cuarroc <47299143+Cuarroc@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:24:23 +0200 Subject: [PATCH 2/2] fix(denkraum): require the root id on V2 even for an empty ledger (R669-O1) OpenAI R1 on #669 found that pending(), inbox() and flushNotifications() only failed closed once an event reached receiptFor(). On a V2 ledger without active events, or with flushNotifications('other-event'), they answered [] / empty lists / 'empty' without a configured root agent id. All three paths now check the root id on V2 right after reading the state, before any event is filtered. V1 stays unchanged (no receipts). A missing transport still returns 'not-configured' first: P1 returns it before reading the ledger, so no root-bound lookup runs (D1). Test-First: tools/denkraum/store/delivery.test.mjs::V2 reads and delivery refuse a missing root agent id even when no event needs a receipt check (R669-O1) Co-Authored-By: Claude Opus 5.5 (1M context) --- tools/denkraum/store/delivery.mjs | 10 +++++++--- tools/denkraum/store/delivery.test.mjs | 11 +++++++++++ 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/tools/denkraum/store/delivery.mjs b/tools/denkraum/store/delivery.mjs index beab52e9..627dc352 100644 --- a/tools/denkraum/store/delivery.mjs +++ b/tools/denkraum/store/delivery.mjs @@ -5,15 +5,19 @@ import { randomUUID } from 'node:crypto'; import { AnswerStore } from './answers.mjs'; import { activeAnswers, activeEvents, ensure, isoTime, receiptFor, validId } from './model.mjs'; +// R669-O1: on V2 every path here checks the root id before reading events, so an empty ledger also answers 503. +const requireRoot = (state, rootAgentId) => ensure(state.schemaVersion !== 2 || validId(rootAgentId), + 'Root-Agent-ID ist nicht konfiguriert; Empfangsprüfung abgelehnt.', 503); + export class DeliveryStore extends AnswerStore { #notificationQueue = Promise.resolve(); async pending() { - const state = await this.read(); + const state = await this.read(); requireRoot(state, this.rootAgentId); return activeAnswers(state).filter(a => state.schemaVersion === 2 ? !receiptFor(state, a, this.rootAgentId) : !a.ack); } async inbox() { - const state = await this.read(); const pending = []; const received = []; + const state = await this.read(); requireRoot(state, this.rootAgentId); const pending = []; const received = []; for (const event of activeEvents(state)) { const receipt = receiptFor(state, event, this.rootAgentId) || null; const progress = state.progress?.find(p => p.eventId === event.eventRef.eventId) ?? null; @@ -31,7 +35,7 @@ export class DeliveryStore extends AnswerStore { if (typeof this.io.notifyEvent !== 'function') return { status: 'not-configured' }; const event = await this.change(state => { if (state.schemaVersion !== 2) return null; - const at = this.io.clock(); + requireRoot(state, this.rootAgentId); const at = this.io.clock(); const eligible = activeEvents(state).filter(e => (eventId === undefined || e.eventRef.eventId === eventId) && !receiptFor(state, e, this.rootAgentId)); const a = eligible.map(e => e.content).filter(a => (a.webhookDelivery?.status !== 'queued' || isoTime(a.webhookDelivery.queuedAt) && at - Date.parse(a.webhookDelivery.queuedAt) >= 7 * 86400000)) diff --git a/tools/denkraum/store/delivery.test.mjs b/tools/denkraum/store/delivery.test.mjs index 65d013d6..e86300dc 100644 --- a/tools/denkraum/store/delivery.test.mjs +++ b/tools/denkraum/store/delivery.test.mjs @@ -109,3 +109,14 @@ test('V2 pending inbox and delivery fail closed without a configured root agent const v1 = await fresh({ v2: false }); assert.equal((await new DeliveryStore(v1.file).pending()).length, 1, 'V1 has no receipts; the DR-02 model binds D2 to V2 only'); }); + +test('V2 reads and delivery refuse a missing root agent id even when no event needs a receipt check (R669-O1)', async () => { + const file = join(await mkdtemp(join(tmpdir(), 'denkraum-delivery-')), 'ledger.json'); const base = store(file); + await base.change(state => { state.questions.push(question()); }); await base.migrate((await base.read()).revision); + const calls = []; const bytes = await readFile(file, 'utf8'); + const rootless = new DeliveryStore(file, { notifyEvent: async event => calls.push(event) }); + for (const call of [() => rootless.pending(), () => rootless.inbox(), () => rootless.flushNotifications(), () => rootless.flushNotifications('other-event')]) + await assert.rejects(call(), status(503)); + assert.deepEqual(calls, []); assert.equal(await readFile(file, 'utf8'), bytes); + assert.equal((await new DeliveryStore(file).flushNotifications()).status, 'not-configured', 'P1: no transport is reported before any ledger read'); +});