Add on-demand transport prompt output

tests/test_cli_open.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from dataclasses import replace
+import io
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+
+from ampbrowser.cli import main
+from ampbrowser.transport_manager import ManagedTransportResult
+from ampbrowser.transports import TransportStatus
+
+
+class CliOpenTest(unittest.TestCase):
+    def test_onion_launch_without_approval_prints_prompt_without_starting_transport(self) -> None:
+        status = TransportStatus(
+            transport="tor",
+            installed=True,
+            running=False,
+            endpoint="socks5://127.0.0.1:9050",
+            adoptable=False,
+            manage_supported=True,
+            note="Tor SOCKS proxy",
+        )
+
+        with patch("ampbrowser.plan.inspect_transport", return_value=status):
+            with patch("ampbrowser.cli.ensure_transport_ready") as ensure_transport_ready:
+                with patch.object(sys, "stdout", new_callable=io.StringIO) as stdout:
+                    code = main(["open", "http://example.onion/", "--launch"])
+
+        self.assertEqual(0, code)
+        ensure_transport_ready.assert_not_called()
+        output = stdout.getvalue()
+        self.assertIn("status=consent-required", output)
+        self.assertIn('setup_prompt_title="Set up Tor?"', output)
+        self.assertIn("setup_prompt_approval_command=", output)
+        self.assertIn("transport_setup_status=-", output)
+
+    def test_onion_launch_with_approval_starts_transport_before_browser_launch(self) -> None:
+        status = TransportStatus(
+            transport="tor",
+            installed=True,
+            running=False,
+            endpoint="socks5://127.0.0.1:9050",
+            adoptable=False,
+            manage_supported=True,
+            note="Tor SOCKS proxy",
+        )
+        transport_result = ManagedTransportResult(
+            transport="tor",
+            provider="arti",
+            status="started",
+            endpoint="socks5://127.0.0.1:9050",
+            owned=True,
+            pid=1234,
+            state_dir=".ampb/transports/tor",
+            command=("/tmp/arti", "proxy"),
+            message="started managed arti transport",
+        )
+
+        def fake_execute(open_plan, *, root: Path):
+            return replace(
+                open_plan,
+                status="launched",
+                dry_run=False,
+                browser_pid=5678,
+                message="launched bundled browser",
+            )
+
+        with patch("ampbrowser.plan.inspect_transport", return_value=status):
+            with patch("ampbrowser.cli.ensure_transport_ready", return_value=transport_result) as ensure_transport_ready:
+                with patch("ampbrowser.cli.execute_open", side_effect=fake_execute) as execute_open:
+                    with patch.object(sys, "stdout", new_callable=io.StringIO) as stdout:
+                        code = main(["open", "http://example.onion/", "--yes", "--launch"])
+
+        self.assertEqual(0, code)
+        ensure_transport_ready.assert_called_once()
+        execute_open.assert_called_once()
+        output = stdout.getvalue()
+        self.assertIn("status=launched", output)
+        self.assertIn("browser_pid=5678", output)
+        self.assertIn("transport_setup_status=started", output)
+        self.assertIn("transport_setup_provider=arti", output)
+        self.assertIn("transport_setup_owned=true", output)
+        self.assertIn("transport_setup_pid=1234", output)
+        self.assertIn("transport_setup_endpoint=socks5://127.0.0.1:9050", output)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_launch.py

@@ -49,6 +49,14 @@ class PrepareOpenTest(unittest.TestCase):
             plan.setup_steps,
         )
         self.assertIn("install and start", plan.message)
+        self.assertEqual("Set up I2P?", plan.setup_prompt_title)
+        self.assertIn("I2P is not installed or running.", plan.setup_prompt_body)
+        self.assertIn("http://example.b32.i2p/", plan.setup_prompt_body)
+        self.assertEqual("Start I2P and open", plan.setup_prompt_approve_label)
+        self.assertEqual(
+            "ampbrowser open http://example.b32.i2p/ --yes --launch",
+            plan.setup_prompt_approval_command,
+        )
 
     def test_yes_approves_setup_without_executing_it(self) -> None:
         status = TransportStatus(
@@ -68,6 +76,9 @@ class PrepareOpenTest(unittest.TestCase):
         self.assertTrue(plan.dry_run)
         self.assertTrue(plan.consent_granted)
         self.assertEqual(("start managed Arti SOCKS proxy", "wait for socks5://127.0.0.1:9050"), plan.setup_steps)
+        self.assertEqual("Set up Tor?", plan.setup_prompt_title)
+        self.assertIn("Tor is not running.", plan.setup_prompt_body)
+        self.assertEqual("Start Tor and open", plan.setup_prompt_approve_label)
 
     def test_adopted_tor_launch_spec_sets_socks_remote_dns(self) -> None:
         status = TransportStatus(

tests/test_transport_manager.py

@@ -139,6 +139,20 @@ class TransportManagerTest(unittest.TestCase):
         self.assertEqual("stale", result.status)
         self.assertFalse(state_path.exists())
 
+    def test_transport_status_reports_stale_cleanup_failure(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            state_dir = root / ".ampb/transports/tor"
+            state_dir.mkdir(parents=True)
+            state_path = state_dir / "ampb-owned.json"
+            state_path.write_text('{"pid": 1234, "provider": "arti", "endpoint": "socks5://127.0.0.1:9050"}\n', encoding="utf-8")
+            with patch("ampbrowser.transport_manager._pid_alive", return_value=False):
+                with patch("ampbrowser.transport_manager._unlink_state", return_value=False):
+                    result = transport_status("tor", config=AppConfig(transport_modes={}), root=root)
+
+        self.assertEqual("stale-cleanup-failed", result.status)
+        self.assertIn("could not remove stale", result.message)
+
     def test_stop_managed_transport_terminates_recorded_pid(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)