omnigent-ai--omnigent/omnigent/inner/loader.py

"""YAML / dict loader for AgentDef."""

from __future__ import annotations

import importlib
import re
from collections.abc import Callable
from pathlib import Path
from typing import Any, TypeAlias

import yaml

from .datamodel import (
    AgentDef,
    ExecutorSpec,
    MemoryConfig,
    OSEnvSandboxSpec,
    OSEnvSpec,
    ParamDef,
    TerminalEnvSpec,
)
from .policies import (
    FunctionPolicy,
    Policy,
    PromptPolicy,
)
from .tools import (
    AgentTool,
    CancellableFunctionTool,
    FunctionTool,
    HandoffTool,
    InheritedTool,
    MCPTool,
    SelfAgentTool,
    SkillTool,
    Tool,
)

# Config-dict shape from agent YAML files. The schema is a nested, open-ended
# tree (tools, policies, executor, os_env, sandbox — each with variant-typed
# children); the loader validates shape field-by-field via isinstance checks,
# so a TypedDict would duplicate that validation without buying us more safety.
YamlData: TypeAlias = dict[str, Any]  # type: ignore[explicit-any]

# Dynamically-loaded Python callables resolved from `callable: mod.func`
# paths in YAML; the loader has no way to know their signatures.
DynamicCallable: TypeAlias = Callable[..., object]  # type: ignore[explicit-any]


class _OmnigentYamlLoader(yaml.SafeLoader):
    """YAML loader with YAML 1.2-style booleans.

    PyYAML's default YAML 1.1 resolver treats unquoted keys like ``on`` as a
    boolean. That breaks policy definitions such as ``on: [tool_call]``.
    Keep ``true``/``false`` boolean parsing, but stop treating ``on``/``off``
    and similar legacy literals as booleans.
    """


_OmnigentYamlLoader.yaml_implicit_resolvers = {
    key: value[:] for key, value in yaml.SafeLoader.yaml_implicit_resolvers.items()
}
for key, resolvers in list(_OmnigentYamlLoader.yaml_implicit_resolvers.items()):
    _OmnigentYamlLoader.yaml_implicit_resolvers[key] = [
        (tag, regexp) for tag, regexp in resolvers if tag != "tag:yaml.org,2002:bool"
    ]
# types-PyYAML declares add_implicit_resolver without return annotations.
_OmnigentYamlLoader.add_implicit_resolver(  # type: ignore[no-untyped-call]
    "tag:yaml.org,2002:bool",
    re.compile(r"^(?:true|True|TRUE|false|False|FALSE)$"),
    list("tTfF"),
)


def load_agent_def(
    path_or_dict: str | Path | YamlData,
    *,
    enforce_handler_allowlist: bool = False,
) -> AgentDef:
    """Load an AgentDef from a YAML file path or a raw dict.

    When *path_or_dict* is a path, the YAML's ``instructions:``
    field (if present) is resolved relative to the YAML's parent
    directory: a value that names an existing file gets its
    contents read in, and any other string is treated as inline
    text. When *path_or_dict* is a raw dict, the same string is
    treated as inline text — there is no anchor for path
    resolution. See :func:`_resolve_instructions` for the rules.

    :param path_or_dict: A YAML file path or already-parsed dict.
    :param enforce_handler_allowlist: When ``True``, reject any
        ``type: function`` policy whose ``handler:`` / ``callable:``
        dotted path is not a registered policy handler, *before*
        ``_parse_agent_def`` resolves and (for factory policies)
        **calls** it. This is the guard for the untrusted
        agent-bundle upload path: ``omnigent.spec.load`` routes a
        single-file omnigent YAML bundle here during
        ``validate_agent_bundle``, and the loader executes policy
        factories at parse time, so an uploaded
        ``handler: subprocess.Popen`` would otherwise run during
        validation. Defaults to ``False`` so trusted callers (local
        ``omnigent run``, operator specs, the CLI) keep working with
        custom handlers — the operator already has code execution, so
        the restriction would add no security there.
    """
    if isinstance(path_or_dict, (str, Path)):
        path = Path(path_or_dict)
        with open(path) as f:
            data = yaml.load(f, Loader=_OmnigentYamlLoader)
        instructions_root: Path | None = path.parent
    else:
        data = path_or_dict
        instructions_root = None
    if enforce_handler_allowlist:
        _reject_unregistered_policy_handlers(data)
    return _parse_agent_def(data, instructions_root=instructions_root)


