buyer: PDF book summarizer (/book) — chunk → map-reduce over suppliers

Adds an authenticated upload flow to the buyer web app: an operator
uploads a PDF, the server extracts + paragraph-aligned-chunks it
(~2.2k tokens), then runs a hierarchical map-reduce summarization by
paying marketplace suppliers per chunk — one on-chain escrow per map
call and per reduce node. One book upload fans out into dozens of real
mainnet escrows (the intended marketplace traffic) while producing a
genuine full-book summary.

Server module (buyer/src/pdf/): PDF extraction (text-only, rejects
scanned), token-aware chunker, supplier discovery/round-robin filtered
to capable models (kimi/deepseek/gpt; excludes qwen2.5:0.5b), pre-spend
cost estimate + wallet-floor guard, and a JobStore orchestrator that
reuses submitPrompt → findSubmittedEscrowRef → runAccept per call.
v1 serializes PostEscrow under a mutex (the shared tx builder can't pin
wallet inputs, so parallel escrows would double-spend); retries failed
calls on a different supplier then continues with gaps. 6 /v1/pdf-*
routes (multipart upload, estimate, confirm-gated start, SSE progress,
result, .md download) behind the existing session auth + page/chunk/
balance caps. SPA adds a 4-step BookSummarizer wizard.

Tests: chunk/estimate/supplier-pool/extract unit tests + orchestrator
(happy/gap/failover) + routes integration. Full buyer suite green
(181 tests); SPA build + typecheck clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
david
2026-06-03 11:37:49 +02:00
parent 2025ab7411
commit c552c4caa3
25 changed files with 2518 additions and 0 deletions

View File

@@ -38,6 +38,29 @@ NETWORK_ID=0
# along over loopback. Production behind TLS should leave this unset / "1".
# COOKIE_SECURE=1
# ─── Optional: PDF Book Summarizer (/book page, /v1/pdf-* routes) ─────────
# Upload a PDF → chunk it → pay suppliers to summarize each chunk (one escrow
# per map + reduce call) → full-book summary. All knobs are optional; defaults
# shown. Tune the caps to bound spend/abuse on mainnet.
#
# Hard caps (a job is refused if it would exceed these):
# PDF_MAX_PAGES=400 # reject books larger than this
# PDF_MAX_CHUNKS=200 # reject books that chunk into more calls
# PDF_MAX_PDF_BYTES=25000000 # multipart upload size limit (bytes)
# Refuse to START a job if projected spend would drop the wallet below this:
# PDF_WALLET_FLOOR_LOVELACE=50000000 # 50 AP3X
# Chunking + reduce shape:
# PDF_CHUNK_TARGET_TOKENS=2200 # target tokens per chunk
# PDF_CHUNK_MAX_TOKENS=3000 # hard ceiling per chunk
# PDF_REDUCE_FANIN=8 # summaries collapsed per reduce node
# PDF_RETRY_K=2 # per-call retries on a different supplier
# Supplier selection (comma-separated model-name substrings):
# PDF_MODEL_ALLOWLIST=kimi,deepseek,gpt
# PDF_MODEL_DENYLIST=qwen2.5:0.5b,qwen2.5-0.5b
# Concurrency. v1 must stay 1 (serial PostEscrow — the shared tx builder
# can't pin wallet inputs yet, so parallel escrows would double-spend):
# PDF_LANE_COUNT=1
# ─── Optional: CIP-33 reference scripts ───────────────────────────────────
# When set, buyer-signed script-spending txs (Accept, Reclaim) read the
# escrow validator from an on-chain reference UTxO instead of inlining

View File

@@ -21,6 +21,9 @@
"bech32": "^2.0.0",
"better-sqlite3": "^11.9.1",
"express": "^4.21.2",
"gpt-tokenizer": "^2.8.1",
"multer": "^1.4.5-lts.1",
"pdf-parse": "^1.1.1",
"react": "^18.3.1",
"react-dom": "^18.3.1",
"react-router-dom": "^6.26.0"
@@ -31,6 +34,8 @@
"@testing-library/user-event": "^14.5.2",
"@types/better-sqlite3": "^7.6.12",
"@types/express": "^5.0.3",
"@types/multer": "^1.4.11",
"@types/pdf-parse": "^1.1.4",
"@types/react": "^18.3.0",
"@types/react-dom": "^18.3.0",
"@vitejs/plugin-react": "^4.3.0",

View File

