diff --git a/AGENTS.md b/AGENTS.md index c9afaa80..8cf1a6ea 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -666,12 +666,14 @@ cd apps/worker && uv run worker.py # Celery worker ### Debug Scripts (Worker) +PDF/PPT debug is track-split (not a single all-format one-shot): + | Script | Purpose | |:---|:---| -| `debug_parse.py` | Unified parsing debug: all formats, `--stop-at profile/hierarchy/full`, `--run-db` | -| `debug_agentic_e2e.py` | End-to-end agentic retrieval test | -| `debug_profiler.py` | Document profiler testing | -| `debug_toc_detection.py` | TOC detection and hierarchy building | +| `debug_text_track.py` | TEXT-TRACK (`chunk`) staged debug: `--stop-at profile/mineru/hierarchy/full`, `--clean` | +| `page_memory/debug_pm_stage0..5.py` | PAGE-TRACK (`page_memory`) staged debug (bootstrap → finalize); shared `--clean` wipes output dir | +| `debug_retrieval.py` | Retrieval debug | +| `_debug_publish.py` | Optional `--run-db` publish helper for the scripts above | ### Quality Checks diff --git a/apps/worker/app/services/document_agent/calibration/phase1.py b/apps/worker/app/services/document_agent/calibration/phase1.py index 8d8eba25..87ae08bd 100644 --- a/apps/worker/app/services/document_agent/calibration/phase1.py +++ b/apps/worker/app/services/document_agent/calibration/phase1.py @@ -1,10 +1,14 @@ """Deterministic calibration Phase-1: regime partition + forward-scan offsets. Entries are partitioned by printed-label kind (the same classifier Phase-2 -uses). Each regime takes its first few entries as probes and scans forward from -the page after this TOC region's ``toc_range`` end until one is confirmed; that -single confirmation fixes the regime's candidate offset. Phase-2 owns tail -verification and bulk anchoring. +uses). Each regime takes leaf probes from its first few distinct printed pages +and scans forward until one is confirmed; that single confirmation fixes the +regime's candidate offset. Phase-2 owns tail verification and bulk anchoring. + +A probe scans from ``max(toc_range end + 1, printed)``: a printed label never +resolves to a physical page before itself, so the offset a scan can yield is +structurally non-negative. Scanning below the printed page would let a section +divider that repeats the heading confirm ahead of the numbered body page. """ from __future__ import annotations @@ -58,13 +62,17 @@ def _regime_probes( *, limit: int, ) -> dict[str, list[_Probe]]: - """Group entries by printed-label kind, keeping the first ``limit`` per kind.""" + """Keep leaf probes from the first distinct printed pages of each kind.""" from app.services.document_parser.structure.body_boundary import ( normalize_heading_text, ) probes: dict[str, list[_Probe]] = {} - for entry in entries: + seen_printed: dict[str, set[int]] = {} + levels = [int(entry.get("level") or 1) for entry in entries] + for index, entry in enumerate(entries): + if index + 1 < len(entries) and levels[index + 1] > levels[index]: + continue label = entry.get("page_number") kind = classify_page_number_kind(label) if len(probes.get(kind, ())) >= limit: @@ -72,10 +80,13 @@ def _regime_probes( printed = parse_printed_page(label, kind=kind) if printed is None: continue + if printed in seen_printed.get(kind, set()): + continue title = normalize_heading_text(str(entry.get("heading") or "")) if not title: continue probes.setdefault(kind, []).append(_Probe(title=title, printed=printed)) + seen_printed.setdefault(kind, set()).add(printed) return probes @@ -116,11 +127,14 @@ def run_calibration_phase1( failure_kind=FAILURE_TOC_EMPTY, region_index=region_index, ) - scan_start = toc_end + 1 - if scan_start > resolved_page_count: + region_scan_start = toc_end + 1 + if region_scan_start > resolved_page_count: return CalibrationResult( status="failed", - notes=f"scan start {scan_start} beyond page_count {resolved_page_count}", + notes=( + f"scan start {region_scan_start} beyond " + f"page_count {resolved_page_count}" + ), failure_kind=FAILURE_NO_OFFSET, region_index=region_index, ) @@ -136,7 +150,7 @@ def run_calibration_phase1( scan = scan_title_forward( ctx=ctx, title=probe.title, - start_page=scan_start, + start_page=max(region_scan_start, probe.printed), page_count=resolved_page_count, ) scans.append(scan) diff --git a/apps/worker/app/services/document_parser/formats/pptx/parser.py b/apps/worker/app/services/document_parser/formats/pptx/parser.py index d26439c4..e7404983 100755 --- a/apps/worker/app/services/document_parser/formats/pptx/parser.py +++ b/apps/worker/app/services/document_parser/formats/pptx/parser.py @@ -77,7 +77,7 @@ def _get_iloveapi_token_lease(): def pptx_to_pdf_api(pptx_path, outdir="."): """ - use iloveapi to convert pptx to pdf (file-path based, used by debug_parse.py) + use iloveapi to convert pptx to pdf (file-path based) API docs: https://www.iloveapi.com/docs/api-reference """ with open(pptx_path, "rb") as f: diff --git a/apps/worker/scripts/debug_parse.py b/apps/worker/scripts/debug_parse.py deleted file mode 100644 index 005d2c04..00000000 --- a/apps/worker/scripts/debug_parse.py +++ /dev/null @@ -1,712 +0,0 @@ -#!/usr/bin/env python3 -"""Unified production-style document parsing debug script. - -Supports all chunk-track Knowhere formats (PDF, DOCX, XLSX, PPTX, MD, Image, -Fragment) through the same checkerboard parser entry used by the worker. Use -``debug_page_memory.py`` for page-track step debugging. - -Pipeline stages: - 1. checkerboard_parse_output → DataFrame - 2. dataframe_to_chunks → list[ChunkPayload] - 3. ZipResultService → chunks.json / manifest.json / doc_nav.json / *.zip - 4. enrich_doc_nav → summary enrichment + top_summary - 5. DB publication → DocumentSection + DocumentChunk (optional, --run-db) - -Output directory: - default → ~/.knowhere/chengke_kb// - -Usage: - cd apps/worker - - # All formats (full pipeline) - python scripts/debug_parse.py --file /path/to/any.pdf - python scripts/debug_parse.py --file /path/to/doc.docx - python scripts/debug_parse.py --file /path/to/sheet.xlsx - python scripts/debug_parse.py --fragment "粘贴的文本..." - - # Options - python scripts/debug_parse.py --spacex --run-db # Enable DB publication -""" - -from __future__ import annotations - -import argparse -import json -import os -import shutil -import sys -import time -import zipfile -from pathlib import Path -from typing import Any - -# ── Bootstrap: path + env ────────────────────────────────────────────────────── -ROOT = Path(__file__).resolve().parents[3] -WORKER_ROOT = ROOT / "apps" / "worker" -sys.path.insert(0, str(WORKER_ROOT)) -sys.path.insert(0, str(ROOT / "packages" / "shared-python")) - -from dotenv import load_dotenv # noqa: E402 - -load_dotenv(WORKER_ROOT / ".env") -os.environ.setdefault("LOCAL_DEBUG", "1") -os.environ.setdefault("OVERSIZED_PDF_SHARD_ENABLED", "true") - -from loguru import logger # noqa: E402 - -from shared.core.config import settings # noqa: E402 - -# ── Constants ────────────────────────────────────────────────────────────────── -DEFAULT_SPACEX_PDF = Path("/Users/wuchengke/Desktop/temp/test_docs/spacex-s1.pdf") -DEFAULT_SJSYJ_PDF = Path( - "/Users/wuchengke/Desktop/temp/test_docs/" - "SJSYJ-SC-2024 企业制度汇编（上册）.pdf" -) - -PRODUCTION_OUTPUT_ROOT = Path("~/.knowhere/chengke_kb").expanduser() -PROFILE_TRANSIENT_DIRS = ( - "_doc_agent", - "coarse_profile_pages", - "calibration_scan", - "calibration_verify", - "toc_pages", - "ocr_pages", - "inspect_pages", - "profile_visuals", - # Legacy dirs from older runs. - "agent_visuals", - "planner_pages", - "page_locate_pages", - "verify_pages", - "calibration_inspect", -) - -# ══════════════════════════════════════════════════════════════════════════════ -# Section A: DB Publication (Stage 10) — preserved from original debug_parse.py -# ══════════════════════════════════════════════════════════════════════════════ - -def _run_db_publication( - chunks: list, - add_dir: str, - source_file_name: str, -): - """Stage 10: publish an already finalized debug result to local DB/S3.""" - from scripts._debug_publish import publish_debug_result_dir - - result = publish_debug_result_dir( - result_dir=add_dir, - source_file_name=source_file_name, - chunks=chunks, - upload_assets=True, - ) - logger.info(" ✅ DB transaction committed (job_id={})", result.job_id) - - -# ══════════════════════════════════════════════════════════════════════════════ -# Section C: Common Post-Parse Pipeline (Stage 7-10) -# ══════════════════════════════════════════════════════════════════════════════ - -def _finalize_output( - parsed_df, - add_dir: str, - source_file_name: str, - *, - run_db: bool = False, - job_metadata: dict[str, Any] | None = None, -) -> list[dict[str, Any]]: - """Stage 7-10: chunks → ZIP → enrich → optional DB. - - Mirrors production flow: - - parse_result_package.py L49 → dataframe_to_chunks - - success_finalization.py L196 → ZipResultService - - success_finalization.py L126 → enrich_doc_nav_summaries - - debug_parse.py _run_db_publication → DB write - - Returns the chunks list. - """ - from shared.services.chunks.dataframe_chunk_converter import dataframe_to_chunks - from shared.services.storage.zip_result_service import ZipResultService - from app.services.connect_builder.summary_builder import ( - build_section_summary_lookup, - enrich_doc_nav_summaries, - ensure_doc_nav_json, - load_nav_top_summary, - ) - - timings: dict[str, float] = {} - - # ── Stage 7: DataFrame → chunks (mirrors parse_result_package.py L49) ── - logger.info("=" * 60) - logger.info("📦 Stage 7: dataframe_to_chunks") - logger.info("=" * 60) - - t0 = time.time() - chunks = dataframe_to_chunks(parsed_df) - timings["Stage 7: chunks"] = time.time() - t0 - - text_count = sum(1 for c in chunks if c.get("type") == "text") - image_count = sum(1 for c in chunks if c.get("type") == "image") - table_count = sum(1 for c in chunks if c.get("type") == "table") - page_count = sum(1 for c in chunks if c.get("type") == "page") - table_ref_count = sum( - 1 - for c in chunks - if c.get("type") == "table" - and str(c.get("content") or "").strip().startswith("tables/") - ) - table_inline_html_count = sum( - 1 - for c in chunks - if c.get("type") == "table" - and " 0: - logger.info("") - logger.info("═" * 58) - logger.info(" 📊 POST-PARSE TIMELINE") - logger.info("═" * 58) - for phase, elapsed in timings.items(): - pct = elapsed / t_total * 100 - logger.info(f" {phase:<35s} │ {elapsed:>7.2f}s ({pct:>5.1f}%)") - logger.info(" " + "─" * 55) - logger.info(f" {'TOTAL':<35s} │ {t_total:>7.2f}s (100.0%)") - logger.info("═" * 58) - - return chunks - - -def _cleanup_agent_transient_dirs(add_dir: str) -> None: - """Remove VLM render caches before packaging debug output.""" - removed: list[str] = [] - for dirname in PROFILE_TRANSIENT_DIRS: - path = os.path.join(add_dir, dirname) - if os.path.isdir(path): - shutil.rmtree(path) - removed.append(dirname) - nested_doc_agent = os.path.join(add_dir, "_doc_agent") - if os.path.isdir(nested_doc_agent): - for dirname in PROFILE_TRANSIENT_DIRS: - path = os.path.join(nested_doc_agent, dirname) - if os.path.isdir(path): - shutil.rmtree(path) - removed.append(f"_doc_agent/{dirname}") - if not os.listdir(nested_doc_agent): - os.rmdir(nested_doc_agent) - removed.append("_doc_agent") - if removed: - logger.info(f" Cleaned transient agent dirs: {', '.join(removed)}") - - -# ══════════════════════════════════════════════════════════════════════════════ -# Section D: Pipeline Entry Points -# ══════════════════════════════════════════════════════════════════════════════ - -def _run_standard_pipeline( - file_path: str, - source_file_name: str, - output_root: str, - *, - run_db: bool = False, - fragment_content: str = "", -) -> dict[str, Any]: - """Standard pipeline for all formats: checkerboard_parse_output → finalize. - - Uses the production black-box entry point. Handles all formats including - oversized PDFs (which are routed internally by parse_pdfs). - - Token/time tracking mirrors production parse_execution.py exactly: - init trackers → parse → collect stats → cleanup. - """ - from app.services.document_parser.parse_service import checkerboard_parse_output - from app.services.document_parser.support.stage_profiler import ( - init_stage_tracker, - cleanup_stage_tracker, - get_current_stage_tracker, - ) - from shared.services.ai.token_tracking import ( - init_token_tracker, - cleanup_token_tracker, - get_current_token_tracker, - ) - - filename = source_file_name - is_fragment = ".fragment" in file_path.lower() - - logger.info("=" * 60) - logger.info(f"📄 Standard pipeline: {filename}") - logger.info(f" Output root: {output_root}") - logger.info("=" * 60) - - # ── Init trackers (same as parse_execution.py; reuse run_pipeline tracker) ── - token_usage_dict = get_current_token_tracker() - owns_token_tracker = token_usage_dict is None - if token_usage_dict is None: - token_usage_dict = init_token_tracker() - - stage_timing_dict = get_current_stage_tracker() - owns_stage_tracker = stage_timing_dict is None - if stage_timing_dict is None: - stage_timing_dict = init_stage_tracker() - - try: - t0 = time.time() - result = checkerboard_parse_output( - file_full_path=file_path, - filename=filename, - output_dir=output_root, - internal_output_filename=filename, - smart_title_parse=True, - summary_image=True, - summary_table=True, - summary_txt=True, - doc_type="auto", - fragment_content=fragment_content if is_fragment else "", - ) - parse_elapsed = time.time() - t0 - - # ── Snapshot stats before cleanup ── - stages_snapshot = { - "timing_ms": dict(stage_timing_dict), - "token_usage": dict(token_usage_dict), - } - finally: - if owns_token_tracker: - cleanup_token_tracker() - if owns_stage_tracker: - cleanup_stage_tracker() - - add_dir = result.output_dir - parsed_df = result.parsed_df - - logger.info("=" * 60) - logger.info(f"✅ Parse complete in {parse_elapsed:.1f}s") - logger.info(f" Output path: {add_dir}") - if parsed_df is not None: - logger.info(f" DataFrame rows: {len(parsed_df)}") - logger.info("=" * 60) - - # ── Print consumption stats ── - logger.info("") - logger.info("═" * 58) - logger.info(" 📊 CONSUMPTION STATS (mirrors manifest.processing.stages)") - logger.info("═" * 58) - token_usage = stages_snapshot["token_usage"] - logger.info( - f" Token usage: prompt={token_usage['prompt_tokens']}, " - f"completion={token_usage['completion_tokens']}, " - f"total={token_usage['total_tokens']}" - ) - timing_ms = stages_snapshot["timing_ms"] - if timing_ms: - logger.info(" Stage timings:") - for stage, ms in sorted(timing_ms.items()): - logger.info(f" {stage:<45s} │ {ms:>8,}ms") - else: - logger.info(" Stage timings: (none recorded)") - logger.info("═" * 58) - - if not add_dir or not os.path.exists(add_dir) or parsed_df is None or parsed_df.empty: - logger.error("❌ Parse returned empty result, cannot proceed") - return {"status": "error", "parse_elapsed": parse_elapsed} - - # Build job_metadata matching production parse_execution.py + success_finalization.py - from datetime import datetime, timezone - - processing_completed_at = datetime.now(timezone.utc) - debug_job_metadata: dict[str, Any] = { - "stages": stages_snapshot, - "processing_started_at": processing_completed_at.isoformat(), - "processing_completed_at": processing_completed_at.isoformat(), - "processing_duration_ms": int(parse_elapsed * 1000), - } - - try: - # Stage 7-10 - chunks = _finalize_output( - parsed_df, add_dir, source_file_name, - run_db=run_db, - job_metadata=debug_job_metadata, - ) - finally: - debug_job_metadata["stages"] = { - "timing_ms": dict(stage_timing_dict), - "token_usage": dict(token_usage_dict), - } - - return { - "status": "success", - "parse_elapsed": round(parse_elapsed, 1), - "output_dir": add_dir, - "chunks_count": len(chunks) if chunks else 0, - "stages": stages_snapshot, - } - - - -def run_pipeline( - file_path: str, - source_file_name: str, - *, - run_db: bool = False, - fragment_content: str = "", - output_root_override: str | None = None, -) -> dict[str, Any]: - """Unified production-style E2E parser entry point.""" - from app.services.document_parser.support.stage_profiler import ( - cleanup_stage_tracker, - get_current_stage_tracker, - init_stage_tracker, - ) - from shared.services.ai.token_tracking import ( - cleanup_token_tracker, - get_current_token_tracker, - init_token_tracker, - ) - - owns_token_tracker = get_current_token_tracker() is None - if owns_token_tracker: - init_token_tracker() - - owns_stage_tracker = get_current_stage_tracker() is None - if owns_stage_tracker: - init_stage_tracker() - - try: - if output_root_override: - output_root = output_root_override - else: - output_root = str(PRODUCTION_OUTPUT_ROOT) - - return _run_standard_pipeline( - file_path, - source_file_name, - output_root, - run_db=run_db, - fragment_content=fragment_content, - ) - finally: - if owns_token_tracker: - cleanup_token_tracker() - if owns_stage_tracker: - cleanup_stage_tracker() - - -# ══════════════════════════════════════════════════════════════════════════════ -# Section E: CLI -# ══════════════════════════════════════════════════════════════════════════════ - -def test_config(): - """Print current configuration.""" - logger.info("=== Current Config ===") - logger.info(f"ENVIRONMENT: {getattr(settings, 'ENVIRONMENT', 'N/A')}") - logger.info( - f"DATABASE_URL: {getattr(settings, 'DATABASE_URL', 'N/A')[:50]}..." - ) - logger.info(f"REDIS_HOST: {getattr(settings, 'REDIS_HOST', 'N/A')}") - logger.info( - f"DS_KEY: {'set' if getattr(settings, 'DS_KEY', None) else 'unset'}" - ) - logger.info( - f"ALI_API_KEYS: {'set' if getattr(settings, 'ALI_API_KEYS', None) else 'unset'}" - ) - logger.info(f"IMAGE_MODEL: {getattr(settings, 'IMAGE_MODEL', 'N/A')}") - logger.info(f"NORMOL_MODEL: {getattr(settings, 'NORMOL_MODEL', 'N/A')}") - logger.info( - f"HIERARCHY_LLM_MODEL: {getattr(settings, 'HIERARCHY_LLM_MODEL', 'N/A')}" - ) - logger.info( - f"MAX_PDF_PAGE_LIMIT: {getattr(settings, 'MAX_PDF_PAGE_LIMIT', 'N/A')}" - ) - - -def _parse_cases(args: argparse.Namespace) -> list[tuple[str, str, str]]: - """Parse CLI args into [(name, file_path, fragment_content), ...]. - - Returns a list of tuples: (job_name, file_path, fragment_content). - For non-fragment cases, fragment_content is "". - """ - cases: list[tuple[str, str, str]] = [] - - for raw_case in args.case or []: - if "=" not in raw_case: - raise ValueError("--case must use name=/path/to/file") - name, path = raw_case.split("=", 1) - resolved = str(Path(path).expanduser().resolve()) - cases.append((name.strip(), resolved, "")) - - if args.file: - file_path = str(Path(args.file).expanduser().resolve()) - job_id = args.job_id or Path(args.file).stem - cases.append((job_id, file_path, "")) - - if args.fragment: - cases.append(("fragment", ".fragment", args.fragment)) - - if args.spacex: - cases.append( - ("spacex-s1", str(DEFAULT_SPACEX_PDF.expanduser().resolve()), "") - ) - if args.sjsyj: - cases.append( - ("sjsyj", str(DEFAULT_SJSYJ_PDF.expanduser().resolve()), "") - ) - - return cases - - -def main() -> int: - parser = argparse.ArgumentParser( - description="Unified document parsing debug script — supports all Knowhere formats.", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog="""examples: - # Full pipeline (any format) - python scripts/debug_parse.py --file /path/to/doc.pdf - python scripts/debug_parse.py --file /path/to/doc.docx - python scripts/debug_parse.py --fragment "粘贴的文本..." - - # Optional DB publication - python scripts/debug_parse.py --spacex --run-db -""", - ) - - # Input sources - input_group = parser.add_argument_group("input") - input_group.add_argument("--file", help="Path to file to parse (any format)") - input_group.add_argument("--job-id", help="Job id override for --file") - input_group.add_argument( - "--fragment", help="Text content for fragment mode parsing" - ) - input_group.add_argument( - "--spacex", - action="store_true", - help=f"SpaceX S-1 fixture: {DEFAULT_SPACEX_PDF}", - ) - input_group.add_argument( - "--sjsyj", - action="store_true", - help=f"企业制度汇编 fixture: {DEFAULT_SJSYJ_PDF}", - ) - input_group.add_argument( - "--case", - action="append", - help="Named fixture: name=/path/to/file", - ) - - # Post-processing - post_group = parser.add_argument_group("post-processing") - post_group.add_argument( - "--run-db", - action="store_true", - help="Enable Stage 10: DB publication (requires running database)", - ) - - # Output control - output_group = parser.add_argument_group("output") - output_group.add_argument( - "--output-root", - default=None, - help=( - "Override output root directory. Default: " - f"{PRODUCTION_OUTPUT_ROOT}" - ), - ) - output_group.add_argument( - "--clean", - action="store_true", - help="Delete existing output before running", - ) - output_group.add_argument( - "--test-config", - action="store_true", - help="Print current configuration and exit", - ) - - args = parser.parse_args() - - if args.test_config: - test_config() - return 0 - - cases = _parse_cases(args) - if not cases: - parser.error( - "provide --file, --fragment, --spacex, --sjsyj, or at least one --case" - ) - - summaries = [] - - for name, file_path, fragment_content in cases: - if file_path != ".fragment" and not os.path.exists(file_path): - raise FileNotFoundError(f"File not found: {file_path}") - - source_file_name = ( - os.path.basename(file_path) if file_path != ".fragment" else "" - ) - - # Clean if requested - if args.clean: - from app.services.document_parser.orchestration.path_segment import ( - build_parser_path_segment, - ) - dir_name = build_parser_path_segment(source_file_name) - clean_root = Path(args.output_root) if args.output_root else PRODUCTION_OUTPUT_ROOT - clean_dir = clean_root / dir_name - if clean_dir.exists(): - logger.info(f"🗑️ Cleaning {clean_dir}") - shutil.rmtree(clean_dir) - - logger.info("") - logger.info("█" * 60) - logger.info(f" CASE: {name}") - logger.info(f" FILE: {file_path}") - logger.info("█" * 60) - - result = run_pipeline( - file_path, - source_file_name, - run_db=args.run_db, - fragment_content=fragment_content, - output_root_override=args.output_root, - ) - result["job_id"] = name - summaries.append(result) - - # Final JSON summary - print(json.dumps({"cases": summaries}, ensure_ascii=False, indent=2)) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/apps/worker/scripts/debug_retrieval.py b/apps/worker/scripts/debug_retrieval.py index 28bf7555..e9ac7438 100644 --- a/apps/worker/scripts/debug_retrieval.py +++ b/apps/worker/scripts/debug_retrieval.py @@ -69,47 +69,6 @@ def _print_box(title: str, body: Any) -> None: print(line) -def _summarize_tool_payload(payload: dict[str, Any] | None) -> str: - """Keep tool logs readable; full LLM prompts/responses are printed separately.""" - if not isinstance(payload, dict): - return str(payload) - - lines: list[str] = [] - if 'top_doc_ids' in payload: - lines.append(f'top_doc_ids={payload.get("top_doc_ids")}') - if 'channel_counts' in payload: - lines.append(f'channel_counts={payload.get("channel_counts")}') - if 'fused_rows' in payload: - rows = payload.get('fused_rows') or [] - lines.append(f'fused_rows={len(rows)}') - for row in rows[:5]: - lines.append( - f' - score={row.get("score", 0):.4f} ' - f'doc={str(row.get("document_id", ""))[:12]} ' - f'path="{row.get("section_path") or row.get("source_chunk_path")}" ' - f'chunk={row.get("chunk_id")}' - ) - if 'candidate_docs' in payload: - docs = payload.get('candidate_docs') or [] - lines.append(f'candidate_docs={len(docs)}') - for doc in docs: - lines.append( - f' - doc={doc.get("document_id")} ' - f'name="{doc.get("source_file_name", "")}" ' - f'confidence={doc.get("confidence")}' - ) - if 'document_id' in payload: - lines.append(f'document_id={payload.get("document_id")}') - if 'has_outline' in payload: - lines.append(f'has_outline={payload.get("has_outline")} ' - f'leaf_count={payload.get("leaf_count", 0)} ' - f'children_count={payload.get("children_count", 0)}') - for key in ('reason', 'document_id', 'raw_response', 'overflowed'): - if key in payload: - lines.append(f'{key}={payload.get(key)}') - return '\n'.join(lines) or str(payload) - - def _decision_stage(kind: str) -> tuple[str, str]: return { 'kg_document_select': ( diff --git a/apps/worker/scripts/page_memory/_debug_pm_shared.py b/apps/worker/scripts/page_memory/_debug_pm_shared.py index 88aa9e62..13270c72 100644 --- a/apps/worker/scripts/page_memory/_debug_pm_shared.py +++ b/apps/worker/scripts/page_memory/_debug_pm_shared.py @@ -2,7 +2,7 @@ # ruff: noqa: E402, F401 """Shared utilities for the staged page-memory debug scripts. -All stage scripts (debug_pm_stage0..6) import from here instead of +All stage scripts (debug_pm_stage0..5) import from here instead of duplicating bootstrap, artifact I/O, and argparse helpers. """ @@ -191,6 +191,11 @@ def base_argparser(description: str) -> argparse.ArgumentParser: parser.add_argument("--model", default=None, help="Override hierarchy/profiler model") parser.add_argument("--vlm-model", default=None, help="VLM model override") parser.add_argument("--no-vlm", action="store_true", help="Disable VLM calls") + parser.add_argument( + "--clean", + action="store_true", + help="Delete the page_memory output dir before running (full retest wipe)", + ) parser.add_argument( "--out-suffix", default="", @@ -217,58 +222,23 @@ def resolve_paths(args: argparse.Namespace) -> tuple[str, str, Path]: safe = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in suffix) dir_name = f"{dir_name}__{safe}" out_dir = OUTPUT_ROOT / dir_name / "page_memory" + if bool(getattr(args, "clean", False)) and out_dir.exists(): + import shutil + + logger.info("🗑️ Cleaning {}", out_dir) + shutil.rmtree(out_dir) out_dir.mkdir(parents=True, exist_ok=True) return pdf_path, filename, out_dir -# ── ToolContext builder ─────────────────────────────────────────────────────── - - -def build_ctx( - *, pdf_path: str, job_id: str, out_dir: Path, - page_count: int, page_texts: dict[int, str], vlm_model: str | None, - asset_extraction_enabled: bool = False, -): - from app.services.document_agent.manifest import ToolContext - from app.services.document_agent.state import ProfileBlackboard - - blackboard = ProfileBlackboard() - blackboard.page_count = page_count - blackboard.page_full_text_cache = dict(page_texts) - - vmodel = vlm_model or os.environ.get("IMAGE_MODEL") - reason_model = os.environ.get("PAGE_LOCATE_REASON_MODEL") or os.environ.get("NORMOL_MODEL") - - return ToolContext( - pdf_path=pdf_path, - job_id=job_id, - blackboard=blackboard, - trace=None, - output_dir=str(out_dir / "_doc_agent"), - settings={ - "vlm_model": vmodel, - "model": reason_model, - "profile_png_dpi": os.environ.get("AGENT_PNG_DPI", "144"), - }, - ) - - # ── Anatomy / doc-profile cache ────────────────────────────────────────────── def resolve_anatomy_cache_path(out_dir: Path) -> Path: - """Prefer package-root ``doc_profile.json``; fall back to legacy paths.""" + """Canonical package-root ``doc_profile.json`` written by Stage 0/1.""" from app.services.document_agent.persist import DOC_PROFILE_FILENAME - candidates = ( - out_dir / DOC_PROFILE_FILENAME, - out_dir / "_doc_agent" / DOC_PROFILE_FILENAME, - out_dir / "_doc_agent" / "anatomy_map.json", - ) - for path in candidates: - if path.is_file(): - return path - return candidates[0] + return out_dir / DOC_PROFILE_FILENAME def load_anatomy_cache(cache_path: Path, pdf_path: str, job_id: str): @@ -392,88 +362,7 @@ def load_anatomy_cache(cache_path: Path, pdf_path: str, job_id: str): ) -# ── Profile ─────────────────────────────────────────────────────────────────── - - -def run_profile( - pdf_path: str, - job_id: str, - out_dir: Path, - model: str | None, - *, - skip_toc_anchoring: bool = False, -): - """Run page-memory profile exactly like production ``memory_service.run``. - - Uses ``profile_document(..., skip_shard_plan=True, oversized_policy="page_memory")`` - so coarse → anatomy matches the live track (no LLM shard planning). - - ``skip_toc_anchoring=True`` stops after TOC extract (legacy monolithic - helper). Prefer staged debug: Stage-0 bootstrap then Stage-1 TOC. - """ - from app.services.document_parser.profiling.doc_profiler import profile_document - from shared.core.config import settings - - logger.info("=" * 70) - logger.info(f"🧬 DOC_PROFILE (page_memory, monolithic) — {job_id}") - logger.info("=" * 70) - if skip_toc_anchoring: - logger.info(" skip_toc_anchoring=True (TOC extract only; no calibration)") - - previous_image_model = settings.IMAGE_MODEL - if model: - settings.IMAGE_MODEL = model - logger.info(f" IMAGE_MODEL override → {model}") - - t0 = time.time() - try: - profile = profile_document( - pdf_path, - job_id, - job_id=job_id, - output_dir=str(out_dir), - skip_shard_plan=True, - oversized_policy="page_memory", - skip_toc_anchoring=skip_toc_anchoring, - ) - finally: - if model: - settings.IMAGE_MODEL = previous_image_model - - logger.info(f" profile done in {time.time() - t0:.1f}s") - logger.info( - " category={} routing={} page_count={} is_atlas={}", - profile.category, - getattr(profile.routing_category, "value", profile.routing_category), - profile.page_count, - profile.is_atlas, - ) - - anatomy = profile.anatomy - if anatomy is None: - raise RuntimeError( - "page_memory profile returned no anatomy " - f"(routing={profile.routing_category}). Atlas / no-anatomy path " - "cannot continue Stage 1." - ) - - from app.services.document_agent.persist import DOC_PROFILE_FILENAME - - profile_path = out_dir / DOC_PROFILE_FILENAME - if not profile_path.exists(): - write_debug_json(profile_path, anatomy.to_dict()) - - asset_pages = sum(1 for f in anatomy.page_features if getattr(f, "has_asset", False)) - logger.info(f" page_count={anatomy.page_count}") - logger.info(f" toc_pages={anatomy.toc_result.toc_pages}") - logger.info(f" has_asset_pages={asset_pages}/{anatomy.page_count}") - logger.info( - " shard_plan.enabled={} shards={}", - anatomy.shard_plan.enabled, - len(anatomy.shard_plan.shards), - ) - logger.info(f" doc_profile → {profile_path}") - return anatomy +# ── Profile helpers (staged Stage-0 / Stage-1) ───────────────────────────────── def _build_debug_coordinator( @@ -912,48 +801,22 @@ def page_text_cache_path(out_dir: Path) -> Path: def load_pipeline_state( state_path: Path, - *, - legacy_locate_cache: Path | None = None, ) -> dict[str, Any]: - """Load the shared Stage 0-6 ledger, with locate-cache compatibility.""" - if state_path.exists(): - data = json.loads(state_path.read_text(encoding="utf-8")) - if not isinstance(data, dict): - raise ValueError(f"pipeline state must be an object: {state_path}") - data.setdefault("version", PIPELINE_STATE_VERSION) - data.setdefault("stages", {}) - return data - - if legacy_locate_cache is not None and legacy_locate_cache.exists(): - rows = json.loads(legacy_locate_cache.read_text(encoding="utf-8")) - if not isinstance(rows, list): - raise ValueError( - f"legacy locate cache must be a list: {legacy_locate_cache}" - ) - logger.warning( - "Legacy locate cache detected; it will be migrated on the next stage write: {}", - legacy_locate_cache, - ) - return { - "version": PIPELINE_STATE_VERSION, - "stages": { - "stage2": { - "status": "legacy", - "skeletons": rows, - } - }, - } - - raise FileNotFoundError(state_path) + """Load the shared Stage 0-6 ledger from ``pipeline_state.json``.""" + if not state_path.exists(): + raise FileNotFoundError(state_path) + data = json.loads(state_path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError(f"pipeline state must be an object: {state_path}") + data.setdefault("version", PIPELINE_STATE_VERSION) + data.setdefault("stages", {}) + return data def _pipeline_skeleton_rows(state: dict[str, Any]) -> list[dict[str, Any]]: stages = state.get("stages") stage2 = stages.get("stage2") if isinstance(stages, dict) else None rows = stage2.get("skeletons") if isinstance(stage2, dict) else None - if rows is None: - # Compatibility with the short-lived ``stage2_state.json`` proposal. - rows = state.get("skeletons") if not isinstance(rows, list): raise ValueError("pipeline state is missing stages.stage2.skeletons[]") return [row for row in rows if isinstance(row, dict)] @@ -961,15 +824,10 @@ def _pipeline_skeleton_rows(state: dict[str, Any]) -> list[dict[str, Any]]: def load_pipeline_skeletons( state_path: Path, - *, - legacy_locate_cache: Path | None = None, ) -> list[Any]: from app.services.page_memory.skeleton_extractor import SectionSkeleton - state = load_pipeline_state( - state_path, - legacy_locate_cache=legacy_locate_cache, - ) + state = load_pipeline_state(state_path) return [ SectionSkeleton( section_path=str(row["section_path"]), @@ -1035,53 +893,6 @@ def update_pipeline_state( return state -def remove_legacy_doc_agent_artifacts( - doc_agent_dir: Path, - *, - include_stage2: bool = False, - keep_resume_cache: bool = True, -) -> None: - """Drop nested doc-agent clutter; keep resume + pipeline history by default. - - Canonical package artifacts live at ``page_memory/`` root - (``doc_profile.json``, ``trace.json``). Nested ``anatomy_map.json`` and - calibration page PNGs are duplicates / inspect leftovers. - """ - import shutil - - names = { - "parser_profile.json", - "toc_hierarchies.json", - "anatomy_map.json", - "trace.json", - "doc_profile.json", - } - if include_stage2: - names.update( - { - "calibration_result.json", - "null_page_parent_locate.json", - "locate_cache.json", - "stage2_state.json", - } - ) - if not keep_resume_cache: - names.update( - { - STAGE0_STATE_NAME, - PAGE_TEXT_CACHE_NAME, - "stage_costs.json", - } - ) - for name in names: - (doc_agent_dir / name).unlink(missing_ok=True) - for dirname in ("coarse_assets", "calibration_inspect"): - legacy_dir = doc_agent_dir / dirname - if legacy_dir.is_dir(): - shutil.rmtree(legacy_dir) - (doc_agent_dir / "coarse_assets.html").unlink(missing_ok=True) - - def record_stage( stages: list[dict[str, Any]], stage: str, @@ -1101,7 +912,7 @@ def record_stage( STAGE_COSTS_VERSION = "1.0" STAGE_COSTS_NAME = "stage_costs.json" -_COST_STAGE_KEYS = tuple(f"stage{number}" for number in range(0, 7)) +_COST_STAGE_KEYS = tuple(f"stage{number}" for number in range(0, 6)) def stage_costs_path(out_dir: Path) -> Path: @@ -1414,7 +1225,6 @@ def stop_with_trace( summary=summary, ) remove_nested_doc_agent_trace(out_dir) - remove_legacy_doc_agent_artifacts(out_dir / "_doc_agent", include_stage2=True) maybe_purge_debug_visuals(out_dir) return 0 @@ -1475,40 +1285,6 @@ def write_top_level_artifacts( (out_dir / "assets.json").unlink(missing_ok=True) -def cleanup_page_memory_artifacts(out_dir: Path) -> None: - stale_files = { - "assets.json", - "chunks.json", - "coarse_scopes.json", - "doc_nav.json", - "hierarchy.json", - "manifest.json", - "node_rows.csv", - "node_rows.json", - "page_plans.json", - "page_rendered.json", - "page_tags.json", - "report.md", - "trace.json", - } - for name in stale_files: - path = out_dir / name - try: - if path.is_file(): - path.unlink() - except Exception: - logger.debug(f"cleanup failed for {path}") - for name in ("asset_annotate", "debug", "images", "pages", "scopes", "tables"): - path = out_dir / name - try: - if path.is_dir(): - import shutil - - shutil.rmtree(path) - except Exception: - logger.debug(f"cleanup failed for {path}") - - # ── Tree helpers ────────────────────────────────────────────────────────────── @@ -1520,21 +1296,6 @@ def walk(nodes: list, depth: int = 0) -> list[tuple[int, Any]]: return rows -def walk_node_count(nodes: list) -> int: - return len(walk(nodes)) - - -def hierarchy_metrics(nodes: list, *, source: str) -> dict[str, Any]: - rows = walk(nodes) - depths = [depth + 1 for depth, _node in rows] - return { - "hierarchy_source": source, - "title_node_count": len(rows), - "title_leaf_count": sum(1 for _depth, node in rows if not node.children), - "title_max_depth": max(depths) if depths else 0, - } - - # ── Artifact loaders ────────────────────────────────────────────────────────── @@ -1569,12 +1330,6 @@ def load_hierarchy_artifact(path: Path) -> tuple[dict[str, Any], list[Any]]: return dict(scope) if isinstance(scope, dict) else {}, sort_skeletons(skeletons) -def load_skeletons_from_hierarchy_artifact(path: Path) -> list[Any]: - """Compatibility reader for callers that only need hierarchy nodes.""" - _scope, skeletons = load_hierarchy_artifact(path) - return skeletons - - def load_page_tags_artifact(path: Path) -> list[Any]: """Load ``page_tags.json`` written by ``serialize_page_tags``.""" from app.services.page_memory.page_tagger import PageTagResult @@ -1724,24 +1479,6 @@ def _scope_meta_from_dir(scope_dir: Path) -> dict[str, Any]: } -def load_locate_cache(locate_cache: Path) -> list[Any]: - from app.services.page_memory.skeleton_extractor import SectionSkeleton - - raw = json.loads(locate_cache.read_text(encoding="utf-8")) - return [ - SectionSkeleton( - section_path=r["section_path"], - title=r["title"], - level=r["level"], - start_page=r["start_page"], - end_page=r["end_page"], - parent_path=r.get("parent_path"), - evidence=r.get("evidence", {}), - ) - for r in raw - ] - - def _serialize_skeletons(skeletons: list[Any]) -> list[dict[str, Any]]: return [ { @@ -1797,7 +1534,7 @@ def build_debug_coarse_scopes( def add_scope_selection_args(parser: argparse.ArgumentParser) -> None: - """Flags shared by stage 4/5/6 for picking one or more coarse scopes.""" + """Flags shared by stage 3/4/5 for picking one or more coarse scopes.""" parser.add_argument( "--scope-id", default=None, @@ -1806,12 +1543,12 @@ def add_scope_selection_args(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--all-scopes", action="store_true", - help="Process every scope under scopes/ (default when no selector is set)", + help="Process every coarse scope (default when no selector is set)", ) parser.add_argument( "--page-range", default=None, - help="Select scope(s) overlapping this page range (e.g. 14-23 or 225)", + help="Page-range selector (e.g. 14-23 or 225)", ) parser.add_argument( "--fat-only", @@ -1900,7 +1637,7 @@ def resolve_debug_scope_ids( logger.error("❌ No scope directories with {} found under {}", require_file, scopes_dir) logger.error( " Run Stage 3 first: uv run python scripts/page_memory/" - "debug_pm_stage3_coarse_scope.py --file ..." + "debug_pm_stage3_scope_fine_hierarchy.py --file ..." ) raise SystemExit(1) diff --git a/apps/worker/scripts/page_memory/debug_pm_stage1_hierarchy.py b/apps/worker/scripts/page_memory/debug_pm_stage1_hierarchy.py index 14e9e3f3..c30c95a8 100644 --- a/apps/worker/scripts/page_memory/debug_pm_stage1_hierarchy.py +++ b/apps/worker/scripts/page_memory/debug_pm_stage1_hierarchy.py @@ -33,7 +33,6 @@ load_anatomy_cache, resolve_anatomy_cache_path, record_stage, - remove_legacy_doc_agent_artifacts, require_file, resolve_paths, run_stage1_toc, @@ -87,7 +86,6 @@ def main() -> int: trace_stages: list[dict] = [] token_cost_tracker = TokenCostTracker() - doc_agent_dir = out_dir / "_doc_agent" anatomy_cache = resolve_anatomy_cache_path(out_dir) if args.reuse_anatomy and anatomy_cache.exists(): @@ -119,7 +117,6 @@ def main() -> int: }, ) token_cost_tracker.snapshot_stage("toc") - remove_legacy_doc_agent_artifacts(doc_agent_dir) logger.info("=" * 70) logger.info("🧠 TOC hierarchy (Stage-1 debug dump)") diff --git a/apps/worker/scripts/page_memory/debug_pm_stage2_calibration.py b/apps/worker/scripts/page_memory/debug_pm_stage2_calibration.py index 32140c90..5034d429 100644 --- a/apps/worker/scripts/page_memory/debug_pm_stage2_calibration.py +++ b/apps/worker/scripts/page_memory/debug_pm_stage2_calibration.py @@ -8,7 +8,7 @@ classify contained/parallel → graft contained → write skeleton_* Also resolves coarse skeletons (C4 resolve-only) into pipeline state so -Stage 3 can resume without re-anchoring. No fine hierarchy. +Stage 3 can resume without re-anchoring. No scope build or fine hierarchy. Requires Stage 0 → Stage 1 first: uv run python scripts/page_memory/debug_pm_stage0_bootstrap.py --file ... diff --git a/apps/worker/scripts/page_memory/debug_pm_stage3_coarse_scope.py b/apps/worker/scripts/page_memory/debug_pm_stage3_coarse_scope.py deleted file mode 100644 index 9ed64bde..00000000 --- a/apps/worker/scripts/page_memory/debug_pm_stage3_coarse_scope.py +++ /dev/null @@ -1,292 +0,0 @@ -#!/usr/bin/env python3 -# ruff: noqa: E402 -"""Stage 3: Coarse scope generation + per-scope directory creation. - -Generates coarse hierarchy scopes from skeletons and creates per-scope -directories with ``skeletons.json`` (meta + coarse nodes) plus empty -``page_tags.json`` / ``assets.json`` placeholders for later stages. - -Requires Stage 2 output: _doc_agent/pipeline_state.json (with skeletons), -doc_profile.json (after production ``run_toc_anchoring``). - -Usage: - cd apps/worker - uv run python scripts/page_memory/debug_pm_stage3_coarse_scope.py --file /path/to/doc.pdf - uv run python scripts/page_memory/debug_pm_stage3_coarse_scope.py --fat-only - uv run python scripts/page_memory/debug_pm_stage3_coarse_scope.py --page-range 225-302 -""" - -import sys -from pathlib import Path as _Path - -sys.path.insert(0, str(_Path(__file__).resolve().parent)) - -import time - -from loguru import logger - -from _debug_pm_shared import ( - TokenCostTracker, - base_argparser, - build_debug_coarse_scopes, - load_anatomy_cache, - resolve_anatomy_cache_path, - load_pipeline_skeletons, - pipeline_state_path, - record_stage, - require_file, - resolve_paths, - scope_id_for_pages, - stop_with_trace, - update_pipeline_state, - write_debug_json, - _serialize_skeletons, - _serialize_scope_skeletons, -) - - -def main() -> int: - parser = base_argparser("Stage 3: Coarse scope generation") - parser.add_argument( - "--fat-only", action="store_true", - help="Auto-select the largest coarse scope only", - ) - parser.add_argument( - "--page-range", default=None, - help="Only process page range, e.g. '225-302'", - ) - parser.add_argument( - "--all-scopes", action="store_true", - help="Process all coarse scopes (default behavior)", - ) - args = parser.parse_args() - - from app.services.page_memory.skeleton_extractor import SectionSkeleton - - pdf_path, filename, out_dir = resolve_paths(args) - doc_agent_dir = out_dir / "_doc_agent" - anatomy_cache = resolve_anatomy_cache_path(out_dir) - state_path = pipeline_state_path(out_dir) - legacy_locate_cache = doc_agent_dir / "locate_cache.json" - - if not state_path.exists() and not legacy_locate_cache.exists(): - require_file( - state_path, - hint="Run Stage 2 first: uv run python scripts/page_memory/debug_pm_stage2_calibration.py --file ...", - ) - require_file( - anatomy_cache, - hint="Run Stage 1 first: uv run python scripts/page_memory/debug_pm_stage1_hierarchy.py --file ...", - ) - - anatomy = load_anatomy_cache(anatomy_cache, pdf_path, filename) - page_count = anatomy.page_count - skeletons = load_pipeline_skeletons( - state_path, - legacy_locate_cache=legacy_locate_cache, - ) - if not state_path.exists(): - update_pipeline_state( - state_path, - stage=2, - document={ - "source_file_name": filename, - "page_count": page_count, - "anatomy_path": str(anatomy_cache), - }, - payload={ - "calibration": {}, - "null_page_parent_locate": {}, - "skeletons": _serialize_skeletons(skeletons), - "migrated_from": str(legacy_locate_cache), - }, - ) - - logger.info("█" * 70) - logger.info(f" STAGE 3: COARSE SCOPE GENERATION — {filename}") - logger.info(f" OUTPUT: {out_dir}") - logger.info("█" * 70) - - t_start = time.time() - trace_stages: list[dict] = [] - token_cost_tracker = TokenCostTracker() - from toc_page_policy import TocPagePolicy - - toc_policy = TocPagePolicy.from_anatomy(anatomy) - - # ── Build coarse scopes ── - coarse_scopes = build_debug_coarse_scopes( - skeletons=skeletons, - filename=filename, - page_count=page_count, - anatomy=anatomy, - ) - - if not coarse_scopes: - root_skel = SectionSkeleton( - section_path=f"{filename}/Root", - level=1, - start_page=1, - end_page=page_count, - title="Root", - parent_path=filename, - evidence={"source": "fallback_root"}, - ) - coarse_scopes = [ - { - "scope_id": scope_id_for_pages(1, page_count), - "skeletons": [root_skel], - "start_page": 1, - "end_page": page_count, - "strategy": "fallback_root", - "processing_pages": toc_policy.filter_processing_pages( - list(range(1, page_count + 1)) - ), - "excluded_toc_pages": sorted(toc_policy.pure_toc_pages), - } - ] - logger.info(" no skeleton hierarchy → fallback Root scope p1-{}", page_count) - - # ── Scope selection ── - if args.fat_only: - selected_scopes = [ - max(coarse_scopes, key=lambda s: int(s["end_page"]) - int(s["start_page"])) - ] - logger.info( - "🎯 --fat-only: 1/{} scopes selected {} p{}-{}", - len(coarse_scopes), - selected_scopes[0]["scope_id"], - selected_scopes[0]["start_page"], - selected_scopes[0]["end_page"], - ) - elif args.page_range: - parts = args.page_range.split("-") - pr_start = int(parts[0]) - pr_end = int(parts[1]) if len(parts) > 1 else pr_start - requested_pages = list(range(pr_start, pr_end + 1)) - pr_skeletons = [ - s for s in skeletons - if s.start_page <= pr_end and s.end_page >= pr_start - ] - selected_scopes = [ - { - "scope_id": scope_id_for_pages(pr_start, pr_end), - "skeletons": pr_skeletons, - "start_page": pr_start, - "end_page": pr_end, - "strategy": "manual_page_range", - "processing_pages": toc_policy.filter_processing_pages( - requested_pages - ), - "excluded_toc_pages": sorted( - set(requested_pages) & toc_policy.pure_toc_pages - ), - } - ] - logger.info(f" --page-range: p{pr_start}-{pr_end} ({len(pr_skeletons)} skeletons)") - else: - selected_scopes = coarse_scopes - logger.info( - " default: all {} scopes selected", len(selected_scopes), - ) - - record_stage( - trace_stages, - "C4.coarse_scopes", - variables={ - "total_coarse_scopes": len(coarse_scopes), - "selected_scopes": len(selected_scopes), - "mode": ( - "fat_only" if args.fat_only - else "page_range" if args.page_range - else "all_scopes" - ), - "scopes": [ - { - "scope_id": s["scope_id"], - "start_page": s["start_page"], - "end_page": s["end_page"], - "strategy": s.get("strategy", ""), - "skeleton_count": len(s["skeletons"]), - "processing_pages": list(s.get("processing_pages") or []), - "excluded_toc_pages": list(s.get("excluded_toc_pages") or []), - } - for s in selected_scopes - ], - }, - ) - - # ── Create per-scope directories ── - scopes_dir = out_dir / "scopes" - scopes_dir.mkdir(parents=True, exist_ok=True) - for s in selected_scopes: - scope_dir = scopes_dir / s["scope_id"] - scope_dir.mkdir(parents=True, exist_ok=True) - write_debug_json( - scope_dir / "skeletons.json", - { - **_serialize_scope_skeletons( - scope_id=str(s["scope_id"]), - start_page=int(s["start_page"]), - end_page=int(s["end_page"]), - strategy=str(s.get("strategy") or ""), - skeletons=s["skeletons"], - ), - "processing_pages": list(s.get("processing_pages") or []), - "excluded_toc_pages": list(s.get("excluded_toc_pages") or []), - }, - ) - # Placeholders for later stages (explicit empty slots for viewing). - write_debug_json(scope_dir / "page_tags.json", []) - write_debug_json(scope_dir / "assets.json", []) - - scope_rows = [ - { - "scope_id": str(scope["scope_id"]), - "start_page": int(scope["start_page"]), - "end_page": int(scope["end_page"]), - "strategy": str(scope.get("strategy") or ""), - "skeleton_count": len(scope["skeletons"]), - "processing_pages": list(scope.get("processing_pages") or []), - "excluded_toc_pages": list(scope.get("excluded_toc_pages") or []), - "artifact_path": str( - scopes_dir / str(scope["scope_id"]) / "skeletons.json" - ), - } - for scope in selected_scopes - ] - update_pipeline_state( - state_path, - stage=3, - payload={ - "selection_mode": ( - "fat_only" - if args.fat_only - else "page_range" - if args.page_range - else "all_scopes" - ), - "total_scope_count": len(coarse_scopes), - "selected_scope_count": len(selected_scopes), - "scopes": scope_rows, - }, - ) - (out_dir / "coarse_scopes.json").unlink(missing_ok=True) - - elapsed = time.time() - t_start - logger.info(f"✅ Stage 3 done in {elapsed:.1f}s") - logger.info(f" {len(selected_scopes)} scope dirs created → {scopes_dir}/") - - return stop_with_trace( - out_dir=out_dir, - stages=trace_stages, - stop_at="scope", - page_count=page_count, - pipeline_stage=3, - elapsed_s=elapsed, - token_cost_tracker=token_cost_tracker, - ) - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/apps/worker/scripts/page_memory/debug_pm_stage4_fine_hierarchy.py b/apps/worker/scripts/page_memory/debug_pm_stage3_scope_fine_hierarchy.py similarity index 60% rename from apps/worker/scripts/page_memory/debug_pm_stage4_fine_hierarchy.py rename to apps/worker/scripts/page_memory/debug_pm_stage3_scope_fine_hierarchy.py index 497a657d..81a2e7e9 100644 --- a/apps/worker/scripts/page_memory/debug_pm_stage4_fine_hierarchy.py +++ b/apps/worker/scripts/page_memory/debug_pm_stage3_scope_fine_hierarchy.py @@ -1,18 +1,20 @@ #!/usr/bin/env python3 # ruff: noqa: E402 -"""Stage 4: Document-level page tagging + per-scope fine hierarchy. +"""Stage 3: Coarse scopes + document page tagging + per-scope fine hierarchy. -Renders and tags each processing page once (global concurrency), then fans +Builds coarse hierarchy scopes, writes ``scopes//skeletons.json``, renders +and tags each selected processing page once (global concurrency), then fans tag subsets into scopes for fine hierarchy refinement. -Requires Stage 3 output: scopes//skeletons.json -Uses Stage 2 skeletons in pipeline_state for ``next_title_by_path``. +Requires Stage 2 output: _doc_agent/pipeline_state.json (with skeletons), +doc_profile.json (after production ``run_toc_anchoring``). Usage: cd apps/worker - uv run python scripts/page_memory/debug_pm_stage4_fine_hierarchy.py \\ - --file /path/to/doc.pdf --scope-id p14-23 --out-suffix boundary_clip - uv run python scripts/page_memory/debug_pm_stage4_fine_hierarchy.py --file ... --all-scopes + uv run python scripts/page_memory/debug_pm_stage3_scope_fine_hierarchy.py --file /path/to/doc.pdf + uv run python scripts/page_memory/debug_pm_stage3_scope_fine_hierarchy.py --fat-only + uv run python scripts/page_memory/debug_pm_stage3_scope_fine_hierarchy.py --all-scopes + uv run python scripts/page_memory/debug_pm_stage3_scope_fine_hierarchy.py --file ... --scope-id p14-23 """ import sys @@ -33,24 +35,27 @@ TraceStageAdapter, add_scope_selection_args, base_argparser, - list_scope_dirs, + build_debug_coarse_scopes, load_anatomy_cache, resolve_anatomy_cache_path, load_pipeline_skeletons, load_scope_skeletons_artifact, + list_scope_dirs, pipeline_state_path, record_stage, require_file, - resolve_debug_scope_ids, resolve_paths, + scope_id_for_pages, sort_skeletons, stop_with_trace, update_pipeline_state, + write_debug_json, write_scope_artifacts, write_top_level_artifacts, page_scope_info, _derive_hierarchy_page_scope, _scope_manifest, + _serialize_scope_skeletons, _serialize_skeletons, ) @@ -111,7 +116,7 @@ def _run_fine_hierarchy_for_scope( token_cost_tracker.register_child_thread() skel_path = scope_dir / "skeletons.json" - require_file(skel_path, hint=f"Run Stage 3 first to create {skel_path}") + require_file(skel_path, hint=f"Stage 3 should have created {skel_path}") scope_meta, active_skeletons = load_scope_skeletons_artifact(skel_path) strategy = str(scope_meta.get("strategy") or "coarse_scope") processing_pages, excluded_toc_pages = _resolve_scope_processing_pages( @@ -231,7 +236,7 @@ def _run_fine_hierarchy_for_scope( def main() -> int: - parser = base_argparser("Stage 4: Combined page tagging + fine hierarchy") + parser = base_argparser("Stage 3: Coarse scopes + fine hierarchy") add_scope_selection_args(parser) parser.add_argument( "--max-workers", type=int, default=5, @@ -242,25 +247,23 @@ def main() -> int: from app.services.document_agent.pdf_text import read_page_texts from app.services.page_memory.fine_hierarchy import build_next_title_by_path from app.services.page_memory.memory_service import _render_and_tag_document_pages + from app.services.page_memory.skeleton_extractor import SectionSkeleton from toc_page_policy import TocPagePolicy from shared.models.schemas.page_memory_config import PageMemoryConfig pdf_path, filename, out_dir = resolve_paths(args) - doc_agent_dir = out_dir / "_doc_agent" anatomy_cache = resolve_anatomy_cache_path(out_dir) state_path = pipeline_state_path(out_dir) - legacy_locate_cache = doc_agent_dir / "locate_cache.json" scopes_dir = out_dir / "scopes" + require_file( + state_path, + hint="Run Stage 2 first: uv run python scripts/page_memory/debug_pm_stage2_calibration.py --file ...", + ) require_file( anatomy_cache, hint="Run Stage 1 first: uv run python scripts/page_memory/debug_pm_stage1_hierarchy.py --file ...", ) - if not state_path.exists() and not legacy_locate_cache.exists(): - require_file( - state_path, - hint="Run Stage 2 first: uv run python scripts/page_memory/debug_pm_stage2_calibration.py --file ...", - ) anatomy = load_anatomy_cache(anatomy_cache, pdf_path, filename) page_count = anatomy.page_count @@ -268,49 +271,195 @@ def main() -> int: page_labels = anatomy.page_labels if anatomy else [] toc_policy = TocPagePolicy.from_anatomy(anatomy) page_memory_config = PageMemoryConfig.default() + skeletons = load_pipeline_skeletons(state_path) - all_skeletons = load_pipeline_skeletons( - state_path, - legacy_locate_cache=legacy_locate_cache, - ) - next_title_by_path = build_next_title_by_path(all_skeletons) - logger.info( - " next_title_by_path: {} paths ({} with tail anchor)", - len(next_title_by_path), - sum(1 for title in next_title_by_path.values() if title), + logger.info("█" * 70) + logger.info(f" STAGE 3: SCOPE + FINE HIERARCHY — {filename}") + logger.info(f" OUTPUT: {out_dir}") + logger.info("█" * 70) + + t_start = time.time() + trace_stages: list[dict] = [] + token_cost_tracker = TokenCostTracker() + + # ── Build coarse scopes ── + coarse_scopes = build_debug_coarse_scopes( + skeletons=skeletons, + filename=filename, + page_count=page_count, + anatomy=anatomy, ) - scope_ids = resolve_debug_scope_ids( - scopes_dir=scopes_dir, - scope_id=args.scope_id, - page_range=args.page_range, - fat_only=args.fat_only, - all_scopes=args.all_scopes, - list_scopes=args.list_scopes, - require_file="skeletons.json", + if not coarse_scopes: + root_skel = SectionSkeleton( + section_path=f"{filename}/Root", + level=1, + start_page=1, + end_page=page_count, + title="Root", + parent_path=filename, + evidence={"source": "fallback_root"}, + ) + coarse_scopes = [ + { + "scope_id": scope_id_for_pages(1, page_count), + "skeletons": [root_skel], + "start_page": 1, + "end_page": page_count, + "strategy": "fallback_root", + "processing_pages": toc_policy.filter_processing_pages( + list(range(1, page_count + 1)) + ), + "excluded_toc_pages": sorted(toc_policy.pure_toc_pages), + } + ] + logger.info(" no skeleton hierarchy → fallback Root scope p1-{}", page_count) + + if args.list_scopes: + logger.info("Available scopes ({}):", len(coarse_scopes)) + for scope in coarse_scopes: + start = int(scope["start_page"]) + end = int(scope["end_page"]) + logger.info( + " {} p{}-{} pages={} skeletons={} {}", + scope["scope_id"], + start, + end, + max(end - start + 1, 0), + len(scope["skeletons"]), + scope.get("strategy") or "", + ) + raise SystemExit(0) + + # ── Scope selection (same priority as resolve_debug_scope_ids) ── + if args.scope_id: + requested = [ + part.strip() for part in str(args.scope_id).split(",") if part.strip() + ] + by_id = {str(scope["scope_id"]): scope for scope in coarse_scopes} + missing = [sid for sid in requested if sid not in by_id] + if missing: + logger.error("❌ Unknown scope-id(s): {}", ", ".join(missing)) + logger.error( + " Available: {}", + ", ".join(str(scope["scope_id"]) for scope in coarse_scopes), + ) + raise SystemExit(1) + selected_scopes = [by_id[sid] for sid in requested] + elif args.fat_only: + selected_scopes = [ + max(coarse_scopes, key=lambda s: int(s["end_page"]) - int(s["start_page"])) + ] + logger.info( + "🎯 --fat-only: 1/{} scopes selected {} p{}-{}", + len(coarse_scopes), + selected_scopes[0]["scope_id"], + selected_scopes[0]["start_page"], + selected_scopes[0]["end_page"], + ) + elif args.page_range: + parts = args.page_range.split("-") + pr_start = int(parts[0]) + pr_end = int(parts[1]) if len(parts) > 1 else pr_start + requested_pages = list(range(pr_start, pr_end + 1)) + pr_skeletons = [ + s for s in skeletons + if s.start_page <= pr_end and s.end_page >= pr_start + ] + selected_scopes = [ + { + "scope_id": scope_id_for_pages(pr_start, pr_end), + "skeletons": pr_skeletons, + "start_page": pr_start, + "end_page": pr_end, + "strategy": "manual_page_range", + "processing_pages": toc_policy.filter_processing_pages( + requested_pages + ), + "excluded_toc_pages": sorted( + set(requested_pages) & toc_policy.pure_toc_pages + ), + } + ] + logger.info(f" --page-range: p{pr_start}-{pr_end} ({len(pr_skeletons)} skeletons)") + else: + selected_scopes = coarse_scopes + logger.info( + " default: all {} scopes selected", len(selected_scopes), + ) + + record_stage( + trace_stages, + "C4.coarse_scopes", + variables={ + "total_coarse_scopes": len(coarse_scopes), + "selected_scopes": len(selected_scopes), + "mode": ( + "scope_id" if args.scope_id + else "fat_only" if args.fat_only + else "page_range" if args.page_range + else "all_scopes" + ), + "scopes": [ + { + "scope_id": s["scope_id"], + "start_page": s["start_page"], + "end_page": s["end_page"], + "strategy": s.get("strategy", ""), + "skeleton_count": len(s["skeletons"]), + "processing_pages": list(s.get("processing_pages") or []), + "excluded_toc_pages": list(s.get("excluded_toc_pages") or []), + } + for s in selected_scopes + ], + }, ) - partial_run = len(scope_ids) < len(list_scope_dirs(scopes_dir)) - logger.info("█" * 70) - logger.info(f" STAGE 4: FINE HIERARCHY — {filename}") - logger.info(f" OUTPUT: {out_dir}") + # ── Create per-scope directories ── + scopes_dir.mkdir(parents=True, exist_ok=True) + for s in selected_scopes: + scope_dir = scopes_dir / s["scope_id"] + scope_dir.mkdir(parents=True, exist_ok=True) + write_debug_json( + scope_dir / "skeletons.json", + { + **_serialize_scope_skeletons( + scope_id=str(s["scope_id"]), + start_page=int(s["start_page"]), + end_page=int(s["end_page"]), + strategy=str(s.get("strategy") or ""), + skeletons=s["skeletons"], + ), + "processing_pages": list(s.get("processing_pages") or []), + "excluded_toc_pages": list(s.get("excluded_toc_pages") or []), + }, + ) + write_debug_json(scope_dir / "page_tags.json", []) + write_debug_json(scope_dir / "assets.json", []) + + scope_ids = [str(s["scope_id"]) for s in selected_scopes] + partial_run = len(scope_ids) < len(list_scope_dirs(scopes_dir)) logger.info(f" SCOPES ({len(scope_ids)}): {scope_ids}") if partial_run: logger.info(" MODE: partial — will not overwrite top-level hierarchy.json") - logger.info("█" * 70) - t_start = time.time() - trace_stages: list[dict] = [] - token_cost_tracker = TokenCostTracker() + next_title_by_path = build_next_title_by_path(skeletons) + logger.info( + " next_title_by_path: {} paths ({} with tail anchor)", + len(next_title_by_path), + sum(1 for title in next_title_by_path.values() if title), + ) selected_processing_pages: set[int] = set() scope_payloads: list[tuple[str, Path, list[int]]] = [] for sid in scope_ids: scope_dir = scopes_dir / sid - scope_meta, skeletons = load_scope_skeletons_artifact(scope_dir / "skeletons.json") + scope_meta, scope_skeletons = load_scope_skeletons_artifact( + scope_dir / "skeletons.json" + ) processing_pages, _excluded = _resolve_scope_processing_pages( scope_meta=scope_meta, - skeletons=skeletons, + skeletons=scope_skeletons, page_count=page_count, toc_policy=toc_policy, ) @@ -412,7 +561,7 @@ def _run_selected_scope(scope_id: str, scope_dir: Path) -> ScopeResult: ) elapsed = time.time() - t_start - logger.info(f"✅ Stage 4 done in {elapsed:.1f}s") + logger.info(f"✅ Stage 3 done in {elapsed:.1f}s") logger.info( f" {len(scope_results)} scopes processed, " f"{len(merged_skeletons)} skeletons this run, " @@ -421,11 +570,38 @@ def _run_selected_scope(scope_id: str, scope_dir: Path) -> ScopeResult: for sid in scope_ids: logger.info(f" → {scopes_dir / sid / 'fine_hierarchy.json'}") + scope_rows = [ + { + "scope_id": str(scope["scope_id"]), + "start_page": int(scope["start_page"]), + "end_page": int(scope["end_page"]), + "strategy": str(scope.get("strategy") or ""), + "skeleton_count": len(scope["skeletons"]), + "processing_pages": list(scope.get("processing_pages") or []), + "excluded_toc_pages": list(scope.get("excluded_toc_pages") or []), + "artifact_path": str( + scopes_dir / str(scope["scope_id"]) / "skeletons.json" + ), + } + for scope in selected_scopes + ] update_pipeline_state( state_path, - stage=4, + stage=3, payload={ + "selection_mode": ( + "scope_id" + if args.scope_id + else "fat_only" + if args.fat_only + else "page_range" + if args.page_range + else "all_scopes" + ), "partial_run": partial_run, + "total_scope_count": len(coarse_scopes), + "selected_scope_count": len(selected_scopes), + "scopes": scope_rows, "processed_scope_ids": scope_ids, "processed_scope_count": len(scope_results), "skeleton_count": len(merged_skeletons), @@ -436,13 +612,14 @@ def _run_selected_scope(scope_id: str, scope_dir: Path) -> ScopeResult: ], }, ) + (out_dir / "coarse_scopes.json").unlink(missing_ok=True) return stop_with_trace( out_dir=out_dir, stages=trace_stages, stop_at="fine_hierarchy", page_count=page_count, - pipeline_stage=4, + pipeline_stage=3, elapsed_s=elapsed, scope_id=scope_ids[0] if len(scope_ids) == 1 else None, token_cost_tracker=token_cost_tracker, diff --git a/apps/worker/scripts/page_memory/debug_pm_stage5_assets.py b/apps/worker/scripts/page_memory/debug_pm_stage4_assets.py similarity index 92% rename from apps/worker/scripts/page_memory/debug_pm_stage5_assets.py rename to apps/worker/scripts/page_memory/debug_pm_stage4_assets.py index 5344ea08..41b06b10 100644 --- a/apps/worker/scripts/page_memory/debug_pm_stage5_assets.py +++ b/apps/worker/scripts/page_memory/debug_pm_stage4_assets.py @@ -1,18 +1,18 @@ #!/usr/bin/env python3 # ruff: noqa: E402 -"""Stage 5: Document-level page asset extraction (C5) — NO page tagging. +"""Stage 4: Document-level page asset extraction (C5) — NO page tagging. Unions processing pages from selected scopes, renders/extracts each unique page once, writes top-level ``assets.json``, and projects references into scope dirs. -Does NOT run page tagging (C3) — Stage 4 already produced shared document-level tags. +Does NOT run page tagging (C3) — Stage 3 already produced shared document-level tags. -Requires Stage 4 output: scopes//fine_hierarchy.json +Requires Stage 3 output: scopes//fine_hierarchy.json Usage: cd apps/worker - uv run python scripts/page_memory/debug_pm_stage5_assets.py --file /path/to/doc.pdf - uv run python scripts/page_memory/debug_pm_stage5_assets.py --scope-id p1-100 - uv run python scripts/page_memory/debug_pm_stage5_assets.py --all-scopes + uv run python scripts/page_memory/debug_pm_stage4_assets.py --file /path/to/doc.pdf + uv run python scripts/page_memory/debug_pm_stage4_assets.py --scope-id p1-100 + uv run python scripts/page_memory/debug_pm_stage4_assets.py --all-scopes """ import sys @@ -69,7 +69,7 @@ def _load_scope_asset_context( fine_hierarchy_path = scope_dir / "fine_hierarchy.json" require_file( fine_hierarchy_path, - hint=f"Run Stage 4 first to produce {fine_hierarchy_path}", + hint=f"Run Stage 3 first to produce {fine_hierarchy_path}", ) prior_scope, active_skeletons = load_hierarchy_artifact(fine_hierarchy_path) if not active_skeletons: @@ -113,13 +113,13 @@ def _load_scope_asset_context( def main() -> int: - parser = base_argparser("Stage 5: Document-level asset extraction (C5)") + parser = base_argparser("Stage 4: Document-level asset extraction (C5)") add_scope_selection_args(parser) parser.add_argument( "--max-workers", type=int, default=5, - help="Kept for CLI compatibility; Stage 5 extracts once at document level", + help="Kept for CLI compatibility; Stage 4 extracts once at document level", ) args = parser.parse_args() @@ -164,7 +164,7 @@ def main() -> int: nonempty_json=True, ) logger.info("█" * 70) - logger.info(f" STAGE 5: DOCUMENT ASSET EXTRACTION — {filename}") + logger.info(f" STAGE 4: DOCUMENT ASSET EXTRACTION — {filename}") logger.info(f" OUTPUT: {out_dir}") logger.info(f" SCOPES: {scope_ids}") logger.info("█" * 70) @@ -276,14 +276,14 @@ def main() -> int: ) elapsed = time.time() - t_start - logger.info(f"✅ Stage 5 done in {elapsed:.1f}s") + logger.info(f"✅ Stage 4 done in {elapsed:.1f}s") logger.info( f" {len(scope_contexts)} scopes, {asset_count} assets, " f"{len(union_pages)} unique pages" ) update_pipeline_state( state_path, - stage=5, + stage=4, payload={ "processed_scope_ids": [context.scope_id for context in scope_contexts], "processed_scope_count": len(scope_contexts), @@ -303,7 +303,7 @@ def main() -> int: stages=trace_stages, stop_at="assets", page_count=page_count, - pipeline_stage=5, + pipeline_stage=4, elapsed_s=elapsed, token_cost_tracker=token_cost_tracker, ) diff --git a/apps/worker/scripts/page_memory/debug_pm_stage6_tagging_finalize.py b/apps/worker/scripts/page_memory/debug_pm_stage5_tagging_finalize.py similarity index 95% rename from apps/worker/scripts/page_memory/debug_pm_stage6_tagging_finalize.py rename to apps/worker/scripts/page_memory/debug_pm_stage5_tagging_finalize.py index cec8ccb0..28104075 100644 --- a/apps/worker/scripts/page_memory/debug_pm_stage6_tagging_finalize.py +++ b/apps/worker/scripts/page_memory/debug_pm_stage5_tagging_finalize.py @@ -1,19 +1,19 @@ #!/usr/bin/env python3 # ruff: noqa: E402 -"""Stage 6: Canonical chunk assembly (C7) + finalize (C9). +"""Stage 5: Canonical chunk assembly (C7) + finalize (C9). -Loads the combined page tags produced by Stage 4, assembles canonical chunks, +Loads the combined page tags produced by Stage 3, assembles canonical chunks, and optionally produces chunks.json / doc_nav.json / manifest.json. -Requires Stage 4 output: scopes//fine_hierarchy.json -Prefer Stage 5 document assets: assets.json -Legacy fallback: scopes//assets.json (deduped by asset_id) +Requires Stage 3 output: scopes//fine_hierarchy.json +Prefer Stage 4 document assets: assets.json +Fallback: scopes//assets.json (deduped by asset_id) Usage: cd apps/worker - uv run python scripts/page_memory/debug_pm_stage6_tagging_finalize.py --file /path/to/doc.pdf - uv run python scripts/page_memory/debug_pm_stage6_tagging_finalize.py --all-scopes --finalize - uv run python scripts/page_memory/debug_pm_stage6_tagging_finalize.py --scope-id p1-100 --finalize --run-db + uv run python scripts/page_memory/debug_pm_stage5_tagging_finalize.py --file /path/to/doc.pdf + uv run python scripts/page_memory/debug_pm_stage5_tagging_finalize.py --all-scopes --finalize + uv run python scripts/page_memory/debug_pm_stage5_tagging_finalize.py --scope-id p1-100 --finalize --run-db """ import sys @@ -75,7 +75,7 @@ def _run_tagging_for_scope( args: Any, token_cost_tracker: TokenCostTracker | None = None, ) -> ScopeResult: - """Load Stage-4 combined tags and rehydrate renders for final assembly.""" + """Load Stage-3 combined tags and rehydrate renders for final assembly.""" from app.services.page_memory.page_renderer import render_document_pages scope_stages: list[dict[str, Any]] = [] @@ -83,7 +83,7 @@ def _run_tagging_for_scope( token_cost_tracker.register_child_thread() fine_hierarchy_path = scope_dir / "fine_hierarchy.json" - require_file(fine_hierarchy_path, hint=f"Run Stage 4 to produce {fine_hierarchy_path}") + require_file(fine_hierarchy_path, hint=f"Run Stage 3 to produce {fine_hierarchy_path}") prior_scope, active_skeletons = load_hierarchy_artifact(fine_hierarchy_path) if not active_skeletons: logger.warning(" [scope {}] no skeletons — skipping", scope_id) @@ -93,7 +93,7 @@ def _run_tagging_for_scope( ) tags_path = scope_dir / "page_tags.json" - require_file(tags_path, hint=f"Run Stage 4 to produce {tags_path}") + require_file(tags_path, hint=f"Run Stage 3 to produce {tags_path}") tags = load_page_tags_artifact(tags_path) # Load existing assets if available @@ -109,7 +109,7 @@ def _run_tagging_for_scope( exc, ) - # Reuse Stage-4's exact scope contract. Fall back only for old artifacts. + # Reuse Stage-3's exact scope contract. Fall back only for older artifacts. recorded_pages = prior_scope.get("processing_pages") final_pages = ( [int(page) for page in recorded_pages] @@ -161,7 +161,7 @@ def _run_tagging_for_scope( }, ) - # Preserve Stage-4 tags while attaching Stage-5 assets. + # Preserve Stage-3 tags while attaching Stage-4 assets. write_scope_artifacts( out_dir=out_dir, scope_id=scope_id, @@ -247,7 +247,7 @@ def _build_report( def main() -> int: - parser = base_argparser("Stage 6: Node assembly + finalize") + parser = base_argparser("Stage 5: Node assembly + finalize") add_scope_selection_args(parser) parser.add_argument( "--max-workers", type=int, default=5, @@ -315,7 +315,7 @@ def main() -> int: nonempty_json=True, ) logger.info("█" * 70) - logger.info(f" STAGE 6: ASSEMBLY + FINALIZE — {filename}") + logger.info(f" STAGE 5: ASSEMBLY + FINALIZE — {filename}") logger.info(f" OUTPUT: {out_dir}") logger.info(f" SCOPES: {scope_ids}") logger.info("█" * 70) @@ -578,7 +578,7 @@ def _load_selected_scope(scope_id: str) -> ScopeResult: stages=trace_stages, stop_at="finalize" if args.finalize else "assembly", page_count=page_count, - pipeline_stage=6, + pipeline_stage=5, elapsed_s=elapsed, token_cost_tracker=token_cost_tracker, final_status="success", @@ -662,7 +662,7 @@ def _load_selected_scope(scope_id: str) -> ScopeResult: update_pipeline_state( state_path, - stage=6, + stage=5, payload={ "processed_scope_ids": [sr.scope_id for sr in scope_results], "finalized": bool(args.finalize), diff --git a/apps/worker/tests/contract/test_calibration_phase1_contract.py b/apps/worker/tests/contract/test_calibration_phase1_contract.py index fdbe2c24..fe649f35 100644 --- a/apps/worker/tests/contract/test_calibration_phase1_contract.py +++ b/apps/worker/tests/contract/test_calibration_phase1_contract.py @@ -96,8 +96,8 @@ def test_offset_is_found_page_minus_printed(patch_scan) -> None: assert result.status == "ok" assert [(r.kind, r.offset) for r in result.regimes] == [("decimal", 5)] - # toc_range=[1, 3] → scan starts at page after TOC end. - assert fake.calls == [("Chapter 1", 4)] + # toc_range=[1, 3], printed=10 → scan starts at the printed page. + assert fake.calls == [("Chapter 1", 10)] def test_first_hit_stops_the_regime(patch_scan) -> None: @@ -111,7 +111,7 @@ def test_first_hit_stops_the_regime(patch_scan) -> None: run_calibration_phase1(ctx=_ctx(), toc_hierarchies=hierarchies, page_count=60) - assert fake.calls == [("Chapter 1", 4)] + assert fake.calls == [("Chapter 1", 10)] def test_second_probe_runs_when_the_first_misses(patch_scan) -> None: @@ -128,7 +128,43 @@ def test_second_probe_runs_when_the_first_misses(patch_scan) -> None: ctx=_ctx(), toc_hierarchies=hierarchies, page_count=60 ) - assert fake.calls == [("Chapter 1", 4), ("Chapter 2", 4)] + assert fake.calls == [("Chapter 1", 10), ("Chapter 2", 20)] + assert [r.offset for r in result.regimes] == [5] + + +def test_probes_use_distinct_printed_pages(patch_scan) -> None: + fake = patch_scan(_FakeScan({"Chapter 2": 25})) + hierarchies = _hierarchy( + [ + {"heading": "Chapter 1", "page_number": "10", "level": 1}, + {"heading": "Chapter 1 Detail", "page_number": "10", "level": 1}, + {"heading": "Chapter 2", "page_number": "20", "level": 1}, + ] + ) + + result = run_calibration_phase1( + ctx=_ctx(), toc_hierarchies=hierarchies, page_count=60 + ) + + assert fake.calls == [("Chapter 1", 10), ("Chapter 2", 20)] + assert [r.offset for r in result.regimes] == [5] + + +def test_probe_prefers_leaf_within_a_printed_page(patch_scan) -> None: + fake = patch_scan(_FakeScan({"A1 Purpose": 15})) + hierarchies = _hierarchy( + [ + {"heading": "Part A", "page_number": "10", "level": 1}, + {"heading": "A1 Purpose", "page_number": "10", "level": 2}, + {"heading": "A2 Scope", "page_number": "20", "level": 2}, + ] + ) + + result = run_calibration_phase1( + ctx=_ctx(), toc_hierarchies=hierarchies, page_count=60 + ) + + assert fake.calls == [("A1 Purpose", 10)] assert [r.offset for r in result.regimes] == [5] @@ -169,7 +205,49 @@ def test_roman_and_decimal_regimes_calibrate_independently(patch_scan) -> None: ("roman", 2), ("decimal", 5), } - assert fake.calls == [("Preface", 4), ("Chapter 1", 4)] + # printed=2 sits inside the TOC range → floor at toc end + 1; printed=10 wins. + assert fake.calls == [("Preface", 4), ("Chapter 1", 10)] + + +def test_scan_floor_is_the_printed_page_so_offset_is_never_negative( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A divider repeating the heading cannot confirm ahead of the printed page.""" + + class _ConfirmFirstPage: + def __init__(self) -> None: + self.calls: list[tuple[str, int]] = [] + + def __call__( + self, + *, + ctx: ToolContext, + title: str, + start_page: int, + page_count: int, + **kwargs: Any, + ) -> TitleScanResult: + self.calls.append((title, start_page)) + return TitleScanResult( + title=title, + found=True, + found_page=start_page, + scanned_pages=[start_page], + next_start=start_page + 1, + ) + + fake = _ConfirmFirstPage() + monkeypatch.setattr(phase1_module, "scan_title_forward", fake) + hierarchies = _hierarchy( + [{"heading": "Chapter 1", "page_number": "10", "level": 1}] + ) + + result = run_calibration_phase1( + ctx=_ctx(), toc_hierarchies=hierarchies, page_count=60 + ) + + assert fake.calls == [("Chapter 1", 10)] + assert [r.offset for r in result.regimes] == [0] def test_confirmed_anchor_is_reported_as_a_sample(patch_scan) -> None: @@ -197,7 +275,7 @@ def test_entries_without_a_parseable_printed_page_are_skipped(patch_scan) -> Non run_calibration_phase1(ctx=_ctx(), toc_hierarchies=hierarchies, page_count=60) - assert fake.calls == [("Chapter 1", 4)] + assert fake.calls == [("Chapter 1", 10)] def test_empty_toc_fails_without_scanning(patch_scan) -> None: