From 4a6d1bf8bb5e998e40fba8ad87689967bcfcaf83 Mon Sep 17 00:00:00 2001 From: Athena Kaminsky Date: Tue, 25 Aug 2026 21:40:41 -0500 Subject: [PATCH] feat: vendor Curry, preloaded and callable across the one universal wheel Vendors curry_core.py from Athena-Pro/Curry (with the str.format()/format_map() sandbox-escape fix from https://github.com/Athena-Pro/Curry/pull/4 already applied) into synapse/, since Curry itself isn't a pip-installable package - it's meant to be pointed at via a config path, which only works from a source checkout. Vendoring a single self-contained, stdlib-only file ships it inside NexusOS's own wheel with no extra dependency to reconcile. synapse/curry_store.py opens it into a module-level singleton (curry_db) at import time, the same pattern as memory.store.store and playbooks.store.playbook_store, and main.py imports it so it's genuinely initialized at process startup - preloaded, not lazy-on-first-use. Backed by its own CURRY_DB file (nexus_config.py), separate from memory.db. NexusOS builds exactly one wheel (py3-none-any, no compiled extensions) - there is no separate Windows/macOS/Linux artifact; platform differences are handled by requirement overlays at install time, not by building different wheels. Verified the same wheel actually carries this correctly: built it, confirmed twine check passes, confirmed synapse/curry_core.py and curry_store.py are present in the archive (bin/check.sh's packaging gate now asserts this too), then installed that exact wheel into a throwaway venv and round-tripped a declare_constant/get_constant_latest call against it with no source checkout present - proving "preloaded and ready to be called" holds from the shipped artifact, not just editable-install execution. Android/Termux is unaffected by this change in either direction: it already has a separate, documented, pre-existing blocker in docs/TERMUX.md (no published Android pydantic-core wheel) that has nothing to do with Curry, which is pure stdlib and adds no new native/binary dependency. Scope: preload only, nothing wired into a chat-facing tool yet - no model or user-authored content reaches declare_function/call_function today. Verified: 216 backend tests pass (4 new in test_curry_store.py, including a regression test proving the vendored sandbox fix survived the copy); the 12 pre-existing C/C++/Rust toolchain failures are unrelated and unchanged. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 3 + bin/check.sh | 2 + synapse/curry_core.py | 1702 +++++++++++++++++++++++++++++++++++++ synapse/curry_store.py | 22 + synapse/main.py | 1 + synapse/nexus_config.py | 8 +- tests/test_curry_store.py | 55 ++ 7 files changed, 1792 insertions(+), 1 deletion(-) create mode 100644 synapse/curry_core.py create mode 100644 synapse/curry_store.py create mode 100644 tests/test_curry_store.py diff --git a/CLAUDE.md b/CLAUDE.md index 6bffd47..eace049 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -126,6 +126,9 @@ React 19 + Vite. No routing library — `App.jsx` manages page state in a single ### Persistent Storage Most data lands in `synapse/memory/memory.db` (SQLite, WAL mode). Tables: memory facts, conversations, messages, app settings. `synapse/memory/store.py` (`PersistentMemoryStore`) owns the schema and all queries. Playbooks are the exception — they live as YAML files in `data/playbooks/` (see Playbook System). `nexus_config.py` defines all paths; it also ensures all required directories exist on import. +### Curry (`synapse/curry_core.py` + `synapse/curry_store.py`) +`curry_core.py` is vendored, unmodified-except-for-one-fix, from [Athena-Pro/Curry](https://github.com/Athena-Pro/Curry) — an immutable, versioned fact store (constants, functions, model registrations, inference provenance) backed by its own SQLite file (`CURRY_DB` in `nexus_config.py`, separate from `memory.db`). `curry_store.py` opens it into a module-level singleton (`curry_db`) at import time — the same pattern as `memory.store.store` / `playbooks.store.playbook_store` — so it's preloaded and callable (`curry_db.declare_constant(...)`, `curry_db.call_function(...)`, etc.) from anywhere in the backend without extra setup. It ships inside the wheel (`bin/check.sh`'s packaging gate asserts this) and has no external dependencies of its own. Nothing currently wires chat/model-authored content into it — it's available, not yet exposed as an action tool. Re-sync `curry_core.py` from upstream by hand, not by script; see the file's own docstring for what changed and why. + ### Logs & Runtime State - `runtime/backend.log`, `runtime/frontend.log`, `runtime/memory.log` — service stdout - `runtime/logs/ollama.log`, `runtime/logs/chat.log` diff --git a/bin/check.sh b/bin/check.sh index a8dbbe4..dbaad84 100644 --- a/bin/check.sh +++ b/bin/check.sh @@ -63,6 +63,8 @@ if not any(n.startswith("synapse/_resources/web/") for n in names): sys.exit("wheel is missing the compiled web UI (cd interface/web && npm run build)") if not any(n.startswith("synapse/_resources/playbooks/") for n in names): sys.exit("wheel is missing the seed playbooks") +if "synapse/curry_core.py" not in names or "synapse/curry_store.py" not in names: + sys.exit("wheel is missing vendored Curry (synapse/curry_core.py / curry_store.py)") print(f"wheel OK: {len(names)} files") PY else diff --git a/synapse/curry_core.py b/synapse/curry_core.py new file mode 100644 index 0000000..71aa3e6 --- /dev/null +++ b/synapse/curry_core.py @@ -0,0 +1,1702 @@ +""" +Curry: A Functional Database for LLM Operations +Core implementation with SQLite backend, type safety, and deterministic execution. + +Vendored from https://github.com/Athena-Pro/Curry (curry_core.py), not written +for NexusOS. Kept as a single self-contained, stdlib-only file specifically so +it can be vendored cleanly like this - no external dependencies, no package +metadata of its own to reconcile with pyproject.toml. + +Includes the fix from https://github.com/Athena-Pro/Curry/pull/4: a function +body could pass validate_function_body's AST check by hiding dunder-attribute +traversal inside a str.format()/str.format_map() field spec (e.g. +'{0.__globals__}'.format(x)), which the AST walk never inspects since it only +looks at literal Attribute/Name nodes, not string constant contents. Any +function with a function_bindings dependency hands eval_context a real Python +closure, and a closure's __globals__ is this module's own namespace - so that +was a working sandbox escape, not a theoretical one. Re-sync deliberately, not +automatically: pull upstream changes by hand and re-diff against this file's +history rather than scripting the sync, so a change here always has a reason +attached to it. + +Only Curry's own database operations execute against this file (declare_*, +call_function, etc.) - nothing in NexusOS wires model-authored content into +declare_function today. See synapse/curry_store.py for how NexusOS opens it. +""" + +import sqlite3 +import json +import hashlib +import uuid +import base64 +import ast +import time +from typing import Any, Dict, List, Optional, Set +from dataclasses import dataclass +from enum import Enum + + +_SAFE_BUILTINS = { + "abs": abs, "all": all, "any": any, "bool": bool, "dict": dict, + "enumerate": enumerate, "filter": filter, "float": float, "int": int, + "len": len, "list": list, "map": map, "max": max, "min": min, + "set": set, "str": str, "sum": sum, "tuple": tuple, "zip": zip, + "round": round +} + +# str.format / str.format_map parse "{0.__class__...}"-style field specs at +# RUNTIME, walking attributes and items on whatever value is passed in via +# getattr/getitem -- including dunder attributes. That traversal happens +# entirely inside the *contents* of a string constant, so validate_function_body's +# AST walk below never sees it: '{0.__globals__}'.format(x) contains no literal +# dunder-prefixed Attribute or Name node anywhere in the source tree, only an +# innocuous-looking .format() call. Any function_bindings entry hands eval_context +# a real Python closure (see call_function), and a closure's __globals__ is the +# whole curry_core module namespace -- so this was a working sandbox escape, not +# a theoretical one. str(), string concatenation, and %-formatting don't support +# attribute/item traversal and stay allowed. +_UNSAFE_STR_METHODS = frozenset({"format", "format_map"}) + + +class TypeSignature(Enum): + """Supported type signatures for constants.""" + FLOAT64 = "Float64" + INT32 = "Int32" + STRING = "String" + BLOB = "Blob" + JSON_TYPE = "Json" + TOKENS = "Tokens" # Token sequences + CURRENCY = "Currency" + BOOL = "Bool" + + +@dataclass +class VersionedRef: + """Reference to a versioned entity (constant, function, or model).""" + name: str + version: int + + def __str__(self): + return f"{self.name}@v{self.version}" + + @staticmethod + def parse(ref_str: str) -> 'VersionedRef': + """Parse 'name@v3' format.""" + if '@v' not in ref_str: + raise ValueError(f"Invalid versioned reference format: {ref_str}") + name, version_str = ref_str.split('@v') + return VersionedRef(name, int(version_str)) + + +class Curry: + """Main Curry database interface.""" + + def __init__(self, db_path: str = ":memory:", fallback_db: Optional['Curry'] = None, uri: bool = False): + """Initialize Curry with SQLite backend.""" + self.db_path = db_path + self.fallback_db = fallback_db + self.conn = sqlite3.connect(db_path, uri=uri) + self.conn.row_factory = sqlite3.Row + self.conn.execute("PRAGMA journal_mode=WAL;") + self._initialize_schema() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + def _initialize_schema(self): + """Create all tables and triggers for Curry.""" + cursor = self.conn.cursor() + + # Skip all DDL on read-only connections (e.g. core_db opened via mode=ro URI). + # The schema is assumed to be current on disk; migrations were applied the last + # time the DB was opened in write mode. SAVEPOINT is the cheapest write probe. + try: + cursor.execute("SAVEPOINT __schema_probe__") + cursor.execute("RELEASE SAVEPOINT __schema_probe__") + except sqlite3.OperationalError: + return # read-only connection — nothing to migrate + + # Retirement tags: group related retirements + cursor.execute(""" + CREATE TABLE IF NOT EXISTS retirement_tags ( + tag_id TEXT PRIMARY KEY, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + reason TEXT NOT NULL, + description TEXT + ) + """) + + # Constants: immutable, versioned values + cursor.execute(""" + CREATE TABLE IF NOT EXISTS constants ( + id TEXT NOT NULL, + version INTEGER NOT NULL, + value BLOB NOT NULL, + type_signature TEXT NOT NULL, + declared_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + retired_at TIMESTAMP, + retirement_tag_id TEXT, + + PRIMARY KEY (id, version), + FOREIGN KEY (retirement_tag_id) REFERENCES retirement_tags(tag_id) + ) + """) + + # Type compatibility: ensure type consistency across versions + cursor.execute(""" + CREATE TABLE IF NOT EXISTS type_compatibility ( + constant_id TEXT NOT NULL, + from_version INTEGER NOT NULL, + to_version INTEGER NOT NULL, + is_compatible BOOLEAN DEFAULT 1, + conversion_function TEXT, + + PRIMARY KEY (constant_id, from_version, to_version), + FOREIGN KEY (constant_id, from_version) REFERENCES constants(id, version), + FOREIGN KEY (constant_id, to_version) REFERENCES constants(id, version) + ) + """) + + # Functions: composed from constants and other functions + cursor.execute(""" + CREATE TABLE IF NOT EXISTS functions ( + name TEXT NOT NULL, + version INTEGER NOT NULL, + body TEXT NOT NULL, + constant_bindings TEXT NOT NULL, -- JSON: {"const_id": "v2", ...} + function_bindings TEXT, -- JSON: {"func_name": "v1", ...} + is_pure BOOLEAN DEFAULT 0, + declared_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + retired_at TIMESTAMP, + retirement_tag_id TEXT, + + PRIMARY KEY (name, version), + FOREIGN KEY (retirement_tag_id) REFERENCES retirement_tags(tag_id) + ) + """) + + try: + cursor.execute("ALTER TABLE functions ADD COLUMN expected_args TEXT") + except sqlite3.OperationalError: + pass + + try: + cursor.execute("ALTER TABLE functions ADD COLUMN description TEXT") + except sqlite3.OperationalError: + pass + + try: + cursor.execute("ALTER TABLE functions ADD COLUMN arg_descriptions TEXT") + except sqlite3.OperationalError: + pass + + try: + cursor.execute("ALTER TABLE constants ADD COLUMN description TEXT") + except sqlite3.OperationalError: + pass + + # Function dependencies: track exact versions used + cursor.execute(""" + CREATE TABLE IF NOT EXISTS function_dependencies ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + function_name TEXT NOT NULL, + function_version INTEGER NOT NULL, + depends_on_constant_id TEXT, + depends_on_constant_version INTEGER, + depends_on_function_name TEXT, + depends_on_function_version INTEGER, + + FOREIGN KEY (function_name, function_version) REFERENCES functions(name, version), + FOREIGN KEY (depends_on_constant_id, depends_on_constant_version) + REFERENCES constants(id, version), + FOREIGN KEY (depends_on_function_name, depends_on_function_version) + REFERENCES functions(name, version) + ) + """) + + # Model versions: LLM checkpoints with locked inference parameters + cursor.execute(""" + CREATE TABLE IF NOT EXISTS model_versions ( + model_name TEXT NOT NULL, + version INTEGER NOT NULL, + checkpoint_hash TEXT NOT NULL, + model_type TEXT, -- 'llama', 'gpt', 'claude', etc. + base_model_name TEXT, + base_model_version INTEGER, + + -- Inference parameters (locked at version time) + temperature REAL, + top_p REAL, + max_tokens INTEGER, + + -- System prompt reference + system_prompt_id TEXT, + system_prompt_version INTEGER, + + -- Training lineage + trained_on_data_id TEXT, + trained_on_data_version INTEGER, + + declared_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + retired_at TIMESTAMP, + retirement_tag_id TEXT, + + PRIMARY KEY (model_name, version), + FOREIGN KEY (retirement_tag_id) REFERENCES retirement_tags(tag_id), + FOREIGN KEY (system_prompt_id, system_prompt_version) + REFERENCES constants(id, version) + ) + """) + + # Prompts: template compositions with input/output schemas + cursor.execute(""" + CREATE TABLE IF NOT EXISTS prompts ( + prompt_id TEXT NOT NULL, + version INTEGER NOT NULL, + name TEXT, + description TEXT, + system_prompt_id TEXT, + system_prompt_version INTEGER, + instruction_template TEXT NOT NULL, + input_schema TEXT, -- JSON + output_schema TEXT, -- JSON + + declared_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + retired_at TIMESTAMP, + retirement_tag_id TEXT, + + PRIMARY KEY (prompt_id, version), + FOREIGN KEY (retirement_tag_id) REFERENCES retirement_tags(tag_id), + FOREIGN KEY (system_prompt_id, system_prompt_version) + REFERENCES constants(id, version) + ) + """) + + # Inferences: LLM inference results with full provenance + cursor.execute(""" + CREATE TABLE IF NOT EXISTS inferences ( + inference_id TEXT PRIMARY KEY, + model_name TEXT NOT NULL, + model_version INTEGER NOT NULL, + input_tokens TEXT, -- JSON or text representation + output_tokens BLOB NOT NULL, + + temperature_used REAL, + top_p_used REAL, + seed INTEGER, + + execution_timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + execution_duration_ms INTEGER, + metadata TEXT, -- JSON: cost, latency details, etc. + + FOREIGN KEY (model_name, model_version) REFERENCES model_versions(model_name, version) + ) + """) + + # Execution cache: deterministic memoization + cursor.execute(""" + CREATE TABLE IF NOT EXISTS execution_cache ( + function_name TEXT NOT NULL, + function_version INTEGER NOT NULL, + input_hash TEXT NOT NULL, + output_hash TEXT NOT NULL, + cached_result BLOB NOT NULL, + cached_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + hit_count INTEGER DEFAULT 1, + + PRIMARY KEY (function_name, function_version, input_hash), + FOREIGN KEY (function_name, function_version) REFERENCES functions(name, version) + ) + """) + + # Note: Validation is done in Python layer for clarity and robustness + # Type checking happens in declare_constant() method + # Dependency validation happens in declare_function() method + + self.conn.commit() + + def _validate_type_signature(self, type_signature: str) -> TypeSignature: + """Validate and normalize a declared type signature.""" + for candidate in TypeSignature: + if candidate.value == type_signature: + return candidate + raise ValueError(f"Unsupported type signature: {type_signature}") + + def _serialize_constant_value(self, value: Any, type_signature: str) -> bytes: + """Serialize a constant value according to its declared type.""" + type_enum = self._validate_type_signature(type_signature) + + if type_enum == TypeSignature.BLOB: + if not isinstance(value, bytes): + raise TypeError("Blob constants must be bytes") + return value + + if type_enum == TypeSignature.FLOAT64: + if not isinstance(value, (int, float)) or isinstance(value, bool): + raise TypeError("Float64 constants must be numeric") + return json.dumps(float(value)).encode("utf-8") + + if type_enum == TypeSignature.INT32: + if not isinstance(value, int) or isinstance(value, bool): + raise TypeError("Int32 constants must be integers") + return json.dumps(value).encode("utf-8") + + if type_enum == TypeSignature.BOOL: + if not isinstance(value, bool): + raise TypeError("Bool constants must be booleans") + return json.dumps(value).encode("utf-8") + + if type_enum == TypeSignature.STRING: + if not isinstance(value, str): + raise TypeError(f"{type_signature} constants must be strings") + return json.dumps(value).encode("utf-8") + + if type_enum == TypeSignature.TOKENS: + if not isinstance(value, (str, list, dict)): + raise TypeError(f"{type_signature} constants must be strings, lists of integers, or dictionaries") + return json.dumps(value).encode("utf-8") + + if type_enum == TypeSignature.JSON_TYPE: + return json.dumps(value).encode("utf-8") + + if type_enum == TypeSignature.CURRENCY: + if not isinstance(value, (int, float, str)) or isinstance(value, bool): + raise TypeError("Currency constants must be numeric or string values") + return json.dumps(value).encode("utf-8") + + raise ValueError(f"Unsupported type signature: {type_signature}") + + def _deserialize_constant_value(self, raw_value: bytes, type_signature: str) -> Any: + """Deserialize a constant value according to its declared type.""" + type_enum = self._validate_type_signature(type_signature) + + if type_enum == TypeSignature.BLOB: + return raw_value + + value = json.loads(raw_value.decode("utf-8")) + + # Validate that the retrieved value still matches the declared type + if type_enum == TypeSignature.FLOAT64 and (not isinstance(value, (int, float)) or isinstance(value, bool)): + raise TypeError("Float64 constants must be numeric") + elif type_enum == TypeSignature.INT32 and (not isinstance(value, int) or isinstance(value, bool)): + raise TypeError("Int32 constants must be integers") + elif type_enum == TypeSignature.BOOL and not isinstance(value, bool): + raise TypeError("Bool constants must be booleans") + elif type_enum == TypeSignature.STRING and not isinstance(value, str): + raise TypeError(f"{type_signature} constants must be strings") + elif type_enum == TypeSignature.TOKENS and not isinstance(value, (str, list, dict)): + raise TypeError(f"{type_signature} constants must be strings, lists of integers, or dictionaries") + elif type_enum == TypeSignature.CURRENCY and (not isinstance(value, (int, float, str)) or isinstance(value, bool)): + raise TypeError("Currency constants must be numeric or string values") + + return value + + def _serialize_cached_result(self, result: Any) -> bytes: + """Serialize a cached function result.""" + if isinstance(result, bytes): + payload = {"encoding": "base64", "value": base64.b64encode(result).decode("ascii")} + else: + payload = {"encoding": "json", "value": result} + + try: + return json.dumps(payload, sort_keys=True).encode("utf-8") + except TypeError as exc: + raise TypeError( + "Function results must be JSON-serializable or bytes to be cached" + ) from exc + + def _deserialize_cached_result(self, raw_value: bytes) -> Any: + """Deserialize a cached function result.""" + payload = json.loads(raw_value.decode("utf-8")) + if payload["encoding"] == "base64": + return base64.b64decode(payload["value"].encode("ascii")) + return payload["value"] + + def _canonicalize_for_hash(self, value: Any) -> Any: + """Convert values into a deterministic, JSON-compatible structure.""" + if value is None or isinstance(value, (str, int, bool)): + return value + + if isinstance(value, float): + # Preserve deterministic float representation for hashing purposes. + return {"__float__": repr(value)} + + if isinstance(value, bytes): + return { + "__bytes__": base64.b64encode(value).decode("ascii") + } + + if isinstance(value, list): + return [self._canonicalize_for_hash(item) for item in value] + + if isinstance(value, tuple): + return { + "__tuple__": [self._canonicalize_for_hash(item) for item in value] + } + + if isinstance(value, dict): + return { + str(key): self._canonicalize_for_hash(val) + for key, val in sorted(value.items(), key=lambda item: str(item[0])) + } + + raise TypeError( + f"Unsupported argument type for deterministic hashing: {type(value).__name__}" + ) + + def _canonical_json_dumps(self, value: Any) -> str: + """Serialize value in a deterministic way suitable for hashing/storage.""" + canonical = self._canonicalize_for_hash(value) + return json.dumps(canonical, sort_keys=True, separators=(",", ":")) + + def _normalize_inference_input(self, input_tokens: Any) -> Dict[str, Any]: + """Normalize inference input into a canonical structure.""" + normalized = { + "raw_text": input_tokens if isinstance(input_tokens, str) else None, + "token_refs": input_tokens if isinstance(input_tokens, (dict, list)) else None, + "source_type": type(input_tokens).__name__, + } + return normalized + + # ============================================================================ + # CONSTANT OPERATIONS + # ============================================================================ + + def declare_constant( + self, + const_id: str, + version: int, + value: Any, + type_signature: str, + description: Optional[str] = None, + ) -> None: + """Declare a new version of a constant.""" + # Serialize early to fail fast on bad types before touching the DB. + value_blob = self._serialize_constant_value(value, type_signature) + + cursor = self.conn.cursor() + + # Validate type consistency + cursor.execute( + "SELECT DISTINCT type_signature FROM constants WHERE id = ? LIMIT 2", + (const_id,) + ) + rows = cursor.fetchall() + if rows and rows[0]["type_signature"] != type_signature: + raise TypeError( + f"Type mismatch for constant {const_id}: " + f"existing type is {rows[0]['type_signature']}, " + f"but attempted to declare {type_signature}" + ) + + # Advisory pre-check for a clear error message; the PRIMARY KEY + # constraint below is the actual guard against concurrent races. + cursor.execute( + "SELECT MAX(version) AS max_version FROM constants WHERE id = ?", + (const_id,) + ) + existing = cursor.fetchone() + if existing and existing["max_version"] is not None and version <= existing["max_version"]: + raise ValueError( + f"Version for constant {const_id} must be greater than existing max " + f"version {existing['max_version']}; got {version}" + ) + + try: + cursor.execute( + """INSERT INTO constants (id, version, value, type_signature, description) + VALUES (?, ?, ?, ?, ?)""", + (const_id, version, value_blob, type_signature, description) + ) + self.conn.commit() + except sqlite3.IntegrityError: + self.conn.rollback() + # Re-read to give an accurate error message after the race. + cursor.execute( + "SELECT MAX(version) AS max_version FROM constants WHERE id = ?", + (const_id,) + ) + current_max = cursor.fetchone()["max_version"] + raise ValueError( + f"Version conflict for constant {const_id}: " + f"version {version} already exists or is not greater than current max " + f"{current_max}" + ) from None + + def retire_constant( + self, + const_id: str, + version: int, + retirement_tag: Optional[str] = None, + ) -> None: + """Mark a constant version as retired.""" + cursor = self.conn.cursor() + # retired_at IS NULL prevents two concurrent agents from silently + # double-retiring the same version (second call would overwrite + # retirement_tag_id with no error). + cursor.execute( + """UPDATE constants + SET retired_at = CURRENT_TIMESTAMP, retirement_tag_id = ? + WHERE id = ? AND version = ? AND retired_at IS NULL""", + (retirement_tag, const_id, version) + ) + if cursor.rowcount == 0: + self.conn.rollback() + # Distinguish "never existed" from "already retired". + cursor.execute( + "SELECT retired_at FROM constants WHERE id = ? AND version = ?", + (const_id, version) + ) + row = cursor.fetchone() + if row is None: + raise KeyError(f"Constant {const_id}@v{version} not found") + raise ValueError(f"Constant {const_id}@v{version} is already retired") + self.conn.commit() + + def retire_constant_with_reason( + self, + const_id: str, + version: int, + reason: str, + description: Optional[str] = None, + ) -> str: + """Create a retirement tag and retire a constant in one step. + + Returns the generated retirement tag ID. + """ + tag_id = f"retire_{const_id}_v{version}_{int(time.time())}" + self.create_retirement_tag(tag_id, reason, description) + self.retire_constant(const_id, version, retirement_tag=tag_id) + return tag_id + + def get_constant( + self, + const_id: str, + version: int, + ) -> Dict[str, Any]: + """Retrieve a constant by exact version.""" + cursor = self.conn.cursor() + cursor.execute( + """SELECT id, version, value, type_signature, declared_at, retired_at, description + FROM constants + WHERE id = ? AND version = ?""", + (const_id, version) + ) + row = cursor.fetchone() + if not row: + if self.fallback_db: + return self.fallback_db.get_constant(const_id, version) + raise KeyError(f"Constant {const_id}@v{version} not found") + + if row["retired_at"]: + raise ValueError(f"Constant {const_id}@v{version} has been retired") + + # Deserialize value + value = self._deserialize_constant_value(row["value"], row["type_signature"]) + + return { + "id": row["id"], + "version": row["version"], + "value": value, + "type_signature": row["type_signature"], + "declared_at": row["declared_at"], + "description": row["description"], + } + + def get_constant_latest(self, const_id: str) -> Dict[str, Any]: + """Get the most recent active version of a constant.""" + cursor = self.conn.cursor() + cursor.execute( + """SELECT id, version, value, type_signature, declared_at, description + FROM constants + WHERE id = ? AND retired_at IS NULL + ORDER BY version DESC + LIMIT 1""", + (const_id,) + ) + row = cursor.fetchone() + if not row: + if self.fallback_db: + return self.fallback_db.get_constant_latest(const_id) + raise KeyError(f"No active version of constant {const_id} found") + + value = self._deserialize_constant_value(row["value"], row["type_signature"]) + + return { + "id": row["id"], + "version": row["version"], + "value": value, + "type_signature": row["type_signature"], + "declared_at": row["declared_at"], + "description": row["description"], + } + + def list_constants(self, active_only: bool = True) -> List[Dict[str, Any]]: + """List all constants with their latest versions.""" + cursor = self.conn.cursor() + query = "SELECT id, MAX(version) as latest_version, type_signature, declared_at FROM constants" + if active_only: + query += " WHERE retired_at IS NULL" + query += " GROUP BY id" + cursor.execute(query) + results = [dict(row) for row in cursor.fetchall()] + + if self.fallback_db: + fallback_results = self.fallback_db.list_constants(active_only) + local_ids = {r["id"] for r in results} + for fr in fallback_results: + if fr["id"] not in local_ids: + results.append(fr) + + return results + + def search_constants( + self, + prefix: Optional[str] = None, + type_signature: Optional[str] = None, + active_only: bool = True, + ) -> List[Dict[str, Any]]: + """Search constants by ID prefix and/or type_signature.""" + cursor = self.conn.cursor() + conditions = [] + params: List[Any] = [] + + if prefix is not None: + conditions.append("id LIKE ?") + params.append(prefix + "%") + if type_signature is not None: + conditions.append("type_signature = ?") + params.append(type_signature) + if active_only: + conditions.append("retired_at IS NULL") + + where = ("WHERE " + " AND ".join(conditions)) if conditions else "" + cursor.execute( + f"""SELECT id, MAX(version) as latest_version, type_signature, declared_at + FROM constants {where} GROUP BY id""", + params, + ) + results = [dict(row) for row in cursor.fetchall()] + + if self.fallback_db: + fallback_results = self.fallback_db.search_constants( + prefix=prefix, type_signature=type_signature, active_only=active_only + ) + local_ids = {r["id"] for r in results} + for fr in fallback_results: + if fr["id"] not in local_ids: + results.append(fr) + + return results + + def compare_constants( + self, + const_id: str, + version_a: int, + version_b: int, + ) -> Dict[str, Any]: + """Structured diff between two versions of a constant. + + Both retired and active versions are compared. Returns a dict with: + - ``same_type``: bool — whether both versions share the same type_signature + - ``same_value``: bool — deep equality of deserialized values + - ``version_a`` / ``version_b``: the input version numbers + - ``declared_a`` / ``declared_b``: ISO timestamps when each was declared + - ``retired_a`` / ``retired_b``: ISO timestamps when each was retired, or None + - ``type_a`` / ``type_b``: type_signature strings + - ``value_a`` / ``value_b``: deserialized values (may be large — callers beware) + """ + cursor = self.conn.cursor() + + def _fetch(ver: int) -> sqlite3.Row: + cursor.execute( + """SELECT id, version, value, type_signature, declared_at, retired_at + FROM constants + WHERE id = ? AND version = ?""", + (const_id, ver), + ) + row = cursor.fetchone() + if row is None: + if self.fallback_db: + fb_cur = self.fallback_db.conn.cursor() + fb_cur.execute( + """SELECT id, version, value, type_signature, declared_at, retired_at + FROM constants WHERE id = ? AND version = ?""", + (const_id, ver), + ) + row = fb_cur.fetchone() + if row is None: + raise KeyError(f"Constant {const_id}@v{ver} not found") + return row + + row_a = _fetch(version_a) + row_b = _fetch(version_b) + + val_a = self._deserialize_constant_value(row_a["value"], row_a["type_signature"]) + val_b = self._deserialize_constant_value(row_b["value"], row_b["type_signature"]) + + return { + "const_id": const_id, + "version_a": version_a, + "version_b": version_b, + "same_type": row_a["type_signature"] == row_b["type_signature"], + "same_value": val_a == val_b, + "type_a": row_a["type_signature"], + "type_b": row_b["type_signature"], + "value_a": val_a, + "value_b": val_b, + "declared_a": row_a["declared_at"], + "declared_b": row_b["declared_at"], + "retired_a": row_a["retired_at"], + "retired_b": row_b["retired_at"], + } + + def get_constant_at_timestamp( + self, + const_id: str, + timestamp: str, + ) -> Dict[str, Any]: + """Return the active version of a constant at a given ISO-8601 UTC timestamp. + + A version is considered active at time T when: + declared_at <= T AND (retired_at IS NULL OR retired_at > T) + + The highest such version is returned (i.e. the one that was declared most + recently before T). Raises ``KeyError`` when no version was active at T. + """ + cursor = self.conn.cursor() + cursor.execute( + """SELECT id, version, value, type_signature, declared_at, retired_at + FROM constants + WHERE id = ? + AND declared_at <= ? + AND (retired_at IS NULL OR retired_at > ?) + ORDER BY version DESC + LIMIT 1""", + (const_id, timestamp, timestamp), + ) + row = cursor.fetchone() + if row is None: + if self.fallback_db: + return self.fallback_db.get_constant_at_timestamp(const_id, timestamp) + raise KeyError( + f"No active version of constant {const_id!r} found at {timestamp!r}" + ) + + value = self._deserialize_constant_value(row["value"], row["type_signature"]) + return { + "id": row["id"], + "version": row["version"], + "value": value, + "type_signature": row["type_signature"], + "declared_at": row["declared_at"], + "retired_at": row["retired_at"], + "query_timestamp": timestamp, + } + + # ============================================================================ + # FUNCTION OPERATIONS + # ============================================================================ + + def validate_function_body( + self, body: str, allowed_names: Set[str], expected_args: Optional[List[str]] = None + ) -> None: + """Statically analyze a function body to prevent unsafe constructs and verify names.""" + try: + tree = ast.parse(body, mode='eval') + except SyntaxError as e: + raise ValueError(f"Function body has syntax error: {e}") + + for node in ast.walk(tree): + if isinstance(node, ast.Attribute): + if node.attr.startswith("__"): + raise ValueError(f"Unsafe dunder attribute access: .{node.attr}") + if isinstance(node.value, ast.Name) and node.value.id.startswith("__"): + raise ValueError(f"Unsafe access on {node.value.id}") + if node.attr in _UNSAFE_STR_METHODS: + raise ValueError( + f"Unsafe method .{node.attr}() -- format-string field specs can reach " + "dunder attributes (e.g. '{0.__class__}') at runtime, invisible to this " + "static check. Build strings with concatenation or f-strings instead." + ) + + if isinstance(node, ast.Name): + if node.id not in allowed_names and node.id not in _SAFE_BUILTINS: + if expected_args is not None and node.id not in expected_args: + raise ValueError(f"Unbound name '{node.id}' not in bindings or expected args") + + def declare_function( + self, + name: str, + version: int, + body: str, + constant_bindings: Optional[Dict[str, int]] = None, + function_bindings: Optional[Dict[str, int]] = None, + is_pure: bool = False, + expected_args: Optional[List[str]] = None, + description: Optional[str] = None, + arg_descriptions: Optional[Dict[str, str]] = None, + ) -> None: + """Declare a versioned function with exact dependency versions. + + description: Human-readable summary of what the function does and which + constants it binds to. Used as the MCP tool description. Example: + 'Apply the standard markup (markup_rate constant) to a wholesale cost.' + + arg_descriptions: Per-argument hint strings surfaced as MCP tool property + descriptions. Non-obvious args (rates, proportions, enums) MUST include + a unit hint and example value. Example: + {'rate': 'Annual rate as a decimal fraction (e.g. 0.06 for 6%)', + 'years': 'Duration in whole years (e.g. 5)'} + """ + constant_bindings = constant_bindings or {} + function_bindings = function_bindings or {} + + allowed_names = set(constant_bindings.keys()) | set(function_bindings.keys()) + try: + self.validate_function_body(body, allowed_names, expected_args) + except ValueError as e: + raise ValueError(f"Function {name}@v{version} body is invalid: {e}") + + cursor = self.conn.cursor() + + # Advisory pre-check for a clear error message; the PRIMARY KEY + # constraint below is the actual guard against concurrent races. + cursor.execute( + "SELECT MAX(version) AS max_version FROM functions WHERE name = ?", + (name,) + ) + existing = cursor.fetchone() + if existing and existing["max_version"] is not None and version <= existing["max_version"]: + raise ValueError( + f"Version for function {name} must be greater than existing max " + f"version {existing['max_version']}; got {version}" + ) + + # Validate all dependencies exist and are active + for const_id, const_version in constant_bindings.items(): + try: + self.get_constant(const_id, const_version) + except KeyError: + raise ValueError( + f"Function {name}@v{version} references non-existent constant {const_id}@v{const_version}" + ) + # get_constant already checks for retired_at and raises ValueError + + for func_name, func_version in function_bindings.items(): + try: + self.get_function(func_name, func_version) + except KeyError: + raise ValueError( + f"Function {name}@v{version} references non-existent function {func_name}@v{func_version}" + ) + + try: + # Insert function + cursor.execute( + """INSERT INTO functions + (name, version, body, constant_bindings, function_bindings, is_pure, + expected_args, description, arg_descriptions) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + name, version, body, + json.dumps(constant_bindings), + json.dumps(function_bindings), + is_pure, + json.dumps(expected_args) if expected_args is not None else None, + description, + json.dumps(arg_descriptions) if arg_descriptions is not None else None, + ) + ) + + # Record dependencies + for const_id, const_version in constant_bindings.items(): + cursor.execute( + """INSERT INTO function_dependencies + (function_name, function_version, depends_on_constant_id, depends_on_constant_version) + VALUES (?, ?, ?, ?)""", + (name, version, const_id, const_version) + ) + + for func_name, func_version in function_bindings.items(): + cursor.execute( + """INSERT INTO function_dependencies + (function_name, function_version, depends_on_function_name, depends_on_function_version) + VALUES (?, ?, ?, ?)""", + (name, version, func_name, func_version) + ) + + self.conn.commit() + except sqlite3.IntegrityError: + self.conn.rollback() + cursor.execute( + "SELECT MAX(version) AS max_version FROM functions WHERE name = ?", + (name,) + ) + current_max = cursor.fetchone()["max_version"] + raise ValueError( + f"Version conflict for function {name}: " + f"version {version} already exists or is not greater than current max " + f"{current_max}" + ) from None + + def get_function(self, name: str, version: int) -> Dict[str, Any]: + """Retrieve a function by exact version.""" + cursor = self.conn.cursor() + cursor.execute( + """SELECT name, version, body, constant_bindings, function_bindings, is_pure, + expected_args, description, arg_descriptions, retired_at + FROM functions + WHERE name = ? AND version = ?""", + (name, version) + ) + row = cursor.fetchone() + if not row: + if self.fallback_db: + return self.fallback_db.get_function(name, version) + raise KeyError(f"Function {name}@v{version} not found") + + if row["retired_at"]: + raise ValueError(f"Function {name}@v{version} has been retired") + + return { + "name": row["name"], + "version": row["version"], + "body": row["body"], + "constant_bindings": json.loads(row["constant_bindings"]), + "function_bindings": json.loads(row["function_bindings"]), + "is_pure": bool(row["is_pure"]), + "expected_args": json.loads(row["expected_args"]) if row["expected_args"] is not None else None, + "description": row["description"], + "arg_descriptions": json.loads(row["arg_descriptions"]) if row["arg_descriptions"] is not None else None, + } + + def retire_function( + self, + name: str, + version: int, + retirement_tag: Optional[str] = None, + ) -> None: + """Mark a function version as retired.""" + cursor = self.conn.cursor() + # retired_at IS NULL prevents two concurrent agents from silently + # double-retiring the same version (second call would overwrite + # retirement_tag_id with no error). + cursor.execute( + """UPDATE functions + SET retired_at = CURRENT_TIMESTAMP, retirement_tag_id = ? + WHERE name = ? AND version = ? AND retired_at IS NULL""", + (retirement_tag, name, version) + ) + if cursor.rowcount == 0: + self.conn.rollback() + # Distinguish "never existed" from "already retired". + cursor.execute( + "SELECT retired_at FROM functions WHERE name = ? AND version = ?", + (name, version) + ) + row = cursor.fetchone() + if row is None: + raise KeyError(f"Function {name}@v{version} not found") + raise ValueError(f"Function {name}@v{version} is already retired") + self.conn.commit() + + def retire_function_with_reason( + self, + name: str, + version: int, + reason: str, + description: Optional[str] = None, + ) -> str: + """Create a retirement tag and retire a function in one step. + + Returns the generated retirement tag ID. + """ + tag_id = f"retire_{name}_v{version}_{int(time.time())}" + self.create_retirement_tag(tag_id, reason, description) + self.retire_function(name, version, retirement_tag=tag_id) + return tag_id + + def list_functions(self, active_only: bool = True) -> List[Dict[str, Any]]: + """List all functions with their latest versions.""" + cursor = self.conn.cursor() + query = ( + "SELECT name, MAX(version) as latest_version, is_pure, " + "expected_args, description, arg_descriptions, declared_at FROM functions" + ) + if active_only: + query += " WHERE retired_at IS NULL" + query += " GROUP BY name" + cursor.execute(query) + results = [] + for row in cursor.fetchall(): + d = dict(row) + d["expected_args"] = json.loads(d["expected_args"]) if d.get("expected_args") else None + d["arg_descriptions"] = json.loads(d["arg_descriptions"]) if d.get("arg_descriptions") else None + results.append(d) + + if self.fallback_db: + fallback_results = self.fallback_db.list_functions(active_only) + local_names = {r["name"] for r in results} + for fr in fallback_results: + if fr["name"] not in local_names: + results.append(fr) + + return results + + # ============================================================================ + # EXECUTION AND COMPOSITION + # ============================================================================ + + def call_function( + self, + name: str, + version: int, + args: Dict[str, Any], + _call_stack: Optional[Set[str]] = None, + ) -> Any: + """Execute a versioned function with locked dependencies.""" + call_key = f"{name}@v{version}" + call_stack = set(_call_stack or set()) + if call_key in call_stack: + raise RuntimeError(f"Cycle detected while executing function call stack at {call_key}") + call_stack.add(call_key) + + func_def = self.get_function(name, version) + + context = {} + for const_id, const_version in func_def["constant_bindings"].items(): + const = self.get_constant(const_id, const_version) + context[const_id] = const["value"] + + for func_name, func_version in func_def["function_bindings"].items(): + context[func_name] = ( + lambda nested_args, fn=func_name, fv=func_version, cs=call_stack: + self.call_function(fn, fv, nested_args, _call_stack=cs) + ) + + eval_context = {**context, **args} + cacheable = bool(func_def["is_pure"]) + input_hash = None + + if cacheable: + hash_payload = { + "args": args, + "constant_bindings": func_def["constant_bindings"], + "function_bindings": func_def["function_bindings"], + } + input_hash = hashlib.sha256( + self._canonical_json_dumps(hash_payload).encode("utf-8") + ).hexdigest() + cursor = self.conn.cursor() + cursor.execute( + """SELECT cached_result FROM execution_cache + WHERE function_name = ? AND function_version = ? AND input_hash = ?""", + (name, version, input_hash) + ) + cached = cursor.fetchone() + if cached: + cursor.execute( + """UPDATE execution_cache + SET hit_count = hit_count + 1 + WHERE function_name = ? AND function_version = ? AND input_hash = ?""", + (name, version, input_hash) + ) + self.conn.commit() + return self._deserialize_cached_result(cached["cached_result"]) + + try: + result = eval(func_def["body"], {"__builtins__": _SAFE_BUILTINS}, eval_context) + except Exception as exc: + raise RuntimeError(f"Failed to execute function {name}@v{version}: {exc}") from exc + + if cacheable and input_hash is not None: + try: + cursor = self.conn.cursor() + cached_result = self._serialize_cached_result(result) + output_hash = hashlib.sha256(cached_result).hexdigest() + cursor.execute( + """INSERT OR REPLACE INTO execution_cache + (function_name, function_version, input_hash, output_hash, cached_result, hit_count) + VALUES (?, ?, ?, ?, ?, COALESCE( + (SELECT hit_count FROM execution_cache + WHERE function_name = ? AND function_version = ? AND input_hash = ?), 1 + ))""", + (name, version, input_hash, output_hash, cached_result, name, version, input_hash) + ) + self.conn.commit() + except Exception: + pass + + return result + + def get_function_lineage(self, name: str, version: int) -> Dict[str, Any]: + """Get complete dependency tree for a function.""" + cursor = self.conn.cursor() + + def get_dependencies( + fn_name: str, + fn_version: int, + path: Optional[Set[str]] = None + ) -> Dict: + current_key = f"{fn_name}@v{fn_version}" + current_path = set(path or set()) + if current_key in current_path: + return { + "constants": [], + "functions": [], + "cycle_detected": True, + "cycle_at": current_key, + } + current_path.add(current_key) + + cursor.execute( + """SELECT depends_on_constant_id, depends_on_constant_version, + depends_on_function_name, depends_on_function_version + FROM function_dependencies + WHERE function_name = ? AND function_version = ?""", + (fn_name, fn_version) + ) + deps = {"constants": [], "functions": []} + + for row in cursor.fetchall(): + if row["depends_on_constant_id"]: + deps["constants"].append({ + "id": row["depends_on_constant_id"], + "version": row["depends_on_constant_version"], + }) + if row["depends_on_function_name"]: + deps["functions"].append({ + "name": row["depends_on_function_name"], + "version": row["depends_on_function_version"], + "lineage": get_dependencies( + row["depends_on_function_name"], + row["depends_on_function_version"], + current_path + ), + }) + + return deps + + return { + "function": f"{name}@v{version}", + "dependencies": get_dependencies(name, version), + } + + # ============================================================================ + # MODEL AND INFERENCE OPERATIONS + # ============================================================================ + + def register_model( + self, + model_name: str, + version: int, + checkpoint_hash: str, + temperature: float = 0.7, + top_p: float = 0.9, + max_tokens: int = 2048, + system_prompt_id: Optional[str] = None, + system_prompt_version: Optional[int] = None, + model_type: Optional[str] = None, + trained_on_data_id: Optional[str] = None, + trained_on_data_version: Optional[int] = None, + ) -> None: + """Register a model version with locked inference parameters.""" + # Validate references before touching the write path. + if system_prompt_id is not None or system_prompt_version is not None: + if system_prompt_id is None or system_prompt_version is None: + raise ValueError("system_prompt_id and system_prompt_version must be provided together") + self.get_constant(system_prompt_id, system_prompt_version) + + if trained_on_data_id is not None or trained_on_data_version is not None: + if trained_on_data_id is None or trained_on_data_version is None: + raise ValueError("trained_on_data_id and trained_on_data_version must be provided together") + self.get_constant(trained_on_data_id, trained_on_data_version) + + cursor = self.conn.cursor() + + # Advisory pre-check for a clear error message; the PRIMARY KEY + # constraint below is the actual guard against concurrent races. + cursor.execute( + "SELECT MAX(version) AS max_version FROM model_versions WHERE model_name = ?", + (model_name,) + ) + existing = cursor.fetchone() + if existing and existing["max_version"] is not None and version <= existing["max_version"]: + raise ValueError( + f"Version for model {model_name} must be greater than existing max " + f"version {existing['max_version']}; got {version}" + ) + + try: + cursor.execute( + """INSERT INTO model_versions + (model_name, version, checkpoint_hash, temperature, top_p, max_tokens, + system_prompt_id, system_prompt_version, model_type, + trained_on_data_id, trained_on_data_version) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (model_name, version, checkpoint_hash, temperature, top_p, max_tokens, + system_prompt_id, system_prompt_version, model_type, + trained_on_data_id, trained_on_data_version) + ) + self.conn.commit() + except sqlite3.IntegrityError: + self.conn.rollback() + cursor.execute( + "SELECT MAX(version) AS max_version FROM model_versions WHERE model_name = ?", + (model_name,) + ) + current_max = cursor.fetchone()["max_version"] + raise ValueError( + f"Version conflict for model {model_name}: " + f"version {version} already exists or is not greater than current max " + f"{current_max}" + ) from None + + def get_model(self, model_name: str, version: int) -> Dict[str, Any]: + """Retrieve model configuration by exact version.""" + cursor = self.conn.cursor() + cursor.execute( + """SELECT * FROM model_versions + WHERE model_name = ? AND version = ? AND retired_at IS NULL""", + (model_name, version) + ) + row = cursor.fetchone() + if not row: + if self.fallback_db: + return self.fallback_db.get_model(model_name, version) + raise KeyError(f"Model {model_name}@v{version} not found") + + return { + "model_name": row["model_name"], + "version": row["version"], + "checkpoint_hash": row["checkpoint_hash"], + "temperature": row["temperature"], + "top_p": row["top_p"], + "max_tokens": row["max_tokens"], + "system_prompt_id": row["system_prompt_id"], + "system_prompt_version": row["system_prompt_version"], + "model_type": row["model_type"], + "trained_on_data_id": row["trained_on_data_id"], + "trained_on_data_version": row["trained_on_data_version"], + } + + def get_model_latest(self, model_name: str) -> Dict[str, Any]: + """Get the most recent active version of a model.""" + cursor = self.conn.cursor() + cursor.execute( + """SELECT * + FROM model_versions + WHERE model_name = ? AND retired_at IS NULL + ORDER BY version DESC + LIMIT 1""", + (model_name,) + ) + row = cursor.fetchone() + if not row: + if self.fallback_db: + return self.fallback_db.get_model_latest(model_name) + raise KeyError(f"No active version of model {model_name} found") + return dict(row) + + def list_models(self, active_only: bool = True) -> List[Dict[str, Any]]: + """List all models with their latest versions.""" + cursor = self.conn.cursor() + query = "SELECT model_name, MAX(version) as latest_version, model_type, declared_at FROM model_versions" + if active_only: + query += " WHERE retired_at IS NULL" + query += " GROUP BY model_name" + cursor.execute(query) + results = [dict(row) for row in cursor.fetchall()] + + if self.fallback_db: + fallback_results = self.fallback_db.list_models(active_only) + local_names = {r["model_name"] for r in results} + for fr in fallback_results: + if fr["model_name"] not in local_names: + results.append(fr) + + return results + + def record_inference( + self, + model_name: str, + model_version: int, + input_tokens: Any, + output_tokens: bytes, + seed: int = 42, + temperature_used: Optional[float] = None, + top_p_used: Optional[float] = None, + duration_ms: Optional[int] = None, + metadata: Optional[Dict] = None, + ) -> str: + """Record an LLM inference result with full provenance.""" + cursor = self.conn.cursor() + inference_id = str(uuid.uuid4()) + + if duration_ms is not None and duration_ms < 0: + raise ValueError("duration_ms must be non-negative when provided") + + # Get model to verify it exists + model = self.get_model(model_name, model_version) + if temperature_used is None: + temperature_used = model["temperature"] + if top_p_used is None: + top_p_used = model["top_p"] + + input_tokens_json = self._canonical_json_dumps( + self._normalize_inference_input(input_tokens) + ) + try: + metadata_json = json.dumps(metadata or {}) + except (TypeError, ValueError) as exc: + raise TypeError(f"metadata must be JSON-serializable: {exc}") from exc + + cursor.execute( + """INSERT INTO inferences + (inference_id, model_name, model_version, input_tokens, output_tokens, + temperature_used, top_p_used, seed, execution_duration_ms, metadata) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (inference_id, model_name, model_version, input_tokens_json, output_tokens, + temperature_used, top_p_used, + seed, duration_ms, metadata_json) + ) + self.conn.commit() + + return inference_id + + def get_inference(self, inference_id: str) -> Dict[str, Any]: + """Retrieve an inference record with full provenance.""" + cursor = self.conn.cursor() + cursor.execute( + """SELECT * FROM inferences WHERE inference_id = ?""", + (inference_id,) + ) + row = cursor.fetchone() + if not row: + raise KeyError(f"Inference {inference_id} not found") + + return { + "inference_id": row["inference_id"], + "model_name": row["model_name"], + "model_version": row["model_version"], + "input_tokens": row["input_tokens"], + "output_tokens": row["output_tokens"], + "temperature_used": row["temperature_used"], + "top_p_used": row["top_p_used"], + "seed": row["seed"], + "execution_timestamp": row["execution_timestamp"], + "execution_duration_ms": row["execution_duration_ms"], + "metadata": json.loads(row["metadata"]) if row["metadata"] else {}, + } + + def search_inferences( + self, + model_name: Optional[str] = None, + model_version: Optional[int] = None, + seed: Optional[int] = None, + start_timestamp: Optional[str] = None, + end_timestamp: Optional[str] = None, + metadata_filters: Optional[Dict[str, Any]] = None, + min_input_tokens_count: Optional[int] = None, + max_input_tokens_count: Optional[int] = None, + min_output_tokens_count: Optional[int] = None, + max_output_tokens_count: Optional[int] = None, + limit: int = 50, + offset: int = 0, + ) -> List[Dict[str, Any]]: + """Search inference rows with deterministic ordering and pagination.""" + if limit <= 0: + raise ValueError("limit must be positive") + if offset < 0: + raise ValueError("offset must be non-negative") + + cursor = self.conn.cursor() + where_clauses = [] + params: List[Any] = [] + + if model_name is not None: + where_clauses.append("model_name = ?") + params.append(model_name) + if model_version is not None: + where_clauses.append("model_version = ?") + params.append(model_version) + if seed is not None: + where_clauses.append("seed = ?") + params.append(seed) + if start_timestamp is not None: + where_clauses.append("execution_timestamp >= ?") + params.append(start_timestamp) + if end_timestamp is not None: + where_clauses.append("execution_timestamp <= ?") + params.append(end_timestamp) + + where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else "" + cursor.execute( + f"""SELECT * FROM inferences + {where_sql} + ORDER BY execution_timestamp ASC, inference_id ASC""", + tuple(params), + ) + + rows = cursor.fetchall() + metadata_filters = metadata_filters or {} + filtered: List[Dict[str, Any]] = [] + + for row in rows: + metadata = json.loads(row["metadata"]) if row["metadata"] else {} + input_tokens_count = metadata.get("input_tokens_count") + output_tokens_count = metadata.get("output_tokens_count") + + metadata_match = all(metadata.get(k) == v for k, v in metadata_filters.items()) + if not metadata_match: + continue + if min_input_tokens_count is not None and ( + input_tokens_count is None or input_tokens_count < min_input_tokens_count + ): + continue + if max_input_tokens_count is not None and ( + input_tokens_count is None or input_tokens_count > max_input_tokens_count + ): + continue + if min_output_tokens_count is not None and ( + output_tokens_count is None or output_tokens_count < min_output_tokens_count + ): + continue + if max_output_tokens_count is not None and ( + output_tokens_count is None or output_tokens_count > max_output_tokens_count + ): + continue + + filtered.append( + { + "inference_id": row["inference_id"], + "model_name": row["model_name"], + "model_version": row["model_version"], + "input_tokens": row["input_tokens"], + "output_tokens": row["output_tokens"], + "temperature_used": row["temperature_used"], + "top_p_used": row["top_p_used"], + "seed": row["seed"], + "execution_timestamp": row["execution_timestamp"], + "execution_duration_ms": row["execution_duration_ms"], + "metadata": metadata, + } + ) + + return filtered[offset:offset + limit] + + def compare_inferences(self, a_id: str, b_id: str) -> Dict[str, Any]: + """Compare two inference records and return structured deltas.""" + a = self.get_inference(a_id) + b = self.get_inference(b_id) + a_output_hash = hashlib.sha256(a["output_tokens"]).hexdigest() + b_output_hash = hashlib.sha256(b["output_tokens"]).hexdigest() + + return { + "a_id": a_id, + "b_id": b_id, + "same_model": ( + a["model_name"] == b["model_name"] and + a["model_version"] == b["model_version"] + ), + "same_seed": a["seed"] == b["seed"], + "same_input_tokens": a["input_tokens"] == b["input_tokens"], + "same_output_hash": a_output_hash == b_output_hash, + "a_output_sha256": a_output_hash, + "b_output_sha256": b_output_hash, + "parameter_diff": { + "temperature_used": [a["temperature_used"], b["temperature_used"]], + "top_p_used": [a["top_p_used"], b["top_p_used"]], + }, + "metadata_diff": { + "a_only_keys": sorted(set(a["metadata"].keys()) - set(b["metadata"].keys())), + "b_only_keys": sorted(set(b["metadata"].keys()) - set(a["metadata"].keys())), + "changed_keys": sorted( + key for key in (set(a["metadata"].keys()) & set(b["metadata"].keys())) + if a["metadata"][key] != b["metadata"][key] + ), + }, + } + + # ============================================================================ + # RETIREMENT AND TAGGING + # ============================================================================ + + def create_retirement_tag(self, tag_id: str, reason: str, description: Optional[str] = None) -> str: + """Create a retirement tag to group related retirements.""" + cursor = self.conn.cursor() + cursor.execute( + """INSERT INTO retirement_tags (tag_id, reason, description) + VALUES (?, ?, ?)""", + (tag_id, reason, description) + ) + self.conn.commit() + return tag_id + + def evict_execution_cache(self, max_entries: int = 1000) -> None: + """Evict execution cache down to max_entries using least recently cached policy.""" + cursor = self.conn.cursor() + cursor.execute( + """DELETE FROM execution_cache + WHERE rowid NOT IN ( + SELECT rowid FROM execution_cache + ORDER BY cached_at DESC LIMIT ? + )""", + (max_entries,) + ) + self.conn.commit() + + def close(self): + """Close the database connection.""" + self.conn.close() + + def backup(self, target_path: str, pages: int = -1, sleep: float = 0.250) -> None: + """Safely backup the database to a target file.""" + with sqlite3.connect(target_path) as dst: + self.conn.backup(dst, pages=pages, sleep=sleep) + + def export_schema(self) -> str: + """Export the current schema as SQL.""" + cursor = self.conn.cursor() + cursor.execute("SELECT sql FROM sqlite_master WHERE type='table'") + tables = [row[0] for row in cursor.fetchall()] + return "\n\n".join(tables) + +import os + +class CurrySession: + """A two-tier session managing a global core DB and a local project DB.""" + + def __init__(self, core_db: Curry, local_db: Curry, config: Dict[str, Any]): + self.core_db = core_db + self.local_db = local_db + self.config = config + + @classmethod + def from_project(cls, project_dir: str) -> 'CurrySession': + config_path = os.path.join(project_dir, ".curry", "config.json") + if not os.path.exists(config_path): + raise FileNotFoundError(f"Curry config not found at {config_path}") + + with open(config_path, "r") as f: + config = json.load(f) + + core_db_path = config.get("core_db") + if not core_db_path: + raise ValueError("config.json must specify 'core_db'") + + # For relative paths in config, resolve them relative to project_dir + local_db_path = config.get("local_db", ".curry/curry.db") + if not os.path.isabs(local_db_path): + local_db_path = os.path.join(project_dir, local_db_path) + + # Open core as read-only — no accidental writes from project sessions + core_db_uri = f"file:{core_db_path.replace(chr(92), '/')}?mode=ro" + core_db = Curry(core_db_uri, uri=True) + + # Ensure local db dir exists + os.makedirs(os.path.dirname(local_db_path), exist_ok=True) + local_db = Curry(local_db_path, fallback_db=core_db) + + return cls(core_db, local_db, config) + + def close(self): + self.local_db.close() + self.core_db.close() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + # Model operations -> core_db + def register_model(self, *args, **kwargs): + raise PermissionError( + "register_model writes to the global core DB and cannot be called from a project session. " + "Use Curry(core_db_path) directly for model registration." + ) + + def get_model(self, *args, **kwargs): + return self.core_db.get_model(*args, **kwargs) + + def get_model_latest(self, *args, **kwargs): + return self.core_db.get_model_latest(*args, **kwargs) + + def list_models(self, *args, **kwargs): + return self.core_db.list_models(*args, **kwargs) + + def retire_model(self, *args, **kwargs): + raise PermissionError( + "retire_model writes to the global core DB and cannot be called from a project session. " + "Use Curry(core_db_path) directly for model registration." + ) + + # Local operations -> local_db + def declare_constant(self, *args, **kwargs): + return self.local_db.declare_constant(*args, **kwargs) + + def get_constant(self, *args, **kwargs): + return self.local_db.get_constant(*args, **kwargs) + + def get_constant_latest(self, *args, **kwargs): + return self.local_db.get_constant_latest(*args, **kwargs) + + def retire_constant(self, *args, **kwargs): + return self.local_db.retire_constant(*args, **kwargs) + + def list_constants(self, *args, **kwargs): + return self.local_db.list_constants(*args, **kwargs) + + def search_constants(self, *args, **kwargs): + return self.local_db.search_constants(*args, **kwargs) + + def compare_constants(self, *args, **kwargs): + return self.local_db.compare_constants(*args, **kwargs) + + def get_constant_at_timestamp(self, *args, **kwargs): + return self.local_db.get_constant_at_timestamp(*args, **kwargs) + + def retire_constant_with_reason(self, *args, **kwargs): + return self.local_db.retire_constant_with_reason(*args, **kwargs) + + def declare_function(self, *args, **kwargs): + return self.local_db.declare_function(*args, **kwargs) + + def get_function(self, *args, **kwargs): + return self.local_db.get_function(*args, **kwargs) + + def retire_function(self, *args, **kwargs): + return self.local_db.retire_function(*args, **kwargs) + + def retire_function_with_reason(self, *args, **kwargs): + return self.local_db.retire_function_with_reason(*args, **kwargs) + + def list_functions(self, *args, **kwargs): + return self.local_db.list_functions(*args, **kwargs) + + def call_function(self, *args, **kwargs): + return self.local_db.call_function(*args, **kwargs) + + def get_function_lineage(self, *args, **kwargs): + return self.local_db.get_function_lineage(*args, **kwargs) + + def record_inference(self, *args, **kwargs): + return self.local_db.record_inference(*args, **kwargs) + + def get_inference(self, *args, **kwargs): + return self.local_db.get_inference(*args, **kwargs) + + def search_inferences(self, *args, **kwargs): + return self.local_db.search_inferences(*args, **kwargs) + + def compare_inferences(self, *args, **kwargs): + return self.local_db.compare_inferences(*args, **kwargs) + + def get_retirement_tag(self, *args, **kwargs): + return self.local_db.get_retirement_tag(*args, **kwargs) + diff --git a/synapse/curry_store.py b/synapse/curry_store.py new file mode 100644 index 0000000..ea3ff12 --- /dev/null +++ b/synapse/curry_store.py @@ -0,0 +1,22 @@ +"""NexusOS's own Curry instance: preloaded at import time, ready to be called. + +Curry (curry_core.py, vendored alongside this file) is an immutable, versioned +fact store - constants, functions, model registrations, and inference +provenance, backed by SQLite. Nothing in NexusOS wires chat/model-authored +content into it yet; this module only makes it available - `from +synapse.curry_store import curry_db` and call `declare_constant`, +`get_constant_latest`, `declare_function`, `call_function`, etc. directly, the +same way `synapse.memory.store.store` and `synapse.playbooks.store.playbook_store` +are used elsewhere in this codebase. + +Kept as a separate database file (CURRY_DB) from the memory/conversation store +on purpose: Curry's schema and lifecycle are independent of the memory store's. +""" +from __future__ import annotations + +from .curry_core import Curry +from .nexus_config import CURRY_DB + +curry_db = Curry(str(CURRY_DB)) + +__all__ = ["curry_db"] diff --git a/synapse/main.py b/synapse/main.py index b405090..d1a9140 100644 --- a/synapse/main.py +++ b/synapse/main.py @@ -182,6 +182,7 @@ async def _generate_conversation_title(first_message: str, model: str) -> Option from .memory.store import store, MemoryItem from .playbooks.store import playbook_store, PlaybookItem +from .curry_store import curry_db # noqa: F401 - import triggers Curry's own preload at startup from .search import needs_web_search, web_search MEMORY_SERVICE = settings.memory_url diff --git a/synapse/nexus_config.py b/synapse/nexus_config.py index 7382adb..118e3bb 100644 --- a/synapse/nexus_config.py +++ b/synapse/nexus_config.py @@ -160,6 +160,11 @@ SEED_PLAYBOOK_DIR = ( # --- DATABASE / STORAGE FILES (match your repo) --- MEMORY_DB = _configured_path("memory_db", "NEXUS_MEMORY_DB", MEMORY_DIR / "memory.db") +# Vendored Curry (synapse/curry_core.py) database: immutable versioned +# constants/functions/models + inference provenance. Separate file from +# MEMORY_DB on purpose - Curry's schema and lifecycle are independent of the +# memory/conversation store. +CURRY_DB = _configured_path("curry_db", "NEXUS_CURRY_DB", DATA_DIR / "curry.db") # --- LOG FILES --- BACKEND_LOG = RUNTIME_DIR / "backend.log" @@ -180,6 +185,7 @@ _REQUIRED_DIRS = ( UPLOADS_DIR, EXPORTS_DIR, MEMORY_DB.parent, + CURRY_DB.parent, ) @@ -424,7 +430,7 @@ __all__ = ["Settings", "settings", "path", "VERSION", "read_user_config", "write_user_config", "init_state", "INITIALIZED_FILES", "DATA_DIR", "MODELS_DIR", "RUNTIME_DIR", "MEMORY_DIR", "LOGS_DIR", "PLAYBOOK_DIR", "UPLOADS_DIR", - "EXPORTS_DIR", "MEMORY_DB", "WEB_DIST_DIR", "FRONTEND_SOURCE_DIR", + "EXPORTS_DIR", "MEMORY_DB", "CURRY_DB", "WEB_DIST_DIR", "FRONTEND_SOURCE_DIR", "ASSETS_DIR", "SEED_PLAYBOOK_DIR", "BACKEND_LOG", "OLLAMA_LOG", "CHAT_LOG", "ALLOWED_HOSTS", "ALLOWED_ORIGINS", diff --git a/tests/test_curry_store.py b/tests/test_curry_store.py new file mode 100644 index 0000000..a1d5e6f --- /dev/null +++ b/tests/test_curry_store.py @@ -0,0 +1,55 @@ +"""synapse/curry_core.py (vendored) + synapse/curry_store.py (NexusOS's preload). + +Two concerns: the vendor sync didn't silently drop the sandbox fix from +https://github.com/Athena-Pro/Curry/pull/4, and curry_store actually gives +NexusOS a live, callable instance without wiring it into any chat-facing tool. +""" +import pytest + +from synapse.curry_core import Curry, TypeSignature +from synapse import curry_store + + +def test_curry_store_is_preloaded_and_callable(): + # curry_store.curry_db is a module-level singleton constructed at import + # time (mirrors synapse.memory.store.store / synapse.playbooks.store.playbook_store) + # - by the time this test runs, it has already opened its database file. + assert isinstance(curry_store.curry_db, Curry) + curry_store.curry_db.declare_constant("t_preload_check", 1, 1, TypeSignature.INT32.value) + assert curry_store.curry_db.get_constant_latest("t_preload_check")["value"] == 1 + curry_store.curry_db.retire_constant("t_preload_check", 1) + + +def test_curry_db_path_matches_nexus_config(tmp_path, monkeypatch): + from synapse import nexus_config + assert str(curry_store.curry_db.db_path) == str(nexus_config.CURRY_DB) + + +def test_vendored_sandbox_fix_rejects_format_dunder_escape(tmp_path): + # Regression test for the vendored fix: a body that hides dunder-attribute + # traversal inside a str.format() field spec must still be rejected at + # declare time, not just the literal '.__class__' form. If a future + # re-vendor from upstream drops the fix, this is what catches it. + db = Curry(str(tmp_path / "sandbox_check.db")) + db.declare_function("helper", 1, "1") + + exploit = "'{0.__globals__}'.format(helper)" + with pytest.raises(ValueError, match="format"): + db.declare_function("evil", 1, exploit, function_bindings={"helper": 1}) + + # the original, always-caught dunder-attribute form stays blocked too + with pytest.raises(ValueError): + db.declare_function("evil2", 1, "x.__class__", expected_args=["x"]) + + db.close() + + +def test_vendored_curry_basic_versioning_roundtrip(tmp_path): + db = Curry(str(tmp_path / "roundtrip.db")) + db.declare_constant("rate", 1, 0.1, TypeSignature.FLOAT64.value) + db.declare_function( + "apply_rate", 1, "amount * (1 + rate)", + constant_bindings={"rate": 1}, expected_args=["amount"], + ) + assert db.call_function("apply_rate", 1, {"amount": 100}) == 110.00000000000001 + db.close()