Merge origin/main into live-touch

Kept both route tables: main's media, agent, assistant and Mac view routes
plus /api/touch.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

tests/assistant_ui.cjs

@@ -0,0 +1,43 @@
+// Run the actual page script with a tiny DOM/fetch fixture; no browser dependency.
+const fs = require('node:fs');
+const vm = require('node:vm');
+const assert = require('node:assert/strict');
+const elements = new Map();
+const events = new Map();
+const requests = [];
+const element = id => {
+  if (!elements.has(id)) elements.set(id, {value:'', checked:false, disabled:false, textContent:'',
+    addEventListener(){}, reset(){}});
+  return elements.get(id);
+};
+const context = {
+  document:{getElementById:element}, location:{hash:''}, URLSearchParams,
+  window:{addEventListener:(name, fn) => events.set(name, fn)},
+  fetch:(path, options) => new Promise(resolve => requests.push({path, options, resolve})),
+};
+const html = fs.readFileSync(process.argv[2], 'utf8');
+vm.runInNewContext(html.match(/<script>([\s\S]*?)<\/script>/)[1].replace('__FRAME_KEY__', '"test"'), context);
+const answer = (index, data) => requests[index].resolve({ok:true,json:async () => data});
+(async () => {
+  context.location.hash = '#confirm=first';
+  const first = events.get('hashchange')();
+  context.location.hash = '#confirm=second';
+  const second = events.get('hashchange')();
+  answer(1, {action:{name:'second'},approved:false});
+  await second;
+  answer(0, {action:{name:'first'},approved:false});
+  await first;
+  assert.match(element('action').textContent, /second/);
+  assert.doesNotMatch(element('action').textContent, /first/);
+  const approved = element('approve').onclick();
+  assert.equal(JSON.parse(requests[2].options.body).confirmation, 'second');
+  context.location.hash = '#confirm=third';
+  const third = events.get('hashchange')();
+  answer(3, {action:{name:'third'},approved:false});
+  await third;
+  answer(2, {message:'Approved for one use'});
+  await approved;
+  assert.equal(element('approval-status').textContent, '');
+  assert.match(element('action').textContent, /third/);
+  console.log('Approval navigation races: pass');
+})().catch(error => { console.error(error); process.exitCode=1; });

tests/e2e/test_agents.py

@@ -0,0 +1,46 @@
+"""Real HTTP/MCP adapter against fake-Frame SSH; no model service needed."""
+import json
+from pathlib import Path
+import sys
+
+import harness
+from harness import api, ok, finished, ssh
+
+sys.path.insert(0, str(harness.ROOT / 'ui'))
+import frame_mcp
+
+
+class Agents(harness.FrameTestCase):
+    def client(self):
+        return frame_mcp.Client('http://127.0.0.1:%d' % harness.Server.port)
+
+    def call(self, name, args):
+        return json.loads(frame_mcp.call(self.client(), name, args)['content'][0]['text'])
+
+    def approve(self, proposal):
+        ok('POST', '/api/agent/approval', {'confirmation': proposal['confirmation'], 'accept': True})
+        return proposal['confirmation']
+
+    def test_status_and_approved_install_job(self):
+        self.assertIn('battery', self.call('status', {}))
+        proposal = self.call('install', {'id': 'org.example.AgentTest'})
+        before = api('POST', '/api/agent/call', {'name': 'install', 'arguments': {'id': 'org.example.AgentTest'}, 'confirmation': proposal['confirmation']})
+        self.assertEqual(before[0], 400)
+        token = self.approve(proposal)
+        job = self.call('install', {'id': 'org.example.AgentTest', 'confirmation': token})
+        self.assertFalse(finished(job).get('error'))
+        self.assertIn('org.example.AgentTest', ssh('flatpak list --app --columns=application'))
+        denied = api('POST', '/api/agent/call', {'name': 'install', 'arguments': {'id': 'org.example.AgentTest'}, 'confirmation': token})
+        self.assertEqual(denied[0], 400)
+
+    def test_approved_file_and_text(self):
+        path = Path(self.path('agent-note.txt'))
+        path.write_text('MCP file content\n')
+        args = {'path': str(path)}
+        token = self.approve(self.call('send_file', args))
+        self.call('send_file', {**args, 'confirmation': token})
+        self.assertEqual(ssh('cat ~/Downloads/agent-note.txt'), path.read_text())
+        args = {'text': 'MCP clipboard text'}
+        token = self.approve(self.call('send_text', args))
+        self.call('send_text', {**args, 'confirmation': token})
+        self.assertEqual(harness.state()['clipboard'], ['MCP clipboard text'])

tests/e2e/test_media_transfer.py

@@ -0,0 +1,21 @@
+"""Media transfer through the real HTTP/SSH path; the fake has no VR renderer."""
+import harness
+from harness import ok, ssh
+
+harness.require()
+
+
+class Media(harness.FrameTestCase):
+    def test_upload_and_list_without_launching_a_viewer(self):
+        # The transfer endpoint doesn't decode. Rendering belongs to the real
+        # headset smoke checks documented in docs/vr-video.md.
+        self.assertEqual(ssh('stat -c "%U:%G %a" ~/.local/share').strip(), 'steamos:steamos 755')
+        result = ok('POST', '/api/upload', raw=b'fake-media', headers={
+            'X-Mode': 'media', 'X-Filename': 'test_SBS.png'})
+        identity = result['id']
+        try:
+            files = ok('POST', '/api/media', {'action': 'list'})['files']
+            self.assertIn(identity, [f['id'] for f in files])
+            self.assertEqual(ssh('cat ~/Videos/FrameControl/'+identity), 'fake-media')
+        finally:
+            ssh('rm -rf ~/Videos/FrameControl/'+identity.split('/')[0])

tests/fakeframe/rootfs/usr/local/lib/fakeframe/init.py

@@ -231,6 +231,13 @@ def sysfs(state):
 
 def runtimes(state):
     """Installed compat tools as Steam app manifests, and the Lepton launcher itself."""
+    # Verified 2026-09-28, BUILD_ID 20260925.6191901: both parents are
+    # steamos:steamos 0755. The root supervisor must not leave them root-owned
+    # when creating Steam's fake manifests; user-account app installs need them.
+    for directory in (HOME + '/.local', HOME + '/.local/share'):
+        os.makedirs(directory, mode=0o755, exist_ok=True)
+        chown(directory)
+        os.chmod(directory, 0o755)
     apps = fs.STEAM_ROOT + '/steamapps'
     for alias, installed in state['runtimes'].items():
         acf = f'{apps}/appmanifest_{fs.RUNTIME_APPIDS[alias]}.acf'

tests/sandbox.py

@@ -0,0 +1,16 @@
+"""Imported first by every test module: nothing a test does reaches this person's
+app data, their telemetry, or the shared compatibility database.
+
+Must run before any ui module is imported, since those read these at import time.
+"""
+import atexit
+import os
+import shutil
+import tempfile
+
+_dir = tempfile.mkdtemp(prefix="frame-control-tests-")
+atexit.register(shutil.rmtree, _dir, ignore_errors=True)
+os.environ["FRAME_CONTROL_DATA_DIR"] = _dir
+os.environ["FRAME_CONTROL_TELEMETRY"] = "0"
+# A maintainer's machine holds the database key; send anything that slips through nowhere.
+os.environ["FRAME_COMPAT_DB_URL"] = "http://127.0.0.1:9"

tests/test_agent.py

