rune/scripts/process_experience.py

#!/usr/bin/env python3
"""process_experience.py — PL0 passive Experience Ledger (TASK I, epic #284).

Observation and evidence capture only. No autonomous self-editing.
Records Process Experience entries from pipeline runs: failures, successes,
inefficiencies — with root-cause attribution, evidence references, and
confidence. These are EVIDENCE RECORDS, not permanent rules.

Root-cause layers:
  MODEL_BEHAVIOR      — the model did something unexpected
  PROMPT_CONTRACT     — the prompt didn't communicate requirements clearly
  SCHEMA              — field names/structure mismatch
  NORMALIZER          — normalization was missing or incorrect
  TOOL_INTERFACE      — tool behavior differed from expectation
  ARTIFACT_ASSEMBLY   — artifact file composition issue
  VALIDATOR           — gate checked the wrong thing or missed something
  ROUTING             — the stage shouldn't have run for this task class
  STAGE_ARCHITECTURE  — the stage design itself is the problem
  DETERMINISTIC_RUNTIME — deterministic code issue (not model)
  EXTERNAL_DEPENDENCY — external service behaved unexpectedly
  UNKNOWN             — attribution uncertain
"""

from __future__ import annotations

import json
import os
import time
from datetime import datetime, timezone

ROOT_CAUSE_LAYERS = {
    "MODEL_BEHAVIOR",
    "PROMPT_CONTRACT",
    "SCHEMA",
    "NORMALIZER",
    "TOOL_INTERFACE",
    "ARTIFACT_ASSEMBLY",
    "VALIDATOR",
    "ROUTING",
    "STAGE_ARCHITECTURE",
    "DETERMINISTIC_RUNTIME",
    "EXTERNAL_DEPENDENCY",
    "UNKNOWN",
}

RESULT_TYPES = {"SUCCESS", "FAILURE", "INEFFICIENCY"}
# (extended below by the PL0 qualification block)


class ExperienceLedger:
    """Append-only ledger of Process Experience records."""

    def __init__(self, path: str):
        self.path = path
        os.makedirs(os.path.dirname(path), exist_ok=True)

    def record(
        self,
        task_run_stage: str,
        result_type: str,
        symptom: str,
        evidence: list[str],
        root_cause: str,
        confidence: float,
        human_intervention: bool = False,
        pipeline_sha: str = "UNKNOWN",
        correction_applied: str = "",
        recurrence_class: str = "",
        wall_cost_s: float | str = 0,
        model_cost_s: float | str = 0,
        tool_cost_s: float | str = 0,
        quality_outcome: str = "",
        reusable: bool = False,
        candidate_status: str = "SEED",
    ) -> dict:
        assert result_type in RESULT_TYPES, f"invalid result_type: {result_type}"
        assert root_cause in ROOT_CAUSE_LAYERS, f"invalid root_cause: {root_cause}"
        assert candidate_status in CANDIDATE_STATUSES, f"invalid status: {candidate_status}"
        entry = {
            "experience_id": f"PE-{int(time.time())}-{len(self.load_all()) + 1:04d}",
            "task_run_stage": task_run_stage,
            "result_type": result_type,
            "symptom": symptom,
            "evidence_references": evidence,
            "root_cause_layer": root_cause,
            "confidence": confidence,
            "human_intervention": human_intervention,
            "pipeline_sha": pipeline_sha,
            "correction_applied": correction_applied,
            "recurrence_class": recurrence_class,
            "wall_cost_s": wall_cost_s,
            "model_cost_s": model_cost_s,
            "tool_cost_s": tool_cost_s,
            "quality_outcome": quality_outcome,
            "reusable_lesson": reusable,
            "candidate_status": candidate_status,
            "recorded_at": datetime.now(timezone.utc).isoformat(),
        }
        with open(self.path, "a") as f:
            f.write(json.dumps(entry) + "\n")
        return entry

    def load_all(self) -> list[dict]:
        if not os.path.exists(self.path):
            return []
        out = []
        with open(self.path) as f:
            for line in f:
                try:
                    out.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
        return out

    def by_root_cause(self, layer: str) -> list[dict]:
        return [e for e in self.load_all() if e["root_cause_layer"] == layer]

    def successes(self) -> list[dict]:
        return [e for e in self.load_all() if e["result_type"] == "SUCCESS"]

    def failures(self) -> list[dict]:
        return [e for e in self.load_all() if e["result_type"] == "FAILURE"]

    def reusable_lessons(self) -> list[dict]:
        return [e for e in self.load_all() if e["reusable_lesson"]]


