Nick
03e994ebb3
- research-dispatcher.yml: label event → API query → stage dispatch (automatic) - research_event_relay.py: live event relay (QueryIssued/SourceRead/ToolFailed etc.) - research_task_tree.py: process lock (idempotency), round sync from persisted state - research_prompt_compiler.py: dynamic stage-scoped prompts from tree state - run_research_stage.py: one-stage executor + tree persistence to master - fake_worker.py: deterministic 8-stage integration harness (15 tests green) - Pipeline extension guide committed 15 relevant pipeline tests passed. Proven: 8-stage chain, MORE_RESEARCH targeted branch, cross-workspace persistence, evidence handoff.
213 lines
8.4 KiB
Python
213 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
|
"""research_task_tree.py — canonical research process state (RSC1).
|
|
|
|
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.
|
|
|
|
Stages (in order):
|
|
DEFINE → INWARD → OUTWARD_DISCOVERY → OUTWARD_READING
|
|
→ ADVERSARIAL → INNOVATION → SYNTHESIS → QUALIFY
|
|
|
|
Stage transitions:
|
|
StageCompleted(current) → advance to next stage
|
|
StageFailed(current, reason) → retry or escalate
|
|
MORE_RESEARCH(unknowns) → create targeted child nodes, re-enter at the
|
|
appropriate stage for each unknown's type
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
STAGES = [
|
|
"DEFINE", "INWARD", "OUTWARD_DISCOVERY", "OUTWARD_READING",
|
|
"ADVERSARIAL", "INNOVATION", "SYNTHESIS", "QUALIFY",
|
|
]
|
|
|
|
STAGE_ARTIFACT_FILES = {
|
|
"DEFINE": "define.json",
|
|
"INWARD": "inward.json",
|
|
"OUTWARD_DISCOVERY": "outward_discovery.json",
|
|
"OUTWARD_READING": "outward_reading.json",
|
|
"ADVERSARIAL": "adversarial.json",
|
|
"INNOVATION": "innovation.json",
|
|
"SYNTHESIS": "synthesis.json",
|
|
"QUALIFY": "qualify.json",
|
|
}
|
|
|
|
|
|
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."""
|
|
stage = self.state["current_stage"]
|
|
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
|
|
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 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]:
|
|
"""Create targeted child research nodes for blocking unknowns.
|
|
Transitions the tree to the targeted branch's first stage.
|
|
Does NOT rerun DEFINE/INWARD unless the unknown invalidates them."""
|
|
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",):
|
|
stages_needed = ["INWARD"]
|
|
elif u_type in ("academic",):
|
|
stages_needed = ["OUTWARD_DISCOVERY", "OUTWARD_READING"]
|
|
elif u_type in ("current-tech", "software-source"):
|
|
stages_needed = ["OUTWARD_DISCOVERY", "OUTWARD_READING"]
|
|
elif u_type == "counter-evidence":
|
|
stages_needed = ["ADVERSARIAL"]
|
|
elif u_type == "direct-verification":
|
|
stages_needed = ["INWARD"]
|
|
else:
|
|
stages_needed = ["OUTWARD_DISCOVERY", "OUTWARD_READING"]
|
|
branches.append({
|
|
"unknown": u,
|
|
"stages": stages_needed,
|
|
"round": self.round,
|
|
})
|
|
all_stages.extend(stages_needed)
|
|
|
|
# Transition to the targeted branch's first stage
|
|
# (after the targeted stages, SYNTHESIS v2 → QUALIFY v2 runs)
|
|
if all_stages:
|
|
self.state["current_stage"] = all_stages[0]
|
|
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": {
|
|
k: {kk: v for kk, v in vv.items() if kk in ("decision", "claims", "sources", "candidates", "disposition")}
|
|
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"),
|
|
}
|