"""CLI agent factory — wraps create_deep_agent() with CLI-specific defaults.""" from __future__ import annotations from collections.abc import Callable from pathlib import Path from typing import TYPE_CHECKING, Any, Literal from pydantic_ai_backends import LocalBackend from apps.cli.config import load_config from apps.cli.model_resolve import resolve_cli_model from apps.cli.reminder import _build_reminder_config from pydantic_deep.agent import create_deep_agent from pydantic_deep.deps import DeepAgentDeps from pydantic_deep.features.forking.capability import LiveForkCapability from pydantic_deep.features.hooks import Hook, HookEvent, HookInput, HookResult from pydantic_deep.features.message_queue import MessageQueue from pydantic_deep.prompts import build_system_prompt if TYPE_CHECKING: from pydantic_ai.models import Model def _detect_fork_test_command(backend: Any) -> str | None: """Auto-detect a test command from the project root for the fork test runner. Checks common test framework markers in `backend.root_dir` (only :class:`~pydantic_ai_backends.LocalBackend` has a `root_dir`). Returns a ready-to-run shell string, or `None` when nothing is detected — the fork runner stays disabled in that case and `auto_with_fallback` falls through to the manual picker as before. Priority order: 1. `pyproject.toml` / `pytest.ini` / `setup.cfg` → pytest 2. `package.json` with a `test` script → npm test 3. `Makefile` with a `test:` or `test :` target → make test """ root_obj = getattr(backend, "root_dir", None) if root_obj is None: return None root = Path(root_obj) # pytest markers pyproject = root / "pyproject.toml" if pyproject.exists(): try: content = pyproject.read_text(encoding="utf-8", errors="ignore") if "[tool.pytest" in content or "[tool.pytest.ini_options]" in content: return "uv run pytest -q --tb=short" except OSError: pass if (root / "pytest.ini").exists() or (root / "setup.cfg").exists(): return "uv run pytest -q --tb=short" # npm / yarn / pnpm pkg = root / "package.json" if pkg.exists(): try: import json data = json.loads(pkg.read_text(encoding="utf-8", errors="ignore")) if isinstance(data.get("scripts"), dict) and "test" in data["scripts"]: return "npm test" except (OSError, ValueError): pass # Makefile with a test target makefile = root / "Makefile" if makefile.exists(): try: for line in makefile.read_text(encoding="utf-8", errors="ignore").splitlines(): if line.startswith("test:") or line.startswith("test "): return "make test" except OSError: pass return None def _make_shell_allow_list_hook(allow_list: list[str]) -> Hook: """Create a hook that blocks shell commands not in the allow list. Args: allow_list: List of allowed command prefixes (e.g., ["python", "pip", "npm"]). Returns: Hook that filters execute tool calls. """ async def _check_command(hook_input: HookInput) -> HookResult: command = str(hook_input.tool_input.get("command", "")) cmd_base = command.strip().split()[0] if command.strip() else "" for allowed in allow_list: if cmd_base == allowed or command.strip().startswith(allowed): return HookResult(allow=True) allowed_str = ", ".join(allow_list) return HookResult( allow=False, reason=( f"Command '{cmd_base}' is not in the allow-list. " f"Allowed commands: {allowed_str}. " f"Try a different approach or use an allowed command." ), ) return Hook( event=HookEvent.PRE_TOOL_USE, handler=_check_command, matcher=r"^execute$", ) def create_cli_agent( # noqa: C901 model: str | Model | None = None, fallback_model: str | Model | None = None, working_dir: str | None = None, shell_allow_list: list[str] | None = None, on_cost_update: Any | None = None, on_context_update: Any | None = None, on_before_compress: Any | None = None, on_after_compress: Any | None = None, on_eviction: Any | None = None, on_reminder: Callable[[int, str], None] | None = None, summarization_model: str | None = None, extra_middleware: list[Any] | None = None, backend: Any | None = None, sandbox: str | None = None, sandbox_image: str | None = None, sandbox_env_vars: dict[str, str] | None = None, sandbox_env_file: str | None = None, workspace: str | None = None, *, include_skills: bool | None = None, include_plan: bool | None = None, include_memory: bool | None = None, include_subagents: bool | None = None, include_todo: bool | None = None, include_local_context: bool = True, context_discovery: bool | None = None, non_interactive: bool = False, lean: bool = False, config_path: Path | None = None, model_settings: dict[str, Any] | None = None, session_id: str | None = None, skills_dir: str | None = None, extra_instructions: str | None = None, web_search: bool | None = None, web_fetch: bool | None = None, thinking: bool | str | None = None, include_teams: bool | None = None, temperature: float | None = None, include_browser: bool | None = None, browser_headless: bool | None = None, include_liteparse: bool | None = None, periodic_reminder: bool | None = None, reminder_mode: Literal["off", "first", "context", "llm"] | None = None, reminder_model: str | None = None, forking: bool | None = None, include_improve: bool | None = None, tool_search: bool | None = None, ) -> tuple[Any, DeepAgentDeps]: """Create a CLI-configured agent with all pydantic-deep capabilities. Configuration precedence: explicit arguments > config file > defaults. Args: model: Model to use. Falls back to the config file's `model`, which itself defaults to `pydantic_deep.models.DEFAULT_MODEL`. working_dir: Filesystem root directory. Defaults to cwd. shell_allow_list: Allowed shell command prefixes. None = all allowed. on_cost_update: Callback for cost updates. on_context_update: Callback for context usage updates. extra_middleware: Additional middleware to include. backend: Override the file storage backend (e.g., DockerSandbox). Takes precedence over `sandbox`. sandbox: Sandbox type: `"local"` or `"docker"`. When `"docker"`, creates a DockerSandbox with the working directory mounted at `/workspace`. Falls back to `config.sandbox`. sandbox_image: Docker image for the sandbox container. Falls back to `config.sandbox_image` (default: `python:3.12-slim`). sandbox_env_vars: Environment variables to inject into the Docker sandbox container. Falls back to `config.sandbox_env_vars`. Only applied when `sandbox="docker"`. Values are passed at container start-time via `RuntimeConfig` with `cache_image=False` so they are not baked permanently into a cached Docker image. sandbox_env_file: Path to a `.env` file whose variables are injected into the Docker sandbox container. Falls back to `config.sandbox_env_file`. Merged with `sandbox_env_vars`; explicit `sandbox_env_vars` take priority over file values. workspace: Named Docker workspace shared across threads. When set, the container persists between sessions so installed packages and any files outside the mounted volume survive restarts. Multiple threads (conversation histories) can share the same workspace. The actual Docker container name is `pydantic-deep-{dir_hash}-{workspace}`. include_skills: Whether to include the skills toolset. include_plan: Whether to include the planner subagent. include_memory: Whether to include persistent agent memory. include_subagents: Whether to include the subagent toolset. include_todo: Whether to include the todo toolset. include_local_context: Whether to include local context (git info, dir tree). Disable for Docker/sandbox backends where the root dir doesn't exist on host. context_discovery: Whether to auto-discover context files (AGENTS.md). config_path: Override config file path (for testing). session_id: Session identifier for per-session plans storage. extra_instructions: Additional instructions appended to the system prompt. skills_dir: Override skills directory path. When None, auto-discovers from `{working_dir}/.pydantic-deep/skills/`. periodic_reminder: Enable periodic task reminders. `None` uses the config default (`True` / `"llm"`). `True`/`False` overrides. reminder_mode: Generator for the reminder text. `"llm"` (default) uses `LLMReminderGenerator`. `"first"` re-states first user message (zero-cost). `"context"` uses a compact transcript (zero-cost). reminder_model: Model used by the `"llm"` reminder generator. Defaults to `config.reminder_model`, then falls back to the main model. Returns: Tuple of (agent, deps) ready for agent.run(). """ config = load_config(config_path) # Apply config defaults — explicit params override effective_model = model or config.model effective_working_dir = working_dir or config.working_dir effective_allow_list = shell_allow_list or config.shell_allow_list or None root = Path(effective_working_dir) if effective_working_dir else Path.cwd() # Resolve sandbox: explicit param > config effective_sandbox = sandbox or config.sandbox if effective_sandbox == "docker" and backend is None: from pydantic_ai_backends import DockerSandbox, RuntimeConfig file_env_vars: dict[str, str] = {} effective_env_file = ( sandbox_env_file if sandbox_env_file is not None else config.sandbox_env_file ) if effective_env_file: from dotenv import dotenv_values file_env_vars = { k: v for k, v in dotenv_values(effective_env_file).items() if v is not None } effective_env_vars = { **config.sandbox_env_vars, **file_env_vars, **(sandbox_env_vars or {}), } docker_kwargs: dict[str, Any] = { "volumes": {str(root.resolve()): "/workspace"}, "work_dir": "/workspace", } if effective_env_vars: docker_kwargs["runtime"] = RuntimeConfig( name="cli-sandbox", base_image=sandbox_image or config.sandbox_image, env_vars=effective_env_vars, cache_image=False, ) else: docker_kwargs["image"] = sandbox_image or config.sandbox_image # Named workspace → reusable container (packages + state persist between threads) # No workspace → ephemeral container (clean slate every time) if workspace: import hashlib dir_hash = hashlib.md5(str(root.resolve()).encode()).hexdigest()[:8] docker_kwargs["container_name"] = f"pydantic-deep-{dir_hash}-{workspace}" effective_backend: Any = DockerSandbox(**docker_kwargs) else: effective_backend = backend or LocalBackend(root_dir=root) hooks: list[Hook] = [] if effective_allow_list is not None: hooks.append(_make_shell_allow_list_hook(effective_allow_list)) middleware: list[Any] = [] if extra_middleware: middleware.extend(extra_middleware) # Forking (and other benchmark-profile flags) default OFF non-interactively; # resolve forking here so the prompt's Forking section matches the capability. _forking = forking if forking is not None else not non_interactive # When using Docker sandbox, the agent operates inside the container at /workspace instruction_root = "/workspace" if effective_sandbox == "docker" else str(root.resolve()) instructions = build_system_prompt( non_interactive=non_interactive, lean=lean, working_dir=instruction_root, forking=_forking, ) if extra_instructions: instructions += "\n\n" + extra_instructions # Add local context toolset (git info + directory tree) # Skipped for Docker/sandbox backends where root_dir doesn't exist on host local_context = None if include_local_context: from apps.cli.local_context import LocalContextToolset local_context = LocalContextToolset(root_dir=root) # Skills directories — searched in order, all matching dirs included: # 1. Bundled skills (shipped with CLI package) # 2. User-level skills (~/.pydantic-deep/skills/) # 3. Project-level skills (.pydantic-deep/skills/) # 4. Explicit override (--skills-dir flag) skill_dirs: list[str] = [] # Gate on the *resolved* skills setting (config default when the flag is # unset), mirroring `effective_skills` below. Gating on the raw # `include_skills` meant headless `pydantic-deep run` — which passes # `include_skills=None` — silently discovered no skill directories, so # skills never loaded despite being enabled by config. _skills_enabled = ( include_skills if include_skills is not None else config.include_skills ) and not lean if _skills_enabled: # Bundled skills (always available) bundled = Path(__file__).resolve().parent / "skills" if bundled.is_dir(): skill_dirs.append(str(bundled)) # User-level skills (home directory) user_skills = Path.home() / ".pydantic-deep" / "skills" if user_skills.is_dir(): skill_dirs.append(str(user_skills)) # Project-level skills (working directory) project_skills_dir = root / ".pydantic-deep" / "skills" if project_skills_dir.is_dir(): skill_dirs.append(str(project_skills_dir)) # Explicit override (highest priority — appended last) if skills_dir: sd = Path(skills_dir) if sd.is_dir(): skill_dirs.append(str(sd)) # In non-interactive mode: no approval needed, disable interactive features # (memory, plan, subagents). Skills stay ON — they're static # instructions that improve benchmark performance. if non_interactive: interrupt_on: dict[str, bool] | None = {"execute": False} else: # Build interrupt_on from config.approve_tools interrupt_on = ( {tool: True for tool in config.approve_tools} if config.approve_tools else None ) # Resolve feature flags: explicit param > config.toml > lean override _skills = include_skills if include_skills is not None else config.include_skills _plan = include_plan if include_plan is not None else config.include_plan _memory = include_memory if include_memory is not None else config.include_memory _subagents = include_subagents if include_subagents is not None else config.include_subagents _todo = include_todo if include_todo is not None else config.include_todo _context_disc = context_discovery if context_discovery is not None else config.context_discovery effective_skills = _skills if not lean else False effective_plan = _plan if not lean else False effective_subagents = _subagents if not lean else False effective_todo = _todo if not lean else False # Benchmark/automation profile: self-improvement, deferred tool search, and # cross-session memory add no value in a single-shot non-interactive run — # default them OFF there (each still overridable). `_forking` is resolved # earlier (needed for the prompt). _improve = include_improve if include_improve is not None else not non_interactive _tool_search = tool_search if tool_search is not None else config.tool_search effective_tool_search = _tool_search and not lean and not non_interactive effective_memory = _memory if (not lean and not non_interactive) else False _browser = include_browser if include_browser is not None else config.include_browser effective_browser = _browser if not lean else False _liteparse = include_liteparse if include_liteparse is not None else config.include_liteparse effective_liteparse = _liteparse if not lean else False # Model settings — explicit param > model_settings dict > non-interactive > config effective_model_settings: dict[str, Any] = {} if non_interactive: effective_model_settings["temperature"] = 0.0 if config.temperature is not None and "temperature" not in (model_settings or {}): effective_model_settings["temperature"] = config.temperature if config.reasoning_effort and "openai_reasoning_effort" not in (model_settings or {}): effective_model_settings["openai_reasoning_effort"] = config.reasoning_effort if model_settings: effective_model_settings.update(model_settings) # Explicit temperature param has highest priority if temperature is not None: effective_model_settings["temperature"] = temperature # Per-session plans directory (relative to backend root) if session_id: plans_dir = f".pydantic-deep/sessions/{session_id}/plans" else: plans_dir = ".pydantic-deep/plans" # Ensure session directory exists (for plans, messages.json) if session_id: from apps.cli.config import get_sessions_dir session_dir = get_sessions_dir() / session_id session_dir.mkdir(parents=True, exist_ok=True) # Build extra capabilities list (browser, future additions) extra_capabilities: list[Any] = [] if effective_browser: try: from pydantic_deep.features.browser import BrowserCapability effective_headless = ( browser_headless if browser_headless is not None else config.browser_headless ) extra_capabilities.append(BrowserCapability(headless=effective_headless)) except ImportError: import warnings warnings.warn( "BrowserCapability requires playwright. " "Install with: pip install 'pydantic-deep[browser]' && playwright install chromium", stacklevel=2, ) queue = MessageQueue() # MCP servers configured via `/mcp` (enabled + authenticated ones only). # Failures (missing optional dep, bad config) degrade to no MCP support. # `mcp_degraded` collects servers that turn out unreachable at runtime so the # TUI can tell the user *why* a server's tools are missing. mcp_servers: list[Any] = [] mcp_degraded: set[str] = set() if not lean: try: from apps.cli.mcp_store import build_mcp_servers_for_agent def _on_mcp_degraded(name: str, _reason: str) -> None: mcp_degraded.add(name) mcp_servers = build_mcp_servers_for_agent(on_degraded=_on_mcp_degraded) except Exception: mcp_servers = [] # Both go through the resolver: the `openai-compatible:` sentinel is a CLI # concept pydantic-ai can't infer, and a local endpoint is as valid a # fallback as it is a primary. model_for_agent = resolve_cli_model(effective_model, config) raw_fallback = fallback_model or config.fallback_model or None fallback_for_agent = resolve_cli_model(raw_fallback, config) if raw_fallback else None agent = create_deep_agent( model=model_for_agent, fallback_model=fallback_for_agent, instructions=instructions, backend=effective_backend, skill_directories=skill_dirs if effective_skills else None, interrupt_on=interrupt_on, model_settings=effective_model_settings or None, include_execute=True, include_filesystem=True, include_todo=effective_todo, include_plan=effective_plan, plans_dir=plans_dir, include_subagents=effective_subagents, include_builtin_subagents=effective_subagents, include_skills=effective_skills, # Memory (store in .pydantic-deep/main/MEMORY.md) include_memory=effective_memory, memory_dir=".pydantic-deep", # Context files (auto-discover AGENTS.md, SOUL.md) context_discovery=_context_disc if not lean else False, include_teams=(include_teams if include_teams is not None else config.include_teams), include_liteparse=effective_liteparse, include_improve=_improve, # Defer the situational tool surface so only the core loop loads upfront. tool_search=effective_tool_search, forking=( LiveForkCapability(test_command=_detect_fork_test_command(effective_backend)) if _forking else False ), # Web tools — explicit params override config web_search=( web_search if web_search is not None else (config.web_search if not lean else False) ), web_fetch=( web_fetch if web_fetch is not None else (config.web_fetch if not lean else False) ), thinking=( thinking if thinking is not None else (config.thinking_effort if not lean else False) ), # History persistence — per-session messages.json history_messages_path=( f".pydantic-deep/sessions/{session_id}/messages.json" if session_id else ".pydantic-deep/messages.json" ), context_manager=not lean, context_manager_max_tokens=None, # auto-detect from genai-prices on_context_update=on_context_update, on_before_compress=on_before_compress, on_after_compress=on_after_compress, summarization_model=summarization_model, eviction_token_limit=20_000, on_eviction=on_eviction, cost_tracking=True, on_cost_update=on_cost_update, output_style="concise" if not lean else None, hooks=hooks or None, middleware=middleware or None, toolsets=[local_context] if local_context else None, mcp_servers=mcp_servers or None, capabilities=extra_capabilities or None, # Message queue for mid-run steering and follow-up delivery message_queue=queue, periodic_reminder=_build_reminder_config( periodic_reminder, reminder_mode, config, on_reminder, # Inherit the resolved runtime model, not config.model (a different # provider than `-m` would crash on a missing key; the raw # `openai-compatible:...` string would break the LLM reminder generator). reminder_model or config.reminder_model or model_for_agent, ), ) # Extract context middleware for CLI commands (/compact, /context) context_mw = getattr(agent, "_context_middleware", None) task_mgr = getattr(agent, "_task_manager", None) deps = DeepAgentDeps( backend=effective_backend, context_middleware=context_mw, message_queue=queue, ) deps._task_manager = task_mgr # type: ignore[attr-defined] # Shared set the resilient MCP wrappers fill when a server is unreachable; # the chat screen surfaces these to the user after a run. deps.mcp_degraded = mcp_degraded # type: ignore[attr-defined] return agent, deps __all__ = ["create_cli_agent"]