rune/scripts/stage_contracts.py

#!/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, {})