feat(agent): support detached exec services

nanobot/agent/tools/shell.py

@@ -6,15 +6,17 @@ import asyncio
 import os
 import re
 import shutil
+import subprocess
 import sys
 import time
+import uuid
 from contextlib import suppress
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 from loguru import logger
-from pydantic import Field
+from pydantic import AliasChoices, Field
 
 from nanobot.agent.tools.base import Tool, tool_parameters
 from nanobot.agent.tools.context import current_request_session_key
@@ -62,6 +64,13 @@ class ExecToolConfig(Base):
     """Shell exec tool configuration."""
     enable: bool = True
     timeout: int = Field(default=60, ge=0)  # Hard timeout (s); 0 = no limit. Not capped by the per-call max.
+    allow_local_service_access: bool = Field(
+        default=False,
+        validation_alias=AliasChoices(
+            "allowLocalServiceAccess",
+            "allow_local_service_access",
+        ),
+    )  # allow shell commands to reach literal localhost/loopback services
     path_prepend: str = ""
     path_append: str = ""
     sandbox: str = ""
@@ -133,6 +142,16 @@ class _PreparedCommand:
             maximum=MAX_OUTPUT_CHARS,
             nullable=True,
         ),
+        detach=BooleanSchema(
+            description=(
+                "Run the command as a detached background process that can "
+                "survive after the agent finishes. Use for local servers, "
+                "dev servers, mock APIs, or other services that must remain "
+                "available for later commands or external verification."
+            ),
+            default=False,
+            nullable=True,
+        ),
     )
 )
 class ExecTool(Tool):
@@ -156,6 +175,7 @@ class ExecTool(Tool):
             working_dir=ctx.workspace,
             timeout=cfg.timeout,
             restrict_to_workspace=ctx.config.restrict_to_workspace,
+            allow_local_service_access=cfg.allow_local_service_access,
             webui_allow_local_service_access=ctx.config.webui_allow_local_service_access,
             sandbox=cfg.sandbox,
             path_prepend=cfg.path_prepend,
@@ -172,6 +192,7 @@ class ExecTool(Tool):
         deny_patterns: list[str] | None = None,
         allow_patterns: list[str] | None = None,
         restrict_to_workspace: bool = False,
+        allow_local_service_access: bool = False,
         webui_allow_local_service_access: bool = True,
         allow_local_preview_access: bool | None = None,
         sandbox: str = "",
@@ -204,6 +225,7 @@ class ExecTool(Tool):
         ]
         self.allow_patterns = allow_patterns or []
         self.restrict_to_workspace = restrict_to_workspace
+        self.allow_local_service_access = allow_local_service_access
         if allow_local_preview_access is not None:
             webui_allow_local_service_access = allow_local_preview_access
         self.webui_allow_local_service_access = webui_allow_local_service_access
@@ -243,8 +265,11 @@ class ExecTool(Tool):
             "Use -y or --yes flags to avoid interactive prompts. "
             "For long-running or interactive commands, pass yield_time_ms; "
             "if the command keeps running, exec returns a session_id that can "
-            "be polled or written to with write_stdin. Output is truncated at "
+            "be polled or written to with write_stdin. For services that "
-            "10 000 chars; timeout defaults to 60s."
+            "must remain available after you finish, pass detach=true instead "
+            "of yield_time_ms; detached output is written to a log file and "
+            "the tool returns a pid. Output is truncated at 10 000 chars; "
+            "timeout defaults to 60s."
         )
 
     @property
@@ -258,6 +283,7 @@ class ExecTool(Tool):
         login: bool | None = None, yield_time_ms: int | None = None,
         max_output_chars: int | None = None,
         max_output_tokens: int | None = None,
+        detach: bool | None = False,
         **kwargs: Any,
     ) -> str:
         command = command or cmd
@@ -271,6 +297,9 @@ class ExecTool(Tool):
         if isinstance(prepared, str):
             return prepared
 
