fix(tui): preserve gateway and cache boundaries

Only reuse the exact managed config/workspace instance so CLI overrides cannot silently attach to another gateway. Revalidate cached release sidecars before execution and recover from corrupted cache entries.

Co-authored-by: Bingxi Zhao <150592536+pancacake@users.noreply.github.com>

nanobot/cli/tui_launcher.py

@@ -24,7 +24,7 @@ from nanobot.cli.webui_support import (
     _webui_endpoint_reachable,
     webui_bootstrap_secret,
 )
-from nanobot.config.paths import get_data_dir
+from nanobot.config.paths import get_data_dir, is_default_workspace
 from nanobot.config.schema import Config
 
 
@@ -140,8 +140,26 @@ def _download_release_tui(asset: str) -> Path | None:
 
     target_dir = get_data_dir() / "bin" / "tui" / version
     target = target_dir / asset
+    cached_checksum = target.with_name(f"{target.name}.sha256")
     if target.is_file():
-        return target
+        try:
+            expected = cached_checksum.read_text(encoding="utf-8").split()[0].lower()
+            actual = hashlib.sha256(target.read_bytes()).hexdigest()
+        except (OSError, IndexError):
+            expected = ""
+            actual = ""
+        if len(expected) == 64 and actual == expected:
+            if os.name != "nt":
+                try:
+                    target.chmod(0o755)
+                except OSError:
+                    return None
+            return target
+        try:
+            target.unlink(missing_ok=True)
+            cached_checksum.unlink(missing_ok=True)
+        except OSError:
+            return None
 
     base = f"https://github.com/HKUDS/nanobot/releases/download/v{version}"
     try:
@@ -156,14 +174,22 @@ def _download_release_tui(asset: str) -> Path | None:
         raise TuiUnavailableError("downloaded TUI binary failed checksum verification")
 
     temporary = target.with_suffix(f"{target.suffix}.tmp-{os.getpid()}")
+    temporary_checksum = cached_checksum.with_suffix(
+        f"{cached_checksum.suffix}.tmp-{os.getpid()}"
+    )
     try:
         target_dir.mkdir(parents=True, exist_ok=True)
         temporary.write_bytes(binary)
+        temporary_checksum.write_text(f"{expected}  {asset}\n", encoding="utf-8")
         if os.name != "nt":
             temporary.chmod(0o755)
         temporary.replace(target)
+        temporary_checksum.replace(cached_checksum)
     except OSError:
         temporary.unlink(missing_ok=True)
+        temporary_checksum.unlink(missing_ok=True)
+        target.unlink(missing_ok=True)
+        cached_checksum.unlink(missing_ok=True)
         return None
     return target
 
