Prune non-success quote lifecycle history

Proof: Emergency cleanup showed normalized quote lifecycle tables, not raw quotes alone, filled the Postgres PVC. History-writer now prunes old non-success quote rows across raw quotes, normalized demand, decisions, commands, executor results, and outcome attributions while preserving completed quote settlement rows and a 30 minute live window. Targeted retention tests, full npm test, and operator dashboard bundle build pass.

Assumptions: completed quote outcomes and attributed successful quote ids are the durable settlement evidence to retain; short recent live rows may be retained temporarily so in-flight attribution is not erased mid-cycle.

Still fake: venue-native terminal fill ids and fee-complete realized PnL remain unavailable; historical non-success quote analytics pruned during emergency recovery are intentionally no longer readable.

src/apps/history-writer.mjs

@@ -26,6 +26,7 @@ import {
   insertHistoryEvents,
   loadLatestPortfolioMetric,
   loadPortfolioMetricInputs,
+  pruneNonSuccessfulQuoteLifecycleHistory,
   pruneRawNearIntentsQuoteHistory,
   refreshIntentRequestOutcomes,
   refreshQuoteOutcomes,
@@ -125,6 +126,10 @@ const rawQuoteHistoryPruneIntervalMs = 60 * 1000;
 const rawQuoteHistoryRetainRecentMs = 30 * 60 * 1000;
 const rawQuoteHistoryPruneBatchSize = 500_000;
 const rawQuoteHistoryPruneMaxBatches = 20;
+const quoteLifecycleHistoryPruneIntervalMs = 60 * 1000;
+const quoteLifecycleHistoryRetainRecentMs = 30 * 60 * 1000;
+const quoteLifecycleHistoryPruneBatchSize = 100_000;
+const quoteLifecycleHistoryPruneMaxBatches = 10;
 const liveEvidenceTopics = [
   config.kafkaTopicNormSwapDemand,
   config.kafkaTopicDecisionTradeDecision,
@@ -197,6 +202,9 @@ const state = {
   last_raw_quote_prune_at: null,
   raw_quote_prune_deleted_count: 0,
   raw_quote_prune_error: null,
+  last_quote_lifecycle_prune_at: null,
+  quote_lifecycle_prune_deleted_count: 0,
+  quote_lifecycle_prune_error: null,
   derived_refresh_skipped_count: 0,
   last_derived_refresh_skipped_at: null,
   last_derived_refresh_skipped_topic: null,
@@ -287,6 +295,7 @@ async function runHistoryConsumer(historyConsumer) {
       if (batch.topic === config.kafkaTopicRawNearIntentsQuote) {
         await maybePruneRawQuoteHistory();
       }
+      await maybePruneQuoteLifecycleHistory();
 
       if (state.draining) {
         setTimeout(() => shutdown(), 0);
@@ -335,6 +344,46 @@ async function maybePruneRawQuoteHistory({ force = false } = {}) {
   }
 }
 
+async function maybePruneQuoteLifecycleHistory({ force = false } = {}) {
+  const nowMs = Date.now();
+  const lastPruneMs = state.last_quote_lifecycle_prune_at
+    ? Date.parse(state.last_quote_lifecycle_prune_at)
+    : 0;
+  if (
+    !force
+    && Number.isFinite(lastPruneMs)
+    && nowMs - lastPruneMs < quoteLifecycleHistoryPruneIntervalMs
+  ) {
+    return null;
+  }
+
+  try {
+    const result = await pruneNonSuccessfulQuoteLifecycleHistory(pool, {
+      now: new Date(nowMs).toISOString(),
+      retainRecentMs: quoteLifecycleHistoryRetainRecentMs,
+      batchSize: quoteLifecycleHistoryPruneBatchSize,
+      maxBatches: quoteLifecycleHistoryPruneMaxBatches,
+    });
+    state.last_quote_lifecycle_prune_at = new Date(nowMs).toISOString();
+    state.quote_lifecycle_prune_deleted_count += result.deletedCount;
+    state.quote_lifecycle_prune_error = null;
+    if (result.deletedCount > 0) {
+      logger.info('non_successful_quote_lifecycle_history_pruned', {
+        details: result,
+      });
+    }
+    return result;
+  } catch (error) {
+    state.quote_lifecycle_prune_error = serializeError(error);
+    logger.error('non_successful_quote_lifecycle_history_prune_failed', {
+      details: {
+        error: state.quote_lifecycle_prune_error,
+      },
+    });
+    return null;
+  }
+}
+
 async function handleWrittenHistoryEvent({
   topic,
   partition,

src/lib/postgres.mjs

@@ -2190,6 +2190,92 @@ export async function pruneRawNearIntentsQuoteHistory(pool, {
   };
 }
 
+export async function pruneNonSuccessfulQuoteLifecycleHistory(pool, {
+  now = new Date().toISOString(),
+  retainRecentMs = 30 * 60 * 1000,
+  batchSize = 100_000,
+  maxBatches = 1,
+} = {}) {
+  const nowMs = Date.parse(now);
+  if (!Number.isFinite(nowMs)) throw new Error('now must be a valid timestamp');
+  if (!Number.isInteger(retainRecentMs) || retainRecentMs <= 0) {
+    throw new Error('retain_recent_ms must be a positive integer');
+  }
+  const boundedBatchSize = Math.max(1, Math.min(500_000, Math.floor(Number(batchSize) || 0)));
+  const boundedMaxBatches = Math.max(1, Math.min(100, Math.floor(Number(maxBatches) || 0)));
+  const cutoff = new Date(nowMs - retainRecentMs).toISOString();
+  const tables = [
+    { table: 'raw_near_intents_quotes', timestampColumn: 'ingested_at' },
+    { table: 'swap_demand_events', timestampColumn: 'ingested_at' },
+    { table: 'trade_decisions', timestampColumn: 'ingested_at' },
+    { table: 'execute_trade_commands', timestampColumn: 'ingested_at' },
+    { table: 'trade_execution_results', timestampColumn: 'ingested_at' },
+    { table: 'quote_outcome_attributions', timestampColumn: 'computed_at', selfOutcome: true },
+  ];
+
+  const perTable = [];
+  let deletedCount = 0;
+  for (const spec of tables) {
+    let tableDeletedCount = 0;
+    let batches = 0;
+    for (let batch = 0; batch < boundedMaxBatches; batch += 1) {
+      const successPredicate = spec.selfOutcome
+        ? `
+            (
+              target.outcome_status = 'completed'
+              OR target.attribution_status IN ('heuristic_match', 'exact_match', 'attributed')
+            )
+          `
+        : `
+            EXISTS (
+              SELECT 1
+              FROM quote_outcome_attributions outcome
+              WHERE outcome.quote_id = target.quote_id
+                AND (
+                  outcome.outcome_status = 'completed'
+                  OR outcome.attribution_status IN ('heuristic_match', 'exact_match', 'attributed')
+                )
+            )
+          `;
+      const result = await pool.query(
+        `
+          WITH stale_rows AS (
+            SELECT target.ctid
+            FROM ${spec.table} target
+            WHERE COALESCE(target.${spec.timestampColumn}, 'epoch'::timestamptz) < $1::timestamptz
+              AND NOT ${successPredicate}
+            ORDER BY target.${spec.timestampColumn} ASC NULLS FIRST
+            LIMIT $2
+          )
+          DELETE FROM ${spec.table} target
+          USING stale_rows
+          WHERE target.ctid = stale_rows.ctid
+        `,
+        [cutoff, boundedBatchSize],
+      );
+      const batchDeleted = Number(result.rowCount || 0);
+      tableDeletedCount += batchDeleted;
+      deletedCount += batchDeleted;
+      batches += 1;
+      if (batchDeleted < boundedBatchSize) break;
+    }
+    perTable.push({
+      table: spec.table,
+      deletedCount: tableDeletedCount,
+      batches,
+    });
+  }
+
+  return {
+    deletedCount,
+    cutoff,
+    retainRecentMs,
+    batchSize: boundedBatchSize,
+    maxBatches: boundedMaxBatches,
+    tables: perTable,
+  };
+}
+
 export async function insertEnvironmentStatusChange(pool, { topic, event, record }) {
   const fingerprint = event.payload?.status_fingerprint || null;
   const environmentKey = event.payload?.environment_key || record.decision_key || null;

test/history-writer-static.test.mjs

@@ -24,9 +24,16 @@ test('history writer replays durable topics but joins the raw quote firehose liv
   assert.match(source, /rawQuoteHistoryRetainRecentMs\s*=\s*30 \* 60 \* 1000/);
   assert.match(source, /rawQuoteHistoryPruneBatchSize\s*=\s*500_000/);
   assert.match(source, /rawQuoteHistoryPruneMaxBatches\s*=\s*20/);
+  assert.match(source, /quoteLifecycleHistoryPruneIntervalMs\s*=\s*60 \* 1000/);
+  assert.match(source, /quoteLifecycleHistoryRetainRecentMs\s*=\s*30 \* 60 \* 1000/);
+  assert.match(source, /quoteLifecycleHistoryPruneBatchSize\s*=\s*100_000/);
+  assert.match(source, /quoteLifecycleHistoryPruneMaxBatches\s*=\s*10/);
   assert.match(source, /pruneRawNearIntentsQuoteHistory/);
+  assert.match(source, /pruneNonSuccessfulQuoteLifecycleHistory/);
   assert.match(source, /maxBatches:\s*rawQuoteHistoryPruneMaxBatches/);
+  assert.match(source, /maxBatches:\s*quoteLifecycleHistoryPruneMaxBatches/);
   assert.match(source, /batch\.topic === config\.kafkaTopicRawNearIntentsQuote/);
+  assert.match(source, /maybePruneQuoteLifecycleHistory/);
 });
 
 test('history writer passes tracked assets into portfolio valuation', () => {

test/inventory-and-history.test.mjs

@@ -3,7 +3,11 @@ import assert from 'node:assert/strict';
 
 import { buildInventorySnapshot } from '../src/core/inventory.mjs';
 import { routeHistoryRecord } from '../src/core/history-records.mjs';
-import { insertHistoryEvents, pruneRawNearIntentsQuoteHistory } from '../src/lib/postgres.mjs';
+import {
+  insertHistoryEvents,
+  pruneNonSuccessfulQuoteLifecycleHistory,
+  pruneRawNearIntentsQuoteHistory,
+} from '../src/lib/postgres.mjs';
 
 test('inventory snapshot keeps pending funding out of spendable balances', () => {
   const snapshot = buildInventorySnapshot({
@@ -157,6 +161,45 @@ test('raw quote retention drains multiple bounded batches when firehose backlog
   }
 });
 
+test('quote lifecycle retention preserves completed trades and drains old non-success rows', async () => {
+  const rowCounts = [100_000, 12, 7, 4, 3, 2, 1];
+  const queries = [];
+  const pool = {
+    async query(sql, params) {
+      queries.push({ sql, params });
+      return { rows: [], rowCount: rowCounts.shift() ?? 0 };
+    },
+  };
+
+  const result = await pruneNonSuccessfulQuoteLifecycleHistory(pool, {
+    now: '2026-06-05T12:00:00.000Z',
+    retainRecentMs: 30 * 60 * 1000,
+    batchSize: 100_000,
+    maxBatches: 10,
+  });
+
+  assert.equal(result.deletedCount, 100_029);
+  assert.equal(result.cutoff, '2026-06-05T11:30:00.000Z');
+  assert.equal(result.batchSize, 100_000);
+  assert.equal(result.maxBatches, 10);
+  assert.deepEqual(result.tables.map((entry) => entry.table), [
+    'raw_near_intents_quotes',
+    'swap_demand_events',
+    'trade_decisions',
+    'execute_trade_commands',
+    'trade_execution_results',
+    'quote_outcome_attributions',
+  ]);
+  assert.equal(queries.length, 7);
+  for (const query of queries) {
+    assert.match(query.sql, /DELETE FROM/);
+    assert.match(query.sql, /outcome_status = 'completed'/);
+    assert.match(query.sql, /attribution_status IN \('heuristic_match', 'exact_match', 'attributed'\)/);
+    assert.deepEqual(query.params, ['2026-06-05T11:30:00.000Z', 100_000]);
+  }
+  assert.match(queries.at(-1).sql, /DELETE FROM quote_outcome_attributions/);
+});
+
 function historyEvent(eventId, payload) {
   return {
     event_id: eventId,