diff --git a/README.md b/README.md index 84a4e5d..7c3763b 100644 --- a/README.md +++ b/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 `, and `/learning rollback ` 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 `, `/learning explain `, and `/learning rollback ` 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 | diff --git a/learning_engine.py b/learning_engine.py index 818f5c0..c857d42 100644 --- a/learning_engine.py +++ b/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,6 +202,13 @@ 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", ""))) @@ -175,6 +216,33 @@ def record_outcome(self, run: dict[str, str], receipts: list[dict[str, Any]], *, 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, } diff --git a/main.py b/main.py index 6aa7d94..106d95c 100644 --- a/main.py +++ b/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//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 | /learning explain ") + console.print("Usage: /learning status [coder|researcher] | /learning metrics [profile] | /learning rollback | /learning explain|evidence|replay ") continue # /forget — show and optionally delete recent learnings diff --git a/server.py b/server.py index 9d2f544..7225ec3 100644 --- a/server.py +++ b/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): diff --git a/skill_registry.py b/skill_registry.py index 775f57a..f7be41d 100644 --- a/skill_registry.py +++ b/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": diff --git a/tests/test_evolution.py b/tests/test_evolution.py index b089b45..0084b20 100644 --- a/tests/test_evolution.py +++ b/tests/test_evolution.py @@ -39,9 +39,16 @@ def _candidate(self): "verification": ["pytest exits zero"], "body": BODY, }) + def _pass_replay(self, candidate): + return self.engine.record_shadow_replay( + candidate["proposal_id"], [{"case_id": "case-1", "verified_success": True}], + [{"case_id": "case-1", "verified_success": True}], + ) + def test_profile_isolation_canary_promotion_and_correction_rollback(self): candidate = self._candidate() self.assertEqual(candidate["status"], "canary") + self._pass_replay(candidate) self.assertEqual(self.registry.match("researcher", "research pytest history"), []) for _ in range(2): @@ -74,6 +81,7 @@ def test_learned_artifact_rejects_secrets_and_cannot_mutate_local_skill(self): def test_refinement_records_predecessor_and_rejects_dynamic_tools(self): first = self._candidate() + self._pass_replay(first) for _ in range(2): run = self.engine.begin_run("coder", "run pytest tests") _, receipts = self.engine.artifact_context(run) @@ -112,6 +120,46 @@ def test_secret_bearing_run_is_never_reviewed(self): self.engine.record_outcome(run, [], verified=True, success=False) self.assertEqual(self.engine.pending_reviews(min_age_seconds=0), []) + def test_promotion_requires_non_regressing_shadow_replay(self): + candidate = self._candidate() + for _ in range(2): + run = self.engine.begin_run("coder", "run pytest tests") + _, receipts = self.engine.artifact_context(run) + self.engine.record_outcome(run, receipts, verified=True, success=True) + skill = next(item for item in self.registry.list() if item["id"] == candidate["skill_id"]) + self.assertEqual(skill["status"], "canary") + replay = self._pass_replay(candidate) + self.assertTrue(replay["non_regressing"]) + run = self.engine.begin_run("coder", "run pytest tests") + _, receipts = self.engine.artifact_context(run) + changes = self.engine.record_outcome(run, receipts, verified=True, success=True) + self.assertIn("Promoted", changes[0]) + card = self.engine.evidence_card(candidate["proposal_id"]) + self.assertEqual(card["revalidation_status"], "valid") + + def test_environment_drift_requires_revalidation(self): + candidate = self._candidate() + self._pass_replay(candidate) + skill = next(item for item in self.registry.list() if item["id"] == candidate["skill_id"]) + skill["manifest"]["applicability"]["environment_hash"] = "different" + self.memory.store.upsert_skill( + name=skill["name"], version=skill["version"], path=skill["path"], + description=skill["description"], permissions=[], status="active", source="learned", + manifest=skill["manifest"], workspace_id="project", + ) + run = self.engine.begin_run("coder", "run pytest tests") + context, receipts = self.engine.artifact_context(run) + self.assertEqual((context, receipts), ("", [])) + self.assertEqual(next(item for item in self.registry.list() if item["id"] == candidate["skill_id"])["status"], "canary") + + def test_correction_creates_regression_case(self): + candidate = self._candidate() + run = self.engine.begin_run("coder", "run pytest tests") + _, receipts = self.engine.artifact_context(run) + self.engine.record_outcome(run, receipts, verified=True, success=False, correction=True) + events = self.memory.store.list_events("learning.regression_case_created", workspace_id="project") + self.assertEqual(events[0]["payload"]["task_signature"], run["task_signature"]) + if __name__ == "__main__": unittest.main()