feat: add causal learning selection

README.md

@@ -351,6 +351,8 @@ Use `/agent auto|coder|researcher` to control routing. `/learning status [profil
 
 Inferred facts and preferences remain candidates until repeated independent evidence supports them. Explicit owner claims activate immediately. Typed global, profile, project, and task scopes resolve deterministically; `/memory why <id>` explains a claim and `/memory forget <id>` removes it together with solely dependent learned behavior.
 
+Selection ranks relevant guidance by verified utility per context character and avoids artifact pairs with repeated verified failures. Paired omission trials can retire guidance only when removing it does not reduce verified completion; retirement is reversible and preserves the learned artifact pre-image.
+
 ### Memory storage
 
 OpenKyrozen v2 uses **SQLite as the source of truth** (`~/.kyrozen/v2/openkyrozen.sqlite3`) and ChromaDB as a rebuildable semantic index. Memories have a kind, scope, confidence, source events, and lifecycle status. Workspaces and sessions are isolated, raw observations are marked as data, and `/forget` removes records by durable ID. If ChromaDB is unavailable, SQLite keeps durable keyword retrieval.
@@ -404,6 +406,9 @@ KYROZEN_SERVER_TOKEN=change-me python server.py --host 0.0.0.0 --port 8000
 | `GET` | `/api/v2/learning/metrics?profile=...` | Profile completion, correction, error, tool, token, and latency metrics |
 | `GET` | `/api/v2/learning/{id}/evidence` | Proof card, applicability, replay, and outcome receipts |
 | `POST` | `/api/v2/learning/{id}/replay` | Record paired sandboxed candidate/predecessor replay results |
+| `POST` | `/api/v2/learning/{id}/omission` | Record paired with/without-artifact results |
+| `POST` | `/api/v2/learning/{id}/retire` | Retire an artifact with non-regressing omission evidence |
+| `POST` | `/api/v2/learning/{id}/restore` | Restore a retired artifact as a canary |
 | `POST` | `/api/v2/learning/{id}/rollback` | Roll back an activated proposal |
 | `GET` | `/api/v2/memory/claims` | Typed memory claims with provenance and scope |
 | `GET/DELETE` | `/api/v2/memory/claims/{id}` | Explain or dependency-completely forget a claim |

learning_benchmark.py

@@ -98,6 +98,8 @@ def main(argv: list[str] | None = None) -> None:
     parser.add_argument("--cases", required=True, type=Path, help="JSONL cases with id, profile, and task")
     parser.add_argument("--clean-runner", required=True, help="Command reading one case JSON from stdin")
     parser.add_argument("--evolved-runner", required=True, help="Command reading one case JSON from stdin")
+    parser.add_argument("--ablation", action="append", default=[], metavar="NAME=COMMAND",
+                        help="Additional no-memory, candidate, predecessor, or omission runner")
     parser.add_argument("--output", type=Path)
     parser.add_argument("--timeout", type=float, default=300.0)
     args = parser.parse_args(argv)
@@ -109,6 +111,13 @@ def main(argv: list[str] | None = None) -> None:
               "clean": {"summary": clean_summary, "results": clean},
               "evolved": {"summary": evolved_summary, "results": evolved},
               "comparison": compare(clean_summary, evolved_summary, clean, evolved)}
+    report["ablations"] = {}
+    for spec in args.ablation:
+        name, separator, command = spec.partition("=")
+        if not separator or not name.strip() or not command.strip():
+            raise ValueError("--ablation requires NAME=COMMAND")
+        results = [_run(command, case, args.timeout) for case in cases]
+        report["ablations"][name.strip()] = {"summary": summarize(results), "results": results}
     output = json.dumps(report, indent=2, ensure_ascii=False) + "\n"
     if args.output:
         args.output.write_text(output, encoding="utf-8")

learning_engine.py

@@ -8,6 +8,7 @@
 import sys
 import uuid
 from datetime import datetime, timezone
+from itertools import combinations
 from typing import Any, TYPE_CHECKING
 
 from event_store import EventStore, stable_hash
@@ -105,11 +106,13 @@ def artifact_context(self, run: dict[str, str]) -> tuple[str, list[dict[str, Any
             compatible.append(item)
         artifacts = compatible
         receipts = [{"skill_id": item["id"], "version": item["version"], "status": item["status"],
-                     "content_hash": item["content_hash"]} for item in artifacts]
+                     "content_hash": item["content_hash"], "chars": len(item["body"]),
+                     "utility_per_char": item.get("utility_per_char")} for item in artifacts]
         for receipt in receipts:
             self.store.append_event("learning.artifact_used", {**run, **receipt}, user_id=self.memory.user_id,
                                     workspace_id=self.memory.workspace_id, session_id=self.memory.session_id)
         preflights = self.negative_preflight(run["profile"], run["task"])
+        run["preflight_ids"] = [item["event_id"] for item in preflights]
         if not artifacts and not preflights:
             return "", receipts
         lines = ["<learned_guidance>",
@@ -208,6 +211,18 @@ def record_outcome(self, run: dict[str, str], receipts: list[dict[str, Any]], *,
             self.store.append_event("learning.review_requested", {"run_id": run["run_id"]},
                                     user_id=self.memory.user_id, workspace_id=self.memory.workspace_id,
                                     session_id=self.memory.session_id)
+        for left, right in combinations(sorted(str(item.get("skill_id")) for item in receipts if item.get("skill_id")), 2):
+            self.store.append_event("learning.artifact_pair", {
+                "pair": [left, right], "run_id": run["run_id"], "verified": bool(verified),
+                "success": bool(success), "correction": bool(correction),
+            }, user_id=self.memory.user_id, workspace_id=self.memory.workspace_id,
+               session_id=self.memory.session_id)
+        if verified and success:
+            for preflight_id in run.get("preflight_ids", []):
+                self.store.append_event("learning.near_miss_prevented", {
+                    "run_id": run["run_id"], "regression_event_id": preflight_id,
+                }, user_id=self.memory.user_id, workspace_id=self.memory.workspace_id,
+                   session_id=self.memory.session_id)
         if correction:
             dependencies = []
             for receipt in receipts:
@@ -254,6 +269,56 @@ def record_shadow_replay(self, proposal_id: str, candidate: list[dict[str, Any]]
                                 session_id=self.memory.session_id)
         return result
 
+    def record_omission_trial(self, proposal_id: str, with_item: list[dict[str, Any]],
+                              without_item: list[dict[str, Any]]) -> dict[str, Any]:
+        proposal = next((item for item in self.store.list_proposals(
+            workspace_id=self.memory.workspace_id, limit=10000) if item["id"] == proposal_id), None)
+        if not proposal or not with_item or len(with_item) != len(without_item):
+            raise ValueError("proposal and equal non-empty omission results are required")
+        left = [str(item.get("case_id", "")) for item in with_item]
+        right = [str(item.get("case_id", "")) for item in without_item]
+        if not all(left) or left != right or len(set(left)) != len(left):
+            raise ValueError("paired omission case ids must be unique and identical")
+        with_successes = sum(bool(item.get("verified_success")) for item in with_item)
+        without_successes = sum(bool(item.get("verified_success")) for item in without_item)
+        result = {"case_ids": left, "with_successes": with_successes, "without_successes": without_successes,
+                  "context_chars_saved": len(proposal["content"]) * len(left),
+                  "retirement_eligible": without_successes >= with_successes}
+        self.store.update_proposal(proposal_id, status=proposal["status"],
+                                   validation={**proposal["validation"], "omission_trial": result})
+        self.store.append_event("learning.omission_trial", {"proposal_id": proposal_id, **result},
+                                user_id=self.memory.user_id, workspace_id=self.memory.workspace_id,
+                                session_id=self.memory.session_id)
+        return result
+
+    def retire_artifact(self, proposal_id: str) -> bool:
+        proposal = next((item for item in self.store.list_proposals(
+            workspace_id=self.memory.workspace_id, limit=10000) if item["id"] == proposal_id), None)
+        validation = (proposal or {}).get("validation", {})
+        skill_id = validation.get("skill_id")
+        if not proposal or not skill_id or not validation.get("omission_trial", {}).get("retirement_eligible"):
+            return False
+        if not self.registry or not self.registry.set_learned_status(skill_id, "retired"):
+            return False
+        self.store.update_proposal(proposal_id, status="retired",
+                                   validation={**validation, "stage": "retired", "preimage_status": proposal["status"]})
+        self.store.append_event("learning.artifact_retired", {"proposal_id": proposal_id, "skill_id": skill_id},
+                                user_id=self.memory.user_id, workspace_id=self.memory.workspace_id)
+        return True
+
+    def restore_retired(self, proposal_id: str) -> bool:
+        proposal = next((item for item in self.store.list_proposals(
+            workspace_id=self.memory.workspace_id, limit=10000) if item["id"] == proposal_id), None)
+        skill_id = (proposal or {}).get("validation", {}).get("skill_id")
+        if not proposal or proposal.get("status") != "retired" or not skill_id or not self.registry:
+            return False
+        if not self.registry.set_learned_status(skill_id, "canary"):
+            return False
+        self.store.update_proposal(proposal_id, status="canary",
+                                   validation={**proposal["validation"], "stage": "canary",
+                                               "revalidation_status": "pending"})
+        return True
+
     def remember_claim(self, *, key: str, value: str, kind: str = "fact", authority: str = "inferred",
                        scope: str = "global", scope_value: str = "", evidence_id: str | None = None,
                        dependencies: list[str] | None = None, valid_until: str | None = None) -> dict[str, Any]:
@@ -368,7 +433,7 @@ def negative_preflight(self, profile: str, task: str) -> list[dict[str, Any]]:
                                             workspace_id=self.memory.workspace_id):
             payload = event["payload"]
             if payload.get("task_signature") == signature and not (set(payload.get("dependencies", [])) & forgotten):
-                result.append(payload)
+                result.append({**payload, "event_id": event["id"]})
         return result[:3]
 
     def _artifact_outcomes(self, skill_id: str) -> list[dict[str, Any]]:
@@ -582,4 +647,8 @@ def metrics(self, profile: str | None = None) -> dict[str, Any]:
             "tokens": sum(int(item.get("tokens", 0)) for item in completed),
             "latency": sum(float(item.get("latency", 0.0)) for item in completed),
             "task_families": families,
+            "context_chars": sum(sum(int(receipt.get("chars", 0)) for receipt in item.get("receipts", []))
+                                 for item in outcomes),
+            "prevented_near_misses": len(self.store.list_events(
+                "learning.near_miss_prevented", limit=10000, workspace_id=self.memory.workspace_id)),
         }

server.py

@@ -597,6 +597,31 @@ async def api_v2_learning_replay(proposal_id: str, request: Request):
         raise HTTPException(400, str(exc)) from exc
 
 
+@app.post("/api/v2/learning/{proposal_id}/omission", dependencies=[Depends(require_api_access)])
+async def api_v2_learning_omission(proposal_id: str, request: Request):
+    body = await request.json()
+    try:
+        return _agent.learning_engine.record_omission_trial(
+            proposal_id, body.get("with_item", []), body.get("without_item", []),
+        )
+    except (AttributeError, ValueError) as exc:
+        raise HTTPException(400, str(exc)) from exc
+
+
+@app.post("/api/v2/learning/{proposal_id}/retire", dependencies=[Depends(require_api_access)])
+async def api_v2_learning_retire(proposal_id: str):
+    if not _agent.learning_engine.retire_artifact(proposal_id):
+        raise HTTPException(409, "Artifact lacks non-regressing omission evidence")
+    return {"status": "retired", "proposal_id": proposal_id}
+
+
+@app.post("/api/v2/learning/{proposal_id}/restore", dependencies=[Depends(require_api_access)])
+async def api_v2_learning_restore(proposal_id: str):
+    if not _agent.learning_engine.restore_retired(proposal_id):
+        raise HTTPException(409, "Artifact is not retired")
+    return {"status": "canary", "proposal_id": proposal_id}
+
+
 @app.post("/api/v2/learning/{proposal_id}/rollback", dependencies=[Depends(require_api_access)])
 async def api_v2_learning_rollback(proposal_id: str):
     if not _agent.learning_engine.rollback(proposal_id):

skill_registry.py

@@ -17,7 +17,7 @@
     "workspace.read", "workspace.write", "shell", "network", "git.read", "git.write", "browser",
 }
 LEARNING_PROFILES = {"coder", "researcher"}
-LEARNED_STATUSES = {"candidate", "canary", "active", "rolled_back", "rejected"}
+LEARNED_STATUSES = {"candidate", "canary", "active", "rolled_back", "rejected", "retired"}
 MAX_LEARNED_CHARS = 8_000
 _SECRET_RE = re.compile(
     r"(?i)(?:api[_-]?key|secret|password|token)\s*[:=]\s*[^\s]{8,}|-----BEGIN [A-Z ]*PRIVATE KEY-----"
@@ -182,31 +182,53 @@ def _terms(text: str) -> set[str]:
     def match(self, profile: str, task: str, *, limit: int = 3, max_chars: int = MAX_LEARNED_CHARS) -> list[dict[str, Any]]:
         """Return active guidance plus at most one matching canary."""
         task_terms = self._terms(task)
-        scored: list[tuple[int, dict[str, Any]]] = []
+        scored: list[tuple[float, dict[str, Any], str]] = []
         for skill in self.list():
             manifest = skill.get("manifest", {})
             if skill.get("source") != "learned" or skill.get("status") not in {"active", "canary"}:
                 continue
             if manifest.get("profiles") != [profile]:
                 continue
             triggers = self._terms(" ".join(str(item) for item in manifest.get("triggers", [])))
-            score = len(task_terms & triggers)
-            if score:
-                scored.append((score, skill))
-        canary = [item for _, item in sorted(scored, key=lambda pair: (-pair[0], pair[1]["name"])) if item["status"] == "canary"][:1]
+            overlap = len(task_terms & triggers)
+            if not overlap:
+                continue
+            try:
+                body = (Path(skill["path"]) / "SKILL.md").read_text(encoding="utf-8")
+            except OSError:
+                continue
+            outcomes = [event["payload"] for event in self.store.list_events(
+                "learning.outcome", limit=10000, workspace_id=self.workspace_id)
+                if any(item.get("skill_id") == skill["id"] for item in event["payload"].get("receipts", []))
+                and event["payload"].get("verified")]
+            successes = sum(bool(item.get("success")) for item in outcomes)
+            utility = (successes + 1) / (len(outcomes) + 2)
+            scored.append((overlap * utility / max(1, len(body)), skill, body))
+        ordered = sorted(scored, key=lambda item: (-item[0], item[1]["name"], item[1]["version"]))
+        canary = [item for _, item, _ in ordered if item["status"] == "canary"][:1]
         canary_names = {item["name"] for item in canary}
-        active = [item for _, item in sorted(scored, key=lambda pair: (-pair[0], pair[1]["name"]))
+        active = [item for _, item, _ in ordered
                   if item["status"] == "active" and item["name"] not in canary_names]
+        bodies = {item["id"]: body for _, item, body in ordered}
+        pair_events = [event["payload"] for event in self.store.list_events(
+            "learning.artifact_pair", limit=10000, workspace_id=self.workspace_id)]
+        blocked = set()
+        for pair in {tuple(item.get("pair", [])) for item in pair_events if len(item.get("pair", [])) == 2}:
+            evidence = [item for item in pair_events if tuple(item.get("pair", [])) == pair and item.get("verified")]
+            if len(evidence) >= 2 and not any(item.get("success") for item in evidence):
+                blocked.add(pair)
         chosen: list[dict[str, Any]] = []
         used = 0
-        for skill in (active[:limit] + canary)[:limit]:
-            try:
-                body = (Path(skill["path"]) / "SKILL.md").read_text(encoding="utf-8")
-            except OSError:
+        for skill in canary + active:
+            if len(chosen) >= limit:
+                break
+            if any(tuple(sorted((skill["id"], item["id"]))) in blocked for item in chosen):
                 continue
+            body = bodies[skill["id"]]
             if used + len(body) > max_chars:
                 continue
-            chosen.append({**skill, "body": body, "content_hash": stable_hash(body)})
+            chosen.append({**skill, "body": body, "content_hash": stable_hash(body),
+                           "utility_per_char": next(score for score, item, _ in ordered if item["id"] == skill["id"])})
             used += len(body)
         return chosen