@@ -0,0 +1,253 @@
+"""MCP protocol, exact-action approvals and explicit assistant data sharing."""
+import io
+import json
+import os
+import shutil
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+from unittest import mock
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
+import frame_agent as agent
+import frame_assistant as assistant
+import frame_mcp as mcp
+import server
+
+
+class Approvals(unittest.TestCase):
+    def test_requires_human_decision_exact_action_and_single_use(self):
+        gate = agent.Approvals()
+        action = {'name': 'power', 'arguments': {'action': 'reboot'}}
+        token = gate.request(action)['confirmation']
+        with self.assertRaises(ValueError):
+            gate.consume(token, action)
+        gate.decide(token, True)
+        with self.assertRaises(ValueError):
+            gate.consume(token, {'name': 'power', 'arguments': {'action': 'poweroff'}})
+        gate.consume(token, action)
+        with self.assertRaises(ValueError):
+            gate.consume(token, action)
+
+    def test_expiry_rejection_and_non_boolean_approval(self):
+        gate = agent.Approvals()
+        token = gate.request({})['confirmation']
+        gate.decide(token, 'true')
+        with self.assertRaises(ValueError):
+            gate.inspect(token)
+        token = gate.request({})['confirmation']
+        with mock.patch.object(agent.time, 'monotonic', return_value=float('inf')):
+            with self.assertRaises(ValueError):
+                gate.decide(token, True)
+
+    def test_concurrent_consumption_executes_once(self):
+        gate = agent.Approvals()
+        token = gate.request({})['confirmation']
+        gate.decide(token, True)
+        results = []
+        def consume():
+            try:
+                gate.consume(token, {})
+                results.append(True)
+            except ValueError:
+                results.append(False)
+        threads = [threading.Thread(target=consume) for _ in range(8)]
+        for thread in threads: thread.start()
+        for thread in threads: thread.join()
+        self.assertEqual(results.count(True), 1)
+
+    def test_action_never_runs_before_approval(self):
+        with mock.patch.object(agent, 'approvals', agent.Approvals()), mock.patch.object(server, 'flatpak') as install:
+            body = {'name': 'install', 'arguments': {'id': 'org.example.App'}}
+            result = agent.call(server, body)
+            install.assert_not_called()
+            body['confirmation'] = result['confirmation']
+            with self.assertRaises(ValueError): agent.call(server, body)
+            agent.approvals.decide(body['confirmation'], True)
+            agent.call(server, body)
+            install.assert_called_once_with({'id': 'org.example.App', 'action': 'install'})
+            with self.assertRaises(ValueError): agent.call(server, body)
+
+    def test_file_content_change_invalidates_approval(self):
+        with tempfile.TemporaryDirectory() as tmp, mock.patch.object(agent, 'approvals', agent.Approvals()), mock.patch.object(server, 'push_file') as push:
+            path = Path(tmp) / 'note.txt'
+            path.write_text('first')
+            body = {'name': 'send_file', 'arguments': {'path': str(path)}}
+            result = agent.call(server, body)
+            agent.approvals.decide(result['confirmation'], True)
+            body['confirmation'] = result['confirmation']
+            path.write_text('second')
+            with self.assertRaises(ValueError): agent.call(server, body)
+            push.assert_not_called()
+
+    def test_no_arbitrary_commands_or_arguments(self):
+        for name, args in [('shell', {'command': 'true'}), ('panel', {'id': 'org.example.App', 'args': '--evil'}),
+                           ('power', {'action': 'factory-reset'}), ('send_text', {'text': ''})]:
+            with self.assertRaises(ValueError): agent.call(server, {'name': name, 'arguments': args})
+
+
+class Assistant(unittest.TestCase):
+    def setUp(self):
+        self.received = []
+        owner = self
+        class Endpoint(BaseHTTPRequestHandler):
+            def log_message(self, *args): pass
+            def do_POST(self):
+                owner.received.append((dict(self.headers), json.loads(self.rfile.read(int(self.headers['Content-Length'])))))
+                if self.path == '/redirect':
+                    self.send_response(302)
+                    self.send_header('Location', '/other')
+                    self.end_headers()
+                    return
+                data = json.dumps({'choices': [{'message': {'content': '<script>not executed</script>'}}]}).encode()
+                self.send_response(200)
+                self.send_header('Content-Length', str(len(data)))
+                self.end_headers()
+                self.wfile.write(data)
+        self.httpd = ThreadingHTTPServer(('127.0.0.1', 0), Endpoint)
+        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
+        self.thread.start()
+        self.body = {'endpoint': 'http://127.0.0.1:%d/chat' % self.httpd.server_port, 'model': 'local', 'prompt': 'Hello', 'consent': True}
+
+    def tearDown(self):
+        self.httpd.shutdown()
+        self.httpd.server_close()
+        self.thread.join()
+
+    def test_no_opt_in_no_request_or_capture(self):
+        capture = mock.Mock()
+        for consent in (False, None, 'true', 1):
+            with self.assertRaises(ValueError): assistant.chat({**self.body, 'consent': consent, 'screenshot': True}, capture)
+        capture.assert_not_called()
+        self.assertEqual(self.received, [])
+
+    def test_text_only_keyless_and_optional_screenshot(self):
+        capture = mock.Mock(return_value=b'png')
+        self.assertIn('script', assistant.chat(self.body, capture)['reply'])
+        capture.assert_not_called()
+        headers, body = self.received[-1]
+        self.assertNotIn('Authorization', headers)
+        self.assertEqual(body['messages'], [{'role': 'user', 'content': 'Hello'}])
+        assistant.chat({**self.body, 'screenshot': True, 'key': 'test-key'}, capture)
+        capture.assert_called_once()
+        headers, body = self.received[-1]
+        self.assertEqual(headers['Authorization'], 'Bearer test-key')
+        self.assertEqual(body['messages'][0]['content'][1]['image_url']['url'], 'data:image/png;base64,cG5n')
+
+    def test_redirects_do_not_forward_context_or_credentials(self):
+        with self.assertRaises(ValueError):
+            assistant.chat({**self.body, 'endpoint': self.body['endpoint'].replace('/chat', '/redirect'), 'key': 'secret'}, mock.Mock())
+        self.assertEqual(len(self.received), 1)
+
+    def test_bad_urls_fail_before_capture(self):
+        for url in ('file:///etc/passwd', 'http://example.com/chat', 'https://user:pass@example.com', 'https://example.com?key=secret'):
+            capture = mock.Mock()
+            with self.assertRaises(ValueError): assistant.chat({**self.body, 'endpoint': url, 'screenshot': True}, capture)
+            capture.assert_not_called()
+
+
+class AssistantPage(unittest.TestCase):
+    @unittest.skipUnless(shutil.which('node'), 'Node is required for the page script regression')
+    def test_approval_navigation_races(self):
+        root = Path(__file__).resolve().parents[1]
+        result = subprocess.run(['node', str(root / 'tests/assistant_ui.cjs'), str(root / 'ui/assistant.html')],
+                                capture_output=True, text=True, timeout=10)
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+
+class Protocol(unittest.TestCase):
+    def test_stdio_initialize_list_call_errors_and_eof(self):
+        messages = [
+            {'jsonrpc': '2.0', 'id': 1, 'method': 'initialize', 'params': {'protocolVersion': '2025-06-18'}},
+            {'jsonrpc': '2.0', 'method': 'notifications/initialized'},
+            {'jsonrpc': '2.0', 'id': 2, 'method': 'tools/list'},
+            {'jsonrpc': '2.0', 'id': 3, 'method': 'tools/call', 'params': {'name': 'shell'}},
+            {'jsonrpc': '2.0', 'id': 4, 'method': 'ping'},
+        ]
+        result = subprocess.run([sys.executable, str(Path(mcp.__file__))], input='\n'.join(map(json.dumps, messages)) + '\n', text=True, capture_output=True, timeout=10)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        replies = list(map(json.loads, result.stdout.splitlines()))
+        self.assertEqual([r['id'] for r in replies], [1, 2, 3, 4])
+        self.assertEqual(replies[0]['result']['protocolVersion'], '2025-06-18')
+        self.assertIn('screenshot', [t['name'] for t in replies[1]['result']['tools']])
+        self.assertTrue(replies[2]['result']['isError'])
+
+    def test_mcp_cannot_approve_and_returns_review_url(self):
+        client = mock.Mock(url='http://127.0.0.1:47810')
+        client.request.return_value = {'approvalPath': '/assistant#confirm=token'}
+        result = mcp.call(client, 'power', {'action': 'reboot'})
+        self.assertIn('http://127.0.0.1:47810/assistant', result['content'][0]['text'])
+        with self.assertRaises(ValueError): mcp.call(client, 'approve', {'confirmation': 'token'})
+        with self.assertRaises(ValueError): mcp.call(client, 'status', {'path': '/api/open'})
+
+    def test_loopback_only_backend(self):
+        for url in ('https://example.com', 'http://127.0.0.1/api', 'http://secret@localhost:1234', 'file:///tmp/x'):
+            with self.assertRaises(ValueError): mcp.Client(url)
+
+
+class ManagedBackend(unittest.TestCase):
+    def test_private_backend_auth_and_cleanup(self):
+        from urllib.error import HTTPError, URLError
+        from urllib.request import urlopen
+        with mock.patch.dict(os.environ, {'FRAME_ALIAS': 'frame-control-test.invalid'}):
+            with mcp.backend() as client:
+                url = client.url
+                self.assertIn('os', client.request('/api/host'))
+                with self.assertRaises(HTTPError) as error:
+                    urlopen(url + '/api/host', timeout=2)
+                self.assertEqual(error.exception.code, 403)
+                error.exception.close()
+                # A second client has its own backend and key.
+                with mcp.backend() as other:
+                    self.assertNotEqual(client.url, other.url)
+                    self.assertNotEqual(client.key, other.key)
+                self.assertIn('os', client.request('/api/host'))
+            with self.assertRaises(URLError):
+                urlopen(url + '/', timeout=2)
+
+    def test_private_ssh_socket_is_not_the_desktop_socket(self):
+        with mock.patch.object(server.frame_host, 'MUX', True), \
+             mock.patch.object(server.frame_host.os, 'getuid', return_value=501, create=True), \
+             mock.patch.object(server.frame_host.os, 'getpid', return_value=123):
+            self.assertEqual(server.frame_host.control_path(), '/tmp/frame-ui-501-%C')
+            self.assertEqual(server.frame_host.control_path(private=True), '/tmp/frame-ui-501-123-%C')
+
+
+class ComputerState(unittest.TestCase):
+    def test_gamescope_triplets_and_empty_focus(self):
+        import frame_computer
+        parsed = frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 16, 42, 123, 32, 55, 999\nGAMESCOPE_FOCUSED_APP(CARDINAL) = \n')
+        self.assertEqual(parsed['windows'], [{'windowId': '0x10', 'appid': 42, 'pid': 123}, {'windowId': '0x20', 'appid': 55, 'pid': 999}])
+        self.assertIsNone(parsed['focusedApp'])
+        with self.assertRaises(ValueError):
+            frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = 1, 2')
+        with self.assertRaises(ValueError):
+            frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS(CARDINAL) = untrusted')
+        with self.assertRaises(ValueError):
+            frame_computer.parse_windows('GAMESCOPE_FOCUSABLE_WINDOWS: no such atom on any window.')
+
+    def test_partial_snapshot_reports_failure_not_empty_success(self):
+        import frame_computer
+        with mock.patch.object(frame_computer.subprocess, 'run', side_effect=OSError('no display')), \
+             mock.patch.object(frame_computer, 'accessibility', side_effect=OSError('no AT-SPI')):
+            result = frame_computer.snapshot()
+        self.assertIn('windowError', result)
+        self.assertIn('accessibilityError', result)
+        self.assertFalse(result['inputEnabled'])
+        self.assertNotIn('windows', result)
+
+    def test_mcp_computer_state_is_read_only(self):
+        client = mock.Mock()
+        client.request.return_value = {'windows': []}
+        mcp.call(client, 'computer_state', {})
+        client.request.assert_called_once_with('/api/computer/state')
+        spec = next(t for t in mcp.TOOLS if t['name'] == 'computer_state')
+        self.assertTrue(spec['annotations']['readOnlyHint'])
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_compat_db.py

@@ -2,6 +2,7 @@
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import os
 import sys
 import tempfile

tests/test_connect.py

@@ -3,6 +3,7 @@ steamos-devkit-service, the ~/.ssh/config block, and the mDNS output parsers.
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import json
 import socket
 import sys

tests/test_frame_apk.py

@@ -1,4 +1,5 @@
 """frame_apk against a small APK built here: binary manifest plus resource table."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import io
 import os
 import struct

tests/test_frame_apk_versions.py

@@ -1,4 +1,5 @@
 """Offline version lookup with small index-v2 fixtures."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import io
 import json
 import os
@@ -242,6 +243,7 @@ class UploadVersionsTest(unittest.TestCase):
             handler.rfile = io.BytesIO(b'x')
             with patch.object(frame_android, 'apk_info', return_value=dict(info)), \
                     patch.object(versions, 'alternatives', side_effect=AssertionError('lookup during upload')) as lookup, \
+                    patch.object(frame_android, 'install_hooks', []), \
                     patch.object(server, 'ensure_master') as ssh:
                 if mode == 'apkinfo':
                     reply = handler.upload()
@@ -256,5 +258,28 @@ class UploadVersionsTest(unittest.TestCase):
                 ssh.assert_not_called()
 
 
+    def test_blocked_uploads_are_reported_like_failed_installs(self):
+        import server
+        info = {'package': 'org.example.app', 'label': 'Example', 'version': '5',
+                'version_code': 5, 'min_sdk': 33, 'abis': [], 'icon_png': None}
+        for apk_info, expected_info in ((dict(info), 'org.example.app'),
+                                        (frame_android.FrameError('not an APK'), None)):
+            handler = object.__new__(server.Handler)
+            handler.headers = {'X-Filename': 'app.apk', 'X-Mode': 'apk', 'Content-Length': '1'}
+            handler.rfile = io.BytesIO(b'x')
+            calls = []
+            patch_info = (patch.object(frame_android, 'apk_info', side_effect=apk_info)
+                          if isinstance(apk_info, Exception) else
+                          patch.object(frame_android, 'apk_info', return_value=apk_info))
+            with patch_info, patch.object(frame_android, 'install_hooks', [lambda *a: calls.append(a)]), \
+                    patch.object(server, 'ensure_master'):
+                with self.assertRaises(server.Failure):
+                    handler.upload()
+            self.assertEqual(len(calls), 1)
+            got_info, meta, error, _ = calls[0]
+            self.assertEqual((got_info or {}).get('package'), expected_info)
+            self.assertIsNone(meta)
+            self.assertIsInstance(error, frame_android.FrameError)
+
 if __name__ == '__main__':
     unittest.main()

tests/test_frame_apk_vr.py

