From a5cb4b03e180e0cda12525b787e5e9ff006a91a8 Mon Sep 17 00:00:00 2001 From: david Date: Fri, 28 Aug 2026 12:29:45 +0200 Subject: [PATCH] feat: backfill Vector proofs from the Surplus support export --- supplier/.env.example | 5 + supplier/src/cli/surplus-seller.ts | 18 +- supplier/src/surplus/config.ts | 9 + supplier/src/surplus/historyExport.ts | 215 ++++++++++++++++++++ supplier/src/surplus/vectorPublisher.ts | 41 +++- tests/unit/surplus-history-export.test.ts | 98 +++++++++ tests/unit/surplus-vector-publisher.test.ts | 48 +++++ 7 files changed, 427 insertions(+), 7 deletions(-) create mode 100644 supplier/src/surplus/historyExport.ts create mode 100644 tests/unit/surplus-history-export.test.ts diff --git a/supplier/.env.example b/supplier/.env.example index c1c9f73..c9c7dd2 100644 --- a/supplier/.env.example +++ b/supplier/.env.example @@ -175,3 +175,8 @@ SURPLUS_PROOF_RESERVE_LOVELACE=5000000 SURPLUS_PROOF_FEE_BUDGET_LOVELACE=1000000 SURPLUS_PROOF_INTERVAL_MS=300000 SURPLUS_PROOF_CONFIRM_TIMEOUT_MS=300000 +# Optional support-export backfill. The CSV must have the exact header: +# request_id,settled_at,model,input_tokens,output_tokens,seller_earned_usd,settlement_status,settlement_type,tx_hash +# The offer id is stamped into each export leaf because the export omits it. +SURPLUS_PROOF_HISTORY_CSV_PATH= +SURPLUS_PROOF_HISTORY_OFFER_ID= diff --git a/supplier/src/cli/surplus-seller.ts b/supplier/src/cli/surplus-seller.ts index aae8c17..51849d7 100644 --- a/supplier/src/cli/surplus-seller.ts +++ b/supplier/src/cli/surplus-seller.ts @@ -7,7 +7,7 @@ import { import { fileURLToPath } from "node:url"; import { resolve } from "node:path"; import { loadSurplusManagerConfig } from "../surplus/config.js"; -import { SurplusClient } from "../surplus/client.js"; +import { SurplusClient, type SurplusSale } from "../surplus/client.js"; import { SurplusSellerController } from "../surplus/controller.js"; import { OpenRouterKeyClient } from "../surplus/openRouterKey.js"; import { FileSurplusStateStore } from "../surplus/state.js"; @@ -16,6 +16,7 @@ import { type SurplusDashboardSnapshot, } from "../surplus/dashboard.js"; import { FileSurplusVectorProofLedger } from "../surplus/vectorProof.js"; +import { loadHistoricalSalesCsv } from "../surplus/historyExport.js"; import { SurplusVectorProofPublisher } from "../surplus/vectorPublisher.js"; import { LiveOgmiosProvider } from "@marketplace/shared/chain"; import type { WalletKey } from "@marketplace/shared/tx"; @@ -63,9 +64,24 @@ export async function runSurplusSeller( privateKeyHex: derived.privateKeyHex, address: derived.address, }; + + let historicalSales: (() => Promise) | undefined; + if (proofConfig.historyCsvPath !== null && proofConfig.historyOfferId !== null) { + const history = await loadHistoricalSalesCsv( + proofConfig.historyCsvPath, + proofConfig.historyOfferId, + ); + console.log( + `surplus-proof: loaded ${history.sales.length} settled export sale(s) from ` + + `${proofConfig.historyCsvPath} (${history.settledWithoutTx} without a Base tx, ` + + `${history.skippedUnsettled} unsettled rows skipped)`, + ); + historicalSales = async () => history.sales; + } publisher = new SurplusVectorProofPublisher({ ledger, earnings: () => client.getEarnings(), + historicalSales, anchor: async (metadata) => { const built = await buildAnchorMetadataTx({ chain, walletKey, metadata }); return { expectedTxHash: built.expectedTxHash }; diff --git a/supplier/src/surplus/config.ts b/supplier/src/surplus/config.ts index b9bc112..d70418b 100644 --- a/supplier/src/surplus/config.ts +++ b/supplier/src/surplus/config.ts @@ -37,6 +37,8 @@ export interface SurplusProofConfig { feeBudgetLovelace: bigint; intervalMs: number; confirmTimeoutMs: number; + historyCsvPath: string | null; + historyOfferId: string | null; } export function loadSurplusManagerConfig( env: Record, @@ -170,6 +172,11 @@ function loadProofConfig( ); } + const historyCsvPath = env.SURPLUS_PROOF_HISTORY_CSV_PATH?.trim() || null; + const historyOfferId = historyCsvPath === null + ? null + : required(env, "SURPLUS_PROOF_HISTORY_OFFER_ID"); + return { ogmiosUrl, walletPrivateKeyHex: walletPrivateKeyHex.toLowerCase(), @@ -202,6 +209,8 @@ function loadProofConfig( 10_000, 3_600_000, ), + historyCsvPath, + historyOfferId, }; } diff --git a/supplier/src/surplus/historyExport.ts b/supplier/src/surplus/historyExport.ts new file mode 100644 index 0000000..589c406 --- /dev/null +++ b/supplier/src/surplus/historyExport.ts @@ -0,0 +1,215 @@ +/** + * historyExport.ts — loads the Surplus support export + * (request_id, settled_at, model, input_tokens, output_tokens, + * seller_earned_usd, settlement_status, settlement_type, tx_hash) and maps the + * settled rows onto the publisher's sale shape. + * + * The export carries less detail than the live API. The mapping records what + * the export proves and nothing more: + * - `createdAt` uses the export's `settled_at` timestamp, + * - `cacheReadTokens` and both effective prices are zero (absent from the export), + * - `offerId` comes from configuration because the export omits it. + * Rows that are not settled (`settlement_status` !== "confirmed") are skipped; + * they surface later through the live window or a newer export. + */ + +import { readFile } from "node:fs/promises"; +import type { SurplusSale } from "./client.js"; + +const EXPECTED_HEADER = [ + "request_id", + "settled_at", + "model", + "input_tokens", + "output_tokens", + "seller_earned_usd", + "settlement_status", + "settlement_type", + "tx_hash", +] as const; + +const BASE_TX_RE = /^0x[0-9a-fA-F]{64}$/; +const TOKENS_RE = /^(?:0|[1-9]\d*)$/; +const USD_RE = /^(?:0|[1-9]\d*)(?:\.(\d{1,6}))?$/; + +export interface HistoricalSalesExport { + sales: SurplusSale[]; + settledWithoutTx: number; + skippedUnsettled: number; +} + +export async function loadHistoricalSalesCsv( + path: string, + offerId: string, +): Promise { + if (offerId.trim() === "") { + throw new Error("The historical export needs a non-empty offer id"); + } + const rows = parseCsv(await readFile(path, "utf8")); + const header = rows.shift(); + if ( + header === undefined || + header.length !== EXPECTED_HEADER.length || + EXPECTED_HEADER.some((name, index) => header[index] !== name) + ) { + throw new Error( + `Surplus export header must be exactly: ${EXPECTED_HEADER.join(",")}`, + ); + } + + const result: HistoricalSalesExport = { + sales: [], + settledWithoutTx: 0, + skippedUnsettled: 0, + }; + const seen = new Set(); + for (const [index, row] of rows.entries()) { + const label = `Surplus export row ${index + 2}`; + if (row.length !== EXPECTED_HEADER.length) { + throw new Error(`${label} has ${row.length} fields, expected 9`); + } + const [ + requestId, + settledAt, + model, + inputTokens, + outputTokens, + earnedUsd, + settlementStatus, + , + txHash, + ] = row; + + if (requestId === "") throw new Error(`${label} is missing request_id`); + if (seen.has(requestId)) { + throw new Error(`${label} repeats request_id ${requestId}`); + } + seen.add(requestId); + if (settlementStatus !== "confirmed") { + result.skippedUnsettled += 1; + continue; + } + + const createdAtMs = Date.parse(settledAt); + if (!Number.isSafeInteger(createdAtMs) || createdAtMs < 0) { + throw new Error(`${label} has an invalid settled_at`); + } + const transactionHash = txHash === "" ? null : normalizeBaseTx(txHash, label); + if (transactionHash === null) result.settledWithoutTx += 1; + + result.sales.push({ + id: requestId, + model: nonEmpty(model, `${label} model`), + offerId, + settlementStatus, + createdAt: new Date(createdAtMs).toISOString(), + createdAtMs, + sellerCostMicroUsd: usdToMicro(earnedUsd, `${label} seller_earned_usd`), + inputTokens: tokens(inputTokens, `${label} input_tokens`), + outputTokens: tokens(outputTokens, `${label} output_tokens`), + cacheReadTokens: 0, + effectiveInputUsdPer1m: 0, + effectiveOutputUsdPer1m: 0, + transactionHash, + }); + } + return result; +} + +function normalizeBaseTx(value: string, label: string): string { + if (!BASE_TX_RE.test(value)) { + throw new Error(`${label} has an invalid tx_hash`); + } + return value.toLowerCase(); +} + +function nonEmpty(value: string, label: string): string { + if (value.trim() === "") throw new Error(`${label} must not be empty`); + return value; +} + +function tokens(value: string, label: string): number { + if (!TOKENS_RE.test(value)) { + throw new Error(`${label} must be a non-negative integer`); + } + const parsed = Number(value); + if (!Number.isSafeInteger(parsed)) { + throw new Error(`${label} exceeds the safe integer range`); + } + return parsed; +} + +/** Exact decimal → micro-USD integer without float rounding. */ +function usdToMicro(value: string, label: string): number { + const match = USD_RE.exec(value); + if (match === null) { + throw new Error(`${label} must be a non-negative USD decimal with at most 6 places`); + } + const whole = Number(value.split(".")[0]); + const fraction = Number((match[1] ?? "").padEnd(6, "0")); + const micro = whole * 1_000_000 + fraction; + if (!Number.isSafeInteger(micro)) { + throw new Error(`${label} exceeds the safe integer range`); + } + return micro; +} + +/** Minimal RFC-4180 parser: quoted fields, escaped quotes, CRLF or LF rows. */ +function parseCsv(text: string): string[][] { + const rows: string[][] = []; + let row: string[] = []; + let field = ""; + let quoted = false; + let index = 0; + + const pushField = (): void => { + row.push(field); + field = ""; + }; + const pushRow = (): void => { + pushField(); + rows.push(row); + row = []; + }; + + while (index < text.length) { + const char = text[index]; + if (quoted) { + if (char === '"') { + if (text[index + 1] === '"') { + field += '"'; + index += 2; + continue; + } + quoted = false; + index += 1; + continue; + } + field += char; + index += 1; + continue; + } + if (char === '"') { + if (field !== "") throw new Error(`Unexpected quote at offset ${index}`); + quoted = true; + index += 1; + continue; + } + if (char === ",") { + pushField(); + index += 1; + continue; + } + if (char === "\n" || char === "\r") { + if (char === "\r" && text[index + 1] === "\n") index += 1; + pushRow(); + index += 1; + continue; + } + field += char; + index += 1; + } + if (quoted) throw new Error("Unterminated quoted field"); + if (field !== "" || row.length > 0) pushRow(); + return rows.filter((entry) => !(entry.length === 1 && entry[0] === "")); +} diff --git a/supplier/src/surplus/vectorPublisher.ts b/supplier/src/surplus/vectorPublisher.ts index 16dd768..02dfb6d 100644 --- a/supplier/src/surplus/vectorPublisher.ts +++ b/supplier/src/surplus/vectorPublisher.ts @@ -18,6 +18,7 @@ import type { SurplusEarnings, SurplusSale } from "./client.js"; import { SURPLUS_SALE_PROOF_PROTOCOL, + SURPLUS_VECTOR_BATCH_SIZE, type CanonicalSurplusSale, type FileSurplusVectorProofLedger, type SurplusPendingBatchIntent, @@ -39,11 +40,18 @@ export type SurplusProofLedgerSeam = Pick< | "pendingBatchIntent" | "confirmPendingBatch" | "failPendingBatch" + | "dashboardProof" >; export interface SurplusVectorProofPublisherOptions { ledger: SurplusProofLedgerSeam; earnings: () => Promise; + /** + * Optional support-export source for sales the live window can no longer + * show. Observed sales win on id collisions; ids already in the ledger are + * never re-ingested, so a later live observation cannot conflict. + */ + historicalSales?: () => Promise; anchor: ( metadata: SurplusProofAnchorMetadata, ) => Promise<{ expectedTxHash: string }>; @@ -122,7 +130,14 @@ export class SurplusVectorProofPublisher { }; const settled = await this.observeSettledSales(); - const { inserted } = await ledger.ingestSales(settled); + const historical = this.options.historicalSales === undefined + ? [] + : await this.options.historicalSales(); + const candidates = dedupeById([...settled, ...historical]); + const fresh = candidates.filter((sale) => + ledger.dashboardProof(sale.id) === null + ); + const { inserted } = await ledger.ingestSales(fresh); result.ingested = inserted; if (inserted > 0) { log(`surplus-proof: captured ${inserted} newly settled sale(s)`); @@ -136,7 +151,7 @@ export class SurplusVectorProofPublisher { ); } - for (const batch of this.planBatches(settled, ledger.selectUnprovedSales())) { + for (const batch of this.planBatches(candidates, ledger.selectUnprovedSales())) { if (this.stopped) break; if (!(await this.fundsAvailable())) { result.fundingBlocked = true; @@ -164,9 +179,9 @@ export class SurplusVectorProofPublisher { } /** - * One batch per observed Base settlement tx (oldest first), then one - * catch-up batch for unproved sales whose settlement tx is no longer - * observable in the Surplus API window. + * One batch per known Base settlement tx (oldest first), then catch-up + * batches for unproved sales with no known settlement tx, chunked to the + * ledger batch-size limit. */ private planBatches( settled: readonly SurplusSale[], @@ -195,7 +210,9 @@ export class SurplusVectorProofPublisher { const batches = [...groups.values()].sort( (left, right) => earliest(left) - earliest(right), ); - if (leftovers.length > 0) batches.push(leftovers); + for (let start = 0; start < leftovers.length; start += SURPLUS_VECTOR_BATCH_SIZE) { + batches.push(leftovers.slice(start, start + SURPLUS_VECTOR_BATCH_SIZE)); + } return batches; } @@ -266,3 +283,15 @@ function earliest(sales: readonly CanonicalSurplusSale[]): number { } return value; } + +/** First occurrence wins; callers put observed live sales before export rows. */ +function dedupeById(sales: readonly SurplusSale[]): SurplusSale[] { + const seen = new Set(); + const unique: SurplusSale[] = []; + for (const sale of sales) { + if (seen.has(sale.id)) continue; + seen.add(sale.id); + unique.push(sale); + } + return unique; +} diff --git a/tests/unit/surplus-history-export.test.ts b/tests/unit/surplus-history-export.test.ts new file mode 100644 index 0000000..bd018ca --- /dev/null +++ b/tests/unit/surplus-history-export.test.ts @@ -0,0 +1,98 @@ +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { loadHistoricalSalesCsv } from "../../supplier/src/surplus/historyExport.js"; + +const HEADER = + "request_id,settled_at,model,input_tokens,output_tokens,seller_earned_usd,settlement_status,settlement_type,tx_hash"; + +const directories: string[] = []; + +async function csvFile(content: string): Promise { + const directory = await mkdtemp(join(tmpdir(), "surplus-history-")); + directories.push(directory); + const path = join(directory, "sales.csv"); + await writeFile(path, content, "utf8"); + return path; +} + +afterEach(async () => { + await Promise.all(directories.splice(0).map((path) => + rm(path, { recursive: true, force: true }) + )); +}); + +describe("loadHistoricalSalesCsv", () => { + it("maps settled export rows onto publisher sales and skips unsettled rows", async () => { + const baseTx = "0x" + "A".repeat(64); + const path = await csvFile([ + HEADER, + `"01SALE1","2026-08-28T07:39:33.869Z","deepseek-v4-flash","151204","805","0.000607","confirmed","onchain","${baseTx}"`, + `"01SALE2","2026-08-28T07:39:22.308Z","deepseek-v4-flash","10","2","1.5","confirmed","onchain",""`, + `"01SALE3","2026-08-28T07:39:48.036Z","deepseek-v4-flash","47257","1761","0.000202","accrued","credit",""`, + ].join("\n")); + + const result = await loadHistoricalSalesCsv(path, "offer-1"); + + expect(result.skippedUnsettled).toBe(1); + expect(result.settledWithoutTx).toBe(1); + expect(result.sales).toHaveLength(2); + expect(result.sales[0]).toMatchObject({ + id: "01SALE1", + offerId: "offer-1", + model: "deepseek-v4-flash", + settlementStatus: "confirmed", + createdAt: "2026-08-28T07:39:33.869Z", + createdAtMs: Date.parse("2026-08-28T07:39:33.869Z"), + sellerCostMicroUsd: 607, + inputTokens: 151_204, + outputTokens: 805, + cacheReadTokens: 0, + effectiveInputUsdPer1m: 0, + effectiveOutputUsdPer1m: 0, + transactionHash: baseTx.toLowerCase(), + }); + expect(result.sales[1]).toMatchObject({ + id: "01SALE2", + sellerCostMicroUsd: 1_500_000, + transactionHash: null, + }); + }); + + it("rejects an unexpected header, duplicate ids, and malformed rows", async () => { + const wrongHeader = await csvFile("request_id,foo\n\"a\",\"b\""); + await expect(loadHistoricalSalesCsv(wrongHeader, "offer-1")).rejects.toThrow( + /header must be exactly/, + ); + + const duplicate = await csvFile([ + HEADER, + `"01SAME","2026-08-28T07:39:33.869Z","m","1","1","0.1","confirmed","onchain",""`, + `"01SAME","2026-08-28T07:39:33.869Z","m","1","1","0.1","confirmed","onchain",""`, + ].join("\n")); + await expect(loadHistoricalSalesCsv(duplicate, "offer-1")).rejects.toThrow( + /repeats request_id/, + ); + + const badTx = await csvFile([ + HEADER, + `"01SALE","2026-08-28T07:39:33.869Z","m","1","1","0.1","confirmed","onchain","0x123"`, + ].join("\n")); + await expect(loadHistoricalSalesCsv(badTx, "offer-1")).rejects.toThrow( + /invalid tx_hash/, + ); + + const badTokens = await csvFile([ + HEADER, + `"01SALE","2026-08-28T07:39:33.869Z","m","-4","1","0.1","confirmed","onchain",""`, + ].join("\n")); + await expect(loadHistoricalSalesCsv(badTokens, "offer-1")).rejects.toThrow( + /input_tokens/, + ); + + await expect(loadHistoricalSalesCsv(badTokens, " ")).rejects.toThrow( + /non-empty offer id/, + ); + }); +}); diff --git a/tests/unit/surplus-vector-publisher.test.ts b/tests/unit/surplus-vector-publisher.test.ts index 4470e0f..471cfaa 100644 --- a/tests/unit/surplus-vector-publisher.test.ts +++ b/tests/unit/surplus-vector-publisher.test.ts @@ -209,6 +209,54 @@ describe("SurplusVectorProofPublisher", () => { ]); expect(ledger.dashboardProof("01EXPIRED")).toMatchObject({ status: "confirmed" }); }); + it("anchors export-only settlements per Base tx and never re-ingests known sale ids", async () => { + const baseTxA = "0x" + "a".repeat(64); + const baseTxB = "0x" + "b".repeat(64); + const observed = sale("01OBSERVED", { transactionHash: baseTxA }); + const { ledger, publisher, anchors } = await harness({ sales: [observed] }); + + await ledger.ingestSales([observed]); + const conflictingExportCopy = sale("01OBSERVED", { + transactionHash: baseTxA, + inputTokens: 999_999, + cacheReadTokens: 0, + }); + const historical = [ + conflictingExportCopy, + sale("01HISTA2", { transactionHash: baseTxA, cacheReadTokens: 0 }), + sale("01HISTB1", { + transactionHash: baseTxB, + cacheReadTokens: 0, + createdAtMs: Date.parse("2026-08-27T01:00:00.000Z"), + createdAt: "2026-08-27T01:00:00.000Z", + }), + ]; + const historicalPublisher = new SurplusVectorProofPublisher({ + ledger, + earnings: async () => earningsWith([observed]), + historicalSales: async () => historical, + anchor: publisher["options"].anchor, + awaitTx: async () => undefined, + balanceLovelace: async () => 25_000_000n, + reserveLovelace: 5_000_000n, + feeBudgetLovelace: 1_000_000n, + settledStatuses: ["confirmed"], + intervalMs: 60_000, + confirmTimeoutMs: 1_000, + log: () => undefined, + }); + + const result = await historicalPublisher.runCycle(); + + expect(result).toMatchObject({ ingested: 2, anchored: 2, failed: false }); + expect(anchors.map((metadata) => metadata.first)).toEqual([ + "01HISTB1", + "01HISTA2", + ]); + expect(ledger.dashboardProof("01OBSERVED")).toMatchObject({ status: "confirmed" }); + expect(ledger.dashboardProof("01HISTA2")).toMatchObject({ status: "confirmed" }); + expect(ledger.dashboardProof("01HISTB1")).toMatchObject({ status: "confirmed" }); + }); it("marks a confirmation timeout as a confirm-stage failure", async () => { const { ledger, publisher } = await harness({