Merge branch 'main' of https://github.com/HaibaraAi137/HalfSeed

Author: HaibaraAi137

src/halfseed/workflow.py

@@ -821,50 +821,63 @@ def _run_iteration_inner(
                 detail=latex_result.detail,
             )
 
-        # Skeptic paper review: after the Writer has produced a fresh
-        # paper.tex, hand it back to the Skeptic to surface citation
-        # gaps, mismatched cites (where the cited paper's abstract
-        # doesn't support the surrounding claim), and prose without
-        # any artifact backing. The critiques are stored as critique
-        # artifacts so the Director sees them when planning the next
-        # iteration; we deliberately do NOT loop the Writer in this
-        # iteration — paper-review feedback is asynchronous, one
-        # iteration of latency, to keep budget predictable.
+        # Post-Writer review fan-out. Three reads-only-on-publish_pool
+        # passes that have no data dependency on each other:
+        #   * Skeptic paper review (citation gaps, prose without
+        #     artifact backing) — comments on paper.tex.
+        #   * Analytical math review (dimensional / index / notation
+        #     errors in the equations).
+        #   * Presentation memo (briefing-friendly summary of the
+        #     curated artifacts).
+        # All three are best-effort, all three only ever WRITE their
+        # own outputs (no shared mutable state), so we run them in
+        # parallel when options.parallel is on. Wall-clock saving is
+        # roughly two LLM round-trips per iteration on a real API.
+        # The Skeptic paper review feedback is intentionally one-
+        # iteration-asynchronous — the Writer is NOT looped here.
         self._check_cancel()
-        paper_critiques = self._run_skeptic_paper_review(project, publish_pool)
+        paper_critiques: list[ResearchArtifact] = []
+        math_critiques: list[ResearchArtifact] = []
+        presentation_memo = ""
+        if self.options.parallel:
+            with ThreadPoolExecutor(max_workers=3) as executor:
+                fut_paper = executor.submit(
+                    self._run_skeptic_paper_review, project, publish_pool,
+                )
+                fut_math = executor.submit(
+                    self._run_analytical_math_review, project, publish_pool,
+                )
+                fut_memo = executor.submit(
+                    self._maybe_run_presentation_memo,
+                    project.research_question, curated, ranked,
+                )
+                paper_critiques = fut_paper.result() or []
+                math_critiques = fut_math.result() or []
+                presentation_memo = fut_memo.result() or ""
+        else:
+            paper_critiques = self._run_skeptic_paper_review(
+                project, publish_pool,
+            ) or []
+            math_critiques = self._run_analytical_math_review(
+                project, publish_pool,
+            ) or []
+            presentation_memo = self._maybe_run_presentation_memo(
+                project.research_question, curated, ranked,
+            ) or ""
+
         if paper_critiques:
             state.artifacts_all.extend(paper_critiques)
             self._emit(
                 "skeptic_paper_review_done",
                 critiques=len(paper_critiques),
             )
-
-        # Math review: parallel pass to the Skeptic's, but specifically
-        # for equation-level structural errors (dimensional, index,
-        # boundary, notation). Different agent because math sanity is
-        # the Analytical role's specialty, not the Skeptic's.
-        self._check_cancel()
-        math_critiques = self._run_analytical_math_review(project, publish_pool)
         if math_critiques:
             state.artifacts_all.extend(math_critiques)
             self._emit(
                 "analytical_math_review_done",
                 critiques=len(math_critiques),
             )
 
-        # Presentation memo: if the architecture has an enabled Presentation
-        # node and the curator can reach it, ask it to draft a concise memo
-        # over the curated report. The memo is added to the iteration
-        # briefing so the user reading briefings/iteration-N.md sees a
-        # human-readable summary, not just metadata. Earlier versions of
-        # this flow only ran Presentation in the legacy single-shot run()
-        # path — this is the iteration-mode equivalent. See
-        # docs/graph-workflow-spec.md §7 (default architecture).
-        self._check_cancel()
-        presentation_memo = self._maybe_run_presentation_memo(
-            project.research_question, curated, ranked
-        )
-
         store.save_artifacts(project.id, iteration.id, state.artifacts_all)
         store.save_threads(state.threads)
         if scheduled_task is not None:

tests/test_post_writer_parallel.py

@@ -0,0 +1,53 @@
+"""Post-Writer review fan-out: parallel vs serial equivalence.
+
+The Skeptic paper review, Analytical math review, and Presentation
+memo are run concurrently when ``parallel=True``. They are read-
+only on ``publish_pool`` and have no mutual data dependency, so
+parallelism must produce the exact same persisted artifact set as
+serial execution. This test exercises both modes against the fake
+backend and asserts the artifact id sets match.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from halfseed.budget import Budget
+from halfseed.config import ModelConfig
+from halfseed.llm.fake import RuleBasedBackend
+from halfseed.state.project import ProjectStore
+from halfseed.workflow import ResearchWorkflow, WorkflowOptions
+
+
+def _run_one_iteration(tmp_path: Path, *, parallel: bool):
+    store = ProjectStore(tmp_path)
+    project = store.init("toy", "Q?")
+    workflow = ResearchWorkflow(
+        backend=RuleBasedBackend(),
+        config=ModelConfig(model="deepseek-v4-flash"),
+        options=WorkflowOptions(
+            max_curated_artifacts=4,
+            parallel=parallel,
+            enable_latex_gate=False,
+        ),
+    )
+    workflow.run_iteration(
+        project, store=store, budget=Budget(token_cap=10_000),
+    )
+    rows = store.list_artifact_records(project.id)
+    return sorted(r[0].id for r in rows)
+
+
+def test_parallel_fanout_produces_same_artifacts_as_serial(
+    tmp_path: Path,
+) -> None:
+    """Same fake-backend run in parallel vs serial must persist the
+    same set of artifact ids. Parallelism only changes wall-clock —
+    the store should be identical."""
+    serial = _run_one_iteration(tmp_path / "s", parallel=False)
+    parallel = _run_one_iteration(tmp_path / "p", parallel=True)
+    assert set(serial) == set(parallel), (
+        f"serial-only: {set(serial) - set(parallel)}\n"
+        f"parallel-only: {set(parallel) - set(serial)}"
+    )