@@ -0,0 +1,297 @@
+"""Local-only VR manifest, raw ZIP and independent signature checks."""
+import io
+import os
+from pathlib import Path
+import struct
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+import zipfile
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
+import frame_apk
+import frame_apk_vr as vr
+import frame_apk_sign as signing
+import frame_android
+from test_frame_apk import pool
+
+
+DEFAULT = 'android.intent.category.DEFAULT'
+
+
+def manifest(utf8=False, launcher=False, category=None, samsung=False, split=False, alias=False, splash=False):
+    strings = ['manifest', 'package', 'org.test.vr', 'application', 'activity', 'intent-filter',
+               'action', 'category', 'name', vr.MAIN, category or next(iter(sorted(vr.VR))),
+               vr.LAUNCHER, 'http://schemas.android.com/apk/res/android', 'meta-data', 'value',
+               'com.samsung.android.vr.application.mode', 'vr_only', 'activity-alias', DEFAULT, next(iter(sorted(vr.VR))), 'targetActivity', '.Splash', '.Game']
+    def start(tag, attrs=()):
+        body = struct.pack('<IIHHHHHH', 0xffffffff, strings.index(tag), 20, 20, len(attrs), 0, 0, 0)
+        for name, value in attrs:
+            ns = 0xffffffff if name == 'package' else 12
+            body += struct.pack('<IIIHBBI', ns, strings.index(name), strings.index(value), 8, 0, 3, strings.index(value))
+        return struct.pack('<HHIII', 0x102, 16, 16 + len(body), 1, 0xffffffff) + body
+    def end(tag):
+        return struct.pack('<HHIIIII', 0x103, 16, 24, 1, 0xffffffff, 0xffffffff, strings.index(tag))
+    def leaf(tag, attrs):
+        return start(tag, attrs) + end(tag)
+    b = pool(strings, utf8) + start('manifest', [('package', 'org.test.vr')]) + start('application')
+    if samsung:
+        b += leaf('meta-data', [('name', strings[15]), ('value', 'vr_only')])
+    if splash:  # a helper activity with its own MAIN filter, ahead of the game's
+        b += start('activity', [('name', '.Splash')]) + start('intent-filter') + leaf('action', [('name', vr.MAIN)])
+        b += leaf('category', [('name', DEFAULT)]) + end('intent-filter') + end('activity')
+    b += start('activity', [('name', '.Game')] if splash else []) + start('intent-filter') + leaf('action', [('name', vr.MAIN)])
+    b += leaf('category', [('name', strings[10])])
+    if split:
+        b += end('intent-filter') + start('intent-filter')
+    if launcher:
+        b += leaf('category', [('name', vr.LAUNCHER)])
+    b += end('intent-filter') + end('activity')
+    if alias:  # Godot 4: LAUNCHER only on an alias of the activity ('vr' or 'flat')
+        b += start('activity-alias', [('targetActivity', '.Game')] if splash else []) + start('intent-filter') + leaf('action', [('name', vr.MAIN)])
+        if alias == 'vr':
+            b += leaf('category', [('name', next(iter(sorted(vr.VR))))])
+        b += leaf('category', [('name', vr.LAUNCHER)])
+        b += end('intent-filter') + end('activity-alias')
+    b += end('application') + end('manifest')
+    return struct.pack('<HHI', 3, 8, len(b) + 8) + b
+
+
+class VRTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tmp = tempfile.TemporaryDirectory()
+        cls.keypath = Path(cls.tmp.name) / 'key.json'
+        cls.key = signing.signing_key(cls.keypath)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.tmp.cleanup()
+
+    def test_manifest_patch(self):
+        for utf8 in (False, True):
+            for category in vr.VR:
+                original = manifest(utf8, category=category)
+                result = vr.add_launcher_category(original)
+                info, filters = vr.inspect(result)
+                self.assertTrue(info['launchable'])
+                self.assertTrue(info['vr'])
+                self.assertIn(vr.LAUNCHER, filters[0]['categories'])
+                self.assertEqual(struct.unpack_from('<I', result, 4)[0], len(result))
+                self.assertEqual(vr.add_launcher_category(result), result)
+                self.assertEqual(vr.add_launcher_category(manifest(utf8, True)), manifest(utf8, True))
+
+    def test_filter_boundaries(self):
+        self.assertFalse(vr.inspect(manifest(launcher=True, split=True))[0]['launchable'])
+        self.assertTrue(vr.inspect(vr.add_launcher_category(manifest(launcher=True, split=True)))[0]['launchable'])
+
+    def test_alias_launcher_is_not_enough(self):
+        # (manifest, still VR after patching)
+        cases = [(manifest(alias='vr'), True),                    # Godot 4 VR export
+                 (manifest(alias='vr', category=DEFAULT), True),  # VR category only on the alias
+                 (manifest(alias='flat', category=DEFAULT), False)]  # Godot 4 flat export
+        for original, is_vr in cases:
+            info, filters = vr.inspect(original)
+            self.assertFalse(info['launchable'])
+            self.assertTrue(info['repairable'])
+            self.assertEqual(len(filters), 1)
+            self.assertFalse(filters[0]['alias'])
+            result = vr.add_launcher_category(original)
+            info, _ = vr.inspect(result)
+            self.assertTrue(info['launchable'])
+            self.assertFalse(info['repairable'])
+            self.assertEqual(info['vr'], is_vr)
+
+    def test_alias_target_activity_is_patched(self):
+        original = manifest(alias='flat', category=DEFAULT, splash=True)
+        info, filters = vr.inspect(original)
+        self.assertTrue(info['repairable'])
+        self.assertEqual(filters[0]['activity'], 'org.test.vr.Game')
+        owner, launchers = None, []
+        for tag, attrs in frame_apk.manifest_elements(vr.add_launcher_category(original)):
+            if tag in ('activity', 'activity-alias'):
+                owner = (tag, attrs.get('name', (0, 0, None))[2])
+            if tag == 'category' and attrs['name'][2] == vr.LAUNCHER:
+                launchers.append(owner)
+        self.assertEqual(launchers, [('activity', '.Game'), ('activity-alias', None)])
+
+    def test_alias_with_launchable_activity_is_left_alone(self):
+        original = manifest(launcher=True, alias='vr')
+        info, _ = vr.inspect(original)
+        self.assertTrue(info['launchable'])
+        self.assertFalse(info['repairable'])
+        self.assertEqual(vr.add_launcher_category(original), original)
+
+    def test_patch_alias_apk(self):
+        with tempfile.TemporaryDirectory() as d:
+            src, dst = Path(d) / 'in.apk', Path(d) / 'out.apk'
+            with zipfile.ZipFile(src, 'w') as z:
+                z.writestr('AndroidManifest.xml', manifest(alias='flat', category=DEFAULT))
+            with patch.object(signing, 'signing_key', return_value=self.key):
+                result = frame_android.patch(src, dst)
+            self.assertEqual(result['patched'], ['launcher'])
+            self.assertTrue(frame_apk.apk_info(dst)['launchable'])
+            self.assertTrue(signing.verify(dst))
+
+    def test_styled_pool(self):
+        for utf8 in (False, True):
+            p = bytearray(pool(['styled'], utf8))
+            old_start = struct.unpack_from('<I', p, 20)[0]
+            p[old_start:old_start] = struct.pack('<I', 0)
+            style_start = len(p)
+            p += struct.pack('<III', 0, 0, 2) + b'\xff' * 12
+            struct.pack_into('<I', p, 4, len(p))
+            struct.pack_into('<I', p, 16, (0x100 if utf8 else 0) | 1)
+            struct.pack_into('<I', p, 12, 1)
+            struct.pack_into('<II', p, 20, old_start + 4, style_start)
+            result, idx = vr._append_string(bytes(p), vr.LAUNCHER)
+            self.assertEqual(frame_apk._string_pool(result, 0), ['styled', vr.LAUNCHER])
+            new_style = struct.unpack_from('<I', result, 24)[0]
+            self.assertEqual(result[new_style:], p[style_start:])
+            self.assertEqual(len(result) % 4, 0)
+
+    def test_detection(self):
+        names = {'lib/arm64-v8a/' + n for n in ('libvrapi.so', 'libopenxr_loader.so', 'libovrplatformloader.so')}
+        info = vr.detection(manifest(category='android.intent.category.LAUNCHER'), names)
+        self.assertTrue(info['vr'])
+        self.assertTrue(info['launchable'])
+        self.assertEqual(len(info['vr_issues']), 3)
+        self.assertFalse(vr.detection(manifest(category=vr.LAUNCHER), set())['vr'])
+        self.assertTrue(vr.detection(manifest(category=vr.LAUNCHER, samsung=True), set())['vr'])
+
+    def test_key_cache(self):
+        with patch.object(signing, '_prime', side_effect=AssertionError('regenerated')):
+            self.assertEqual(signing.signing_key(self.keypath), self.key)
+        if os.name != 'nt':
+            self.assertEqual(self.keypath.stat().st_mode & 0o777, 0o600)
+
+    def test_repack_sign_tamper(self):
+        with tempfile.TemporaryDirectory() as d:
+            src, dst = Path(d) / 'in.apk', Path(d) / 'out.apk'
+            with zipfile.ZipFile(src, 'w') as z:
+                z.writestr('AndroidManifest.xml', manifest(), compress_type=8)
+                z.writestr('classes.dex', b'compress me' * 10000, compress_type=8)
+                z.writestr('resources.arsc', b'1234')
+                z.writestr('lib/arm64-v8a/libx.so', b'ELF' * 500000)
+                z.writestr('META-INF/OLD.RSA', b'old')
+                z.writestr('META-INF/MANIFEST.MF', b'old')
+            with patch.object(signing, 'signing_key', return_value=self.key):
+                result = frame_android.patch(src, dst, {'assets/layer.json': b'{}', 'lib/arm64-v8a/liblayer.so': b'layer'})
+            self.assertEqual(result['patched'], ['launcher'])
+            self.assertTrue(frame_apk.apk_info(dst)['launchable'])
+            self.assertTrue(signing.verify(dst))
+            def compressed(path, info):
+                data = path.read_bytes()
+                nl, el = struct.unpack_from('<HH', data, info.header_offset + 26)
+                start = info.header_offset + 30 + nl + el
+                return data[start:start + info.compress_size], start
+            with zipfile.ZipFile(src) as a, zipfile.ZipFile(dst) as b:
+                self.assertNotIn('META-INF/OLD.RSA', b.namelist())
+                self.assertNotIn('META-INF/MANIFEST.MF', b.namelist())
+                for i in b.infolist():
+                    raw, start = compressed(dst, i)
+                    if i.compress_type == 0:
+                        self.assertEqual(start % (16384 if i.filename.endswith('.so') else 4), 0)
+                    if i.filename in ('classes.dex', 'resources.arsc', 'lib/arm64-v8a/libx.so'):
+                        self.assertEqual(raw, compressed(src, a.getinfo(i.filename))[0])
+                _, off = compressed(dst, b.getinfo('resources.arsc'))
+            original = dst.read_bytes()
+            data = bytearray(original)
+            eo, cd = signing._eocd(data)
+            size = struct.unpack_from('<Q', data, cd - 24)[0]
+            block_start = cd - size - 8
+            value = data[block_start + 20:cd - 24]
+            signer = signing._parts(signing._parts(value)[0])[0]
+            signed, signatures, pub = signing._parts(signer)
+            signature = signing._parts(signing._parts(signatures)[0][4:])[0]
+            sig_offset = data.index(signature, block_start)
+            data[sig_offset] ^= 1
+            dst.write_bytes(data)
+            with self.assertRaisesRegex(ValueError, 'RSA signature'):
+                signing.verify(dst)
+            data = bytearray(original)
+            data[off] ^= 1
+            dst.write_bytes(data)
+            with self.assertRaisesRegex(ValueError, 'digest'):
+                signing.verify(dst)
+
+    @unittest.skipUnless(Path('/usr/bin/openssl').exists(), 'openssl absent')
+    def test_openssl(self):
+        with tempfile.TemporaryDirectory() as d:
+            d = Path(d)
+            (d / 'cert.der').write_bytes(signing.certificate(self.key))
+            (d / 'message').write_bytes(b'independent signature check')
+            (d / 'signature').write_bytes(signing.rsa_sign(b'independent signature check', self.key))
+            def run(*args):
+                return subprocess.run(['/usr/bin/openssl', *args], check=True, capture_output=True).stdout
+            run('x509', '-inform', 'DER', '-in', str(d / 'cert.der'), '-out', str(d / 'cert.pem'))
+            run('verify', '-check_ss_sig', '-CAfile', str(d / 'cert.pem'), str(d / 'cert.pem'))
+            (d / 'pub.pem').write_bytes(run('x509', '-in', str(d / 'cert.pem'), '-pubkey', '-noout'))
+            output = run('dgst', '-sha256', '-verify', str(d / 'pub.pem'), '-signature', str(d / 'signature'), str(d / 'message'))
+            self.assertIn(b'Verified OK', output)
+
+    def test_install_auto_and_override(self):
+        base = {'package': 'org.test.vr', 'label': 'VR', 'abis': [], 'min_sdk': None,
+                'vr': True, 'vr_activity': True, 'launchable': False, 'repairable': True}
+        with patch.object(frame_android, 'apk_info', return_value=base), \
+                patch.object(frame_android, 'xr_compat_files', return_value={}), \
+                patch.object(frame_android, 'patch', return_value={'patched': ['launcher']}) as repair, \
+                patch.object(frame_android, '_install', return_value={}) as install:
+            frame_android.install('original.apk')
+            repair.assert_called_once()
+            self.assertFalse(install.call_args.args[3])
+            self.assertEqual(install.call_args.args[1]['patched'], ['launcher'])
+            frame_android.install('original.apk', flatscreen=True)
+            self.assertTrue(install.call_args.args[3])
+
+    def test_xr_compat_layer(self):
+        with tempfile.TemporaryDirectory() as d:
+            apk = Path(d) / 'a.apk'
+            with zipfile.ZipFile(apk, 'w') as z:
+                z.writestr('lib/arm64-v8a/libopenxr_loader.so', b'')
+            add = frame_android.xr_compat_files(str(apk))
+            self.assertEqual(set(add), set(frame_android.XR_COMPAT_FILES))
+            self.assertIn(b'XR_APILAYER_FRAME_compat', add['assets/openxr/1/api_layers/implicit.d/XrApiLayer_FRAME_compat.json'])
+            self.assertTrue(add['lib/arm64-v8a/libXrApiLayer_FRAME_compat.so'].startswith(b'\x7fELF'))
+            with zipfile.ZipFile(apk, 'a') as z:  # already injected: nothing more to add
+                z.writestr('lib/arm64-v8a/libXrApiLayer_FRAME_compat.so', b'')
+            self.assertEqual(frame_android.xr_compat_files(str(apk)), {})
+            flat = Path(d) / 'flat.apk'
+            with zipfile.ZipFile(flat, 'w') as z:
+                z.writestr('classes.dex', b'')
+            self.assertEqual(frame_android.xr_compat_files(str(flat)), {})
+
+    def test_xr_compat_layer_missing(self):
+        with tempfile.TemporaryDirectory() as d:
+            apk = Path(d) / 'a.apk'
+            with zipfile.ZipFile(apk, 'w') as z:
+                z.writestr('lib/arm64-v8a/libopenxr_loader.so', b'')
+            with patch.object(frame_android, 'XR_COMPAT', d):
+                with self.assertRaisesRegex(frame_android.FrameError, 'build.sh'):
+                    frame_android.xr_compat_files(str(apk))
+
+    def test_patch_rejects_corrupt_manifest_cleanly(self):
+        with patch.object(frame_android, 'apk_info', side_effect=struct.error('bad')):
+            with self.assertRaises(frame_android.FrameError):
+                frame_android.patch('x.apk', 'y.apk')
+
+    def test_install_adds_layer_to_vr_apps(self):
+        base = {'package': 'org.test.vr', 'label': 'VR', 'abis': [], 'min_sdk': None,
+                'vr': True, 'vr_activity': True, 'launchable': True, 'repairable': False}
+        layer = {'x': b''}
+        with patch.object(frame_android, 'apk_info', return_value=dict(base)), \
+                patch.object(frame_android, 'xr_compat_files', return_value=layer), \
+                patch.object(frame_android, 'patch', return_value={'patched': ['openxr-compat']}) as repair, \
+                patch.object(frame_android, '_install', return_value={}) as install:
+            frame_android.install('game.apk')
+            self.assertIs(repair.call_args.args[2], layer)
+            self.assertEqual(install.call_args.args[1]['patched'], ['openxr-compat'])
+            repair.reset_mock()
+            frame_android.install('game.apk', xr_compat=False)  # launchable, no layer: install as is
+            repair.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_frame_titles.py