def record_operator_outcomes(task_state_path: str | None = None, ledger_path: str | None = None) -> list[dict]:
    """Consume operator task outcomes into the Experience Ledger (Runtime
    Reachability P0, 2026-09-10).

    The operator's send_task persists every task in the TaskStateRepository
    checkpoint (task_state.json colocated with operator.json) and emits a
    durable task.outcome journal event per terminal transition. This
    consumer reads the CHECKPOINT (the authoritative task store — the
    journal is a one-way integrity journal: payload_hash only, no payload)
    and appends a SEED experience record for every terminal task not yet
    recorded (matched by evidence task_id — idempotent).

    Attribution stays UNKNOWN: attribution is PL1's job, not the ledger's.
    Statuses stay SEED/CANDIDATE — never auto-VALIDATED.
    """
    path = task_state_path or os.path.expanduser("~/.rune/hot/task_state.json")
    ledger = ExperienceLedger(
        ledger_path
        or os.environ.get("RUNE_EXPERIENCE_LEDGER", os.path.expanduser("~/.rune/state/process_experience.jsonl"))
    )
    with open(path) as f:
        checkpoint = json.load(f)
    tasks = checkpoint.get("tasks", [])
    seen_tasks = set()
    for rec in ledger.load_all():
        for ev in rec.get("evidence_references", []):
            if ev.startswith("task:"):
                seen_tasks.add(ev.split(":", 1)[1])
    recorded = []
    for t in tasks:
        status = t.get("status", "")
        if status not in ("completed", "failed"):
            continue
        # research_required tasks: routing disposition, NOT failure evidence
        # (handoff §2/§C). The dispatch lives in route.jsonl; recording it
        # as FAILURE would poison PL0 with false failure-learning pressure.
        if t.get("outcome_ref") == "research_dispatch" or t.get("result") == "RESEARCH_REQUIRED":
            continue
        task_id = t.get("task_id", "")
        if not task_id or task_id in seen_tasks:
            continue
        result_type = "SUCCESS" if status == "completed" else "FAILURE"
        rec = ledger.record(
            task_run_stage=f"operator:{status}",
            result_type=result_type,
            symptom=(t.get("title") or "")[:200],
            evidence=[f"task:{task_id}", f"model:{t.get('assigned_model')}", f"outcome_ref:{t.get('outcome_ref')}"],
            root_cause="UNKNOWN",
            confidence=0.3,
            pipeline_sha="operator",
            reusable=False,
            candidate_status="SEED",
        )
        recorded.append(rec)
    return recorded


if __name__ == "__main__":  # pragma: no cover — manual consumer invocation
    import argparse

    ap = argparse.ArgumentParser(description="Record operator task outcomes as experience evidence")
    ap.add_argument("--task-state", default=None)
    ap.add_argument("--ledger", default=None)
    args = ap.parse_args()
    recs = record_operator_outcomes(args.task_state, args.ledger)
    print(f"recorded {len(recs)} experience records")


# PL0 qualification (handoff 2026-09-10 §9): raw outcomes are preserved
# verbatim, but only QUALIFIED records exert learning pressure in PL1.
# The distinction is cheap and rule-based — no model call.
CANDIDATE_STATUSES = {"SEED", "CANDIDATE", "EVALUATED", "PROMOTED", "REJECTED"}

# Task titles that are outcome telemetry, not learning material: trivial
# greetings, arithmetic/canary probes, bootstrap self-checks.
_TELEMETRY_TITLE_PATTERNS = (
    "hello",
    "how are you",
    "hi ",
    "thanks",
    "thank you",
    "reply with exactly",
    "what is ",
    "check",
    "status",
)


def qualify_experience(record: dict) -> str:
    """Classify a raw outcome record: TELEMETRY vs CANDIDATE.

    TELEMETRY = simple successful conversation / canary probes — kept as
    raw outcome evidence, excluded from PL1 evaluation pressure.
    CANDIDATE = failures, verification failures, retries, corrections,
    inefficiencies, architectural omissions, reusable procedures, route
    misclassifications — the material PL1 evaluates.
    """
    if record.get("result_type") in ("FAILURE", "INEFFICIENCY"):
        return "CANDIDATE"
    symptom = (record.get("symptom") or "").lower()
    if any(p in symptom for p in _TELEMETRY_TITLE_PATTERNS):
        return "TELEMETRY"
    # Simple short successes with no verification/failure signal are
    # telemetry; anything substantive defaults to CANDIDATE (PL1 decides).
    if len(symptom) < 60 and not record.get("quality_outcome"):
        return "TELEMETRY"
    return "CANDIDATE"


def qualified_candidates(ledger_path: str | None = None) -> list[dict]:
    """All CANDIDATE-qualified records — the PL1 evaluation input."""
    ledger = ExperienceLedger(
        ledger_path
        or os.environ.get("RUNE_EXPERIENCE_LEDGER", os.path.expanduser("~/.rune/state/process_experience.jsonl"))
    )
    return [r for r in ledger.load_all() if qualify_experience(r) == "CANDIDATE"]