@@ -16,6 +16,8 @@ import { createApp } from "./server.js";
import { buildChainProvider } from "./chain.js";
import { Marketplace, MemoryTaskHistoryStore } from "./sdk/index.js";
import { ResponseArchive } from "./db/archive.js";
import { JobStore } from "./pdf/summarize-job.js";
import { loadPdfCaps } from "./pdf/caps.js";
import type { WalletKey } from "@marketplace/shared/tx";
// Wire ed25519 sha512 hook (idempotent — same as receipt/sign.ts).
@@ -91,6 +93,18 @@ export async function runMain(env: Record<string, string | undefined>): Promise<
);
}
// PDF book summarizer: one job registry, sharing the live marketplace SDK,
// chain, wallet, and (optional) archive. Caps come from PDF_* env vars.
const pdfCaps = loadPdfCaps(env);
const jobStore = new JobStore({
marketplace,
chain,
walletKey,
indexerUrl: config.indexerUrl,
archive,
caps: pdfCaps,
});
const app = createApp({
distPath,
chain,
@@ -102,6 +116,8 @@ export async function runMain(env: Record<string, string | undefined>): Promise<
password: config.password,
sessionSecret: config.sessionSecret,
cookieSecure: config.cookieSecure,
jobStore,
pdfCaps,
});
const server = app.listen(config.port, () => {

58
buyer/src/pdf/caps.ts Normal file
View File

@@ -0,0 +1,58 @@
/**
* buyer/src/pdf/caps.ts — resolve PdfCaps (safety + sizing knobs) from env.
*
* Every knob is optional with a sane default, so the PDF summarizer works out
* of the box; operators tune spend/abuse limits via PDF_* env vars. See
* pdf/types.ts for field docs.
*/
import type { PdfCaps } from "./types.js";
function intEnv(env: Record<string, string | undefined>, name: string, dflt: number): number {
const v = env[name];
if (v === undefined || v === "") return dflt;
const n = Number(v);
if (!Number.isFinite(n) || n < 0) {
throw new Error(`loadPdfCaps: ${name} must be a non-negative number (got: ${v})`);
}
return Math.floor(n);
}
function bigintEnv(env: Record<string, string | undefined>, name: string, dflt: bigint): bigint {
const v = env[name];
if (v === undefined || v === "") return dflt;
if (!/^\d+$/.test(v)) {
throw new Error(`loadPdfCaps: ${name} must be a non-negative integer (got: ${v})`);
}
return BigInt(v);
}
function listEnv(env: Record<string, string | undefined>, name: string, dflt: string[]): string[] {
const v = env[name];
if (v === undefined) return dflt;
return v
.split(",")
.map((s) => s.trim())
.filter((s) => s.length > 0);
}
/** Built-in defaults — used as the fallback when createApp gets no pdfCaps. */
export function defaultPdfCaps(): PdfCaps {
return loadPdfCaps({});
}
export function loadPdfCaps(env: Record<string, string | undefined>): PdfCaps {
return {
maxPages: intEnv(env, "PDF_MAX_PAGES", 400),
maxChunks: intEnv(env, "PDF_MAX_CHUNKS", 200),
walletFloorLovelace: bigintEnv(env, "PDF_WALLET_FLOOR_LOVELACE", 50_000_000n),
laneCount: Math.max(1, intEnv(env, "PDF_LANE_COUNT", 1)),
chunkTargetTokens: intEnv(env, "PDF_CHUNK_TARGET_TOKENS", 2200),
chunkMaxTokens: intEnv(env, "PDF_CHUNK_MAX_TOKENS", 3000),
reduceFanin: Math.max(2, intEnv(env, "PDF_REDUCE_FANIN", 8)),
retryK: intEnv(env, "PDF_RETRY_K", 2),
modelAllowlist: listEnv(env, "PDF_MODEL_ALLOWLIST", ["kimi", "deepseek", "gpt"]),
modelDenylist: listEnv(env, "PDF_MODEL_DENYLIST", ["qwen2.5:0.5b", "qwen2.5-0.5b"]),
maxPdfBytes: intEnv(env, "PDF_MAX_PDF_BYTES", 25_000_000),
};
}

100
buyer/src/pdf/chunk.ts Normal file
View File

@@ -0,0 +1,100 @@
/**
* buyer/src/pdf/chunk.ts — paragraph-aligned chunker.
*
* Splits extracted book text into ~targetTokens chunks WITHOUT ever cutting a
* sentence in half (the unit of summarization should be coherent). Strategy:
*
* 1. Decompose text into "units" each ≤ maxTokens:
* paragraph → if too big → sentences → if a sentence is still too big
* (rare) → hard word-split as a last resort.
* 2. Greedily pack units into chunks, flushing before a unit would push the
* running chunk past targetTokens.
*
* Each chunk therefore contains whole sentences; the only place a sentence is
* ever broken is a single sentence longer than maxTokens, which essentially
* never happens in real prose.
*/
import { countTokens } from "./tokens.js";
import type { Chunk } from "./types.js";
export interface ChunkOptions {
targetTokens: number;
maxTokens: number;
}
/** Blank-line separated paragraphs, internal whitespace collapsed. */
function splitParagraphs(text: string): string[] {
return text
.split(/\n\s*\n+/)
.map((p) => p.replace(/\s+/g, " ").trim())
.filter((p) => p.length > 0);
}
/** Sentence-ish split: keep terminal punctuation attached to its sentence. */
function splitSentences(paragraph: string): string[] {
const matches = paragraph.match(/[^.!?]+[.!?]+(?:\s|$)|[^.!?]+$/g);
if (!matches) return [paragraph];
return matches.map((s) => s.trim()).filter((s) => s.length > 0);
}
/** Last-resort split of a single oversize sentence by word count. */
function hardSplitByTokens(s: string, maxTokens: number): string[] {
const words = s.split(/\s+/).filter(Boolean);
const total = countTokens(s) || 1;
const wordsPerChunk = Math.max(1, Math.floor((words.length * maxTokens) / total));
const out: string[] = [];
for (let i = 0; i < words.length; i += wordsPerChunk) {
out.push(words.slice(i, i + wordsPerChunk).join(" "));
}
return out.length > 0 ? out : [s];
}
/** Decompose text into coherent units each ≤ maxTokens. */
function toUnits(text: string, maxTokens: number): string[] {
const units: string[] = [];
for (const para of splitParagraphs(text)) {
if (countTokens(para) <= maxTokens) {
units.push(para);
continue;
}
for (const sent of splitSentences(para)) {
if (countTokens(sent) <= maxTokens) {
units.push(sent);
} else {
units.push(...hardSplitByTokens(sent, maxTokens));
}
}
}
return units;
}
export function chunkText(text: string, opts: ChunkOptions): Chunk[] {
const { targetTokens, maxTokens } = opts;
const units = toUnits(text, maxTokens);
const chunks: Chunk[] = [];
let buf: string[] = [];
let bufTokens = 0;
const flush = (): void => {
if (buf.length === 0) return;
chunks.push({
index: chunks.length,
text: buf.join("\n\n"),
tokenEstimate: bufTokens,
});
buf = [];
bufTokens = 0;
};
for (const unit of units) {
const ut = countTokens(unit);
if (bufTokens > 0 && bufTokens + ut > targetTokens) flush();
buf.push(unit);
bufTokens += ut;
}
flush();
return chunks;
}

67
buyer/src/pdf/estimate.ts Normal file
View File

@@ -0,0 +1,67 @@
/**
* buyer/src/pdf/estimate.ts — pure cost math, computed BEFORE any escrow is
* posted so the operator can confirm the spend.
*
* A job's call count = mapCalls (one per chunk) + reduceCalls (the hierarchical
* reduce tree). Each call pays exactly one supplier's flat price_lovelace; the
* pool mixes prices, so we bound the total with the priciest supplier.
*/
import type { SupplierPool } from "./supplier-pool.js";
/**
* Number of reduce calls in an F-ary reduce tree over `n` leaves.
* e.g. n=90,F=8 → 12 + 2 + 1 = 15; n=8 → 1; n=1 → 0.
*/
export function reduceCallCount(n: number, fanin: number): number {
if (n <= 1 || fanin < 2) return 0;
let total = 0;
let level = n;
while (level > 1) {
level = Math.ceil(level / fanin);
total += level;
}
return total;
}
export interface JobEstimate {
mapCalls: number;
reduceCalls: number;
totalCalls: number;
/** Upper-bound per-call price (priciest supplier), lovelace. */
perCallMaxLovelace: string;
/** Upper-bound total payment, lovelace. */
totalLovelace: string;
/** Same, expressed in AP3X (1 AP3X = 1e6 lovelace), 2dp. */
totalAp3x: string;
}
export function estimateJob(
chunkCount: number,
fanin: number,
pool: SupplierPool,
): JobEstimate {
const mapCalls = chunkCount;
const reduceCalls = reduceCallCount(chunkCount, fanin);
const totalCalls = mapCalls + reduceCalls;
const priceMax = pool.maxPrice();
const totalLovelace = priceMax * BigInt(totalCalls);
return {
mapCalls,
reduceCalls,
totalCalls,
perCallMaxLovelace: priceMax.toString(),
totalLovelace: totalLovelace.toString(),
totalAp3x: (Number(totalLovelace) / 1e6).toFixed(2),
};
}
/**
* Conservative lovelace headroom for chain fees + transient bonds across a
* whole job. The buyer pays PostEscrow + Accept fees per call and locks a
* 1 AP3X bond transiently; this over-estimates so the wallet floor check
* errs safe.
*/
export function feeHeadroomLovelace(totalCalls: number): bigint {
return BigInt(totalCalls) * 1_000_000n + 2_000_000n;
}

57
buyer/src/pdf/extract.ts Normal file
View File

@@ -0,0 +1,57 @@
/**
* buyer/src/pdf/extract.ts — text extraction from an uploaded PDF buffer.
*
* Text-based PDFs only (v1 has no OCR). Scanned/image-only PDFs extract
* almost no text; we detect that via a chars-per-page floor and reject with
* a clear `scanned_or_image_pdf` reason so the UI can tell the operator why.
*
* We require `pdf-parse/lib/pdf-parse.js` directly (not the package index)
* because pdf-parse's index.js runs a debug block that reads a bundled test
* PDF when `module.parent` is falsy — which is exactly the case under ESM/tsx
* and would crash on import. The lib entry is the pure function.
*/
import { createRequire } from "module";
import { PdfExtractionError } from "./types.js";
const require = createRequire(import.meta.url);
const pdfParse = require("pdf-parse/lib/pdf-parse.js") as (
data: Buffer,
opts?: Record<string, unknown>,
) => Promise<{ text: string; numpages: number }>;
/** Below this many extracted chars per page we treat the PDF as scanned. */
const MIN_CHARS_PER_PAGE = 50;
export interface ExtractResult {
text: string;
pageCount: number;
}
export async function extractPdfText(buf: Buffer): Promise<ExtractResult> {
let parsed: { text: string; numpages: number };
try {
parsed = await pdfParse(buf);
} catch (err) {
throw new PdfExtractionError(
"pdf_parse_failed",
err instanceof Error ? err.message : String(err),
);
}
const text = (parsed.text ?? "").trim();
const pageCount = parsed.numpages ?? 0;
if (
text.length === 0 ||
(pageCount > 0 && text.length / pageCount < MIN_CHARS_PER_PAGE)
) {
throw new PdfExtractionError(
"scanned_or_image_pdf",
`extracted only ${text.length} chars across ${pageCount} page(s) — ` +
`looks scanned/image-only, which v1 can't summarize (no OCR)`,
);
}
return { text, pageCount };
}

208
buyer/src/pdf/routes.ts Normal file
View File

@@ -0,0 +1,208 @@
/**
* buyer/src/pdf/routes.ts — HTTP surface for the PDF book summarizer.
*
* Mounted under /v1/pdf-* so it inherits the buyer app's session auth
* (server.ts gates every /v1/* route except /v1/auth/*). The upload route
* uses multer (multipart) so the global express.json({limit:"1mb"}) body
* parser never sees the PDF.
*
* Endpoints:
* POST /v1/pdf-upload multipart "file" → {job_id, chunk_count, ...}
* GET /v1/pdf-jobs/:id/estimate pre-spend cost + wallet floor check
* POST /v1/pdf-jobs/:id/start {confirm:true} → 202 (posts escrows)
* GET /v1/pdf-jobs/:id/events text/event-stream live progress
* GET /v1/pdf-jobs/:id job view (status, coverage, summary)
* GET /v1/pdf-jobs/:id/summary.md downloadable markdown
*/
import type { Express, Request, Response } from "express";
import multer from "multer";
import { extractPdfText } from "./extract.js";
import { chunkText } from "./chunk.js";
import { PdfExtractionError, type PdfCaps } from "./types.js";
import type { JobStore } from "./summarize-job.js";
function jsonError(res: Response, status: number, reason: string, message: string): Response {
return res.status(status).json({ error: reason, message });
}
export function registerPdfRoutes(
app: Express,
jobStore: JobStore | undefined,
caps: PdfCaps,
): void {
// Without a JobStore (chain/marketplace deps missing, or archive off) the
// feature is disabled — respond 503 on every pdf route so the SPA renders a
// clear banner instead of 404s.
if (!jobStore) {
const disabled = (_req: Request, res: Response): Response =>
jsonError(
res,
503,
"service_unavailable",
"PDF summarizer disabled (buyer booted without chain/marketplace/archive deps)",
);
app.post("/v1/pdf-upload", disabled);
app.all(/^\/v1\/pdf-jobs(\/.*)?$/, disabled);
return;
}
const upload = multer({
storage: multer.memoryStorage(),
limits: { fileSize: caps.maxPdfBytes },
});
// ── POST /v1/pdf-upload ─────────────────────────────────────────────
app.post("/v1/pdf-upload", (req: Request, res: Response) => {
upload.single("file")(req, res, async (err: unknown) => {
if (err) {
const code = (err as { code?: string }).code;
if (code === "LIMIT_FILE_SIZE") {
return jsonError(res, 413, "pdf_too_large", `max upload is ${caps.maxPdfBytes} bytes`);
}
return jsonError(res, 400, "upload_failed", err instanceof Error ? err.message : String(err));
}
const file = (req as Request & { file?: { buffer: Buffer; originalname?: string } }).file;
if (!file) {
return jsonError(res, 400, "file_required", 'multipart form field "file" is required');
}
let extracted;
try {
extracted = await extractPdfText(file.buffer);
} catch (e) {
if (e instanceof PdfExtractionError) {
return jsonError(res, 422, e.reason, e.message);
}
return jsonError(res, 500, "extract_failed", e instanceof Error ? e.message : String(e));
}
if (extracted.pageCount > caps.maxPages) {
return jsonError(
res,
422,
"too_many_pages",
`${extracted.pageCount} pages exceeds the ${caps.maxPages}-page cap`,
);
}
const chunks = chunkText(extracted.text, {
targetTokens: caps.chunkTargetTokens,
maxTokens: caps.chunkMaxTokens,
});
if (chunks.length === 0) {
return jsonError(res, 422, "no_text", "no extractable text found in the PDF");
}
if (chunks.length > caps.maxChunks) {
return jsonError(
res,
422,
"too_many_chunks",
`${chunks.length} chunks exceeds the ${caps.maxChunks}-chunk cap`,
);
}
const job = jobStore.createJob(file.originalname || "book.pdf", extracted.pageCount, chunks);
return res.status(200).json({
job_id: job.jobId,
filename: job.filename,
page_count: extracted.pageCount,
chunk_count: chunks.length,
sample_chunk: chunks[0].text.slice(0, 500),
});
});
});
// ── GET /v1/pdf-jobs/:id/estimate ───────────────────────────────────
app.get("/v1/pdf-jobs/:id/estimate", async (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id));
if (!job) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`);
try {
const est = await jobStore.estimate(job);
return res.status(200).json(est);
} catch (err) {
return jsonError(res, 502, "estimate_failed", err instanceof Error ? err.message : String(err));
}
});
// ── POST /v1/pdf-jobs/:id/start ─────────────────────────────────────
app.post("/v1/pdf-jobs/:id/start", async (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id));
if (!job) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`);
const body = (typeof req.body === "object" && req.body !== null ? req.body : {}) as {
confirm?: unknown;
};
if (body.confirm !== true) {
return jsonError(res, 400, "not_confirmed", "body must include { confirm: true }");
}
if (job.status !== "estimated") {
return jsonError(res, 409, "already_started", `job is already ${job.status}`);
}
let est;
try {
est = await jobStore.estimate(job);
} catch (err) {
return jsonError(res, 502, "estimate_failed", err instanceof Error ? err.message : String(err));
}
if (est.no_capable_suppliers) {
return jsonError(res, 409, "no_capable_suppliers", "no capable suppliers available right now");
}
if (est.would_drop_below_floor) {
return jsonError(
res,
402,
"below_wallet_floor",
`projected spend would drop the wallet below the ${est.wallet_floor_lovelace} lovelace floor`,
);
}
jobStore.start(job);
return res.status(202).json({ job_id: job.jobId, status: "running" });
});
// ── GET /v1/pdf-jobs/:id/events (SSE) ───────────────────────────────
app.get("/v1/pdf-jobs/:id/events", (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id));
if (!job) {
jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`);
return;
}
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache, no-transform");
res.setHeader("Connection", "keep-alive");
// Defeat proxy buffering (Traefik/nginx/Cloudflare) so frames flush live.
res.setHeader("X-Accel-Buffering", "no");
if (typeof (res as Response & { flushHeaders?: () => void }).flushHeaders === "function") {
(res as Response & { flushHeaders: () => void }).flushHeaders();
}
const unsubscribe = job.subscribe((frame) => res.write(frame));
const keepAlive = setInterval(() => res.write(": keep-alive\n\n"), 15_000);
req.on("close", () => {
clearInterval(keepAlive);
unsubscribe();
});
});
// ── GET /v1/pdf-jobs/:id/summary.md ─────────────────────────────────
app.get("/v1/pdf-jobs/:id/summary.md", (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id));
if (!job) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`);
if (!job.finalSummary) {
return jsonError(res, 404, "not_ready", "summary not available yet");
}
const safeName = job.filename.replace(/\.pdf$/i, "").replace(/[^A-Za-z0-9._-]+/g, "_") || "summary";
res.setHeader("Content-Type", "text/markdown; charset=utf-8");
res.setHeader("Content-Disposition", `attachment; filename="${safeName}-summary.md"`);
return res.send(job.finalSummary);
});
// ── GET /v1/pdf-jobs/:id ────────────────────────────────────────────
app.get("/v1/pdf-jobs/:id", (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id));
if (!job) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`);
return res.status(200).json(job.view());
});
}

View File

@@ -0,0 +1,525 @@
/**
* buyer/src/pdf/summarize-job.ts — the summarization orchestrator + in-memory
* job registry.
*
* One job = one uploaded book. The worker:
* MAP — summarize each chunk (one paid escrow per chunk).
* REDUCE — hierarchically collapse chunk summaries (fan-in F) up a tree to a
* single book summary; each reduce node is another paid escrow.
*
* Each call reuses the exact run-cycle.ts sequence: submitPrompt (PostEscrow →
* supplier → receipt-verify) → findSubmittedEscrowRef (indexer) → runAccept.
*
* Concurrency: v1 is serial. submitPrompt+accept run under a global mutex
* because the shared tx builder can't pin wallet inputs yet, so two in-flight
* PostEscrow txs would select the same wallet UTxO and double-spend. laneCount
* is honored for dispatch but the mutex keeps chain ops serialized regardless
* — the parallel fast-follow lifts the mutex once PostEscrow takes
* presetWalletInputs.
*
* Failure policy: a failed call retries on a *different* supplier up to
* caps.retryK times; if still failing it becomes a "gap" (excluded, job
* continues). Orphaned escrows from failed calls are recovered by the existing
* `reclaim:orphans` cron (submitPrompt throws before returning a ref, so we
* can't reclaim inline).
*
* Job state is in-memory only (lost on restart); already-posted escrows remain
* recoverable via the indexer + reclaim cron.
*/
import { randomUUID } from "crypto";
import { canonicalize } from "@marketplace/shared/cbor";
import type { ChainProvider, OutputReference } from "@marketplace/shared/chain";
import type { WalletKey } from "@marketplace/shared/tx";
import type { Marketplace } from "../sdk/Marketplace.js";
import type { ResponseArchive } from "../db/archive.js";
import { runAccept } from "../cli/acceptFlow.js";
import { SUMMARIZE_CAPABILITY_ID, buildSupplierPool, SupplierPool } from "./supplier-pool.js";
import { estimateJob, feeHeadroomLovelace, type JobEstimate } from "./estimate.js";
import type { Chunk, ChunkResult, JobPhase, JobView, PdfCaps, PoolSupplier } from "./types.js";
const ESCROW_REF_RE = /^([0-9a-f]{64})#(\d+)$/;
const SUBMITTED_LOOKUP_TIMEOUT_MS = 30_000;
/** The result of running one supplier call's full on-chain lifecycle. */
export interface CallOutcome {
response: string;
escrowRef: string;
supplierPkh: string;
model: string;
receipt: Record<string, unknown>;
receiptSignature: string;
}
/** Runs one call (PostEscrow → supplier → verify → Accept). Injectable for
* tests so the orchestrator can be exercised without lucid/chain. */
export type RunCallFn = (supplier: PoolSupplier, prompt: string) => Promise<CallOutcome>;
export interface JobDeps {
marketplace: Marketplace;
chain: ChainProvider;
walletKey: WalletKey;
indexerUrl: string;
archive?: ResponseArchive;
caps: PdfCaps;
/** Override the per-call lifecycle (tests). Defaults to the real
* submitPrompt → findSubmittedEscrowRef → runAccept sequence. */
runCall?: RunCallFn;
/** Override the wallet-balance read (tests). Defaults to chain query. */
walletBalance?: () => Promise<bigint>;
}
export interface ProgressEvent {
phase: "map" | "reduce" | "done" | "error";
label?: string;
completed: number;
failed: number;
total: number;
reduce_level?: number;
running_cost_lovelace: string;
coverage: { done: number; total: number };
message?: string;
status?: JobPhase;
}
export interface EstimateView extends JobEstimate {
suppliers: { model: string; price_lovelace: string }[];
wallet_balance_lovelace: string;
wallet_floor_lovelace: string;
would_drop_below_floor: boolean;
no_capable_suppliers: boolean;
}
/** Minimal async mutex — serializes the chain-touching critical section. */
class Mutex {
private tail: Promise<void> = Promise.resolve();
run<T>(fn: () => Promise<T>): Promise<T> {
const result = this.tail.then(fn);
// Keep the chain alive but swallow errors so one failure doesn't poison
// the queue; callers observe the real result/throw via `result`.
this.tail = result.then(
() => undefined,
() => undefined,
);
return result;
}
}
async function walletLovelace(chain: ChainProvider, address: string): Promise<bigint> {
const utxos = await chain.queryUtxosByAddress(address);
return utxos.reduce((acc, u) => acc + BigInt(u.lovelace), 0n);
}
/**
* Poll the indexer for the current Submitted-state escrow ref matching this
* prompt_hash. submitPrompt returns the Open ref, which Claim+Submit have
* already spent — runAccept needs the live Submitted ref. Lifted from
* buyer/src/cli/run-cycle.ts.
*/
async function findSubmittedEscrowRef(
indexerUrl: string,
buyerPkh: string,
promptHash: string,
timeoutMs: number,
): Promise<OutputReference> {
const deadline = Date.now() + timeoutMs;
const base = indexerUrl.replace(/\/+$/, "");
while (Date.now() < deadline) {
const resp = await fetch(`${base}/escrows?buyer=${buyerPkh}`);
if (resp.ok) {
const rows = (await resp.json()) as Array<{
utxo_ref: string;
prompt_hash: string;
state: string;
submitted_at: number | null;
}>;
const match = rows
.filter((r) => r.prompt_hash === promptHash && r.state === "Submitted")
.sort((a, b) => (b.submitted_at ?? 0) - (a.submitted_at ?? 0))[0];
if (match) {
const m = ESCROW_REF_RE.exec(match.utxo_ref);
if (m) return { txHash: m[1], index: Number(m[2]) };
}
}
await new Promise((r) => setTimeout(r, 2_000));
}
throw new Error(
`indexer: no Submitted escrow with prompt_hash=${promptHash} within ${timeoutMs}ms`,
);
}
function mapPrompt(chunk: Chunk): string {
// The "[chunk N]" prefix salts the prompt so prompt_hash is unique even when
// two chunks share identical text — the indexer Submitted lookup matches on
// hash and would otherwise be ambiguous.
return (
`[chunk ${chunk.index + 1}] You are summarizing one section of a longer book. ` +
`Write a faithful, self-contained summary of the passage below in clear prose — ` +
`capture the key events, arguments, characters, and conclusions, and add nothing ` +
`that is not present.\n\nPASSAGE:\n${chunk.text}`
);
}
function reducePrompt(level: number, node: number, summaries: string[]): string {
const joined = summaries
.map((s, i) => `--- Section ${i + 1} ---\n${s}`)
.join("\n\n");
return (
`[reduce L${level}.${node}] Below are summaries of consecutive sections of a book, ` +
`in order. Synthesize them into a single coherent summary that preserves the overall ` +
`narrative/argument and chronological flow. Be faithful; do not invent.\n\nSECTIONS:\n${joined}`
);
}
function groupBy<T>(items: T[], size: number): T[][] {
const out: T[][] = [];
for (let i = 0; i < items.length; i += size) out.push(items.slice(i, i + size));
return out;
}
async function runWithConcurrency<T>(
items: T[],
worker: (item: T) => Promise<void>,
concurrency: number,
): Promise<void> {
let next = 0;
const lanes = Array.from({ length: Math.max(1, concurrency) }, async () => {
for (;;) {
const idx = next++;
if (idx >= items.length) return;
await worker(items[idx]);
}
});
await Promise.all(lanes);
}
export class Job {
readonly jobId: string;
readonly filename: string;
readonly pageCount: number;
readonly chunks: Chunk[];
status: JobPhase = "estimated";
runningCost = 0n;
finalSummary?: string;
coverageDone = 0;
failedCount = 0;
readonly escrowRefs: string[] = [];
/** Map results indexed by chunk index; reduce results appended after. */
readonly chunkResults: ChunkResult[];
private subscribers = new Set<(frame: string) => void>();
private frames: string[] = [];
constructor(filename: string, pageCount: number, chunks: Chunk[]) {
this.jobId = `job_${randomUUID()}`;
this.filename = filename;
this.pageCount = pageCount;
this.chunks = chunks;
this.chunkResults = chunks.map((c) => ({
index: c.index,
label: `map:${c.index}`,
status: "pending" as const,
}));
}
get coverageTotal(): number {
return this.chunks.length;
}
emit(ev: ProgressEvent): void {
const eventName = ev.phase === "done" ? "done" : "progress";
const frame = `event: ${eventName}\ndata: ${JSON.stringify(ev)}\n\n`;
this.frames.push(frame);
if (this.frames.length > 1000) this.frames.shift();
for (const send of this.subscribers) {
try {
send(frame);
} catch {
/* dropped client; route's close handler unsubscribes */
}
}
}
/** Attach an SSE writer; replays buffered frames so reconnects catch up. */
subscribe(send: (frame: string) => void): () => void {
for (const f of this.frames) send(f);
this.subscribers.add(send);
return () => {
this.subscribers.delete(send);
};
}
view(): JobView {
return {
job_id: this.jobId,
filename: this.filename,
status: this.status,
page_count: this.pageCount,
chunk_count: this.chunks.length,
coverage: { done: this.coverageDone, total: this.coverageTotal },
running_cost_lovelace: this.runningCost.toString(),
final_summary_md: this.finalSummary,
chunk_results: this.chunkResults,
escrow_refs: this.escrowRefs,
};
}
}
export class JobStore {
private readonly jobs = new Map<string, Job>();
private readonly started = new Set<string>();
private readonly chainMutex = new Mutex();
constructor(private readonly deps: JobDeps) {}
createJob(filename: string, pageCount: number, chunks: Chunk[]): Job {
const job = new Job(filename, pageCount, chunks);
this.jobs.set(job.jobId, job);
return job;
}
get(id: string): Job | undefined {
return this.jobs.get(id);
}
/** Discover suppliers + read balance to produce the pre-spend estimate. */
async estimate(job: Job): Promise<EstimateView> {
const pool = await buildSupplierPool(this.deps.marketplace, this.deps.caps);
const est = estimateJob(job.chunks.length, this.deps.caps.reduceFanin, pool);
const balance = this.deps.walletBalance
? await this.deps.walletBalance()
: await walletLovelace(this.deps.chain, this.deps.walletKey.address);
const projected = BigInt(est.totalLovelace) + feeHeadroomLovelace(est.totalCalls);
const wouldDrop = balance - projected < this.deps.caps.walletFloorLovelace;
return {
...est,
suppliers: pool.all().map((s) => ({
model: s.model,
price_lovelace: s.priceLovelace.toString(),
})),
wallet_balance_lovelace: balance.toString(),
wallet_floor_lovelace: this.deps.caps.walletFloorLovelace.toString(),
would_drop_below_floor: wouldDrop,
no_capable_suppliers: pool.size === 0,
};
}
/** Kick off the background worker. Idempotent: a second call is a no-op. */
start(job: Job): void {
if (this.started.has(job.jobId)) return;
this.started.add(job.jobId);
job.status = "running";
void this.run(job).catch((err) => {
job.status = "failed";
// Terminal: emit a `done` frame so SSE clients stop waiting; carries the
// failure message + status so the UI can surface it.
job.emit(
this.event(job, "done", {
status: "failed",
message: err instanceof Error ? err.message : String(err),
}),
);
});
}
private async run(job: Job): Promise<void> {
const { caps } = this.deps;
const pool = await buildSupplierPool(this.deps.marketplace, caps);
if (pool.size === 0) {
job.status = "failed";
job.emit(this.event(job, "done", { message: "no_capable_suppliers", status: "failed" }));
return;
}
// ── MAP ────────────────────────────────────────────────────────────
await runWithConcurrency(
job.chunks,
async (chunk) => {
const r = await this.doOneCall(job, pool, "map", `map:${chunk.index}`, mapPrompt(chunk));
const row = job.chunkResults[chunk.index];
if (r.ok) {
row.status = "ok";
row.summary = r.summary;
row.escrowRef = r.escrowRef;
row.supplierModel = r.model;
job.coverageDone++;
} else {
row.status = "gap";
job.failedCount++;
}
job.emit(this.event(job, "map", { label: row.label }));
},
caps.laneCount,
);
// ── REDUCE (hierarchical) ───────────────────────────────────────────
let summaries = job.chunkResults
.filter((r) => r.status === "ok" && typeof r.summary === "string")
.map((r) => r.summary as string);
let level = 0;
while (summaries.length > 1) {
level++;
const groups = groupBy(summaries, caps.reduceFanin);
const nextLevel: (string | undefined)[] = new Array(groups.length);
await runWithConcurrency(
groups.map((g, i) => ({ g, i })),
async ({ g, i }) => {
const label = `reduce:${level}.${i}`;
const r = await this.doOneCall(job, pool, "reduce", label, reducePrompt(level, i, g));
const row: ChunkResult = {
index: -1,
label,
status: r.ok ? "ok" : "gap",
escrowRef: r.escrowRef,
supplierModel: r.model,
summary: r.summary,
};
job.chunkResults.push(row);
if (r.ok) {
nextLevel[i] = r.summary;
} else {
job.failedCount++;
}
job.emit(this.event(job, "reduce", { label, reduce_level: level }));
},
caps.laneCount,
);
summaries = nextLevel.filter((s): s is string => typeof s === "string");
}
job.finalSummary = summaries[0] ?? this.assembleFromPartials(job);
const hasSummary = typeof job.finalSummary === "string" && job.finalSummary.length > 0;
if (!hasSummary) {
job.status = "failed";
} else if (job.coverageDone < job.coverageTotal || job.failedCount > 0) {
job.status = "completed_with_gaps";
} else {
job.status = "completed";
}
// Per-call chats are already persisted in doOneCall; the final summary
// lives in job state and is downloadable via GET .../summary.md.
job.emit(this.event(job, "done", { status: job.status }));
}
/** The real per-call lifecycle: submitPrompt → resolve Submitted → Accept. */
private async defaultRunCall(sup: PoolSupplier, prompt: string): Promise<CallOutcome> {
const submit = await this.deps.marketplace.submitPrompt({
advertRef: sup.advertRef,
messages: [{ role: "user", content: prompt }],
payment_lovelace: sup.priceLovelace,
max_output_tokens: sup.maxOutputTokens,
});
const submittedRef = await findSubmittedEscrowRef(
this.deps.indexerUrl,
this.deps.walletKey.pubKeyHash,
submit.receipt.prompt_hash,
SUBMITTED_LOOKUP_TIMEOUT_MS,
);
await runAccept({
chain: this.deps.chain,
walletKey: this.deps.walletKey,
escrowRef: submittedRef,
});
return {
response: submit.response,
escrowRef: `${submit.escrowRef.txHash}#${submit.escrowRef.index}`,
supplierPkh: submit.receipt.supplier_pkh,
model: submit.receipt.model,
receipt: submit.receipt as unknown as Record<string, unknown>,
receiptSignature: submit.receiptSignature,
};
}
/** One map/reduce call: round-robin a supplier, run the full lifecycle.
* Serialized through chainMutex so concurrent PostEscrow txs never select
* the same wallet UTxO (v1 has no input pinning). */
private async doOneCall(
job: Job,
pool: SupplierPool,
phase: "map" | "reduce",
label: string,
prompt: string,
): Promise<{ ok: boolean; summary?: string; escrowRef?: string; model?: string }> {
const runCall: RunCallFn = this.deps.runCall ?? ((s, p) => this.defaultRunCall(s, p));
const tried = new Set<string>();
for (let attempt = 0; attempt <= this.deps.caps.retryK; attempt++) {
const sup = pool.next(tried);
if (!sup) break; // ran out of distinct suppliers
tried.add(sup.utxoRef);
try {
const result = await this.chainMutex.run(() => runCall(sup, prompt));
job.runningCost += sup.priceLovelace;
job.escrowRefs.push(result.escrowRef);
if (this.deps.archive) {
try {
const canonicalAssistant = canonicalize({
role: "assistant",
content: result.response,
});
this.deps.archive.persistChat({
escrow_ref: result.escrowRef,
posted_at: Date.now(),
capability_id: SUMMARIZE_CAPABILITY_ID,
supplier_pkh: result.supplierPkh,
model: result.model,
payment_lovelace: sup.priceLovelace.toString(),
request_messages: [{ role: "user", content: prompt }],
response_canonical: canonicalAssistant,
receipt: result.receipt,
receipt_signature: result.receiptSignature,
});
} catch (err) {
// archive is best-effort; the on-chain receipt is authoritative
// eslint-disable-next-line no-console
console.error(`[pdf] archive.persistChat failed for ${label}:`, err);
}
}
return { ok: true, summary: result.response, escrowRef: result.escrowRef, model: sup.model };
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
job.emit(
this.event(job, phase, {
label,
message: `retry ${attempt + 1}/${this.deps.caps.retryK + 1} on new supplier: ${message}`,
}),
);
// Orphaned escrow (if PostEscrow landed) is recovered by reclaim:orphans.
}
}
return { ok: false };
}
/** Fallback book summary built from surviving map summaries, gaps marked. */
private assembleFromPartials(job: Job): string {
const parts: string[] = [];
for (const r of job.chunkResults.filter((x) => x.index >= 0)) {
if (r.status === "ok" && r.summary) {
parts.push(`## Section ${r.index + 1}\n\n${r.summary}`);
} else {
parts.push(`## Section ${r.index + 1}\n\n_[gap: section omitted — summarization failed]_`);
}
}
return parts.join("\n\n");
}
private event(
job: Job,
phase: ProgressEvent["phase"],
extra: Partial<ProgressEvent> = {},
): ProgressEvent {
return {
phase,
completed: job.coverageDone,
failed: job.failedCount,
total: job.coverageTotal,
running_cost_lovelace: job.runningCost.toString(),
coverage: { done: job.coverageDone, total: job.coverageTotal },
...extra,
};
}
}