def _reject_unregistered_policy_handlers(data: YamlData) -> None:
    """Reject ``type: function`` policies whose handler is not registered.

    Scans the raw YAML ``policies:`` mapping for handler dotted paths
    that are not in the policy registry and raises before any import or
    factory call. Tool ``callable:`` paths are intentionally *not*
    scanned — they are a separate surface and are not invoked at parse
    time. See :func:`load_agent_def` for why this only runs on the
    untrusted bundle-upload path.

    :param data: The raw agent YAML dict (pre-parse). Non-dict input
        (malformed YAML) is ignored here and left for the parser to
        reject.
    :raises ValueError: If a function policy names an unregistered
        handler, e.g. ``"subprocess.Popen"``.
    """
    from omnigent.policies.registry import is_registered_handler

    if not isinstance(data, dict):
        return
    policies = data.get("policies")
    if not isinstance(policies, dict):
        return
    for pname, pdata in policies.items():
        if not isinstance(pdata, dict):
            continue
        if pdata.get("type", "function") != "function":
            continue
        handler = pdata.get("handler") or pdata.get("callable")
        if isinstance(handler, str) and not is_registered_handler(handler):
            raise ValueError(
                f"Policy {pname!r}: handler {handler!r} is not a registered policy "
                f"handler. Uploaded agent bundles may only use handlers from the "
                f"policy registry; a server admin must add custom handlers via the "
                f"'policy_modules' config."
            )


def _read_contained_file(root: Path, value: str) -> str | None:
    """
    Read an *instructions_root*-relative file named by *value*, if contained.

    Mirrors :func:`omnigent.spec.parser._read_contained_file`: resolves
    symlinks and ``..`` and confirms the target stays within *root* before
    reading, so a crafted ``instructions: ../../etc/passwd`` in an uploaded
    bundle cannot read files outside the bundle. Returns ``None`` for a
    non-contained or non-existent path so the caller falls back to treating
    *value* as inline text.

    :param root: Directory the value is anchored to, e.g. the spec file's
        parent directory.
    :param value: The single-line ``instructions:`` value, e.g.
        ``"AGENTS.md"``.
    :returns: The file contents if *value* names a contained file, else
        ``None``.
    """
    candidate = root / value
    try:
        resolved = candidate.resolve()
        if resolved.is_relative_to(root.resolve()) and resolved.is_file():
            return resolved.read_text()
    except OSError:
        # Path too long or invalid characters — fall through to inline text.
        pass
    return None


def _resolve_instructions(
    raw_value: object,
    instructions_root: Path | None,
) -> str | None:
    """
    Resolve the ``instructions:`` field to a system-prompt string.

    Mirrors :func:`omnigent.spec.parser._resolve_instructions`
    so omnigent-flavored YAMLs and native Omnigent YAMLs treat the
    field identically:

    - Single-line value that names an existing file relative to
      *instructions_root*: the file's contents are read.
    - Multi-line value (contains ``\\n``): treated as inline text.
    - File-path-shaped value that doesn't resolve: silently
      treated as inline text. Matches native Omnigent behavior so users
      who type ``instructions: AGENTS.md`` and forget the file
      get the literal string back rather than a misleading error.
    - ``None`` / non-string raw values return ``None``.

    :param raw_value: The raw YAML value at the ``instructions:``
        key, or ``None`` if the key was absent.
    :param instructions_root: Directory to anchor relative path
        lookups. ``None`` skips the file-read attempt and treats
        every value as inline text — used when the loader has no
        on-disk anchor (raw-dict input path).
    :returns: The resolved instruction text, or ``None`` if no
        ``instructions:`` was supplied.
    """
    if raw_value is None:
        return None
    if not isinstance(raw_value, str):
        return None
    if "\n" in raw_value:
        return raw_value
    if instructions_root is not None:
        contained = _read_contained_file(instructions_root, raw_value)
        if contained is not None:
            return contained
    return raw_value


