diff --git a/README.md b/README.md index abcd99a..1ceef61 100644 --- a/README.md +++ b/README.md @@ -353,6 +353,8 @@ Inferred facts and preferences remain candidates until repeated independent evid 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. +Learned guidance is bound to the provider/model family that produced its evidence unless paired replay validates it across models. Verifier reliability, evidence-adaptive review priority, and a user-owned `KYROZEN_LEARNING_CONSTITUTION` file constrain evolution. Redacted experience capsules are portable JSON evidence, but imports always remain inactive candidates until local validation. + ### 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. @@ -409,6 +411,9 @@ KYROZEN_SERVER_TOKEN=change-me python server.py --host 0.0.0.0 --port 8000 | `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 | +| `GET` | `/api/v2/learning/{id}/capsule` | Export a redacted, harness-neutral experience capsule | +| `POST` | `/api/v2/learning/capsules` | Import a capsule as an inactive candidate | +| `GET` | `/api/v2/learning/constitution` | Inspect the immutable user-owned learning policy | | `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 | diff --git a/learning_engine.py b/learning_engine.py index 863b4fe..bb3e229 100644 --- a/learning_engine.py +++ b/learning_engine.py @@ -3,12 +3,14 @@ from __future__ import annotations import json +import os import platform import re import sys import uuid from datetime import datetime, timezone from itertools import combinations +from pathlib import Path from typing import Any, TYPE_CHECKING from event_store import EventStore, stable_hash @@ -26,6 +28,12 @@ _SECRET_RE = re.compile( r"(?i)(?:api[_-]?key|secret|password|token)\s*[:=]\s*[^\s]{8,}|-----BEGIN [A-Z ]*PRIVATE KEY-----" ) +CAPSULE_PROTOCOL = "openkyrozen-experience-capsule-v1" +DEFAULT_CONSTITUTION = { + "allowed_artifact_types": ["policy", "skill"], "allow_dynamic_tools": False, + "allow_permission_expansion": False, "allow_harness_edits": False, + "minimum_live_successes": 2, "require_shadow_replay": True, +} class LearningEngine: @@ -71,13 +79,13 @@ def feedback_signal(text: str) -> str | None: return "success" return None - def begin_run(self, profile: str, task: str) -> dict[str, str]: + def begin_run(self, profile: str, task: str, *, provider_model: str | None = None) -> 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], - "environment_hash": environment["hash"]} + "environment_hash": environment["hash"], "provider_model": provider_model or "unspecified"} 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 @@ -89,6 +97,16 @@ def environment_fingerprint(self, extra: dict[str, Any] | None = None) -> dict[s "workspace": self.memory.workspace_id, **(extra or {})} return {"values": values, "hash": stable_hash(json.dumps(values, sort_keys=True, default=str))} + def constitution(self) -> dict[str, Any]: + path = os.environ.get("KYROZEN_LEARNING_CONSTITUTION", "").strip() + if not path: + return dict(DEFAULT_CONSTITUTION) + try: + data = json.loads(Path(path).expanduser().read_text(encoding="utf-8")) + except (OSError, ValueError): + return dict(DEFAULT_CONSTITUTION) + return {**DEFAULT_CONSTITUTION, **data} if isinstance(data, dict) else dict(DEFAULT_CONSTITUTION) + def artifact_context(self, run: dict[str, str]) -> tuple[str, list[dict[str, Any]]]: if self.registry is None: return "", [] @@ -96,7 +114,12 @@ def artifact_context(self, run: dict[str, str]) -> tuple[str, list[dict[str, Any compatible = [] for item in artifacts: expected = item.get("manifest", {}).get("applicability", {}).get("environment_hash") - if expected and expected != run.get("environment_hash"): + model_scope = item.get("manifest", {}).get("applicability", {}).get("provider_model") + proposal = self._proposal_for_skill(item["id"]) + cross_model = bool(((proposal or {}).get("validation", {}).get("shadow_replay") or {}).get( + "cross_model_validated")) + if ((expected and expected != run.get("environment_hash")) + or (model_scope and model_scope != run.get("provider_model") and not cross_model)): 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, @@ -174,6 +197,14 @@ def propose_artifact(self, run_id: str, artifact: dict[str, Any]) -> dict[str, A "profile": profile, }, } + constitution = self.constitution() + if manifest["artifact_type"] not in constitution.get("allowed_artifact_types", []): + return {"status": "rejected", "reason": "learning constitution forbids artifact type"} + source_run = next((event["payload"] for event in self.store.list_events( + "learning.run_started", limit=10000, workspace_id=self.memory.workspace_id) + if event["payload"].get("run_id") == run_id), None) + if source_run and source_run.get("provider_model") != "unspecified": + manifest["applicability"]["provider_model"] = source_run["provider_model"] try: installed = self.registry.install_learned(body, manifest, status="canary") except (OSError, ValueError) as exc: @@ -255,11 +286,13 @@ def record_shadow_replay(self, proposal_id: str, candidate: list[dict[str, Any]] 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) + models = sorted({str(item.get("provider_model")) for item in candidate if item.get("provider_model")}) result = { "case_ids": candidate_ids, "candidate_successes": candidate_successes, "predecessor_successes": predecessor_successes, "non_regressing": candidate_successes >= predecessor_successes, "environment": self.environment_fingerprint(), + "model_scope": models, "cross_model_validated": len(models) >= 2, } validation = {**proposal.get("validation", {}), "shadow_replay": result, "revalidation_status": "valid" if result["non_regressing"] else "failed"} @@ -269,6 +302,46 @@ def record_shadow_replay(self, proposal_id: str, candidate: list[dict[str, Any]] session_id=self.memory.session_id) return result + def verifier_reliability(self, verifier_id: str) -> dict[str, Any]: + outcomes = [event["payload"] for event in self.store.list_events( + "learning.outcome", limit=10000, workspace_id=self.memory.workspace_id)] + accepted = {item.get("run_id") for item in outcomes + if item.get("source") == verifier_id and item.get("verified") and item.get("success")} + corrected = {item.get("run_id") for item in outcomes if item.get("correction")} + false_accepts = len(accepted & corrected) + return {"verifier_id": verifier_id, "accepted": len(accepted), "false_accepts": false_accepts, + "reliability": 1.0 - (false_accepts / len(accepted)) if accepted else None} + + def export_capsule(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 + capsule = {"protocol": CAPSULE_PROTOCOL, "exported_at": datetime.now(timezone.utc).isoformat(), + "kind": proposal["kind"], "content": proposal["content"], + "content_hash": stable_hash(proposal["content"]), "profile": proposal.get("profile"), + "evidence_card": self.evidence_card(proposal_id)} + if _SECRET_RE.search(json.dumps(capsule, ensure_ascii=False)): + raise ValueError("capsule contains secret-like content") + return capsule + + def import_capsule(self, capsule: dict[str, Any]) -> dict[str, Any]: + if not isinstance(capsule, dict) or capsule.get("protocol") != CAPSULE_PROTOCOL: + raise ValueError("unsupported experience capsule") + content = str(capsule.get("content", "")).strip() + if not content or stable_hash(content) != capsule.get("content_hash") or _SECRET_RE.search(content): + raise ValueError("invalid or secret-bearing capsule") + proposal_id = self.store.create_proposal( + str(capsule.get("kind", "skill")), content, confidence=0.0, + evidence=[], workspace_id=self.memory.workspace_id, user_id=self.memory.user_id, + ) + self.store.update_proposal(proposal_id, status="candidate", validation={ + "stage": "imported", "source": "imported", "active": False, + "profile": capsule.get("profile"), "imported_evidence_card": capsule.get("evidence_card"), + }) + self.store.append_event("learning.capsule_imported", {"proposal_id": proposal_id, "active": False}, + user_id=self.memory.user_id, workspace_id=self.memory.workspace_id) + return {"proposal_id": proposal_id, "status": "candidate", "active": False} + 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( @@ -467,8 +540,14 @@ def _reconcile_artifact(self, skill_id: str) -> str | None: return f"Rolled back learned {skill['name']} after verified regression." successful_runs = {item["run_id"] for item in verified if item.get("success")} replay = (proposal or {}).get("validation", {}).get("shadow_replay") or {} + verifier_ids = {item.get("source") for item in verified if item.get("success") and item.get("source")} + verifiers_reliable = all( + self.verifier_reliability(verifier)["reliability"] in {None, 1.0} + or self.verifier_reliability(verifier)["reliability"] >= 0.5 + for verifier in verifier_ids + ) if (skill["status"] == "canary" and len(successful_runs) >= 2 and not any_failure and not corrected - and replay.get("non_regressing") is True): + and replay.get("non_regressing") is True and verifiers_reliable): if self.registry.activate_learned(skill_id): if proposal: self.store.update_proposal(proposal["id"], status="active", confidence=0.9, @@ -486,7 +565,7 @@ def pending_reviews(self, *, min_age_seconds: float = 30.0, limit: int = 1) -> l "learning.run_reviewed", limit=10000, workspace_id=self.memory.workspace_id)} requested = {event["payload"].get("run_id") for event in self.store.list_events( "learning.review_requested", limit=10000, workspace_id=self.memory.workspace_id)} - now, result = datetime.now(timezone.utc), [] + now, candidates = datetime.now(timezone.utc), [] for event in reversed(completed): payload = event["payload"] if payload["run_id"] in reviewed or payload.get("provider_error") or payload.get("contains_secret"): @@ -495,10 +574,13 @@ def pending_reviews(self, *, min_age_seconds: float = 30.0, limit: int = 1) -> l continue if not payload.get("eligible") and payload["run_id"] not in requested: continue - result.append(payload) - if len(result) >= limit: - break - return result + recurrence = sum(item["payload"].get("task_signature") == payload.get("task_signature") + for item in completed) + score = recurrence * 2 + int(payload["run_id"] in requested) * 5 + int(payload.get("errors", 0)) * 2 + score += min(5, int(payload.get("tokens", 0)) // 1000) + candidates.append((score, event["created_at"], payload)) + candidates.sort(key=lambda item: (-item[0], item[1])) + return [payload for _, _, payload in candidates[:limit]] def mark_reviewed(self, run_id: str, *, result: str) -> None: self.store.append_event("learning.run_reviewed", {"run_id": run_id, "result": str(result)[:500]}, @@ -633,11 +715,13 @@ 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]] = {} + families: dict[str, dict[str, Any]] = {} 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"))) + for family in families.values(): + family["completion_rate"] = family["successes"] / family["uses"] 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, diff --git a/main.py b/main.py index b369a03..95c507a 100644 --- a/main.py +++ b/main.py @@ -2830,7 +2830,9 @@ def _chat_turn(user_input: str, clear_tasks: bool = False, profile: str | None = )) _last_learning_run = None resolved_profile = learning_engine.route_profile(user_input, profile or _agent_profile_mode) - learning_run = learning_engine.begin_run(resolved_profile, user_input) + DEEPSEEK_MODEL = _select_model(user_input) + provider_model = f"{_provider_config.provider}:{DEEPSEEK_MODEL}" if _provider_config else f"unknown:{DEEPSEEK_MODEL}" + learning_run = learning_engine.begin_run(resolved_profile, user_input, provider_model=provider_model) learned_context, learning_receipts = learning_engine.artifact_context(learning_run) _execution_capability_token = issue_capability_token( f"surface:{_EXECUTION_SURFACE}", @@ -2847,7 +2849,6 @@ def _chat_turn(user_input: str, clear_tasks: bool = False, profile: str | None = tasks.clear() # Auto-select the best model for this turn based on task complexity - DEEPSEEK_MODEL = _select_model(user_input) complexity = _classify_complexity(user_input) turn_start = time.time() diff --git a/server.py b/server.py index 2fd35b3..be75c4e 100644 --- a/server.py +++ b/server.py @@ -622,6 +622,30 @@ async def api_v2_learning_restore(proposal_id: str): return {"status": "canary", "proposal_id": proposal_id} +@app.get("/api/v2/learning/{proposal_id}/capsule", dependencies=[Depends(require_api_access)]) +async def api_v2_learning_export_capsule(proposal_id: str): + try: + capsule = _agent.learning_engine.export_capsule(proposal_id) + except ValueError as exc: + raise HTTPException(409, str(exc)) from exc + if capsule is None: + raise HTTPException(404, "Learning proposal not found") + return capsule + + +@app.post("/api/v2/learning/capsules", dependencies=[Depends(require_api_access)]) +async def api_v2_learning_import_capsule(request: Request): + try: + return _agent.learning_engine.import_capsule(await request.json()) + except ValueError as exc: + raise HTTPException(400, str(exc)) from exc + + +@app.get("/api/v2/learning/constitution", dependencies=[Depends(require_api_access)]) +async def api_v2_learning_constitution(): + return _agent.learning_engine.constitution() + + @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/tests/test_evolution_robustness.py b/tests/test_evolution_robustness.py new file mode 100644 index 0000000..dbb4b9b --- /dev/null +++ b/tests/test_evolution_robustness.py @@ -0,0 +1,97 @@ +import json +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from learning_engine import LearningEngine +from memory import MemoryBank +from skill_registry import SkillRegistry + + +BODY = """# Robust guidance + +## Trigger +Use for pytest. + +## Steps +1. Run the focused test. + +## Verify +Require a zero exit status. +""" + + +class EvolutionRobustnessTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + root = Path(self.directory.name) + self.memory = MemoryBank(root / "state.sqlite3", workspace_id="project") + self.registry = SkillRegistry(self.memory.store, workspace_id="project", root=root / "skills") + self.engine = LearningEngine(self.memory, registry=self.registry) + + def tearDown(self): + self.directory.cleanup() + + def _candidate(self, run_id="source"): + return self.engine.propose_artifact(run_id, { + "name": "robust-guide", "profile": "coder", "triggers": ["pytest"], + "verification": ["pytest succeeds"], "body": BODY, + }) + + def test_model_bound_artifact_requires_matching_model_or_cross_model_proof(self): + source = self.engine.begin_run("coder", "pytest", provider_model="provider:model-a") + candidate = self._candidate(source["run_id"]) + self.registry.set_learned_status(candidate["skill_id"], "active") + other = self.engine.begin_run("coder", "pytest", provider_model="provider:model-b") + self.assertEqual(self.engine.artifact_context(other), ("", [])) + self.registry.set_learned_status(candidate["skill_id"], "active") + paired = [ + {"case_id": "a", "verified_success": True, "provider_model": "provider:model-a"}, + {"case_id": "b", "verified_success": True, "provider_model": "provider:model-b"}, + ] + self.engine.record_shadow_replay(candidate["proposal_id"], paired, paired) + context, receipts = self.engine.artifact_context(other) + self.assertIn("Robust guidance", context) + self.assertEqual(receipts[0]["skill_id"], candidate["skill_id"]) + + def test_verifier_reliability_detects_later_correction(self): + run = self.engine.begin_run("coder", "pytest") + self.engine.record_outcome(run, [], verified=True, success=True, source="test-verifier") + self.engine.record_outcome(run, [], verified=True, success=False, correction=True, source="user_feedback") + reliability = self.engine.verifier_reliability("test-verifier") + self.assertEqual(reliability["false_accepts"], 1) + self.assertEqual(reliability["reliability"], 0.0) + + def test_constitution_is_read_only_and_blocks_disallowed_artifact(self): + path = Path(self.directory.name) / "constitution.json" + path.write_text(json.dumps({"allowed_artifact_types": ["policy"]})) + before = path.read_bytes() + with patch.dict(os.environ, {"KYROZEN_LEARNING_CONSTITUTION": str(path)}): + result = self._candidate() + self.assertEqual(result["status"], "rejected") + self.assertEqual(self.engine.constitution()["allowed_artifact_types"], ["policy"]) + self.assertEqual(path.read_bytes(), before) + + def test_capsule_round_trip_never_activates_import(self): + candidate = self._candidate() + capsule = self.engine.export_capsule(candidate["proposal_id"]) + imported = self.engine.import_capsule(capsule) + self.assertFalse(imported["active"]) + proposal = next(item for item in self.engine.status(100) if item["id"] == imported["proposal_id"]) + self.assertEqual(proposal["status"], "candidate") + self.assertEqual(proposal["validation"]["stage"], "imported") + + def test_review_budget_prefers_verified_failure(self): + routine = self.engine.begin_run("coder", "routine pytest") + self.engine.complete_run(routine, result="done", receipts=[], tools=[{"success": True}, {"success": True}], + tokens=10, latency=0.1) + failure = self.engine.begin_run("coder", "important pytest") + self.engine.complete_run(failure, result="failed", receipts=[], tools=[], tokens=5000, latency=0.1) + self.engine.record_outcome(failure, [], verified=True, success=False) + self.assertEqual(self.engine.pending_reviews(min_age_seconds=0, limit=1)[0]["run_id"], failure["run_id"]) + + +if __name__ == "__main__": + unittest.main()