requirements-ci.txt

@@ -82,7 +82,7 @@ pygments==2.21.0
     #   pytest
     #   rich
     #   textual
-pyright==1.1.412
+pyright==1.1.413
     # via rune (pyproject.toml)
 pytest==9.1.1
     # via

scripts/research_stage_executor.py

@@ -48,11 +48,13 @@ def artifact_hash(artifact: dict) -> str:
 class StageExecutor:
     """Executes exactly one research stage. No self-advancing."""
 
-    def __init__(self, issue: int, artifact_dir: str, round_no: int = 1):
+    def __init__(self, issue: int, artifact_dir: str, round_no: int = 1,
+                 allow_stdout_fallback: bool = False):
         self.tree = ResearchTaskTree(issue=issue, round_no=round_no, artifact_dir=artifact_dir)
         self.issue = issue
         self.artifact_dir = artifact_dir
         self.round_no = round_no
+        self.allow_stdout_fallback = allow_stdout_fallback
         self.event_log = []
 
     def execute(self, event: dict) -> dict:
@@ -257,6 +259,47 @@ class StageExecutor:
             "artifact": artifact,
         }
 
+    def _resolve_worker_artifact(self, stage: str, relay_result: dict) -> tuple[dict | None, str]:
+        """ADAPT-9/#298 artifact authority.
+
+        The declared structured artifact FILE is the truth; stdout is
+        observability only. Resolution order:
+          1. declared stage artifact file (STAGE_ARTIFACT_FILES)
+          2. LEGACY_COMPATIBILITY_ONLY: JSON extracted from stdout — used only
+             when the file is absent, so historical tools/tests keep working;
+             canonical execution never depends on this path.
+        Returns (worker_artifact_or_None, channel) where channel is
+        "file" | "stdout-legacy" | "none".
+        """
+        from research_task_tree import STAGE_ARTIFACT_FILES
+        artifact_file = os.path.join(
+            self.artifact_dir, STAGE_ARTIFACT_FILES.get(stage, f"{stage.lower()}.json"))
+        if os.path.exists(artifact_file):
+            try:
+                with open(artifact_file) as f:
+                    worker_artifact = json.load(f)
+                if isinstance(worker_artifact, dict) and worker_artifact:
+                    return worker_artifact, "file"
+            except (json.JSONDecodeError, OSError):
+                pass  # unreadable file → ARTIFACT_MISSING
+        # LEGACY_COMPATIBILITY_ONLY (NOT CANONICAL — REMOVE AFTER legacy tests
+        # are rewritten): stdout JSON extraction. Canonical execution (all
+        # current stage paths) treats stdout as observability and requires the
+        # declared file. This branch exists so historical fixtures still parse;
+        # it is reachable only when allow_stdout_fallback=True is passed.
+        if self.allow_stdout_fallback and relay_result.get("texts"):
+            import re as _re
+            text = "\n\n".join(relay_result["texts"])
+            m = _re.search(r"\{.*\}", text, _re.DOTALL)
+            if m:
+                try:
+                    worker_artifact = json.loads(m.group(0))
+                    if isinstance(worker_artifact, dict):
+                        return worker_artifact, "stdout-legacy"
+                except json.JSONDecodeError:
+                    pass
+        return None, "none"
+
     def _execute_outward(self, event: dict) -> dict:
         """OUTWARD: one coherent MiMo external-research pass (discovery + reading + evidence extraction)."""
         import re as _re
@@ -367,13 +410,15 @@ class StageExecutor:
                 return {"status": "STAGE_COMPLETED", "stage": "OUTWARD",
                         "artifact_hash": artifact_digest, "next_stage": next_stage,
                         "artifact": artifact}
-            error_detail = ""
-            if os.path.exists(err_path):
-                with open(err_path) as f:
-                    error_detail = f.read()[-300:]
-            self.tree.fail("no text output. stderr: " + error_detail)
+            # ADAPT-9/#298: stdout was prose (or empty) and the declared
+            # artifact file is absent. stdout content cannot establish stage
+            # success — this is ARTIFACT_MISSING, full stop.
+            self.tree.fail("ARTIFACT_MISSING: declared structured artifact "
+                           f"{STAGE_ARTIFACT_FILES.get('OUTWARD', 'outward.json')} "
+                           "was not written and stdout carried no usable JSON")
             return {"status": "STAGE_FAILED", "stage": "OUTWARD",
-                    "error": "no text output", "stderr": error_detail}
+                    "error": "ARTIFACT_MISSING",
+                    "stderr": self._tail_err(err_path)}
 
         # Prose-only stdout with a valid dossier on disk: prefer the file.
         if relay_result["texts"] and not _re.search(r"\{.*\}", "\n\n".join(relay_result["texts"]), _re.DOTALL):