def _parse_agent_def(
    data: YamlData,
    *,
    instructions_root: Path | None = None,
) -> AgentDef:
    agent = AgentDef()
    # ``AgentDef.name`` and ``AgentDef.prompt`` are both ``str | None``;
    # missing YAML keys flow through as ``None``.
    agent.name = data.get("name")
    agent.prompt = data.get("prompt")
    # ``instructions:`` is resolved into the agent's system-prompt
    # text right here (file path → contents, or inline string).
    # Translator code downstream only sees the resolved string
    # in ``agent.instructions`` and the raw user-supplied
    # ``agent.prompt`` — it doesn't have to re-walk the path.
    agent.instructions = _resolve_instructions(data.get("instructions"), instructions_root)
    agent.input_type = data.get("input_type")
    agent.output_type = data.get("output_type")
    if "async_enabled" in data:
        agent.async_enabled = data["async_enabled"]
    elif "async" in data:
        agent.async_enabled = data["async"]
    if "cancellable" in data:
        agent.cancellable = bool(data["cancellable"])
    agent.runtime = data.get("runtime", False)
    agent.timers = data.get("timers", False)
    agent.spawn = data.get("spawn", False)
    agent.agent_session_sharing = data.get("agent_session_sharing", "none")
    agent.os_env = _parse_os_env_spec(data.get("os_env"))

    # Executor
    executor_data = data.get("executor")
    if executor_data:
        agent.executor = _parse_executor_spec(executor_data)

    # Params
    for pname, pdata in data.get("params", {}).items():
        if isinstance(pdata, dict):
            agent.params[pname] = ParamDef(
                type=pdata.get("type", "string"),
                description=pdata.get("description"),
                default=pdata.get("default"),
            )
        else:
            agent.params[pname] = ParamDef(type="string", default=pdata)

    # Tools
    for tname, tdata in data.get("tools", {}).items():
        agent.tools[tname] = _parse_tool(tname, tdata)

    # Policies
    for pname, pdata in data.get("policies", {}).items():
        agent.policies[pname] = _parse_policy(pname, pdata)

    # Memories
    for mname, mdata in data.get("memories", {}).items():
        if isinstance(mdata, dict):
            agent.memories[mname] = MemoryConfig(
                scope=mdata.get("scope", "per_session"),
            )
        else:
            agent.memories[mname] = MemoryConfig()

    # Labels
    agent.labels = {str(k): str(v) for k, v in data.get("labels", {}).items()}

    # ASK timeout
    if "ask_timeout" in data:
        agent.ask_timeout = float(data["ask_timeout"]) if data["ask_timeout"] is not None else None

    # Label schema
    from .datamodel import LabelSchemaRule

    for ls_name, ls_data in data.get("label_schema", {}).items():
        if isinstance(ls_data, dict):
            agent.label_schema[str(ls_name)] = LabelSchemaRule(
                values=[str(v) for v in ls_data.get("values", [])],
            )

    # Policy transparency
    if "policy_transparency" in data:
        agent.policy_transparency = bool(data["policy_transparency"])

    # Terminals
    for term_name, term_data in data.get("terminals", {}).items():
        agent.terminals[str(term_name)] = _parse_terminal_env_spec(term_data)
    # Cross-field validation: a terminal with ``allow_sandbox_override``
    # lets the LLM pass an arbitrary ``sandbox`` arg to
    # ``sys_terminal_launch``. The override only mutates ``sandbox.type``
    # — egress_rules stay on the policy object but an override to
    # ``none`` (an accepted override value) can't enforce them, so the
    # LLM effectively drops the egress allow-list. Reject the
    # combination at parse time rather than silently accept a spec
    # that's only as strong as the LLM lets it be.
    for term_name, term_spec in agent.terminals.items():
        if not term_spec.allow_sandbox_override:
            continue
        effective_sandbox = _effective_terminal_sandbox(term_spec, agent.os_env)
        if effective_sandbox is not None and effective_sandbox.egress_rules:
            raise ValueError(
                f"terminal {term_name!r}: allow_sandbox_override=true is "
                "incompatible with egress_rules on the effective sandbox. "
                "An override of sandbox.type to 'none' "
                "drops hard network enforcement while egress_rules remain "
                "as inert decoration on the policy, so the LLM could "
                "silently bypass the network allow-list. Either remove "
                "allow_sandbox_override or remove egress_rules from the "
                "effective sandbox."
            )

    # Workflow
    agent.workflow = data.get("workflow")

    # Metadata
    agent.metadata = data.get("metadata", {})

    return agent


