diff --git a/docs/adr/0012-installation-runs-at-first-launch.md b/docs/adr/0012-installation-runs-at-first-launch.md index 09e64d8..9e81d0a 100644 --- a/docs/adr/0012-installation-runs-at-first-launch.md +++ b/docs/adr/0012-installation-runs-at-first-launch.md @@ -47,3 +47,12 @@ At the time of writing there is one item, `claude_hooks.py`. The Codex login `La **There is no uninstall in the menu bar**, because the panel is view-only in v1.0. Uninstall is `bridge-install uninstall`, a console script that lands in the bundle's `engine/bin/` beside `bridgectl` by the same pipeline rule that needs no list of script names. **A test may not reach the machine it is running on, and one line of design was not enough to hold that (#83).** Two drafts of the Codex item loaded a real login job into the author's own launchd — the second started the real shared Codex daemon — because `Launchd` and `home` had defaults that resolved to the real ones and a test that omitted them passed. Defaults were removed and `bridge-install`'s test-only keyword arguments grew to five, but the guard that actually holds is in `tests/conftest.py`: the real `launchctl` runner is taken away from the whole suite, so a test that forgets fails loudly instead of silently changing the machine. Every item added to this boundary that runs a subprocess belongs behind that same fixture. + + +## Upgrade exception — #383 (2026-09-22) + +The app-owned upgrade module may stop the identified shared login job after +validating a user-installed newer candidate. Reconcile and uninstall keep their +non-stopping behavior, and installation still imports no adapter or Bridge Core. +The direct-switch, recovery, and storage rules live in +[ADR 0022's upgrade amendment](0022-the-shared-app-server-is-the-users-own-codex.md#upgrade-exception--383-2026-09-22). diff --git a/docs/adr/0022-the-shared-app-server-is-the-users-own-codex.md b/docs/adr/0022-the-shared-app-server-is-the-users-own-codex.md index bb27854..51df959 100644 --- a/docs/adr/0022-the-shared-app-server-is-the-users-own-codex.md +++ b/docs/adr/0022-the-shared-app-server-is-the-users-own-codex.md @@ -117,3 +117,33 @@ Stated rather than implied, because this ADR is written on measurements and thes 3. **Homebrew.** The `PATH` rule is argued for a native binary, not measured against one. And one for whoever re-runs the acceptance harness's codex lane: **`approvalPolicy: "on-request"` with a `readOnly` sandbox no longer forces an approval request** on 0.153.4. #82's gate prompt produced none twice (90 s and 120 s waits) — the model answered in prose and never called the shell. `"untrusted"` produces the request immediately. A gate written on the old assumption **hangs rather than fails**. + + +## Upgrade exception — #383 (2026-09-22) + +The app checks for a user-installed Codex upgrade at startup and on filesystem +notifications. It validates the candidate on an isolated socket, then directly +switches only its identified shared login job. Active turns may be interrupted; +neither task completion nor approval/input waits delay the switch. The socket +peer must belong to the launchd job's process group. That group is stopped, +launchd's old definition is removed, and the group must be gone before loading +the replacement. `bootout` returning zero alone does not prove process exit. + +The app-owned update module composes installation and existing Codex discovery; +installation still imports no adapter or Bridge Core. Reconcile and uninstall +retain their previous non-stopping behavior. The updater reuses terminal and +VoiceCoding reconnect, reads the original live Session identities back, and +never resumes historical Sessions or replays commands. Connection recovery does +not mean an interrupted task continued. Closed terminals cease to be required. + +Using one executable path does not imply one running version: npm replaces +files underneath a still-running process. The server's initialize userAgent +contains its build version (the originator prefix can vary by client). The +update check compares that version with the runnable candidate, without adding +a version pin to the agent adapter. + +Only an old executable that remains present and passes validation can be used +for recovery. No version is copied, cached, or downloaded. Failed attempts are +remembered only for the app's lifetime to prevent repeated interruption from +duplicate file events. Shutdown stops listening and discards queued checks; +an already-started bounded switch finishes before the app exits. diff --git a/pyproject.toml b/pyproject.toml index 1a5319c..701a977 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "gpt-voicecoding" -version = "1.0.5" +version = "1.0.6" description = "Voice-controlling terminal coding agents through a realtime voice call." readme = "README.md" license = { file = "LICENSE" } diff --git a/shell/Sources/GPTVoiceCodingShell/ShellModel.swift b/shell/Sources/GPTVoiceCodingShell/ShellModel.swift index 5394b87..eabdea0 100644 --- a/shell/Sources/GPTVoiceCodingShell/ShellModel.swift +++ b/shell/Sources/GPTVoiceCodingShell/ShellModel.swift @@ -240,6 +240,8 @@ final class ShellModel { private var telegramRequest: Task? private var telegramCancellation: Task? private let runCommand: @Sendable (EngineCommand) async -> InstallationReport + private let runCodexUpdate: CodexUpdate.Check + private var codexUpdate: CodexUpdate? private(set) var codexCheck: CodexCheck = .checking private(set) var codexCheckFailure: String? private(set) var installationReport: InstallationReport? @@ -378,6 +380,17 @@ final class ShellModel { runCommand ?? { command in await InstallationRunner(report: { pathOutcomes.record($0) }).run(command) } + self.runCodexUpdate = { attempted in + do { + let command = try EngineCommand.resolveCodexUpdate( + resources: Bundle.main.resourceURL, attempted: attempted) + if let runCommand { return await runCommand(command) } + return await InstallationRunner(report: { pathOutcomes.record($0) }).run( + command, deadline: CodexUpdate.deadline, separateOutput: true) + } catch { + return InstallationReport(ok: false, lines: ["\(error)"]) + } + } self.wallNow = wallNow self.messageNow = wallNow() self.now = now @@ -507,6 +520,13 @@ final class ShellModel { // launchd hands an app opened from Finder (#272, ADR 0022). installationReport = await runCommand(command) installationFailure = installationReport?.failure + guard !Task.isCancelled && !stopping else { return } + if codexUpdate == nil { + codexUpdate = CodexUpdate( + check: runCodexUpdate, + report: { [weak self] failure in self?.installationFailure = failure }) + } + codexUpdate?.checkForUpdate() } func checkCodex() async { @@ -883,6 +903,7 @@ final class ShellModel { preparation?.cancel() credentialStartRecovery.cancel() stopCredentialObservation() + await codexUpdate?.stop() await supervisor.shutDown() } diff --git a/shell/Sources/ShellCore/CodexUpdate.swift b/shell/Sources/ShellCore/CodexUpdate.swift new file mode 100644 index 0000000..0e0aa07 --- /dev/null +++ b/shell/Sources/ShellCore/CodexUpdate.swift @@ -0,0 +1,179 @@ +import CoreServices +import Foundation + +/// The app owns this event source and one serialized check. Python owns all +/// update decisions; file events only request another reading of the facts. +@MainActor +public final class CodexUpdate { + public static let deadline = 4 * Installation.deadline + public typealias Check = @Sendable (String) async -> InstallationReport + + private struct Result: Decodable { + var status: String + var failure: String + var attempted: String + var watchPaths: [String] + enum CodingKeys: String, CodingKey { + case status, failure, attempted + case watchPaths = "watch_paths" + } + } + + private let check: Check + private let report: (String?) -> Void + private var observer: InstallationChanges? + private var work: Task? + private var pending = false + private var stopped = false + private var attempted = "" + private var lastFailure: String? + private var watchedPaths: [String] = [] + + public init(check: @escaping Check, report: @escaping (String?) -> Void) { + self.check = check + self.report = report + } + + /// Startup and filesystem notifications enter through the same method. + public func checkForUpdate() { + guard !stopped else { return } + pending = true + guard work == nil else { return } + work = Task { [weak self] in + guard let self else { return } + while pending && !stopped { + pending = false + let response = await check(attempted) + guard !stopped else { break } + do { + let output = (response.standardOutput ?? response.lines).joined(separator: "\n") + let result = try JSONDecoder().decode(Result.self, from: Data(output.utf8)) + attempted = result.attempted + if watchedPaths != result.watchPaths { + watchedPaths = result.watchPaths + try watch() + // Close the startup/retarget registration gap once. This + // is event reconciliation, not a periodic timer. + pending = true + } + if result.status == "failed" { + showFailure(result.failure) + } else if result.status == "updated" { + showFailure(nil) + } + } catch { + showFailure(response.failure ?? "Codex update check failed: \(error)") + } + } + work = nil + } + } + + /// Cancel queued checks and listening, but finish an already-started switch + /// before the app exits. Its runner and Python operations both have bounds. + public func stop() async { + stopped = true + pending = false + observer?.cancel() + observer = nil + await work?.value + work = nil + } + + private func watch() throws { + observer?.cancel() + observer = try InstallationChanges(paths: watchedPaths) { [weak self] in + Task { @MainActor in + guard let self, !self.stopped else { return } + do { try self.watch() } catch { self.showFailure("Codex watch failed: \(error)") } + self.checkForUpdate() + } + } + } + + private func showFailure(_ failure: String?) { + guard failure != lastFailure else { return } + lastFailure = failure + report(failure) + } +} + +/// FSEvents follows package contents as well as directory replacement. A watch +/// on only the npm shim misses a replaced native dependency below the package. +private final class InstallationChanges: @unchecked Sendable { + private final class Callback: @unchecked Sendable { + let paths: [String] + let changed: @Sendable () -> Void + + init(paths: [String], changed: @escaping @Sendable () -> Void) { + self.paths = paths + self.changed = changed + } + + func receive(_ events: [String]) { + let normalized = events.flatMap { event in + [event, URL(fileURLWithPath: event).resolvingSymlinksInPath().path] + } + if normalized.contains(where: { event in + paths.contains { path in + event == path || event.hasPrefix(path + "/") || path.hasPrefix(event + "/") + } + }) { + changed() + } + } + } + + private let callback: Callback + private var stream: FSEventStreamRef? + + init(paths: [String], changed: @escaping @Sendable () -> Void) throws { + let normalized = Array( + Set( + paths.flatMap { path in + let url = URL(fileURLWithPath: path).standardizedFileURL + return [url.path, url.resolvingSymlinksInPath().path] + })) + callback = Callback(paths: normalized, changed: changed) + guard !paths.isEmpty else { return } + let roots = Set( + normalized.map { path in + var parent = URL(fileURLWithPath: path).deletingLastPathComponent() + while !FileManager.default.fileExists(atPath: parent.path) && parent.path != "/" { + parent.deleteLastPathComponent() + } + return parent.path + }) + var context = FSEventStreamContext( + version: 0, info: Unmanaged.passUnretained(callback).toOpaque(), + retain: nil, release: nil, copyDescription: nil) + stream = FSEventStreamCreate( + nil, + { _, info, _, rawPaths, _, _ in + guard let info else { return } + let callback = Unmanaged.fromOpaque(info).takeUnretainedValue() + let events = unsafeBitCast(rawPaths, to: NSArray.self) as? [String] ?? [] + callback.receive(events) + }, &context, Array(roots) as CFArray, + FSEventStreamEventId(kFSEventStreamEventIdSinceNow), 0.2, + FSEventStreamCreateFlags( + kFSEventStreamCreateFlagUseCFTypes | kFSEventStreamCreateFlagFileEvents + | kFSEventStreamCreateFlagWatchRoot)) + guard let stream else { throw CocoaError(.fileReadUnknown) } + FSEventStreamSetDispatchQueue(stream, .main) + guard FSEventStreamStart(stream) else { + cancel() + throw CocoaError(.fileReadUnknown) + } + } + + func cancel() { + guard let stream else { return } + FSEventStreamStop(stream) + FSEventStreamInvalidate(stream) + FSEventStreamRelease(stream) + self.stream = nil + } + + deinit { cancel() } +} diff --git a/shell/Sources/ShellCore/EngineCommand.swift b/shell/Sources/ShellCore/EngineCommand.swift index 104eb3a..96c52d7 100644 --- a/shell/Sources/ShellCore/EngineCommand.swift +++ b/shell/Sources/ShellCore/EngineCommand.swift @@ -80,6 +80,16 @@ public struct EngineCommand: Equatable, Sendable { searchPath: searchPath) } + public static func resolveCodexUpdate( + resources: URL?, attempted: String, + environment: [String: String] = ProcessInfo.processInfo.environment + ) throws -> EngineCommand { + try resolve( + resources: resources, + arguments: ["-m", "gpt_voicecoding.codex_update", "--attempted", attempted], + environment: environment, searchPath: nil) + } + private static func resolve( resources: URL?, arguments: [String], diff --git a/shell/Tests/ShellCoreTests/CodexUpdateTests.swift b/shell/Tests/ShellCoreTests/CodexUpdateTests.swift new file mode 100644 index 0000000..cc7b11a --- /dev/null +++ b/shell/Tests/ShellCoreTests/CodexUpdateTests.swift @@ -0,0 +1,98 @@ +import Foundation +import Testing + +@testable import ShellCore + +@Suite(.serialized) struct CodexUpdateTests { + actor Checks { + var calls = 0 + var active = 0 + var peak = 0 + var release: CheckedContinuation? + + func run(_ attempted: String) async -> InstallationReport { + calls += 1 + active += 1 + peak = max(peak, active) + if calls == 1 { await withCheckedContinuation { release = $0 } } + active -= 1 + return InstallationReport( + ok: true, + lines: [ + #"{"status":"unchanged","failure":"","attempted":"","watch_paths":[]}"# + ]) + } + + func unblock() { + release?.resume() + release = nil + } + } + + @Test @MainActor func repeatedEventsSerializeAndQuitDiscardsQueuedCheck() async { + let checks = Checks() + let updater = CodexUpdate(check: { await checks.run($0) }, report: { _ in }) + updater.checkForUpdate() + while await checks.calls == 0 { await Task.yield() } + updater.checkForUpdate() + updater.checkForUpdate() + let stopping = Task { await updater.stop() } + await Task.yield() + await checks.unblock() + await stopping.value + updater.checkForUpdate() + #expect(await checks.calls == 1) + #expect(await checks.peak == 1) + } + + actor FileChecks { + let paths: [String] + var calls = 0 + init(_ paths: [String]) { self.paths = paths } + func run() -> InstallationReport { + calls += 1 + let json = try! JSONSerialization.data(withJSONObject: [ + "status": "unchanged", "failure": "", "attempted": "", "watch_paths": paths, + ]) + return InstallationReport(ok: true, lines: [String(decoding: json, as: UTF8.self)]) + } + } + + @Test @MainActor func fileWritesAndReplacedDirectoriesKeepTriggeringChecks() async throws { + let root = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString) + let package = root.appendingPathComponent("package") + let file = package.appendingPathComponent("codex") + try FileManager.default.createDirectory(at: package, withIntermediateDirectories: true) + try Data("first".utf8).write(to: file) + defer { try? FileManager.default.removeItem(at: root) } + let checks = FileChecks([package.path]) + let updater = CodexUpdate(check: { _ in await checks.run() }, report: { _ in }) + updater.checkForUpdate() + try await wait(checks, after: 1) + var previous = await checks.calls + try Data("second".utf8).write(to: file) + try await wait(checks, after: previous) + previous = await checks.calls + try FileManager.default.removeItem(at: package) + try FileManager.default.createDirectory(at: package, withIntermediateDirectories: true) + try Data("third".utf8).write(to: file) + try await wait(checks, after: previous) + previous = await checks.calls + try Data("fourth".utf8).write(to: file) + try await wait(checks, after: previous) + await updater.stop() + previous = await checks.calls + try Data("after stop".utf8).write(to: file) + try await Task.sleep(for: .milliseconds(350)) + #expect(await checks.calls == previous) + } + + private func wait(_ checks: FileChecks, after previous: Int) async throws { + let deadline = ContinuousClock.now + .seconds(5) + while await checks.calls <= previous && ContinuousClock.now < deadline { + try await Task.sleep(for: .milliseconds(20)) + } + #expect(await checks.calls > previous) + } + +} diff --git a/src/gpt_voicecoding/codex_update.py b/src/gpt_voicecoding/codex_update.py new file mode 100644 index 0000000..1f7d420 --- /dev/null +++ b/src/gpt_voicecoding/codex_update.py @@ -0,0 +1,445 @@ +"""App-owned Codex upgrade checks; installation never imports an agent adapter. + +The shell calls ``check`` at launch and on installation changes. This module +owns the decision and reports the result; it never submits or resumes a turn. +""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import os +import plistlib +import re +import signal +import socket +import tempfile +from contextlib import suppress +from dataclasses import asdict, dataclass, replace +from pathlib import Path + +from gpt_voicecoding import __version__ +from gpt_voicecoding.adapters.agent.codex import discovery +from gpt_voicecoding.adapters.agent.codex.processes import enumerate_runs +from gpt_voicecoding.adapters.codex_app_server.process import initialise +from gpt_voicecoding.adapters.codex_app_server.wire import AppServerConnection, WireError +from gpt_voicecoding.installation import codex_launch_agent as agent +from gpt_voicecoding.installation import codex_runtime, read_intent +from gpt_voicecoding.installation.codex_runtime import ( + HANDSHAKE_TIMEOUT_SECONDS, + CodexRuntime, + Resolution, +) +from gpt_voicecoding.locations import codex_daemon_log_path, installation_path + + +@dataclass(frozen=True) +class UpdateResult: + status: str + failure: str = "" + attempted: str = "" + watch_paths: tuple[str, ...] = () + + +class CodexUpdate: + def __init__( + self, + *, + runtime: CodexRuntime, + launchd: agent.Launchd | None = None, + directory: Path | None = None, + record_path: Path | None = None, + log_path: Path | None = None, + list_sessions=enumerate_runs, + process_identity=None, + attempted: str = "", + recovery_timeout: float = agent.SHELL_RECONCILE_DEADLINE_SECONDS, + ) -> None: + self.runtime = runtime + self.launchd = launchd or agent.default_launchd() + self.directory = directory or agent.default_launch_agents_directory() + self.record_path = record_path or installation_path() + self.log_path = log_path or codex_daemon_log_path() + self.list_sessions = list_sessions + self.process_identity = process_identity or _process_identity + self.recovery_timeout = recovery_timeout + self._lock = asyncio.Lock() + self.attempted = attempted + + async def check(self) -> UpdateResult: + async with self._lock: + paths = _watch_paths(self.runtime) + try: + async with asyncio.timeout(3 * agent.SHELL_RECONCILE_DEADLINE_SECONDS): + result = await self._check() + except (OSError, ValueError, TimeoutError, WireError) as error: + result = UpdateResult( + "failed", f"Codex update failed: {error or 'operation timed out'}" + ) + return replace(result, attempted=self.attempted, watch_paths=paths) + + async def _check(self) -> UpdateResult: + try: + candidate = await _version(self.runtime) + except (OSError, ValueError, TimeoutError) as error: + return UpdateResult("failed", f"Codex installation is not ready: {error}") + client = AppServerConnection(self.runtime.control_socket) + try: + async with asyncio.timeout(HANDSHAKE_TIMEOUT_SECONDS): + await client.connect() + answer = await initialise(client, experimental=False, version=__version__) + running = _server_version(answer) + if running == candidate: + return UpdateResult("unchanged") + fingerprint = _fingerprint(self.runtime, candidate) + if self.attempted == fingerprint: + return UpdateResult("unchanged") + try: + await _probe(self.runtime, candidate) + except (OSError, ValueError, TimeoutError, WireError) as error: + return UpdateResult("failed", f"Codex candidate is not ready: {error}") + held = await self._owned_job() + before = await self._sessions(client) + # Validation may have overlapped another package replacement. + if ( + await _version(self.runtime) != candidate + or _fingerprint(self.runtime, candidate) != fingerprint + ): + raise ValueError("Codex installation changed during validation") + self.attempted = fingerprint + try: + await self._switch(held, self.runtime) + await client.aclose() + await self._recovered(candidate, before) + return UpdateResult("updated") + except (OSError, ValueError, TimeoutError, WireError) as error: + recovery = await self._restore(held, running, before) + return UpdateResult( + "failed", f"Codex update failed: {error or 'recovery timed out'}. {recovery}" + ) + except (OSError, ValueError, TimeoutError, WireError) as error: + return UpdateResult("failed", f"Codex update failed: {error}") + finally: + await client.aclose() + + async def _restore(self, previous: agent.HeldJob, version: str, before) -> str: + client = AppServerConnection(self.runtime.control_socket) + try: + async with asyncio.timeout(HANDSHAKE_TIMEOUT_SECONDS): + await client.connect() + await initialise(client, experimental=False, version=__version__) + return "The server responds, but Session recovery was not confirmed." + except (OSError, WireError, TimeoutError): + pass + finally: + await client.aclose() + old = replace(self.runtime, executable=Path(previous.program)) + try: + if await _version(old) != version: + return "The old executable is no longer available; no rollback was attempted." + await _probe(old, version) + # A failed new job may still be loaded, but an unresponsive process + # cannot prove socket ownership: refuse to stop it on a guess. + held = await asyncio.to_thread(self.launchd.held_job) + if held is not None: + if ( + held.program != str(self.runtime.executable) + or held.arguments != (held.program, *self.runtime.server_arguments) + or held.login != previous.login + ): + return "The login job identity changed; recovery was refused." + if held.pid is not None and await _group_alive(held.pid): + return "The new job is still present; automatic recovery was refused." + code, reason = await asyncio.to_thread( + self.launchd.ask, + [str(agent.LAUNCHCTL), "bootout", f"{self.launchd.domain}/{agent.LABEL}"], + ) + if code: + return f"The failed job could not be removed: {reason}" + outcome = await asyncio.to_thread( + agent.install, + self.directory, + Resolution(runtime=old, reason=""), + self.log_path, + self.record_path, + self.launchd, + ) + if not outcome.ok: + return f"Old-version recovery failed: {outcome.note}" + await self._recovered(version, before, executable=old.executable) + return "The existing old version was restored." + except (OSError, ValueError, WireError, TimeoutError) as error: + return f"No runnable old version could be restored: {error}" + + async def _owned_job(self, executable: Path | None = None) -> agent.HeldJob: + held = await asyncio.to_thread(self.launchd.held_job) + document = plistlib.loads(agent.plist_path(self.directory).read_bytes()) + if document.get("Label") != agent.LABEL or document.get("ProgramArguments") != [ + str(executable or self.runtime.executable), + *self.runtime.server_arguments, + ]: + raise ValueError("the shared login job file is not the expected Codex installation") + if ( + held is None + or held.pid is None + or held.login is None + or held.arguments != (held.program, *self.runtime.server_arguments) + ): + raise ValueError("the shared server is not an identified VoiceCoding login job") + # Darwin sys/un.h: SOL_LOCAL=0, LOCAL_PEERPID=2. This asks the kernel + # which process owns the connected endpoint, not which file was written. + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as peer: + peer.settimeout(HANDSHAKE_TIMEOUT_SECONDS) + peer.connect(str(self.runtime.control_socket)) + peer_pid = peer.getsockopt(0, 2) + if os.getpgid(peer_pid) != held.pid or os.getpgid(held.pid) != held.pid: + raise ValueError("the socket peer does not belong to the identified login job") + return held + + async def _switch(self, held: agent.HeldJob, runtime: CodexRuntime) -> None: + if await self._owned_job() != held: + raise ValueError("the login job changed before the switch") + # bootout returns before all processes exit; a single SIGTERM can wait + # on a business turn. The kernel-proven job group is the interruption + # boundary, including the npm shim's native child, never a name match. + code, reason = await asyncio.to_thread( + self.launchd.ask, ["/bin/kill", "-KILL", "--", f"-{held.pid}"] + ) + if code: + raise ValueError(f"the identified process group could not be stopped: {reason}") + code, reason = await asyncio.to_thread( + self.launchd.ask, + [str(agent.LAUNCHCTL), "bootout", f"{self.launchd.domain}/{agent.LABEL}"], + ) + if code: + raise ValueError(f"launchd did not remove the old job: {reason}") + # Removal is asynchronous even after bootout returns. Wait for the + # definition and its group to disappear, not for business work to end. + async with asyncio.timeout(agent.COMMAND_TIMEOUT_SECONDS): + while await asyncio.to_thread(self.launchd.held_job) is not None or await _group_alive( + held.pid + ): + await asyncio.sleep(0.05) + outcome = await asyncio.to_thread( + agent.install, + self.directory, + Resolution(runtime=runtime, reason=""), + self.log_path, + self.record_path, + self.launchd, + ) + if not outcome.ok: + raise ValueError(outcome.note) + + async def _sessions(self, client: AppServerConnection): + candidates = await self.list_sessions() + + async def listed(): + return candidates + + report = await discovery.discover( + client, + evidence=discovery.ProcessEvidence(list_sessions=listed, home=self.runtime.codex_home), + ) + if report.error: + raise ValueError(report.error) + loaded = await client.request("thread/loaded/list", {}) + if not isinstance(loaded.get("data"), list): + raise ValueError("the server did not return its loaded threads") + found = {} + for row in report.rows: + if ( + row.target.pid is not None + and row.child.is_main + and row.has_controlling_terminal is True + and row.target.session_id in loaded["data"] + ): + identity = await self.process_identity(row.target.pid) + if identity is not None: + found[row.target] = identity + return found + + async def _recovered(self, expected: str, before, *, executable: Path | None = None) -> None: + async with asyncio.timeout(self.recovery_timeout): + while True: + client = AppServerConnection(self.runtime.control_socket) + try: + await client.connect() + answer = await initialise(client, experimental=False, version=__version__) + if _server_version(answer) != expected: + raise ValueError("the shared server is not running the validated version") + await self._owned_job(executable) + remaining = set() + for target, original in before.items(): + # Discovery may omit a living terminal when lsof fails. + # Only a direct process identity reading proves exit or + # PID reuse; an elapsed-time estimate cannot do that. + identity = await self.process_identity(target.pid) + if identity == original: + remaining.add(target) + recovered = await self._sessions(client) + if remaining.issubset(recovered): + return + except (WireError, OSError): + pass + finally: + await client.aclose() + await asyncio.sleep(0.05) + + +async def _process_identity(pid: int) -> str | None: + return await _ps(["-p", str(pid), "-o", "lstart="], missing_ok=True) + + +async def _group_alive(pid: int | None) -> bool: + if pid is None: + raise ValueError("the old process group was not identified") + output = await _ps(["-axo", "pgid=,stat="]) + return any( + fields[0] == str(pid) and not fields[1].startswith("Z") + for line in output.splitlines() + if len(fields := line.split()) == 2 + ) + + +async def _ps(arguments: list[str], *, missing_ok: bool = False) -> str | None: + process = await asyncio.create_subprocess_exec( + "/bin/ps", + *arguments, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + output, error = await asyncio.wait_for(process.communicate(), HANDSHAKE_TIMEOUT_SECONDS) + finally: + if process.returncode is None: + process.kill() + await process.wait() + if missing_ok and process.returncode == 1 and not output.strip() and not error.strip(): + return None + if process.returncode or not output.strip(): + raise OSError("could not read process identity: " + error.decode("utf-8", "replace")) + return output.decode().strip() + + +def _server_version(answer: dict) -> str: + user_agent = answer.get("userAgent") + match = re.match(r"[^/]+/(\S+) \(", user_agent) if isinstance(user_agent, str) else None + if match is None: + raise ValueError("the server did not identify its running version") + return match[1] + + +async def _version(runtime: CodexRuntime) -> str: + process = await asyncio.create_subprocess_exec( + str(runtime.executable), + "--version", + env={**os.environ, **runtime.launch_environment}, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + start_new_session=True, + ) + try: + output, error = await asyncio.wait_for(process.communicate(), HANDSHAKE_TIMEOUT_SECONDS) + finally: + if process.returncode is None: + os.killpg(process.pid, signal.SIGKILL) + await process.wait() + match = re.fullmatch(r"codex-cli (\S+)\s*", output.decode("utf-8", "replace")) + if process.returncode or match is None: + raise ValueError(error.decode("utf-8", "replace")[:200] or "unreadable Codex version") + return match[1] + + +async def _probe(runtime: CodexRuntime, expected: str) -> None: + # No credentials, user config, model calls, or persistent copies. The short + # socket root is required by Darwin's sockaddr_un limit. + with tempfile.TemporaryDirectory(prefix="vc-codex-", dir="/tmp") as directory: + root = Path(directory) + socket_path = root / "probe.sock" + environment = {"HOME": directory, "CODEX_HOME": directory, "PATH": runtime.path} + process = await asyncio.create_subprocess_exec( + str(runtime.executable), + "app-server", + "--listen", + f"unix://{socket_path}", + env=environment, + cwd=directory, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + start_new_session=True, + ) + try: + async with asyncio.timeout(HANDSHAKE_TIMEOUT_SECONDS): + while True: + if process.returncode is not None: + raise ValueError(f"candidate app-server exited with {process.returncode}") + client = AppServerConnection(socket_path) + try: + await client.connect() + answer = await initialise(client, experimental=False, version=__version__) + if _server_version(answer) != expected: + raise ValueError("candidate app-server version disagrees with CLI") + return + except WireError: + await asyncio.sleep(0.05) + finally: + await client.aclose() + finally: + with suppress(ProcessLookupError): + os.killpg(process.pid, signal.SIGKILL) + await process.wait() + + +def _watch_paths(runtime: CodexRuntime) -> tuple[str, ...]: + paths = {runtime.executable.absolute(), runtime.executable.resolve()} + # The npm shim lives under the package whose dependencies it starts. Watch + # that package, not only the symlink or shim, which can remain unchanged. + for parent in runtime.executable.resolve().parents: + if (parent / "package.json").is_file(): + paths.add(parent) + break + return tuple(sorted(map(str, paths))) + + +def _fingerprint(runtime: CodexRuntime, version: str) -> str: + digest = hashlib.sha256(version.encode()) + for named in _watch_paths(runtime): + path = Path(named) + entries = [path, *path.rglob("*")] if path.is_dir() else [path] + for entry in sorted(entries): + stat = entry.lstat() + digest.update(f"{entry}:{stat.st_ino}:{stat.st_size}:{stat.st_mtime_ns}".encode()) + return digest.hexdigest() + + +def main() -> int: + parser = argparse.ArgumentParser(description="Check and follow a user's Codex upgrade.") + parser.add_argument("--attempted", default="") + args = parser.parse_args() + directory = agent.default_launch_agents_directory() + resolved = codex_runtime.resolve(os.environ, recorded_path=agent.recorded_path(directory)) + if not read_intent().install_wanted: + result = UpdateResult("unchanged") + elif resolved.runtime is None: + path = codex_runtime.machine_path(os.environ, agent.recorded_path(directory)) or "" + result = UpdateResult( + "unchanged", + watch_paths=tuple( + str(Path(part) / codex_runtime.EXECUTABLE_NAME) + for part in path.split(os.pathsep) + if part + ), + ) + else: + result = asyncio.run( + CodexUpdate(runtime=resolved.runtime, attempted=args.attempted).check() + ) + print(json.dumps(asdict(result))) + return 0 if result.status != "failed" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/gpt_voicecoding/installation/codex_launch_agent.py b/src/gpt_voicecoding/installation/codex_launch_agent.py index 12a7140..6f3546b 100644 --- a/src/gpt_voicecoding/installation/codex_launch_agent.py +++ b/src/gpt_voicecoding/installation/codex_launch_agent.py @@ -245,6 +245,8 @@ class HeldJob: program: str login_asid: int | None boot_session: str | None + pid: int | None = None + arguments: tuple[str, ...] = () @property def login(self) -> tuple[int, str] | None: @@ -313,10 +315,16 @@ def held_job(self) -> HeldJob | None: return None found_program = re.search(r"^\s*program\s*=\s*(.+?)\s*$", said, re.MULTILINE) found_asid = re.search(r"^\s*asid\s*=\s*(\d+)\s*$", said, re.MULTILINE) + found_pid = re.search(r"^\s*pid\s*=\s*(\d+)\s*$", said, re.MULTILINE) + arguments = re.search(r"\n\s*arguments = \{\n(.*?)\n\s*\}", said, re.DOTALL) return HeldJob( program=found_program.group(1) if found_program else "", login_asid=int(found_asid.group(1)) if found_asid else None, boot_session=self.ask_boot_session(), + pid=int(found_pid[1]) if found_pid else None, + arguments=tuple(line.strip() for line in arguments[1].splitlines()) + if arguments + else (), ) def bootstrap(self, path: Path) -> tuple[HeldJob | None, str]: diff --git a/tests/codex_update_fake.py b/tests/codex_update_fake.py new file mode 100644 index 0000000..fecfac8 --- /dev/null +++ b/tests/codex_update_fake.py @@ -0,0 +1,94 @@ +"""An executable protocol peer and an isolated launchd command boundary.""" + +import os +import plistlib +import signal +import subprocess +import sys +from pathlib import Path + +from gpt_voicecoding.installation import codex_launch_agent as agent + + +class UpdateMachine: + def __init__(self, root: Path) -> None: + self.root = root + self.executable = root / "codex" + self.version = root / "version" + self.version.write_text("1.2.3") + self.threads = root / "threads.json" + self.threads.write_text("[]") + self.refuse_start = False + self.removing = 0 + self.removed_output = "" + import_paths = [str(Path(__file__).parent), str(Path(__file__).parents[1] / "src")] + self.executable.write_text( + f"#!{sys.executable}\n" + "import asyncio, sys, json\nfrom pathlib import Path\n" + f"sys.path[:0] = {import_paths!r}\n" + "from codex_fake import FakeAppServer\n" + f"version = Path({str(self.version)!r}).read_text()\n" + "if '--version' in sys.argv:\n print('codex-cli '+version); sys.exit(0)\n" + "async def main():\n" + " path = Path(sys.argv[sys.argv.index('--listen')+1].removeprefix('unix://'))\n" + " path.parent.mkdir(parents=True, exist_ok=True)\n" + f" threads = json.loads(Path({str(self.threads)!r}).read_text())\n" + " async with FakeAppServer(path) as server:\n" + " server.answers('initialize', {'userAgent':'gpt-voicecoding/'+version+' (test)'})\n" + " server.answers('thread/loaded/list', {'data': [t['id'] for t in threads]})\n" + " server.answers('thread/read', lambda p: {'thread': " + "next(t for t in threads if t['id']==p['threadId'])})\n" + " await asyncio.Event().wait()\n" + "asyncio.run(main())\n" + ) + self.executable.chmod(0o700) + self.process: subprocess.Popen | None = None + self.arguments: list[str] = [] + self.commands: list[str] = [] + self.launchd = agent.Launchd("gui/test", self.run, lambda: "test-boot") + + def run(self, argv): + if argv[0] == "/bin/kill": + assert self.process is not None and argv[-1] == f"-{self.process.pid}" + os.killpg(self.process.pid, signal.SIGKILL) + self.process.wait(timeout=5) + return 0, "" + verb = argv[1] + self.commands.append(verb) + if verb == "print": + if self.removing: + self.removing -= 1 + return 0, self.removed_output + if self.process is None: + return 1, "not loaded" + return 0, ( + f"program = {self.arguments[0]}\nasid = 1\npid = {self.process.pid}\n" + "arguments = {\n" + "\n".join(self.arguments) + "\n}\n" + ) + if verb == "bootout": + _, self.removed_output = self.run(["launchctl", "print"]) + self.close() + self.removing = 2 + return 0, "" + if verb == "bootstrap": + if self.refuse_start: + return 1, "candidate refused to start" + document = plistlib.loads(Path(argv[-1]).read_bytes()) + self.arguments = document["ProgramArguments"] + Path(self.arguments[-1].removeprefix("unix://")).unlink(missing_ok=True) + self.process = subprocess.Popen( + self.arguments, + env={**os.environ, **document["EnvironmentVariables"]}, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + return 0, "" + raise AssertionError(argv) + + def close(self): + if self.process is not None: + if self.process.poll() is None: + os.killpg(self.process.pid, signal.SIGKILL) + self.process.wait(timeout=5) + self.process = None diff --git a/tests/test_codex_update.py b/tests/test_codex_update.py new file mode 100644 index 0000000..be47084 --- /dev/null +++ b/tests/test_codex_update.py @@ -0,0 +1,293 @@ +"""The app's single update entry point, with isolated files and protocol peers.""" + +import asyncio +import json +import os +from pathlib import Path + +import pytest + +from codex_fake import FakeAppServer +from codex_update_fake import UpdateMachine +from gpt_voicecoding.adapters.agent.codex.processes import Candidate +from gpt_voicecoding.codex_update import CodexUpdate +from gpt_voicecoding.installation import codex_launch_agent as agent +from gpt_voicecoding.installation.codex_runtime import CodexRuntime + + +def test_matching_version_does_not_restart(socket_root: Path) -> None: + async def scenario() -> None: + executable = socket_root / "codex" + executable.write_text("#!/bin/sh\nprintf 'codex-cli 1.2.3\\n'\n") + executable.chmod(0o700) + runtime = CodexRuntime( + executable, socket_root, socket_root / "server.sock", "/usr/bin:/bin" + ) + async with FakeAppServer(runtime.control_socket) as server: + server.answers("initialize", {"userAgent": "codex_cli_rs/1.2.3 (Mac OS)"}) + updater = CodexUpdate(runtime=runtime) + result = await updater.check() + assert result.status == "unchanged" + assert not result.failure + + asyncio.run(scenario()) + + +def test_valid_upgrade_switches_once_and_reads_new_server(socket_root: Path) -> None: + async def scenario() -> None: + machine = UpdateMachine(socket_root) + runtime = CodexRuntime( + machine.executable, socket_root, socket_root / "s.sock", os.environ["PATH"] + ) + path = agent.plist_path(socket_root) + path.write_text(agent.render(agent.job(runtime, socket_root / "server.log"))) + machine.launchd.bootstrap(path) + try: + for _ in range(100): + if runtime.control_socket.exists(): + break + await asyncio.sleep(0.01) + machine.version.write_text("1.2.4") + updater = CodexUpdate( + runtime=runtime, + launchd=machine.launchd, + directory=socket_root, + record_path=socket_root / "record.json", + log_path=socket_root / "server.log", + list_sessions=lambda: empty_sessions(), + ) + first, second = await asyncio.gather(updater.check(), updater.check()) + assert [first.status, second.status] == ["updated", "unchanged"] + assert machine.commands.count("bootout") == 1 + finally: + machine.close() + + asyncio.run(scenario()) + + +async def empty_sessions(): + return () + + +def test_incomplete_installation_keeps_old_server(socket_root: Path) -> None: + async def scenario() -> None: + runtime = CodexRuntime( + socket_root / "missing", socket_root, socket_root / "s.sock", "/usr/bin:/bin" + ) + async with FakeAppServer(runtime.control_socket) as server: + server.answers("initialize", {"userAgent": "codex_cli_rs/1.2.3 (Mac OS)"}) + result = await CodexUpdate(runtime=runtime).check() + assert result.status == "failed" + assert result.failure + assert server.path.exists() + assert not server.calls_to("turn/start") + + asyncio.run(scenario()) + + +def test_version_output_alone_does_not_prove_candidate_ready(socket_root: Path) -> None: + async def scenario() -> None: + executable = socket_root / "codex" + executable.write_text( + '#!/bin/sh\nif [ "$1" = "--version" ]; then echo "codex-cli 1.2.4"; ' + 'else echo "missing runtime dependency" >&2; exit 1; fi\n' + ) + executable.chmod(0o700) + runtime = CodexRuntime(executable, socket_root, socket_root / "s.sock", "/usr/bin:/bin") + async with FakeAppServer(runtime.control_socket) as server: + server.answers("initialize", {"userAgent": "codex_cli_rs/1.2.3 (Mac OS)"}) + result = await CodexUpdate(runtime=runtime).check() + assert result.status == "failed" + assert "candidate" in result.failure.lower() + assert server.path.exists() + + asyncio.run(scenario()) + + +@pytest.mark.parametrize("closed", [False, True]) +@pytest.mark.parametrize("reading", ["steady", "missing", "drifting"]) +def test_recovery_requires_original_live_session( + socket_root: Path, closed: bool, reading: str +) -> None: + async def scenario(): + machine = UpdateMachine(socket_root) + machine.threads.write_text( + json.dumps( + [ + { + "id": "original", + "threadSource": "user", + "cwd": str(socket_root), + "createdAt": 100, + "status": {"type": "active"}, + } + ] + ) + ) + runtime = CodexRuntime( + machine.executable, socket_root, socket_root / "s.sock", os.environ["PATH"] + ) + path = agent.plist_path(socket_root) + path.write_text(agent.render(agent.job(runtime, socket_root / "server.log"))) + machine.launchd.bootstrap(path) + + async def sessions(): + if (closed or reading == "missing") and "bootout" in machine.commands: + return () + started = 200 if reading == "drifting" and "bootout" in machine.commands else 99 + return ( + Candidate( + pid=123, workspace=socket_root, session_id="original", started_at=started + ), + ) + + async def identity(pid): + return None if closed and "bootout" in machine.commands else "same live process" + + try: + while not runtime.control_socket.exists(): + await asyncio.sleep(0.01) + machine.version.write_text("1.2.4") + # The terminal never resumes on the new server. Only closing it + # legitimately removes it from the required recovery set. + machine.threads.write_text("[]") + updater = CodexUpdate( + runtime=runtime, + launchd=machine.launchd, + directory=socket_root, + record_path=socket_root / "record.json", + log_path=socket_root / "log", + list_sessions=sessions, + process_identity=identity, + recovery_timeout=0.5, + ) + result = await updater.check() + assert result.status == ("updated" if closed else "failed"), result + assert machine.commands.count("bootout") == 1 + again = await updater.check() + assert again.status == "unchanged" + assert machine.commands.count("bootout") == 1 + finally: + machine.close() + + asyncio.run(scenario()) + + +def test_failed_start_without_old_binary_does_not_retry(socket_root: Path): + async def scenario(): + machine = UpdateMachine(socket_root) + runtime = CodexRuntime( + machine.executable, socket_root, socket_root / "s.sock", os.environ["PATH"] + ) + path = agent.plist_path(socket_root) + path.write_text(agent.render(agent.job(runtime, socket_root / "log"))) + machine.launchd.bootstrap(path) + try: + while not runtime.control_socket.exists(): + await asyncio.sleep(0.01) + machine.version.write_text("1.2.4") + machine.refuse_start = True + updater = CodexUpdate( + runtime=runtime, + launchd=machine.launchd, + directory=socket_root, + record_path=socket_root / "record.json", + log_path=socket_root / "log", + list_sessions=empty_sessions, + ) + result = await updater.check() + assert result.status == "failed" + assert "old executable is no longer available" in result.failure + assert machine.commands.count("bootstrap") == 2 + assert result.attempted + finally: + machine.close() + + asyncio.run(scenario()) + + +def test_failed_candidate_restores_only_a_still_runnable_old_version(socket_root: Path): + async def scenario(): + old_root, new_root = socket_root / "old", socket_root / "new" + old_root.mkdir() + new_root.mkdir() + machine, candidate = UpdateMachine(old_root), UpdateMachine(new_root) + candidate.version.write_text("1.2.4") + runtime = CodexRuntime( + candidate.executable, socket_root, socket_root / "s.sock", os.environ["PATH"] + ) + old_runtime = CodexRuntime( + machine.executable, socket_root, runtime.control_socket, runtime.path + ) + path = agent.plist_path(socket_root) + path.write_text(agent.render(agent.job(old_runtime, socket_root / "log"))) + machine.launchd.bootstrap(path) + path.write_text(agent.render(agent.job(runtime, socket_root / "log"))) + + def launchctl(arguments): + if arguments[1] == "bootstrap": + import plistlib + + program = plistlib.loads(Path(arguments[-1]).read_bytes())["ProgramArguments"][0] + if program == str(candidate.executable): + return 1, "new candidate cannot start in the real environment" + return machine.run(arguments) + + launchd = agent.Launchd("gui/test", launchctl, lambda: "test-boot") + try: + while not runtime.control_socket.exists(): + await asyncio.sleep(0.01) + result = await CodexUpdate( + runtime=runtime, + launchd=launchd, + directory=socket_root, + record_path=socket_root / "record.json", + log_path=socket_root / "log", + list_sessions=empty_sessions, + ).check() + assert result.status == "failed" + assert "existing old version was restored" in result.failure, result + assert machine.arguments[0] == str(machine.executable) + finally: + machine.close() + + asyncio.run(scenario()) + + +def test_foreign_socket_peer_is_never_stopped(socket_root: Path): + async def scenario(): + machine = UpdateMachine(socket_root) + runtime = CodexRuntime( + machine.executable, socket_root, socket_root / "s.sock", os.environ["PATH"] + ) + path = agent.plist_path(socket_root) + path.write_text(agent.render(agent.job(runtime, socket_root / "log"))) + machine.launchd.bootstrap(path) + + def launchctl(arguments): + code, output = machine.run(arguments) + if arguments[1] == "print": + output = output.replace(f"pid = {machine.process.pid}", f"pid = {os.getpid()}") + return code, output + + try: + while not runtime.control_socket.exists(): + await asyncio.sleep(0.01) + machine.version.write_text("1.2.4") + updater = CodexUpdate( + runtime=runtime, + launchd=agent.Launchd("gui/test", launchctl, lambda: "test-boot"), + directory=socket_root, + record_path=socket_root / "record.json", + log_path=socket_root / "log", + list_sessions=empty_sessions, + ) + result = await updater.check() + assert result.status == "failed" + assert "socket peer does not belong" in result.failure + assert machine.process.poll() is None + assert "bootout" not in machine.commands + finally: + machine.close() + + asyncio.run(scenario())