feat(traces): durable trace publication queue — wait for on-chain market, retry, confirm via indexer (#131)

.env.example

@@ -108,6 +108,13 @@ TRACE_AUDIT_MAX_RPC=500
 TRACE_AUDIT_MAX_FETCHES=100
 TRACE_AUDIT_MAX_BYTES=2097152
 
+# Trace publication queue (#131): pins traces, waits for the market to be
+# on-chain, publishes once, and confirms via the indexer's TracePublished event.
+TRACE_PUBLISH_ENABLED=true
+TRACE_PUBLISH_CRON=*/30 * * * * *
+TRACE_PUBLISH_MAX_ATTEMPTS=8
+TRACE_CONFIRM_TIMEOUT_MS=600000
+
 # ── Copy Trading ───────────────────────────────────────────────────────────
 COPY_TRADE_TTL_MINUTES=30
 COPY_TRADE_MAX_PENDING=5

docs/api.md

@@ -602,17 +602,24 @@ The old payload had `amount` (float USDC), `price` (0–1) and `probChange`
 (string). Frontends derive display values from `collateralRaw` and
 `priceYesBps`.
 
+> A trace's `verified` flag means its `trace_hash` was confirmed on-chain: set
+> by the indexer when it sees the matching `TracePublished` event (or by
+> `POST /traces/verify`), never from a write result alone and never in dry-run.
+
 ### `REASONING_PUBLISHED`
 
 Emitted when the indexer stores a reasoning-registry publish event.
+`action` is one of the backend's `create_market`, `buy_yes`, `buy_no`
+(`src/lib/trace-actions.ts`); traces published by other agents may use other
+strings, such as `buy` from the contracts repo's TypeScript agents.
 
 ```json
 {
   "onChainTraceId": "3",
   "onChainMarketId": "0",
   "traceId": "uuid or null",
   "agent": "G…",
-  "action": "buy",
+  "action": "buy_yes",
   "ipfsCid": "bafy…",
   "traceHash": "64-hex",
   "txHash": "64-hex",

docs/architecture.md

@@ -36,6 +36,19 @@ Backend categories are mapped to the contract enum by `toContractCategory()`.
 
 A trace is canonicalised once (sorted keys, same rules as `agents/core/trace.ts`), pinned byte for byte with Pinata's `pinFileToIPFS`, and `trace_hash` is the sha256 of those bytes. It is published to reasoning-registry once its market is on-chain. `POST /traces/verify` re-fetches the bytes by the on-chain CID and compares them with the on-chain hash, like `x402/trace-verification.ts`.
 
+Publication is durable (#131). Each trace that should be on-chain gets a `trace_publications` row holding the exact document, worked by `src/jobs/trace-publication.worker.ts` every `TRACE_PUBLISH_CRON` (default 30 s):
+
+| Status | Meaning |
+|---|---|
+| `AWAITING_PIN` | Pin the stored payload. A missing Pinata config parks the row for an hour; other pin errors retry. |
+| `AWAITING_MARKET` | Wait for `Market.onChainMarketId` (re-checked every 5 min, not counted as an attempt). A `CANCELLED` market fails the row with `MARKET_CANCELLED`. |
+| `READY` | Never publish twice: a trace already linked by the indexer, or the same agent + `trace_hash` among the market's 10 newest registry traces (`market_trace_at`), is confirmed without a write. Otherwise `publish_trace` is called once. |
+| `SUBMITTED` | Waiting for the indexer. After `TRACE_CONFIRM_TIMEOUT_MS` the submitted trace id is re-read on-chain before any retry, so a landed transaction isn't repeated. |
+| `CONFIRMED` | Set by the indexer's `TracePublished` event (matching `trace_hash`) or by the chain checks above. |
+| `FAILED` | `AgentNotRegistered` / `EmptyAction` / `EmptyCid`, a voided market, or `TRACE_PUBLISH_MAX_ATTEMPTS` reached. `lastError` says which. |
+
+Transient errors retry with backoff `min(2^attempts × 30 s, 1 h)`. Rows are claimed with `FOR UPDATE SKIP LOCKED` and leased, and chain calls run after the claim commits, so several processes can run the worker. `verified` is set only by indexer confirmation (or the duplicate checks, which read the same on-chain record) and by `POST /traces/verify`; dry-run never confirms a row or sets `verified`. The creation trace is pinned with `onChainMarketId: null` when its market is still pending and is **not** re-pinned later: that would change `trace_hash`; the `market_id` argument of `publish_trace` provides the link. Action strings live in `src/lib/trace-actions.ts`.
+
 ## Payments
 
 A daily pass is granted only after `payment-verification.service.ts` reads the transaction with Stellar RPC `getTransaction` and confirms all of the following:

docs/contract-requests.md

@@ -62,3 +62,10 @@ duplicate it, but today the adapter:
 If dry-run simulated (and returned the simulated result), the bindings were
 injectable, and `agentSell` existed, `src/services/chain.service.ts` here
 could become a thin wrapper around it. Its method names already match.
+
+## Note: `publish_trace` action strings differ between repos
+
+The backend publishes traces with `create_market`, `buy_yes` and `buy_no`
+(`src/lib/trace-actions.ts`). The contracts repo's TypeScript agents use
+`create_market` and `buy`. Consumers that group traces by action should treat
+`buy` as either direction, or the two sides should agree on one vocabulary.

prisma/migrations/20261017000008_trace_publications/migration.sql

@@ -0,0 +1,54 @@
+-- Migration: 20261017000008_trace_publications (#131)
+-- Adds the durable reasoning-trace publication queue:
+-- - TracePublicationStatus enum
+-- - trace_publications table (one row per trace that should be on-chain)
+--
+-- DATA MIGRATION (see CONTRIBUTING.md): the final INSERT backfills queue rows
+-- for traces that were pinned but never published. Only MARKET_CREATION traces
+-- are backfilled: trade traces don't record their direction, so their
+-- publish_trace action (buy_yes / buy_no) can't be reconstructed. Backfilled
+-- rows have a NULL payload because the trace is already pinned.
+
+CREATE TYPE "TracePublicationStatus" AS ENUM (
+  'AWAITING_PIN',
+  'AWAITING_MARKET',
+  'READY',
+  'SUBMITTED',
+  'CONFIRMED',
+  'FAILED'
+);
+
+CREATE TABLE "trace_publications" (
+  "id" TEXT NOT NULL,
+  "traceId" TEXT NOT NULL,
+  "action" TEXT NOT NULL,
+  "payload" JSONB,
+  "status" "TracePublicationStatus" NOT NULL DEFAULT 'AWAITING_PIN',
+  "attempts" INTEGER NOT NULL DEFAULT 0,
+  "lastError" TEXT,
+  "nextAttemptAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "submittedTxHash" TEXT,
+  "submittedTraceId" BIGINT,
+  "confirmedAt" TIMESTAMP(3),
+  "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  "updatedAt" TIMESTAMP(3) NOT NULL,
+
+  CONSTRAINT "trace_publications_pkey" PRIMARY KEY ("id")
+);
+
+CREATE UNIQUE INDEX "trace_publications_traceId_key" ON "trace_publications"("traceId");
+CREATE INDEX "trace_publications_status_nextAttemptAt_idx" ON "trace_publications"("status", "nextAttemptAt");
+
+ALTER TABLE "trace_publications"
+  ADD CONSTRAINT "trace_publications_traceId_fkey"
+  FOREIGN KEY ("traceId") REFERENCES "reasoning_traces"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- DATA: backfill pinned-but-unpublished market-creation traces.
+INSERT INTO "trace_publications" ("id", "traceId", "action", "payload", "status", "nextAttemptAt", "createdAt", "updatedAt")
+SELECT gen_random_uuid()::text, t."id", 'create_market', NULL, 'AWAITING_MARKET', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
+FROM "reasoning_traces" t
+WHERE t."onChainTraceId" IS NULL
+  AND t."traceHash" IS NOT NULL
+  AND t."ipfsCid" IS NOT NULL
+  AND t."decisionType" = 'MARKET_CREATION'
+ON CONFLICT ("traceId") DO NOTHING;

prisma/schema.prisma

@@ -316,6 +316,7 @@ model ReasoningTrace {
   subscriptions Subscription[]
   paymentEvents PaymentEvent[]
   audits        TraceAudit[]
+  publication   TracePublication?
 
   @@index([marketId])
   @@index([ipfsCid])
@@ -549,6 +550,40 @@ model ConstraintQuarantine {
   @@map("constraint_quarantine")
 }
 
+/// Lifecycle of a reasoning trace's on-chain publication (#131).
+enum TracePublicationStatus {
+  AWAITING_PIN
+  AWAITING_MARKET
+  READY
+  SUBMITTED
+  CONFIRMED
+  FAILED
+}
+
+/// Durable queue row: one per trace that should be published to
+/// reasoning-registry. Processed by src/jobs/trace-publication.worker.ts.
+model TracePublication {
+  id               String                 @id @default(uuid())
+  traceId          String                 @unique
+  action           String
+  /// Exact trace document; pinned bytes = canonicalBytes(payload). Null only
+  /// for backfilled rows whose trace was already pinned.
+  payload          Json?
+  status           TracePublicationStatus @default(AWAITING_PIN)
+  attempts         Int                    @default(0)
+  lastError        String?
+  nextAttemptAt    DateTime               @default(now())
+  submittedTxHash  String?
+  /// Trace id returned by the publish_trace write, used to re-check chain state.
+  submittedTraceId BigInt?
+  confirmedAt      DateTime?
+  createdAt        DateTime               @default(now())
+  updatedAt        DateTime               @updatedAt
+
+  trace ReasoningTrace @relation(fields: [traceId], references: [id], onDelete: Cascade)
+
+  @@index([status, nextAttemptAt])
+  @@map("trace_publications")
 /// Idempotency-Key records for state-changing endpoints (#123). One row per
 /// (scope, method, route, key); see src/middlewares/idempotency.middleware.ts.
 model IdempotencyKey {

src/agents/market-maker.agent.ts

@@ -6,7 +6,8 @@ import {
 } from '../services/market.service';
 import { generateReasoningTrace } from '../services/reasoning.service';
 import { createMarketOnChain } from '../services/market-chain.service';
-import { pinAndPublishTrace } from '../services/trace-publish.service';
+import { enqueueTracePublication } from '../services/trace-publish.service';
+import { TRACE_ACTIONS } from '../lib/trace-actions';
 import { prisma } from '../lib/prisma';
 import { logger } from '../lib/logger';
 import { config } from '../config';
@@ -110,16 +111,17 @@ export async function runMarketMakerCycle(opts?: { signal?: AbortSignal }): Prom
 
     signal.throwIfAborted();
 
-    // Step 7: Pin the trace and publish its hash to reasoning-registry
+    // Step 7: Queue the trace for pinning + publication to reasoning-registry.
+    // The queue publishes it once the market is on-chain (#131). The payload is
+    // fixed now: re-pinning later with a new onChainMarketId would change traceHash.
     try {
-      await pinAndPublishTrace({
+      await enqueueTracePublication({
         traceId: trace.id,
         payload: { ...tracePayload, onChainMarketId: onChainMarketId?.toString() ?? null },
-        action: 'create_market',
-        onChainMarketId,
+        action: TRACE_ACTIONS.createMarket,
       });
     } catch (err) {
-      logger.error({ err, traceId: trace.id }, 'Market Maker Agent: trace pin/publish failed');
+      logger.error({ err, traceId: trace.id }, 'Market Maker Agent: failed to queue trace publication');
     }
 
     logger.info({ marketId: market.id, traceId: trace.id }, 'Market Maker Agent: cycle complete');

src/config/index.ts

@@ -126,6 +126,20 @@ const envSchema = z
     TRACE_AUDIT_MAX_FETCHES: z.string().default('100').transform(Number),
     TRACE_AUDIT_MAX_BYTES: z.string().default('2097152').transform(Number),
 
+    // ── Trace publication queue (Issue #131) ──────────────────────────────
+    TRACE_PUBLISH_ENABLED: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'),
+    TRACE_PUBLISH_CRON: z.string().default('*/30 * * * * *'),
+    TRACE_PUBLISH_MAX_ATTEMPTS: z
+      .string()
+      .default('8')
+      .transform(Number)
+      .refine((n) => Number.isInteger(n) && n >= 1 && n <= 100, 'must be an integer between 1 and 100'),
+    TRACE_CONFIRM_TIMEOUT_MS: z
+      .string()
+      .default('600000')
+      .transform(Number)
+      .refine((n) => Number.isInteger(n) && n >= 10_000, 'must be an integer >= 10000'),
+
     // ── Copy Trading (Issue #151) ─────────────────────────────────────────
     COPY_TRADE_TTL_MINUTES: z.string().default('30').transform(Number),
     COPY_TRADE_MAX_PENDING: z.string().default('5').transform(Number),

src/cron/trace-publication.cron.ts

@@ -0,0 +1,33 @@
+import cron from 'node-cron';
+import { processTracePublications } from '../jobs/trace-publication.worker';
+import { logger } from '../lib/logger';
+import { config } from '../config';
+
+let isRunning = false;
+
+/**
+ * Trace publication cron (#131). Every TRACE_PUBLISH_CRON (default 30 s) it
+ * advances queued trace publications: pin, wait for the on-chain market,
+ * publish once, and re-check unconfirmed submissions. Rows are claimed with
+ * FOR UPDATE SKIP LOCKED, so several processes can run this safely.
+ */
+export function startTracePublicationCron(): void {
+  if (!config.TRACE_PUBLISH_ENABLED) {
+    logger.info('Trace publication cron is disabled by config (TRACE_PUBLISH_ENABLED=false)');
+    return;
+  }
+
+  cron.schedule(config.TRACE_PUBLISH_CRON, async () => {
+    if (isRunning) return;
+    isRunning = true;
+    try {
+      await processTracePublications();
+    } catch (err) {
+      logger.error({ err }, 'Trace publication cron encountered an error');
+    } finally {
+      isRunning = false;
+    }
+  });
+
+  logger.info({ cronSchedule: config.TRACE_PUBLISH_CRON }, 'Trace publication cron scheduled');
+}