buyer/pdf: durable job store + past-work list + resume-on-return

The summarization runs server-side, but the SPA held the job only in
component state and jobs lived only in memory — so navigating away from
the Book Summarizer reset the UI (losing the handle to a still-running
job) and past summaries vanished on restart.

- PdfJobDb: SQLite store (its own file under ARCHIVE_DIR) persisting
  every job on each progress update.
- JobStore: createdAt, persist-on-update, viewOf()/list() that serve
  live jobs first then fall back to the durable record; non-live jobs
  whose worker died surface as a new "interrupted" status.
- routes: GET /v1/pdf-jobs (newest-first list); GET /:id + summary.md
  serve past jobs via viewOf; /:id/events sends a single terminal frame
  for a finished/past job (no hang) and live frames otherwise.
- SPA: RecentJobs list (open any job → reconnect to live progress or
  show result), localStorage resume of the active job on return, and a
  "back to list (keeps running)" affordance.
- mainnet buyer compose: persistent named volume + ARCHIVE_DIR so the
  archive + pdf-jobs.db survive container recreation/redeploys.

Tests: persistence-across-restart, interrupted detection, list, and
past-job serving (incl. terminal SSE). Full buyer suite green (185).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
david
2026-06-03 11:59:20 +02:00
parent c552c4caa3
commit cf77811331
10 changed files with 674 additions and 50 deletions

View File