@@ -1,4 +1,5 @@
 """frame_titles without a headset: executable headers, launch targets, zips, runtimes."""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import json
 import os
 import shutil

tests/test_macview.py

@@ -0,0 +1,421 @@
+"""Mac in the headset (ui/frame_macview.py and the frame-mac-view agent).
+
+The Python checks run anywhere. On macOS the agent is built and driven over
+HTTP and WebSocket with its test pattern, which needs no Screen Recording
+permission: status, the token, the viewer page, H.264 keyframes, pointer
+input reaching the source, and closing.
+
+Run: python3 -m unittest discover -s tests
+"""
+import base64
+import http.client
+import json
+import os
+import shutil
+import socket
+import struct
+import subprocess
+import sys
+import tempfile
+import time
+import threading
+import unittest
+from unittest import mock
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "ui"))
+
+import frame_macview  # noqa: E402
+
+
+class Helpers(unittest.TestCase):
+    def test_panel_id_is_stable_and_in_range(self):
+        a = frame_macview.panel_id("window:123")
+        self.assertEqual(a, frame_macview.panel_id("window:123"))
+        self.assertNotEqual(a, frame_macview.panel_id("window:124"))
+        self.assertTrue(2_001_000_000 <= a < 2_002_000_000)
+
+    def test_fit_keeps_aspect_inside_the_panel(self):
+        self.assertEqual(frame_macview.fit(2560, 1440), (1920, 1080))
+        w, h = frame_macview.fit(800, 1600)
+        self.assertEqual(h, 1080)
+        self.assertAlmostEqual(w / h, 0.5, places=2)
+
+    @unittest.skipUnless(shutil.which("bash"), "needs bash")
+    @unittest.skipIf(os.name == "nt", "Windows' bash.exe is WSL's launcher, and runners have no distribution")
+    def test_launch_script_parses(self):
+        r = subprocess.run(["bash", "-n"], input=frame_macview.LAUNCH, text=True, capture_output=True)
+        self.assertEqual(r.returncode, 0, r.stderr)
+
+    def test_show_checks_the_source_before_anything_else(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
+        with self.assertRaises(frame_macview.MacViewError):
+            mv.show("rm -rf /")
+
+    def test_missing_browser_is_explained(self):
+        calls = []
+
+        def run(remote, stdin=None, timeout=30):
+            calls.append(remote)
+            e = RuntimeError("exit 3")
+            e.stdout = "NO_BROWSER\n"
+            raise e
+
+        mv = frame_macview.MacView(["ssh"], run, "frame")
+        mv.call = lambda path, **kw: {"screen": True, "ticket": "tk"}
+        mv.ensure_tunnel = lambda **kw: None
+        mv.remote_port = 47900
+        with self.assertRaises(frame_macview.MacViewError) as cm:
+            mv.show("window:5")
+        self.assertIn("Chromium", str(cm.exception))
+        self.assertTrue(calls[0].startswith("bash -s -- "))
+
+    def test_separate_windows_need_accessibility(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
+        mv.call = lambda path, **kw: {"screen": True, "accessibility": False}
+        with self.assertRaises(frame_macview.MacViewError) as cm:
+            mv.show("separate:42")
+        self.assertIn("Accessibility", str(cm.exception))
+
+    def test_screen_permission_is_checked_before_the_headset(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: self.fail("no ssh"), "frame")
+        mv.call = lambda path, **kw: {"screen": False}
+        with self.assertRaises(frame_macview.MacViewError) as cm:
+            mv.show("display:1")
+        self.assertIn("Screen Recording", str(cm.exception))
+
+    def test_stop_all_ends_the_viewer_browser_unless_shown_again(self):
+        calls = []
+        mv = frame_macview.MacView(["ssh"], lambda remote, **kw: calls.append(remote) or "", "frame")
+        mv.agent = mock.Mock(poll=lambda: None)
+        mv.call = lambda path, **kw: {"closed": 1}
+        mv.shown = {"window:5"}
+        with mock.patch.object(frame_macview.time, "sleep"):
+            gen = mv.shows
+            mv.shown.clear()
+            mv._end_viewer_browser(gen)
+            self.assertEqual(len(calls), 1)
+            self.assertIn("pkill -f '[f]rame-control/mac-view", calls[0])
+            mv.shows += 1  # Show pressed during the wait: leave the new viewer alone
+            mv._end_viewer_browser(gen)
+            self.assertEqual(len(calls), 1)
+            mv.launching = 1  # a Show replacing its own stream is still launching
+            mv._end_viewer_browser(mv.shows)
+            self.assertEqual(len(calls), 1)
+        # A Show waits while the cleanup checks and runs pkill.
+        mv.launching = 0
+        in_pkill, release, entered = threading.Event(), threading.Event(), threading.Event()
+
+        def blocking_pkill(remote, **kw):
+            in_pkill.set()
+            release.wait(5)
+        mv.run = blocking_pkill
+        mv._show = lambda *a: entered.set() or "shown"
+        with mock.patch.object(frame_macview.time, "sleep"):
+            cleanup = threading.Thread(target=mv._end_viewer_browser, args=(mv.shows,))
+            cleanup.start()
+            self.assertTrue(in_pkill.wait(2))
+            shower = threading.Thread(target=mv.show, args=("window:5",))
+            shower.start()
+            self.assertFalse(entered.wait(0.3), "Show started while the cleanup held the lock")
+            release.set()
+            self.assertTrue(entered.wait(2))
+            cleanup.join()
+            shower.join()
+
+    def test_tunnel_prefers_the_usb_c_network_when_plugged_in(self):
+        mv = frame_macview.MacView(["ssh"], lambda remote, **kw: "13: usb0 inet 10.86.200.233/29 scope global", "frame")
+        ssh_g = mock.Mock(stdout="user steamos\nhostname frame.example.ts.net\n")
+        with mock.patch.object(frame_macview.socket, "create_connection") as conn, \
+                mock.patch.object(frame_macview.subprocess, "run", return_value=ssh_g):
+            self.assertEqual(mv._usb_route(), ["-o", "HostName=10.86.200.233", "-o", "HostKeyAlias=frame.example.ts.net"])
+            conn.assert_called_once_with(("10.86.200.233", 22), timeout=1)
+            ssh_g.stdout = "hostname frame.example.ts.net\nhostkeyalias paired-frame\n"  # a configured alias wins
+            conn.side_effect = None
+            self.assertIn("HostKeyAlias=paired-frame", mv._usb_route())
+            conn.side_effect = OSError("unplugged")
+            self.assertEqual(mv._usb_route(), [])
+        mv.run = lambda remote, **kw: ""  # no usb0
+        self.assertEqual(mv._usb_route(), [])
+        mv.prefer_usb = False
+        mv.run = lambda remote, **kw: self.fail("no ssh when USB is off")
+        self.assertEqual(mv._usb_route(), [])
+
+    def test_a_failed_usb_tunnel_falls_back_to_the_network(self):
+        mv = frame_macview.MacView(["ssh"], lambda *a, **k: "", "frame")
+        mv._usb_route = lambda: ["-o", "HostName=10.86.200.233"]
+        tried = []
+
+        def attempt(via, ports):
+            tried.append(list(via))
+            mv._last_tunnel_error = "Connection refused"
+            return not via  # USB fails, the normal path works
+        mv._open_tunnel = attempt
+        mv.tunnel_up = lambda: False
+        mv.ensure_tunnel()
+        self.assertEqual(tried, [["-o", "HostName=10.86.200.233"], []])
+        self.assertEqual(mv.route, "network")
+
+
+class WS:
+    """A minimal WebSocket client (masked frames out, plain frames in)."""
+
+    def __init__(self, port, path):
+        self.s = socket.create_connection(("127.0.0.1", port), timeout=10)
+        key = base64.b64encode(os.urandom(16)).decode()
+        self.s.sendall(f"GET {path} HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
+                       f"Sec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n\r\n".encode())
+        head = b""
+        while b"\r\n\r\n" not in head:
+            head += self.s.recv(1)
+        self.status = int(head.split()[1])
+        self.buf = b""
+
+    def _read(self, n):
+        while len(self.buf) < n:
+            chunk = self.s.recv(65536)
+            if not chunk:
+                raise EOFError
+            self.buf += chunk
+        out, self.buf = self.buf[:n], self.buf[n:]
+        return out
+
+    def recv(self):
+        b0, b1 = self._read(2)
+        n = b1 & 0x7F
+        if n == 126:
+            n = struct.unpack(">H", self._read(2))[0]
+        elif n == 127:
+            n = struct.unpack(">Q", self._read(8))[0]
+        return b0 & 0x0F, self._read(n)
+
+    def send_text(self, text):
+        data, mask = text.encode(), os.urandom(4)
+        head = bytes([0x81, 0x80 | len(data)]) if len(data) < 126 else bytes([0x81, 0xFE]) + struct.pack(">H", len(data))
+        self.s.sendall(head + mask + bytes(c ^ mask[i % 4] for i, c in enumerate(data)))
+
+    def close(self):
+        self.s.close()
+
+
+@unittest.skipUnless(sys.platform == "darwin" and shutil.which("xcrun"), "the agent is macOS-only")
+class Agent(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tmp = tempfile.mkdtemp()
+        cls.bin = Path(cls.tmp) / "frame-mac-view"
+        r = subprocess.run(["/bin/sh", str(ROOT / "mac" / "frame-mac-view" / "build.sh"), str(cls.bin)],
+                           capture_output=True, text=True, timeout=600)
+        if r.returncode:  # a real failure on a Mac with Xcode: don't hide it as a skip
+            raise AssertionError("agent didn't build:\n" + (r.stderr or r.stdout)[-2000:])
+        cls.token = "t0ken-" + os.urandom(6).hex()
+        cls.proc = subprocess.Popen([str(cls.bin), "serve", "--port", "0", "--page", str(ROOT / "ui" / "mac-view.html"),
+                                     "--exit-on-eof"], env={**os.environ, "FRAME_MAC_VIEW_TOKEN": cls.token},
+                                    stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
+        line = cls.proc.stdout.readline()
+        cls.port = int(line.rsplit(":", 1)[1])
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.proc.stdin.close()  # --exit-on-eof
+        cls.proc.stdout.close()
+        try:
+            cls.proc.wait(5)
+        except subprocess.TimeoutExpired:
+            cls.proc.kill()
+        shutil.rmtree(cls.tmp, ignore_errors=True)
+
+    def get(self, path, method="GET"):
+        c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
+        c.request(method, path)
+        r = c.getresponse()
+        return r.status, r.read()
+
+    def test_token_is_required(self):
+        self.assertEqual(self.get("/status")[0], 403)
+        self.assertEqual(self.get("/status?k=wrong")[0], 403)
+        status, body = self.get(f"/status?k={self.token}")
+        self.assertEqual(status, 200)
+        self.assertIn("screen", json.loads(body))
+
+    def test_serves_the_viewer_page(self):
+        status, body = self.get(f"/view?k={self.token}&src=test")
+        self.assertEqual(status, 200)
+        self.assertIn(b"VideoDecoder", body)
+
+    def test_snapshot_needs_the_key(self):
+        self.assertEqual(self.get("/snapshot?display=1")[0], 403)
+
+    def test_separate_without_accessibility_explains(self):
+        if json.loads(self.get(f"/status?k={self.token}")[1])["accessibility"]:
+            self.skipTest("this Mac allows Accessibility here")
+        ws = WS(self.port, f"/stream?k={self.token}&src=separate:1")
+        self.assertEqual(ws.status, 101)
+        for _ in range(5):
+            op, data = ws.recv()
+            msg = json.loads(data) if op == 1 else {}
+            if msg.get("t") in ("error", "closed"):
+                break
+        self.assertIn("Accessibility", msg.get("message", msg.get("reason", "")))
+        ws.close()
+
+    def test_lists_displays(self):
+        status, body = self.get(f"/displays?k={self.token}")
+        self.assertEqual(status, 200)
+        self.assertIsInstance(json.loads(body)["displays"], list)
+
+    def ticket(self, src):
+        status, body = self.get(f"/ticket?k={self.token}&src={src}", method="POST")
+        self.assertEqual(status, 200)
+        return json.loads(body)["ticket"]
+
+    def test_ping_and_page_are_open_but_streams_are_not(self):
+        self.assertEqual(self.get("/ping"), (200, b"frame-mac-view"))
+        self.assertEqual(WS(self.port, "/stream?src=test").status, 403)
+        self.assertEqual(self.get("/ticket?src=test", method="POST")[0], 403)
+
+    def test_tickets_are_single_use_and_tied_to_one_source(self):
+        t = self.ticket("test")
+        self.assertEqual(WS(self.port, f"/stream?src=display:1&t={t}").status, 403)  # wrong source
+        ws = WS(self.port, f"/stream?src=test&t={t}")
+        self.assertEqual(ws.status, 101)
+        hello = json.loads(ws.recv()[1])
+        self.assertEqual(hello["t"], "hello")
+        # Until the viewer acknowledges, a retry (the hello got lost) gets the
+        # same key, and replaces the first viewer rather than adding one.
+        retry = WS(self.port, f"/stream?src=test&t={t}")
+        self.assertEqual(retry.status, 101)
+        self.assertEqual(json.loads(retry.recv()[1])["r"], hello["r"])
+        time.sleep(0.3)
+        _, body = self.get(f"/status?k={self.token}")
+        self.assertEqual(len(json.loads(body)["streams"]), 1)
+        ws.close()
+        ws = retry
+        ws.send_text(json.dumps({"t": "ack"}))
+        time.sleep(0.3)
+        self.assertEqual(WS(self.port, f"/stream?src=test&t={t}").status, 403)  # spent
+        again = WS(self.port, f"/stream?src=test&r={hello['r']}")  # the viewer reconnecting
+        self.assertEqual(again.status, 101)
+        again.close()
+        ws.close()
+        # Stop revokes the reconnect key.
+        self.get(f"/close?k={self.token}&src=test", method="POST")
+        self.assertEqual(WS(self.port, f"/stream?src=test&r={hello['r']}").status, 403)
+
+    def test_stop_revokes_tickets_not_yet_used(self):
+        t = self.ticket("test")
+        self.get(f"/close?k={self.token}&src=test", method="POST")
+        self.assertEqual(WS(self.port, f"/stream?src=test&t={t}").status, 403)
+
+    def test_bad_frame_lengths_close_the_socket_not_the_agent(self):
+        ws = WS(self.port, f"/stream?src=test&t={self.ticket('test')}")
+        self.assertEqual(ws.status, 101)
+        # A masked frame claiming 2^63 bytes.
+        ws.s.sendall(bytes([0x81, 0xFF]) + struct.pack(">Q", 1 << 63) + os.urandom(4))
+        with self.assertRaises((EOFError, OSError)):
+            for _ in range(1000):
+                ws.recv()
+        ws.close()
+        self.assertEqual(self.get(f"/status?k={self.token}")[0], 200)
+
+    def test_stream_input_and_close(self):
+        ws = WS(self.port, f"/stream?k={self.token}&src=test&codec=h264&fps=30&max=640")
+        self.assertEqual(ws.status, 101)
+        info = None
+        key = None
+        for _ in range(200):
+            op, data = ws.recv()
+            if op == 1:
+                msg = json.loads(data)
+                if msg["t"] == "hello":
+                    continue
+                if msg["t"] == "error":
+                    self.skipTest("no H.264 encoder here: " + msg["message"])
+                if msg["t"] == "info":
+                    info = msg
+            elif op == 2 and data[0] == 1:
+                key = data
+            if info and key:
+                break
+        self.assertEqual(info["src"], "test")
+        self.assertTrue(info["input"])
+        # A keyframe: flags, 8-byte timestamp, sequence number, input echoed,
+        # then Annex B with the SPS first.
+        self.assertEqual(key[17:21], b"\x00\x00\x00\x01")
+        self.assertEqual(key[21] & 0x1F, 7)  # NAL type 7, SPS
+        ws.send_text(json.dumps({"t": "m", "e": "down", "b": 0, "x": 0.5, "y": 0.5}))
+        ws.send_text(json.dumps({"t": "key-frame"}))
+        got_key = False
+        for _ in range(200):
+            op, data = ws.recv()
+            if op == 2 and data[0] == 1:
+                got_key = True
+                break
+        self.assertTrue(got_key, "no keyframe after asking for one")
+        _, body = self.get(f"/status?k={self.token}")
+        self.assertEqual([s["src"] for s in json.loads(body)["streams"]], ["test"])
+        status, body = self.get(f"/close?k={self.token}", method="POST")
+        self.assertEqual(json.loads(body)["closed"], 1)
+        for _ in range(400):
+            op, data = ws.recv()
+            if op == 1 and json.loads(data)["t"] == "close":
+                break
+        self.assertEqual(json.loads(data)["t"], "close")
+        # Without the viewer doing anything, the agent ends the stream itself.
+        time.sleep(1)
+        _, body = self.get(f"/status?k={self.token}")
+        self.assertEqual(json.loads(body)["streams"], [])
+        ws.close()
+
+    def test_timing_reports_and_input_echo(self):
+        # What a viewer does: sync clocks, report each frame, stamp input.
+        ws = WS(self.port, f"/stream?k={self.token}&src=test&codec=h264&fps=30&max=640")
+        self.assertEqual(ws.status, 101)
+        ws.send_text(json.dumps({"t": "w"}))  # keep-warm filler: ignored
+        ws.send_text(json.dumps({"t": "ping", "c": 1.5}))
+        pong, echoed, seqs, info = None, None, [], None
+        clicked = False
+        deadline = time.time() + 10
+        while time.time() < deadline and not (pong and echoed):
+            op, data = ws.recv()
+            if op == 1:
+                msg = json.loads(data)
+                if msg["t"] == "error":
+                    self.skipTest("no H.264 encoder here: " + msg["message"])
+                if msg["t"] == "pong":
+                    pong = msg
+                if msg["t"] == "info":
+                    info = msg
+            elif op == 2:
+                seq, echo = struct.unpack(">II", data[9:17])
+                seqs.append(seq)
+                now = pong["a"] if pong else 0
+                ws.send_text(json.dumps({"t": "rx", "s": seq, "r": now}))
+                ws.send_text(json.dumps({"t": "fd", "f": [[seq, now + 1, now + 2, now + 3]], "drop": 0}))
+                if echo == 7:
+                    echoed = seq
+                if len(seqs) == 3 and not clicked:
+                    ws.send_text(json.dumps({"t": "m", "e": "down", "b": 0, "x": 0.5, "y": 0.5, "i": 7, "tv": now}))
+                    clicked = True
+        self.assertEqual(pong["c"], 1.5)
+        self.assertGreater(pong["a"], 0)
+        self.assertEqual(seqs[:3], sorted(seqs[:3]))
+        self.assertIsNotNone(echoed, "no frame was tagged as the first reply to the click")
+        time.sleep(1.7)  # frames are reported once the viewer has had time
+        stream = json.loads(self.get(f"/stats?k={self.token}")[1])["streams"][0]
+        first = stream["frames"][0]
+        self.assertGreater(first["e1"], first["e0"])
+        self.assertGreaterEqual(first["e0"], first["cap"])
+        self.assertEqual(first["vs"] - first["rx"], 3)
+        self.assertEqual([i["frame"] for i in stream["inputs"] if i["id"] == 7], [echoed])
+        self.assertIn("total", stream["summary"])
+        self.assertEqual(info["warm"], 0)  # off unless FRAME_MAC_VIEW_WARM is set
+        live = json.loads(self.get(f"/status?k={self.token}")[1])["streams"][0]
+        self.assertEqual(live["controller"]["tier"], 0)
+        self.assertIn("fps", live["stats"])
+        ws.close()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_media.py

@@ -0,0 +1,141 @@
+"""Owned media planning, eye isolation, decoder choice and fake-Frame ownership."""
+import json
+import os
+from pathlib import Path
+import struct
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
+import frame_media as media
+import frame_media_player as player
+import frame_media_remote as remote
+import frame_splat as splat
+import server
+
+
+class Media(unittest.TestCase):
+    def test_layout_evidence_and_override(self):
+        for name, layout in [('film_SBS.mp4', 'sbs'), ('film.OU.mkv', 'ou'),
+                             ('film_FSBS.mp4', 'full-sbs'), ('photo_TB.png', 'ou')]:
+            self.assertEqual(media.plan(name)['layout'], layout)
+        self.assertEqual(media.plan('film_SBS_OU.mp4', 'mono')['source'], 'explicit')
+        self.assertEqual(media.plan('film.mkv', metadata={'stereo_mode': 'top_bottom'})['layout'], 'full-ou')
+        for name in ('film.mp4', 'film_SBS_OU.mp4', 'businessbs.mp4'):
+            with self.assertRaises(ValueError):
+                media.plan(name)
+        with self.assertRaises(ValueError):
+            media.plan('film.mkv', metadata={'stereo_mode': 'right_left'})
+
+    def test_unsupported_containers_do_not_flatten_spatial_photos(self):
+        for name in ('spatial.HEIC', 'stereo.mpo', 'cloud.ply', 'cloud.spz', 'app.exe'):
+            with self.assertRaises(ValueError):
+                media.plan(name, 'sbs')
+
+    def test_ou_pixels_keep_each_eye_and_row(self):
+        a, b, c, d = [bytes([n])*8 for n in (1, 2, 3, 4)]
+        data, width, height = media.stereo_pixels(a+b+c+d, 2, 4, 'ou')
+        self.assertEqual((data, width, height), (a+c+b+d, 4, 2))
+        with self.assertRaises(ValueError):
+            media.stereo_pixels(b'bad', 2, 4, 'sbs')
+        self.assertEqual(media.geometry(3840, 2160, 'sbs'), (1920, 1080, 2))
+        self.assertEqual(media.geometry(1920, 1080, 'ou'), (1920, 1080, .5))
+
+    def test_decode_is_hardware_and_one_clock_for_audio(self):
+        for codec, decoder in [('h264', 'h264_v4l2m2m'), ('hevc', 'hevc_v4l2m2m')]:
+            cmd = player.decoder_command(Path('/tmp/a file.mp4'), {'codec_name': codec}, 1280, 720, True)
+            self.assertIn(decoder, cmd)
+            self.assertIn('-re', cmd)
+            self.assertIn('pulse', cmd)
+            self.assertIn(str(Path('/tmp/a file.mp4')), cmd)
+        with self.assertRaises(ValueError):
+            player.decoder_command(Path('x.webm'), {'codec_name': 'vp9'}, 640, 480, False)
+        cmd = player.decoder_command(Path('x.png'), {'codec_name': 'png'}, 640, 480, False, True)
+        self.assertNotIn('-re', cmd)
+        self.assertIn('-frames:v', cmd)
+
+    def test_fake_frame_library_and_traversal(self):
+        with tempfile.TemporaryDirectory() as d, patch.object(remote, 'ROOT', Path(d)):
+            identity = 'a'*32+'/space and quote\'.png'
+            path = Path(d)/identity
+            path.parent.mkdir()
+            path.write_bytes(b'test')
+            self.assertEqual(remote.media_path(identity), path.resolve())
+            for bad in ('../../etc/passwd', '/etc/passwd', 'a'*32+'/..', 'a'*32+'/x/y', None):
+                with self.assertRaises((ValueError, FileNotFoundError)):
+                    remote.media_path(bad)
+            link = path.parent/'link.png'
+            link.symlink_to(path)
+            with self.assertRaises(ValueError):
+                remote.media_path('a'*32+'/link.png')
+            with patch.object(remote, 'status', return_value={'state': 'idle'}):
+                files = remote.run({'action': 'list'})['files']
+                self.assertEqual([f['id'] for f in files], [identity])
+
+    def test_server_rejects_bad_actions_before_ssh(self):
+        with patch.object(server, 'ssh') as ssh:
+            for body in ({'action': 'delete'}, {'action': 'play', 'id': '../x'},
+                         {'action': 'play', 'id': 'a'*32+'/x', 'layout': 'invalid'},
+                         {'action': 'play', 'id': 'a'*32+'/x', 'theatre': 'false'}):
+                with self.assertRaises(server.Failure):
+                    server.media(body)
+            ssh.assert_not_called()
+
+    def test_fake_frame_stop_only_owns_our_unit(self):
+        for rc in (0, 5):  # 5: already collected ("not loaded"), a no-op
+            with patch.object(remote.subprocess, 'run') as run, patch.object(remote, 'status', return_value={'state': 'ended'}):
+                run.return_value.returncode = rc
+                self.assertEqual(remote.run({'action': 'stop'})['state'], 'ended')
+                self.assertEqual(run.call_args.args[0], ['systemctl', '--user', 'stop', 'frame-control-media.service'])
+        with patch.object(remote.subprocess, 'run') as run, patch.object(remote, 'status', return_value={}):
+            run.return_value.returncode, run.return_value.stderr = 1, 'Access denied'
+            with self.assertRaisesRegex(RuntimeError, 'Access denied'):
+                remote.run({'action': 'stop'})
+
+    def test_start_failure_reports_systemd_error(self):
+        with tempfile.TemporaryDirectory() as d, patch.object(remote, 'ROOT', Path(d)), \
+                patch.object(remote, 'STATUS', Path(d)/'status.json'), \
+                patch.object(remote, 'active', return_value=False), \
+                patch.object(remote.subprocess, 'run') as run:
+            identity = 'b'*32+'/still_SBS.png'
+            (Path(d)/identity).parent.mkdir()
+            (Path(d)/identity).write_bytes(b'x')
+            run.return_value.returncode, run.return_value.stderr = 1, 'Unit already exists'
+            with patch.object(remote, 'probe', return_value=({}, False)), \
+                    self.assertRaisesRegex(RuntimeError, 'Unit already exists'):
+                remote.run({'action': 'play', 'id': identity})
+            self.assertEqual(run.call_args.args[0][0], 'systemd-run')  # reset-failed's result is ignored
+
+    def test_upload_rejects_unplayable_names_and_keeps_copy_error(self):
+        with patch.object(server, 'ssh') as ssh, patch.object(server, 'push_file') as push:
+            # On Windows a backslash is a separator, so such a name can't reach here.
+            for name in ('.hidden.mp4',) + (('a\\b_SBS.mp4',) if os.sep == '/' else ()):
+                with self.assertRaises(server.Failure):
+                    server.push_media(Path('/tmp')/name)
+            ssh.assert_not_called()
+            push.side_effect = server.Failure('copy failed')
+            ssh.side_effect = [None, server.Failure('link down')]
+            with self.assertRaisesRegex(server.Failure, 'copy failed'):
+                server.push_media(Path('/tmp/film_SBS.mp4'))
+
+    def test_splat_invalid_records_and_stereo_parallax(self):
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d)/'small.splat'
+            path.write_bytes(struct.pack('<6f8B', 0, 0, 0, .001, .001, .001,
+                                         255, 0, 0, 255, 255, 128, 128, 128))
+            data, w, h = splat.render(path, 64, 48)
+            self.assertEqual((len(data), w, h), (128*48*4, 128, 48))
+            def centroid(eye):
+                weights = [(x, data[(y*w+x+eye*64)*4]) for y in range(h) for x in range(64)]
+                return sum(x*v for x,v in weights)/sum(v for x,v in weights)
+            self.assertGreater(centroid(0), centroid(1))
+            for bad in (b'', b'bad', struct.pack('<6f8B', float('nan'), 0, 0, 1, 1, 1, *([128]*8))):
+                path.write_bytes(bad)
+                with self.assertRaises(ValueError):
+                    splat.read(path)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_server.py

@@ -5,6 +5,7 @@ request guards and input validation, which all run before any SSH call.
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import http.client
 import io
 import json

tests/test_steam.py

@@ -2,6 +2,7 @@
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import json
 import subprocess
 import sys
@@ -13,6 +14,7 @@ ROOT = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(ROOT / "ui"))
 
 import frame_store  # noqa: E402
+import frame_steam  # noqa: E402
 import test_server  # noqa: E402  (not `from … import`, or unittest runs ServerGuards twice)
 
 
@@ -49,6 +51,39 @@ class FrameSteamHelper(unittest.TestCase):
             self.assertEqual(out.returncode, 1, args)
             self.assertIn("error", json.loads(out.stdout), args)
 
+    def test_installed_game_is_not_reinstalled(self):
+        with mock.patch.object(frame_steam, "Page") as page, \
+                mock.patch.object(frame_steam, "steam_url") as launch:
+            page.return_value.eval.return_value = {"name": "Gravitas", "installed": True}
+            self.assertEqual(frame_steam.install(1067310)["state"], "installed")
+        launch.assert_not_called()
+
+    def test_license_and_eula_wait_for_headset(self):
+        # Seen during #26's free Gravitas install: state 3 is not permission
+        # to click through the license. The same guard applies to an EULA.
+        for state in (3, 8):
+            with self.subTest(state=state), \
+                    mock.patch.object(frame_steam, "Page") as page, \
+                    mock.patch.object(frame_steam, "steam_url") as launch, \
+                    mock.patch.object(frame_steam.time, "sleep"):
+                page.return_value.eval.side_effect = [
+                    None, {"app": 1067310, "state": state, "need": 1, "free": 100}]
+                self.assertEqual(frame_steam.install(1067310)["state"], "headset")
+                launch.assert_called_once_with("steam://install/1067310")
+                self.assertNotIn(mock.call("SteamClient.Installs.ContinueInstall()"),
+                                 page.return_value.eval.call_args_list)
+
+    def test_insufficient_space_leaves_options_open(self):
+        with mock.patch.object(frame_steam, "Page") as page, \
+                mock.patch.object(frame_steam, "steam_url"), \
+                mock.patch.object(frame_steam.time, "sleep"):
+            page.return_value.eval.side_effect = [
+                {"name": "Gravitas", "installed": False},
+                {"app": 1067310, "state": 7, "need": 100, "free": 10}]
+            self.assertEqual(frame_steam.install(1067310)["state"], "headset")
+            self.assertNotIn(mock.call("SteamClient.Installs.ContinueInstall()"),
+                             page.return_value.eval.call_args_list)
+
 
 class HelperErrors(unittest.TestCase):
     def test_json_error_found_despite_ssh_stderr(self):

tests/test_telemetry.py

@@ -0,0 +1,448 @@
+"""Anonymous analytics (ui/frame_telemetry.py): what's collected at each level,
+what's scrubbed, and that nothing is sent without a key, the notice, or consent.
+
+Run: python3 -m unittest discover -s tests
+"""
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
+import json
+import os
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "ui"))
+
+import frame_compat_db as db  # noqa: E402
+import frame_report as fr  # noqa: E402
+import frame_telemetry as tm  # noqa: E402
+
+
+class Base(unittest.TestCase):
+    """A packaged build with a key, its state in a temp folder."""
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        state = Path(tmp.name)
+        for name, value in (("STATE", state), ("SETTINGS", state / "settings.json"),
+                            ("OUTBOX", state / "outbox.jsonl"), ("SENT", state / "sent.jsonl")):
+            p = mock.patch.object(tm, name, value)
+            p.start()
+            self.addCleanup(p.stop)
+        env = mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_KEY": "phc_test", "FRAME_CONTROL_PACKAGED": "1",
+                                           "FRAME_CONTROL_POSTHOG_HOST": "http://127.0.0.1:9",
+                                           "FRAME_CONTROL_VERSION": "9.9.9"})
+        env.start()
+        self.addCleanup(env.stop)
+        for k in ("DO_NOT_TRACK", "FRAME_CONTROL_TELEMETRY"):
+            os.environ.pop(k, None)
+        tm._seen_errors.clear()
+
+    def queued(self):
+        return tm._read_lines(tm.OUTBOX)
+
+
+class Gates(Base):
+    def test_blocked_without_key_or_in_a_checkout_or_by_do_not_track(self):
+        self.assertIsNone(tm.blocked())
+        with mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_KEY": ""}), \
+                mock.patch.object(tm, "HERE", Path(tempfile.gettempdir()) / "no-config-here"):
+            self.assertIn("key", tm.blocked())
+        with mock.patch.dict(os.environ, {"FRAME_CONTROL_PACKAGED": ""}):
+            self.assertIn("source checkout", tm.blocked())
+        with mock.patch.dict(os.environ, {"DO_NOT_TRACK": "1"}):
+            self.assertIn("DO_NOT_TRACK", tm.blocked())
+            self.assertFalse(tm.capture("app_opened"))
+        self.assertFalse(tm.OUTBOX.exists())
+
+    def test_usage_is_on_by_default_the_others_are_opt_in(self):
+        self.assertTrue(tm.capture("app_opened"))
+        self.assertFalse(tm.capture("compat_report", {}, level="compat"))
+        self.assertFalse(tm.capture("$exception", {}, level="diagnostics"))
+        self.assertEqual([e["event"] for e in self.queued()], ["app_opened"])
+
+    def test_events_are_anonymous(self):
+        tm.capture("app_opened")
+        e = self.queued()[0]
+        self.assertEqual(e["distinct_id"], tm.settings()["id"])
+        self.assertIs(e["properties"]["$process_person_profile"], False)
+        self.assertIs(e["properties"]["$geoip_disable"], True)
+        self.assertEqual(e["properties"]["app_version"], "9.9.9")
+
+    def test_turning_a_level_off_drops_its_unsent_events(self):
+        tm.update_settings({"diagnostics": True})
+        tm.capture("app_opened")
+        tm.diagnostic("somewhere", RuntimeError("boom"))
+        self.assertEqual(len(self.queued()), 2)
+        tm.update_settings({"diagnostics": False})
+        self.assertEqual([e["event"] for e in self.queued()], ["app_opened"])
+        tm.update_settings({"usage": False})
+        self.assertEqual(self.queued(), [])
+        self.assertFalse(tm.capture("app_opened"))
+
+    def test_the_same_error_is_sent_once_in_a_while(self):
+        tm.update_settings({"diagnostics": True})
+        for _ in range(3):
+            tm.diagnostic("POST /api/android install", RuntimeError("boom"))
+        self.assertEqual(len(self.queued()), 1)
+
+    def test_page_events_are_checked(self):
+        self.assertTrue(tm.page_event({"event": "tab_viewed", "properties": {"tab": "android", "extra": "x"}})["queued"])
+        self.assertEqual(self.queued()[0]["properties"].get("extra"), None)
+        with self.assertRaises(ValueError):
+            tm.page_event({"event": "anything_else"})
+        with self.assertRaises(ValueError):
+            tm.page_event({"event": "tab_viewed", "properties": {"tab": "/Users/me/secret"}})
+
+
+class Lifecycle(Base):
+    def test_install_update_and_one_open_a_day(self):
+        tm.app_started()
+        tm.app_started()
+        self.assertEqual([e["event"] for e in self.queued()], ["app_installed", "app_opened"])
+        with mock.patch.dict(os.environ, {"FRAME_CONTROL_VERSION": "10.0.0"}):
+            tm.app_started()
+        e = self.queued()[-1]
+        self.assertEqual((e["event"], e["properties"]["from_version"]), ("app_updated", "9.9.9"))
+
+    def test_frame_build_once(self):
+        tm.frame_seen("20260922.1", "3.8")
+        tm.frame_seen("20260922.1", "3.8")
+        self.assertEqual(len(self.queued()), 1)
+
+
+class Sending(Base):
+    def serve(self, status=200):
+        got = []
+
+        class H(BaseHTTPRequestHandler):
+            def do_POST(self):
+                got.append((self.path, json.loads(self.rfile.read(int(self.headers["Content-Length"])))))
+                self.send_response(status)
+                self.end_headers()
+                self.wfile.write(b'{"status": 1}')
+
+            def log_message(self, *a):
+                pass
+
+        httpd = HTTPServer(("127.0.0.1", 0), H)
+        threading.Thread(target=httpd.serve_forever, daemon=True).start()
+        self.addCleanup(httpd.server_close)
+        self.addCleanup(httpd.shutdown)
+        os.environ["FRAME_CONTROL_POSTHOG_HOST"] = f"http://127.0.0.1:{httpd.server_port}"
+        return got
+
+    def test_nothing_is_sent_before_the_notice_was_shown(self):
+        got = self.serve()
+        tm.capture("app_opened")
+        self.assertEqual(tm.flush(), 0)
+        self.assertEqual(got, [])
+        tm.update_settings({"noticeShown": True})
+        self.assertEqual(tm.flush(), 1)
+        path, body = got[0]
+        self.assertEqual((path, body["api_key"], body["batch"][0]["event"]), ("/batch/", "phc_test", "app_opened"))
+        self.assertEqual(self.queued(), [])
+        self.assertEqual([e["event"] for e in tm.state()["sent"]], ["app_opened"])
+
+    def test_a_failed_send_keeps_the_events(self):
+        self.serve(status=500)
+        tm.update_settings({"noticeShown": True})
+        tm.capture("app_opened")
+        self.assertEqual(tm.flush(), 0)
+        self.assertEqual(len(self.queued()), 1)
+
+
+class Scrub(unittest.TestCase):
+    def test_personal_details_are_removed(self):
+        home = str(Path.home())
+        text = (f"open {home}/Downloads/My Game.apk failed; ssh alex@192.168.1.20 (frame.local) "
+                "key ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIM steam 76561198000000000 mac 3c:22:fb:12:34:56 "
+                "url https://example.com/private/path?token=abc phc_abcdefghijklmnopqrstu C:\\Users\\Bob\\x "
+                "/home/carol/y")
+        out = tm.scrub(text)
+        for leaked in (home, "192.168.1.20", "frame.local", "AAAAC3Nza", "76561198000000000", "3c:22:fb",
+                       "private/path", "phc_abcdefghijklmnopqrstu", "Bob", "carol", "alex@"):
+            self.assertNotIn(leaked, out)
+        self.assertIn("https://example.com/…", out)
+        self.assertIn("~/Downloads", out)
+
+    def test_categories(self):
+        self.assertEqual(tm.categorize("adb: failed to install: INSTALL_FAILED_NO_MATCHING_ABIS: x"),
+                         ("android_installer", "INSTALL_FAILED_NO_MATCHING_ABIS"))
+        self.assertEqual(tm.categorize("X has no arm64-v8a build (armeabi-v7a)")[0], "apk_wrong_abi")
+        self.assertEqual(tm.categorize("ssh: connect to host 10.0.0.2 port 22: Connection refused")[0],
+                         "frame_unreachable")
+        self.assertEqual(tm.categorize("something new")[0], "other")
+
+
+class Compat(Base):
+    def test_reports_are_shared_only_after_opting_in_without_file_names(self):
+        r = {"id": "r1", "package": "org.example", "version": "1.0", "rating": "works", "via": "user",
+             "notes": f"from {Path.home()}/x", "source": "MyPrivateBuild.apk", "date": "2026-09-28T10:00:00"}
+        self.assertFalse(tm.compat_report(r))
+        tm.update_settings({"compat": True})
+        self.assertTrue(tm.compat_report(r))
+        p = self.queued()[-1]["properties"]
+        self.assertEqual((p["package"], p["rating"], p["id"]), ("org.example", "works", "r1"))
+        self.assertNotIn("source", p)
+        self.assertNotIn(str(Path.home()), p["notes"])
+        r2 = dict(r, id="r2", source="https://f-droid.org/repo/org.example_1.apk")
+        tm.compat_report(r2)
+        self.assertEqual(self.queued()[-1]["properties"]["source"], "https://f-droid.org/…")
+
+    def test_opting_in_shares_earlier_local_reports(self):
+        with mock.patch.object(db, "shared", return_value=False), \
+                mock.patch.object(db, "_outbox", return_value=[{"id": "old1", "package": "org.a", "rating": "works",
+                                                                "date": "2026-09-01T00:00:00"}]):
+            tm.update_settings({"compat": True})
+            tm.update_settings({"compat": True})  # already sent: not again
+        self.assertEqual([e["properties"]["id"] for e in self.queued() if e["event"] == "compat_report"], ["old1"])
+
+
+class ApkInstallReports(unittest.TestCase):
+    """server.apk_installed: an APK that won't install is reported; connection trouble isn't."""
+
+    def setUp(self):
+        import server
+        self.server = server
+        for target, name in ((server.frame_catalog, "add_report"), (server.frame_telemetry, "install_finished")):
+            p = mock.patch.object(target, name)
+            setattr(self, name, p.start())
+            self.addCleanup(p.stop)
+
+    def test_wrong_abi_is_an_install_failed_report(self):
+        info = {"package": "org.x", "version": "2.0", "label": "X"}
+        self.server.apk_installed(info, None, self.server.frame_android.FrameError(
+            "X has no arm64-v8a build (armeabi-v7a); Lepton is 64-bit ARM only"), 3.0)
+        args, kw = self.add_report.call_args
+        self.assertEqual((args[0], args[1], kw["result"], kw["via"]), ("org.x", "2.0", "install_failed", "install"))
+        self.assertIs(self.install_finished.call_args[0][1], False)
+
+    def test_connection_trouble_is_not_reported(self):
+        self.server.apk_installed({"package": "org.x", "version": "2.0"}, None,
+                                  self.server.frame_android.FrameError("timed out talking to frame"), 3.0)
+        self.add_report.assert_not_called()
+
+    def test_private_package_names_stay_here(self):
+        with mock.patch.dict(self.server.frame_catalog._cache, {"by_pkg": {"org.public": {}}}):
+            self.server.apk_installed({"package": "com.private.thing", "version": "1"}, {"package": "com.private.thing"},
+                                      None, 2.0)
+            self.assertIsNone(self.install_finished.call_args[1]["package"])
+            self.server.apk_installed({"package": "org.public", "version": "1"}, {"package": "org.public"}, None, 2.0)
+            self.assertEqual(self.install_finished.call_args[1]["package"], "org.public")
+
+
+class CommunitySync(unittest.TestCase):
+    def ev(self, i, who="a", day="2026-09-28", **kw):
+        return ({"id": f"id{i}", "package": "org.x", "rating": "works", "date": f"{day}T00:00:00",
+                 "via": "probe", **kw}, who, f"{day} 10:00:00")
+
+    def test_rows_are_validated_marked_and_capped_per_reporter(self):
+        events = [self.ev(i) for i in range(5)] + [self.ev(9, who="b", rating="nonsense"), self.ev(10, who="b")]
+        rows, skipped = db.community_rows(events, {}, cap=3)
+        self.assertEqual([r["id"] for r in rows], ["id0", "id1", "id2", "id10"])
+        self.assertTrue(all(r["via"] == "community-probe" for r in rows))
+        self.assertEqual(len(skipped), 3)
+
+    def test_the_cap_and_duplicates_hold_across_syncs(self):
+        state = {}
+        rows, _ = db.community_rows([self.ev(i) for i in range(3)], state, cap=3)
+        self.assertEqual(len(rows), 3)
+        rows, skipped = db.community_rows([self.ev(i) for i in range(6)], state, cap=3)  # overlapping re-read
+        self.assertEqual(rows, [])
+        self.assertEqual([why for _, why in skipped], ["over the daily limit for one reporter"] * 3)
+
+    def test_a_malformed_event_is_skipped_not_fatal(self):
+        rows, skipped = db.community_rows([self.ev(1, via=["probe"]), ("not json", "a", "2026-09-28"), self.ev(2)], {})
+        self.assertEqual([r["id"] for r in rows], ["id2"])
+        self.assertEqual(len(skipped), 2)
+
+
+class Regressions(Base):
+    """Findings from the cross-provider review."""
+
+    def test_urls_lose_credentials_paths_and_private_hosts(self):
+        for text, leaked in (("https://alice:secret@example.com/private.apk?token=credential", ("alice", "secret", "private", "credential")),
+                             ("https://alice:secret@192.168.1.4/private.apk", ("alice", "192.168", "private")),
+                             ("fe80::1234 and 2001:db8::5", ("fe80", "2001:db8")),
+                             ("sk-proj-abcdefghijklmnopqrstuv", ("abcdefghijk",)),
+                             ("http://frame.local:8080/x", ("frame.local", "8080"))):
+            out = tm.scrub(text)
+            for s in leaked:
+                self.assertNotIn(s, out, (text, out))
+
+    def test_compat_labels_versions_and_sources_are_scrubbed(self):
+        tm.update_settings({"compat": True})
+        tm.compat_report({"id": "r9", "package": "org.x", "rating": "works", "date": "2026-09-28T00:00:00",
+                          "label": "alice@example.com build", "version": "1.0-alice@example.com",
+                          "source": "https://alice:secret@192.168.1.4/private.apk"})
+        p = self.queued()[-1]["properties"]
+        self.assertNotIn("alice", json.dumps(p))
+        self.assertNotIn("source", p)
+
+    def test_an_unsent_report_is_shared_again_after_opting_out_and_in(self):
+        with mock.patch.object(db, "shared", return_value=False), \
+                mock.patch.object(db, "_outbox", return_value=[{"id": "q1", "package": "org.a", "rating": "works",
+                                                                "date": "2026-09-01T00:00:00"}]):
+            tm.update_settings({"compat": True})
+            tm.update_settings({"compat": False})
+            self.assertEqual(self.queued(), [])
+            tm.update_settings({"compat": True})
+        self.assertEqual([e["properties"]["id"] for e in self.queued() if e["event"] == "compat_report"], ["q1"])
+
+    def test_opting_out_waits_for_a_send_in_progress(self):
+        tm.update_settings({"noticeShown": True})
+        tm.capture("app_opened")
+        order = []
+        started = threading.Event()
+
+        def slow_open(req, timeout):
+            started.set()
+            time.sleep(0.3)
+            order.append("sent")
+            return mock.MagicMock(__enter__=lambda s: s, __exit__=lambda *a: False, read=lambda: b"{}")
+
+        with mock.patch.object(tm.urllib.request, "urlopen", side_effect=slow_open):
+            th = threading.Thread(target=tm.flush)
+            th.start()
+            started.wait(2)
+            tm.update_settings({"usage": False})
+            order.append("opted out")
+            th.join()
+        self.assertEqual(order, ["sent", "opted out"])
+
+    def test_project_id_comes_from_the_config(self):
+        with mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_PROJECT": "12345"}):
+            self.assertEqual(tm.config()["project"], "12345")
+
+
+class ReportProblem(Base):
+    """Report a problem: diagnostics are scrubbed and bounded; the report goes privately to PostHog."""
+
+    def serve(self, status=200):
+        got = []
+
+        class H(BaseHTTPRequestHandler):
+            def do_POST(self):
+                got.append((self.path, json.loads(self.rfile.read(int(self.headers["Content-Length"])))))
+                self.send_response(status)
+                self.end_headers()
+                self.wfile.write(b'{"status":"Ok"}')
+
+            def log_message(self, *a):
+                pass
+
+        httpd = HTTPServer(("127.0.0.1", 0), H)
+        threading.Thread(target=httpd.serve_forever, daemon=True).start()
+        self.addCleanup(httpd.server_close)
+        self.addCleanup(httpd.shutdown)
+        p = mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_HOST": f"http://127.0.0.1:{httpd.server_port}"})
+        p.start()
+        self.addCleanup(p.stop)
+        return got
+
+    def test_diagnostics_are_scrubbed_and_include_the_log(self):
+        log = tm.STATE / "server.log"
+        log.write_text("GET /api/status 200\nTraceback: ssh alice@192.168.1.9 failed in %s/x\n" % Path.home())
+        with mock.patch.dict(os.environ, {"FRAME_CONTROL_LOG": str(log)}):
+            text = fr.diagnostics(["16:00  Install failed: https://bob:pw@example.com/a.apk"], include_logs=True, limit=5000)
+        self.assertIn("Frame Control 9.9.9", text)
+        self.assertIn("Traceback", text)
+        self.assertNotIn("GET /api/status", text)
+        for leaked in ("alice", "192.168.1.9", str(Path.home()), "bob", "pw@"):
+            self.assertNotIn(leaked, text)
+
+    def test_a_report_is_bounded_in_utf16_units(self):
+        body = {"title": "Live view stops", "message": "It stops 😀 " * 800, "diagnostics": "log 😀 line\n" * 2000}
+        title, text, diag = fr.compose(body)
+        self.assertLessEqual(fr.u16(text), fr.TEXT_MAX)
+        self.assertLessEqual(fr.u16(diag), fr.DIAG_MAX)
+        self.assertTrue(text.startswith("It stops"))
+        with self.assertRaises(ValueError):
+            fr.compose({"title": "hi", "message": "It stops after a minute."})
+
+    def test_logs_only_when_asked_and_environment_is_kept_first(self):
+        log = tm.STATE / "server.log"
+        log.write_text("".join(f"old line {i}\n" for i in range(200)) + "newest line\n")
+        with mock.patch.dict(os.environ, {"FRAME_CONTROL_LOG": str(log)}):
+            plain = fr.diagnostics(["Copy Jane Doe tax return.pdf to ~/Downloads"])
+            full = fr.diagnostics(["Install failed"], include_logs=True, limit=400)
+        self.assertNotIn("Jane Doe", plain)
+        self.assertNotIn("line", plain)
+        self.assertTrue(full.startswith("Frame Control 9.9.9"))
+        self.assertIn("Install failed", full)
+        self.assertIn("newest line", full)
+        self.assertLessEqual(fr.u16(full), 400)
+
+    def test_the_previewed_diagnostics_are_what_is_sent(self):
+        got = self.serve()
+        fr.send({"title": "Live view stops", "message": "It stops after a minute.",
+                 "diagnostics": "Frame Control 9.9.9\nssh janes-mac.tail12345.ts.net failed"})
+        diag = got[0][1]["batch"][0]["properties"]["diagnostics"]
+        self.assertIn("Frame Control 9.9.9", diag)
+        self.assertNotIn("janes-mac", diag)
+
+    def test_send_is_a_private_posthog_event_whatever_the_settings(self):
+        got = self.serve()
+        tm.update_settings({"usage": False})  # analytics off: a deliberate report still goes
+        res = fr.send({"kind": "idea", "title": "Live view stops", "message": "It stops after a minute.",
+                       "contact": "me@example.com"})
+        path, body = got[0]
+        event = body["batch"][0]
+        self.assertEqual((path, body["api_key"], event["event"]), ("/batch/", "phc_test", "problem_report"))
+        props = event["properties"]
+        self.assertEqual((props["kind"], props["title"], props["message"], props["contact"], props["report_id"]),
+                         ("idea", "Live view stops", "It stops after a minute.", "me@example.com", res["id"]))
+        self.assertEqual((props["$process_person_profile"], props["$geoip_disable"]), (False, True))
+        self.assertNotEqual(event["distinct_id"], tm.settings()["id"])  # not linked to the analytics
+        self.assertIn(res["id"], res["message"])
+        self.assertEqual([e["event"] for e in tm._read_lines(tm.SENT)], ["problem_report"])
+
+    def test_a_sent_report_is_not_an_error_if_the_local_log_fails(self):
+        self.serve()
+        with mock.patch.object(tm, "record_sent", side_effect=OSError("disk full")):
+            res = fr.send({"title": "Live view stops", "message": "It stops after a minute."})
+        self.assertTrue(res["id"])
+
+    def test_events_queued_by_older_versions_get_the_placeholder_address(self):
+        got = self.serve()
+        tm.update_settings({"noticeShown": True})
+        tm._write_lines(tm.OUTBOX, [{"event": "app_opened", "distinct_id": "x", "uuid": "u1",
+                                     "properties": {"level": "usage"}}])
+        self.assertEqual(tm.flush(), 1)
+        self.assertEqual(got[0][1]["batch"][0]["properties"]["$ip"], "0.0.0.0")
+        self.assertEqual(tm._read_lines(tm.SENT)[0]["properties"]["$ip"], "0.0.0.0")
+
+    def test_the_inbox_skips_malformed_reports(self):
+        good = ["2026-09-28T09:50:00Z", "AB12CD34", "bug", "Live view stops", "It stops.", None,
+                "0.4.0", "macOS", "", ""]
+        rows = [["2026-09-28T10:00:00Z", "X", "bug", "Hand-made", None, None, None, None, None, None], ["short"], good]
+        with mock.patch.object(db, "_posthog_query", return_value={"results": rows}), \
+             mock.patch.object(sys, "argv", ["frame_report.py", "inbox"]), \
+             mock.patch("builtins.print") as out:
+            fr.main()
+        printed = " ".join(str(c.args[0]) for c in out.call_args_list if c.args)
+        self.assertIn("AB12CD34", printed)
+        self.assertIn("Hand-made", printed)
+
+    def test_a_refused_report_is_an_error(self):
+        self.serve(status=401)
+        with self.assertRaisesRegex(fr.ReportError, "HTTP 401"):
+            fr.send({"title": "Live view stops", "message": "It stops after a minute."})
+        self.assertEqual(tm._read_lines(tm.SENT), [])
+
+    def test_no_key_means_no_report(self):
+        with mock.patch.dict(os.environ, {"FRAME_CONTROL_POSTHOG_KEY": ""}), \
+             mock.patch.object(tm, "HERE", tm.STATE):
+            with self.assertRaisesRegex(fr.ReportError, "no PostHog project key"):
+                fr.send({"title": "Live view stops", "message": "It stops after a minute."})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_webinstall.py

@@ -4,6 +4,7 @@ the localhost-testing rule allows.
 
 Run: python3 -m unittest discover -s tests
 """
+import sandbox  # noqa: F401  (first: keeps tests off real data and services)
 import hashlib
 import json
 import os