"""Tool execution boundary used by the Nexum CLI and server.""" from __future__ import annotations import ast import fcntl import glob as globlib import hashlib import hmac import http.client import io import ipaddress import json import os import re import shlex import socket import ssl # Required only at the isolated execution boundary. import subprocess # nosec B404 import time import tokenize import urllib.parse from dataclasses import replace from pathlib import Path from typing import Any, cast from .tooling.artifacts import ArtifactStore from .tooling.browser import ( BROWSER_TOOL_NAMES, BROWSER_TOOL_SPECS, execute_browser_tool, ) from .tooling.contracts import ( ToolCall, ToolExecutionContext, ToolExecutionResult, ToolParameter, ToolSpec, ) from .tooling.delegation import DELEGATED_AGENT_TOOL_SPECS from .tooling.drafting import ( DRAFTING_TOOL_NAMES, DRAFTING_TOOL_SPECS, execute_drafting_tool, ) from .tooling.engineering import ( ENGINEERING_TOOL_NAMES, ENGINEERING_TOOL_SPECS, execute_engineering_tool, ) from .tooling.events import EventLog from .tooling.idempotency import IdempotencyStore from .tooling.language_packs import analyze_language_file, language_pack_catalog from .tooling.repository import ( REPOSITORY_TOOLS, execute_repository_tool, repository_tool_names, ) from .tooling.sandbox import ( control_root, sandbox_argv, sandbox_environment, ) from .tooling.scheduler import execute_call_batch from .tooling.security import ( ApprovalStore, SecretRedactor, ToolPolicy, redact_sensitive_value, ) from .tooling.tasks import TaskStore from .tooling.transactions import ( TRANSACTION_TOOL_NAMES, TRANSACTION_TOOL_SPECS, TransactionStore, ) TOOL_CALL_START = "<|tool_call_start|>" TOOL_CALL_END = "<|tool_call_end|>" _DYNAMIC_TOOL_SCHEMA = "nexum.dynamic-tool.v2" MARKED_CALL_RE = re.compile( rf"{re.escape(TOOL_CALL_START)}(?P.*?){re.escape(TOOL_CALL_END)}", re.DOTALL, ) CORE_TOOLS: tuple[ToolSpec, ...] = ( ToolSpec( "Bash", "terminal", "Run a shell command in the selected workspace and return stdout, stderr, and exit status.", "Bash(command='pwd')", (ToolParameter("command", "string", "Shell command to execute."),), risk="workspace_write", parallel_safe=False, idempotent=False, task_support="optional", ), ToolSpec( "Read", "filesystem", "Read a UTF-8 text file contained in the selected workspace.", "Read(path='README.md')", (ToolParameter("path", "string", "Workspace-relative file path."),), ), ToolSpec( "Write", "filesystem", "Write a UTF-8 text file contained in the selected workspace.", "Write(path='out.txt', content='ok')", ( ToolParameter("path", "string", "Workspace-relative file path."), ToolParameter("content", "string", "Complete file content."), ), risk="workspace_write", parallel_safe=False, ), ToolSpec( "Edit", "filesystem", "Replace one exact text occurrence in a workspace file.", "Edit(path='a.txt', old_string='x', new_string='y')", ( ToolParameter("path", "string", "Workspace-relative file path."), ToolParameter("old_string", "string", "Exact text to replace."), ToolParameter("new_string", "string", "Replacement text."), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "Glob", "filesystem", "Find workspace files by a recursive glob pattern.", "Glob(pattern='**/*.py')", (ToolParameter("pattern", "string", "Workspace-relative glob pattern."),), ), ToolSpec( "Grep", "filesystem", "Search workspace file contents and return matching lines with locations.", "Grep(pattern='error', path='.')", ( ToolParameter( "pattern", "string", "Text or regular expression to search for." ), ToolParameter( "path", "string", "Workspace-relative search root.", required=False ), ), ), ToolSpec( "WebFetch", "web", "Fetch an HTTP or HTTPS resource and return its response body.", "WebFetch(url='https://example.com')", (ToolParameter("url", "string", "HTTP or HTTPS URL."),), source_trust="untrusted_content", ), ToolSpec( "WebSearch", "web", "Search the public web and return the result page for evidence gathering.", "WebSearch(query='python release notes')", (ToolParameter("query", "string", "Search query."),), source_trust="untrusted_content", ), ToolSpec( "ToolCatalog", "orchestration", "List available local tools, optionally filtered by a query.", "ToolCatalog(query='file')", (ToolParameter("query", "string", "Optional catalog filter.", required=False),), ), ToolSpec( "ToolDescribe", "orchestration", "Return the exact schema and execution annotations for one available tool.", "ToolDescribe(name='Read')", (ToolParameter("name", "string", "Exact tool name."),), ), ToolSpec( "LanguagePacks", "orchestration", "Discover source-intelligence packs and native language toolchains in the current runtime.", "LanguagePacks(query='python')", ( ToolParameter( "query", "string", "Optional language name, alias, identifier, or file extension.", required=False, ), ), ), ToolSpec( "LanguageInspect", "engineering", "Inspect one workspace source file with its real language parser or declared lexical backend.", "LanguageInspect(path='src/main.py', language='python')", ( ToolParameter("path", "string", "Workspace-relative source file."), ToolParameter( "language", "string", "Optional language name, alias, or pack identifier.", required=False, ), ), ), ToolSpec( "RequestInput", "interaction", "Pause the open task and request structured information from the caller.", "RequestInput(prompt='Choose a deployment region', schema={})", ( ToolParameter("prompt", "string", "Question presented to the caller."), ToolParameter( "schema", "object", "JSON Schema describing the requested response.", required=False, ), ), parallel_safe=False, idempotent=False, ), ToolSpec( "TaskStart", "terminal", "Start a durable terminal task that continues until the command exits or is cancelled.", "TaskStart(command='python -m http.server')", (ToolParameter("command", "string", "Shell command to run."),), risk="workspace_write", parallel_safe=False, idempotent=False, task_support="required", ), ToolSpec( "TaskStatus", "orchestration", "Read the durable status and result of a long-running task.", "TaskStatus(task_id='task_...')", (ToolParameter("task_id", "string", "Durable task identifier."),), ), ToolSpec( "TaskCancel", "orchestration", "Cancel one running task while preserving its execution record.", "TaskCancel(task_id='task_...')", (ToolParameter("task_id", "string", "Durable task identifier."),), risk="destructive", parallel_safe=False, idempotent=True, ), ToolSpec( "ArtifactList", "artifacts", "List content-addressed artifacts created in the selected workspace.", "ArtifactList()", ), ToolSpec( "ArtifactRead", "artifacts", "Read a model-selected range from a content-addressed artifact.", "ArtifactRead(artifact_id='art_...', offset=0)", ( ToolParameter("artifact_id", "string", "Content-addressed artifact identifier."), ToolParameter("offset", "integer", "Byte offset.", required=False), ToolParameter("length", "integer", "Number of bytes.", required=False), ), ), ToolSpec( "CreateTool", "orchestration", "Register a new reusable workspace-local command tool without running its command.", "CreateTool(name='disk_usage', command='df -h')", ( ToolParameter("name", "string", "Local tool name."), ToolParameter("command", "string", "Command implemented by the tool."), ToolParameter( "description", "string", "Purpose of the tool.", required=False ), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "UpgradeTool", "orchestration", "Create a validated generation of an existing workspace-local tool without running its command.", "UpgradeTool(name='disk_usage', command='df -hT', expected_sha256='...')", ( ToolParameter("name", "string", "Existing local tool name."), ToolParameter( "command", "string", "Updated command implemented by the tool." ), ToolParameter("description", "string", "Updated purpose.", required=False), ToolParameter( "expected_sha256", "string", "Exact current definition digest returned by ToolDescribe.", ), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "RetireTool", "orchestration", "Retire a workspace-local tool while retaining its complete version history.", "RetireTool(name='disk_usage', expected_sha256='...')", ( ToolParameter("name", "string", "Existing local tool name."), ToolParameter( "expected_sha256", "string", "Exact current definition digest returned by ToolDescribe.", ), ), risk="workspace_write", parallel_safe=False, idempotent=False, ), ToolSpec( "RunDynamicTool", "orchestration", "Run a workspace-local tool created earlier in the same workspace.", "RunDynamicTool(name='disk_usage')", ( ToolParameter("name", "string", "Local tool name."), ToolParameter( "args", "string", "Optional argument string.", required=False ), ), risk="workspace_write", parallel_safe=False, idempotent=False, task_support="optional", ), ) + DRAFTING_TOOL_SPECS + ENGINEERING_TOOL_SPECS + REPOSITORY_TOOLS + TRANSACTION_TOOL_SPECS + BROWSER_TOOL_SPECS ADVERTISED_TOOLS: tuple[ToolSpec, ...] = CORE_TOOLS + DELEGATED_AGENT_TOOL_SPECS def list_tools() -> list[dict[str, Any]]: return [tool.to_dict() for tool in ADVERTISED_TOOLS] def tool_schemas() -> list[dict[str, Any]]: return [tool.openai_schema() for tool in ADVERTISED_TOOLS] def runtime_tool_schemas() -> list[dict[str, Any]]: """Return only tools implemented by this local runtime process.""" return [tool.openai_schema() for tool in CORE_TOOLS] def tool_names() -> set[str]: return {tool.name for tool in CORE_TOOLS} def tool_spec(name: str) -> ToolSpec | None: return next((tool for tool in CORE_TOOLS if tool.name == name), None) def advertised_tool_spec(name: str) -> ToolSpec | None: return next((tool for tool in ADVERTISED_TOOLS if tool.name == name), None) def _state_root(cwd: str | Path | None = None) -> Path: return control_root(cwd or os.getcwd()) def _dynamic_tool_dir(cwd: str | Path | None = None) -> Path: return _state_root(cwd) / "tools" def _dynamic_current_dir(cwd: str | Path | None = None) -> Path: return _dynamic_tool_dir(cwd) / "current" def _dynamic_history_dir(cwd: str | Path | None = None) -> Path: return _dynamic_tool_dir(cwd) / "history" def _dynamic_name(name: str) -> str: if not name or re.fullmatch(r"[A-Za-z0-9_]+", name) is None: raise ValueError( "dynamic tool name must contain only letters, numbers, and underscores" ) return name def _dynamic_definition_sha256(payload: dict[str, Any]) -> str: canonical = dict(payload) canonical.pop("definition_sha256", None) return hashlib.sha256( json.dumps( canonical, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") ).hexdigest() def _dynamic_record( *, name: str, command: str, description: str, status: str, generation: int, previous_sha256: str, ) -> dict[str, Any]: payload: dict[str, Any] = { "schema": _DYNAMIC_TOOL_SCHEMA, "name": name, "command": command, "description": description, "status": status, "generation": generation, "previous_sha256": previous_sha256, } payload["definition_sha256"] = _dynamic_definition_sha256(payload) return payload def _validate_dynamic_record( payload: Any, *, expected_name: str, ) -> dict[str, Any]: if not isinstance(payload, dict): raise ValueError("dynamic tool definition must be an object") if payload.get("schema") != _DYNAMIC_TOOL_SCHEMA: raise ValueError("dynamic tool definition schema is unsupported") name = str(payload.get("name") or "") command = str(payload.get("command") or "") description = str(payload.get("description") or "") status = str(payload.get("status") or "") generation = payload.get("generation") previous_sha256 = str(payload.get("previous_sha256") or "") supplied_sha256 = str(payload.get("definition_sha256") or "") if name != expected_name or _dynamic_name(name) != _dynamic_name(expected_name): raise ValueError("dynamic tool definition name changed") if not command: raise ValueError("dynamic tool command is empty") if status not in {"active", "retired"}: raise ValueError("dynamic tool status is invalid") if isinstance(generation, bool) or not isinstance(generation, int) or generation <= 0: raise ValueError("dynamic tool generation is invalid") for digest in (previous_sha256, supplied_sha256): if digest and ( len(digest) != 64 or any(value not in "0123456789abcdef" for value in digest) ): raise ValueError("dynamic tool definition digest is invalid") normalized = _dynamic_record( name=name, command=command, description=description, status=status, generation=generation, previous_sha256=previous_sha256, ) if not hmac.compare_digest( supplied_sha256, str(normalized["definition_sha256"]), ): raise ValueError("dynamic tool definition integrity check failed") return normalized def _atomic_json_write(path: Path, payload: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) with temporary.open("rb") as handle: os.fsync(handle.fileno()) os.replace(temporary, path) directory_fd = os.open(path.parent, os.O_RDONLY) try: os.fsync(directory_fd) finally: os.close(directory_fd) def _dynamic_path(name: str, cwd: str) -> Path: return _dynamic_current_dir(cwd) / f"{_dynamic_name(name)}.json" def _legacy_dynamic_path(name: str, cwd: str) -> Path: return _dynamic_tool_dir(cwd) / f"{_dynamic_name(name)}.json" def _dynamic_history_path(name: str, generation: int, cwd: str) -> Path: return ( _dynamic_history_dir(cwd) / _dynamic_name(name) / f"{generation:06d}.json" ) def _dynamic_lock_path(cwd: str) -> Path: return _dynamic_tool_dir(cwd) / "lifecycle.lock" def _archive_dynamic_record(record: dict[str, Any], cwd: str) -> None: path = _dynamic_history_path( str(record["name"]), int(record["generation"]), cwd, ) if path.is_file(): observed = _validate_dynamic_record( json.loads(path.read_text(encoding="utf-8")), expected_name=str(record["name"]), ) if not hmac.compare_digest( str(observed["definition_sha256"]), str(record["definition_sha256"]), ): raise RuntimeError("dynamic tool history conflicts with current state") return _atomic_json_write(path, record) def _load_dynamic_record(name: str, cwd: str) -> dict[str, Any]: path = _dynamic_path(name, cwd) if not path.is_file(): legacy = _legacy_dynamic_path(name, cwd) if not legacy.is_file(): raise FileNotFoundError(f"dynamic tool not found: {name}") payload = json.loads(legacy.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise ValueError("legacy dynamic tool definition must be an object") record = _dynamic_record( name=str(payload.get("name") or name), command=str(payload.get("command") or ""), description=str(payload.get("description") or ""), status="active", generation=1, previous_sha256="", ) record = _validate_dynamic_record(record, expected_name=name) _atomic_json_write(path, record) _archive_dynamic_record(record, cwd) legacy.unlink() return record return _validate_dynamic_record( json.loads(path.read_text(encoding="utf-8")), expected_name=name, ) def _dynamic_catalog_row(record: dict[str, Any]) -> dict[str, Any]: return { "name": str(record["name"]), "namespace": "workspace", "surface": "dynamic", "description": str(record["description"]), "invocation": "RunDynamicTool", "status": str(record["status"]), "generation": int(record["generation"]), "definition_sha256": str(record["definition_sha256"]), } def _dynamic_tool_catalog(cwd: str | Path | None = None) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] workspace = str(Path(cwd or os.getcwd()).expanduser().resolve()) root = _dynamic_current_dir(workspace) legacy_root = _dynamic_tool_dir(workspace) if not root.is_dir() and not legacy_root.is_dir(): return rows lock_path = _dynamic_lock_path(workspace) lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with lock_path.open("a+b") as lock_handle: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) try: names = { path.stem for directory in (root, legacy_root) for path in directory.glob("*.json") } for name in sorted(names): record = _load_dynamic_record(name, workspace) if record["status"] == "active": rows.append(_dynamic_catalog_row(record)) finally: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) return rows def _tool_call_from_node(node: ast.AST, raw: str) -> ToolCall: if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Name): raise ValueError("tool entries must be direct function calls") if node.args: raise ValueError("positional tool arguments are not supported") out: dict[str, Any] = {} dependencies: tuple[str, ...] = () for keyword in node.keywords: if keyword.arg is None: raise ValueError("expanded tool arguments are not supported") value = ast.literal_eval(keyword.value) if keyword.arg == "depends_on": if not isinstance(value, list) or any( not isinstance(item, str) for item in value ): raise ValueError("depends_on must be an array of strings") dependencies = tuple(item for item in value if item) continue out[keyword.arg] = value return ToolCall(name=node.func.id, args=out, raw=raw, depends_on=dependencies) def _without_single_spurious_closer(source: str) -> str | None: """Remove one unmatched closer only when the remaining delimiters are exact.""" opener_for = {")": "(", "]": "[", "}": "{"} openers = set(opener_for.values()) stack: list[str] = [] spurious_index: int | None = None try: tokens = list(tokenize.generate_tokens(io.StringIO(source).readline)) except tokenize.TokenError: return None for index, token in enumerate(tokens): if token.type != tokenize.OP: continue if token.string in openers: stack.append(token.string) continue expected = opener_for.get(token.string) if expected is None: continue if stack and stack[-1] == expected: stack.pop() continue if spurious_index is not None: return None spurious_index = index if spurious_index is None or stack: return None return cast( str, tokenize.untokenize( token for index, token in enumerate(tokens) if index != spurious_index ), ) def _parse_call_expression(source: str, raw: str) -> list[ToolCall]: try: node = ast.parse(source.strip(), mode="eval").body except SyntaxError as exc: normalized = _without_single_spurious_closer(source) if normalized is None: raise ValueError(f"invalid tool call syntax: {exc}") from exc try: node = ast.parse(normalized.strip(), mode="eval").body except SyntaxError: raise ValueError(f"invalid tool call syntax: {exc}") from exc entries = node.elts if isinstance(node, (ast.List, ast.Tuple)) else [node] return [_tool_call_from_node(entry, raw) for entry in entries] def _tool_call_from_json(entry: Any, raw: str) -> ToolCall: if not isinstance(entry, dict): raise ValueError("JSON tool call must be an object") function = entry.get("function", entry) if not isinstance(function, dict): raise ValueError("JSON tool call function must be an object") name = function.get("name") if not isinstance(name, str) or not name.strip(): raise ValueError("JSON tool call is missing a function name") arguments = function.get("arguments", function.get("parameters", {})) if isinstance(arguments, str): try: arguments = json.loads(arguments or "{}") except json.JSONDecodeError as exc: raise ValueError("JSON tool arguments are not valid JSON") from exc if not isinstance(arguments, dict): raise ValueError("JSON tool arguments must be an object") args = dict(arguments) dependencies_value = args.pop( "depends_on", function.get("depends_on", entry.get("depends_on", [])), ) if dependencies_value is None: dependencies_value = [] if not isinstance(dependencies_value, list) or any( not isinstance(item, str) for item in dependencies_value ): raise ValueError("depends_on must be an array of strings") call_id = entry.get("id", "") if not isinstance(call_id, str): raise ValueError("JSON tool call id must be a string") return ToolCall( name=name.strip(), args=args, raw=raw, call_id=call_id, depends_on=tuple(item for item in dependencies_value if item), ) def _parse_json_tool_calls(source: str, raw: str) -> list[ToolCall]: try: decoded = json.loads(source.strip()) except json.JSONDecodeError as exc: raise ValueError(f"invalid JSON tool call syntax: {exc}") from exc entries = decoded if isinstance(decoded, list) else [decoded] if not entries: raise ValueError("JSON tool call array is empty") return [_tool_call_from_json(entry, raw) for entry in entries] def _parse_tool_call_payload(source: str, raw: str) -> list[ToolCall]: try: return _parse_call_expression(source, raw) except ValueError as expression_error: try: return _parse_json_tool_calls(source, raw) except ValueError: raise expression_error def parse_tool_calls(text: str) -> list[ToolCall]: calls: list[ToolCall] = [] marked = list(MARKED_CALL_RE.finditer(text)) for match in marked: calls.extend(_parse_tool_call_payload(match.group("body"), match.group(0))) if marked: return calls candidate = text.strip() if not candidate: return [] try: return _parse_tool_call_payload(candidate, candidate) except ValueError: return [] def _string_arg(args: dict[str, Any], key: str, default: str = "") -> str: value = args.get(key, default) if value is None: return default return str(value) def _string_tuple_arg(args: dict[str, Any], key: str) -> tuple[str, ...]: value = args.get(key) if not isinstance(value, list) or any(not isinstance(item, str) for item in value): raise ValueError(f"{key} must be an array of strings") return tuple(value) def _result( call: ToolCall, *, ok: bool, output: str = "", error: str = "", stdout: str = "", stderr: str = "", exit_code: int | None = None, executed: bool = False, started: float, ) -> ToolExecutionResult: spec = tool_spec(call.name) rendered = output or stdout or stderr return ToolExecutionResult( name=call.name, args=call.args, ok=ok, tool_call_id=call.call_id, output=output, error=error, stdout=stdout, stderr=stderr, exit_code=exit_code, executed=executed, elapsed_s=round(time.perf_counter() - started, 4), source_trust=( spec.source_trust if spec is not None else "trusted_execution" ), output_sha256=hashlib.sha256(rendered.encode("utf-8")).hexdigest(), ) def _run_command( command: str, cwd: str, timeout_s: float ) -> tuple[bool, str, str, int | None]: workspace = Path(cwd).expanduser().resolve() if not workspace.is_dir(): return False, "", "workspace directory does not exist", None # The executable is fixed and the requested command is a positional sandbox input. proc = subprocess.run( # nosec B603 sandbox_argv(workspace, command), cwd=str(workspace), env=sandbox_environment(), text=True, capture_output=True, timeout=timeout_s if timeout_s > 0 else None, ) return proc.returncode == 0, proc.stdout, proc.stderr, int(proc.returncode) def _workspace_path(path: str, cwd: str) -> Path: target = Path(path).expanduser() base = Path(cwd).expanduser().resolve() if target.is_absolute(): target = target.resolve() else: target = (base / target).resolve() try: relative = target.relative_to(base) except ValueError as exc: raise ValueError( f"path leaves workspace: {path}; use a workspace-relative path" ) from exc if relative.parts and relative.parts[0] == ".nexum": raise ValueError("workspace control paths require dedicated runtime tools") return target def _workspace_relative_path(path: Path, cwd: str) -> str: base = Path(cwd).expanduser().resolve() return path.resolve().relative_to(base).as_posix() def _control_relative_path(path: Path, cwd: str) -> str: relative = path.resolve().relative_to(_state_root(cwd)) return (Path(".nexum") / relative).as_posix() def _workspace_glob_pattern(pattern: str, cwd: str) -> str: base = Path(cwd).expanduser().resolve() raw = Path(pattern).expanduser() lexical_parts = tuple(part for part in raw.parts if part not in {"", "."}) if lexical_parts and lexical_parts[0] == ".nexum": raise ValueError("workspace control paths require dedicated runtime tools") if raw.is_absolute(): prefix_parts: list[str] = [] for part in raw.parts: if any(marker in part for marker in "*?["): break prefix_parts.append(part) prefix = Path(*prefix_parts).resolve() try: prefix.relative_to(base) except ValueError as exc: raise ValueError("glob pattern leaves workspace") from exc return str(raw) candidate = base / raw prefix = base for part in raw.parts: if any(marker in part for marker in "*?["): break prefix = prefix / part try: prefix.resolve().relative_to(base) except ValueError as exc: raise ValueError("glob pattern leaves workspace") from exc return str(candidate) def _read(path: str, cwd: str) -> tuple[bool, str, str]: target = _workspace_path(path, cwd) if not target.exists() or not target.is_file(): return False, "", f"file not found: {path}" return True, target.read_text(encoding="utf-8", errors="replace"), "" def _write(path: str, content: str, cwd: str) -> tuple[bool, str, str]: target = _workspace_path(path, cwd) target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") return True, _workspace_relative_path(target, cwd), "" def _edit(path: str, old: str, new: str, cwd: str) -> tuple[bool, str, str]: target = _workspace_path(path, cwd) if not target.exists() or not target.is_file(): return False, "", f"file not found: {path}" text = target.read_text(encoding="utf-8", errors="replace") if old not in text: return False, "", "old_string not found" target.write_text(text.replace(old, new, 1), encoding="utf-8") return True, _workspace_relative_path(target, cwd), "" def _validated_public_target( url: str, ) -> tuple[urllib.parse.SplitResult, tuple[str, ...]]: parsed = urllib.parse.urlsplit(url) if parsed.scheme.lower() not in {"http", "https"}: raise ValueError("WebFetch accepts only HTTP and HTTPS URLs") if parsed.username is not None or parsed.password is not None: raise ValueError("WebFetch URL credentials are not allowed") hostname = parsed.hostname if not hostname: raise ValueError("WebFetch URL must include a hostname") try: addresses = {ipaddress.ip_address(hostname)} except ValueError: try: rows = socket.getaddrinfo( hostname, parsed.port or (443 if parsed.scheme.lower() == "https" else 80), type=socket.SOCK_STREAM, ) except socket.gaierror as exc: raise ValueError("WebFetch hostname could not be resolved") from exc addresses = {ipaddress.ip_address(row[4][0]) for row in rows} if not addresses or any(not address.is_global for address in addresses): raise ValueError("WebFetch target must resolve only to public addresses") return parsed, tuple(sorted(str(address) for address in addresses)) class _PinnedHTTPSConnection(http.client.HTTPSConnection): def __init__( self, connect_address: str, server_hostname: str, port: int, timeout: float, ) -> None: context = ssl.create_default_context() super().__init__( server_hostname, port=port, timeout=timeout, context=context, ) self._connect_address = connect_address self._nexum_timeout = timeout self._nexum_context = context def connect(self) -> None: raw_socket = socket.create_connection( (self._connect_address, self.port), self._nexum_timeout ) self.sock = self._nexum_context.wrap_socket( raw_socket, server_hostname=self.host ) def _public_http_response( parsed: urllib.parse.SplitResult, addresses: tuple[str, ...], timeout_s: float, ) -> tuple[http.client.HTTPConnection, http.client.HTTPResponse]: hostname = parsed.hostname if hostname is None: raise ValueError("WebFetch URL must include a hostname") port = parsed.port or (443 if parsed.scheme.lower() == "https" else 80) path = parsed.path or "/" if parsed.query: path += "?" + parsed.query display_host = hostname.encode("idna").decode("ascii") if ":" in display_host: display_host = f"[{display_host}]" default_port = 443 if parsed.scheme.lower() == "https" else 80 host_header = display_host if port == default_port else f"{display_host}:{port}" last_error: OSError | None = None for address in addresses: connection: http.client.HTTPConnection if parsed.scheme.lower() == "https": connection = _PinnedHTTPSConnection(address, hostname, port, timeout_s) else: connection = http.client.HTTPConnection( address, port=port, timeout=timeout_s ) try: connection.request( "GET", path, headers={"Host": host_header, "User-Agent": "nexum-runtime/0.1"}, ) response = connection.getresponse() peer = connection.sock.getpeername()[0] if connection.sock else address if not ipaddress.ip_address(peer).is_global: connection.close() raise ValueError("WebFetch connected peer is not public") return connection, response except OSError as exc: connection.close() last_error = exc raise ConnectionError( "WebFetch could not connect to a validated address" ) from last_error def _web_fetch(url: str, timeout_s: float) -> tuple[bool, str, str]: current = url seen: set[str] = set() timeout = timeout_s if timeout_s > 0 else 20.0 while True: if current in seen: raise ValueError("WebFetch redirect loop detected") if len(seen) >= 10: raise ValueError("WebFetch redirect chain is too long") seen.add(current) parsed, addresses = _validated_public_target(current) connection, response = _public_http_response(parsed, addresses, timeout) try: if response.status in {301, 302, 303, 307, 308}: location = response.getheader("Location") if not location: return False, "", "HTTP redirect did not include a location" current = urllib.parse.urljoin(current, location) continue data = response.read(1024 * 1024 + 1) if len(data) > 1024 * 1024: return False, "", "HTTP response exceeded one mebibyte" if not 200 <= response.status < 300: return False, "", f"HTTP status {response.status}" return True, data.decode("utf-8", errors="replace"), "" finally: connection.close() def _web_search(query: str, timeout_s: float) -> tuple[bool, str, str]: url = "https://html.duckduckgo.com/html/?q=" + urllib.parse.quote_plus(query) return _web_fetch(url, timeout_s) def _grep( pattern: str, path: str, cwd: str, timeout_s: float ) -> tuple[bool, str, str, int]: target = _workspace_path(path, cwd) relative_target = _workspace_relative_path(target, cwd) or "." # Fixed executable and positional arguments; no shell expansion. process = subprocess.run( # nosec B603 [ "/usr/bin/grep", "-R", "--line-number", "--exclude-dir=.nexum", "--", pattern, relative_target, ], cwd=cwd, text=True, capture_output=True, timeout=timeout_s if timeout_s > 0 else None, ) if process.returncode == 1: return True, "", "", 1 return ( process.returncode == 0, process.stdout, process.stderr, int(process.returncode), ) def _mutate_dynamic( args: dict[str, Any], cwd: str, *, operation: str, ) -> tuple[bool, str, str]: name = _string_arg(args, "name") _dynamic_name(name) path = _dynamic_path(name, cwd) legacy_path = _legacy_dynamic_path(name, cwd) lock_path = _dynamic_lock_path(cwd) lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with lock_path.open("a+b") as lock_handle: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) try: current: dict[str, Any] | None = None if path.is_file() or legacy_path.is_file(): current = _load_dynamic_record(name, cwd) _archive_dynamic_record(current, cwd) if operation == "create": if current is not None: return False, "", "dynamic tool already exists" command = _string_arg(args, "command") if not command: return False, "", "name and command are required" next_record = _dynamic_record( name=name, command=command, description=_string_arg(args, "description"), status="active", generation=1, previous_sha256="", ) else: if current is None: return False, "", f"dynamic tool not found: {name}" expected_sha256 = _string_arg(args, "expected_sha256") if not hmac.compare_digest( expected_sha256, str(current["definition_sha256"]), ): return False, "", "dynamic tool definition changed before mutation" if operation == "upgrade": command = _string_arg(args, "command") if not command: return False, "", "name and command are required" description = _string_arg( args, "description", str(current["description"]), ) if ( current["status"] == "active" and command == current["command"] and description == current["description"] ): return False, "", "dynamic tool upgrade is a no-op" status = "active" elif operation == "retire": if current["status"] == "retired": return False, "", "dynamic tool is already retired" command = str(current["command"]) description = str(current["description"]) status = "retired" else: raise ValueError("dynamic tool mutation is unsupported") next_record = _dynamic_record( name=name, command=command, description=description, status=status, generation=int(current["generation"]) + 1, previous_sha256=str(current["definition_sha256"]), ) next_history = _dynamic_history_path( name, int(next_record["generation"]), cwd, ) if next_history.is_file(): prior = _validate_dynamic_record( json.loads(next_history.read_text(encoding="utf-8")), expected_name=name, ) if not hmac.compare_digest( str(prior["definition_sha256"]), str(next_record["definition_sha256"]), ): raise RuntimeError( "dynamic tool history conflicts with proposed state" ) _atomic_json_write(path, next_record) _archive_dynamic_record(next_record, cwd) finally: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) return ( True, json.dumps( { "name": name, "path": _control_relative_path(path, cwd), "registered": next_record["status"] == "active", "retired": next_record["status"] == "retired", "generation": next_record["generation"], "definition_sha256": next_record["definition_sha256"], "previous_sha256": next_record["previous_sha256"], "dynamic_command_executed": False, }, sort_keys=True, ), "", ) def _run_dynamic( args: dict[str, Any], cwd: str, timeout_s: float ) -> tuple[bool, str, str, int | None, bool]: name = _string_arg(args, "name") lock_path = _dynamic_lock_path(cwd) lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with lock_path.open("a+b") as lock_handle: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) try: try: record = _load_dynamic_record(name, cwd) _archive_dynamic_record(record, cwd) except FileNotFoundError: return False, "", f"dynamic tool not found: {name}", None, False except (json.JSONDecodeError, RuntimeError, ValueError) as exc: return ( False, "", f"{type(exc).__name__}: {exc}", None, False, ) if record["status"] != "active": return False, "", f"dynamic tool is retired: {name}", None, False command = str(record["command"]) finally: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) extra = _string_arg(args, "args") if extra: command = f"{command} {shlex.quote(extra)}" ok, stdout, stderr, code = _run_command(command, cwd, timeout_s) return ok, stdout, stderr, code, True def _describe_dynamic_tool(name: str, cwd: str) -> dict[str, Any] | None: lock_path = _dynamic_lock_path(cwd) lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with lock_path.open("a+b") as lock_handle: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) try: try: record = _load_dynamic_record(name, cwd) except FileNotFoundError: return None _archive_dynamic_record(record, cwd) row = _dynamic_catalog_row(record) if record["status"] == "retired": row["invocation"] = "UpgradeTool" return row finally: fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) def _execute_tool_call_impl( call: ToolCall, *, cwd: str = ".", timeout_s: float = 0.0, session_id: str = "", external_effect_authorized: bool = False, ) -> ToolExecutionResult: started = time.perf_counter() if "depends_on" in call.args: call = replace( call, args={ key: value for key, value in call.args.items() if key != "depends_on" }, ) spec = tool_spec(call.name) if spec is None: return _result( call, ok=False, error=f"unsupported tool: {call.name}", started=started ) try: spec.validate_arguments(call.args) except ValueError as exc: return _result( call, ok=False, error=f"invalid arguments: {exc}", started=started, ) try: if call.name == "Bash": ok, stdout, stderr, code = _run_command( _string_arg(call.args, "command"), cwd, timeout_s ) return _result( call, ok=ok, output=stdout or stderr, stdout=stdout, stderr=stderr, exit_code=code, executed=True, started=started, ) if call.name == "Read": ok, output, error = _read(_string_arg(call.args, "path"), cwd) return _result( call, ok=ok, output=output, error=error, executed=True, started=started ) if call.name == "Write": ok, output, error = _write( _string_arg(call.args, "path"), _string_arg(call.args, "content"), cwd ) return _result( call, ok=ok, output=output, error=error, executed=True, started=started ) if call.name == "Edit": ok, output, error = _edit( _string_arg(call.args, "path"), _string_arg(call.args, "old_string"), _string_arg(call.args, "new_string"), cwd, ) return _result( call, ok=ok, output=output, error=error, executed=True, started=started ) if call.name == "Glob": pattern = _string_arg(call.args, "pattern") search_pattern = _workspace_glob_pattern(pattern, cwd) matches = sorted( _workspace_relative_path(Path(match), cwd) for match in globlib.glob(search_pattern, recursive=True) if not _workspace_relative_path(Path(match), cwd).startswith( ".nexum/" ) and _workspace_relative_path(Path(match), cwd) != ".nexum" ) return _result( call, ok=True, output="\n".join(matches), executed=True, started=started ) if call.name == "Grep": pattern = _string_arg(call.args, "pattern") path = _string_arg(call.args, "path", ".") ok, stdout, stderr, code = _grep(pattern, path, cwd, timeout_s) return _result( call, ok=ok, output=stdout or stderr, stdout=stdout, stderr=stderr, exit_code=code, executed=True, started=started, ) if call.name == "WebFetch": ok, output, error = _web_fetch(_string_arg(call.args, "url"), timeout_s) return _result( call, ok=ok, output=output, error=error, executed=True, started=started ) if call.name == "WebSearch": ok, output, error = _web_search(_string_arg(call.args, "query"), timeout_s) return _result( call, ok=ok, output=output, error=error, executed=True, started=started ) if call.name in BROWSER_TOOL_NAMES: return execute_browser_tool( call, ToolExecutionContext( workspace=str(Path(cwd).expanduser().resolve()), timeout_s=timeout_s, session_id=session_id, ), ) if call.name in ENGINEERING_TOOL_NAMES: return execute_engineering_tool( call, ToolExecutionContext( workspace=str(Path(cwd).expanduser().resolve()), timeout_s=timeout_s, session_id=session_id, ), run_command=_run_command, ) if call.name in DRAFTING_TOOL_NAMES: return execute_drafting_tool( call, ToolExecutionContext( workspace=str(Path(cwd).expanduser().resolve()), timeout_s=timeout_s, session_id=session_id, ), ) if call.name in repository_tool_names(): return execute_repository_tool( call, ToolExecutionContext( workspace=str(Path(cwd).expanduser().resolve()), timeout_s=timeout_s, session_id=session_id, ), github_token=os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN"), approved_external_effect=external_effect_authorized, ) if call.name in TRANSACTION_TOOL_NAMES: store = TransactionStore(cwd) if call.name == "TransactionBegin": record = store.begin( _string_tuple_arg(call.args, "paths"), session_id=session_id, ) elif call.name == "TransactionStatus": record = store.get( _string_arg(call.args, "transaction_id"), session_id=session_id, ) elif call.name == "TransactionCommit": record = store.commit( _string_arg(call.args, "transaction_id"), session_id=session_id, ) else: expected = call.args.get("expected_current") if not isinstance(expected, dict) or any( not isinstance(key, str) or not isinstance(value, str) for key, value in expected.items() ): raise ValueError("expected_current must map paths to digests") record = store.rollback( _string_arg(call.args, "transaction_id"), dict(expected), session_id=session_id, ) return _result( call, ok=True, output=json.dumps(record.to_dict(), sort_keys=True), executed=True, started=started, ) if call.name == "ToolCatalog": query = _string_arg(call.args, "query").lower() tools = [ tool for tool in [*list_tools(), *_dynamic_tool_catalog(cwd)] if not query or query in json.dumps(tool).lower() ] return _result( call, ok=True, output=json.dumps({"tools": tools}, indent=2), executed=True, started=started, ) if call.name == "ToolDescribe": requested_name = _string_arg(call.args, "name") selected = advertised_tool_spec(requested_name) dynamic = ( None if selected is not None else _describe_dynamic_tool(requested_name, cwd) ) return _result( call, ok=selected is not None or dynamic is not None, output=json.dumps( selected.to_dict() if selected is not None else dynamic, indent=2, ) if selected is not None or dynamic is not None else "", error="" if selected is not None or dynamic is not None else "tool is not available", executed=True, started=started, ) if call.name == "LanguagePacks": return _result( call, ok=True, output=json.dumps( language_pack_catalog(_string_arg(call.args, "query")), indent=2, sort_keys=True, ), executed=True, started=started, ) if call.name == "LanguageInspect": target = _workspace_path(_string_arg(call.args, "path"), cwd) if not target.is_file(): raise ValueError("language inspection target is not a file") return _result( call, ok=True, output=json.dumps( analyze_language_file( target, language=_string_arg(call.args, "language"), ), indent=2, sort_keys=True, ), executed=True, started=started, ) if call.name == "RequestInput": return ToolExecutionResult( name=call.name, args=call.args, ok=False, tool_call_id=call.call_id, output=json.dumps( { "prompt": _string_arg(call.args, "prompt"), "schema": call.args.get("schema") or {}, }, sort_keys=True, ), executed=False, status="input_required", source_trust=spec.source_trust, ) if call.name == "TaskStart": task = TaskStore(cwd).start_terminal( session_id=session_id, command=_string_arg(call.args, "command"), workspace=cwd, ) return _result( call, ok=True, output=json.dumps(task.to_dict(), sort_keys=True), executed=True, started=started, ) if call.name == "TaskStatus": task = TaskStore(cwd).status( _string_arg(call.args, "task_id"), session_id=session_id, ) return _result( call, ok=True, output=json.dumps(task.to_dict(), sort_keys=True), executed=True, started=started, ) if call.name == "TaskCancel": task = TaskStore(cwd).cancel( _string_arg(call.args, "task_id"), session_id=session_id, ) return _result( call, ok=task.status == "cancelled", output=json.dumps(task.to_dict(), sort_keys=True), error="" if task.status == "cancelled" else f"task is {task.status}", executed=True, started=started, ) if call.name == "ArtifactList": records = [ record.to_dict() for record in ArtifactStore(cwd).list(session_id=session_id) ] return _result( call, ok=True, output=json.dumps({"artifacts": records}, sort_keys=True), executed=True, started=started, ) if call.name == "ArtifactRead": offset = int(call.args.get("offset") or 0) raw_length = call.args.get("length") length = int(raw_length) if raw_length is not None else None artifact_record, data = ArtifactStore(cwd).read( _string_arg(call.args, "artifact_id"), offset=offset, length=length, session_id=session_id, ) return _result( call, ok=True, output=json.dumps( { "artifact": artifact_record.to_dict(), "offset": offset, "content": data.decode("utf-8", errors="replace"), }, sort_keys=True, ), executed=True, started=started, ) if call.name in {"CreateTool", "UpgradeTool", "RetireTool"}: operation = { "CreateTool": "create", "UpgradeTool": "upgrade", "RetireTool": "retire", }[call.name] ok, output, error = _mutate_dynamic( call.args, cwd, operation=operation, ) return _result( call, ok=ok, output=output, error=error, executed=True, started=started ) if call.name == "RunDynamicTool": ok, stdout, stderr, code, executed = _run_dynamic( call.args, cwd, timeout_s, ) return _result( call, ok=ok, output=stdout or stderr, error=stderr if not executed else "", stdout=stdout, stderr=stderr, exit_code=code, executed=executed, started=started, ) return _result( call, ok=False, error=f"unsupported tool: {call.name}", started=started ) except subprocess.TimeoutExpired: return _result( call, ok=False, error=f"timeout after {timeout_s}s", executed=True, started=started, ) except Exception as exc: return _result( call, ok=False, error=f"{type(exc).__name__}: {exc}", executed=True, started=started, ) def _safe_result( result: ToolExecutionResult, *, context: ToolExecutionContext, ) -> ToolExecutionResult: redactor = SecretRedactor() output = redactor.redact(result.output) stdout = redactor.redact(result.stdout) stderr = redactor.redact(result.stderr) error = redactor.redact(result.error) rendered = output or stdout or stderr output_sha256 = hashlib.sha256(rendered.encode("utf-8")).hexdigest() artifact_id = result.artifact_id if rendered: artifact = ArtifactStore(context.workspace).put_text( rendered, source=result.source_trust, session_id=context.session_id, ) artifact_id = artifact.artifact_id return replace( result, args=redact_sensitive_value(result.args, redactor=redactor), output=output, stdout=stdout, stderr=stderr, error=error, output_sha256=output_sha256, artifact_id=artifact_id, ) def execute_tool_call( call: ToolCall, *, cwd: str = ".", timeout_s: float = 0.0, session_id: str = "", ) -> ToolExecutionResult: effective_session_id = session_id or "direct" context = ToolExecutionContext( workspace=str(Path(cwd).expanduser().resolve()), timeout_s=timeout_s, session_id=effective_session_id, ) spec = tool_spec(call.name) if spec is None: return _safe_result( _execute_tool_call_impl( call, cwd=context.workspace, timeout_s=timeout_s, session_id=effective_session_id, ), context=context, ) policy = ToolPolicy.load(context.workspace) decision = policy.decision(spec) external_effect_authorized = decision == "allow" approval_store = ApprovalStore(context.workspace) if decision == "deny": result = ToolExecutionResult( name=call.name, args=call.args, ok=False, tool_call_id=call.call_id, error="tool policy denied this action", executed=False, status="denied", source_trust=spec.source_trust, ) return _safe_result(result, context=context) if decision == "approve": approval_session = session_id or "direct" if not call.approval_id: approval = approval_store.request(approval_session, call, spec) result = ToolExecutionResult( name=call.name, args=call.args, ok=False, tool_call_id=call.call_id, error="approval required for this exact action", executed=False, status="input_required", source_trust=spec.source_trust, approval_id=approval.approval_id, ) return _safe_result(result, context=context) try: approval_store.consume( call.approval_id, session_id=approval_session, call=call, ) external_effect_authorized = True except (OSError, PermissionError, RuntimeError, ValueError) as exc: result = ToolExecutionResult( name=call.name, args=call.args, ok=False, tool_call_id=call.call_id, error=f"approval rejected: {exc}", executed=False, status="denied", source_trust=spec.source_trust, approval_id=call.approval_id, ) return _safe_result(result, context=context) idempotency: IdempotencyStore | None = None if session_id and (call.call_id or call.idempotency_key): idempotency = IdempotencyStore(context.workspace) intent, created = idempotency.begin(session_id, call) if not created and intent.status == "completed" and intent.result is not None: return replace(ToolExecutionResult(**intent.result), replayed=True) if not created: result = ToolExecutionResult( name=call.name, args=call.args, ok=False, tool_call_id=call.call_id, error=( "a prior execution started without a durable result; inspect the " "environment before selecting a recovery action" ), executed=False, status="input_required", source_trust=spec.source_trust, ) return _safe_result(result, context=context) result = _safe_result( _execute_tool_call_impl( call, cwd=context.workspace, timeout_s=timeout_s, session_id=effective_session_id, external_effect_authorized=external_effect_authorized, ), context=context, ) if idempotency is not None: idempotency.complete(session_id, call, result) if session_id: EventLog(context.workspace).append( "tool_result", session_id=session_id, tool_call_id=call.call_id, status=result.status if result.status != "completed" else ("ok" if result.ok else "failed"), detail={ "tool": call.name, "executed": result.executed, "ok": result.ok, "output_sha256": result.output_sha256, "artifact_id": result.artifact_id, }, ) return result def execute_tool_text( text: str, *, cwd: str = ".", timeout_s: float = 0.0, session_id: str = "" ) -> list[ToolExecutionResult]: try: calls = parse_tool_calls(text) except Exception as exc: return [ ToolExecutionResult( name="ParseToolCall", args={}, ok=False, output="", error=f"{type(exc).__name__}: {exc}", executed=False, ) ] if not calls: return [ ToolExecutionResult( name="ParseToolCall", args={}, ok=False, output="", error="no parseable tool call found", executed=False, ) ] return list( execute_tool_calls( tuple(calls), cwd=cwd, timeout_s=timeout_s, session_id=session_id, ) ) def execute_tool_calls( calls: tuple[ToolCall, ...], *, cwd: str = ".", timeout_s: float = 0.0, session_id: str = "", ) -> tuple[ToolExecutionResult, ...]: return execute_call_batch( calls, execute=lambda call: execute_tool_call( call, cwd=cwd, timeout_s=timeout_s, session_id=session_id, ), resolve_spec=tool_spec, ) __all__ = [ "TOOL_CALL_END", "TOOL_CALL_START", "ToolCall", "ToolExecutionResult", "execute_tool_call", "execute_tool_calls", "execute_tool_text", "list_tools", "parse_tool_calls", "runtime_tool_schemas", "tool_names", "tool_schemas", "tool_spec", ]