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"}
CANDIDATE_STATUSES = {"SEED", "EVALUATED", "PROMOTED", "REJECTED"}


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"]]