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/routes/round-tables.js b/src/routes/round-tables.js index 9733956..41b5e93 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,11 @@ 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' }); + } + 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({ @@ -53,7 +52,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..344451f 100644 --- a/src/server.js +++ b/src/server.js @@ -36,6 +36,10 @@ config(); const PORT = process.env.PORT || 8080; const CLEANUP_INTERVAL_MS = parseInt(process.env.CLEANUP_INTERVAL_MS) || 60000; +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) { @@ -221,14 +225,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..c38d239 100644 --- a/src/server.test.js +++ b/src/server.test.js @@ -11,6 +11,8 @@ 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'; +import { groupService } from './services/group.service.js'; let createMechStorage = null; try { @@ -4359,3 +4361,231 @@ 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 facilitator and 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; + + // 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() + }); + + 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'); + + // 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`) + .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'); +}); + +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) + .post('/api/round-tables') + .set('X-Agent-ID', facilitator.agent_id) + .send({ + topic: 'Zero enrollment test', + goal: 'All participants unknown', + 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'); +}); + +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); + + // 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/services/round-table.service.js b/src/services/round-table.service.js index 88c960a..1475738 100644 --- a/src/services/round-table.service.js +++ b/src/services/round-table.service.js @@ -11,32 +11,42 @@ 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. + * Returns `excluded_participants` so callers know who was dropped. */ 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)}`; 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, @@ -45,13 +55,38 @@ 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 = []; + 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); + } + } + + if (enrolledParticipants.length === 0) { + 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); + } + + // 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) { - // 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({ groupId: group.id, err: err.message }, '[RoundTable] Could not update group max_members after partial enrollment'); } } @@ -60,7 +95,7 @@ export class RoundTableService { topic, goal, facilitator, - participants: uniqueParticipants, + participants: enrolledParticipants, group_id: group.id, status: 'open', thread: [], @@ -71,10 +106,12 @@ 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({ + version: '1.0', + id: uuid(), from: facilitator, to: participantId, type: 'work_order', @@ -84,7 +121,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.` }, @@ -95,6 +132,11 @@ export class RoundTableService { } } + // 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; } @@ -107,7 +149,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 +161,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 +183,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 +196,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 +209,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 +247,8 @@ export class RoundTableService { /** * Mark expired Round Tables (called by cleanup loop). + * 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' }); @@ -212,34 +256,73 @@ 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 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]; + await Promise.allSettled(toNotify.map(async (recipientId) => { + try { + await inboxService.send({ + version: '1.0', + id: uuid(), + from: rt.facilitator, + 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: rt.expires_at + }, { verify_signature: false }); + } catch (err) { + logger.warn({ recipientId, err: err.message }, '[RoundTable] Could not notify recipient of expiry'); + } + })); + // Clean up backing group try { await groupService.delete(rt.group_id, rt.facilitator); } 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'); } } 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 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/index.js b/src/storage/index.js index 4eae9b9..bc187fa 100644 --- a/src/storage/index.js +++ b/src/storage/index.js @@ -62,11 +62,17 @@ 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'); } +// 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)'); +} const storage = new Proxy(_storage, { get(target, prop) { diff --git a/src/storage/memory.js b/src/storage/memory.js index fad288d..a6f4c03 100644 --- a/src/storage/memory.js +++ b/src/storage/memory.js @@ -611,6 +611,24 @@ 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') { + // 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); + purged++; + } + } + } + return purged; + } } // Singleton instance