View File

@@ -0,0 +1,120 @@
/**
* buyer/src/pdf/supplier-pool.ts — discover + filter + round-robin the
* capable suppliers a summarization job pays.
*
* "Capable" = a llm.text.generate.v1 supplier whose model passes the
* allow/deny filter (e.g. exclude the weak local qwen2.5:0.5b). The pool
* round-robins across them so traffic is spread over multiple agents and a
* retry always lands on a *different* supplier than the one that just failed.
*/
import type { Marketplace } from "../sdk/Marketplace.js";
import type { SupplierView } from "../sdk/types.js";
import type { PdfCaps, PoolSupplier } from "./types.js";
export const SUMMARIZE_CAPABILITY_ID = "llm.text.generate.v1";
const REF_RE = /^([0-9a-f]{64})#(\d+)$/;
function parseRef(ref: string): { txHash: string; index: number } | null {
const m = REF_RE.exec(ref);
return m ? { txHash: m[1], index: Number(m[2]) } : null;
}
/** Is `model` capable given the allow/deny substring lists? Deny wins. */
export function isCapableModel(
model: string,
allow: string[],
deny: string[],
): boolean {
const m = model.toLowerCase();
if (deny.some((d) => d.length > 0 && m.includes(d.toLowerCase()))) return false;
if (allow.length === 0) return true;
return allow.some((a) => a.length > 0 && m.includes(a.toLowerCase()));
}
/** Filter raw indexer views down to capable, payable suppliers. */
export function filterSuppliers(
views: SupplierView[],
caps: PdfCaps,
): PoolSupplier[] {
const out: PoolSupplier[] = [];
const seen = new Set<string>();
for (const v of views) {
if (v.capability_id !== SUMMARIZE_CAPABILITY_ID) continue;
if (!isCapableModel(v.model, caps.modelAllowlist, caps.modelDenylist)) continue;
if (seen.has(v.utxo_ref)) continue;
const ref = parseRef(v.utxo_ref);
if (!ref) continue; // demo/synthetic entries have non-parseable refs
let price: bigint;
try {
price = BigInt(v.price_lovelace);
} catch {
continue;
}
seen.add(v.utxo_ref);
out.push({
advertRef: ref,
utxoRef: v.utxo_ref,
supplierPkh: v.supplier_pkh,
model: v.model,
priceLovelace: price,
maxOutputTokens: v.max_output_tokens,
endpointUrl: v.endpoint_url,
});
}
return out;
}
export class SupplierPool {
private readonly suppliers: PoolSupplier[];
private cursor = 0;
constructor(suppliers: PoolSupplier[]) {
this.suppliers = suppliers;
}
get size(): number {
return this.suppliers.length;
}
all(): PoolSupplier[] {
return this.suppliers.slice();
}
/** Upper bound for cost estimates: the priciest supplier in the pool. */
maxPrice(): bigint {
return this.suppliers.reduce(
(max, s) => (s.priceLovelace > max ? s.priceLovelace : max),
0n,
);
}
/**
* Round-robin the next supplier, skipping any whose utxoRef is in `exclude`.
* Returns null only when every supplier is excluded (caller treats as
* "out of suppliers" → gap).
*/
next(exclude?: Set<string>): PoolSupplier | null {
const n = this.suppliers.length;
if (n === 0) return null;
for (let i = 0; i < n; i++) {
const s = this.suppliers[(this.cursor + i) % n];
if (!exclude || !exclude.has(s.utxoRef)) {
this.cursor = (this.cursor + i + 1) % n;
return s;
}
}
return null;
}
}
export async function buildSupplierPool(
marketplace: Marketplace,
caps: PdfCaps,
): Promise<SupplierPool> {
const views = await marketplace.discoverSuppliers({
capability_id: SUMMARIZE_CAPABILITY_ID,
});
return new SupplierPool(filterSuppliers(views, caps));
}

