Nexum-Expanded/runtime/src/nexum_runtime/executor.py

File size: 65,299 Bytes

9a70a84

"""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<body>.*?){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",
]