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