feat: add proof-carrying evolution

README.md

@@ -345,9 +345,9 @@ OpenKyrozen learns reusable policies and skills only from completed multi-step w
 
 ### How it works
 
-Every run records its profile, task signature, tool/error receipts, acceptance evidence, latency, and tokens in SQLite. Matching is deterministic and injects at most three artifacts (8,000 characters total), including at most one canary. A canary promotes after two distinct verified successes; one linked correction, or two verified failures among its last five active uses, rolls it back to its predecessor. Learned artifacts cannot add permissions or dynamic tools, and user/bundled/plugin skills are immutable to evolution.
+Every run records its profile, task signature, tool/error receipts, acceptance evidence, latency, and tokens in SQLite. Matching is deterministic and injects at most three artifacts (8,000 characters total), including at most one canary. A canary promotes only after two distinct verified successes and a non-regressing paired shadow replay; one linked correction, or two verified failures among its last five active uses, rolls it back to its predecessor. Learned artifacts cannot add permissions or dynamic tools, and user/bundled/plugin skills are immutable to evolution.
 
-Use `/agent auto|coder|researcher` to control routing. `/learning status [profile]`, `/learning metrics [profile]`, `/learning explain <id>`, and `/learning rollback <id>` expose lifecycle state and evidence. Project indexing remains separate knowledge ingestion.
+Use `/agent auto|coder|researcher` to control routing. `/learning status [profile]`, `/learning metrics [profile]`, `/learning evidence <id>`, `/learning explain <id>`, and `/learning rollback <id>` expose lifecycle state and proof. Shadow replay accepts paired frozen results through the authenticated API and never executes replay commands. Project indexing remains separate knowledge ingestion.
 
 ### Memory storage
 
@@ -400,6 +400,8 @@ KYROZEN_SERVER_TOKEN=change-me python server.py --host 0.0.0.0 --port 8000
 | `GET/POST` | `/api/v2/tasks` | Durable task listing and creation |
 | `GET` | `/api/v2/learning` | Learning proposal status |
 | `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}/rollback` | Roll back an activated proposal |
 | `GET` | `/api/v2/events` | Auditable runtime, task, session, and learning events |
 | `GET/POST` | `/api/v2/schedules` | Durable interval and one-shot Gateway jobs |

learning_engine.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import json
+import platform
 import re
+import sys
 import uuid
 from datetime import datetime, timezone
 from typing import Any, TYPE_CHECKING
@@ -69,16 +71,37 @@ def feedback_signal(text: str) -> str | None:
     def begin_run(self, profile: str, task: str) -> dict[str, str]:
         if profile not in EVOLUTION_PROFILES:
             raise ValueError("profile must be coder or researcher")
+        environment = self.environment_fingerprint()
         run = {"run_id": f"learnrun_{uuid.uuid4().hex}", "profile": profile,
-               "task_signature": self.task_signature(profile, task), "task": str(task)[:4000]}
+               "task_signature": self.task_signature(profile, task), "task": str(task)[:4000],
+               "environment_hash": environment["hash"]}
         self.store.append_event("learning.run_started", run, user_id=self.memory.user_id,
                                 workspace_id=self.memory.workspace_id, session_id=self.memory.session_id)
         return run
 
+    def environment_fingerprint(self, extra: dict[str, Any] | None = None) -> dict[str, Any]:
+        """Return a stable, secret-free applicability fingerprint."""
+        values = {"system": platform.system(), "machine": platform.machine(),
+                  "python": f"{sys.version_info.major}.{sys.version_info.minor}",
+                  "workspace": self.memory.workspace_id, **(extra or {})}
+        return {"values": values, "hash": stable_hash(json.dumps(values, sort_keys=True, default=str))}
+
     def artifact_context(self, run: dict[str, str]) -> tuple[str, list[dict[str, Any]]]:
         if self.registry is None:
             return "", []
         artifacts = self.registry.match(run["profile"], run["task"])
+        compatible = []
+        for item in artifacts:
+            expected = item.get("manifest", {}).get("applicability", {}).get("environment_hash")
+            if expected and expected != run.get("environment_hash"):
+                self.registry.set_learned_status(item["id"], "canary")
+                self.store.append_event("learning.artifact_revalidation_required", {
+                    **run, "skill_id": item["id"], "expected_environment_hash": expected,
+                }, user_id=self.memory.user_id, workspace_id=self.memory.workspace_id,
+                   session_id=self.memory.session_id)
+                continue
+            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]
         for receipt in receipts:
@@ -134,6 +157,14 @@ def propose_artifact(self, run_id: str, artifact: dict[str, Any]) -> dict[str, A
             "artifact_type": str(artifact.get("artifact_type", "skill")), "profiles": [profile],
             "triggers": [str(item).lower() for item in artifact.get("triggers", [])],
             "verification": artifact.get("verification", []), "parent_version": parent_version,
+            "verification_contract": {
+                "requirements": artifact.get("verification", []),
+                "side_effects": "forbidden_during_replay",
+            },
+            "applicability": artifact.get("applicability") or {
+                "environment_hash": self.environment_fingerprint()["hash"],
+                "profile": profile,
+            },
         }
         try:
             installed = self.registry.install_learned(body, manifest, status="canary")
