From 5a03d8d5cab4dc688fa130a3930b169d390d4521 Mon Sep 17 00:00:00 2001 From: Evan Date: Mon, 31 Aug 2026 21:58:46 +0800 Subject: [PATCH] feat: add causal learning selection --- README.md | 5 ++ learning_benchmark.py | 9 ++++ learning_engine.py | 73 ++++++++++++++++++++++++++- server.py | 25 ++++++++++ skill_registry.py | 46 ++++++++++++----- tests/test_causal_selection.py | 84 ++++++++++++++++++++++++++++++++ tests/test_learning_benchmark.py | 21 +++++++- 7 files changed, 248 insertions(+), 15 deletions(-) create mode 100644 tests/test_causal_selection.py diff --git a/README.md b/README.md index 088ad3a..abcd99a 100644 --- a/README.md +++ b/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 ` explains a claim and `/memory forget ` 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 | diff --git a/learning_benchmark.py b/learning_benchmark.py index 474b6ac..1d181e1 100644 --- a/learning_benchmark.py +++ b/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") diff --git a/learning_engine.py b/learning_engine.py index b14b3e0..863b4fe 100644 --- a/learning_engine.py +++ b/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 = ["", @@ -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)), } diff --git a/server.py b/server.py index 98e3d0c..2fd35b3 100644 --- a/server.py +++ b/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): diff --git a/skill_registry.py b/skill_registry.py index f7be41d..f87f663 100644 --- a/skill_registry.py +++ b/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,7 +182,7 @@ 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"}: @@ -190,23 +190,45 @@ def match(self, profile: str, task: str, *, limit: int = 3, max_chars: int = MAX 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 diff --git a/tests/test_causal_selection.py b/tests/test_causal_selection.py new file mode 100644 index 0000000..5d9c7b6 --- /dev/null +++ b/tests/test_causal_selection.py @@ -0,0 +1,84 @@ +import tempfile +import unittest +from pathlib import Path + +from learning_engine import LearningEngine +from memory import MemoryBank +from skill_registry import SkillRegistry + + +BODY = """# Guidance + +## Trigger +Use for pytest failures. + +## Steps +1. Inspect the shared cause. + +## Verify +Run pytest successfully. +""" + + +class CausalSelectionTests(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 _artifact(self, name): + result = self.engine.propose_artifact(name, { + "name": name, "profile": "coder", "triggers": ["pytest"], + "verification": ["pytest succeeds"], "body": BODY.replace("Guidance", name), + }) + self.registry.set_learned_status(result["skill_id"], "active") + return result + + def test_repeated_pair_failure_prevents_co_injection(self): + left, right = self._artifact("left-guide"), self._artifact("right-guide") + receipts = [{"skill_id": left["skill_id"]}, {"skill_id": right["skill_id"]}] + for _ in range(2): + run = self.engine.begin_run("coder", "pytest") + self.engine.record_outcome(run, receipts, verified=True, success=False) + self.registry.set_learned_status(left["skill_id"], "active") + self.registry.set_learned_status(right["skill_id"], "active") + selected = self.registry.match("coder", "pytest") + self.assertEqual(len(selected), 1) + + def test_non_regressing_omission_allows_reversible_retirement(self): + artifact = self._artifact("retire-guide") + cases = [{"case_id": "one", "verified_success": True}] + result = self.engine.record_omission_trial(artifact["proposal_id"], cases, cases) + self.assertTrue(result["retirement_eligible"]) + self.assertGreater(result["context_chars_saved"], 0) + self.assertTrue(self.engine.retire_artifact(artifact["proposal_id"])) + self.assertEqual(self.registry.match("coder", "pytest"), []) + self.assertTrue(self.engine.restore_retired(artifact["proposal_id"])) + self.assertEqual(len(self.registry.match("coder", "pytest")), 1) + + def test_regressing_omission_cannot_retire(self): + artifact = self._artifact("keep-guide") + with_item = [{"case_id": "one", "verified_success": True}] + without_item = [{"case_id": "one", "verified_success": False}] + result = self.engine.record_omission_trial(artifact["proposal_id"], with_item, without_item) + self.assertFalse(result["retirement_eligible"]) + self.assertFalse(self.engine.retire_artifact(artifact["proposal_id"])) + + def test_success_after_preflight_counts_prevented_near_miss(self): + artifact = self._artifact("preflight-guide") + first = self.engine.begin_run("coder", "pytest") + _, receipts = self.engine.artifact_context(first) + self.engine.record_outcome(first, receipts, verified=True, success=False, correction=True) + second = self.engine.begin_run("coder", "pytest") + self.engine.artifact_context(second) + self.engine.record_outcome(second, [], verified=True, success=True) + self.assertEqual(self.engine.metrics()["prevented_near_misses"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_learning_benchmark.py b/tests/test_learning_benchmark.py index 722ea4c..ae8e223 100644 --- a/tests/test_learning_benchmark.py +++ b/tests/test_learning_benchmark.py @@ -1,6 +1,10 @@ +import json +import sys +import tempfile import unittest +from pathlib import Path -from learning_benchmark import compare, summarize +from learning_benchmark import compare, main, summarize class LearningBenchmarkTests(unittest.TestCase): @@ -14,6 +18,21 @@ def test_summary_and_comparison_are_harness_neutral(self): self.assertTrue(result["credible_secondary_gain"]) self.assertTrue(result["public_superiority_claim_supported"]) + def test_cli_exports_named_ablation(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + cases = root / "cases.jsonl" + runner = root / "runner.py" + output = root / "result.json" + cases.write_text(json.dumps({"id": "one", "profile": "coder", "task": "test"}) + "\n") + runner.write_text( + "import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'verified_success': True, 'corrections': 0, 'repeated_errors': 0, 'tool_calls': 1, 'tokens': 1, 'latency': 0.1}))\n" + ) + command = f"{sys.executable} {runner}" + main(["--cases", str(cases), "--clean-runner", command, "--evolved-runner", command, + "--ablation", f"no-memory={command}", "--output", str(output)]) + self.assertIn("no-memory", json.loads(output.read_text())["ablations"]) + if __name__ == "__main__": unittest.main()