--- name: chapter-lead-writer description: 'Write H2 chapter lead blocks (`sections/S_lead.md`) that preview the chapter''s comparison lens and connect its H3 subsections, without adding new facts. **Trigger**: chapter lead writer, section lead writer, H2 lead, lead paragraph, 章节导读, 章节导语. **Use when**: you have H2 chapters with multiple H3 subsections and the draft reads like paragraph islands across subsections. **Skip if**: the outline has no H3 subsections, or `outline/chapter_briefs.jsonl` is missing. **Network**: none. **Guardrail**: no new facts/citations; no headings; no narration templates; use only citation keys present in `citations/ref.bib`.' version: 0.1.0 metadata: openclaw: requires: anyBins: - python3 - python --- # Chapter Lead Writer ## Purpose This skill writes the body-only lead block that sits under an H2 heading and makes a chapter with multiple H3 subsections read like one argument. This `SKILL.md` is now the **package router**, not the full method manual. ## Migration status This package is in **P0 compatibility-preserving migration**: - `references/` and `assets/` now hold the intended knowledge and contract layers. - `scripts/run.py` remains in **compatibility mode** for active generation. - a later script-thinning pass should move more judgment and exemplars out of Python and leave the script with deterministic execution and validation only. For now, preserve the existing output contract and treat `scripts/run.py` as the execution source of truth. ## Inputs Required: - `outline/outline.yml` - `outline/chapter_briefs.jsonl` - `citations/ref.bib` Optional: - `outline/writer_context_packs.jsonl` ## Outputs For each H2 section with H3 subsections: - `sections/S_lead.md` ## Output contract Keep these file-shape rules stable: - each lead file is body-only and contains no headings - each lead file previews the chapter lens and connects multiple H3s as one argument - each lead file stays within the chapter's existing citation scope - each lead file adds no new facts that are not supported later in the chapter ## Load Order Always read: - `references/overview.md` - `references/lead_block_archetypes.md` Read by task: - `references/throughline_patterns.md` — when chapter briefs are thin or hard to convert into a throughline - `references/bridge_examples.md` — when the lead needs stronger H3 transitions without slide narration - `references/bad_narration_examples.md` — when removing table-of-contents narration, planner talk, count-based openers Machine-readable assets: - `assets/lead_block_contract.json` — stable package contract for lead-block shape - `assets/lead_block_compatibility_defaults.json` — fallback phrasing, item limits, joiners, sentence cadence ## Routing rules Use this skill in the following order: 1. Confirm the chapter is eligible - identify H2 sections with H3 subsections from `outline/outline.yml` - locate the corresponding chapter brief in `outline/chapter_briefs.jsonl` 2. Load the method - read `references/overview.md` - read `references/lead_block_archetypes.md` - load the other reference files only if the chapter brief or current prose needs them 3. Check citation scope - if `outline/writer_context_packs.jsonl` exists, use it for cross-cutting chapter citations - keep any citations inside the existing chapter scope and validate keys against `citations/ref.bib` 4. Execute - **current phase**: use `scripts/run.py` in compatibility mode to preserve active behavior and output shape - **future phase**: keep `scripts/run.py` for deterministic execution only, with the writing method and anti-pattern inventory living in `references/` ## Compatibility mode note `scripts/run.py` still contains active lead-generation logic. That is temporary. For now: - do not treat the current script wording as the target architecture - do treat `assets/lead_block_compatibility_defaults.json` as the primary compatibility-mode wording source - do not copy large prose instructions back into `SKILL.md` - do preserve the current output contract while reducing obvious narration stems in the active path ## What this skill should guarantee Regardless of where the detailed method lives, this skill should produce chapter leads that: - state the chapter's comparison lens rather than narrating the outline - connect the H3 subsections as one argument, not as isolated stops on a tour - introduce recurring contrasts without slash-list jargon - keep the evaluation or calibration lens visible at a high level - avoid slide narration, planner talk, and repeated stock openers - choose from multiple candidate lead frames when possible (lens-first / sequence-first / comparison-first) and keep the least narrated option instead of reusing one stock cadence everywhere ## Block conditions Stop and route upstream if any of these are true: - `outline/chapter_briefs.jsonl` is missing - the target H2 section has no H3 subsections - the chapter brief is too incomplete to infer a throughline safely - the requested lead would require new facts or out-of-scope citations ## Script role `scripts/run.py` should currently be treated as a **compatibility executor**. Its long-term role after script thinning is narrower: - chapter discovery - brief loading and normalization - contract validation - deterministic report writing It is not the long-term home for lead archetypes, bridge examples, or narration anti-patterns. ## Script ### Quick Start - `python scripts/run.py --workspace ` ### All Options - `--workspace ` - `--unit-id ` - `--inputs ` - `--outputs ` - `--checkpoint ` ### Examples - `python scripts/run.py --workspace workspaces/` ## Troubleshooting - If `outline/chapter_briefs.jsonl` is missing or too thin, rebuild chapter briefs first. - If `outline/writer_context_packs.jsonl` is missing, the script will still run but with a thinner citation pool. - If a generated lead sounds narrated, patch the compatibility asset and references before changing Python. FILE:AGENTS.md # AGENTS.md This exported skill uses `AGENTS.md` only as a local repo-root marker for bundled helper scripts. Source skill: `chapter-lead-writer` from `research-units-pipeline-skills`. Export slug: `chapter-lead-writer`. FILE:assets/lead_block_compatibility_defaults.json { "compatibility_mode": true, "default_archetype": "lens-first", "limits": { "throughline_items": 2, "contrast_items": 3, "subsection_preview_items": 3 }, "fallbacks": { "section_title": "chapter {section_id}", "comparison_problem": "the comparison pressure that the subsections keep reopening", "recurring_contrasts": "protocol assumptions, resource constraints, and evaluation scope", "subsection_preview": "the chapter subsections" }, "templates": { "lens-first": { "paragraph_1": [ "In {section_title}, the central question is how {comparison_problem} change the meaning of reported progress{citation}.", "{section_title} is where differences in {comparison_problem} stop looking like implementation detail and start changing what the evidence can support{citation}.", "The core issue in {section_title} is not breadth alone, but how {comparison_problem} reshape what counts as a fair comparison{citation}." ], "paragraph_2": [ "{subsection_preview} each expose that pressure from a different angle, showing why {recurring_contrasts} cannot be treated as interchangeable background choices{citation}.", "Across {subsection_preview}, the same problem keeps returning: {recurring_contrasts} alter the interpretation of otherwise similar results{citation}.", "Taken together, {subsection_preview} show where {recurring_contrasts} become substantive constraints rather than secondary implementation detail{citation}." ] } } } FILE:assets/lead_block_contract.json { "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://example.local/chapter-lead-writer/lead-block-contract.schema.json", "title": "Chapter Lead Block Contract", "type": "object", "additionalProperties": false, "required": [ "section_id", "section_title", "output_path", "body_only", "max_paragraphs", "must_not_have_headings", "may_add_new_facts", "citation_scope", "compatibility_mode" ], "properties": { "section_id": { "type": "string", "minLength": 1 }, "section_title": { "type": "string", "minLength": 1 }, "output_path": { "type": "string", "pattern": "^sections/S[^/]+_lead\\.md$" }, "body_only": { "type": "boolean", "const": true }, "max_paragraphs": { "type": "integer", "minimum": 1, "maximum": 3 }, "must_not_have_headings": { "type": "boolean", "const": true }, "may_add_new_facts": { "type": "boolean", "const": false }, "citation_scope": { "type": "string", "enum": [ "chapter-only", "chapter-or-selected-crosscutting" ] }, "compatibility_mode": { "type": "boolean" } } } 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/bad_narration_examples.md # Bad Narration Examples Use this file to remove high-signal generator voice from chapter leads. ## Table-of-contents narration Bad: - `In this section, we discuss planning, memory, and adaptation.` - `This chapter surveys several design choices and then reviews evaluation.` Why it fails: - it narrates structure instead of stating the chapter's comparison problem ## Slide navigation Bad: - `Next, we move from interfaces to planning.` - `We now turn to evaluation.` Why it fails: - it sounds like a presentation script rather than paper prose ## Planner talk Bad: - `To keep the chapter coherent, we first focus on representation and then expand to evaluation.` - `The remaining uncertainty is resolved by looking at feedback mechanisms.` Why it fails: - it exposes internal planning language instead of the literature-facing argument ## Count-based opener slots Bad: - `Two key points organize this chapter.` - `Three takeaways define the next comparison.` Why it fails: - it creates a reusable shell that quickly feels generated when repeated across chapters ## Title narration Bad: - `From representation to adaptation, the chapter covers several themes.` Why it fails: - it restates heading movement instead of making a claim about the chapter lens FILE:references/bridge_examples.md # Bridge Examples Use these examples to connect H3 subsections without sounding like slide navigation. Paraphrase them; do not copy them verbatim. ## Good bridge: from interface to decision policy - `Once the interface fixes what actions are executable, the next comparison is how different systems choose among those actions under limited feedback and bounded cost.` Why it works: - it makes the next subsection feel necessary - it names the hinge of the comparison instead of narrating movement ## Good bridge: from representation to evaluation - `Richer internal state only matters if the evaluator can still tell what evidence a decision relied on, which is why the chapter shifts from representation choices to comparison under explicit protocol assumptions.` Why it works: - it links a design choice to the chapter's calibration lens ## Good bridge: from local mechanism to chapter-level contrast - `These mechanisms differ in implementation, but they are comparable because each changes the same trade-off between runtime flexibility and predictable behavior under constraint.` Why it works: - it connects local variation back to one recurring contrast ## Good bridge: from one H3 to the next without ToC talk - `That bottleneck narrows the comparison to what the system can revise after acting, which is why the next subsection focuses on feedback and update structure rather than on architecture labels alone.` Why it works: - it points forward through argument, not through narration FILE:references/lead_block_archetypes.md # Lead Block Archetypes Use these archetypes as drafting shapes. Paraphrase them; do not paste them verbatim. Compatibility note: - the active compatibility-mode sentence defaults now live in `assets/lead_block_compatibility_defaults.json` - update that asset when the issue is fallback cadence or reusable sentence wording, and update this file when the issue is the underlying chapter-lead method ## 1. Lens-first Use when: - the chapter brief already has a strong throughline - the reader mainly needs the chapter's comparison lens up front Move pattern: - sentence 1: state the shared comparison problem - sentence 2: name the recurring contrasts - optional sentence 3: calibrate how evidence or protocol assumptions affect comparison Example: - `What matters most in this chapter is not whether the systems share a surface architecture, but which assumptions make their decisions comparable under the same evaluation setting.` ## 2. Contrast-first Use when: - the chapter is held together by 2-3 recurring trade-offs - the H3 subsections are easy to over-read as isolated methods Move pattern: - sentence 1: introduce the recurring trade-off - sentence 2: show how different subsections expose different sides of it - sentence 3: state why that contrast matters for interpretation Example: - `The central contrast in this chapter is between flexibility and control: each subsection changes how much structure the system fixes in advance and how much it leaves to runtime adaptation.` ## 3. Calibration-first Use when: - the comparison only makes sense if the reader keeps protocol or evidence assumptions in view - the chapter includes evaluation, reproducibility, or deployment constraints Move pattern: - sentence 1: state the high-level calibration rule - sentence 2: connect that rule to the chapter's subsections - sentence 3: preview the design choices that the chapter will compare under that rule Example: - `The comparisons in this chapter only hold if protocol assumptions stay visible, because the same design choice looks very different once budget, access, or observability changes.` ## 4. Problem-first Use when: - the chapter brief defines a shared bottleneck or failure surface - the subsections are best read as different answers to the same question Move pattern: - sentence 1: name the bottleneck - sentence 2: show how each subsection addresses it differently - sentence 3: explain the comparison lens that keeps those answers on the same scale Example: - `This chapter is unified by a single bottleneck: once the system must act under limited feedback, design choices that seem local become coupled to evidence quality, cost, and stability.` FILE:references/overview.md # Overview ## Purpose This package writes the short H2 lead block that turns several H3 subsections into one chapter-level argument. Read this file before applying the skill or editing the package. ## Role split ### `SKILL.md` Use `SKILL.md` as the router: - when the skill triggers - required inputs and outputs - when to read which reference file - what the active script still does in compatibility mode ### `references/` Use `references/` for the judgment layer: - lead-block archetypes - throughline construction patterns - bridge examples - narration anti-patterns ### `assets/` Use `assets/` for machine-readable constraints that later validation can consume. In compatibility mode, use `assets/lead_block_compatibility_defaults.json` as the editable source for fallback lead phrasing, joiners, and sentence cadence. ### `scripts/` Use the script only for deterministic work and compatibility-mode execution. It should not remain the long-term home for voice rules or reusable prose patterns. If a fixed fallback phrase or lead cadence needs tuning, change the compatibility asset before editing Python. ## What a chapter lead must do A good chapter lead does four jobs: - state the chapter's comparison lens - connect multiple H3 subsections as one argument - preview 2-3 recurring contrasts without turning into an axis dump - calibrate how the chapter should be read at a high level, especially when protocol or evidence assumptions affect comparison ## What it must not do Avoid: - table-of-contents narration - slide-navigation phrasing - planner-talk transitions - count-based opener slots used as a reusable shell - title narration that sounds like a generated heading bridge - new claims that the downstream H3s do not support ## Recommended read order 1. `lead_block_archetypes.md` - choose the shape of the lead block 2. `throughline_patterns.md` - map chapter brief fields into a chapter-level argument 3. `bridge_examples.md` - strengthen local transitions between subsections without narrating the outline 4. `bad_narration_examples.md` - remove weak stems and generator voice before finalizing the block ## Compatibility mode boundary In this migration phase, the script still writes the active output. Use the references to guide edits and future refactors, but preserve: - body-only output - existing file naming - the report file - the same high-level contract of a short lead block per eligible H2 chapter FILE:references/throughline_patterns.md # Throughline Patterns Use this file when `outline/chapter_briefs.jsonl` is hard to convert into one lead block. ## Field-to-move mapping ### `throughline` Use it to answer: - what shared question this chapter is really asking - why the H3 subsections belong together Good transformation: - convert a topic list into one comparison problem Example transformation: - weak input: `state management`, `deliberation`, `feedback loops` - stronger throughline: `The chapter compares how systems preserve decision quality once intermediate state, delayed feedback, and multi-step control must all be managed together.` ### `key_contrasts` Use it to choose 2-3 recurring contrasts that can appear in the lead. Good contrasts are: - comparative - reusable across several H3 subsections - readable as prose rather than labels Prefer: - `fixed structure versus runtime adaptation` - `richer state versus easier verification` - `local gains versus transfer under tighter constraints` Avoid: - copied slash lists - internal planner labels - contrast inventories longer than the lead can hold naturally ### `lead_paragraph_plan` Use it to decide progression. Good use: - interpret the plan as movement in the argument, not as a miniature table of contents Example: - weak use: `first cover setup, then methods, then evaluation` - stronger use: `move from the source of the bottleneck to the design choices it induces, then to the conditions under which those choices remain comparable` ## Compression rule If a brief contains many bullets, compress them into: - one chapter-level question - two or three recurring contrasts - one progression sentence if needed If the brief cannot be compressed that way, the lead should stay conservative and avoid over-claiming coherence. FILE:scripts/run.py from __future__ import annotations import argparse import hashlib import json import re import sys from pathlib import Path from typing import Any def _load_jsonl(path: Path) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] if not path.exists() or path.stat().st_size <= 0: return out import json for raw in path.read_text(encoding='utf-8', errors='ignore').splitlines(): raw = raw.strip() if not raw: continue try: rec = json.loads(raw) except Exception: continue if isinstance(rec, dict): out.append(rec) return out def _load_json_asset(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 {} def _uniq(items: list[str]) -> list[str]: out: list[str] = [] seen: set[str] = set() for item in items: v = str(item or '').strip() if not v or v in seen: continue seen.add(v) out.append(v) return out def _cite(keys: list[str], n: int = 3) -> str: vals = _uniq(keys)[:n] return f"[@{'; '.join(vals)}]" if vals else "" def _clean_phrase(text: str) -> str: s = str(text or '').strip().replace('/', ' and ') s = ' '.join(s.split()) return s.rstrip(' .;:，；。') def _natural_phrase(text: str) -> str: s = _clean_phrase(text).replace('&', 'and') s = re.sub(r'$[^)]*$', '', s) s = re.sub(r'\s+', ' ', s).strip(' ,;:') low = s.lower() blocked = ( 'core mechanism and architecture', 'evaluation protocol', 'datasets, metrics, human eval', ) if any(token in low for token in blocked): return '' return low if low else '' def _content_phrase(text: str) -> str: s = _clean_phrase(text) low = s.lower() prefixes = [ 'pin scope to goal:', 'compare approaches along:', 'para 1:', 'para 2:', 'para 3:', 'para 3 (optional):', ] for prefix in prefixes: if low.startswith(prefix): s = s.split(':', 1)[1].strip() if ':' in s else '' low = s.lower() meta_starts = ( 'state the chapter', 'preview the h3', 'highlight evaluation anchors', ) if any(low.startswith(prefix) for prefix in meta_starts): return '' return s def _skill_root() -> Path: return Path(__file__).resolve().parents[1] def _compatibility_defaults() -> dict[str, Any]: asset_path = _skill_root() / 'assets' / 'lead_block_compatibility_defaults.json' return _load_json_asset(asset_path) def _limit(value: Any, *, fallback: int) -> int: try: num = int(value) except Exception: return fallback return num if num > 0 else fallback def _series(items: list[str], *, limit: int, fallback: str) -> str: vals = [str(item).strip() for item in items if str(item).strip()][:limit] if not vals: return fallback if len(vals) == 1: return vals[0] if len(vals) == 2: return f'{vals[0]} and {vals[1]}' return f"{', '.join(vals[:-1])}, and {vals[-1]}" def _citation_suffix(keys: list[str], n: int = 3) -> str: cite = _cite(keys, n) return f' {cite}' if cite else '' def _render(template: str, **kwargs: str) -> str: text = template.format(**kwargs) text = ' '.join(text.split()) text = text.replace(' .', '.').replace(' ,', ',').replace(' ;', ';').replace(' :', ':') return text def _ordered_options(seed: str, value: Any) -> list[str]: if not isinstance(value, list): single = str(value or '').strip() return [single] if single else [] options = [str(item or '').strip() for item in value if str(item or '').strip()] if not options: return [] digest = hashlib.sha1(str(seed or '').encode('utf-8', errors='ignore')).hexdigest() idx = int(digest[:12], 16) % len(options) return options[idx:] + options[:idx] def _pick_template(seed: str, value: Any, fallback: str) -> str: options = _ordered_options(seed, value) return options[0] if options else fallback def main() -> int: parser = argparse.ArgumentParser() parser.add_argument('--workspace', required=True) parser.add_argument('--unit-id', default='') parser.add_argument('--inputs', default='') parser.add_argument('--outputs', default='') parser.add_argument('--checkpoint', default='') args = parser.parse_args() repo_root = Path(__file__).resolve() for _ in range(10): if (repo_root / "AGENTS.md").exists(): break parent = repo_root.parent if parent == repo_root: break repo_root = parent sys.path.insert(0, str(repo_root)) from tooling.common import atomic_write_text, ensure_dir, load_yaml, now_iso_seconds from tooling.pipeline_text import slug_unit_id workspace = Path(args.workspace).resolve() compat = _compatibility_defaults() limits = compat.get('limits') if isinstance(compat.get('limits'), dict) else {} fallbacks = compat.get('fallbacks') if isinstance(compat.get('fallbacks'), dict) else {} templates = compat.get('templates') if isinstance(compat.get('templates'), dict) else {} archetype = str(compat.get('default_archetype') or 'lens-first').strip() or 'lens-first' template_pack = templates.get(archetype) if isinstance(templates.get(archetype), dict) else {} sections_dir = workspace / 'sections' ensure_dir(sections_dir) report_path = workspace / 'output' / 'CHAPTER_LEADS_REPORT.md' ensure_dir(report_path.parent) outline = load_yaml(workspace / 'outline' / 'outline.yml') if (workspace / 'outline' / 'outline.yml').exists() else [] briefs = {str(r.get('section_id') or '').strip(): r for r in _load_jsonl(workspace / 'outline' / 'chapter_briefs.jsonl') if str(r.get('section_id') or '').strip()} packs = _load_jsonl(workspace / 'outline' / 'writer_context_packs.jsonl') cites_by_sec: dict[str, list[str]] = {} for rec in packs: sec_id = str(rec.get('section_id') or '').strip() if not sec_id: continue bucket = cites_by_sec.setdefault(sec_id, []) for field in ['allowed_bibkeys_selected', 'allowed_bibkeys_chapter']: for key in rec.get(field) or []: bucket.append(str(key).strip()) written: list[str] = [] if isinstance(outline, list): for sec in outline: if not isinstance(sec, dict): continue subs = [x for x in (sec.get('subsections') or []) if isinstance(x, dict)] if not subs: continue sec_id = str(sec.get('id') or '').strip() sec_title = str(sec.get('title') or '').strip() or str(fallbacks.get('section_title') or 'chapter {section_id}').format(section_id=sec_id) if not sec_id: continue brief = briefs.get(sec_id) or {} throughline = [_natural_phrase(_content_phrase(x)) for x in (brief.get('throughline') or []) if _natural_phrase(_content_phrase(x))] contrasts = [_natural_phrase(x) for x in (brief.get('key_contrasts') or []) if _natural_phrase(x)] sub_titles = [str(x.get('title') or '').strip() for x in subs if str(x.get('title') or '').strip()] cites = cites_by_sec.get(sec_id) or [] comparison_problem = _series( contrasts or throughline, limit=_limit(limits.get('throughline_items'), fallback=2), fallback=str(fallbacks.get('comparison_problem') or 'the comparison pressure that the subsections keep reopening'), ) recurring_contrasts = _series( contrasts, limit=_limit(limits.get('contrast_items'), fallback=3), fallback=str(fallbacks.get('recurring_contrasts') or 'protocol assumptions, resource constraints, and evaluation scope'), ) subsection_preview = _series( sub_titles, limit=_limit(limits.get('subsection_preview_items'), fallback=3), fallback=str(fallbacks.get('subsection_preview') or 'the chapter subsections'), ) paragraph_1 = _render( _pick_template( f'lead:p1:{sec_id}:{sec_title}', template_pack.get('paragraph_1'), 'What binds {section_title} together is a shared comparison problem: {comparison_problem}{citation}.', ), section_title=sec_title, comparison_problem=comparison_problem, citation=_citation_suffix(cites, 4), ) paragraph_2 = _render( _pick_template( f'lead:p2:{sec_id}:{subsection_preview}', template_pack.get('paragraph_2'), 'Taken together, {subsection_preview} show why {recurring_contrasts} cannot be treated as background implementation detail{citation}.', ), subsection_preview=subsection_preview, recurring_contrasts=recurring_contrasts, citation=_citation_suffix(cites[4:], 4) or _citation_suffix(cites, 4), ) paragraphs = [paragraph_1, paragraph_2] path = sections_dir / f"{slug_unit_id(sec_id)}_lead.md" atomic_write_text(path, '\n\n'.join(paragraphs).rstrip() + '\n') written.append(str(path.relative_to(workspace))) report = '\n'.join([ '# Chapter leads report', '', '- Status: PASS', f'- Generated at: `{now_iso_seconds()}`', ] + [f'- `{p}`' for p in written]) + '\n' atomic_write_text(report_path, report) return 0 if __name__ == '__main__': raise SystemExit(main()) FILE:tooling/__init__.py FILE:tooling/common.py from __future__ import annotations import csv import json import os import re import shutil import tempfile from dataclasses import dataclass from datetime import date, datetime from pathlib import Path from typing import Any, Iterable import yaml def today_iso() -> str: return date.today().isoformat() def now_iso_seconds() -> str: return datetime.now().replace(microsecond=0).isoformat() def ensure_dir(path: Path) -> None: path.mkdir(parents=True, exist_ok=True) def atomic_write_text(path: Path, content: str) -> None: ensure_dir(path.parent) fd, tmp_path = tempfile.mkstemp(prefix=path.name, dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: handle.write(content) os.replace(tmp_path, path) except Exception: try: os.unlink(tmp_path) except OSError: pass raise def backup_existing(path: Path) -> Path: """Rename an existing file to a timestamped `.bak.*` sibling and return the backup path.""" if not path.exists(): return path stamp = datetime.now().replace(microsecond=0).isoformat().replace("-", "").replace(":", "") backup = path.with_name(f"{path.name}.bak.{stamp}") counter = 1 while backup.exists(): backup = path.with_name(f"{path.name}.bak.{stamp}.{counter}") counter += 1 path.replace(backup) return backup def parse_semicolon_list(value: str | None) -> list[str]: if not value: return [] return [item.strip() for item in value.split(";") if item.strip()] def normalize_title_for_dedupe(title: str) -> str: title = title.lower() title = re.sub(r"[^a-z0-9]+", " ", title) title = re.sub(r"\s+", " ", title).strip() return title def normalize_axis_label(text: str) -> str: text = re.sub(r"\s+", " ", (text or "").strip().lower()) text = text.rstrip(" .;:，；。") text = re.sub(r"\s*/\s*", " ", text) text = re.sub(r"[^a-z0-9]+", " ", text) return text.strip() def subsection_brief_generic_axis_norms() -> set[str]: """Axis labels that strict survey gates treat as scaffold-level defaults. Keep this aligned across brief generation and quality-gate checks so the generator does not promote an axis that the gate later classifies as generic. """ axes = { "core mechanism and system architecture", "training and data setup", "evaluation protocol", "evaluation protocol (benchmarks / metrics / human)", "evaluation protocol (datasets / metrics / human)", "evaluation protocol (datasets, metrics, human evaluation)", "compute and efficiency", "compute and latency constraints", "efficiency and compute", "tool interface contract (schemas / protocols)", "tool selection / routing policy", "sandboxing / permissions / observability", "failure modes and limitations", } return {normalize_axis_label(axis) for axis in axes} def read_jsonl(path: Path) -> list[dict[str, Any]]: records: list[dict[str, Any]] = [] if not path.exists(): return records with path.open("r", encoding="utf-8") as handle: for line in handle: line = line.strip() if not line: continue records.append(json.loads(line)) return records def write_jsonl(path: Path, records: Iterable[dict[str, Any]]) -> None: ensure_dir(path.parent) lines = [json.dumps(record, ensure_ascii=False) for record in records] atomic_write_text(path, "\n".join(lines) + ("\n" if lines else "")) def read_tsv(path: Path) -> list[dict[str, str]]: if not path.exists(): return [] with path.open("r", encoding="utf-8", newline="") as handle: reader = csv.DictReader(handle, delimiter="\t") return [dict(row) for row in reader] def write_tsv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str]) -> None: ensure_dir(path.parent) fd, tmp_path = tempfile.mkstemp(prefix=path.name, dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames, delimiter="\t") writer.writeheader() for row in rows: writer.writerow({key: row.get(key, "") for key in fieldnames}) os.replace(tmp_path, path) except Exception: try: os.unlink(tmp_path) except OSError: pass raise @dataclass(frozen=True) class UnitsTable: fieldnames: list[str] rows: list[dict[str, str]] @staticmethod def load(path: Path) -> "UnitsTable": with path.open("r", encoding="utf-8", newline="") as handle: reader = csv.DictReader(handle) fieldnames = list(reader.fieldnames or []) rows = [dict(row) for row in reader] return UnitsTable(fieldnames=fieldnames, rows=rows) def save(self, path: Path) -> None: ensure_dir(path.parent) fd, tmp_path = tempfile.mkstemp(prefix=path.name, dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=self.fieldnames) writer.writeheader() for row in self.rows: writer.writerow({key: row.get(key, "") for key in self.fieldnames}) os.replace(tmp_path, path) except Exception: try: os.unlink(tmp_path) except OSError: pass raise def load_yaml(path: Path) -> Any: with path.open("r", encoding="utf-8") as handle: return yaml.safe_load(handle) def dump_yaml(path: Path, data: Any) -> None: ensure_dir(path.parent) text = yaml.safe_dump( data, allow_unicode=True, sort_keys=False, default_flow_style=False, width=120, ) atomic_write_text(path, text) def copy_tree(src_dir: Path, dst_dir: Path, *, overwrite: bool) -> None: if not src_dir.is_dir(): raise ValueError(f"Template directory not found: {src_dir}") ensure_dir(dst_dir) for src_path in src_dir.rglob("*"): rel = src_path.relative_to(src_dir) dst_path = dst_dir / rel if src_path.is_dir(): ensure_dir(dst_path) continue ensure_dir(dst_path.parent) if dst_path.exists() and not overwrite: continue shutil.copy2(src_path, dst_path) def tokenize(text: str) -> list[str]: text = text.lower() text = re.sub(r"[^a-z0-9]+", " ", text) return [token for token in text.split() if token] _EN_STOPWORDS = { "a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "in", "is", "it", "of", "on", "or", "that", "the", "this", "to", "with", "we", "our", "via", "towards", "toward", "using", "use", "based", "new", "towards", "into", "over", "under", "between", "within", "without", "beyond", } _GENERIC_PAPER_WORDS = { "survey", "review", "tutorial", "paper", "approach", "method", "methods", "model", "models", "framework", "frameworks", "system", "systems", "learning", "deep", "neural", "network", "networks", "analysis", "benchmark", "benchmarks", "dataset", "datasets", "evaluation", "evaluating", "towards", "using", "based", "study", "studies", } def candidate_keywords(titles: Iterable[str], *, top_k: int, min_freq: int) -> list[str]: freq: dict[str, int] = {} for title in titles: for token in tokenize(title): if token in _EN_STOPWORDS or token in _GENERIC_PAPER_WORDS: continue if len(token) < 3: continue freq[token] = freq.get(token, 0) + 1 candidates = [t for t, c in sorted(freq.items(), key=lambda kv: (-kv[1], kv[0])) if c >= min_freq] return candidates[:top_k] def update_status_log(status_path: Path, line: str) -> None: ensure_dir(status_path.parent) if status_path.exists(): existing = status_path.read_text(encoding="utf-8") else: existing = "# Status\n" if "## Run log" not in existing: existing = existing.rstrip() + "\n\n## Run log\n" updated = existing.rstrip() + f"\n- {line}\n" atomic_write_text(status_path, updated) def update_status_field(status_path: Path, heading: str, value: str) -> None: heading_line = f"## {heading}".strip() bullet_line = f"- `{value}`" if status_path.exists(): lines = status_path.read_text(encoding="utf-8").splitlines() else: lines = ["# Status"] out: list[str] = [] i = 0 updated = False while i < len(lines): line = lines[i] out.append(line) if line.strip() == heading_line: if i + 1 < len(lines) and lines[i + 1].lstrip().startswith("-"): out.append(bullet_line) i += 2 updated = True continue out.append(bullet_line) updated = True i += 1 if not updated: out.extend(["", heading_line, bullet_line]) atomic_write_text(status_path, "\n".join(out).rstrip() + "\n") def decisions_has_approval(decisions_path: Path, checkpoint: str) -> bool: if not checkpoint: return False if not decisions_path.exists(): return False text = decisions_path.read_text(encoding="utf-8") pattern = rf"^\s*-\s*\[[xX]\]\s*(?:Approve\s*)?{re.escape(checkpoint)}\b" return re.search(pattern, text, flags=re.MULTILINE) is not None def ensure_decisions_approval_checklist(decisions_path: Path) -> None: if decisions_path.exists(): text = decisions_path.read_text(encoding="utf-8") else: text = "# Decisions log\n" if re.search(r"^##\s+Approvals\b", text, flags=re.MULTILINE): return workspace = decisions_path.parent checkpoints = _human_checkpoints_from_units(workspace) if not checkpoints: return checklist_lines = ["## Approvals (check to unblock)"] for checkpoint in checkpoints: hint = _approval_hint(checkpoint) suffix = f" ({hint})" if hint else "" checklist_lines.append(f"- [ ] Approve {checkpoint}{suffix}") checklist_lines.append("") checklist = "\n".join(checklist_lines) lines = text.splitlines() if lines and lines[0].startswith("#"): new_text = "\n".join([lines[0], "", checklist] + lines[1:]).rstrip() + "\n" else: new_text = (checklist + "\n" + text).rstrip() + "\n" atomic_write_text(decisions_path, new_text) def set_decisions_approval(decisions_path: Path, checkpoint: str, *, approved: bool) -> None: checkpoint = checkpoint.strip() if not checkpoint: raise ValueError("checkpoint must be non-empty") ensure_decisions_approval_checklist(decisions_path) text = decisions_path.read_text(encoding="utf-8") lines = text.splitlines() pattern = re.compile(rf"^\s*-\s*\[\s*[xX ]\s*\]\s*(?:Approve\s*)?{re.escape(checkpoint)}\b") updated = False for idx, line in enumerate(lines): if pattern.search(line): lines[idx] = re.sub( r"\[\s*[xX ]\s*\]", "[x]" if approved else "[ ]", line, count=1, ) updated = True break if not updated: insert_at = None for idx, line in enumerate(lines): if line.strip().startswith("## Approvals"): insert_at = idx + 1 break if insert_at is None: lines.append("") lines.append("## Approvals (check to unblock)") insert_at = len(lines) lines.insert(insert_at, f"- [{'x' if approved else ' '}] Approve {checkpoint}") atomic_write_text(decisions_path, "\n".join(lines).rstrip() + "\n") def _human_checkpoints_from_units(workspace: Path) -> list[str]: units_path = workspace / "UNITS.csv" if not units_path.exists(): return [] try: table = UnitsTable.load(units_path) except Exception: return [] seen: set[str] = set() out: list[str] = [] for row in table.rows: owner = (row.get("owner") or "").strip().upper() if owner != "HUMAN": continue checkpoint = (row.get("checkpoint") or "").strip() if checkpoint and checkpoint not in seen: seen.add(checkpoint) out.append(checkpoint) return out def _approval_hint(checkpoint: str) -> str: hints = { "C0": "kickoff: scope/sources/time window/constraints", "C1": "retrieval + core set", "C2": "scope + outline", "C3": "evidence ready", "C4": "citations verified", "C5": "allow prose writing", } return hints.get(checkpoint, "") def upsert_checkpoint_block(decisions_path: Path, checkpoint: str, markdown_block: str) -> None: begin = f"" 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 '')

Chapter Lead Writer

Comments (0)