# ---------------------------------------------------------------------------
# Tool parsing
# ---------------------------------------------------------------------------


def _parse_tool(name: str, data: str | YamlData) -> Tool:
    if isinstance(data, str):
        if data == "inherit":
            return InheritedTool(name=name)
        if data == "self":
            # ``tools.<name>: self`` shorthand: the sub-agent's spec
            # is a clone of the parent's. See
            # :class:`SelfAgentTool` — the translator materializes
            # the sub-spec at translation time.
            return SelfAgentTool(name=name)
        return FunctionTool(name=name, description=data)

    if not isinstance(data, dict):
        return FunctionTool(name=name, description=str(data))

    tool_type = data.get("type", "function")

    if tool_type == "function":
        # Reject typos like ``runtime: clinet`` at load time.
        runtime_raw = data.get("runtime", "server")
        if runtime_raw not in ("server", "client"):
            raise ValueError(
                f"Tool '{name}': invalid runtime {runtime_raw!r}; must be 'server' or 'client'.",
            )
        tool = FunctionTool(
            name=name,
            description=data.get("description"),
            catalog_path=data.get("catalog_path"),
            input_schema=data.get("parameters") or data.get("input_schema"),
            cancellable=bool(data.get("cancellable", False)),
            runtime=runtime_raw,
            warehouse_id=data.get("warehouse_id"),
        )
        callable_path = data.get("callable")
        if callable_path and isinstance(callable_path, str):
            # Catch the conflict before the validator resolves the
            # import, so client-runtime YAMLs never load the module.
            if runtime_raw == "client":
                raise ValueError(
                    f"Tool '{name}': 'runtime: client' tools must not "
                    f"declare a 'callable:'. The SDK consumer provides "
                    f"the implementation at stream-start time; declaring "
                    f"a server-side callable here would silently never "
                    f"run.",
                )
            tool.callable = _resolve_callable(callable_path)
        return tool

    if tool_type == "cancellable_function":
        raw_runner = data.get("runner")
        return CancellableFunctionTool(
            name=name,
            description=data.get("description"),
            input_schema=data.get("parameters") or data.get("input_schema"),
            runner=_resolve_callable(raw_runner) if isinstance(raw_runner, str) else None,
            cancellable=bool(data.get("cancellable", True)),
        )

    if tool_type == "mcp":
        # ``profile`` can be declared directly (``profile: myprof``)
        # or inside an ``auth:`` block (``auth: {type: databricks,
        # profile: myprof}``). The ``auth:`` block is the preferred
        # form — it matches the executor auth shape.
        mcp_profile = data.get("profile")
        raw_auth = data.get("auth")
        if mcp_profile is None and isinstance(raw_auth, dict):
            if str(raw_auth.get("type", "")) == "databricks":
                mcp_profile = raw_auth.get("profile")
        return MCPTool(
            name=name,
            description=data.get("description"),
            url=data.get("url"),
            command=data.get("command"),
            args=data.get("args"),
            env=data.get("env"),
            tools=data.get("tools"),
            tool_name=data.get("tool_name"),
            profile=mcp_profile,
            databricks_server=data.get("databricks_server"),
            headers=data.get("headers"),
        )

    if tool_type == "agent":
        # ``spec: self`` dict form: the sub-agent's spec is a clone
        # of the parent's. Other agent fields (prompt, tools,
        # executor, os_env, ...) would conflict with the cloned
        # parent, so reject them up-front rather than silently
        # ignoring them.
        if data.get("spec") == "self":
            for conflicting in (
                "prompt",
                "tools",
                "executor",
                "os_env",
                "pass_history",
                "pass_histories",
                "max_sessions",
            ):
                if conflicting in data:
                    raise ValueError(
                        f"Tool '{name}': 'spec: self' cannot be combined with "
                        f"'{conflicting}'. The sub-agent's configuration is "
                        f"cloned from the parent — declare a regular "
                        f"'type: agent' tool with explicit fields if you "
                        f"need overrides."
                    )
            return SelfAgentTool(
                name=name,
                description=data.get("description"),
            )

        sub_tools: dict[str, Tool] = {}
        for sname, sdata in data.get("tools", {}).items():
            sub_tools[sname] = _parse_tool(sname, sdata)
        raw_max_sessions = data.get("max_sessions")
        max_sessions: int | None = None
        if raw_max_sessions is not None:
            if not isinstance(raw_max_sessions, int) or isinstance(raw_max_sessions, bool):
                raise ValueError(
                    f"Tool '{name}': 'max_sessions' must be an integer, "
                    f"got {type(raw_max_sessions).__name__!r} ({raw_max_sessions!r})."
                )
            if raw_max_sessions < 1:
                raise ValueError(
                    f"Tool '{name}': 'max_sessions' must be >= 1, got {raw_max_sessions!r}."
                )
            max_sessions = raw_max_sessions
        return AgentTool(
            name=name,
            description=data.get("description"),
            prompt=data.get("prompt"),
            tools=sub_tools,
            executor=_parse_executor_spec(data.get("executor")),
            os_env=(
                data.get("os_env")
                if data.get("os_env") == "inherit"
                else _parse_os_env_spec(data.get("os_env"))
            ),
            pass_history=data.get("pass_history", False),
            pass_histories=data.get("pass_histories"),
            max_sessions=max_sessions,
        )

    if tool_type == "inherit":
        return InheritedTool(name=name)

    if tool_type == "skill":
        # ``content`` is an alias for ``path`` — either is accepted. ``None``
        # when neither is configured; ``SkillTool.path`` is ``str | None``.
        path_value = data.get("content") or data.get("path")
        if path_value is not None and not isinstance(path_value, str):
            raise TypeError(f"Tool {name!r}: skill 'path'/'content' must be a string")
        return SkillTool(
            name=name,
            description=data.get("description"),
            path=path_value,
        )

    if tool_type == "handoff":
        return HandoffTool(
            name=name,
            description=data.get("description"),
            target_agent=data.get("target_agent"),
            pass_history=data.get("pass_history", True),
            bidirectional=data.get("bidirectional", True),
        )

    # Default: treat as function.
    return FunctionTool(name=name, description=data.get("description"))