@@ -150,7 +181,10 @@ def propose_artifact(self, run_id: str, artifact: dict[str, Any]) -> dict[str, A
                                                  evidence=[run_id], workspace_id=self.memory.workspace_id,
                                                  user_id=self.memory.user_id)
         validation = {"success": True, "stage": "canary", "profile": profile,
-                      "skill_id": installed["id"], "manifest": installed["manifest"]}
+                      "skill_id": installed["id"], "manifest": installed["manifest"],
+                      "verification_contract": manifest["verification_contract"],
+                      "applicability": manifest["applicability"],
+                      "revalidation_status": "pending", "shadow_replay": None}
         self.store.update_proposal(proposal_id, status="canary", validation=validation, confidence=0.5)
         self.store.append_event("learning.artifact_created", {"run_id": run_id, "proposal_id": proposal_id,
                                                                "skill_id": installed["id"], "profile": profile},
@@ -168,13 +202,47 @@ 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)
+        if correction:
+            self.store.append_event("learning.regression_case_created", {
+                "run_id": run["run_id"], "profile": run["profile"],
+                "task_signature": run["task_signature"], "task": self._clean(run.get("task", "")),
+                "receipts": receipts, "required_outcome": "must not repeat corrected behavior",
+            }, user_id=self.memory.user_id, workspace_id=self.memory.workspace_id,
+               session_id=self.memory.session_id)
         changes = []
         for receipt in receipts:
             change = self._reconcile_artifact(str(receipt.get("skill_id", "")))
             if change:
                 changes.append(change)
         return changes
 
