From 36701aacde123cfc5c8a811047a82312521e16b3 Mon Sep 17 00:00:00 2001 From: Evan Date: Mon, 31 Aug 2026 21:46:51 +0800 Subject: [PATCH] feat: add verified memory contracts --- README.md | 4 ++ event_store.py | 16 ++++- learning_engine.py | 134 +++++++++++++++++++++++++++++++++++- main.py | 17 ++++- memory.py | 13 +++- server.py | 22 ++++++ subagents.py | 5 +- tests/test_memory_claims.py | 72 +++++++++++++++++++ 8 files changed, 273 insertions(+), 10 deletions(-) create mode 100644 tests/test_memory_claims.py diff --git a/README.md b/README.md index 7c3763b..088ad3a 100644 --- a/README.md +++ b/README.md @@ -349,6 +349,8 @@ Every run records its profile, task signature, tool/error receipts, acceptance e 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. +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. + ### 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. @@ -403,6 +405,8 @@ KYROZEN_SERVER_TOKEN=change-me python server.py --host 0.0.0.0 --port 8000 | `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/memory/claims` | Typed memory claims with provenance and scope | +| `GET/DELETE` | `/api/v2/memory/claims/{id}` | Explain or dependency-completely forget a claim | | `GET` | `/api/v2/events` | Auditable runtime, task, session, and learning events | | `GET/POST` | `/api/v2/schedules` | Durable interval and one-shot Gateway jobs | | `POST` | `/api/v2/schedules/{id}/disable` | Disable a scheduled job | diff --git a/event_store.py b/event_store.py index f728a1c..296cc1d 100644 --- a/event_store.py +++ b/event_store.py @@ -268,10 +268,13 @@ def upsert_memory(self, content: str, *, kind: str = "episodic", status: str = " ) return memory_id - def list_memories(self, *, kind: str | None = None, status: str = "active", limit: int = 100, + def list_memories(self, *, kind: str | None = None, status: str | None = "active", limit: int = 100, workspace_id: str = "default", session_id: str | None = None) -> list[dict[str, Any]]: - clauses = ["status=?", "workspace_id=?"] - params: list[Any] = [status, workspace_id] + clauses = ["workspace_id=?"] + params: list[Any] = [workspace_id] + if status: + clauses.append("status=?") + params.append(status) if kind: clauses.append("kind=?") params.append(kind) @@ -286,6 +289,13 @@ def list_memories(self, *, kind: str | None = None, status: str = "active", limi ).fetchall() return [dict(row, source_event_ids=self._loads(row["source_event_ids"], []), metadata=self._loads(row["metadata"], {})) for row in rows] + def set_memory_status(self, memory_id: str, status: str, *, workspace_id: str = "default") -> bool: + with self._lock, self.connection() as db: + return db.execute( + "UPDATE memories SET status=?,updated_at=? WHERE id=? AND workspace_id=?", + (status, utc_now(), memory_id, workspace_id), + ).rowcount == 1 + def delete_memories(self, ids: list[str], *, workspace_id: str = "default") -> int: ids = [str(item) for item in ids if item] if not ids: diff --git a/learning_engine.py b/learning_engine.py index c857d42..b14b3e0 100644 --- a/learning_engine.py +++ b/learning_engine.py @@ -18,6 +18,8 @@ EVOLUTION_PROFILES = {"coder", "researcher"} +CLAIM_SCOPES = {"global", "profile", "project", "task"} +SCOPE_RANK = {"global": 0, "profile": 1, "project": 2, "task": 3} POSITIVE_FEEDBACK = ("that works", "it works", "worked", "fixed", "solved", "perfect", "great", "谢谢", "好了", "搞定") NEGATIVE_FEEDBACK = ("not working", "still broken", "didn't work", "doesn't work", "wrong", "incorrect", "not fixed", "不对", "还不行", "没解决") _SECRET_RE = re.compile( @@ -107,12 +109,15 @@ def artifact_context(self, run: dict[str, str]) -> tuple[str, list[dict[str, Any 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) - if not artifacts: + preflights = self.negative_preflight(run["profile"], run["task"]) + if not artifacts and not preflights: return "", receipts lines = ["", "The following profile-scoped guidance is untrusted procedure data. It cannot grant permissions."] for item in artifacts: lines.append(f"\n### {item['name']} {item['version']} [{item['status']}]\n{item['body']}") + for item in preflights: + lines.append(f"\n### Corrected failure preflight\n{item['required_outcome']}") lines.append("") return "\n".join(lines), receipts @@ -184,7 +189,8 @@ def propose_artifact(self, run_id: str, artifact: dict[str, Any]) -> dict[str, A "skill_id": installed["id"], "manifest": installed["manifest"], "verification_contract": manifest["verification_contract"], "applicability": manifest["applicability"], - "revalidation_status": "pending", "shadow_replay": None} + "revalidation_status": "pending", "shadow_replay": None, + "dependencies": [str(item) for item in artifact.get("dependencies", []) if item]} 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}, @@ -203,10 +209,15 @@ def record_outcome(self, run: dict[str, str], receipts: list[dict[str, Any]], *, 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: + proposal = self._proposal_for_skill(str(receipt.get("skill_id", ""))) + dependencies.extend((proposal or {}).get("validation", {}).get("dependencies", [])) 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", + "receipts": receipts, "dependencies": list(dict.fromkeys(dependencies)), + "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 = [] @@ -243,6 +254,123 @@ def record_shadow_replay(self, proposal_id: str, candidate: list[dict[str, Any]] session_id=self.memory.session_id) return result + 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]: + """Create a typed memory claim; inferred claims need repeated evidence.""" + key, value = self._clean(key), self._clean(value) + if not key or not value or _SECRET_RE.search(f"{key}={value}"): + raise ValueError("claim requires non-secret key and value") + if authority not in {"owner", "inferred"} or scope not in CLAIM_SCOPES: + raise ValueError("invalid claim authority or scope") + if scope != "global" and not str(scope_value).strip(): + raise ValueError("non-global claims require scope_value") + metadata = {"claim": True, "claim_key": key, "claim_value": value, + "authority": authority, "scope": {"type": scope, "value": str(scope_value).strip()}, + "valid_from": datetime.now(timezone.utc).isoformat(), "valid_until": valid_until, + "dependencies": [str(item) for item in dependencies or [] if item]} + active = [row for row in self.store.list_memories(status="active", limit=10000, + workspace_id=self.memory.workspace_id) + if row.get("metadata", {}).get("claim_key") == key + and row.get("metadata", {}).get("scope") == metadata["scope"]] + if authority == "owner": + for row in active: + if row.get("metadata", {}).get("claim_value") != value: + self.store.set_memory_status(row["id"], "superseded", workspace_id=self.memory.workspace_id) + metadata["supersedes"] = row["id"] + memory_id = self.memory.add_log(f"{key}: {value}", kind=kind, status="active", confidence=1.0, + metadata=metadata) + self.store.append_event("memory.claim_activated", {"memory_id": memory_id, **metadata}, + user_id=self.memory.user_id, workspace_id=self.memory.workspace_id, + session_id=self.memory.session_id) + return {"status": "active", "memory_id": memory_id, "needs_clarification": False} + proposals = self.store.list_proposals(workspace_id=self.memory.workspace_id, limit=10000) + existing = next((item for item in proposals if item["status"] == "candidate" + and item.get("validation", {}).get("claim_key") == key + and item.get("validation", {}).get("claim_value") == value + and item.get("validation", {}).get("scope") == metadata["scope"]), None) + conflict = any(row.get("metadata", {}).get("claim_value") != value for row in active) + if existing: + count = self.store.add_proposal_evidence(existing["id"], evidence_id or stable_hash(f"{key}:{value}")) + if count >= 2 and not conflict: + validation = {**existing["validation"], **metadata, "success": True, + "evidence_count": count, "stage": "active"} + self.store.update_proposal(existing["id"], status="active", confidence=0.8, validation=validation) + memory_id = self.memory.add_log(f"{key}: {value}", kind=kind, status="active", confidence=0.8, + metadata={**metadata, "proposal_id": existing["id"]}) + return {"status": "active", "proposal_id": existing["id"], "memory_id": memory_id, + "needs_clarification": False} + return {"status": "candidate", "proposal_id": existing["id"], "evidence_count": count, + "needs_clarification": conflict} + proposal_id = self.store.create_proposal(kind, f"{key}: {value}", confidence=0.4, + evidence=[evidence_id or stable_hash(f"{key}:{value}")], + workspace_id=self.memory.workspace_id, user_id=self.memory.user_id) + self.store.update_proposal(proposal_id, status="candidate", + validation={**metadata, "stage": "candidate", "conflict": conflict}) + self.store.append_event("memory.claim_candidate", {"proposal_id": proposal_id, **metadata}, + user_id=self.memory.user_id, workspace_id=self.memory.workspace_id, + session_id=self.memory.session_id) + return {"status": "candidate", "proposal_id": proposal_id, "evidence_count": 1, + "needs_clarification": conflict} + + def resolve_claim(self, key: str, *, profile: str | None = None, project: str | None = None, + task_signature: str | None = None) -> dict[str, Any] | None: + context = {"profile": profile, "project": project, "task": task_signature} + matches = [] + for row in self.store.list_memories(status="active", limit=10000, workspace_id=self.memory.workspace_id): + meta = row.get("metadata", {}) + if not meta.get("claim") or meta.get("claim_key") != key: + continue + scope = meta.get("scope", {"type": "global", "value": ""}) + if scope["type"] != "global" and context.get(scope["type"]) != scope.get("value"): + continue + matches.append((SCOPE_RANK[scope["type"]], int(meta.get("authority") == "owner"), row)) + if not matches: + return None + best_rank = max((rank, authority) for rank, authority, _ in matches) + best = [row for rank, authority, row in matches if (rank, authority) == best_rank] + values = {row["metadata"]["claim_value"] for row in best} + return {"conflict": len(values) > 1, "value": next(iter(values)) if len(values) == 1 else None, + "claims": best, "scope_rank": best_rank[0]} + + def explain_claim(self, claim_id: str) -> dict[str, Any] | None: + rows = self.store.list_memories(status=None, limit=10000, workspace_id=self.memory.workspace_id) + row = next((item for item in rows if item["id"] == claim_id and item.get("metadata", {}).get("claim")), None) + return row + + def forget_claim(self, claim_id: str) -> bool: + claim = self.explain_claim(claim_id) + if not claim: + return False + self.memory.delete_logs([claim_id]) + for proposal in self.store.list_proposals(workspace_id=self.memory.workspace_id, limit=10000): + if claim_id not in proposal.get("validation", {}).get("dependencies", []): + continue + skill_id = proposal.get("validation", {}).get("skill_id") + if skill_id and self.registry: + self.registry.rollback_learned(skill_id) + self.store.update_proposal(proposal["id"], status="rejected", + validation={**proposal["validation"], "reason": "source claim deleted"}) + self.store.append_event("memory.claim_forgotten", {"memory_id": claim_id}, + user_id=self.memory.user_id, workspace_id=self.memory.workspace_id, + session_id=self.memory.session_id) + self.store.append_event("learning.regression_cases_deactivated", {"dependency": claim_id}, + user_id=self.memory.user_id, workspace_id=self.memory.workspace_id, + session_id=self.memory.session_id) + return True + + def negative_preflight(self, profile: str, task: str) -> list[dict[str, Any]]: + signature = self.task_signature(profile, task) + forgotten = {event["payload"].get("dependency") for event in self.store.list_events( + "learning.regression_cases_deactivated", limit=10000, workspace_id=self.memory.workspace_id)} + result = [] + for event in self.store.list_events("learning.regression_case_created", limit=10000, + 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) + return result[:3] + 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) diff --git a/main.py b/main.py index 106d95c..b369a03 100644 --- a/main.py +++ b/main.py @@ -2054,7 +2054,11 @@ def _search_memory(args: str) -> str: def _build_memory_context(query: str, n: int = 3) -> str: """Return bounded, explicitly untrusted memory data for the model.""" - recalled = memory_bank.recall_records(query, n_results=n) + profile = learning_engine.route_profile(query, _agent_profile_mode) + recalled = memory_bank.recall_records( + query, n_results=n, profile=profile, + task_signature=learning_engine.task_signature(profile, query), + ) if not recalled: return "" lines = [ @@ -3745,6 +3749,17 @@ def main() -> None: console.print("Usage: /learning status [coder|researcher] | /learning metrics [profile] | /learning rollback | /learning explain|evidence|replay ") continue + if user_input.lower().startswith("/memory "): + parts = user_input.split(maxsplit=2) + if len(parts) == 3 and parts[1].lower() == "why": + claim = learning_engine.explain_claim(parts[2].strip()) + console.print(json.dumps(claim, ensure_ascii=False, indent=2) if claim else "Memory claim not found.") + elif len(parts) == 3 and parts[1].lower() == "forget": + console.print("Memory claim forgotten." if learning_engine.forget_claim(parts[2].strip()) else "Memory claim not found.") + else: + console.print("Usage: /memory why|forget ") + continue + # /forget — show and optionally delete recent learnings if user_input.lower().startswith("/forget"): parts = user_input.split(maxsplit=1) diff --git a/memory.py b/memory.py index 1a3f71f..8ec8681 100644 --- a/memory.py +++ b/memory.py @@ -188,7 +188,8 @@ def recall(self, query: str, n_results: int = 2) -> list[str]: scored.sort(key=lambda item: (item[0], item[1]), reverse=True) return [item[2] for item in scored[:limit]] - def recall_records(self, query: str, n_results: int = 2) -> list[dict[str, Any]]: + def recall_records(self, query: str, n_results: int = 2, *, profile: str | None = None, + task_signature: str | None = None) -> list[dict[str, Any]]: """Return recalled data with provenance and trust metadata.""" documents = self.recall(query, n_results=n_results) if not documents: @@ -198,7 +199,15 @@ def recall_records(self, query: str, n_results: int = 2) -> list[dict[str, Any]] by_content: dict[str, dict[str, Any]] = {} for row in rows: by_content.setdefault(row["content"], row) - return [by_content[doc] for doc in documents if doc in by_content] + records = [by_content[doc] for doc in documents if doc in by_content] + context = {"profile": profile, "project": self.workspace_id, "task": task_signature} + visible = [] + for row in records: + scope = row.get("metadata", {}).get("scope") + if scope and scope.get("type") != "global" and context.get(scope.get("type")) != scope.get("value"): + continue + visible.append(row) + return visible def get_recent(self, n: int = 10) -> list[str]: rows = self.store.list_memories(status="active", limit=max(1, min(n, 10000)), workspace_id=self.workspace_id, diff --git a/server.py b/server.py index 7225ec3..98e3d0c 100644 --- a/server.py +++ b/server.py @@ -512,6 +512,28 @@ async def api_v2_memory(q: str = "", limit: int = 10, session_id: str | None = N }} +@app.get("/api/v2/memory/claims", dependencies=[Depends(require_api_access)]) +async def api_v2_memory_claims(): + rows = _agent.memory_bank.store.list_memories(status=None, limit=10000, + workspace_id=_agent.memory_bank.workspace_id) + return {"claims": [row for row in rows if row.get("metadata", {}).get("claim")]} + + +@app.get("/api/v2/memory/claims/{claim_id}", dependencies=[Depends(require_api_access)]) +async def api_v2_memory_claim(claim_id: str): + claim = _agent.learning_engine.explain_claim(claim_id) + if claim is None: + raise HTTPException(404, "Memory claim not found") + return claim + + +@app.delete("/api/v2/memory/claims/{claim_id}", dependencies=[Depends(require_api_access)]) +async def api_v2_memory_forget_claim(claim_id: str): + if not _agent.learning_engine.forget_claim(claim_id): + raise HTTPException(404, "Memory claim not found") + return {"status": "forgotten", "memory_id": claim_id} + + @app.get("/api/v2/tasks", dependencies=[Depends(require_api_access)]) async def api_v2_tasks(session_id: str | None = None): session = _normalise_session_id(session_id) if session_id else None diff --git a/subagents.py b/subagents.py index b23f14b..6793c1f 100644 --- a/subagents.py +++ b/subagents.py @@ -66,7 +66,10 @@ def run(self, profile_name: str, task: str, *, workspace_id: str | None = None) scoped_memory = MemoryBank(self.memory.db_path, user_id=self.memory.user_id, workspace_id=workspace_id, session_id=run_id if profile.memory_scope == "session" else None) - context = scoped_memory.recall_records(task, n_results=5) if profile.memory_scope != "none" else [] + context = (scoped_memory.recall_records( + task, n_results=5, profile=profile.name, + task_signature=self.learning_engine.task_signature(profile.name, task) if self.learning_engine else None, + ) if profile.memory_scope != "none" else []) learning_run = None receipts: list[dict[str, Any]] = [] if self.learning_engine and profile.evolution_enabled and profile.name in {"coder", "researcher"}: diff --git a/tests/test_memory_claims.py b/tests/test_memory_claims.py new file mode 100644 index 0000000..d6f9faf --- /dev/null +++ b/tests/test_memory_claims.py @@ -0,0 +1,72 @@ +import tempfile +import unittest +from pathlib import Path + +from learning_engine import LearningEngine +from memory import MemoryBank +from skill_registry import SkillRegistry + + +class MemoryClaimTests(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 test_inferred_claim_needs_two_observations(self): + first = self.engine.remember_claim(key="editor", value="vim", evidence_id="one") + self.assertEqual(first["status"], "candidate") + self.assertIsNone(self.engine.resolve_claim("editor")) + second = self.engine.remember_claim(key="editor", value="vim", evidence_id="two") + self.assertEqual(second["status"], "active") + self.assertEqual(self.engine.resolve_claim("editor")["value"], "vim") + + def test_owner_claim_supersedes_and_scope_precedence_is_deterministic(self): + old = self.engine.remember_claim(key="language", value="Python", authority="owner") + self.engine.remember_claim(key="language", value="Rust", authority="owner") + self.engine.remember_claim(key="language", value="Swift", authority="owner", + scope="profile", scope_value="coder") + self.assertEqual(self.engine.explain_claim(old["memory_id"])["status"], "superseded") + self.assertEqual(self.engine.resolve_claim("language", profile="coder")["value"], "Swift") + self.assertEqual(self.engine.resolve_claim("language", profile="researcher")["value"], "Rust") + + def test_conflicting_inference_requests_clarification(self): + self.engine.remember_claim(key="format", value="table", authority="owner") + candidate = self.engine.remember_claim(key="format", value="prose", evidence_id="one") + self.assertTrue(candidate["needs_clarification"]) + again = self.engine.remember_claim(key="format", value="prose", evidence_id="two") + self.assertEqual(again["status"], "candidate") + self.assertTrue(again["needs_clarification"]) + + def test_forget_deactivates_dependent_artifact(self): + claim = self.engine.remember_claim(key="test command", value="pytest", authority="owner") + body = """# Test\n\n## Trigger\nUse pytest.\n\n## Steps\n1. Run it.\n\n## Verify\nRequire success.\n""" + artifact = self.engine.propose_artifact("run", { + "name": "claim-dependent", "profile": "coder", "triggers": ["pytest"], + "verification": ["pytest succeeds"], "body": body, "dependencies": [claim["memory_id"]], + }) + self.assertTrue(self.engine.forget_claim(claim["memory_id"])) + proposal = next(item for item in self.engine.status(100) if item["id"] == artifact["proposal_id"]) + self.assertEqual(proposal["status"], "rejected") + self.assertEqual(proposal["lifecycle_stage"], "rolled_back") + + def test_secret_claim_is_rejected(self): + with self.assertRaises(ValueError): + self.engine.remember_claim(key="api_key", value="supersecretvalue", authority="owner") + + def test_profile_scoped_claim_is_not_recalled_by_other_profile(self): + self.engine.remember_claim(key="test runner", value="pytest", authority="owner", + scope="profile", scope_value="coder") + coder = self.memory.recall_records("test runner pytest", n_results=5, profile="coder") + researcher = self.memory.recall_records("test runner pytest", n_results=5, profile="researcher") + self.assertEqual(len(coder), 1) + self.assertEqual(researcher, []) + + +if __name__ == "__main__": + unittest.main()