# ---------------------------------------------------------------------------
# Policy parsing
# ---------------------------------------------------------------------------


def _parse_policy(name: str, data: YamlData | str | bool | None) -> Policy:
    if not isinstance(data, dict):
        return Policy(name=name)

    policy_type = data.get("type", "function")
    # Default ``on`` for function policies is all four phases — the callable
    # self-selects which to act on by returning ALLOW for events it ignores.
    # Prompt and label policies retain their old defaults (see their branches).
    _function_all_phases = ["request", "response", "tool_call", "tool_result"]
    on = data.get(
        "on", _function_all_phases if policy_type == "function" else ["request", "response"]
    )

    if policy_type == "function":
        callable_obj = None
        # Accept both ``handler:`` (new name) and ``callable:`` (legacy alias).
        callable_path = data.get("handler") or data.get("callable")
        if callable_path and isinstance(callable_path, str):
            callable_obj = _resolve_callable(callable_path)
        has_factory_params = "factory_params" in data
        factory_params = data.get("factory_params", {})
        factory: DynamicCallable | None = None
        if has_factory_params and callable_obj is not None:
            factory = callable_obj
            factory_result = callable_obj(**factory_params)
            if not callable(factory_result):
                raise ValueError(
                    f"Policy factory '{callable_path}' returned non-callable: "
                    f"{type(factory_result)}"
                )
            callable_obj = factory_result
        return FunctionPolicy(
            name=name,
            on=on,
            callable=callable_obj,
            # Only carry factory_params when the factory was
            # successfully resolved. If _resolve_callable returned
            # None (import error on the server side), passing
            # factory_params without factory would trip
            # FunctionPolicy.__post_init__'s invariant check.
            factory_params=factory_params if factory is not None else {},
            factory=factory,
        )

    if policy_type == "prompt":
        return PromptPolicy(
            name=name,
            on=on,
            # `None` means no prompt was configured; PromptPolicy.evaluate()
            # fails closed in that case.
            prompt=data.get("prompt"),
            executor=_parse_executor_spec(data.get("executor")),
            allow_set_labels=bool(data.get("allow_set_labels", False)),
            allowed_label_keys=(
                [str(key) for key in data.get("allowed_label_keys", [])]
                if isinstance(data.get("allowed_label_keys"), list)
                else None
            ),
        )

    return Policy(name=name, on=on)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _resolve_callable(dotted_path: str) -> DynamicCallable | None:
    """Try to import a dotted Python path like ``mypackage.module.func``.

    Returns ``None`` on failure (we don't want the loader to crash just
    because a function isn't importable in this environment).
    """
    parts = dotted_path.rsplit(".", 1)
    if len(parts) != 2:
        return None
    module_path, attr_name = parts
    try:
        mod = importlib.import_module(module_path)
        return getattr(mod, attr_name, None)
    except (ImportError, AttributeError):
        return None


