Wire authenticated voice watch notices through an owned native launcher

apps/desktop/assets/codex-bridge/test_native_speech_mux.py

@@ -0,0 +1,358 @@
+"""Real pipes/processes and real loopback HTTP; no native runtime activation."""
+import concurrent.futures
+import http.client
+import json
+import os
+from pathlib import Path
+import queue
+import subprocess
+import sys
+import threading
+import time
+
+import native_speech_mux as muxmod
+
+
+# Fake engine is a separate process speaking complete JSONL, not mocked Popen.
+ENGINE = r'''
+import json, os, sys, time
+if '--probe' in sys.argv:
+    print(json.dumps({'argv':sys.argv[1:], 'cwd':os.getcwd(), 'marker':os.getenv('MUX_TEST_MARKER')}))
+    sys.stderr.write('original-stderr\n')
+    sys.exit(23)
+for raw in sys.stdin.buffer:
+    try: m=json.loads(raw)
+    except ValueError:
+        sys.stdout.buffer.write(raw); sys.stdout.buffer.flush(); continue
+    method=m.get('method')
+    def emit(v):
+        sys.stdout.buffer.write(json.dumps(v).encode()+b'\n'); sys.stdout.buffer.flush()
+    if method=='thread/start':
+        emit({'id':m['id'],'result':{'thread':{'id':'thread-A'}}})
+    elif method=='thread/realtime/start':
+        emit({'method':'thread/realtime/started','params':{'threadId':'thread-A','realtimeSessionId':'rt'}})
+    elif method=='thread/realtime/stop':
+        emit({'method':'thread/realtime/closed','params':{'threadId':'thread-A','reason':None}})
+    elif method=='error-event':
+        emit({'method':'thread/realtime/error','params':{'threadId':'thread-A','message':'failed'}})
+    elif method=='thread/realtime/appendSpeech':
+        emit({'method':'item/commandExecution/requestApproval','id':m['id'],'params':{'command':'test'}})
+        emit({'method':'seen-speech','params':m['params']})
+        emit({'id':m['id'], 'error':{'code':-1,'message':'no'}} if m['params']['text']=='rpc-error' else {'id':m['id'],'result':{}})
+    elif method=='delay':
+        time.sleep(.25)
+        sys.stdout.buffer.write(raw); sys.stdout.buffer.flush()
+    elif method=='pause':
+        emit({'method':'paused'})
+        while not os.path.exists(m['params']['release']): time.sleep(.01)
+        emit({'method':'released'})
+    elif method=='exit':
+        sys.exit(17)
+    else:
+        sys.stdout.buffer.write(raw); sys.stdout.buffer.flush()
+sys.stdout.buffer.write(b'{"method":"saw-eof"}\n'); sys.stdout.buffer.flush()
+sys.stderr.write('engine-eof\n')
+sys.exit(19)
+'''
+
+
+class Harness:
+    def __init__(self, tmp_path):
+        engine = tmp_path / 'engine.py'
+        engine.write_text(ENGINE)
+        self.mux = muxmod.NativeSpeechMux([sys.executable, '-u', str(engine), 'app-server'], observer=True)
+        ir, iw = os.pipe()
+        outr, outw = os.pipe()
+        self.source = os.fdopen(ir, 'rb', buffering=0)
+        self.input = os.fdopen(iw, 'wb', buffering=0)
+        self.target = os.fdopen(outw, 'wb', buffering=0)
+        self.output = os.fdopen(outr, 'rb')
+        self.lines = queue.Queue()
+        self.code = None
+        self.error = None
+        self.url, self.token = self.mux.start_inlet()
+        def run():
+            try:
+                self.code = self.mux.run(self.source, self.target)
+            except BaseException as e:
+                self.error = e
+            finally:
+                self.target.close()
+        self.thread = threading.Thread(target=run, daemon=True)
+        self.thread.start()
+        def collect():
+            for raw in self.output:
+                self.lines.put(raw)
+        self.collector = threading.Thread(target=collect, daemon=True)
+        self.collector.start()
+
+    def send(self, value):
+        raw = value if isinstance(value, bytes) else json.dumps(value).encode()+b'\n'
+        self.input.write(raw)
+
+    def recv(self):
+        return self.lines.get(timeout=8)
+
+    def http(self, path, body, token=None):
+        conn = http.client.HTTPConnection('127.0.0.1', int(self.url.rsplit(':', 1)[1]), timeout=5)
+        conn.request('POST', path, json.dumps(body), {'Authorization':'Bearer '+(token if token is not None else self.token)})
+        response = conn.getresponse()
+        data = response.read()
+        conn.close()
+        return response.status, json.loads(data) if response.status == 200 else data
+
+    def start(self):
+        self.send({'id':1, 'method':'thread/start','params':{}})
+        assert json.loads(self.recv())['result']['thread']['id'] == 'thread-A'
+        assert self.mux.snapshot()['threads']['thread-A']['active'] is False
+        self.send({'id':2,'method':'thread/realtime/start','params':{'threadId':'thread-A'}})
+        assert json.loads(self.recv())['method'] == 'thread/realtime/started'
+        return self.bind()
+
+    def bind(self):
+        generation = self.mux.snapshot()['threads']['thread-A']['generation']
+        b = muxmod.Binding('tenant','bekci','rev',self.mux.connection_id,'thread-A',generation)
+        assert self.mux.bind(b)
+        return self.notice(generation)
+
+    def notice(self, generation=1, **changes):
+        now = time.time()
+        return dict(notice_id='notice-1',tenant_id='tenant',bekci_session_id='bekci',history_revision='rev',
+                    connection_id=self.mux.connection_id,thread_id='thread-A',expected_generation=generation,
+                    issued_at=now,expires_at=now+30,text='düzeltme', **changes)
+
+    def close(self):
+        self.input.close()
+        self.thread.join(10)
+        if self.thread.is_alive():
+            self.mux._process.kill()
+            self.thread.join(5)
+            raise AssertionError('mux failed to exit after EOF')
+        self.collector.join(3)
+        self.source.close()
+        self.output.close()
+        assert self.error is None
+
+
+def test_native_notice_rpc_and_approval_are_separate(tmp_path):
+    h = Harness(tmp_path)
+    try:
+        notice = h.start()
+        assert h.http('/notice', notice, 'wrong')[0] == 401
+        assert h.http('/notice', notice)[1]['status'] == 'queued'
+        approval = json.loads(h.recv())
+        assert approval['method'] == 'item/commandExecution/requestApproval'
+        assert approval['id'].startswith('bekci-speech-')
+        assert json.loads(h.recv()) == {'method':'seen-speech','params':{'threadId':'thread-A','text':'düzeltme'}}
+        answer = (json.dumps({'id':approval['id'],'result':{'decision':'decline'}}, separators=(', ', ': '))+'\r\n').encode()
+        h.send(answer)
+        assert h.recv() == answer
+        assert h.http('/notice', notice)[1]['status'] == 'rpc_accepted'
+        assert h.lines.empty()  # native own-result never leaked to desktop
+        assert h.http('/rpc', {'method':'turn/start'})[0] == 404
+        assert h.http('/notice', {**notice,'notice_id':'err','text':'rpc-error'})[1]['status'] == 'queued'
+        h.recv(); h.recv()
+        h.send(b'{"method":"barrier"}\n')
+        assert h.recv() == b'{"method":"barrier"}\n'
+        assert h.http('/notice', {**notice,'notice_id':'err','text':'rpc-error'})[1]['status'] == 'rpc_error'
+    finally:
+        h.close()
+    assert h.code == 19
+    assert json.loads(h.recv())['method'] == 'saw-eof'
+
+
+def test_unbound_schema_ttl_tenant_generation_and_restart(tmp_path):
+    h = Harness(tmp_path)
+    try:
+        n = h.notice()
+        assert h.http('/notice', n)[1]['status'] == 'binding_not_established'
+        n = h.start()
+        for delta, status in [({'method':'turn/start'},'invalid_schema'),
+                              ({'expected_generation':True},'invalid_schema'),
+                              ({'expires_at':time.time()-1},'expired_or_invalid_ttl'),
+                              ({'issued_at':float('nan')},'expired_or_invalid_ttl'),
+                              ({'tenant_id':'other'},'binding_not_established'),
+                              ({'history_revision':'other'},'binding_mismatch'),
+                              ({'expected_generation':2},'binding_mismatch'),
+                              ({'connection_id':'other'},'binding_mismatch')]:
+            assert h.http('/notice', {**n, **delta})[1]['status'] == status
+        h.send({'method':'thread/realtime/stop','params':{'threadId':'thread-A'}})
+        assert json.loads(h.recv())['method'] == 'thread/realtime/closed'
+        assert h.http('/notice', n)[1]['status'] == 'binding_not_established'
+        h.send({'method':'thread/realtime/start','params':{'threadId':'thread-A'}})
+        h.recv()
+        fresh = h.bind()
+        assert fresh['expected_generation'] == n['expected_generation']+1
+        assert h.http('/notice', n)[1]['status'] == 'binding_mismatch'
+        h.send({'method':'error-event'})
+        h.recv()
+        assert h.http('/notice', fresh)[1]['status'] == 'binding_not_established'
+    finally:
+        h.close()
+
+
+def test_concurrent_notices_and_raw_backpressure_order(tmp_path):
+    h = Harness(tmp_path)
+    try:
+        n = h.start()
+        h.send({'method':'delay'})
+        records = [(json.dumps({'method':'echo','i':i,'padding':'x'*16000})+'\r\n').encode() for i in range(90)]
+        producer = threading.Thread(target=lambda: [h.send(r) for r in records], daemon=True)
+        producer.start()
+        with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+            results = list(pool.map(lambda i: h.http('/notice',{**n,'notice_id':str(i)})[1]['status'], range(12)))
+        assert set(results) <= {'queued','busy'}
+        accepted = results.count('queued')
+        echoes = []
+        speech = 0
+        approvals = 0
+        for _ in range(1 + len(records) + accepted*2):
+            raw = h.recv()
+            method = json.loads(raw).get('method')
+            if method == 'echo': echoes.append(raw)
+            elif method == 'seen-speech': speech += 1
+            elif method == 'item/commandExecution/requestApproval': approvals += 1
+        producer.join(5)
+        assert not producer.is_alive()
+        assert echoes == records
+        assert speech == approvals == accepted
+        assert h.mux._queue.maxsize == 32
+    finally:
+        h.close()
+
+
+def test_fragmented_oversized_jsonl_and_partial_eof_preserved(tmp_path):
+    h = Harness(tmp_path)
+    try:
+        h.start()
+        raw = b'{ "method":"huge", "value":"' + b'x'*200000 + b'"}\r\n'
+        producer = threading.Thread(target=lambda: h.send(raw),daemon=True)
+        producer.start()
+        assert h.recv() == raw
+        producer.join(5)
+        assert h.mux.snapshot()['injection_disabled']
+        h.send(b'{"partial":true}')
+    finally:
+        h.close()
+    # Engine echoes an unterminated record, then its EOF notification: byte exact.
+    assert h.recv() == b'{"partial":true}{"method":"saw-eof"}\n'
+    assert h.code == 19
+
+
+def test_non_appserver_passthrough_argv_env_cwd_stderr_exit(tmp_path, monkeypatch):
+    engine = tmp_path / 'engine.py'
+    engine.write_text(ENGINE)
+    module = Path(muxmod.__file__).resolve()
+    monkeypatch.setenv('MUX_TEST_MARKER','inherited')
+    args = [sys.executable,str(module),sys.executable,'--','-u',str(engine),'--probe','space arg','app-server']
+    p = subprocess.run(args,cwd=tmp_path,capture_output=True,timeout=10)
+    assert p.returncode == 23
+    result = json.loads(p.stdout)
+    assert result == {'argv':['--probe','space arg','app-server'],'cwd':str(tmp_path),'marker':'inherited'}
+    direct = subprocess.run([sys.executable, '-u', str(engine), '--probe', 'space arg', 'app-server'],
+                            cwd=tmp_path, capture_output=True, timeout=10)
+    assert (p.stdout, p.stderr, p.returncode) == (direct.stdout, direct.stderr, direct.returncode)
+
+
+def test_engine_exit_without_desktop_eof(tmp_path):
+    h = Harness(tmp_path)
+    try:
+        h.send({'method':'exit'})
+        h.thread.join(5)
+        assert not h.thread.is_alive()
+        assert h.code == 17
+        assert h.mux.snapshot()['closed']
+    finally:
+        h.close()
+
+
+def test_owned_namespace_collision_disables_injection_but_forwards(tmp_path):
+    h = Harness(tmp_path)
+    try:
+        n = h.start()
+        raw = json.dumps({'id':h.mux._prefix+'collision','method':'echo'}).encode()+b'\n'
+        h.send(raw)
+        assert h.recv() == raw
+        assert h.http('/notice', n)[1]['status'] == 'injection_disabled'
+    finally:
+        h.close()
+
+
+def test_queued_notice_revalidated_after_observed_stop(tmp_path):
+    h = Harness(tmp_path)
+    release = tmp_path / 'release'
+    try:
+        n = h.start()
+        h.send({'method':'pause','params':{'release':str(release)}})
+        assert json.loads(h.recv())['method'] == 'paused'
+        # Fill the OS pipe while the real engine is paused; leave bounded queue
+        # capacity for the notice and stop record behind it.
+        for i in range(8):
+            h.send({'method':'echo','i':i,'padding':'x'*32000})
+        assert h.http('/notice', n)[1]['status'] == 'queued'
+        h.send({'method':'thread/realtime/stop','params':{'threadId':'thread-A'}})
+        deadline = time.monotonic()+5
+        while h.mux.snapshot()['threads']['thread-A']['active'] and time.monotonic() < deadline:
+            time.sleep(.01)
+        assert not h.mux.snapshot()['threads']['thread-A']['active']
+        release.touch()
+        methods = [json.loads(h.recv())['method'] for _ in range(10)]
+        assert methods == ['released'] + ['echo']*8 + ['thread/realtime/closed']
+        assert h.mux._receipts[('tenant','bekci','notice-1')]['status'] == 'binding_not_established'
+        assert h.lines.empty()
+    finally:
+        release.touch()
+        h.close()
+
+def test_real_request_correlation_and_opaque_receipt_contract(tmp_path):
+    h = Harness(tmp_path)
+    try:
+        n = h.start()
+        adapter = muxmod.WatchReceiptAdapter(h.mux)
+        headers = {'Session-Id':'thread-A','Thread-Id':'thread-A'}
+        correlation = adapter.capture(headers, {'prompt_cache_key':'thread-A'})
+        assert correlation.thread_id == 'thread-A'
+        assert adapter.capture({**headers,'Thread-Id':'other'}, {'prompt_cache_key':'thread-A'}) is None
+        assert adapter.capture(headers, {}) is None
+        assert adapter.accept('bekci.watch', {}, correlation) == 'backend_contract_not_established'
+        # Internal contract fixture, explicitly NOT a claim about backend wire format.
+        receipt = muxmod.WatchReceipt('opaque','tenant','bekci','rev','thread-A',time.time()+60)
+        adapter.decode = lambda payload: payload
+        adapter.resolve = lambda handle: receipt
+        assert adapter.accept('bekci.watch','opaque',correlation) == 'bound'
+        assert adapter.accept('_bekci_watch','opaque',correlation) == 'bound'
+        assert adapter.accept('other','opaque',correlation) == 'unsupported_metadata_transport'
+        assert adapter.notice('unbound',n)['status'] == 'binding_not_established'
+        assert adapter.notice('opaque',{**n,'tenant_id':'other'})['status'] == 'binding_mismatch'
+        assert adapter.notice('opaque',n)['status'] == 'queued'
+        assert json.loads(h.recv())['method'] == 'item/commandExecution/requestApproval'
+        assert json.loads(h.recv())['method'] == 'seen-speech'
+        h.send({'method':'thread/realtime/stop','params':{'threadId':'thread-A'}}); h.recv()
+        h.send({'method':'thread/realtime/start','params':{'threadId':'thread-A'}}); h.recv()
+        assert adapter.accept('bekci.watch','opaque',correlation) == 'stale_generation'
+        fresh = adapter.capture(headers, {'prompt_cache_key':'thread-A'})
+        adapter.resolve = lambda handle: muxmod.WatchReceipt('opaque','tenant','bekci','rev','wrong',time.time()+60)
+        assert adapter.accept('bekci.watch','opaque',fresh) == 'receipt_mismatch_or_expired'
+        adapter.resolve = lambda handle: muxmod.WatchReceipt('opaque','tenant','bekci','rev','thread-A',time.time()-1)
+        assert adapter.accept('bekci.watch','opaque',fresh) == 'receipt_mismatch_or_expired'
+    finally:
+        h.close()
+
+
+def test_verified_original_launcher_fallback_actual_process(tmp_path):
+    import hashlib
+    executable = str(Path(sys.executable).resolve())
+    with open(executable,'rb') as f:
+        digest = hashlib.file_digest(f,'sha256').hexdigest()
+    assert muxmod.verified_original(executable,digest) == executable
+    import pytest
+    with pytest.raises(ValueError): muxmod.verified_original(executable,'0'*64)
+    with pytest.raises(ValueError): muxmod.verified_original('python.exe',digest)
+    # A value that happens to say app-server must not trigger observer/attachment.
+    marker = tmp_path/'original-ran'
+    code = 'from pathlib import Path; import sys; Path(sys.argv[1]).write_text("ok"); sys.exit(21)'
+    def forbidden(adapter):
+        raise AssertionError('not an app-server invocation')
+    assert muxmod.launch_original(executable,digest,['-c',code,str(marker),'app-server'],attach=forbidden) == 21
+    assert marker.read_text() == 'ok'
+