Codex

@@ -12,6 +12,7 @@ import sys import threading import time import signal+import tempfile  sys.path.insert(0, str(Path(__file__).resolve().parent / 'vendor')) @@ -119,6 +120,66 @@ def extension_version(binary):     return tuple(map(int, match[1].split('.'))) if match else None  +def protocol_fingerprint(binary, directory):+    """Cache the complete native schema by executable identity, not VSIX version."""+    before = identity(binary)+    key = hashlib.sha256(json.dumps(before).encode()).hexdigest()+    path = directory / 'protocols.json'+    try:+        cache = json.loads(path.read_text())+        if not isinstance(cache, dict): cache = {}+    except (OSError, ValueError):+        cache = {}+    if key in cache: return cache[key]+    with tempfile.TemporaryDirectory(prefix='codex-protocol-') as temporary:+        subprocess.run([str(binary), 'app-server', 'generate-json-schema',+                        '--experimental', '--out', temporary], check=True,+                       stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,+                       stderr=subprocess.PIPE, timeout=15)+        files = sorted(Path(temporary).rglob('*.json'))+        if not files: raise RuntimeError('Native app-server schema generation returned no schemas')+        digest = hashlib.sha256()+        for file in files:+            schema = json.loads(file.read_text())+            digest.update(json.dumps([str(file.relative_to(temporary)), schema],+                                     sort_keys=True, separators=(',', ':')).encode())+        if identity(binary) != before:+            raise RuntimeError('Codex binary changed while checking its protocol')+    cache[key] = digest.hexdigest()+    cache = dict(list(cache.items())[-8:])+    temporary = path.with_suffix('.tmp')+    temporary.write_text(json.dumps(cache) + '\n')+    temporary.replace(path)+    return cache[key]+++def compatible_backend(metadata, binary, directory):+    running = Path(metadata.get('binary', ''))+    if not extension_version(binary) or not extension_version(running): return False+    try:+        running_protocol = metadata.get('protocolFingerprint')+        if not running_protocol:+            # Legacy servers have no saved attestation. A replaced/deleted path+            # cannot prove which protocol their running image implements.+            if metadata.get('identity') != identity(running): return False+            running_protocol = protocol_fingerprint(running, directory)+        return running_protocol == protocol_fingerprint(binary, directory)+    except (OSError, ValueError, RuntimeError, subprocess.SubprocessError):+        return False+++def defer_upgrade(directory, metadata, binary, reason):+    record = {'state': 'deferred', 'reason': reason, 'protocolCompatible': True,+              'runningBinary': metadata['binary'], 'requestedBinary': str(binary),+              'pid': metadata['pid'], 'checkedAt': time.time(),+              'message': 'Connected to the existing compatible backend. Active work is preserved; '+                         'the next connection can upgrade after all views close and work is idle.'}+    path = directory / 'upgrade.json'+    temporary = path.with_suffix('.tmp')+    temporary.write_text(json.dumps(record, indent=2) + '\n')+    temporary.replace(path)++ def backend_idle(sock):     """Ask the backend; disk history can lag an active turn."""     with unix_connect(str(sock), compression=None, max_size=None, open_timeout=5,@@ -174,15 +235,30 @@ def ensure_server(binary, args, client_guard=None):             if requested_version and running_version and requested_version < running_version:                 raise RuntimeError('This editor uses an older Codex extension than the running backend. '                                    'Reload the editor to use the updated extension; backend downgrade refused.')-            # An extension upgrade may switch protocols. Drain the old version-            # only with no other proxy clients and an authoritatively idle server.+            # Matching generated schemas permit reconnecting without replacing+            # the writer. Different/unknown protocols retain the conservative gate.+            compatible = compatible_backend(metadata, binary, directory)             if client_guard is None:                 raise RuntimeError('Backend version differs; upgrade requires exclusive client ownership')             try: fcntl.flock(client_guard, fcntl.LOCK_EX | fcntl.LOCK_NB)             except BlockingIOError:+                if compatible:+                    # Failed lock conversion may drop our shared lock on Linux.+                    fcntl.flock(client_guard, fcntl.LOCK_SH)+                    defer_upgrade(directory, metadata, binary, 'connected_views')+                    return sock                 raise RuntimeError('Codex updated; the previous version still has connected views. '                                    'Its work is preserved. Close those views after work finishes, then retry.')-            if not backend_idle(sock):+            try:+                idle = backend_idle(sock)+            except Exception:+                if not compatible: raise+                idle = False+            if not idle:+                if compatible:+                    fcntl.flock(client_guard, fcntl.LOCK_SH)+                    defer_upgrade(directory, metadata, binary, 'backend_busy_or_unverified')+                    return sock                 raise RuntimeError('Codex updated while a turn is running. Its work is preserved; '                                    'retry when it finishes to activate the new version.')             if process_matches(metadata): os.kill(metadata['pid'], signal.SIGTERM)@@ -193,6 +269,10 @@ def ensure_server(binary, args, client_guard=None):         if socket_alive(sock):             raise RuntimeError('An unrecognized process owns the Codex runtime socket')         sock.unlink(missing_ok=True)+        protocol = None+        if extension_version(binary):+            try: protocol = protocol_fingerprint(binary, directory)+            except (OSError, ValueError, RuntimeError, subprocess.SubprocessError): pass         log = directory / 'server.log'         with log.open('ab') as output:             process = subprocess.Popen(@@ -201,6 +281,7 @@ def ensure_server(binary, args, client_guard=None):                 start_new_session=True, close_fds=True,             )         metadata = {'pid': process.pid, 'binary': str(binary), 'identity': identity(binary), 'args': args,+                    'protocolFingerprint': protocol,                     'socket': str(sock), 'started_at': time.time(),                     'start_ticks': Path(f'/proc/{process.pid}/stat').read_text()                         .rsplit(')', 1)[1].split()[19]}@@ -212,6 +293,7 @@ def ensure_server(binary, args, client_guard=None):             if process.poll() is not None:                 raise RuntimeError(f'Codex backend exited ({process.returncode}); inspect {log}')             if socket_alive(sock):+                (directory / 'upgrade.json').unlink(missing_ok=True)                 return sock             time.sleep(0.05)         raise RuntimeError(f'Codex backend is still starting; inspect {log}')@@ -250,7 +332,10 @@ def main():     if argv == ['--runtime-status']:         path = state_dir() / 'server.json'         metadata = json.loads(path.read_text()) if path.exists() else {}-        print(json.dumps({**metadata, 'alive': process_matches(metadata)}, indent=2))+        pending = path.with_name('upgrade.json')+        upgrade = json.loads(pending.read_text()) if pending.exists() else None+        if upgrade and upgrade.get('pid') != metadata.get('pid'): upgrade = None+        print(json.dumps({**metadata, 'alive': process_matches(metadata), 'upgrade': upgrade}, indent=2))         return     binary = native_binary()     args = server_args(argv)