def _parse_executor_spec(data: YamlData | str | bool | None) -> ExecutorSpec | None:
    if not data:
        return None
    if isinstance(data, str):
        return ExecutorSpec(model=data)
    if isinstance(data, dict):
        # ``ExecutorSpec.{model,harness,profile}`` are ``str | None``;
        # missing keys map to ``None`` directly. ``data.get`` happens to
        # already return ``None`` for missing keys, so the assignment
        # flows through unchanged.
        #
        # Parse ``executor.auth`` into a typed auth dataclass so that
        # inline AgentTool sub-agents can declare auth (e.g. api_key +
        # base_url for mock LLM routing) and have it flow through to the
        # child spec's executor. Without this, auth blocks on inline
        # sub-agent executors are silently dropped.
        auth = None
        raw_auth = data.get("auth")
        if isinstance(raw_auth, dict):
            from omnigent.spec.parser import _parse_executor_auth

            auth = _parse_executor_auth(data, expand_env=True)
        return ExecutorSpec(
            model=data.get("model"),
            harness=data.get("harness"),
            profile=data.get("profile"),
            auth=auth,
        )
    return None


def _parse_os_env_spec(data: YamlData | str | bool | None) -> OSEnvSpec | None:
    if data is None:
        return None
    if isinstance(data, str):
        return OSEnvSpec(type=data)
    if not isinstance(data, dict):
        raise TypeError("os_env must be a string or mapping")

    sandbox_data = data.get("sandbox")
    sandbox = None
    if sandbox_data is not None:
        sandbox = _parse_os_env_sandbox_spec(sandbox_data)
    fork = bool(data.get("fork", False))
    start_in_scratch = bool(data.get("start_in_scratch", False))
    # Mirror the Omnigent YAML parser's cross-field validation
    # (omnigent/spec/parser.py:_parse_os_env). Keeping the two
    # loaders in lockstep prevents a class of "silently weakens
    # the sandbox on the legacy path" bug — an operator who used
    # ``load_agent_def`` would otherwise get past parse time with
    # a spec that ``create_os_environment`` later rejects (or, worse,
    # accepts with weaker semantics than the Omnigent parser would have
    # allowed).
    if start_in_scratch and fork:
        raise ValueError(
            "os_env.start_in_scratch and os_env.fork are mutually exclusive: "
            "fork already provides a writable workspace by copying cwd"
        )
    if start_in_scratch and sandbox is not None and sandbox.type == "none":
        raise ValueError(
            "os_env.start_in_scratch requires an active sandbox; "
            "sandbox.type=none does not create a scratch tmpdir"
        )
    return OSEnvSpec(
        type=data.get("type", "caller_process"),
        cwd=data.get("cwd"),
        sandbox=sandbox,
        fork=fork,
        start_in_scratch=start_in_scratch,
    )


def _effective_terminal_sandbox(
    term_spec: TerminalEnvSpec, agent_os_env: OSEnvSpec | None
) -> OSEnvSandboxSpec | None:
    """Return the sandbox a terminal would end up with at launch time.

    Mirrors the runtime resolution in
    :func:`omnigent.inner.terminal.build_terminal_os_env_spec`: if the
    terminal has its own ``os_env`` spec, that wins; if it inherits
    (``os_env: inherit`` or omitted), the agent's ``os_env`` is used.
    Used at parse time for cross-field validation that needs both
    halves of the spec.
    """
    if isinstance(term_spec.os_env, OSEnvSpec):
        return term_spec.os_env.sandbox
    if agent_os_env is not None:
        return agent_os_env.sandbox
    return None


