rune/scripts/research_define.py

#!/usr/bin/env python3
"""research_define.py — deterministic DEFINE stage (R1).

DEFINE compiles the research contract from the Gitea issue.
It is DETERMINISTIC — it reads the issue body and produces the contract.
MiMo is invoked ONLY if the issue is ambiguous (no clear objective/constraints).

DEFINE does NOT: inspect code, search externally, or produce a design.
DEFINE does: classify the question, identify sub-questions, define the evidence
contract, and produce the required-stage plan.
"""

from __future__ import annotations

import json
import re
from typing import Any

# Keywords that suggest external research is needed
EXTERNAL_INDICATORS = [
    "best practices",
    "state of the art",
    "alternative",
    "compare",
    "benchmark",
    "library",
    "framework",
    "prior art",
    "existing solutions",
    "how do other",
    "state of the art",
    "literature",
    "approach",
]

# Keywords that suggest repo-specific work
REPO_INDICATORS = [
    "fix",
    "bug",
    "refactor",
    "add",
    "implement",
    "remove",
    "optimize",
    "update",
    "migrate",
    "clean up",
    "wire",
    "integrate",
    "test",
]


def compile_define_contract(issue: dict) -> dict:
    """Deterministically compile the issue contract from a Gitea issue.

    ISSUE is a deterministic intake/gate pseudo-stage (canonical handoff
    2026-09-08): validates fields, classifies, compiles the contract.
    No model invocation, no research, no repository browsing.
    INNOVATION is CONDITIONAL — included only when the route warrants it;
    the standard researched mechanism may already be sufficient.
    """
    title = issue.get("title", "")
    body = issue.get("body", "") or ""
    labels = [l["name"] for l in issue.get("labels", []) if isinstance(l, dict)]
    full_text = f"{title}\n{body}".lower()

    # Classify question type
    has_external = any(ind in full_text for ind in EXTERNAL_INDICATORS)
    has_repo = any(ind in full_text for ind in REPO_INDICATORS)

    if has_external and has_repo:
        question_type = "mixed"
    elif has_external:
        question_type = "current-tech"
    elif has_repo:
        question_type = "repo-specific"
    else:
        question_type = "repo-specific"  # default: issues on this repo are repo-specific

    # Extract sub-questions from the body (lines with "?" or numbered items)
    sub_questions = []
    for line in body.split("\n"):
        line = line.strip()
        if line.endswith("?"):
            sub_questions.append(line)
        elif re.match(r"^\d+\.", line):
            sub_questions.append(line)

    if not sub_questions:
        sub_questions = [f"Address: {title}"]

    # Canonical required stages (ISSUE gate replaces DEFINE; OUTWARD is unified)
    # Every issue gets INWARD (what does Rune currently do?) and SYNTHESIS + QUALIFY
    required_stages = ["ISSUE", "INWARD"]

    if question_type in ("mixed", "current-tech", "academic"):
        required_stages.append("OUTWARD")

    if question_type in ("mixed", "current-tech") or "adversarial" in full_text:
        required_stages.append("ADVERSARIAL")

    # INNOVATION is CONDITIONAL: only when design-choice indicators suggest
    # a credible opportunity for something materially better. Never mandatory.
    design_indicators = ["design", "architecture", "approach", "strategy", "pattern", "choose"]
    innovation_warranted = any(ind in full_text for ind in design_indicators) or question_type == "mixed"
    if innovation_warranted:
        required_stages.append("INNOVATION")

    required_stages.extend(["SYNTHESIS", "QUALIFY"])

    # Skipped stages with reasons (canonical stage names only)
    all_stages = {"ISSUE", "INWARD", "OUTWARD", "APPLICABILITY", "ADVERSARIAL", "INNOVATION", "SYNTHESIS", "QUALIFY"}
    skipped = {}
    for s in sorted(all_stages - set(required_stages)):
        if s == "OUTWARD":
            skipped[s] = "No external factual uncertainty identified in the issue."
        elif s == "ADVERSARIAL":
            skipped[s] = "No design choice requiring adversarial evaluation."
        elif s == "INNOVATION":
            # Conditional stage: standard solution may be sufficient
            skipped[s] = "STANDARD_SOLUTION_SUFFICIENT — INNOVATION is conditional, not ceremonial."
        elif s == "APPLICABILITY":
            skipped[s] = "No OUTWARD research to ground against Rune."

    # Ambiguity signal: recorded for the process, NOT a MiMo trigger.
    # ISSUE never invokes a model; downstream stages decide how to handle
    # an ambiguous contract via the blocking-unknowns mechanism.
    ambiguous = question_type == "repo-specific" and not has_repo and not has_external and len(body) < 50

    return {
        "issue": issue.get("number"),
        "title": title,
        "question_type": question_type,
        "sub_questions": sub_questions,
        "required_stages": required_stages,
        "skipped_stages": skipped,
        "evidence_classes": _evidence_classes(question_type),
        "sufficient_evidence": "≥2 code-path citations for INWARD; ≥2 unique URLs for OUTWARD (if required)",
        "expected_artifact": "synthesis.json",
        "ambiguous": ambiguous,
        # COMPATIBILITY ONLY — always False. ISSUE never invokes MiMo;
        # ambiguity is carried by the `ambiguous` flag instead.
        # REMOVE AFTER no caller reads needs_mimo.
        "needs_mimo": False,
        "source": "deterministic",
    }


def _evidence_classes(question_type: str) -> list[str]:
    classes = ["repo"]
    if question_type in ("mixed", "current-tech", "academic"):
        classes.append("external")
    if question_type == "academic":
        classes.append("academic")
    return classes


def is_ambiguous(issue: dict) -> bool:
    """Check if the issue is too ambiguous for deterministic classification.
    Ambiguity is a signal for downstream stages — never a MiMo trigger."""
    contract = compile_define_contract(issue)
    return contract["ambiguous"]


def main_with(argv: list[str]) -> int:
    import argparse

    parser = argparse.ArgumentParser(description="Deterministic DEFINE stage")
    parser.add_argument("--issue-file", required=True, help="Path to issue JSON")
    parser.add_argument("--out", required=True, help="Output path for define.json")
    args = parser.parse_args(argv[1:])

    with open(args.issue_file) as f:
        issue = json.load(f)

    contract = compile_define_contract(issue)
    ambiguous = is_ambiguous(issue)

    result = {
        "contract": contract,
        "needs_mimo": ambiguous,
    }

    with open(args.out, "w") as f:
        json.dump(result, f, indent=2, default=str)

    print(
        f"DEFINE contract: type={contract['question_type']}, "
        f"stages={contract['required_stages']}, "
        f"mimo_needed={ambiguous}"
    )
    return 0


if __name__ == "__main__":
    import sys

    sys.exit(main_with(sys.argv))