@@ -73,6 +73,69 @@ class RoutingTests(unittest.TestCase):             with patch.dict(os.environ, {'ADOM_CODEX_RUNTIME_STATE': directory, 'CODEX_HOME': '/tmp/b'}):                 self.assertNotEqual(runtime.state_dir(), first) +    def test_compatible_upgrade_preserves_busy_backend_and_connected_views(self):+        for connected, idle in [(False, False), (True, True), (False, RuntimeError('probe failed'))]:+            with self.subTest(connected=connected, idle=idle), tempfile.TemporaryDirectory() as tmp:+                directory = Path(tmp)+                old = directory/'openai.chatgpt-1.0.0-linux-x64/codex'+                new = directory/'openai.chatgpt-1.0.1-linux-x64/codex'+                for binary in [old, new]:+                    binary.parent.mkdir(); binary.write_text('fixture')+                metadata = {'pid': 123, 'binary': str(old), 'identity': runtime.identity(old), 'args': ['app-server']}+                (directory/'server.json').write_text(json.dumps(metadata))+                with (directory/'clients.lock').open('a') as guard, (directory/'clients.lock').open('a') as other:+                    runtime.fcntl.flock(guard, runtime.fcntl.LOCK_SH)+                    if connected: runtime.fcntl.flock(other, runtime.fcntl.LOCK_SH)+                    with patch.object(runtime, 'state_dir', return_value=directory), \+                         patch.object(runtime, 'process_matches', return_value=True), \+                         patch.object(runtime, 'socket_alive', return_value=True), \+                         patch.object(runtime, 'protocol_fingerprint', return_value='same-schema'), \+                         patch.object(runtime, 'backend_idle', **({'side_effect': idle} if isinstance(idle, Exception) else {'return_value': idle})), \+                         patch.object(runtime.os, 'kill') as kill, \+                         patch.object(runtime.subprocess, 'Popen') as spawn:+                        self.assertEqual(runtime.ensure_server(new, ['app-server'], guard), directory/'server.sock')+                        kill.assert_not_called(); spawn.assert_not_called()+                    record = json.loads((directory/'upgrade.json').read_text())+                    self.assertEqual(record['pid'], 123)+                    self.assertTrue(record['protocolCompatible'])+                    runtime.fcntl.flock(other, runtime.fcntl.LOCK_UN)+                    with self.assertRaises(BlockingIOError):+                        runtime.fcntl.flock(other, runtime.fcntl.LOCK_EX | runtime.fcntl.LOCK_NB)++    def test_unknown_changed_or_different_protocol_does_not_attest_compatibility(self):+        with tempfile.TemporaryDirectory() as tmp:+            directory = Path(tmp)+            old = directory/'openai.chatgpt-1.0.0-linux-x64/codex'+            new = directory/'openai.chatgpt-1.0.1-linux-x64/codex'+            for binary in [old, new]:+                binary.parent.mkdir(); binary.write_text('fixture')+            metadata = {'binary': str(old), 'identity': runtime.identity(old)}+            with patch.object(runtime, 'protocol_fingerprint', side_effect=['old-schema', 'new-schema']):+                self.assertFalse(runtime.compatible_backend(metadata, new, directory))+            with patch.object(runtime, 'protocol_fingerprint', side_effect=RuntimeError('unsupported')):+                self.assertFalse(runtime.compatible_backend(metadata, new, directory))+            old.write_text('replacement')+            with patch.object(runtime, 'protocol_fingerprint') as fingerprint:+                self.assertFalse(runtime.compatible_backend(metadata, new, directory))+                fingerprint.assert_not_called()+            metadata['protocolFingerprint'] = 'attested-at-start'+            old.unlink()+            with patch.object(runtime, 'protocol_fingerprint', return_value='attested-at-start'):+                self.assertTrue(runtime.compatible_backend(metadata, new, directory))++    def test_schema_fingerprint_is_cached_by_binary_identity(self):+        with tempfile.TemporaryDirectory() as tmp:+            directory = Path(tmp); binary = directory/'codex'; binary.write_text('fixture')+            def generate(args, **kwargs):+                (Path(args[-1])/'schema.json').write_text('{"type":"object","properties":{}}')+            with patch.object(runtime.subprocess, 'run', side_effect=generate) as run:+                first = runtime.protocol_fingerprint(binary, directory)+                self.assertEqual(runtime.protocol_fingerprint(binary, directory), first)+                self.assertEqual(run.call_count, 1)+                binary.write_text('changed fixture')+                self.assertEqual(runtime.protocol_fingerprint(binary, directory), first)+                self.assertEqual(run.call_count, 2)+  class Client:     def __init__(self, env):@@ -160,6 +223,8 @@ class ReconnectTest(unittest.TestCase):                 'wire_api = "responses"\nrequires_openai_auth = false\n')             env = dict(os.environ, CODEX_HOME=str(home), ADOM_CODEX_RUNTIME_STATE=str(test_root / 'state'))             env.pop('OPENAI_API_KEY', None)+            if os.environ.get('ADOM_TEST_CODEX_OLD'):+                env['ADOM_CODEX_NATIVE'] = os.environ['ADOM_TEST_CODEX_OLD']             try:                 # Simultaneous clients must share one server, even before any                 # conversation exists. IDs are scoped to each connection.@@ -182,6 +247,20 @@ class ReconnectTest(unittest.TestCase):                     'input': [{'type': 'text', 'text': 'Local fixture', 'text_elements': []}]})['turn']                 self.assertTrue(started.wait(10), 'Fixture did not receive the model request')                 self.assertFalse(runtime.backend_idle(Path(json.loads(metadata_path.read_text())['socket'])))+                if os.environ.get('ADOM_TEST_CODEX_NEW'):+                    upgrade_env = dict(env, ADOM_CODEX_NATIVE=os.environ['ADOM_TEST_CODEX_NEW'])+                    newer = Client(upgrade_env)+                    clients.append(newer)+                    observed = newer.rpc(2, 'thread/read', {'threadId': thread_id, 'includeTurns': True})['thread']+                    self.assertEqual(observed['turns'][-1]['id'], turn['id'])+                    self.assertEqual(json.loads(metadata_path.read_text())['pid'], original_pid)+                    self.assertEqual(json.loads(metadata_path.with_name('upgrade.json').read_text())['reason'], 'connected_views')+                    newer.close(); first.close(); second.close()+                    env = upgrade_env+                    first = Client(env)+                    clients.append(first)+                    self.assertEqual(json.loads(metadata_path.read_text())['pid'], original_pid)+                    self.assertEqual(json.loads(metadata_path.with_name('upgrade.json').read_text())['reason'], 'backend_busy_or_unverified')                 second.close()                 for index in range(3):                     first.close(abrupt=index % 2 == 0)@@ -193,11 +272,11 @@ class ReconnectTest(unittest.TestCase):                     self.assertEqual(resumed['turns'][-1]['id'], turn['id'])                     self.assertEqual(resumed['turns'][-1]['status'], 'inProgress')                     self.assertEqual(json.loads(metadata_path.read_text())['pid'], original_pid)-                # A store update must not connect its new frontend to an old-                # backend, nor kill the active turn even with zero clients.+                # An unattested binary change must still fail closed, preserving+                # the active turn even when there are no proxy clients.                 first.close()                 original_metadata = json.loads(metadata_path.read_text())-                changed_metadata = {**original_metadata, 'identity': ['older-extension']}+                changed_metadata = {**original_metadata, 'identity': ['older-extension'], 'protocolFingerprint': None}                 metadata_path.write_text(json.dumps(changed_metadata))                 rejected = subprocess.run([sys.executable, str(ROOT/'runtime/runtime.py'),                     '-c', 'features.code_mode_host=true', 'app-server', '--analytics-default-enabled'],