fix(preview): render macOS text and own complete POSIX launches

Enable the native font backend, fit initial windows inside the usable display, and require terminal text in installed macOS screenshots. Isolate POSIX launch groups and reject runtime endpoints outside the current launch so AppImage restarts cannot reconnect to orphaned instances.

scripts/conformance/harness.py

@@ -10,6 +10,7 @@ import json
 import os
 import re
 import shutil
+import signal
 import socket
 import subprocess
 import sys
@@ -521,6 +522,13 @@ class ConformanceContext:
                     )
                     snapshot = snapshot_response.get("result")
                     require(isinstance(snapshot, dict), "runtime snapshot is not an object")
+                    if os.name == "posix":
+                        runtime_pid = snapshot.get("process_id")
+                        require(
+                            type(runtime_pid) is int and runtime_pid > 0
+                            and os.getpgid(runtime_pid) == self.process.pid,
+                            "runtime process does not belong to this launch",
+                        )
                     windows = snapshot.get("windows") or []
                     ready = any(
                         any(
@@ -548,7 +556,9 @@ class ConformanceContext:
             "stderr": subprocess.STDOUT,
         }
         if os.name != "nt":
-            return subprocess.Popen(command, stdin=subprocess.DEVNULL, **options)
+            return subprocess.Popen(
+                command, stdin=subprocess.DEVNULL, start_new_session=True, **options,
+            )
         job = _WindowsJob()
         process = None
         try:
@@ -586,15 +596,20 @@ class ConformanceContext:
                 if process is not None:
                     process.wait(timeout=max(0, deadline - time.monotonic()))
                 self._windows_job.close()
-            elif process is not None and process.poll() is None:
-                if force:
-                    process.kill()
-                else:
-                    process.terminate()
+            elif process is not None:
+                # An AppImage launcher forks the application. Its private
+                # group remains ours even after the launcher itself exits.
+                try:
+                    os.killpg(process.pid, signal.SIGKILL if force else signal.SIGTERM)
+                except ProcessLookupError:
+                    pass
                 try:
                     process.wait(timeout=timeout)
                 except subprocess.TimeoutExpired:
-                    process.kill()
+                    try:
+                        os.killpg(process.pid, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
                     process.wait(timeout=timeout)
         finally:
             self._close_log()

scripts/conformance/macos_launch.py

@@ -23,6 +23,25 @@ if str(SCRIPTS) not in sys.path:
 from conformance.harness import ConformanceContext, ResolvedApp, RuntimeClient, require
 
 
+def verify_rendered_text(ctx: ConformanceContext, probe: Path, log: BinaryIO) -> None:
+    marker = "PEBREL TEXT READY"
+    # Clear the echoed command before printing the marker; only rendered output
+    # can satisfy the screenshot observation.
+    ctx.prompt("printf '\\033[2J\\033[H%s%s\\n' 'PEBREL TEXT ' 'READY'")
+    ctx.wait_for_line(re.compile(r"^" + re.escape(marker) + r"$"))
+    screenshot = ctx.artifact_dir / "installed-app.png"
+    deadline = time.monotonic() + 15
+    while time.monotonic() < deadline:
+        require(capture_screenshot(screenshot, log) == "captured", "installed app screenshot is unavailable")
+        result = subprocess.run([str(probe), str(screenshot), marker],
+                                stdout=log, stderr=subprocess.STDOUT, timeout=15, check=False)
+        require(result.returncode in (0, 1), "screenshot text recognition failed")
+        if result.returncode == 0:
+            return
+        time.sleep(0.1)
+    raise RuntimeError("installed app did not render the terminal text marker")
+
+
 def capture_screenshot(destination: Path, log: BinaryIO) -> str:
     try:
         capture = subprocess.run(["/usr/sbin/screencapture", "-x", str(destination)],
@@ -72,6 +91,11 @@ def main() -> int:
             launcher = None
             with (logs / "launchservices.log").open("wb") as log:
                 try:
+                    text_probe = root / "macos-text"
+                    subprocess.run([
+                        "/usr/bin/xcrun", "swiftc", str(SCRIPTS / "conformance/macos_text.swift"),
+                        "-module-cache-path", str(root / "swift-cache"), "-o", str(text_probe),
+                    ], stdout=log, stderr=subprocess.STDOUT, timeout=90, check=True)
                     launcher = subprocess.Popen(
                         ["/usr/bin/open", "-n", "-W", "-a", str(installed),
                          "--env", f"NEBULA_CONFIG_DIR={ctx.config_dir}",
@@ -102,7 +126,8 @@ def main() -> int:
                     ctx.wait_for_line(re.compile(r"NEBULA_GUI_LOCALE=[^\r\n]*UTF-?8_END", re.IGNORECASE))
                     ctx.prompt('printf "NEBULA_GUI_CWD=%s_END\\n" "$PWD"')
                     ctx.wait_for_line(re.compile(r"NEBULA_GUI_CWD=" + re.escape(str(Path.home())) + r"_END"))
-                    report["screenshot"] = capture_screenshot(logs / "installed-app.png", log)
+                    verify_rendered_text(ctx, text_probe, log)
+                    report.update(screenshot="captured", rendered_text=True)
                     for window in ctx.snapshot().get("windows", []):
                         ctx.api("window.close", {"window_id": window["id"]})
                     deadline = time.monotonic() + 5

scripts/conformance/macos_text.swift

@@ -0,0 +1,26 @@
+import Foundation
+import Vision
+
+guard CommandLine.arguments.count == 3 else {
+    fputs("usage: macos-text IMAGE EXPECTED_TEXT\n", stderr)
+    exit(2)
+}
+
+do {
+    let request = VNRecognizeTextRequest()
+    request.recognitionLevel = .accurate
+    request.usesLanguageCorrection = false
+    request.recognitionLanguages = ["en-US"]
+    let handler = VNImageRequestHandler(url: URL(fileURLWithPath: CommandLine.arguments[1]))
+    try handler.perform([request])
+    let expected = CommandLine.arguments[2].uppercased()
+    let found = (request.results ?? []).contains { observation in
+        observation.topCandidates(1).contains { candidate in
+            candidate.string.uppercased().contains(expected)
+        }
+    }
+    exit(found ? 0 : 1)
+} catch {
+    fputs("screenshot text recognition failed: \(error)\n", stderr)
+    exit(2)
+}

scripts/conformance/tests/test_posix_lifecycle.py

@@ -0,0 +1,206 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import select
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+SCRIPTS_DIR = Path(__file__).resolve().parents[2]
+if str(SCRIPTS_DIR) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS_DIR))
+
+from conformance.harness import ConformanceContext, ConformanceError
+
+
+REAL_POPEN = subprocess.Popen
+FIXTURE = """
+from pathlib import Path
+import os, subprocess, sys, time
+
+root = Path(sys.argv[1])
+role = sys.argv[2]
+if role != 'grandchild':
+    next_role = 'child' if role == 'parent' else 'grandchild'
+    subprocess.Popen([sys.executable, '-I', '-S', __file__, str(root), next_role])
+(root / (role + '.pid')).write_text(str(os.getpid()))
+while True:
+    if role == 'parent' and (root / 'exit').exists():
+        code = root / 'exit-code'
+        raise SystemExit(int(code.read_text()) if code.exists() else 0)
+    time.sleep(0.01)
+"""
+
+
+@unittest.skipUnless(os.name == "posix", "POSIX launch process groups")
+class PosixLifecycleTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temporary = tempfile.TemporaryDirectory(prefix="nebula-posix-lifecycle-")
+        self.root = Path(self.temporary.name)
+        self.fixture = self.root / "fixture.py"
+        self.fixture.write_text(FIXTURE, encoding="utf-8")
+        self.app = SimpleNamespace(executable=self.root / "fixture-app")
+        self.context = ConformanceContext(
+            self.app, "linux-test", self.root / "config", self.root / "work",
+            self.root / "artifacts", startup_timeout=2,
+        )
+        self.context.prepare()
+        self.context.port_file.write_text("1 fixture-token", encoding="utf-8")
+        self.parents: list[subprocess.Popen] = []
+        self.launches: list[Path] = []
+        self.descendants: list[int] = []
+        self.unrelated: subprocess.Popen | None = None
+        self.addCleanup(self.cleanup_owned_processes)
+
+        def launch_fixture(command, **kwargs):
+            launch = self.root / f"launch-{len(self.launches) + 1}"
+            launch.mkdir()
+            self.launches.append(launch)
+            # All fixture descendants inherit this pipe. EOF proves they have
+            # actually exited even when their former parent is already reaped.
+            kwargs["stdout"] = subprocess.PIPE
+            process = REAL_POPEN([
+                sys.executable, "-I", "-S", os.fspath(self.fixture),
+                os.fspath(launch), "parent",
+            ], **kwargs)
+            self.parents.append(process)
+            return process
+
+        self.client = SimpleNamespace(request=lambda *args, **kwargs: {
+            "result": {
+                "process_id": self.parents[-1].pid,
+                "windows": [{"tabs": [{"kind": "shell", "panes": [{}]}]}],
+            },
+        })
+        self.launch_patch = patch("conformance.harness.subprocess.Popen", side_effect=launch_fixture)
+        self.client_patch = patch(
+            "conformance.harness.RuntimeClient.from_port_file", return_value=self.client,
+        )
+        self.launch_patch.start()
+        self.client_patch.start()
+        self.addCleanup(self.launch_patch.stop)
+        self.addCleanup(self.client_patch.stop)
+
+    def capture_descendants(self) -> None:
+        for role in ("child", "grandchild"):
+            path = self.launches[-1] / f"{role}.pid"
+            deadline = time.monotonic() + 5
+            while not path.is_file() and time.monotonic() < deadline:
+                time.sleep(0.01)
+            self.assertTrue(path.is_file(), f"fixture never created {role}")
+            self.descendants.append(int(path.read_text()))
+
+    def assert_tree_exited(self, process: subprocess.Popen) -> None:
+        self.assertTrue(select.select([process.stdout], [], [], 2)[0], "launch descendants survived")
+        self.assertEqual(os.read(process.stdout.fileno(), 65536), b"")
+
+    def cleanup_owned_processes(self) -> None:
+        self.context.stop(force=True)
+        # Keep the negative fixture bounded when running against the old harness.
+        for launch in self.launches:
+            for role in ("child", "grandchild"):
+                path = launch / f"{role}.pid"
+                if path.is_file() and path.read_text():
+                    self.descendants.append(int(path.read_text()))
+        for pid in set(self.descendants):
+            try:
+                os.kill(pid, signal.SIGKILL)
+            except ProcessLookupError:
+                pass
+        for process in [*self.parents, self.unrelated]:
+            if process is not None:
+                if process.poll() is None:
+                    process.kill()
+                process.wait(timeout=5)
+                if process.stdout is not None:
+                    process.stdout.close()
+        self.temporary.cleanup()
+
+    def test_force_stop_terminates_the_launch_but_not_an_unrelated_process(self) -> None:
+        self.unrelated = REAL_POPEN(
+            [sys.executable, "-I", "-S", "-c", "import time; time.sleep(60)"],
+            stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        )
+        self.context.start()
+        self.capture_descendants()
+        self.context.stop(force=True)
+        self.assert_tree_exited(self.parents[-1])
+        self.assertIsNone(self.unrelated.poll())
+
+    def test_stop_terminates_children_after_the_wrapper_has_exited(self) -> None:
+        self.context.start()
+        self.capture_descendants()
+        (self.launches[-1] / "exit").touch()
+        self.context.process.wait(timeout=5)
+        self.context.stop(force=True)
+        self.assert_tree_exited(self.parents[-1])
+
+    def test_restart_terminates_the_previous_launch(self) -> None:
+        self.context.start()
+        self.capture_descendants()
+        self.context.restart()
+        self.capture_descendants()
+        self.assert_tree_exited(self.parents[0])
+        self.assertFalse(select.select([self.parents[1].stdout], [], [], 0)[0])
+
+    def test_normal_close_terminates_leftover_children_and_preserves_exit_code(self) -> None:
+        for code in (0, 7):
+            with self.subTest(code=code):
+                self.context.start()
+                self.capture_descendants()
+                (self.launches[-1] / "exit-code").write_text(str(code))
+                with patch.object(
+                    self.context, "api",
+                    side_effect=lambda *args, **kwargs: (self.launches[-1] / "exit").touch(),
+                ):
+                    if code:
+                        with self.assertRaisesRegex(ConformanceError, "Nebula exited with code 7"):
+                            self.context.close_and_wait()
+                    else:
+                        self.context.close_and_wait()
+                self.assert_tree_exited(self.parents[-1])
+
+    def test_start_rejects_an_endpoint_from_another_launch(self) -> None:
+        self.context.startup_timeout = 0.15
+        self.client.request = lambda *args, **kwargs: {
+            "result": {
+                "process_id": os.getpid(),
+                "windows": [{"tabs": [{"kind": "shell", "panes": [{}]}]}],
+            },
+        }
+        with self.assertRaisesRegex(ConformanceError, "runtime process does not belong to this launch"):
+            self.context.start()
+        self.assert_tree_exited(self.parents[-1])
+
+    def test_start_accepts_the_application_beneath_a_wrapper(self) -> None:
+        self.client.request = lambda *args, **kwargs: {
+            "result": {
+                "process_id": int((self.launches[-1] / "child.pid").read_text()),
+                "windows": [{"tabs": [{"kind": "shell", "panes": [{}]}]}],
+            },
+        }
+        self.context.start()
+        self.capture_descendants()
+        self.context.stop(force=True)
+        self.assert_tree_exited(self.parents[-1])
+
+    def test_cleanup_failure_retains_the_owned_group_for_retry(self) -> None:
+        self.context.start()
+        self.capture_descendants()
+        process = self.context.process
+        with patch("conformance.harness.os.killpg", side_effect=OSError("cleanup unavailable")):
+            with self.assertRaisesRegex(OSError, "cleanup unavailable"):
+                self.context.stop(force=True)
+        self.assertIs(self.context.process, process)
+        self.context.stop(force=True)
+        self.assert_tree_exited(process)
+
+
+if __name__ == "__main__":
+    unittest.main()