Nick
ca4f7593d7
CORRECTION 1: #314 reopened — recall_with_associations mode field defect CONFIRMED in current source and FIXED (mode: association_merged). CORRECTION 3: Gitea credential removed from remote URL (credential helper). TOKEN EXPOSED: yes (in diagnostic output). TOKEN ROTATED: not yet (requires Gitea web UI access). Remote URL now shows username only, no secret. CORRECTION 4: duplicate-release-commit defect ROOT CAUSE: convergence computed warranted bump from commits-since-tag without checking if master version already satisfies the bump. FIX: already-converged guard in version_convergence.py (master version > last tag → no commit). Repo-aware (args.repo passed to read_version/latest_tag). CORRECTION 6: lifecycle terminalization — execution exception now guarantees task terminalizes as failed with evidence (no immortal RUNNING). CORRECTION 7: route instruction task-scoped on RuneTask.route_instruction (NOT shared operator state). Read-only leading-verb override. Isolation tests prove no cross-contamination between interleaved tasks. PL1-01/02/03/05: pl1_records.py (route resolution, proof level, failure class, candidate status), metrics with honest NOT_YET_MEASURABLE semantics, route-instruction learning loop (PE-1789102658-0269 SEED). 1253 tests green; ratchet 352/80 (shrink); henxels clean.
437 lines
18 KiB
Python
437 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(f"blocking adversarial unknowns present: {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 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"),
|
||
}
|