+        if detach:
+            return await self._execute_detached(prepared)
+
         if yield_time_ms is not None:
             return await self._execute_session(prepared, yield_time_ms, max_output_chars)
 
@@ -387,6 +416,61 @@ class ExecTool(Tool):
         except Exception as exc:
             return f"Error executing command: {exc}"
 
+    async def _execute_detached(self, prepared: _PreparedCommand) -> str:
+        log_dir = Path(prepared.cwd) / ".nanobot" / "exec-logs"
+        try:
+            log_dir.mkdir(parents=True, exist_ok=True)
+            log_path = log_dir / f"detached-{uuid.uuid4().hex[:12]}.log"
+        except Exception as exc:
+            return f"Error preparing detached command log directory: {exc}"
+
+        log_handle = None
+        try:
+            log_handle = open(log_path, "ab", buffering=0)
+            process = await self._spawn(
+                prepared.command,
+                prepared.cwd,
+                prepared.env,
+                prepared.shell_program,
+                prepared.login,
+                stdout=log_handle,
+                stderr=log_handle,
+                start_new_session=not _IS_WINDOWS,
+                creationflags=(
+                    subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS
+                    if _IS_WINDOWS
+                    else 0
+                ),
+            )
+        except Exception as exc:
+            return f"Error starting detached command: {exc}"
+        finally:
+            if log_handle is not None:
+                with suppress(Exception):
+                    log_handle.close()
+
+        try:
+            exit_code = await asyncio.wait_for(process.wait(), timeout=0.2)
+        except asyncio.TimeoutError:
+            return (
+                "Detached process started.\n"
+                f"pid: {process.pid}\n"
+                f"cwd: {prepared.cwd}\n"
+                f"log: {log_path}\n"
+                "Poll the log or run a health check to verify the service is ready."
+            )
+
+        log_text = ""
+        with suppress(Exception):
+            log_text = log_path.read_text(encoding="utf-8", errors="replace")
+        if len(log_text) > 4000:
+            log_text = log_text[-4000:]
+        return (
+            f"Detached process exited immediately with code {exit_code}.\n"
+            f"log: {log_path}\n"
+            f"{log_text}"
+        )
+
     def _resolve_timeout(self, timeout: int | None) -> int | None:
         """Resolve the effective hard timeout in seconds (None = no limit).
 
@@ -508,6 +592,10 @@ class ExecTool(Tool):
         login: bool = False,
         *,
         stdin: int = asyncio.subprocess.DEVNULL,
+        stdout: Any = asyncio.subprocess.PIPE,
+        stderr: Any = asyncio.subprocess.PIPE,
+        start_new_session: bool = False,
+        creationflags: int = 0,
     ) -> asyncio.subprocess.Process:
         """Launch *command* in a platform-appropriate shell."""
         if _IS_WINDOWS:
@@ -515,18 +603,20 @@ class ExecTool(Tool):
                 return await asyncio.create_subprocess_exec(
                     "powershell", "-NoProfile", "-Command", command,
                     stdin=stdin,
-                    stdout=asyncio.subprocess.PIPE,
+                    stdout=stdout,
-                    stderr=asyncio.subprocess.PIPE,
+                    stderr=stderr,
                     cwd=cwd,
                     env=env,
+                    creationflags=creationflags,
                 )
             return await asyncio.create_subprocess_shell(
                 command,
                 stdin=stdin,
-                stdout=asyncio.subprocess.PIPE,
+                stdout=stdout,
-                stderr=asyncio.subprocess.PIPE,
+                stderr=stderr,
                 cwd=cwd,
                 env=env,
+                creationflags=creationflags,
             )
         shell_program = shell_program or shutil.which("bash") or "/bin/bash"
         args = [shell_program]
@@ -537,10 +627,11 @@ class ExecTool(Tool):
         return await asyncio.create_subprocess_exec(
             *args,
             stdin=stdin,
-            stdout=asyncio.subprocess.PIPE,
+            stdout=stdout,
-            stderr=asyncio.subprocess.PIPE,
+            stderr=stderr,
             cwd=cwd,
             env=env,
+            start_new_session=start_new_session,
         )
 
     @staticmethod
@@ -658,11 +749,12 @@ class ExecTool(Tool):
                 return "Error: Command blocked by allowlist filter (not in allowlist)"
 
         from nanobot.security.network import contains_internal_url
+        allow_loopback = self.allow_local_service_access or current_scope_allows_loopback(
+            enabled=self.webui_allow_local_service_access,
+        )
         if contains_internal_url(
             cmd,
-            allow_loopback=current_scope_allows_loopback(
+            allow_loopback=allow_loopback,
-                enabled=self.webui_allow_local_service_access,
-            ),
         ):
             # The runner turns this marker into a non-retryable security hint.
             return "Error: Command blocked by safety guard (internal/private URL detected)"

nanobot/agent/verification_state.py

@@ -43,6 +43,19 @@ _TEST_COMMAND_RE = re.compile(
     r"\bmake\s+(?:[^;&|]*\s+)?test\b"
     r")"
 )
+_ARTIFACT_CHECK_COMMAND_RE = re.compile(
+    r"(?ix)"
+    r"("
+    r"\bcmp\b|"
+    r"\bdiff\b|"
+    r"\bsha(?:1|224|256|384|512)?sum\b|"
+    r"\bmd5sum\b|"
+    r"\bgcc\b.*(?:&&|;).*\./|"
+    r"\bclang\b.*(?:&&|;).*\./|"
+    r"\bpython3?\b.*<<['\"]?PY\b.*\bassert\b"
+    r")"
+)
+_COMPARISON_COMMAND_RE = re.compile(r"(?i)\b(?:cmp|diff)\b")
 _FAILURE_RE = re.compile(
     r"(?im)"
     r"("
@@ -66,6 +79,20 @@ _SUCCESS_RE = re.compile(
     r"\bExit code:\s*0\b"
     r")"
 )
+_ARTIFACT_SUCCESS_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b(?:cmp|diff|test|verify)_exit:\s*0\b|"
+    r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*0\s*$"
+    r")"
+)
+_ARTIFACT_FAILURE_RE = re.compile(
+    r"(?im)"
+    r"("
+    r"\b(?:cmp|diff|test|verify)_exit:\s*[1-9]\d*\b|"
+    r"^\s*(?:cmp|diff|match|same|image|ppm|stdout|stderr|out|err)[\w.-]*:\s*[1-9]\d*\s*$"
+    r")"
+)
 _FAILED_TEST_RE = re.compile(r"(?m)^FAILED\s+([^\s]+)")
 _PYTEST_SHORT_RE = re.compile(r"(?m)^_{3,}\s+([A-Za-z0-9_./:-]+)\s+_{3,}$")
 _ERROR_LINE_RE = re.compile(
@@ -98,13 +125,28 @@ def analyze_verification_result(
 
     command = " ".join((command or "").split())
     looks_like_test_command = bool(_TEST_COMMAND_RE.search(command))
+    looks_like_artifact_check = bool(_ARTIFACT_CHECK_COMMAND_RE.search(command))
+    looks_like_comparison_command = bool(_COMPARISON_COMMAND_RE.search(command))
+    looks_like_verification = looks_like_test_command or looks_like_artifact_check
     failure_seen = bool(_FAILURE_RE.search(output))
     success_seen = bool(_SUCCESS_RE.search(output))
+    artifact_success_seen = bool(_ARTIFACT_SUCCESS_RE.search(output)) and (
+        looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*0\b", output, flags=re.I))
+    )
+    artifact_failure_seen = bool(_ARTIFACT_FAILURE_RE.search(output)) and (
+        looks_like_comparison_command or bool(re.search(r"\b(?:test|verify)_exit:\s*[1-9]\d*\b", output, flags=re.I))
+    )
 
     if not looks_like_test_command and not failure_seen:
+        if not (looks_like_artifact_check and artifact_success_seen and exit_code == 0):
             return None
 
-    if (timed_out and looks_like_test_command) or (exit_code not in (None, 0) and (looks_like_test_command or failure_seen)) or failure_seen:
+    if (
+        (timed_out and looks_like_verification)
+        or (exit_code not in (None, 0) and (looks_like_verification or failure_seen))
+        or failure_seen
+        or artifact_failure_seen
+    ):
         return VerificationAnalysis(
             status="failed",
             command=command,
@@ -122,6 +164,13 @@ def analyze_verification_result(
             exit_code=exit_code,
         )
 
+    if looks_like_artifact_check and exit_code == 0 and artifact_success_seen:
+        return VerificationAnalysis(
+            status="passed",
+            command=command,
+            exit_code=exit_code,
+        )
+
     return None
 
 

tests/agent/test_verification_state.py

@@ -71,3 +71,105 @@ Exit code: 127
     assert analysis is not None
     assert analysis.status == "failed"
     assert any("command not found" in item for item in analysis.primary_errors)
+
+
+def test_analyze_artifact_comparison_success_records_pass():
+    output = """\
+run_exit:0
+0d115b98  /app/image.ppm
+0d115b98  /tmp/orig.ppm
+cmp_exit:0
+      7      21    1024
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "cd /usr/bin && gcc -static -o /app/reversed_final /app/mystery.c -lm "
+            "&& (cd /app && ./reversed_final >/tmp/final_out 2>/tmp/final_err); "
+            "sha256sum /app/image.ppm /tmp/orig.ppm; "
+            "cmp -s /app/image.ppm /tmp/orig.ppm; echo cmp_exit:$?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+    assert append_verification_feedback("ok", analysis) == "ok"
+
+
+def test_analyze_plain_checksum_without_success_marker_is_ignored():
+    analysis = analyze_verification_result(
+        command="sha256sum /app/image.ppm /tmp/orig.ppm",
+        output="0d115b98  /app/image.ppm\n0d115b98  /tmp/orig.ppm\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is None
+
+
+def test_analyze_named_comparison_markers_record_pass():
+    output = """\
+ppm:0
+stderr:0
+stdout:0
+      4      26    1011
+1821 mystery.c
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "gcc -static -O2 -o reversed mystery.c -lm\n"
+            "./reversed > vrout.txt 2> vrerr.txt\n"
+            "cp image.ppm rev.ppm\n"
+            "./mystery > voout.txt 2> voerr.txt\n"
+            "cmp image.ppm rev.ppm\n"
+            "printf 'ppm:%s\\n' $?\n"
+            "cmp voerr.txt vrerr.txt\n"
+            "printf 'stderr:%s\\n' $?\n"
+            "cmp voout.txt vrout.txt\n"
+            "printf 'stdout:%s\\n' $?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "passed"
+
+
+def test_analyze_named_comparison_marker_failure_records_failed():
+    output = """\
+ppm:0
+stderr:1
+stdout:0
+
+Exit code: 0
+"""
+
+    analysis = analyze_verification_result(
+        command=(
+            "cmp image.ppm rev.ppm; printf 'ppm:%s\\n' $?; "
+            "cmp voerr.txt vrerr.txt; printf 'stderr:%s\\n' $?; "
+            "cmp voout.txt vrout.txt; printf 'stdout:%s\\n' $?"
+        ),
+        output=output,
+        exit_code=0,
+    )
+
+    assert analysis is not None
+    assert analysis.status == "failed"
+
+
+def test_analyze_plain_run_status_marker_without_comparison_is_ignored():
+    analysis = analyze_verification_result(
+        command="gcc -static -O2 -o reversed mystery.c -lm && ./reversed",
+        output="rc:0\nExit code: 0",
+        exit_code=0,
+    )
+
+    assert analysis is None

tests/config/test_config_migration.py

@@ -246,3 +246,16 @@ def test_load_config_accepts_legacy_local_preview_access(tmp_path) -> None:
     config = load_config(config_path)
 
     assert config.tools.webui_allow_local_service_access is False
+
+
+def test_load_config_accepts_exec_local_service_access(tmp_path) -> None:
+    config_path = tmp_path / "config.json"
+    config_path.write_text(
+        json.dumps({"tools": {"exec": {"allowLocalServiceAccess": True}}}),
+        encoding="utf-8",
+    )
+
+    config = load_config(config_path)
+
+    assert config.tools.exec.allow_local_service_access is True
+    assert not hasattr(config.tools, "allow_local_service_access")

tests/tools/test_exec_security.py

@@ -9,7 +9,11 @@ from unittest.mock import patch
 import pytest
 
 from nanobot.agent.tools.shell import ExecTool
-from nanobot.security.workspace_access import bind_workspace_scope, build_workspace_scope, reset_workspace_scope
+from nanobot.security.workspace_access import (
+    bind_workspace_scope,
+    build_workspace_scope,
+    reset_workspace_scope,
+)
 
 
 def _fake_resolve_private(hostname, port, family=0, type_=0):
@@ -68,6 +72,21 @@ def test_exec_core_full_workspace_scope_blocks_loopback(tmp_path):
     assert "internal/private" in error
 
 
+def test_exec_explicit_local_service_access_allows_loopback(tmp_path):
+    tool = ExecTool(working_dir=str(tmp_path), allow_local_service_access=True)
+    with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_localhost):
+        error = tool._guard_command("curl http://localhost:8765/", str(tmp_path))
+    assert error is None
+
+
+def test_exec_explicit_local_service_access_still_blocks_metadata(tmp_path):
+    tool = ExecTool(working_dir=str(tmp_path), allow_local_service_access=True)
+    with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve_private):
+        error = tool._guard_command("curl http://169.254.169.254/latest/meta-data/", str(tmp_path))
+    assert error is not None
+    assert "internal/private" in error
+
+
 def test_exec_full_workspace_scope_blocks_loopback_when_local_service_disabled(tmp_path):
     tool = ExecTool(working_dir=str(tmp_path), webui_allow_local_service_access=False)
     scope = build_workspace_scope(tmp_path, "full", source_channel="websocket")

tests/tools/test_exec_session_tools.py

@@ -104,6 +104,42 @@ def test_exec_one_shot_accepts_max_output_tokens_alias(tmp_path):
     assert "Exit code: 0" in result
 
 
+def test_exec_detach_starts_background_process(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        ready_path = tmp_path / "ready.txt"
+        command = _python_command(
+            "import pathlib, time; "
+            "pathlib.Path('ready.txt').write_text('ok'); "
+            "time.sleep(0.6)"
+        )
+        result = await tool.execute(command=command, detach=True)
+        for _ in range(20):
+            if ready_path.exists():
+                break
+            await asyncio.sleep(0.05)
+        return result
+
+    result = asyncio.run(run())
+
+    assert "Detached process started." in result
+    assert "pid:" in result
+    assert "log:" in result
+    assert (tmp_path / "ready.txt").read_text() == "ok"
+
+
+def test_exec_detach_reports_immediate_exit(tmp_path):
+    async def run() -> str:
+        tool = ExecTool(working_dir=str(tmp_path), timeout=5)
+        command = _python_command("print('boom'); raise SystemExit(7)")
+        return await tool.execute(command=command, detach=True)
+
+    result = asyncio.run(run())
+
+    assert "Detached process exited immediately with code 7" in result
+    assert "boom" in result
+
+
 def test_exec_long_output_summary_includes_failure_signals(tmp_path):
     async def run() -> str:
         tool = ExecTool(working_dir=str(tmp_path), timeout=5)