def _parse_terminal_env_spec(data: YamlData | str | bool | None) -> TerminalEnvSpec:
    if isinstance(data, str):
        return TerminalEnvSpec(command=data)
    if not isinstance(data, dict):
        raise TypeError("terminal spec must be a string or mapping")

    os_env_val = data.get("os_env")
    os_env: OSEnvSpec | str | None = None
    if os_env_val == "inherit":
        os_env = "inherit"
    elif os_env_val is not None:
        os_env = _parse_os_env_spec(os_env_val)

    env_val = data.get("env", {})
    if not isinstance(env_val, dict):
        raise TypeError("terminal 'env' must be a mapping of string -> string")
    env = {str(k): str(v) for k, v in env_val.items()}

    return TerminalEnvSpec(
        command=data.get("command", "bash"),
        args=list(data.get("args", [])),
        env=env,
        os_env=os_env,
        allow_cwd_override=bool(data.get("allow_cwd_override", False)),
        allow_sandbox_override=bool(data.get("allow_sandbox_override", False)),
        log_file=data.get("log_file"),
        scrollback=int(data.get("scrollback", 10000)),
        session_prefix=data.get("session_prefix", "omni_"),
    )


def _parse_os_env_sandbox_spec(data: YamlData | str | bool | None) -> OSEnvSandboxSpec:
    if isinstance(data, str):
        return OSEnvSandboxSpec(type=data)
    if data is False:
        return OSEnvSandboxSpec(type="none")
    if not isinstance(data, dict):
        raise TypeError("os_env.sandbox must be a string, false, or mapping")
    raw_type = data.get("type")
    if raw_type is None:
        # No ``type:`` field -- resolve via the platform default
        # (same behavior as the Omnigent YAML parser, kept in sync so legacy
        # and Omnigent loaders agree on what an "untyped" sandbox block means).
        from .sandbox import _default_sandbox_for_platform

        sandbox_type = _default_sandbox_for_platform().type
    else:
        sandbox_type = raw_type
    egress_rules = data.get("egress_rules")
    # Mirror the Omnigent parser's hard reject of ``egress_rules`` paired with
    # a backend that cannot enforce them at spawn time. Without this
    # check the loader would happily accept a YAML that declares an
    # egress allow-list on ``none``, where the
    # MITM proxy is never wired up — the rules would sit in the spec
    # object as inert decoration while ``curl`` reaches the open
    # internet, looking exactly like the bug class fixed in
    # ``terminal._clone_sandbox_spec``.
    if egress_rules and sandbox_type not in ("linux_bwrap", "darwin_seatbelt"):
        raise ValueError(
            "os_env.sandbox.egress_rules requires sandbox.type=linux_bwrap "
            "(Linux) or sandbox.type=darwin_seatbelt (macOS) for hard "
            "network enforcement: those backends restrict network access "
            "at spawn time so the MITM proxy is the only egress path. "
            f"Got sandbox.type={sandbox_type!r}."
        )
    allow_private = data.get("egress_allow_private_destinations", False)
    if not isinstance(allow_private, bool):
        raise TypeError(
            "os_env.sandbox.egress_allow_private_destinations must be a "
            f"boolean, got {type(allow_private).__name__}"
        )
    # Secretless credential proxy. Reuse the single canonical parser
    # (``omnigent.spec.parser._parse_credential_proxy``) rather than a
    # second copy so the single-file omnigent-YAML path and the
    # bundle/config.yaml path can never drift — a duplicated parser here
    # is exactly what silently dropped ``credential_proxy`` on this path
    # before. Lazy-imported to avoid an import-time cycle with the spec
    # layer (which imports inner.datamodel). The two cross-field guards
    # below mirror the spec parser so an inert credential_proxy (no
    # hardened backend / no egress rules) is rejected on both paths.
    from omnigent.spec.parser import (
        _credential_proxy_macos_unsupported_reason,
        _parse_credential_proxy,
    )

    credential_proxy = _parse_credential_proxy(data.get("credential_proxy"))
    if credential_proxy is not None and sandbox_type not in ("linux_bwrap", "darwin_seatbelt"):
        raise ValueError(
            "os_env.sandbox.credential_proxy requires sandbox.type=linux_bwrap "
            "(Linux) or sandbox.type=darwin_seatbelt (macOS) so credentials are "
            "bound to a hardened helper boundary. "
            f"Got sandbox.type={sandbox_type!r}."
        )
    if credential_proxy is not None and not egress_rules:
        raise ValueError(
            "os_env.sandbox.credential_proxy requires os_env.sandbox.egress_rules: "
            "the MITM egress proxy is what swaps the synthetic placeholder for the "
            "real credential and rejects placeholder leaks, so it must be active."
        )
    macos_reason = _credential_proxy_macos_unsupported_reason(credential_proxy, sandbox_type)
    if macos_reason is not None:
        raise ValueError(macos_reason)
    # Defer the absent-field defaults to the dataclass so there is a single
    # source of truth: re-stating literals here (e.g. ``"warn"``, ``50000``)
    # silently drifts the moment the OSEnvSandboxSpec defaults change. An
    # explicit YAML ``null`` is treated the same as a missing key.
    fields = OSEnvSandboxSpec.__dataclass_fields__
    max_entries_raw = data.get("cwd_hidden_scan_max_entries")
    overflow_raw = data.get("cwd_hidden_scan_overflow")
    return OSEnvSandboxSpec(
        type=sandbox_type,
        read_paths=data.get("read_paths"),
        write_paths=(
            list(data["write_paths"])
            if "write_paths" in data and data.get("write_paths") is not None
            else None
        ),
        write_files=(
            list(data["write_files"])
            if "write_files" in data and data.get("write_files") is not None
            else None
        ),
        allow_network=data.get("allow_network", True),
        cwd_allow_hidden=data.get("cwd_allow_hidden"),
        cwd_hidden_scan_max_entries=(
            int(max_entries_raw)
            if max_entries_raw is not None
            else fields["cwd_hidden_scan_max_entries"].default
        ),
        cwd_hidden_scan_overflow=(
            str(overflow_raw)
            if overflow_raw is not None
            else fields["cwd_hidden_scan_overflow"].default
        ),
        env_passthrough=data.get("env_passthrough"),
        egress_rules=egress_rules,
        egress_allow_private_destinations=allow_private,
        credential_proxy=credential_proxy,
    )


