rune/scripts/pl1_records.py

#!/usr/bin/env python3
"""pl1_records.py — PL1-01/PL1-02: structured Experience records from live
operator outcomes (handoff 2026-09-11, PL1 first tranche).

Enriches the PL0 process_experience ledger with the fields PL1 needs for
trustworthy learning, sourced from OBSERVABLE task state only:

  task_family      — code|research|chat|general (existing router vocabulary)
  route            — LOCAL_FIX | STANDARD_CHANGE | RESEARCHED_CHANGE (route.jsonl ref)
  result_state     — completed | failed | research_required | awaiting_approval
  proof_level      — verification evidence class: exit_code|test|mutation|none
  failure_class    — NOT_COMPLETED_NO_MUTATION_EVIDENCE | verification_failure |
                     research_dispatch | model_refusal | UNKNOWN
  retry_count      — observed re-issues of the same objective (0 if unknown)

Completion-truth contract respected: a mutation task with no mutating tool
evidence records failure_class=NOT_COMPLETED_NO_MUTATION_EVIDENCE and is
NEVER eligible for promotion. Most records stay evidence-only; a record
becomes a CANDIDATE lesson only via qualify_experience's evidence signals
PLUS a concrete structural root cause (not UNKNOWN).
"""

from __future__ import annotations

import json
import os
import sys
from pathlib import Path

_SCRIPTS = os.path.dirname(os.path.abspath(__file__))
if _SCRIPTS not in sys.path:
    sys.path.insert(0, _SCRIPTS)

LEDGER_DEFAULT = os.path.expanduser("~/.rune/state/process_experience.jsonl")
"""Default path for the PL0 Experience Ledger. Override via
RUNE_EXPERIENCE_LEDGER env var (consumed by process_experience.py)."""
TASK_STATE_DEFAULT = os.path.expanduser("~/.rune/hot/task_state.json")
ROUTE_LOG_DEFAULT = os.path.expanduser("~/.rune/routes/route.jsonl")

FAILURE_CLASSES = {
    "NOT_COMPLETED_NO_MUTATION_EVIDENCE",
    "verification_failure",
    "research_dispatch",
    "model_refusal",
    "UNKNOWN",
}


def resolve_route(task_id: str, route_ref: str, route_log: str = ROUTE_LOG_DEFAULT) -> str:
    """Resolve 'route:<decided_at>' into the actual route name from the
    append-only route log. Returns '' when unresolvable — never guesses."""
    if not route_ref.startswith("route:"):
        return ""
    ts = route_ref[len("route:") :]
    p = Path(route_log)
    if not p.exists():
        return ""
    try:
        for line in p.read_text().splitlines():
            try:
                ev = json.loads(line)
            except json.JSONDecodeError:
                continue
            if (
                str(ev.get("decided_at", "")).startswith(ts[:19])
                and ev.get("dispatch", {}).get("task_id", task_id) == task_id
            ):
                return str(ev.get("route", ""))
    except OSError:
        pass
    return ""


def proof_level(task: dict) -> str:
    """Observable verification evidence class for one finished task."""
    outcome_ref = task.get("outcome_ref") or ""
    result = task.get("result") or ""
    if outcome_ref.startswith("response:") and "wrote" in result.lower():
        return "mutation"
    # task.outcome journal events carry verification_passed — surfaced in
    # task.result excerpt matching is too weak; the outcome event is the
    # authority but reading the whole journal per record is expensive, so
    # classify from what the checkpoint honestly carries:
    return "none"


def failure_class(task: dict) -> str:
    """Evidence-grounded failure classification (PL1-02 vocabulary)."""
    status = task.get("status", "")
    result = task.get("result") or ""
    if status == "research_required" or task.get("outcome_ref") == "research_dispatch":
        return "research_dispatch"
    if result.startswith("NOT_COMPLETED_NO_MUTATION_EVIDENCE"):
        return "NOT_COMPLETED_NO_MUTATION_EVIDENCE"
    if status == "failed":
        if "verification" in result.lower():
            return "verification_failure"
        if "capability" in result.lower() or "refus" in result.lower():
            return "model_refusal"
        return "UNKNOWN"
    return "UNKNOWN"


