fix(gateway): harden shared client lifecycle

docs/cli-reference.md

@@ -121,7 +121,7 @@ nanobot sessions restore-workspace --config ./bot-a/config.json --workspace ./bo
 The command never deletes the external store and refuses to overwrite a different existing
 workspace file. Back up both the config directory and workspace before changing versions.
 
-Interactive mode uses nanobot's native TypeScript terminal UI. It talks to the same local gateway as the WebUI, so streaming, tool progress, and WebSocket sessions share one protocol instead of maintaining a second agent loop. If no gateway is running, either client starts it on demand. Exiting one TUI or WebUI launcher releases only that client; the last interactive launcher stops the on-demand gateway. Only an explicit `nanobot gateway --background` promotes it to persistent mode. `nanobot gateway restart` preserves the current lifecycle mode, and `nanobot gateway stop` ends either mode.
+Interactive mode uses nanobot's native TypeScript terminal UI. It talks to the same local gateway as the WebUI, so streaming, tool progress, and WebSocket sessions share one protocol instead of maintaining a second agent loop. If no gateway is running, either client starts it on demand. Exiting one TUI or WebUI launcher releases only that client; the last interactive launcher stops the on-demand gateway. A small gateway watchdog also reclaims an on-demand process if its last client crashes. Only an explicit `nanobot gateway --background` promotes it to persistent mode. `nanobot gateway restart` restarts a detached gateway without changing that lifetime; restart an attached foreground gateway in its owning terminal. `nanobot gateway stop` ends either mode.
 
 The default `--theme auto` mode probes the terminal's real foreground and background colors before first paint and follows supported live appearance changes. Use `--theme light` or `--theme dark` when a terminal or multiplexer does not report its colors reliably. The model preset and workspace access labels above the composer can be clicked to open their selectors; arrow keys, `Enter`, and `Esc` provide the same controls without a mouse. Access changes still pass through the gateway's local-trust and active-turn policy checks.
 