@@ -716,29 +761,14 @@ class StageExecutor:
             return {"status": "STAGE_FAILED", "stage": "ADVERSARIAL",
                     "error": "mimo exited " + str(relay_result["returncode"])}
 
-        worker_artifact = None
-        if relay_result["texts"]:
-            import re as _re
-            text = "\n\n".join(relay_result["texts"])
-            m = _re.search(r"\{.*\}", text, _re.DOTALL)
-            if m:
-                try:
-                    worker_artifact = json.loads(m.group(0))
-                except json.JSONDecodeError:
-                    worker_artifact = None
-        # File channel: checked whenever stdout parse yields no usable artifact
-        if not worker_artifact:
-            artifact_file = os.path.join(self.artifact_dir, "adversarial.json")
-            if os.path.exists(artifact_file):
-                try:
-                    with open(artifact_file) as f:
-                        worker_artifact = json.load(f)
-                except (json.JSONDecodeError, OSError):
-                    pass
-        if not worker_artifact:
-            self.tree.fail("ADVERSARIAL: no dossier produced (stdout or file)")
+        # ADAPT-9/#298 artifact authority (declared file = truth)
+        worker_artifact, channel = self._resolve_worker_artifact("ADVERSARIAL", relay_result)
+        if worker_artifact is None:
+            self.tree.fail("ARTIFACT_MISSING: declared structured artifact "
+                           "adversarial.json was not written and stdout carried "
+                           "no usable JSON")
             return {"status": "STAGE_FAILED", "stage": "ADVERSARIAL",
-                    "error": "no artifact produced", "stderr": self._tail_err(err_path)}
+                    "error": "ARTIFACT_MISSING", "stderr": self._tail_err(err_path)}
 
         artifact = self._normalize_adversarial_artifact(worker_artifact, relay_result, event)
         gate_errors = self._validate_adversarial(artifact)
@@ -893,28 +923,14 @@ class StageExecutor:
             return {"status": "STAGE_FAILED", "stage": "INNOVATION",
                     "error": "mimo exited " + str(relay_result["returncode"])}
 
-        worker_artifact = None
-        if relay_result["texts"]:
-            import re as _re
-            text = "\n\n".join(relay_result["texts"])
-            m = _re.search(r"\{.*\}", text, _re.DOTALL)
-            if m:
-                try:
-                    worker_artifact = json.loads(m.group(0))
-                except json.JSONDecodeError:
-                    worker_artifact = None
-        if not worker_artifact:
-            f = os.path.join(self.artifact_dir, "innovation.json")
-            if os.path.exists(f):
-                try:
-                    with open(f) as fh:
-                        worker_artifact = json.load(fh)
-                except (json.JSONDecodeError, OSError):
-                    pass
-        if not worker_artifact:
-            self.tree.fail("INNOVATION: no dossier produced (stdout or file)")
+        # ADAPT-9/#298 artifact authority (declared file = truth)
+        worker_artifact, channel = self._resolve_worker_artifact("INNOVATION", relay_result)
+        if worker_artifact is None:
+            self.tree.fail("ARTIFACT_MISSING: declared structured artifact "
+                           "innovation.json was not written and stdout carried "
+                           "no usable JSON")
             return {"status": "STAGE_FAILED", "stage": "INNOVATION",
-                    "error": "no artifact produced", "stderr": self._tail_err(err_path)}
+                    "error": "ARTIFACT_MISSING", "stderr": self._tail_err(err_path)}
 
         artifact = self._normalize_innovation_artifact(worker_artifact, relay_result, event)
         gate_errors = self._validate_innovation(artifact)
@@ -1062,28 +1078,14 @@ class StageExecutor:
             return {"status": "STAGE_FAILED", "stage": "SYNTHESIS",
                     "error": "mimo exited " + str(relay_result["returncode"])}
 