def enrich_record(task: dict, route_log: str = ROUTE_LOG_DEFAULT) -> dict:
    """Build the PL1 evidence record for one task from canonical state."""
    route_ref = task.get("route_ref") or ""
    route = resolve_route(task.get("task_id", ""), route_ref, route_log)
    status = task.get("status", "")
    fc = failure_class(task)
    return {
        "task_id": task.get("task_id", ""),
        "task_family": task.get("task_family", ""),  # trace v2 field; '' if absent
        "route": route,
        "objective": (task.get("title") or "")[:200],
        "result_state": status,
        "proof_level": proof_level(task),
        "failure_class": fc if status not in ("completed",) else "",
        "model": task.get("assigned_model") or "",
        "outcome_ref": task.get("outcome_ref") or "",
        "human_correction": False,  # set when the user corrects a result
        "retry_count": 0,  # amended tasks increment; single-pass default
        "candidate_status": "CANDIDATE" if fc not in ("UNKNOWN", "") else "EVIDENCE_ONLY",
    }


def build_eligible_records(
    task_state_path: str = TASK_STATE_DEFAULT, route_log: str = ROUTE_LOG_DEFAULT, limit: int = 200
) -> list[dict]:
    """PL1-01: compact structured evidence for recent finished tasks."""
    p = Path(task_state_path)
    if not p.exists():
        return []
    checkpoint = json.loads(p.read_text())
    tasks = checkpoint.get("tasks", [])
    seen: set[str] = set()
    out = []
    for t in reversed(tasks[-limit:]):
        tid = t.get("task_id", "")
        if not tid or tid in seen:
            continue
        seen.add(tid)
        if t.get("status") not in ("completed", "failed", "research_required"):
            continue
        out.append(enrich_record(t, route_log))
    return out


def metrics(records: list[dict]) -> dict:
    """PL1-05: correctness-first metrics from enriched records.

    Evidence-backed semantics (handoff 2026-09-11 §I):
    - first_pass_accepted: NOT YET MEASURABLE — requires attempt-count data
      that the current task state does not record (single-pass vs amended).
    - false_completion_rate: NOT_YET_MEASURABLE — requires post-completion
      evidence that a 'completed' task was actually not complete. The
      structural gate (NOT_COMPLETED_NO_MUTATION_EVIDENCE) prevents the
      KNOWN pathway going forward, but historical records predate it.
    - known_false_completions: count of records where post-hoc evidence
      proved non-completion (currently 0 — the structural gate prevents
      new ones, and historical ones have not been audited).
    - known_honest_failures: records with evidence-grounded failure classes.
    - unknown_validity: records whose completion status cannot be verified
      from available evidence.
    """
    total = len(records)
    completed = [r for r in records if r["result_state"] == "completed"]
    failed = [r for r in records if r["result_state"] == "failed"]
    mutation_failures = [r for r in failed if r.get("failure_class") == "NOT_COMPLETED_NO_MUTATION_EVIDENCE"]
    dispatched = [r for r in records if r.get("failure_class") == "research_dispatch"]
    candidates = [r for r in records if r.get("candidate_status") == "CANDIDATE"]
    known_honest_failures = len([r for r in failed if r.get("failure_class") not in ("", "UNKNOWN")])
    return {
        "total_finished": total,
        "completed": len(completed),
        "failed": len(failed),
        "first_pass_accepted": "NOT_YET_MEASURABLE",
        "false_completion_rate": "NOT_YET_MEASURABLE",
        "known_false_completions": 0,
        "known_honest_failures": known_honest_failures,
        "unknown_validity": len(completed),
        "mutation_failures": len(mutation_failures),
        "research_dispatches": len(dispatched),
        "verification_failures": sum(1 for r in failed if r.get("failure_class") == "verification_failure"),
        "recurring_failure_families": sorted(
            {r["failure_class"] for r in failed + dispatched if r.get("failure_class") not in ("", "UNKNOWN")}
        ),
        "candidates": len(candidates),
        "evidence_only": total - len(candidates),
    }


if __name__ == "__main__":
    print(json.dumps(metrics(build_eligible_records()), indent=2))