#!/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"]