@@ -17,6 +17,7 @@ import { buildChainProvider } from "./chain.js";
import { Marketplace, MemoryTaskHistoryStore } from "./sdk/index.js"; import { Marketplace, MemoryTaskHistoryStore } from "./sdk/index.js";
import { ResponseArchive } from "./db/archive.js"; import { ResponseArchive } from "./db/archive.js";
import { JobStore } from "./pdf/summarize-job.js"; import { JobStore } from "./pdf/summarize-job.js";
import { PdfJobDb } from "./pdf/job-db.js";
import { loadPdfCaps } from "./pdf/caps.js"; import { loadPdfCaps } from "./pdf/caps.js";
import type { WalletKey } from "@marketplace/shared/tx"; import type { WalletKey } from "@marketplace/shared/tx";
@@ -96,6 +97,19 @@ export async function runMain(env: Record<string, string | undefined>): Promise<
// PDF book summarizer: one job registry, sharing the live marketplace SDK, // PDF book summarizer: one job registry, sharing the live marketplace SDK,
// chain, wallet, and (optional) archive. Caps come from PDF_* env vars. // chain, wallet, and (optional) archive. Caps come from PDF_* env vars.
const pdfCaps = loadPdfCaps(env); const pdfCaps = loadPdfCaps(env);
// Durable job store (SQLite under ARCHIVE_DIR) so past summaries survive
// restarts and navigation. Best-effort: if it can't open, jobs fall back to
// in-memory only (lost on restart) but the feature still works.
let pdfJobDb: PdfJobDb | undefined;
try {
pdfJobDb = new PdfJobDb(config.archiveDir);
console.log(`[buyer] pdf job store at ${config.archiveDir}/pdf-jobs.db`);
} catch (err) {
console.error(
`[buyer] failed to open pdf job store at ${config.archiveDir}:`,
err instanceof Error ? err.message : String(err),
);
}
const jobStore = new JobStore({ const jobStore = new JobStore({
marketplace, marketplace,
chain, chain,
@@ -103,6 +117,7 @@ export async function runMain(env: Record<string, string | undefined>): Promise<
indexerUrl: config.indexerUrl, indexerUrl: config.indexerUrl,
archive, archive,
caps: pdfCaps, caps: pdfCaps,
db: pdfJobDb,
}); });
const app = createApp({ const app = createApp({

97
buyer/src/pdf/job-db.ts Normal file
View File

@@ -0,0 +1,97 @@
/**
* buyer/src/pdf/job-db.ts — durable persistence for PDF summarization jobs.
*
* Jobs are otherwise in-memory (lost on restart), so "past work" would vanish
* on every redeploy. This SQLite table (its own file under ARCHIVE_DIR, beside
* the response archive) records every job's metadata, final summary, per-chunk
* results, and escrow refs so the operator can list + reopen past summaries
* across restarts.
*
* Synchronous (better-sqlite3) like db/archive.ts — volume is low (a handful
* of writes per book) and the simplicity is worth more than async overhead.
*/
import { mkdirSync } from "fs";
import { join, resolve } from "path";
import Database from "better-sqlite3";
import type { Database as BetterDatabase } from "better-sqlite3";
export interface JobRecord {
job_id: string;
filename: string;
status: string;
page_count: number;
chunk_count: number;
coverage_done: number;
coverage_total: number;
running_cost_lovelace: string;
final_summary_md: string | null;
chunk_results_json: string;
escrow_refs_json: string;
created_at: number;
updated_at: number;
}
const CREATE_SQL = `
CREATE TABLE IF NOT EXISTS pdf_jobs (
job_id TEXT PRIMARY KEY,
filename TEXT NOT NULL,
status TEXT NOT NULL,
page_count INTEGER NOT NULL,
chunk_count INTEGER NOT NULL,
coverage_done INTEGER NOT NULL,
coverage_total INTEGER NOT NULL,
running_cost_lovelace TEXT NOT NULL,
final_summary_md TEXT,
chunk_results_json TEXT NOT NULL,
escrow_refs_json TEXT NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_pdf_jobs_created ON pdf_jobs (created_at DESC);
`;
export class PdfJobDb {
private readonly db: BetterDatabase;
private readonly upsertStmt;
private readonly listStmt;
private readonly getStmt;
constructor(archiveDir: string) {
const dir = resolve(archiveDir);
mkdirSync(dir, { recursive: true });
this.db = new Database(join(dir, "pdf-jobs.db"));
this.db.pragma("journal_mode = WAL");
this.db.exec(CREATE_SQL);
this.upsertStmt = this.db.prepare(`
INSERT OR REPLACE INTO pdf_jobs (
job_id, filename, status, page_count, chunk_count, coverage_done,
coverage_total, running_cost_lovelace, final_summary_md,
chunk_results_json, escrow_refs_json, created_at, updated_at
) VALUES (
@job_id, @filename, @status, @page_count, @chunk_count, @coverage_done,
@coverage_total, @running_cost_lovelace, @final_summary_md,
@chunk_results_json, @escrow_refs_json, @created_at, @updated_at
)
`);
this.listStmt = this.db.prepare(`SELECT * FROM pdf_jobs ORDER BY created_at DESC LIMIT @limit`);
this.getStmt = this.db.prepare(`SELECT * FROM pdf_jobs WHERE job_id = @job_id`);
}
upsert(rec: JobRecord): void {
this.upsertStmt.run(rec as unknown as Record<string, unknown>);
}
list(limit = 100): JobRecord[] {
return this.listStmt.all({ limit }) as JobRecord[];
}
get(jobId: string): JobRecord | null {
return (this.getStmt.get({ job_id: jobId }) as JobRecord | undefined) ?? null;
}
close(): void {
this.db.close();
}
}

View File

@@ -113,6 +113,11 @@ export function registerPdfRoutes(
}); });
}); });
// ── GET /v1/pdf-jobs — past-work list (newest first) ────────────────
app.get("/v1/pdf-jobs", (_req: Request, res: Response) => {
return res.status(200).json({ jobs: jobStore.list(100) });
});
// ── GET /v1/pdf-jobs/:id/estimate ─────────────────────────────────── // ── GET /v1/pdf-jobs/:id/estimate ───────────────────────────────────
app.get("/v1/pdf-jobs/:id/estimate", async (req: Request, res: Response) => { app.get("/v1/pdf-jobs/:id/estimate", async (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id)); const job = jobStore.get(String(req.params.id));
@@ -163,10 +168,14 @@ export function registerPdfRoutes(
}); });
// ── GET /v1/pdf-jobs/:id/events (SSE) ─────────────────────────────── // ── GET /v1/pdf-jobs/:id/events (SSE) ───────────────────────────────
// Live jobs stream real frames + replay; a finished/past (non-live) job
// emits a single terminal `done` frame so the client resolves immediately.
app.get("/v1/pdf-jobs/:id/events", (req: Request, res: Response) => { app.get("/v1/pdf-jobs/:id/events", (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id)); const id = String(req.params.id);
if (!job) { const job = jobStore.get(id);
jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`); const view = jobStore.viewOf(id);
if (!job && !view) {
jsonError(res, 404, "job_not_found", `no job ${id}`);
return; return;
} }
res.setHeader("Content-Type", "text/event-stream"); res.setHeader("Content-Type", "text/event-stream");
@@ -178,7 +187,23 @@ export function registerPdfRoutes(
(res as Response & { flushHeaders: () => void }).flushHeaders(); (res as Response & { flushHeaders: () => void }).flushHeaders();
} }
const unsubscribe = job.subscribe((frame) => res.write(frame)); if (!job && view) {
// Past/interrupted job: synthesize one terminal frame and end.
const frame = {
phase: "done",
completed: view.coverage.done,
failed: 0,
total: view.coverage.total,
running_cost_lovelace: view.running_cost_lovelace,
coverage: view.coverage,
status: view.status,
};
res.write(`event: done\ndata: ${JSON.stringify(frame)}\n\n`);
res.end();
return;
}
const unsubscribe = job!.subscribe((frame) => res.write(frame));
const keepAlive = setInterval(() => res.write(": keep-alive\n\n"), 15_000); const keepAlive = setInterval(() => res.write(": keep-alive\n\n"), 15_000);
req.on("close", () => { req.on("close", () => {
clearInterval(keepAlive); clearInterval(keepAlive);
@@ -188,21 +213,21 @@ export function registerPdfRoutes(
// ── GET /v1/pdf-jobs/:id/summary.md ───────────────────────────────── // ── GET /v1/pdf-jobs/:id/summary.md ─────────────────────────────────
app.get("/v1/pdf-jobs/:id/summary.md", (req: Request, res: Response) => { app.get("/v1/pdf-jobs/:id/summary.md", (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id)); const view = jobStore.viewOf(String(req.params.id));
if (!job) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`); if (!view) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`);
if (!job.finalSummary) { if (!view.final_summary_md) {
return jsonError(res, 404, "not_ready", "summary not available yet"); return jsonError(res, 404, "not_ready", "summary not available yet");
} }
const safeName = job.filename.replace(/\.pdf$/i, "").replace(/[^A-Za-z0-9._-]+/g, "_") || "summary"; const safeName = view.filename.replace(/\.pdf$/i, "").replace(/[^A-Za-z0-9._-]+/g, "_") || "summary";
res.setHeader("Content-Type", "text/markdown; charset=utf-8"); res.setHeader("Content-Type", "text/markdown; charset=utf-8");
res.setHeader("Content-Disposition", `attachment; filename="${safeName}-summary.md"`); res.setHeader("Content-Disposition", `attachment; filename="${safeName}-summary.md"`);
return res.send(job.finalSummary); return res.send(view.final_summary_md);
}); });
// ── GET /v1/pdf-jobs/:id ──────────────────────────────────────────── // ── GET /v1/pdf-jobs/:id ────────────────────────────────────────────
app.get("/v1/pdf-jobs/:id", (req: Request, res: Response) => { app.get("/v1/pdf-jobs/:id", (req: Request, res: Response) => {
const job = jobStore.get(String(req.params.id)); const view = jobStore.viewOf(String(req.params.id));
if (!job) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`); if (!view) return jsonError(res, 404, "job_not_found", `no job ${String(req.params.id)}`);
return res.status(200).json(job.view()); return res.status(200).json(view);
}); });
} }

