Strengthened the disposable reorg rehearsal

server/market/canonical-inbox.ts

@@ -12,7 +12,7 @@ const pageSchema = z.object({ streamId: z.string().uuid(), previousAnchor: hex,
   head: z.number().int().nonnegative().safe(), caughtUp: z.boolean() }).strict();
 export type CanonicalInboxPage = z.infer<typeof pageSchema>;
 export type CanonicalInboxResult = { streamId: string; cursor: number; head: number; height: number; tip: string;
-  replayed: boolean; appendedEvents: z.infer<typeof eventSchema>[] };
+  replayed: boolean; appendedEvents: z.infer<typeof eventSchema>[]; reviewRequired?: boolean };
 
 export class CanonicalInboxError extends Error {
   readonly code: string;
@@ -127,4 +127,25 @@ export async function ingestCanonicalPage(client: PoolClient, raw: unknown): Pro
 /** Full append-only inbox verification. Expected to run under the caller's coherent audit snapshot. */
 export async function auditCanonicalInbox(client: PoolClient): Promise<void> {
   await readCanonicalHistory(client);
+  const reviews = await client.query(`SELECT key,value FROM wk_custody.custody_meta
+    WHERE key LIKE 'sale-reorg-review:%' ORDER BY key`);
+  for (const row of reviews.rows) {
+    let record: unknown;
+    try { record = JSON.parse(row.value); } catch { throw new CanonicalInboxError('sale-review-record-invalid'); }
+    const schema = z.object({ version: z.literal(1), streamId: z.string().uuid(), orderId: z.string().uuid(),
+      terminalTransactionId: hex, stateRoot: hex, eventSequence: z.number().int().positive().safe(), eventHash: hex }).strict();
+    const parsed = schema.safeParse(record), keyMatch = /^sale-reorg-review:([a-f0-9-]{36}):(\d+)$/.exec(row.key);
+    if (!parsed.success || !keyMatch || keyMatch[1] !== parsed.data.orderId || Number(keyMatch[2]) !== parsed.data.eventSequence) {
+      throw new CanonicalInboxError('sale-review-record-invalid');
+    }
+    const event = await client.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1',
+      [eventKey(parsed.data.eventSequence)]);
+    if (!event.rowCount) throw new CanonicalInboxError('sale-review-event-missing');
+    const disconnected = storedEvent(event.rows[0].value);
+    if (disconnected.type !== 'disconnect' || disconnected.eventHash !== parsed.data.eventHash ||
+        !disconnected.transactionIds.includes(parsed.data.terminalTransactionId)) {
+      throw new CanonicalInboxError('sale-review-event-mismatch');
+    }
+  }
+  if (reviews.rowCount) throw new CanonicalInboxError('posted-sale-terminal-disconnected');
 }

server/market/postgres-custody.ts

@@ -348,27 +348,46 @@ export class PostgresCustody {
     });
   }
   /** Disposable observer rehearsal only; input must come from Ledger.canonicalEvents(), never peers/HTTP. */
-  observeCanonicalEvents(page: unknown) {
+  async observeCanonicalEvents(page: unknown) {
     check(this.options.disposableSaleRehearsal === true, 'Canonical event inbox rehearsal is unavailable on this custody instance');
-    return this.transaction(true, async c => {
+    const result = await this.transaction(true, async c => {
       check(!this.diagnostics().mutationHold, 'Custody reconciliation hold');
       check(!(await c.query("SELECT 1 FROM wk_custody.custody_meta WHERE key='integrity_incident_v1'")).rowCount, 'Custody integrity incident unresolved');
       try {
         const result = await ingestCanonicalPage(c, page);
+        const reviews: { orderId: string; terminalTransactionId: string; stateRoot: string; eventSequence: number; eventHash: string }[] = [];
         for (const event of result.appendedEvents) {
           if (event.type !== 'disconnect') continue;
           for (const txid of event.transactionIds) {
             const terminal = await c.query('SELECT value FROM wk_custody.custody_meta WHERE key=$1', [`sale-rehearsal-tx:${txid}`]);
-            if (terminal.rowCount) throw new CustodyIntegrityError('posted-sale-terminal-disconnected');
+            if (terminal.rowCount) {
+              const link = JSON.parse(terminal.rows[0].value);
+              if (link.version !== 1 || typeof link.orderId !== 'string' || typeof link.stateRoot !== 'string') {
+                throw new CustodyIntegrityError('sale-terminal-index-invalid');
+              }
+              reviews.push({ orderId: link.orderId, terminalTransactionId: txid, stateRoot: link.stateRoot,
+                eventSequence: event.sequence, eventHash: event.eventHash });
+            }
           }
         }
-        return result;
+        for (const review of reviews) {
+          await c.query('INSERT INTO wk_custody.custody_meta(key,value) VALUES($1,$2) ON CONFLICT(key) DO NOTHING',
+            [`sale-reorg-review:${review.orderId}:${review.eventSequence}`, JSON.stringify({ version: 1,
+              streamId: result.streamId, ...review })]);
+        }
+        if (reviews.length) await recordIntegrityIncident(c, 'posted-sale-terminal-disconnected');
+        return { ...result, appendedEvents: [], reviewRequired: reviews.length > 0 };
       }
       catch (error) {
         if (error instanceof CanonicalInboxError) throw new CustodyIntegrityError(`canonical-inbox-${error.code}`);
         throw error;
       }
     });
+    if (result.reviewRequired) {
+      this.integrityHold = true;
+      this.auditState = { ...this.auditState, status: 'failed', lastAttemptAt: Date.now(), failureCode: 'posted-sale-terminal-disconnected' };
+    }
+    return result;
   }
   /** Consume one bounded page directly from this process's locally validated ledger; no HTTP/peer feed. */
   async observeLedgerCanonicalEvents(ledger: Ledger) {