+    def record_shadow_replay(self, proposal_id: str, candidate: list[dict[str, Any]],
+                             predecessor: list[dict[str, Any]]) -> dict[str, Any]:
+        """Record paired, already-sandboxed replay results without executing commands."""
+        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 candidate or len(candidate) != len(predecessor):
+            raise ValueError("proposal and equal non-empty paired replay results are required")
+        candidate_ids = [str(item.get("case_id", "")) for item in candidate]
+        predecessor_ids = [str(item.get("case_id", "")) for item in predecessor]
+        if not all(candidate_ids) or candidate_ids != predecessor_ids or len(set(candidate_ids)) != len(candidate_ids):
+            raise ValueError("paired replay case ids must be unique and identical")
+        candidate_successes = sum(bool(item.get("verified_success")) for item in candidate)
+        predecessor_successes = sum(bool(item.get("verified_success")) for item in predecessor)
+        result = {
+            "case_ids": candidate_ids, "candidate_successes": candidate_successes,
+            "predecessor_successes": predecessor_successes,
+            "non_regressing": candidate_successes >= predecessor_successes,
+            "environment": self.environment_fingerprint(),
+        }
+        validation = {**proposal.get("validation", {}), "shadow_replay": result,
+                      "revalidation_status": "valid" if result["non_regressing"] else "failed"}
+        self.store.update_proposal(proposal_id, status=proposal["status"], validation=validation)
+        self.store.append_event("learning.shadow_replay", {"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 _artifact_outcomes(self, skill_id: str) -> list[dict[str, Any]]:
         events = self.store.list_events("learning.outcome", limit=10000, workspace_id=self.memory.workspace_id)
         return [event["payload"] for event in reversed(events)
@@ -205,7 +273,9 @@ def _reconcile_artifact(self, skill_id: str) -> str | None:
                                         user_id=self.memory.user_id, workspace_id=self.memory.workspace_id)
                 return f"Rolled back learned {skill['name']} after verified regression."
         successful_runs = {item["run_id"] for item in verified if item.get("success")}
-        if skill["status"] == "canary" and len(successful_runs) >= 2 and not any_failure and not corrected:
+        replay = (proposal or {}).get("validation", {}).get("shadow_replay") or {}
+        if (skill["status"] == "canary" and len(successful_runs) >= 2 and not any_failure and not corrected
+                and replay.get("non_regressing") is True):
             if self.registry.activate_learned(skill_id):
                 if proposal:
                     self.store.update_proposal(proposal["id"], status="active", confidence=0.9,
@@ -344,6 +414,23 @@ def status(self, limit: int = 100, *, profile: str | None = None,
                              }})
         return enriched
 
+    def evidence_card(self, proposal_id: str) -> dict[str, Any] | None:
+        proposal = next((item for item in self.status(10000) if item["id"] == proposal_id), None)
+        if not proposal:
+            return None
+        validation = proposal.get("validation", {})
+        return {
+            "proposal_id": proposal_id, "profile": proposal.get("profile"),
+            "stage": proposal.get("lifecycle_stage"), "source_evidence": proposal.get("evidence", []),
+            "verification_contract": validation.get("verification_contract"),
+            "applicability": validation.get("applicability"),
+            "revalidation_status": validation.get("revalidation_status"),
+            "shadow_replay": validation.get("shadow_replay"),
+            "outcomes": proposal.get("evidence_receipts", []),
+            "metrics": proposal.get("artifact_metrics", {}),
+            "predecessor": proposal.get("predecessor"),
+        }
+
     def metrics(self, profile: str | None = None) -> dict[str, Any]:
         completed = [event["payload"] for event in self.store.list_events(
             "learning.run_completed", limit=10000, workspace_id=self.memory.workspace_id)]
@@ -353,6 +440,11 @@ def metrics(self, profile: str | None = None) -> dict[str, Any]:
             completed = [item for item in completed if item.get("profile") == profile]
             outcomes = [item for item in outcomes if item.get("profile") == profile]
         verified = [item for item in outcomes if item.get("verified")]
+        families: dict[str, dict[str, int]] = {}
+        for item in verified:
+            family = families.setdefault(str(item.get("task_signature", "unknown")), {"uses": 0, "successes": 0})
+            family["uses"] += 1
+            family["successes"] += int(bool(item.get("success")))
         return {
             "profile": profile or "all", "runs": len(completed), "verified_outcomes": len(verified),
             "completion_rate": (sum(bool(item.get("success")) for item in verified) / len(verified)) if verified else None,
@@ -361,4 +453,5 @@ def metrics(self, profile: str | None = None) -> dict[str, Any]:
             "tool_calls": sum(int(item.get("tool_calls", 0)) for item in completed),
             "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,
         }

main.py

@@ -3733,11 +3733,16 @@ def main() -> None:
             elif subcommand == "explain" and len(parts) > 2:
                 proposals = [p for p in learning_engine.status(1000) if p["id"] == parts[2].strip()]
                 console.print(json.dumps(proposals[0], ensure_ascii=False, indent=2) if proposals else "Proposal not found.")
+            elif subcommand == "evidence" and len(parts) > 2:
+                card = learning_engine.evidence_card(parts[2].strip())
+                console.print(json.dumps(card, ensure_ascii=False, indent=2) if card else "Proposal not found.")
+            elif subcommand == "replay" and len(parts) > 2:
+                console.print("Replay accepts paired frozen results through POST /api/v2/learning/<id>/replay; it never runs live commands.")
             elif subcommand == "metrics":
                 profile_filter = parts[2].strip() if len(parts) > 2 and parts[2].strip() in {"coder", "researcher"} else None
                 console.print(json.dumps(learning_engine.metrics(profile_filter), ensure_ascii=False, indent=2))
             else:
-                console.print("Usage: /learning status [coder|researcher] | /learning metrics [profile] | /learning rollback <id> | /learning explain <id>")
+                console.print("Usage: /learning status [coder|researcher] | /learning metrics [profile] | /learning rollback <id> | /learning explain|evidence|replay <id>")
             continue
 
         # /forget — show and optionally delete recent learnings

server.py

@@ -554,6 +554,27 @@ async def api_v2_learning_metrics(profile: str | None = None):
     return _agent.learning_engine.metrics(profile)
 
 
+@app.get("/api/v2/learning/{proposal_id}/evidence", dependencies=[Depends(require_api_access)])
+async def api_v2_learning_evidence(proposal_id: str):
+    card = _agent.learning_engine.evidence_card(proposal_id)
+    if card is None:
+        raise HTTPException(404, "Learning proposal not found")
+    return card
+
+
+@app.post("/api/v2/learning/{proposal_id}/replay", dependencies=[Depends(require_api_access)])
+async def api_v2_learning_replay(proposal_id: str, request: Request):
+    body = await request.json()
+    if not isinstance(body, dict):
+        raise HTTPException(400, "JSON object required")
+    try:
+        return _agent.learning_engine.record_shadow_replay(
+            proposal_id, body.get("candidate", []), body.get("predecessor", []),
+        )
+    except ValueError as exc:
+        raise HTTPException(400, str(exc)) from exc
+
+
 @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

@@ -219,6 +219,13 @@ def activate_learned(self, skill_id: str) -> bool:
                 self.store.set_skill_status(item["id"], "rolled_back", workspace_id=self.workspace_id)
         return self.store.set_skill_status(skill_id, "active", workspace_id=self.workspace_id)
 
+    def set_learned_status(self, skill_id: str, status: str) -> bool:
+        if status not in LEARNED_STATUSES:
+            return False
+        skill = next((item for item in self.list() if item["id"] == skill_id), None)
+        return bool(skill and skill.get("source") == "learned" and
+                    self.store.set_skill_status(skill_id, status, workspace_id=self.workspace_id))
+
     def rollback_learned(self, skill_id: str) -> bool:
         skill = next((item for item in self.list() if item["id"] == skill_id), None)
         if not skill or skill.get("source") != "learned":