From f0e0daae7701b7fbac81c4eb57a6f12fd1d460a0 Mon Sep 17 00:00:00 2001 From: Richard R Date: Wed, 3 Jun 2026 04:44:55 -0600 Subject: [PATCH] feat: improve PDF parsing recovery and reader UI (#98) * refactor(ui): replace anchor tags with next/link in SidebarNavLink and UserMenu Update SidebarNavLink to use next/link for navigation instead of anchor tags, ensuring proper routing and improved accessibility in Next.js. Refactor UserMenu to remove legacy Link wrappers and directly use SidebarNavLink for signin and signup links. This streamlines navigation components and aligns with Next.js best practices. * fix(pdf): handle non-zero viewport origins and improve layout model ordering Update normalizeTextItemsForLayout to correctly apply viewport transforms, including non-zero page origins, ensuring accurate mapping of PDF text items to top-left coordinates. Refactor runLayoutModel to implement a custom order sequence builder for layout regions using model order logits, improving region ordering consistency with model semantics. Update related tests to cover viewport transform edge cases. * refactor(pdf): improve text normalization with font ascent and vertical overlap logic Enhance text normalization by incorporating font ascent and descent data to more accurately position glyphs, especially for decorative initials. Update merge logic to better detect line membership using vertical overlap, ensuring drop caps and overlapping glyphs are merged correctly. Extend tests to cover these layout scenarios. * style(range): redesign slider with precision gauge and ruler ticks Revamp the range input to feature a minimalist "precision gauge" style. Introduce a hairline rail, ruler notches for discrete steps, and a slim needle thumb. Add CSS variables and logic for per-instance tick sizing and coloring. Remove bulky inline class-based styling in favor of centralized CSS for improved maintainability and visual clarity. * refactor(ui): modularize PDF loader and range slider visuals Move PDF layout scan visualization and range slider styles into dedicated CSS modules, isolating their styles from the global scope. Integrate PdfLayoutScan component into the PDF viewer loader UI for animated parse progress. Refactor progress bars to use a reusable progress-fill class with animated sheen effect. Update range input to use CSS module for precision gauge styling. * style(reader): remove grid overlay from PdfLayoutScan visualization * refactor(api): add staleness detection for inflight worker operation states Integrate isWorkerOperationStateStale checks into document parse API endpoints to ensure inflight worker operation states are not reused if stale. Introduce helper for staleness detection and corresponding unit tests. Enhance SSE event streaming with keepalive intervals and improve progress acknowledgment error handling. * feat(worker): recover and fail stale in-flight pdf ops on startup Add orphaned operation recovery logic to detect and mark stale in-flight pdf_layout jobs as failed during worker startup. Extend OperationStateStore with listOpStates for state enumeration. Update tests and documentation to cover recovery behavior and new environment variable COMPUTE_PDF_JOB_ATTEMPTS. * refactor(worker): distinguish staleness thresholds for running and queued pdf ops Update orphan recovery logic to apply separate timeouts for 'running' and 'queued' pdf_layout operations. Adjust tests to verify that only stale 'running' operations are failed, while stale 'queued' operations remain untouched. * feat(worker): extend orphan recovery to handle whisper_align ops and improve logging Update orphan recovery to detect and fail stale 'running' whisper_align operations in addition to pdf_layout. Refactor recovery logic to generalize staleness checks across operation kinds and enhance log output with detailed operation info. Expand tests to verify correct handling of both whisper_align and pdf_layout operations in running and queued states. * refactor(control-plane): introduce revision-based CAS for operation state updates Add revision tracking and compare-and-set (CAS) semantics to operation state stores, enabling atomic state transitions and preventing lost updates. Extend the OperationStateStore interface with getOpStateRecord and compareAndSetOpState methods. Update orchestrator and worker runtime to utilize CAS for marking operations as failed only if the state is unchanged. Enhance in-memory, JetStream, and test control plane implementations to support revision logic. This change improves concurrency safety and correctness of operation state management across distributed components. * feat(ui): add parse failure state to PDF layout scan animation Display a distinct "parse halted" visual state in the PDF layout scan component and PDF viewer page when parsing fails. The loader animation is replaced by a static, dimmed page with an alert glyph and updated styling, ensuring users are not misled by an active animation after a failure. CSS and component logic updated to support the new state. * feat(worker): extract orphaned operation recovery to module with periodic sweep Move orphaned operation recovery logic into a dedicated orphan-recovery module, introducing a periodic sweep timer that triggers recovery every 15 seconds while the worker is connected. Refactor runtime to delegate orphan detection and handling to the new module, improving modularity and maintainability. Add unit tests for orphan-recovery to ensure correctness. * fix(ui): adjust PDF viewer layout and update parse loader description * refactor(pdf): streamline layout model region extraction and update test coverage - Replace custom order sequence logic with softmax-based class selection in runLayoutModel - Remove unused sigmoid and buildOrderSequence functions - Simplify detection loop to filter and map regions directly - Add targeted tests for layout model extraction logic - Update CSS animation naming for consistency - Clarify test description and add inline comments for orphan recovery scenario * fix(pdf): add strict validation for layout model output shapes and extend test coverage Add explicit error handling for invalid or inconsistent pred_boxes and logits array lengths in runLayoutModel to prevent silent failures. Expand test suite to verify correct region filtering and error scenarios, ensuring only labeled regions are returned and malformed outputs are handled robustly. --- .env.example | 1 + compute/core/src/control-plane/index.ts | 1 - .../core/src/control-plane/orchestrator.ts | 30 +++ compute/core/src/control-plane/types.ts | 11 + compute/core/src/pdf/merge.ts | 6 +- compute/core/src/pdf/normalize-text.ts | 64 ++++- compute/core/src/pdf/parse.ts | 3 +- compute/core/src/pdf/runLayoutModel.ts | 33 ++- .../tests/control-plane/orchestrator.test.ts | 47 +++- .../control-plane/run-layout-model.test.ts | 175 ++++++++++++++ .../helpers/in-memory-control-plane.ts} | 29 ++- compute/worker/src/control-plane/jetstream.ts | 42 +++- compute/worker/src/orphan-recovery.ts | 100 ++++++++ compute/worker/src/runtime.ts | 121 +++++++++- compute/worker/tests/api/routes.test.ts | 75 ++++++ .../tests/fixtures/fake-control-plane.ts | 36 +++ .../worker/tests/unit/orphan-recovery.test.ts | 59 +++++ docs-site/docs/deploy/compute-worker.md | 3 +- docs-site/docs/deploy/local-development.md | 4 +- .../docs/reference/environment-variables.md | 8 + src/app/(app)/pdf/[id]/page.tsx | 106 ++++++--- src/app/(app)/pdf/[id]/usePdfDocument.ts | 7 + .../api/documents/[id]/parsed/events/route.ts | 225 ++++++++++-------- src/app/api/documents/[id]/parsed/route.ts | 17 +- src/app/globals.css | 28 +++ src/components/ProgressCard.tsx | 2 +- src/components/auth/UserMenu.tsx | 21 +- .../documents/DexieMigrationModal.tsx | 2 +- .../reader/PdfLayoutScan.module.css | 211 ++++++++++++++++ src/components/reader/PdfLayoutScan.tsx | 172 +++++++++++++ src/components/ui/range.module.css | 115 +++++++++ src/components/ui/range.tsx | 46 ++-- src/components/ui/sidebar-nav.tsx | 13 +- src/lib/server/compute/worker-parse-state.ts | 16 ++ ...pdf-merge-text-with-regions.vitest.spec.ts | 15 ++ ...-parse-normalize-text-items.vitest.spec.ts | 51 +++- tests/unit/worker-parse-state.vitest.spec.ts | 27 +++ 37 files changed, 1719 insertions(+), 203 deletions(-) create mode 100644 compute/core/tests/control-plane/run-layout-model.test.ts rename compute/core/{src/control-plane/in-memory.ts => tests/helpers/in-memory-control-plane.ts} (80%) create mode 100644 compute/worker/src/orphan-recovery.ts create mode 100644 compute/worker/tests/unit/orphan-recovery.test.ts create mode 100644 src/components/reader/PdfLayoutScan.module.css create mode 100644 src/components/reader/PdfLayoutScan.tsx create mode 100644 src/components/ui/range.module.css diff --git a/.env.example b/.env.example index 8173d6d..02c3e57 100644 --- a/.env.example +++ b/.env.example @@ -65,6 +65,7 @@ S3_BUCKET= # COMPUTE_JOB_CONCURRENCY=1 # COMPUTE_WHISPER_TIMEOUT_MS=30000 # COMPUTE_PDF_TIMEOUT_MS=300000 +# COMPUTE_PDF_JOB_ATTEMPTS=1 # COMPUTE_OP_STALE_MS=1800000 # WHISPER_MODEL_BASE_URL=https://huggingface.co/onnx-community/whisper-base_timestamped/resolve/main # PDF_LAYOUT_MODEL_BASE_URL=https://huggingface.co/Bei0001/PP-DocLayoutV3-ONNX/resolve/main diff --git a/compute/core/src/control-plane/index.ts b/compute/core/src/control-plane/index.ts index 8e6b18c..5b8255d 100644 --- a/compute/core/src/control-plane/index.ts +++ b/compute/core/src/control-plane/index.ts @@ -1,5 +1,4 @@ export * from './types'; export * from './state-machine'; export * from './orchestrator'; -export * from './in-memory'; export * from './sse'; diff --git a/compute/core/src/control-plane/orchestrator.ts b/compute/core/src/control-plane/orchestrator.ts index 950fe71..7f621a4 100644 --- a/compute/core/src/control-plane/orchestrator.ts +++ b/compute/core/src/control-plane/orchestrator.ts @@ -263,6 +263,36 @@ export class OperationOrchestrator { return next; } + async markFailedIfUnchanged(input: { + current: WorkerOperationState; + expectedRevision: number; + error: WorkerJobErrorShape | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise { + const now = input.updatedAt ?? this.clock.now(); + const shape = typeof input.error === 'string' ? { message: input.error } : input.error; + + const next: WorkerOperationState = { + ...input.current, + status: 'failed', + startedAt: input.current.startedAt ?? now, + updatedAt: now, + error: shape, + ...(input.timing ? { timing: input.timing } : {}), + }; + + const updated = await this.stateStore.compareAndSetOpState({ + opId: input.current.opId, + expectedRevision: input.expectedRevision, + newState: next, + }); + if (!updated) return null; + + await this.eventStream.append(next.opId, next); + return next; + } + async explainReuseDecision(input: { current: WorkerOperationState; requestKind: WorkerOperationKind; diff --git a/compute/core/src/control-plane/types.ts b/compute/core/src/control-plane/types.ts index 79fd16c..9fc66ff 100644 --- a/compute/core/src/control-plane/types.ts +++ b/compute/core/src/control-plane/types.ts @@ -26,9 +26,20 @@ export interface OperationIndexEntry { opId: string; } +export interface OperationStateRecord { + state: OperationState; + revision: number; +} + export interface OperationStateStore { getOpState(opId: string): Promise | null>; + getOpStateRecord(opId: string): Promise | null>; putOpState(state: OperationState): Promise; + compareAndSetOpState(input: { + opId: string; + expectedRevision: number; + newState: OperationState; + }): Promise; getOpIndex(opKey: string): Promise; compareAndSetOpIndex(input: { opKey: string; diff --git a/compute/core/src/pdf/merge.ts b/compute/core/src/pdf/merge.ts index aeff9ab..e1f71b2 100644 --- a/compute/core/src/pdf/merge.ts +++ b/compute/core/src/pdf/merge.ts @@ -53,7 +53,11 @@ function joinText(items: PdfTextItem[]): string { const prevEndX = prev.x + prev.width; const gap = item.x - prevEndX; const lineJump = item.y - prev.y; - const lineBreak = lineJump > Math.max(2, Math.min(prev.height, item.height) * 0.6); + const prevBottom = prev.y + prev.height; + const itemBottom = item.y + item.height; + const verticalOverlap = Math.max(0, Math.min(prevBottom, itemBottom) - Math.max(prev.y, item.y)); + const sharesLineBand = verticalOverlap >= Math.max(1, Math.min(prev.height, item.height) * 0.5); + const lineBreak = !sharesLineBand && lineJump > Math.max(2, Math.min(prev.height, item.height) * 0.6); const avgCharWidth = item.width / Math.max(1, item.text.length); const needsSpace = lineBreak || gap > Math.max(avgCharWidth * 0.3, 2); out += needsSpace ? ` ${item.text}` : item.text; diff --git a/compute/core/src/pdf/normalize-text.ts b/compute/core/src/pdf/normalize-text.ts index 2c2eca7..1e91101 100644 --- a/compute/core/src/pdf/normalize-text.ts +++ b/compute/core/src/pdf/normalize-text.ts @@ -1,7 +1,52 @@ import type { TextItem } from 'pdfjs-dist/types/src/display/api'; import type { PdfTextItem } from './types'; -export function normalizeTextItemsForLayout(items: TextItem[], pageHeight: number): PdfTextItem[] { +interface ViewportLike { + height: number; + transform: readonly number[]; +} + +interface TextStyleLike { + ascent?: number; + descent?: number; +} + +function applyViewportTransform( + x: number, + y: number, + transform: readonly number[], +): { x: number; y: number } { + const a = Number(transform[0] ?? 1); + const b = Number(transform[1] ?? 0); + const c = Number(transform[2] ?? 0); + const d = Number(transform[3] ?? 1); + const e = Number(transform[4] ?? 0); + const f = Number(transform[5] ?? 0); + return { + x: (a * x) + (c * y) + e, + y: (b * x) + (d * y) + f, + }; +} + +function resolveTopOffset(height: number, style: TextStyleLike | undefined): number { + const ascent = Number(style?.ascent); + if (Number.isFinite(ascent) && ascent > 0) { + return height * ascent; + } + + const descent = Number(style?.descent); + if (Number.isFinite(descent) && descent < 0) { + return height * (1 + descent); + } + + return height; +} + +export function normalizeTextItemsForLayout( + items: TextItem[], + viewport: ViewportLike, + styles: Record = {}, +): PdfTextItem[] { return items .filter((item) => { if (!(typeof item.str === 'string' && item.str.trim().length > 0)) return false; @@ -17,13 +62,20 @@ export function normalizeTextItemsForLayout(items: TextItem[], pageHeight: numbe return true; }) .map((item) => { - const x = Number(item.transform[4] ?? 0); + const origin = applyViewportTransform( + Number(item.transform[4] ?? 0), + Number(item.transform[5] ?? 0), + viewport.transform, + ); const width = Math.max(0, Number(item.width ?? 0)); const height = Math.max(1, Math.abs(Number(item.transform[3] ?? 1))); - const baselineY = Number(item.transform[5] ?? 0); - // pdf.js text transforms are in PDF user-space (origin bottom-left). - // Normalize into top-left page coordinates to match rendered image/model boxes. - const y = Math.max(0, pageHeight - baselineY - height); + const topOffset = resolveTopOffset(height, styles[item.fontName ?? '']); + // pdf.js text transforms are in PDF user-space and may include non-zero + // page origins via the viewport transform. Map the text baseline into + // viewport coordinates first, then adjust upward using font ascent, + // which matches how pdf.js positions glyph runs in its text layer. + const x = Math.max(0, origin.x); + const y = Math.max(0, origin.y - topOffset); return { text: item.str, x, diff --git a/compute/core/src/pdf/parse.ts b/compute/core/src/pdf/parse.ts index a6dbe22..2b3a6f7 100644 --- a/compute/core/src/pdf/parse.ts +++ b/compute/core/src/pdf/parse.ts @@ -72,7 +72,8 @@ export async function parsePdf(input: ParsePdfInput): Promise const textContent = await page.getTextContent(); const textItems = normalizeTextItemsForLayout( textContent.items.filter((item): item is TextItem => 'str' in item && 'transform' in item), - viewport.height, + viewport, + textContent.styles, ); if (textItems.length > 0) sawText = true; diff --git a/compute/core/src/pdf/runLayoutModel.ts b/compute/core/src/pdf/runLayoutModel.ts index 5b8fb43..5c726e0 100644 --- a/compute/core/src/pdf/runLayoutModel.ts +++ b/compute/core/src/pdf/runLayoutModel.ts @@ -196,8 +196,10 @@ function preprocessResized( const ctx = canvas.getContext('2d'); ctx.fillStyle = '#ffffff'; ctx.fillRect(0, 0, preprocessor.inputWidth, preprocessor.inputHeight); - ctx.imageSmoothingEnabled = true; - ctx.imageSmoothingQuality = 'high'; + // Match the upstream image processor more closely. The official + // implementation explicitly disables antialiasing during resize to stay + // close to OpenCV semantics for PP-DocLayoutV3 inputs. + ctx.imageSmoothingEnabled = false; ctx.drawImage(image, 0, 0, preprocessor.inputWidth, preprocessor.inputHeight); const imageData = ctx.getImageData(0, 0, preprocessor.inputWidth, preprocessor.inputHeight); @@ -284,12 +286,29 @@ export async function runLayoutModel(input: RunLayoutInput): Promise { let firstAttempt = true; const conflictStore = { getOpState: store.getOpState.bind(store), + getOpStateRecord: store.getOpStateRecord.bind(store), putOpState: store.putOpState.bind(store), + compareAndSetOpState: store.compareAndSetOpState.bind(store), getOpIndex: store.getOpIndex.bind(store), compareAndSetOpIndex: async (input: { opKey: string; newOpId: string; expectedOpId: string | null }) => { if (firstAttempt && input.expectedOpId === null) { @@ -92,4 +94,45 @@ describe('operation orchestrator', () => { expect(created.opId).toMatch(/^op-/); expect(await store.getOpIndex('cas-key')).toEqual({ opId: created.opId }); }); + + test('markFailedIfUnchanged only writes once for the expected revision', async () => { + const queue = new InMemoryOperationQueue(); + const stateStore = new InMemoryOperationStateStore(); + const eventStream = new InMemoryOperationEventStream(); + const orchestrator = new OperationOrchestrator({ + queue, + stateStore, + eventStream, + config: { opStaleMs: 2_000, maxCasRetries: 5 }, + }); + + const created = await orchestrator.enqueueOrReuse(buildRequest('stale-op')); + await orchestrator.markRunning({ opId: created.opId, updatedAt: 2_000 }); + + const record = await stateStore.getOpStateRecord(created.opId); + expect(record).not.toBeNull(); + + const first = await orchestrator.markFailedIfUnchanged({ + current: record!.state, + expectedRevision: record!.revision, + error: { code: 'WORKER_ORPHANED_OP', message: 'stale op' }, + updatedAt: 3_000, + }); + const second = await orchestrator.markFailedIfUnchanged({ + current: record!.state, + expectedRevision: record!.revision, + error: { code: 'WORKER_ORPHANED_OP', message: 'stale op' }, + updatedAt: 3_000, + }); + + expect(first).toMatchObject({ + opId: created.opId, + status: 'failed', + error: { code: 'WORKER_ORPHANED_OP' }, + }); + expect(second).toBeNull(); + + const events = await eventStream.listSince(created.opId, 0); + expect(events.filter((event) => event.snapshot.status === 'failed')).toHaveLength(1); + }); }); diff --git a/compute/core/tests/control-plane/run-layout-model.test.ts b/compute/core/tests/control-plane/run-layout-model.test.ts new file mode 100644 index 0000000..e951bf3 --- /dev/null +++ b/compute/core/tests/control-plane/run-layout-model.test.ts @@ -0,0 +1,175 @@ +import { beforeEach, describe, expect, test, vi } from 'vitest'; + +const mockState = vi.hoisted(() => ({ + runOutput: { + logits: { data: new Float32Array() }, + pred_boxes: { data: new Float32Array() }, + }, +})); + +vi.mock('onnxruntime-node', () => ({ + InferenceSession: { + create: vi.fn(async () => ({ + run: vi.fn(async () => mockState.runOutput), + })), + }, + Tensor: class Tensor { + constructor( + public type: string, + public data: Float32Array, + public dims: number[], + ) {} + }, +})); + +vi.mock('fs/promises', () => ({ + readFile: vi.fn(async (path: string) => { + if (path === '/tmp/model-config.json') { + return JSON.stringify({ + id2label: { + 0: 'text', + 1: 'table', + }, + }); + } + + if (path === '/tmp/model-preprocessor.json') { + return JSON.stringify({ + size: { width: 2, height: 2 }, + rescale_factor: 1 / 255, + image_mean: [0, 0, 0], + image_std: [1, 1, 1], + }); + } + + throw new Error(`unexpected readFile path: ${path}`); + }), +})); + +vi.mock('@napi-rs/canvas', () => { + const createCanvas = (width: number, height: number) => ({ + getContext: () => ({ + fillStyle: '#ffffff', + fillRect: () => {}, + drawImage: () => {}, + imageSmoothingEnabled: true, + getImageData: () => ({ + data: new Uint8ClampedArray(width * height * 4).fill(255), + }), + }), + }); + const loadImage = vi.fn(async () => ({ width: 2, height: 2 })); + + return { + createCanvas, + loadImage, + default: { + createCanvas, + loadImage, + }, + }; +}); + +vi.mock('../../src/pdf/model', () => ({ + ensureModel: vi.fn(async () => '/tmp/model.onnx'), + MODEL_CONFIG_PATH: '/tmp/model-config.json', + MODEL_PREPROCESSOR_PATH: '/tmp/model-preprocessor.json', +})); + +vi.mock('../../src/config/cpu-budget', () => ({ + getOnnxThreadsPerJob: vi.fn(() => 1), +})); + +describe('runLayoutModel', () => { + beforeEach(() => { + vi.resetModules(); + mockState.runOutput = { + logits: { data: new Float32Array() }, + pred_boxes: { data: new Float32Array() }, + }; + }); + + test('keeps one winner per query instead of dropping later queries behind duplicate class rows', async () => { + mockState.runOutput = { + logits: { + data: new Float32Array([ + 3, + 4, + 2.5, + 0.1, + ]), + }, + pred_boxes: { + data: new Float32Array([ + 0.25, 0.25, 0.3, 0.3, + 0.75, 0.75, 0.3, 0.3, + ]), + }, + }; + + const { runLayoutModel } = await import('../../src/pdf/runLayoutModel'); + const regions = await runLayoutModel({ + pageWidth: 100, + pageHeight: 100, + textItems: [{} as never], + pageImage: Buffer.from([1]), + }); + + expect(regions).toHaveLength(2); + expect(regions[0]?.label).toBe('text'); + expect(regions[0]?.confidence).toBeCloseTo(0.9168273, 6); + expect(regions[0]?.bbox).toEqual([ + expect.closeTo(60, 5), + expect.closeTo(60, 5), + expect.closeTo(90, 5), + expect.closeTo(90, 5), + ]); + expect(regions[1]?.label).toBe('table'); + expect(regions[1]?.confidence).toBeCloseTo(0.73105858, 6); + expect(regions[1]?.bbox).toEqual([ + expect.closeTo(10, 5), + expect.closeTo(10, 5), + expect.closeTo(40, 5), + expect.closeTo(40, 5), + ]); + }); + + test('drops unlabeled query winners and keeps only labeled regions', async () => { + mockState.runOutput = { + logits: { + data: new Float32Array([ + 0.1, + 0.2, + 5, + 4, + 0.1, + 0.1, + ]), + }, + pred_boxes: { + data: new Float32Array([ + 0.25, 0.25, 0.3, 0.3, + 0.75, 0.75, 0.3, 0.3, + ]), + }, + }; + + const { runLayoutModel } = await import('../../src/pdf/runLayoutModel'); + const regions = await runLayoutModel({ + pageWidth: 100, + pageHeight: 100, + textItems: [{} as never], + pageImage: Buffer.from([1]), + }); + + expect(regions).toHaveLength(1); + expect(regions[0]?.label).toBe('text'); + expect(regions[0]?.confidence).toBeCloseTo(0.96109135, 5); + expect(regions[0]?.bbox).toEqual([ + expect.closeTo(60, 5), + expect.closeTo(60, 5), + expect.closeTo(90, 5), + expect.closeTo(90, 5), + ]); + }); +}); diff --git a/compute/core/src/control-plane/in-memory.ts b/compute/core/tests/helpers/in-memory-control-plane.ts similarity index 80% rename from compute/core/src/control-plane/in-memory.ts rename to compute/core/tests/helpers/in-memory-control-plane.ts index fc77c0e..cc9b9ce 100644 --- a/compute/core/src/control-plane/in-memory.ts +++ b/compute/core/tests/helpers/in-memory-control-plane.ts @@ -1,4 +1,5 @@ import { EventEmitter } from 'node:events'; +import type { WorkerOperationKind } from '../../src/api-contracts'; import type { OperationEvent, OperationEventStream, @@ -7,8 +8,7 @@ import type { OperationState, OperationStateStore, QueuedOperation, -} from './types'; -import type { WorkerOperationKind } from '../api-contracts'; +} from '../../src/control-plane/types'; function topicFor(opId: string): string { return `op.${opId}`; @@ -49,14 +49,39 @@ export class InMemoryOperationQueue implements OperationQueue { export class InMemoryOperationStateStore implements OperationStateStore { private readonly stateByOpId = new Map(); + private readonly revisionByOpId = new Map(); private readonly opIndexByKey = new Map(); async getOpState(opId: string): Promise { return this.stateByOpId.get(opId) ?? null; } + async getOpStateRecord(opId: string): Promise<{ state: OperationState; revision: number } | null> { + const state = this.stateByOpId.get(opId); + if (!state) return null; + return { + state, + revision: this.revisionByOpId.get(opId) ?? 0, + }; + } + async putOpState(state: OperationState): Promise { this.stateByOpId.set(state.opId, state); + this.revisionByOpId.set(state.opId, (this.revisionByOpId.get(state.opId) ?? 0) + 1); + } + + async compareAndSetOpState(input: { + opId: string; + expectedRevision: number; + newState: OperationState; + }): Promise { + const currentState = this.stateByOpId.get(input.opId); + if (!currentState) return false; + const currentRevision = this.revisionByOpId.get(input.opId) ?? 0; + if (currentRevision !== input.expectedRevision) return false; + this.stateByOpId.set(input.opId, input.newState); + this.revisionByOpId.set(input.opId, currentRevision + 1); + return true; } async getOpIndex(opKey: string): Promise { diff --git a/compute/worker/src/control-plane/jetstream.ts b/compute/worker/src/control-plane/jetstream.ts index 642906d..9225d96 100644 --- a/compute/worker/src/control-plane/jetstream.ts +++ b/compute/worker/src/control-plane/jetstream.ts @@ -27,6 +27,7 @@ export interface KvStoreLike { put(key: string, data: Uint8Array): Promise; create(key: string, data: Uint8Array): Promise; update(key: string, data: Uint8Array, version: number): Promise; + keys(filter?: string | string[]): Promise>; } function toErrorMessage(error: unknown): string { @@ -80,10 +81,18 @@ export class JetStreamOperationStateStore implements Operation } async getOpState(opId: string): Promise | null> { + const record = await this.getOpStateRecord(opId); + return record?.state ?? null; + } + + async getOpStateRecord(opId: string): Promise<{ state: OperationState; revision: number } | null> { const kv = await this.getKv(); const entry = await kv.get(opStateKvKey(opId)); if (!isPut(entry)) return null; - return this.opStateCodec.decode(entry.value); + return { + state: this.opStateCodec.decode(entry.value), + revision: entry.revision, + }; } async putOpState(state: OperationState): Promise { @@ -91,6 +100,37 @@ export class JetStreamOperationStateStore implements Operation await kv.put(opStateKvKey(state.opId), this.opStateCodec.encode(state)); } + async compareAndSetOpState(input: { + opId: string; + expectedRevision: number; + newState: OperationState; + }): Promise { + const kv = await this.getKv(); + try { + await kv.update( + opStateKvKey(input.opId), + this.opStateCodec.encode(input.newState), + input.expectedRevision, + ); + return true; + } catch (error) { + if (isCasConflictError(error)) return false; + throw error; + } + } + + async listOpStates(): Promise[]> { + const kv = await this.getKv(); + const keys = await kv.keys('op_state.*'); + const states: OperationState[] = []; + for await (const key of keys) { + const entry = await kv.get(key); + if (!isPut(entry)) continue; + states.push(this.opStateCodec.decode(entry.value)); + } + return states; + } + async getOpIndex(opKey: string): Promise<{ opId: string } | null> { const kv = await this.getKv(); const entry = await kv.get(opIndexKvKey(opKey)); diff --git a/compute/worker/src/orphan-recovery.ts b/compute/worker/src/orphan-recovery.ts new file mode 100644 index 0000000..882ea94 --- /dev/null +++ b/compute/worker/src/orphan-recovery.ts @@ -0,0 +1,100 @@ +import type { + PdfLayoutJobResult, + WhisperAlignJobResult, + WorkerJobTiming, + WorkerJobState, + WorkerOperationState, +} from '@openreader/compute-core/api-contracts'; + +export type StreamedOperationState = WorkerOperationState; + +export interface OrphanRecoveryStateStore { + getOpStateRecord(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>; + listOpStates(): Promise; +} + +export interface OrphanRecoveryOrchestrator { + markFailedIfUnchanged(input: { + current: StreamedOperationState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise; +} + +export interface RecoverOrphanedOperationsInput { + operationStateStore: OrphanRecoveryStateStore; + orchestrator: OrphanRecoveryOrchestrator; + whisperTimeoutMs: number; + pdfTimeoutMs: number; + opStaleMs: number; + nowMs?: number; +} + +function isInflightStatus(status: WorkerJobState): boolean { + return status === 'queued' || status === 'running'; +} + +export function getOrphanRecoveryThresholdMs(input: { + state: StreamedOperationState; + whisperTimeoutMs: number; + pdfTimeoutMs: number; + opStaleMs: number; +}): number | null { + if (!isInflightStatus(input.state.status)) return null; + if (input.state.status === 'running') { + return input.state.kind === 'whisper_align' ? input.whisperTimeoutMs : input.pdfTimeoutMs; + } + if (input.state.kind !== 'pdf_layout') return null; + return input.opStaleMs; +} + +export async function recoverOrphanedOperations( + input: RecoverOrphanedOperationsInput, +): Promise>> { + const nowMs = input.nowMs ?? Date.now(); + const states = await input.operationStateStore.listOpStates(); + const candidateStates = states.filter((state) => ( + getOrphanRecoveryThresholdMs({ + state, + whisperTimeoutMs: input.whisperTimeoutMs, + pdfTimeoutMs: input.pdfTimeoutMs, + opStaleMs: input.opStaleMs, + }) !== null + )); + const recoveredStates: Array> = []; + + for (const candidate of candidateStates) { + const record = await input.operationStateStore.getOpStateRecord(candidate.opId); + if (!record) continue; + const staleAfterMs = getOrphanRecoveryThresholdMs({ + state: record.state, + whisperTimeoutMs: input.whisperTimeoutMs, + pdfTimeoutMs: input.pdfTimeoutMs, + opStaleMs: input.opStaleMs, + }); + if (staleAfterMs === null) continue; + const ageMs = nowMs - record.state.updatedAt; + if (ageMs <= staleAfterMs) continue; + + const recovered = await input.orchestrator.markFailedIfUnchanged({ + current: record.state, + expectedRevision: record.revision, + error: { + code: 'WORKER_ORPHANED_OP', + message: `Worker stopped before completion; stale operation recovered during reconciliation after ${staleAfterMs}ms`, + }, + updatedAt: nowMs, + }); + if (!recovered) continue; + + recoveredStates.push({ + opId: recovered.opId, + kind: recovered.kind, + status: record.state.status, + }); + } + + return recoveredStates; +} diff --git a/compute/worker/src/runtime.ts b/compute/worker/src/runtime.ts index 2a43774..e208e63 100644 --- a/compute/worker/src/runtime.ts +++ b/compute/worker/src/runtime.ts @@ -56,6 +56,10 @@ import { hashOpKey, } from './control-plane/jetstream'; import { type JsonCodec, createJsonCodec } from './control-plane/json-codec'; +import { + recoverOrphanedOperations, + type StreamedOperationState, +} from './orphan-recovery'; import { buildInferProgressForPageParsed, buildInferProgressForPageStart } from './pdf-progress'; import { buildQueueWaitTiming, decideRetryAction } from './worker-loop-policy'; @@ -69,6 +73,7 @@ const COMPUTE_STATE_BUCKET = 'compute_state'; const COMPUTE_STATE_TTL_MS = 24 * 60 * 60 * 1000; const LOOP_ERROR_BACKOFF_MS = 500; const RUNNING_HEARTBEAT_MS = 5000; +const OP_EVENTS_KEEPALIVE_MS = 15_000; const DOCUMENT_ID_REGEX = /^[a-f0-9]{64}$/i; const SAFE_NAMESPACE_REGEX = /^[a-zA-Z0-9._-]{1,128}$/; const WHISPER_MAX_DELIVER = 1; @@ -78,6 +83,7 @@ const NATS_API_TIMEOUT_MS = 60_000; // put it to sleep. Reconnect happens lazily on the next inbound request. const IDLE_DISCONNECT_MS = 120_000; const IDLE_CHECK_INTERVAL_MS = 5_000; +const ORPHAN_SWEEP_INTERVAL_MS = 15_000; // Bounded pull window so consumer loops yield periodically and can be stopped // cleanly when going idle, instead of blocking on a long-lived pull. const PULL_EXPIRES_MS = 5_000; @@ -106,8 +112,6 @@ interface NatsSession { layoutConsumer: Consumer; } -type StreamedOperationState = WorkerOperationState; - interface OperationEventStreamLike { subscribe(input: { opId: string; @@ -119,6 +123,8 @@ interface OperationEventStreamLike { interface OperationStateStoreLike { getOpState(opId: string): Promise; + getOpStateRecord?(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>; + listOpStates?(): Promise; } interface OrchestratorLike { @@ -147,6 +153,13 @@ interface OrchestratorLike { updatedAt?: number; timing?: WorkerJobTiming; }): Promise; + markFailedIfUnchanged?(input: { + current: StreamedOperationState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise; } export interface ComputeWorkerRouteDeps { @@ -540,6 +553,7 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti let connecting: Promise | null = null; let workerLoops: Promise[] = []; let idleTimer: NodeJS.Timeout | null = null; + let orphanSweepTimer: NodeJS.Timeout | null = null; let stopping = false; let loopStopRequested = false; @@ -568,6 +582,19 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti idleTimer.unref?.(); } + function startOrphanSweepTimer(): void { + if (orphanSweepTimer) return; + orphanSweepTimer = setInterval(() => { + if (!session || stopping) return; + void runOrphanedOpRecovery({ force: true }).catch((error) => { + app.log.error({ + error: toErrorMessage(error), + }, 'periodic orphaned operation recovery failed'); + }); + }, ORPHAN_SWEEP_INTERVAL_MS); + orphanSweepTimer.unref?.(); + } + async function disconnect(reason: string): Promise { const current = session; if (!current) return; @@ -579,6 +606,10 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti clearInterval(idleTimer); idleTimer = null; } + if (orphanSweepTimer) { + clearInterval(orphanSweepTimer); + orphanSweepTimer = null; + } try { await current.nc.close(); } catch { @@ -616,9 +647,12 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti const layoutConsumer = await js.consumers.get(JOBS_STREAM_NAME, LAYOUT_CONSUMER_NAME); const next: NatsSession = { nc, js, jsm, kv, whisperConsumer, layoutConsumer }; session = next; + sessionGeneration += 1; + orphanRecoveryDoneForGeneration = -1; markActivity(); startWorkerLoops(next); startIdleTimer(); + startOrphanSweepTimer(); // Safety net: if the connection closes for any reason (network drop after // exhausting reconnects, or our own disconnect), drop the stale session so // the next request reconnects cleanly. @@ -730,8 +764,65 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti const operationStateStore = options.routeDeps?.operationStateStore ?? defaultOperationStateStore; const operationEventStream = options.routeDeps?.operationEventStream ?? defaultOperationEventStream; const orchestrator = options.routeDeps?.orchestrator ?? defaultOrchestrator; + let orphanRecoveryPromise: Promise | null = null; + let orphanRecoveryDoneForGeneration = -1; + let sessionGeneration = options.routeDeps ? 0 : -1; + + const runOrphanedOpRecovery = async (options?: { force?: boolean }): Promise => { + if (typeof operationStateStore.listOpStates !== 'function') return; + if (typeof operationStateStore.getOpStateRecord !== 'function') return; + if (typeof orchestrator.markFailedIfUnchanged !== 'function') return; + if (!options?.force && orphanRecoveryDoneForGeneration === sessionGeneration) return; + if (orphanRecoveryPromise) { + await orphanRecoveryPromise; + return; + } + + orphanRecoveryPromise = (async () => { + const recoveredStates = await recoverOrphanedOperations({ + operationStateStore: operationStateStore as typeof operationStateStore & { + getOpStateRecord(opId: string): Promise<{ state: StreamedOperationState; revision: number } | null>; + listOpStates(): Promise; + }, + orchestrator: orchestrator as typeof orchestrator & { + markFailedIfUnchanged(input: { + current: StreamedOperationState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: WorkerJobTiming; + }): Promise; + }, + whisperTimeoutMs, + pdfTimeoutMs, + opStaleMs, + }); + + if (recoveredStates.length > 0) { + app.log.warn({ + recoveredCount: recoveredStates.length, + ops: recoveredStates.map((state) => ({ + opId: state.opId, + kind: state.kind, + status: state.status, + })), + }, 'recovered stale in-flight operations during reconciliation'); + } + + orphanRecoveryDoneForGeneration = sessionGeneration; + })().finally(() => { + orphanRecoveryPromise = null; + }); + + await orphanRecoveryPromise; + }; + + const ensureOrphanedOpRecovery = async (): Promise => { + await runOrphanedOpRecovery(); + }; const getOpState = async (opId: string): Promise => { + await ensureOrphanedOpRecovery(); return await operationStateStore.getOpState(opId); }; @@ -797,6 +888,7 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti } const requestOp = parsed.data as WorkerOperationRequest; + await ensureOrphanedOpRecovery(); const op = await orchestrator.enqueueOrReuse(requestOp); app.log.info({ kind: requestOp.kind, @@ -863,6 +955,7 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti let closed = false; let unsubscribe: (() => void) | null = null; + let keepalive: NodeJS.Timeout | null = null; const writeSnapshot = (snapshot: StreamedOperationState, eventId: number): void => { if (closed || reply.raw.writableEnded) return; @@ -884,6 +977,10 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti unsubscribe(); unsubscribe = null; } + if (keepalive) { + clearInterval(keepalive); + keepalive = null; + } activeSse = Math.max(0, activeSse - 1); markActivity(); if (!reply.raw.writableEnded) { @@ -903,6 +1000,11 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti return reply; } + keepalive = setInterval(() => { + if (closed || reply.raw.writableEnded) return; + reply.raw.write(': keepalive\n\n'); + }, OP_EVENTS_KEEPALIVE_MS); + unsubscribe = await operationEventStream.subscribe({ opId: params.data.opId, sinceEventId, @@ -1247,6 +1349,17 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti const result = await input.run(decoded.payload, context.queueWaitTiming?.queueWaitMs ?? 0, { onProgress: async (progress) => { + try { + input.msg.working(); + } catch (ackError) { + app.log.warn({ + worker: input.workerLabel, + kind: context?.decoded.kind, + opId: context?.decoded.opId, + jobId: context?.decoded.jobId, + error: toErrorMessage(ackError), + }, 'failed to extend JetStream ack wait on progress'); + } await markProgress(context!, progress, Date.now()); }, }); @@ -1380,6 +1493,10 @@ export async function createComputeWorkerApp(options: CreateComputeWorkerAppOpti clearInterval(idleTimer); idleTimer = null; } + if (orphanSweepTimer) { + clearInterval(orphanSweepTimer); + orphanSweepTimer = null; + } await app.close(); await Promise.allSettled(workerLoops); const current = session; diff --git a/compute/worker/tests/api/routes.test.ts b/compute/worker/tests/api/routes.test.ts index 77c4d05..dad89d9 100644 --- a/compute/worker/tests/api/routes.test.ts +++ b/compute/worker/tests/api/routes.test.ts @@ -119,4 +119,79 @@ describe('compute worker API routes', () => { expect(stream.body).toContain('id: 7'); expect(stream.body).toContain('"status":"succeeded"'); }); + + test('marks stale running whisper and pdf ops failed during request-time orphan recovery but leaves queued ops on the conservative path', async () => { + const now = Date.now(); + fake.seedState({ + opId: 'op-stale-whisper-running', + opKey: 'k-stale-whisper-running', + kind: 'whisper_align', + jobId: 'job-op-stale-whisper-running', + status: 'running', + queuedAt: 1, + updatedAt: now - 40_000, + }); + fake.seedState({ + opId: 'op-stale-whisper-queued', + opKey: 'k-stale-whisper-queued', + kind: 'whisper_align', + jobId: 'job-op-stale-whisper-queued', + status: 'queued', + queuedAt: 1, + updatedAt: now - 40_000, + }); + fake.seedState({ + opId: 'op-stale-pdf-running', + opKey: 'k-stale-pdf-running', + kind: 'pdf_layout', + jobId: 'job-op-stale-pdf-running', + status: 'running', + queuedAt: 1, + updatedAt: now - 310_000, + }); + fake.seedState({ + opId: 'op-stale-pdf-queued', + opKey: 'k-stale-pdf-queued', + kind: 'pdf_layout', + jobId: 'job-op-stale-pdf-queued', + status: 'queued', + queuedAt: 1, + updatedAt: now - 310_000, + }); + + // GET /ops/:opId resolves via getOpState(), which first awaits the shared + // orphanRecoveryPromise path through ensureOrphanedOpRecovery(). + const fetch = await runtime.app.inject({ + method: 'GET', + url: '/ops/op-stale-whisper-running', + headers: AUTH, + }); + + expect(fetch.statusCode).toBe(200); + expect(fetch.json()).toMatchObject({ + opId: 'op-stale-whisper-running', + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + expect(fake.getState('op-stale-whisper-running')).toMatchObject({ + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + expect(fake.getState('op-stale-whisper-queued')).toMatchObject({ + status: 'queued', + }); + expect(fake.getState('op-stale-pdf-running')).toMatchObject({ + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + expect(fake.getState('op-stale-pdf-queued')).toMatchObject({ + status: 'queued', + }); + }); }); diff --git a/compute/worker/tests/fixtures/fake-control-plane.ts b/compute/worker/tests/fixtures/fake-control-plane.ts index 2d18f2b..9822523 100644 --- a/compute/worker/tests/fixtures/fake-control-plane.ts +++ b/compute/worker/tests/fixtures/fake-control-plane.ts @@ -13,6 +13,7 @@ type ComputeEvent = WorkerOperationEvent; export class FakeControlPlane { private readonly stateByOpId = new Map(); + private readonly revisionByOpId = new Map(); private readonly opIdByOpKey = new Map(); private readonly eventsByOpId = new Map(); private nextOpId = 1; @@ -44,9 +45,19 @@ export class FakeControlPlane { updatedAt: input.updatedAt, timing: input.timing, }), + markFailedIfUnchanged: async (input) => this.compareAndSetFailed(input), }, operationStateStore: { getOpState: async (opId) => this.stateByOpId.get(opId) ?? null, + getOpStateRecord: async (opId) => { + const state = this.stateByOpId.get(opId); + if (!state) return null; + return { + state, + revision: this.revisionByOpId.get(opId) ?? 0, + }; + }, + listOpStates: async () => Array.from(this.stateByOpId.values()), }, operationEventStream: { subscribe: async ({ opId, sinceEventId, onEvent }) => { @@ -62,6 +73,7 @@ export class FakeControlPlane { seedState(state: ComputeState): void { this.stateByOpId.set(state.opId, state); + this.revisionByOpId.set(state.opId, (this.revisionByOpId.get(state.opId) ?? 0) + 1); this.opIdByOpKey.set(state.opKey, state.opId); } @@ -71,6 +83,10 @@ export class FakeControlPlane { this.eventsByOpId.set(opId, list); } + getState(opId: string): ComputeState | null { + return this.stateByOpId.get(opId) ?? null; + } + private async enqueueOrReuse(request: WorkerOperationRequest): Promise { const existingId = this.opIdByOpKey.get(request.opKey); if (existingId) { @@ -91,11 +107,30 @@ export class FakeControlPlane { }; this.stateByOpId.set(opId, state); + this.revisionByOpId.set(opId, 1); this.opIdByOpKey.set(request.opKey, opId); this.seedEvent(opId, { eventId: 1, snapshot: state }); return state; } + private async compareAndSetFailed(input: { + current: ComputeState; + expectedRevision: number; + error: { message: string; code?: string } | string; + updatedAt?: number; + timing?: ComputeState['timing']; + }): Promise { + const currentRevision = this.revisionByOpId.get(input.current.opId) ?? 0; + if (currentRevision !== input.expectedRevision) return null; + return this.updateState(input.current.opId, { + ...input.current, + status: 'failed', + error: typeof input.error === 'string' ? { message: input.error } : input.error, + updatedAt: input.updatedAt, + timing: input.timing, + }); + } + private async updateState( opId: string, patch: Partial, @@ -110,6 +145,7 @@ export class FakeControlPlane { updatedAt: patch.updatedAt ?? Date.now(), }; this.stateByOpId.set(opId, next); + this.revisionByOpId.set(opId, (this.revisionByOpId.get(opId) ?? 0) + 1); const currentEvents = this.eventsByOpId.get(opId) ?? []; const nextEventId = (currentEvents.at(-1)?.eventId ?? 0) + 1; currentEvents.push({ eventId: nextEventId, snapshot: next }); diff --git a/compute/worker/tests/unit/orphan-recovery.test.ts b/compute/worker/tests/unit/orphan-recovery.test.ts new file mode 100644 index 0000000..712fdf6 --- /dev/null +++ b/compute/worker/tests/unit/orphan-recovery.test.ts @@ -0,0 +1,59 @@ +import { afterEach, describe, expect, test, vi } from 'vitest'; +import { recoverOrphanedOperations } from '../../src/orphan-recovery'; +import { FakeControlPlane } from '../fixtures/fake-control-plane'; + +describe('orphan recovery', () => { + afterEach(() => { + vi.useRealTimers(); + }); + + test('recovers a running whisper op when a later sweep crosses the timeout in the same session', async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date('2026-06-03T08:00:00.000Z')); + + const fake = new FakeControlPlane(); + const startedAt = Date.now(); + fake.seedState({ + opId: 'op-whisper-running', + opKey: 'k-whisper-running', + kind: 'whisper_align', + jobId: 'job-op-whisper-running', + status: 'running', + queuedAt: startedAt, + updatedAt: startedAt, + }); + + const firstSweep = await recoverOrphanedOperations({ + operationStateStore: fake.deps.operationStateStore, + orchestrator: fake.deps.orchestrator, + whisperTimeoutMs: 30_000, + pdfTimeoutMs: 300_000, + opStaleMs: 1_800_000, + }); + expect(firstSweep).toEqual([]); + expect(fake.getState('op-whisper-running')).toMatchObject({ + status: 'running', + }); + + vi.advanceTimersByTime(31_000); + + const secondSweep = await recoverOrphanedOperations({ + operationStateStore: fake.deps.operationStateStore, + orchestrator: fake.deps.orchestrator, + whisperTimeoutMs: 30_000, + pdfTimeoutMs: 300_000, + opStaleMs: 1_800_000, + }); + expect(secondSweep).toEqual([{ + opId: 'op-whisper-running', + kind: 'whisper_align', + status: 'running', + }]); + expect(fake.getState('op-whisper-running')).toMatchObject({ + status: 'failed', + error: { + code: 'WORKER_ORPHANED_OP', + }, + }); + }); +}); diff --git a/docs-site/docs/deploy/compute-worker.md b/docs-site/docs/deploy/compute-worker.md index b5ec152..cd279db 100644 --- a/docs-site/docs/deploy/compute-worker.md +++ b/docs-site/docs/deploy/compute-worker.md @@ -41,7 +41,7 @@ S3_SECRET_ACCESS_KEY=... - Embedded/local mode: configure the root `.env` only. - External worker mode: set `COMPUTE_WORKER_URL` and `COMPUTE_WORKER_TOKEN` on the app, and worker runtime values on the worker service. -- Keep shared values aligned across app and worker: `COMPUTE_WORKER_TOKEN`, `S3_*`, `COMPUTE_WHISPER_TIMEOUT_MS`, `COMPUTE_PDF_TIMEOUT_MS`, and `COMPUTE_OP_STALE_MS`. +- Keep shared values aligned across app and worker: `COMPUTE_WORKER_TOKEN`, `S3_*`, `COMPUTE_WHISPER_TIMEOUT_MS`, `COMPUTE_PDF_TIMEOUT_MS`, `COMPUTE_PDF_JOB_ATTEMPTS`, and `COMPUTE_OP_STALE_MS`. ::: Common optional variables: @@ -76,6 +76,7 @@ COMPUTE_WORKER_TOKEN= # Optional shared overrides: # COMPUTE_WHISPER_TIMEOUT_MS=30000 # COMPUTE_PDF_TIMEOUT_MS=300000 +# COMPUTE_PDF_JOB_ATTEMPTS=1 # COMPUTE_OP_STALE_MS=1800000 ``` diff --git a/docs-site/docs/deploy/local-development.md b/docs-site/docs/deploy/local-development.md index 338aeb0..7454b7b 100644 --- a/docs-site/docs/deploy/local-development.md +++ b/docs-site/docs/deploy/local-development.md @@ -189,9 +189,11 @@ COMPUTE_WORKER_TOKEN= ``` Ownership in external worker mode: -- root `.env`: app routing/auth (`COMPUTE_WORKER_URL`, `COMPUTE_WORKER_TOKEN`) plus optional shared timeout/stale overrides +- root `.env`: app routing/auth (`COMPUTE_WORKER_URL`, `COMPUTE_WORKER_TOKEN`) plus optional shared timeout/stale/retry overrides such as `COMPUTE_PDF_JOB_ATTEMPTS` - `compute/worker/.env*` (or worker platform env): worker runtime variables (`NATS_*`, `S3_*`, model base URLs, worker tuning) +For embedded worker startup (`COMPUTE_WORKER_URL` unset), worker tuning values such as `COMPUTE_PDF_JOB_ATTEMPTS` must be set in the root `.env` because `compute/worker/.env*` is ignored in that mode. + Worker mode requires worker-reachable shared object storage (S3-compatible endpoint). For external worker mode, object storage must be shared/reachable by both app and worker services. diff --git a/docs-site/docs/reference/environment-variables.md b/docs-site/docs/reference/environment-variables.md index 45786fc..d80bd8e 100644 --- a/docs-site/docs/reference/environment-variables.md +++ b/docs-site/docs/reference/environment-variables.md @@ -48,6 +48,7 @@ Runtime site features are seeded with `RUNTIME_SEED_JSON` / `RUNTIME_SEED_JSON_P | `COMPUTE_JOB_CONCURRENCY` | Compute | `1` | Shared compute concurrency cap | | `COMPUTE_WHISPER_TIMEOUT_MS` | Compute | `30000` | Whisper alignment timeout budget | | `COMPUTE_PDF_TIMEOUT_MS` | Compute | `300000` | PDF parse timeout budget | +| `COMPUTE_PDF_JOB_ATTEMPTS` | Compute | `1` | Max JetStream deliveries for PDF layout jobs | | `COMPUTE_OP_STALE_MS` | Compute | `max(30m, 4x max compute timeout)` | Shared stale window for compute op replacement | | `WHISPER_MODEL_BASE_URL` | Compute model source | onnx-community default | Override Whisper ONNX model base URL | | `PDF_LAYOUT_MODEL_BASE_URL` | Compute model source | PP-DocLayoutV3 default | Override PDF layout ONNX model base URL | @@ -307,6 +308,13 @@ PDF parse timeout budget. - Default: `300000` +### COMPUTE_PDF_JOB_ATTEMPTS + +Max JetStream deliveries for PDF layout jobs. + +- Default: `1` +- In embedded worker mode, set this in the root `.env` + ### COMPUTE_OP_STALE_MS Stale operation window before worker/app cleanup logic can replace an op. diff --git a/src/app/(app)/pdf/[id]/page.tsx b/src/app/(app)/pdf/[id]/page.tsx index 0d1b3b9..79d4060 100644 --- a/src/app/(app)/pdf/[id]/page.tsx +++ b/src/app/(app)/pdf/[id]/page.tsx @@ -19,6 +19,7 @@ import { RateLimitBanner } from '@/components/auth/RateLimitBanner'; import { useAuthRateLimit } from '@/contexts/AuthRateLimitContext'; import { useFeatureFlag } from '@/contexts/RuntimeConfigContext'; import { LoadingSpinner } from '@/components/Spinner'; +import { PdfLayoutScan } from '@/components/reader/PdfLayoutScan'; import { Button, ButtonLink } from '@/components/ui'; import { FORCE_REPARSE_CONFIRM_MESSAGE, @@ -279,21 +280,13 @@ export default function PDFViewerPage() { const isMerging = parseProgress?.phase === 'merge'; let statusText = 'Loading PDF...'; - let statusSubText = 'Initializing document renderer'; if (!isLoading) { if (parseUiState === 'pending') { statusText = 'Preparing PDF layout...'; - statusSubText = parseProgress?.phase === 'merge' - ? 'Finalizing stitched block structure' - : 'Queueing parser and preparing page extraction'; } else if (parseUiState === 'running') { statusText = 'Parsing PDF layout blocks...'; - statusSubText = parseProgress?.phase === 'merge' - ? 'Merging cross-page sections' - : 'Inferring reading order and text regions'; } else if (parseUiState === 'failed') { statusText = 'PDF parsing failed. Retry to continue.'; - statusSubText = 'The parser could not build a usable layout map'; } } @@ -305,34 +298,89 @@ export default function PDFViewerPage() { return (
-
+
{showDetailedParseLoader ? ( -
-
-
-
+
+ {/* prism top edge + corner glow for depth */} +
+
+ {/* dotted texture spanning the whole loader */} +
+ +
+ {/* header: status badge + model attribution */} +
- + {parseUiState === 'failed' ? ( + + + + ) : ( + + )} PDF Layout Parse
-

{statusText}

+
+ + PP-DocLayout-V3 +
-
-
-

- {hasMeasuredProgress ? `Page ${pagesParsed} / ${totalPages}` : 'Awaiting first page'} -

-

{stageLabel}

+
+ {/* animated layout scanner — static "halted" view when failed */} +
+
-
-
-
-

- {hasMeasuredProgress ? `${Math.round(progressPercent)}% complete` : statusSubText} + + {/* live status + progress */} + {parseUiState === 'failed' ? ( +

+

{statusText}

+

{stageLabel}

+
+ ) : ( +
+
+

+ {hasMeasuredProgress ? `Page ${pagesParsed} / ${totalPages}` : 'Awaiting first page'} +

+

{stageLabel}

+
+
+
+
+

+ {hasMeasuredProgress ? `${Math.round(progressPercent)}% complete` : 'Calibrating layout pass'} +

+
+ )} +
+ + {/* attribution footer */} +
+ + + +

+ Classifying footnotes, titles, tables, figures, formulas, & more with PP-DocLayout-V3.

diff --git a/src/app/(app)/pdf/[id]/usePdfDocument.ts b/src/app/(app)/pdf/[id]/usePdfDocument.ts index 1720f63..b252618 100644 --- a/src/app/(app)/pdf/[id]/usePdfDocument.ts +++ b/src/app/(app)/pdf/[id]/usePdfDocument.ts @@ -291,6 +291,10 @@ export function usePdfDocument(): PdfDocumentState { pdfDocumentRef.current = pdfDocument; }, [pdfDocument]); + useEffect(() => { + pageTextCacheRef.current.clear(); + }, [parsedDocument, documentSettings.pdf?.skipBlockKinds]); + useEffect(() => { setCurrDocPage(currDocPageNumber); }, [currDocPageNumber]); @@ -538,7 +542,10 @@ export function usePdfDocument(): PdfDocumentState { if (!currDocId) return; try { const forced = await forceReparsePdfDocument(currDocId); + loadSeqRef.current += 1; + pageTextCacheRef.current.clear(); setParsedDocument(null); + setCurrDocText(undefined); setParseStatus(forced.status); setParseProgress(null); setActiveParseOpId(forced.opId ?? null); diff --git a/src/app/api/documents/[id]/parsed/events/route.ts b/src/app/api/documents/[id]/parsed/events/route.ts index 31af7fe..2649ae5 100644 --- a/src/app/api/documents/[id]/parsed/events/route.ts +++ b/src/app/api/documents/[id]/parsed/events/route.ts @@ -5,7 +5,7 @@ import { documents } from '@/db/schema'; import { requireAuthContext } from '@/lib/server/auth/auth'; import { getWorkerClientConfigFromEnv } from '@/lib/server/compute/worker'; import { isAbortLikeError } from '@/lib/server/compute/abort-like-error'; -import { snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state'; +import { isWorkerOperationStateStale, snapshotFromWorkerState } from '@/lib/server/compute/worker-parse-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; import { isValidDocumentId } from '@/lib/server/documents/blobstore'; import { normalizeParseStatus, parseDocumentParseState } from '@/lib/server/documents/parse-state'; @@ -16,6 +16,7 @@ import { errorResponse } from '@/lib/server/errors/next-response'; import { logDegraded, logServerError } from '@/lib/server/errors/logging'; import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; import { parseSseEventId, parseSsePayload } from '@openreader/compute-core'; +import { getComputeOpStaleMs } from '@openreader/compute-core'; import type { PdfLayoutJobResult, WorkerOperationEvent, WorkerOperationState } from '@openreader/compute-core/api-contracts'; export const dynamic = 'force-dynamic'; @@ -55,6 +56,7 @@ function sleep(ms: number): Promise { } async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Promise { + const opStaleMs = getComputeOpStaleMs(); const state = await healStaleDocumentParseState({ documentId: row.id, userId: row.userId, @@ -68,7 +70,11 @@ async function toSnapshotState(row: ParseRow, preferredOpId?: string | null): Pr // the per-user document row currently says "ready" or has a different opId. if (opId && (requestedOpId !== null || parseStatus !== 'ready')) { const workerState = await fetchWorkerOperationState(opId); - if (workerState && workerState.opId === opId) { + if ( + workerState + && workerState.opId === opId + && !isWorkerOperationStateStale(workerState, opStaleMs) + ) { return { snapshot: { ...snapshotFromWorkerState(workerState), @@ -325,119 +331,136 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string continue; } - workerAbort = new AbortController(); - const query = lastEventId && lastEventId > 0 - ? `?sinceEventId=${encodeURIComponent(String(lastEventId))}` - : ''; - const response = await fetch( - `${workerCfg.baseUrl}/ops/${encodeURIComponent(currentOpId)}/events${query}`, - { - method: 'GET', - headers: { - Authorization: `Bearer ${workerCfg.token}`, - Accept: 'text/event-stream', - ...(lastEventId && lastEventId > 0 ? { 'Last-Event-ID': String(lastEventId) } : {}), + try { + workerAbort = new AbortController(); + const query = lastEventId && lastEventId > 0 + ? `?sinceEventId=${encodeURIComponent(String(lastEventId))}` + : ''; + const response = await fetch( + `${workerCfg.baseUrl}/ops/${encodeURIComponent(currentOpId)}/events${query}`, + { + method: 'GET', + headers: { + Authorization: `Bearer ${workerCfg.token}`, + Accept: 'text/event-stream', + ...(lastEventId && lastEventId > 0 ? { 'Last-Event-ID': String(lastEventId) } : {}), + }, + cache: 'no-store', + signal: workerAbort.signal, }, - cache: 'no-store', - signal: workerAbort.signal, - }, - ); + ); - if (closed) return; + if (closed) return; - if (!response.ok) { - const upstreamResponseBody = await response.text().catch(() => ''); - logger.warn({ - event: 'documents.parsed.events.worker_stream_request_failed', - degraded: true, - step: 'worker_stream_request', - documentId: id, - opId: currentOpId, - status: response.status, - upstreamResponseBody, - error: { - name: 'WorkerStreamRequestFailed', - message: `Worker stream request failed with status ${response.status}`, - }, - }, 'Worker stream request failed'); - await sleep(500); - continue; - } - if (!response.body) { - logger.warn({ - event: 'documents.parsed.events.worker_stream_missing_body', - degraded: true, - step: 'worker_stream_body', - documentId: id, - opId: currentOpId, - error: { - name: 'WorkerStreamMissingBody', - message: 'Worker stream response missing body', - }, - }, 'Worker stream response missing body'); - await sleep(500); - continue; - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ''; - let streamEnded = false; - - while (!closed && !streamEnded) { - const read = await reader.read(); - if (read.done) { - streamEnded = true; - break; + if (!response.ok) { + const upstreamResponseBody = await response.text().catch(() => ''); + logger.warn({ + event: 'documents.parsed.events.worker_stream_request_failed', + degraded: true, + step: 'worker_stream_request', + documentId: id, + opId: currentOpId, + status: response.status, + upstreamResponseBody, + error: { + name: 'WorkerStreamRequestFailed', + message: `Worker stream request failed with status ${response.status}`, + }, + }, 'Worker stream request failed'); + await sleep(500); + continue; + } + if (!response.body) { + logger.warn({ + event: 'documents.parsed.events.worker_stream_missing_body', + degraded: true, + step: 'worker_stream_body', + documentId: id, + opId: currentOpId, + error: { + name: 'WorkerStreamMissingBody', + message: 'Worker stream response missing body', + }, + }, 'Worker stream response missing body'); + await sleep(500); + continue; } - buffer += decoder.decode(read.value, { stream: true }); + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + let streamEnded = false; - while (true) { - const frameEnd = buffer.indexOf('\n\n'); - if (frameEnd < 0) break; - const frame = buffer.slice(0, frameEnd); - buffer = buffer.slice(frameEnd + 2); - - const eventId = parseSseEventId(frame); - if (eventId && eventId > 0) { - lastEventId = eventId; + while (!closed && !streamEnded) { + const read = await reader.read(); + if (read.done) { + streamEnded = true; + break; } - const payload = parseSsePayload(frame); - if (!payload) continue; + buffer += decoder.decode(read.value, { stream: true }); - let parsed: WorkerOperationEvent | WorkerOperationState; - try { - parsed = JSON.parse(payload) as WorkerOperationEvent | WorkerOperationState; - } catch { - continue; - } + while (true) { + const frameEnd = buffer.indexOf('\n\n'); + if (frameEnd < 0) break; + const frame = buffer.slice(0, frameEnd); + buffer = buffer.slice(frameEnd + 2); - const workerSnapshot: WorkerOperationState = ( - parsed && typeof parsed === 'object' && 'snapshot' in parsed - ? parsed.snapshot - : parsed as WorkerOperationState - ); - if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; + const eventId = parseSseEventId(frame); + if (eventId && eventId > 0) { + lastEventId = eventId; + } - const nextSnapshot: ParsedSnapshot = { - ...snapshotFromWorkerState(workerSnapshot), - opId: workerSnapshot.opId, - }; - const nextSignature = JSON.stringify(nextSnapshot); - if (nextSignature !== signature) { - current = nextSnapshot; - signature = nextSignature; - currentFromWorker = true; - writeSnapshot(current); - } + const payload = parseSsePayload(frame); + if (!payload) continue; - if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { - closeStream(); - return; + let parsed: WorkerOperationEvent | WorkerOperationState; + try { + parsed = JSON.parse(payload) as WorkerOperationEvent | WorkerOperationState; + } catch { + continue; + } + + const workerSnapshot: WorkerOperationState = ( + parsed && typeof parsed === 'object' && 'snapshot' in parsed + ? parsed.snapshot + : parsed as WorkerOperationState + ); + if (!workerSnapshot || workerSnapshot.opId !== currentOpId) continue; + + const nextSnapshot: ParsedSnapshot = { + ...snapshotFromWorkerState(workerSnapshot), + opId: workerSnapshot.opId, + }; + const nextSignature = JSON.stringify(nextSnapshot); + if (nextSignature !== signature) { + current = nextSnapshot; + signature = nextSignature; + currentFromWorker = true; + writeSnapshot(current); + } + + if (shouldCloseForTerminalSnapshot(current, currentFromWorker)) { + closeStream(); + return; + } } } + } catch (error) { + if (closed || isAbortLikeError(error)) return; + logDegraded(logger, { + event: 'documents.parsed.events.worker_stream_read_failed', + msg: 'Worker stream read failed; reconnecting', + step: 'worker_stream_read', + context: { + documentId: id, + opId: currentOpId, + requestId, + }, + error, + }); + await sleep(500); + continue; } if (closed) return; diff --git a/src/app/api/documents/[id]/parsed/route.ts b/src/app/api/documents/[id]/parsed/route.ts index 9738b90..0a783f2 100644 --- a/src/app/api/documents/[id]/parsed/route.ts +++ b/src/app/api/documents/[id]/parsed/route.ts @@ -7,6 +7,7 @@ import { requireAuthContext } from '@/lib/server/auth/auth'; import { createOrReusePdfWorkerOperation } from '@/lib/server/compute/worker-op-create'; import { documentParseStateFromWorkerState, + isWorkerOperationStateStale, snapshotFromWorkerState, } from '@/lib/server/compute/worker-parse-state'; import { fetchWorkerOperationState } from '@/lib/server/compute/worker-op-state'; @@ -33,6 +34,7 @@ import { createRequestLogger, hashForLog, type ServerLogger } from '@/lib/server import { errorResponse } from '@/lib/server/errors/next-response'; import { logDegraded } from '@/lib/server/errors/logging'; import type { ParsedPdfDocument } from '@/types/parsed-pdf'; +import { getComputeOpStaleMs } from '@openreader/compute-core'; import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; export const dynamic = 'force-dynamic'; @@ -192,6 +194,7 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string request: req, }); try { + const opStaleMs = getComputeOpStaleMs(); if (!isS3Configured()) return s3NotConfiguredResponse(); const authCtxOrRes = await requireAuthContext(req); @@ -255,7 +258,11 @@ export async function GET(req: NextRequest, ctx: { params: Promise<{ id: string if (effectiveOpId && effectiveStatus !== 'ready') { const workerState = await fetchWorkerOperationState(effectiveOpId); - if (workerState && workerState.opId === effectiveOpId) { + if ( + workerState + && workerState.opId === effectiveOpId + && !isWorkerOperationStateStale(workerState, opStaleMs) + ) { return finalizeFromWorkerState({ workerState, row, @@ -348,6 +355,7 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string request: req, }); try { + const opStaleMs = getComputeOpStaleMs(); if (!isS3Configured()) return s3NotConfiguredResponse(); const authCtxOrRes = await requireAuthContext(req); @@ -388,7 +396,12 @@ export async function POST(req: NextRequest, ctx: { params: Promise<{ id: string const existingOpId = normalizeOpId(state.opId); if (existingOpId) { const existing = await fetchWorkerOperationState(existingOpId); - if (existing && (existing.status === 'queued' || existing.status === 'running') && !replace) { + if ( + existing + && !isWorkerOperationStateStale(existing, opStaleMs) + && (existing.status === 'queued' || existing.status === 'running') + && !replace + ) { const snapshot = snapshotFromWorkerState(existing); return NextResponse.json({ error: 'Parse operation already in progress', diff --git a/src/app/globals.css b/src/app/globals.css index 284d34c..1d69c28 100644 --- a/src/app/globals.css +++ b/src/app/globals.css @@ -424,3 +424,31 @@ h1, h2, h3, h4, h5, h6 { -webkit-mask: linear-gradient(90deg, transparent 0%, black 8%, black 92%, transparent 100%); border-radius: 999px; /* subtle rounding to reinforce taper */ } + +/* Generic determinate progress fill with a moving sheen. Apply to the colored + fill element (the one whose width tracks progress); its track should clip. */ +.progress-fill { + position: relative; + overflow: hidden; +} +.progress-fill::after { + content: ''; + position: absolute; + inset: 0; + background: linear-gradient( + 90deg, + transparent, + color-mix(in srgb, var(--background) 55%, transparent), + transparent + ); + transform: translateX(-100%); + animation: progress-sheen 2.2s linear infinite; +} +@keyframes progress-sheen { + 0% { transform: translateX(-100%); } + /* sweep across at constant speed, then rest off-screen so the loop reset + (from the right edge back to the left) is never visible */ + 65% { transform: translateX(220%); } + 100% { transform: translateX(220%); } +} + diff --git a/src/components/ProgressCard.tsx b/src/components/ProgressCard.tsx index f8341d0..2933c9d 100644 --- a/src/components/ProgressCard.tsx +++ b/src/components/ProgressCard.tsx @@ -66,7 +66,7 @@ export function ProgressCard({ {/* Progress bar */}
diff --git a/src/components/auth/UserMenu.tsx b/src/components/auth/UserMenu.tsx index d9e44ea..6f31aa5 100644 --- a/src/components/auth/UserMenu.tsx +++ b/src/components/auth/UserMenu.tsx @@ -1,6 +1,5 @@ 'use client'; -import Link from 'next/link'; import { useAuthConfig } from '@/contexts/AuthRateLimitContext'; import { useFeatureFlag } from '@/contexts/RuntimeConfigContext'; import { useAuthSession } from '@/hooks/useAuthSession'; @@ -35,21 +34,19 @@ export function UserMenu({ if (variant === 'sidebar') { return (
- + } + label="Connect" + /> + {enableUserSignups && ( } - label="Connect" + label="Create account" /> - - {enableUserSignups && ( - - } - label="Create account" - /> - )}
); diff --git a/src/components/documents/DexieMigrationModal.tsx b/src/components/documents/DexieMigrationModal.tsx index df469d1..2efa278 100644 --- a/src/components/documents/DexieMigrationModal.tsx +++ b/src/components/documents/DexieMigrationModal.tsx @@ -155,7 +155,7 @@ export function DexieMigrationModal({

{status}

-
+
)} diff --git a/src/components/reader/PdfLayoutScan.module.css b/src/components/reader/PdfLayoutScan.module.css new file mode 100644 index 0000000..5ee5a31 --- /dev/null +++ b/src/components/reader/PdfLayoutScan.module.css @@ -0,0 +1,211 @@ +/* Scanner visualization for the PDF layout-parse loader. Scoped via CSS Module + so it stays out of the global stylesheet and works across the helper that + renders region content. Used by PdfLayoutScan.tsx. */ + +.stage { + position: relative; + width: 100%; + aspect-ratio: 1 / 1; + display: grid; + place-items: center; + background: transparent; +} + +.page { + position: relative; + width: 68%; + aspect-ratio: 3 / 4; + border-radius: 6px; + background: linear-gradient(180deg, color-mix(in srgb, var(--background) 92%, var(--foreground)), var(--background)); + border: 1px solid var(--line); + box-shadow: + var(--elev-2), + 0 0 0 1px color-mix(in srgb, var(--background) 60%, transparent), + inset 0 1px 0 color-mix(in srgb, var(--foreground) 6%, transparent); + overflow: hidden; +} + +/* ── readout chip (in the stage band, above the page) ─────────────────── */ +.tagRow { + position: absolute; + top: 4%; + left: 0; + right: 0; + display: flex; + justify-content: center; + z-index: 5; + padding: 0 4%; +} +.tag { + display: inline-flex; + align-items: center; + max-width: 100%; + padding: 3px 9px; + border-radius: 6px; + font-family: var(--font-display), system-ui, sans-serif; + font-size: 10px; + font-weight: 700; + line-height: 1.25; + letter-spacing: 0.01em; + white-space: nowrap; + color: var(--background); + background: var(--accent); + box-shadow: 0 2px 10px color-mix(in srgb, var(--accent) 50%, transparent); + animation: tag-in 0.34s var(--ease); +} +@keyframes tag-in { + from { opacity: 0; transform: translateY(-4px) scale(0.92); } + to { opacity: 1; transform: translateY(0) scale(1); } +} + +/* ── failed / halted state ────────────────────────────────────────────── */ +.tagFailed { + color: var(--foreground); + background: color-mix(in srgb, var(--foreground) 12%, transparent); + box-shadow: none; + animation: none; +} + +.pageFailed { + display: grid; + place-items: center; + opacity: 0.55; + filter: grayscale(0.4); +} +.alert { + width: 38%; + height: 38%; + stroke: var(--accent); + stroke-width: 1.6; + color: var(--accent); +} + +/* ── region boxes (one shown at a time, centered) ─────────────────────── */ +.solos { + position: absolute; + inset: 8%; + display: grid; + place-items: center; +} +.solo { + grid-area: 1 / 1; /* stack so regions cross-fade in place */ + display: flex; + align-items: center; + justify-content: center; + border-radius: 6px; + border: 1.5px solid var(--accent); + background: color-mix(in srgb, var(--foreground) 8%, transparent); + box-shadow: + 0 0 0 1px var(--accent-line), + 0 0 22px color-mix(in srgb, var(--accent) 30%, transparent); + padding: 4px 7px; + opacity: 0; + transform: scale(0.9); + transition: opacity 0.32s var(--ease), transform 0.32s var(--ease); + will-change: opacity, transform; +} +.solo.active { + opacity: 1; + transform: scale(1); +} + +/* inner content so each region reads as real document material */ +.lines { + display: flex; + flex-direction: column; + justify-content: center; + gap: 11%; + width: 100%; + height: 100%; +} +.linesHeading { gap: 22%; } +.lines > span { + display: block; + height: 3px; + width: 100%; + border-radius: 999px; + background: color-mix(in srgb, var(--foreground) 26%, transparent); +} +.linesHeading > span { + height: 6px; + background: color-mix(in srgb, var(--foreground) 46%, transparent); +} + +.glyph { + width: 100%; + height: 100%; + stroke: color-mix(in srgb, var(--foreground) 36%, transparent); + stroke-width: 2; + stroke-linejoin: round; + fill: none; +} +.glyph circle { + fill: color-mix(in srgb, var(--foreground) 30%, transparent); + stroke: none; +} + +.table { + display: grid; + grid-template-columns: repeat(4, 1fr); + grid-template-rows: repeat(4, 1fr); + gap: 2px; + width: 100%; + height: 100%; + background: color-mix(in srgb, var(--foreground) 16%, transparent); + border: 1px solid color-mix(in srgb, var(--foreground) 16%, transparent); + border-radius: 3px; + overflow: hidden; +} +.table > span { + background: color-mix(in srgb, var(--background) 82%, var(--foreground)); +} +.table > span:nth-child(-n + 4) { + background: color-mix(in srgb, var(--foreground) 22%, transparent); +} + +/* ── scan beam ────────────────────────────────────────────────────────── */ +.beam { + position: absolute; + left: -4%; + right: -4%; + height: 2px; + border-radius: 999px; + background: linear-gradient(90deg, transparent, var(--accent), transparent); + box-shadow: 0 0 14px 2px color-mix(in srgb, var(--accent) 55%, transparent); + animation: beam 1.05s var(--ease) infinite; + will-change: top, opacity; +} +.beam::after { + content: ''; + position: absolute; + left: 0; + right: 0; + bottom: 1px; + height: 30px; + background: linear-gradient(to top, color-mix(in srgb, var(--accent) 18%, transparent), transparent); +} +@keyframes beam { + 0% { top: -4%; opacity: 0; } + 8% { opacity: 1; } + 90% { opacity: 1; } + 100% { top: 104%; opacity: 0; } +} + +/* camera-style corner reticles */ +.corner { + position: absolute; + width: 10px; + height: 10px; + border: 1.5px solid var(--accent-line); + pointer-events: none; +} +.cornerTl { top: 5px; left: 5px; border-right: 0; border-bottom: 0; border-top-left-radius: 3px; } +.cornerTr { top: 5px; right: 5px; border-left: 0; border-bottom: 0; border-top-right-radius: 3px; } +.cornerBl { bottom: 5px; left: 5px; border-right: 0; border-top: 0; border-bottom-left-radius: 3px; } +.cornerBr { bottom: 5px; right: 5px; border-left: 0; border-top: 0; border-bottom-right-radius: 3px; } + +@media (prefers-reduced-motion: reduce) { + .beam { animation: none; opacity: 0; } + .tag { animation: none; } + .solo { transition: none; } +} diff --git a/src/components/reader/PdfLayoutScan.tsx b/src/components/reader/PdfLayoutScan.tsx new file mode 100644 index 0000000..7f46d00 --- /dev/null +++ b/src/components/reader/PdfLayoutScan.tsx @@ -0,0 +1,172 @@ +'use client'; + +import { useEffect, useState } from 'react'; +import styles from './PdfLayoutScan.module.css'; + +/** + * PdfLayoutScan — ambient visualization of PP-DocLayout-V3 at work. + * + * A miniature document page sits directly on the loader's dotted background + * while a scan beam sweeps across it. One layout region is "detected" at a time + * — a box rendered with one + * of a few simple shapes — while a readout chip above the page names its class. + * It cycles through the complete set of region classes PP-DocLayout-V3 emits + * (see PARSED_PDF_BLOCK_KINDS in types/parsed-pdf). Purely decorative; honours + * prefers-reduced-motion by freezing on a single region. Styles live in the + * adjacent CSS module so they stay out of the global stylesheet. + * + * When `failed` is set the animation is replaced by a static "halted" view — + * the beam stops, the page dims, and an alert glyph sits on it — so the loader + * never implies active work after a parse failure. + */ + +// Only a handful of real shapes — most regions are just text lines. +type BlockShape = 'heading' | 'text' | 'small' | 'image' | 'table'; + +interface ScanBlock { + label: string; + shape: BlockShape; +} + +// One consistent box size per shape, so every region renders the same way and +// nothing gets clipped — only the label changes. +const SHAPE_SIZE: Record = { + heading: { width: 82, height: 22 }, + text: { width: 88, height: 66 }, + small: { width: 72, height: 24 }, + image: { width: 80, height: 68 }, + table: { width: 88, height: 64 }, +}; + +// Every PP-DocLayout-V3 class, in document-flow order, mapped to a simple shape. +const SCAN_BLOCKS: ScanBlock[] = [ + { label: 'Header', shape: 'small' }, + { label: 'Doc title', shape: 'heading' }, + { label: 'Abstract', shape: 'text' }, + { label: 'Paragraph title', shape: 'heading' }, + { label: 'Text', shape: 'text' }, + { label: 'Formula', shape: 'text' }, + { label: 'Formula number', shape: 'small' }, + { label: 'Figure title', shape: 'small' }, + { label: 'Image', shape: 'image' }, + { label: 'Chart', shape: 'image' }, + { label: 'Table', shape: 'table' }, + { label: 'Algorithm', shape: 'text' }, + { label: 'Content', shape: 'text' }, + { label: 'Aside text', shape: 'text' }, + { label: 'Number', shape: 'small' }, + { label: 'Reference', shape: 'small' }, + { label: 'Reference content', shape: 'text' }, + { label: 'Footnote', shape: 'small' }, + { label: 'Vision footnote', shape: 'small' }, + { label: 'Seal', shape: 'image' }, + { label: 'Footer', shape: 'small' }, +]; + +const STEP_MS = 1050; + +const cx = (...parts: Array) => parts.filter(Boolean).join(' '); + +function BlockContent({ shape }: { shape: BlockShape }) { + if (shape === 'image') { + return ( + + + + + ); + } + if (shape === 'table') { + return ( +
+ {Array.from({ length: 16 }).map((_, i) => ( + + ))} +
+ ); + } + const isHeading = shape === 'heading'; + const lines = isHeading ? 2 : shape === 'small' ? 2 : 5; + return ( +
+ {Array.from({ length: lines }).map((_, i) => ( + + ))} +
+ ); +} + +export function PdfLayoutScan({ failed = false }: { failed?: boolean }) { + const [active, setActive] = useState(0); + + useEffect(() => { + if (typeof window === 'undefined' || failed) return; // no cycling once halted + const reduced = window.matchMedia?.('(prefers-reduced-motion: reduce)').matches; + if (reduced) return; // freeze on the first region for reduced-motion users + const id = window.setInterval(() => { + setActive((i) => (i + 1) % SCAN_BLOCKS.length); + }, STEP_MS); + return () => window.clearInterval(id); + }, [failed]); + + if (failed) { + return ( +
+
+ Parse halted +
+ +
+ + + + + + + + + + +
+
+ ); + } + + return ( +
+ {/* readout chip: floats in the stage band above the page so long class + names are never clipped by the page's rounded overflow */} +
+ + {SCAN_BLOCKS[active].label} + +
+ +
+
+ {SCAN_BLOCKS.map((block, i) => { + const size = SHAPE_SIZE[block.shape]; + return ( +
+ +
+ ); + })} +
+ +
+ + + + +
+
+ ); +} diff --git a/src/components/ui/range.module.css b/src/components/ui/range.module.css new file mode 100644 index 0000000..4191c45 --- /dev/null +++ b/src/components/ui/range.module.css @@ -0,0 +1,115 @@ +/* Range slider — "precision gauge": a hairline rail with ruler notches per step + and a slim accent needle instead of a ball. Scoped via CSS Module so it stays + out of the global stylesheet. Per-instance values (--range-progress, + --range-tick-size, --range-tick-color) come from inline styles set in + range.tsx. Used by RangeInput. */ + +.range { + --range-track-h: 2px; + --range-needle-h: 1rem; + --range-needle-w: 3px; + --range-tick-size: 25%; /* segment width; only used when ticks are enabled */ + --range-tick-color: transparent; /* set to a real color inline for discrete sliders */ + --range-fill: var(--muted); + --range-empty: color-mix(in srgb, var(--offbase) 82%, var(--background)); + -webkit-appearance: none; + appearance: none; + width: 100%; + height: 1.25rem; + background: transparent; + cursor: pointer; +} +.range:disabled { + cursor: not-allowed; + opacity: 0.6; +} +.range:focus-visible { + outline: none; +} + +/* Ticks ride above the muted fill / empty rail (first listed background = top). */ +.range::-webkit-slider-runnable-track { + height: var(--range-track-h); + border-radius: 999px; + background: + repeating-linear-gradient( + to right, + var(--range-tick-color) 0, + var(--range-tick-color) 1.5px, + transparent 1.5px, + transparent var(--range-tick-size) + ), + linear-gradient( + to right, + var(--range-fill) 0, + var(--range-fill) var(--range-progress), + var(--range-empty) var(--range-progress), + var(--range-empty) 100% + ); +} +.range::-webkit-slider-thumb { + -webkit-appearance: none; + appearance: none; + width: var(--range-needle-w); + height: var(--range-needle-h); + margin-top: calc((var(--range-track-h) - var(--range-needle-h)) / 2); + border-radius: 999px; + background: var(--accent); + box-shadow: + 0 0 0 1px color-mix(in srgb, var(--background) 70%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); + transition: transform 150ms ease, box-shadow 150ms ease; +} +.range:hover::-webkit-slider-thumb { + transform: scaleY(1.3); +} +.range:active::-webkit-slider-thumb { + transform: scaleY(1.55); +} +.range:focus-visible::-webkit-slider-thumb { + box-shadow: + 0 0 0 4px color-mix(in srgb, var(--accent) 28%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); +} + +/* Firefox: track + native progress fill + matching needle. */ +.range::-moz-range-track { + height: var(--range-track-h); + border-radius: 999px; + background: + repeating-linear-gradient( + to right, + var(--range-tick-color) 0, + var(--range-tick-color) 1.5px, + transparent 1.5px, + transparent var(--range-tick-size) + ), + var(--range-empty); +} +.range::-moz-range-progress { + height: var(--range-track-h); + border-radius: 999px; + background: var(--range-fill); +} +.range::-moz-range-thumb { + width: var(--range-needle-w); + height: var(--range-needle-h); + border: 0; + border-radius: 999px; + background: var(--accent); + box-shadow: + 0 0 0 1px color-mix(in srgb, var(--background) 70%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); + transition: transform 150ms ease, box-shadow 150ms ease; +} +.range:hover::-moz-range-thumb { + transform: scaleY(1.3); +} +.range:active::-moz-range-thumb { + transform: scaleY(1.55); +} +.range:focus-visible::-moz-range-thumb { + box-shadow: + 0 0 0 4px color-mix(in srgb, var(--accent) 28%, transparent), + 0 1px 6px color-mix(in srgb, var(--accent) 55%, transparent); +} diff --git a/src/components/ui/range.tsx b/src/components/ui/range.tsx index 033fadb..e0f1ae0 100644 --- a/src/components/ui/range.tsx +++ b/src/components/ui/range.tsx @@ -1,8 +1,11 @@ import type { CSSProperties, InputHTMLAttributes } from 'react'; import { cn } from './cn'; +import styles from './range.module.css'; type RangeStyle = CSSProperties & { '--range-progress'?: string; + '--range-tick-size'?: string; + '--range-tick-color'?: string; }; function toNumber(value: string | number | readonly string[] | undefined, fallback: number): number { @@ -24,30 +27,16 @@ function resolveRangeProgress(props: InputHTMLAttributes): num return Math.min(100, Math.max(0, progress)); } -const rangeInputClass = cn( - 'h-6 w-full cursor-pointer appearance-none bg-transparent [--range-track-h:0.5rem] [--range-thumb-h:1.25rem]', - 'focus-visible:outline-none focus-visible:[&::-webkit-slider-thumb]:ring-4 focus-visible:[&::-webkit-slider-thumb]:ring-accent/25', - 'focus-visible:[&::-moz-range-thumb]:ring-4 focus-visible:[&::-moz-range-thumb]:ring-accent/25', - 'disabled:cursor-not-allowed disabled:opacity-60', - '[&::-webkit-slider-runnable-track]:h-[var(--range-track-h)] [&::-webkit-slider-runnable-track]:rounded-full', - '[&::-webkit-slider-runnable-track]:bg-[linear-gradient(to_right,var(--secondary-accent)_0%,var(--secondary-accent)_var(--range-progress),color-mix(in_srgb,var(--offbase)_82%,var(--background))_var(--range-progress),color-mix(in_srgb,var(--offbase)_82%,var(--background))_100%)]', - '[&::-webkit-slider-runnable-track]:shadow-[inset_0_1px_0_color-mix(in_srgb,var(--foreground)_10%,transparent)]', - '[&::-webkit-slider-thumb]:mt-[calc((var(--range-track-h)-var(--range-thumb-h))/2)] [&::-webkit-slider-thumb]:h-[var(--range-thumb-h)] [&::-webkit-slider-thumb]:w-[var(--range-thumb-h)] [&::-webkit-slider-thumb]:appearance-none [&::-webkit-slider-thumb]:rounded-full', - '[&::-webkit-slider-thumb]:border-2 [&::-webkit-slider-thumb]:border-[color-mix(in_srgb,var(--background)_78%,white)]', - '[&::-webkit-slider-thumb]:bg-accent', - '[&::-webkit-slider-thumb]:transition-transform [&::-webkit-slider-thumb]:duration-150 [&::-webkit-slider-thumb]:ease-out', - 'active:[&::-webkit-slider-thumb]:scale-[1.07]', - '[&::-moz-range-track]:h-[var(--range-track-h)] [&::-moz-range-track]:rounded-full [&::-moz-range-track]:border-0', - '[&::-moz-range-track]:bg-[color-mix(in_srgb,var(--offbase)_82%,var(--background))]', - '[&::-moz-range-progress]:h-[var(--range-track-h)] [&::-moz-range-progress]:rounded-full [&::-moz-range-progress]:border-0 [&::-moz-range-progress]:bg-secondary-accent', - '[&::-moz-range-thumb]:h-[var(--range-thumb-h)] [&::-moz-range-thumb]:w-[var(--range-thumb-h)] [&::-moz-range-thumb]:appearance-none [&::-moz-range-thumb]:rounded-full [&::-moz-range-thumb]:border-2', - '[&::-moz-range-thumb]:border-[color-mix(in_srgb,var(--background)_78%,white)] [&::-moz-range-thumb]:bg-accent', - '[&::-moz-range-thumb]:transition-transform [&::-moz-range-thumb]:duration-150 [&::-moz-range-thumb]:ease-out', - 'active:[&::-moz-range-thumb]:scale-[1.07]', -); - -function rangeInputClassName(className?: string) { - return cn(rangeInputClass, className); +// Discrete sliders get a ruler: one notch per step. Returns the segment width +// (as a %) for the repeating tick gradient, or null for continuous/large ranges. +function resolveTickSize(props: InputHTMLAttributes): string | null { + const min = toNumber(props.min, 0); + const max = toNumber(props.max, 100); + const step = toNumber(props.step, 1); + if (step <= 0) return null; + const segments = Math.round((max - min) / step); + if (!Number.isFinite(segments) || segments < 2 || segments > 24) return null; + return `${100 / segments}%`; } export function RangeInput({ @@ -55,10 +44,17 @@ export function RangeInput({ style, ...props }: Omit, 'type'>) { + const tickSize = resolveTickSize(props); const rangeStyle: RangeStyle = { ...style, '--range-progress': `${resolveRangeProgress(props)}%`, + ...(tickSize + ? { + '--range-tick-size': tickSize, + '--range-tick-color': 'color-mix(in srgb, var(--foreground) 22%, transparent)', + } + : {}), }; - return ; + return ; } diff --git a/src/components/ui/sidebar-nav.tsx b/src/components/ui/sidebar-nav.tsx index b556d96..9ff0da7 100644 --- a/src/components/ui/sidebar-nav.tsx +++ b/src/components/ui/sidebar-nav.tsx @@ -1,4 +1,5 @@ -import { forwardRef, type AnchorHTMLAttributes, type ButtonHTMLAttributes, type HTMLAttributes, type ReactNode } from 'react'; +import Link from 'next/link'; +import { forwardRef, type ButtonHTMLAttributes, type ComponentPropsWithoutRef, type HTMLAttributes, type ReactNode } from 'react'; import { cn } from './cn'; import { focusRing, motionColors } from './tokens'; @@ -170,7 +171,7 @@ export function SidebarNavItem({ ); } -export const SidebarNavLink = forwardRef & { +type SidebarNavLinkProps = ComponentPropsWithoutRef & { active?: boolean; icon?: ReactNode; label?: ReactNode; @@ -179,7 +180,9 @@ export const SidebarNavLink = forwardRef(function SidebarNavLink({ +}; + +export const SidebarNavLink = forwardRef(function SidebarNavLink({ active = false, icon, label, @@ -193,7 +196,7 @@ export const SidebarNavLink = forwardRef {children} - + ); }); diff --git a/src/lib/server/compute/worker-parse-state.ts b/src/lib/server/compute/worker-parse-state.ts index 0e2b861..cce75a4 100644 --- a/src/lib/server/compute/worker-parse-state.ts +++ b/src/lib/server/compute/worker-parse-state.ts @@ -2,6 +2,10 @@ import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compu import type { PdfParseProgress, PdfParseStatus } from '@/types/parsed-pdf'; import type { DocumentParseState } from '@/lib/server/documents/parse-state'; +function isInflightWorkerStatus(status: WorkerOperationState['status']): boolean { + return status === 'queued' || status === 'running'; +} + export function mapWorkerStatusToParseStatus(status: WorkerOperationState['status']): PdfParseStatus { switch (status) { case 'queued': @@ -44,6 +48,18 @@ export function documentParseStateFromWorkerState( }; } +export function isWorkerOperationStateStale( + state: WorkerOperationState, + staleMs: number, + nowMs = Date.now(), +): boolean { + if (!isInflightWorkerStatus(state.status)) return false; + if (!Number.isFinite(staleMs) || staleMs <= 0) return false; + const updatedAt = Number(state.updatedAt ?? 0); + if (!Number.isFinite(updatedAt) || updatedAt <= 0) return false; + return (nowMs - updatedAt) > staleMs; +} + export function mergeNonReadyParseSnapshot(input: { parseStatus: PdfParseStatus; parseProgress: PdfParseProgress | null; diff --git a/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts b/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts index 135e9c3..449824e 100644 --- a/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts +++ b/tests/unit/pdf-merge-text-with-regions.vitest.spec.ts @@ -34,4 +34,19 @@ describe('mergeTextWithRegions', () => { expect(merged).toHaveLength(1); expect(merged[0].text).toBe('inside'); }); + + test('keeps decorative drop caps attached to the same line when boxes overlap vertically', () => { + const regions = [ + { bbox: [0, 0, 200, 120] as [number, number, number, number], label: 'text' as const }, + ]; + + const textItems = [ + { text: 'I', x: 0, y: 10, width: 12, height: 60 }, + { text: 't’s funny,', x: 12, y: 40, width: 50, height: 12 }, + ]; + + const merged = mergeTextWithRegions(regions, textItems); + expect(merged).toHaveLength(1); + expect(merged[0].text).toBe('It’s funny,'); + }); }); diff --git a/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts b/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts index 3ceccd0..b61bd26 100644 --- a/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts +++ b/tests/unit/pdf-parse-normalize-text-items.vitest.spec.ts @@ -32,7 +32,12 @@ describe('normalizeTextItemsForLayout', () => { [0, 10, -10, 0, 30, 400], ); - const normalized = normalizeTextItemsForLayout([horizontal, rotated], 800); + const normalized = normalizeTextItemsForLayout([horizontal, rotated], { + height: 800, + transform: [1, 0, 0, -1, 0, 800], + }, { + test: { ascent: 0.8, descent: -0.2 }, + }); expect(normalized).toHaveLength(1); expect(normalized[0]?.text).toBe('Powered by large language models'); }); @@ -41,7 +46,49 @@ describe('normalizeTextItemsForLayout', () => { const vertical = makeTextItem('Side label', [0, 10, -10, 0, 30, 200]); const skewed = makeTextItem('Watermark', [10, 5, 2, 10, 200, 500]); - const normalized = normalizeTextItemsForLayout([vertical, skewed], 800); + const normalized = normalizeTextItemsForLayout([vertical, skewed], { + height: 800, + transform: [1, 0, 0, -1, 0, 800], + }, { + test: { ascent: 0.8, descent: -0.2 }, + }); expect(normalized).toEqual([]); }); + + test('accounts for non-zero page origins in the viewport transform', () => { + const croppedPageLine = makeTextItem( + 'Vasher turned away.', + [11.2, 0, 0, 11.2, 127.5, 644.4128], + 100, + ); + + const normalized = normalizeTextItemsForLayout([croppedPageLine], { + height: 666.0074, + transform: [1, 0, 0, -1, -53.4352, 720.565], + }, { + test: { ascent: 0.716, descent: -0.269 }, + }); + + expect(normalized).toHaveLength(1); + expect(normalized[0]?.x).toBeCloseTo(74.0648, 4); + expect(normalized[0]?.y).toBeCloseTo(68.1356, 2); + }); + + test('uses font ascent to place decorative initials closer to the visible glyph top', () => { + const dropCap = makeTextItem( + 'I', + [60, 0, 0, 60, 111.1326, 434.46], + 9.18, + ); + + const normalized = normalizeTextItemsForLayout([dropCap], { + height: 666.0074, + transform: [1, 0, 0, -1, -53.4352, 720.565], + }, { + test: { ascent: 0.638, descent: -0.134 }, + }); + + expect(normalized).toHaveLength(1); + expect(normalized[0]?.y).toBeCloseTo(247.825, 3); + }); }); diff --git a/tests/unit/worker-parse-state.vitest.spec.ts b/tests/unit/worker-parse-state.vitest.spec.ts index 062cf87..dfdd37e 100644 --- a/tests/unit/worker-parse-state.vitest.spec.ts +++ b/tests/unit/worker-parse-state.vitest.spec.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from 'vitest'; import type { PdfLayoutJobResult, WorkerOperationState } from '@openreader/compute-core/api-contracts'; import { documentParseStateFromWorkerState, + isWorkerOperationStateStale, snapshotFromWorkerState, } from '../../src/lib/server/compute/worker-parse-state'; @@ -88,4 +89,30 @@ describe('worker parse state mapping', () => { error: 'layout model crashed', }); }); + + test('treats old inflight worker states as stale', () => { + const workerState = makeWorkerState({ + status: 'running', + updatedAt: 1_000, + progress: { + totalPages: 500, + pagesParsed: 250, + currentPage: 251, + phase: 'infer', + }, + }); + + expect(isWorkerOperationStateStale(workerState, 5_000, 6_001)).toBe(true); + expect(isWorkerOperationStateStale(workerState, 5_000, 5_999)).toBe(false); + }); + + test('never treats terminal worker states as stale', () => { + const failedState = makeWorkerState({ + status: 'failed', + updatedAt: 1_000, + error: { code: 'PDF_PARSE_FAILED', message: 'crashed' }, + }); + + expect(isWorkerOperationStateStale(failedState, 5_000, 99_999)).toBe(false); + }); });