16
buyer/src/pdf/tokens.ts Normal file
View File

@@ -0,0 +1,16 @@
/**
* buyer/src/pdf/tokens.ts — single token-counting helper, shared by the
* chunker (chunk.ts) and the cost estimator (estimate.ts) so the estimate
* matches the chunk count exactly.
*
* Uses gpt-tokenizer's default cl100k_base encoder — pure JS, no native deps,
* no network. The exact tokenizer a given supplier model uses may differ, but
* cl100k is a good universal proxy for chunk sizing and cost bounds.
*/
import { encode } from "gpt-tokenizer";
export function countTokens(s: string): number {
if (s.length === 0) return 0;
return encode(s).length;
}

103
buyer/src/pdf/types.ts Normal file
View File

@@ -0,0 +1,103 @@
/**
* buyer/src/pdf/types.ts — shared types + errors for the PDF book summarizer.
*
* The summarizer is a server-side feature bolted onto the buyer web app: an
* operator uploads a PDF, the server chunks it and runs a hierarchical
* map-reduce summarization by paying marketplace suppliers per chunk (one
* escrow per map call + per reduce node). See pdf/summarize-job.ts for the
* orchestrator and pdf/routes.ts for the HTTP surface.
*/
import type { OutputReference } from "@marketplace/shared/chain";
/** One paragraph-aligned slice of the book, fed to a supplier as one prompt. */
export interface Chunk {
/** 0-based position in the book — also salts the map prompt so prompt_hash
* is unique even for identical text (the indexer lookup matches on hash). */
index: number;
text: string;
tokenEstimate: number;
}
/** Operator-tunable safety + sizing knobs, resolved from env at boot. */
export interface PdfCaps {
/** Reject books with more pages than this (hard cap). */
maxPages: number;
/** Reject books that chunk into more than this many calls (hard cap). */
maxChunks: number;
/** Refuse to start a job if projected spend would drop the wallet below
* this lovelace balance. */
walletFloorLovelace: bigint;
/** Concurrency for the orchestrator. v1 ships 1 (serial PostEscrow — the
* shared tx builder can't pin wallet inputs yet, so parallel escrows would
* double-spend). Kept configurable for the parallel fast-follow. */
laneCount: number;
/** Target tokens per chunk (greedy packing). */
chunkTargetTokens: number;
/** Hard ceiling per chunk; oversize paragraphs split on sentence bounds. */
chunkMaxTokens: number;
/** Reduce-tree fan-in: how many summaries collapse into one per level. */
reduceFanin: number;
/** Per-call retries on a *different* supplier before a chunk is a gap. */
retryK: number;
/** Model-name substrings considered capable (empty = allow all). */
modelAllowlist: string[];
/** Model-name substrings to always exclude (e.g. weak local models). */
modelDenylist: string[];
/** Max upload size in bytes (multer fileSize). */
maxPdfBytes: number;
}
/** A discovered, capable supplier ready to be paid. */
export interface PoolSupplier {
advertRef: OutputReference;
utxoRef: string; // "<txhash>#<index>"
supplierPkh: string;
model: string;
priceLovelace: bigint;
maxOutputTokens: number;
endpointUrl: string;
}
export type JobPhase =
| "estimated"
| "running"
| "completed"
| "completed_with_gaps"
| "failed";
/** Per-call result row, surfaced to the UI. */
export interface ChunkResult {
/** chunk index for map calls; -1 for reduce nodes. */
index: number;
/** "map:12" or "reduce:1.3". */
label: string;
status: "ok" | "gap" | "pending";
escrowRef?: string;
supplierModel?: string;
summary?: string;
}
/** JSON view returned by GET /v1/pdf-jobs/:id. */
export interface JobView {
job_id: string;
filename: string;
status: JobPhase;
page_count: number;
chunk_count: number;
coverage: { done: number; total: number };
running_cost_lovelace: string;
final_summary_md?: string;
chunk_results: ChunkResult[];
escrow_refs: string[];
}
/** Thrown by extractPdfText for unusable PDFs (e.g. scanned/image-only). */
export class PdfExtractionError extends Error {
readonly reason: string;
constructor(reason: string, message?: string) {
super(message ?? reason);
this.name = "PdfExtractionError";
this.reason = reason;
}
}