@@ -189,25 +215,42 @@ def _ensure_gateway(
     from nanobot.gateway import GatewayRuntime, GatewayRuntimePaths, GatewayStartOptions
 
     base_url = _webui_browser_url(config).split("/#/", 1)[0].rstrip("/")
-    if _webui_endpoint_reachable(base_url):
-        return _GatewayLease(runtime=None, owned=False, base_url=base_url)
-
-    workspace = (
+    workspace_override_path = (
         str(Path(workspace_override).expanduser().resolve(strict=False))
         if workspace_override
         else None
     )
+    effective_workspace = config.workspace_path.resolve(strict=False)
+    runtime_workspace = (
+        None if is_default_workspace(effective_workspace) else str(effective_workspace)
+    )
     runtime = GatewayRuntime(
         paths=GatewayRuntimePaths.for_instance(
             data_dir=config_path.parent,
-            workspace=workspace,
+            workspace=runtime_workspace,
             config_path=str(config_path),
         )
     )
+    status = runtime.status()
+    endpoint_reachable = _webui_endpoint_reachable(base_url)
+    if status.running:
+        if status.port not in {None, config.gateway.port}:
+            raise TuiUnavailableError(
+                "the matching gateway instance is running on a different port; "
+                "restart it or use `nanobot agent --classic`"
+            )
+        if endpoint_reachable:
+            return _GatewayLease(runtime=runtime, owned=False, base_url=base_url)
+    elif endpoint_reachable:
+        raise TuiUnavailableError(
+            "the configured gateway port belongs to a different nanobot instance; "
+            "stop that instance or use `nanobot agent --classic`"
+        )
+
     result = runtime.start_background(
         GatewayStartOptions(
             port=config.gateway.port,
-            workspace=workspace,
+            workspace=workspace_override_path,
             config_path=str(config_path),
         )
     )
@@ -220,7 +263,10 @@ def _ensure_gateway(
     deadline = time.monotonic() + 20
     while time.monotonic() < deadline:
         if _webui_endpoint_reachable(base_url):
-            return _GatewayLease(runtime=runtime, owned=owned, base_url=base_url)
+            current = runtime.status()
+            if current.running and current.port in {None, config.gateway.port}:
+                return _GatewayLease(runtime=runtime, owned=owned, base_url=base_url)
+            break
         if not runtime.status().running and not _gateway_health_ready(
             config.gateway.host,
             config.gateway.port,

tests/cli/test_tui_launcher.py

@@ -9,6 +9,7 @@ from nanobot.cli.tui_launcher import (
     TuiUnavailableError,
     _authenticated_ws_url,
     _download_release_tui,
+    _ensure_gateway,
     _resolve_tui_command,
     _websocket_chat_id,
 )
@@ -102,12 +103,41 @@ def test_release_tui_is_verified_and_cached(
 
     assert target == tmp_path / "bin" / "tui" / "9.9.9" / "nanobot-tui-linux-x64"
     assert target.read_bytes() == binary
+    assert target.with_name(f"{target.name}.sha256").read_text().startswith(digest.decode())
     assert len(downloads) == 2
 
     assert _download_release_tui("nanobot-tui-linux-x64") == target
     assert len(downloads) == 2
 
 
+def test_release_tui_replaces_a_corrupted_cached_binary(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    binary = b"native-tui"
+    digest = hashlib.sha256(binary).hexdigest().encode()
+    asset = "nanobot-tui-linux-x64"
+    target = tmp_path / "bin" / "tui" / "9.9.9" / asset
+    target.parent.mkdir(parents=True)
+    target.write_bytes(b"corrupted")
+    target.with_name(f"{target.name}.sha256").write_text(
+        f"{hashlib.sha256(binary).hexdigest()}  {asset}\n"
+    )
+    downloads: list[str] = []
+
+    def read_asset(url: str, *, max_bytes: int) -> bytes:
+        downloads.append(url)
+        return digest if url.endswith(".sha256") else binary
+
+    monkeypatch.setattr("nanobot.cli.tui_launcher.__version__", "9.9.9")
+    monkeypatch.setattr("nanobot.cli.tui_launcher.get_data_dir", lambda: tmp_path)
+    monkeypatch.setattr("nanobot.cli.tui_launcher._read_release_asset", read_asset)
+
+    assert _download_release_tui(asset) == target
+    assert target.read_bytes() == binary
+    assert len(downloads) == 2
+
+
 def test_release_tui_rejects_bad_checksum(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -121,3 +151,66 @@ def test_release_tui_rejects_bad_checksum(
 
     with pytest.raises(TuiUnavailableError, match="checksum"):
         _download_release_tui("nanobot-tui-linux-x64")
+
+
+def test_gateway_reuse_requires_the_matching_managed_instance(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    config = Config()
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    config.agents.defaults.workspace = str(workspace)
+
+    class FakeRuntime:
+        def __init__(self, *, paths: object) -> None:
+            self.paths = paths
+
+        def status(self) -> SimpleNamespace:
+            return SimpleNamespace(running=False, port=None)
+
+    monkeypatch.setattr("nanobot.gateway.GatewayRuntime", FakeRuntime)
+    monkeypatch.setattr("nanobot.cli.tui_launcher._webui_endpoint_reachable", lambda _url: True)
+
+    with pytest.raises(TuiUnavailableError, match="different nanobot instance"):
+        _ensure_gateway(
+            config,
+            config_path=tmp_path / "config.json",
+            workspace_override=str(workspace),
+        )
+
+
+def test_gateway_reuses_the_matching_managed_instance(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    config = Config()
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    config.agents.defaults.workspace = str(workspace)
+    runtime: object | None = None
+
+    class FakeRuntime:
+        def __init__(self, *, paths: object) -> None:
+            nonlocal runtime
+            self.paths = paths
+            runtime = self
+
+        def status(self) -> SimpleNamespace:
+            return SimpleNamespace(running=True, port=config.gateway.port)
+
+        def stop(self, *, timeout_s: int) -> None:
+            raise AssertionError(f"unowned gateway stopped with timeout {timeout_s}")
+
+    monkeypatch.setattr("nanobot.gateway.GatewayRuntime", FakeRuntime)
+    monkeypatch.setattr("nanobot.cli.tui_launcher._webui_endpoint_reachable", lambda _url: True)
+
+    lease = _ensure_gateway(
+        config,
+        config_path=tmp_path / "config.json",
+        workspace_override=str(workspace),
+    )
+
+    assert lease.runtime is runtime
+    assert lease.owned is False
+    lease.close()