--- name: appendix-table-writer description: 'Curate reader-facing survey tables for the Appendix (clean layout + high information density), using only in-scope evidence and existing citation keys. **Trigger**: appendix tables, publishable tables, survey tables, reader tables, 附录表格, 可发表表格, 综述表格. **Use when**: you have C4 artifacts (evidence packs + anchor sheet + citations) and want tables that look like a real survey (not internal logs). **Skip if**: `outline/tables_appendix.md` already exists and is refined (>=2 tables; citation-backed; no placeholders; not index-y). **Network**: none. **Guardrail**: no invented facts; no pipeline jargon; no paragraph cells; use only keys present in `citations/ref.bib`.' version: 0.1.0 metadata: openclaw: requires: anyBins: - python3 - python --- # Appendix Table Writer (publishable survey tables) ## Why this exists The pipeline can produce index tables that are useful for planning/debugging, but read like internal artifacts. This skill writes publishable, reader-facing tables that can live in an Appendix: - cleaner layout - higher information density - survey-style organization (methods/benchmarks/risks), not intermediate state Index tables remain in `outline/tables_index.md` and should not be copied verbatim into the paper. ## Inputs - `outline/table_schema.md` (table intent + evidence mapping) - `outline/tables_index.md` (internal index; optional but recommended) - `outline/subsection_briefs.jsonl` - `outline/evidence_drafts.jsonl` - `outline/anchor_sheet.jsonl` - `citations/ref.bib` - Optional: `GOAL.md` Read as needed: - `references/table_cell_hygiene.md` when Appendix table cells still copy raw paper self-narration or generic result wrappers Machine-readable assets: - `assets/table_cell_hygiene.json` ## Output - `outline/tables_appendix.md` ## Roles (use explicitly) ### Survey Table Curator (reader lens) Mission: choose tables a reader actually wants in a survey Appendix. Do: - prefer 2-3 tables that answer big questions (methods, evaluation, risks) - make rows comparable (same row unit across the table) - make the table legible without reading the whole paper Avoid: - one-row-per-H3 index dumps - columns named like internal axes ("axes", "blocking_missing", "evidence readiness") ### Production Editor (layout) Mission: make the table look publishable in LaTeX. Do: - keep columns <= 4 - keep cells short (phrases, not sentences) - use `
` sparingly (0-1 per cell; never a list dump) Avoid: - 6-8 columns with tiny unreadable text - cells that look like notes (semicolon chains + slash lists + long parentheticals) - slash-separated axis markers (A/B/C) in captions/headers/cells (post-merge voice gate will flag them); use commas or 'and' instead - internal axis jargon that reads like an intermediate artifact once printed (e.g., calling table columns "tokens"); prefer "protocol details/metadata/assumptions" ### Evidence Steward (verifiability) Mission: prevent hallucinations. Do: - every row must include citations in a dedicated column (e.g., "Key refs") - only restate what appears in evidence packs / anchor sheet - when evidence is thin, prefer fewer rows with stronger grounding Avoid: - "representative works" with no supporting claim in packs/anchors - adding benchmark/method details not present upstream ## Table contract (publishable, Appendix-ready) `outline/tables_appendix.md` must: - contain >=2 Markdown tables - use a caption line before each table, e.g. `**Appendix Table A1. ...**` - contain no headings (`#`, `##`, `###`) inside the file (the merger adds an Appendix heading) - contain no placeholders (`TODO`, `TBD`, `FIXME`, `...`, unicode ellipsis) - contain citations in rows using `[@BibKey]` (keys must exist in `citations/ref.bib`) - avoid pipeline jargon and index-like column names ## Workflow (explicit inputs) - Start from `GOAL.md` (scope) and `outline/table_schema.md` (what each table must answer). - Use `outline/tables_index.md` as a shortlist source, but do not paste it verbatim. - Fill rows/cells using `outline/subsection_briefs.jsonl`, `outline/evidence_drafts.jsonl`, and `outline/anchor_sheet.jsonl` (no guessing). - Validate every cited key against `citations/ref.bib`. ## Recommended Appendix tables (default set) If you are unsure what to build, start with these two: 1) Method/architecture map (representative works) - Row unit: work/system line (not H3 id) - Columns (example): - Work (short name) - Core idea (1 short phrase) - Loop + interface assumptions (1 short phrase; reader-facing) - Key refs (2-4 cite keys) 2) Evaluation protocol / benchmark map - Row unit: benchmark / evaluation setting (or a canonical protocol dimension if benchmarks are thin) - Columns (example): - Benchmark / setting - Task + metric (phrases, not definitions) - Key protocol constraints (budget/cost/latency/steps/tool access/threat model) - Key refs (2-4 cite keys) Optional third (only if it stays clean): 3) Risk / threat-surface map - Row unit: threat/failure mode category - Columns: surface; why it matters; mitigation pattern; key refs ## Positive / negative examples (style) Bad (index table / internal notes): - Column: "Axes" - Cell: `planning / memory / tools / eval / safety` (slash dump) - Rows: every H3 id with 5+ `
` lines Good (survey table): - Column labels are reader-facing ("Core idea", "Task + metric", "Constraint") - Cells are short phrases (no narration) - A reader can scan and compare rows quickly Also good (avoid intermediate-artifact tells): - Don't label columns as "token(s)". If you need the idea, rewrite as "protocol details/metadata/assumptions". - Avoid ASCII arrows like `->` inside cells; prefer natural phrasing (e.g., "interleaves reasoning traces with tool actions"). ## When to stop / route upstream If you cannot fill a row without guessing: - remove the row (prefer fewer, solid rows), and - route upstream: strengthen `evidence-draft` / `anchor-sheet` for that area. ## Script (generator + validator) ### Quick Start - `python scripts/run.py --help` - `python scripts/run.py --workspace workspaces/` ### All Options - `--workspace ` (required) - `--unit-id ` (optional; used only for runner bookkeeping) - `--inputs ` (optional; ignored by the validator; kept for runner compatibility) - `--outputs ` (optional; defaults to `outline/tables_appendix.md`) - `--checkpoint ` (optional; ignored by the validator) ### Examples - Validate the default appendix tables file: `python scripts/run.py --workspace workspaces/e2e-agent-survey-latex-verify-YYYYMMDD-HHMMSS` - Validate a workspace that writes appendix tables to a non-standard path: `python scripts/run.py --workspace workspaces/ --outputs outline/tables_appendix.md` Notes: - This script writes `outline/tables_appendix.md` from the existing evidence artifacts and then validates the result. - It always writes a short report to `output/TABLES_APPENDIX_REPORT.md`. FILE:AGENTS.md # AGENTS.md This exported skill uses `AGENTS.md` only as a local repo-root marker for bundled helper scripts. Source skill: `appendix-table-writer` from `research-units-pipeline-skills`. Export slug: `appendix-table-writer`. FILE:assets/table_cell_hygiene.json { "leading_context_pattern": "(?i)^(?:building on this foundation|against this backdrop|in this context|to this end|towards this end|specifically|for example|for instance|in addition|moreover|furthermore|additionally|then|crucially)[: ,]+\\s*", "leading_author_finding_pattern": "(?i)^(?:our\\s+(?:extensive\\s+)?evaluation(?:s)?(?:\\s*\$[^)]*\$)?|evaluations?\\s+across\\s+[^,]{0,160}|through\\s+(?:simulated|extensive|comprehensive)\\s+[^,]{0,160}\\s+experiments?|we|our\\s+(?:results|analysis|study|experiments?))\\s+(?:(?:also|further|furthermore|empirically|experimentally)\\s+)?(?:show|shows|find|finds|demonstrate|demonstrates|validate|validates|reveal|reveals|indicate|indicates)\\s+(?:that\\s+)?", "leading_author_action_pattern": "(?i)^(?:we|our\\s+(?:approach|method|framework|system)|specifically,\\s*we|building on this foundation,\\s*we)\\s+(?:introduce|present|propose|develop|describe|analyze|review|explore|evaluate|design|conduct)\\b[^,]{0,220},\\s*", "generic_summary_patterns": [ "(?i)^our extensive evaluation\\b", "(?i)^evaluations? across\\b", "(?i)^through simulated and real-world experiments\\b", "(?i)^through extensive benchmarking\\b", "(?i)^building on this foundation, we present\\b", "(?i)^specifically, we introduce\\b", "(?i)^we propose\\s+1\\)", "(?i)^our main contribution is\\b", "(?i)^from a detailed search of \\d+ articles\\b", "(?i)^evaluation mentions include\\b", "(?i)^when comparing results, anchor the paragraph with\\b", "(?i)^three key findings mapped to these axes\\b" ], "drop_if_starts_with": [ "we ", "our ", "this paper ", "this survey " ] } FILE:pipelines/arxiv-survey-latex.pipeline.md --- name: arxiv-survey-latex version: 3.8 variant_of: arxiv-survey variant_overrides: routing_hints: [latex, pdf, tex, 可编译, 编译] routing_default: false routing_priority: 20 units_template: templates/UNITS.arxiv-survey-latex.csv target_artifacts: __append__: - latex/main.tex - latex/main.pdf - output/LATEX_BUILD_REPORT.md stages: C5: title: Draft + PDF required_skills: __append__: - latex-scaffold - latex-compile-qa optional_skills: __remove__: - latex-scaffold - latex-compile-qa produces: __append__: - latex/main.tex - latex/main.pdf - output/LATEX_BUILD_REPORT.md --- # Pipeline: arXiv survey / review (MD-first + LaTeX/PDF) Variant of `arxiv-survey`. Use this pipeline only when the default deliverable must include: - `latex/main.tex` - `latex/main.pdf` - `output/LATEX_BUILD_REPORT.md` All non-PDF survey behavior, defaults, checkpoints, and earlier stages inherit from `arxiv-survey`. FILE:pipelines/arxiv-survey.pipeline.md --- name: arxiv-survey version: 3.8 profile: arxiv-survey routing_hints: [survey, review, 综述, 调研, literature review] routing_default: true routing_priority: 10 target_artifacts: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md - papers/papers_raw.jsonl - papers/papers_dedup.jsonl - papers/core_set.csv - papers/retrieval_report.md - outline/taxonomy.yml - outline/chapter_skeleton.yml - outline/section_bindings.jsonl - outline/section_binding_report.md - outline/section_briefs.jsonl - outline/outline.yml - outline/mapping.tsv - outline/coverage_report.md - outline/outline_state.jsonl - output/REROUTE_STATE.json - outline/subsection_briefs.jsonl - outline/chapter_briefs.jsonl - outline/transitions.md - papers/fulltext_index.jsonl - papers/paper_notes.jsonl - papers/evidence_bank.jsonl - outline/evidence_bindings.jsonl - outline/evidence_binding_report.md - outline/claim_evidence_matrix.md - outline/table_schema.md - outline/tables_index.md - outline/tables_appendix.md - output/TABLES_APPENDIX_REPORT.md - outline/evidence_drafts.jsonl - outline/anchor_sheet.jsonl - outline/writer_context_packs.jsonl - citations/ref.bib - citations/verified.jsonl - sections/sections_manifest.jsonl - sections/h3_bodies.refined.ok - sections/paragraphs_curated.refined.ok - sections/style_harmonized.refined.ok - sections/opener_varied.refined.ok - sections/abstract.md - sections/S1.md - sections/S2.md - sections/discussion.md - sections/conclusion.md - output/QUALITY_GATE.md - output/RUN_ERRORS.md - output/SCHEMA_NORMALIZATION_REPORT.md - output/EVIDENCE_SELFLOOP_TODO.md - output/WRITER_SELFLOOP_TODO.md - output/EVAL_ANCHOR_REPORT.md - output/ARGUMENT_SELFLOOP_TODO.md - output/SECTION_ARGUMENT_SUMMARIES.jsonl - output/ARGUMENT_SKELETON.md - output/PARAGRAPH_CURATION_REPORT.md - output/FRONT_MATTER_REPORT.md - output/CHAPTER_LEADS_REPORT.md - output/SECTION_LOGIC_REPORT.md - output/GLOBAL_REVIEW.md - output/DRAFT.md - output/MERGE_REPORT.md - output/POST_MERGE_VOICE_REPORT.md - output/CITATION_BUDGET_REPORT.md - output/CITATION_INJECTION_REPORT.md - output/AUDIT_REPORT.md - output/CONTRACT_REPORT.md default_checkpoints: [C0,C1,C2,C3,C4,C5] units_template: templates/UNITS.arxiv-survey.csv contract_model: pipeline.frontmatter/v1 structure_mode: section_first pre_retrieval_shell: enabled: true approval_surface: false allowed_h2: [Introduction, Related Work, Core Chapters, Discussion, Conclusion] binding_layers: [chapter_skeleton, section_bindings, section_briefs, subsection_mapping] core_chapter_h3_target: 3 query_defaults: max_results: 1800 core_size: 300 per_subsection: 28 global_citation_min_subsections: 4 draft_profile: survey citation_target: recommended evidence_mode: abstract overridable_query_fields: - keywords - exclude - max_results - core_size - per_subsection - global_citation_min_subsections - draft_profile - citation_target - enrich_metadata - evidence_mode - fulltext_max_papers - fulltext_max_pages - fulltext_min_chars - time_window.from - time_window.to quality_contract: citation_policy: unique_hard_floor: 150 unique_recommended: 165 structure_policy: max_final_h2_by_profile: survey: 8 deep: 9 max_h3_by_profile: survey: 10 deep: 12 front_matter_policy: survey: introduction: min_cites: 35 min_paras: 5 min_chars: 2600 related_work: min_cites: 50 min_paras: 6 min_chars: 3200 deep: introduction: min_cites: 40 min_paras: 6 min_chars: 3000 related_work: min_cites: 55 min_paras: 7 min_chars: 3600 subsection_policy: survey: min_unique_citations: 12 min_chars: 4200 deep: min_unique_citations: 14 min_chars: 5200 loop_policy: stage_retry_budget: C1: 2 C2: 2 C3: 1 C4: 1 max_reroutes: 4 require_human_on_retry_after_approval: true stages: C0: title: Init mode: no_prose required_skills: [workspace-init, pipeline-router] optional_skills: [] produces: [STATUS.md, UNITS.csv, CHECKPOINTS.md, DECISIONS.md, GOAL.md, queries.md, output/QUALITY_GATE.md, output/RUN_ERRORS.md] C1: title: Retrieval & core set mode: no_prose required_skills: [literature-engineer, dedupe-rank] optional_skills: [keyword-expansion, survey-seed-harvest] produces: [papers/papers_raw.jsonl, papers/retrieval_report.md, papers/papers_dedup.jsonl, papers/core_set.csv] C2: title: Structure mode: no_prose required_skills: [taxonomy-builder, chapter-skeleton, section-bindings, section-briefs, outline-builder, section-mapper, outline-refiner, pipeline-router, human-checkpoint] optional_skills: [outline-budgeter] produces: [outline/taxonomy.yml, outline/chapter_skeleton.yml, outline/section_bindings.jsonl, outline/section_binding_report.md, outline/section_briefs.jsonl, outline/outline.yml, outline/mapping.tsv, outline/coverage_report.md, outline/outline_state.jsonl, output/REROUTE_STATE.json, DECISIONS.md] human_checkpoint: approve: scope + section skeleton + outline write_to: DECISIONS.md C3: title: Evidence mode: no_prose required_skills: [pdf-text-extractor, paper-notes, subsection-briefs, chapter-briefs] optional_skills: [] produces: [papers/fulltext_index.jsonl, papers/paper_notes.jsonl, papers/evidence_bank.jsonl, outline/subsection_briefs.jsonl, outline/chapter_briefs.jsonl] C4: title: Citations + evidence packs mode: no_prose required_skills: [citation-verifier, evidence-binder, evidence-draft, table-schema, anchor-sheet, table-filler, appendix-table-writer, schema-normalizer, writer-context-pack, evidence-selfloop, claim-matrix-rewriter] optional_skills: [survey-visuals] produces: [citations/ref.bib, citations/verified.jsonl, outline/evidence_bindings.jsonl, outline/evidence_binding_report.md, outline/table_schema.md, outline/tables_index.md, outline/tables_appendix.md, output/TABLES_APPENDIX_REPORT.md, outline/evidence_drafts.jsonl, outline/anchor_sheet.jsonl, output/SCHEMA_NORMALIZATION_REPORT.md, outline/writer_context_packs.jsonl, output/EVIDENCE_SELFLOOP_TODO.md, outline/claim_evidence_matrix.md] C5: title: Draft mode: prose_allowed required_skills: [front-matter-writer, chapter-lead-writer, subsection-writer, writer-selfloop, section-logic-polisher, argument-selfloop, paragraph-curator, style-harmonizer, opener-variator, evaluation-anchor-checker, transition-weaver, section-merger, post-merge-voice-gate, citation-diversifier, citation-injector, draft-polisher, global-reviewer, pipeline-auditor, artifact-contract-auditor] optional_skills: [prose-writer, subsection-polisher, redundancy-pruner, terminology-normalizer, limitation-weaver, latex-scaffold, latex-compile-qa] produces: [outline/transitions.md, sections/sections_manifest.jsonl, sections/h3_bodies.refined.ok, sections/paragraphs_curated.refined.ok, sections/style_harmonized.refined.ok, sections/opener_varied.refined.ok, sections/abstract.md, sections/S1.md, sections/S2.md, sections/discussion.md, sections/conclusion.md, output/WRITER_SELFLOOP_TODO.md, output/EVAL_ANCHOR_REPORT.md, output/ARGUMENT_SELFLOOP_TODO.md, output/SECTION_ARGUMENT_SUMMARIES.jsonl, output/ARGUMENT_SKELETON.md, output/PARAGRAPH_CURATION_REPORT.md, output/FRONT_MATTER_REPORT.md, output/CHAPTER_LEADS_REPORT.md, output/SECTION_LOGIC_REPORT.md, output/MERGE_REPORT.md, output/DRAFT.md, output/POST_MERGE_VOICE_REPORT.md, output/CITATION_BUDGET_REPORT.md, output/CITATION_INJECTION_REPORT.md, output/GLOBAL_REVIEW.md, output/AUDIT_REPORT.md, output/CONTRACT_REPORT.md] --- # Pipeline: arXiv survey / review (MD-first) Default contract (survey-grade, A150++): - `queries.md` defaults are set for a *survey deliverable* (no silent downgrade): `core_size=300`, `per_subsection=28`, global unique citations hard floor `>=150` (recommended `>=165` when `core_size=300`; default `citation_target=recommended`). - `draft_profile` controls **writing strictness** (`survey` vs `deep`), not “speed mode”. - `evidence_mode` controls **evidence strength** (`abstract` default; `fulltext` optional and heavier). ## Stage 0 - Init (C0) required_skills: - workspace-init - pipeline-router produces: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md ## Stage 1 - Retrieval & core set (C1) required_skills: - literature-engineer - dedupe-rank optional_skills: - keyword-expansion - survey-seed-harvest produces: - papers/papers_raw.jsonl - papers/retrieval_report.md - papers/papers_dedup.jsonl - papers/core_set.csv Notes: - `queries.md` may specify `max_results` and a year `time window`; `arxiv-search` will paginate and attach arXiv metadata (categories, arxiv_id, etc.) when online. - If you import an offline export but later have network, you can set `enrich_metadata: true` in `queries.md` (or run `arxiv-search --enrich-metadata`) to backfill missing abstracts/authors/categories via arXiv `id_list`. - Evidence-first expectation (A150++): aim for a large dedup pool (target >=1200, not ~200) and a stable, verifiable core set (`core_size=300`) so later stages can bind wide in-scope citation pools without forcing out-of-scope drift. ## Stage 2 - Structure (C2) [NO PROSE] required_skills: - taxonomy-builder - chapter-skeleton - section-bindings - section-briefs - outline-builder - section-mapper - outline-refiner optional_skills: - outline-budgeter produces: - outline/taxonomy.yml - outline/chapter_skeleton.yml - outline/section_bindings.jsonl - outline/section_binding_report.md - outline/section_briefs.jsonl - outline/outline.yml - outline/mapping.tsv - outline/coverage_report.md - outline/outline_state.jsonl - output/REROUTE_STATE.json human_checkpoint: - approve: scope + section skeleton + outline - write_to: DECISIONS.md Notes: - `chapter-skeleton` is the first retrieval-informed chapter contract; it stays chapter-level and does not emit stable H3 ids. - `section-bindings` measures chapter saturation before H3 decomposition and writes PASS/BLOCKED signals into `outline/section_binding_report.md`. - `section-briefs` turns the chapter layer into decomposition guidance plus subsection seeds; `outline-builder` then derives the first stable `outline/outline.yml` from that section layer. - `outline-refiner` now writes both `outline/outline_state.jsonl` and `output/REROUTE_STATE.json`; section-first reroute/block information should be read from those artifacts instead of inferred from prose notes. - Evidence-first expectation: each subsection should be written as a *question to answer* (RQ) plus *evidence needs* (what kind of citations/results are required), not just generic scaffold bullets. - Coverage default: `section-mapper` uses `queries.md:per_subsection` as the per-H3 mapping contract (A150++ default: 28) so later evidence binding and writing have enough in-scope citations to choose from. - Diversity expectation: mapping should not over-reuse a few papers across unrelated H3s; reserve “global” works for genuinely cross-cutting citations (controlled by `global_citation_min_subsections`). - Budget policy (paper-like): avoid H3 explosion; the outline gate uses `queries.md:draft_profile` to set max H3 (survey<=10, deep<=12). - If the outline is over-fragmented, use `outline-budgeter` (NO PROSE) to merge adjacent H3s into fewer, thicker units, then rerun `section-mapper` → `outline-refiner` before `Approve C2`. ## Stage 3 - Evidence (C3) [NO PROSE] required_skills: - pdf-text-extractor - paper-notes - subsection-briefs - chapter-briefs produces: - papers/fulltext_index.jsonl - papers/paper_notes.jsonl - papers/evidence_bank.jsonl - outline/subsection_briefs.jsonl - outline/chapter_briefs.jsonl Notes: - `queries.md` can set `evidence_mode: "abstract"|"fulltext"` (A150++ default: `abstract`). - `queries.md` can set `draft_profile: "survey"|"deep"` to control writing gate strictness (A150++ default: `survey`). - If `evidence_mode: "fulltext"`, `pdf-text-extractor` can be tuned via `fulltext_max_papers`, `fulltext_max_pages`, `fulltext_min_chars`. - `subsection-briefs` converts each H3 into a verifiable writing card (scope_rule/rq/axes/clusters/paragraph_plan) so writing does not copy outline scaffolds. - Optional refinement markers (recommended): treat briefs as *contracts*, not scaffolds. If you manually refine them and want to prevent regeneration, create: - `outline/subsection_briefs.refined.ok` - `outline/chapter_briefs.refined.ok` These markers are used as explicit “reviewed/refined” signals and as a freeze switch (scripts won’t overwrite refined briefs). ## Stage 4 - Citations + evidence packs (C4) [NO PROSE] required_skills: - citation-verifier - evidence-binder - evidence-draft - table-schema - anchor-sheet - table-filler - appendix-table-writer - schema-normalizer - writer-context-pack - evidence-selfloop - claim-matrix-rewriter optional_skills: - survey-visuals produces: - citations/ref.bib - citations/verified.jsonl - outline/evidence_bindings.jsonl - outline/evidence_binding_report.md - outline/table_schema.md - outline/tables_index.md - outline/tables_appendix.md - output/TABLES_APPENDIX_REPORT.md - outline/evidence_drafts.jsonl - outline/anchor_sheet.jsonl - output/SCHEMA_NORMALIZATION_REPORT.md - outline/writer_context_packs.jsonl - output/EVIDENCE_SELFLOOP_TODO.md - outline/claim_evidence_matrix.md Notes: - `evidence-draft` turns paper notes into per-subsection evidence packs (claim candidates + concrete comparisons + eval protocol + limitations) that the writer must follow. - `claim-matrix-rewriter` makes `outline/claim_evidence_matrix.md` a projection/index of evidence packs (not an outline expansion), so writer guidance stays evidence-first. - `writer-context-pack` builds a deterministic per-H3 drafting pack (briefs + evidence + anchors + allowed cites), reducing hollow writing and making C5 more debuggable. - Tables are part of the default survey deliverable, but split into two layers: - `outline/tables_index.md` (internal index; produced by `table-filler`; useful for planning/debugging; NOT inserted into the paper) - `outline/tables_appendix.md` (reader-facing; produced by `appendix-table-writer`; clean/publishable; inserted into the draft as an Appendix block by `section-merger`) - Optional: `survey-visuals` can still produce timeline/figure specs as intermediate artifacts. - Optional refinement markers (recommended): after you spot-check/refine C4 artifacts and want to freeze them, create: - `outline/evidence_bindings.refined.ok` - `outline/evidence_drafts.refined.ok` - `outline/anchor_sheet.refined.ok` - `outline/writer_context_packs.refined.ok` These markers make “reviewed/refined” explicit and prevent accidental regeneration/overwrite; strict mode relies on content checks (placeholders/blocking_missing/scope), not marker presence. ## Stage 5 - Draft (C5) [PROSE AFTER C2] required_skills: - front-matter-writer - chapter-lead-writer - subsection-writer - writer-selfloop - section-logic-polisher - argument-selfloop - paragraph-curator - style-harmonizer - opener-variator - evaluation-anchor-checker - transition-weaver - section-merger - post-merge-voice-gate - citation-diversifier - citation-injector - draft-polisher - global-reviewer - pipeline-auditor - artifact-contract-auditor optional_skills: - prose-writer - subsection-polisher - redundancy-pruner - terminology-normalizer - limitation-weaver - evaluation-anchor-checker - latex-scaffold - latex-compile-qa produces: - sections/sections_manifest.jsonl - sections/abstract.md - sections/discussion.md - sections/conclusion.md - output/WRITER_SELFLOOP_TODO.md - output/EVAL_ANCHOR_REPORT.md - output/SECTION_LOGIC_REPORT.md - output/ARGUMENT_SELFLOOP_TODO.md - output/SECTION_ARGUMENT_SUMMARIES.jsonl - output/ARGUMENT_SKELETON.md - output/MERGE_REPORT.md - output/DRAFT.md - output/POST_MERGE_VOICE_REPORT.md - output/CITATION_BUDGET_REPORT.md - output/CITATION_INJECTION_REPORT.md - output/GLOBAL_REVIEW.md - output/AUDIT_REPORT.md - output/CONTRACT_REPORT.md Notes: - C5 writing system (semantic + minimal artifacts; no extra machinery): - **Unit of work**: `sections/*.md` (front matter, H2 leads, H3 bodies). Avoid editing `output/DRAFT.md` directly until after merge. - **Single source of truth (口径锁定)**: `output/ARGUMENT_SKELETON.md` → `## Consistency Contract` (terminology, scope boundary, evaluation protocol fields, baseline naming). - **Write → check → fix (five gates)**: 1) `writer-selfloop` → `output/WRITER_SELFLOOP_TODO.md`: file existence, depth, citation scope, paper voice. 2) `section-logic-polisher` → `output/SECTION_LOGIC_REPORT.md`: paragraph linkage (no jump cuts / “paragraph islands”). 3) `argument-selfloop` → `output/ARGUMENT_SELFLOOP_TODO.md` + `output/ARGUMENT_SKELETON.md` + `output/SECTION_ARGUMENT_SUMMARIES.jsonl`: section-level closure + premise/definition stability. 4) `paragraph-curator` → `output/PARAGRAPH_CURATION_REPORT.md`: **select → evaluate → subset → fuse** so sections converge (reduce redundancy, strengthen synthesis) without changing citation keys. 5) `evaluation-anchor-checker` → `output/EVAL_ANCHOR_REPORT.md`: final section-level numeric hygiene sweep so surviving numeric claims keep same-sentence task/metric/constraint context before merge. - **Openers-last**: draft the middle first; rewrite paragraph 1 last so it reflects real content (front matter + H3). - Writing self-loop gate: `subsection-writer` ensures the full `sections/` file set exists (and emits `sections/sections_manifest.jsonl`); `writer-selfloop` blocks until depth/citation-scope/paper-voice checks pass, writing `output/WRITER_SELFLOOP_TODO.md` (PASS/FAIL). - Argument self-loop gate: `argument-selfloop` blocks “smooth but hollow” sections by making the argument chain explicit (per-section paragraph moves + a global dependency skeleton). Its ledgers are intermediate artifacts and must never be merged into the paper. - Style hygiene (C5 hard gate for `survey`/`deep`): treat `output/WRITER_SELFLOOP_TODO.md` Style Smells as mandatory fixes. Run `style-harmonizer` + `opener-variator` on flagged files, then rerun `writer-selfloop` before merge. - Numeric hygiene gate: run `evaluation-anchor-checker` after the last section-level rewrites (`paragraph-curator` + `style-harmonizer` + `opener-variator`) and before merge. If later section rewrites touch the same H3 files, rerun it; do not wait for `pipeline-auditor` to discover underspecified numbers in the merged draft. - Micro-fix routing (preferred over broad rewrites): if Style Smells are specific, use targeted micro-skills before a general harmonize pass: - opener cadence / “overview” narration → `opener-variator` - count-based limitation slots (“Two limitations…”) → `limitation-weaver` - underspecified numeric/performance claims (missing task/metric/budget) → `evaluation-anchor-checker` - Triage rule (prevents “写作补洞”): if `writer-selfloop` FAILs because a subsection cannot meet `must_use` *in-scope* (thin packs / missing anchors / out-of-scope citation pressure), stop and rerun the evidence loop (`evidence-selfloop` + upstream C2/C3/C4) instead of padding prose. - WebWeaver-style “planner vs writer” split (single agent, two passes): - Planner pass: for each section/subsection, pick the exact citation IDs to use from the evidence bank (`outline/evidence_drafts.jsonl`) and keep scope consistent with the outline. - Writer pass: write that section using only those citation IDs; avoid dumping the whole notes set into context. - Treat this stage as an iteration loop: draft per H3 → logic-polish (thesis + connectors) → weave transitions → merge → de-template/cohere → global review → (if gaps) back to C3/C4 → regenerate. - Post-merge voice gate: `post-merge-voice-gate` treats `outline/transitions.md` as a high-frequency injection source. If it FAILs, fix the *source* (usually transitions via `transition-weaver`, or the owning `sections/*.md`) and re-merge; do not “patch around it” in `draft-polisher`. - Depth target (profile-aware): each H3 should be “少而厚” (avoid stubs). Use `queries.md:draft_profile` as the contract: - `survey`: >=10 paragraphs + >=12 unique cites - `deep`: >=11 paragraphs + >=14 unique cites In all profiles, require >=2 concrete contrasts + evaluation anchoring + a cross-paper synthesis paragraph + an explicit limitation. - Profile semantics: `survey` is the default deliverable contract; `deep` is stricter (and typically pairs well with `evidence_mode: fulltext`). - Coherence target (paper-like): for every H2 chapter with H3 subsections, write a short **chapter lead** block (`sections/S_lead.md`) that previews the comparison axes and how the H3s connect (no new headings; avoid generic glue). - Anti-template style contract (paper-like, not “outline narration”): - Avoid meta openers like “This subsection surveys/argues …” and slide-like navigation (“Next, we move from … / We now turn to …”). - Keep signposting light: avoid repeating a literal opener label across many subsections (e.g., `Key takeaway:`); vary opener phrasing and cadence. - Tone target: calm, academic, understated; delete hype words (`clearly`, `obviously`) and “PPT speaker notes”. - Keep evidence-policy disclaimers **once** in front matter (not repeated across H3s). - If you cite numbers, include minimal evaluation context (task + metric + constraint/budget/cost) in the same paragraph. - Citation shape must be reader-facing: no adjacent citation blocks (e.g., `[@a] [@b]`), no duplicate keys in one block (e.g., `[@a; @a]`), and avoid tail-only citation style by keeping mid-sentence citations in each H3. - `section-merger` merges `sections/*.md` plus `outline/transitions.md` (within-chapter H3→H3 by default). Between-H2 transition insertion is optional: create `outline/transitions.insert_h2.ok` in the workspace if you want narrator-style handoffs included. - Tables are part of the default deliverable: `outline/tables_appendix.md` is inserted into the draft by `section-merger` as a single Appendix block (index tables in `outline/tables_index.md` remain intermediate) unless `outline/tables.insert.off` exists. Other visuals (`outline/timeline.md`, `outline/figures.md`) remain intermediate by default. - Citation scope policy: citations are subsection-first (from `outline/evidence_bindings.jsonl`), with limited reuse allowed within the same H2 chapter to reduce brittleness; avoid cross-chapter “free cite” drift. - Controlled flexibility: bibkeys mapped to >= `queries.md:global_citation_min_subsections` subsections (A150++ default: 4) are treated as cross-cutting/global; see `allowed_bibkeys_global` in writer packs / `sections_manifest.jsonl`. - If global unique citations are low, run `citation-diversifier` → `citation-injector` *before* `draft-polisher` (the polisher treats citation keys as immutable). - `queries.md` can set `citation_target: recommended|hard` to control whether the recommended target is enforced as blocking (default: `recommended` for A150++). - If you intentionally add/remove citations after an earlier polish run, reset the citation-anchoring baseline before rerunning `draft-polisher`: - delete `output/citation_anchors.prepolish.jsonl` (workspace-local), then rerun `draft-polisher`. - Recommended skills (toolkit, not a rigid one-shot chain): - Modular drafting: `subsection-writer` → `writer-selfloop` → `section-logic-polisher` → `argument-selfloop` → `paragraph-curator` → `style-harmonizer` → `opener-variator` → `evaluation-anchor-checker` → `transition-weaver` → `section-merger` → `draft-polisher` → `global-reviewer` → `pipeline-auditor`. - Legacy one-shot drafting: `prose-writer` (kept for quick experiments; less debuggable). - If the draft reads like “paragraph islands”, run `section-logic-polisher` and patch only failing `sections/S*.md` until PASS, then merge. - Add `pipeline-auditor` after `global-reviewer` as a regression test (blocks on ellipsis, repeated boilerplate, and citation hygiene). - If you also need a PDF deliverable, use `latex-scaffold` + `latex-compile-qa` (see `arxiv-survey-latex`). ## Quality gates (strict mode) - Citation coverage: expect a large, verifiable bibliography (A150++ default: `core_size=300` → `ref.bib` ~300) and high cite density: - Per-H3: `survey` profile expects >=12 unique citations per H3 (and deeper profiles may require more). - Front matter: `survey` profile expects Introduction>=35 and Related Work>=50 unique citations (dense positioning; no cite dumps). - Global: `pipeline-auditor` gates on **global unique citations across the full draft**. A150++ defaults: hard `>=150`; recommended `>=165` (when bib=300). `queries.md:citation_target` controls which is blocking (default: `recommended`). If it fails, prefer `citation-diversifier` → `citation-injector` (in-scope, NO NEW FACTS) using each H3’s `allowed_bibkeys_selected` / `allowed_bibkeys_mapped` from `outline/writer_context_packs.jsonl`. - Anti-template: drafts containing ellipsis placeholders (`…`) or leaked scaffold instructions (e.g., "enumerate 2-4 ...") should block and be regenerated from improved outline/mapping/evidence artifacts. - Final polish hard gates (`survey`/`deep`): block on narration-template openers (e.g., `This subsection ...`), slide navigation phrasing, repeated opener stems/口癖, adjacent citation blocks (`[@a] [@b]`), duplicate keys in one block (`[@a; @a]`), and low H3 mid-sentence citation ratio (<30%). FILE:pipelines/graduate-paper-pipeline.md # Pipeline：中文毕业论文重构与定稿（研究阶段草案） > 状态：研究阶段草案 > 当前定位：先研究流程与所需 skills，**暂不绑定 UNITS，不强行做成可执行合同** > 面向对象：已有学校模板、既有论文、Overleaf 源稿、PDF、BibTeX、实验图表与中间工作文档的中文毕业论文写作 > 核心目标：把“已有材料”重构为“主线统一、结构顺畅、证据完整、术语一致、可编译交付”的中文毕业论文 ## 1. 这条 Pipeline 的本质这不是一条“从零写论文”的线性流程，也不是“把几篇 paper 拼起来”的拼装流程，而是一条**以问题清单驱动、以 Markdown 中间层为主、以 TeX 交付层收口**的论文工程 Pipeline。它的真实主循环是： > `问题清单 -> 语义定位 -> Markdown 重构 -> TeX 回写 -> 编译复查 -> 问题回写 -> 再进入下一轮` 因此，这条 Pipeline 的首要目标不是“赶紧写出一版正文”，而是先把以下几件事做对： 1. 先把已有材料还原出来，而不是直接在 `tex` 里盲改。 2. 先围绕毕业论文主线重构章节角色，而不是照搬原论文叙事。 3. 先在 Markdown 中间层把结构、证据、术语与图表想清楚，再回写到 TeX。 4. 先解决结构与证据问题，再解决文风与去 AI 味问题。 5. 真正的收敛点不是“写完一遍”，而是“问题清单被逐轮清空、编译与复查稳定通过”。 ## 2. 这条 Pipeline 和 survey Pipeline 的根本区别它与 `arxiv-survey` / `arxiv-survey-latex` 的差异很大，必须单独建模： - survey pipeline 的起点是“根据主题检索文献并逐层收敛结构”； - graduate-paper pipeline 的起点是“已有材料盘点与重构”； - survey pipeline 的核心中间工件是 `outline/evidence_drafts/...`； - graduate-paper pipeline 的核心中间工件是 `codex_md/` 里的大纲、问题清单、章节中间稿、图表计划和复查记录； - survey pipeline 的主目标是“生成一篇综述”； - graduate-paper pipeline 的主目标是“把已有研究工作重构成一篇符合中文学位论文规范的定稿论文”。一句话说： > survey 更像“检索驱动的证据写作流程”； > graduate-paper 更像“已有材料驱动的论文工程重构流程”。 ## 3. 工作区分层与关键工件这条 Pipeline 需要明确区分**思考层**与**交付层**。 ### 3.1 目录职责 - `main.tex`：论文总入口，负责装配封面、摘要、目录、正文、附录、致谢、成果与参考文献。 - `chapters/`：最终交付版正文 `.tex`。 - `abstract/`、`preface/`、`acknowledgement/`、`achievements/`：非正文正式交付部分。 - `references/`：参考文献库与样式文件。 - `pdf/`：已发表或已投稿论文 PDF。 - `Overleaf_ref/`：既有源稿、修回稿、补充材料。 - `codex_md/`：中间工作层，是**论文重构工作区**。 - `claude_md/`：复查清单、终稿核对记录。 - `tmp_layout/`、`tmp_layout2/`：图表试排与版面预演。其中最关键的是： - `codex_md/` 是**思考区 / 重构区** - `chapters/` 是**交付区 / 定稿区** 不要把二者混用。 ### 3.2 建议固定的中间工件建议在这条 Pipeline 里把以下工件长期固定下来： - `codex_md/material_index.md`：已有材料盘点与索引 - `codex_md/material_readiness.md`：材料就绪度与缺口提示 - `codex_md/missing_info.md`：缺失信息清单 - `codex_md/question_list.md`：本轮问题单、优先级与验收口径 - `codex_md/00_thesis_outline.md`：毕业论文主线、大纲、章节角色 - `codex_md/chapter_role_map.md`：来源材料 -> 章节 -> 角色映射表 - `codex_md/chapter_rewrite_rules.md`：章节重构准则表 - `codex_md/terminology_glossary.md`：术语统一表 - `codex_md/symbol_metric_table.md`：符号 / 指标统一表 - `codex_md/figure_plan.md`：图表与版面计划 - `claude_md/review_checklist.md`：编译、排版、数据、模板项复查清单 - `output/THESIS_BUILD_REPORT.md`：编译与终稿检查报告 ## 4. 这条 Pipeline 真正需要的 skills 当前先不考虑复用和泛化，按这条毕业论文 Pipeline 自身所需来拆，第一版建议使用以下 skills。 ### 4.1 主链 skills 1. `thesis-workspace-init` 职责：初始化论文工程、提醒用户放置材料、建立工作区、检查 `main.tex` 是否可编译、生成材料盘点。 2. `thesis-question-list` 职责：建立并维护 `question_list.md`，把每一轮修改目标、问题优先级、边界、验收口径固定下来。这是整条 Pipeline 的**控制面 skill**。 3. `thesis-source-role-mapper` 职责：把既有论文 / 模板 / 源稿 / PDF / 图表材料映射到“毕业论文角色”，不是只做 `paper -> chapter`。 4. `thesis-chapter-reconstructor` 职责：围绕毕业论文主线，重构每一章的目标、比重、承接与叙事方式。这是整条 Pipeline 的**核心重构 skill**。 5. `thesis-markdown-aligner` 职责：统一主线、术语、符号、指标、图表口径，并让各章在 Markdown 中间层收敛成一篇论文而不是一组 paper。 6. `thesis-tex-writeback` 职责：把已经在 Markdown 层理顺的内容回写到 `chapters/*.tex`，并同步图表、公式、交叉引用与章节承接。 7. `thesis-compile-review` 职责：执行编译、warning 分级、模板模式检查、数据口径核验、问题回写。它负责形成最后的质量闭环。 8. `thesis-style-polisher` 职责：在结构、证据、数据稳定后，做中文学位论文风格润色与去 AI 味处理。 ### 4.2 并行支线 skills 9. `thesis-visual-layout-planner` 职责：图表规划、Mermaid 草图、临时拼版、图文节奏检查。 10. `thesis-frontmatter-sync` 职责：同步摘要、封面、附录、成果、致谢、名单等非正文部分，避免这些内容拖到最后才补。 11. `thesis-citation-enhance-review` 职责：定位必须有引用支撑的句子，扩充候选文献，核验引用与论断匹配关系，回写 `references/*.bib` 与正文引用。 ### 4.3 这 11 个 skills 的关系主链是： `thesis-workspace-init -> thesis-question-list -> thesis-source-role-mapper -> thesis-chapter-reconstructor -> thesis-markdown-aligner -> thesis-tex-writeback -> thesis-compile-review -> thesis-style-polisher` 并行支线是： - `thesis-visual-layout-planner` - `thesis-frontmatter-sync` - `thesis-citation-enhance-review` 其中： - `thesis-question-list` 是全流程控制面 - `thesis-chapter-reconstructor` 是核心重构面 - `thesis-compile-review` 是质量闭环面 ## 5. Pipeline 总览（按阶段） | 阶段 | 目标 | 核心 skills | 主要输出 | |---|---|---|---| | 阶段 0 | 工程初始化与材料入仓 | `thesis-workspace-init` | 目录骨架、初始编译、材料索引、材料就绪度提示 | | 阶段 1 | 还原已有材料到 Markdown 中间层 | `thesis-workspace-init` | 初始 Markdown 材料、缺失信息清单、材料缺口提示 | | 阶段 1.5 | 锁定本轮问题与边界 | `thesis-question-list` | `question_list.md` | | 阶段 2 | 建立来源材料与毕业论文角色映射 | `thesis-source-role-mapper` | 章节角色映射表、按章材料归类 | | 阶段 2.5 | 围绕主线重构章节 | `thesis-chapter-reconstructor` | 重构后的章节 Markdown | | 阶段 3 | 对齐、归并并统一全篇结构 | `thesis-markdown-aligner` | 稳定 outline、术语表、符号表、证据缺口清单 | | 阶段 3.5 | 图表与版面预演 | `thesis-visual-layout-planner` | 图表计划、图文映射、临时版式结果 | | 阶段 4 | 回写 TeX 交付层 | `thesis-tex-writeback` | 可编译章节 `.tex`、初版完整 `main.pdf` | | 阶段 4.5 | 同步非正文部分 | `thesis-frontmatter-sync` | 摘要、附录、封面、成果等同步版本 | | 阶段 5 | 引用增强与核验 | `thesis-citation-enhance-review` | 引用补强结果、核验记录 | | 阶段 6 | 编译、排版与终稿复查 | `thesis-compile-review` | `THESIS_BUILD_REPORT`、review checklist | | 阶段 7 | 最终润色与去 AI 味 | `thesis-style-polisher` | 更自然的中文终稿 | ## 6. 分阶段详细说明 ### 阶段 0：工程初始化与材料入仓 **目标** 先把论文变成一个可维护的工程，而不是一堆零散文件。 **主要输入** - 学校模板 - 现有仓库 / 源稿 - 学号、年份、中英文题目等基础元信息 - 现有 `main.tex` **核心 skill** - `thesis-workspace-init` **主要输出** - 基础目录骨架 - 初始 `main.tex` / 初始 `main.pdf` - `codex_md/material_index.md` - `codex_md/material_readiness.md` **衔接关系** - 如果 `main.tex` 还不能编译，不要进入正文重构阶段 - 如果材料还没归位，不要开始问题清单与章节重构 **执行重点** - 明确哪些是材料区，哪些是工作区，哪些是交付区 - 先保证“能编译”，再谈“写得好” ### 阶段 1：还原已有材料到 Markdown 中间层 **目标** 把既有 `template / tex / Overleaf / PDF / bib` 中可复用的内容还原出来，形成 Markdown 中间层，而不是直接在 TeX 层硬改。 **主要输入** - 学校模板 - 既有 `.tex` - Overleaf 源稿 - PDF - 参考文献库 **核心 skill** - `thesis-workspace-init` **主要输出** - 初始 Markdown 章节材料 - `codex_md/material_readiness.md` - `codex_md/missing_info.md` - 材料索引与待补信息清单 **衔接关系** - 阶段 1 的输出会直接喂给 `thesis-question-list` 和 `thesis-source-role-mapper` **执行重点** - 重点是“材料资产盘点”，不是“先写漂亮句子” - 要显式标出待补实验细节、待补图注、待补引用、待补数字核验 ### 阶段 1.5：锁定问题清单与本轮目标 **目标** 建立本轮控制面，明确这轮到底修什么，不修什么。 **主要输入** - 初始 Markdown 材料 - 上一轮 review 反馈 - 当前编译 / 结构 / 文风问题 **核心 skill** - `thesis-question-list` **主要输出** - `codex_md/question_list.md` **衔接关系** - 这一步是之后所有重构与回写的起点 - 任何结构性争议，都应该先回到这里重新排序 **执行重点** - 每个问题都要写清楚：是什么、为什么、怎么改、验收到什么程度 - 问题单不是备忘录，而是迭代入口 ### 阶段 2：建立来源材料与毕业论文角色映射 **目标** 把已有材料映射到毕业论文的章节角色，而不是只做“paper 对 chapter”的机械分配。 **主要输入** - 现有论文、PDF、源稿、图表 - 初始 Markdown 材料 - `question_list.md` **核心 skill** - `thesis-source-role-mapper` **主要输出** - `codex_md/chapter_role_map.md` - 按章归类的 Markdown 材料 - 各章内容边界 **衔接关系** - 这是 `thesis-chapter-reconstructor` 的直接输入 **执行重点** - 区分：方法主体 / 背景支撑 / 验证证据 / 系统实现 / 局限与展望素材 - 这里定错了，后面整篇论文都会歪 ### 阶段 2.5：围绕主线重构章节 **目标** 把原论文式叙事改造成毕业论文式叙事。 **主要输入** - 章节角色映射表 - `question_list.md` - 各章 Markdown 草稿 **核心 skill** - `thesis-chapter-reconstructor` **主要输出** - 重构后的章节 Markdown - `codex_md/chapter_rewrite_rules.md` **衔接关系** - 这是整条 Pipeline 的核心阶段 - 后续一切对齐、回写、编译，都是在这里的重构成果之上进行 **执行重点** - 明确每章要回答什么问题 - 决定哪些内容强化、弱化、上移、下沉 - 重写与前后章节的承接关系 - 把“卖点叙事”改成“毕业论文主线叙事” ### 阶段 3：Markdown 对齐、归并与统一 **目标** 让整篇论文在中间层先变成“一篇论文”，而不是几篇论文的拼接。 **主要输入** - 重构后的章节 Markdown - `question_list.md` **核心 skill** - `thesis-markdown-aligner` **主要输出** - 稳定版 `codex_md/00_thesis_outline.md` - `codex_md/terminology_glossary.md` - `codex_md/symbol_metric_table.md` - 证据缺口清单 **衔接关系** - 结构不稳、术语不稳、指标不稳时，不要进入 `tex-writeback` **执行重点** - 统一研究主线、术语、缩写、符号、指标、图表口径 - 决定哪些内容在第 2 章统一解释，哪些在第 3-5 章展开，哪些进入附录 ### 阶段 3.5：图表与版面预演 **目标** 在正文定稿前提前规划图表与版式，不要把图表拖成最后的补丁工程。 **主要输入** - 稳定版 outline - 章节 Markdown - 已有图表与实验结果 **核心 skill** - `thesis-visual-layout-planner` **主要输出** - `codex_md/figure_plan.md` - `mermaid/` 图示草稿 - `tmp_layout/`、`tmp_layout2/` 版面试排结果 **衔接关系** - 图表规划结果会直接影响 `tex-writeback` 的图文结构 **执行重点** - 明确每章需要哪些图表支撑主线 - 先做草图，再做临时拼版，再决定最终落位与图注 ### 阶段 4：回写 TeX 交付层 **目标** 把已经在 Markdown 层理顺的内容回写到 `chapters/*.tex`，并进入交付层。 **主要输入** - 稳定版 Markdown 章节 - 图表计划 - 术语 / 符号 / 指标统一表 **核心 skill** - `thesis-tex-writeback` **主要输出** - `chapters/*.tex` - 初版完整 `main.pdf` **衔接关系** - 如果在这个阶段发现结构问题，原则上优先回 Markdown 层修，不在 TeX 层重新发明结构 **执行重点** - 同步图表、公式、交叉引用、章首导言与章末小结 - TeX 层负责交付，不负责重新思考结构 ### 阶段 4.5：同步非正文部分 **目标** 并行维护摘要、附录、封面、成果、致谢等非正文部分，避免它们被拖到最后变成低质量补写。 **主要输入** - 正文稳定版本 - 学校模板项 - 中英文题目与摘要信息 **核心 skill** - `thesis-frontmatter-sync` **主要输出** - `abstract/abstract.tex` - `abstract/abstract-en.tex` - `preface/...` - `appendix/...` - `acknowledgement/...` - `achievements/...` **衔接关系** - 阶段 6 的终稿复查会把这些部分与正文一并检查 **执行重点** - 中英文题目与摘要口径一致 - 摘要中的数字、指标、方法名与正文一致 - 附录只放补充性内容，不重复主线 ### 阶段 5：文献增强与引用核验 **目标** 把“应该有引用支撑的句子”系统补齐，并确保引用与论断匹配。 **主要输入** - 当前 Markdown / TeX 版本 - `references/*.bib` - 现有 PDF / 参考工作 **核心 skill** - `thesis-citation-enhance-review` **主要输出** - 增强后的参考文献库 - 引用补强记录 - 引用核验记录 **衔接关系** - 文献增强可以与阶段 3、4 并行，但收口必须早于最终润色 **执行重点** - 优先找事实性描述、方法分类、经典结论、指标定义、数据来源这类必须有引文支撑的句子 - 先补对，再补多 ### 阶段 6：编译、排版与终稿复查 **目标** 形成质量闭环，不只是“能编译”，而是“适合提交”。 **主要输入** - 完整 TeX 工程 - 非正文同步版本 - review checklist **核心 skill** - `thesis-compile-review` **主要输出** - `output/THESIS_BUILD_REPORT.md` - `claude_md/review_checklist.md` - 已关闭 / 待关闭问题清单 **衔接关系** - 阶段 6 的问题会决定回到阶段 1.5 / 2.5 / 3 / 4 的哪个层面修复 **执行重点** - 编译通过只是最低要求 - 需要分级处理 warning：阻断提交 / 必须修 / 可记录观察 - 还要检查：数据一致性、cite key 有效性、模板参数是否正确、术语与缩写是否统一 ### 阶段 7：最终润色与去 AI 味 **目标** 在结构、证据、数据稳定之后，再做中文学位论文风格润色。 **主要输入** - 通过编译与复查的论文版本 - `中文写作要求.md` - `GPT口癖与高频用词调研.md` **核心 skill** - `thesis-style-polisher` **主要输出** - 更自然的中文终稿 - 去模板化后的导言、小结、总结与展望 **衔接关系** - 阶段 7 必须放在最后 - 如果润色暴露出结构或证据问题，应回退到前面阶段，而不是硬润色遮盖 **执行重点** - 优先处理章首导言、章末小结、贡献描述、总结与展望 - 压低模板腔、宣传腔、AI 常见口癖 - 目标是“更像学位论文”，不是“更像 AI 优化后的论文” ## 7. 这条 Pipeline 的四个固定回环为了避免它再次退化成“线性 SOP”，建议把以下四个回环固定下来： ### 7.1 结构闭环 `outline -> 某章 md -> 发现章节失衡 -> 回 outline / question_list` ### 7.2 内容闭环 `paper / Overleaf / PDF -> 抽取内容 -> 重构叙事 -> 发现仍像原论文 -> 再重构` ### 7.3 排版闭环 `tex -> 编译 -> 发现 warning / 图表位置 / 交叉引用问题 -> 回 tex 或图表规划` ### 7.4 文风闭环 `正文稳定 -> AI 润色 -> 发现模板腔 / AI 味 -> 回写作规范与措辞控制` ## 8. 这条 Pipeline 的执行优先级如果实际运行时资源有限，优先级应当是： 1. `thesis-question-list` 2. `thesis-source-role-mapper` 3. `thesis-chapter-reconstructor` 4. `thesis-markdown-aligner` 5. `thesis-tex-writeback` 6. `thesis-compile-review` 换句话说： - 真正不能省的是“问题清单 + 材料角色映射 + 章节重构 + 编译复查” - 真正不该最早做的是“润色与去 AI 味” ## 9. 当前结论这条 graduate-paper pipeline 的第一原则应当是： > **先重构，再成文；先中间层收敛，再 TeX 交付；先证据与结构正确，再文风与润色。** 因此，这条 Pipeline 后续如果要正式执行化，最值得优先实现的不是“一个大而全的 runner”，而是先把以下 skills 做稳： - `thesis-workspace-init` - `thesis-question-list` - `thesis-source-role-mapper` - `thesis-chapter-reconstructor` - `thesis-compile-review` 这五个 skill 稳了，这条中文毕业论文 Pipeline 才真正有落地价值。 FILE:pipelines/idea-brainstorm.pipeline.md --- name: idea-brainstorm version: 4.0 profile: idea-brainstorm routing_hints: [idea, ideation, brainstorm, 点子, 选题, 找方向, 找 idea] routing_priority: 30 target_artifacts: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md - output/trace/IDEA_BRIEF.md - papers/papers_raw.jsonl - papers/papers_dedup.jsonl - papers/core_set.csv - papers/retrieval_report.md - outline/taxonomy.yml - papers/paper_notes.jsonl - papers/evidence_bank.jsonl - output/trace/IDEA_SIGNAL_TABLE.md - output/trace/IDEA_SIGNAL_TABLE.jsonl - output/trace/IDEA_DIRECTION_POOL.md - output/trace/IDEA_DIRECTION_POOL.jsonl - output/trace/IDEA_SCREENING_TABLE.md - output/trace/IDEA_SCREENING_TABLE.jsonl - output/trace/IDEA_SHORTLIST.md - output/trace/IDEA_SHORTLIST.jsonl - output/REPORT.md - output/APPENDIX.md - output/REPORT.json - output/DELIVERABLE_SELFLOOP_TODO.md - output/QUALITY_GATE.md - output/RUN_ERRORS.md - output/CONTRACT_REPORT.md default_checkpoints: [C0,C1,C2,C3,C4,C5] units_template: templates/UNITS.idea-brainstorm.csv contract_model: pipeline.frontmatter/v1 query_defaults: draft_profile: idea_brainstorm max_results: 1800 core_size: 100 evidence_mode: abstract direction_pool_min: 12 direction_pool_max: 24 idea_screen_top_n: 10 idea_shortlist_size: 5 report_top_n: 3 overridable_query_fields: - keywords - exclude - max_results - core_size - evidence_mode - direction_pool_min - direction_pool_max - idea_screen_top_n - idea_shortlist_size - report_top_n - time_window.from - time_window.to quality_contract: memo_bundle: required_outputs: [output/REPORT.md, output/APPENDIX.md, output/REPORT.json] signal_policy: min_rows: 10 direction_policy: shortlist_min: 3 shortlist_max: 5 keep_min: 3 cluster_diversity_min: 2 lead_diversity_target: 3 lead_diversity_axes: [cluster, direction_type, program_kind] screening_policy: keep_rank_max: 7 maybe_rank_max: 12 score_weights: discussion_worthiness: 0.24 academic_value: 0.22 evidence_grounding: 0.18 direction_distinctness: 0.16 first_probe_clarity: 0.10 thesis_potential: 0.10 loop_policy: stage_retry_budget: C1: 2 C3: 1 C4: 1 max_reroutes: 4 require_human_on_retry_after_approval: true stages: C0: title: Init + idea brief mode: no_prose required_skills: [workspace-init, pipeline-router, idea-brief, human-checkpoint] optional_skills: [] produces: [STATUS.md, UNITS.csv, CHECKPOINTS.md, DECISIONS.md, GOAL.md, queries.md, output/trace/IDEA_BRIEF.md, output/QUALITY_GATE.md, output/RUN_ERRORS.md] human_checkpoint: approve: brainstorm brief write_to: DECISIONS.md C1: title: Retrieval + core set mode: no_prose required_skills: [literature-engineer, dedupe-rank] optional_skills: [] produces: [papers/papers_raw.jsonl, papers/papers_dedup.jsonl, papers/core_set.csv, papers/retrieval_report.md] C2: title: Idea landscape / focus mode: no_prose required_skills: [taxonomy-builder, pipeline-router, human-checkpoint] optional_skills: [] produces: [outline/taxonomy.yml, DECISIONS.md] human_checkpoint: approve: focus clusters / lenses + exclusions write_to: DECISIONS.md C3: title: Evidence signals mode: no_prose required_skills: [paper-notes, idea-signal-mapper] optional_skills: [] produces: [papers/paper_notes.jsonl, papers/evidence_bank.jsonl, output/trace/IDEA_SIGNAL_TABLE.md, output/trace/IDEA_SIGNAL_TABLE.jsonl] C4: title: Direction pool + screening mode: short_prose_ok required_skills: [idea-direction-generator, idea-screener] optional_skills: [] produces: [output/trace/IDEA_DIRECTION_POOL.md, output/trace/IDEA_DIRECTION_POOL.jsonl, output/trace/IDEA_SCREENING_TABLE.md, output/trace/IDEA_SCREENING_TABLE.jsonl] C5: title: Shortlist + memo synthesis + self-loop mode: prose_allowed required_skills: [idea-shortlist-curator, idea-memo-writer, deliverable-selfloop, artifact-contract-auditor] optional_skills: [] produces: [output/trace/IDEA_SHORTLIST.md, output/trace/IDEA_SHORTLIST.jsonl, output/REPORT.md, output/APPENDIX.md, output/REPORT.json, output/DELIVERABLE_SELFLOOP_TODO.md, output/CONTRACT_REPORT.md] --- # Pipeline: research idea brainstorm (signals -> directions -> memo) Goal: produce a **discussion-ready research-idea memo** for PI / PhD readers. This pipeline is for “找 research idea / brainstorm / 选题 / 找方向”. It is **not** for writing a survey draft and **not** for generating execution-grade project specs. The terminal deliverable is a single default entrypoint: - `output/REPORT.md` That memo should feel like a mature brainstorm artifact: - grounded in literature, - small enough to discuss, - clear about what is promising, - honest about uncertainty, - and rich enough to guide the next discussion round. Artifact policy: - Reader-facing terminal artifacts: - `output/REPORT.md` - `output/APPENDIX.md` - `output/REPORT.json` - Trace artifacts stay under `output/trace/`. - The run is only shareable when both layers exist and pass contract audit. Default profile: - Retrieval route: `literature-engineer` - `core_size=100` - Evidence mode: `abstract` - Signal table: 10-20 rows - Direction pool: 12-24 - Shortlist: 3-5 - Final memo lead directions: 3 ## Stage 0 - Init + idea brief (C0) required_skills: - workspace-init - pipeline-router - idea-brief - human-checkpoint produces: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md - output/trace/IDEA_BRIEF.md Notes: - The brief is the single source of truth for topic, audience, constraints, exclusions, targets, and query buckets. - Default behavior: block once at C0 for human approval of the brief before retrieval. ## Stage 1 - Retrieval + core set (C1) required_skills: - literature-engineer - dedupe-rank produces: - papers/papers_raw.jsonl - papers/papers_dedup.jsonl - papers/core_set.csv - papers/retrieval_report.md Notes: - Use multi-query buckets from `queries.md`. - If recall is too low/noisy, fix it upstream and rerun C1. ## Stage 2 - Idea landscape / focus (C2) [NO PROSE] required_skills: - taxonomy-builder - pipeline-router - human-checkpoint produces: - outline/taxonomy.yml - DECISIONS.md human_checkpoint: - approve: focus clusters / lenses + exclusions - write_to: DECISIONS.md Notes: - Taxonomy is an idea landscape, not a paper outline. - Default behavior: pause at C2 so the human can choose a few promising focus lenses. ## Stage 3 - Evidence signals (C3) [table-first] required_skills: - paper-notes - idea-signal-mapper produces: - papers/paper_notes.jsonl - papers/evidence_bank.jsonl - output/trace/IDEA_SIGNAL_TABLE.md - output/trace/IDEA_SIGNAL_TABLE.jsonl Notes: - The signal table should capture tensions, missing pieces, and academically meaningful axes rather than proposal-ready wedges. - Keep this stage compact and table-first; no long prose. ## Stage 4 - Direction pool + screening (C4) required_skills: - idea-direction-generator - idea-screener produces: - output/trace/IDEA_DIRECTION_POOL.md - output/trace/IDEA_DIRECTION_POOL.jsonl - output/trace/IDEA_SCREENING_TABLE.md - output/trace/IDEA_SCREENING_TABLE.jsonl Notes: - Expansion should generate discussion-worthy research directions, not an operator cartesian product. - Screening should score discussion value, distinctness, evidence grounding, and thesis potential. ## Stage 5 - Shortlist + memo synthesis + self-loop (C5) required_skills: - idea-shortlist-curator - idea-memo-writer - deliverable-selfloop - artifact-contract-auditor produces: - output/trace/IDEA_SHORTLIST.md - output/trace/IDEA_SHORTLIST.jsonl - output/REPORT.md - output/APPENDIX.md - output/REPORT.json - output/DELIVERABLE_SELFLOOP_TODO.md - output/CONTRACT_REPORT.md Notes: - `output/trace/IDEA_SHORTLIST.md` is an internal convergence layer, not the user-facing final answer. - `output/REPORT.md` is the terminal deliverable and should read like a discussion-ready research idea brainstorm memo. - `deliverable-selfloop` should evaluate the full memo bundle plus its trace chain. FILE:pipelines/lit-snapshot.pipeline.md --- name: lit-snapshot version: 2.3 profile: lit-snapshot routing_hints: [snapshot, 快照, one-page, one page, 48h] routing_priority: 30 target_artifacts: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md - papers/papers_raw.jsonl - papers/papers_dedup.jsonl - papers/core_set.csv - outline/taxonomy.yml - outline/outline.yml - output/SNAPSHOT.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/QUALITY_GATE.md - output/RUN_ERRORS.md - output/CONTRACT_REPORT.md default_checkpoints: [C0,C1,C2,C3] units_template: templates/UNITS.lit-snapshot.csv --- # Pipeline: literature snapshot (24-48h) [bullets-first] Goal: a compact, reader-facing snapshot (`output/SNAPSHOT.md`) with a small but usable paper set and a paper-like structure. This is intentionally lighter than `arxiv-survey*` (no evidence packs, no BibTeX, no LaTeX). ## Stage 0 - Init (C0) required_skills: - workspace-init - pipeline-router produces: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md ## Stage 1 - Retrieval & core set (C1) required_skills: - arxiv-search - dedupe-rank optional_skills: - keyword-expansion produces: - papers/papers_raw.jsonl - papers/papers_dedup.jsonl - papers/core_set.csv Notes: - Snapshot default: aim for a smaller but diverse set (e.g., core_set ~20-40) rather than a survey-scale pool. - Practical retrieval loop (multi-query, then refine): - Start broad with multiple query buckets (synonyms/acronyms/subtopics) and merge the results. - If too few: add buckets + widen time window; if too noisy: rewrite keywords + add exclusions; rerun C1. - If the snapshot feels generic, the fix is almost always upstream: broaden `queries.md` (more synonyms, fewer excludes, wider time window) and rerun C1. ## Stage 2 - Structure (C2) [NO PROSE] required_skills: - taxonomy-builder - outline-builder optional_skills: - outline-budgeter produces: - outline/taxonomy.yml - outline/outline.yml human_checkpoint: - approve: scope + outline (snapshot ToC) - write_to: DECISIONS.md Notes: - Paper-like default: prefer fewer, thicker sections. For snapshots, avoid H3 explosion; treat H2 as the main organizing unit. - If the outline is over-fragmented, use `outline-budgeter` (NO PROSE) to merge adjacent nodes and keep the ToC readable before writing. ## Stage 3 - Snapshot (C3) [SHORT PROSE OK] required_skills: - snapshot-writer - deliverable-selfloop - artifact-contract-auditor optional_skills: - prose-writer produces: - output/SNAPSHOT.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/CONTRACT_REPORT.md Notes: - The deliverable is bullets-first and pointer-heavy: every non-trivial claim should attach 1-2 concrete paper pointers from `papers/core_set.csv`. - Avoid outline narration (e.g., `This section surveys ...`); write content claims + why-it-matters + pointers. FILE:pipelines/peer-review.pipeline.md --- name: peer-review version: 2.3 profile: peer-review routing_hints: [peer review, review report, referee, 审稿] routing_priority: 30 target_artifacts: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md - output/PAPER.md - output/CLAIMS.md - output/MISSING_EVIDENCE.md - output/NOVELTY_MATRIX.md - output/REVIEW.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/QUALITY_GATE.md - output/RUN_ERRORS.md - output/CONTRACT_REPORT.md default_checkpoints: [C0,C1,C2,C3] units_template: templates/UNITS.peer-review.csv --- # Pipeline: peer review / referee report Goal: produce an actionable referee report (`output/REVIEW.md`) that is traceable to the submitted paper’s claims and evidence (no free-floating advice, no invented comparisons). ## Stage 0 - Init (C0) required_skills: - workspace-init - pipeline-router produces: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md ## Stage 1 - Claims (C1) required_skills: - manuscript-ingest - claims-extractor produces: - output/PAPER.md - output/CLAIMS.md Notes: - Input expectation: you must have the manuscript text available as `output/PAPER.md` (or equivalent extracted text) before running this stage. - Contract: every claim must include a source pointer (section/page/quote) so later critique is auditable. ## Stage 2 - Evidence audit (C2) required_skills: - evidence-auditor - novelty-matrix produces: - output/MISSING_EVIDENCE.md - output/NOVELTY_MATRIX.md Notes: - Evidence-auditor should write gaps/risks/verification steps, not “fix the paper”. - Novelty-matrix should be conservative: compare against cited/known related work; if you lack sources, record that limitation explicitly. ## Stage 3 - Rubric write-up (C3) required_skills: - rubric-writer - deliverable-selfloop - artifact-contract-auditor produces: - output/REVIEW.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/CONTRACT_REPORT.md Notes: - Prefer concrete, minimal fixes (what experiment/ablation/analysis would resolve the concern) over generic “needs more experiments”. FILE:pipelines/systematic-review.pipeline.md --- name: systematic-review version: 2.3 profile: systematic-review routing_hints: [systematic review, systematic, prisma, 系统综述] routing_priority: 30 target_artifacts: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md - output/PROTOCOL.md - papers/papers_raw.jsonl - papers/retrieval_report.md - papers/papers_dedup.jsonl - papers/core_set.csv - papers/screening_log.csv - papers/extraction_table.csv - output/SYNTHESIS.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/QUALITY_GATE.md - output/RUN_ERRORS.md - output/CONTRACT_REPORT.md default_checkpoints: [C0,C1,C2,C3,C4,C5] units_template: templates/UNITS.systematic-review.csv --- # Pipeline: systematic review (PRISMA-style) ## Stage 0 - Init (C0) required_skills: - workspace-init - pipeline-router produces: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md ## Stage 1 - Protocol (C1) required_skills: - protocol-writer produces: - output/PROTOCOL.md human_checkpoint: - approve: protocol locked (query, inclusion/exclusion, databases, time window) - write_to: DECISIONS.md ## Stage 2 - Retrieval & candidate pool (C2) required_skills: - literature-engineer - dedupe-rank optional_skills: - keyword-expansion - arxiv-search produces: - papers/papers_raw.jsonl - papers/retrieval_report.md - papers/papers_dedup.jsonl - papers/core_set.csv Notes: - Systematic-review contract: keep the candidate pool auditable. Prefer dedupe (good) over aggressive ranking/filtering (dangerous). - `dedupe-rank` default: for this pipeline, it keeps the full deduped pool in `papers/core_set.csv` unless you explicitly set `queries.md:core_size` (screening should not silently drop papers). ## Stage 3 - Screening (C3) required_skills: - screening-manager produces: - papers/screening_log.csv Notes: - Use `papers/papers_dedup.jsonl` (or `papers/core_set.csv`) as the candidate list; `papers/screening_log.csv` must include protocol-grounded reasons for every decision. - Practical auditability: number protocol clauses (`I1..` / `E1..`) and require screening rows to cite them (e.g., `reason_codes=E3`). ## Stage 4 - Extraction (C4) required_skills: - extraction-form - bias-assessor produces: - papers/extraction_table.csv Notes: - Extraction is schema-driven: do not write narrative synthesis here; keep all fields consistent and fill bias fields (low/unclear/high + notes). ## Stage 5 - Synthesis (C5) [PROSE ALLOWED] required_skills: - synthesis-writer - deliverable-selfloop - artifact-contract-auditor produces: - output/SYNTHESIS.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/CONTRACT_REPORT.md FILE:pipelines/tutorial.pipeline.md --- name: tutorial version: 2.3 profile: tutorial routing_hints: [tutorial, 教程] routing_priority: 30 target_artifacts: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md - output/TUTORIAL_SPEC.md - outline/concept_graph.yml - outline/module_plan.yml - output/TUTORIAL.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/QUALITY_GATE.md - output/RUN_ERRORS.md - output/CONTRACT_REPORT.md default_checkpoints: [C0,C1,C2,C3] units_template: templates/UNITS.tutorial.csv --- # Pipeline: tutorial (teaching loop) Goal: a tutorial deliverable (`output/TUTORIAL.md`) that has a consistent running example and a real teaching loop (objectives -> steps -> exercises -> verification), not just a blog-style explanation. ## Stage 0 - Init (C0) required_skills: - workspace-init - pipeline-router produces: - STATUS.md - UNITS.csv - CHECKPOINTS.md - DECISIONS.md - GOAL.md - queries.md ## Stage 1 - Spec (C1) required_skills: - tutorial-spec produces: - output/TUTORIAL_SPEC.md Notes: - The spec is the scope contract: audience, prerequisites, learning objectives, and a running example (if applicable). ## Stage 2 - Structure (C2) [NO PROSE] required_skills: - concept-graph - module-planner - exercise-builder produces: - outline/concept_graph.yml - outline/module_plan.yml human_checkpoint: - approve: target audience + scope + running example - write_to: DECISIONS.md Notes: - Treat `outline/module_plan.yml` as the execution contract for writing: modules must have concrete outputs and at least one verifiable exercise each. ## Stage 3 - Writing (C3) [PROSE ALLOWED] required_skills: - tutorial-module-writer - deliverable-selfloop - artifact-contract-auditor produces: - output/TUTORIAL.md - output/DELIVERABLE_SELFLOOP_TODO.md - output/CONTRACT_REPORT.md Notes: - Write only within the approved scope; if you discover missing prerequisites, record them as a follow-up instead of expanding scope silently. FILE:references/table_cell_hygiene.md # Table Cell Hygiene ## Purpose Appendix tables are reader-facing. They should not expose raw evidence snippets that still sound like paper self-narration. ## Drop or rewrite - `we propose ...` - `our extensive evaluation ...` - `evaluations across ... validate ...` - `through extensive benchmarking ...` - `specifically, we introduce ...` - `building on this foundation, we present ...` - writer-instruction cells such as `Evaluation mentions include ...` or `When comparing results, anchor ...` - survey-meta or scope-setting lines that belong in prose, not in a table cell ## Keep - short result clauses - benchmark / metric anchors - compact limitation or risk statements - row-comparable evidence phrases ## Boundary The hygiene policy belongs in `assets/table_cell_hygiene.json`. The script should apply it deterministically and skip unusable cells rather than preserving reader-unfriendly wrappers. FILE:scripts/run.py from __future__ import annotations import argparse import json import re import sys from pathlib import Path from typing import Any _TABLE_SEP = re.compile(r"(?m)^\|?\s*:?-{3,}:?\s*(\|\s*:?-{3,}:?\s*)+\|?$") _ASSET_ROOT = Path(__file__).resolve().parents[1] / "assets" def _load_json(path: Path) -> dict[str, Any]: if not path.exists() or path.stat().st_size <= 0: return {} try: data = json.loads(path.read_text(encoding="utf-8", errors="ignore")) except Exception: return {} return data if isinstance(data, dict) else {} _CELL_HYGIENE = _load_json(_ASSET_ROOT / "table_cell_hygiene.json") def _compile_pattern(key: str, default: str) -> re.Pattern[str]: return re.compile(str(_CELL_HYGIENE.get(key) or default).strip()) def _compile_pattern_list(key: str, default: list[str]) -> list[re.Pattern[str]]: raw = _CELL_HYGIENE.get(key) if not isinstance(raw, list): raw = default return [re.compile(str(item).strip()) for item in raw if str(item).strip()] _LEADING_CONTEXT_RE = _compile_pattern( "leading_context_pattern", r"(?i)^(?:building on this foundation|against this backdrop|in this context|to this end|towards this end|specifically|for example|for instance|in addition|moreover|furthermore|additionally|then|crucially)[: ,]+\s*", ) _LEADING_AUTHOR_FINDING_RE = _compile_pattern( "leading_author_finding_pattern", r"(?i)^(?:we|our\s+(?:results|analysis|study|experiments?))\s+(?:show|shows|find|finds|demonstrate|demonstrates|validate|validates|reveal|reveals|indicate|indicates)\s+(?:that\s+)?", ) _LEADING_AUTHOR_ACTION_RE = _compile_pattern( "leading_author_action_pattern", r"(?i)^(?:we|our\s+(?:approach|method|framework|system))\s+(?:introduce|present|propose|develop|describe|analyze|review|explore|evaluate|design|conduct)\b[^,]{0,220},\s*", ) _GENERIC_SUMMARY_PATTERNS = _compile_pattern_list( "generic_summary_patterns", [r"(?i)^our extensive evaluation\b", r"(?i)^we propose\s+1\)"], ) _DROP_PREFIXES = [str(x).strip().lower() for x in (_CELL_HYGIENE.get("drop_if_starts_with") or []) if str(x).strip()] def _is_placeholder(text: str) -> bool: low = (text or "").strip().lower() if not low: return True if "" end = f"" block = "\n".join([begin, markdown_block.rstrip(), end, ""]).rstrip() + "\n" if decisions_path.exists(): text = decisions_path.read_text(encoding="utf-8") else: text = "# Decisions log\n\n" ensure_decisions_approval_checklist(decisions_path) text = decisions_path.read_text(encoding="utf-8") pattern = re.compile( rf"{re.escape(begin)}.*?{re.escape(end)}\n?", flags=re.DOTALL, ) if pattern.search(text): new_text = pattern.sub(block, text) else: new_text = text.rstrip() + "\n\n" + block atomic_write_text(decisions_path, new_text) def seed_queries_from_topic(queries_path: Path, topic: str) -> None: topic = topic.strip() if not topic: return if queries_path.exists(): lines = queries_path.read_text(encoding="utf-8").splitlines() else: lines = [ "# Queries", "", "## Primary query", "- keywords:", " - \"\"", "- exclude:", " - \"\"", "- max_results: \"\"", "- core_size: \"\"", "- time window:", " - from: \"\"", " - to: \"\"", "", "## Notes", "-", ] def _has_nonempty_values(token: str) -> bool: in_block = False for raw in lines: stripped = raw.strip() if stripped.startswith(f"- {token}:"): in_block = True continue if not in_block: continue if raw.startswith(" - "): value = stripped[2:].strip().strip('"').strip("'") if value: return True continue if stripped.startswith("- "): break return False def _has_nonempty_scalar(token: str) -> bool: for raw in lines: stripped = raw.strip() if not stripped.startswith(f"- {token}:"): continue value = stripped.split(":", 1)[1].split("#", 1)[0].strip().strip('"').strip("'") return bool(value) return False def _has_nonempty_time_field(field: str) -> bool: # Looks for lines like: ' - from: "2022"' for raw in lines: stripped = raw.strip() if not stripped.startswith(f"- {field}:"): continue value = stripped.split(":", 1)[1].strip().strip('"').strip("'") return bool(value) return False has_keywords = _has_nonempty_values("keywords") has_excludes = _has_nonempty_values("exclude") has_time_from = _has_nonempty_time_field("from") has_time_to = _has_nonempty_time_field("to") has_max_results = _has_nonempty_scalar("max_results") has_core_size = _has_nonempty_scalar("core_size") workspace = queries_path.parent profile = pipeline_profile(workspace) query_defaults = pipeline_query_defaults(workspace) raw_tlow = topic.lower() topic_for_queries = _sanitize_topic_for_query_seed(topic) keyword_suggestions = [topic_for_queries] tlow = topic_for_queries.lower() is_agent = any(t in tlow for t in ("agent", "agents", "agentic")) is_embodied = any( t in tlow for t in ( "embodied ai", "embodied intelligence", "embodied agent", "embodied robotics", "robot foundation model", "robot learning", "robot manipulation", "vision-language-action", "vla", "generalist robot", ) ) is_text_to_image = any(t in tlow for t in ("text-to-image", "text to image", "t2i")) is_text_to_video = any(t in tlow for t in ("text-to-video", "text to video", "t2v")) is_diffusion = "diffusion" in tlow is_generative = is_text_to_image or is_text_to_video or is_diffusion or ("image generation" in tlow) or ("generative" in tlow) if is_agent: keyword_suggestions.extend( [ "LLM agent", "language model agent", "tool use", "function calling", "tool-using agent", "planning", "memory", "multi-agent", "benchmark", "safety", ] ) exclude_suggestions: list[str] = [] if is_agent: exclude_suggestions.append("agent-based modeling") exclude_suggestions.extend(["react hooks", "perovskite", "banach", "coxeter"]) if is_embodied: keyword_suggestions.extend( [ "embodied AI survey", "embodied AI review", "embodied intelligence survey", "embodied agent survey", "robot foundation model survey", "robot learning survey", "robot manipulation survey", "embodied robotics survey", "vision-language-action survey", "vision-language-action model", "robot foundation model", "generalist robot policy", "world model robot", ] ) stripped_output_terms = topic_for_queries.lower().strip() != raw_tlow.strip() if stripped_output_terms and any(t in raw_tlow for t in ("latex", "pdf", "markdown", "typesetting")): exclude_suggestions.extend(["latex", "pdf", "typesetting", "document layout"]) if is_generative: keyword_suggestions.extend( [ "text-to-image generation", "text-guided image generation", "diffusion model", "denoising diffusion probabilistic model", "latent diffusion", "stable diffusion", "classifier-free guidance", "diffusion transformer", "DiT", "masked generative transformer", "MaskGIT", "autoregressive image generation", "VQGAN", "VQ-VAE", "ControlNet", "DreamBooth", "textual inversion", "LoRA fine-tuning", ] ) default_max_results = query_defaults.get("max_results") max_results_suggestion = str(default_max_results) if str(default_max_results or "").strip() else ( 1800 if profile == "arxiv-survey" else (800 if (is_agent or is_generative or is_embodied) else 300) ) time_from_suggestion = ( "2018" if is_embodied else ("2022" if (is_agent and ("llm" in tlow or "language model" in tlow)) else ("2020" if is_generative else "")) ) core_size_suggestion = str(query_defaults.get("core_size") or "").strip() or ("300" if profile == "arxiv-survey" else "") out: list[str] = [] i = 0 while i < len(lines): line = lines[i] stripped = line.strip() if stripped.startswith("- keywords:") and not has_keywords: out.append(line) i += 1 while i < len(lines) and lines[i].startswith(" - "): i += 1 for kw in _dedupe_preserve_order(keyword_suggestions)[:14]: out.append(f" - \"{kw}\"") continue if stripped.startswith("- exclude:") and not has_excludes: out.append(line) i += 1 while i < len(lines) and lines[i].startswith(" - "): i += 1 for ex in _dedupe_preserve_order(exclude_suggestions)[:10]: out.append(f" - \"{ex}\"") continue if stripped.startswith("- max_results:") and not has_max_results and max_results_suggestion: out.append(f"- max_results: \"{max_results_suggestion}\"") i += 1 continue if stripped.startswith("- core_size:") and not has_core_size and core_size_suggestion: out.append(f"- core_size: \"{core_size_suggestion}\"") i += 1 continue if stripped.startswith("- time window:") and not (has_time_from or has_time_to) and time_from_suggestion: out.append(line) i += 1 # Skip existing from/to lines if present. while i < len(lines) and lines[i].startswith(" -"): i += 1 out.append(f" - from: \"{time_from_suggestion}\"") out.append(" - to: \"\"") continue out.append(line) i += 1 out = _materialize_missing_query_defaults(out, query_defaults, allowed_fields=pipeline_overridable_query_fields(workspace)) atomic_write_text(queries_path, "\n".join(out).rstrip() + "\n") def _dedupe_preserve_order(items: list[str]) -> list[str]: seen: set[str] = set() out: list[str] = [] for item in items: item = item.strip() if not item or item in seen: continue seen.add(item) out.append(item) return out def _sanitize_topic_for_query_seed(topic: str) -> str: text = str(topic or "").strip() if not text: return "" patterns = [ r"(?i)\bwith\s+latex\s*/\s*pdf\s+output\b", r"(?i)\bwith\s+latex\s+output\b", r"(?i)\bwith\s+pdf\s+output\b", r"(?i)\bwith\s+markdown\s+output\b", r"(?i)\blatex\s*/\s*pdf\s+output\b", r"(?i)\bpdf\s+output\b", r"(?i)\blatex\s+output\b", r"(?i)\bmarkdown\s+output\b", r"(?i)\bfor\s+latex\s*/\s*pdf\b", ] for pattern in patterns: text = re.sub(pattern, "", text) text = re.sub(r"\s+", " ", text).strip(" ,;:-") return text or topic _LEGACY_PIPELINE_ALIASES = { "idea-finder": "idea-brainstorm", "idea-finder.pipeline.md": "idea-brainstorm", "pipelines/idea-finder.pipeline.md": "idea-brainstorm", } def find_repo_root(start: Path | None = None) -> Path: """Walk up from *start* (default: this file) looking for AGENTS.md.""" candidate = (start or Path(__file__)).resolve() for _ in range(10): if (candidate / "AGENTS.md").exists(): return candidate parent = candidate.parent if parent == candidate: break candidate = parent raise FileNotFoundError("Could not find repo root (AGENTS.md marker)") def _normalize_pipeline_lock_value(value: str) -> str: raw = str(value or "").strip() return _LEGACY_PIPELINE_ALIASES.get(raw, raw) def resolve_pipeline_spec_path(*, repo_root: Path, pipeline_value: str) -> Path | None: value = _normalize_pipeline_lock_value(pipeline_value) if not value: return None candidate = Path(value) if candidate.is_absolute() and candidate.exists(): return candidate.resolve() rel_candidate = repo_root / value if rel_candidate.exists(): return rel_candidate.resolve() filename = Path(value).name if filename: direct = repo_root / "pipelines" / filename if direct.exists(): return direct.resolve() stem = filename if stem.endswith(".pipeline.md"): stem = stem[: -len(".pipeline.md")] if stem: direct = repo_root / "pipelines" / f"{stem}.pipeline.md" if direct.exists(): return direct.resolve() return None def load_workspace_pipeline_spec(workspace: Path): from tooling.pipeline_spec import PipelineSpec try: repo_root = find_repo_root(workspace) except FileNotFoundError: repo_root = Path(__file__).resolve().parents[1] lock_path = workspace / "PIPELINE.lock.md" if not lock_path.exists(): return None pipeline_name = "" try: for raw in lock_path.read_text(encoding="utf-8", errors="ignore").splitlines(): line = raw.strip() if line.startswith("pipeline:"): pipeline_name = line.split(":", 1)[1].strip() break except Exception: return None if not pipeline_name: return None spec_path = resolve_pipeline_spec_path(repo_root=repo_root, pipeline_value=pipeline_name) if spec_path is None: return None try: return PipelineSpec.load(spec_path) except Exception: return None def pipeline_query_defaults(workspace: Path) -> dict[str, Any]: spec = load_workspace_pipeline_spec(workspace) return dict(spec.query_defaults) if spec is not None else {} def pipeline_quality_contract(workspace: Path) -> dict[str, Any]: spec = load_workspace_pipeline_spec(workspace) return dict(spec.quality_contract) if spec is not None else {} def pipeline_quality_contract_value(workspace: Path, *keys: str, default: Any = None) -> Any: current: Any = pipeline_quality_contract(workspace) for key in keys: if not isinstance(current, dict): return default current = current.get(str(key)) if current is None: return default return current def pipeline_query_default(workspace: Path, key: str, default: Any = None) -> Any: spec = load_workspace_pipeline_spec(workspace) if spec is None: return default return spec.query_default(key, default) def pipeline_overridable_query_fields(workspace: Path) -> set[str]: spec = load_workspace_pipeline_spec(workspace) if spec is None: return set() return set(spec.overridable_query_fields) def pipeline_profile(workspace: Path) -> str: """Return the pipeline profile for a workspace. Reads PIPELINE.lock.md to get the pipeline name, then loads the pipeline spec file and reads its ``profile`` frontmatter field. Falls back to ``"default"`` if anything is missing. """ spec = load_workspace_pipeline_spec(workspace) if spec is None: return "default" return str(spec.profile or "default").strip() or "default" def latest_outline_state(workspace: Path) -> dict[str, Any]: path = Path(workspace).resolve() / "outline" / "outline_state.jsonl" records = [rec for rec in read_jsonl(path) if isinstance(rec, dict)] return dict(records[-1]) if records else {} def _materialize_missing_query_defaults(lines: list[str], query_defaults: dict[str, Any], *, allowed_fields: set[str] | None = None) -> list[str]: if not query_defaults: return lines existing_keys: set[str] = set() for raw in lines: stripped = raw.strip() if not stripped.startswith("- ") or ":" not in stripped: continue key = stripped[2:].split(":", 1)[0].strip().lower().replace(" ", "_").replace("-", "_") if key: existing_keys.add(key) additions: list[str] = [] for key, value in query_defaults.items(): norm_key = str(key or "").strip().lower().replace(" ", "_").replace("-", "_") if not norm_key or norm_key in existing_keys: continue if allowed_fields and norm_key not in allowed_fields: continue rendered = _render_query_scalar(value) if rendered is None: continue additions.append(f'- {norm_key}: "{rendered}"') if not additions: return lines out = list(lines) if out and out[-1].strip(): out.append("") out.extend(additions) return out def _render_query_scalar(value: Any) -> str | None: if value is None: return None if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float)): return str(value) if isinstance(value, str): text = value.strip() return text or None return None FILE:tooling/executor.py from __future__ import annotations import subprocess import sys from dataclasses import dataclass from pathlib import Path from tooling.common import ( UnitsTable, atomic_write_text, decisions_has_approval, ensure_dir, latest_outline_state, load_workspace_pipeline_spec, now_iso_seconds, parse_semicolon_list, set_decisions_approval, update_status_field, update_status_log, ) @dataclass(frozen=True) class RunResult: unit_id: str | None status: str message: str def _section_first_cutover_block_message(*, workspace: Path, outputs: list[str]) -> str | None: spec = load_workspace_pipeline_spec(workspace) if spec is None or str(spec.structure_mode or "").strip().lower() != "section_first": return None if "outline/outline_state.jsonl" not in outputs: return None latest = latest_outline_state(workspace) if not latest: return "Section-first cutover is not actionable yet: `outline/outline_state.jsonl` is missing or empty. Rerun `outline-refiner` after fixing the chapter/section layer." structure_phase = str(latest.get("structure_phase") or "").strip() h3_status = str(latest.get("h3_status") or "").strip().lower() reroute_target = str(latest.get("reroute_target") or "").strip() retry_budget = str(latest.get("retry_budget_remaining") or "").strip() reroute_reason = str(latest.get("reroute_reason") or "").strip() if structure_phase.lower() == "decomposed" and h3_status == "stable": return None missing: list[str] = [] for key in ("structure_phase", "h3_status", "approval_status", "reroute_target", "retry_budget_remaining"): if key not in latest: missing.append(key) continue if key in {"structure_phase", "h3_status"} and not str(latest.get(key) or "").strip(): missing.append(key) if missing: return ( "Section-first cutover cannot advance because the latest `outline_state.jsonl` record is missing " f"required fields: {', '.join(missing)}." ) reroute_label = reroute_target or "unknown" retry_label = retry_budget or "unknown" reason_suffix = f" Reason: {reroute_reason}." if reroute_reason else "" return ( "Section-first cutover is still blocked; " f"latest outline state has structure_phase={structure_phase or 'missing'}, " f"h3_status={h3_status or 'missing'}, reroute_target={reroute_label}, " f"retry_budget_remaining={retry_label}.{reason_suffix} Fix the reroute target explicitly, then rerun this unit." ) def _append_run_error(*, workspace: Path, unit_id: str, skill: str, kind: str, message: str, log_rel: str | None) -> None: """Append a short failure record to `output/RUN_ERRORS.md` (workspace-local). This is a human-facing error sink that survives reruns and makes BLOCKED states debuggable. """ try: ensure_dir(workspace / "output") out_path = workspace / "output" / "RUN_ERRORS.md" stamp = now_iso_seconds() log_hint = f" (log: `{log_rel}`)" if log_rel else "" line = f"- {stamp} `{unit_id}` `{skill}` `{kind}`: {message}{log_hint}" if out_path.exists() and out_path.stat().st_size > 0: prev = out_path.read_text(encoding="utf-8", errors="ignore").rstrip() + "\n" else: prev = "# Run errors\n\n" atomic_write_text(out_path, prev + line + "\n") except Exception: # Never let the runner crash while trying to log an error. return def run_one_unit( *, workspace: Path, repo_root: Path, strict: bool = False, auto_approve: set[str] | None = None, ) -> RunResult: units_path = workspace / "UNITS.csv" status_path = workspace / "STATUS.md" if not units_path.exists(): return RunResult(unit_id=None, status="ERROR", message=f"Missing {units_path}") table = UnitsTable.load(units_path) runnable_idx = _find_first_runnable(table) if runnable_idx is None: return RunResult(unit_id=None, status="IDLE", message="No runnable unit found") row = table.rows[runnable_idx] unit_id = row.get("unit_id", "").strip() skill = row.get("skill", "").strip() owner = row.get("owner", "").strip().upper() row["status"] = "DOING" table.save(units_path) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} DOING {skill}") auto_approve_set = {str(x or "").strip().upper() for x in (auto_approve or set()) if str(x or "").strip()} if owner == "HUMAN": checkpoint = row.get("checkpoint", "").strip() if checkpoint and decisions_has_approval(workspace / "DECISIONS.md", checkpoint): row["status"] = "DONE" table.save(units_path) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} DONE (HUMAN approved {checkpoint})") _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="DONE", message=f"HUMAN approved {checkpoint}") if checkpoint and checkpoint.upper() in auto_approve_set: set_decisions_approval(workspace / "DECISIONS.md", checkpoint, approved=True) row["status"] = "DONE" table.save(units_path) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} DONE (AUTO approved {checkpoint})") _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="DONE", message=f"AUTO approved {checkpoint}") row["status"] = "BLOCKED" table.save(units_path) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} BLOCKED (await HUMAN approval {checkpoint})") _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="BLOCKED", message=f"Await HUMAN approval {checkpoint} in DECISIONS.md") script_path = repo_root / "scripts" / "run.py" if not script_path.exists(): row["status"] = "BLOCKED" table.save(units_path) skill_md = "SKILL.md" update_status_log( status_path, ( f"{now_iso_seconds()} {unit_id} BLOCKED " f"(no script for {skill}; run manually per {skill_md} then mark DONE)" ), ) _refresh_status_checkpoint(status_path, table) return RunResult( unit_id=unit_id, status="BLOCKED", message=( f"No executable script for skill '{skill}'. " f"Run it manually by following `{skill_md}`, write the required outputs, " f"then mark the unit DONE (e.g., `python scripts/pipeline.py mark --workspace {workspace} --unit-id {unit_id} --status DONE`)." ), ) inputs = parse_semicolon_list(row.get("inputs")) raw_outputs = parse_semicolon_list(row.get("outputs")) outputs = [_strip_optional_marker(rel) for rel in raw_outputs] required_outputs = [outputs[i] for i, rel in enumerate(raw_outputs) if not rel.strip().startswith("?")] checkpoint = row.get("checkpoint", "").strip() cmd = [ sys.executable, str(script_path), "--workspace", str(workspace), "--unit-id", unit_id, "--inputs", ";".join(inputs), "--outputs", ";".join(outputs), "--checkpoint", checkpoint, ] log_rel = f"output/unit_logs/{unit_id}.{skill}.log" log_path = workspace / log_rel try: completed = subprocess.run(cmd, check=False, capture_output=True, text=True) if completed.stdout or completed.stderr or completed.returncode != 0: ensure_dir(log_path.parent) body = [ f"# Unit log\n", f"- unit_id: {unit_id}\n", f"- skill: {skill}\n", f"- exit: {completed.returncode}\n", f"- cmd: {' '.join(cmd)}\n", "\n## stdout\n\n", (completed.stdout or "(empty)") + "\n", "\n## stderr\n\n", (completed.stderr or "(empty)") + "\n", ] atomic_write_text(log_path, "".join(body)) except Exception as exc: # pragma: no cover row["status"] = "BLOCKED" table.save(units_path) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} BLOCKED (exec error)") _append_run_error( workspace=workspace, unit_id=unit_id, skill=skill, kind="exec_error", message=f"{type(exc).__name__}: {exc}", log_rel=None, ) _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="BLOCKED", message=str(exc)) missing = [rel for rel in required_outputs if rel and not (workspace / rel).exists()] if completed.returncode == 0 and not missing: if strict: from tooling.quality_gate import check_unit_outputs, write_quality_report try: issues = check_unit_outputs(skill=skill, workspace=workspace, outputs=outputs) except Exception as exc: # pragma: no cover from tooling.quality_gate import QualityIssue issues = [ QualityIssue( code="quality_gate_exception", message=f"Quality gate crashed: {type(exc).__name__}: {exc}", ) ] # Avoid confusing stale QUALITY_GATE.md after a successful run. report_path = workspace / "output" / "QUALITY_GATE.md" if issues or report_path.exists(): write_quality_report(workspace=workspace, unit_id=unit_id, skill=skill, issues=issues) if issues: row["status"] = "BLOCKED" table.save(units_path) rel_report = str((workspace / "output" / "QUALITY_GATE.md").relative_to(workspace)) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} BLOCKED (quality gate: {rel_report})") _refresh_status_checkpoint(status_path, table) reroute_hint = _reroute_hint(workspace) return RunResult( unit_id=unit_id, status="BLOCKED", message=f"Quality gate failed; see {rel_report}" + (f"; {reroute_hint}" if reroute_hint else ""), ) cutover_block = _section_first_cutover_block_message(workspace=workspace, outputs=outputs) if cutover_block: row["status"] = "BLOCKED" table.save(units_path) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} BLOCKED (section-first cutover)") _append_run_error( workspace=workspace, unit_id=unit_id, skill=skill, kind="section_first_cutover", message=cutover_block, log_rel=log_rel if log_path.exists() else None, ) _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="BLOCKED", message=cutover_block) row["status"] = "DONE" table.save(units_path) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} DONE {skill}") _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="DONE", message="OK") row["status"] = "BLOCKED" table.save(units_path) if missing: update_status_log(status_path, f"{now_iso_seconds()} {unit_id} BLOCKED (missing outputs: {', '.join(missing)})") _append_run_error( workspace=workspace, unit_id=unit_id, skill=skill, kind="missing_outputs", message=f"Missing outputs: {', '.join(missing)}", log_rel=log_rel if log_path.exists() else None, ) _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="BLOCKED", message=f"Missing outputs: {', '.join(missing)}" + (f"; see {log_rel}" if log_path.exists() else "")) update_status_log(status_path, f"{now_iso_seconds()} {unit_id} BLOCKED (script failed)") _append_run_error( workspace=workspace, unit_id=unit_id, skill=skill, kind="script_failed", message=f"Skill script failed (exit {completed.returncode})", log_rel=log_rel if log_path.exists() else None, ) _refresh_status_checkpoint(status_path, table) return RunResult(unit_id=unit_id, status="BLOCKED", message=f"Skill script failed (exit {completed.returncode})" + (f"; see {log_rel}" if log_path.exists() else "")) def _find_first_runnable(table: UnitsTable) -> int | None: status_ok = {"DONE", "SKIP"} unit_by_id = {row.get("unit_id", ""): row for row in table.rows} for idx, row in enumerate(table.rows): if row.get("status", "").strip().upper() not in {"TODO", "BLOCKED"}: continue deps = parse_semicolon_list(row.get("depends_on")) if not deps: return idx deps_done = True for dep_id in deps: dep = unit_by_id.get(dep_id) if not dep: deps_done = False break if dep.get("status", "").strip().upper() not in status_ok: deps_done = False break if deps_done: return idx return None def _refresh_status_checkpoint(status_path: Path, table: UnitsTable) -> None: checkpoint = _compute_current_checkpoint(table) update_status_field(status_path, "Current checkpoint", checkpoint) def _compute_current_checkpoint(table: UnitsTable) -> str: for row in table.rows: if row.get("status", "").strip().upper() not in {"DONE", "SKIP"}: return (row.get("checkpoint") or "").strip() or "C0" return "DONE" def invalidate_downstream_units(table: UnitsTable, *, root_unit_id: str) -> list[str]: """Reset all transitive downstream dependents of `root_unit_id` to TODO. This is used when a previously satisfied upstream unit is reopened for rerun. Keeping downstream units as DONE would otherwise leave stale artifacts in place and make later `run` invocations stop too early. """ root = str(root_unit_id or "").strip() if not root: return [] direct_children: dict[str, list[dict[str, str]]] = {} for row in table.rows: unit_id = str(row.get("unit_id") or "").strip() for dep in parse_semicolon_list(row.get("depends_on")): direct_children.setdefault(dep, []).append(row) affected: list[str] = [] seen: set[str] = set() stack = [root] while stack: current = stack.pop() for child in direct_children.get(current, []): child_id = str(child.get("unit_id") or "").strip() if not child_id or child_id in seen: continue seen.add(child_id) if str(child.get("status") or "").strip().upper() != "TODO": child["status"] = "TODO" affected.append(child_id) stack.append(child_id) return affected def _strip_optional_marker(relpath: str) -> str: relpath = (relpath or "").strip() if relpath.startswith("?"): return relpath[1:].strip() return relpath def _reroute_hint(workspace: Path) -> str: path = workspace / "output" / "REROUTE_STATE.json" if not path.exists() or path.stat().st_size <= 0: return "" try: import json data = json.loads(path.read_text(encoding="utf-8", errors="ignore") or "{}") except Exception: return "" if not isinstance(data, dict): return "" target = str(data.get("reroute_target") or "").strip() status = str(data.get("status") or "").strip() phase = str(data.get("structure_phase") or "").strip() h3 = str(data.get("h3_status") or "").strip() reason = str(data.get("reroute_reason") or "").strip() if not any([target, status, phase, h3]): return "" parts = [] if status: parts.append(f"reroute_status={status}") if target: parts.append(f"reroute_target={target}") if phase: parts.append(f"structure_phase={phase}") if h3: parts.append(f"h3_status={h3}") if reason: parts.append(f"reason={reason}") return ", ".join(parts) FILE:tooling/ideation.py from __future__ import annotations import json import re from dataclasses import asdict, dataclass, is_dataclass from pathlib import Path from typing import Any, Iterable from tooling.common import ( atomic_write_text, ensure_dir, load_workspace_pipeline_spec, pipeline_overridable_query_fields, pipeline_query_default, read_jsonl, ) DEFAULT_IDEA_RUBRIC: list[tuple[str, float, str]] = [ ("discussion_worthiness", 0.24, "is this direction worth a serious PI/PhD discussion right now?"), ("academic_value", 0.22, "could it open a meaningful paper/thesis-worthy research angle?"), ("evidence_grounding", 0.18, "can we point to concrete literature tensions rather than pure speculation?"), ("direction_distinctness", 0.16, "is it genuinely different from neighboring directions, not just a wording variant?"), ("first_probe_clarity", 0.10, "is there a plausible low-cost first probe that would teach us something?"), ("thesis_potential", 0.10, "does it plausibly grow beyond a one-off note into a stronger line of work?"), ] STOPWORDS = { "a", "an", "and", "the", "of", "to", "for", "in", "on", "with", "via", "by", "from", "or", "work", "works", "study", "studies", "survey", "review", "benchmarks", "benchmark", "evaluation", "agents", "agent", "model", "models", "llm", "large", "language", "using", "based", } CLUSTER_AXIS_HINTS: list[tuple[list[str], list[str], str, str]] = [ (["agent loop", "action"], ["observability granularity", "action-space design", "tool/environment boundary"], "mechanism", "Could sharpen how we think about the basic agent loop rather than adding yet another full stack."), (["tool", "orchestration", "interface"], ["tool routing", "permission model", "API reliability assumptions"], "systems", "Could clarify whether interface design is the real bottleneck behind seemingly agent-level gains."), (["planning", "reasoning"], ["search depth", "verification loop", "partial-observability handling"], "mechanism", "Could turn vague talk about reasoning quality into a sharper research question about what actually drives robust planning."), (["memory", "retrieval", "rag"], ["retrieval policy", "state summarization cadence", "memory horizon"], "mechanism", "Could reveal when memory is a genuine capability lever versus a reporting convenience."), (["self-improvement", "adaptation"], ["feedback type", "adaptation horizon", "evaluation signal quality"], "adaptation", "Could distinguish genuine improvement from evaluation-sensitive self-optimization."), (["multi-agent", "coordination"], ["role specialization", "communication budget", "aggregation rule"], "coordination", "Could separate coordination benefits from simple redundancy or verification effects."), (["benchmark", "evaluation"], ["metric choice", "task slice design", "failure-analysis lens"], "evaluation", "Could make evaluation choices themselves a research object instead of hidden background assumptions."), (["safety", "security", "governance"], ["threat model", "monitoring granularity", "permission boundary"], "governance", "Could connect technical agent behaviors to deployment-relevant risk questions in a more principled way."), ] GENERIC_LIMITATION_PATTERNS = [ "abstract-level evidence only", "validate assumptions", "full paper", "before relying on this as key evidence", ] BENCHMARK_HINTS = [ "HotpotQA", "FEVER", "ALFWorld", "WebShop", "HumanEval", "WebArena", "AgentBench", "GSM8K", "MATH", "Wikipedia", "wiki", "API", "question answering", "fact verification", ] AXIS_INSIGHT_LIBRARY: dict[str, dict[str, str]] = { "observability granularity": { "question": "whether several agent-loop gains are really planner gains, or whether they mostly come from changing what the planner gets to observe and when", "confound": "planner quality and broader agent competence", "thesis": "Several agent-loop gains remain hard to interpret because papers often improve observation access at the same time they improve the planner; before crediting planner depth, we should ask what the system was allowed to see.", "insight": "A convincing result would not just move aggregate score. It would show which published gains survive once observation access is fixed, ideally producing a regime map of when observability—not planner depth—changes the failure story.", "contribution_shape": "Could yield a causal-attribution result plus a reporting rule for agent-loop papers: claims about planning quality should specify and control observation access.", "demotion": "This direction weakens sharply if the strongest prior work already holds observation access fixed while planner quality changes and still reports the same gain pattern.", "kill_signal": "an anchor paper already fixes observation access while varying planner quality and the main conclusion still survives", "missing_piece": "What is missing is a fixed-interface, fixed-budget comparison that varies only observation access and tracks whether the failure taxonomy—not just average score—changes.", "program_kind": "causal attribution", "time_to_clarity": "fast", "priority_note": "Fastest to falsify on public tasks, and the answer would immediately change how several agent-loop results are interpreted.", "reading_extract": "what the agent sees at each step, which ablations vary planner depth, and whether the action interface stays fixed", "title": "Observability granularity vs planner depth", }, "action-space design": { "question": "whether robustness comes from better reasoning or from giving the agent a cleaner action interface", "confound": "the shape of the action vocabulary and interface design", "thesis": "Some agent-loop gains may be interface gains in disguise: when action vocabularies become cleaner or narrower, papers can attribute robustness to reasoning improvements that partly come from action-space design.", "insight": "A useful result would show whether the same nominal planner still behaves very differently once the action space is widened, normalized, or made less ergonomic.", "contribution_shape": "Could produce an action-space normalization protocol and a clearer account of which agent claims survive once interface ergonomics are controlled.", "demotion": "This direction weakens if current papers already compare equivalent action spaces and still obtain the same ranking.", "kill_signal": "the key anchor papers already normalize action vocabularies or API surfaces and still see the same ordering", "missing_piece": "What is missing is an action-space-normalized comparison that keeps planner prompts fixed while changing only interface granularity and affordances.", "program_kind": "interface normalization", "time_to_clarity": "medium", "priority_note": "Interesting and still thesis-relevant, but less immediate than observability because interface normalization usually requires more careful task redesign.", "reading_extract": "how many actions are available, how semantically aligned the actions are, and whether interface cleanup happens together with reasoning changes", "title": "Action-space design or agent competence?", }, "tool/environment boundary": { "question": "whether reliability depends more on boundary design between tool and environment than on the nominal reasoning loop", "confound": "tool abstraction and environment modeling", "thesis": "A number of agent results may really be about boundary design: where the system stops reasoning internally and starts delegating to tools can change reliability without changing the nominal planner much.", "insight": "A strong result would show that several agent-level conclusions are actually boundary-design conclusions once tool abstraction is normalized.", "contribution_shape": "Could yield a boundary-design taxonomy and a cleaner experimental recipe for separating planner quality from tool/interface scaffolding.", "demotion": "This direction weakens if changing the boundary carefully barely affects the failure story.", "kill_signal": "careful tool-boundary changes leave both the metric and the qualitative failure modes essentially unchanged", "missing_piece": "What is missing is a comparison that keeps task semantics fixed while moving the tool/environment boundary in a controlled way.", "program_kind": "systems boundary", "time_to_clarity": "medium", "priority_note": "Valuable as a systems-facing thesis line, but slower to turn into a decisive first readout than the lead mechanism questions.", "reading_extract": "where tool calls begin, what state is exposed to the planner, and whether environment modeling changes together with the tool boundary", "title": "Where the tool boundary really matters", }, "search depth": { "question": "whether apparent planning gains reflect deeper search or simply more inference-time budget to recover from weak initial choices", "confound": "inference-time compute budget", "thesis": "Many planning results still bundle depth with budget: deeper search often spends more tokens, branches, or retries, so it remains unclear whether depth changes reasoning quality or just buys more recovery opportunities.", "insight": "A convincing result would show whether depth changes the nature of planning failures under a matched budget, rather than merely delaying failure by spending more compute.", "contribution_shape": "Could produce a compute-normalized planning benchmark slice and a regime map for when search depth matters beyond extra inference budget.", "demotion": "This direction weakens if prior work already equalizes budget and still finds depth-specific gains.", "kill_signal": "the anchor papers already normalize token or wall-clock budget and the depth advantage remains intact", "missing_piece": "What is missing is a compute-normalized study that holds token or wall-clock budget fixed while varying depth or branching, then checks whether failure modes actually change.", "program_kind": "budget-normalized mechanism", "time_to_clarity": "medium", "priority_note": "Still thesis-sized, but it ranks behind observability because compute normalization is harder to defend and easier for prior work to have addressed already.", "reading_extract": "the reported depth or branching settings, the effective compute budget, and whether shallow baselines were budget-matched", "title": "Search depth or compute budget?", }, "verification loop": { "question": "whether verification contributes a distinct reasoning mechanism or mostly acts as expensive redundancy", "confound": "extra compute and repeated checking", "thesis": "Verification-heavy pipelines may look stronger because they retry, re-check, or filter more often, not necessarily because they add a distinct reasoning mechanism.", "insight": "A useful result would separate verification as a mechanism from verification as a compute-expensive retry policy.", "contribution_shape": "Could produce a cleaner protocol for comparing verification loops under matched retry or compute budgets.", "demotion": "This direction weakens if verification can be replaced by simple repetition with little change in behavior.", "kill_signal": "repeat-sampling or retry baselines already match the reported verification gain", "missing_piece": "What is missing is a matched-budget comparison between explicit verification and simple repetition or reranking baselines.", "program_kind": "verification audit", "time_to_clarity": "medium", "priority_note": "Decision-relevant when the group suspects redundancy masquerading as reasoning, though the probe is less clean than the top two directions.", "reading_extract": "how many extra passes are spent on checking, whether retry baselines exist, and which errors verification actually fixes", "title": "Verification or just expensive redundancy?", }, "partial-observability handling": { "question": "whether planning systems fail because they reason badly or because they are brittle under missing state information", "confound": "state visibility", "thesis": "Some planning failures may be epistemic before they are algorithmic: systems can look like weak planners when they are actually brittle under missing state information.", "insight": "A strong result would show whether improved visibility changes the failure taxonomy more than planner changes do.", "contribution_shape": "Could produce a planning-under-partial-observability benchmark slice and a clearer division between epistemic and algorithmic failure modes.", "demotion": "This direction weakens if stronger visibility barely changes the failure pattern.", "kill_signal": "improved visibility leaves both success rates and error types almost unchanged", "missing_piece": "What is missing is a task slice that varies only state visibility while keeping planner scaffolding steady.", "program_kind": "epistemic failure analysis", "time_to_clarity": "medium", "priority_note": "Useful if the group wants a deeper failure-analysis program, but it is currently less concrete than the lead directions.", "reading_extract": "what information is hidden, which observations are restored by tools, and whether planner quality is tested separately from visibility", "title": "Is planning failure really an observability problem?", }, "retrieval policy": { "question": "whether memory gains come from better stored knowledge or from better decisions about when and how retrieval is triggered", "confound": "memory content versus retrieval timing", "thesis": "Reported memory gains are often hard to interpret because memory content and retrieval triggers move together; some improvements may be protocol effects about when retrieval happens, not capability effects about what memory stores.", "insight": "A convincing result would separate memory-content effects from retrieval-trigger effects and show whether the interpretation of current RAG-style gains changes once trigger policy is normalized.", "contribution_shape": "Could yield a cleaner memory-evaluation protocol and a reusable result on when retrieval policy, rather than memory content, is the true hidden variable.", "demotion": "This direction weakens if current work already varies retrieval policy independently of memory content and sees the same story.", "kill_signal": "the main memory papers already keep the memory store fixed while varying retrieval triggers or timing and the conclusion does not move", "missing_piece": "What is missing is a fixed-memory comparison that varies retrieval trigger and timing only, then checks whether the gain survives and which errors it actually removes.", "program_kind": "protocol sensitivity", "time_to_clarity": "medium", "priority_note": "Different enough from the planning directions to stay in the lead set, but it ranks lower because the current evidence is thinner and more protocol-sensitive.", "reading_extract": "what is stored in memory, when retrieval fires, what retrieval budget is allowed, and whether trigger policy is ever ablated independently", "title": "Retrieval policy or memory content?", }, "state summarization cadence": { "question": "whether summarization helps because it improves reasoning or because it selectively compresses away troublesome state information", "confound": "what gets forgotten versus what gets highlighted", "thesis": "Summarization may help less because the agent reasons better and more because the system filters state in a favorable way; cadence choices can quietly redefine what information survives.", "insight": "A useful result would show whether different summarization cadences preserve the same conclusions and failure taxonomy once memory content is held steady.", "contribution_shape": "Could produce a summarization-control protocol for long-horizon agents and a clearer account of how state compression alters conclusions.", "demotion": "This direction weakens if different summarization cadences preserve the same conclusions and failure taxonomy.", "kill_signal": "changing summarization cadence leaves both retrieval behavior and downstream errors largely unchanged", "missing_piece": "What is missing is a fixed-memory, fixed-task comparison that varies only summarization cadence and records what state is lost or amplified.", "program_kind": "state compression", "time_to_clarity": "medium", "priority_note": "Promising but currently less mature than retrieval-policy framing because the concrete prior-work hooks are thinner.", "reading_extract": "what gets summarized away, how often summaries are rewritten, and whether failure cases correlate with state compression choices", "title": "What summarization cadence is really changing", }, "memory horizon": { "question": "whether longer memory helps because more context is useful or because evaluations reward persistence over selectivity", "confound": "context length and evaluation design", "thesis": "Longer memory can look like better capability even when the benchmark mainly rewards persistence or repeated access to earlier state, so horizon effects may partly be evaluation-design effects.", "insight": "A convincing result would show whether extending horizon changes task framing and error patterns, not just aggregate score.", "contribution_shape": "Could produce a regime map for when horizon length matters, versus when selective retrieval or better task design explains the gain.", "demotion": "This direction weakens if memory horizon has little effect once retrieval policy is controlled.", "kill_signal": "memory-length changes stop mattering once retrieval triggers or evaluation design are normalized", "missing_piece": "What is missing is a horizon study that matches retrieval policy and benchmark framing while varying how much state is retained.", "program_kind": "regime mapping", "time_to_clarity": "medium", "priority_note": "Conceptually useful, but currently less decisive than retrieval-policy framing for a first discussion round.", "reading_extract": "whether longer context actually changes retrieval choices, and whether the benchmark rewards persistence more than selective recall", "title": "Does longer memory really help?", }, } @dataclass(frozen=True) class IdeaSignal: signal_id: str cluster: str direction_type: str theme: str claim_or_observation: str tension: str missing_piece: str possible_axis: str academic_value: str evidence_confidence: str paper_ids: list[str] @dataclass(frozen=True) class DirectionCard: direction_id: str cluster: str direction_type: str title: str focus_axis: str main_confound: str program_kind: str contribution_shape: str time_to_clarity: str one_line_thesis: str why_interesting: str literature_suggests: list[str] closest_prior_gap: list[str] missing_piece: str possible_variants: list[str] academic_value: str first_probes: list[str] what_counts_as_insight: str weakness_conditions: list[str] kill_criteria: list[str] what_would_change_mind: list[str] best_fit: str why_this_ranks_here: str evidence_confidence: str paper_ids: list[str] signal_ids: list[str] anchor_reading_notes: list[dict[str, str]] @dataclass(frozen=True) class ScreenedDirection: direction_id: str cluster: str direction_type: str title: str total_score: float discussion_worthiness: int academic_value_score: int evidence_grounding: int direction_distinctness: int first_probe_clarity: int thesis_potential: int recommendation: str rationale: str def read_core_set(path: Path) -> list[dict[str, str]]: import csv if not path.exists(): return [] with path.open("r", encoding="utf-8", newline="") as handle: return [dict(row) for row in csv.DictReader(handle)] def write_jsonl(path: Path, rows: Iterable[dict[str, Any] | Any]) -> None: ensure_dir(path.parent) encoded: list[str] = [] for row in rows: value = asdict(row) if is_dataclass(row) else row encoded.append(json.dumps(value, ensure_ascii=False)) atomic_write_text(path, "\n".join(encoded).rstrip() + ("\n" if encoded else "")) def write_json(path: Path, data: Any) -> None: ensure_dir(path.parent) atomic_write_text(path, json.dumps(data, ensure_ascii=False, indent=2).rstrip() + "\n") def uniq_keep_order(items: Iterable[str]) -> list[str]: out: list[str] = [] seen: set[str] = set() for item in items: value = str(item or "").strip() if not value or value in seen: continue seen.add(value) out.append(value) return out def slugify(text: str) -> str: raw = re.sub(r"[^A-Za-z0-9]+", "-", str(text or "").strip().lower()) return raw.strip("-") or "x" def clean_text(text: str, *, limit: int = 220) -> str: s = str(text or "").strip() s = s.replace("\n", " ") s = re.sub(r"\s+", " ", s) s = s.replace("|", ", ") s = s.strip(" \"'`") if len(s) <= limit: return s clipped = s[:limit].rsplit(" ", 1)[0].strip() return clipped if clipped else s[:limit].strip() def clean_sentence(text: str, *, limit: int = 180) -> str: s = clean_text(text, limit=max(limit * 2, limit)) if len(s) <= limit: return s window = s[:limit + 40] pieces = re.split(r'(?<=[.!?;])\s+', window) acc: list[str] = [] total = 0 for piece in pieces: piece = piece.strip() if not piece: continue nxt = total + len(piece) + (1 if acc else 0) if nxt > limit and acc: break acc.append(piece) total = nxt if total >= int(limit * 0.65): break if acc: return " ".join(acc).strip() clipped = s[:limit].rsplit(" ", 1)[0].strip() return clipped if clipped else s[:limit].strip() def markdown_table(headers: list[str], rows: list[list[str]]) -> str: head = "| " + " | ".join(headers) + " |" sep = "| " + " | ".join(["---"] * len(headers)) + " |" body = ["| " + " | ".join(str(cell).replace("\n", " ").strip() for cell in row) + " |" for row in rows] return "\n".join([head, sep] + body) def write_markdown(path: Path, text: str) -> None: ensure_dir(path.parent) atomic_write_text(path, text.rstrip() + "\n") def extract_goal_from_goal_md(path: Path) -> str: if not path.exists(): return "research ideas" lines = [ln.strip() for ln in path.read_text(encoding="utf-8", errors="ignore").splitlines() if ln.strip() and not ln.startswith("#")] return lines[0] if lines else "research ideas" def _idea_workspace_from_brief(path: Path) -> Path | None: try: return path.resolve().parents[2] except Exception: return None def _require_positive_float(value: Any, *, field_name: str) -> float: try: parsed = float(value) except Exception as exc: raise ValueError(f"Missing or invalid ideation contract field: {field_name}") from exc if parsed <= 0: raise ValueError(f"Missing or invalid ideation contract field: {field_name}") return parsed def _query_int_override(workspace: Path, key: str, default: int) -> int: queries_path = workspace / "queries.md" normalized = str(key or "").strip().lower().replace(" ", "_").replace("-", "_") if not normalized: return int(default) if normalized not in pipeline_overridable_query_fields(workspace): return int(default) if not queries_path.exists(): return int(default) try: for raw in queries_path.read_text(encoding="utf-8", errors="ignore").splitlines(): line = raw.strip() if not line.startswith("- ") or ":" not in line: continue key, value = line[2:].split(":", 1) key = key.strip().lower().replace(" ", "_").replace("-", "_") if key != normalized: continue cleaned = value.split("#", 1)[0].strip().strip('"').strip("'") if not cleaned: raise ValueError(f"Invalid ideation override: `{normalized}` is empty in queries.md.") try: parsed = int(cleaned) except Exception as exc: raise ValueError(f"Invalid ideation override: `{normalized}` must be an integer in queries.md.") from exc if parsed <= 0: raise ValueError(f"Invalid ideation override: `{normalized}` must be a positive integer in queries.md.") return parsed except Exception: raise return int(default) def _validate_score_weights(value: Any, *, field_name: str) -> dict[str, float]: required = { "discussion_worthiness": 0.24, "academic_value": 0.22, "evidence_grounding": 0.18, "direction_distinctness": 0.16, "first_probe_clarity": 0.10, "thesis_potential": 0.10, } if not isinstance(value, dict): raise ValueError(f"Missing or invalid ideation contract field: {field_name}") weights: dict[str, float] = {} for key in required: if key not in value: raise ValueError(f"Missing or invalid ideation contract field: {field_name}.{key}") weights[key] = _require_positive_float(value.get(key), field_name=f"{field_name}.{key}") total = sum(weights.values()) if total <= 0: raise ValueError(f"Missing or invalid ideation contract field: {field_name}") return {key: round(weight / total, 6) for key, weight in weights.items()} def _validate_diversity_axes(value: Any, *, field_name: str) -> list[str]: allowed = {"cluster", "direction_type", "program_kind"} if not isinstance(value, list): raise ValueError(f"Missing or invalid ideation contract field: {field_name}") axes: list[str] = [] seen: set[str] = set() for item in value: axis = str(item or "").strip().lower().replace("-", "_") if not axis: continue if axis not in allowed: raise ValueError(f"Missing or invalid ideation contract field: {field_name}.{axis}") if axis in seen: continue seen.add(axis) axes.append(axis) if not axes: raise ValueError(f"Missing or invalid ideation contract field: {field_name}") return axes def resolve_idea_contract(workspace: Path) -> dict[str, Any]: spec = load_workspace_pipeline_spec(workspace) if spec is None: raise ValueError("Missing active pipeline contract.") query_defaults = dict(spec.query_defaults) quality_contract = dict(spec.quality_contract) def _require_positive_int(value: Any, *, field_name: str) -> int: try: parsed = int(value) except Exception as exc: raise ValueError(f"Missing or invalid ideation contract field: {field_name}") from exc if parsed <= 0: raise ValueError(f"Missing or invalid ideation contract field: {field_name}") return parsed signal_policy = quality_contract.get("signal_policy") or {} direction_policy = quality_contract.get("direction_policy") or {} screening_policy = quality_contract.get("screening_policy") or {} brief = parse_idea_brief(workspace / "output" / "trace" / "IDEA_BRIEF.md") focus_clusters = [str(x).strip() for x in (brief.get("focus_clusters") or []) if str(x).strip()] direction_pool_min_default = _require_positive_int(query_defaults.get("direction_pool_min"), field_name="query_defaults.direction_pool_min") direction_pool_max_default = _require_positive_int(query_defaults.get("direction_pool_max"), field_name="query_defaults.direction_pool_max") shortlist_size_default = _require_positive_int(query_defaults.get("idea_shortlist_size"), field_name="query_defaults.idea_shortlist_size") report_top_n_default = _require_positive_int(query_defaults.get("report_top_n"), field_name="query_defaults.report_top_n") idea_screen_top_n_default = _require_positive_int(query_defaults.get("idea_screen_top_n"), field_name="query_defaults.idea_screen_top_n") shortlist_min = _require_positive_int(direction_policy.get("shortlist_min"), field_name="quality_contract.direction_policy.shortlist_min") shortlist_max = _require_positive_int(direction_policy.get("shortlist_max"), field_name="quality_contract.direction_policy.shortlist_max") keep_min = _require_positive_int(direction_policy.get("keep_min"), field_name="quality_contract.direction_policy.keep_min") cluster_diversity_min = _require_positive_int(direction_policy.get("cluster_diversity_min"), field_name="quality_contract.direction_policy.cluster_diversity_min") lead_diversity_target = _require_positive_int(direction_policy.get("lead_diversity_target"), field_name="quality_contract.direction_policy.lead_diversity_target") lead_diversity_axes = _validate_diversity_axes( direction_policy.get("lead_diversity_axes"), field_name="quality_contract.direction_policy.lead_diversity_axes", ) signal_table_min = _require_positive_int(signal_policy.get("min_rows"), field_name="quality_contract.signal_policy.min_rows") keep_rank_max = _require_positive_int(screening_policy.get("keep_rank_max"), field_name="quality_contract.screening_policy.keep_rank_max") maybe_rank_max = _require_positive_int(screening_policy.get("maybe_rank_max"), field_name="quality_contract.screening_policy.maybe_rank_max") score_weights = _validate_score_weights( screening_policy.get("score_weights"), field_name="quality_contract.screening_policy.score_weights", ) direction_pool_min = _query_int_override(workspace, "direction_pool_min", direction_pool_min_default) direction_pool_max = _query_int_override(workspace, "direction_pool_max", direction_pool_max_default) shortlist_size = _query_int_override(workspace, "idea_shortlist_size", shortlist_size_default) report_top_n = _query_int_override(workspace, "report_top_n", report_top_n_default) idea_screen_top_n = _query_int_override(workspace, "idea_screen_top_n", idea_screen_top_n_default) if direction_pool_min > direction_pool_max: raise ValueError(f"Invalid ideation contract: direction_pool_min ({direction_pool_min}) exceeds direction_pool_max ({direction_pool_max}).") if shortlist_size < shortlist_min or shortlist_size > shortlist_max: raise ValueError( f"Invalid ideation contract: idea_shortlist_size ({shortlist_size}) must stay within " f"[{shortlist_min}, {shortlist_max}]." ) if report_top_n > shortlist_size: raise ValueError(f"Invalid ideation contract: report_top_n ({report_top_n}) exceeds idea_shortlist_size ({shortlist_size}).") if maybe_rank_max < keep_rank_max: raise ValueError( f"Invalid ideation contract: maybe_rank_max ({maybe_rank_max}) must be >= keep_rank_max ({keep_rank_max})." ) if idea_screen_top_n > direction_pool_max: raise ValueError(f"Invalid ideation contract: idea_screen_top_n ({idea_screen_top_n}) exceeds direction_pool_max ({direction_pool_max}).") return { "focus_clusters": focus_clusters, "direction_pool_min": direction_pool_min, "direction_pool_max": direction_pool_max, "idea_screen_top_n": idea_screen_top_n, "shortlist_size": shortlist_size, "report_top_n": report_top_n, "shortlist_min": shortlist_min, "shortlist_max": shortlist_max, "signal_table_min": signal_table_min, "keep_min": keep_min, "cluster_diversity_min": cluster_diversity_min, "lead_diversity_target": lead_diversity_target, "lead_diversity_axes": lead_diversity_axes, "keep_rank_max": keep_rank_max, "maybe_rank_max": maybe_rank_max, "score_weights": score_weights, } def parse_idea_brief(path: Path) -> dict[str, Any]: text = path.read_text(encoding="utf-8", errors="ignore") if path.exists() else "" out: dict[str, Any] = { "goal": extract_goal_from_goal_md(path), "focus_clusters": [], "query_buckets": [], "exclusions": [], "constraints": [], "targets": {}, } cur = None for raw in text.splitlines(): line = raw.rstrip() if line.startswith("## "): cur = line[3:].strip().lower() continue if cur == "goal" and line.startswith("- Topic:"): out["goal"] = line.split(":", 1)[1].strip() if cur in {"focus after c2", "focus lenses after c2"} and line.startswith("- Focus clusters:"): clusters = [x.strip() for x in line.split(":", 1)[1].split(";") if x.strip()] cleaned: list[str] = [] for item in clusters: low = item.lower() if "to be filled" in low or "fill after c2" in low or "placeholder" in low: continue cleaned.append(item) out["focus_clusters"] = cleaned if cur == "query buckets" and re.match(r"^\d+\.\s+", line): out["query_buckets"].append(re.sub(r"^\d+\.\s+", "", line).strip()) if cur in {"exclude terms", "exclusions"} and line.startswith("- "): value = line[2:].strip() if value and not value.lower().startswith("none"): out["exclusions"].append(value) if cur == "constraints" and line.startswith("- "): out["constraints"].append(line[2:].strip()) if cur == "targets" and line.startswith("- ") and ":" in line: key, value = line[2:].split(":", 1) out["targets"][key.strip().lower().replace(" ", "_")] = value.strip() return out def collect_note_index(path: Path) -> dict[str, dict[str, Any]]: out: dict[str, dict[str, Any]] = {} for rec in read_jsonl(path): if not isinstance(rec, dict): continue pid = str(rec.get("paper_id") or "").strip() if pid: out[pid] = rec return out def keywords_from_cluster(name: str) -> list[str]: toks = [t for t in re.findall(r"[A-Za-z0-9]+", str(name or "").lower()) if t not in STOPWORDS and len(t) > 2] return uniq_keep_order(toks) def score_note_to_cluster(cluster_name: str, note: dict[str, Any]) -> int: keys = keywords_from_cluster(cluster_name) blob_parts = [str(note.get("title") or "")] for field in ["summary_bullets", "limitations", "key_results", "method", "abstract"]: val = note.get(field) if isinstance(val, list): blob_parts.extend([str(x) for x in val]) elif isinstance(val, str): blob_parts.append(val) blob = " ".join(blob_parts).lower() return sum(1 for key in keys if key in blob) def map_notes_to_clusters(taxonomy_path: Path, notes_path: Path) -> dict[str, list[dict[str, Any]]]: import yaml taxonomy = yaml.safe_load(taxonomy_path.read_text(encoding="utf-8", errors="ignore")) if taxonomy_path.exists() else [] notes = [r for r in read_jsonl(notes_path) if isinstance(r, dict)] out: dict[str, list[dict[str, Any]]] = {} for top in taxonomy or []: if not isinstance(top, dict): continue for child in top.get("children") or []: if not isinstance(child, dict): continue name = str(child.get("name") or "").strip() if not name: continue scored: list[tuple[int, dict[str, Any]]] = [] for note in notes: s = score_note_to_cluster(name, note) if s > 0: scored.append((s, note)) scored.sort(key=lambda x: (-x[0], str(x[1].get("paper_id") or ""))) out[name] = [n for _, n in scored[:8]] return out def _cluster_profile(cluster: str) -> tuple[str, list[str], str]: low = cluster.lower() for keys, axes, direction_type, academic_value in CLUSTER_AXIS_HINTS: if any(key in low for key in keys): return direction_type, axes, academic_value return "research", ["assumption sensitivity", "failure analysis", "scope boundary"], "Could sharpen the way this sub-area is framed and compared." def _note_bullets(note: dict[str, Any], field: str) -> list[str]: val = note.get(field) if isinstance(val, list): return [clean_text(x, limit=220) for x in val if clean_text(x, limit=220)] if isinstance(val, str) and clean_text(val, limit=220): return [clean_text(val, limit=220)] return [] def _specific_limitations(note: dict[str, Any]) -> list[str]: items = _note_bullets(note, "limitations") specific = [] for item in items: low = item.lower() if any(pat in low for pat in GENERIC_LIMITATION_PATTERNS): continue specific.append(item) return specific or items def _evidence_confidence(notes: list[dict[str, Any]]) -> str: if not notes: return "low" if any(str(n.get("evidence_level") or "").strip() == "fulltext" for n in notes): return "medium-high" specific = sum( 1 for n in notes if _specific_limitations(n) and not any(p in (_specific_limitations(n)[0].lower()) for p in GENERIC_LIMITATION_PATTERNS) ) if len(notes) >= 3 and specific >= 2: return "medium" return "low-medium" def _axis_profile(axis: str, cluster: str) -> dict[str, str]: base = AXIS_INSIGHT_LIBRARY.get(axis, {}) confound = base.get("confound", "nearby design choices and evaluation framing") return { "question": base.get("question", f"whether {axis} is doing more conceptual work in {cluster.lower()} than current papers make explicit"), "confound": confound, "thesis": base.get("thesis", f"Current results in {cluster.lower()} may be hard to interpret because {axis} still moves together with {confound}."), "insight": base.get("insight", f"A meaningful result would show whether {axis} changes how we interpret the current literature rather than merely how we narrate it."), "contribution_shape": base.get("contribution_shape", f"Could turn {axis} into a cleaner explanatory variable for {cluster.lower()} rather than a background convenience."), "demotion": base.get("demotion", f"This direction would weaken if the strongest prior work already isolates {axis} cleanly."), "kill_signal": base.get("kill_signal", f"the strongest prior work already isolates {axis} against {confound}"), "missing_piece": base.get("missing_piece", f"What is missing is a cleaner comparison that varies {axis} while keeping {confound} fixed enough to change interpretation, not just presentation."), "program_kind": base.get("program_kind", "mechanism clarification"), "time_to_clarity": base.get("time_to_clarity", "medium"), "priority_note": base.get("priority_note", f"Worth discussion because it could turn {axis} into a sharper explanatory wedge for {cluster.lower()}."), "reading_extract": base.get("reading_extract", f"which variables change together with {axis}, what comparator is used, and whether any ablation already fixes {confound}"), "title": base.get("title", f"What {axis} is really doing"), } def _result_fact(note: dict[str, Any]) -> str: candidates = _note_bullets(note, "key_results") + _note_bullets(note, "summary_bullets") scored: list[str] = [] for cand in candidates: low = cand.lower() if any(token in low for token in ["pass@1", "accuracy", "success rate", "exact match", "f1", "outperform", "achiev", "%"]) or any(ch.isdigit() for ch in cand): scored.append(cand) chosen = scored[0] if scored else _claim_text(note) chosen = re.sub(r"^(for instance|for example|e\.g\.)[:,]?\s*", "", chosen, flags=re.IGNORECASE) low = chosen.lower() tasks = _extract_task_mentions(note) task_text = "/".join(tasks[:2]) if tasks else "reported setting" percents = re.findall(r"\d+(?:\.\d+)?%", chosen) if "pass@1" in low and percents: if len(percents) >= 2: return clean_text(f"{task_text}: {percents[0]} pass@1 vs {percents[1]} in the reported comparison.", limit=95) return clean_text(f"{task_text}: {percents[0]} pass@1 in the reported comparison.", limit=95) if "success rate" in low and len(percents) >= 2: baseline = " over imitation/RL baselines" if "imitation" in low or "reinforcement learning" in low else " in the reported comparison" return clean_text(f"{task_text}: {percents[0]}/{percents[1]} success-rate gains{baseline}.", limit=95) if len(percents) >= 2: return clean_text(f"{task_text}: {percents[0]} vs {percents[1]} in the reported comparison.", limit=95) if len(percents) == 1: return clean_text(f"{task_text}: reported result reaches {percents[0]} in the abstract.", limit=95) return clean_text(chosen, limit=95) def _metric_phrase(notes: list[dict[str, Any]]) -> str: blob = " ".join(_result_fact(note) for note in notes) low = blob.lower() if "pass@1" in low: return "pass@1 plus failure-type shifts" if "success rate" in low: return "success rate plus failure-type shifts" if "accuracy" in low: return "accuracy plus failure-type shifts" if "exact match" in low: return "exact match plus failure-type shifts" if "f1" in low: return "F1 plus failure-type shifts" if "%" in blob: return "the reported benchmark metric plus failure-type shifts" return "task success plus failure-type shifts" def _sentence_has_concrete_hook(text: str) -> bool: low = str(text or "").lower() if any(ch.isdigit() for ch in str(text or "")): return True if any(token in low for token in ["pass@1", "accuracy", "success rate", "exact match", "f1", "alfworld", "webshop", "hotpotqa", "fever", "humaneval", "webarena", "agentbench"]): return True return False def _extract_task_mentions(note: dict[str, Any]) -> list[str]: blob_parts = [] for field in ["key_results", "summary_bullets", "method", "abstract", "title"]: val = note.get(field) if isinstance(val, list): blob_parts.extend([str(x) for x in val]) elif isinstance(val, str): blob_parts.append(val) blob = " ".join(blob_parts) low = blob.lower() positions: list[tuple[int, str]] = [] for hint in BENCHMARK_HINTS: pos = low.find(hint.lower()) if pos >= 0: positions.append((pos, hint)) for match in re.finditer(r"\b(?:HumanEval|ALFWorld|WebShop|HotpotQA|FEVER|WebArena|AgentBench|GSM8K|MATH)\b", blob): positions.append((match.start(), match.group(0))) positions.sort(key=lambda item: (item[0], item[1])) return uniq_keep_order(name for _, name in positions) def _task_phrase(notes: list[dict[str, Any]]) -> str: tasks: list[str] = [] for note in notes: tasks.extend(_extract_task_mentions(note)) uniq = uniq_keep_order(tasks) if not uniq: return "a small public task slice" if len(uniq) == 1: return uniq[0] return "/".join(uniq[:2]) def _claim_text(note: dict[str, Any]) -> str: candidates = _note_bullets(note, "key_results") + _note_bullets(note, "summary_bullets") for cand in candidates: if len(cand.split()) >= 8: return cand return clean_text(note.get("method") or note.get("title") or "", limit=220) def _limitation_text(note: dict[str, Any], axis: str, cluster: str) -> str: limitations = _specific_limitations(note) if limitations and not any(pat in limitations[0].lower() for pat in GENERIC_LIMITATION_PATTERNS): return limitations[0] profile = _axis_profile(axis, cluster) return f"The paper still leaves unclear whether {axis} is genuinely responsible for the reported story in {cluster.lower()}, or whether that story is really driven by {profile['confound']}." def _anchor_notes(note_index: dict[str, dict[str, Any]], paper_ids: list[str]) -> list[dict[str, Any]]: return [note_index[pid] for pid in paper_ids if pid in note_index] def _paper_annotation(note: dict[str, Any], axis: str, cluster: str) -> str: title = clean_text(note.get("title") or note.get("paper_id") or "paper", limit=110) profile = _axis_profile(axis, cluster) result = _result_fact(note) return clean_sentence( f"{title}: {result} Open gap: {axis} still moves with {profile['confound']}.", limit=300, ) def _synthesis_annotation(notes: list[dict[str, Any]], axis: str, cluster: str) -> str: profile = _axis_profile(axis, cluster) task_phrase = _task_phrase(notes) return clean_sentence( f"Across the anchor papers, the live question is whether {axis} changes interpretation itself or merely rides along with {profile['confound']}, especially on {task_phrase}.", limit=260, ) def build_signal_rows(*, cluster: str, notes: list[dict[str, Any]]) -> list[IdeaSignal]: if not notes: return [] direction_type, axes, academic_value = _cluster_profile(cluster) evidence_confidence = _evidence_confidence(notes) anchors = notes[:3] paper_ids = uniq_keep_order(str(n.get("paper_id") or "").strip() for n in anchors) signals: list[IdeaSignal] = [] for idx, axis in enumerate(axes[:3], start=1): anchor = anchors[(idx - 1) % len(anchors)] claim = _claim_text(anchor) tension = _limitation_text(anchor, axis, cluster) profile = _axis_profile(axis, cluster) missing_piece = clean_sentence( f"What is still missing is a direction that isolates whether {axis} is the real explanatory variable in {cluster.lower()}, rather than leaving it entangled with {profile['confound']}.", limit=240, ) signals.append(IdeaSignal( signal_id=f"SIG-{slugify(cluster)[:16]}-{idx}", cluster=cluster, direction_type=direction_type, theme=f"{profile['title']} in {cluster}", claim_or_observation=claim, tension=clean_text(tension, limit=220), missing_piece=missing_piece, possible_axis=axis, academic_value=academic_value, evidence_confidence=evidence_confidence, paper_ids=paper_ids, )) return signals def signal_table_markdown(rows: list[IdeaSignal]) -> str: table_rows = [] for row in rows: table_rows.append([ row.signal_id, row.cluster, row.theme, row.claim_or_observation, row.tension, row.missing_piece, row.possible_axis, row.academic_value, row.evidence_confidence, ", ".join(row.paper_ids), ]) return "\n".join([ "# IDEA_SIGNAL_TABLE", "", "## Research signals", "", markdown_table(["Signal ID", "Cluster", "Theme", "Claim / observation", "Tension", "Missing piece", "Possible axis", "Academic value", "Confidence", "Paper IDs"], table_rows), "", ]) def _title_from_signal(signal: IdeaSignal) -> str: profile = _axis_profile(signal.possible_axis, signal.cluster) return clean_sentence(profile["title"], limit=72) def _best_fit(direction_type: str, axis: str) -> str: if direction_type == "governance": return "Best fit when the group wants a deployment-relevant reading of agent behavior rather than another internal benchmark story." if direction_type == "evaluation": return "Best fit when the discussion is really about what counts as convincing evidence, not just how to raise scores." if axis in {"search depth", "verification loop", "retrieval policy"}: return "Best fit for a PhD discussion that wants one sharper mechanism/confound question rather than a broad systems buildout." return "Best fit for PI/PhD discussions that want a thesis-worthy mechanism question rather than a generic benchmark wrapper." def signals_to_direction_cards(signals: list[IdeaSignal], *, note_index: dict[str, dict[str, Any]], focus_clusters: list[str], pool_min: int, pool_max: int) -> list[DirectionCard]: focus = {x.strip() for x in focus_clusters if str(x).strip()} cards: list[DirectionCard] = [] for idx, signal in enumerate(signals, start=1): profile = _axis_profile(signal.possible_axis, signal.cluster) anchors = _anchor_notes(note_index, signal.paper_ids) task_phrase = _task_phrase(anchors) metric_phrase = _metric_phrase(anchors) nearest = anchors[0] if anchors else {} nearest_title = clean_text((nearest.get("title") if isinstance(nearest, dict) else "") or (signal.paper_ids[0] if signal.paper_ids else signal.cluster), limit=90) nearest_task_phrase = _task_phrase([nearest]) if nearest else task_phrase literature_suggests = [_paper_annotation(note, signal.possible_axis, signal.cluster) for note in anchors[:2]] literature_suggests.append(_synthesis_annotation(anchors, signal.possible_axis, signal.cluster)) closest_prior_gap = [ clean_sentence( f"{nearest_title} is the closest prior anchor because it already reports concrete behavior on {nearest_task_phrase}. The unresolved point is whether those gains survive once {profile['confound']} is held fixed while {signal.possible_axis} is varied.", limit=220, ), clean_sentence( "The novelty test here is narrow, not rhetorical: if a strong anchor paper already runs that single-variable control, this direction should collapse quickly rather than stay alive as a vague confound story.", limit=220, ), ] one_line_thesis = clean_sentence(profile["thesis"], limit=230) why_interesting = clean_sentence( f"This is not just another benchmark wedge. It opens a {profile['program_kind']} line around {signal.possible_axis}: {profile['question']}.", limit=220, ) missing_piece = clean_sentence(profile["missing_piece"], limit=230) possible_variants = [ clean_sentence(f"Single-variable control on {task_phrase}: vary {signal.possible_axis} while holding {profile['confound']} fixed as far as the setup allows.", limit=210), clean_sentence("Replace score-only reporting with a failure-type comparison after the same intervention, so the result says more than whether the average went up.", limit=210), clean_sentence("Check whether the conclusion survives on one simple public task slice and one more tool- or environment-heavy setting before treating it as a general claim.", limit=210), ] first_probes = [ clean_sentence( f"Intervention: vary {signal.possible_axis} while holding {profile['confound']} as fixed as possible on {task_phrase}. Readout: {metric_phrase}. Decisive if the interpretation changes even after the control.", limit=220, ), clean_sentence( f"Prior-work audit: inspect {nearest_title} for any ablation that already fixes {profile['confound']}, and if the conclusion survives, demote this direction.", limit=180, ), ] what_counts_as_insight = clean_sentence(profile['insight'], limit=190) kill_criteria = [ clean_sentence(f"Kill quickly if {profile['kill_signal']}.", limit=170), clean_sentence(f"Kill if the first controlled probe leaves both {metric_phrase} and the failure taxonomy essentially unchanged.", limit=170), ] weakness_conditions = [ clean_sentence(f"This direction is weaker if changing {signal.possible_axis} mostly rescales the aggregate metric without changing which failure modes appear or disappear.", limit=175), clean_sentence(profile['demotion'], limit=170), ] anchor_reading_notes: list[dict[str, str]] = [] for note in anchors[:3]: title = clean_text(note.get("title") or note.get("paper_id") or "paper", limit=110) result = _result_fact(note) anchor_reading_notes.append({ "paper_title": title, "why_read": clean_sentence(f"Closest {signal.cluster.lower()} anchor with a concrete result hook on {task_phrase}: {result}", limit=180), "what_to_extract": clean_sentence(f"Extract the metric and comparator, then check whether {profile['reading_extract']}.", limit=180), "current_hook": clean_sentence(result, limit=180), "kill_signal": clean_sentence(f"Weaken this direction if the paper already shows {profile['kill_signal']}.", limit=180), }) why_this_ranks_here = clean_sentence(profile['priority_note'], limit=170) cards.append(DirectionCard( direction_id=f"DIR-{idx:03d}", cluster=signal.cluster, direction_type=signal.direction_type, title=_title_from_signal(signal), focus_axis=signal.possible_axis, main_confound=profile['confound'], program_kind=profile['program_kind'], contribution_shape=profile['contribution_shape'], time_to_clarity=profile['time_to_clarity'], one_line_thesis=one_line_thesis, why_interesting=why_interesting, literature_suggests=literature_suggests, closest_prior_gap=closest_prior_gap, missing_piece=missing_piece, possible_variants=possible_variants, academic_value=profile['contribution_shape'], first_probes=first_probes, what_counts_as_insight=what_counts_as_insight, weakness_conditions=weakness_conditions, kill_criteria=kill_criteria, what_would_change_mind=kill_criteria, best_fit=_best_fit(signal.direction_type, signal.possible_axis), why_this_ranks_here=why_this_ranks_here, evidence_confidence=signal.evidence_confidence, paper_ids=signal.paper_ids, signal_ids=[signal.signal_id], anchor_reading_notes=anchor_reading_notes, )) cards.sort(key=lambda c: (0 if c.cluster in focus else 1, c.cluster, c.program_kind, c.title)) target = max(pool_min, min(pool_max, len(cards))) return cards[:target] def direction_pool_markdown(cards: list[DirectionCard]) -> str: rows: list[list[str]] = [] for card in cards: rows.append([ card.direction_id, card.cluster, card.direction_type, card.program_kind, card.title, clean_sentence(card.one_line_thesis, limit=100), clean_sentence(card.why_interesting, limit=100), clean_sentence(card.missing_piece, limit=100), clean_sentence(" / ".join(card.possible_variants), limit=120), clean_sentence(card.academic_value, limit=90), clean_sentence(" / ".join(card.first_probes), limit=120), card.evidence_confidence, ", ".join(card.paper_ids), ]) return "\n".join([ "# IDEA_DIRECTION_POOL", "", "## Candidate research directions", "", markdown_table(["Direction ID", "Cluster", "Type", "Program", "Title", "One-line thesis", "Why interesting", "Missing piece", "Possible variants", "Academic value", "First probes", "Confidence", "Paper IDs"], rows), "", ]) def _thesis_potential(direction_type: str, cluster: str) -> int: if direction_type in {"mechanism", "coordination", "adaptation"}: return 5 if direction_type in {"systems", "governance"}: return 4 if "benchmark" in cluster.lower() or direction_type == "evaluation": return 3 return 4 def score_direction_cards( cards: list[DirectionCard], *, focus_clusters: list[str], keep_rank_max: int, maybe_rank_max: int, score_weights: dict[str, float] | None = None, ) -> list[ScreenedDirection]: focus = {x.strip() for x in focus_clusters if str(x).strip()} keep_rank_max = int(keep_rank_max) maybe_rank_max = int(maybe_rank_max) if keep_rank_max <= 0: raise ValueError("Invalid ideation scoring contract: keep_rank_max must be positive.") if maybe_rank_max < keep_rank_max: raise ValueError("Invalid ideation scoring contract: maybe_rank_max must be >= keep_rank_max.") weights = _validate_score_weights(score_weights or {}, field_name="score_direction_cards.score_weights") cluster_counts: dict[str, int] = {} program_counts: dict[str, int] = {} for card in cards: cluster_counts[card.cluster] = cluster_counts.get(card.cluster, 0) + 1 program_counts[card.program_kind] = program_counts.get(card.program_kind, 0) + 1 provisional: list[tuple[DirectionCard, float, int, int, int, int, int, int]] = [] for card in cards: discussion_worthiness = 5 if card.cluster in focus or card.time_to_clarity == "fast" else 4 academic_value_score = 5 if any(token in card.contribution_shape.lower() for token in ["rule", "protocol", "regime map", "causal"]) else 4 concrete_hooks = sum(1 for item in card.literature_suggests if _sentence_has_concrete_hook(item)) evidence_grounding = 5 if len(card.paper_ids) >= 3 and concrete_hooks >= 2 else 4 if len(card.paper_ids) >= 2 and concrete_hooks >= 1 else 3 if program_counts.get(card.program_kind, 0) == 1 and cluster_counts.get(card.cluster, 0) == 1: direction_distinctness = 5 elif program_counts.get(card.program_kind, 0) == 1 or cluster_counts.get(card.cluster, 0) == 1: direction_distinctness = 4 else: direction_distinctness = 3 probe_blob = " ".join(card.first_probes).lower() first_probe_clarity = 5 if all(token in probe_blob for token in ["intervention:", "readout:", "decisive if"]) else 4 if "intervention:" in probe_blob else 3 thesis_potential = _thesis_potential(card.direction_type, card.cluster) total = round( discussion_worthiness * weights["discussion_worthiness"] + academic_value_score * weights["academic_value"] + evidence_grounding * weights["evidence_grounding"] + direction_distinctness * weights["direction_distinctness"] + first_probe_clarity * weights["first_probe_clarity"] + thesis_potential * weights["thesis_potential"], 2, ) provisional.append((card, total, discussion_worthiness, academic_value_score, evidence_grounding, direction_distinctness, first_probe_clarity, thesis_potential)) provisional.sort(key=lambda row: (-row[1], 0 if row[0].time_to_clarity == "fast" else 1, row[0].cluster, row[0].title)) rows: list[ScreenedDirection] = [] for idx, (card, total, dw, av, eg, dd, fp, tp) in enumerate(provisional, start=1): recommendation = "keep" if idx <= keep_rank_max else "maybe" if idx <= maybe_rank_max else "drop" strengths: list[str] = [] if eg >= 5: strengths.append("concrete anchor evidence") if dd >= 5: strengths.append(f"distinct {card.program_kind} wedge") if fp >= 5: strengths.append("clean first probe") if av >= 5: strengths.append("thesis-sized payoff") if not strengths: strengths.append("discussion value") risk = "still abstract-first" if str(card.evidence_confidence).startswith("low") else f"main risk is controlling {card.main_confound} cleanly" rationale = clean_sentence(f"Strongest on {' + '.join(strengths[:2])}; {risk}.", limit=160) rows.append(ScreenedDirection( direction_id=card.direction_id, cluster=card.cluster, direction_type=card.direction_type, title=card.title, total_score=total, discussion_worthiness=dw, academic_value_score=av, evidence_grounding=eg, direction_distinctness=dd, first_probe_clarity=fp, thesis_potential=tp, recommendation=recommendation, rationale=rationale, )) return rows def screening_table_markdown(rows: list[ScreenedDirection]) -> str: data: list[list[str]] = [] for row in rows: data.append([ row.direction_id, row.cluster, row.direction_type, row.title, f"{row.total_score:.2f}", str(row.discussion_worthiness), str(row.academic_value_score), str(row.evidence_grounding), str(row.direction_distinctness), str(row.first_probe_clarity), str(row.thesis_potential), row.recommendation, row.rationale, ]) return "\n".join([ "# IDEA_SCREENING_TABLE", "", "## Discussion-first screening", "", markdown_table(["Direction ID", "Cluster", "Type", "Title", "Total", "Discussion", "Academic value", "Evidence", "Distinctness", "First probe", "Thesis potential", "Decision", "Rationale"], data), "", ]) def shortlist_markdown(records: list[dict[str, Any]]) -> str: lines = ["# IDEA_SHORTLIST", "", "## Prioritized directions", ""] for record in records: lines.extend([ f"### Direction {record['rank']}. {record['title']}", f"- Cluster: {record['cluster']}", f"- Type: {record['direction_type']}", f"- Focus axis: {record.get('focus_axis')}", f"- Program kind: {record.get('program_kind')}", f"- Main confound: {record.get('main_confound')}", f"- Time to clarity: {record.get('time_to_clarity')}", f"- One-line thesis: {record['one_line_thesis']}", f"- Why this is interesting: {record['why_interesting']}", f"- Why this ranks here: {record['why_this_ranks_here']}", f"- Contribution shape: {record.get('contribution_shape')}", "- What the literature already suggests:", ]) for item in record.get("literature_suggests") or []: lines.append(f" - {item}") lines.extend([ "- Closest prior work and why it does not settle the question:", ]) for item in record.get("closest_prior_gap") or []: lines.append(f" - {item}") lines.extend([ f"- What is still missing: {record['missing_piece']}", "- Possible variants:", ]) for item in record.get("possible_variants") or []: lines.append(f" - {item}") lines.extend([ f"- Why this could matter academically: {record['academic_value']}", "- First probes:", ]) for item in record.get("first_probes") or []: lines.append(f" - {item}") lines.extend([ f"- What would count as actual insight: {record['what_counts_as_insight']}", "- What would make this weak or unconvincing:", ]) for item in record.get("weakness_conditions") or []: lines.append(f" - {item}") lines.extend([ "- Quick kill criteria:", ]) for item in record.get("kill_criteria") or record.get("what_would_change_mind") or []: lines.append(f" - {item}") lines.extend([ f"- Best fit: {record['best_fit']}", f"- Evidence confidence: {record['evidence_confidence']}", "- Anchor papers: " + ", ".join(f"`{pid}`" for pid in record.get("paper_ids") or []), f"- Why prioritized now: {record['why_prioritized']}", "", ]) return "\n".join(lines) def shortlist_snapshot_table(records: list[dict[str, Any]]) -> str: rows = [] for rec in records: rows.append([ str(rec.get("rank") or ""), rec.get("title", ""), clean_sentence(rec.get("why_this_ranks_here", ""), limit=52), clean_sentence(rec.get("contribution_shape", rec.get("academic_value", "")), limit=56), clean_sentence((rec.get("kill_criteria") or rec.get("what_would_change_mind") or [""])[0], limit=54), ]) return markdown_table(["Rank", "Direction", "Why now", "If it survives", "Fast kill signal"], rows) def build_report_payload(*, topic: str, shortlist: list[dict[str, Any]], deferred: list[dict[str, Any]], trace_paths: dict[str, str]) -> dict[str, Any]: top = list(shortlist) takeaways: list[str] = [] if top: takeaways.append("The strongest directions are the ones most likely to change how existing results are interpreted, not just add another benchmark win.") takeaways.append("The current rank order reflects a tradeoff between time-to-clarity, thesis-sized payoff, and how concrete the nearest prior-work gap already looks.") program_kinds = uniq_keep_order(str(rec.get("program_kind") or "").strip() for rec in top) if len(program_kinds) >= 2: takeaways.append(f"The lead set is intentionally not one confound template repeated three times: it spans {', '.join(program_kinds[:3])} rather than a single explanatory mold.") else: takeaways.append("The lead set still sits in one tight neighborhood, so the next reading pass should stress-test whether those directions are genuinely distinct thesis lines.") if any(str(rec.get("evidence_confidence") or "").startswith("low") for rec in top): takeaways.append("Most anchors are still abstract-first, so the memo is best used to decide the next reading and falsification pass rather than to lock a project immediately.") else: takeaways.append("The evidence is already concrete enough for a serious PI/PhD discussion, though one deeper paper pass could still reshuffle the exact order.") lead = top[0] if top else {} lead_anchor = (lead.get("anchor_reading_notes") or [{}])[0] if top else {} lead_anchor_title = lead_anchor.get("paper_title") if isinstance(lead_anchor, dict) else "the first anchor paper" discussion_questions = [ "Which direction still survives once we ask for a single-variable control rather than a suggestive confound story?", "Which lead direction would still matter if the nearest prior work already addressed the obvious control we are worried about?", "Which candidate could plausibly turn into a thesis line with a reusable method, protocol, or regime map rather than a one-off empirical note?", "Which ranking would change most after one full-paper reading pass on the anchor set?", ] uncertainties = [ "The main remaining risk is novelty risk, not idea scarcity: closer reading may reveal that one lead direction is already settled by an existing control or ablation.", "Evidence confidence is still bounded by abstract-first notes for part of the lead set, so paper-specific details could either strengthen or kill a direction quickly.", "The memo is more reliable as a discussion and triage artifact than as a final commitment on exact project order.", ] next_steps = [] if top: next_steps.append(clean_sentence(f"Start with {lead.get('title')}: read {lead_anchor_title or 'the first anchor paper'} looking specifically for whether {lead.get('focus_axis')} is already isolated against {lead.get('main_confound')}.", limit=220)) first_probe = (lead.get("first_probes") or [""])[0] if first_probe: next_steps.append(clean_sentence(first_probe, limit=220)) next_steps.append(clean_sentence(f"Re-rank only after checking the quick kill criteria for {lead.get('title')} and at least one competing direction in the lead set.", limit=220)) else: next_steps.extend([ "Read the anchor papers for the current top direction in full and test whether the memo's stated hidden variable still looks unresolved.", "In the next PI/PhD discussion, use the ranking arguments and kill criteria to decide whether any direction deserves promotion into a sharper thesis candidate.", ]) return { "topic": topic, "takeaways": takeaways, "top_directions": top, "deferred_directions": deferred, "discussion_questions": discussion_questions, "uncertainties": uncertainties, "next_steps": next_steps, "trace_artifacts": trace_paths, } def report_markdown(payload: dict[str, Any]) -> str: top_dirs = payload.get("top_directions") or [] deferred_idx = 3 + len(top_dirs) discussion_idx = deferred_idx + 1 uncertainty_idx = deferred_idx + 2 next_idx = deferred_idx + 3 appendix_idx = deferred_idx + 4 lines = [ "# Research Idea Brainstorm Memo", "", "## 0. Scope and framing", "", f"- Topic: {payload.get('topic') or 'research ideas'}", "- Intended readers: PI / PhD", "- Goal: surface a small number of discussion-worthy research directions rather than force a final project choice.", "- Current evidence basis: abstract-first notes unless otherwise stated.", "- What this memo is not: not a final project spec, not a survey draft, and not a symmetric top-3 proposal pack.", "", "## 1. Big-picture takeaways", "", ] for item in payload.get("takeaways") or []: lines.append(f"- {item}") lines.extend([ "", "## 2. Top directions at a glance", "", shortlist_snapshot_table(payload.get("top_directions") or []), "", ]) for idx, record in enumerate(top_dirs, start=1): lines.extend([ f"## {idx + 2}. Direction {idx} — {record.get('title')}", "", "### One-line thesis", f"- {record.get('one_line_thesis')}", "", "### Why it belongs in the lead set", f"- {record.get('why_this_ranks_here')}", f"- Time to clarity: {record.get('time_to_clarity')}", "", "### What the current literature actually shows", ]) for item in record.get("literature_suggests") or []: lines.append(f"- {item}") lines.extend([ "", "### Closest prior work and remaining gap", ]) for item in record.get("closest_prior_gap") or []: lines.append(f"- {item}") lines.extend([ f"- Missing piece: {record.get('missing_piece')}", "", "### If this direction is right, what contribution emerges", f"- {record.get('contribution_shape') or record.get('academic_value')}", f"- {record.get('what_counts_as_insight')}", "", "### Smallest decisive probe", ]) for item in record.get("first_probes") or []: lines.append(f"- {item}") lines.extend([ "", "### Quick kill criteria", ]) for item in record.get("kill_criteria") or record.get("what_would_change_mind") or []: lines.append(f"- {item}") lines.extend(["",]) lines.extend([ f"## {deferred_idx}. Other promising but not prioritized directions", "", ]) deferred = payload.get("deferred_directions") or [] if deferred: for record in deferred: lines.append(f"- **{record.get('title')}** — {record.get('one_line_thesis')} (why not prioritized now: {record.get('why_not_prioritized')})") else: lines.append("- No additional deferred directions were retained in the current memo.") lines.extend([ "", f"## {discussion_idx}. Cross-cutting discussion questions", "", ]) for item in payload.get("discussion_questions") or []: lines.append(f"- {item}") lines.extend([ "", f"## {uncertainty_idx}. Uncertainty and disagreement", "", ]) for item in payload.get("uncertainties") or []: lines.append(f"- {item}") lines.extend([ "", f"## {next_idx}. Suggested next reading / next discussion step", "", ]) for item in payload.get("next_steps") or []: lines.append(f"- {item}") lines.extend([ "", f"## {appendix_idx}. Appendix guide", "", "- `output/APPENDIX.md` for anchor-paper reading notes and deferred directions.", "- `output/REPORT.json` for the structured version of this memo.", ]) return "\n".join(lines) def appendix_markdown(payload: dict[str, Any], *, core_titles: dict[str, str]) -> str: lines = [ "# Appendix to the Research Idea Brainstorm Memo", "", "## A. Deferred but still promising directions", "", ] deferred = payload.get("deferred_directions") or [] if deferred: for rec in deferred: lines.extend([ f"### {rec.get('title')}", f"- One-line thesis: {rec.get('one_line_thesis')}", f"- Program kind: {rec.get('program_kind')}", f"- Why interesting: {rec.get('why_interesting')}", f"- Why not prioritized now: {rec.get('why_not_prioritized')}", "", ]) else: lines.append("- (none)") lines.extend([ "## B. Anchor papers and what to extract", "", ]) for rec in payload.get("top_directions") or []: lines.append(f"### {rec.get('title')}") lines.append(f"- Program kind: {rec.get('program_kind')}") lines.append(f"- Lead-set reason: {rec.get('why_this_ranks_here')}") notes = rec.get("anchor_reading_notes") or [] if notes: rows: list[list[str]] = [] for item in notes: if not isinstance(item, dict): continue rows.append([ item.get("paper_title", "paper"), item.get("why_read", ""), item.get("what_to_extract", ""), item.get("current_hook", ""), item.get("kill_signal", ""), ]) if rows: lines.append(markdown_table(["Anchor paper", "Why read now", "What to extract", "Current evidence hook", "Kill signal"], rows)) else: for pid in rec.get("paper_ids") or []: title = core_titles.get(pid, pid) lines.append(f"- {title}") else: for pid in rec.get("paper_ids") or []: title = core_titles.get(pid, pid) lines.append(f"- {title}") lines.append("") return "\n".join(lines) FILE:tooling/pipeline_spec.py from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any import yaml @dataclass(frozen=True) class PipelineStage: id: str title: str checkpoint: str mode: str required_skills: tuple[str, ...] optional_skills: tuple[str, ...] produces: tuple[str, ...] human_checkpoint: dict[str, Any] @staticmethod def from_mapping(stage_id: str, data: Any, *, path: Path) -> "PipelineStage": if not isinstance(data, dict): raise ValueError(f"`stages.{stage_id}` must be a mapping in {path}") return PipelineStage( id=str(stage_id or "").strip(), title=str(data.get("title") or stage_id).strip() or str(stage_id), checkpoint=str(data.get("checkpoint") or stage_id).strip() or str(stage_id), mode=str(data.get("mode") or "").strip(), required_skills=_string_tuple(data.get("required_skills"), field_name=f"stages.{stage_id}.required_skills", path=path), optional_skills=_string_tuple(data.get("optional_skills"), field_name=f"stages.{stage_id}.optional_skills", path=path), produces=_string_tuple(data.get("produces"), field_name=f"stages.{stage_id}.produces", path=path), human_checkpoint=_mapping(data.get("human_checkpoint"), field_name=f"stages.{stage_id}.human_checkpoint", path=path, required=False), ) @dataclass(frozen=True) class PipelineSpec: path: Path name: str version: str units_template: str default_checkpoints: tuple[str, ...] profile: str routing_hints: tuple[str, ...] routing_default: bool routing_priority: int target_artifacts: tuple[str, ...] contract_model: str structure_mode: str pre_retrieval_shell: dict[str, Any] binding_layers: tuple[str, ...] core_chapter_h3_target: int query_defaults: dict[str, Any] overridable_query_fields: tuple[str, ...] quality_contract: dict[str, Any] loop_policy: dict[str, Any] stages: dict[str, PipelineStage] variant_of: str variant_overrides: dict[str, Any] @staticmethod def load(path: Path) -> "PipelineSpec": resolved = path.resolve() raw_frontmatter, frontmatter = _load_variant_aware_frontmatter(resolved) name = str(frontmatter.get("name") or path.stem.replace(".pipeline", "")) units_template = str(frontmatter.get("units_template") or "") version = str(frontmatter.get("version") or "") default_checkpoints = _string_tuple(frontmatter.get("default_checkpoints"), field_name="default_checkpoints", path=resolved) profile = str(frontmatter.get("profile") or "default").strip() or "default" routing_hints = _string_tuple(frontmatter.get("routing_hints"), field_name="routing_hints", path=resolved) routing_default = bool(frontmatter.get("routing_default")) try: routing_priority = int(frontmatter.get("routing_priority") or 0) except Exception: routing_priority = 0 target_artifacts = _string_tuple(frontmatter.get("target_artifacts"), field_name="target_artifacts", path=resolved) contract_model = str(frontmatter.get("contract_model") or "").strip() structure_mode = str(frontmatter.get("structure_mode") or "").strip() pre_retrieval_shell = _mapping(frontmatter.get("pre_retrieval_shell"), field_name="pre_retrieval_shell", path=resolved, required=False) binding_layers = _string_tuple(frontmatter.get("binding_layers"), field_name="binding_layers", path=resolved) try: core_chapter_h3_target = int(frontmatter.get("core_chapter_h3_target") or 0) except Exception: core_chapter_h3_target = 0 query_defaults = _mapping(frontmatter.get("query_defaults"), field_name="query_defaults", path=resolved, required=False) overridable_query_fields = _string_tuple( frontmatter.get("overridable_query_fields"), field_name="overridable_query_fields", path=resolved, ) quality_contract = _mapping(frontmatter.get("quality_contract"), field_name="quality_contract", path=resolved, required=False) loop_policy = _mapping(frontmatter.get("loop_policy"), field_name="loop_policy", path=resolved, required=False) stages = _parse_stages(frontmatter.get("stages"), path=resolved) variant_of = str(raw_frontmatter.get("variant_of") or "").strip() variant_overrides = _mapping(raw_frontmatter.get("variant_overrides"), field_name="variant_overrides", path=resolved, required=False) if not units_template: raise ValueError(f"Missing units_template in pipeline front matter: {resolved}") return PipelineSpec( path=resolved, name=name, version=version, units_template=units_template, default_checkpoints=default_checkpoints, profile=profile, routing_hints=routing_hints, routing_default=routing_default, routing_priority=routing_priority, target_artifacts=target_artifacts, contract_model=contract_model, structure_mode=structure_mode, pre_retrieval_shell=pre_retrieval_shell, binding_layers=binding_layers, core_chapter_h3_target=core_chapter_h3_target, query_defaults=query_defaults, overridable_query_fields=overridable_query_fields, quality_contract=quality_contract, loop_policy=loop_policy, stages=stages, variant_of=variant_of, variant_overrides=variant_overrides, ) def query_default(self, key: str, default: Any = None) -> Any: return self.query_defaults.get(str(key or "").strip(), default) def allows_query_override(self, key: str) -> bool: return str(key or "").strip() in set(self.overridable_query_fields) def _parse_frontmatter(text: str) -> dict[str, Any]: lines = text.splitlines() if not lines or lines[0].strip() != "---": raise ValueError("Pipeline file must start with YAML front matter '---'") end_idx = None for idx in range(1, len(lines)): if lines[idx].strip() == "---": end_idx = idx break if end_idx is None: raise ValueError("Unterminated YAML front matter (missing closing '---')") raw = "\n".join(lines[1:end_idx]) data = yaml.safe_load(raw) or {} if not isinstance(data, dict): raise ValueError("Pipeline YAML front matter must be a mapping") return data def _load_variant_aware_frontmatter(path: Path, seen: set[Path] | None = None) -> tuple[dict[str, Any], dict[str, Any]]: resolved = path.resolve() active = set(seen or set()) if resolved in active: chain = " -> ".join(str(p) for p in [*active, resolved]) raise ValueError(f"Cyclic `variant_of` chain detected: {chain}") active.add(resolved) text = resolved.read_text(encoding="utf-8") raw = _parse_frontmatter(text) variant_of = str(raw.get("variant_of") or "").strip() if not variant_of: return raw, dict(raw) _validate_raw_variant_frontmatter(raw, path=resolved) base_path = _resolve_pipeline_reference(resolved, variant_of) _, base_effective = _load_variant_aware_frontmatter(base_path, seen=active) current = {key: raw[key] for key in ("name", "version") if key in raw} merged = _deep_merge(base_effective, current) merged = _deep_merge( merged, _mapping(raw.get("variant_overrides"), field_name="variant_overrides", path=resolved, required=False), ) merged["variant_of"] = variant_of merged["variant_overrides"] = raw.get("variant_overrides") or {} return raw, merged def _validate_raw_variant_frontmatter(raw: dict[str, Any], *, path: Path) -> None: allowed_raw_keys = {"name", "version", "variant_of", "variant_overrides"} extra_keys = sorted(str(key) for key in raw.keys() if str(key) not in allowed_raw_keys) if extra_keys: raise ValueError( "Variant pipeline files may only keep top-level keys " "`name`, `version`, `variant_of`, `variant_overrides`; " f"move the rest under `variant_overrides` in {path}: {', '.join(extra_keys)}" ) def _resolve_pipeline_reference(path: Path, ref: str) -> Path: value = str(ref or "").strip() if not value: raise ValueError(f"Empty `variant_of` reference in {path}") candidate = Path(value) if candidate.is_absolute() and candidate.exists(): return candidate.resolve() repo_root = path.parent.parent for base in (path.parent, repo_root, repo_root / "pipelines"): direct = (base / value).resolve() if direct.exists(): return direct stem = Path(value).name if stem.endswith(".pipeline.md"): stem = stem[: -len(".pipeline.md")] if stem: for base in (path.parent, repo_root / "pipelines"): direct = (base / f"{stem}.pipeline.md").resolve() if direct.exists(): return direct raise ValueError(f"Could not resolve `variant_of: {value}` from {path}") def _deep_merge(base: Any, override: Any) -> Any: if isinstance(base, list) and isinstance(override, dict) and any(str(key).startswith("__") for key in override.keys()): return _apply_list_patch(base, override) if isinstance(base, dict) and isinstance(override, dict): merged: dict[str, Any] = {str(k): v for k, v in base.items()} for key, value in override.items(): if key in merged: merged[key] = _deep_merge(merged[key], value) else: merged[key] = value return merged return override def _apply_list_patch(base: list[Any], override: dict[str, Any]) -> list[Any]: allowed_keys = {"__append__", "__prepend__", "__remove__", "__replace__"} bad_keys = sorted(str(key) for key in override.keys() if str(key) not in allowed_keys) if bad_keys: raise ValueError( "List patch overrides only support " "`__append__`, `__prepend__`, `__remove__`, `__replace__`; " f"got: {', '.join(bad_keys)}" ) if "__replace__" in override: replacement = override.get("__replace__") if not isinstance(replacement, list): raise ValueError("List patch `__replace__` must be a YAML list") return list(replacement) current = list(base) remove_values = override.get("__remove__", []) prepend_values = override.get("__prepend__", []) append_values = override.get("__append__", []) for field_name, value in ( ("__remove__", remove_values), ("__prepend__", prepend_values), ("__append__", append_values), ): if not isinstance(value, list): raise ValueError(f"List patch `{field_name}` must be a YAML list") if remove_values: current = [item for item in current if item not in remove_values] if prepend_values: current = list(prepend_values) + current if append_values: current = current + list(append_values) return current def _mapping(value: Any, *, field_name: str, path: Path, required: bool) -> dict[str, Any]: if value is None: return {} if not isinstance(value, dict): raise ValueError(f"`{field_name}` must be a mapping in {path}") return {str(key): item for key, item in value.items()} def _string_tuple(value: Any, *, field_name: str, path: Path) -> tuple[str, ...]: if value is None: return () if not isinstance(value, list): raise ValueError(f"`{field_name}` must be a YAML list in {path}") out: list[str] = [] for item in value: text = str(item or "").strip() if text: out.append(text) return tuple(out) def _parse_stages(value: Any, *, path: Path) -> dict[str, PipelineStage]: if value is None: return {} items: list[tuple[str, Any]] = [] if isinstance(value, dict): items = [(str(stage_id or "").strip(), data) for stage_id, data in value.items()] elif isinstance(value, list): for idx, item in enumerate(value): if not isinstance(item, dict): raise ValueError(f"`stages[{idx}]` must be a mapping in {path}") stage_id = str(item.get("id") or item.get("stage") or "").strip() if not stage_id: raise ValueError(f"`stages[{idx}]` is missing `id` in {path}") items.append((stage_id, item)) else: raise ValueError(f"`stages` must be a mapping or list in {path}") stages: dict[str, PipelineStage] = {} for stage_id, data in items: stage = PipelineStage.from_mapping(stage_id, data, path=path) if not stage.id: raise ValueError(f"`stages` contains an empty stage id in {path}") stages[stage.id] = stage return stages FILE:tooling/pipeline_text.py from __future__ import annotations import json import re from pathlib import Path from typing import Any from tooling.common import load_yaml, read_jsonl def slug_unit_id(unit_id: str) -> str: raw = str(unit_id or '').strip() out: list[str] = [] for ch in raw: out.append(ch if ch.isalnum() else '_') safe = ''.join(out).strip('_') return f'S{safe}' if safe else 'S' def load_outline_sections(path: Path) -> list[dict[str, Any]]: outline = load_yaml(path) if path.exists() else [] if not isinstance(outline, list): return [] out: list[dict[str, Any]] = [] for sec in outline: if not isinstance(sec, dict): continue sec_id = str(sec.get('id') or '').strip() sec_title = str(sec.get('title') or '').strip() subsections: list[dict[str, str]] = [] for sub in sec.get('subsections') or []: if not isinstance(sub, dict): continue sub_id = str(sub.get('id') or '').strip() sub_title = str(sub.get('title') or '').strip() if sub_id and sub_title: subsections.append({'id': sub_id, 'title': sub_title}) if sec_id and sec_title: out.append({'id': sec_id, 'title': sec_title, 'subsections': subsections}) return out def iter_h3_units(path: Path) -> list[dict[str, str]]: units: list[dict[str, str]] = [] for sec in load_outline_sections(path): sec_id = str(sec.get('id') or '').strip() sec_title = str(sec.get('title') or '').strip() for sub in sec.get('subsections') or []: sub_id = str(sub.get('id') or '').strip() sub_title = str(sub.get('title') or '').strip() if sub_id and sub_title: units.append( { 'section_id': sec_id, 'section_title': sec_title, 'sub_id': sub_id, 'title': sub_title, } ) return units def read_jsonl_map(path: Path, key: str) -> dict[str, dict[str, Any]]: out: dict[str, dict[str, Any]] = {} for rec in read_jsonl(path): if not isinstance(rec, dict): continue val = str(rec.get(key) or '').strip() if val: out[val] = rec return out def read_bib_keys(path: Path) -> set[str]: if not path.exists() or path.stat().st_size <= 0: return set() text = path.read_text(encoding='utf-8', errors='ignore') return set(re.findall(r'(?im)^@\w+\s*\{\s*([^,\s]+)\s*,', text)) def uniq_keep_order(items: list[str]) -> list[str]: out: list[str] = [] seen: set[str] = set() for item in items: value = str(item or '').strip() if not value or value in seen: continue seen.add(value) out.append(value) return out def clean_excerpt(text: str, *, limit: int = 220) -> str: s = str(text or '').strip() s = s.replace('\n', ' ') s = re.sub(r'\s+', ' ', s) s = s.strip(' "\'`') s = s.replace('|', ', ') if len(s) <= limit: return s clipped = s[:limit].rsplit(' ', 1)[0].strip() return clipped if clipped else s[:limit].strip() def citation_list(keys: list[str], *, max_keys: int = 3) -> str: items = uniq_keep_order(keys)[:max_keys] if not items: return '' cites = [f'[@{k}]' for k in items] if len(cites) == 1: return cites[0] if len(cites) == 2: return f'{cites[0]} and {cites[1]}' return ', '.join(cites[:-1]) + f', and {cites[-1]}' def inline_evidence_phrase(keys: list[str], *, max_keys: int = 3) -> str: items = uniq_keep_order(keys)[:max_keys] if not items: return '' cites = [f'in [@{k}]' for k in items] if len(cites) == 1: return cites[0] if len(cites) == 2: return f'{cites[0]} and {cites[1]}' return ', '.join(cites[:-1]) + f', and {cites[-1]}' def heading_blocks(md: str) -> list[tuple[str, str]]: blocks: list[tuple[str, str]] = [] current = '' lines: list[str] = [] for raw in (md or '').splitlines(): if raw.startswith('### '): if current: blocks.append((current, '\n'.join(lines).strip())) current = raw[4:].strip() lines = [] continue if raw.startswith('## '): if current: blocks.append((current, '\n'.join(lines).strip())) current = '' lines = [] continue if current: lines.append(raw) if current: blocks.append((current, '\n'.join(lines).strip())) return blocks def dump_jsonl_lines(records: list[dict[str, Any]]) -> str: return '\n'.join(json.dumps(r, ensure_ascii=False) for r in records).rstrip() + ('\n' if records else '')

Appendix Table Writer

Comments (0)