Nick
b2a43a3a92
- research-dispatcher.yml: triggers on issues.labeled, checks label via API (deterministic), dispatches research-stage via workflow_dispatch (proven). Fixes the label-event payload quirk that silently skipped API-applied labels. - research-stage-headless.yml: dispatch-only trigger (dispatcher handles label events); structured artifact generation from mimo events (receipt + URLs + code paths + disposition) validated by research_artifact.py before commit. - The prose comment stays as human-readable rendering; the ARTIFACT is the gate authority (Nick: replace prose-regex authority with structured artifact validation).
105 lines
4.6 KiB
Python
105 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""research_artifact.py — structured research-round schema + validation (P0.4).
|
|
|
|
Per the Forge Cognitive Research Loop handoff §8: machine-readable research
|
|
round with causal evidence links. Phases are FROZEN artifacts — the later
|
|
phase's input is the persisted prior artifact (§7: "the later phase must not
|
|
be able to pretend it saw research that the pipeline never recorded").
|
|
|
|
Schema (minimal, causally sufficient):
|
|
issue, round, receipt (from run_agent_stage.build_receipt),
|
|
phases: {inward: {claims: [{id, claim, evidence[]}]},
|
|
outward: {sources: [{id, url, title, retrieved_at, tool,
|
|
content_hash, learned}],
|
|
claims: [{id, claim, supports[]}]},
|
|
innovation: {candidates: [{id, proposal, derived_from[],
|
|
expected_gain, risks[]}]},
|
|
synthesis: {decision, evidence_used[], blocking_unknowns[],
|
|
feasibility, kill_test,
|
|
disposition: PROCEED|MORE_RESEARCH}}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def validate_outward(outward: dict) -> list[str]:
|
|
"""§9: OUTWARD valid only with real discovery — ≥2 sources, each with
|
|
url + tool receipt + learned note; ≥1 search query recorded."""
|
|
errs = []
|
|
sources = outward.get("sources", [])
|
|
if len(sources) < 2:
|
|
errs.append(f"OUTWARD: {len(sources)} sources < 2 (real discovery required)")
|
|
for s_ in sources:
|
|
for field in ("id", "url", "tool", "learned"):
|
|
if not s_.get(field):
|
|
errs.append(f"OUTWARD source {s_.get('id', '?')}: missing {field}")
|
|
if not re.match(r"https?://", s_.get("url", "")):
|
|
errs.append(f"OUTWARD source {s_.get('id', '?')}: url not http(s)")
|
|
if not outward.get("queries"):
|
|
errs.append("OUTWARD: no search queries recorded (discovery, not recall)")
|
|
return errs
|
|
|
|
|
|
def validate_causal_links(artifact: dict) -> list[str]:
|
|
"""§11/§12: innovation candidates derive_from prior claims; synthesis
|
|
decisions cite evidence_used. At least one candidate depends on OUTWARD
|
|
evidence unless synthesis records 'no external mechanism relevant'."""
|
|
errs = []
|
|
phases = artifact.get("phases", {})
|
|
inward_ids = {c["id"] for c in phases.get("inward", {}).get("claims", [])}
|
|
outward_ids = {c["id"] for c in phases.get("outward", {}).get("claims", [])}
|
|
candidates = phases.get("innovation", {}).get("candidates", [])
|
|
synthesis = phases.get("synthesis", {})
|
|
known = inward_ids | outward_ids
|
|
for c in candidates:
|
|
refs = set(c.get("derived_from", []))
|
|
if not refs:
|
|
errs.append(f"INNOVATION {c.get('id', '?')}: no derived_from (decorative, not causal)")
|
|
continue
|
|
if not refs & known:
|
|
errs.append(f"INNOVATION {c.get('id', '?')}: derived_from references unknown claims {sorted(refs - known)}")
|
|
outward_dependent = any(set(c.get("derived_from", [])) & outward_ids for c in candidates)
|
|
if not outward_dependent and "no external mechanism relevant" not in synthesis.get("decision", ""):
|
|
errs.append("no INNOVATION candidate derives from OUTWARD evidence — research was decorative")
|
|
if not synthesis.get("evidence_used"):
|
|
errs.append("SYNTHESIS: evidence_used empty — decisions untraceable")
|
|
if synthesis.get("disposition") == "MORE_RESEARCH" and not synthesis.get("blocking_unknowns"):
|
|
errs.append("MORE_RESEARCH disposition without named blocking_unknowns")
|
|
return errs
|
|
|
|
|
|
def validate(artifact: dict) -> list[str]:
|
|
errs = validate_outward(artifact.get("phases", {}).get("outward", {}))
|
|
errs += validate_causal_links(artifact)
|
|
if artifact.get("phases", {}).get("synthesis", {}).get("disposition") not in ("PROCEED", "MORE_RESEARCH"):
|
|
errs.append("synthesis.disposition must be PROCEED or MORE_RESEARCH")
|
|
return errs
|
|
|
|
|
|
def main_with(argv: list[str]) -> int:
|
|
artifact = json.loads(Path(argv[1]).read_text())
|
|
errs = validate(artifact)
|
|
if errs:
|
|
print("research artifact INVALID:")
|
|
for e in errs:
|
|
print(f" - {e}")
|
|
return 1
|
|
print("research artifact VALID (outward discovery + causal links verified)")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main_with(sys.argv))
|
|
|
|
|
|
def extract_disposition(final_text: str) -> str:
|
|
"""Extract disposition from the agent's final text (the agent self-declares
|
|
the disposition; the artifact validation checks causal consistency)."""
|
|
m = re.search(r"Disposition:\s*(PROCEED|MORE_RESEARCH)", final_text, re.IGNORECASE)
|
|
return m.group(1).upper() if m else ""
|