Nick
a553fb1363
ADVERSARIAL owns epistemic refinement. QUALIFY owns implementation-readiness. - ADVERSARIAL_DISPOSITIONS enum: SURVIVES / SURVIVES_WITH_MITIGATIONS / EVIDENCE_DEFICIENT / APPLICABILITY_DEFICIENT / INNOVATION_WARRANTED / NO_VIABLE_CANDIDATE. Only SURVIVES* unlock SYNTHESIS; every other disposition routes targeted refinement to the earliest repairable stage (EVIDENCE_DEFICIENT/NO_VIABLE_CANDIDATE → OUTWARD, APPLICABILITY_DEFICIENT → APPLICABILITY, INNOVATION_WARRANTED → INNOVATION) - advance() REQUIRES a valid adversarial disposition — an adversarial pass without a disposition cannot unlock anything - synthesis_readiness() deterministic invariant (§7): disposition unlocks AND ≥1 surviving candidate AND mitigations represented (for SURVIVES_WITH_MITIGATIONS) AND blocking unknowns = 0 - Falsification tests + required mitigations are first-class capsule fields — they reach SYNTHESIS as mandatory inputs, not optional prose (§5) - QUALIFY narrowed (§8–10): outcomes PROCEED / REFINE_SYNTHESIS (routes to SYNTHESIS only) / REOPEN_EARLIER_STAGE (exceptional; requires reason, return_to, new_blocking_fact, why_previous_gate_could_not_reasonably_ have_caught_it). Generic MORE_RESEARCH removed as a QUALIFY outcome. - Fake worker + all chain tests updated to the disposition/outcome contract Acceptance §18: 18/18 PASS (test_adversarial_gated_refinement.py). 148 tests pass, ruff clean.
408 lines
18 KiB
Python
408 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""research_task_tree.py — canonical research process state (RP1/RP2).
|
||
|
||
The ResearchTaskTree is the single source of truth for a research process:
|
||
which stage is current, which artifacts are accepted, which questions are
|
||
blocking. It persists state in the research artifact directory and syncs
|
||
to Gitea issue labels.
|
||
|
||
The tree is the authority. The TUI, MiMo capsules, and Gitea comments are
|
||
projections of this state — not independent truths.
|
||
|
||
Canonical pipeline (handoff 2026-09-08 — supersedes RSC0-9 stage design):
|
||
ISSUE → INWARD → OUTWARD → APPLICABILITY
|
||
→ optional INNOVATION → ADVERSARIAL → SYNTHESIS → QUALIFY
|
||
|
||
ISSUE replaces DEFINE: the Gitea issue is the authoritative statement of
|
||
intent; a deterministic issue-quality gate checks it is buildable, but
|
||
there is no cognitive DEFINE stage. OUTWARD is ONE unified natural MiMo
|
||
research task (OUTWARD_DISCOVERY/OUTWARD_READING are retired as stages;
|
||
their concepts survive only as artifact subsections).
|
||
|
||
Stage transitions:
|
||
StageCompleted(current) → advance to next stage
|
||
StageFailed(current, reason) → retry or escalate
|
||
REFINE(unknowns) → targeted return to the earliest deficient stage
|
||
(no full-pipeline restart; INWARD is not rerun unless Rune reality changed)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import time
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
STAGES = [
|
||
"ISSUE", "INWARD", "OUTWARD", "APPLICABILITY",
|
||
"INNOVATION", "ADVERSARIAL", "SYNTHESIS", "QUALIFY",
|
||
]
|
||
|
||
# Legacy stage names that must never become READY through the canonical path.
|
||
RETIRED_STAGES = {"DEFINE", "OUTWARD_DISCOVERY", "OUTWARD_READING"}
|
||
|
||
STAGE_ARTIFACT_FILES = {
|
||
"ISSUE": "issue_gate.json",
|
||
"INWARD": "inward.json",
|
||
"OUTWARD": "outward.json",
|
||
"APPLICABILITY": "applicability.json",
|
||
"ADVERSARIAL": "adversarial.json",
|
||
"INNOVATION": "innovation.json",
|
||
"SYNTHESIS": "synthesis.json",
|
||
"QUALIFY": "qualify.json",
|
||
}
|
||
|
||
# Candidate classifications for APPLICABILITY (§9 of canonical handoff).
|
||
APPLICABILITY_CLASSES = {
|
||
"DIRECTLY_APPLICABLE",
|
||
"APPLICABLE_WITH_ADAPTATION",
|
||
"CONFLICTS_WITH_RUNE",
|
||
"TOO_COMPLEX_FOR_VALUE",
|
||
"INSUFFICIENT_EVIDENCE",
|
||
"PROMISING_FOR_INNOVATION",
|
||
}
|
||
|
||
# APPLICABILITY routing decision (conditional INNOVATION correction).
|
||
# The default is SKIP — INNOVATION runs only when warranted.
|
||
APPLICABILITY_ROUTES = {
|
||
"STANDARD_SOLUTION_SUFFICIENT", # → ADVERSARIAL (INNOVATION skipped)
|
||
"INNOVATION_WARRANTED", # → INNOVATION → ADVERSARIAL
|
||
}
|
||
|
||
# Valid INNOVATION conclusions. NO_INNOVATION_WARRANTED is a SUCCESS result.
|
||
INNOVATION_CONCLUSIONS = {
|
||
"ADOPT", "ADAPT", "BORROW", "BUILD",
|
||
"NO_INNOVATION_WARRANTED",
|
||
}
|
||
|
||
# ── ADVERSARIAL disposition contract (pipeline correction 2026-09-09) ──────
|
||
# ADVERSARIAL owns epistemic refinement; QUALIFY owns implementation-readiness.
|
||
# Only SURVIVES* dispositions unlock SYNTHESIS. Everything else triggers
|
||
# targeted refinement at the earliest stage that can repair the defect.
|
||
ADVERSARIAL_DISPOSITIONS = {
|
||
"SURVIVES", # → SYNTHESIS READY
|
||
"SURVIVES_WITH_MITIGATIONS", # → SYNTHESIS READY (mitigations are required inputs)
|
||
"EVIDENCE_DEFICIENT", # → targeted OUTWARD → APPLICABILITY → ADVERSARIAL
|
||
"APPLICABILITY_DEFICIENT", # → APPLICABILITY → ADVERSARIAL
|
||
"INNOVATION_WARRANTED", # → INNOVATION → ADVERSARIAL
|
||
"NO_VIABLE_CANDIDATE", # → targeted OUTWARD (expand candidate space)
|
||
}
|
||
SYNTHESIS_UNLOCKING_DISPOSITIONS = {"SURVIVES", "SURVIVES_WITH_MITIGATIONS"}
|
||
|
||
# Targeted refinement routes per disposition (§4 of the correction).
|
||
ADVERSARIAL_REFINEMENT_ROUTES = {
|
||
"EVIDENCE_DEFICIENT": "OUTWARD",
|
||
"APPLICABILITY_DEFICIENT": "APPLICABILITY",
|
||
"INNOVATION_WARRANTED": "INNOVATION",
|
||
"NO_VIABLE_CANDIDATE": "OUTWARD",
|
||
}
|
||
|
||
# QUALIFY outcomes (narrowed semantics, §8–10 of the correction).
|
||
QUALIFY_OUTCOMES = {
|
||
"PROCEED", # implementation-ready → MiMo Code
|
||
"REFINE_SYNTHESIS", # build instructions weak → SYNTHESIS only
|
||
"REOPEN_EARLIER_STAGE", # exceptional: genuinely new blocking fact
|
||
}
|
||
|
||
|
||
class ResearchTaskTree:
|
||
"""Canonical research process state. One instance per research issue."""
|
||
|
||
def __init__(self, issue: int, round_no: int = 1, artifact_dir: str = ""):
|
||
self.issue = issue
|
||
self.round = round_no
|
||
self.artifact_dir = artifact_dir or f"research/issue-{issue}"
|
||
self.state = self._load()
|
||
|
||
def _state_file(self) -> str:
|
||
return os.path.join(self.artifact_dir, "task_tree.json")
|
||
|
||
def _lock_file(self) -> str:
|
||
return os.path.join(self.artifact_dir, "process.lock")
|
||
|
||
def acquire_process_lock(self) -> bool:
|
||
"""Idempotency guard: only one active ResearchProcess per issue.
|
||
Uses a lock file with TTL — stale locks from crashed workers expire."""
|
||
lock_path = Path(self._lock_file())
|
||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||
if lock_path.exists():
|
||
age = time.time() - lock_path.stat().st_mtime
|
||
ttl = int(os.environ.get("RESEARCH_LOCK_TTL", "7200")) # 2h default
|
||
if age < ttl:
|
||
return False # another process holds the lock
|
||
lock_path.unlink() # stale — take over
|
||
lock_path.write_text(f"pid={os.getpid()} acquired={time.time():.0f}")
|
||
return True
|
||
|
||
def release_process_lock(self) -> None:
|
||
lock_path = Path(self._lock_file())
|
||
if lock_path.exists():
|
||
lock_path.unlink()
|
||
|
||
def _load(self) -> dict:
|
||
path = os.path.join(self.artifact_dir, "task_tree.json")
|
||
if os.path.exists(path):
|
||
with open(path) as f:
|
||
state = json.load(f)
|
||
self.round = state.get("round", self.round)
|
||
return state
|
||
return {
|
||
"issue": self.issue,
|
||
"round": self.round,
|
||
"current_stage": STAGES[0],
|
||
"completed_stages": [],
|
||
"artifacts": {}, # stage -> artifact dict
|
||
"blocking_unknowns": [],
|
||
"status": "ACTIVE", # RUNNING | STALLED | COMPLETE | FAILED
|
||
"stage_history": [],
|
||
}
|
||
|
||
def save(self) -> None:
|
||
os.makedirs(self.artifact_dir, exist_ok=True)
|
||
path = os.path.join(self.artifact_dir, "task_tree.json")
|
||
with open(path, "w") as f:
|
||
json.dump(self.state, f, indent=2)
|
||
|
||
@property
|
||
def current_stage(self) -> str:
|
||
return self.state["current_stage"]
|
||
|
||
@property
|
||
def is_complete(self) -> bool:
|
||
return self.state["status"] == "COMPLETE"
|
||
|
||
def advance(self, artifact: dict) -> str:
|
||
"""Accept the current stage's artifact, advance to the next stage.
|
||
Returns the next stage name. Raises if already at QUALIFY.
|
||
Retired stage names cannot be advanced (cannot become READY)."""
|
||
stage = self.state["current_stage"]
|
||
if stage in RETIRED_STAGES:
|
||
raise ValueError(
|
||
f"Stage '{stage}' is retired (canonical handoff 2026-09-08); "
|
||
f"it cannot be executed or advanced. Canonical stages: {STAGES}"
|
||
)
|
||
if stage not in self.state["completed_stages"] or stage != "QUALIFY":
|
||
self.state["completed_stages"].append(stage)
|
||
self.state["artifacts"][stage] = artifact
|
||
self.state["stage_history"].append({
|
||
"stage": stage, "round": self.round,
|
||
"completed_at": "auto", "artifact_keys": list(artifact.keys()),
|
||
})
|
||
# Check if we're in a targeted branch
|
||
targeted = self.state.get("targeted_branch_stages", [])
|
||
if targeted and stage in targeted:
|
||
# In a targeted branch: advance within the branch
|
||
t_idx = targeted.index(stage)
|
||
if t_idx < len(targeted) - 1:
|
||
self.state["current_stage"] = targeted[t_idx + 1]
|
||
else:
|
||
# Targeted branch complete → return to SYNTHESIS v2
|
||
self.state["current_stage"] = self.state.get("targeted_return_stage", "SYNTHESIS")
|
||
self.state["targeted_branch_stages"] = []
|
||
else:
|
||
# Normal pipeline: advance to next stage in order — with
|
||
# stage-specific routing decisions:
|
||
if stage == "APPLICABILITY":
|
||
# APPLICABILITY decides whether INNOVATION runs (conditional).
|
||
self.state["current_stage"] = self._route_after_applicability(artifact)
|
||
elif stage == "ADVERSARIAL":
|
||
# ADVERSARIAL is the FINAL EPISTEMIC GATE (correction 2026-09-09):
|
||
# its disposition decides whether SYNTHESIS unlocks or refinement
|
||
# routes to the earliest repairable stage.
|
||
self.state["current_stage"] = self._route_after_adversarial(artifact)
|
||
elif stage == "QUALIFY":
|
||
self.state["current_stage"] = self._route_after_qualify(artifact)
|
||
else:
|
||
idx = STAGES.index(stage)
|
||
if idx < len(STAGES) - 1:
|
||
self.state["current_stage"] = STAGES[idx + 1]
|
||
else:
|
||
self.state["current_stage"] = "QUALIFY"
|
||
self.save()
|
||
return self.state["current_stage"]
|
||
|
||
def _route_after_adversarial(self, artifact: dict) -> str:
|
||
"""ADVERSARIAL disposition routing (correction §3–4).
|
||
|
||
SURVIVES / SURVIVES_WITH_MITIGATIONS → SYNTHESIS (gated by the
|
||
readiness invariant). Everything else routes to the earliest stage
|
||
that can repair the defect. Raises on missing/invalid disposition —
|
||
an adversarial pass without a disposition cannot unlock anything.
|
||
"""
|
||
disposition = artifact.get("disposition")
|
||
if disposition not in ADVERSARIAL_DISPOSITIONS:
|
||
raise ValueError(
|
||
f"ADVERSARIAL artifact must carry a valid disposition "
|
||
f"({sorted(ADVERSARIAL_DISPOSITIONS)}), got: {disposition!r}"
|
||
)
|
||
self.state["adversarial_disposition"] = disposition
|
||
if disposition in SYNTHESIS_UNLOCKING_DISPOSITIONS:
|
||
ok, problems = self.synthesis_readiness(artifact)
|
||
if not ok:
|
||
raise ValueError(
|
||
"SYNTHESIS readiness invariant failed: " + "; ".join(problems)
|
||
)
|
||
return "SYNTHESIS"
|
||
return ADVERSARIAL_REFINEMENT_ROUTES[disposition]
|
||
|
||
def _route_after_qualify(self, artifact: dict) -> str:
|
||
"""Narrowed QUALIFY routing (correction §8–10).
|
||
|
||
QUALIFY owns IMPLEMENTATION-READINESS, not epistemic refinement.
|
||
|
||
PROCEED → complete (implementation handoff)
|
||
REFINE_SYNTHESIS → SYNTHESIS only (never an OUTWARD pass)
|
||
REOPEN_EARLIER_STAGE → earliest affected stage, ONLY with a genuinely
|
||
new blocking fact + why earlier gates couldn't
|
||
have caught it (exceptional escape hatch)
|
||
"""
|
||
outcome = artifact.get("outcome", "PROCEED")
|
||
if outcome not in QUALIFY_OUTCOMES:
|
||
raise ValueError(
|
||
f"QUALIFY outcome must be one of {sorted(QUALIFY_OUTCOMES)}, got: {outcome!r}"
|
||
)
|
||
self.state["qualify_outcome"] = outcome
|
||
if outcome == "PROCEED":
|
||
self.state["status"] = "COMPLETE"
|
||
return "QUALIFY" # terminal
|
||
if outcome == "REFINE_SYNTHESIS":
|
||
return "SYNTHESIS"
|
||
# REOPEN_EARLIER_STAGE: all four required fields must be present
|
||
required_fields = ("reason", "return_to", "new_blocking_fact",
|
||
"why_previous_gate_could_not_reasonably_have_caught_it")
|
||
missing = [f for f in required_fields if not artifact.get(f)]
|
||
if missing:
|
||
raise ValueError(
|
||
f"REOPEN_EARLIER_STAGE requires {required_fields}; missing: {missing}"
|
||
)
|
||
return_to = artifact["return_to"]
|
||
if return_to not in STAGES or return_to == "QUALIFY":
|
||
raise ValueError(f"REOPEN_EARLIER_STAGE return_to must be an epistemic stage, got: {return_to!r}")
|
||
return return_to
|
||
|
||
def synthesis_readiness(self, adversarial_artifact: dict) -> tuple[bool, list[str]]:
|
||
"""Deterministic SYNTHESIS READY invariant (correction §7).
|
||
|
||
SYNTHESIS READY IFF:
|
||
- ADVERSARIAL disposition ∈ {SURVIVES, SURVIVES_WITH_MITIGATIONS}
|
||
- at least one surviving candidate exists
|
||
- required mitigations are represented (when disposition
|
||
is SURVIVES_WITH_MITIGATIONS)
|
||
- blocking adversarial unknowns = 0
|
||
"""
|
||
problems = []
|
||
disposition = adversarial_artifact.get("disposition")
|
||
if disposition not in SYNTHESIS_UNLOCKING_DISPOSITIONS:
|
||
problems.append(f"disposition {disposition!r} does not unlock SYNTHESIS")
|
||
survivors = adversarial_artifact.get("surviving_candidates", [])
|
||
if not survivors:
|
||
problems.append("no surviving candidate")
|
||
if disposition == "SURVIVES_WITH_MITIGATIONS" and not adversarial_artifact.get("required_mitigations"):
|
||
problems.append("SURVIVES_WITH_MITIGATIONS without required_mitigations — "
|
||
"mitigations are mandatory synthesis inputs, not optional prose")
|
||
if self.state.get("blocking_unknowns"):
|
||
problems.append("blocking adversarial unknowns present: "
|
||
f"{len(self.state['blocking_unknowns'])}")
|
||
return (len(problems) == 0, problems)
|
||
|
||
def _route_after_applicability(self, artifact: dict) -> str:
|
||
"""APPLICABILITY decides the INNOVATION branch (canonical correction).
|
||
|
||
STANDARD_SOLUTION_SUFFICIENT → ADVERSARIAL (INNOVATION skipped)
|
||
INNOVATION_WARRANTED → INNOVATION → ADVERSARIAL
|
||
|
||
INNOVATION exists to answer: can useful aspects be combined/borrowed
|
||
to produce something materially better? The standard researched
|
||
mechanism may already be the best solution. Default when the artifact
|
||
is silent: STANDARD_SOLUTION_SUFFICIENT (skip — do not manufacture
|
||
novelty for its own sake).
|
||
"""
|
||
route = artifact.get("route", "STANDARD_SOLUTION_SUFFICIENT")
|
||
if route not in ("STANDARD_SOLUTION_SUFFICIENT", "INNOVATION_WARRANTED"):
|
||
raise ValueError(
|
||
f"APPLICABILITY route must be STANDARD_SOLUTION_SUFFICIENT or "
|
||
f"INNOVATION_WARRANTED, got: {route!r}"
|
||
)
|
||
self.state["applicability_route"] = route
|
||
return "INNOVATION" if route == "INNOVATION_WARRANTED" else "ADVERSARIAL"
|
||
|
||
def fail(self, reason: str) -> None:
|
||
self.state["stage_history"].append({
|
||
"stage": self.state["current_stage"], "round": self.round,
|
||
"failed": True, "reason": reason,
|
||
})
|
||
self.save()
|
||
|
||
def more_research(self, unknowns: list[dict]) -> list[dict]:
|
||
"""Targeted refinement (QUALIFY → REFINE, canonical §14).
|
||
|
||
Routes to the EARLIEST deficient stage. No full-pipeline restart:
|
||
INWARD is not rerun unless Rune reality itself changed. Retired
|
||
stage names (OUTWARD_DISCOVERY/OUTWARD_READING) route to the
|
||
unified OUTWARD stage.
|
||
"""
|
||
self.state["blocking_unknowns"] = unknowns
|
||
self.round += 1
|
||
self.state["round"] = self.round
|
||
branches = []
|
||
all_stages = []
|
||
for u in unknowns:
|
||
u_type = u.get("type", "mixed")
|
||
if u_type in ("repo/inward", "direct-verification"):
|
||
stages_needed = ["INWARD"]
|
||
elif u_type in ("academic", "current-tech", "software-source"):
|
||
stages_needed = ["OUTWARD"]
|
||
elif u_type == "counter-evidence":
|
||
stages_needed = ["ADVERSARIAL"]
|
||
elif u_type == "applicability":
|
||
stages_needed = ["APPLICABILITY"]
|
||
elif u_type == "innovation":
|
||
stages_needed = ["INNOVATION"]
|
||
else:
|
||
stages_needed = ["OUTWARD"]
|
||
branches.append({
|
||
"unknown": u,
|
||
"stages": stages_needed,
|
||
"round": self.round,
|
||
})
|
||
all_stages.extend(stages_needed)
|
||
|
||
# Route to the earliest deficient stage in canonical order,
|
||
# then refresh downstream stages (targeted refinement path).
|
||
if all_stages:
|
||
first = min(all_stages, key=lambda s: STAGES.index(s))
|
||
self.state["current_stage"] = first
|
||
self.state["targeted_branch_stages"] = all_stages
|
||
self.state["targeted_return_stage"] = "SYNTHESIS"
|
||
self.save()
|
||
return branches
|
||
|
||
def get_prior_artifacts(self, for_stage: str) -> dict:
|
||
"""Return the accepted artifacts this stage needs."""
|
||
idx = STAGES.index(for_stage) if for_stage in STAGES else len(STAGES)
|
||
return {STAGES[i]: self.state["artifacts"].get(STAGES[i], {})
|
||
for i in range(idx)}
|
||
|
||
def capsule(self) -> dict:
|
||
"""The process capsule for MiMo invocation (§14)."""
|
||
return {
|
||
"process_id": f"research-{self.issue}-r{self.round}",
|
||
"issue": self.issue,
|
||
"round": self.round,
|
||
"current_stage": self.current_stage,
|
||
"completed_stages": self.state["completed_stages"],
|
||
"accepted_artifacts": {
|
||
# Adversarial-gated refinement (correction §5–6): mitigations
|
||
# and falsification tests MUST reach SYNTHESIS as required inputs.
|
||
k: {kk: v for kk, v in vv.items() if kk in (
|
||
"decision", "claims", "sources", "candidates", "disposition",
|
||
"required_mitigations", "falsification_tests",
|
||
"surviving_candidates", "rejected_candidates", "route",
|
||
)}
|
||
for k, vv in self.state["artifacts"].items()
|
||
},
|
||
"blocking_unknowns": self.state.get("blocking_unknowns", []),
|
||
"allowed_action_now": self.current_stage,
|
||
"expected_completion_artifact": STAGE_ARTIFACT_FILES.get(self.current_stage, "artifact.json"),
|
||
}
|