def load_agent_def_from_path(path_str: str) -> AgentDef:
    """
    Resolve a CLI target path to an :class:`AgentDef`.

    Supports three shapes:

    - ``foo.yaml`` (single-file omnigent YAML) — parsed directly.
    - ``my-agent/`` (omnigent AGENTSPEC directory with
      ``config.yaml``, or a directory containing exactly one
      omnigent YAML at its root).
    - ``my-agent.tar.gz`` (tarball bundle) — extracted, then same
      directory dispatch.
    """
    path = Path(path_str)
    if path.is_file() and path.suffix not in {".tgz", ".gz", ".tar"}:
        return load_agent_def(path_str)

    if not (path.is_dir() or path.is_file()):
        raise FileNotFoundError(f"agent spec not found: {path}")

    import tempfile

    from omnigent.spec import _find_omnigent_yaml_in_dir

    if path.is_dir():
        resolved_dir = path
        cleanup: tempfile.TemporaryDirectory[str] | None = None
    else:
        from omnigent.spec.tar_utils import extract_safe

        cleanup = tempfile.TemporaryDirectory(prefix="omnigent-bundle-")
        resolved_dir = Path(cleanup.name)
        extract_safe(path.read_bytes(), resolved_dir)

    try:
        config_yaml = resolved_dir / "config.yaml"
        if config_yaml.is_file():
            from omnigent.spec import load as load_agent_spec
            from omnigent.spec.omnigent import (
                agent_spec_to_agent_def,
            )

            agent_spec = load_agent_spec(resolved_dir)
            return agent_spec_to_agent_def(agent_spec)
        single = _find_omnigent_yaml_in_dir(resolved_dir)
        if single is None:
            raise FileNotFoundError(
                f"{path}: directory has neither a config.yaml "
                f"nor a single-file omnigent YAML at its root"
            )
        return load_agent_def(str(single))
    finally:
        if cleanup is not None:
            cleanup.cleanup()