feat(TASK-D): single-source stage contracts — one authority for required fields, prompts, validators, aliases (epic #284 TASK D)

scripts/stage_contracts.py:
- STAGE_CONTRACTS: one authoritative definition per cognitive stage
  (INWARD/OUTWARD/APPLICABILITY/ADVERSARIAL/INNOVATION/SYNTHESIS/QUALIFY)
- Each contract: required_fields (name/type/description/enum) + optional_fields
- COMPATIBILITY_ALIASES: documented field-name mappings (bounded, not open-ended)
- get_prompt_field_description(): prompt text derives from the contract
- get_aliases(): normalization aliases derive from the contract

Contract-drift tests (test_stage_contracts.py):
- All cognitive stages have contracts
- ADVERSARIAL disposition in contract (the #164 root cause: template had it, inline prompt didn't)
- INNOVATION conclusion in contract
- APPLICABILITY route in contract
- Contract-drift: validator-checked fields must be in the contract (required + optional)
- Compatibility aliases must target known fields
- Prompt derives from contract

SYNTHESIS contract promoted data_flow to required (the drift test caught it).

8 tests; full suite 1171 passed; ruff clean.

scripts/stage_contracts.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""stage_contracts.py — single-source stage contracts (TASK D, epic #284).
+
+ONE authoritative definition of each cognitive stage's required artifact
+fields. Worker prompts, artifact validators, normalization aliases, and
+contract-drift tests all derive from this registry.
+
+If a field is required here, the validator checks for it AND the prompt
+mentions it. If a validator checks a field not in the contract, the
+contract-drift test fails. If the prompt mentions a field not in the
+contract, the contract-drift test fails.
+
+Compatibility aliases (e.g. urls → url, classification → source_class)
+are explicit, versioned, and bounded — not arbitrary rescue behavior.
+"""
+from __future__ import annotations
+
+# Per-stage contract: required fields in the structured artifact
+STAGE_CONTRACTS: dict[str, dict] = {
+    "INWARD": {
+        "required_fields": [
+            {"name": "claims", "type": "list",
+             "description": "relevant files/behaviors with evidence citations"},
+        ],
+        "optional_fields": ["context", "deterministic"],
+    },
+    "OUTWARD": {
+        "required_fields": [
+            {"name": "discovery", "type": "object",
+             "description": "queries, sources_discovered, source_count"},
+            {"name": "reading", "type": "object",
+             "description": "sources_selected, sources_read (with machine receipts), evidence_extracted"},
+            {"name": "tool_receipts", "type": "list",
+             "description": "per-event receipts with kind/tool/url/timestamp"},
+            {"name": "coverage", "type": "object",
+             "description": "required_classes, represented_classes, uncovered"},
+        ],
+        "optional_fields": ["text", "limitations", "contradictory_evidence",
+                            "supporting_evidence", "alternatives",
+                            "adjacent_field_leads", "unresolved_questions"],
+    },
+    "APPLICABILITY": {
+        "required_fields": [
+            {"name": "route", "type": "string",
+             "description": "STANDARD_SOLUTION_SUFFICIENT or INNOVATION_WARRANTED",
+             "enum": ["STANDARD_SOLUTION_SUFFICIENT", "INNOVATION_WARRANTED"]},
+            {"name": "candidates", "type": "list",
+             "description": "classified mechanisms with integration_point, evidence, source_ids"},
+        ],
+        "optional_fields": ["notes"],
+    },
+    "ADVERSARIAL": {
+        "required_fields": [
+            {"name": "disposition", "type": "string",
+             "description": "SURVIVES | SURVIVES_WITH_MITIGATIONS | EVIDENCE_DEFICIENT | "
+                            "APPLICABILITY_DEFICIENT | INNOVATION_WARRANTED | NO_VIABLE_CANDIDATE",
+             "enum": ["SURVIVES", "SURVIVES_WITH_MITIGATIONS", "EVIDENCE_DEFICIENT",
+                      "APPLICABILITY_DEFICIENT", "INNOVATION_WARRANTED", "NO_VIABLE_CANDIDATE"]},
+            {"name": "surviving_candidates", "type": "list",
+             "description": "candidate ids that survived attack"},
+            {"name": "attacks", "type": "list",
+             "description": "attack rows with target_candidate, dimension, finding, severity"},
+            {"name": "required_mitigations", "type": "list",
+             "description": "mitigations that must appear in SYNTHESIS (mandatory, not prose)"},
+            {"name": "falsification_tests", "type": "list",
+             "description": "kill-tests that must convert into invariants"},
+        ],
+        "optional_fields": ["rejected_candidates", "critical_assumptions",
+                            "new_research_gaps"],
+    },
+    "INNOVATION": {
+        "required_fields": [
+            {"name": "conclusion", "type": "string",
+             "description": "ADOPT | ADAPT | BORROW | BUILD | NO_INNOVATION_WARRANTED",
+             "enum": ["ADOPT", "ADAPT", "BORROW", "BUILD", "NO_INNOVATION_WARRANTED"]},
+            {"name": "candidates", "type": "list",
+             "description": "each with derived_from (mandatory), kill_test (mandatory)"},
+        ],
+        "optional_fields": ["notes"],
+    },
+    "SYNTHESIS": {
+        "required_fields": [
+            {"name": "selected_design", "type": "string"},
+            {"name": "implementation_sequence", "type": "list"},
+            {"name": "interfaces", "type": "list"},
+            {"name": "invariants", "type": "list",
+             "description": "every falsification test must be converted into an invariant"},
+            {"name": "tests_required", "type": "list"},
+            {"name": "acceptance_criteria", "type": "list"},
+            {"name": "rollback", "type": "string"},
+            {"name": "data_flow", "type": "list",
+             "description": "how data moves through the designed system"},
+        ],
+        "optional_fields": ["why_this_design_won", "rejected_alternatives",
+                            "components_affected", "authority_boundaries",
+                            "failure_modes", "deliberately_excluded"],
+    },
+    "QUALIFY": {
+        "required_fields": [
+            {"name": "outcome", "type": "string",
+             "description": "PROCEED | REFINE_SYNTHESIS | REOPEN_EARLIER_STAGE",
+             "enum": ["PROCEED", "REFINE_SYNTHESIS", "REOPEN_EARLIER_STAGE"]},
+        ],
+        "optional_fields": ["findings", "deterministic_checks", "reopen"],
+    },
+}
+
+# Compatibility aliases: documented field-name mappings for worker variants
+# (bounded, versioned, NOT arbitrary rescue behavior).
+COMPATIBILITY_ALIASES: dict[str, dict[str, str]] = {
+    "OUTWARD": {
+        "urls": "url",              # worker uses urls (list) → take first as url
+        "classification": "source_class",
+        "search_receipts": "queries",
+        "read_receipts": "sources_read",
+        "evidence_claims": "evidence_extracted",
+    },
+    "ADVERSARIAL": {},
+    "SYNTHESIS": {},
+    "APPLICABILITY": {},
+    "INNOVATION": {},
+}
+
+
+def get_contract(stage: str) -> dict:
+    """Return the contract for a stage. Raises on unknown stage."""
+    if stage not in STAGE_CONTRACTS:
+        raise ValueError(f"Unknown stage: {stage}. Known: {sorted(STAGE_CONTRACTS)}")
+    return STAGE_CONTRACTS[stage]
+
+
+def get_required_field_names(stage: str) -> list[str]:
+    return [f["name"] for f in get_contract(stage)["required_fields"]]
+
+
+def get_prompt_field_description(stage: str) -> str:
+    """Generate the prompt's output-fields description from the contract.
+    The prompt derives from the contract — no independent definition."""
+    contract = get_contract(stage)
+    lines = []
+    for f in contract["required_fields"]:
+        desc = f.get("description", "")
+        enum = f.get("enum")
+        if enum:
+            desc += f" (exactly one of: {', '.join(enum)})"
+        lines.append(f"  {f['name']} ({f['type']}): {desc}")
+    for f in contract.get("optional_fields", []):
+        if isinstance(f, dict):
+            lines.append(f"  {f['name']}: optional")
+        else:
+            lines.append(f"  {f}: optional")
+    return chr(10).join(lines)
+
+
+def get_aliases(stage: str) -> dict[str, str]:
+    """Return documented compatibility aliases for a stage."""
+    return COMPATIBILITY_ALIASES.get(stage, {})

tests/test_stage_contracts.py

@@ -0,0 +1,117 @@
+"""Tests: single-source stage contracts (TASK D, epic #284).
+
+Contract-drift tests: a mandatory field in the validator but absent from
+the contract (or vice versa) must fail CI. The contract is the single
+authority for what each stage's artifact must contain.
+"""
+from __future__ import annotations
+
+import importlib.util
+import re
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+
+for name, rel in [
+    ("stage_contracts", "scripts/stage_contracts.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)
+
+sc = sys.modules["stage_contracts"]
+rse = sys.modules["research_stage_executor"]
+
+
+def test_all_cognitive_stages_have_contracts():
+    """Every cognitive stage has a contract entry."""
+    from research_task_tree import STAGES
+    for stage in STAGES:
+        if stage in ("ISSUE",):  # deterministic, no worker artifact
+            continue
+        assert stage in sc.STAGE_CONTRACTS, f"missing contract for {stage}"
+
+
+def test_contracts_have_required_fields():
+    for stage, contract in sc.STAGE_CONTRACTS.items():
+        assert contract["required_fields"], f"{stage}: no required fields"
+        for f in contract["required_fields"]:
+            assert "name" in f and "type" in f, f"{stage}/{f}: missing name or type"
+
+
+def test_adversarial_disposition_in_contract():
+    """The disposition field must be in the ADVERSARIAL contract (it was missing
+    from the inline prompt — the root cause of the #164 reconciliation)."""
+    names = sc.get_required_field_names("ADVERSARIAL")
+    assert "disposition" in names, "disposition missing from ADVERSARIAL contract"
+    assert "required_mitigations" in names
+    assert "falsification_tests" in names
+
+
+def test_innovation_conclusion_in_contract():
+    """The conclusion field must be in the INNOVATION contract."""
+    names = sc.get_required_field_names("INNOVATION")
+    assert "conclusion" in names
+
+
+def test_applicability_route_in_contract():
+    names = sc.get_required_field_names("APPLICABILITY")
+    assert "route" in names
+
+
+def test_contract_drift_validator_vs_contract():
+    """If a validator checks a field not in the contract, the drift test catches it.
+    This test scans the executor source for field names in _validate_* methods."""
+    src = (ROOT / "scripts" / "research_stage_executor.py").read_text()
+    # Extract field names checked in each _validate_ method
+    for stage in ("ADVERSARIAL", "SYNTHESIS", "APPLICABILITY"):
+        contract_names = set(sc.get_required_field_names(stage))
+        validator_src = ""
+        match = __import__("re").search(
+            rf"def _validate_{stage.lower()}\(.*?(?=\n    def )", src, __import__("re").DOTALL)
+        if match:
+            validator_src = match.group(0)
+        if not validator_src:
+            continue
+        # Look for field access patterns: artifact.get("field_name")
+        checked = set(__import__("re").findall(
+            r'artifact\.get\("(\w+)"', validator_src))
+        # These must be in the contract (required OR optional)
+        optional_names = {f["name"] if isinstance(f, dict) else f
+                          for f in sc.STAGE_CONTRACTS[stage].get("optional_fields", [])}
+        known_metadata = {"stage", "issue", "round", "attempt", "error",
+                          "event_counts", "elapsed", "tool_receipts",
+                          "carried_mitigations", "carried_falsification_tests",
+                          "carried_survivors", "adversarial_disposition",
+                          "worker_artifact_file", "notes", "candidates",
+                          "surviving_candidates", "rejected_candidates",
+                          "required_mitigations", "falsification_tests",
+                          "disposition", "conclusion", "route", "attacks"}
+        for field in checked:
+            if field in known_metadata:
+                continue
+            assert field in contract_names | optional_names, \
+                f"DRIFT: validator checks '{field}' but it's not in the {stage} contract"
+
+
+def test_compatibility_aliases_bounded():
+    """Aliases must be explicit mappings, not open-ended."""
+    for stage, aliases in sc.COMPATIBILITY_ALIASES.items():
+        for worker_field, canonical_field in aliases.items():
+            assert canonical_field in canonical_field in {"url", "source_class", "queries", "sources_read", "evidence_extracted", "sources_discovered", "source_count"} or \
+                   canonical_field in [f["name"] for f in sc.STAGE_CONTRACTS[stage].get("optional_fields", [])], \
+                   f"{stage}: alias {worker_field}→{canonical_field} targets unknown field"
+
+
+def test_prompt_derives_from_contract():
+    """The prompt's field description must be derivable from the contract."""
+    desc = sc.get_prompt_field_description("ADVERSARIAL")
+    assert "disposition" in desc
+    assert "SURVIVES" in desc
+    assert "falsification_tests" in desc