diff --git a/.agentworkforce/trajectories/completed/2026-08/traj_pzkvekhsexoq/summary.md b/.agentworkforce/trajectories/completed/2026-08/traj_pzkvekhsexoq/summary.md new file mode 100644 index 00000000..899ca28e --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-08/traj_pzkvekhsexoq/summary.md @@ -0,0 +1,32 @@ +# Trajectory: Implement atomic ResolveOrReserve for deterministic 1:1 DMs + +> **Status:** ✅ Completed +> **Confidence:** 95% +> **Started:** August 2, 2026 at 08:11 AM +> **Completed:** August 2, 2026 at 08:26 AM + +--- + +## Summary + +Added atomic 1:1 DM ResolveOrReserve reservations with migration backfill, fail-closed digest collision handling, and concurrent negative-control coverage. + +**Approach:** Standard approach + +--- + +## Key Decisions + +### Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows +- **Chose:** Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows +- **Reasoning:** The reservation must name workspace plus sorted pair in one conflict-detecting upsert. Reserving first prevents colliding requests from creating shared metadata before a winner exists, and a separate table can backfill existing DMs without changing nullable group-DM columns. + +--- + +## Chapters + +### 1. Work +*Agent: default* + +- Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows: Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows +- The reservation-first design passes the real concurrent send path and its negative control. Full engine build, lint, and 536-test suite are green; root/package-wide unrelated runner/load failures were isolated and the SDK timeout rerun passed 416/416. diff --git a/.agentworkforce/trajectories/completed/2026-08/traj_pzkvekhsexoq/trajectory.json b/.agentworkforce/trajectories/completed/2026-08/traj_pzkvekhsexoq/trajectory.json new file mode 100644 index 00000000..b44bff0f --- /dev/null +++ b/.agentworkforce/trajectories/completed/2026-08/traj_pzkvekhsexoq/trajectory.json @@ -0,0 +1,73 @@ +{ + "id": "traj_pzkvekhsexoq", + "version": 1, + "task": { + "title": "Implement atomic ResolveOrReserve for deterministic 1:1 DMs" + }, + "status": "completed", + "startedAt": "2026-08-02T06:11:12.788Z", + "completedAt": "2026-08-02T06:26:48.981Z", + "agents": [ + { + "name": "default", + "role": "lead", + "joinedAt": "2026-08-02T06:13:21.452Z" + } + ], + "chapters": [ + { + "id": "chap_hdc34f62008i", + "title": "Work", + "agentName": "default", + "startedAt": "2026-08-02T06:13:21.452Z", + "endedAt": "2026-08-02T06:26:48.981Z", + "events": [ + { + "ts": 1785651201456, + "type": "decision", + "content": "Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows: Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows", + "raw": { + "question": "Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows", + "chosen": "Reserve deterministic 1:1 DM ids in a dedicated tuple table before creating channel or conversation rows", + "alternatives": [], + "reasoning": "The reservation must name workspace plus sorted pair in one conflict-detecting upsert. Reserving first prevents colliding requests from creating shared metadata before a winner exists, and a separate table can backfill existing DMs without changing nullable group-DM columns." + }, + "significance": "high" + }, + { + "ts": 1785652008647, + "type": "reflection", + "content": "The reservation-first design passes the real concurrent send path and its negative control. Full engine build, lint, and 536-test suite are green; root/package-wide unrelated runner/load failures were isolated and the SDK timeout rerun passed 416/416.", + "raw": { + "focalPoints": [ + "atomicity", + "negative-control", + "validation" + ], + "confidence": 0.95 + }, + "significance": "high", + "tags": [ + "focal:atomicity", + "focal:negative-control", + "focal:validation", + "confidence:0.95" + ] + } + ] + } + ], + "retrospective": { + "summary": "Added atomic 1:1 DM ResolveOrReserve reservations with migration backfill, fail-closed digest collision handling, and concurrent negative-control coverage.", + "approach": "Standard approach", + "confidence": 0.95 + }, + "commits": [], + "filesChanged": [], + "projectId": "AgentWorkforce/relaycast", + "tags": [], + "_trace": { + "startRef": "21a5390929e16c21058e08256dcdc33434e8e108", + "endRef": "21a5390929e16c21058e08256dcdc33434e8e108" + } +} \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index e3b1b398..468e533f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,11 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). Packages without a separate changelog are covered by the cross-package notes below. -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- 1:1 DM resolution now atomically reserves each deterministic conversation ID for its workspace and participant pair. A collision returns `409 dm_conversation_id_collision` instead of silently aliasing another conversation. ## [6.3.1] - 2026-07-31 diff --git a/README.md b/README.md index f2ccae1f..7850cf30 100644 --- a/README.md +++ b/README.md @@ -411,6 +411,13 @@ GET /workspace/events Activity feed channel-message items include `channel_id` and `channel_name`; DM items include `conversation_id`. +`POST /dm` can return **`409 dm_conversation_id_collision`**. A 1:1 conversation id is derived +deterministically from `(workspace, sorted agent pair)`, and that binding is reserved +atomically before any conversation state is created, so a derivation that would name another +pair's conversation fails closed instead of resolving to it. Two cases produce the error: the +identifier is already bound to a different pair, or the pair is already bound to a different +identifier. It is not retryable — the same inputs collide again. + `GET /workspace/events?since=&limit=` is the durable, cursor-based log behind the workspace stream (30-day retention by default): every published stream frame is appended with a per-workspace monotonic `seq` (also stamped on the live frame), so observers can reconcile after diff --git a/openapi.yaml b/openapi.yaml index c0d6c86d..072f15d2 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -2687,6 +2687,20 @@ paths: type: boolean data: $ref: '#/components/schemas/DmSendResponse' + '409': + description: | + `dm_conversation_id_collision` — the deterministic 1:1 conversation + identifier could not be reserved for this exact + `(workspace, sorted agent pair)` tuple. A 1:1 DM id is derived from + that tuple, and the reservation is atomic, so the request fails + closed rather than resolving to another pair's conversation. Two + cases produce it: the identifier is already bound to a different + pair, or the pair is already bound to a different identifier. It is + not retryable — the same inputs will collide again. + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorResponse' /dm/conversations: get: diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index cba6c630..9f0acda1 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Patch] + +### Fixed + +- `sendDm` atomically reserves each deterministic 1:1 conversation ID for its workspace and sorted participant pair, so a collision returns `409 dm_conversation_id_collision` on every driver instead of resolving to another pair's conversation or failing with an uncaught database error. Migration `0033` adds `dm_conversation_reservations` and backfills existing 1:1 DMs; see the migration header for the pre-flight audit to run before applying it. ## [6.3.1] - 2026-07-31 diff --git a/packages/engine/src/db/migrations/0033_dm_conversation_reservations.sql b/packages/engine/src/db/migrations/0033_dm_conversation_reservations.sql new file mode 100644 index 00000000..40681abd --- /dev/null +++ b/packages/engine/src/db/migrations/0033_dm_conversation_reservations.sql @@ -0,0 +1,124 @@ +-- Bind every deterministic 1:1 DM id to the workspace and sorted participant +-- pair that derived it. The primary key is the atomic reservation seam: a +-- conflicting digest can never overwrite or alias another pair's conversation. +CREATE TABLE dm_conversation_reservations ( + conversation_id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, + participant_one_id TEXT NOT NULL, + participant_two_id TEXT NOT NULL, + created_at INTEGER NOT NULL DEFAULT (unixepoch()), + CONSTRAINT dm_conversation_reservations_sorted_pair_check + CHECK (participant_one_id <= participant_two_id) +); + +CREATE UNIQUE INDEX dm_conversation_reservations_pair_unique + ON dm_conversation_reservations (workspace_id, participant_one_id, participant_two_id); + +-- PRE-FLIGHT AUDIT (run BOTH read-only queries BEFORE applying this migration). +-- The backfill can abort for two independent reasons, and each needs its own +-- check. An earlier version of this comment shipped only query (a) and claimed an +-- empty result meant the migration would apply cleanly. That was wrong: a +-- duplicate pair passes (a) and still aborts the backfill on the pair-uniqueness +-- index. Raised in review of PR #303. +-- +-- (a) Roster shape. The backfill aborts if any single legacy 1:1 has a roster +-- that is not 1 or 2 distinct participants. That is deliberate - inventing a +-- tuple would bind a conversation to the wrong pair - but one corrupt row +-- blocks the whole deployment. +-- +-- SELECT dc.id, dc.workspace_id, COUNT(DISTINCT dp.agent_id) AS participants +-- FROM dm_conversations dc +-- LEFT JOIN dm_participants dp ON dp.conversation_id = dc.id +-- WHERE dc.dm_type = '1:1' +-- GROUP BY dc.id, dc.workspace_id +-- HAVING participants NOT BETWEEN 1 AND 2; +-- +-- (b) Duplicate pairs. Two DISTINCT legacy 1:1 conversations in one workspace +-- that resolve to the same sorted pair both satisfy (a), but only one can be +-- reserved - the second violates dm_conversation_reservations_pair_unique and +-- aborts the migration. This is exactly what a pre-deterministic id scheme +-- leaves behind, so it is the likelier of the two in practice. +-- +-- SELECT workspace_id, participant_one_id, participant_two_id, +-- COUNT(*) AS conversations +-- FROM ( +-- SELECT dc.workspace_id, +-- MIN(dp.agent_id) AS participant_one_id, +-- MAX(dp.agent_id) AS participant_two_id +-- FROM dm_conversations dc +-- JOIN dm_participants dp ON dp.conversation_id = dc.id +-- WHERE dc.dm_type = '1:1' +-- GROUP BY dc.id, dc.workspace_id +-- HAVING COUNT(DISTINCT dp.agent_id) BETWEEN 1 AND 2 +-- ) +-- GROUP BY workspace_id, participant_one_id, participant_two_id +-- HAVING conversations > 1; +-- +-- (c) Ids that do not match the CURRENT derivation. This one cannot be written +-- in SQL - it needs SHA-256, which SQLite does not have - and it is the check +-- that matters most operationally, because it is the only failure that is +-- INVISIBLE AT MIGRATION TIME. (a) and (b) abort the migration loudly. (c) +-- lets it succeed, and then every subsequent DM between that pair returns 409 +-- forever, because the backfill reserved the pair under an id the send path +-- will never re-derive. +-- +-- Run: node scripts/audit-dm-reservations.mjs --sqlite +-- D1: wrangler d1 execute --json --command "" \ +-- | node scripts/audit-dm-reservations.mjs --stdin +-- +-- That script also re-runs (a) and (b), so it is the single command to trust. +-- +-- All three clean means this migration will apply AND no existing pair will start +-- failing afterwards. Remediate anything any of them returns - for (b) that means +-- deciding which conversation survives, since the reservation can only bind one; +-- for (c) it means re-keying the conversation to the derived id, or seeding its +-- reservation under the derived id, before deploying the code that reserves. + +-- Backfill ONLY conversations with exactly two distinct participants. +-- +-- A one-row roster is ambiguous and MUST NOT be reserved. It looks like a +-- self-DM, but `dm_participants.agent_id` cascades on agent deletion, so a +-- perfectly ordinary two-party 1:1 collapses to a single row the moment one +-- participant's agent is deleted - while its id still encodes the ORIGINAL pair. +-- +-- Reading those as (X, X) is wrong twice over. Several orphans belonging to the +-- same surviving agent all collapse to the same (workspace, X, X) tuple and +-- collide on the pair-uniqueness index, aborting the migration; and any that +-- survived would reserve a self-DM tuple against an id no derivation produces, +-- so that agent's next self-DM would 409 forever. +-- +-- This is not hypothetical. An earlier version of this backfill used +-- MIN/MAX over 1-2 participants; audited against production it produced 4 +-- colliding pair groups and 30 mismatched ids, all of them orphaned two-party +-- conversations, and would have failed the deployment. Restricted to exactly two +-- participants the same data yields zero findings across 3425 conversations. +-- +-- Skipping is safe rather than merely convenient. An unreserved conversation is +-- in exactly the state every conversation was in before this migration: the +-- first send through the reservation path claims it, and because a genuine +-- self-DM's id already equals its derivation, that claim adopts the existing +-- conversation instead of creating a second one. Orphaned two-party rows are +-- simply never re-derived, so they stay readable and inert. +-- +-- Malformed rosters (zero, or more than two) are skipped for the same reason. +-- The earlier version aborted the whole migration on them; skipping avoids +-- inventing a tuple just as effectively without blocking a deployment, and any +-- future send still goes through the reservation path. +INSERT INTO dm_conversation_reservations ( + conversation_id, + workspace_id, + participant_one_id, + participant_two_id, + created_at +) +SELECT + dc.id, + dc.workspace_id, + MIN(dp.agent_id), + MAX(dp.agent_id), + dc.created_at +FROM dm_conversations dc +JOIN dm_participants dp ON dp.conversation_id = dc.id +WHERE dc.dm_type = '1:1' +GROUP BY dc.id, dc.workspace_id, dc.created_at +HAVING COUNT(DISTINCT dp.agent_id) = 2; diff --git a/packages/engine/src/db/schema.ts b/packages/engine/src/db/schema.ts index bfcd6f57..27cf3a9c 100644 --- a/packages/engine/src/db/schema.ts +++ b/packages/engine/src/db/schema.ts @@ -7,6 +7,7 @@ import { uniqueIndex, primaryKey, foreignKey, + check, } from 'drizzle-orm/sqlite-core'; import { sql } from 'drizzle-orm'; import type { AnySQLiteColumn } from 'drizzle-orm/sqlite-core'; @@ -602,6 +603,33 @@ export const dmConversations = sqliteTable( ], ); +// ============================================ +// 1:1 DM Conversation Reservations +// ============================================ +export const dmConversationReservations = sqliteTable( + 'dm_conversation_reservations', + { + conversationId: text('conversation_id').primaryKey(), + workspaceId: text('workspace_id') + .notNull() + .references(() => workspaces.id, { onDelete: 'cascade' }), + participantOneId: text('participant_one_id').notNull(), + participantTwoId: text('participant_two_id').notNull(), + createdAt: integer('created_at', { mode: 'timestamp' }).notNull().default(sql`(unixepoch())`), + }, + (table) => [ + check( + 'dm_conversation_reservations_sorted_pair_check', + sql`${table.participantOneId} <= ${table.participantTwoId}`, + ), + uniqueIndex('dm_conversation_reservations_pair_unique').on( + table.workspaceId, + table.participantOneId, + table.participantTwoId, + ), + ], +); + // ============================================ // DM Participants // ============================================ diff --git a/packages/engine/src/engine/__tests__/dm.test.ts b/packages/engine/src/engine/__tests__/dm.test.ts index 5913bfd6..77a526e6 100644 --- a/packages/engine/src/engine/__tests__/dm.test.ts +++ b/packages/engine/src/engine/__tests__/dm.test.ts @@ -9,12 +9,18 @@ * are a compatibility surface, not just internal behaviour. */ -import { afterEach, describe, expect, it } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import { and, eq, isNotNull } from 'drizzle-orm'; import { getSqliteDb, runMigrations, type SqliteDbHandle } from '../../adapters/node/database.js'; -import { agents, dmConversations, dmParticipants, workspaces } from '../../db/schema.js'; -import { sendDm } from '../dm.js'; +import { + agents, + dmConversationReservations, + dmConversations, + dmParticipants, + workspaces, +} from '../../db/schema.js'; +import { isPairReservationConflict, sendDm } from '../dm.js'; type Db = SqliteDbHandle['db']; @@ -56,7 +62,7 @@ function seed(): Fixture { return { db: handle.db, ws, alice, bob }; } -/** Mirrors `getDmPairKey` — kept independent so a change to the algorithm fails here. */ +/** Mirrors the derivation helper — kept independent so an algorithm change fails here. */ async function expectedConversationId(ws: string, a: string, b: string): Promise { const [first, second] = [a, b].sort(); const digest = await crypto.subtle.digest( @@ -104,6 +110,261 @@ describe('1:1 DM conversation identity', () => { .where(eq(dmParticipants.conversationId, first.conversation_id)); expect(participants).toHaveLength(2); }); + + /** + * Migration 0033 deliberately does NOT reserve one-row rosters, because agent + * deletion cascades `dm_participants` and collapses an ordinary two-party 1:1 + * into a single row whose id still encodes the original pair. Reserving those + * as (X, X) collided on production data. + * + * The safety of skipping them rests entirely on this: an existing, UNRESERVED + * conversation must be adopted by the first send, not duplicated. A genuine + * self-DM's id already equals its derivation, so the reservation claims that + * same id and the conversation is reused. + * + * If this ever regressed, every pre-migration self-DM would silently fork into + * a second conversation on first use and the old history would disappear from + * the user's view. + */ + it('adopts an existing unreserved conversation instead of duplicating it', async () => { + const { db, ws, alice } = seed(); + + // Create it, then drop the reservation to model a pre-0033 conversation. + await sendDm(db, ws, alice, { to: '@self', text: 'before the migration' }); + const original = db.select().from(dmConversations).all(); + expect(original).toHaveLength(1); + db.delete(dmConversationReservations).run(); + + await sendDm(db, ws, alice, { to: '@self', text: 'after the migration' }); + + const after = db.select().from(dmConversations).all(); + expect(after).toHaveLength(1); + expect(after[0].id).toBe(original[0].id); + expect(db.select().from(dmConversationReservations).all()).toHaveLength(1); + }); + + /** + * A self-DM is one roster row, so the reservation stores the same agent as + * both participants. That has to satisfy the sorted-pair CHECK + * (participant_one_id <= participant_two_id), which it does only because the + * comparison is non-strict. Untested before, and `@self` is a documented + * request shape, so a stricter constraint would have broken a live feature. + */ + it('reserves a self-DM without violating the sorted-pair constraint', async () => { + const { db, ws, alice } = seed(); + await sendDm(db, ws, alice, { to: '@self', text: 'note to self' }); + + const rows = db.select().from(dmConversationReservations).all(); + expect(rows).toHaveLength(1); + expect(rows[0].participantOneId).toBe(rows[0].participantTwoId); + }); + + /** + * The reservation and the conversation/roster inserts are NOT one transaction, + * so a crash between them leaves a reservation with no conversation behind it. + * That must be self-healing rather than a permanent 409: the retry presents the + * identical tuple, which the conditional upsert accepts. + * + * Asserted rather than assumed. If re-resolution ever stopped accepting an + * identical tuple, a single mid-write crash would lock that pair out of DMs + * for good, and nothing else in this suite would notice. + */ + it('recovers when a reservation outlives its conversation', async () => { + const { db, ws, alice } = seed(); + await sendDm(db, ws, alice, { to: 'bob', text: 'one' }); + const reserved = db.select().from(dmConversationReservations).all()[0].conversationId; + + db.delete(dmConversations).run(); // conversation gone, reservation remains + + await expect(sendDm(db, ws, alice, { to: 'bob', text: 'two' })).resolves.toBeTruthy(); + expect(db.select().from(dmConversations).all()[0].id).toBe(reserved); + }); + + /** + * The 409 above must survive the driver change between self-hosted and hosted. + * + * This engine has already regressed this exact class once: PR #193 added + * `isUniqueConstraintError` to agent.ts after clean 409 handling became an + * uncaught 500 on D1, because detection only matched better-sqlite3's shape. + * The first version of the reservation handler made the same mistake, and the + * suite could not see it because the suite runs better-sqlite3. + * + * So the shapes are asserted directly rather than only through the send path. + */ + it('recognises the pair conflict across driver error shapes', () => { + // better-sqlite3 (what this suite actually runs) + expect(isPairReservationConflict({ + code: 'SQLITE_CONSTRAINT_UNIQUE', + message: 'UNIQUE constraint failed: dm_conversation_reservations.workspace_id', + })).toBe(true); + + // D1, top level + expect(isPairReservationConflict({ + code: 'SQLITE_CONSTRAINT_UNIQUE', + message: 'D1_ERROR: UNIQUE constraint failed: dm_conversation_reservations.workspace_id', + })).toBe(true); + + // D1 re-wrapped by drizzle: neither code nor table name at the top level + expect(isPairReservationConflict({ + message: 'Failed query: insert into "dm_conversation_reservations"', + cause: { + code: 'SQLITE_CONSTRAINT_UNIQUE', + message: 'D1_ERROR: UNIQUE constraint failed: dm_conversation_reservations.workspace_id', + }, + })).toBe(true); + + // Split across the chain: code on the wrapper, table name on the cause + expect(isPairReservationConflict({ + code: 'SQLITE_CONSTRAINT_UNIQUE', + message: 'Failed query', + cause: { message: 'UNIQUE constraint failed: dm_conversation_reservations.workspace_id' }, + })).toBe(true); + }); + + it('does not launder unrelated failures into a pair conflict', () => { + // A different table's unique violation is not ours. + expect(isPairReservationConflict({ + code: 'SQLITE_CONSTRAINT_UNIQUE', + message: 'UNIQUE constraint failed: observer_tokens.workspace_id', + })).toBe(false); + + // A non-unique constraint on OUR table is not a pair collision either - + // reporting a FK failure as a participant-pair conflict would be a lie. + expect(isPairReservationConflict({ + code: 'SQLITE_CONSTRAINT_FOREIGNKEY', + message: 'FOREIGN KEY constraint failed: dm_conversation_reservations.workspace_id', + })).toBe(false); + + expect(isPairReservationConflict(null)).toBe(false); + expect(isPairReservationConflict(new Error('boom'))).toBe(false); + }); + + it('terminates on a cyclic cause chain instead of blowing the stack', () => { + const a: { message: string; cause?: unknown } = { message: 'wrapper a' }; + const b: { message: string; cause?: unknown } = { message: 'wrapper b', cause: a }; + a.cause = b; // A -> B -> A, which a self-reference-only guard would miss + expect(() => isPairReservationConflict(a)).not.toThrow(); + expect(isPairReservationConflict(a)).toBe(false); + }); + + /** + * A legacy 1:1 conversation whose id is NOT the current derivation is backfilled + * by migration 0033 under its own id, so the PAIR is reserved while the + * deterministic id is not. The next send derives a different conversation_id for + * the same pair, which violates the pair_unique index rather than the primary + * key. The upsert only names conversation_id as its conflict target, so that + * surfaced as a raw SQLITE_CONSTRAINT_UNIQUE (500) instead of the controlled + * 409 this whole seam exists to produce. + * + * Raised in review of PR #303. Failing closed is not enough on its own - it has + * to fail closed with the documented code, or callers cannot distinguish it + * from an engine fault. + */ + it('reports a controlled collision when the pair is reserved under a different id', async () => { + const { db, ws, alice, bob } = seed(); + const [first, second] = [alice, bob].sort(); + + // Simulate the 0033 backfill of a pre-deterministic conversation id. + db.insert(dmConversationReservations).values({ + conversationId: 'dm_legacy_nondeterministic_id', + workspaceId: ws, + participantOneId: first, + participantTwoId: second, + }).run(); + + let code: string | undefined; + let status: number | undefined; + let raw: string | undefined; + try { + await sendDm(db, ws, alice, { to: 'bob', text: 'hello' }); + } catch (err) { + const e = err as { code?: string; status?: number; message?: string }; + code = e.code; + status = e.status; + raw = e.message; + } + + expect(code, `expected a coded collision, got: ${raw}`).toBe('dm_conversation_id_collision'); + expect(status).toBe(409); + }); + + /** + * NOTE ON WHAT THIS DOES AND DOES NOT PROVE, raised in review of PR #303. + * + * Both real `sendDm` paths are launched before either is awaited, so this + * exercises the production path with a forced digest collision. But + * better-sqlite3 is a single synchronous connection, so the two sends + * serialize and there is no genuine database contention. It proves the + * collision branch and that the reservation is consulted before any + * conversation state is created. It does NOT prove multi-writer atomicity + * against a networked engine. + * + * The DB constraint is what makes the operation atomic; this test proves the + * constraint is load-bearing, which the negative control (dropping to a silent + * ON CONFLICT DO NOTHING) confirms by producing two winners. Named for what it + * demonstrates rather than for the mechanism it relies on. + */ + it('rejects the losing pair when two colliding sends interleave', async () => { + const { db, ws, alice, bob } = seed(); + const suffix = ws.slice(3); + const carol = `ag_carol_${suffix}`; + const dave = `ag_dave_${suffix}`; + + db.insert(agents).values({ + id: carol, + workspaceId: ws, + name: 'carol', + tokenHash: `tok_c_${suffix}`, + }).run(); + db.insert(agents).values({ + id: dave, + workspaceId: ws, + name: 'dave', + tokenHash: `tok_d_${suffix}`, + }).run(); + + const forcedDigest = new Uint8Array(32).fill(0x5a).buffer; + const digestSpy = vi.spyOn(globalThis.crypto.subtle, 'digest').mockResolvedValue(forcedDigest); + + let outcomes: PromiseSettledResult>>[]; + try { + // Deliberately launch both real send paths before awaiting either one. + outcomes = await Promise.allSettled([ + sendDm(db, ws, alice, { to: 'bob', text: 'alice to bob' }), + sendDm(db, ws, carol, { to: 'dave', text: 'carol to dave' }), + ]); + } finally { + digestSpy.mockRestore(); + } + + const winners = outcomes.filter((outcome) => outcome.status === 'fulfilled'); + const losers = outcomes.filter((outcome) => outcome.status === 'rejected'); + + expect(winners).toHaveLength(1); + expect(losers).toHaveLength(1); + expect((losers[0] as PromiseRejectedResult).reason).toMatchObject({ + code: 'dm_conversation_id_collision', + status: 409, + }); + + const winner = (winners[0] as PromiseFulfilledResult>>).value; + const roster = await db + .select({ agentId: dmParticipants.agentId }) + .from(dmParticipants) + .where(eq(dmParticipants.conversationId, winner.conversation_id)); + const rosterIds = roster.map((row) => row.agentId).sort(); + expect([ + [alice, bob].sort(), + [carol, dave].sort(), + ]).toContainEqual(rosterIds); + + const reservations = await db.select().from(dmConversationReservations); + expect(reservations).toHaveLength(1); + expect([ + reservations[0].participantOneId, + reservations[0].participantTwoId, + ]).toEqual(rosterIds); + }); }); describe('invariant DM-1: a 1:1 conversation never has a departed participant', () => { diff --git a/packages/engine/src/engine/dm.ts b/packages/engine/src/engine/dm.ts index 1dbb3fe6..029c8312 100644 --- a/packages/engine/src/engine/dm.ts +++ b/packages/engine/src/engine/dm.ts @@ -6,6 +6,7 @@ import { channels, agents, dmConversations, + dmConversationReservations, dmParticipants, messageAttachments, } from '../db/schema.js'; @@ -30,75 +31,193 @@ interface SendDmOptions { mailbox?: MailboxConfig; } -async function getDmPairKey(workspaceId: string, agentA: string, agentB: string): Promise { +/** + * Derivation only: this digest does not resolve or claim a conversation. + * Callers must atomically reserve the derived id before using it. + */ +async function deriveDmPairKey(workspaceId: string, agentA: string, agentB: string): Promise { const [first, second] = [agentA, agentB].sort(); return (await sha256Hex(`${workspaceId}:${first}:${second}`)).slice(0, 24); } +/** + * Recognize a unique-constraint violation on `dm_conversation_reservations` + * across every driver this engine actually runs on. + * + * THIS ENGINE HAS ALREADY REGRESSED THIS EXACT BUG CLASS ONCE. `engine/agent.ts` + * gained `isUniqueConstraintError` in PR #193 after clean 409 handling silently + * became an uncaught 500 against D1, because detection only matched + * better-sqlite3's error shape. `engine/observerToken.ts` documents the same + * trap. A first version of this handler matched `.code` and `.message` on the + * top-level error only — which passes against better-sqlite3 in tests and would + * have reproduced that regression in the hosted engine, where the driver differs + * from the one the test suite exercises. + * + * Self-hosted runs on better-sqlite3: `.code` is `SQLITE_CONSTRAINT_UNIQUE` and + * `.message` reads `UNIQUE constraint failed: dm_conversation_reservations...`. + * The hosted engine runs Cloudflare D1 via `drizzle-orm/d1`, which prefixes + * `D1_ERROR: ` and may re-wrap the driver error under `.cause` rather than + * surfacing it at the top level. So the chain has to be walked. + * + * Two conditions must BOTH hold somewhere in the chain: the failure is a unique + * violation, and it names this table. The table check is what stops an unrelated + * constraint failure on this insert — a FK or NOT NULL on `workspace_id` — from + * being laundered into a tidy 409 that says something untrue about participant + * pairs. They are tracked independently across the walk because a wrapper may + * carry the code while only the wrapped cause carries the message. + * + * The walk is iterative and records every object visited in a `WeakSet`, + * breaking on any revisit rather than only a direct self-reference: a multi-step + * cycle (`A -> B -> A`) would otherwise blow the stack and turn the check meant + * to prevent a 500 into one itself. Same reasoning as `isObserverTokenNameConflict`. + */ +export function isPairReservationConflict(err: unknown): boolean { + const visited = new WeakSet(); + let current: unknown = err; + let sawUniqueViolation = false; + let namesReservationTable = false; + + while (current && typeof current === 'object') { + if (visited.has(current)) break; + visited.add(current); + + const candidate = current as { code?: string; message?: string; cause?: unknown }; + const message = candidate.message ?? ''; + const lowerMessage = message.toLowerCase(); + + if ( + candidate.code === 'SQLITE_CONSTRAINT_UNIQUE' + || lowerMessage.includes('unique constraint failed') + || (candidate.code === 'SQLITE_CONSTRAINT' && lowerMessage.includes('unique')) + ) { + sawUniqueViolation = true; + } + if (lowerMessage.includes('dm_conversation_reservations')) { + namesReservationTable = true; + } + + current = candidate.cause; + } + + return sawUniqueViolation && namesReservationTable; +} + +/** + * Atomically resolve or reserve a deterministic 1:1 DM id for one exact tuple. + * + * The primary-key conflict and conditional no-op update are one SQL statement. + * An identical tuple returns the existing reservation; a digest collision makes + * the conflict predicate false, returns no row, and fails closed. + * + * TWO DISTINCT COLLISIONS, and both must fail closed with the same coded 409: + * + * 1. Same conversation_id, different pair. Caught by the PRIMARY KEY conflict + * target: the conditional update predicate is false, no row is returned. + * + * 2. Same pair, different conversation_id. This violates the pair_unique + * index, which is NOT the conflict target - SQLite only accepts one - so + * the statement raises SQLITE_CONSTRAINT_UNIQUE. Raised in review of PR + * #303 and reachable in practice: migration 0033 backfills whatever `dc.id` + * a legacy 1:1 already had, without requiring it to equal the current + * derivation, so the pair can be reserved under an id the next send will + * not re-derive. Left unhandled that surfaced as a 500. + * + * Failing closed is not sufficient on its own. It has to fail closed with the + * documented code, or a caller cannot tell a refused collision from an engine + * fault - which is the same distinction the rest of this seam exists to make. + */ +async function resolveOrReserveConversation( + db: Db, + conversationId: string, + workspaceId: string, + sortedPair: readonly [string, string], +): Promise { + const [participantOneId, participantTwoId] = sortedPair; + + let reservation: { conversationId: string } | undefined; + try { + [reservation] = await db + .insert(dmConversationReservations) + .values({ + conversationId, + workspaceId, + participantOneId, + participantTwoId, + }) + .onConflictDoUpdate({ + target: dmConversationReservations.conversationId, + set: { conversationId: sql`excluded.conversation_id` }, + setWhere: and( + eq(dmConversationReservations.workspaceId, workspaceId), + eq(dmConversationReservations.participantOneId, participantOneId), + eq(dmConversationReservations.participantTwoId, participantTwoId), + ), + }) + .returning({ conversationId: dmConversationReservations.conversationId }); + } catch (err) { + // Collision (2) above. + if (!isPairReservationConflict(err)) throw err; + + throw codedError( + 'DM participant pair is already reserved under a different conversation identifier', + 'dm_conversation_id_collision', + 409, + ); + } + + if (!reservation) { + throw codedError( + 'DM conversation identifier is already reserved for a different participant pair', + 'dm_conversation_id_collision', + 409, + ); + } +} + async function resolveConversation( db: Db, workspaceId: string, fromAgentId: string, toAgentId: string, ) { - const existing = await db.all<{ id: string; channel_id: string }>(sql` - SELECT dc.id, dc.channel_id - FROM dm_conversations dc - JOIN dm_participants p1 - ON p1.conversation_id = dc.id - AND p1.agent_id = ${fromAgentId} - AND p1.left_at IS NULL - JOIN dm_participants p2 - ON p2.conversation_id = dc.id - AND p2.agent_id = ${toAgentId} - AND p2.left_at IS NULL - WHERE dc.workspace_id = ${workspaceId} - AND dc.dm_type = '1:1' - LIMIT 1 - `); - - let conversationId = existing[0]?.id; - - if (!conversationId) { - const pairKey = await getDmPairKey(workspaceId, fromAgentId, toAgentId); - const deterministicConversationId = `dm_${pairKey}`; - const deterministicChannelId = `dmch_${pairKey}`; - - await db.insert(channels).values({ - id: deterministicChannelId, - workspaceId, - name: `dm-${pairKey}`, - channelType: 1, - }).onConflictDoNothing(); - - await db.insert(dmConversations).values({ - id: deterministicConversationId, - workspaceId, - channelId: deterministicChannelId, - dmType: '1:1', - }).onConflictDoNothing(); - - // Clear `left_at` rather than no-op on conflict. Reaching this branch for an - // id that already exists means the lookup above missed, and the only way it - // can miss for an existing 1:1 is a participant marked departed. Leaving the - // marker set would resolve the conversation while its roster disagreed — - // see invariant DM-1 in __tests__/dm.test.ts. - const rejoin = { - target: [dmParticipants.conversationId, dmParticipants.agentId], - set: { leftAt: null }, - }; - - await db.insert(dmParticipants).values({ - conversationId: deterministicConversationId, - agentId: fromAgentId, - }).onConflictDoUpdate(rejoin); - await db.insert(dmParticipants).values({ - conversationId: deterministicConversationId, - agentId: toAgentId, - }).onConflictDoUpdate(rejoin); + const sortedPair = [fromAgentId, toAgentId].sort() as [string, string]; + const pairKey = await deriveDmPairKey(workspaceId, sortedPair[0], sortedPair[1]); + const conversationId = `dm_${pairKey}`; + const channelId = `dmch_${pairKey}`; + + // This is the mandatory resolution seam. It must happen before any metadata + // creation so exactly one tuple can win a digest collision. + await resolveOrReserveConversation(db, conversationId, workspaceId, sortedPair); + + await db.insert(channels).values({ + id: channelId, + workspaceId, + name: `dm-${pairKey}`, + channelType: 1, + }).onConflictDoNothing(); + + await db.insert(dmConversations).values({ + id: conversationId, + workspaceId, + channelId, + dmType: '1:1', + }).onConflictDoNothing(); + + // A deterministic 1:1 is a durable relationship. Re-resolution restores a + // stale departure marker instead of letting roster state disagree with it. + const rejoin = { + target: [dmParticipants.conversationId, dmParticipants.agentId], + set: { leftAt: null }, + }; - conversationId = deterministicConversationId; - } + await db.insert(dmParticipants).values({ + conversationId, + agentId: fromAgentId, + }).onConflictDoUpdate(rejoin); + await db.insert(dmParticipants).values({ + conversationId, + agentId: toAgentId, + }).onConflictDoUpdate(rejoin); const [conv] = await db .select({ id: dmConversations.id, channelId: dmConversations.channelId }) diff --git a/scripts/audit-dm-reservations.mjs b/scripts/audit-dm-reservations.mjs new file mode 100755 index 00000000..669b8524 --- /dev/null +++ b/scripts/audit-dm-reservations.mjs @@ -0,0 +1,267 @@ +#!/usr/bin/env node +/** + * Pre-flight audit for migration 0033 (dm_conversation_reservations). + * + * READ-ONLY. Run this against a production replica BEFORE applying 0033. + * + * WHY THIS EXISTS AS A SCRIPT AND NOT SQL + * ─────────────────────────────────────── + * The interesting check needs SHA-256 to recompute the deterministic + * conversation id, and SQLite does not have it. It is also the check whose + * failure is INVISIBLE AT MIGRATION TIME: + * + * (b) duplicate pair -> migration aborts loudly. You find out immediately. + * (c) id != derivation -> migration SUCCEEDS, and then every subsequent DM + * between that pair returns 409 forever, because the + * backfill reserved the pair under an id the send + * path will never re-derive. + * + * IT ALSO EARNED ITS KEEP. Run against production it found 4 colliding pair + * groups and 30 mismatched ids that would have aborted the deployment. All 30 + * were ORPHANED TWO-PARTY conversations, not self-DMs: `dm_participants.agent_id` + * cascades on agent deletion, so an ordinary 1:1 collapses to a one-row roster + * while its id still encodes the original pair. The backfill was reading those + * as (X, X). Migration 0033 now reserves only exactly-two-participant + * conversations, and the same production data yields zero findings. + * + * So one-row rosters are reported, not flagged — and the derivation is used to + * say which are genuine self-DMs and which are orphans, since the roster alone + * cannot tell them apart. + * + * Git history says the derivation never changed — the only edit swapped node + * crypto for web crypto with a byte-identical input string, and production + * confirms it: 3425 two-party conversations, zero mismatches. This script exists + * because "stable in git history" and "zero rows in production disagree" are + * different claims, and only the second one is evidence. + * + * USAGE + * ───── + * Self-hosted (SQLite file): + * node scripts/audit-dm-reservations.mjs --sqlite /path/to/relay.db + * + * Hosted (Cloudflare D1) — feed it the rows, since D1 is not a local file: + * wrangler d1 execute --json --command \ + * "SELECT dc.id, dc.workspace_id, dp.agent_id \ + * FROM dm_conversations dc \ + * JOIN dm_participants dp ON dp.conversation_id = dc.id \ + * WHERE dc.dm_type = '1:1'" \ + * | node scripts/audit-dm-reservations.mjs --stdin + * + * Exit codes: 0 = clean, 1 = findings, 2 = usage/read error. + */ + +import { createHash } from 'node:crypto'; + +/** Must stay byte-identical to `deriveDmPairKey` in packages/engine/src/engine/dm.ts. */ +function deriveConversationId(workspaceId, agentA, agentB) { + const [first, second] = [agentA, agentB].sort(); + const key = createHash('sha256') + .update(`${workspaceId}:${first}:${second}`) + .digest('hex') + .slice(0, 24); + return `dm_${key}`; +} + +function usage(message) { + if (message) console.error(`error: ${message}\n`); + console.error('usage: audit-dm-reservations.mjs --sqlite | --stdin'); + process.exit(2); +} + +async function readStdin() { + const chunks = []; + for await (const chunk of process.stdin) chunks.push(chunk); + return Buffer.concat(chunks).toString('utf8'); +} + +/** Normalise either input source to [{ id, workspace_id, agent_id }]. */ +async function loadRows(argv) { + if (argv.includes('--stdin')) { + const raw = await readStdin(); + if (!raw.trim()) usage('nothing on stdin'); + // Tolerate a preamble. wrangler emits banners and warnings on stdout + // depending on version and TTY attachment, and an operator running this as a + // deploy gate should not have to care. A brittle parser here fails exactly + // when it matters most. + const start = raw.search(/[[{]/); + if (start === -1) usage(`stdin contained no JSON. First 200 chars:\n${raw.slice(0, 200)}`); + + let parsed; + try { + parsed = JSON.parse(raw.slice(start)); + } catch (err) { + usage(`stdin was not valid JSON (${err.message}). First 200 chars:\n${raw.slice(start, start + 200)}`); + } + + // Envelope shapes seen in the wild: [{results:[...]}], {results:[...]}, + // {result:[{results:[...]}]}, and a bare array of rows. Rather than enumerate + // them forever, find the first array whose members look like our rows. + const looksLikeRows = (v) => + Array.isArray(v) && v.length > 0 && typeof v[0] === 'object' && v[0] !== null + && ('agent_id' in v[0]) && ('workspace_id' in v[0]); + + const seen = new Set(); + const find = (node, depth = 0) => { + if (depth > 6 || node === null || typeof node !== 'object') return null; + if (seen.has(node)) return null; + seen.add(node); + if (looksLikeRows(node)) return node; + for (const value of Array.isArray(node) ? node : Object.values(node)) { + const hit = find(value, depth + 1); + if (hit) return hit; + } + return null; + }; + + const rows = find(parsed); + if (!rows) { + // An empty result set is legitimate: a deployment with no 1:1 DMs at all. + if (JSON.stringify(parsed).includes('"results":[]')) return []; + usage( + 'could not find a row array containing agent_id and workspace_id.\n' + + `Received (first 300 chars): ${JSON.stringify(parsed).slice(0, 300)}\n` + + 'Check the SELECT aliases match: id, workspace_id, agent_id.', + ); + } + return rows; + } + + const i = argv.indexOf('--sqlite'); + if (i === -1 || !argv[i + 1]) usage('pass --sqlite or --stdin'); + const path = argv[i + 1]; + + let Database; + try { + ({ default: Database } = await import('better-sqlite3')); + } catch { + usage('better-sqlite3 is not installed here; use --stdin instead'); + } + const db = new Database(path, { readonly: true, fileMustExist: true }); + try { + return db.prepare(` + SELECT dc.id AS id, dc.workspace_id AS workspace_id, dp.agent_id AS agent_id + FROM dm_conversations dc + JOIN dm_participants dp ON dp.conversation_id = dc.id + WHERE dc.dm_type = '1:1' + `).all(); + } finally { + db.close(); + } +} + +const rows = await loadRows(process.argv.slice(2)); + +// Group the roster per conversation. +const conversations = new Map(); +for (const row of rows) { + const id = row.id ?? row.conversation_id; + const workspaceId = row.workspace_id; + const agentId = row.agent_id; + if (!id || !workspaceId || !agentId) continue; + if (!conversations.has(id)) conversations.set(id, { id, workspaceId, agents: new Set() }); + conversations.get(id).agents.add(agentId); +} + +const malformed = []; +const mismatched = []; +const pairIndex = new Map(); // workspace|a|b -> [conversationId] + +const skipped = []; + +for (const conv of conversations.values()) { + const agents = [...conv.agents].sort(); + + // Migration 0033 reserves ONLY exactly-two-participant conversations. Anything + // else is skipped by the backfill, so it cannot abort the migration and cannot + // create a wrong binding. Recorded, not flagged. + // + // A one-row roster is genuinely ambiguous: `dm_participants.agent_id` cascades + // on agent deletion, so an ordinary two-party 1:1 collapses to one row while + // its id still encodes the original pair. It is indistinguishable from a real + // self-DM by roster alone -- but the derivation tells them apart, so this + // reports which is which. + if (agents.length !== 2) { + const solo = agents[0]; + const looksLikeSelfDm = agents.length === 1 + && conv.id === deriveConversationId(conv.workspaceId, solo, solo); + skipped.push({ + id: conv.id, + workspaceId: conv.workspaceId, + participants: agents.length, + kind: agents.length === 1 ? (looksLikeSelfDm ? 'self-DM' : 'orphaned two-party') : 'malformed', + }); + if (agents.length === 0 || agents.length > 2) { + malformed.push({ id: conv.id, workspaceId: conv.workspaceId, participants: agents.length }); + } + continue; + } + + const [first, second] = agents; + + // (b) duplicate pair within a workspace. + const pairKey = `${conv.workspaceId}|${first}|${second}`; + if (!pairIndex.has(pairKey)) pairIndex.set(pairKey, []); + pairIndex.get(pairKey).push(conv.id); + + // (c) the one only this script can check. + const expected = deriveConversationId(conv.workspaceId, first, second); + if (conv.id !== expected) { + mismatched.push({ id: conv.id, expected, workspaceId: conv.workspaceId, pair: [first, second] }); + } +} + +const duplicates = [...pairIndex.entries()] + .filter(([, ids]) => ids.length > 1) + .map(([key, ids]) => ({ pair: key, conversations: ids })); + +// ── report ─────────────────────────────────────────────────────────────────── +console.log(`Scanned ${conversations.size} 1:1 conversation(s).\n`); + +const section = (label, items, render) => { + if (items.length === 0) { + console.log(` ok ${label}: none`); + return 0; + } + console.log(` FAIL ${label}: ${items.length}`); + for (const item of items.slice(0, 20)) console.log(` ${render(item)}`); + if (items.length > 20) console.log(` ... and ${items.length - 20} more`); + return items.length; +}; + +const reserved = conversations.size - skipped.length; +console.log(` ${reserved} will be reserved by the backfill; ${skipped.length} skipped (not exactly two participants).\n`); + +let findings = 0; +section('(a) malformed rosters — skipped by the backfill, but worth knowing about', malformed, + (m) => `${m.id} (workspace ${m.workspaceId}, ${m.participants} participants)`); +findings += section('(b) duplicate pairs among reserved conversations — migration will ABORT', duplicates, + (d) => `${d.pair} -> ${d.conversations.join(', ')}`); +findings += section('(c) id does not match the current derivation — migration SUCCEEDS, then DMs 409', mismatched, + (m) => `${m.id} should be ${m.expected} (workspace ${m.workspaceId}, pair ${m.pair.join(' + ')})`); + +const byKind = skipped.reduce((acc, s) => ({ ...acc, [s.kind]: (acc[s.kind] ?? 0) + 1 }), {}); +if (skipped.length > 0) { + console.log(''); + console.log(' skipped breakdown (informational — none of these block the migration):'); + for (const [kind, count] of Object.entries(byKind)) console.log(` ${count} ${kind}`); + if (byKind['orphaned two-party']) { + console.log(' an orphaned two-party 1:1 is one whose peer agent was deleted;'); + console.log(' dm_participants cascades on agent delete, leaving a one-row roster.'); + } +} + +console.log(''); +if (findings === 0) { + console.log('Clean. Migration 0033 will apply, and no existing pair will start failing afterwards.'); + process.exit(0); +} + +console.log(`${findings} finding(s). Remediate before applying migration 0033.`); +console.log(''); +console.log(' (a) decide the correct roster for each conversation.'); +console.log(' (b) decide which conversation survives — a pair can hold only one reservation.'); +console.log(' (c) THIS IS THE QUIET ONE. The migration will not complain, but every'); +console.log(' subsequent DM between that pair returns 409. Either re-key the'); +console.log(' conversation to the derived id, or seed its reservation under the'); +console.log(' derived id, before deploying the code that reserves.'); +process.exit(1);