feat(gateway): add background and service controls

README.md

@@ -362,6 +362,7 @@ nanobot gateway
 ```
 
 Leave that terminal open, then open `http://127.0.0.1:8765` in your browser. Enter the WebUI password you set in the wizard, then send your first message there.
+Prefer not to keep a terminal open? Use `nanobot gateway --background`, then manage it with `nanobot gateway status`, `logs`, `restart`, and `stop`.
 
 For manual or terminal-only setup, test one CLI message:
 
@@ -417,6 +418,8 @@ Merge this block into your existing config:
 nanobot gateway
 ```
 
+Use `nanobot gateway --background` for a local background process you can manage later with `nanobot gateway status`, `logs`, `restart`, and `stop`.
+
 **3. Open the WebUI**
 
 Visit [`http://127.0.0.1:8765`](http://127.0.0.1:8765) in your browser. To open it from another device on your LAN, see [WebUI docs -> LAN access](./docs/webui.md#lan-access).

docs/cli-reference.md

@@ -12,7 +12,7 @@ Use this page when you know what you want to run and need the command shape. For
 | Check config without calling a model | `nanobot status` | Reads the default config and summarizes the active model/provider |
 | Send one test message | `nanobot agent -m "Hello!"` | First proof that install, config, provider, model, and workspace all work |
 | Chat in the terminal | `nanobot agent` | Interactive local chat; exit with `exit`, `/exit`, `:q`, or `Ctrl+D` |
-| Use WebUI or chat apps | `nanobot gateway` | Keep this terminal running while those surfaces are in use |
+| Use WebUI or chat apps | `nanobot gateway` | Keep this terminal running, or use `nanobot gateway --background` |
 | Serve an OpenAI-compatible API | `nanobot serve` | Starts `/v1/chat/completions`, `/v1/models`, and `/health` |
 | Check chat channel setup | `nanobot channels status` | Useful before starting `nanobot gateway` |
 | Log in to QR/OAuth-style channels | `nanobot channels login <channel>` | Used by channels such as WhatsApp and WeChat |
@@ -46,7 +46,9 @@ nanobot gateway --verbose
 nanobot serve --verbose
 ```
 
-Long-running commands keep working until you stop them. Press `Ctrl+C` in that terminal to stop `nanobot gateway` or `nanobot serve`.
+Long-running commands keep working until you stop them. Press `Ctrl+C` in that terminal
+to stop foreground `nanobot gateway` or `nanobot serve`. If you started the gateway
+with `--background`, use `nanobot gateway stop`.
 
 ## Setup
 
@@ -79,15 +81,38 @@ Interactive mode exits with `exit`, `quit`, `/exit`, `/quit`, `:q`, or `Ctrl+D`.
 
 ## Gateway
 
-`nanobot gateway` starts enabled chat channels, WebUI/WebSocket when configured, cron-backed system jobs, Dream, heartbeat, and the health endpoint.
+`nanobot gateway` starts enabled chat channels, WebUI/WebSocket when configured, cron-backed system jobs, Dream, heartbeat, and the health endpoint. By default it runs in the foreground, which keeps existing scripts and terminal workflows unchanged. Use `--background` when you want a local macOS, Linux, or Windows process that you can manage from the CLI.
 
 | Command | Description |
 |---|---|
-| `nanobot gateway` | Start the gateway with config defaults |
+| `nanobot gateway` | Start the gateway in the foreground with config defaults |
 | `nanobot gateway --verbose` | Show verbose runtime output |
 | `nanobot gateway --port <port>` | Override `gateway.port` for the health endpoint |
 | `nanobot gateway --workspace <path>` | Override workspace |
 | `nanobot gateway --config <path>` | Use a specific config file |
+| `nanobot gateway --background` | Start the gateway as a background process |
+| `nanobot gateway status` | Show the recorded background gateway PID, state file, and log file |
+| `nanobot gateway logs --no-follow` | Print recent background gateway logs and exit |
+| `nanobot gateway logs` | Follow background gateway logs |
+| `nanobot gateway restart` | Restart the recorded background gateway with the current config |
+| `nanobot gateway stop` | Stop the recorded background gateway |
+| `nanobot gateway install-service` | Install a systemd user service or macOS LaunchAgent |
+| `nanobot gateway install-service --dry-run` | Preview the generated service file and system commands |
+| `nanobot gateway uninstall-service` | Remove the installed system service |
+
+For custom instances, pass the same selector flags to management commands:
+
+```bash
+nanobot gateway --background --config ./bot-a/config.json --workspace ./bot-a/workspace
+nanobot gateway status --config ./bot-a/config.json --workspace ./bot-a/workspace
+nanobot gateway stop --config ./bot-a/config.json --workspace ./bot-a/workspace
+nanobot gateway install-service --config ./bot-a/config.json --workspace ./bot-a/workspace --name bot-a
+```
+
+`--background` is a lightweight detached process. `install-service` is for
+login/startup integration: Linux uses a systemd user service; macOS uses a
+LaunchAgent plist. System services run the foreground gateway under the OS
+supervisor rather than nesting another background process.
 
 Default health endpoint:
 

docs/deployment.md

@@ -106,48 +106,41 @@ docker run -v ~/.nanobot:/home/nanobot/.nanobot --rm nanobot status
 
 Run the gateway as a systemd user service so it starts automatically and restarts on failure.
 
-**1. Find the nanobot binary path:**
+Preview the generated unit first:
 
 ```bash
-which nanobot   # e.g. /home/user/.local/bin/nanobot
+nanobot gateway install-service --manager systemd --dry-run
 ```
 
-**2. Create the service file** at `~/.config/systemd/user/nanobot-gateway.service` (replace `ExecStart` path if needed):
+Install, enable, and start it:
-
-```ini
-[Unit]
-Description=Nanobot Gateway
-After=network.target
-
-[Service]
-Type=simple
-ExecStart=%h/.local/bin/nanobot gateway
-Restart=always
-RestartSec=10
-NoNewPrivileges=yes
-ProtectSystem=strict
-ReadWritePaths=%h
-
-[Install]
-WantedBy=default.target
-```
-
-**3. Enable and start:**
 
 ```bash
-systemctl --user daemon-reload
+nanobot gateway install-service --manager systemd
-systemctl --user enable --now nanobot-gateway
 ```
 
-**Common operations:**
+For a custom instance, pass the same config/workspace selector you use to run the gateway:
+
+```bash
+nanobot gateway install-service \
+  --manager systemd \
+  --name nanobot-telegram \
+  --config ~/.nanobot-telegram/config.json \
+  --workspace ~/.nanobot-telegram/workspace
+```
+
+Common operations:
 
 ```bash
 systemctl --user status nanobot-gateway        # check status
 systemctl --user restart nanobot-gateway       # restart after config changes
 journalctl --user -u nanobot-gateway -f        # follow logs
+nanobot gateway uninstall-service --manager systemd
 ```
 
-If you edit the `.service` file itself, run `systemctl --user daemon-reload` before restarting.
+The installer writes `~/.config/systemd/user/nanobot-gateway.service`, runs
+`systemctl --user daemon-reload`, enables the unit, and restarts it. It uses the
+current Python executable with `python -m nanobot gateway --foreground`, so the
+service runs in the same environment you used to install nanobot.
 
 > **Note:** User services only run while you are logged in. To keep the gateway running after logout, enable lingering:
 >
@@ -159,70 +152,38 @@ If you edit the `.service` file itself, run `systemctl --user daemon-reload` bef
 
 Use a LaunchAgent when you want `nanobot gateway` to stay online after you log in, without keeping a terminal open.
 
-**1. Get the absolute `nanobot` path:**
+Preview the generated plist first:
 
 ```bash
-which nanobot   # e.g. /Users/youruser/.local/bin/nanobot
+nanobot gateway install-service --manager launchd --dry-run
 ```
 
-Use that exact path in the plist. It keeps the Python environment from your install method.
+Install, load, enable, and start it:
-
-**2. Create `~/Library/LaunchAgents/ai.nanobot.gateway.plist`:**
-
-```xml
-<?xml version="1.0" encoding="UTF-8"?>
-<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
-<plist version="1.0">
-<dict>
-  <key>Label</key>
-  <string>ai.nanobot.gateway</string>
-
-  <key>ProgramArguments</key>
-  <array>
-    <string>/Users/youruser/.local/bin/nanobot</string>
-    <string>gateway</string>
-    <string>--workspace</string>
-    <string>/Users/youruser/.nanobot/workspace</string>
-  </array>
-
-  <key>WorkingDirectory</key>
-  <string>/Users/youruser/.nanobot/workspace</string>
-
-  <key>RunAtLoad</key>
-  <true/>
-
-  <key>KeepAlive</key>
-  <dict>
-    <key>SuccessfulExit</key>
-    <false/>
-  </dict>
-
-  <key>StandardOutPath</key>
-  <string>/Users/youruser/.nanobot/logs/gateway.log</string>
-
-  <key>StandardErrorPath</key>
-  <string>/Users/youruser/.nanobot/logs/gateway.error.log</string>
-</dict>
-</plist>
-```
-
-**3. Load and start it:**
 
 ```bash