View File

@@ -28,6 +28,10 @@ import type { ChatMessage } from "@marketplace/shared/tx";
import { canonicalize } from "@marketplace/shared/cbor";
import { runAccept } from "./cli/acceptFlow.js";
import type { ResponseArchive } from "./db/archive.js";
import { registerPdfRoutes } from "./pdf/routes.js";
import { defaultPdfCaps } from "./pdf/caps.js";
import type { PdfCaps } from "./pdf/types.js";
import type { JobStore } from "./pdf/summarize-job.js";
import {
SESSION_COOKIE_NAME,
SESSION_MAX_AGE_MS,
@@ -99,6 +103,12 @@ export interface AppDeps {
* brute-force rate-limit window. Defaults to Date.now. Tests use a fake
* clock to drive expiry and rolling-renewal assertions deterministically. */
nowMs?: () => number;
/** PDF book summarizer job registry. When provided, the /v1/pdf-* routes
* are live; when omitted they respond 503 (feature disabled). Built in
* runMain from the same marketplace/chain/wallet/archive deps. */
jobStore?: JobStore;
/** Sizing/safety knobs for the PDF summarizer. Defaults applied when absent. */
pdfCaps?: PdfCaps;
}
interface ResolvedDeps {
@@ -943,6 +953,12 @@ export function createApp(deps: AppDeps): Express {
});
}
// ── PDF book summarizer (/v1/pdf-*) ─────────────────────────────────
// Registered after the other /v1/* routes (so it's behind the same auth
// gate) and before the SPA static fallback. When jobStore is absent the
// routes respond 503 (feature disabled).
registerPdfRoutes(app, deps.jobStore, deps.pdfCaps ?? defaultPdfCaps());
if (deps.distPath) {
let isDir = false;
try {

View File

@@ -13,6 +13,7 @@ import Dashboard from "./pages/Dashboard.js";
import TaskHistory from "./pages/TaskHistory.js";
import Wallet from "./pages/Wallet.js";
import PendingReceipts from "./pages/PendingReceipts.js";
import BookSummarizer from "./pages/BookSummarizer.js";
import { RequireAuth, useAuth } from "./state/AuthContext.js";
function SignOutButton() {
@@ -34,6 +35,7 @@ export default function App() {
<div className="min-h-screen bg-gray-50 text-gray-900">
<nav className="bg-white border-b border-gray-200 px-6 py-3 flex items-center gap-6">
<Link to="/" className="font-medium hover:underline">Dashboard</Link>
<Link to="/book" className="font-medium hover:underline">Book Summarizer</Link>
<Link to="/tasks" className="font-medium hover:underline">Tasks</Link>
<Link to="/pending" className="font-medium hover:underline">Pending</Link>
<Link to="/wallet" className="font-medium hover:underline">Wallet</Link>
@@ -42,6 +44,7 @@ export default function App() {
<main className="p-6">
<Routes>
<Route path="/" element={<Dashboard />} />
<Route path="/book" element={<BookSummarizer />} />
<Route path="/tasks" element={<TaskHistory />} />
<Route path="/pending" element={<PendingReceipts />} />
<Route path="/wallet" element={<Wallet />} />

View File

@@ -0,0 +1,133 @@
/**
* buyer/src/ui/components/JobEstimate.tsx — step 2: show the pre-spend cost
* estimate and require an explicit "Proceed" before any escrow is posted.
*
* GET /v1/pdf-jobs/:id/estimate returns the map+reduce call count, an
* upper-bound AP3X cost, the wallet balance, and whether the spend would drop
* the wallet below its floor (in which case Proceed is disabled).
*/
import { useEffect, useState } from "react";
import type { UploadInfo } from "./PdfUploadForm.js";
interface EstimateData {
mapCalls: number;
reduceCalls: number;
totalCalls: number;
perCallMaxLovelace: string;
totalLovelace: string;
totalAp3x: string;
suppliers: { model: string; price_lovelace: string }[];
wallet_balance_lovelace: string;
wallet_floor_lovelace: string;
would_drop_below_floor: boolean;
no_capable_suppliers: boolean;
}
function ap3x(lovelace: string): string {
return (Number(lovelace) / 1e6).toFixed(2);
}
export default function JobEstimate({
info,
onStarted,
}: {
info: UploadInfo;
onStarted: () => void;
}) {
const [est, setEst] = useState<EstimateData | null>(null);
const [error, setError] = useState<string | null>(null);
const [starting, setStarting] = useState(false);
useEffect(() => {
let cancelled = false;
fetch(`/v1/pdf-jobs/${info.job_id}/estimate`, { credentials: "same-origin" })
.then(async (r) => {
const body = await r.json();
if (!r.ok) throw new Error(`${body.error}: ${body.message ?? ""}`);
if (!cancelled) setEst(body as EstimateData);
})
.catch((e: Error) => !cancelled && setError(e.message));
return () => {
cancelled = true;
};
}, [info.job_id]);
const proceed = async (): Promise<void> => {
setStarting(true);
setError(null);
try {
const r = await fetch(`/v1/pdf-jobs/${info.job_id}/start`, {
method: "POST",
credentials: "same-origin",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ confirm: true }),
});
const body = await r.json();
if (!r.ok) throw new Error(`${body.error}: ${body.message ?? ""}`);
onStarted();
} catch (e) {
setError((e as Error).message);
setStarting(false);
}
};
if (error) {
return (
<div role="alert" className="rounded border border-red-300 bg-red-50 p-3 text-sm text-red-700">
{error}
</div>
);
}
if (!est) return <div className="text-sm text-gray-500">Estimating cost…</div>;
const blocked = est.would_drop_below_floor || est.no_capable_suppliers || est.totalCalls === 0;
return (
<div className="space-y-4">
<div className="rounded border border-gray-200 bg-white p-4 text-sm">
<h3 className="mb-2 font-medium">{info.filename}</h3>
<dl className="grid grid-cols-2 gap-x-6 gap-y-1 text-gray-700">
<dt>Pages</dt>
<dd className="text-right">{info.page_count}</dd>
<dt>Chunks (map calls)</dt>
<dd className="text-right">{est.mapCalls}</dd>
<dt>Reduce calls</dt>
<dd className="text-right">{est.reduceCalls}</dd>
<dt className="font-medium">Total paid calls</dt>
<dd className="text-right font-medium">{est.totalCalls}</dd>
<dt className="font-medium">Est. max cost</dt>
<dd className="text-right font-medium">{est.totalAp3x} AP3X</dd>
<dt>Wallet balance</dt>
<dd className="text-right">{ap3x(est.wallet_balance_lovelace)} AP3X</dd>
<dt>Wallet floor</dt>
<dd className="text-right">{ap3x(est.wallet_floor_lovelace)} AP3X</dd>
</dl>
<p className="mt-3 text-xs text-gray-500">
Suppliers: {est.suppliers.map((s) => `${s.model} (${ap3x(s.price_lovelace)})`).join(", ") || "none"}
</p>
</div>
{est.no_capable_suppliers && (
<div className="rounded border border-yellow-300 bg-yellow-50 p-3 text-sm text-yellow-900">
No capable suppliers are available right now — try again shortly.
</div>
)}
{est.would_drop_below_floor && (
<div className="rounded border border-red-300 bg-red-50 p-3 text-sm text-red-700">
This job would drop the wallet below its floor. Fund the wallet or
reduce the book size before proceeding.
</div>
)}
<button
type="button"
onClick={() => void proceed()}
disabled={blocked || starting}
className="rounded bg-indigo-600 px-4 py-2 text-sm font-medium text-white hover:bg-indigo-700 disabled:bg-gray-400"
>
{starting ? "Starting…" : `Proceed & pay (~${est.totalAp3x} AP3X)`}
</button>
</div>
);
}

View File

@@ -0,0 +1,86 @@
/**
* buyer/src/ui/components/JobProgress.tsx — step 3: live progress via SSE.
*
* Opens EventSource('/v1/pdf-jobs/:id/events'). The server replays buffered
* frames on connect, so a refresh mid-job catches up. On the terminal `done`
* event we close the stream and hand control back to the page.
*/
import { useEffect, useRef, useState } from "react";
interface ProgressFrame {
phase: "map" | "reduce" | "done" | "error";
label?: string;
completed: number;
failed: number;
total: number;
reduce_level?: number;
running_cost_lovelace: string;
coverage: { done: number; total: number };
message?: string;
status?: string;
}
function ap3x(lovelace: string): string {
return (Number(lovelace) / 1e6).toFixed(2);
}
export default function JobProgress({
jobId,
total,
onDone,
}: {
jobId: string;
total: number;
onDone: () => void;
}) {
const [frame, setFrame] = useState<ProgressFrame | null>(null);
const [log, setLog] = useState<string[]>([]);
const onDoneRef = useRef(onDone);
onDoneRef.current = onDone;
useEffect(() => {
const es = new EventSource(`/v1/pdf-jobs/${jobId}/events`);
es.addEventListener("progress", (e: MessageEvent) => {
const f = JSON.parse(e.data) as ProgressFrame;
setFrame(f);
if (f.message) setLog((prev) => [...prev.slice(-20), `${f.label ?? f.phase}: ${f.message}`]);
});
es.addEventListener("done", (e: MessageEvent) => {
setFrame(JSON.parse(e.data) as ProgressFrame);
es.close();
onDoneRef.current();
});
es.onerror = () => {
// Transient drop — EventSource auto-reconnects; the server replays
// buffered frames so we don't lose state.
};
return () => es.close();
}, [jobId]);
const done = frame?.coverage.done ?? 0;
const totalCount = frame?.total ?? total;
const pct = totalCount > 0 ? Math.round((done / totalCount) * 100) : 0;
return (
<div className="space-y-3">
<div className="h-3 w-full overflow-hidden rounded bg-gray-200">
<div className="h-full bg-indigo-600 transition-all" style={{ width: `${pct}%` }} />
</div>
<div className="flex justify-between text-sm text-gray-700">
<span>
{frame?.phase === "reduce" ? `Reducing (level ${frame.reduce_level ?? "?"})` : "Summarizing chunks"} —{" "}
{done}/{totalCount} ({pct}%)
</span>
<span>
{frame ? `${frame.failed} gaps · ${ap3x(frame.running_cost_lovelace)} AP3X spent` : "starting…"}
</span>
</div>
{log.length > 0 && (
<pre className="max-h-40 overflow-auto rounded border border-gray-200 bg-gray-50 p-2 text-xs text-gray-600">
{log.join("\n")}
</pre>
)}
</div>
);
}

View File

@@ -0,0 +1,81 @@
/**
* buyer/src/ui/components/PdfUploadForm.tsx — step 1 of the book summarizer.
*
* Posts a PDF as multipart/form-data to POST /v1/pdf-upload. The server
* extracts + chunks it (no escrow yet) and returns a job_id + chunk count.
* No private key or chain access in the browser — everything goes through the
* same-origin /v1/* server routes.
*/
import { useState } from "react";
export interface UploadInfo {
job_id: string;
filename: string;
page_count: number;
chunk_count: number;
sample_chunk: string;
}
export default function PdfUploadForm({ onUploaded }: { onUploaded: (info: UploadInfo) => void }) {
const [file, setFile] = useState<File | null>(null);
const [loading, setLoading] = useState(false);
const [error, setError] = useState<string | null>(null);
const handleSubmit = async (event: React.FormEvent): Promise<void> => {
event.preventDefault();
if (!file) return;
setLoading(true);
setError(null);
try {
const fd = new FormData();
fd.append("file", file);
const resp = await fetch("/v1/pdf-upload", {
method: "POST",
credentials: "same-origin",
body: fd,
});
const body = (await resp.json()) as UploadInfo & { error?: string; message?: string };
if (!resp.ok) {
throw new Error(`${body.error ?? resp.statusText}: ${body.message ?? ""}`);
}
onUploaded(body);
} catch (err) {
setError((err as Error).message ?? "upload failed");
} finally {
setLoading(false);
}
};
return (
<form onSubmit={handleSubmit} className="space-y-4">
<p className="text-sm text-gray-600">
Upload a text-based PDF book. The buyer will chunk it and pay marketplace
suppliers to summarize each chunk, then synthesize a full-book summary.
</p>
<input
type="file"
accept="application/pdf"
onChange={(e) => setFile(e.target.files?.[0] ?? null)}
disabled={loading}
aria-label="pdf file"
className="block text-sm text-gray-700 file:mr-3 file:rounded file:border-0 file:bg-indigo-600 file:px-4 file:py-2 file:text-white hover:file:bg-indigo-700"
/>
<div className="flex items-center gap-3">
<button
type="submit"
disabled={loading || !file}
className="rounded bg-indigo-600 px-4 py-2 text-sm font-medium text-white hover:bg-indigo-700 disabled:bg-gray-400"
>
Upload &amp; analyze
</button>
{loading && <span className="text-sm text-gray-500">Extracting &amp; chunking…</span>}
</div>
{error && (
<div role="alert" className="rounded border border-red-300 bg-red-50 p-3 text-sm text-red-700">
{error}
</div>
)}
</form>
);
}

View File

@@ -0,0 +1,164 @@
/**
* buyer/src/ui/pages/BookSummarizer.tsx — the PDF book summarizer wizard.
*
* Upload → Estimate/Confirm → Live progress (SSE) → Result.
*
* Each step talks only to the same-origin /v1/pdf-* server routes (the browser
* holds no key). One book upload fans out into many on-chain escrows — the
* traffic the marketplace wants — while producing a real book summary.
*/
import { useState } from "react";
import PdfUploadForm, { type UploadInfo } from "../components/PdfUploadForm.js";
import JobEstimate from "../components/JobEstimate.js";
import JobProgress from "../components/JobProgress.js";
type Step = "upload" | "estimate" | "running" | "done";
interface ChunkResult {
index: number;
label: string;
status: "ok" | "gap" | "pending";
escrowRef?: string;
supplierModel?: string;
}
interface JobView {
job_id: string;
filename: string;
status: string;
coverage: { done: number; total: number };
running_cost_lovelace: string;
final_summary_md?: string;
chunk_results: ChunkResult[];
escrow_refs: string[];
}
function ap3x(lovelace: string): string {
return (Number(lovelace) / 1e6).toFixed(2);
}
export default function BookSummarizer() {
const [step, setStep] = useState<Step>("upload");
const [info, setInfo] = useState<UploadInfo | null>(null);
const [view, setView] = useState<JobView | null>(null);
const [error, setError] = useState<string | null>(null);
const reset = (): void => {
setStep("upload");
setInfo(null);
setView(null);
setError(null);
};
const loadResult = async (jobId: string): Promise<void> => {
try {
const r = await fetch(`/v1/pdf-jobs/${jobId}`, { credentials: "same-origin" });
const body = await r.json();
if (!r.ok) throw new Error(`${body.error}: ${body.message ?? ""}`);
setView(body as JobView);
setStep("done");
} catch (e) {
setError((e as Error).message);
setStep("done");
}
};
return (
<div className="max-w-3xl space-y-6">
<div>
<h1 className="text-2xl font-semibold">Book Summarizer</h1>
<p className="mt-1 text-sm text-gray-500">
Upload a PDF → pay suppliers to summarize each chunk → get a full-book summary.
</p>
</div>
{step === "upload" && (
<PdfUploadForm
onUploaded={(i) => {
setInfo(i);
setStep("estimate");
}}
/>
)}
{step === "estimate" && info && (
<JobEstimate info={info} onStarted={() => setStep("running")} />
)}
{step === "running" && info && (
<div className="space-y-4">
<h2 className="text-lg font-medium">Summarizing “{info.filename}”…</h2>
<JobProgress
jobId={info.job_id}
total={info.chunk_count}
onDone={() => void loadResult(info.job_id)}
/>
<p className="text-xs text-gray-400">
Each chunk is a separate on-chain escrow (PostEscrow → supplier → Accept),
so this can take a while — that’s real marketplace traffic.
</p>
</div>
)}
{step === "done" && (
<div className="space-y-4">
{error && (
<div role="alert" className="rounded border border-red-300 bg-red-50 p-3 text-sm text-red-700">
{error}
</div>
)}
{view && (
<>
<div className="flex flex-wrap items-center gap-x-6 gap-y-1 text-sm">
<span className="font-medium">
{view.status === "completed"
? "✅ Completed"
: view.status === "completed_with_gaps"
? "⚠️ Completed with gaps"
: "❌ Failed"}
</span>
<span>
Coverage: {view.coverage.done}/{view.coverage.total} chunks
</span>
<span>Spent: {ap3x(view.running_cost_lovelace)} AP3X</span>
<span>{view.escrow_refs.length} escrows</span>
</div>
{view.final_summary_md && (
<a
href={`/v1/pdf-jobs/${view.job_id}/summary.md`}
className="inline-block rounded bg-indigo-600 px-4 py-2 text-sm font-medium text-white hover:bg-indigo-700"
>
Download summary (.md)
</a>
)}
{view.final_summary_md && (
<pre className="max-h-[28rem] overflow-auto whitespace-pre-wrap rounded border border-gray-200 bg-white p-4 text-sm text-gray-800">
{view.final_summary_md}
</pre>
)}
<details className="text-xs text-gray-500">
<summary className="cursor-pointer">On-chain escrows ({view.escrow_refs.length})</summary>
<ul className="mt-2 space-y-0.5 font-mono">
{view.escrow_refs.map((ref) => (
<li key={ref}>{ref}</li>
))}
</ul>
</details>
</>
)}
<button
type="button"
onClick={reset}
className="rounded border border-gray-300 px-4 py-2 text-sm font-medium text-gray-700 hover:bg-gray-100"
>
Summarize another book
</button>
</div>
)}
</div>
);
}