@@ -166,7 +166,7 @@ only when no other interactive client still holds it.
 | `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 status` | Show PID, foreground/background launch mode, explicit/on-demand lifetime, live client count, state, and logs |
 | `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/cli/gateway.py

@@ -128,6 +128,10 @@ def create_gateway_app(
             console.print(f"Port: {status.port}")
         if status.started_at is not None:
             console.print(f"Started At: {status.started_at}")
+        if status.running:
+            console.print(f"Launch Mode: {status.launch_mode}")
+            console.print(f"Lifetime: {status.lifetime}")
+            console.print(f"Clients: {status.clients}")
         console.print(f"State: {status.state_path}")
         console.print(f"Logs: {status.log_path}")
 
@@ -176,6 +180,35 @@ def create_gateway_app(
                     loaded_config=cfg,
                 )
             )
+            if (
+                result.message == "gateway_already_running"
+                and result.status.launch_mode == "foreground"
+            ):
+                console.print(
+                    "[yellow]Gateway is already running in the foreground; "
+                    "an attached process cannot be detached in place.[/yellow]"
+                )
+                console.print(
+                    "[dim]Stop it in its current terminal, then run "
+                    "`nanobot gateway --background`.[/dim]"
+                )
+                print_status(result.status)
+                raise typer.Exit(1)
+            if (
+                result.message == "gateway_already_running"
+                and result.status.launch_mode == "unknown"
+                and result.status.lifetime == "explicit"
+            ):
+                console.print(
+                    "[yellow]Gateway is already running, but this older process did "
+                    "not record whether it is attached or detached.[/yellow]"
+                )
+                console.print(
+                    "[dim]Stop it first, then rerun `nanobot gateway --background` "
+                    "to establish an unambiguous lifecycle.[/dim]"
+                )
+                print_status(result.status)
+                raise typer.Exit(1)
             promoted = False
             if result.ok or result.message == "gateway_already_running":
                 promoted = GatewayClientLease(
@@ -184,7 +217,7 @@ def create_gateway_app(
                 ).mark_persistent()
             if result.ok:
                 console.print("[green]Gateway started in the background.[/green]")
-                print_status(result.status)
+                print_status(runtime.status())
                 return
             if result.message == "gateway_already_running":
                 if promoted:
@@ -201,7 +234,7 @@ def create_gateway_app(
                         "[yellow]Gateway is already running in persistent "
                         "background mode.[/yellow]"
                     )
-                print_status(result.status)
+                print_status(runtime.status())
                 return
             console.print(f"[yellow]Gateway was not started: {result.message}[/yellow]")
             print_status(result.status)
@@ -303,6 +336,17 @@ def create_gateway_app(
             )
             print_status(result.status)
             raise typer.Exit(1)
+        if result.message == "gateway_foreground_restart_required":
+            console.print(
+                "[yellow]Gateway is attached to a foreground terminal and cannot "
+                "be restarted as a background process.[/yellow]"
+            )
+            console.print(
+                "[dim]Restart it in that terminal, or stop it and run "
+                "`nanobot gateway --background`.[/dim]"
+            )
+            print_status(result.status)
+            raise typer.Exit(1)
         console.print(f"[red]Gateway restart failed: {result.message}[/red]")
         print_status(result.status)
         raise typer.Exit(1)

nanobot/cli/gateway_runtime.py

@@ -389,7 +389,13 @@ def _run_gateway(
 
     # Use the same runtime identity for foreground and managed gateway processes.
     from nanobot.config.loader import get_config_path
-    from nanobot.gateway.runtime import GatewayRuntime, GatewayRuntimePaths, GatewayStartOptions
+    from nanobot.gateway.runtime import (
+        GatewayClientLease,
+        GatewayRuntime,
+        GatewayRuntimePaths,
+        GatewayStartOptions,
+        monitor_gateway_clients,
+    )
 
     config_path = str(get_config_path().resolve(strict=False))
     gateway_workspace = (
@@ -872,6 +878,14 @@ def _run_gateway(
                 finally:
                     await mcp_provider.aclose()
 
+            async def _monitor_local_clients() -> None:
+                orphaned = await monitor_gateway_clients(
+                    GatewayClientLease(gateway_runtime, kind="gateway-monitor"),
+                    shutdown_event,
+                )
+                if orphaned:
+                    logger.info("Last local client disappeared; stopping on-demand gateway")
+
             tasks = [
                 asyncio.create_task(
                     watch_config_file(
@@ -890,6 +904,10 @@ def _run_gateway(
                     ),
                     name="nanobot-local-triggers",
                 ),
+                asyncio.create_task(
+                    _monitor_local_clients(),
+                    name="nanobot-gateway-client-monitor",
+                ),
             ]
             if health_server_enabled:
                 tasks.append(asyncio.create_task(

nanobot/cli/tui_launcher.py

@@ -279,16 +279,13 @@ def _ensure_gateway(
                 "stop that instance or use `nanobot agent --classic`"
             )
 
-        result = runtime.start_background(
+        result = lease.ensure_on_demand_gateway(
             GatewayStartOptions(
                 port=config.gateway.port,
                 workspace=workspace_override_path,
                 config_path=str(config_path),
             )
         )
-        started_here = result.ok
-        if started_here:
-            lease.mark_ephemeral()
         if not result.ok and result.message != "gateway_already_running":
             raise TuiUnavailableError(
                 f"could not start the local gateway ({result.message}); "
@@ -309,8 +306,6 @@ def _ensure_gateway(
                 break
             time.sleep(0.1)
 
-        if started_here:
-            runtime.stop(timeout_s=5)
         raise TuiUnavailableError(
             f"local gateway did not become ready; logs: {result.status.log_path}"
         )

nanobot/cli/webui.py

@@ -234,16 +234,13 @@ def webui(
         config_path=config_arg,
     )
 
-    def ensure_shared_gateway(*, client_lease: GatewayClientLease | None = None) -> None:
+    def ensure_shared_gateway(*, client_lease: GatewayClientLease) -> None:
         """Start or refresh the one managed gateway shared by local clients."""
         _prepare_webui_bundle_for_gateway(
             runtime_config,
             mode="skip" if dev else webui_bundle_mode,
         )
-        result = runtime.start_background(start_options)
-        started_fresh = result.ok
-        if started_fresh and client_lease is not None:
-            client_lease.mark_ephemeral()
+        result = client_lease.ensure_on_demand_gateway(start_options)
         restarted = False
         restart_attempted = False
         if not result.ok and result.message == "gateway_already_running" and changed_webui:

nanobot/gateway/runtime.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import json
 import os
@@ -14,7 +15,7 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any, Generator, cast
+from typing import Any, Generator, Literal, cast
 
 from filelock import FileLock
 
@@ -28,8 +29,33 @@ from nanobot.process_runtime import (
 )
 
 GatewayStartOptions = ProcessStartOptions
-GatewayStatus = ProcessStatus
-RuntimeResult = ProcessResult
+
+GatewayLaunchMode = Literal["foreground", "background", "unknown"]
+GatewayLifetime = Literal["explicit", "on_demand"]
+
+
+@dataclass(frozen=True)
+class GatewayStatus(ProcessStatus):
+    """Observable lifecycle state for one shared local gateway."""
+
+    launch_mode: GatewayLaunchMode = "unknown"
+    lifetime: GatewayLifetime = "explicit"
+    clients: int = 0
+
+
+@dataclass(frozen=True)
+class GatewayLeaseSnapshot:
+    """Live local clients and the gateway lifetime they imply."""
+
+    auto_stop: bool
+    clients: int
+
+
+@dataclass(frozen=True)
+class RuntimeResult(ProcessResult):
+    """Result of a gateway lifecycle operation."""
+
+    status: GatewayStatus
 
 
 def build_gateway_command(python_executable: str, options: GatewayStartOptions) -> list[str]:
@@ -104,19 +130,78 @@ class GatewayRuntime(ManagedProcessRuntime[ProcessStartOptions]):
     def _build_child_command(self, options: ProcessStartOptions) -> list[str]:
         return build_gateway_command(self.python_executable, options)
 
+    def start_background(self, options: ProcessStartOptions) -> RuntimeResult:
+        """Start the gateway detached from the current terminal."""
+        with self._lifecycle_lock():
+            return self._start_background(options)
+
+    def start_on_demand(self, options: ProcessStartOptions) -> RuntimeResult:
+        """Atomically reuse a gateway or start one owned by local client leases."""
+        with self._lifecycle_lock():
+            status = self.status()
+            if status.running:
+                return RuntimeResult(False, "gateway_already_running", status)
+            GatewayClientLease(self, kind="gateway-start").mark_ephemeral()
+            return self._start_background(options)
+
+    def _start_background(self, options: ProcessStartOptions) -> RuntimeResult:
+        result = super()._start_background(options)
+        if not result.ok:
+            return self._result(result)
+        state = self._read_state()
+        if state and result.status.pid == state.get("pid"):
+            state["launch_mode"] = "background"
+            self._write_state(state)
+        return RuntimeResult(True, result.message, self.status())
+
+    def stop(self, *, timeout_s: int = 20) -> RuntimeResult:
+        """Stop the gateway recorded by this runtime."""
+        with self._lifecycle_lock():
+            return self._result(self._stop(timeout_s=timeout_s))
+
+    def status(self, *, reason: str | None = None) -> GatewayStatus:
+        """Return process, launch, and client lifetime state in one snapshot."""
+        process = super().status(reason=reason)
+        state = self._read_state() if process.running else None
+        raw_mode = state.get("launch_mode") if state else None
+        launch_mode: GatewayLaunchMode = (
+            raw_mode if raw_mode in {"foreground", "background"} else "unknown"
+        )
+        lease = GatewayClientLease(self, kind="gateway-status").snapshot()
+        return GatewayStatus(
+            running=process.running,
+            pid=process.pid,
+            state_path=process.state_path,
+            log_path=process.log_path,
+            started_at=process.started_at,
+            port=process.port,
+            command=process.command,
+            reason=process.reason,
+            launch_mode=launch_mode,
+            lifetime="on_demand" if lease.auto_stop else "explicit",
+            clients=lease.clients,
+        )
+
     @contextmanager
     def foreground_instance(self, options: ProcessStartOptions) -> Generator[None]:
         """Publish this foreground gateway while it is available to local clients."""
-        self._claim_current_process(options)
+        launch_mode = self._claim_current_process(options)
+        if launch_mode == "foreground":
+            GatewayClientLease(self, kind="gateway-foreground").mark_persistent()
         try:
             yield
         finally:
             self._release_current_process()
 
-    def _claim_current_process(self, options: ProcessStartOptions) -> None:
+    def _claim_current_process(self, options: ProcessStartOptions) -> GatewayLaunchMode:
         with self._lifecycle_lock():
             state = self._read_state() or {}
             pid = os.getpid()
+            launch_mode = (
+                "background"
+                if state.get("pid") == pid and state.get("launch_mode") == "background"
+                else "foreground"
+            )
             state.update(
                 {
                     "pid": pid,
@@ -128,9 +213,11 @@ class GatewayRuntime(ManagedProcessRuntime[ProcessStartOptions]):
                     "config_path": options.config_path,
                     "command": self._build_child_command(options),
                     "log_path": str(self.paths.log_path),
+                    "launch_mode": launch_mode,
                 }
             )
             self._write_state(state)
+            return launch_mode
 
     def _release_current_process(self) -> None:
         with self._lifecycle_lock():
@@ -138,17 +225,24 @@ class GatewayRuntime(ManagedProcessRuntime[ProcessStartOptions]):
             if state and self._record_matches_process(state, os.getpid()):
                 self._clear_state()
 
-    def restart(self, options: ProcessStartOptions, *, timeout_s: int = 20) -> ProcessResult:
+    def restart(self, options: ProcessStartOptions, *, timeout_s: int = 20) -> RuntimeResult:
         """Restart an existing gateway without creating a new persistent instance."""
         with self._lifecycle_lock():
             status = self.status()
             if not status.running:
-                return ProcessResult(False, "gateway_not_running", status)
+                return RuntimeResult(False, "gateway_not_running", status)
+            if status.launch_mode == "foreground":
+                return RuntimeResult(False, "gateway_foreground_restart_required", status)
             stop_result = self._stop(timeout_s=timeout_s)
             if not stop_result.ok:
-                return stop_result
+                return self._result(stop_result)
             return self._start_background(options)
 
+    def _result(self, result: ProcessResult) -> RuntimeResult:
+        status = result.status
+        gateway_status = status if isinstance(status, GatewayStatus) else self.status()
+        return RuntimeResult(result.ok, result.message, gateway_status)
+
 
 class GatewayClientLease:
     """Reference-count an on-demand gateway shared by local interactive clients."""
@@ -180,10 +274,17 @@ class GatewayClientLease:
             clients[self.token] = {
                 "pid": self.pid,
                 "kind": self.kind,
+                "identity": self._process_identity(self.pid),
             }
             self._write_state(state)
             self._acquired = True
 
+    def ensure_on_demand_gateway(self, options: GatewayStartOptions) -> RuntimeResult:
+        """Atomically reuse a gateway or start one owned by local client leases."""
+        if not self._acquired:
+            raise RuntimeError("gateway client lease must be acquired before startup")
+        return self.runtime.start_on_demand(options)
+
     def mark_ephemeral(self) -> None:
         """Mark a gateway started by a client for last-client shutdown."""
         with self.lock:
@@ -205,23 +306,43 @@ class GatewayClientLease:
         with self.lock:
             self.state_path.unlink(missing_ok=True)
 
+    def snapshot(self) -> GatewayLeaseSnapshot:
+        """Prune dead clients and return current lifetime state."""
+        with self.lock:
+            state = self._live_state()
+            self._write_or_clear(state)
+            return GatewayLeaseSnapshot(
+                auto_stop=bool(state.get("auto_stop")),
+                clients=len(self._clients(state)),
+            )
+
+    def orphaned_on_demand(self) -> bool:
+        """Return whether an on-demand gateway has lost every live client."""
+        snapshot = self.snapshot()
+        return snapshot.auto_stop and snapshot.clients == 0
+
     def release(self, *, timeout_s: int = 20) -> bool:
         """Release this client and stop an ephemeral gateway when it was the last."""
         if not self._acquired:
             return False
+        should_stop = False
         with self.lock:
             state = self._live_state()
             clients = self._clients(state)
             clients.pop(self.token, None)
             self._acquired = False
-            if clients or not bool(state.get("auto_stop")):
-                self._write_or_clear(state)
-                return False
-            result = self.runtime.stop(timeout_s=timeout_s)
+            should_stop = not clients and bool(state.get("auto_stop"))
+            self._write_or_clear(state)
+        if not should_stop:
+            return False
+        result = self.runtime.stop(timeout_s=timeout_s)
+        with self.lock:
             stopped = result.ok or result.message == "gateway_not_running"
             if stopped:
                 self.state_path.unlink(missing_ok=True)
             else:
+                state = self._live_state()
+                state["auto_stop"] = True
                 self._write_state(state)
             return stopped
 
@@ -234,12 +355,27 @@ class GatewayClientLease:
                 stale.append(token)
                 continue
             record = cast(dict[str, object], value)
-            if not _pid_is_running(record.get("pid")):
+            pid = record.get("pid")
+            identity = record.get("identity")
+            if (
+                not isinstance(pid, int)
+                or not self._process_is_running(pid)
+                or (identity is not None and identity != self._process_identity(pid))
+            ):
                 stale.append(token)
         for token in stale:
             clients.pop(token, None)
         return state
 
+    def _process_identity(self, pid: int) -> str | int | None:
+        resolver = getattr(self.runtime, "process_identity", None)
+        value = resolver(pid) if callable(resolver) else None
+        return value if isinstance(value, (str, int)) else None
+
+    def _process_is_running(self, pid: int) -> bool:
+        checker = getattr(self.runtime, "process_is_running", None)
+        return bool(checker(pid)) if callable(checker) else _pid_is_running(pid)
+
     @staticmethod
     def _clients(state: dict[str, object]) -> dict[str, object]:
         value = state.get("clients")
@@ -284,6 +420,23 @@ class GatewayClientLease:
             temporary.unlink(missing_ok=True)
 
 
+async def monitor_gateway_clients(
+    lease: GatewayClientLease,
+    shutdown_event: asyncio.Event,
+    *,
+    poll_interval_s: float = 1.0,
+) -> bool:
+    """Stop waiting when an on-demand gateway loses every live client."""
+    while not shutdown_event.is_set():
+        try:
+            await asyncio.wait_for(shutdown_event.wait(), timeout=poll_interval_s)
+        except TimeoutError:
+            if lease.orphaned_on_demand():
+                shutdown_event.set()
+                return True
+    return False
+
+
 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:
@@ -291,11 +444,11 @@ def _instance_suffix(*, workspace: str | None, config_path: str | None) -> str |
     return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16]
 
 
-def _pid_is_running(value: object) -> bool:
-    if not isinstance(value, int) or value <= 0:
+def _pid_is_running(pid: int) -> bool:
+    if pid <= 0:
         return False
     try:
-        os.kill(value, 0)
+        os.kill(pid, 0)
     except ProcessLookupError:
         return False
     except PermissionError:

nanobot/process_runtime.py

@@ -255,6 +255,14 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
         except KeyboardInterrupt:
             return 130
 
+    def process_identity(self, pid: int) -> str | int | None:
+        """Return an identity that changes when an operating-system PID is reused."""
+        return self._process_identity(pid)
+
+    def process_is_running(self, pid: int) -> bool:
+        """Return whether the recorded operating-system process is still live."""
+        return self._is_pid_running(pid)
+
     def _message(self, event: str) -> str:
         return f"{self.service_name}_{event}"
 
@@ -364,9 +372,36 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
         if self.platform_name == "Windows":
             return _windows_process_identity(pid)
         try:
-            return os.getpgid(pid)
+            process_group = os.getpgid(pid)
         except OSError:
             return None
+        started_at = self._posix_process_started_at(pid)
+        return f"{process_group}:{started_at}" if started_at else process_group
+
+    def _posix_process_started_at(self, pid: int) -> str | None:
+        if self.platform_name == "Linux":
+            try:
+                stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
+            except OSError:
+                return None
+            closing_paren = stat.rfind(")")
+            fields = stat[closing_paren + 2 :].split() if closing_paren >= 0 else []
+            # /proc/<pid>/stat fields after comm begin at field 3; starttime is field 22.
+            return fields[19] if len(fields) > 19 else None
+        if self.platform_name == "Darwin":
+            try:
+                result = self._subprocess_run(
+                    ["ps", "-o", "lstart=", "-p", str(pid)],
+                    check=False,
+                    capture_output=True,
+                    text=True,
+                    timeout=1,
+                )
+            except (OSError, subprocess.SubprocessError):
+                return None
+            started_at = getattr(result, "stdout", "").strip()
+            return started_at or None
+        return None
 
     def _record_matches_process(self, state: dict[str, Any] | None, pid: int) -> bool:
         if not state:
@@ -374,7 +409,13 @@ class ManagedProcessRuntime(Generic[_StartOptionsT]):
         recorded = state.get("identity")
         if recorded is None:
             return True
-        return recorded == self._process_identity(pid)
+        current = self._process_identity(pid)
+        if recorded == current:
+            return True
+        # Older POSIX state files stored only the process group id.
+        return isinstance(recorded, int) and isinstance(current, str) and current.startswith(
+            f"{recorded}:"
+        )
 
     def _read_state(self) -> dict[str, Any] | None:
         try:

tests/cli/test_commands.py

@@ -1977,6 +1977,12 @@ def _patch_webui_managed_gateway(
             )
             return RuntimeResult(True, "gateway_started_background", status)
 
+        def start_on_demand(self, options):
+            from nanobot.gateway import GatewayClientLease
+
+            GatewayClientLease(self, kind="test-webui").mark_ephemeral()
+            return self.start_background(options)
+
         def status(self):
             return SimpleNamespace(running=self.running)
 
@@ -2528,6 +2534,8 @@ def test_webui_foreground_attaches_to_existing_managed_gateway(monkeypatch, tmp_
                 status=SimpleNamespace(log_path=self.paths.log_path),
             )
 
+        start_on_demand = start_background
+
         def restart(self, options, *, timeout_s: int):
             seen["restart_options"] = options
             seen["restart_timeout"] = timeout_s

tests/cli/test_gateway_commands.py

@@ -36,6 +36,9 @@ class FakeRuntime:
             started_at="2026-06-22T00:00:00Z",
             port=18790,
             reason="running",
+            launch_mode="background",
+            lifetime="explicit",
+            clients=0,
         )
         self.started_options: GatewayStartOptions | None = None
         self.restarted_options: GatewayStartOptions | None = None
@@ -246,6 +249,29 @@ def test_gateway_background_reports_an_existing_persistent_gateway(tmp_path):
     assert "already running in persistent background mode" in result.stdout
 
 
+def test_gateway_background_does_not_claim_a_foreground_gateway(tmp_path):
+    app, fake_runtime, _service, _calls, _prepare_calls = _test_app(tmp_path)
+    fake_runtime.status_value = GatewayStatus(
+        running=True,
+        pid=12345,
+        state_path=fake_runtime.paths.state_path,
+        log_path=fake_runtime.paths.log_path,
+        launch_mode="foreground",
+    )
+
+    def already_running(_options: GatewayStartOptions) -> RuntimeResult:
+        return RuntimeResult(False, "gateway_already_running", fake_runtime.status_value)
+
+    fake_runtime.start_background = already_running  # type: ignore[method-assign]
+
+    result = runner.invoke(app, ["gateway", "--background"])
+
+    assert result.exit_code == 1
+    assert "cannot be" in result.stdout
+    assert "detached in place" in result.stdout
+    assert "Stop it in its current terminal" in result.stdout
+
+
 def test_gateway_rejects_conflicting_modes(tmp_path):
     app, _runtime, _service, _calls, _prepare_calls = _test_app(tmp_path)
 
@@ -263,6 +289,9 @@ def test_gateway_status_uses_runtime(tmp_path):
     assert result.exit_code == 0
     assert "Running: yes" in result.stdout
     assert "PID: 12345" in result.stdout
+    assert "Launch Mode: background" in result.stdout
+    assert "Lifetime: explicit" in result.stdout
+    assert "Clients: 0" in result.stdout
 
 
 def test_gateway_logs_can_read_without_following(tmp_path):
@@ -330,6 +359,33 @@ def test_gateway_restart_does_not_create_a_persistent_gateway(tmp_path):
     assert "nanobot gateway --background" in result.stdout
 
 
+def test_gateway_restart_explains_foreground_lifecycle(tmp_path):
+    app, fake_runtime, _service, _calls, _prepare_calls = _test_app(tmp_path)
+
+    def foreground_restart(
+        _options: GatewayStartOptions, *, timeout_s: int
+    ) -> RuntimeResult:
+        fake_runtime.stop_timeout = timeout_s
+        return RuntimeResult(
+            False,
+            "gateway_foreground_restart_required",
+            GatewayStatus(
+                running=True,
+                pid=12345,
+                state_path=fake_runtime.paths.state_path,
+                log_path=fake_runtime.paths.log_path,
+                launch_mode="foreground",
+            ),
+        )
+
+    fake_runtime.restart = foreground_restart  # type: ignore[method-assign]
+
+    result = runner.invoke(app, ["gateway", "restart"])
+
+    assert result.exit_code == 1
+    assert "attached to a foreground terminal" in result.stdout
+
+
 def test_gateway_install_service_uses_service_installer(tmp_path):
     config = Config()
     config.gateway.port = 18794

tests/cli/test_tui_launcher.py

@@ -485,6 +485,12 @@ def test_gateway_started_for_tui_stops_when_its_last_lease_exits(
                 status=SimpleNamespace(log_path=tmp_path / "gateway.log"),
             )
 
+        def start_on_demand(self, options: object) -> SimpleNamespace:
+            from nanobot.gateway import GatewayClientLease
+
+            GatewayClientLease(self, kind="test-tui").mark_ephemeral()
+            return self.start_background(options)
+
         def stop(self, *, timeout_s: int) -> SimpleNamespace:
             nonlocal stopped
             assert timeout_s == 20

tests/gateway/test_runtime.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import os
 import signal
@@ -16,6 +17,7 @@ from nanobot.gateway import (
     GatewayStartOptions,
     GatewayStatus,
 )
+from nanobot.gateway.runtime import monitor_gateway_clients
 
 
 class FakeProcess:
@@ -102,6 +104,8 @@ def test_start_background_writes_state_and_child_command(tmp_path, monkeypatch):
     assert state["pid"] == 12345
     assert state["identity"] == 12345
     assert state["port"] == 18790
+    assert state["launch_mode"] == "background"
+    assert result.status.launch_mode == "background"
 
 
 def test_foreground_gateway_claim_is_discoverable_and_released(tmp_path, monkeypatch):
@@ -123,12 +127,26 @@ def test_foreground_gateway_claim_is_discoverable_and_released(tmp_path, monkeyp
         assert status.running is True
         assert status.pid == os.getpid()
         assert status.port == 18790
+        assert status.launch_mode == "foreground"
+        assert status.lifetime == "explicit"
         assert status.command == tuple(runtime._build_child_command(options))
 
     assert runtime.status().running is False
     assert not runtime.paths.state_path.exists()
 
 
+def test_explicit_foreground_gateway_clears_stale_auto_stop_state(tmp_path, monkeypatch):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Darwin")
+    monkeypatch.setattr(runtime, "_process_identity", lambda pid: pid)
+    lease = GatewayClientLease(runtime, kind="stale")
+    lease.mark_ephemeral()
+
+    with runtime.foreground_instance(GatewayStartOptions(port=18790)):
+        assert runtime.status().lifetime == "explicit"
+
+    assert not lease.state_path.exists()
+
+
 def test_foreground_gateway_release_preserves_a_replacement_state(tmp_path, monkeypatch):
     runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Darwin")
     monkeypatch.setattr(runtime, "_process_identity", lambda pid: pid)
@@ -250,6 +268,28 @@ def test_restart_does_not_start_a_gateway_that_is_not_running(tmp_path):
     assert spawned == []
 
 
+def test_restart_does_not_detach_a_foreground_gateway(tmp_path, monkeypatch):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
+    status = GatewayStatus(
+        running=True,
+        pid=12345,
+        state_path=runtime.paths.state_path,
+        log_path=runtime.paths.log_path,
+        launch_mode="foreground",
+    )
+    monkeypatch.setattr(runtime, "status", lambda **_kwargs: status)
+    monkeypatch.setattr(
+        runtime,
+        "_stop",
+        lambda **_kwargs: pytest.fail("foreground gateway must not be stopped"),
+    )
+
+    result = runtime.restart(GatewayStartOptions(port=18790))
+
+    assert result.ok is False
+    assert result.message == "gateway_foreground_restart_required"
+
+
 def test_last_interactive_client_stops_an_on_demand_gateway(tmp_path, monkeypatch):
     runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
     stopped: list[int] = []
@@ -273,6 +313,34 @@ def test_last_interactive_client_stops_an_on_demand_gateway(tmp_path, monkeypatc
     assert not webui.state_path.exists()
 
 
+def test_on_demand_lifetime_is_recorded_before_the_gateway_spawns(tmp_path, monkeypatch):
+    observed_auto_stop: list[bool] = []
+
+    def fake_popen(*_args, **_kwargs):
+        lease_path = runtime.paths.state_path.with_name("gateway.clients.json")
+        observed_auto_stop.append(
+            json.loads(lease_path.read_text(encoding="utf-8"))["auto_stop"]
+        )
+        return FakeProcess()
+
+    runtime = GatewayRuntime(
+        paths=_paths(tmp_path),
+        platform_name="Linux",
+        popen=fake_popen,
+        sleep=lambda _seconds: None,
+    )
+    monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
+    monkeypatch.setattr(runtime, "_process_identity", lambda _pid: 12345)
+    client = GatewayClientLease(runtime, kind="tui", token="client")
+    client.acquire()
+
+    result = client.ensure_on_demand_gateway(GatewayStartOptions(port=18790))
+
+    assert result.ok is True
+    assert observed_auto_stop == [True]
+    assert result.status.lifetime == "on_demand"
+
+
 def test_explicit_background_gateway_survives_the_last_client(tmp_path, monkeypatch):
     runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
     stopped: list[int] = []
@@ -308,6 +376,42 @@ def test_failed_last_client_shutdown_remains_retryable(tmp_path, monkeypatch):
     assert state == {"auto_stop": True, "clients": {}}
 
 
+def test_lease_snapshot_prunes_a_reused_client_pid(tmp_path, monkeypatch):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
+    identity = "same-pid:first-process"
+    monkeypatch.setattr(runtime, "_is_pid_running", lambda _pid: True)
+    monkeypatch.setattr(runtime, "_process_identity", lambda _pid: identity)
+    client = GatewayClientLease(runtime, kind="tui", pid=12345, token="client")
+
+    client.acquire()
+    client.mark_ephemeral()
+    identity = "same-pid:replacement-process"
+
+    snapshot = client.snapshot()
+    assert snapshot.auto_stop is True
+    assert snapshot.clients == 0
+    assert json.loads(client.state_path.read_text(encoding="utf-8")) == {
+        "auto_stop": True,
+        "clients": {},
+    }
+
+
+async def test_client_monitor_stops_an_orphaned_on_demand_gateway(tmp_path):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
+    lease = GatewayClientLease(runtime, kind="gateway-monitor")
+    lease.mark_ephemeral()
+    shutdown_event = asyncio.Event()
+
+    orphaned = await monitor_gateway_clients(
+        lease,
+        shutdown_event,
+        poll_interval_s=0.001,
+    )
+
+    assert orphaned is True
+    assert shutdown_event.is_set()
+
+
 def test_start_background_uses_windows_process_group_flags(tmp_path, monkeypatch):
     calls: list[dict] = []
 
@@ -359,6 +463,18 @@ def test_status_clears_state_when_pid_identity_changes(tmp_path, monkeypatch):
     assert not runtime.paths.state_path.exists()
 
 
+def test_posix_process_identity_includes_start_time_and_accepts_legacy_state(
+    tmp_path,
+    monkeypatch,
+):
+    runtime = GatewayRuntime(paths=_paths(tmp_path), platform_name="Linux")
+    monkeypatch.setattr("nanobot.process_runtime.os.getpgid", lambda _pid: 42)
+    monkeypatch.setattr(runtime, "_posix_process_started_at", lambda _pid: "987654")
+
+    assert runtime.process_identity(12345) == "42:987654"
+    assert runtime._record_matches_process({"identity": 42}, 12345) is True
+
+
 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)