-mkdir -p ~/Library/LaunchAgents ~/.nanobot/logs
+nanobot gateway install-service --manager launchd
-launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/ai.nanobot.gateway.plist
-launchctl enable gui/$(id -u)/ai.nanobot.gateway
-launchctl kickstart -k gui/$(id -u)/ai.nanobot.gateway
 ```
 
-**Common operations:**
+For a custom instance:
+
+```bash
+nanobot gateway install-service \
+  --manager launchd \
+  --name nanobot-telegram \
+  --config ~/.nanobot-telegram/config.json \
+  --workspace ~/.nanobot-telegram/workspace
+```
+
+Common operations:
 
 ```bash
 launchctl list | grep ai.nanobot.gateway
-launchctl kickstart -k gui/$(id -u)/ai.nanobot.gateway   # restart
+launchctl kickstart -k gui/$(id -u)/ai.nanobot.gateway
-launchctl bootout gui/$(id -u) ~/Library/LaunchAgents/ai.nanobot.gateway.plist
+nanobot gateway uninstall-service --manager launchd
 ```
 
-After editing the plist, run `launchctl bootout ...` and `launchctl bootstrap ...` again.
+The installer writes `~/Library/LaunchAgents/ai.nanobot.gateway.plist`, uses the
+current Python executable with `python -m nanobot gateway --foreground`, and
+writes LaunchAgent logs under `~/.nanobot/logs/`.
 
 > **Note:** if startup fails with "address already in use", stop the manually started `nanobot gateway` process first.

nanobot/cli/commands.py

@@ -50,6 +50,7 @@ from rich.text import Text  # noqa: E402
 
 from nanobot import __logo__, __version__  # noqa: E402
 from nanobot.agent.loop import AgentLoop  # noqa: E402
+from nanobot.cli.gateway import create_gateway_app  # noqa: E402
 from nanobot.cli.stream import StreamRenderer, ThinkingSpinner  # noqa: E402
 from nanobot.config.paths import get_workspace_path, is_default_workspace  # noqa: E402
 from nanobot.config.schema import Config  # noqa: E402
@@ -714,32 +715,6 @@ def serve(
 # ============================================================================
 
 
-@app.command()
-def gateway(
-    port: int | None = typer.Option(None, "--port", "-p", help="Gateway port"),
-    workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
-    verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
-    config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
-):
-    """Start the nanobot gateway."""
-    if verbose:
-        logger.remove(_log_handler_id)
-        logger.add(
-            sys.stderr,
-            format=(
-                "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
-                "<level>{level: <5}</level> | "
-                "<cyan>{extra[channel]}</cyan> | "
-                "<level>{message}</level>"
-            ),
-            level="DEBUG",
-            colorize=None,
-            filter=lambda record: record["extra"].setdefault("channel", "-") or True,
-        )
-    cfg = _load_runtime_config(config, workspace)
-    _run_gateway(cfg, port=port)
-
-
 def _run_gateway(
     config: Config,
     *,
@@ -1163,6 +1138,17 @@ def _run_gateway(
     asyncio.run(run())
 
 
+app.add_typer(
+    create_gateway_app(
+        console=console,
+        log_handler_id=_log_handler_id,
+        load_runtime_config=_load_runtime_config,
+        run_gateway=_run_gateway,
+    ),
+    name="gateway",
+)
+
+
 # ============================================================================
 # Agent Commands
 # ============================================================================

nanobot/cli/gateway.py

@@ -0,0 +1,291 @@
+"""Typer commands for foreground and background gateway control."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+import typer
+from loguru import logger
+from rich.console import Console
+
+from nanobot.config.schema import Config
+from nanobot.gateway import (
+    GatewayRuntime,
+    GatewayRuntimePaths,
+    GatewayStartOptions,
+    GatewayStatus,
+)
+from nanobot.gateway.service import (
+    GatewayServiceInstaller,
+    GatewayServiceOptions,
+    GatewayServiceResult,
+    ServiceManagerKind,
+)
+
+RuntimeConfigLoader = Callable[[str | None, str | None], Config]
+GatewayRunner = Callable[..., None]
+GatewayRuntimeFactory = Callable[..., Any]
+GatewayServiceFactory = Callable[[], Any]
+
+
+def create_gateway_app(
+    *,
+    console: Console,
+    log_handler_id: int,
+    load_runtime_config: RuntimeConfigLoader,
+    run_gateway: GatewayRunner,
+    runtime_factory: GatewayRuntimeFactory | None = None,
+    service_factory: GatewayServiceFactory | None = None,
+) -> typer.Typer:
+    gateway_app = typer.Typer(
+        help="Start and manage the nanobot gateway.",
+        invoke_without_command=True,
+        no_args_is_help=False,
+    )
+
+    def configure_logging(verbose: bool) -> None:
+        if not verbose:
+            return
+        logger.remove(log_handler_id)
+        logger.add(
+            sys.stderr,
+            format=(
+                "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
+                "<level>{level: <5}</level> | "
+                "<cyan>{extra[channel]}</cyan> | "
+                "<level>{message}</level>"
+            ),
+            level="DEBUG",
+            colorize=None,
+            filter=lambda record: record["extra"].setdefault("channel", "-") or True,
+        )
+
+    def runtime_for_instance(*, workspace: str | None = None, config: str | None = None):
+        if runtime_factory is not None:
+            return runtime_factory(workspace=workspace, config=config)
+        config_path = str(Path(config).expanduser().resolve(strict=False)) if config else None
+        workspace_path = str(Path(workspace).expanduser().resolve(strict=False)) if workspace else None
+        data_dir = Path(config_path).parent if config_path else None
+        return GatewayRuntime(
+            paths=GatewayRuntimePaths.for_instance(
+                data_dir=data_dir,
+                workspace=workspace_path,
+                config_path=config_path,
+            )
+        )
+
+    def service_installer():
+        return service_factory() if service_factory is not None else GatewayServiceInstaller()
+
+    def start_options(
+        *,
+        port: int | None,
+        verbose: bool,
+        workspace: str | None,
+        config: str | None,
+    ) -> GatewayStartOptions:
+        cfg = load_runtime_config(config, workspace)
+        resolved_config = str(Path(config).expanduser().resolve()) if config else None
+        resolved_workspace = str(Path(workspace).expanduser().resolve(strict=False)) if workspace else None
+        return GatewayStartOptions(
+            port=port if port is not None else cfg.gateway.port,
+            verbose=verbose,
+            workspace=resolved_workspace,
+            config_path=resolved_config,
+        )
+
+    def print_status(status: GatewayStatus) -> None:
+        console.print(f"Running: {'yes' if status.running else 'no'}")
+        console.print(f"Reason: {status.reason}")
+        if status.pid is not None:
+            console.print(f"PID: {status.pid}")
+        if status.port is not None:
+            console.print(f"Port: {status.port}")
+        if status.started_at is not None:
+            console.print(f"Started At: {status.started_at}")
+        console.print(f"State: {status.state_path}")
+        console.print(f"Logs: {status.log_path}")
+
+    def print_service_result(result: GatewayServiceResult) -> None:
+        console.print(f"Manager: {result.manager}")
+        if result.path is not None:
+            console.print(f"Path: {result.path}")
+        if result.commands:
+            console.print("Commands:")
+            for command in result.commands:
+                console.print("  " + " ".join(command))
+        if result.content is not None:
+            console.print()
+            console.print(result.content)
+
+    @gateway_app.callback(invoke_without_command=True)
+    def gateway(
+        ctx: typer.Context,
+        port: int | None = typer.Option(None, "--port", "-p", help="Gateway port"),
+        workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
+        verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
+        config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
+        foreground: bool = typer.Option(False, "--foreground", help="Run in the foreground"),
+        background: bool = typer.Option(False, "--background", help="Start as a background process"),
+    ) -> None:
+        """Start the nanobot gateway."""
+        if ctx.invoked_subcommand is not None:
+            return
+        if foreground and background:
+            console.print("[red]Error: --foreground and --background cannot be used together.[/red]")
+            raise typer.Exit(1)
+        if background:
+            runtime = runtime_for_instance(workspace=workspace, config=config)
+            result = runtime.start_background(
+                start_options(
+                    port=port,
+                    verbose=verbose,
+                    workspace=workspace,
+                    config=config,
+                )
+            )
+            if result.ok:
+                console.print("[green]Gateway started in the background.[/green]")
+                print_status(result.status)
+                return
+            console.print(f"[yellow]Gateway was not started: {result.message}[/yellow]")
+            print_status(result.status)
+            raise typer.Exit(1)
+
+        configure_logging(verbose)
+        cfg = load_runtime_config(config, workspace)
+        run_gateway(cfg, port=port)
+
+    @gateway_app.command("status")
+    def gateway_status(
+        workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
+        config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
+    ) -> None:
+        """Show the background gateway status."""
+        print_status(runtime_for_instance(workspace=workspace, config=config).status())
+
+    @gateway_app.command("logs")
+    def gateway_logs(
+        tail: int = typer.Option(200, "--tail", help="Number of recent lines to show"),
+        follow: bool = typer.Option(True, "--follow/--no-follow", help="Follow new log output"),
+        workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
+        config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
+    ) -> None:
+        """Show background gateway logs."""
+        runtime = runtime_for_instance(workspace=workspace, config=config)
+        if follow:
+            raise typer.Exit(runtime.follow_logs(tail=tail))
+        lines = runtime.read_log_tail(tail=tail)
+        if not lines:
+            console.print("[dim]No gateway log output available yet.[/dim]")
+            return
+        for line in lines:
+            console.print(line)
+
+    @gateway_app.command("stop")
+    def gateway_stop(
+        timeout: int = typer.Option(20, "--timeout", help="Stop timeout in seconds"),
+        workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
+        config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
+    ) -> None:
+        """Stop the background gateway."""
+        result = runtime_for_instance(workspace=workspace, config=config).stop(timeout_s=timeout)
+        if result.ok:
+            console.print("[green]Gateway stopped.[/green]")
+        else:
+            console.print(f"[yellow]Gateway was not stopped: {result.message}[/yellow]")
+        print_status(result.status)
+        if not result.ok and result.message != "gateway_not_running":
+            raise typer.Exit(1)
+
+    @gateway_app.command("restart")
+    def gateway_restart(
+        port: int | None = typer.Option(None, "--port", "-p", help="Gateway port"),
+        workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
+        verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
+        config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
+        timeout: int = typer.Option(20, "--timeout", help="Restart timeout in seconds"),
+    ) -> None:
+        """Restart the background gateway."""
+        runtime = runtime_for_instance(workspace=workspace, config=config)
+        result = runtime.restart(
+            start_options(
+                port=port,
+                verbose=verbose,
+                workspace=workspace,
+                config=config,
+            ),
+            timeout_s=timeout,
+        )
+        if result.ok:
+            console.print("[green]Gateway restarted in the background.[/green]")
+            print_status(result.status)
+            return
+        console.print(f"[red]Gateway restart failed: {result.message}[/red]")
+        print_status(result.status)
+        raise typer.Exit(1)
+
+    @gateway_app.command("install-service")
+    def gateway_install_service(
+        port: int | None = typer.Option(None, "--port", "-p", help="Gateway port"),
+        workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
+        verbose: bool = typer.Option(False, "--verbose", "-v", help="Verbose output"),
+        config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
+        name: str = typer.Option("nanobot-gateway", "--name", help="Service name"),
+        manager: ServiceManagerKind = typer.Option("auto", "--manager", help="auto, systemd, or launchd"),
+        enable: bool = typer.Option(True, "--enable/--no-enable", help="Enable the service after writing it"),
+        start_now: bool = typer.Option(True, "--start/--no-start", help="Start the service after writing it"),
+        dry_run: bool = typer.Option(False, "--dry-run", help="Print generated service without installing"),
+    ) -> None:
+        """Install a systemd user service or macOS LaunchAgent for the gateway."""
+        options = GatewayServiceOptions(
+            start=start_options(port=port, verbose=verbose, workspace=workspace, config=config),
+            name=name,
+            manager=manager,
+            enable=enable,
+            start_now=start_now,
+        )
+        try:
+            result = service_installer().install(options, dry_run=dry_run)
+        except subprocess.CalledProcessError as exc:
+            console.print(f"[red]Service install failed while running: {' '.join(exc.cmd)}[/red]")
+            raise typer.Exit(exc.returncode or 1) from exc
+        except OSError as exc:
+            console.print(f"[red]Service install failed: {exc}[/red]")
+            raise typer.Exit(1) from exc
+        if result.ok:
+            console.print("[green]Gateway service installed.[/green]" if not dry_run else "[green]Gateway service dry run.[/green]")
+            print_service_result(result)
+            return
+        console.print(f"[red]Gateway service was not installed: {result.message}[/red]")
+        print_service_result(result)
+        raise typer.Exit(1)
+
+    @gateway_app.command("uninstall-service")
+    def gateway_uninstall_service(
+        name: str = typer.Option("nanobot-gateway", "--name", help="Service name"),
+        manager: ServiceManagerKind = typer.Option("auto", "--manager", help="auto, systemd, or launchd"),
+        dry_run: bool = typer.Option(False, "--dry-run", help="Print actions without uninstalling"),
+    ) -> None:
+        """Uninstall the system gateway service."""
+        try:
+            result = service_installer().uninstall(name=name, manager=manager, dry_run=dry_run)
+        except subprocess.CalledProcessError as exc:
+            console.print(f"[red]Service uninstall failed while running: {' '.join(exc.cmd)}[/red]")
+            raise typer.Exit(exc.returncode or 1) from exc
+        except OSError as exc:
+            console.print(f"[red]Service uninstall failed: {exc}[/red]")
+            raise typer.Exit(1) from exc
+        if result.ok:
+            console.print("[green]Gateway service uninstalled.[/green]" if not dry_run else "[green]Gateway service uninstall dry run.[/green]")
+            print_service_result(result)
+            return
+        console.print(f"[red]Gateway service was not uninstalled: {result.message}[/red]")
+        print_service_result(result)
+        raise typer.Exit(1)
+
+    return gateway_app

nanobot/gateway/__init__.py

@@ -0,0 +1,19 @@
+"""Lightweight background runtime for the nanobot gateway."""
+
+from nanobot.gateway.runtime import (
+    GatewayRuntime,
+    GatewayRuntimePaths,
+    GatewayStartOptions,
+    GatewayStatus,
+    RuntimeResult,
+    build_gateway_command,
+)
+
+__all__ = [
+    "GatewayRuntime",
+    "GatewayRuntimePaths",
+    "GatewayStartOptions",
+    "GatewayStatus",
+    "RuntimeResult",
+    "build_gateway_command",
+]

nanobot/gateway/runtime.py

@@ -0,0 +1,448 @@
+"""Background process control for ``nanobot gateway``.
+
+This module intentionally stays small: the CLI owns command wording, while this
+runtime owns process state, log files, and platform-specific detach/stop details.
+"""
+
+from __future__ import annotations
+
+import ctypes
+import json
+import os
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+from collections.abc import Callable
+from contextlib import suppress
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from nanobot.config.paths import get_data_dir
+
+
+@dataclass(frozen=True)
+class GatewayStartOptions:
+    """Options needed to start a background gateway instance."""
+
+    port: int
+    verbose: bool = False
+    workspace: str | None = None
+    config_path: str | None = None
+
+
+@dataclass(frozen=True)
+class GatewayStatus:
+    """Current background gateway status."""
+
+    running: bool
+    pid: int | None
+    state_path: Path
+    log_path: Path
+    started_at: str | None = None
+    port: int | None = None
+    command: tuple[str, ...] = ()
+    reason: str = "not_started"
+
+
+@dataclass(frozen=True)
+class RuntimeResult:
+    """Result from a gateway runtime control operation."""
+
+    ok: bool
+    message: str
+    status: GatewayStatus
+
+
+def build_gateway_command(python_executable: str, options: GatewayStartOptions) -> list[str]:
+    """Build a foreground gateway command for process supervisors."""
+    command = [
+        python_executable,
+        "-m",
+        "nanobot",
+        "gateway",
+        "--foreground",
+        "--port",
+        str(options.port),
+    ]
+    if options.verbose:
+        command.append("--verbose")
+    if options.workspace:
+        command.extend(["--workspace", options.workspace])
+    if options.config_path:
+        command.extend(["--config", options.config_path])
+    return command
+
+
+@dataclass(frozen=True)
+class GatewayRuntimePaths:
+    """Filesystem layout for one gateway runtime instance."""
+
+    run_dir: Path
+    logs_dir: Path
+    state_path: Path
+    log_path: Path
+
+    @classmethod
+    def for_instance(
+        cls,
+        *,
+        data_dir: Path | None = None,
+        workspace: str | None = None,
+        config_path: str | None = None,
+    ) -> "GatewayRuntimePaths":
+        base = data_dir or get_data_dir()
+        suffix = _instance_suffix(workspace=workspace, config_path=config_path)
+        run_dir = base / "run"
+        logs_dir = base / "logs"
+        stem = "gateway" if suffix is None else f"gateway.{suffix}"
+        return cls(
+            run_dir=run_dir,
+            logs_dir=logs_dir,
+            state_path=run_dir / f"{stem}.json",
+            log_path=logs_dir / f"{stem}.log",
+        )
+
+
+class GatewayRuntime:
+    """Manage a background ``nanobot gateway`` process."""
+
+    def __init__(
+        self,
+        *,
+        paths: GatewayRuntimePaths | None = None,
+        platform_name: str | None = None,
+        python_executable: str | None = None,
+        popen: Callable[..., Any] = subprocess.Popen,
+        subprocess_run: Callable[..., Any] = subprocess.run,
+        sleep: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self.paths = paths or GatewayRuntimePaths.for_instance()
+        self.platform_name = platform_name or _platform_name()
+        self.python_executable = python_executable or sys.executable
+        self._popen = popen
+        self._subprocess_run = subprocess_run
+        self._sleep = sleep
+
+    def start_background(self, options: GatewayStartOptions) -> RuntimeResult:
+        """Start gateway as a detached background process."""
+        current = self.status()
+        if current.running:
+            return RuntimeResult(False, "gateway_already_running", current)
+
+        command = self._build_child_command(options)
+        self.paths.run_dir.mkdir(parents=True, exist_ok=True)
+        self.paths.logs_dir.mkdir(parents=True, exist_ok=True)
+
+        with self.paths.log_path.open("a", encoding="utf-8") as log_handle:
+            process = self._popen(
+                command,
+                stdin=subprocess.DEVNULL,
+                stdout=log_handle,
+                stderr=subprocess.STDOUT,
+                **self._popen_platform_kwargs(),
+            )
+
+        pid = int(process.pid)
+        self._sleep(0.2)
+        if not self._is_pid_running(pid):
+            return RuntimeResult(False, "gateway_exited_during_startup", self.status())
+
+        identity = self._process_identity(pid)
+        self._write_state(
+            {
+                "pid": pid,
+                "identity": identity,
+                "started_at": _utc_now(),
+                "platform": self.platform_name,
+                "port": options.port,
+                "workspace": options.workspace,
+                "config_path": options.config_path,
+                "command": command,
+                "log_path": str(self.paths.log_path),
+            }
+        )
+        return RuntimeResult(True, "gateway_started_background", self.status())
+
+    def stop(self, *, timeout_s: int = 20) -> RuntimeResult:
+        """Stop the recorded background gateway process."""
+        status = self.status()
+        if not status.pid:
+            return RuntimeResult(False, "gateway_not_running", status)
+
+        state = self._read_state()
+        if not self._record_matches_process(state, status.pid):
+            self._clear_state()
+            return RuntimeResult(False, "gateway_state_stale", self.status(reason="stale_state"))
+
+        self._terminate(status.pid, timeout_s=timeout_s)
+        self._clear_state()
+        return RuntimeResult(True, "gateway_stopped", self.status(reason="stopped"))
+
+    def restart(self, options: GatewayStartOptions, *, timeout_s: int = 20) -> RuntimeResult:
+        """Restart the background gateway."""
+        stop_result = self.stop(timeout_s=timeout_s)
+        if not stop_result.ok and stop_result.message not in {"gateway_not_running", "gateway_state_stale"}:
+            return stop_result
+        return self.start_background(options)
+
+    def status(self, *, reason: str | None = None) -> GatewayStatus:
+        """Return live status, clearing stale state when needed."""
+        state = self._read_state()
+        pid = _as_int(state.get("pid")) if state else None
+        if pid is None:
+            return GatewayStatus(
+                running=False,
+                pid=None,
+                state_path=self.paths.state_path,
+                log_path=self.paths.log_path,
+                reason=reason or "not_started",
+            )
+
+        if not self._is_pid_running(pid) or not self._record_matches_process(state, pid):
+            self._clear_state()
+            return GatewayStatus(
+                running=False,
+                pid=None,
+                state_path=self.paths.state_path,
+                log_path=self.paths.log_path,
+                reason=reason or "stale_state",
+            )
+
+        command = state.get("command")
+        return GatewayStatus(
+            running=True,
+            pid=pid,
+            state_path=self.paths.state_path,
+            log_path=self.paths.log_path,
+            started_at=_as_str(state.get("started_at")),
+            port=_as_int(state.get("port")),
+            command=tuple(command) if isinstance(command, list) else (),
+            reason=reason or "running",
+        )
+
+    def read_log_tail(self, *, tail: int = 200) -> list[str]:
+        """Return the last ``tail`` log lines."""
+        if tail <= 0 or not self.paths.log_path.exists():
+            return []
+        try:
+            lines = self.paths.log_path.read_text(encoding="utf-8", errors="replace").splitlines()
+        except OSError:
+            return []
+        return lines[-tail:]
+
+    def follow_logs(self, *, tail: int = 200) -> int:
+        """Print existing log tail and follow new log lines."""
+        for line in self.read_log_tail(tail=tail):
+            print(line)
+        self.paths.logs_dir.mkdir(parents=True, exist_ok=True)
+        self.paths.log_path.touch(exist_ok=True)
+        try:
+            with self.paths.log_path.open("r", encoding="utf-8", errors="replace") as handle:
+                handle.seek(0, os.SEEK_END)
+                while True:
+                    line = handle.readline()
+                    if line:
+                        print(line.rstrip("\n"))
+                    else:
+                        self._sleep(0.5)
+        except KeyboardInterrupt:
+            return 130
+
+    def _build_child_command(self, options: GatewayStartOptions) -> list[str]:
+        return build_gateway_command(self.python_executable, options)
+
+    def _popen_platform_kwargs(self) -> dict[str, Any]:
+        if self.platform_name == "Windows":
+            flags = 0
+            flags |= getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
+            flags |= getattr(subprocess, "CREATE_NO_WINDOW", 0)
+            return {"creationflags": flags}
+        return {"start_new_session": True}
+
+    def _terminate(self, pid: int, *, timeout_s: int) -> None:
+        if self.platform_name == "Windows":
+            self._terminate_windows(pid, timeout_s=timeout_s)
+        else:
+            self._terminate_posix(pid, timeout_s=timeout_s)
+
+    def _terminate_posix(self, pid: int, *, timeout_s: int) -> None:
+        try:
+            pgid = os.getpgid(pid)
+        except OSError:
+            pgid = None
+        try:
+            if pgid is not None:
+                os.killpg(pgid, signal.SIGTERM)
+            else:
+                os.kill(pid, signal.SIGTERM)
+        except ProcessLookupError:
+            return
+        if self._wait_for_exit(pid, timeout_s):
+            return
+        with suppress(ProcessLookupError):
+            if pgid is not None:
+                os.killpg(pgid, signal.SIGKILL)
+            else:
+                os.kill(pid, signal.SIGKILL)
+        self._wait_for_exit(pid, 2)
+
+    def _terminate_windows(self, pid: int, *, timeout_s: int) -> None:
+        ctrl_break = getattr(signal, "CTRL_BREAK_EVENT", None)
+        if ctrl_break is not None:
+            with suppress(ProcessLookupError):
+                os.kill(pid, ctrl_break)
+            if self._wait_for_exit(pid, timeout_s):
+                return
+        self._subprocess_run(["taskkill", "/PID", str(pid), "/T"], check=False)
+        if self._wait_for_exit(pid, 2):
+            return
+        self._subprocess_run(["taskkill", "/PID", str(pid), "/T", "/F"], check=False)
+        self._wait_for_exit(pid, 2)
+
+    def _wait_for_exit(self, pid: int, timeout_s: int | float) -> bool:
+        deadline = time.monotonic() + max(float(timeout_s), 0.0)
+        while time.monotonic() < deadline:
+            if not self._is_pid_running(pid):
+                return True
+            self._sleep(0.1)
+        return not self._is_pid_running(pid)
+
+    def _is_pid_running(self, pid: int) -> bool:
+        if pid <= 0:
+            return False
+        if self.platform_name == "Windows":
+            return _windows_process_identity(pid) is not None
+        try:
+            os.kill(pid, 0)
+        except ProcessLookupError:
+            return False
+        except PermissionError:
+            return True
+        except OSError:
+            return False
+        return True
+
+    def _process_identity(self, pid: int) -> str | int | None:
+        if self.platform_name == "Windows":
+            return _windows_process_identity(pid)
+        try:
+            return os.getpgid(pid)
+        except OSError:
+            return None
+
+    def _record_matches_process(self, state: dict[str, Any] | None, pid: int) -> bool:
+        if not state:
+            return False
+        recorded = state.get("identity")
+        if recorded is None:
+            return True
+        return recorded == self._process_identity(pid)
+
+    def _read_state(self) -> dict[str, Any] | None:
+        try:
+            with self.paths.state_path.open(encoding="utf-8") as handle:
+                payload = json.load(handle)
+        except (OSError, json.JSONDecodeError, ValueError):
+            return None
+        return payload if isinstance(payload, dict) else None
+
+    def _write_state(self, payload: dict[str, Any]) -> None:
+        self.paths.run_dir.mkdir(parents=True, exist_ok=True)
+        fd, tmp_name = tempfile.mkstemp(
+            prefix=f"{self.paths.state_path.name}.",
+            suffix=".tmp",
+            dir=self.paths.run_dir,
+        )
+        tmp_path = Path(tmp_name)
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as handle:
+                json.dump(payload, handle, indent=2, ensure_ascii=False)
+                handle.write("\n")
+                handle.flush()
+                os.fsync(handle.fileno())
+            tmp_path.replace(self.paths.state_path)
+        finally:
+            tmp_path.unlink(missing_ok=True)
+
+    def _clear_state(self) -> None:
+        self.paths.state_path.unlink(missing_ok=True)
+
+
+def _instance_suffix(*, workspace: str | None, config_path: str | None) -> str | None:
+    raw = "|".join(value for value in (workspace, config_path) if value)
+    if not raw:
+        return None
+    import hashlib
+
+    return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]
+
+
+def _platform_name() -> str:
+    if sys.platform.startswith("win"):
+        return "Windows"
+    if sys.platform == "darwin":
+        return "Darwin"
+    return "Linux"
+
+
+def _utc_now() -> str:
+    return datetime.now(UTC).isoformat().replace("+00:00", "Z")
+
+
+def _as_int(value: object) -> int | None:
+    if isinstance(value, int):
+        return value
+    if isinstance(value, str):
+        try:
+            return int(value)
+        except ValueError:
+            return None
+    return None
+
+
+def _as_str(value: object) -> str | None:
+    return value if isinstance(value, str) else None
+
+
+def _windows_process_identity(pid: int) -> str | None:
+    if os.name != "nt":
+        return None
+
+    class FileTime(ctypes.Structure):
+        _fields_ = [("low", ctypes.c_uint32), ("high", ctypes.c_uint32)]
+
+        @property
+        def value(self) -> int:
+            return (int(self.high) << 32) | int(self.low)
+
+    process_query_limited_information = 0x1000
+    kernel32 = ctypes.windll.kernel32
+    handle = kernel32.OpenProcess(process_query_limited_information, False, pid)
+    if not handle:
+        return None
+    try:
+        creation_time = FileTime()
+        exit_time = FileTime()
+        kernel_time = FileTime()
+        user_time = FileTime()
+        ok = kernel32.GetProcessTimes(
+            handle,
+            ctypes.byref(creation_time),
+            ctypes.byref(exit_time),
+            ctypes.byref(kernel_time),
+            ctypes.byref(user_time),
+        )
+        if not ok:
+            return None
+        exit_code = ctypes.c_uint32()
+        if not kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)):
+            return None
+        if exit_code.value != 259:
+            return None
+        return str(creation_time.value)
+    finally:
+        kernel32.CloseHandle(handle)

nanobot/gateway/service.py

@@ -0,0 +1,286 @@
+"""Install and manage OS-level gateway services."""
+
+from __future__ import annotations
+
+import os
+import plistlib
+import re
+import subprocess
+import sys
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+from nanobot.gateway import GatewayStartOptions, build_gateway_command
+
+ServiceManagerKind = Literal["auto", "systemd", "launchd"]
+
+
+@dataclass(frozen=True)
+class GatewayServiceOptions:
+    """Inputs used to render one system service."""
+
+    start: GatewayStartOptions
+    name: str = "nanobot-gateway"
+    manager: ServiceManagerKind = "auto"
+    enable: bool = True
+    start_now: bool = True
+    python_executable: str = sys.executable
+
+
+@dataclass(frozen=True)
+class GatewayServiceResult:
+    """Result from service install/uninstall operations."""
+
+    ok: bool
+    message: str
+    manager: str
+    path: Path | None
+    commands: tuple[tuple[str, ...], ...] = ()
+    content: str | None = None
+
+
+class GatewayServiceInstaller:
+    """Render and install systemd user services or macOS LaunchAgents."""
+
+    def __init__(
+        self,
+        *,
+        platform_name: str | None = None,
+        subprocess_run: Callable[..., Any] = subprocess.run,
+        home: Path | None = None,
+    ) -> None:
+        self.platform_name = platform_name or _platform_name()
+        self._subprocess_run = subprocess_run
+        self.home = home or Path.home()
+
+    def install(self, options: GatewayServiceOptions, *, dry_run: bool = False) -> GatewayServiceResult:
+        manager = self._resolve_manager(options.manager)
+        if manager == "systemd":
+            return self._install_systemd(options, dry_run=dry_run)
+        if manager == "launchd":
+            return self._install_launchd(options, dry_run=dry_run)
+        return GatewayServiceResult(False, f"unsupported_service_manager:{manager}", manager, None)
+
+    def uninstall(
+        self,
+        *,
+        name: str = "nanobot-gateway",
+        manager: ServiceManagerKind = "auto",
+        dry_run: bool = False,
+    ) -> GatewayServiceResult:
+        resolved = self._resolve_manager(manager)
+        if resolved == "systemd":
+            return self._uninstall_systemd(name=name, dry_run=dry_run)
+        if resolved == "launchd":
+            return self._uninstall_launchd(name=name, dry_run=dry_run)
+        return GatewayServiceResult(False, f"unsupported_service_manager:{resolved}", resolved, None)
+
+    def _install_systemd(
+        self,
+        options: GatewayServiceOptions,
+        *,
+        dry_run: bool,
+    ) -> GatewayServiceResult:
+        unit_name = _systemd_unit_name(options.name)
+        path = self.home / ".config" / "systemd" / "user" / unit_name
+        command = build_gateway_command(options.python_executable, options.start)
+        content = _systemd_unit_content(
+            description=f"Nanobot Gateway ({options.name})",
+            command=command,
+            working_directory=_working_directory_text(options.start),
+        )
+        commands: list[tuple[str, ...]] = [("systemctl", "--user", "daemon-reload")]
+        if options.enable:
+            commands.append(("systemctl", "--user", "enable", unit_name))
+        if options.start_now:
+            commands.append(("systemctl", "--user", "restart", unit_name))
+        if dry_run:
+            return GatewayServiceResult(True, "service_install_dry_run", "systemd", path, tuple(commands), content)
+
+        _working_directory(options.start).mkdir(parents=True, exist_ok=True)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        for command_args in commands:
+            self._subprocess_run(list(command_args), check=True)
+        return GatewayServiceResult(True, "service_installed", "systemd", path, tuple(commands), content)
+
+    def _uninstall_systemd(
+        self,
+        *,
+        name: str,
+        dry_run: bool,
+    ) -> GatewayServiceResult:
+        unit_name = _systemd_unit_name(name)
+        path = self.home / ".config" / "systemd" / "user" / unit_name
+        commands = (
+            ("systemctl", "--user", "disable", "--now", unit_name),
+            ("systemctl", "--user", "daemon-reload"),
+        )
+        if dry_run:
+            return GatewayServiceResult(True, "service_uninstall_dry_run", "systemd", path, commands)
+
+        self._run_best_effort(commands[0])
+        path.unlink(missing_ok=True)
+        self._subprocess_run(list(commands[1]), check=True)
+        return GatewayServiceResult(True, "service_uninstalled", "systemd", path, commands)
+
+    def _install_launchd(
+        self,
+        options: GatewayServiceOptions,
+        *,
+        dry_run: bool,
+    ) -> GatewayServiceResult:
+        label = _launchd_label(options.name)
+        path = self.home / "Library" / "LaunchAgents" / f"{label}.plist"
+        log_stem = _safe_service_name(options.name)
+        stdout_path = self.home / ".nanobot" / "logs" / f"{log_stem}.launchd.log"
+        stderr_path = self.home / ".nanobot" / "logs" / f"{log_stem}.launchd.err.log"
+        payload = {
+            "Label": label,
+            "ProgramArguments": build_gateway_command(options.python_executable, options.start),
+            "WorkingDirectory": _working_directory_text(options.start),
+            "RunAtLoad": bool(options.start_now),
+            "KeepAlive": {"SuccessfulExit": False},
+            "StandardOutPath": str(stdout_path),
+            "StandardErrorPath": str(stderr_path),
+        }
+        content = plistlib.dumps(payload, sort_keys=False).decode("utf-8")
+        domain = _launchd_domain()
+        commands: list[tuple[str, ...]] = []
+        if options.enable or options.start_now:
+            commands.append(("launchctl", "bootstrap", domain, str(path)))
+        if options.enable:
+            commands.append(("launchctl", "enable", f"{domain}/{label}"))
+        if options.start_now:
+            commands.append(("launchctl", "kickstart", "-k", f"{domain}/{label}"))
+        if dry_run:
+            return GatewayServiceResult(True, "service_install_dry_run", "launchd", path, tuple(commands), content)
+
+        _working_directory(options.start).mkdir(parents=True, exist_ok=True)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        stdout_path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content, encoding="utf-8")
+        if options.enable or options.start_now:
+            self._run_best_effort(("launchctl", "bootout", domain, str(path)))
+        for command_args in commands:
+            self._subprocess_run(list(command_args), check=True)
+        return GatewayServiceResult(True, "service_installed", "launchd", path, tuple(commands), content)
+
+    def _uninstall_launchd(
+        self,
+        *,
+        name: str,
+        dry_run: bool,
+    ) -> GatewayServiceResult:
+        label = _launchd_label(name)
+        path = self.home / "Library" / "LaunchAgents" / f"{label}.plist"
+        domain = _launchd_domain()
+        commands = (
+            ("launchctl", "bootout", domain, str(path)),
+            ("launchctl", "disable", f"{domain}/{label}"),
+        )
+        if dry_run:
+            return GatewayServiceResult(True, "service_uninstall_dry_run", "launchd", path, commands)
+
+        for command_args in commands:
+            self._run_best_effort(command_args)
+        path.unlink(missing_ok=True)
+        return GatewayServiceResult(True, "service_uninstalled", "launchd", path, commands)
+
+    def _resolve_manager(self, manager: ServiceManagerKind) -> str:
+        if manager != "auto":
+            return manager
+        if self.platform_name == "Darwin":
+            return "launchd"
+        if self.platform_name == "Linux":
+            return "systemd"
+        return self.platform_name.lower()
+
+    def _run_best_effort(self, command_args: tuple[str, ...]) -> None:
+        self._subprocess_run(list(command_args), check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+
+
+def _platform_name() -> str:
+    if sys.platform == "darwin":
+        return "Darwin"
+    if sys.platform.startswith("linux"):
+        return "Linux"
+    if sys.platform.startswith("win"):
+        return "Windows"
+    return sys.platform
+
+
+def _working_directory(options: GatewayStartOptions) -> Path:
+    if options.workspace:
+        return Path(options.workspace).expanduser()
+    return Path.home()
+
+
+def _working_directory_text(options: GatewayStartOptions) -> str:
+    if options.workspace:
+        return os.path.expanduser(options.workspace)
+    return str(Path.home())
+
+
+def _systemd_unit_name(name: str) -> str:
+    stem = _safe_service_name(name)
+    return stem if stem.endswith(".service") else f"{stem}.service"
+
+
+def _launchd_label(name: str) -> str:
+    if name.startswith("ai.nanobot."):
+        return name
+    suffix = _safe_service_name(name).removeprefix("nanobot-").replace("-", ".")
+    return f"ai.nanobot.{suffix}"
+
+
+def _safe_service_name(name: str) -> str:
+    value = name.strip().lower()
+    value = re.sub(r"[^a-z0-9_.-]+", "-", value)
+    value = value.strip(".-")
+    return value or "nanobot-gateway"
+
+
+def _launchd_domain() -> str:
+    getuid = getattr(os, "getuid", None)
+    if getuid is None:
+        return "gui/current"
+    return f"gui/{getuid()}"
+
+
+def _systemd_unit_content(
+    *,
+    description: str,
+    command: list[str],
+    working_directory: str,
+) -> str:
+    quoted_command = " ".join(_systemd_quote(part) for part in command)
+    return "\n".join(
+        [
+            "[Unit]",
+            f"Description={description}",
+            "After=network-online.target",
+            "Wants=network-online.target",
+            "",
+            "[Service]",
+            "Type=simple",
+            f"WorkingDirectory={_systemd_quote(str(working_directory))}",
+            f"ExecStart={quoted_command}",
+            "Restart=always",
+            "RestartSec=10",
+            "Environment=PYTHONUNBUFFERED=1",
+            "NoNewPrivileges=yes",
+            "",
+            "[Install]",
+            "WantedBy=default.target",
+            "",
+        ]
+    )
+
+
+def _systemd_quote(value: str) -> str:
+    if value and not re.search(r"\s|['\"\\]", value):
+        return value
+    return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'

tests/cli/test_gateway_commands.py

@@ -0,0 +1,218 @@
+from pathlib import Path
+
+import typer
+from rich.console import Console
+from typer.testing import CliRunner
+
+from nanobot.cli.gateway import create_gateway_app
+from nanobot.config.schema import Config
+from nanobot.gateway import GatewayStartOptions, GatewayStatus, RuntimeResult
+from nanobot.gateway.service import GatewayServiceOptions, GatewayServiceResult
+
+runner = CliRunner()
+
+
+class FakeRuntime:
+    def __init__(self, tmp_path: Path):
+        self.status_value = GatewayStatus(
+            running=True,
+            pid=12345,
+            state_path=tmp_path / "gateway.json",
+            log_path=tmp_path / "gateway.log",
+            started_at="2026-06-22T00:00:00Z",
+            port=18790,
+            reason="running",
+        )
+        self.started_options: GatewayStartOptions | None = None
+        self.restarted_options: GatewayStartOptions | None = None
+        self.stop_timeout: int | None = None
+        self.follow_tail: int | None = None
+
+    def start_background(self, options: GatewayStartOptions) -> RuntimeResult:
+        self.started_options = options
+        return RuntimeResult(True, "gateway_started_background", self.status_value)
+
+    def restart(self, options: GatewayStartOptions, *, timeout_s: int) -> RuntimeResult:
+        self.restarted_options = options
+        self.stop_timeout = timeout_s
+        return RuntimeResult(True, "gateway_started_background", self.status_value)
+
+    def stop(self, *, timeout_s: int) -> RuntimeResult:
+        self.stop_timeout = timeout_s
+        return RuntimeResult(True, "gateway_stopped", self.status_value)
+
+    def status(self) -> GatewayStatus:
+        return self.status_value
+
+    def read_log_tail(self, *, tail: int) -> list[str]:
+        return [f"line {tail}"]
+
+    def follow_logs(self, *, tail: int) -> int:
+        self.follow_tail = tail
+        return 0
+
+
+class FakeServiceInstaller:
+    def __init__(self, tmp_path: Path):
+        self.tmp_path = tmp_path
+        self.installed_options: GatewayServiceOptions | None = None
+        self.install_dry_run: bool | None = None
+        self.uninstalled_name: str | None = None
+        self.uninstall_manager: str | None = None
+
+    def install(self, options: GatewayServiceOptions, *, dry_run: bool) -> GatewayServiceResult:
+        self.installed_options = options
+        self.install_dry_run = dry_run
+        return GatewayServiceResult(
+            True,
+            "service_install_dry_run" if dry_run else "service_installed",
+            "systemd",
+            self.tmp_path / "nanobot-gateway.service",
+            (("systemctl", "--user", "daemon-reload"),),
+            "[Unit]\nDescription=Nanobot Gateway\n",
+        )
+
+    def uninstall(self, *, name: str, manager: str, dry_run: bool) -> GatewayServiceResult:
+        self.uninstalled_name = name
+        self.uninstall_manager = manager
+        return GatewayServiceResult(
+            True,
+            "service_uninstall_dry_run" if dry_run else "service_uninstalled",
+            "systemd",
+            self.tmp_path / "nanobot-gateway.service",
+            (("systemctl", "--user", "disable", "--now", "nanobot-gateway.service"),),
+        )
+
+
+def _test_app(tmp_path: Path, config: Config | None = None):
+    app = typer.Typer()
+    fake_runtime = FakeRuntime(tmp_path)
+    fake_service = FakeServiceInstaller(tmp_path)
+    run_calls: list[tuple[Config, int | None]] = []
+
+    def load_runtime_config(_config_path: str | None, _workspace: str | None) -> Config:
+        return config or Config()
+
+    def run_gateway(config: Config, *, port: int | None = None) -> None:
+        run_calls.append((config, port))
+
+    app.add_typer(
+        create_gateway_app(
+            console=Console(),
+            log_handler_id=0,
+            load_runtime_config=load_runtime_config,
+            run_gateway=run_gateway,
+            runtime_factory=lambda **_kwargs: fake_runtime,
+            service_factory=lambda: fake_service,
+        ),
+        name="gateway",
+    )
+    return app, fake_runtime, fake_service, run_calls
+
+
+def test_gateway_default_still_runs_foreground(tmp_path):
+    app, _runtime, _service, calls = _test_app(tmp_path)
+
+    result = runner.invoke(app, ["gateway", "--port", "18791"])
+
+    assert result.exit_code == 0
+    assert len(calls) == 1
+    assert calls[0][1] == 18791
+
+
+def test_gateway_background_starts_detached_runtime(tmp_path):
+    config = Config()
+    config.gateway.port = 18792
+    app, fake_runtime, _service, _calls = _test_app(tmp_path, config=config)
+
+    result = runner.invoke(app, ["gateway", "--background"])
+
+    assert result.exit_code == 0
+    assert "Gateway started in the background" in result.stdout
+    assert fake_runtime.started_options == GatewayStartOptions(port=18792)
+
+
+def test_gateway_rejects_conflicting_modes(tmp_path):
+    app, _runtime, _service, _calls = _test_app(tmp_path)
+
+    result = runner.invoke(app, ["gateway", "--foreground", "--background"])
+
+    assert result.exit_code == 1
+    assert "--foreground and --background cannot be used together" in result.stdout
+
+
+def test_gateway_status_uses_runtime(tmp_path):
+    app, _runtime, _service, _calls = _test_app(tmp_path)
+
+    result = runner.invoke(app, ["gateway", "status"])
+
+    assert result.exit_code == 0
+    assert "Running: yes" in result.stdout
+    assert "PID: 12345" in result.stdout
+
+
+def test_gateway_logs_can_read_without_following(tmp_path):
+    app, _runtime, _service, _calls = _test_app(tmp_path)
+
+    result = runner.invoke(app, ["gateway", "logs", "--tail", "12", "--no-follow"])
+
+    assert result.exit_code == 0
+    assert "line 12" in result.stdout
+
+
+def test_gateway_stop_treats_not_running_as_clean(tmp_path):
+    app, fake_runtime, _service, _calls = _test_app(tmp_path)
+
+    def fake_stop(*, timeout_s: int) -> RuntimeResult:
+        fake_runtime.stop_timeout = timeout_s
+        return RuntimeResult(False, "gateway_not_running", fake_runtime.status_value)
+
+    fake_runtime.stop = fake_stop  # type: ignore[method-assign]
+
+    result = runner.invoke(app, ["gateway", "stop", "--timeout", "3"])
+
+    assert result.exit_code == 0
+    assert "gateway_not_running" in result.stdout
+    assert fake_runtime.stop_timeout == 3
+
+
+def test_gateway_restart_starts_background_runtime(tmp_path):
+    config = Config()
+    config.gateway.port = 18793
+    app, fake_runtime, _service, _calls = _test_app(tmp_path, config=config)
+
+    result = runner.invoke(app, ["gateway", "restart", "--timeout", "9", "--verbose"])
+
+    assert result.exit_code == 0
+    assert "Gateway restarted in the background" in result.stdout
+    assert fake_runtime.stop_timeout == 9
+    assert fake_runtime.restarted_options == GatewayStartOptions(port=18793, verbose=True)
+
+
+def test_gateway_install_service_uses_service_installer(tmp_path):
+    config = Config()
+    config.gateway.port = 18794
+    app, _runtime, service, _calls = _test_app(tmp_path, config=config)
+
+    result = runner.invoke(app, ["gateway", "install-service", "--dry-run", "--manager", "systemd"])
+
+    assert result.exit_code == 0
+    assert "Gateway service dry run" in result.stdout
+    assert service.install_dry_run is True
+    assert service.installed_options is not None
+    assert service.installed_options.start.port == 18794
+    assert service.installed_options.manager == "systemd"
+
+
+def test_gateway_uninstall_service_uses_service_installer(tmp_path):
+    app, _runtime, service, _calls = _test_app(tmp_path)
+
+    result = runner.invoke(
+        app,
+        ["gateway", "uninstall-service", "--dry-run", "--name", "custom-gateway", "--manager", "systemd"],
+    )
+
+    assert result.exit_code == 0
+    assert "Gateway service uninstall dry run" in result.stdout
+    assert service.uninstalled_name == "custom-gateway"
+    assert service.uninstall_manager == "systemd"

tests/gateway/test_gateway_service.py

@@ -0,0 +1,190 @@
+import os
+import plistlib
+
+from nanobot.gateway import GatewayStartOptions
+from nanobot.gateway.service import GatewayServiceInstaller, GatewayServiceOptions
+
+
+def _expected_launchd_domain() -> str:
+    getuid = getattr(os, "getuid", None)
+    if getuid is None:
+        return "gui/current"
+    return f"gui/{getuid()}"
+
+
+def test_systemd_install_dry_run_renders_user_unit(tmp_path):
+    installer = GatewayServiceInstaller(platform_name="Linux", home=tmp_path)
+
+    result = installer.install(
+        GatewayServiceOptions(
+            start=GatewayStartOptions(
+                port=18790,
+                verbose=True,
+                workspace="/tmp/nanobot workspace",
+                config_path="/tmp/nanobot/config.json",
+            ),
+            python_executable="/venv/bin/python",
+        ),
+        dry_run=True,
+    )
+
+    assert result.ok is True
+    assert result.manager == "systemd"
+    assert result.path == tmp_path / ".config/systemd/user/nanobot-gateway.service"
+    assert ("systemctl", "--user", "daemon-reload") in result.commands
+    assert ("systemctl", "--user", "enable", "nanobot-gateway.service") in result.commands
+    assert ("systemctl", "--user", "restart", "nanobot-gateway.service") in result.commands
+    assert result.content is not None
+    assert 'WorkingDirectory="/tmp/nanobot workspace"' in result.content
+    assert 'ExecStart=/venv/bin/python -m nanobot gateway --foreground --port 18790 --verbose' in result.content
+    assert '--workspace "/tmp/nanobot workspace" --config /tmp/nanobot/config.json' in result.content
+
+
+def test_systemd_install_writes_unit_and_runs_commands(tmp_path):
+    commands: list[list[str]] = []
+    workspace = tmp_path / "missing-workspace"
+    installer = GatewayServiceInstaller(
+        platform_name="Linux",
+        home=tmp_path,
+        subprocess_run=lambda command, **_kwargs: commands.append(command),
+    )
+
+    result = installer.install(
+        GatewayServiceOptions(
+            start=GatewayStartOptions(port=18790, workspace=str(workspace)),
+            enable=False,
+            start_now=True,
+            python_executable="/python",
+        )
+    )
+
+    assert result.ok is True
+    assert result.path is not None
+    assert result.path.exists()
+    assert workspace.exists()
+    assert commands == [
+        ["systemctl", "--user", "daemon-reload"],
+        ["systemctl", "--user", "restart", "nanobot-gateway.service"],
+    ]
+
+
+def test_launchd_install_dry_run_renders_plist(tmp_path):
+    installer = GatewayServiceInstaller(platform_name="Darwin", home=tmp_path)
+
+    result = installer.install(
+        GatewayServiceOptions(
+            start=GatewayStartOptions(
+                port=18791,
+                workspace="/Users/test/.nanobot/workspace",
+                config_path="/Users/test/.nanobot/config.json",
+            ),
+            python_executable="/opt/homebrew/bin/python3",
+        ),
+        dry_run=True,
+    )
+
+    assert result.ok is True
+    assert result.manager == "launchd"
+    assert result.path == tmp_path / "Library/LaunchAgents/ai.nanobot.gateway.plist"
+    assert result.content is not None
+    payload = plistlib.loads(result.content.encode("utf-8"))
+    assert payload["Label"] == "ai.nanobot.gateway"
+    assert payload["ProgramArguments"] == [
+        "/opt/homebrew/bin/python3",
+        "-m",
+        "nanobot",
+        "gateway",
+        "--foreground",
+        "--port",
+        "18791",
+        "--workspace",
+        "/Users/test/.nanobot/workspace",
+        "--config",
+        "/Users/test/.nanobot/config.json",
+    ]
+    assert payload["KeepAlive"] == {"SuccessfulExit": False}
+    assert ("launchctl", "bootstrap", _expected_launchd_domain(), str(result.path)) in result.commands
+
+
+def test_launchd_no_enable_start_still_bootstraps(tmp_path):
+    installer = GatewayServiceInstaller(platform_name="Darwin", home=tmp_path)
+
+    result = installer.install(
+        GatewayServiceOptions(
+            start=GatewayStartOptions(port=18790),
+            enable=False,
+            start_now=True,
+        ),
+        dry_run=True,
+    )
+
+    assert result.commands[0][:2] == ("launchctl", "bootstrap")
+    assert not any(command[1] == "enable" for command in result.commands)
+    assert any(command[1] == "kickstart" for command in result.commands)
+
+
+def test_launchd_no_enable_start_reinstall_boots_out_existing_label(tmp_path):
+    commands: list[list[str]] = []
+    installer = GatewayServiceInstaller(
+        platform_name="Darwin",
+        home=tmp_path,
+        subprocess_run=lambda command, **_kwargs: commands.append(command),
+    )
+
+    result = installer.install(
+        GatewayServiceOptions(
+            start=GatewayStartOptions(port=18790),
+            enable=False,
+            start_now=True,
+        )
+    )
+
+    assert result.ok is True
+    assert commands[0][:2] == ["launchctl", "bootout"]
+    assert commands[1][:2] == ["launchctl", "bootstrap"]
+
+
+def test_launchd_dry_run_does_not_require_posix_getuid(tmp_path, monkeypatch):
+    monkeypatch.delattr(os, "getuid", raising=False)
+    installer = GatewayServiceInstaller(platform_name="Darwin", home=tmp_path)
+
+    result = installer.install(
+        GatewayServiceOptions(start=GatewayStartOptions(port=18790)),
+        dry_run=True,
+    )
+
+    assert result.ok is True
+    assert result.commands[0][:3] == ("launchctl", "bootstrap", "gui/current")
+
+
+def test_uninstall_systemd_removes_unit_and_reloads(tmp_path):
+    commands: list[list[str]] = []
+    installer = GatewayServiceInstaller(
+        platform_name="Linux",
+        home=tmp_path,
+        subprocess_run=lambda command, **_kwargs: commands.append(command),
+    )
+    unit = tmp_path / ".config/systemd/user/nanobot-gateway.service"
+    unit.parent.mkdir(parents=True)
+    unit.write_text("[Unit]\n", encoding="utf-8")
+
+    result = installer.uninstall()
+
+    assert result.ok is True
+    assert not unit.exists()
+    assert commands == [
+        ["systemctl", "--user", "disable", "--now", "nanobot-gateway.service"],
+        ["systemctl", "--user", "daemon-reload"],
+    ]
+
+
+def test_auto_manager_rejects_windows_services(tmp_path):
+    installer = GatewayServiceInstaller(platform_name="Windows", home=tmp_path)
+
+    result = installer.install(
+        GatewayServiceOptions(start=GatewayStartOptions(port=18790)),
+        dry_run=True,
+    )
+
+    assert result.ok is False
+    assert result.message == "unsupported_service_manager:windows"

tests/gateway/test_runtime.py

@@ -0,0 +1,148 @@
+import json
+from pathlib import Path
+
+from nanobot.gateway import GatewayRuntime, GatewayRuntimePaths, GatewayStartOptions
+
+
+class FakeProcess:
+    def __init__(self, pid: int = 12345):
+        self.pid = pid
+
+
+def _paths(tmp_path: Path) -> GatewayRuntimePaths:
+    return GatewayRuntimePaths.for_instance(data_dir=tmp_path)
+
+
+def test_paths_use_stable_instance_suffix_for_custom_selectors(tmp_path):
+    default_paths = GatewayRuntimePaths.for_instance(data_dir=tmp_path)
+    first_paths = GatewayRuntimePaths.for_instance(
+        data_dir=tmp_path,
+        workspace="/tmp/workspace-a",
+        config_path="/tmp/config-a.json",
+    )
+    second_paths = GatewayRuntimePaths.for_instance(
+        data_dir=tmp_path,
+        workspace="/tmp/workspace-b",
+        config_path="/tmp/config-b.json",
+    )
+
+    assert default_paths.state_path.name == "gateway.json"
+    assert first_paths.state_path.name.startswith("gateway.")
+    assert first_paths.state_path != second_paths.state_path
+    assert first_paths.log_path != second_paths.log_path
+
+
+def test_start_background_writes_state_and_child_command(tmp_path, monkeypatch):
+    calls: list[dict] = []
+
+    def fake_popen(command, **kwargs):
+        calls.append({"command": command, "kwargs": kwargs})
+        return FakeProcess()
+
+    runtime = GatewayRuntime(
+        paths=_paths(tmp_path),
+        platform_name="Linux",
+        python_executable="/python",
+        popen=fake_popen,
+        sleep=lambda _seconds: None,
+    )
+    monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
+    monkeypatch.setattr(runtime, "_process_identity", lambda _pid: 12345)
+
+    result = runtime.start_background(
+        GatewayStartOptions(
+            port=18790,
+            verbose=True,
+            workspace="/tmp/workspace",
+            config_path="/tmp/config.json",
+        )
+    )
+
+    assert result.ok is True
+    assert result.status.running is True
+    assert calls[0]["command"] == [
+        "/python",
+        "-m",
+        "nanobot",
+        "gateway",
+        "--foreground",
+        "--port",
+        "18790",
+        "--verbose",
+        "--workspace",
+        "/tmp/workspace",
+        "--config",
+        "/tmp/config.json",
+    ]
+    assert calls[0]["kwargs"]["start_new_session"] is True
+    state = json.loads(runtime.paths.state_path.read_text(encoding="utf-8"))
+    assert state["pid"] == 12345
+    assert state["identity"] == 12345
+    assert state["port"] == 18790
+
+
+def test_start_background_uses_windows_process_group_flags(tmp_path, monkeypatch):
+    calls: list[dict] = []
+
+    def fake_popen(command, **kwargs):
+        calls.append({"command": command, "kwargs": kwargs})
+        return FakeProcess()
+
+    runtime = GatewayRuntime(
+        paths=_paths(tmp_path),
+        platform_name="Windows",
+        python_executable="python.exe",
+        popen=fake_popen,
+        sleep=lambda _seconds: None,
+    )
+    monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
+    monkeypatch.setattr(runtime, "_process_identity", lambda _pid: "created-at")
+
+    result = runtime.start_background(GatewayStartOptions(port=18790))
+
+    assert result.ok is True
+    assert "creationflags" in calls[0]["kwargs"]
+    assert "start_new_session" not in calls[0]["kwargs"]
+
+
+def test_status_clears_stale_state(tmp_path, monkeypatch):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
+    runtime.paths.run_dir.mkdir(parents=True)
+    runtime.paths.state_path.write_text('{"pid": 12345, "identity": 12345}', encoding="utf-8")
+    monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: False)
+
+    status = runtime.status()
+
+    assert status.running is False
+    assert status.reason == "stale_state"
+    assert not runtime.paths.state_path.exists()
+
+
+def test_status_clears_state_when_pid_identity_changes(tmp_path, monkeypatch):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
+    runtime.paths.run_dir.mkdir(parents=True)
+    runtime.paths.state_path.write_text('{"pid": 12345, "identity": 111}', encoding="utf-8")
+    monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
+    monkeypatch.setattr(runtime, "_process_identity", lambda _pid: 222)
+
+    status = runtime.status()
+
+    assert status.running is False
+    assert status.reason == "stale_state"
+    assert not runtime.paths.state_path.exists()
+
+
+def test_stop_terminates_recorded_process(tmp_path, monkeypatch):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
+    runtime.paths.run_dir.mkdir(parents=True)
+    runtime.paths.state_path.write_text('{"pid": 12345, "identity": 12345}', encoding="utf-8")
+    monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
+    monkeypatch.setattr(runtime, "_process_identity", lambda _pid: 12345)
+    terminated: list[int] = []
+    monkeypatch.setattr(runtime, "_terminate", lambda pid, timeout_s: terminated.append(pid))
+
+    result = runtime.stop()
+
+    assert result.ok is True
+    assert terminated == [12345]
+    assert not runtime.paths.state_path.exists()