Make the market-maker cycle a resumable saga with explicit step state #152

Overview

runMarketMakerCycle() in src/agents/market-maker.agent.ts is OracleDesk's main write path: DB market → question pin and spec commitment → on-chain creation through the treasury → reasoning trace → trace pin and registry publication. The steps run sequentially in one try block, several failures are swallowed, and nothing records which steps finished. A crash or a Pinata/RPC hiccup leaves partial state that nothing revisits:

• a PENDING market with questionHash/metaUri set but no on-chain market, because createMarketOnChain threw after pinJson and the DB update;
• a live on-chain market (onChainMarketId set) with no published creation trace, because pinAndPublishTrace threw and the error was swallowed;
• a trace pinned before its market was on-chain, which pinAndPublishTrace never publishes later;
• in the worst case, an on-chain market whose DB row never received onChainMarketId because the process died between signAndSend() returning and prisma.market.update. A naive retry then creates a second market and spends a second 100 USDC seed from treasury capital.

This issue turns the cycle into a saga. Every step is idempotent, checkpoints its output in a persisted run record, and can be resumed by a resumer after a crash or transient failure. Compensating actions apply where they are safe.

Background & current behavior

Step by step, in runMarketMakerCycle():

1. aggregateSignals() → generateMarketQuestion(signals) → isDuplicateMarket(question) (each may end the cycle with null).
2. createMarketFromProposal(proposal) → PENDING Market plus AgentLog MARKET_CREATED.
3. createMarketOnChain(market) (src/services/market-chain.service.ts): resolutionHash(spec) → pinJson(questionDocument(...)) → prisma.market.update({ resolutionSpec, resolutionHash, questionHash, metaUri, seedAmountRaw }) → agentCreateMarket(...) → in live mode, update onChainMarketId, creationTxHash, status: 'ACTIVE'. Errors are caught in the agent and logged ("market stays PENDING").
4. generateReasoningTrace({...}) (src/services/reasoning.service.ts), which makes an LLM call for hedge conditions and creates a ReasoningTrace.
5. pinAndPublishTrace({ traceId, payload: {...tracePayload, onChainMarketId}, action: 'create_market', onChainMarketId }) (src/services/trace-publish.service.ts). Errors are caught and logged.

Related facts:

• pinJson is idempotent for identical bytes. questionDocument() is deterministic, but the trace payload includes timestamp: new Date().toISOString().
• In live mode a write() throw in chain.service.ts is ambiguous (the transaction may have landed).
• Callers: src/cron/ingestion.cron.ts and triggerMarketGeneration in src/controllers/market.controller.ts.
• Indexer (src/services/indexer.service.ts) stores chain_events but does not project MarketCreated. The market-core MarketCreated event has no question_hash, so it cannot be used to find a lost creation. get_market(id).question_hash/meta_uri can.

Goals

• A persisted market_generation_runs record with an explicit state machine and per-step inputs and outputs.
• Each step is idempotent and checkpoints its output before the next starts.
• Lost on-chain creations are detected and recovered by matching question_hash/meta_uri before any re-submission, so a seed is never spent twice.
• Pinned-but-unpublished traces are published once their market is on-chain, within the same saga.
• A resumer (startup + periodic) picks up incomplete runs, with bounded retries and exponential back-off, and moves runs that cannot continue to NEEDS_ATTENTION.
• Compensations: unreachable PENDING markets are marked CANCELLED with a reason. Nothing on-chain is ever compensated automatically.

Non-goals

