server: commit operations before enqueueing and answer seq_no races with PREV_HASH_MISMATCH

The operation and receipt now commit before the queue send. A Postgres
23505 on operations_agent_id_seq_no_key is answered with 400
PREV_HASH_MISMATCH carrying the stored chain hash so hook retries can
resync. A failed enqueue after the commit is logged as
operation.enqueue_failed and the receipt is still returned.

Every producer now names its message: MessageQueue.send takes the
durable message id, and the BullMQ adapter passes it as the job id, so
the receipt's queue_message_id is the real queue identifier.

Claude-Session: https://claude.ai/code/session_01QoUjdc7HGKTsgj5GhG8HvF

AGENTS.md

@@ -56,6 +56,7 @@ This repository publishes the open-source server, console, integration catalog,
 - Keep production source files at or below 500 lines.
 - Keep functions focused on one ownership boundary.
 - Propagate unexpected errors to the CLI or request boundary.
+- Commit the operation and receipt before enqueueing. Log a failed enqueue as `operation.enqueue_failed` and still return the receipt; answer a `UNIQUE(agent_id, seq_no)` conflict with `PREV_HASH_MISMATCH` carrying the stored chain hash.
 - Keep technical claims executable through tests or generated validation.
 - Keep the mirrored Python package, distribution, and CLI version in `sdks/python/elydora/_version.py`; Setuptools metadata must read that literal dynamically.
 - Ship `sdks/python/elydora/py.typed` in every mirrored wheel and verify public annotations from an installed-wheel consumer.

packages/server/src/adapters/bullmq.ts

@@ -9,7 +9,6 @@
 import { Queue as BullQueue } from 'bullmq';
 import type { ConnectionOptions } from 'bullmq';
 import type { MessageQueue } from './interfaces.js';
-import { generateUUIDv7 } from '../utils/uuid.js';
 
 export const QUEUE_NAME = 'elydora-queue';
 
@@ -20,14 +19,14 @@ export class BullMQAdapter implements MessageQueue {
     this.queue = new BullQueue(QUEUE_NAME, { connection });
   }
 