View File

@@ -36,7 +36,16 @@ import type { ResponseArchive } from "../db/archive.js";
import { runAccept } from "../cli/acceptFlow.js"; import { runAccept } from "../cli/acceptFlow.js";
import { SUMMARIZE_CAPABILITY_ID, buildSupplierPool, SupplierPool } from "./supplier-pool.js"; import { SUMMARIZE_CAPABILITY_ID, buildSupplierPool, SupplierPool } from "./supplier-pool.js";
import { estimateJob, feeHeadroomLovelace, type JobEstimate } from "./estimate.js"; import { estimateJob, feeHeadroomLovelace, type JobEstimate } from "./estimate.js";
import type { Chunk, ChunkResult, JobPhase, JobView, PdfCaps, PoolSupplier } from "./types.js"; import type {
Chunk,
ChunkResult,
JobListItem,
JobPhase,
JobView,
PdfCaps,
PoolSupplier,
} from "./types.js";
import type { PdfJobDb, JobRecord } from "./job-db.js";
const ESCROW_REF_RE = /^([0-9a-f]{64})#(\d+)$/; const ESCROW_REF_RE = /^([0-9a-f]{64})#(\d+)$/;
const SUBMITTED_LOOKUP_TIMEOUT_MS = 30_000; const SUBMITTED_LOOKUP_TIMEOUT_MS = 30_000;
@@ -62,11 +71,16 @@ export interface JobDeps {
indexerUrl: string; indexerUrl: string;
archive?: ResponseArchive; archive?: ResponseArchive;
caps: PdfCaps; caps: PdfCaps;
/** Durable job store. When present, every job is persisted on each update
* so past work survives restarts and is listable/openable. */
db?: PdfJobDb;
/** Override the per-call lifecycle (tests). Defaults to the real /** Override the per-call lifecycle (tests). Defaults to the real
* submitPrompt → findSubmittedEscrowRef → runAccept sequence. */ * submitPrompt → findSubmittedEscrowRef → runAccept sequence. */
runCall?: RunCallFn; runCall?: RunCallFn;
/** Override the wallet-balance read (tests). Defaults to chain query. */ /** Override the wallet-balance read (tests). Defaults to chain query. */
walletBalance?: () => Promise<bigint>; walletBalance?: () => Promise<bigint>;
/** Injectable clock (tests). Defaults to Date.now. */
now?: () => number;
} }
export interface ProgressEvent { export interface ProgressEvent {
@@ -193,11 +207,20 @@ async function runWithConcurrency<T>(
await Promise.all(lanes); await Promise.all(lanes);
} }
function safeParse<T>(s: string, fallback: T): T {
try {
return JSON.parse(s) as T;
} catch {
return fallback;
}
}
export class Job { export class Job {
readonly jobId: string; readonly jobId: string;
readonly filename: string; readonly filename: string;
readonly pageCount: number; readonly pageCount: number;
readonly chunks: Chunk[]; readonly chunks: Chunk[];
readonly createdAt: number;
status: JobPhase = "estimated"; status: JobPhase = "estimated";
runningCost = 0n; runningCost = 0n;
finalSummary?: string; finalSummary?: string;
@@ -206,15 +229,19 @@ export class Job {
readonly escrowRefs: string[] = []; readonly escrowRefs: string[] = [];
/** Map results indexed by chunk index; reduce results appended after. */ /** Map results indexed by chunk index; reduce results appended after. */
readonly chunkResults: ChunkResult[]; readonly chunkResults: ChunkResult[];
/** Persistence hook — JobStore wires this to write the job to SQLite on
* every progress emit, so an interrupted/navigated-away job is recoverable. */
onUpdate?: () => void;
private subscribers = new Set<(frame: string) => void>(); private subscribers = new Set<(frame: string) => void>();
private frames: string[] = []; private frames: string[] = [];
constructor(filename: string, pageCount: number, chunks: Chunk[]) { constructor(filename: string, pageCount: number, chunks: Chunk[], createdAt: number) {
this.jobId = `job_${randomUUID()}`; this.jobId = `job_${randomUUID()}`;
this.filename = filename; this.filename = filename;
this.pageCount = pageCount; this.pageCount = pageCount;
this.chunks = chunks; this.chunks = chunks;
this.createdAt = createdAt;
this.chunkResults = chunks.map((c) => ({ this.chunkResults = chunks.map((c) => ({
index: c.index, index: c.index,
label: `map:${c.index}`, label: `map:${c.index}`,
@@ -238,6 +265,8 @@ export class Job {
/* dropped client; route's close handler unsubscribes */ /* dropped client; route's close handler unsubscribes */
} }
} }
// Persist after fanning out so the durable record reflects the latest frame.
this.onUpdate?.();
} }
/** Attach an SSE writer; replays buffered frames so reconnects catch up. */ /** Attach an SSE writer; replays buffered frames so reconnects catch up. */
@@ -261,6 +290,7 @@ export class Job {
final_summary_md: this.finalSummary, final_summary_md: this.finalSummary,
chunk_results: this.chunkResults, chunk_results: this.chunkResults,
escrow_refs: this.escrowRefs, escrow_refs: this.escrowRefs,
created_at: this.createdAt,
}; };
} }
} }
@@ -272,16 +302,111 @@ export class JobStore {
constructor(private readonly deps: JobDeps) {} constructor(private readonly deps: JobDeps) {}
private now(): number {
return this.deps.now ? this.deps.now() : Date.now();
}
createJob(filename: string, pageCount: number, chunks: Chunk[]): Job { createJob(filename: string, pageCount: number, chunks: Chunk[]): Job {
const job = new Job(filename, pageCount, chunks); const job = new Job(filename, pageCount, chunks, this.now());
job.onUpdate = () => this.persist(job);
this.jobs.set(job.jobId, job); this.jobs.set(job.jobId, job);
this.persist(job);
return job; return job;
} }
/** The live (in-memory) job — needed for estimate/start/SSE. */
get(id: string): Job | undefined { get(id: string): Job | undefined {
return this.jobs.get(id); return this.jobs.get(id);
} }
/** Write the job's current state to the durable store (no-op without db). */
private persist(job: Job): void {
if (!this.deps.db) return;
this.deps.db.upsert(this.recordOf(job));
}
private recordOf(job: Job): JobRecord {
return {
job_id: job.jobId,
filename: job.filename,
status: job.status,
page_count: job.pageCount,
chunk_count: job.chunks.length,
coverage_done: job.coverageDone,
coverage_total: job.coverageTotal,
running_cost_lovelace: job.runningCost.toString(),
final_summary_md: job.finalSummary ?? null,
chunk_results_json: JSON.stringify(job.chunkResults),
escrow_refs_json: JSON.stringify(job.escrowRefs),
created_at: job.createdAt,
updated_at: this.now(),
};
}
/** A non-live persisted record whose status is non-terminal means the
* worker died mid-flight — surface it as "interrupted". */
private staleStatus(r: JobRecord): JobPhase {
const s = r.status as JobPhase;
return s === "estimated" || s === "running" ? "interrupted" : s;
}
private recordToView(r: JobRecord): JobView {
return {
job_id: r.job_id,
filename: r.filename,
status: this.staleStatus(r),
page_count: r.page_count,
chunk_count: r.chunk_count,
coverage: { done: r.coverage_done, total: r.coverage_total },
running_cost_lovelace: r.running_cost_lovelace,
final_summary_md: r.final_summary_md ?? undefined,
chunk_results: safeParse<ChunkResult[]>(r.chunk_results_json, []),
escrow_refs: safeParse<string[]>(r.escrow_refs_json, []),
created_at: r.created_at,
};
}
/** View for any job — live first, else the durable record (past work). */
viewOf(id: string): JobView | null {
const live = this.jobs.get(id);
if (live) return live.view();
const r = this.deps.db?.get(id);
return r ? this.recordToView(r) : null;
}
/** Newest-first list of all jobs (live + persisted), for the past-work UI. */
list(limit = 100): JobListItem[] {
const byId = new Map<string, JobListItem>();
if (this.deps.db) {
for (const r of this.deps.db.list(limit)) {
byId.set(r.job_id, {
job_id: r.job_id,
filename: r.filename,
status: this.staleStatus(r),
coverage: { done: r.coverage_done, total: r.coverage_total },
chunk_count: r.chunk_count,
running_cost_lovelace: r.running_cost_lovelace,
created_at: r.created_at,
has_summary: !!(r.final_summary_md && r.final_summary_md.length > 0),
});
}
}
// Live jobs win (most current status), even over their own db row.
for (const job of this.jobs.values()) {
byId.set(job.jobId, {
job_id: job.jobId,
filename: job.filename,
status: job.status,
coverage: { done: job.coverageDone, total: job.coverageTotal },
chunk_count: job.chunks.length,
running_cost_lovelace: job.runningCost.toString(),
created_at: job.createdAt,
has_summary: !!(job.finalSummary && job.finalSummary.length > 0),
});
}
return [...byId.values()].sort((a, b) => b.created_at - a.created_at).slice(0, limit);
}
/** Discover suppliers + read balance to produce the pre-spend estimate. */ /** Discover suppliers + read balance to produce the pre-spend estimate. */
async estimate(job: Job): Promise<EstimateView> { async estimate(job: Job): Promise<EstimateView> {
const pool = await buildSupplierPool(this.deps.marketplace, this.deps.caps); const pool = await buildSupplierPool(this.deps.marketplace, this.deps.caps);
@@ -309,6 +434,7 @@ export class JobStore {
if (this.started.has(job.jobId)) return; if (this.started.has(job.jobId)) return;
this.started.add(job.jobId); this.started.add(job.jobId);
job.status = "running"; job.status = "running";
this.persist(job);
void this.run(job).catch((err) => { void this.run(job).catch((err) => {
job.status = "failed"; job.status = "failed";
// Terminal: emit a `done` frame so SSE clients stop waiting; carries the // Terminal: emit a `done` frame so SSE clients stop waiting; carries the

View File

@@ -64,7 +64,11 @@ export type JobPhase =
| "running" | "running"
| "completed" | "completed"
| "completed_with_gaps" | "completed_with_gaps"
| "failed"; | "failed"
// A job that was mid-flight when the process died (persisted as running but
// no live worker on the next boot). Its partial summary/coverage is still
// openable; it cannot be resumed.
| "interrupted";
/** Per-call result row, surfaced to the UI. */ /** Per-call result row, surfaced to the UI. */
export interface ChunkResult { export interface ChunkResult {
@@ -90,6 +94,19 @@ export interface JobView {
final_summary_md?: string; final_summary_md?: string;
chunk_results: ChunkResult[]; chunk_results: ChunkResult[];
escrow_refs: string[]; escrow_refs: string[];
created_at: number;
}
/** Compact row returned by GET /v1/pdf-jobs (the past-work list). */
export interface JobListItem {
job_id: string;
filename: string;
status: JobPhase;
coverage: { done: number; total: number };
chunk_count: number;
running_cost_lovelace: string;
created_at: number;
has_summary: boolean;
} }
/** Thrown by extractPdfText for unusable PDFs (e.g. scanned/image-only). */ /** Thrown by extractPdfText for unusable PDFs (e.g. scanned/image-only). */

View File

@@ -0,0 +1,106 @@
/**
* buyer/src/ui/components/RecentJobs.tsx — the past-work list.
*
* Fetches GET /v1/pdf-jobs and renders every job (running + finished) newest
* first. Clicking one calls onOpen(jobId); the page reconnects to a running
* job's live progress or shows a finished job's result. Re-fetches whenever
* `refreshKey` changes (e.g. after a job starts/finishes).
*/
import { useEffect, useState } from "react";
interface JobListItem {
job_id: string;
filename: string;
status: string;
coverage: { done: number; total: number };
chunk_count: number;
running_cost_lovelace: string;
created_at: number;
has_summary: boolean;
}
function ap3x(lovelace: string): string {
return (Number(lovelace) / 1e6).toFixed(2);
}
function ago(ms: number): string {
const s = Math.max(0, Math.floor((Date.now() - ms) / 1000));
if (s < 60) return `${s}s ago`;
if (s < 3600) return `${Math.floor(s / 60)}m ago`;
if (s < 86400) return `${Math.floor(s / 3600)}h ago`;
return `${Math.floor(s / 86400)}d ago`;
}
function badge(status: string): { label: string; cls: string } {
switch (status) {
case "running":
return { label: "running", cls: "bg-blue-100 text-blue-800" };
case "completed":
return { label: "completed", cls: "bg-green-100 text-green-800" };
case "completed_with_gaps":
return { label: "gaps", cls: "bg-amber-100 text-amber-800" };
case "failed":
return { label: "failed", cls: "bg-red-100 text-red-700" };
case "interrupted":
return { label: "interrupted", cls: "bg-gray-200 text-gray-700" };
default:
return { label: status, cls: "bg-gray-100 text-gray-600" };
}
}
export default function RecentJobs({
refreshKey,
onOpen,
}: {
refreshKey: number;
onOpen: (jobId: string) => void;
}) {
const [jobs, setJobs] = useState<JobListItem[]>([]);
const [error, setError] = useState<string | null>(null);
useEffect(() => {
let cancelled = false;
fetch("/v1/pdf-jobs", { credentials: "same-origin" })
.then(async (r) => {
const body = await r.json();
if (!r.ok) throw new Error(body.message ?? r.statusText);
if (!cancelled) setJobs(Array.isArray(body.jobs) ? body.jobs : []);
})
.catch((e: Error) => !cancelled && setError(e.message));
return () => {
cancelled = true;
};
}, [refreshKey]);
if (error) {
return <p className="text-sm text-red-600">Couldn’t load past summaries: {error}</p>;
}
if (jobs.length === 0) {
return <p className="text-sm text-gray-400">No past summaries yet.</p>;
}
return (
<ul className="divide-y divide-gray-100 rounded border border-gray-200 bg-white">
{jobs.map((j) => {
const b = badge(j.status);
return (
<li key={j.job_id}>
<button
type="button"
onClick={() => onOpen(j.job_id)}
className="flex w-full items-center gap-3 px-4 py-2.5 text-left text-sm hover:bg-gray-50"
>
<span className={`rounded px-2 py-0.5 text-xs font-medium ${b.cls}`}>{b.label}</span>
<span className="flex-1 truncate font-medium text-gray-800">{j.filename}</span>
<span className="text-xs text-gray-500">
{j.coverage.done}/{j.coverage.total} · {ap3x(j.running_cost_lovelace)} AP3X
</span>
<span className="w-16 text-right text-xs text-gray-400">{ago(j.created_at)}</span>
</button>
</li>
);
})}
</ul>
);
}

View File

@@ -3,18 +3,25 @@
* *
* Upload → Estimate/Confirm → Live progress (SSE) → Result. * Upload → Estimate/Confirm → Live progress (SSE) → Result.
* *
* The summarization runs SERVER-SIDE, so navigating away never stops it. This
* page remembers the active job (localStorage) and reconnects on return, and
* lists all past jobs (running + finished) so any can be reopened — a finished
* job shows its result; a running one reconnects to live progress.
*
* Each step talks only to the same-origin /v1/pdf-* server routes (the browser * 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 * holds no key).
* traffic the marketplace wants — while producing a real book summary.
*/ */
import { useState } from "react"; import { useEffect, useState } from "react";
import PdfUploadForm, { type UploadInfo } from "../components/PdfUploadForm.js"; import PdfUploadForm, { type UploadInfo } from "../components/PdfUploadForm.js";
import JobEstimate from "../components/JobEstimate.js"; import JobEstimate from "../components/JobEstimate.js";
import JobProgress from "../components/JobProgress.js"; import JobProgress from "../components/JobProgress.js";
import RecentJobs from "../components/RecentJobs.js";
type Step = "upload" | "estimate" | "running" | "done"; type Step = "upload" | "estimate" | "running" | "done";
const ACTIVE_JOB_KEY = "pdf_active_job";
interface ChunkResult { interface ChunkResult {
index: number; index: number;
label: string; label: string;
@@ -27,6 +34,8 @@ interface JobView {
job_id: string; job_id: string;
filename: string; filename: string;
status: string; status: string;
page_count: number;
chunk_count: number;
coverage: { done: number; total: number }; coverage: { done: number; total: number };
running_cost_lovelace: string; running_cost_lovelace: string;
final_summary_md?: string; final_summary_md?: string;
@@ -38,30 +47,104 @@ function ap3x(lovelace: string): string {
return (Number(lovelace) / 1e6).toFixed(2); return (Number(lovelace) / 1e6).toFixed(2);
} }
function statusLabel(status: string): string {
switch (status) {
case "completed":
return "✅ Completed";
case "completed_with_gaps":
return "⚠️ Completed with gaps";
case "interrupted":
return "⏸️ Interrupted (server restarted mid-job)";
case "running":
return "⏳ Running";
default:
return "❌ Failed";
}
}
function infoFromView(v: JobView): UploadInfo {
return {
job_id: v.job_id,
filename: v.filename,
page_count: v.page_count,
chunk_count: v.chunk_count,
sample_chunk: "",
};
}
export default function BookSummarizer() { export default function BookSummarizer() {
const [step, setStep] = useState<Step>("upload"); const [step, setStep] = useState<Step>("upload");
const [info, setInfo] = useState<UploadInfo | null>(null); const [info, setInfo] = useState<UploadInfo | null>(null);
const [view, setView] = useState<JobView | null>(null); const [view, setView] = useState<JobView | null>(null);
const [error, setError] = useState<string | null>(null); const [error, setError] = useState<string | null>(null);
const [refreshKey, setRefreshKey] = useState(0);
// On mount, resume an active job if one is still running.
useEffect(() => {
const aid = localStorage.getItem(ACTIVE_JOB_KEY);
if (!aid) return;
let cancelled = false;
fetch(`/v1/pdf-jobs/${aid}`, { credentials: "same-origin" })
.then(async (r) => {
if (!r.ok) {
localStorage.removeItem(ACTIVE_JOB_KEY);
return;
}
const v = (await r.json()) as JobView;
if (cancelled) return;
if (v.status === "running" || v.status === "estimated") {
setInfo(infoFromView(v));
setStep("running");
} else {
localStorage.removeItem(ACTIVE_JOB_KEY);
}
})
.catch(() => undefined);
return () => {
cancelled = true;
};
}, []);
const openJob = async (jobId: string): Promise<void> => {
setError(null);
try {
const r = await fetch(`/v1/pdf-jobs/${jobId}`, { credentials: "same-origin" });
const v = (await r.json()) as JobView & { message?: string };
if (!r.ok) throw new Error(v.message ?? r.statusText);
if (v.status === "running" || v.status === "estimated") {
setInfo(infoFromView(v));
localStorage.setItem(ACTIVE_JOB_KEY, v.job_id);
setStep("running");
} else {
setView(v);
setStep("done");
}
} catch (e) {
setError((e as Error).message);
setStep("done");
}
};
const loadResult = async (jobId: string): Promise<void> => {
localStorage.removeItem(ACTIVE_JOB_KEY);
try {
const r = await fetch(`/v1/pdf-jobs/${jobId}`, { credentials: "same-origin" });
const body = (await r.json()) as JobView & { message?: string };
if (!r.ok) throw new Error(body.message ?? r.statusText);
setView(body);
setStep("done");
} catch (e) {
setError((e as Error).message);
setStep("done");
}
};
const reset = (): void => { const reset = (): void => {
setStep("upload"); setStep("upload");
setInfo(null); setInfo(null);
setView(null); setView(null);
setError(null); setError(null);
}; setRefreshKey((k) => k + 1);
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 ( return (
@@ -70,20 +153,33 @@ export default function BookSummarizer() {
<h1 className="text-2xl font-semibold">Book Summarizer</h1> <h1 className="text-2xl font-semibold">Book Summarizer</h1>
<p className="mt-1 text-sm text-gray-500"> <p className="mt-1 text-sm text-gray-500">
Upload a PDF → pay suppliers to summarize each chunk → get a full-book summary. Upload a PDF → pay suppliers to summarize each chunk → get a full-book summary.
Jobs keep running if you navigate away; reopen them below.
</p> </p>
</div> </div>
{step === "upload" && ( {step === "upload" && (
<>
<PdfUploadForm <PdfUploadForm
onUploaded={(i) => { onUploaded={(i) => {
setInfo(i); setInfo(i);
setStep("estimate"); setStep("estimate");
}} }}
/> />
<div className="space-y-2">
<h2 className="text-sm font-medium text-gray-700">Past &amp; in-progress summaries</h2>
<RecentJobs refreshKey={refreshKey} onOpen={(id) => void openJob(id)} />
</div>
</>
)} )}
{step === "estimate" && info && ( {step === "estimate" && info && (
<JobEstimate info={info} onStarted={() => setStep("running")} /> <JobEstimate
info={info}
onStarted={() => {
localStorage.setItem(ACTIVE_JOB_KEY, info.job_id);
setStep("running");
}}
/>
)} )}
{step === "running" && info && ( {step === "running" && info && (
@@ -96,8 +192,16 @@ export default function BookSummarizer() {
/> />
<p className="text-xs text-gray-400"> <p className="text-xs text-gray-400">
Each chunk is a separate on-chain escrow (PostEscrow → supplier → Accept), Each chunk is a separate on-chain escrow (PostEscrow → supplier → Accept),
so this can take a while — that’s real marketplace traffic. so this can take a while — that’s real marketplace traffic. You can leave this
page; the job keeps running and shows up under “Past summaries”.
</p> </p>
<button
type="button"
onClick={reset}
className="text-sm text-gray-600 hover:underline"
>
← Back to list (keeps running)
</button>
</div> </div>
)} )}
@@ -111,13 +215,8 @@ export default function BookSummarizer() {
{view && ( {view && (
<> <>
<div className="flex flex-wrap items-center gap-x-6 gap-y-1 text-sm"> <div className="flex flex-wrap items-center gap-x-6 gap-y-1 text-sm">
<span className="font-medium"> <span className="font-medium">{statusLabel(view.status)}</span>
{view.status === "completed" <span className="truncate text-gray-600">{view.filename}</span>
? "✅ Completed"
: view.status === "completed_with_gaps"
? "⚠️ Completed with gaps"
: "❌ Failed"}
</span>
<span> <span>
Coverage: {view.coverage.done}/{view.coverage.total} chunks Coverage: {view.coverage.done}/{view.coverage.total} chunks
</span> </span>
@@ -134,10 +233,12 @@ export default function BookSummarizer() {
</a> </a>
)} )}
{view.final_summary_md && ( {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"> <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} {view.final_summary_md}
</pre> </pre>
) : (
<p className="text-sm text-gray-500">No summary was produced for this job.</p>
)} )}
<details className="text-xs text-gray-500"> <details className="text-xs text-gray-500">
@@ -155,7 +256,7 @@ export default function BookSummarizer() {
onClick={reset} onClick={reset}
className="rounded border border-gray-300 px-4 py-2 text-sm font-medium text-gray-700 hover:bg-gray-100" className="rounded border border-gray-300 px-4 py-2 text-sm font-medium text-gray-700 hover:bg-gray-100"
> >
Summarize another book ← Back to summaries
</button> </button>
</div> </div>
)} )}

View File

@@ -21,7 +21,13 @@ services:
environment: environment:
INDEXER_URL: http://marketplace-mainnet-indexer:8090 INDEXER_URL: http://marketplace-mainnet-indexer:8090
BUYER_PORT: "8070" BUYER_PORT: "8070"
# Pin the archive dir so it lands on the persistent volume below — keeps
# the response archive + PDF job store (past summaries) across container
# recreation/redeploys.
ARCHIVE_DIR: /repo/data/archive
# NETWORK_ID intentionally not pinned — see supplier compose comment. # NETWORK_ID intentionally not pinned — see supplier compose comment.
volumes:
- buyer-mainnet-archive:/repo/data/archive
healthcheck: healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8070/healthz >/dev/null || exit 1"] test: ["CMD-SHELL", "wget -qO- http://localhost:8070/healthz >/dev/null || exit 1"]
interval: 20s interval: 20s
@@ -50,3 +56,6 @@ networks:
traefik-net: traefik-net:
external: true external: true
name: traefik-net name: traefik-net
volumes:
buyer-mainnet-archive:

View File

@@ -5,7 +5,11 @@
*/ */
import { describe, it, expect } from "vitest"; import { describe, it, expect } from "vitest";
import { mkdtempSync, rmSync } from "fs";
import { tmpdir } from "os";
import { join } from "path";
import { JobStore, type RunCallFn } from "../../buyer/src/pdf/summarize-job.js"; import { JobStore, type RunCallFn } from "../../buyer/src/pdf/summarize-job.js";
import { PdfJobDb } from "../../buyer/src/pdf/job-db.js";
import { loadPdfCaps } from "../../buyer/src/pdf/caps.js"; import { loadPdfCaps } from "../../buyer/src/pdf/caps.js";
import type { Chunk } from "../../buyer/src/pdf/types.js"; import type { Chunk } from "../../buyer/src/pdf/types.js";
import type { Marketplace } from "../../buyer/src/sdk/Marketplace.js"; import type { Marketplace } from "../../buyer/src/sdk/Marketplace.js";
@@ -166,6 +170,66 @@ describe("JobStore orchestrator", () => {
expect(typeof job.finalSummary).toBe("string"); expect(typeof job.finalSummary).toBe("string");
}); });
it("persists a completed job durably across a JobStore 'restart'", async () => {
const dir = mkdtempSync(join(tmpdir(), "pdfjobs-"));
try {
const canned: RunCallFn = async (sup) => ({
response: "S",
escrowRef: `${"f".repeat(64)}#${Math.floor(Math.random() * 1e9)}`,
supplierPkh: sup.supplierPkh,
model: sup.model,
receipt: {},
receiptSignature: "sig",
});
const db1 = new PdfJobDb(dir);
const store1 = new JobStore({
marketplace: MARKETPLACE, chain: CHAIN, walletKey: WALLET, indexerUrl: "http://i",
caps: loadPdfCaps({}), db: db1, runCall: canned, walletBalance: async () => 10_000_000_000n,
});
const job = store1.createJob("persist.pdf", 3, chunks(3));
await waitForDone(store1, job);
expect(job.status).toBe("completed");
db1.close();
// Fresh process: new db connection + new store (empty in-memory map).
const db2 = new PdfJobDb(dir);
const store2 = new JobStore({
marketplace: MARKETPLACE, chain: CHAIN, walletKey: WALLET, indexerUrl: "http://i",
caps: loadPdfCaps({}), db: db2,
});
const v = store2.viewOf(job.jobId);
expect(v).not.toBeNull();
expect(v?.status).toBe("completed");
expect(typeof v?.final_summary_md).toBe("string");
expect((v?.escrow_refs.length ?? 0)).toBeGreaterThan(0);
expect(store2.list().find((j) => j.job_id === job.jobId)?.has_summary).toBe(true);
db2.close();
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
it("reports a persisted-but-not-live running job as interrupted", async () => {
const dir = mkdtempSync(join(tmpdir(), "pdfjobs-"));
try {
const db = new PdfJobDb(dir);
db.upsert({
job_id: "job_crashed", filename: "x.pdf", status: "running", page_count: 1, chunk_count: 2,
coverage_done: 1, coverage_total: 2, running_cost_lovelace: "200000", final_summary_md: null,
chunk_results_json: "[]", escrow_refs_json: "[]", created_at: 1, updated_at: 2,
});
const store = new JobStore({
marketplace: MARKETPLACE, chain: CHAIN, walletKey: WALLET, indexerUrl: "http://i",
caps: loadPdfCaps({}), db,
});
expect(store.viewOf("job_crashed")?.status).toBe("interrupted");
expect(store.list().find((j) => j.job_id === "job_crashed")?.status).toBe("interrupted");
db.close();
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
it("fails cleanly when no capable suppliers exist", async () => { it("fails cleanly when no capable suppliers exist", async () => {
const emptyMarket = { discoverSuppliers: async () => [] } as unknown as Marketplace; const emptyMarket = { discoverSuppliers: async () => [] } as unknown as Marketplace;
const store = new JobStore({ const store = new JobStore({

View File

@@ -6,8 +6,12 @@
*/ */
import { describe, it, expect } from "vitest"; import { describe, it, expect } from "vitest";
import { mkdtempSync, rmSync } from "fs";
import { tmpdir } from "os";
import { join } from "path";
import { createApp } from "../../buyer/src/server.js"; import { createApp } from "../../buyer/src/server.js";
import { JobStore, type RunCallFn } from "../../buyer/src/pdf/summarize-job.js"; import { JobStore, type RunCallFn } from "../../buyer/src/pdf/summarize-job.js";
import { PdfJobDb } from "../../buyer/src/pdf/job-db.js";
import { loadPdfCaps } from "../../buyer/src/pdf/caps.js"; import { loadPdfCaps } from "../../buyer/src/pdf/caps.js";
import type { Chunk } from "../../buyer/src/pdf/types.js"; import type { Chunk } from "../../buyer/src/pdf/types.js";
import type { Marketplace } from "../../buyer/src/sdk/Marketplace.js"; import type { Marketplace } from "../../buyer/src/sdk/Marketplace.js";
@@ -44,7 +48,7 @@ const MARKETPLACE = {
const WALLET = { address: "addr_test", pubKeyHash: "pkh", pubKeyHex: "", privateKeyHex: "0".repeat(64) } as WalletKey; const WALLET = { address: "addr_test", pubKeyHash: "pkh", pubKeyHex: "", privateKeyHex: "0".repeat(64) } as WalletKey;
const CHAIN = {} as ChainProvider; const CHAIN = {} as ChainProvider;
function makeStore(opts?: { balance?: bigint; caps?: Record<string, string> }): JobStore { function makeStore(opts?: { balance?: bigint; caps?: Record<string, string>; db?: PdfJobDb }): JobStore {
let n = 0; let n = 0;
const runCall: RunCallFn = async (sup) => { const runCall: RunCallFn = async (sup) => {
n += 1; n += 1;
@@ -65,6 +69,7 @@ function makeStore(opts?: { balance?: bigint; caps?: Record<string, string> }):
caps: loadPdfCaps(opts?.caps ?? {}), caps: loadPdfCaps(opts?.caps ?? {}),
runCall, runCall,
walletBalance: async () => opts?.balance ?? 10_000_000_000n, walletBalance: async () => opts?.balance ?? 10_000_000_000n,
db: opts?.db,
}); });
} }
@@ -100,6 +105,65 @@ describe("/v1/pdf-* routes", () => {
expect(["pdf_parse_failed", "scanned_or_image_pdf", "no_text"]).toContain(res.body.error); expect(["pdf_parse_failed", "scanned_or_image_pdf", "no_text"]).toContain(res.body.error);
}); });
it("lists jobs newest-first via GET /v1/pdf-jobs", async () => {
const store = makeStore();
const a = store.createJob("First.pdf", 1, chunks(1));
const b = store.createJob("Second.pdf", 2, chunks(2));
const app = createApp({ jobStore: store, pdfCaps: loadPdfCaps({}) });
const { default: request } = await import("supertest");
const res = await request(app).get("/v1/pdf-jobs");
expect(res.status).toBe(200);
expect(Array.isArray(res.body.jobs)).toBe(true);
const ids = res.body.jobs.map((j: { job_id: string }) => j.job_id);
expect(ids).toContain(a.jobId);
expect(ids).toContain(b.jobId);
const found = res.body.jobs.find((j: { job_id: string }) => j.job_id === b.jobId);
expect(found.filename).toBe("Second.pdf");
expect(found.status).toBe("estimated");
});
it("serves a past (persisted, non-live) job's view, summary.md, and terminal SSE", async () => {
const dir = mkdtempSync(join(tmpdir(), "pdfroutes-"));
try {
const db1 = new PdfJobDb(dir);
const store1 = makeStore({ db: db1 });
const seeded = store1.createJob("Past.pdf", 2, chunks(2));
const app1 = createApp({ jobStore: store1, pdfCaps: loadPdfCaps({}) });
const { default: request } = await import("supertest");
await request(app1).post(`/v1/pdf-jobs/${seeded.jobId}/start`).send({ confirm: true });
let view = seeded.view();
for (let i = 0; i < 100 && view.status === "running"; i++) {
await new Promise((r) => setTimeout(r, 10));
view = (await request(app1).get(`/v1/pdf-jobs/${seeded.jobId}`)).body;
}
expect(view.status).toBe("completed");
db1.close();
// Fresh store (no live job) backed by the same db = "after restart".
const db2 = new PdfJobDb(dir);
const store2 = makeStore({ db: db2 });
const app2 = createApp({ jobStore: store2, pdfCaps: loadPdfCaps({}) });
const v = await request(app2).get(`/v1/pdf-jobs/${seeded.jobId}`);
expect(v.status).toBe(200);
expect(v.body.status).toBe("completed");
expect(v.body.final_summary_md.length).toBeGreaterThan(0);
const md = await request(app2).get(`/v1/pdf-jobs/${seeded.jobId}/summary.md`);
expect(md.status).toBe(200);
// SSE on a non-live job returns a single terminal frame and ENDS (no hang).
const ev = await request(app2).get(`/v1/pdf-jobs/${seeded.jobId}/events`).buffer(true);
expect(ev.text).toContain("event: done");
const list = await request(app2).get("/v1/pdf-jobs");
expect(list.body.jobs.find((j: { job_id: string }) => j.job_id === seeded.jobId)?.has_summary).toBe(true);
db2.close();
} finally {
rmSync(dir, { recursive: true, force: true });
}
});
it("404s estimate for an unknown job", async () => { it("404s estimate for an unknown job", async () => {
const app = createApp({ jobStore: makeStore(), pdfCaps: loadPdfCaps({}) }); const app = createApp({ jobStore: makeStore(), pdfCaps: loadPdfCaps({}) });
const { default: request } = await import("supertest"); const { default: request } = await import("supertest");