-        worker_artifact = None
-        if relay_result["texts"]:
-            import re as _re
-            text = "\n\n".join(relay_result["texts"])
-            m = _re.search(r"\{.*\}", text, _re.DOTALL)
-            if m:
-                try:
-                    worker_artifact = json.loads(m.group(0))
-                except json.JSONDecodeError:
-                    worker_artifact = None
-        if not worker_artifact:
-            artifact_file = os.path.join(self.artifact_dir, "synthesis.json")
-            if os.path.exists(artifact_file):
-                try:
-                    with open(artifact_file) as f:
-                        worker_artifact = json.load(f)
-                except (json.JSONDecodeError, OSError):
-                    pass
-        if not worker_artifact:
-            self.tree.fail("SYNTHESIS: no dossier produced (stdout or file)")
+        # ADAPT-9/#298 artifact authority (declared file = truth)
+        worker_artifact, channel = self._resolve_worker_artifact("SYNTHESIS", relay_result)
+        if worker_artifact is None:
+            self.tree.fail("ARTIFACT_MISSING: declared structured artifact "
+                           "synthesis.json was not written and stdout carried "
+                           "no usable JSON")
             return {"status": "STAGE_FAILED", "stage": "SYNTHESIS",
-                    "error": "no artifact produced", "stderr": self._tail_err(err_path)}
+                    "error": "ARTIFACT_MISSING", "stderr": self._tail_err(err_path)}
 
         artifact = self._normalize_synthesis_artifact(worker_artifact, relay_result, event)
         gate_errors = self._validate_synthesis(artifact)

tests/test_adapt9_artifact_authority.py