-  async send(body: unknown): Promise<{ messageId: string }> {
+  async send(messageId: string, body: unknown): Promise<void> {
-    const job = await this.queue.add('message', body as Record<string, unknown>, {
+    await this.queue.add('message', body as Record<string, unknown>, {
+      jobId: messageId,
       attempts: 3,
       backoff: { type: 'exponential', delay: 1000 },
       removeOnComplete: { count: 100 },
       removeOnFail: { count: 500 },
     });
-    return { messageId: job.id ?? generateUUIDv7() };
   }
 
   async close(): Promise<void> {

packages/server/src/adapters/interfaces.ts

@@ -117,6 +117,6 @@ export interface Cache {
  * Mirrors the subset of Cloudflare Queue methods used by Elydora services.
  */
 export interface MessageQueue {
-  /** Enqueue a message. Returns the assigned message ID. */
+  /** Enqueue a message under the caller's durable message ID. */
-  send(body: unknown): Promise<{ messageId: string }>;
+  send(messageId: string, body: unknown): Promise<void>;
 }

packages/server/src/scheduler.ts

@@ -12,6 +12,7 @@ import { RedisCacheAdapter } from './adapters/redis-cache.js';
 import { BullMQAdapter } from './adapters/bullmq.js';
 import { createEpoch } from './services/epoch-service.js';
 import { DEFAULT_EPOCH_INTERVAL_MS } from './shared/index.js';
+import { generateUUIDv7 } from './utils/uuid.js';
 
 // ---------------------------------------------------------------------------
 // Initialize adapters
@@ -65,7 +66,7 @@ async function runEpochCreation(): Promise<void> {
 
       if (epoch) {
         try {
-          await queue.send({
+          await queue.send(generateUUIDv7(), {
             type: 'tsa',
             epoch_id: epoch.epoch_id,
             org_id: epoch.org_id,

packages/server/src/services/export-service.ts

@@ -99,7 +99,7 @@ export async function createExport(
     .run();
 
   // Enqueue the export job
-  await queue.send({
+  await queue.send(generateUUIDv7(), {
     type: 'export',
     export_id: exportId,
     org_id: orgId,

packages/server/src/services/operation-service.ts

@@ -201,20 +201,10 @@ export async function submitOperation(
   });
 
   // ------------------------------------------------------------------
-  // Step 10: Enqueue for async processing
+  // Step 10: Reserve the queue message ID; the send follows the commit
   // ------------------------------------------------------------------
-  const queueMessageResult = await queue.send({
+  const queueMessageId = generateUUIDv7();
-    type: 'operation',
-    operation_id: eor.operation_id,
-    org_id: eor.org_id,
-    agent_id: eor.agent_id,
-  });
-
-  // D1 batch insert: operation + receipt
   const receiptId = generateUUIDv7();
-  const queueMessageId = typeof queueMessageResult === 'object' && queueMessageResult !== null
-    ? String((queueMessageResult as Record<string, unknown>).messageId ?? receiptId)
-    : receiptId;
 
   // ------------------------------------------------------------------
   // Step 11: Generate EAR receipt
@@ -291,11 +281,45 @@ export async function submitOperation(
       .bind(receiptId, eor.operation_id, r2ReceiptKey, receivedAt),
   );
 
-  await db.batch(statements);
+  try {
+    await db.batch(statements);
+  } catch (error) {
+    if (!isSeqNoConflict(error)) throw error;
+    const winner = await db
+      .prepare('SELECT chain_hash FROM operations WHERE agent_id = ? ORDER BY seq_no DESC LIMIT 1')
+      .bind(eor.agent_id)
+      .first<{ chain_hash: string }>();
+    if (!winner) throw new Error('seq_no conflict without a stored operation.');
+    throw new AppError(400, 'PREV_HASH_MISMATCH', {
+      key: 'operation.prevHashMismatch',
+      params: { expected: winner.chain_hash, actual: eor.prev_chain_hash },
+    });
+  }
+
+  try {
+    await queue.send(queueMessageId, {
+      type: 'operation',
+      operation_id: eor.operation_id,
+      org_id: eor.org_id,
+      agent_id: eor.agent_id,
+    });
+  } catch (error) {
+    console.error('operation.enqueue_failed', {
+      operation_id: eor.operation_id,
+      queue_message_id: queueMessageId,
+      error: error instanceof Error ? error.message : String(error),
+    });
+  }
 
   return { receipt: ear };
 }
 
+function isSeqNoConflict(error: unknown): boolean {
+  if (typeof error !== 'object' || error === null) return false;
+  const { code, constraint } = error as { code?: unknown; constraint?: unknown };
+  return code === '23505' && constraint === 'operations_agent_id_seq_no_key';
+}
+
 // ---------------------------------------------------------------------------
 // Get operation
 // ---------------------------------------------------------------------------

packages/server/test/operation-conflict.test.mjs

@@ -0,0 +1,113 @@
+import assert from 'node:assert/strict';
+import { generateKeyPairSync, randomUUID } from 'node:crypto';
+import test from 'node:test';
+
+import { AppError } from '../dist/middleware/error-handler.js';
+import { submitOperation } from '../dist/services/operation-service.js';
+import { jcsCanonicalise, sha256Base64url, signEd25519 } from '../dist/utils/crypto.js';
+
+const GENESIS = 'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA';
+const WINNER = 'Rxlf4j36C3KvIQ3hWuOkX698BR5iDypUFuB70JjEuvM';
+
+function keyPair() {
+  const pair = generateKeyPairSync('ed25519');
+  return {
+    seed: pair.privateKey.export({ format: 'jwk' }).d,
+    publicKey: pair.publicKey.export({ format: 'jwk' }).x,
+  };
+}
+
+async function signedEor(agentId, orgId, kid, seed) {
+  const payload = { tool: 'Bash' };
+  const unsigned = {
+    op_version: '1.0',
+    operation_id: randomUUID(),
+    org_id: orgId,
+    agent_id: agentId,
+    issued_at: Date.now(),
+    ttl_ms: 30_000,
+    nonce: randomUUID(),
+    operation_type: 'ai.tool_use',
+    subject: { session_id: 'unit' },
+    action: { tool: 'Bash' },
+    payload,
+    payload_hash: await sha256Base64url(jcsCanonicalise(payload)),
+    prev_chain_hash: GENESIS,
+    agent_pubkey_kid: kid,
+  };
+  const signature = await signEd25519(seed, new TextEncoder().encode(jcsCanonicalise(unsigned)));
+  return { ...unsigned, signature };
+}
+
+function pgConflict() {
+  return Object.assign(new Error('duplicate key value violates unique constraint "operations_agent_id_seq_no_key"'), {
+    code: '23505',
+    constraint: 'operations_agent_id_seq_no_key',
+  });
+}
+
+function fakes(batchError, agentId, orgId, kid, publicKey) {
+  const sent = [];
+  const db = {
+    prepare: (sql) => ({
+      bind: () => ({
+        first: async () => {
+          if (sql.includes('FROM agents')) return { agent_id: agentId, org_id: orgId, status: 'active' };
+          if (sql.includes('FROM agent_keys')) return { kid, agent_id: agentId, public_key: publicKey, status: 'active' };
+          if (sql.includes('SELECT chain_hash, seq_no')) return null;
+          if (sql.includes('SELECT chain_hash FROM operations')) return { chain_hash: WINNER };
+          throw new Error(`unexpected query: ${sql}`);
+        },
+      }),
+    }),
+    batch: async () => {
+      if (batchError) throw batchError;
+      return [];
+    },
+  };
+  const r2 = { put: async () => undefined };
+  const kv = { get: async () => null, put: async () => undefined };
+  const queue = { send: async (messageId, body) => { sent.push({ messageId, body }); } };
+  return { db, r2, kv, queue, sent };
+}
+
+test('a seq_no conflict on insert reports the winning chain hash as PREV_HASH_MISMATCH', async () => {
+  const agent = keyPair();
+  const server = keyPair();
+  const eor = await signedEor('agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.seed);
+  const { db, r2, kv, queue, sent } = fakes(pgConflict(), 'agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.publicKey);
+
+  await assert.rejects(
+    submitOperation(db, r2, kv, queue, eor, server.seed),
+    (error) => error instanceof AppError
+      && error.statusCode === 400
+      && error.errorCode === 'PREV_HASH_MISMATCH'
+      && error.message.includes(`Expected prev_chain_hash "${WINNER}"`),
+  );
+  assert.equal(sent.length, 0);
+});
+
+test('other insert failures propagate unchanged and nothing is enqueued', async () => {
+  const agent = keyPair();
+  const server = keyPair();
+  const eor = await signedEor('agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.seed);
+  const { db, r2, kv, queue, sent } = fakes(new Error('connection reset'), 'agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.publicKey);
+
+  await assert.rejects(
+    submitOperation(db, r2, kv, queue, eor, server.seed),
+    (error) => error instanceof Error && !(error instanceof AppError) && error.message === 'connection reset',
+  );
+  assert.equal(sent.length, 0);
+});
+
+test('a committed operation is enqueued with its reserved queue message id', async () => {
+  const agent = keyPair();
+  const server = keyPair();
+  const eor = await signedEor('agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.seed);
+  const { db, r2, kv, queue, sent } = fakes(null, 'agent-conflict', 'org-conflict', 'agent-conflict-key-1', agent.publicKey);
+
+  const { receipt } = await submitOperation(db, r2, kv, queue, eor, server.seed);
+  assert.equal(sent.length, 1);
+  assert.equal(sent[0].messageId, receipt.queue_message_id);
+  assert.equal(sent[0].body.operation_id, eor.operation_id);
+});