• BullMQ or distributed locking (separate issue); a Postgres row lock is enough here.
• A general chain-write outbox (separate issue); only the minimal create-market recovery is added.
• Admin-supplied generation parameters (wave draft Record indexer ledger gaps instead of silently skipping past RPC retention #10) and the trader cycle.

Proposed implementation

1. Schema (new migration).

   enum GenerationRunState { STARTED SIGNALS_READY PROPOSED MARKET_PERSISTED QUESTION_PINNED CHAIN_SUBMITTED CHAIN_CONFIRMED TRACE_PERSISTED TRACE_PINNED TRACE_PUBLISHED COMPLETED SKIPPED NEEDS_ATTENTION CANCELLED }
   model MarketGenerationRun {
     id           String             @id @default(uuid())
     trigger      String             // 'cron' | 'admin' | 'resume'
     state        GenerationRunState @default(STARTED)
     attempts     Int                @default(0)
     nextAttemptAt DateTime?
     lockedBy     String?
     lockedUntil  DateTime?
     inputs       Json?              // signals snapshot ids, proposal
     outputs      Json               @default("{}") // { questionCid, questionHash, resolutionHash, chainTxHash, onChainMarketId, traceId, traceCid, traceHash, traceTxHash, onChainTraceId }
     marketId     String?            @unique
     lastError    Json?
     createdAt    DateTime           @default(now())
     updatedAt    DateTime           @updatedAt
     market Market? @relation(fields: [marketId], references: [id], onDelete: SetNull)
     @@index([state, nextAttemptAt])
     @@map("market_generation_runs")
   }

2. src/services/market-saga.service.ts (new). It owns the state machine.

   type StepFn = (run: RunCtx) => Promise<Partial<RunOutputs> & { next: GenerationRunState }>;
   const STEPS: Record<GenerationRunState, StepFn | undefined>;
   export async function startRun(trigger: 'cron'|'admin'): Promise<string>;
   export async function advance(runId: string, opts?: { maxSteps?: number }): Promise<MarketGenerationRun>;
   export async function resumeIncomplete(limit = 5): Promise<number>;

   advance claims the run with UPDATE … SET lockedBy, lockedUntil = now()+5min WHERE id = $1 AND (lockedUntil IS NULL OR lockedUntil < now()) RETURNING *, runs one step, and persists outputs + state in one update before moving on. Steps read inputs only from the run record and the DB.
3. Steps and their idempotency.
   • STARTED → SIGNALS_READY → PROPOSED: store a compact signal snapshot and the validated proposal in inputs. No signals or a duplicate → SKIPPED.
   • PROPOSED → MARKET_PERSISTED: if run.marketId is set, reuse it. Otherwise call createMarketFromProposal and set run.marketId in the same $transaction (refactor createMarketFromProposal to accept a tx client).
   • MARKET_PERSISTED → QUESTION_PINNED: split createMarketOnChain into prepareMarketCommitment(market, spec) (resolution hash, pin, DB update; idempotent because the bytes are deterministic, and if market.questionHash is already set, verify instead of re-pinning) and submitMarketCreation(market).
   • QUESTION_PINNED → CHAIN_SUBMITTED → CHAIN_CONFIRMED: before submitting, run lost-creation recovery: scan market_count() downward over the last SAGA_RECOVERY_SCAN ids (default 50) with getOnChainMarket(id) and match question_hash === outputs.questionHash and creator === CONTRACTS.treasury. A match means the market already exists, so adopt the id. Otherwise call agentCreateMarket. In live mode, record chainTxHash before awaiting confirmation (add an optional onBuilt(hash) callback to write() in chain.service.ts). On an ambiguous ChainError, stay at CHAIN_SUBMITTED and confirm on the next attempt via getTransaction(chainTxHash) plus recovery, never by blind re-submission. Dry-run reaches CHAIN_CONFIRMED with onChainMarketId: null.
   • CHAIN_CONFIRMED → TRACE_PERSISTED: if outputs.traceId exists, reuse it. Otherwise call generateReasoningTrace. Store the exact tracePayload (including its timestamp and traceId) in outputs.tracePayload so later pins hash identical bytes.
   • TRACE_PERSISTED → TRACE_PINNED: pinJson(outputs.tracePayload). Compare with an existing traceHash if set.
   • TRACE_PINNED → TRACE_PUBLISHED: only when onChainMarketId is set. Before publishing, check market_trace_count(id) plus market_trace_at for an existing trace with the same trace_hash (recovery, as above), then publishTrace. Dry-run → COMPLETED without publishing.
     Split pinAndPublishTrace in trace-publish.service.ts into pinTrace and publishPinnedTrace so the saga can checkpoint between them. Keep pinAndPublishTrace as a wrapper for the trader path.
4. Retry and back-off. On a step error, increment attempts, store lastError (code, message, step), and set nextAttemptAt = now + min(2^attempts × 30 s, 1 h). After SAGA_MAX_ATTEMPTS (default 8), set NEEDS_ATTENTION. Permanent errors (IPFS_NOT_CONFIGURED, TradeCapExceeded, DailyCapExceeded) go straight to NEEDS_ATTENTION.
5. Compensation. A run in NEEDS_ATTENTION before CHAIN_SUBMITTED whose market is PENDING gets an admin action POST /admin/generation-runs/:id/cancel that sets the market to CANCELLED (with an AgentLog) and the run to CANCELLED. After CHAIN_SUBMITTED, cancelling is refused, because the on-chain state belongs to the contracts.
6. Wiring.
   • runMarketMakerCycle() becomes const id = await startRun(trigger); return summarize(await advance(id, { maxSteps: Infinity })), keeping the MarketCycleResult shape for market.controller.ts.
   • src/cron/ingestion.cron.ts also calls resumeIncomplete() before starting a new run, and src/server.ts calls it once at boot.
   • Admin routes GET /admin/generation-runs and GET /admin/generation-runs/:id, documented in docs/api.md.
7. Fault injection. A test-only setSagaFaultHook(step, fn) seam throws before or after a named step.

Edge cases & pitfalls

• Two processes resuming the same run: the conditional lock update must be the only way to claim a run. Lock expiry must be longer than the slowest step (LLM plus chain confirmation).
• generateReasoningTrace writes the DB row and calls the LLM. If the process crashes after the insert but before the checkpoint, a retry creates a second trace. Wrap it to accept a pre-generated id (traceId stored in outputs before the insert), or look up an existing trace by (marketId, decisionType='MARKET_CREATION').
• Recovery scans must check creator, since another creator could reuse the same question_hash.
• Schema evolution: inputs/outputs are JSON. Version them (outputs.v = 1) so future steps can migrate them.

Testing plan

• tests/market-saga.test.ts (new), with in-memory Prisma (as in tests/indexer.test.ts), a fake IpfsTransport, and stubbed chain and LLM calls:
  • happy path reaches COMPLETED with all outputs recorded
  • crash after QUESTION_PINNED resumes without re-pinning different bytes
  • ambiguous agent_create_market failure is recovered by question_hash scan, not resubmitted
  • trace pinned before market on-chain is published after CHAIN_CONFIRMED
  • existing on-chain trace with same hash is adopted on resume
  • permanent error moves run to NEEDS_ATTENTION
  • concurrent advance calls cannot claim the same run
  • fault injected at every step boundary eventually COMPLETES (property-style loop over steps)
• Commands: npx prisma migrate dev --name market_generation_runs, npm run build, npm test, npm run check:contracts.

Acceptance criteria

• market_generation_runs persists state, inputs, outputs, attempts, locks and errors for every cycle.
• Each step checkpoints its output before the next starts, and re-running any step after a simulated crash produces no duplicate side effects.
• On-chain creation recovery matches question_hash and the treasury creator before any re-submission, so no seed is ever spent twice in tests.
• Traces pinned before the market was on-chain are published once it is, and existing on-chain traces with the same hash are adopted.
• A resumer runs at boot and on each ingestion tick, with exponential back-off and a NEEDS_ATTENTION terminal state.
• createMarketOnChain and pinAndPublishTrace are split into checkpointable parts, and existing callers keep working.
• Admin endpoints list and inspect runs and cancel pre-chain runs, documented in docs/api.md.
• Fault-injection tests cover every step boundary.
• npm run build, npm test and npm run check:contracts pass.

Resources

• src/agents/market-maker.agent.ts, src/services/market-chain.service.ts, src/services/trace-publish.service.ts
• src/services/market.service.ts, src/services/reasoning.service.ts, src/services/ipfs.service.ts, src/services/chain.service.ts
• src/cron/ingestion.cron.ts, src/controllers/market.controller.ts, src/server.ts, prisma/schema.prisma, docs/architecture.md (ADR 0001 section)
• docs/wave-issues/10-generation-params.md, docs/wave-issues/13-persistent-generation-jobs.md (related, not overlapping)
• Saga pattern: https://microservices.io/patterns/data/saga.html
• Stellar RPC getTransaction: https://developers.stellar.org/docs/data/apis/rpc/api-reference/methods/getTransaction
• PostgreSQL row locking: https://www.postgresql.org/docs/16/explicit-locking.html#LOCKING-ROWS

Generated by Claude Code