feat: propagate request and job correlation ids through logs, agent logs and errors

.env.example

@@ -74,6 +74,9 @@ RESOLUTION_SIGNERS=
 RESOLUTION_THRESHOLD=1
 RESOLUTION_DISPUTE_WINDOW_SECS=86400
 
+# Accept a well-formed inbound X-Request-Id; set false without a trusted proxy.
+TRUST_INBOUND_REQUEST_ID=true
+
 # Event indexer (Stellar RPC getEvents poller).
 INDEXER_ENABLED=true
 INDEXER_POLL_MS=5000

docs/api.md

@@ -14,9 +14,11 @@ Every response, success or error:
 
 ```json
 { "ok": true,  "data": { }, "error": null, "meta": { } }
-{ "ok": false, "data": null, "error": { "code": "SOME_CODE", "message": "Readable text", "details": { } } }
+{ "ok": false, "data": null, "error": { "code": "SOME_CODE", "message": "Readable text", "details": { }, "requestId": "req_018f..." } }
 ```
 
+Every response carries an `X-Request-Id` header (exposed via CORS), and error envelopes repeat it as `error.requestId`; quote it when reporting a problem. A client may send its own `X-Request-Id` (8 to 64 characters of `A-Z a-z 0-9 . _ : -`); any other value is ignored and a new id is generated.
+
 `meta` is optional. List endpoints put pagination there:
 `{ "page": 1, "limit": 20, "total": 57, "totalPages": 3 }`.
 

docs/architecture.md

@@ -6,6 +6,10 @@ The backend is the API, indexer and LLM pipeline around the OracleDesk Soroban c
 
 Express 5 with helmet, hpp, an env CORS allowlist, rate limits (global, `/auth/*`, `/markets/generate`), JSON bodies capped at 1 MB, and pino logging with secrets redacted. BigInts serialise as decimal strings through the app's JSON replacer. Every response uses the `{ ok, data, error, meta }` envelope. The full contract with the frontend is [api.md](api.md).
 
+## Correlation ids
+
+`src/lib/context.ts` keeps an `AsyncLocalStorage` context with a `requestId` and, where relevant, a `cycleId` or `jobId`. The first Express middleware assigns it: a well-formed inbound `X-Request-Id` (`[A-Za-z0-9._:-]{8,64}`, only when `TRUST_INBOUND_REQUEST_ID=true`) is kept, anything else is replaced and never logged. The id is echoed in the `X-Request-Id` response header and in `error.requestId`. Each cron tick and indexer poll runs in its own context with a fresh `cycleId`, and `POST /markets/generate` runs the cycle with a `jobId` plus the originating `requestId`. A pino mixin adds these ids to every log line, and `writeAgentLog()` (`src/lib/agent-log.ts`, the only writer of `AgentLog` rows) stores them in `data.correlation`. To follow one request or cycle, search the logs for `requestId=<id>` (or the JSON field).
+
 ## Login (SEP-10 style)
 
 `auth.service.ts` builds a challenge transaction with `WebAuth.buildChallengeTx`, signed by `AUTH_SIGNING_SECRET` (a key that holds nothing and is not the agent key). It stores the transaction hash in Redis for 5 minutes. `/auth/verify` reads the challenge, enforces its time bounds without the SDK's grace period, consumes it atomically (`GETDEL`), checks the user's signature with `verifyChallengeTxSigners`, and issues a JWT. Admin rights are an env allowlist of G-addresses checked server-side.

src/agents/market-maker.agent.ts

@@ -8,7 +8,7 @@ import { generateReasoningTrace } from '../services/reasoning.service';
 import { createMarketOnChain } from '../services/market-chain.service';
 import { enqueueTracePublication } from '../services/trace-publish.service';
 import { TRACE_ACTIONS } from '../lib/trace-actions';
-import { prisma } from '../lib/prisma';
+import { writeAgentLog } from '../lib/agent-log';
 import { logger } from '../lib/logger';
 import { config } from '../config';
 import { deadline } from '../lib/resilience';
