From e401137a6d68856f0a27b8dac36147041d1b4856 Mon Sep 17 00:00:00 2001 From: dundas Date: Sat, 28 Feb 2026 13:32:49 -0600 Subject: [PATCH 1/6] =?UTF-8?q?fix(round-tables):=20address=20second=20rev?= =?UTF-8?q?iew=20=E2=80=94=20split-brain,=20error=20codes,=20expiry=20noti?= =?UTF-8?q?fication,=20TTL=20purge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix split-brain: only store participants that successfully enrolled in the backing ADMP group (enrolledParticipants vs uniqueParticipants) - Fail-fast when zero participants enroll; clean up orphaned group - Add makeError() helper so routes use error.statusCode, not string-matching - Use nullish coalescing (??) for timeout_minutes default (0 was falsy) - Require integer timeout_minutes; reject floats (e.g. 1.5) - Notify participants via inbox on expireStale() so they know the session ended - Add purgeStaleRoundTables() to memory.js and purgeStale() service method to prevent unbounded storage growth; hooked into cleanup loop with configurable ROUND_TABLE_PURGE_TTL_MS env var (default: 7 days) - Add 5 missing tests: facilitator speak, expireStale lifecycle, dedup, goal validation, non-integer timeout validation Co-Authored-By: Claude Sonnet 4.6 --- src/routes/round-tables.js | 12 +-- src/server.js | 7 +- src/server.test.js | 113 ++++++++++++++++++++++++++++ src/services/round-table.service.js | 87 +++++++++++++++------ src/storage/memory.js | 15 ++++ 5 files changed, 202 insertions(+), 32 deletions(-) diff --git a/src/routes/round-tables.js b/src/routes/round-tables.js index 9733956..10b9479 100644 --- a/src/routes/round-tables.js +++ b/src/routes/round-tables.js @@ -10,11 +10,7 @@ import { authenticateAgent } from '../middleware/auth.js'; const router = express.Router(); function getErrorStatusCode(error) { - const msg = error.message || ''; - if (msg.includes('not found')) return 404; - if (msg.includes('Not a participant') || msg.includes('Only the facilitator')) return 403; - if (msg.includes('already resolved') || msg.includes('has expired') || msg.includes('maximum of 200')) return 409; - return 400; + return error.statusCode || 400; } /** @@ -44,8 +40,8 @@ router.post('/', authenticateAgent, async (req, res) => { if (invalidParticipant !== undefined) { return res.status(400).json({ error: 'INVALID_PARTICIPANT_ID', message: 'each participant must be a non-empty string of 255 characters or less' }); } - if (timeout_minutes !== undefined && (typeof timeout_minutes !== 'number' || !Number.isFinite(timeout_minutes))) { - return res.status(400).json({ error: 'INVALID_TIMEOUT', message: 'timeout_minutes must be a number' }); + if (timeout_minutes !== undefined && (typeof timeout_minutes !== 'number' || !Number.isFinite(timeout_minutes) || !Number.isInteger(timeout_minutes))) { + return res.status(400).json({ error: 'INVALID_TIMEOUT', message: 'timeout_minutes must be an integer' }); } const rt = await roundTableService.create({ @@ -53,7 +49,7 @@ router.post('/', authenticateAgent, async (req, res) => { goal: goal.trim(), facilitator: req.agent.agent_id, participants, - timeout_minutes: timeout_minutes || 30 + timeout_minutes: timeout_minutes ?? 30 }); res.status(201).json(rt); diff --git a/src/server.js b/src/server.js index 59fc767..86f142b 100644 --- a/src/server.js +++ b/src/server.js @@ -36,6 +36,7 @@ config(); const PORT = process.env.PORT || 8080; const CLEANUP_INTERVAL_MS = parseInt(process.env.CLEANUP_INTERVAL_MS) || 60000; +const ROUND_TABLE_PURGE_TTL_MS = parseInt(process.env.ROUND_TABLE_PURGE_TTL_MS) || 7 * 24 * 60 * 60 * 1000; // Warn about insecure outbox webhook configuration if (process.env.MAILGUN_API_KEY && !process.env.MAILGUN_WEBHOOK_SIGNING_KEY) { @@ -221,14 +222,16 @@ function startBackgroundJobs() { const messagesDeleted = await storage.cleanupExpiredMessages(); const ephemeralPurged = await inboxService.purgeExpiredEphemeralMessages(); const roundTablesExpired = await roundTableService.expireStale(); + const roundTablesPurged = await roundTableService.purgeStale(ROUND_TABLE_PURGE_TTL_MS); - if (leasesReclaimed > 0 || messagesExpired > 0 || messagesDeleted > 0 || ephemeralPurged > 0 || roundTablesExpired > 0) { + if (leasesReclaimed > 0 || messagesExpired > 0 || messagesDeleted > 0 || ephemeralPurged > 0 || roundTablesExpired > 0 || roundTablesPurged > 0) { logger.debug({ leasesReclaimed, messagesExpired, messagesDeleted, ephemeralPurged, - roundTablesExpired + roundTablesExpired, + roundTablesPurged }, 'Cleanup job completed'); } } catch (error) { diff --git a/src/server.test.js b/src/server.test.js index 4dbf047..0698ffb 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -11,6 +11,7 @@ import { requireApiKey } from './middleware/auth.js'; import { webhookService } from './services/webhook.service.js'; import { outboxService } from './services/outbox.service.js'; import { storage } from './storage/index.js'; +import { roundTableService } from './services/round-table.service.js'; let createMechStorage = null; try { @@ -4359,3 +4360,115 @@ test('round table: missing required fields return 400', async () => { .send({ topic: 'Missing participants', goal: 'Test', participants: [] }); assert.equal(noParticipants.status, 400); }); + +test('round table: facilitator can speak in their own session', async () => { + const facilitator = await registerAgent('rt-fac-speak'); + const participant = await registerAgent('rt-fac-speak-p'); + + const createRes = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', facilitator.agent_id) + .send({ + topic: 'Facilitator speech test', + goal: 'Verify facilitator can speak', + participants: [participant.agent_id], + timeout_minutes: 30 + }); + + assert.equal(createRes.status, 201); + const rtId = createRes.body.id; + + const speakRes = await request(app) + .post(`/api/round-tables/${rtId}/speak`) + .set('X-Agent-ID', facilitator.agent_id) + .send({ message: 'I am the facilitator and I can speak.' }); + + assert.equal(speakRes.status, 201); + assert.equal(speakRes.body.thread_length, 1); +}); + +test('round table: expireStale marks session expired and notifies participants', async () => { + const facilitator = await registerAgent('rt-expire-fac'); + const participant = await registerAgent('rt-expire-p'); + + const createRes = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', facilitator.agent_id) + .send({ + topic: 'Expiry test', + goal: 'Verify expiry', + participants: [participant.agent_id], + timeout_minutes: 60 + }); + + assert.equal(createRes.status, 201); + const rtId = createRes.body.id; + + // Backdate the expiry to force expiration + await storage.updateRoundTable(rtId, { + expires_at: new Date(Date.now() - 1000).toISOString() + }); + + const expired = await roundTableService.expireStale(); + assert.ok(expired >= 1, 'at least one session should be expired'); + + // Confirm status is now expired via storage + const rt = await storage.getRoundTable(rtId); + assert.equal(rt.status, 'expired'); + + // Confirm speak returns 409 + const lateSpeak = await request(app) + .post(`/api/round-tables/${rtId}/speak`) + .set('X-Agent-ID', participant.agent_id) + .send({ message: 'Too late.' }); + assert.equal(lateSpeak.status, 409); +}); + +test('round table: duplicate participants are deduplicated', async () => { + const facilitator = await registerAgent('rt-dedup-fac'); + const participant = await registerAgent('rt-dedup-p'); + + const createRes = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', facilitator.agent_id) + .send({ + topic: 'Dedup test', + goal: 'Verify dedup', + participants: [participant.agent_id, participant.agent_id, participant.agent_id], + timeout_minutes: 30 + }); + + assert.equal(createRes.status, 201); + assert.equal(createRes.body.participants.length, 1, 'duplicates should be removed'); + assert.equal(createRes.body.participants[0], participant.agent_id); +}); + +test('round table: missing goal returns 400', async () => { + const agent = await registerAgent('rt-no-goal'); + + const res = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', agent.agent_id) + .send({ topic: 'No goal here', participants: ['other-agent'] }); + + assert.equal(res.status, 400); + assert.ok(res.body.error === 'INVALID_GOAL'); +}); + +test('round table: non-integer timeout_minutes returns 400', async () => { + const agent = await registerAgent('rt-float-timeout'); + const participant = await registerAgent('rt-float-timeout-p'); + + const res = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', agent.agent_id) + .send({ + topic: 'Float timeout', + goal: 'Test integer validation', + participants: [participant.agent_id], + timeout_minutes: 1.5 + }); + + assert.equal(res.status, 400); + assert.ok(res.body.error === 'INVALID_TIMEOUT'); +}); diff --git a/src/services/round-table.service.js b/src/services/round-table.service.js index 88c960a..a3e730a 100644 --- a/src/services/round-table.service.js +++ b/src/services/round-table.service.js @@ -11,25 +11,33 @@ import { inboxService } from './inbox.service.js'; const logger = pino({ level: process.env.NODE_ENV === 'production' ? 'info' : 'debug' }); +function makeError(message, statusCode) { + const err = new Error(message); + err.statusCode = statusCode; + return err; +} + export class RoundTableService { /** * Create a new Round Table session. * Automatically creates an ADMP group for multicast routing and sends * a work_order to each participant inviting them to join. + * Only participants successfully enrolled in the backing group are stored, + * preventing a split-brain between rt.participants and group membership. */ async create({ topic, goal, facilitator, participants, timeout_minutes = 30 }) { if (!topic || !goal || !facilitator) { - throw new Error('topic, goal, and facilitator are required'); + throw makeError('topic, goal, and facilitator are required', 400); } if (!Array.isArray(participants) || participants.length === 0) { - throw new Error('participants must be a non-empty array'); + throw makeError('participants must be a non-empty array', 400); } - if (timeout_minutes < 1 || timeout_minutes > 10080) { - throw new Error('timeout_minutes must be between 1 and 10080 (7 days)'); + if (!Number.isInteger(timeout_minutes) || timeout_minutes < 1 || timeout_minutes > 10080) { + throw makeError('timeout_minutes must be an integer between 1 and 10080 (7 days)', 400); } const uniqueParticipants = [...new Set(participants)]; if (uniqueParticipants.length > 20) { - throw new Error('Round Table supports at most 20 participants'); + throw makeError('Round Table supports at most 20 participants', 400); } const id = `rt_${uuid().replace(/-/g, '').slice(0, 12)}`; @@ -45,22 +53,30 @@ export class RoundTableService { settings: { max_members: uniqueParticipants.length + 1, message_ttl_sec: timeout_minutes * 60 } }); - // Add all participants to the group + // Add participants to the group — only those successfully enrolled are stored. + // A participant that doesn't exist yet cannot be enrolled and is excluded rather + // than silently included in rt.participants (split-brain prevention). + const enrolledParticipants = []; for (const participantId of uniqueParticipants) { try { await groupService.addMember(group.id, facilitator, participantId, 'member'); + enrolledParticipants.push(participantId); } catch (err) { - // Log but don't fail creation if a participant doesn't exist yet - logger.warn({ participantId, err: err.message }, '[RoundTable] Could not add participant'); + logger.warn({ participantId, err: err.message }, '[RoundTable] Could not enroll participant — excluded from session'); } } + if (enrolledParticipants.length === 0) { + try { await groupService.delete(group.id, facilitator); } catch (_) {} + throw makeError('No participants could be enrolled; round table not created', 400); + } + const rt = { id, topic, goal, facilitator, - participants: uniqueParticipants, + participants: enrolledParticipants, group_id: group.id, status: 'open', thread: [], @@ -71,8 +87,8 @@ export class RoundTableService { await storage.createRoundTable(rt); - // Notify each participant with a work_order via ADMP inbox - for (const participantId of uniqueParticipants) { + // Notify each enrolled participant with a work_order via ADMP inbox + for (const participantId of enrolledParticipants) { try { await inboxService.send({ from: facilitator, @@ -84,7 +100,7 @@ export class RoundTableService { topic, goal, facilitator, - participants: uniqueParticipants, + participants: enrolledParticipants, expires_at, instructions: `You have been invited to a Round Table deliberation session. POST to /api/round-tables/${id}/speak with {"message":"..."} to contribute. The facilitator will resolve with an outcome when consensus is reached.` }, @@ -107,7 +123,7 @@ export class RoundTableService { this._requireParticipant(rt, from); if (rt.thread.length >= 200) { - throw new Error('Round Table thread has reached the maximum of 200 entries'); + throw makeError('Round Table thread has reached the maximum of 200 entries', 409); } const entry = { @@ -119,7 +135,7 @@ export class RoundTableService { const thread = [...rt.thread, entry]; const updated = await storage.updateRoundTable(id, { thread }); - if (!updated) throw new Error(`Round table ${id} not found`); + if (!updated) throw makeError(`Round table ${id} not found`, 404); // Multicast to all participants via the backing group try { @@ -141,7 +157,7 @@ export class RoundTableService { */ async get(id, requesterId) { const rt = await storage.getRoundTable(id); - if (!rt) throw new Error(`Round table ${id} not found`); + if (!rt) throw makeError(`Round table ${id} not found`, 404); this._requireParticipant(rt, requesterId); return rt; } @@ -154,10 +170,10 @@ export class RoundTableService { const rt = await this._getOpen(id); if (rt.facilitator !== facilitator) { - throw new Error('Only the facilitator can resolve a Round Table'); + throw makeError('Only the facilitator can resolve a Round Table', 403); } if (!outcome) { - throw new Error('outcome is required to resolve'); + throw makeError('outcome is required to resolve', 400); } const now = new Date().toISOString(); @@ -167,7 +183,7 @@ export class RoundTableService { decision: decision || 'approved', resolved_at: now }); - if (!updated) throw new Error(`Round table ${id} not found`); + if (!updated) throw makeError(`Round table ${id} not found`, 404); // Multicast resolution to all participants try { @@ -205,6 +221,7 @@ export class RoundTableService { /** * Mark expired Round Tables (called by cleanup loop). + * Notifies all participants of expiry and cleans up backing groups. */ async expireStale() { const tables = await storage.listRoundTables({ status: 'open' }); @@ -214,6 +231,24 @@ export class RoundTableService { for (const rt of tables) { if (rt.expires_at && new Date(rt.expires_at).getTime() < now) { await storage.updateRoundTable(rt.id, { status: 'expired' }); + + // Notify participants of expiry + const expiredAt = new Date().toISOString(); + for (const participantId of rt.participants) { + try { + await inboxService.send({ + from: rt.facilitator, + to: participantId, + type: 'notification', + subject: `Round Table expired: ${rt.topic}`, + body: { round_table_id: rt.id, topic: rt.topic, reason: 'timeout', expires_at: rt.expires_at }, + timestamp: expiredAt + }, { verify_signature: false }); + } catch (err) { + logger.warn({ participantId, err: err.message }, '[RoundTable] Could not notify participant of expiry'); + } + } + // Clean up backing group try { await groupService.delete(rt.group_id, rt.facilitator); @@ -227,19 +262,27 @@ export class RoundTableService { return expired; } + /** + * Purge resolved/expired Round Tables older than olderThanMs (default: 7 days). + * Called by the cleanup loop to prevent unbounded storage growth. + */ + async purgeStale(olderThanMs = 7 * 24 * 60 * 60 * 1000) { + return await storage.purgeStaleRoundTables(olderThanMs); + } + // ---- internal helpers ---- async _getOpen(id) { const rt = await storage.getRoundTable(id); - if (!rt) throw new Error(`Round table ${id} not found`); - if (rt.status === 'resolved') throw new Error('Round table is already resolved'); - if (rt.status === 'expired') throw new Error('Round table has expired'); + if (!rt) throw makeError(`Round table ${id} not found`, 404); + if (rt.status === 'resolved') throw makeError('Round table is already resolved', 409); + if (rt.status === 'expired') throw makeError('Round table has expired', 409); return rt; } _requireParticipant(rt, agentId) { if (rt.facilitator !== agentId && !(rt.participants || []).includes(agentId)) { - throw new Error('Not a participant of this Round Table'); + throw makeError('Not a participant of this Round Table', 403); } } } diff --git a/src/storage/memory.js b/src/storage/memory.js index fad288d..11c23c2 100644 --- a/src/storage/memory.js +++ b/src/storage/memory.js @@ -611,6 +611,21 @@ export class MemoryStorage { } return tables; } + + async purgeStaleRoundTables(olderThanMs) { + const cutoff = Date.now() - olderThanMs; + let purged = 0; + for (const [id, rt] of this.roundTables.entries()) { + if (rt.status === 'resolved' || rt.status === 'expired') { + const closedAt = rt.resolved_at || rt.expires_at; + if (closedAt && new Date(closedAt).getTime() < cutoff) { + this.roundTables.delete(id); + purged++; + } + } + } + return purged; + } } // Singleton instance From 6b1a4bd7781e998cde36b5221c3795f6b0e03b32 Mon Sep 17 00:00:00 2001 From: dundas Date: Sat, 28 Feb 2026 13:57:25 -0600 Subject: [PATCH 2/6] =?UTF-8?q?fix(round-tables):=20address=20third=20revi?= =?UTF-8?q?ew=20=E2=80=94=20facilitator=20expiry=20notify,=20resilient=20e?= =?UTF-8?q?xpiry=20loop,=20excluded=5Fparticipants,=20max=5Fmembers=20fix?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Notify facilitator AND participants on expiry (facilitator was excluded) - Add version:1.0 + id to all inboxService.send envelopes (required fields) - Wrap per-record expireStale in try-catch so one failure does not abort rest - Return excluded_participants in create response so callers know who was dropped - Align group max_members to enrolledParticipants.length+1 after partial enrollment via groupService.update() (prevents unauthorized join slots) - Fix parseInt(env) || default -> NaN-safe pattern for ROUND_TABLE_PURGE_TTL_MS - Use rt.expires_at (not new Date()) as canonical timestamp in expiry notifications - Remove unnecessary return await in purgeStale - Strengthen expiry test: assert inbox delivery to both participant and facilitator - Add zero-enrollment test: all-ghost participants returns 400 Co-Authored-By: Claude Sonnet 4.6 --- src/server.js | 3 +- src/server.test.js | 42 ++++++++++++++++++++++++- src/services/round-table.service.js | 49 ++++++++++++++++++++++------- 3 files changed, 81 insertions(+), 13 deletions(-) diff --git a/src/server.js b/src/server.js index 86f142b..b2f3f0c 100644 --- a/src/server.js +++ b/src/server.js @@ -36,7 +36,8 @@ config(); const PORT = process.env.PORT || 8080; const CLEANUP_INTERVAL_MS = parseInt(process.env.CLEANUP_INTERVAL_MS) || 60000; -const ROUND_TABLE_PURGE_TTL_MS = parseInt(process.env.ROUND_TABLE_PURGE_TTL_MS) || 7 * 24 * 60 * 60 * 1000; +const _purgeTtlParsed = parseInt(process.env.ROUND_TABLE_PURGE_TTL_MS); +const ROUND_TABLE_PURGE_TTL_MS = Number.isNaN(_purgeTtlParsed) ? 7 * 24 * 60 * 60 * 1000 : _purgeTtlParsed; // Warn about insecure outbox webhook configuration if (process.env.MAILGUN_API_KEY && !process.env.MAILGUN_WEBHOOK_SIGNING_KEY) { diff --git a/src/server.test.js b/src/server.test.js index 0698ffb..2cd17ee 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -4387,7 +4387,7 @@ test('round table: facilitator can speak in their own session', async () => { assert.equal(speakRes.body.thread_length, 1); }); -test('round table: expireStale marks session expired and notifies participants', async () => { +test('round table: expireStale marks session expired and notifies facilitator and participants', async () => { const facilitator = await registerAgent('rt-expire-fac'); const participant = await registerAgent('rt-expire-p'); @@ -4404,6 +4404,11 @@ test('round table: expireStale marks session expired and notifies participants', assert.equal(createRes.status, 201); const rtId = createRes.body.id; + // Drain the work_order invitation from participant inbox before backdating + await request(app) + .post(`/api/agents/${encodeURIComponent(participant.agent_id)}/inbox/pull`) + .set('X-Agent-ID', participant.agent_id); + // Backdate the expiry to force expiration await storage.updateRoundTable(rtId, { expires_at: new Date(Date.now() - 1000).toISOString() @@ -4416,6 +4421,23 @@ test('round table: expireStale marks session expired and notifies participants', const rt = await storage.getRoundTable(rtId); assert.equal(rt.status, 'expired'); + // Participant inbox should have an expiry notification + const participantPull = await request(app) + .post(`/api/agents/${encodeURIComponent(participant.agent_id)}/inbox/pull`) + .set('X-Agent-ID', participant.agent_id); + assert.equal(participantPull.status, 200); + assert.equal(participantPull.body.envelope.type, 'notification'); + assert.equal(participantPull.body.envelope.body.round_table_id, rtId); + assert.equal(participantPull.body.envelope.body.reason, 'timeout'); + + // Facilitator inbox should also have an expiry notification + const facilitatorPull = await request(app) + .post(`/api/agents/${encodeURIComponent(facilitator.agent_id)}/inbox/pull`) + .set('X-Agent-ID', facilitator.agent_id); + assert.equal(facilitatorPull.status, 200); + assert.equal(facilitatorPull.body.envelope.type, 'notification'); + assert.equal(facilitatorPull.body.envelope.body.round_table_id, rtId); + // Confirm speak returns 409 const lateSpeak = await request(app) .post(`/api/round-tables/${rtId}/speak`) @@ -4472,3 +4494,21 @@ test('round table: non-integer timeout_minutes returns 400', async () => { assert.equal(res.status, 400); assert.ok(res.body.error === 'INVALID_TIMEOUT'); }); + +test('round table: zero-enrollment returns 400 and leaves no orphaned records', async () => { + const facilitator = await registerAgent('rt-zero-enroll-fac'); + + // All fake participants — addMember will throw "Agent not found" for each + const res = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', facilitator.agent_id) + .send({ + topic: 'Zero enrollment test', + goal: 'All participants unknown', + participants: ['ghost-agent-1', 'ghost-agent-2'], + timeout_minutes: 30 + }); + + assert.equal(res.status, 400); + assert.ok(res.body.message.toLowerCase().includes('no participants')); +}); diff --git a/src/services/round-table.service.js b/src/services/round-table.service.js index a3e730a..c9f7e6e 100644 --- a/src/services/round-table.service.js +++ b/src/services/round-table.service.js @@ -24,6 +24,7 @@ export class RoundTableService { * a work_order to each participant inviting them to join. * Only participants successfully enrolled in the backing group are stored, * preventing a split-brain between rt.participants and group membership. + * Returns `excluded_participants` so callers know who was dropped. */ async create({ topic, goal, facilitator, participants, timeout_minutes = 30 }) { if (!topic || !goal || !facilitator) { @@ -44,7 +45,8 @@ export class RoundTableService { const now = new Date(); const expires_at = new Date(now.getTime() + timeout_minutes * 60 * 1000).toISOString(); - // Create backing ADMP group (invite-only, facilitator is owner) + // Create backing ADMP group (invite-only, facilitator is owner). + // max_members is initially the upper bound; adjusted after enrollment below. const groupName = `round-table-${id}`; const group = await groupService.create({ name: groupName, @@ -57,12 +59,14 @@ export class RoundTableService { // A participant that doesn't exist yet cannot be enrolled and is excluded rather // than silently included in rt.participants (split-brain prevention). const enrolledParticipants = []; + const excludedParticipants = []; for (const participantId of uniqueParticipants) { try { await groupService.addMember(group.id, facilitator, participantId, 'member'); enrolledParticipants.push(participantId); } catch (err) { logger.warn({ participantId, err: err.message }, '[RoundTable] Could not enroll participant — excluded from session'); + excludedParticipants.push(participantId); } } @@ -71,6 +75,17 @@ export class RoundTableService { throw makeError('No participants could be enrolled; round table not created', 400); } + // Align max_members to actual enrolled count + facilitator + if (enrolledParticipants.length < uniqueParticipants.length) { + try { + await groupService.update(group.id, facilitator, { + settings: { max_members: enrolledParticipants.length + 1, message_ttl_sec: timeout_minutes * 60 } + }); + } catch (err) { + logger.warn({ groupId: group.id, err: err.message }, '[RoundTable] Could not update group max_members after partial enrollment'); + } + } + const rt = { id, topic, @@ -91,6 +106,8 @@ export class RoundTableService { for (const participantId of enrolledParticipants) { try { await inboxService.send({ + version: '1.0', + id: uuid(), from: facilitator, to: participantId, type: 'work_order', @@ -111,7 +128,8 @@ export class RoundTableService { } } - return rt; + // Include excluded participants in the response so callers know who was dropped + return { ...rt, excluded_participants: excludedParticipants }; } /** @@ -221,7 +239,8 @@ export class RoundTableService { /** * Mark expired Round Tables (called by cleanup loop). - * Notifies all participants of expiry and cleans up backing groups. + * Notifies facilitator and all participants of expiry, cleans up backing groups. + * Each record is processed independently — a failure on one does not abort the rest. */ async expireStale() { const tables = await storage.listRoundTables({ status: 'open' }); @@ -229,23 +248,28 @@ export class RoundTableService { let expired = 0; for (const rt of tables) { - if (rt.expires_at && new Date(rt.expires_at).getTime() < now) { + if (!rt.expires_at || new Date(rt.expires_at).getTime() >= now) continue; + + try { await storage.updateRoundTable(rt.id, { status: 'expired' }); - // Notify participants of expiry - const expiredAt = new Date().toISOString(); - for (const participantId of rt.participants) { + // Notify facilitator and all participants of expiry. + // Use rt.expires_at as the canonical close timestamp. + const toNotify = [rt.facilitator, ...rt.participants]; + for (const recipientId of toNotify) { try { await inboxService.send({ + version: '1.0', + id: uuid(), from: rt.facilitator, - to: participantId, + to: recipientId, type: 'notification', subject: `Round Table expired: ${rt.topic}`, body: { round_table_id: rt.id, topic: rt.topic, reason: 'timeout', expires_at: rt.expires_at }, - timestamp: expiredAt + timestamp: rt.expires_at }, { verify_signature: false }); } catch (err) { - logger.warn({ participantId, err: err.message }, '[RoundTable] Could not notify participant of expiry'); + logger.warn({ recipientId, err: err.message }, '[RoundTable] Could not notify recipient of expiry'); } } @@ -255,7 +279,10 @@ export class RoundTableService { } catch (err) { logger.warn({ groupId: rt.group_id, err: err.message }, '[RoundTable] Group cleanup on expiry failed'); } + expired++; + } catch (err) { + logger.warn({ id: rt.id, err: err.message }, '[RoundTable] Failed to expire record — will retry next cycle'); } } @@ -267,7 +294,7 @@ export class RoundTableService { * Called by the cleanup loop to prevent unbounded storage growth. */ async purgeStale(olderThanMs = 7 * 24 * 60 * 60 * 1000) { - return await storage.purgeStaleRoundTables(olderThanMs); + return storage.purgeStaleRoundTables(olderThanMs); } // ---- internal helpers ---- From 0d05ecb7a4c2a22f56c8b5dd69faae9c07ef5af5 Mon Sep 17 00:00:00 2001 From: dundas Date: Sat, 28 Feb 2026 14:10:28 -0600 Subject: [PATCH 3/6] =?UTF-8?q?fix(round-tables):=20address=20fourth=20rev?= =?UTF-8?q?iew=20=E2=80=94=20storage=20contract,=20zero-enroll=20log,=20te?= =?UTF-8?q?st=20robustness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add startup assertion for purgeStaleRoundTables in storage/index.js so any backend missing the method fails fast at startup instead of at runtime - Log groupService.delete failure on zero-enrollment path (was silently swallowed; inconsistent with every other error path in the service) - Fix zero-enrollment test: use groupService.listForAgent for backend-agnostic group cleanup assertion (storage.groups Map only exists in memory backend) - Use unique ghost agent IDs in zero-enrollment test to prevent collision with agents registered by other tests - Import groupService in test file for direct service-level assertions Co-Authored-By: Claude Sonnet 4.6 --- src/server.test.js | 13 +++++++++++-- src/services/round-table.service.js | 6 +++++- src/storage/index.js | 7 +++++-- 3 files changed, 21 insertions(+), 5 deletions(-) diff --git a/src/server.test.js b/src/server.test.js index 2cd17ee..483c88a 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -12,6 +12,7 @@ import { webhookService } from './services/webhook.service.js'; import { outboxService } from './services/outbox.service.js'; import { storage } from './storage/index.js'; import { roundTableService } from './services/round-table.service.js'; +import { groupService } from './services/group.service.js'; let createMechStorage = null; try { @@ -4495,8 +4496,12 @@ test('round table: non-integer timeout_minutes returns 400', async () => { assert.ok(res.body.error === 'INVALID_TIMEOUT'); }); -test('round table: zero-enrollment returns 400 and leaves no orphaned records', async () => { +test('round table: zero-enrollment returns 400 and leaves no orphaned groups', async () => { const facilitator = await registerAgent('rt-zero-enroll-fac'); + const unique = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; + + // Count groups the facilitator belongs to before the attempt + const groupsBefore = (await groupService.listForAgent(facilitator.agent_id)).length; // All fake participants — addMember will throw "Agent not found" for each const res = await request(app) @@ -4505,10 +4510,14 @@ test('round table: zero-enrollment returns 400 and leaves no orphaned records', .send({ topic: 'Zero enrollment test', goal: 'All participants unknown', - participants: ['ghost-agent-1', 'ghost-agent-2'], + participants: [`ghost-${unique}-1`, `ghost-${unique}-2`], timeout_minutes: 30 }); assert.equal(res.status, 400); assert.ok(res.body.message.toLowerCase().includes('no participants')); + + // No orphaned round-table groups should remain + const groupsAfter = (await groupService.listForAgent(facilitator.agent_id)).length; + assert.equal(groupsAfter, groupsBefore, 'group created during enrollment should be cleaned up'); }); diff --git a/src/services/round-table.service.js b/src/services/round-table.service.js index c9f7e6e..43c3ce3 100644 --- a/src/services/round-table.service.js +++ b/src/services/round-table.service.js @@ -71,7 +71,11 @@ export class RoundTableService { } if (enrolledParticipants.length === 0) { - try { await groupService.delete(group.id, facilitator); } catch (_) {} + try { + await groupService.delete(group.id, facilitator); + } catch (err) { + logger.warn({ groupId: group.id, err: err.message }, '[RoundTable] Could not clean up group on zero-enrollment'); + } throw makeError('No participants could be enrolled; round table not created', 400); } diff --git a/src/storage/index.js b/src/storage/index.js index 4eae9b9..29b1ee3 100644 --- a/src/storage/index.js +++ b/src/storage/index.js @@ -62,11 +62,14 @@ switch (backend) { // re-validated here — this guard only fires on new writes via createAgent(). const STORAGE_AGENT_ID_RE = /^[a-zA-Z0-9._:/-]+$/; -// Startup assertion: if the storage interface renames createAgent, the Proxy guard -// silently becomes a no-op. Crashing at startup is better than a silent bypass. +// Startup assertions: if the storage interface renames required methods, crash at +// startup rather than silently failing at runtime. if (typeof _storage.createAgent !== 'function') { throw new Error('storage: createAgent is missing — update the Proxy guard in storage/index.js'); } +if (typeof _storage.purgeStaleRoundTables !== 'function') { + throw new Error('storage: purgeStaleRoundTables is missing — implement it in the storage adapter (see memory.js for reference)'); +} const storage = new Proxy(_storage, { get(target, prop) { From 2b8755100eae70f032063a1cb93c2c1fb81c7825 Mon Sep 17 00:00:00 2001 From: dundas Date: Sat, 28 Feb 2026 14:30:22 -0600 Subject: [PATCH 4/6] fix(round-tables): partial enrollment test, excluded_participants shape, encodeURIComponent in test - Add partial enrollment test: verifies split-brain prevention is exercised (one valid + one ghost participant; asserts participants=[valid], excluded_participants=[ghost], group max_members=2) - Only include excluded_participants in create response when non-empty, keeping create and GET response shapes consistent on the happy path - Fix group ID URL encoding in test (group:// IDs contain slashes) Co-Authored-By: Claude Sonnet 4.6 --- src/server.test.js | 34 +++++++++++++++++++++++++++++ src/services/round-table.service.js | 8 +++++-- 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/src/server.test.js b/src/server.test.js index 483c88a..d6fa4e0 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -4521,3 +4521,37 @@ test('round table: zero-enrollment returns 400 and leaves no orphaned groups', a const groupsAfter = (await groupService.listForAgent(facilitator.agent_id)).length; assert.equal(groupsAfter, groupsBefore, 'group created during enrollment should be cleaned up'); }); + +test('round table: partial enrollment — only enrolled participants stored, excluded_participants returned', async () => { + const facilitator = await registerAgent('rt-partial-fac'); + const validParticipant = await registerAgent('rt-partial-p'); + const unique = `${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; + + const createRes = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', facilitator.agent_id) + .send({ + topic: 'Partial enrollment test', + goal: 'Verify split-brain prevention', + participants: [validParticipant.agent_id, `ghost-${unique}`], + timeout_minutes: 30 + }); + + assert.equal(createRes.status, 201); + + // Only the valid participant should be in participants + assert.equal(createRes.body.participants.length, 1); + assert.equal(createRes.body.participants[0], validParticipant.agent_id); + + // The ghost agent should be in excluded_participants + assert.ok(Array.isArray(createRes.body.excluded_participants)); + assert.equal(createRes.body.excluded_participants.length, 1); + assert.ok(createRes.body.excluded_participants[0].startsWith('ghost-')); + + // The backing group's max_members should be aligned to enrolled count + 1 (= 2) + const groupRes = await request(app) + .get(`/api/groups/${encodeURIComponent(createRes.body.group_id)}`) + .set('X-Agent-ID', facilitator.agent_id); + assert.equal(groupRes.status, 200); + assert.equal(groupRes.body.settings.max_members, 2); +}); diff --git a/src/services/round-table.service.js b/src/services/round-table.service.js index 43c3ce3..3abcf7d 100644 --- a/src/services/round-table.service.js +++ b/src/services/round-table.service.js @@ -132,8 +132,12 @@ export class RoundTableService { } } - // Include excluded participants in the response so callers know who was dropped - return { ...rt, excluded_participants: excludedParticipants }; + // Include excluded_participants when non-empty so callers know who was dropped. + // Omit the field on the happy path to keep the create and GET response shapes consistent. + if (excludedParticipants.length > 0) { + return { ...rt, excluded_participants: excludedParticipants }; + } + return rt; } /** From 838d09915f09c1ff46a81242c6f30108c426192f Mon Sep 17 00:00:00 2001 From: dundas Date: Sat, 28 Feb 2026 14:44:38 -0600 Subject: [PATCH 5/6] fix(round-tables): parallelize expiry notifications, add comments, docs update - Replace sequential notification loop in expireStale with Promise.allSettled for parallel delivery to all recipients (up to 21 for a full session) - Add inline comments explaining: - why from=rt.facilitator is used for expiry notifications (logical author) - why facilitator receives a self-addressed copy (documented intent) - why verify_signature:false is required (server holds no private keys) - why rt.expires_at is used as timestamp (canonical, not processing time) - Update docs/AGENT-GUIDE.md: add Round Tables endpoint table and behavior notes documenting excluded_participants, expiry notifications, partial enrollment, storage growth/purge, and ROUND_TABLE_PURGE_TTL_MS env var Co-Authored-By: Claude Sonnet 4.6 --- docs/AGENT-GUIDE.md | 16 ++++++++++++++++ src/services/round-table.service.js | 13 +++++++++---- 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/docs/AGENT-GUIDE.md b/docs/AGENT-GUIDE.md index adfd162..86a28b7 100644 --- a/docs/AGENT-GUIDE.md +++ b/docs/AGENT-GUIDE.md @@ -347,6 +347,22 @@ See [docs/API-REFERENCE.md](./API-REFERENCE.md) for the complete endpoint docume | `POST /api/agents/:id/messages/:msgId/reply` | HTTP Sig (self only) | Reply | | `GET /api/messages/:msgId/status` | API Key | Delivery status | +### Round Tables (ephemeral multi-agent deliberation) +| Endpoint | Auth | Description | +|----------|------|-------------| +| `POST /api/round-tables` | Agent ID | Create session | +| `GET /api/round-tables` | Agent ID | List my sessions | +| `GET /api/round-tables/:id` | Agent ID (participants only) | Get session | +| `POST /api/round-tables/:id/speak` | Agent ID (participants only) | Add message | +| `POST /api/round-tables/:id/resolve` | Agent ID (facilitator only) | Close session | + +**Round Table behavior notes:** + +- **`excluded_participants`** — When some requested participants cannot be enrolled in the backing ADMP group (e.g. unregistered agent IDs), the create response includes an `excluded_participants` array listing those that were dropped. This field is only present when at least one participant was excluded; it is absent in the happy path. The stored session record does not carry this field — it is returned at create time only. +- **Expiry notifications** — When a session times out, the server automatically sends a `notification` message (type `notification`, body `{ reason: "timeout" }`) to the facilitator and all participants. The notification envelope has `from` set to the facilitator's agent ID (the logical author of the session). Facilitators receive a self-addressed copy; agents should not rely on `from === self` as a filter to suppress expiry notifications. +- **Partial enrollment** — If only some participants enroll successfully, the session is created with the enrolled subset. `rt.participants` and the backing group membership are kept in sync. The group's `max_members` is updated to reflect the actual enrolled count after partial enrollment. +- **Storage growth** — Resolved and expired sessions accumulate in storage. The server periodically purges records older than `ROUND_TABLE_PURGE_TTL_MS` (default: 7 days). Set this env var to control retention. Custom storage adapters must implement `purgeStaleRoundTables(olderThanMs)`. + --- ## 7. Error Handling diff --git a/src/services/round-table.service.js b/src/services/round-table.service.js index 3abcf7d..1475738 100644 --- a/src/services/round-table.service.js +++ b/src/services/round-table.service.js @@ -261,10 +261,15 @@ export class RoundTableService { try { await storage.updateRoundTable(rt.id, { status: 'expired' }); - // Notify facilitator and all participants of expiry. - // Use rt.expires_at as the canonical close timestamp. + // Notify facilitator and all participants of expiry in parallel. + // Use rt.expires_at as the canonical close timestamp (not processing time). + // Notifications are sent from the facilitator's identity since they created + // the session. The facilitator will receive a self-addressed notification — + // this is intentional and preserves a single consistent sender for all recipients. + // verify_signature: false is required because the server does not hold private + // keys on behalf of agents; signature is omitted and delivery is trusted internally. const toNotify = [rt.facilitator, ...rt.participants]; - for (const recipientId of toNotify) { + await Promise.allSettled(toNotify.map(async (recipientId) => { try { await inboxService.send({ version: '1.0', @@ -279,7 +284,7 @@ export class RoundTableService { } catch (err) { logger.warn({ recipientId, err: err.message }, '[RoundTable] Could not notify recipient of expiry'); } - } + })); // Clean up backing group try { From 2327fbd1d8332f33251c2a353195fe879747b3e1 Mon Sep 17 00:00:00 2001 From: dundas Date: Sat, 28 Feb 2026 15:02:36 -0600 Subject: [PATCH 6/6] fix(round-tables): facilitator-in-participants guard, IIFE for purge TTL, rolling deploy note - Add FACILITATOR_IN_PARTICIPANTS validation: reject creates where the facilitator is listed as a participant (addMember would throw 'already owner', landing the facilitator in excluded_participants or aborting on zero-enrollment path) - Replace _purgeTtlParsed temp var with IIFE to avoid underscore-private convention - Add BREAKING CHANGE comment to storage/index.js purgeStaleRoundTables assertion so operators deploying custom adapters see upgrade instructions at the assertion site - Add canonical-close-time comment to memory.js purgeStaleRoundTables explaining why resolved_at/expires_at is used instead of Date.now() - Strengthen partial enrollment test: assert enrolled agent can speak (201) and a non-enrolled registered agent gets 403 (not just unregistered ghost) - Add facilitator-in-participants test: verify 400 + FACILITATOR_IN_PARTICIPANTS error Co-Authored-By: Claude Sonnet 4.6 --- src/routes/round-tables.js | 3 +++ src/server.js | 6 ++++-- src/server.test.js | 34 ++++++++++++++++++++++++++++++++++ src/storage/index.js | 3 +++ src/storage/memory.js | 3 +++ 5 files changed, 47 insertions(+), 2 deletions(-) diff --git a/src/routes/round-tables.js b/src/routes/round-tables.js index 10b9479..41b5e93 100644 --- a/src/routes/round-tables.js +++ b/src/routes/round-tables.js @@ -43,6 +43,9 @@ router.post('/', authenticateAgent, async (req, res) => { if (timeout_minutes !== undefined && (typeof timeout_minutes !== 'number' || !Number.isFinite(timeout_minutes) || !Number.isInteger(timeout_minutes))) { return res.status(400).json({ error: 'INVALID_TIMEOUT', message: 'timeout_minutes must be an integer' }); } + if (participants.includes(req.agent.agent_id)) { + return res.status(400).json({ error: 'FACILITATOR_IN_PARTICIPANTS', message: 'facilitator cannot be listed as a participant' }); + } const rt = await roundTableService.create({ topic: topic.trim(), diff --git a/src/server.js b/src/server.js index b2f3f0c..344451f 100644 --- a/src/server.js +++ b/src/server.js @@ -36,8 +36,10 @@ config(); const PORT = process.env.PORT || 8080; const CLEANUP_INTERVAL_MS = parseInt(process.env.CLEANUP_INTERVAL_MS) || 60000; -const _purgeTtlParsed = parseInt(process.env.ROUND_TABLE_PURGE_TTL_MS); -const ROUND_TABLE_PURGE_TTL_MS = Number.isNaN(_purgeTtlParsed) ? 7 * 24 * 60 * 60 * 1000 : _purgeTtlParsed; +const ROUND_TABLE_PURGE_TTL_MS = (() => { + const parsed = parseInt(process.env.ROUND_TABLE_PURGE_TTL_MS); + return Number.isNaN(parsed) ? 7 * 24 * 60 * 60 * 1000 : parsed; +})(); // Warn about insecure outbox webhook configuration if (process.env.MAILGUN_API_KEY && !process.env.MAILGUN_WEBHOOK_SIGNING_KEY) { diff --git a/src/server.test.js b/src/server.test.js index d6fa4e0..c38d239 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -4554,4 +4554,38 @@ test('round table: partial enrollment — only enrolled participants stored, exc .set('X-Agent-ID', facilitator.agent_id); assert.equal(groupRes.status, 200); assert.equal(groupRes.body.settings.max_members, 2); + + // Enrolled participant can speak + const rtId = createRes.body.id; + const speakRes = await request(app) + .post(`/api/round-tables/${rtId}/speak`) + .set('X-Agent-ID', validParticipant.agent_id) + .send({ message: 'I am enrolled and can speak.' }); + assert.equal(speakRes.status, 201); + + // A real registered agent that was not enrolled cannot speak (enforced by _requireParticipant) + const nonEnrolled = await registerAgent('rt-partial-non-enrolled'); + const nonEnrolledSpeakRes = await request(app) + .post(`/api/round-tables/${rtId}/speak`) + .set('X-Agent-ID', nonEnrolled.agent_id) + .send({ message: 'I was not enrolled and should not speak.' }); + assert.equal(nonEnrolledSpeakRes.status, 403); +}); + +test('round table: facilitator cannot be listed as a participant', async () => { + const facilitator = await registerAgent('rt-fac-as-participant'); + const otherParticipant = await registerAgent('rt-fac-as-participant-p'); + + const res = await request(app) + .post('/api/round-tables') + .set('X-Agent-ID', facilitator.agent_id) + .send({ + topic: 'Self-inclusion test', + goal: 'Verify facilitator cannot be a participant', + participants: [facilitator.agent_id, otherParticipant.agent_id], + timeout_minutes: 30 + }); + + assert.equal(res.status, 400); + assert.equal(res.body.error, 'FACILITATOR_IN_PARTICIPANTS'); }); diff --git a/src/storage/index.js b/src/storage/index.js index 29b1ee3..bc187fa 100644 --- a/src/storage/index.js +++ b/src/storage/index.js @@ -67,6 +67,9 @@ const STORAGE_AGENT_ID_RE = /^[a-zA-Z0-9._:/-]+$/; if (typeof _storage.createAgent !== 'function') { throw new Error('storage: createAgent is missing — update the Proxy guard in storage/index.js'); } +// BREAKING CHANGE for custom storage adapters: purgeStaleRoundTables(olderThanMs) is now +// required. On a rolling deployment, update your storage adapter before deploying new +// application instances to avoid startup failure. See memory.js for the reference implementation. if (typeof _storage.purgeStaleRoundTables !== 'function') { throw new Error('storage: purgeStaleRoundTables is missing — implement it in the storage adapter (see memory.js for reference)'); } diff --git a/src/storage/memory.js b/src/storage/memory.js index 11c23c2..a6f4c03 100644 --- a/src/storage/memory.js +++ b/src/storage/memory.js @@ -617,6 +617,9 @@ export class MemoryStorage { let purged = 0; for (const [id, rt] of this.roundTables.entries()) { if (rt.status === 'resolved' || rt.status === 'expired') { + // Use resolved_at/expires_at as the canonical close time, not processing time. + // Records are retained for the full TTL measured from when the session logically + // closed, even if the cleanup job runs late. const closedAt = rt.resolved_at || rt.expires_at; if (closedAt && new Date(closedAt).getTime() < cutoff) { this.roundTables.delete(id);