@@ -0,0 +1,227 @@
+"""Tests: ADAPT-9/#298 — artifact authority + retry authority.
+
+Artifact authority: the declared structured artifact file is the truth;
+stdout content cannot establish stage success. Prose stdout + missing file
+→ STAGE_FAILED / ARTIFACT_MISSING, no advancement.
+
+Retry authority: attempt/event identity is append-only; retry = new attempt;
+no .consumed_* deletion; failed attempts cannot advance; history never
+erased; duplicate delivery is a no-op.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+import unittest.mock as mock
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+
+for name, rel in [
+    ("research_task_tree", "scripts/research_task_tree.py"),
+    ("research_stage_executor", "scripts/research_stage_executor.py"),
+]:
+    if name not in sys.modules:
+        spec = importlib.util.spec_from_file_location(name, ROOT / rel)
+        mod = importlib.util.module_from_spec(spec)
+        sys.modules[name] = mod
+        spec.loader.exec_module(mod)
+
+rtt = sys.modules["research_task_tree"]
+rse = sys.modules["research_stage_executor"]
+
+
+@pytest.fixture
+def mock_gitea(monkeypatch):
+    ISSUE_DATA = {"number": 100, "title": "t", "body": "b", "labels": []}
+    monkeypatch.setattr(rse.StageExecutor, "_fetch_issue", lambda self: ISSUE_DATA)
+
+
+@pytest.fixture
+def tree(tmp_path):
+    t = rtt.ResearchTaskTree(issue=995, round_no=1,
+                             artifact_dir=str(tmp_path / "research/issue-995"))
+    for stage in ("ISSUE", "INWARD", "OUTWARD"):
+        t.advance({"test": True})
+    t.advance({"candidates": [{"id": "M1", "classification": "DIRECTLY_APPLICABLE"}],
+               "route": "STANDARD_SOLUTION_SUFFICIENT"})
+    assert t.current_stage == "ADVERSARIAL"
+    return t
+
+
+def _relay_result(texts):
+    return {"returncode": 0, "texts": texts, "event_counts": {},
+            "receipts": [], "elapsed": 0.0}
+
+
+def test_plausible_stdout_json_without_file_is_artifact_missing(tree, mock_gitea):
+    """THE nasty case: MiMo stdout carries plausible JSON claiming success,
+    but the declared artifact file is absent → FAILED, no advancement."""
+    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=1)
+    texts = ['ADVERSARIAL completed successfully {"surviving_candidates": ["M1"], '
+             '"attacks": [{"id": "A1", "target_candidate": "M1", '
+             '"dimension": "failure modes", "finding": "x"}], '
+             '"disposition": "SURVIVES"}']
+    worker, channel = ex._resolve_worker_artifact("ADVERSARIAL", _relay_result(texts))
+    # stdout JSON exists but the FILE is the authority — the helper returns the
+    # legacy-compat parse only for backward compatibility; the executor must
+    # mark the channel and the ADVERSARIAL path requires the file.
+    # Assert the declared artifact file does NOT exist:
+    assert not (Path(tree.artifact_dir) / "adversarial.json").exists()
+    # Execute the stage via the real path with relay mocked:
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result(texts)):
+        result = ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                             "attempt": 1, "event_id": "nasty-1"})
+    assert result["status"] == "STAGE_FAILED"
+    assert result["error"] == "ARTIFACT_MISSING"
+    # No advancement, no accepted artifact
+    assert tree.current_stage == "ADVERSARIAL"
+    assert "ADVERSARIAL" not in tree.state["completed_stages"]
+    assert tree.state["artifacts"].get("ADVERSARIAL") in (None, {})
+
+
+def test_prose_stdout_with_valid_file_passes(tree, mock_gitea, tmp_path):
+    """The inverse: stdout is ordinary prose, artifact file is valid → PASS."""
+    dossier = {"disposition": "SURVIVES", "surviving_candidates": ["M1"],
+               "rejected_candidates": [], "required_mitigations": [],
+               "falsification_tests": ["FT1: mode switch must be labeled"],
+               "attacks": GOOD_ATTACKS}
+    (Path(tree.artifact_dir) / "adversarial.json").write_text(json.dumps(dossier))
+    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=1)
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result(["working on it...", "done"])):
+        result = ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                             "attempt": 1, "event_id": "inverse-1"})
+    assert result["status"] == "STAGE_COMPLETED"
+    # The executor holds its own tree instance (same persisted dir) — assert there
+    assert ex.tree.current_stage == "SYNTHESIS"
+    assert ex.tree.state["adversarial_disposition"] == "SURVIVES"
+
+
+def test_file_takes_precedence_over_stdout_json(tree, mock_gitea):
+    """Both channels present: the FILE wins (single authority)."""
+    file_dossier = {"disposition": "SURVIVES", "surviving_candidates": ["M1"],
+                    "rejected_candidates": [], "required_mitigations": [],
+                    "falsification_tests": [],
+                    "attacks": [{"id": "A1", "target_candidate": "M1",
+                                 "dimension": "failure modes", "finding": "from file"}]}
+    (Path(tree.artifact_dir) / "adversarial.json").write_text(json.dumps(file_dossier))
+    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=1)
+    worker, channel = ex._resolve_worker_artifact(
+        "ADVERSARIAL",
+        _relay_result(['{"disposition": "SURVIVES", "surviving_candidates": ["OTHER"], '
+                       '"attacks": []}']))
+    assert channel == "file"
+    assert worker["surviving_candidates"] == ["M1"]
+
+
+# ── Retry authority ──────────────────────────────────────────────────────────
+
+def test_retry_uses_new_event_id_no_marker_deletion(tree, mock_gitea):
+    """FAILED → retry(new event) → COMPLETED. No .consumed_* deletion; all
+    attempts auditable; exactly one advancement."""
+    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=1)
+    # attempt 1: FAIL via artifact missing
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result(["prose only, no file"])):
+        r1 = ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                         "attempt": 1, "event_id": "retry-ADVERSARIAL-1"})
+    assert r1["status"] == "STAGE_FAILED"
+    assert (Path(tree.artifact_dir) / ".consumed_retry-ADVERSARIAL-1").exists()
+
+    # attempt 2: GATE_FAILED (file present but gate-invalid: no disposition)
+    bad = {"surviving_candidates": ["M1"], "attacks": GOOD_ATTACKS.copy()}
+    # missing disposition → tree.advance raises... but gate validates first?
+    # The normalize passes disposition=None; the executor's gate is dimensional,
+    # the disposition check lives in tree.advance. Simulate GATE_FAILED via
+    # invalid attacks (dimension coverage failure):
+    bad["attacks"] = [{"id": "A1", "target_candidate": "M1",
+                       "dimension": "failure modes", "finding": "x"}]
+    (Path(tree.artifact_dir) / "adversarial.json").write_text(json.dumps(bad))
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result([])):
+        r2 = ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                         "attempt": 2, "event_id": "retry-ADVERSARIAL-2"})
+    # Missing 6 of 7 dimensions + no falsification → GATE_FAILED
+    assert r2["status"] in ("GATE_FAILED", "GATE_RETRY")
+
+    # attempt 3: COMPLETED (full valid dossier)
+    good = {"disposition": "SURVIVES", "surviving_candidates": ["M1"],
+            "rejected_candidates": [], "required_mitigations": [],
+            "falsification_tests": ["FT1: mode switch must be labeled"],
+            "attacks": GOOD_ATTACKS}
+    (Path(tree.artifact_dir) / "adversarial.json").write_text(json.dumps(good))
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result([])):
+        r3 = ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                         "attempt": 3, "event_id": "retry-ADVERSARIAL-3"})
+    assert r3["status"] == "STAGE_COMPLETED"
+    # executor holds its own tree instance (same persisted dir)
+    assert ex.tree.current_stage == "SYNTHESIS"
+    # exactly one advancement for ADVERSARIAL
+    assert ex.tree.state["completed_stages"].count("ADVERSARIAL") == 1
+
+    # No marker deletion ever required: all three consumed markers still exist
+    for ev in ("retry-ADVERSARIAL-1", "retry-ADVERSARIAL-2", "retry-ADVERSARIAL-3"):
+        assert (Path(tree.artifact_dir) / f".consumed_{ev}").exists()
+
+    # Duplicate delivery of the COMPLETED event → no-op
+    r4 = ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                     "attempt": 3, "event_id": "retry-ADVERSARIAL-3"})
+    assert r4["status"] == "DUPLICATE"
+    assert ex.tree.state["completed_stages"].count("ADVERSARIAL") == 1
+
+
+GOOD_ATTACKS = [
+    {"id": "A1", "target_candidate": "M1", "dimension": "contradicting evidence",
+     "finding": "f1", "severity": "minor"},
+    {"id": "A2", "target_candidate": "M1", "dimension": "weak assumptions",
+     "finding": "f2", "severity": "minor"},
+    {"id": "A3", "target_candidate": "M1", "dimension": "failure modes",
+     "finding": "f3", "severity": "moderate"},
+    {"id": "A4", "target_candidate": "M1", "dimension": "simpler alternative",
+     "finding": "f4", "severity": "minor"},
+    {"id": "A5", "target_candidate": "M1", "dimension": "duplicate authority",
+     "finding": "f5", "severity": "minor"},
+    {"id": "A6", "target_candidate": "M1", "dimension": "constraint violations",
+     "finding": "none", "severity": "none"},
+]
+
+
+def test_failed_attempt_cannot_advance(tree, mock_gitea):
+    """A failed attempt never advances the tree, even on retry of same stage."""
+    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=1)
+    before = tree.current_stage
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result(["prose"])):
+        r = ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                        "attempt": 1, "event_id": "noadv-1"})
+    assert r["status"] == "STAGE_FAILED"
+    assert ex.tree.current_stage == before
+    assert "ADVERSARIAL" not in ex.tree.state["completed_stages"]
+
+
+def test_retry_does_not_overwrite_history(tree, mock_gitea):
+    """The failed attempt's history stays in stage_history after a successful retry."""
+    ex = rse.StageExecutor(issue=tree.issue, artifact_dir=tree.artifact_dir, round_no=1)
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result(["prose"])):
+        ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                    "attempt": 1, "event_id": "hist-1"})
+    fails_before = len([h for h in tree.state["stage_history"] if h.get("failed")])
+    good = {"disposition": "SURVIVES", "surviving_candidates": ["M1"],
+            "rejected_candidates": [], "required_mitigations": [],
+            "falsification_tests": ["FT1: mode switch must be labeled"],
+            "attacks": GOOD_ATTACKS}
+    (Path(tree.artifact_dir) / "adversarial.json").write_text(json.dumps(good))
+    with mock.patch("research_event_relay.ResearchEventRelay.run",
+                    return_value=_relay_result([])):
+        ex.execute({"stage": "ADVERSARIAL", "issue": tree.issue,
+                    "attempt": 2, "event_id": "hist-2"})
+    fails_after = len([h for h in tree.state["stage_history"] if h.get("failed")])
+    assert fails_after >= fails_before, "failed history must not be erased"