@@ -135,13 +135,11 @@ export async function runMarketMakerCycle(opts?: { signal?: AbortSignal }): Prom
   } catch (err) {
     logger.error({ err }, 'Market Maker Agent: unhandled cycle error');
 
-    await prisma.agentLog.create({
-      data: {
-        agentType: 'MARKET_MAKER',
-        level:     'ERROR',
-        action:    'CYCLE_FAILED',
-        error:     String(err),
-      },
+    await writeAgentLog({
+      agentType: 'MARKET_MAKER',
+      level:     'ERROR',
+      action:    'CYCLE_FAILED',
+      error:     String(err),
     }).catch(() => {});
 
     throw err; // Re-throw so the job tracker records FAILED status

src/agents/trader.agent.ts

@@ -5,6 +5,7 @@ import { checkCorrelation } from '../services/correlation.service';
 import { generateReasoningTrace } from '../services/reasoning.service';
 import { detectHedgeOpportunities, autoHedge, checkDailyDrawdown, enforceStopLoss } from '../services/hedging.service';
 import { prisma } from '../lib/prisma';
+import { writeAgentLog } from '../lib/agent-log';
 import { logger } from '../lib/logger';
 import { config } from '../config';
 import { deadline } from '../lib/resilience';
@@ -99,9 +100,7 @@ export async function runTraderCycle(opts?: { signal?: AbortSignal }): Promise<v
     logger.info('Trader Agent: cycle complete');
   } catch (err) {
     logger.error({ err }, 'Trader Agent: unhandled cycle error');
-    await prisma.agentLog.create({
-      data: { agentType: 'TRADER', level: 'ERROR', action: 'CYCLE_FAILED', error: String(err) },
-    });
+    await writeAgentLog({ agentType: 'TRADER', level: 'ERROR', action: 'CYCLE_FAILED', error: String(err) });
   } finally {
     handle.cleanup?.();
   }

src/app.ts

@@ -2,11 +2,12 @@ import express from 'express';
 import cors from 'cors';
 import helmet from 'helmet';
 import compression from 'compression';
-import morgan from 'morgan';
 import hpp from 'hpp';
 
 import { config } from './config';
 import { logger } from './lib/logger';
+import { acceptInboundId, currentContext, newId, runInContext, runWithContext } from './lib/context';
+import { sendError } from './utils/response.util';
 import { errorMiddleware } from './middlewares/error.middleware';
 import { globalLimiter } from './middlewares/rate-limit.middleware';
 import { abortMiddleware } from './middlewares/abort.middleware';
@@ -40,7 +41,7 @@ app.use(cors({
   credentials: true,
   // #123: let browsers send Idempotency-Key and read Idempotent-Replayed.
   allowedHeaders: ['Content-Type', 'Authorization', 'Idempotency-Key'],
-  exposedHeaders: ['Idempotent-Replayed', 'Retry-After'],
+  exposedHeaders: ['Idempotent-Replayed', 'Retry-After', 'X-Request-Id'],
 }));
 
 // ─── Rate Limiting ───
@@ -49,15 +50,34 @@ app.use(globalLimiter);
 // ─── Body Parsing + Compression ───
 app.use(compression());
 app.use(express.json({ limit: '1mb' }));
+// Stream-based body parsing can resume handlers outside the request's async
+// context; re-enter it so everything downstream stays correlated.
+app.use((_req, res, next) => runInContext(res.locals.requestContext, next));
 
 // ─── HTTP Logging ───
-// Paths only: morgan's "combined" format would log query strings and referrers.
-app.use(
-  morgan(':method :url :status :res[content-length] - :response-time ms', {
-    stream: { write: (msg) => logger.info(msg.trim()) },
-    skip:   () => config.NODE_ENV === 'test',
-  }),
-);
+// Paths only: the route template, never the query string or referrer.
+app.use((req, res, next) => {
+  if (config.NODE_ENV !== 'test') {
+    const ctx = currentContext();
+    res.on('finish', () => {
+      const log = () =>
+        logger.info(
+          {
+            method: req.method,
+            route: req.route ? req.baseUrl + req.route.path : 'unmatched',
+            status: res.statusCode,
+            durationMs: Date.now() - (ctx?.startedAt ?? Date.now()),
+            contentLength: res.getHeader('content-length'),
+          },
+          'http',
+        );
+      // 'finish' listeners may not run inside the request's context.
+      if (ctx) runInContext(ctx, log);
+      else log();
+    });
+  }
+  next();
+});
 
 // ─── Request Timeout & Abort Signal ───
 app.use(abortMiddleware);
@@ -75,11 +95,7 @@ app.use('/api/v1', router);
 
 // ─── 404 Handler ───
 app.use((_req, res) => {
-  res.status(404).json({
-    ok: false,
-    data: null,
-    error: { code: 'NOT_FOUND', message: 'Route not found' },
-  });
+  sendError(res, 404, 'NOT_FOUND', 'Route not found');
 });
 
 // ─── Global Error Handler ───

src/config/index.ts

@@ -104,6 +104,7 @@ const envSchema = z
     RESOLUTION_THRESHOLD: z.string().default('1').transform(Number),
     RESOLUTION_DISPUTE_WINDOW_SECS: z.string().default('86400').transform(Number),
 
+    TRUST_INBOUND_REQUEST_ID: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'),
     INDEXER_ENABLED: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'),
     INDEXER_POLL_MS: z.string().default('5000').transform(Number),
 

src/controllers/market.controller.ts

@@ -1,5 +1,7 @@
 import type { Request, Response } from 'express';
 import { prisma } from '../lib/prisma';
+import { writeAgentLog } from '../lib/agent-log';
+import { bindDetached } from '../lib/context';
 import { v4 as uuidv4 } from 'uuid';
 import type { JobRecord } from '../types';
 import { JobStatus } from '../types';
@@ -219,25 +221,23 @@ export async function triggerMarketGeneration(
 
   // ── Persist audit log ────────────────────────────────────────
 
-  prisma.agentLog
-    .create({
-      data: {
+  bindDetached(
+    () =>
+      writeAgentLog({
         agentType: 'MARKET_MAKER',
         level: 'INFO',
         action: 'GENERATION_JOB_STARTED',
-
-        data: {
-          jobId,
-        },
-      },
-    })
+        data: { jobId },
+      }),
+    { jobId },
+  )
     .catch(() => {
       // Non-blocking
     });
 
   // ── Run generation in background ─────────────────────────────
 
-  runMarketMakerCycle()
+  bindDetached(() => runMarketMakerCycle(), { jobId, kind: 'job' })
     .then((result: any) => {
       const existingJob = jobTracker.get(jobId);
 

src/cron/ingestion.cron.ts

@@ -3,24 +3,27 @@ import { runMarketMakerCycle } from '../agents/market-maker.agent';
 import { logger } from '../lib/logger';
 import { recordCronRun } from './registry';
 
-let isRunning = false;
+let runningCycleId: string | null = null;
 
 /**
  * Schedules the Market Maker Agent to run every 15 minutes.
  *
- * The `isRunning` guard prevents overlapping executions.
+ * The `runningCycleId` guard prevents overlapping executions.
  * If a cycle is still running when the next one is scheduled,
  * the new trigger is skipped.
  *
  * Cron expression: "0,15,30,45 * * * *"
  *   Runs at :00, :15, :30, :45 of every hour
  */
 export function startIngestionCron(): void {
-  cron.schedule('0,15,30,45 * * * *', async () => {
-    if (isRunning) {
-      logger.warn('Ingestion cron: previous cycle still running — skipping');
-      return;
-    }
+  // The context is entered inside the callback: node-cron fires it from its own timer.
+  cron.schedule('0,15,30,45 * * * *', () => {
+    const cycleId = newId('cyc');
+    return runWithContext({ requestId: cycleId, cycleId, kind: 'cron', job: 'market_maker' }, async () => {
+      if (runningCycleId) {
+        logger.warn({ skippedJob: 'market_maker', runningCycleId }, 'Ingestion cron: previous cycle still running — skipping');
+        return;
+      }
 
     isRunning = true;
     const startedAt = new Date();

src/cron/monitor.cron.ts

@@ -2,10 +2,17 @@ import cron from 'node-cron';
 import { runTraderCycle } from '../agents/trader.agent';
 import { expireStaleSubscriptions } from '../services/subscription.service';
 import { logger } from '../lib/logger';
+import { newId, runWithContext } from '../lib/context';
 import { purgeExpiredIdempotencyKeys } from '../middlewares/idempotency.middleware';
 import { recordCronRun } from './registry';
 
-let traderRunning = false;
+let runningTraderCycleId: string | null = null;
+
+/** One correlated context per cron tick. */
+function tick(job: string, fn: () => Promise<void>): Promise<void> {
+  const cycleId = newId('cyc');
+  return runWithContext({ requestId: cycleId, cycleId, kind: 'cron', job }, fn);
+}
 
 /**
  * Position Monitor Cron — runs every 15 minutes.
@@ -18,9 +25,9 @@ let traderRunning = false;
  */
 export function startMonitorCron(): void {
   // Trader Agent — every 15 minutes
-  cron.schedule('5,20,35,50 * * * *', async () => {
-    if (traderRunning) {
-      logger.warn('Monitor cron: trader cycle still running — skipping');
+  cron.schedule('5,20,35,50 * * * *', () => tick('trader', async () => {
+    if (runningTraderCycleId) {
+      logger.warn({ skippedJob: 'trader', runningCycleId: runningTraderCycleId }, 'Monitor cron: trader cycle still running — skipping');
       return;
     }
 
@@ -33,9 +40,9 @@ export function startMonitorCron(): void {
       recordCronRun('trader', 'failure', startedAt, new Date(), err);
       logger.error({ err }, 'Monitor cron: trader cycle failed');
     } finally {
-      traderRunning = false;
+      runningTraderCycleId = null;
     }
-  });
+  }));
 
   // Subscription expiry — every hour
   cron.schedule('0 * * * *', async () => {
@@ -47,7 +54,7 @@ export function startMonitorCron(): void {
       recordCronRun('subscription-expiry', 'failure', startedAt, new Date(), err);
       logger.error({ err }, 'Monitor cron: subscription expiry failed');
     }
-  });
+  }));
 
   // Expired Idempotency-Key rows (24h TTL, #123) — hourly at :10, in batches of 5 000
   cron.schedule('10 * * * *', async () => {
@@ -60,12 +67,10 @@ export function startMonitorCron(): void {
       recordCronRun('idempotency-purge', 'failure', startedAt, new Date(), err);
       logger.error({ err }, 'Monitor cron: idempotency key purge failed');
     }
-  });
+  }));
 
   // Copy trade sweep — every 5 minutes
-  cron.schedule('*/5 * * * *', async () => {
-    await sweepCopyTrades();
-  });
+  cron.schedule('*/5 * * * *', () => tick('copy_trade_sweep', sweepCopyTrades));
 
   logger.info('Monitor cron started');
 }

src/lib/agent-log.ts

@@ -0,0 +1,17 @@
+import type { AgentLog, Prisma, PrismaClient } from '@prisma/client';
+import { prisma } from './prisma';
+import { correlation } from './context';
+
+export type AgentLogRow = Omit<Prisma.AgentLogUncheckedCreateInput, 'data'> & {
+  data?: Record<string, unknown>;
+};
+
+/**
+ * The only place that writes AgentLog rows: stamps the current request/cycle/job
+ * ids into `data.correlation` so a row can be tied back to its log lines.
+ */
+export function writeAgentLog(row: AgentLogRow, db: Pick<PrismaClient, 'agentLog'> = prisma): Promise<AgentLog> {
+  return db.agentLog.create({
+    data: { ...row, data: { ...row.data, correlation: correlation() } as any },
+  });
+}