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 diff --git a/CLAUDE.md b/CLAUDE.md index 6bffd47..177f469 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -126,6 +126,12 @@ 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 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 - `runtime/logs/ollama.log`, `runtime/logs/chat.log` diff --git a/bin/check.sh b/bin/check.sh index a8dbbe4..21ecf4b 100644 --- a/bin/check.sh +++ b/bin/check.sh @@ -63,6 +63,10 @@ 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)") +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 new file mode 100644 index 0000000..c101af2 --- /dev/null +++ b/synapse/curry_core.py @@ -0,0 +1,1720 @@ +""" +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. + +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: + +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 +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 + # 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() + + 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..edb3fdf 100644 --- a/synapse/main.py +++ b/synapse/main.py @@ -182,7 +182,9 @@ 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 +from . import slash_commands as _slash_commands MEMORY_SERVICE = settings.memory_url @@ -460,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. @@ -469,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)) @@ -483,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/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/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_curry_store.py b/tests/test_curry_store.py new file mode 100644 index 0000000..c142555 --- /dev/null +++ b/tests/test_curry_store.py @@ -0,0 +1,53 @@ +"""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 gives NexusOS a +live instance for the registered chat tools. +""" +import pytest + +from synapse.curry_core import Curry, TypeSignature +from synapse import curry_store + + +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) + assert curry_store.curry_db.conn.execute("SELECT 1").fetchone()[0] == 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() 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())