From c1db8564c21694be77bd9cd8d9e0cd7e41b422e7 Mon Sep 17 00:00:00 2001 From: abayomiwav <277284682+abayomiwav@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:51:20 +0100 Subject: [PATCH 1/3] feat(db): add trace_publications queue table and status enum (#131) Durable queue row per reasoning trace that should be published to reasoning-registry: exact payload, status, attempts/backoff, submitted tx and trace id, confirmation time. The migration backfills pinned-but- unpublished MARKET_CREATION traces (commented as a data migration); trade traces can't be backfilled because their direction isn't recorded. Co-Authored-By: Claude Opus 5.5 --- .../migration.sql | 54 +++++++++++++++++++ prisma/schema.prisma | 37 +++++++++++++ 2 files changed, 91 insertions(+) create mode 100644 prisma/migrations/20261017000008_trace_publications/migration.sql diff --git a/prisma/migrations/20261017000008_trace_publications/migration.sql b/prisma/migrations/20261017000008_trace_publications/migration.sql new file mode 100644 index 0000000..c7a6f7e --- /dev/null +++ b/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; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 6b5c264..6c50b17 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -316,6 +316,7 @@ model ReasoningTrace { subscriptions Subscription[] paymentEvents PaymentEvent[] audits TraceAudit[] + publication TracePublication? @@index([marketId]) @@index([ipfsCid]) @@ -548,3 +549,39 @@ model ConstraintQuarantine { @@index([tableName, checkName]) @@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") +} From d443df16df551269f1e393edbcbc8a35047e5d74 Mon Sep 17 00:00:00 2001 From: abayomiwav <277284682+abayomiwav@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:51:20 +0100 Subject: [PATCH 2/3] feat(traces): queue trace publication until the market is on-chain and confirm via indexer (#131) - enqueueTracePublication replaces pinAndPublishTrace in runMarketMakerCycle (step 7) and executeTrade (step 5); trade traces are no longer fire-and- forget or skipped when the market isn't on-chain yet. - src/jobs/trace-publication.worker.ts: claims due rows with FOR UPDATE SKIP LOCKED, pins, waits for onChainMarketId (voided -> FAILED), checks the trace link and the registry's market_trace_at index before publishing so a hash is never published twice, retries with capped exponential backoff, parks AgentNotRegistered/EmptyAction/EmptyCid as FAILED, and never confirms in dry-run. Runs every TRACE_PUBLISH_CRON (default 30 s). - Indexer TracePublished now sets verified and confirms the queue row. - Action strings come from src/lib/trace-actions.ts. - New env: TRACE_PUBLISH_ENABLED, TRACE_PUBLISH_CRON, TRACE_PUBLISH_MAX_ATTEMPTS, TRACE_CONFIRM_TIMEOUT_MS (validated). Co-Authored-By: Claude Opus 5.5 --- .env.example | 7 + src/agents/market-maker.agent.ts | 14 +- src/config/index.ts | 14 ++ src/cron/trace-publication.cron.ts | 33 +++ src/jobs/trace-publication.worker.ts | 342 ++++++++++++++++++++++++++ src/lib/trace-actions.ts | 17 ++ src/server.ts | 2 + src/services/indexer.service.ts | 18 +- src/services/trace-publish.service.ts | 73 +++--- src/services/trade.service.ts | 23 +- 10 files changed, 485 insertions(+), 58 deletions(-) create mode 100644 src/cron/trace-publication.cron.ts create mode 100644 src/jobs/trace-publication.worker.ts create mode 100644 src/lib/trace-actions.ts diff --git a/.env.example b/.env.example index 5420fdc..2a09575 100644 --- a/.env.example +++ b/.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 diff --git a/src/agents/market-maker.agent.ts b/src/agents/market-maker.agent.ts index 145df5a..b73e741 100644 --- a/src/agents/market-maker.agent.ts +++ b/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'); diff --git a/src/config/index.ts b/src/config/index.ts index f64abc3..9d9390b 100644 --- a/src/config/index.ts +++ b/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), diff --git a/src/cron/trace-publication.cron.ts b/src/cron/trace-publication.cron.ts new file mode 100644 index 0000000..37b1203 --- /dev/null +++ b/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'); +} diff --git a/src/jobs/trace-publication.worker.ts b/src/jobs/trace-publication.worker.ts new file mode 100644 index 0000000..be08a30 --- /dev/null +++ b/src/jobs/trace-publication.worker.ts @@ -0,0 +1,342 @@ +/** + * Trace publication worker (#131). + * + * Drives every `trace_publications` row through + * AWAITING_PIN → AWAITING_MARKET → READY → SUBMITTED → CONFIRMED (or FAILED) + * + * - AWAITING_PIN: pin the stored payload; the pinned bytes define traceHash. + * - AWAITING_MARKET: wait until the market has an on-chain id (voided → FAILED). + * - READY: make sure the hash isn't already on-chain (trace row link, + * then the registry's per-market index), then publish once. + * - SUBMITTED: confirmation comes from the indexer's TracePublished event + * (indexer.service.ts); after TRACE_CONFIRM_TIMEOUT_MS the + * chain is re-checked before any retry, so a hash is never + * published twice. + * + * Chain calls always run outside any database transaction: rows are claimed + * with FOR UPDATE SKIP LOCKED, leased, committed, then worked on. + * Dry-run never confirms a row or sets `verified`. + */ +import { TracePublicationStatus } from '@prisma/client'; +import { prisma as defaultPrisma } from '../lib/prisma'; +import { logger } from '../lib/logger'; +import { config } from '../config'; +import { AppError } from '../middlewares/error.middleware'; +import { pinJson } from '../services/ipfs.service'; +import { publishTrace } from '../services/chain.service'; +import { ChainError } from '../services/stellar/errors'; +import { agentSigner } from '../services/stellar/clients'; +import { defaultRegistryReader, type RegistryReader } from '../services/trace-audit.service'; + +const S = TracePublicationStatus; + +/** Contract errors that retrying can't fix. */ +const PERMANENT_CONTRACT_ERRORS = new Set(['AgentNotRegistered', 'EmptyAction', 'EmptyCid']); +const BASE_BACKOFF_MS = 30_000; +const MAX_BACKOFF_MS = 60 * 60_000; +const MARKET_WAIT_MS = 5 * 60_000; +const DRY_RUN_RECHECK_MS = 6 * 60 * 60_000; +const CLAIM_LEASE_MS = 2 * 60_000; +/** How many of a market's newest registry traces the duplicate check reads. */ +const DUPLICATE_SCAN_LIMIT = 10; +/** Max state transitions handled for one row in one pass. */ +const MAX_STEPS = 6; + +export interface TracePublisher { + agent(): string; + publish(args: { + market_id: bigint; + action: string; + trace_hash: Buffer; + ipfs_cid: string; + }): Promise<{ mode: 'dry-run' | 'live'; value: bigint; txHash: string | null }>; +} + +export const defaultPublisher: TracePublisher = { + agent: () => agentSigner().publicKey as string, + publish: (args) => publishTrace({ agent: agentSigner().publicKey as string, ...args }), +}; + +export interface PublicationRow { + id: string; + traceId: string; + action: string; + payload: unknown; + status: TracePublicationStatus; + attempts: number; + submittedTraceId: bigint | null; +} + +export interface WorkerDeps { + prisma?: any; + publisher?: TracePublisher; + reader?: RegistryReader; + pin?: (payload: Record, name: string) => Promise<{ cid: string; hash: string }>; + dryRun?: boolean; + maxAttempts?: number; + confirmTimeoutMs?: number; + claim?: (batch: number, now: Date) => Promise; +} + +interface Ctx { + prisma: any; + publisher: TracePublisher; + reader: RegistryReader; + pin: NonNullable; + dryRun: boolean; + maxAttempts: number; + confirmTimeoutMs: number; + now: Date; +} + +function ctxFrom(deps: WorkerDeps, now: Date): Ctx { + return { + prisma: deps.prisma ?? defaultPrisma, + publisher: deps.publisher ?? defaultPublisher, + reader: deps.reader ?? defaultRegistryReader, + pin: deps.pin ?? ((payload, name) => pinJson(payload, name)), + dryRun: deps.dryRun ?? config.CHAIN_EXECUTION_MODE === 'dry-run', + maxAttempts: deps.maxAttempts ?? config.TRACE_PUBLISH_MAX_ATTEMPTS, + confirmTimeoutMs: deps.confirmTimeoutMs ?? config.TRACE_CONFIRM_TIMEOUT_MS, + now, + }; +} + +const after = (now: Date, ms: number) => new Date(now.getTime() + ms); + +export const backoffMs = (attempts: number) => + Math.min(2 ** attempts * BASE_BACKOFF_MS, MAX_BACKOFF_MS); + +const hexOf = (h: Buffer | Uint8Array | string) => + (typeof h === 'string' ? h : Buffer.from(h).toString('hex')).toLowerCase(); + +/** + * Claims due rows with FOR UPDATE SKIP LOCKED and leases them (pushes + * nextAttemptAt out) inside one short transaction, so concurrent workers never + * pick the same row. Chain work happens after this commits. + */ +export async function claimDuePublications(batch: number, now: Date, prisma: any = defaultPrisma): Promise { + return prisma.$transaction(async (tx: any) => { + const rows: PublicationRow[] = await tx.$queryRaw` + SELECT "id", "traceId", "action", "payload", "status", "attempts", "submittedTraceId" + FROM "trace_publications" + WHERE "status" IN ('AWAITING_PIN', 'AWAITING_MARKET', 'READY', 'SUBMITTED') + AND "nextAttemptAt" <= ${now} + ORDER BY "nextAttemptAt" + LIMIT ${batch} + FOR UPDATE SKIP LOCKED`; + if (rows.length > 0) { + await tx.tracePublication.updateMany({ + where: { id: { in: rows.map((r) => r.id) } }, + data: { nextAttemptAt: after(now, CLAIM_LEASE_MS) }, + }); + } + return rows; + }); +} + +/** Processes due rows. Safe to run from several processes at once. */ +export async function processTracePublications( + opts: { batch?: number; now?: Date } & WorkerDeps = {}, +): Promise<{ processed: number; byStatus: Record }> { + const now = opts.now ?? new Date(); + const ctx = ctxFrom(opts, now); + const rows = await (opts.claim ?? ((b, n) => claimDuePublications(b, n, ctx.prisma)))(opts.batch ?? 20, now); + const byStatus: Record = {}; + for (const row of rows) { + const status = await processPublication(row, ctx); + byStatus[status] = (byStatus[status] ?? 0) + 1; + } + if (rows.length > 0) logger.info({ processed: rows.length, byStatus }, 'Trace publication worker pass complete'); + return { processed: rows.length, byStatus }; +} + +/** Runs one row's state machine until it parks, finishes, or fails. */ +export async function processPublication(row: PublicationRow, ctxOrDeps: Ctx | WorkerDeps): Promise { + const ctx = 'now' in ctxOrDeps && ctxOrDeps.now instanceof Date + ? (ctxOrDeps as Ctx) + : ctxFrom(ctxOrDeps as WorkerDeps, new Date()); + let current: PublicationRow = { ...row }; + try { + for (let step = 0; step < MAX_STEPS; step++) { + const next = await advance(current, ctx); + if (!next) return current.status; + current = { ...current, ...next.row }; + if (next.park) return current.status; + } + return current.status; + } catch (err) { + return handleError(current, err, ctx); + } +} + +type Advance = { row: Partial; park: boolean } | null; + +async function update(ctx: Ctx, id: string, data: Record) { + await ctx.prisma.tracePublication.update({ where: { id }, data }); +} + +async function loadTrace(ctx: Ctx, traceId: string) { + return ctx.prisma.reasoningTrace.findUnique({ + where: { id: traceId }, + select: { + id: true, + traceHash: true, + ipfsCid: true, + onChainTraceId: true, + market: { select: { onChainMarketId: true, status: true } }, + }, + }); +} + +async function confirm(ctx: Ctx, row: PublicationRow, onChainTraceId: bigint | null, alreadyLinked: boolean): Promise { + if (!alreadyLinked && onChainTraceId !== null) { + await ctx.prisma.reasoningTrace.update({ + where: { id: row.traceId }, + data: { onChainTraceId, verified: true }, + }); + } + await update(ctx, row.id, { status: S.CONFIRMED, confirmedAt: ctx.now, lastError: null }); + return { row: { status: S.CONFIRMED }, park: true }; +} + +/** Newest-first scan of the market's registry index for this agent + hash. */ +async function findOnChainDuplicate(ctx: Ctx, marketId: bigint, traceHash: string): Promise { + const agent = ctx.publisher.agent().toUpperCase(); + const count = await ctx.reader.marketTraceCount(marketId); + for (let i = count - 1n, seen = 0; i >= 0n && seen < DUPLICATE_SCAN_LIMIT; i--, seen++) { + const id = await ctx.reader.marketTraceAt(marketId, i); + const onChain = await ctx.reader.getOnChainTraceOrNull(id); + if (!onChain) continue; + if (String(onChain.agent).toUpperCase() === agent && hexOf(onChain.trace_hash) === traceHash.toLowerCase()) { + return id; + } + } + return null; +} + +async function advance(row: PublicationRow, ctx: Ctx): Promise { + switch (row.status) { + case S.AWAITING_PIN: { + if (row.payload == null) { + const trace = await loadTrace(ctx, row.traceId); + if (trace?.traceHash && trace.ipfsCid) { + await update(ctx, row.id, { status: S.AWAITING_MARKET }); + return { row: { status: S.AWAITING_MARKET }, park: false }; + } + await update(ctx, row.id, { status: S.FAILED, lastError: 'NO_PAYLOAD' }); + return { row: { status: S.FAILED }, park: true }; + } + try { + const pinned = await ctx.pin(row.payload as Record, `oracledesk-trace-${row.traceId}`); + await ctx.prisma.reasoningTrace.update({ + where: { id: row.traceId }, + data: { ipfsCid: pinned.cid, traceHash: pinned.hash }, + }); + } catch (err) { + if (err instanceof AppError && err.code === 'IPFS_NOT_CONFIGURED') { + // Not a failure of this row: keep the payload and check back later. + await update(ctx, row.id, { lastError: 'IPFS_NOT_CONFIGURED', nextAttemptAt: after(ctx.now, MAX_BACKOFF_MS) }); + return { row: {}, park: true }; + } + throw err; + } + await update(ctx, row.id, { status: S.AWAITING_MARKET, lastError: null }); + return { row: { status: S.AWAITING_MARKET }, park: false }; + } + + case S.AWAITING_MARKET: { + const trace = await loadTrace(ctx, row.traceId); + if (!trace) { + await update(ctx, row.id, { status: S.FAILED, lastError: 'TRACE_NOT_FOUND' }); + return { row: { status: S.FAILED }, park: true }; + } + if (trace.market?.status === 'CANCELLED') { + await update(ctx, row.id, { status: S.FAILED, lastError: 'MARKET_CANCELLED' }); + return { row: { status: S.FAILED }, park: true }; + } + if (trace.market?.onChainMarketId == null) { + // Waiting isn't a failed attempt. + await update(ctx, row.id, { nextAttemptAt: after(ctx.now, MARKET_WAIT_MS) }); + return { row: {}, park: true }; + } + await update(ctx, row.id, { status: S.READY }); + return { row: { status: S.READY }, park: false }; + } + + case S.READY: { + const trace = await loadTrace(ctx, row.traceId); + if (!trace?.traceHash || !trace.ipfsCid || trace.market?.onChainMarketId == null) { + await update(ctx, row.id, { status: S.AWAITING_MARKET }); + return { row: { status: S.AWAITING_MARKET }, park: false }; + } + if (!ctx.dryRun) { + // Never publish a hash twice: trust the indexer link first, then chain state. + if (trace.onChainTraceId != null) return confirm(ctx, row, trace.onChainTraceId, true); + const dup = await findOnChainDuplicate(ctx, trace.market.onChainMarketId, trace.traceHash); + if (dup !== null) return confirm(ctx, row, dup, false); + } + const result = await ctx.publisher.publish({ + market_id: trace.market.onChainMarketId, + action: row.action, + trace_hash: Buffer.from(trace.traceHash, 'hex'), + ipfs_cid: trace.ipfsCid, + }); + if (result.mode === 'dry-run') { + logger.info({ traceId: row.traceId, simulatedTraceId: String(result.value) }, 'dry-run: trace publish simulated, row stays READY'); + await update(ctx, row.id, { nextAttemptAt: after(ctx.now, DRY_RUN_RECHECK_MS) }); + return { row: {}, park: true }; + } + await update(ctx, row.id, { + status: S.SUBMITTED, + submittedTxHash: result.txHash, + submittedTraceId: result.value, + nextAttemptAt: after(ctx.now, ctx.confirmTimeoutMs), + lastError: null, + }); + return { row: { status: S.SUBMITTED, submittedTraceId: result.value }, park: true }; + } + + case S.SUBMITTED: { + // The indexer didn't confirm within TRACE_CONFIRM_TIMEOUT_MS. Check chain + // state before trying again so a landed transaction isn't repeated. + const trace = await loadTrace(ctx, row.traceId); + if (!ctx.dryRun && trace?.onChainTraceId != null) return confirm(ctx, row, trace.onChainTraceId, true); + if (!ctx.dryRun && trace?.traceHash && row.submittedTraceId != null) { + const onChain = await ctx.reader.getOnChainTraceOrNull(row.submittedTraceId); + if (onChain && hexOf(onChain.trace_hash) === trace.traceHash.toLowerCase()) { + return confirm(ctx, row, row.submittedTraceId, false); + } + } + const attempts = row.attempts + 1; + if (attempts >= ctx.maxAttempts) { + await update(ctx, row.id, { status: S.FAILED, attempts, lastError: 'NOT_CONFIRMED' }); + return { row: { status: S.FAILED, attempts }, park: true }; + } + await update(ctx, row.id, { status: S.READY, attempts, nextAttemptAt: after(ctx.now, backoffMs(attempts)) }); + return { row: { status: S.READY, attempts }, park: true }; + } + + default: + return null; + } +} + +async function handleError(row: PublicationRow, err: unknown, ctx: Ctx): Promise { + const contractError = err instanceof ChainError ? err.contractError : null; + const message = contractError ?? (err instanceof Error ? err.message : String(err)); + if (contractError && PERMANENT_CONTRACT_ERRORS.has(contractError)) { + await update(ctx, row.id, { status: S.FAILED, lastError: contractError }); + logger.error({ traceId: row.traceId, contractError }, 'Trace publication failed permanently'); + return S.FAILED; + } + const attempts = row.attempts + 1; + if (attempts >= ctx.maxAttempts) { + await update(ctx, row.id, { status: S.FAILED, attempts, lastError: message }); + logger.error({ traceId: row.traceId, attempts, err }, 'Trace publication gave up after max attempts'); + return S.FAILED; + } + await update(ctx, row.id, { attempts, lastError: message, nextAttemptAt: after(ctx.now, backoffMs(attempts)) }); + logger.warn({ traceId: row.traceId, attempts, err }, 'Trace publication attempt failed; will retry'); + return row.status; +} diff --git a/src/lib/trace-actions.ts b/src/lib/trace-actions.ts new file mode 100644 index 0000000..be861ee --- /dev/null +++ b/src/lib/trace-actions.ts @@ -0,0 +1,17 @@ +/** + * reasoning-registry `publish_trace` action strings (#131). The single source + * of truth for the backend's agents; docs/api.md shows the same values. + * + * Note: the contracts repo's TypeScript agents use `buy` for trades, while this + * backend distinguishes `buy_yes` / `buy_no` (see docs/contract-requests.md). + */ +export const TRACE_ACTIONS = { + createMarket: 'create_market', + buyYes: 'buy_yes', + buyNo: 'buy_no', +} as const; + +export type TraceAction = (typeof TRACE_ACTIONS)[keyof typeof TRACE_ACTIONS]; + +export const tradeTraceAction = (direction: 'YES' | 'NO'): TraceAction => + direction === 'YES' ? TRACE_ACTIONS.buyYes : TRACE_ACTIONS.buyNo; diff --git a/src/server.ts b/src/server.ts index 3e17612..2236aef 100644 --- a/src/server.ts +++ b/src/server.ts @@ -7,6 +7,7 @@ import { prisma } from './lib/prisma'; import { startIngestionCron } from './cron/ingestion.cron'; import { startMonitorCron } from './cron/monitor.cron'; import { startTraceAuditCron } from './cron/trace-audit.cron'; +import { startTracePublicationCron } from './cron/trace-publication.cron'; import { startEventListener } from './services/indexer.service'; import { closeRedis } from './lib/redis'; @@ -46,6 +47,7 @@ async function bootstrap(): Promise { startIngestionCron(); startMonitorCron(); startTraceAuditCron(); + startTracePublicationCron(); io.on('connection', (socket) => { logger.info({ socketId: socket.id }, 'New client connected to socket.io'); diff --git a/src/services/indexer.service.ts b/src/services/indexer.service.ts index 40a232f..94e50fb 100644 --- a/src/services/indexer.service.ts +++ b/src/services/indexer.service.ts @@ -171,10 +171,20 @@ async function applySideEffects(event: DecodedEvent, emit: IndexerDeps['emit']): }, select: { id: true, onChainTraceId: true }, }); - if (trace && trace.onChainTraceId === null) { - await prisma.reasoningTrace.update({ - where: { id: trace.id }, - data: { onChainTraceId, publishTxHash: event.txHash }, + if (trace && (trace.onChainTraceId === null || trace.onChainTraceId === onChainTraceId)) { + // The event's trace_hash equals our stored traceHash, which is exactly + // what verification means for our own traces (#131). + if (trace.onChainTraceId === null) { + await prisma.reasoningTrace.update({ + where: { id: trace.id }, + data: { onChainTraceId, publishTxHash: event.txHash, verified: true }, + }); + } else { + await prisma.reasoningTrace.update({ where: { id: trace.id }, data: { verified: true } }); + } + await prisma.tracePublication.updateMany({ + where: { traceId: trace.id, status: { in: ['AWAITING_MARKET', 'READY', 'SUBMITTED'] } }, + data: { status: 'CONFIRMED', confirmedAt: new Date(), lastError: null }, }); } emit('REASONING_PUBLISHED', { diff --git a/src/services/trace-publish.service.ts b/src/services/trace-publish.service.ts index cf9645a..09e8a77 100644 --- a/src/services/trace-publish.service.ts +++ b/src/services/trace-publish.service.ts @@ -1,52 +1,47 @@ /** - * Reasoning trace lifecycle after generation: - * pin the canonical JSON → traceHash = sha256(pinned bytes) → - * reasoning_registry.publish_trace (dry-run simulates) → verify. + * Reasoning trace lifecycle after generation (#131): + * enqueue → pin the canonical JSON (traceHash = sha256(pinned bytes)) → + * wait for the market's on-chain id → reasoning_registry.publish_trace → + * confirmed by the indexer's TracePublished event → verified. + * See src/jobs/trace-publication.worker.ts. */ import { prisma } from '../lib/prisma'; import { logger } from '../lib/logger'; import { AppError } from '../middlewares/error.middleware'; import { verifyTraceHash, sha256Hex } from '../utils/hash.util'; -import { fetchIpfsBytes, pinJson } from './ipfs.service'; -import { getOnChainTrace, publishTrace } from './chain.service'; -import { agentSigner } from './stellar/clients'; +import { Prisma } from '@prisma/client'; +import { fetchIpfsBytes } from './ipfs.service'; +import { getOnChainTrace, printable } from './chain.service'; +import { processPublication } from '../jobs/trace-publication.worker'; +import type { TraceAction } from '../lib/trace-actions'; import type { TraceVerification } from '../types'; -export async function pinAndPublishTrace(params: { - traceId: string; - payload: Record; - action: string; - onChainMarketId: bigint | null; -}): Promise<{ cid: string; traceHash: string; onChainTraceId: bigint | null; publishTxHash: string | null }> { - const pinned = await pinJson(params.payload, `oracledesk-trace-${params.traceId}`); - - await prisma.reasoningTrace.update({ - where: { id: params.traceId }, - data: { ipfsCid: pinned.cid, traceHash: pinned.hash }, - }); - - if (params.onChainMarketId === null) { - logger.info({ traceId: params.traceId }, 'Trace pinned; market not on-chain yet, so not published'); - return { cid: pinned.cid, traceHash: pinned.hash, onChainTraceId: null, publishTxHash: null }; - } - - const result = await publishTrace({ - agent: agentSigner().publicKey as string, - market_id: params.onChainMarketId, - action: params.action, - trace_hash: Buffer.from(pinned.hash, 'hex'), - ipfs_cid: pinned.cid, +/** + * Records that a trace must be published (#131). The row is durable: if + * pinning, the market's on-chain creation, or the publish call isn't possible + * yet, src/jobs/trace-publication.worker.ts finishes it later. An immediate + * best-effort pass runs in the background so the common case stays fast. + */ +export async function enqueueTracePublication( + params: { traceId: string; payload: Record; action: TraceAction }, + deps: { prisma?: any; processNow?: boolean } = {}, +): Promise<{ id: string }> { + const db = deps.prisma ?? prisma; + const row = await db.tracePublication.upsert({ + where: { traceId: params.traceId }, + create: { + traceId: params.traceId, + action: params.action, + payload: printable(params.payload) as Prisma.InputJsonValue, + }, + update: {}, }); - - if (result.mode === 'dry-run') { - return { cid: pinned.cid, traceHash: pinned.hash, onChainTraceId: null, publishTxHash: null }; + if (deps.processNow ?? true) { + void processPublication(row, { prisma: db }).catch((err) => + logger.warn({ err, traceId: params.traceId }, 'Immediate trace publication pass failed; the worker will retry'), + ); } - - await prisma.reasoningTrace.update({ - where: { id: params.traceId }, - data: { onChainTraceId: result.value, publishTxHash: result.txHash }, - }); - return { cid: pinned.cid, traceHash: pinned.hash, onChainTraceId: result.value, publishTxHash: result.txHash }; + return { id: row.id }; } /** diff --git a/src/services/trade.service.ts b/src/services/trade.service.ts index 17e67d8..09c67a9 100644 --- a/src/services/trade.service.ts +++ b/src/services/trade.service.ts @@ -3,7 +3,8 @@ import { logger } from '../lib/logger'; import { AppError } from '../middlewares/error.middleware'; import type { KellyResult, TradePayload, KellyInput } from '../types'; import { agentBuy, Outcomes, quoteBuy } from './chain.service'; -import { pinAndPublishTrace } from './trace-publish.service'; +import { enqueueTracePublication } from './trace-publish.service'; +import { tradeTraceAction } from '../lib/trace-actions'; import { usdcFromNumber } from '../lib/amounts'; // ─── Risk constants ─── @@ -175,14 +176,18 @@ export async function executeTrade(payload: TradePayload): Promise<{ }), ]); - // ── Step 5: Publish the trace hash to reasoning-registry (non-blocking) ── - if (payload.traceId && payload.tracePayload && market.onChainMarketId !== null) { - pinAndPublishTrace({ - traceId: payload.traceId, - payload: payload.tracePayload, - action: payload.direction === 'YES' ? 'buy_yes' : 'buy_no', - onChainMarketId: market.onChainMarketId, - }).catch(err => logger.warn({ err }, 'Failed to pin/publish trade trace')); + // ── Step 5: Queue the trace for publication to reasoning-registry (#131) ── + // Durable: the worker pins, waits for the on-chain market and retries. + if (payload.traceId && payload.tracePayload) { + try { + await enqueueTracePublication({ + traceId: payload.traceId, + payload: payload.tracePayload, + action: tradeTraceAction(payload.direction), + }); + } catch (err) { + logger.error({ err, traceId: payload.traceId }, 'Failed to queue trade trace publication'); + } } await logAgentAction('TRADER', 'INFO', 'TRADE_EXECUTED', payload.marketId, { From c046882113908d113097c826a160f83591ae4cc8 Mon Sep 17 00:00:00 2001 From: abayomiwav <277284682+abayomiwav@users.noreply.github.com> Date: Sat, 17 Oct 2026 10:51:21 +0100 Subject: [PATCH 3/3] test+docs(traces): cover the publication queue; document lifecycle and actions (#131) 13 worker tests (pending market, backoff vs permanent errors, pin retry, missing Pinata config, already-linked and on-chain duplicates, SUBMITTED re-check, dry-run, voided market, max attempts, claim loop, idempotent enqueue) plus an indexer assertion that TracePublished sets verified and confirms the queue row. docs/architecture.md gains the lifecycle; docs/api.md fixes the REASONING_PUBLISHED action example and defines verified; docs/contract-requests.md notes the buy vs buy_yes/buy_no mismatch. Co-Authored-By: Claude Opus 5.5 --- docs/api.md | 9 +- docs/architecture.md | 13 ++ docs/contract-requests.md | 7 + tests/indexer.test.ts | 10 ++ tests/trace-publication.test.ts | 276 ++++++++++++++++++++++++++++++++ 5 files changed, 314 insertions(+), 1 deletion(-) create mode 100644 tests/trace-publication.test.ts diff --git a/docs/api.md b/docs/api.md index c888fe6..c6ac2c8 100644 --- a/docs/api.md +++ b/docs/api.md @@ -551,9 +551,16 @@ 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 { @@ -561,7 +568,7 @@ Emitted when the indexer stores a reasoning-registry publish event. "onChainMarketId": "0", "traceId": "uuid or null", "agent": "G…", - "action": "buy", + "action": "buy_yes", "ipfsCid": "bafy…", "traceHash": "64-hex", "txHash": "64-hex", diff --git a/docs/architecture.md b/docs/architecture.md index 7d574a2..269dfdc 100644 --- a/docs/architecture.md +++ b/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: diff --git a/docs/contract-requests.md b/docs/contract-requests.md index 9f78060..fb3db54 100644 --- a/docs/contract-requests.md +++ b/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. diff --git a/tests/indexer.test.ts b/tests/indexer.test.ts index a5a33ee..3b5a266 100644 --- a/tests/indexer.test.ts +++ b/tests/indexer.test.ts @@ -30,6 +30,7 @@ async function setup(events = FIXTURES.map(toEventResponse)) { ['m0', { id: 'm0', onChainMarketId: 0n, question: 'Market zero?', status: 'ACTIVE' }], ]), traces: [{ id: 't0', marketId: 'm0', market: { onChainMarketId: 0n }, traceHash: 'f4b58d4ac4068163f34bd9486f0563c70873afa5f64e915ab7e85af7dee15e4e', onChainTraceId: null as bigint | null }], + publicationUpdates: [] as Array<{ where: any; data: any }>, copyTrades: [] as any[], closedPositions: 0, }; @@ -55,6 +56,10 @@ async function setup(events = FIXTURES.map(toEventResponse)) { }) ?? null; }; (prisma.reasoningTrace.update as any) = async ({ where, data }: any) => Object.assign(db.traces.find((t) => t.id === where.id)!, data); + (prisma.tracePublication.updateMany as any) = async ({ where, data }: any) => { + db.publicationUpdates.push({ where, data }); + return { count: 1 }; + }; (prisma.copyTrade.findFirst as any) = async ({ where }: any) => { return db.copyTrades.find((ct) => { if (where.txHash && ct.txHash !== where.txHash) return false; @@ -113,6 +118,11 @@ test('indexes trades, resolutions and traces from recorded events', async () => assert.equal(published.onChainTraceId, '0'); assert.equal(published.traceId, 't0'); assert.equal(db.traces[0].onChainTraceId, 0n); + // #131: indexer confirmation verifies the trace and confirms its queue row. + assert.equal((db.traces[0] as any).verified, true); + assert.equal(db.publicationUpdates.length, 1); + assert.equal(db.publicationUpdates[0].where.traceId, 't0'); + assert.equal(db.publicationUpdates[0].data.status, 'CONFIRMED'); }); test('re-reading the same ledgers is idempotent', async () => { diff --git a/tests/trace-publication.test.ts b/tests/trace-publication.test.ts new file mode 100644 index 0000000..8065a71 --- /dev/null +++ b/tests/trace-publication.test.ts @@ -0,0 +1,276 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { setupTestEnv } from './helpers/env'; +import type { PublicationRow, TracePublisher } from '../src/jobs/trace-publication.worker'; +import type { RegistryReader } from '../src/services/trace-audit.service'; + +// config validates the environment on import, so load modules after setup. +setupTestEnv(); +async function load() { + const worker = await import('../src/jobs/trace-publication.worker'); + const service = await import('../src/services/trace-publish.service'); + const { ChainError } = await import('../src/services/stellar/errors'); + const { AppError } = await import('../src/middlewares/error.middleware'); + const { TRACE_ACTIONS } = await import('../src/lib/trace-actions'); + return { ...worker, ...service, ChainError, AppError, TRACE_ACTIONS }; +} + +const AGENT = 'GBRPYHIL2CI3FNQ4BXLFMNDLFJUNPU2HY3ZMFSHONUCEOASW7QC7OX2H'; +const HASH = 'ab'.repeat(32); + +interface Store { + pubs: Map; + traces: Map; + markets: Map; +} + +function makeDb(opts: { onChainMarketId?: bigint | null; marketStatus?: string; trace?: Record; pub?: Record } = {}) { + const store: Store = { pubs: new Map(), traces: new Map(), markets: new Map() }; + store.markets.set('m1', { id: 'm1', onChainMarketId: opts.onChainMarketId ?? null, status: opts.marketStatus ?? 'PENDING' }); + store.traces.set('t1', { id: 't1', marketId: 'm1', traceHash: null, ipfsCid: null, onChainTraceId: null, verified: false, ...opts.trace }); + store.pubs.set('p1', { + id: 'p1', traceId: 't1', action: 'create_market', payload: { decision: 'x' }, + status: 'AWAITING_PIN', attempts: 0, submittedTraceId: null, lastError: null, nextAttemptAt: new Date(0), + ...opts.pub, + }); + const prisma = { + tracePublication: { + update: async ({ where, data }: any) => Object.assign(store.pubs.get(where.id), data), + upsert: async ({ where, create }: any) => { + const existing = [...store.pubs.values()].find((p) => p.traceId === where.traceId); + if (existing) return existing; + const row = { id: `p${store.pubs.size + 1}`, status: 'AWAITING_PIN', attempts: 0, submittedTraceId: null, ...create }; + store.pubs.set(row.id, row); + return row; + }, + }, + reasoningTrace: { + findUnique: async ({ where }: any) => { + const t = store.traces.get(where.id); + if (!t) return null; + const m = store.markets.get(t.marketId); + return { ...t, market: m ? { onChainMarketId: m.onChainMarketId, status: m.status } : null }; + }, + update: async ({ where, data }: any) => Object.assign(store.traces.get(where.id), data), + }, + }; + return { store, prisma }; +} + +function fakePublisher(impl?: (args: any) => Promise): TracePublisher & { calls: any[] } { + const calls: any[] = []; + return { + calls, + agent: () => AGENT, + publish: async (args) => { + calls.push(args); + return impl ? impl(args) : { mode: 'live', value: 42n, txHash: 'cd'.repeat(32) }; + }, + }; +} + +function fakeReader(onChain: Array<{ id: bigint; agent: string; trace_hash: string }> = []): RegistryReader { + return { + traceCount: async () => BigInt(onChain.length), + marketTraceCount: async () => BigInt(onChain.length), + marketTraceAt: async (_m, i) => onChain[Number(i)].id, + getOnChainTraceOrNull: async (id) => onChain.find((t) => t.id === id) ?? null, + }; +} + +const pinOk = async () => ({ cid: 'QmPinned', hash: HASH }); + +function deps(prisma: any, extra: Record = {}) { + return { prisma, publisher: fakePublisher(), reader: fakeReader(), pin: pinOk, dryRun: false, maxAttempts: 5, confirmTimeoutMs: 600_000, ...extra }; +} + +const row = (store: Store): PublicationRow => ({ ...store.pubs.get('p1') }); + +test('trace for a pending market waits, then publishes once the market gets an on-chain id', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ onChainMarketId: null }); + const publisher = fakePublisher(); + + await processPublication(row(store), deps(prisma, { publisher })); + assert.equal(store.pubs.get('p1').status, 'AWAITING_MARKET'); + assert.equal(store.traces.get('t1').traceHash, HASH, 'pinned while waiting'); + assert.ok(store.pubs.get('p1').nextAttemptAt > new Date(), 'parked for later'); + assert.equal(store.pubs.get('p1').attempts, 0, 'waiting is not an attempt'); + assert.equal(publisher.calls.length, 0); + + store.markets.get('m1').onChainMarketId = 7n; + store.markets.get('m1').status = 'ACTIVE'; + await processPublication(row(store), deps(prisma, { publisher })); + assert.equal(store.pubs.get('p1').status, 'SUBMITTED'); + assert.equal(publisher.calls.length, 1); + assert.equal(publisher.calls[0].market_id, 7n); + assert.equal(publisher.calls[0].action, 'create_market'); + assert.equal(publisher.calls[0].trace_hash.toString('hex'), HASH); + assert.equal(store.pubs.get('p1').submittedTraceId, 42n); + assert.equal(store.traces.get('t1').verified, false, 'a write result alone does not verify'); +}); + +test('transient publish failure backs off and retries; AgentNotRegistered fails permanently', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const ready = { onChainMarketId: 7n, marketStatus: 'ACTIVE', trace: { traceHash: HASH, ipfsCid: 'QmPinned' }, pub: { status: 'READY' } }; + const { store, prisma } = makeDb(ready); + + const flaky = fakePublisher(async () => { throw new ChainError('publish_trace submission failed', null); }); + await processPublication(row(store), deps(prisma, { publisher: flaky })); + assert.equal(store.pubs.get('p1').status, 'READY'); + assert.equal(store.pubs.get('p1').attempts, 1); + assert.ok(store.pubs.get('p1').nextAttemptAt > new Date()); + + const denied = fakePublisher(async () => { throw new ChainError('publish_trace simulation failed', 'AgentNotRegistered'); }); + await processPublication(row(store), deps(prisma, { publisher: denied })); + assert.equal(store.pubs.get('p1').status, 'FAILED'); + assert.equal(store.pubs.get('p1').lastError, 'AgentNotRegistered'); +}); + +test('pin failure keeps the payload and pins on the next pass', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ onChainMarketId: null }); + const failingPin = async () => { throw new AppError(502, 'IPFS_UPLOAD_FAILED', 'Failed to pin document to IPFS'); }; + + await processPublication(row(store), deps(prisma, { pin: failingPin })); + assert.equal(store.pubs.get('p1').status, 'AWAITING_PIN'); + assert.equal(store.pubs.get('p1').attempts, 1); + assert.deepEqual(store.pubs.get('p1').payload, { decision: 'x' }, 'payload is kept for the retry'); + + await processPublication(row(store), deps(prisma)); + assert.equal(store.pubs.get('p1').status, 'AWAITING_MARKET'); + assert.equal(store.traces.get('t1').ipfsCid, 'QmPinned'); +}); + +test('missing Pinata config parks the row without counting an attempt', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb(); + const notConfigured = async () => { throw new AppError(503, 'IPFS_NOT_CONFIGURED', 'not set'); }; + await processPublication(row(store), deps(prisma, { pin: notConfigured })); + assert.equal(store.pubs.get('p1').status, 'AWAITING_PIN'); + assert.equal(store.pubs.get('p1').attempts, 0); + assert.equal(store.pubs.get('p1').lastError, 'IPFS_NOT_CONFIGURED'); +}); + +test('already-linked trace (indexer saw it first) is confirmed without publishing', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ + onChainMarketId: 7n, marketStatus: 'ACTIVE', + trace: { traceHash: HASH, ipfsCid: 'QmPinned', onChainTraceId: 5n }, + pub: { status: 'READY' }, + }); + const publisher = fakePublisher(); + await processPublication(row(store), deps(prisma, { publisher })); + assert.equal(store.pubs.get('p1').status, 'CONFIRMED'); + assert.equal(publisher.calls.length, 0); +}); + +test('on-chain duplicate found via market_trace_at is confirmed without publishing', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ + onChainMarketId: 7n, marketStatus: 'ACTIVE', + trace: { traceHash: HASH, ipfsCid: 'QmPinned' }, + pub: { status: 'READY' }, + }); + const publisher = fakePublisher(); + const reader = fakeReader([ + { id: 10n, agent: AGENT, trace_hash: HASH }, + { id: 11n, agent: 'GOTHERAGENT', trace_hash: 'ef'.repeat(32) }, + ]); + await processPublication(row(store), deps(prisma, { publisher, reader })); + assert.equal(store.pubs.get('p1').status, 'CONFIRMED'); + assert.equal(store.traces.get('t1').onChainTraceId, 10n); + assert.equal(store.traces.get('t1').verified, true); + assert.equal(publisher.calls.length, 0, 'the same hash is never published twice'); +}); + +test('a submission the indexer never confirmed is re-checked on-chain before retrying', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const submitted = { + onChainMarketId: 7n, marketStatus: 'ACTIVE', + trace: { traceHash: HASH, ipfsCid: 'QmPinned' }, + pub: { status: 'SUBMITTED', submittedTraceId: 42n }, + }; + // Landed: confirm from chain state. + let { store, prisma } = makeDb(submitted); + const publisher = fakePublisher(); + await processPublication(row(store), deps(prisma, { publisher, reader: fakeReader([{ id: 42n, agent: AGENT, trace_hash: HASH }]) })); + assert.equal(store.pubs.get('p1').status, 'CONFIRMED'); + assert.equal(store.traces.get('t1').onChainTraceId, 42n); + assert.equal(publisher.calls.length, 0); + + // Not landed: back to READY with backoff, still without publishing in this pass. + ({ store, prisma } = makeDb(submitted)); + await processPublication(row(store), deps(prisma, { publisher })); + assert.equal(store.pubs.get('p1').status, 'READY'); + assert.equal(store.pubs.get('p1').attempts, 1); + assert.equal(publisher.calls.length, 0); +}); + +test('dry-run never confirms or sets verified', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ + onChainMarketId: 7n, marketStatus: 'ACTIVE', + trace: { traceHash: HASH, ipfsCid: 'QmPinned', onChainTraceId: 5n }, + pub: { status: 'READY' }, + }); + const publisher = fakePublisher(async () => ({ mode: 'dry-run', value: 99n, txHash: null })); + const reader = fakeReader([{ id: 10n, agent: AGENT, trace_hash: HASH }]); + await processPublication(row(store), deps(prisma, { publisher, reader, dryRun: true })); + assert.equal(store.pubs.get('p1').status, 'READY'); + assert.equal(store.traces.get('t1').verified, false); + assert.ok(store.pubs.get('p1').nextAttemptAt > new Date()); +}); + +test('voided market moves AWAITING_MARKET to FAILED', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ marketStatus: 'CANCELLED', trace: { traceHash: HASH, ipfsCid: 'QmPinned' }, pub: { status: 'AWAITING_MARKET' } }); + const publisher = fakePublisher(); + await processPublication(row(store), deps(prisma, { publisher })); + assert.equal(store.pubs.get('p1').status, 'FAILED'); + assert.equal(store.pubs.get('p1').lastError, 'MARKET_CANCELLED'); + assert.equal(publisher.calls.length, 0); +}); + +test('retries stop after TRACE_PUBLISH_MAX_ATTEMPTS', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ + onChainMarketId: 7n, marketStatus: 'ACTIVE', + trace: { traceHash: HASH, ipfsCid: 'QmPinned' }, pub: { status: 'READY', attempts: 4 }, + }); + const flaky = fakePublisher(async () => { throw new Error('rpc timeout'); }); + await processPublication(row(store), deps(prisma, { publisher: flaky, maxAttempts: 5 })); + assert.equal(store.pubs.get('p1').status, 'FAILED'); + assert.equal(store.pubs.get('p1').lastError, 'rpc timeout'); +}); + +test('backoff is exponential and capped at one hour', async () => { + const { backoffMs } = await load(); + assert.equal(backoffMs(1), 60_000); + assert.equal(backoffMs(2), 120_000); + assert.equal(backoffMs(20), 60 * 60_000); +}); + +test('processTracePublications works through claimed rows', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb({ onChainMarketId: 7n, marketStatus: 'ACTIVE' }); + const publisher = fakePublisher(); + const res = await processTracePublications({ ...deps(prisma, { publisher }), claim: async () => [row(store)] }); + assert.equal(res.processed, 1); + assert.equal(res.byStatus.SUBMITTED, 1); + assert.equal(publisher.calls.length, 1); +}); + +test('enqueueTracePublication stores the payload and action once per trace', async () => { + const { processPublication, processTracePublications, backoffMs, enqueueTracePublication, ChainError, AppError, TRACE_ACTIONS } = await load(); + const { store, prisma } = makeDb(); + store.pubs.clear(); + const payload = { decision: 'buy', traceId: 't1' }; + const first = await enqueueTracePublication({ traceId: 't1', payload, action: TRACE_ACTIONS.buyYes }, { prisma, processNow: false }); + const again = await enqueueTracePublication({ traceId: 't1', payload, action: TRACE_ACTIONS.buyYes }, { prisma, processNow: false }); + assert.equal(first.id, again.id, 'enqueue is idempotent per trace'); + assert.equal(store.pubs.size, 1); + const stored = store.pubs.get(first.id); + assert.equal(stored.action, 'buy_yes'); + assert.deepEqual(stored.payload, payload); +});