fix(mercury): close 4 critical silent failures — discriminated union, atomic audit, real sovereignty, fail-closed (PR #108)

Real-money attack/loss vectors fixed in the mercury_payment executor:

1. Mercury post() helper now returns a discriminated PostResult — every
   failure mode (network / http / parse / 409 idempotency_collision) is
   surfaced to the executor instead of collapsing to null. runMercuryPayment
   and the audit row carry httpStatus + bodySnippet + failureKind so
   operators can diagnose without re-calling Mercury.

2. Executor maps Mercury's 2xx body.status into the audit vocabulary:
   sent/posted/delivered -> completed, pending -> in_progress,
   failed -> failed (refusal), requires_review -> pending_review.
   No more blanket "completed" stamp on 2xx-with-error-envelope.

3. Audit row is PRE-WRITTEN as in_flight BEFORE the Mercury call and
   UPDATED in place afterwards. Idempotency key is now deterministic on
   intent_id only (NOT attempt), so the partial unique index on
   (intent_id, idempotency_key) blocks duplicate placeholders across
   retries and Mercury de-dupes on the same key end-to-end. A retry that
   finds a prior in_flight row refuses with in_flight_unknown (the only
   safe default — Mercury state must be reconciled by an operator).

4. Chat surface (src/agents/tools/actions.ts::execute_payment) no longer
   passes a synthetic { decision: 'autonomous' } snapshot to
   runMercuryPayment. The chat tool factory has no access to the chat
   actor ChittyID and therefore cannot perform a real assessSovereignty()
   call, so it REFUSES Mercury payments with a "use the dashboard"
   message and audits the attempt. This closes the silent-bypass of the
   money-path sovereignty gate.

5. failIntent() failures in dispatch.ts are no longer swallowed by
   .catch(() => null). A new safeFailIntent() wrapper logs a stable
   "audit_write_failed_during_failIntent" token to console.error and
   re-throws so the daemon's outer loop catches and applies backoff.

6. account_slug is normalized (lowercase + [a-z0-9-]) BEFORE the KV
   lookup; if normalization changes the value the input is refused with
   invalid_account_slug — no silent fallback to a different token.

Tests: 8 new failure-path tests drive runMercuryPayment with an injected
fetch double (real Response objects — no mocks of Mercury or the DB).
Existing 3 DB integration tests untouched and still pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: chitcommit

meta/executors/dispatch.ts

@@ -41,14 +41,24 @@ function getSql(env: Env): NeonQueryFunction<false, false> {
 
 /**
  * Stable, content-addressable idempotency key.
- * Formula: sha256("{intent.id}:{attempt}:{intent.intentType}")
+ *
+ * Formula: sha256("{intent.id}:{intent.intentType}")
+ *
+ * The key is deterministic on `intent.id` (NOT `attempt`) so that:
+ *   1. The partial unique index on `cc_actions_log (intent_id, idempotency_key)`
+ *      prevents duplicate audit rows across daemon retries — every retry of
+ *      the same intent reuses the same key and the index rejects the second
+ *      INSERT.
+ *   2. Mercury de-dupes on the same value end-to-end — a retry after a Neon
+ *      blip cannot cause double-spend because Mercury sees the same key.
+ *   3. The pre-write in_flight row uses this key, and a successful run
+ *      UPDATEs that same row in place (see writePreAudit / updateAuditRow).
  */
 async function computeIdempotencyKey(
   intentId: string,
-  attempt: number,
   intentType: string,
 ): Promise<string> {
-  const data = new TextEncoder().encode(`${intentId}:${attempt}:${intentType}`);
+  const data = new TextEncoder().encode(`${intentId}:${intentType}`);
   const digest = await crypto.subtle.digest('SHA-256', data);
   return [...new Uint8Array(digest)]
     .map((b) => b.toString(16).padStart(2, '0'))
@@ -83,18 +93,23 @@ export async function dispatch(
 ): Promise<ExecutorResult> {
   const sql = getSql(env);
   const freshnessMs = options.freshnessMs ?? SOVEREIGNTY_FRESHNESS_MS;
+  const idempotencyKey = await computeIdempotencyKey(intent.id, intent.intentType);
 
-  // 1. Replay short-circuit: if any prior cc_actions_log row exists for this
-  //    intent that hit a terminal status ('completed' or 'failed'), the intent
-  //    has already been dispatched and the second call must NOT re-execute.
-  //    The unique partial index on (intent_id, idempotency_key) backs this
-  //    invariant for retry attempts; the latest-terminal-row lookup backs the
-  //    "intent already done" case.
+  // 1. Replay / safety lookup. With the key deterministic on intent_id, every
+  //    prior row for this intent shares the same idempotency_key. We look up
+  //    by (intent_id, idempotency_key) regardless of status because:
+  //      - 'completed' / 'failed' / 'pending_review' / 'in_progress' →
+  //        terminal-or-known; short-circuit and replay the prior outcome.
+  //      - 'in_flight' → the previous attempt's outcome is UNKNOWN. Mercury
+  //        may or may not have moved money. We MUST NOT re-call Mercury.
+  //        Refuse with `in_flight_unknown`; the operator runbook resolves by
+  //        querying Mercury directly using this idempotency key and then
+  //        manually setting the row to its true terminal state.
   const priorRows = (await sql`
     SELECT id, status, response_payload, error_message, idempotency_key, attempt
     FROM cc_actions_log
     WHERE intent_id = ${intent.id}::uuid
-      AND status IN ('completed', 'failed')
+      AND idempotency_key = ${idempotencyKey}
     ORDER BY executed_at DESC
     LIMIT 1
   `) as unknown as Array<{
@@ -107,29 +122,37 @@ export async function dispatch(
   }>;
   if (priorRows[0]) {
     const prior = priorRows[0];
+    if (prior.status === 'in_flight') {
+      const errMsg =
+        `prior attempt is in_flight (audit_log id=${prior.id}, idempotency_key=${idempotencyKey}) — ` +
+        `Mercury state is unknown; operator must reconcile before retry`;
+      console.error(`[meta/executors/dispatch] in_flight_unknown for intent ${intent.id}: ${errMsg}`);
+      return {
+        ok: false,
+        idempotencyKey,
+        actionLogId: prior.id,
+        error: errMsg,
+        replayed: true,
+      };
+    }
+    // Any other status → known terminal outcome, replay it.
     return {
       ok: prior.status === 'completed',
-      idempotencyKey: prior.idempotency_key,
+      idempotencyKey,
       actionLogId: prior.id,
       data: prior.response_payload ?? undefined,
       error: prior.error_message ?? undefined,
       replayed: true,
     };
   }
 
-  // 2. Compute attempt number (prior rows + 1) and idempotency key for the
-  //    new audit row. The partial unique index on (intent_id, idempotency_key)
-  //    prevents two concurrent dispatchers from writing duplicate rows for
-  //    the same attempt.
+  // 2. Compute attempt number. With the deterministic key, `attempt` is now
+  //    purely audit metadata (the unique partial index dedupes us, not the
+  //    attempt number). Still useful for operators inspecting retry history.
   const [{ count: priorCount } = { count: 0 }] = (await sql`
     SELECT COUNT(*)::int AS count FROM cc_actions_log WHERE intent_id = ${intent.id}::uuid
   `) as unknown as Array<{ count: number }>;
   const attempt = (priorCount ?? 0) + 1;
-  const idempotencyKey = await computeIdempotencyKey(
-    intent.id,
-    attempt,
-    intent.intentType,
-  );
 
   // 3. Re-reckon sovereignty if snapshot stale.
   let sovereignty: SovereigntyAssessmentSnapshot;
@@ -161,7 +184,7 @@ export async function dispatch(
         requestPayload: intent.payload,
         metadata: { reason: 'no_actor_for_reckon' },
       });
-      await failIntent(env, intent.id, errMsg).catch(() => null);
+      await safeFailIntent(env, intent.id, errMsg);
       return { ok: false, idempotencyKey, error: errMsg };
     }
     const result = await assessSovereignty(
@@ -191,7 +214,7 @@ export async function dispatch(
         requestPayload: intent.payload,
         metadata: { sovereignty },
       });
-      await failIntent(env, intent.id, refusal).catch(() => null);
+      await safeFailIntent(env, intent.id, refusal);
       return {
         ok: false,
         idempotencyKey,
@@ -208,7 +231,43 @@ export async function dispatch(
     throw new Error(errMsg);
   }
 
-  // 5. Execute.
+  // 5. PRE-WRITE the audit row as `in_flight` BEFORE invoking the executor.
+  //    This is the atomicity guarantee for the money path: if the executor
+  //    moves money and the post-update fails (e.g., Neon outage), the
+  //    in_flight row still exists and a retry will see it via the prior-row
+  //    lookup above and refuse with `in_flight_unknown`. The operator runbook
+  //    then reconciles by querying Mercury with `idempotencyKey`.
+  //
+  //    The partial unique index on (intent_id, idempotency_key) prevents two
+  //    concurrent dispatchers from racing to create the same in_flight row —
+  //    the second INSERT will fail with a unique violation. We let that throw
+  //    so the daemon's outer loop treats it as a retryable error.
+  let auditId: string;
+  try {
+    auditId = await writeAuditRow(sql, {
+      intentId: intent.id,
+      attempt,
+      idempotencyKey,
+      actionType: 'payment_in_flight',
+      targetType: 'intent',
+      targetId: intent.id,
+      description: `mercury_payment in_flight for intent ${intent.id}`,
+      status: 'in_flight',
+      errorMessage: null,
+      responsePayload: null,
+      requestPayload: intent.payload,
+      metadata: { sovereignty, canonicalUri: executor.canonicalUri, phase: 'pre_execute' },
+    });
+  } catch (err) {
+    // Pre-write failed — money has NOT moved. Surface to caller; daemon
+    // retries with backoff. failIntent is NOT called (the intent stays
+    // claimable on the next pass).
+    const errMsg = err instanceof Error ? err.message : String(err);
+    console.error(`[meta/executors/dispatch] pre-write audit failed for intent ${intent.id}: ${errMsg}`);
+    throw new Error(`audit_write_failed_pre_execute: ${errMsg}`);
+  }
+
+  // 6. Execute.
   const ctx: ExecutorContext = {
     env,
     sql,
@@ -222,42 +281,61 @@ export async function dispatch(
     runOutput = await executor.run(ctx);
   } catch (err) {
     const errMsg = err instanceof Error ? err.message : String(err);
-    const auditId = await writeAuditRow(sql, {
-      intentId: intent.id,
-      attempt,
-      idempotencyKey,
-      actionType: 'executor_error',
-      targetType: 'intent',
-      targetId: intent.id,
-      description: `executor threw: ${errMsg}`,
-      status: 'failed',
-      errorMessage: errMsg,
-      responsePayload: null,
-      requestPayload: intent.payload,
-      metadata: { sovereignty, canonicalUri: executor.canonicalUri },
-    });
+    // Executor threw — update the in_flight row to failed (NOT a new row;
+    // same idempotency key, same record). If the update itself throws, we
+    // surface to the daemon so the row remains `in_flight` and a retry will
+    // hit the `in_flight_unknown` branch.
+    try {
+      await updateAuditRow(sql, auditId, {
+        actionType: 'executor_error',
+        description: `executor threw: ${errMsg}`,
+        status: 'failed',
+        errorMessage: errMsg,
+        responsePayload: null,
+        metadata: { sovereignty, canonicalUri: executor.canonicalUri, phase: 'executor_threw' },
+      });
+    } catch (updateErr) {
+      const updateMsg = updateErr instanceof Error ? updateErr.message : String(updateErr);
+      console.error(
+        `[meta/executors/dispatch] AUDIT_UPDATE_FAILED_AFTER_EXECUTOR_THREW intent=${intent.id} key=${idempotencyKey}: ${updateMsg}`,
+      );
+      throw new Error(
+        `audit_update_failed_after_executor_threw: original=${errMsg}; update_error=${updateMsg}`,
+      );
+    }
     return { ok: false, idempotencyKey, actionLogId: auditId, error: errMsg };
   }
 
-  // 6. Write audit row.
-  const auditId = await writeAuditRow(sql, {
-    intentId: intent.id,
-    attempt,
-    idempotencyKey,
-    actionType: runOutput.actionType,
-    targetType: runOutput.targetType,
-    targetId: runOutput.targetId ?? null,
-    description: runOutput.description,
-    status: runOutput.status,
-    errorMessage: runOutput.errorMessage ?? null,
-    responsePayload: runOutput.responsePayload ?? null,
-    requestPayload: intent.payload,
-    metadata: {
-      sovereignty,
-      canonicalUri: executor.canonicalUri,
-      ...(runOutput.metadata ?? {}),
-    },
-  });
+  // 7. UPDATE the same audit row in place with the executor result. This is
+  //    the atomicity completion: in_flight → terminal status. Failure here
+  //    leaves the row in_flight, which a retry will treat as
+  //    `in_flight_unknown` (safe — Mercury de-dupes on idempotencyKey, so
+  //    operator reconciliation reveals the true state).
+  try {
+    await updateAuditRow(sql, auditId, {
+      actionType: runOutput.actionType,
+      description: runOutput.description,
+      status: runOutput.status,
+      errorMessage: runOutput.errorMessage ?? null,
+      responsePayload: runOutput.responsePayload ?? null,
+      metadata: {
+        sovereignty,
+        canonicalUri: executor.canonicalUri,
+        phase: 'post_execute',
+        ...(runOutput.metadata ?? {}),
+      },
+    });
+  } catch (err) {
+    const errMsg = err instanceof Error ? err.message : String(err);
+    console.error(
+      `[meta/executors/dispatch] AUDIT_UPDATE_FAILED_POST_EXECUTE intent=${intent.id} key=${idempotencyKey} executor_ok=${runOutput.ok}: ${errMsg}`,
+    );
+    // Surface to daemon — the row is still in_flight, so a retry will be
+    // refused as in_flight_unknown (correct, since money may have moved).
+    throw new Error(
+      `audit_update_failed_post_execute: executor_ok=${runOutput.ok}; update_error=${errMsg}`,
+    );
+  }
 
   return {
     ok: runOutput.ok,
@@ -268,6 +346,26 @@ export async function dispatch(
   };
 }
 
+/**
+ * failIntent wrapper that, if failIntent itself throws (e.g., Neon outage),
+ * (a) logs to console.error with a stable token for operator alerting and
+ * (b) re-throws so the daemon's outer loop catches and applies backoff.
+ * Replaces the prior `.catch(() => null)` pattern which silently dropped
+ * failure information.
+ */
+async function safeFailIntent(env: Env, intentId: string, reason: string): Promise<void> {
+  try {
+    await failIntent(env, intentId, reason);
+  } catch (err) {
+    const errMsg = err instanceof Error ? err.message : String(err);
+    // Stable token so log aggregators / Alchemist alerting can match.
+    console.error(
+      `[meta/executors/dispatch] audit_write_failed_during_failIntent intent=${intentId} reason="${reason}" error=${errMsg}`,
+    );
+    throw new Error(`failIntent_threw: intent=${intentId}: ${errMsg}`);
+  }
+}
+
 interface AuditRow {
   intentId: string;
   attempt: number;
@@ -283,6 +381,38 @@ interface AuditRow {
   metadata: Record<string, unknown>;
 }
 
+interface AuditUpdate {
+  actionType: string;
+  description: string;
+  status: string;
+  errorMessage: string | null;
+  responsePayload: Record<string, unknown> | null;
+  metadata: Record<string, unknown>;
+}
+
+/**
+ * UPDATE an existing audit row by id. Used to transition the in_flight
+ * placeholder into its terminal status after the executor returns. We
+ * intentionally do NOT touch intent_id / attempt / idempotency_key /
+ * request_payload — those were set at pre-write and are immutable.
+ */
+async function updateAuditRow(
+  sql: NeonQueryFunction<false, false>,
+  id: string,
+  patch: AuditUpdate,
+): Promise<void> {
+  await sql`
+    UPDATE cc_actions_log
+    SET action_type = ${patch.actionType},
+        description = ${patch.description},
+        status = ${patch.status},
+        error_message = ${patch.errorMessage},
+        response_payload = ${patch.responsePayload ? JSON.stringify(patch.responsePayload) : null}::jsonb,
+        metadata = ${JSON.stringify(patch.metadata)}::jsonb
+    WHERE id = ${id}::uuid
+  `;
+}
+
 async function writeAuditRow(
   sql: NeonQueryFunction<false, false>,
   row: AuditRow,