feat(task-012): orchestrate parallel section production jobs

2026-05-21 23:10:34 +03:00
parent 7e1b05067c
commit 8ebb5ad623
15 changed files with 756 additions and 39 deletions
@@ -90,7 +90,7 @@ class EvidenceMatrixPublicApiTest(unittest.TestCase):
        )
        self.assertEqual(200, evidence_response.status_code, evidence_response.text)
        body = evidence_response.json()
-        self.assertEqual("RESEARCH_RUNNING", body["article"]["status"])
+        self.assertEqual("EVIDENCE_MATRIX_READY", body["article"]["status"])
        self.assertFalse(body["insufficient_evidence_reasons"])
        self.assertTrue(body["evidence"])
        self.assertTrue(body["claims"])
@@ -0,0 +1,321 @@
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from typing import Any
+
+from fastapi.testclient import TestClient
+
+
+BACKEND_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(BACKEND_ROOT))
+
+from src.application.seed_data import seed_reference_data  # noqa: E402
+from src.infrastructure.repositories import open_backend_repository  # noqa: E402
+from src.presentation.dependencies import get_repository  # noqa: E402
+from src.presentation.main import app  # noqa: E402
+
+
+DEMO_EDITOR_EMAIL = "editor@example.com"
+DEMO_ADMIN_EMAIL = "admin@example.com"
+DEMO_USER_EMAIL_HEADER = "X-Demo-User-Email"
+
+
+class ParallelProductionPublicApiTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp_dir = tempfile.TemporaryDirectory()
+        os.environ["OBJECT_STORAGE_LOCAL_ROOT"] = str(Path(self.tmp_dir.name) / "objects")
+        dsn = f"sqlite:///{Path(self.tmp_dir.name) / 'parallel-production.db'}"
+        self.repository = open_backend_repository(dsn)
+        self.repository.setup()
+        seed_reference_data(self.repository)
+        app.dependency_overrides[get_repository] = lambda: self.repository
+        self.client = TestClient(app)
+
+    def tearDown(self) -> None:
+        app.dependency_overrides.clear()
+        os.environ.pop("OBJECT_STORAGE_LOCAL_ROOT", None)
+        self.tmp_dir.cleanup()
+
+    def test_start_production_requires_evidence_matrix_ready(self) -> None:
+        article_id, _ = self._create_article_with_approved_plan()
+        research_response = self.client.post(
+            f"/api/articles/{article_id}/research/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, research_response.status_code, research_response.text)
+
+        start_response = self.client.post(
+            f"/api/articles/{article_id}/draft/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(409, start_response.status_code, start_response.text)
+        self.assertIn("Evidence matrix must be ready", start_response.text)
+
+    def test_start_production_creates_one_section_job_per_approved_plan_section(self) -> None:
+        article_id, approved_section_count = self._prepare_article_for_parallel_production()
+
+        start_production_response = self.client.post(
+            f"/api/articles/{article_id}/draft/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(202, start_production_response.status_code, start_production_response.text)
+
+        jobs = start_production_response.json()["jobs"]
+        section_jobs = [job for job in jobs if job["job_type"] == "SECTION_SCAFFOLD"]
+        self.assertEqual(
+            approved_section_count,
+            len(section_jobs),
+            "Expected one SECTION_SCAFFOLD job per approved plan section",
+        )
+
+    def test_section_jobs_run_independently_and_can_fail_independently(self) -> None:
+        article_id, _ = self._prepare_article_for_parallel_production()
+        created_jobs = self._start_parallel_production(article_id)
+        section_jobs = [job for job in created_jobs if job["job_type"] == "SECTION_SCAFFOLD"]
+        self.assertGreaterEqual(len(section_jobs), 2)
+
+        success_job = section_jobs[0]
+        failed_job = section_jobs[1]
+        complete_success = self._complete_job(
+            success_job["id"],
+            output={
+                "status": "SUCCEEDED",
+                "output_files": [{"path": "outputs/section-success.md"}],
+                "payload": {
+                    "used_evidence_ids": success_job["payload"]["used_evidence_ids"],
+                    "unsupported_claims": [],
+                    "draft_markdown": "Supported section draft",
+                },
+            },
+        )
+        self.assertEqual("SUCCEEDED", complete_success["status"])
+        self.assertIsNone(complete_success["error_category"])
+
+        unsupported_claims = [
+            {
+                "claim_text": "Unverified benchmark introduced during scaffolding.",
+                "risk_level": "high",
+            }
+        ]
+        complete_failure = self._complete_job(
+            failed_job["id"],
+            output={
+                "status": "SUCCEEDED",
+                "output_files": [{"path": "outputs/section-failed.md"}],
+                "payload": {
+                    "used_evidence_ids": failed_job["payload"]["used_evidence_ids"],
+                    "unsupported_claims": unsupported_claims,
+                    "draft_markdown": "Draft includes unsupported claim",
+                },
+            },
+        )
+        self.assertEqual("FAILED", complete_failure["status"])
+        self.assertEqual("UNSUPPORTED_CLAIMS_FOUND", complete_failure["error_category"])
+        self.assertEqual(unsupported_claims, complete_failure["payload"]["unsupported_claims"])
+
+    def test_retry_failed_section_does_not_rerun_successful_sections(self) -> None:
+        article_id, _ = self._prepare_article_for_parallel_production()
+        created_jobs = self._start_parallel_production(article_id)
+        section_jobs = [job for job in created_jobs if job["job_type"] == "SECTION_SCAFFOLD"]
+        self.assertGreaterEqual(len(section_jobs), 2)
+
+        success_job = self._complete_job(
+            section_jobs[0]["id"],
+            output={
+                "status": "SUCCEEDED",
+                "output_files": [{"path": "outputs/section-success.md"}],
+                "payload": {"unsupported_claims": []},
+            },
+        )
+        failed_job = self._complete_job(
+            section_jobs[1]["id"],
+            output={
+                "status": "FAILED",
+                "error_category": "CLI_EXIT_CODE_FAILURE",
+                "error_message": "Runner failed for one section",
+                "payload": {"unsupported_claims": []},
+            },
+        )
+
+        retry_response = self.client.post(
+            f"/api/agent-jobs/{failed_job['id']}/retry",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_ADMIN_EMAIL},
+        )
+        self.assertEqual(201, retry_response.status_code, retry_response.text)
+        retry_job = retry_response.json()["job"]
+        self.assertEqual("QUEUED", retry_job["status"])
+        self.assertEqual(failed_job["id"], retry_job["parent_job_id"])
+        self.assertEqual(2, retry_job["attempt"])
+
+        detail = self.client.get(
+            f"/api/articles/{article_id}",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()
+        successful_descendants = [
+            job
+            for job in detail["agent_jobs"]
+            if job.get("parent_job_id") == success_job["id"]
+        ]
+        failed_descendants = [
+            job
+            for job in detail["agent_jobs"]
+            if job.get("parent_job_id") == failed_job["id"]
+        ]
+        self.assertEqual([], successful_descendants)
+        self.assertEqual(1, len(failed_descendants))
+        self.assertEqual("QUEUED", failed_descendants[0]["status"])
+
+    def test_each_scaffold_payload_lists_used_evidence_ids(self) -> None:
+        article_id, _ = self._prepare_article_for_parallel_production()
+        evidence = self.client.get(
+            f"/api/articles/{article_id}/evidence",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()["evidence"]
+        evidence_ids = {item["id"] for item in evidence}
+
+        jobs = self._start_parallel_production(article_id)
+        section_jobs = [job for job in jobs if job["job_type"] == "SECTION_SCAFFOLD"]
+        self.assertTrue(section_jobs)
+        self.assertTrue(any(job["payload"]["used_evidence_ids"] for job in section_jobs))
+        for job in section_jobs:
+            self.assertIn("used_evidence_ids", job["payload"])
+            self.assertIsInstance(job["payload"]["used_evidence_ids"], list)
+            for evidence_id in job["payload"]["used_evidence_ids"]:
+                self.assertIn(evidence_id, evidence_ids)
+
+    def test_unsupported_claims_from_scaffold_are_captured_in_article_detail(self) -> None:
+        article_id, _ = self._prepare_article_for_parallel_production()
+        jobs = self._start_parallel_production(article_id)
+        section_job = [job for job in jobs if job["job_type"] == "SECTION_SCAFFOLD"][0]
+        unsupported_claims = [
+            {
+                "claim_text": "Unverified migration timeline claim",
+                "risk_level": "medium",
+            }
+        ]
+
+        completed_job = self._complete_job(
+            section_job["id"],
+            output={
+                "status": "SUCCEEDED",
+                "output_files": [{"path": "outputs/section-with-unsupported.md"}],
+                "payload": {
+                    "used_evidence_ids": section_job["payload"]["used_evidence_ids"],
+                    "unsupported_claims": unsupported_claims,
+                    "draft_markdown": "Draft with unsupported claim",
+                },
+            },
+        )
+        self.assertEqual("FAILED", completed_job["status"])
+        self.assertEqual("UNSUPPORTED_CLAIMS_FOUND", completed_job["error_category"])
+
+        detail_response = self.client.get(
+            f"/api/articles/{article_id}",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, detail_response.status_code, detail_response.text)
+        article_jobs = detail_response.json()["agent_jobs"]
+        refreshed_job = [job for job in article_jobs if job["id"] == section_job["id"]][0]
+        self.assertEqual(unsupported_claims, refreshed_job["payload"]["unsupported_claims"])
+        self.assertIn("Unsupported claims introduced during scaffolding", refreshed_job["error_message"])
+
+    def _start_parallel_production(self, article_id: str) -> list[dict[str, Any]]:
+        response = self.client.post(
+            f"/api/articles/{article_id}/draft/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(202, response.status_code, response.text)
+        return response.json()["jobs"]
+
+    def _complete_job(self, job_id: str, *, output: dict[str, Any]) -> dict[str, Any]:
+        response = self.client.post(
+            f"/internal/agent-jobs/{job_id}/complete",
+            json={
+                "workspace_path": f"/tmp/{job_id}",
+                "stdout": "fake section scaffolding runner\n",
+                "stderr": "",
+                "exit_code": 0,
+                "duration_ms": 2,
+                "output": output,
+            },
+        )
+        self.assertEqual(200, response.status_code, response.text)
+        return response.json()["job"]
+
+    def _prepare_article_for_parallel_production(self) -> tuple[str, int]:
+        article_id, approved_section_count = self._create_article_with_approved_plan()
+
+        research_response = self.client.post(
+            f"/api/articles/{article_id}/research/start",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, research_response.status_code, research_response.text)
+
+        evidence_response = self.client.get(
+            f"/api/articles/{article_id}/evidence",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, evidence_response.status_code, evidence_response.text)
+        self.assertFalse(evidence_response.json()["insufficient_evidence_reasons"])
+        self.assertEqual("EVIDENCE_MATRIX_READY", evidence_response.json()["article"]["status"])
+        return article_id, approved_section_count
+
+    def _create_article_with_approved_plan(self) -> tuple[str, int]:
+        site = self.client.get(
+            "/api/sites",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()[0]["site"]
+        article = self.client.post(
+            "/api/articles",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+            json={
+                "target_site_id": site["id"],
+                "brief_description": "Run parallel production jobs from approved sections.",
+                "working_title": "Parallel Production Jobs",
+                "content_type": "longform_guide",
+                "primary_keyword": "parallel production jobs",
+            },
+        ).json()["article"]
+        article_id = article["id"]
+
+        questions = self.client.post(
+            f"/api/articles/{article_id}/boundary-questions/generate",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        ).json()["questions"]
+        for question in questions:
+            if question["is_required"]:
+                patch_response = self.client.patch(
+                    f"/api/articles/{article_id}/boundary-questions/{question['id']}",
+                    headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+                    json={"answer": f"Answer for {question['category']}"},
+                )
+                self.assertEqual(200, patch_response.status_code, patch_response.text)
+
+        submit_response = self.client.post(
+            f"/api/articles/{article_id}/boundary-questions/submit",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, submit_response.status_code, submit_response.text)
+
+        plan_response = self.client.post(
+            f"/api/articles/{article_id}/plan/generate",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(201, plan_response.status_code, plan_response.text)
+        plan = plan_response.json()["plan"]
+        self.assertGreaterEqual(len(plan["sections"]), 1)
+
+        approve_response = self.client.post(
+            f"/api/articles/{article_id}/plans/{plan['id']}/approve",
+            headers={DEMO_USER_EMAIL_HEADER: DEMO_EDITOR_EMAIL},
+        )
+        self.assertEqual(200, approve_response.status_code, approve_response.text)
+        return article_id, len(plan["sections"])
+
+
+if __name__ == "__main__":
+    unittest.main()