From 4a6d1bf8bb5e998e40fba8ad87689967bcfcaf83 Mon Sep 17 00:00:00 2001 From: Athena Kaminsky Date: Tue, 25 Aug 2026 21:40:41 -0500 Subject: [PATCH 1/5] 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() -- 2.39.5 From 1183ab5282c06b1b015d3b154fc272f598c8c6e9 Mon Sep 17 00:00:00 2001 From: Athena Kaminsky Date: Tue, 25 Aug 2026 22:11:00 -0500 Subject: [PATCH 2/5] feat: direct tool invocation via /tool_name(arg=val) + wire Curry as real tools Adds synapse/slash_commands.py: a chat message that's nothing but /tool_name(arg=val, arg=val) dispatches straight through tools.dispatch(), skipping model selection, RAG/playbook context assembly, and the ask-policy approval round-trip entirely. A human typing this IS the approval - there's no one else to ask - so it's a deliberate, reviewed bypass of the approval step specifically, not of anything a tool validates internally (path boundaries, size caps, Curry's own sandbox checks all still run). Argument values parse via ast.literal_eval only: strings/numbers/bools/None/literal containers, no names, no calls, no attribute access - a malformed or hostile-looking argument fails to parse rather than executing anything. Wired into chat_stream_endpoint (main.py) as an early short-circuit, before any of the RAG/model-selection work that a slash-command doesn't need. Web needed no changes (it already forwards raw text unchanged); the TUI previously swallowed every leading "/" locally and never reached the backend with it, so tui_app.py's _handle_slash now falls through to _start_chat for anything shaped like a tool call while still handling its own local meta-commands (/help, /model, /new, ...) exactly as before. Also finally wires Curry in as ten real tools (curry_declare_constant, curry_get_constant/_latest, curry_list_constants, curry_retire_constant, curry_declare_function, curry_get_function, curry_list_functions, curry_call_function, curry_retire_function) - deferred from the vendoring pass. The five write/execute ones are ACTION tools in the same always-ask-regardless-of-global-policy floor as edit_source (ALWAYS_ASK_ACTION_TOOLS, generalized in tools.py from the old self_edit-only ALWAYS_ASK_TOOLS so future tool families share one place to register into). curry_call_function is gated as an action for the same reason run_snippet is: it executes code, even sandboxed. Fixed a real bug surfaced while wiring this up: curry_db is a long-lived singleton holding one sqlite3 connection (unlike NexusOS's own memory store, which opens/closes a fresh connection per call specifically to dodge this), and sqlite3 forbids using a connection from a different thread than created it. That's a non-issue in production (uvicorn's single event-loop thread), but Starlette's TestClient runs the ASGI app through an anyio portal thread, so it broke immediately under test. Fixed at the source (curry_core.py, Curry.__init__) with check_same_thread=False, documented as a second deliberate vendoring deviation alongside the PR #4 sandbox fix - there was never real concurrent access here, just an overly strict same-thread assertion tripping on a thread-identity change with only one logical caller. Verified: 244 backend tests pass (18 new for the parser + endpoint wiring + curry tool registration, 4 new for the TUI passthrough); the 12 pre-existing C/C++/Rust toolchain failures are unrelated and unchanged. Confirmed by hand over the real HTTP endpoint: successful dispatch, zero tool_request events (approval bypass working as designed), a format()-dunder exploit attempt still rejected by the vendored sandbox fix even through the new tool registration, malformed arguments rejected before ever reaching dispatch, and an unknown tool name rejected cleanly. Wheel rebuilt and content-checked (bin/check.sh's gate now also asserts slash_commands.py ships). Co-Authored-By: Claude Sonnet 5 --- bin/check.sh | 2 + nexusos_cli/tui_app.py | 14 ++ synapse/chat.py | 10 +- synapse/curry_core.py | 48 +++-- synapse/main.py | 64 ++++++- synapse/slash_commands.py | 94 ++++++++++ synapse/tools.py | 350 ++++++++++++++++++++++++++++++++++- tests/test_slash_commands.py | 204 ++++++++++++++++++++ tests/test_tui.py | 83 +++++++++ 9 files changed, 842 insertions(+), 27 deletions(-) create mode 100644 synapse/slash_commands.py create mode 100644 tests/test_slash_commands.py diff --git a/bin/check.sh b/bin/check.sh index dbaad84..21ecf4b 100644 --- a/bin/check.sh +++ b/bin/check.sh @@ -65,6 +65,8 @@ 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)") +if "synapse/slash_commands.py" not in names: + sys.exit("wheel is missing synapse/slash_commands.py") print(f"wheel OK: {len(names)} files") PY else diff --git a/nexusos_cli/tui_app.py b/nexusos_cli/tui_app.py index 31c4135..68b7e3a 100644 --- a/nexusos_cli/tui_app.py +++ b/nexusos_cli/tui_app.py @@ -15,6 +15,7 @@ from typing import Any import httpx from synapse.nexus_config import settings +from synapse.slash_commands import parse_slash_command from .monitor import collect_snapshot @@ -344,6 +345,19 @@ class NexusTUI: log.write( f"[dim]model:[/] {_escape(self._model or '(auto)')}" ) + elif parse_slash_command(text) is not None: + # Shaped like /tool_name(arg=val, ...) rather than one of + # the local meta-commands above — not handled here, sent + # to the backend as-is. chat_stream_endpoint recognizes + # and dispatches it directly (see synapse/slash_commands.py); + # a malformed one still goes through so the user sees the + # backend's own error, with full context, in one place. + if self._busy: + log.write( + "[yellow]Still streaming — wait or Ctrl+C to interrupt[/]" + ) + else: + self._start_chat(text) else: log.write( f"[red]unknown command[/] /{_escape(cmd)} — try /help" diff --git a/synapse/chat.py b/synapse/chat.py index e479407..f15b81f 100644 --- a/synapse/chat.py +++ b/synapse/chat.py @@ -167,10 +167,16 @@ async def _run_tool_loop(manager, messages, model, tool_schemas, temperature, nu break messages.append(msg) - # If any action tool needs per-call approval, pause and wait for the user. + # Curry write/execute tools always require approval when model-issued, + # even if the global policy allows lower-risk actions. A human-typed + # /tool(...) command is dispatched separately by main.py. decisions = None action_calls = [c for c in calls if _tools.is_action(c.get("function", {}).get("name", ""))] - if policy == "ask" and action_calls: + needs_approval = policy == "ask" or any( + c.get("function", {}).get("name", "") in _tools.ALWAYS_ASK_ACTION_TOOLS + for c in action_calls + ) + if needs_approval and action_calls: event = asyncio.Event() # Single-use capability token, delivered only to the client that owns # this stream. /chat/approve requires it, so knowing the (guessable, diff --git a/synapse/curry_core.py b/synapse/curry_core.py index 71aa3e6..c101af2 100644 --- a/synapse/curry_core.py +++ b/synapse/curry_core.py @@ -7,21 +7,27 @@ 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. +Two deliberate deviations from upstream, both explained at their call site +rather than just here — re-sync by hand and re-diff against this file's +history rather than scripting the sync, so every change here keeps its reason +attached: -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. +1. The fix from https://github.com/Athena-Pro/Curry/pull/4 (validate_function_body, + below): a function body could pass the 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 — + a working sandbox escape, not a theoretical one. +2. check_same_thread=False on the connection (Curry.__init__, below) — a + long-lived singleton created at import time can legitimately be called + from a different OS thread than it was constructed on (Starlette's + TestClient runs the ASGI app through an anyio portal thread); nothing here + adds genuinely concurrent access, it relaxes an overly strict assertion. + +curry_declare_function/curry_call_function ARE reachable from model-issued +tool calls in NexusOS (see synapse/tools.py) — both are ACTION tools requiring +per-call human approval (synapse/tools.py's ALWAYS_ASK_ACTION_TOOLS), same as +run_snippet. See synapse/curry_store.py for how NexusOS opens this file. """ import sqlite3 @@ -95,7 +101,19 @@ class Curry: """Initialize Curry with SQLite backend.""" self.db_path = db_path self.fallback_db = fallback_db - self.conn = sqlite3.connect(db_path, uri=uri) + # NexusOS deviation: check_same_thread=False. self.conn is held for + # this object's whole lifetime (unlike NexusOS's own memory store, + # which opens/closes a fresh connection per call specifically to avoid + # this), and a long-lived singleton created at import time can + # legitimately be called from a different OS thread than it was + # constructed on — e.g. Starlette's TestClient runs the ASGI app + # through an anyio portal thread, and any future to_thread-offloaded + # caller would too. There is still only ever one logical caller at a + # time here (asyncio's single event loop + the GIL serialize access; + # nothing in NexusOS calls curry_db from two threads concurrently) — + # this relaxes sqlite3's same-thread assertion, it does not add real + # concurrent access that wasn't already being serialized. + self.conn = sqlite3.connect(db_path, uri=uri, check_same_thread=False) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA journal_mode=WAL;") self._initialize_schema() diff --git a/synapse/main.py b/synapse/main.py index d1a9140..edb3fdf 100644 --- a/synapse/main.py +++ b/synapse/main.py @@ -184,6 +184,7 @@ 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 +from . import slash_commands as _slash_commands MEMORY_SERVICE = settings.memory_url @@ -461,6 +462,38 @@ async def _resume_dropped_extractions() -> None: # ------------------------- # Chat (streaming) # ------------------------- +async def _slash_command_stream( + slash: "_slash_commands.SlashCommand | _slash_commands.SlashCommandError", + conversation_id: str, +) -> AsyncGenerator[str, None]: + """Dispatch an explicit slash-command without a model or approval round-trip.""" + if isinstance(slash, _slash_commands.SlashCommandError): + yield f"event: error\ndata: {_json.dumps({'detail': slash.text})}\n\n" + return + + if slash.tool not in _tools.REGISTRY: + detail = f"unknown tool: {slash.tool}" + yield f"event: error\ndata: {_json.dumps({'detail': detail})}\n\n" + return + + yield f"event: status\ndata: {_json.dumps({'tool': slash.tool})}\n\n" + raw_result = await _tools.dispatch(slash.tool, slash.args) + + content = raw_result + try: + parsed = _json.loads(raw_result) + if isinstance(parsed, dict) and isinstance(parsed.get("fence"), str): + content = parsed["fence"] + else: + content = _json.dumps(parsed, indent=2, ensure_ascii=False) + except (TypeError, ValueError): + pass + + store.add_message(conversation_id, "assistant", content) + yield f"data: {_json.dumps(content)}\n\n" + yield "event: done\ndata: {}\n\n" + + @app.post("/chat/stream") async def chat_stream_endpoint(payload: Dict[str, Any]): # Bound concurrent chats so a flood can't fan out unlimited model inference. @@ -470,13 +503,39 @@ async def chat_stream_endpoint(payload: Dict[str, Any]): _chat_slot_held = True try: message = payload.get("message", "") + conversation_id = payload.get("conversation_id") or str(_uuid.uuid4()) + + if not message: + raise HTTPException(status_code=400, detail="Missing 'message'") + + # A whole-message /tool_name(arg=val, ...) command is an explicit human + # action. It skips model selection and approval but not the tool's own + # validation; slash_commands.py accepts literal keyword values only. + slash = _slash_commands.parse_slash_command(message) + if slash is not None: + project_id = store.conversation_project(conversation_id) + if project_id is None: + project_id = store.get_settings().get("active_project", "") + store.create_conversation(conversation_id, project_id or "") + store.add_message(conversation_id, "user", message) + slash_stream = _slash_command_stream(slash, conversation_id) + + async def _slash_guarded() -> AsyncGenerator[str, None]: + try: + async for chunk in slash_stream: + yield chunk + finally: + _CHAT_INFLIGHT.release() + + _chat_slot_held = False + return StreamingResponse(_slash_guarded(), media_type="text/event-stream") + app_settings = store.get_settings() # Model precedence: explicit request > active playbook's pinned model > auto-select. _active_pb = playbook_manager.get_main_playbook() _pb_model = _active_pb.model if (_active_pb and _active_pb.model) else "" model = payload.get("model") or _pb_model or await _auto_select_model(message) context = payload.get("context", {}) - conversation_id = payload.get("conversation_id") or str(_uuid.uuid4()) history = payload.get("history", []) temperature = payload.get("temperature", app_settings.get("temperature")) num_ctx = payload.get("num_ctx", app_settings.get("num_ctx", 0)) @@ -484,9 +543,6 @@ async def chat_stream_endpoint(payload: Dict[str, Any]): gpu_offload = payload.get("gpu_offload", app_settings.get("gpu_offload", -1)) num_gpu = await get_ollama_manager().resolve_num_gpu(gpu_offload, model) - if not message: - raise HTTPException(status_code=400, detail="Missing 'message'") - # Resolve the project scope: an existing conversation keeps its bound project; # a brand-new one inherits the current workspace (active_project setting). # Everything project-scoped below (instructions, memory facts, RAG) uses it. diff --git a/synapse/slash_commands.py b/synapse/slash_commands.py new file mode 100644 index 0000000..2a573d5 --- /dev/null +++ b/synapse/slash_commands.py @@ -0,0 +1,94 @@ +"""Direct tool invocation from chat input: `/tool_name(arg=val, arg=val)`. + +A human typing this IS the approval — there's no one else to ask — so a +recognized slash-command skips the ask-policy round-trip entirely and +dispatches straight through `tools.dispatch()`, the same entry point a +model-issued tool call already goes through. It does not bypass anything a +tool validates internally (path boundaries, size caps, Curry's own sandbox +checks, etc.) — only the human-approval step, which this message already is. + +Argument values are parsed with `ast.literal_eval`, not `eval()`: strings, +numbers, booleans, None, and literal lists/dicts/tuples only. There is no way +to reference a name, call a function, or access an attribute in this syntax — +a malformed or hostile-looking argument fails to parse rather than executing +anything, which is the "lint, not run" property that makes this different +from just typing Python. + +The whole message must be nothing but the command — this is a deliberate +command line, not a directive embedded in prose. Anything else (including a +message that merely starts with `/` but isn't shaped like this) falls through +to the normal chat/model path unchanged. +""" +from __future__ import annotations + +import ast +import re +from dataclasses import dataclass +from typing import Any, Optional + +# name(args) where name is a plain identifier — the same shape as a Python +# function call, so it reads the way the tool's own schema already documents +# it. re.DOTALL: argument values (e.g. a multi-line body= string) may +# legitimately contain newlines. +_COMMAND_RE = re.compile(r"^/([A-Za-z_][A-Za-z0-9_]*)\((.*)\)\s*$", re.DOTALL) + + +@dataclass +class SlashCommand: + tool: str + args: dict[str, Any] + + +@dataclass +class SlashCommandError: + text: str + + +def parse_slash_command(message: str) -> Optional[SlashCommand | SlashCommandError]: + """Parse `/tool_name(arg=val, ...)`. + + Returns None when `message` isn't shaped like a slash-command at all (the + caller should treat it as an ordinary chat message). Returns + SlashCommandError when it looks like one but is malformed — that's worth + telling the user about rather than silently sending "/curry_call_fnction(...)" + to the model as if it were prose. + """ + stripped = (message or "").strip() + match = _COMMAND_RE.match(stripped) + if not match: + return None + + tool_name, raw_args = match.group(1), match.group(2).strip() + if not raw_args: + return SlashCommand(tool=tool_name, args={}) + + # Parse "k1=v1, k2=v2" as keyword arguments to a call with no positional + # arguments and no function to actually call — ast.parse(mode='eval') on a + # synthetic call expression reuses Python's own keyword-argument grammar + # (quoting, nesting, trailing commas) instead of hand-rolling a parser for + # it, while call() as a bare name is never resolved or invoked. + try: + tree = ast.parse(f"call({raw_args})", mode="eval") + except SyntaxError as e: + return SlashCommandError(f"could not parse arguments for /{tool_name}(...): {e}") + + call_node = tree.body + if not isinstance(call_node, ast.Call) or call_node.args: + return SlashCommandError( + f"/{tool_name}(...) arguments must be keyword form: arg=value, arg=value" + ) + + args: dict[str, Any] = {} + for kw in call_node.keywords: + if kw.arg is None: # **mapping unpacking — no source for that here + return SlashCommandError(f"/{tool_name}(...) does not support **-unpacking") + try: + args[kw.arg] = ast.literal_eval(kw.value) + except (ValueError, SyntaxError): + return SlashCommandError( + f"/{tool_name}(...): argument '{kw.arg}' must be a literal " + "(string, number, bool, None, list, dict, or tuple) — not an " + "expression, name, or call" + ) + + return SlashCommand(tool=tool_name, args=args) diff --git a/synapse/tools.py b/synapse/tools.py index 38863a4..d773232 100644 --- a/synapse/tools.py +++ b/synapse/tools.py @@ -3,16 +3,20 @@ Ollama drives the calling: `/api/chat` with a `tools` param returns `message.tool_calls`, and this module is just the registry + dispatch. -Most tools READ local state (memory, history, documents, models). A few act: -`web_search`/`fetch_url` make outbound HTTP requests, and `remember` WRITES a -memory fact. The per-playbook allowlist (`PlaybookItem.tools`) is the security -boundary — an action tool only fires when a playbook explicitly lists it. +Most tools READ local state (memory, history, documents, models). Some act: +`web_search`/`fetch_url` make outbound HTTP requests, `remember` writes a +memory fact, and `curry_*` reads or writes NexusOS's vendored Curry ledger. +The per-playbook allowlist (`PlaybookItem.tools`) is the first gate. Curry +write/execute tools additionally require per-call approval when model-issued. +A message consisting only of `/tool_name(arg=val, ...)` dispatches directly; +see `synapse/slash_commands.py` for that explicit-human-command boundary. """ from __future__ import annotations import json from typing import Awaitable, Callable +from .curry_store import curry_db from .memory.store import store, MemoryItem from .ollama_manager import get_ollama_manager @@ -215,6 +219,124 @@ async def _list_files(pattern: str = "", **_) -> str: return json.dumps(sorted(hits)) +# Curry (synapse/curry_core.py, vendored) — immutable, versioned constants and +# functions. Expected caller errors keep the same structured JSON shape as the +# other tools instead of falling through dispatch()'s generic error envelope. +_CURRY_FENCE_LANG = "nexus-curry" + + +def _curry_fence(payload: dict) -> str: + body = json.dumps(payload, ensure_ascii=False, default=str).replace("`", "\\u0060") + return f"```{_CURRY_FENCE_LANG}\n{body}\n```" + + +async def _curry_call(fn, *args, **kwargs) -> dict: + # Curry holds one SQLite connection. Calls stay on the event-loop thread, + # where these local database operations are short and naturally serialized. + try: + result = fn(*args, **kwargs) + return {"ok": True, "result": result} + except (KeyError, ValueError, TypeError, RuntimeError) as exc: + return {"ok": False, "error": str(exc)} + + +async def _curry_declare_constant( + id: str = "", version: int = 0, value=None, type_signature: str = "", + description: str = "", **_, +) -> str: + """ACTION tool: declare a new, immutable version of a named constant.""" + out = await _curry_call( + curry_db.declare_constant, id, version, value, type_signature, description or None + ) + if out["ok"]: + out = {"ok": True, "id": id, "version": version} + out["fence"] = _curry_fence({"kind": "declare_constant", **out}) + return json.dumps(out) + + +async def _curry_get_constant(id: str = "", version: int = 0, **_) -> str: + return json.dumps(await _curry_call(curry_db.get_constant, id, version)) + + +async def _curry_get_constant_latest(id: str = "", **_) -> str: + return json.dumps(await _curry_call(curry_db.get_constant_latest, id)) + + +async def _curry_list_constants(active_only: bool = True, **_) -> str: + return json.dumps(await _curry_call(curry_db.list_constants, active_only)) + + +async def _curry_retire_constant( + id: str = "", version: int = 0, reason: str = "", **_, +) -> str: + out = await _curry_call( + curry_db.retire_constant_with_reason, + id, + version, + reason or "retired via tool call", + ) + return json.dumps(out) + + +async def _curry_declare_function( + name: str = "", version: int = 0, body: str = "", + constant_bindings: dict | None = None, function_bindings: dict | None = None, + is_pure: bool = False, expected_args: list | None = None, + description: str = "", arg_descriptions: dict | None = None, **_, +) -> str: + """ACTION tool: declare one statically validated expression.""" + out = await _curry_call( + curry_db.declare_function, + name, + version, + body, + constant_bindings or {}, + function_bindings or {}, + is_pure, + expected_args, + description or None, + arg_descriptions, + ) + if out["ok"]: + out = {"ok": True, "name": name, "version": version} + out["fence"] = _curry_fence({"kind": "declare_function", **out}) + return json.dumps(out) + + +async def _curry_get_function(name: str = "", version: int = 0, **_) -> str: + return json.dumps(await _curry_call(curry_db.get_function, name, version)) + + +async def _curry_list_functions(active_only: bool = True, **_) -> str: + return json.dumps(await _curry_call(curry_db.list_functions, active_only)) + + +async def _curry_call_function( + name: str = "", version: int = 0, args: dict | None = None, **_, +) -> str: + out = await _curry_call(curry_db.call_function, name, version, args or {}) + if out["ok"]: + out["fence"] = _curry_fence({ + "kind": "call_function", + "name": name, + "version": version, + **out, + }) + return json.dumps(out) + + +async def _curry_retire_function( + name: str = "", version: int = 0, reason: str = "", **_, +) -> str: + out = await _curry_call( + curry_db.retire_function_with_reason, + name, + version, + reason or "retired via tool call", + ) + return json.dumps(out) + + # name -> (schema, callable). Schema is the OpenAI/Ollama function-tool format. REGISTRY: dict[str, tuple[dict, Callable[..., Awaitable[str]]]] = { "search_memory": ( @@ -360,13 +482,229 @@ REGISTRY: dict[str, tuple[dict, Callable[..., Awaitable[str]]]] = { }, _remember, ), + "curry_declare_constant": ( + { + "type": "function", + "function": { + "name": "curry_declare_constant", + "description": ( + "Declare a new immutable version of a Curry constant. " + "Requires per-call human approval when model-issued." + ), + "parameters": { + "type": "object", + "properties": { + "id": {"type": "string", "description": "Constant identifier."}, + "version": {"type": "integer", "description": "A new, higher version."}, + "value": {"description": "Value matching type_signature."}, + "type_signature": { + "type": "string", + "description": ( + "Float64 | Int32 | String | Blob | Json | Tokens | " + "Currency | Bool" + ), + }, + "description": {"type": "string"}, + }, + "required": ["id", "version", "value", "type_signature"], + }, + }, + }, + _curry_declare_constant, + ), + "curry_get_constant": ( + { + "type": "function", + "function": { + "name": "curry_get_constant", + "description": "Retrieve a Curry constant by exact id and version.", + "parameters": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "version": {"type": "integer"}, + }, + "required": ["id", "version"], + }, + }, + }, + _curry_get_constant, + ), + "curry_get_constant_latest": ( + { + "type": "function", + "function": { + "name": "curry_get_constant_latest", + "description": "Retrieve the latest active version of a Curry constant.", + "parameters": { + "type": "object", + "properties": {"id": {"type": "string"}}, + "required": ["id"], + }, + }, + }, + _curry_get_constant_latest, + ), + "curry_list_constants": ( + { + "type": "function", + "function": { + "name": "curry_list_constants", + "description": "List Curry constants.", + "parameters": { + "type": "object", + "properties": {"active_only": {"type": "boolean"}}, + }, + }, + }, + _curry_list_constants, + ), + "curry_retire_constant": ( + { + "type": "function", + "function": { + "name": "curry_retire_constant", + "description": ( + "Retire, but do not delete, a Curry constant version. " + "Requires per-call human approval when model-issued." + ), + "parameters": { + "type": "object", + "properties": { + "id": {"type": "string"}, + "version": {"type": "integer"}, + "reason": {"type": "string"}, + }, + "required": ["id", "version"], + }, + }, + }, + _curry_retire_constant, + ), + "curry_declare_function": ( + { + "type": "function", + "function": { + "name": "curry_declare_function", + "description": ( + "Declare a new immutable Curry function version. The body is one " + "statically validated Python expression. Requires per-call human " + "approval when model-issued." + ), + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "version": {"type": "integer"}, + "body": {"type": "string"}, + "constant_bindings": {"type": "object"}, + "function_bindings": {"type": "object"}, + "is_pure": {"type": "boolean"}, + "expected_args": { + "type": "array", + "items": {"type": "string"}, + }, + "description": {"type": "string"}, + "arg_descriptions": {"type": "object"}, + }, + "required": ["name", "version", "body"], + }, + }, + }, + _curry_declare_function, + ), + "curry_get_function": ( + { + "type": "function", + "function": { + "name": "curry_get_function", + "description": "Retrieve a Curry function by exact name and version.", + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "version": {"type": "integer"}, + }, + "required": ["name", "version"], + }, + }, + }, + _curry_get_function, + ), + "curry_list_functions": ( + { + "type": "function", + "function": { + "name": "curry_list_functions", + "description": "List Curry functions and their expected arguments.", + "parameters": { + "type": "object", + "properties": {"active_only": {"type": "boolean"}}, + }, + }, + }, + _curry_list_functions, + ), + "curry_call_function": ( + { + "type": "function", + "function": { + "name": "curry_call_function", + "description": ( + "Execute an exact Curry function version with runtime arguments. " + "Requires per-call human approval when model-issued." + ), + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "version": {"type": "integer"}, + "args": {"type": "object"}, + }, + "required": ["name", "version"], + }, + }, + }, + _curry_call_function, + ), + "curry_retire_function": ( + { + "type": "function", + "function": { + "name": "curry_retire_function", + "description": ( + "Retire, but do not delete, a Curry function version. " + "Requires per-call human approval when model-issued." + ), + "parameters": { + "type": "object", + "properties": { + "name": {"type": "string"}, + "version": {"type": "integer"}, + "reason": {"type": "string"}, + }, + "required": ["name", "version"], + }, + }, + }, + _curry_retire_function, + ), } # Tools that act (write local state or reach the network). These require an # explicit consent gate (settings.allow_action_tools) on top of the per-playbook -# allowlist — a playbook granting one isn't enough on its own. -ACTION_TOOLS = frozenset({"web_search", "fetch_url", "remember"}) +# allowlist — a playbook granting one isn't enough on its own. Curry writes and +# execution additionally require per-call approval for model-issued calls. +CURRY_ALWAYS_ASK_TOOLS = frozenset({ + "curry_declare_constant", + "curry_retire_constant", + "curry_declare_function", + "curry_retire_function", + "curry_call_function", +}) +ACTION_TOOLS = frozenset({"web_search", "fetch_url", "remember"}) | CURRY_ALWAYS_ASK_TOOLS +ALWAYS_ASK_ACTION_TOOLS = CURRY_ALWAYS_ASK_TOOLS def is_action(name: str) -> bool: diff --git a/tests/test_slash_commands.py b/tests/test_slash_commands.py new file mode 100644 index 0000000..faeef03 --- /dev/null +++ b/tests/test_slash_commands.py @@ -0,0 +1,204 @@ +"""synapse/slash_commands.py (the /tool_name(arg=val) parser) and its wiring +into chat_stream_endpoint (direct dispatch, no model call, no approval +round-trip) plus the ten curry_* tools it can now reach. +""" +import json + +import pytest +from fastapi.testclient import TestClient + +from synapse.slash_commands import SlashCommand, SlashCommandError, parse_slash_command +from synapse.main import app +from synapse import tools + + +# --------------------------------------------------------------------------- +# Parser +# --------------------------------------------------------------------------- + +def test_parses_keyword_arguments_as_python_literals(): + result = parse_slash_command('/curry_call_function(name="x", version=1, args={"a": 1})') + assert result == SlashCommand( + tool="curry_call_function", + args={"name": "x", "version": 1, "args": {"a": 1}}, + ) + + +def test_parses_no_arguments(): + assert parse_slash_command("/curry_list_functions()") == SlashCommand(tool="curry_list_functions", args={}) + + +def test_non_slash_message_returns_none(): + assert parse_slash_command("just chatting, not a command") is None + + +def test_slash_without_parens_returns_none(): + # The TUI's own local commands (/model foo, /new) use this shape — must + # never be mistaken for a tool call. + assert parse_slash_command("/model gpt") is None + + +def test_slash_embedded_in_prose_returns_none(): + assert parse_slash_command('hey /curry_call_function(name="x", version=1) run this') is None + + +def test_name_or_call_as_argument_value_is_rejected(): + # ast.literal_eval only accepts literals — a bare name or a call is a + # parse failure, not a value, so nothing here is ever evaluated. + result = parse_slash_command("/curry_call_function(x=some_name)") + assert isinstance(result, SlashCommandError) + result2 = parse_slash_command('/curry_call_function(x=__import__("os"))') + assert isinstance(result2, SlashCommandError) + + +def test_positional_arguments_are_rejected(): + result = parse_slash_command("/curry_call_function(1, 2)") + assert isinstance(result, SlashCommandError) + + +def test_double_star_unpacking_is_rejected(): + result = parse_slash_command('/curry_call_function(**{"a": 1})') + assert isinstance(result, SlashCommandError) + + +def test_malformed_syntax_is_rejected(): + result = parse_slash_command("/curry_call_function(name=)") + assert isinstance(result, SlashCommandError) + + +# --------------------------------------------------------------------------- +# Curry tool registration +# --------------------------------------------------------------------------- + +_CURRY_ACTION_TOOLS = { + "curry_declare_constant", "curry_retire_constant", + "curry_declare_function", "curry_retire_function", "curry_call_function", +} +_CURRY_READ_TOOLS = { + "curry_get_constant", "curry_get_constant_latest", "curry_list_constants", + "curry_get_function", "curry_list_functions", +} + + +def test_all_curry_tools_registered(): + for name in _CURRY_ACTION_TOOLS | _CURRY_READ_TOOLS: + assert name in tools.REGISTRY + + +def test_curry_write_and_execute_tools_are_gated_actions(): + for name in _CURRY_ACTION_TOOLS: + assert tools.is_action(name), name + assert name in tools.ALWAYS_ASK_ACTION_TOOLS, name + + +def test_curry_read_tools_are_not_actions(): + for name in _CURRY_READ_TOOLS: + assert not tools.is_action(name), name + + +# --------------------------------------------------------------------------- +# End-to-end HTTP: direct dispatch, no model call, no approval round-trip +# --------------------------------------------------------------------------- + +@pytest.fixture +def client(): + return TestClient(app) + + +def _sse_events(body: str) -> list[tuple[str, str]]: + events = [] + event_type = "message" + for block in body.split("\n\n"): + for line in block.splitlines(): + if line.startswith("event: "): + event_type = line[len("event: "):].strip() + elif line.startswith("data: "): + events.append((event_type, line[len("data: "):])) + event_type = "message" + return events + + +def test_slash_command_dispatches_without_model_call(client, monkeypatch): + from synapse import chat as chatmod + + async def _boom(*a, **k): + raise AssertionError("the model must not be called for a slash-command") + monkeypatch.setattr(chatmod, "stream_chat_response", _boom) + + resp = client.post("/chat/stream", json={ + "message": '/curry_list_functions()', + "conversation_id": "test-slash-http-1", + }) + events = _sse_events(resp.text) + assert ("status", json.dumps({"tool": "curry_list_functions"})) in events + assert any(t == "done" for t, _ in events) + + +def test_slash_command_skips_approval_round_trip(client, monkeypatch): + async def _fake_dispatch(name, args): + return json.dumps({"ok": True, "result": "did it"}) + monkeypatch.setattr(tools, "dispatch", _fake_dispatch) + + resp = client.post("/chat/stream", json={ + "message": '/curry_call_function(name="x", version=1, args={})', + "conversation_id": "test-slash-http-2", + }) + events = _sse_events(resp.text) + assert not any(t == "tool_request" for t, _ in events) + assert any(t == "done" for t, _ in events) + + +def test_slash_command_uses_fence_from_result_when_present(client, monkeypatch): + async def _fake_dispatch(name, args): + return json.dumps({"ok": True, "fence": "```nexus-curry\n{\"kind\": \"x\"}\n```"}) + monkeypatch.setattr(tools, "dispatch", _fake_dispatch) + + resp = client.post("/chat/stream", json={ + "message": '/curry_call_function(name="x", version=1, args={})', + "conversation_id": "test-slash-http-3", + }) + events = _sse_events(resp.text) + content = [d for t, d in events if t == "message"] + assert content and "nexus-curry" in content[0] + + +def test_slash_command_unknown_tool_yields_error_not_a_chat_reply(client): + resp = client.post("/chat/stream", json={ + "message": "/not_a_real_tool(a=1)", + "conversation_id": "test-slash-http-4", + }) + events = _sse_events(resp.text) + assert any(t == "error" for t, _ in events) + assert not any(t == "status" for t, _ in events) + + +def test_slash_command_malformed_yields_error(client): + resp = client.post("/chat/stream", json={ + "message": "/curry_call_function(x=some_name)", + "conversation_id": "test-slash-http-5", + }) + events = _sse_events(resp.text) + assert any(t == "error" for t, _ in events) + + +def test_message_with_leading_slash_but_not_command_shaped_goes_to_chat(client, monkeypatch): + # e.g. "/model gpt" or plain prose starting with "/" - must still reach + # the normal model path, not be swallowed as a broken slash-command. + called = {} + + async def _fake_stream(*a, **k): + called["hit"] = True + return + yield # pragma: no cover - make this an async generator + + # main.py did `from .chat import stream_chat_response`, a separate name + # binding from chat.stream_chat_response - patch the one main.py actually + # calls. + from synapse import main as mainmod + monkeypatch.setattr(mainmod, "stream_chat_response", _fake_stream) + + client.post("/chat/stream", json={ + "message": "/model gpt", + "conversation_id": "test-slash-http-6", + }) + assert called.get("hit") is True diff --git a/tests/test_tui.py b/tests/test_tui.py index 9d11f32..d235e71 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -301,3 +301,86 @@ def test_interrupt_cancels_silent_stream_and_accepts_next_message(monkeypatch): def test_escape_round_trip_helper(): assert "[" in _escape("x[y]") or "\\[" in _escape("x[y]") + + +def test_slash_tool_call_shape_forwards_to_start_chat(monkeypatch): + """/tool_name(arg=val) isn't a local meta-command — it must reach the + backend (synapse/slash_commands.py + chat_stream_endpoint dispatch it), + not fall into the generic 'unknown command' branch.""" + pytest.importorskip("textual") + from nexusos_cli.tui_app import NexusTUI + + app = NexusTUI.build_app(api_url="http://127.0.0.1:9") + calls: list[str] = [] + monkeypatch.setattr(app, "_start_chat", lambda text: calls.append(text)) + + async def _run(): + async with app.run_test(): + text = '/curry_call_function(name="double", version=1, args={"x": 21})' + app._handle_slash(text) + assert calls == [text] + log = app.query_one("#log") + assert not any("unknown command" in line.text for line in log.lines) + + asyncio.run(_run()) + + +def test_slash_malformed_tool_call_still_forwards_for_the_backend_error(monkeypatch): + """Even a malformed /tool(...) is forwarded rather than swallowed locally + — the backend's parser gives a clearer, more specific error than the + TUI's generic 'unknown command' would.""" + pytest.importorskip("textual") + from nexusos_cli.tui_app import NexusTUI + + app = NexusTUI.build_app(api_url="http://127.0.0.1:9") + calls: list[str] = [] + monkeypatch.setattr(app, "_start_chat", lambda text: calls.append(text)) + + async def _run(): + async with app.run_test(): + text = "/curry_call_function(x=__import__('os'))" + app._handle_slash(text) + assert calls == [text] + + asyncio.run(_run()) + + +def test_slash_local_meta_commands_still_handled_locally(monkeypatch): + """A known local command must still be handled in-TUI, never forwarded — + the new tool-call passthrough is strictly the fallback branch.""" + pytest.importorskip("textual") + from nexusos_cli.tui_app import NexusTUI + + app = NexusTUI.build_app(api_url="http://127.0.0.1:9") + calls: list[str] = [] + monkeypatch.setattr(app, "_start_chat", lambda text: calls.append(text)) + + async def _run(): + async with app.run_test(): + app._handle_slash("/help") + assert calls == [] + log = app.query_one("#log") + assert any("this list" in line.text for line in log.lines) + + asyncio.run(_run()) + + +def test_slash_unknown_bare_command_still_rejected(monkeypatch): + """A genuinely unknown command (no parens, not a local command) keeps the + existing 'unknown command' behavior rather than silently forwarding + anything that starts with /.""" + pytest.importorskip("textual") + from nexusos_cli.tui_app import NexusTUI + + app = NexusTUI.build_app(api_url="http://127.0.0.1:9") + calls: list[str] = [] + monkeypatch.setattr(app, "_start_chat", lambda text: calls.append(text)) + + async def _run(): + async with app.run_test(): + app._handle_slash("/frobnicate") + assert calls == [] + log = app.query_one("#log") + assert any("unknown command" in line.text for line in log.lines) + + asyncio.run(_run()) -- 2.39.5 From 841464f9b107b4401aca7e37ff60aa21c618cfcd Mon Sep 17 00:00:00 2001 From: Athena Kaminsky Date: Tue, 25 Aug 2026 22:11:37 -0500 Subject: [PATCH 3/5] docs: update CLAUDE.md for curry tool wiring + slash commands Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/CLAUDE.md b/CLAUDE.md index eace049..177f469 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -127,7 +127,10 @@ React 19 + Vite. No routing library — `App.jsx` manages page state in a single 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. +`curry_core.py` is vendored from [Athena-Pro/Curry](https://github.com/Athena-Pro/Curry), with two deliberate deviations from upstream documented in the file's own docstring (a sandbox-escape fix and a `check_same_thread=False` connection fix) — 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 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. Ten `curry_*` tools in `tools.py` expose it to chat (`curry_declare_constant`, `curry_call_function`, etc.); the five that write or execute are ACTION tools in `ALWAYS_ASK_ACTION_TOOLS`, same approval floor as `edit_source`. Re-sync `curry_core.py` from upstream by hand, not by script. + +### Direct tool invocation (`synapse/slash_commands.py`) +A chat message that's nothing but `/tool_name(arg=val, ...)` (Python-call-shaped, arguments parsed via `ast.literal_eval` only — no names, no calls, no attribute access) dispatches straight through `tools.dispatch()`, skipping model selection, context assembly, and the ask-policy approval round-trip. A human typing it is the approval. Wired into `chat_stream_endpoint` as an early short-circuit; the TUI's `_handle_slash` falls through to the backend for anything shaped like a tool call that isn't one of its own local meta-commands (`/help`, `/model`, `/new`). ### Logs & Runtime State - `runtime/backend.log`, `runtime/frontend.log`, `runtime/memory.log` — service stdout -- 2.39.5 From 13bffc41e97d044c20d8b97b8e686e617ef99c6e Mon Sep 17 00:00:00 2001 From: Athena Kaminsky Date: Wed, 26 Aug 2026 08:25:30 -0500 Subject: [PATCH 4/5] chore(curry): ignore the runtime ledger --- .gitignore | 1 + 1 file changed, 1 insertion(+) diff --git a/.gitignore b/.gitignore index b88e465..689b5d1 100644 --- a/.gitignore +++ b/.gitignore @@ -13,6 +13,7 @@ synapse/memory/memory.db synapse/memory/memory.db-wal synapse/memory/memory.db-shm assets/gitnexus-logo.svg +/data/curry.db *.db-wal *.db-shm .DS_Store -- 2.39.5 From a9ff8c0c24684943ad639ba11e4375e82d26715a Mon Sep 17 00:00:00 2001 From: Athena Kaminsky Date: Wed, 26 Aug 2026 08:27:52 -0500 Subject: [PATCH 5/5] test(curry): keep preload check repeatable --- tests/test_curry_store.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/tests/test_curry_store.py b/tests/test_curry_store.py index a1d5e6f..c142555 100644 --- a/tests/test_curry_store.py +++ b/tests/test_curry_store.py @@ -1,8 +1,8 @@ """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. +https://github.com/Athena-Pro/Curry/pull/4, and curry_store gives NexusOS a +live instance for the registered chat tools. """ import pytest @@ -10,14 +10,12 @@ from synapse.curry_core import Curry, TypeSignature from synapse import curry_store -def test_curry_store_is_preloaded_and_callable(): +def test_curry_store_is_preloaded_and_open(): # 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) + assert curry_store.curry_db.conn.execute("SELECT 1").fetchone()[0] == 1 def test_curry_db_path_matches_nexus_config(tmp_path, monkeypatch): -- 2.39.5