Skip to content

fix(codex): restore Desktop discovery and lifecycle tracking #280

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hyiiiii wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(codex): restore Desktop discovery and lifecycle tracking #280

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 108 additions & 14 deletions Sources/CodeIsland/AppState+CodexAppServer.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,16 @@ import AppKit
import SwiftUI
import CodeIslandCore

enum CodexAppServerExitStrategy: Equatable {
case reconnectPreservingSessions
case stopAndRemoveSessions
}

extension AppState {
/// Session ID prefix applied to Codex threads surfaced via the app-server.
/// The rollout-file discovery path uses the raw UUID; the `codexapp:` prefix
/// keeps the two channels' session namespaces disjoint so a user running
/// Codex Desktop AND Codex CLI simultaneously doesn't see them collapse.
static let codexAppSessionPrefix = "codexapp:"
/// Desktop app-server, rollout fallback, and state-DB discovery all use this
/// prefix; Codex CLI keeps the raw UUID so the two modes remain distinct.
nonisolated static let codexAppSessionPrefix = "codexapp:"
// Plain `let` constants — nonisolated so NSWorkspace notification
// handlers on background queues can read them without Swift 6
// Sendable warnings.
Expand Down Expand Up @@ -54,6 +58,8 @@ extension AppState {
}

func stopCodexAppServerWatcher() {
codexAppServerReconnectTask?.cancel()
codexAppServerReconnectTask = nil
if let observers = codexAppServerObservers {
let center = NSWorkspace.shared.notificationCenter
for observer in observers { center.removeObserver(observer) }
Expand All @@ -66,16 +72,20 @@ extension AppState {

private func startCodexAppServerClientIfPossible() {
guard codexAppServerClient == nil else { return }
guard let executable = Self.codexAppServerExecutableURL() else { return }
guard let executable = Self.codexAppServerExecutableURL() else {
scheduleCodexAppServerReconnect()
return
}

let client = CodexAppServerClient(executableURL: executable)
client.onMessage = { [weak self] message in
Task { @MainActor in self?.handleCodexAppServerMessage(message) }
}
client.onExit = { [weak self] _ in
client.onExit = { [weak self, weak client] _ in
Task { @MainActor in
self?.codexAppServerClient = nil
self?.removeCodexAppServerSessions()
guard let self, let client,
self.codexAppServerClient === client else { return }
self.handleCodexAppServerExit()
}
}

Expand All @@ -85,11 +95,93 @@ extension AppState {
try client.initializeHandshake(clientName: "CodeIsland", clientVersion: version)
} catch {
client.stop()
scheduleCodexAppServerReconnect()
return
}
codexAppServerClient = client
}

nonisolated static func codexAppServerExitStrategy(
hostRunning: Bool,
watcherActive: Bool
) -> CodexAppServerExitStrategy {
hostRunning && watcherActive
? .reconnectPreservingSessions
: .stopAndRemoveSessions
}

private func handleCodexAppServerExit() {
codexAppServerClient = nil
let hostRunning = NSWorkspace.shared.runningApplications.contains {
$0.bundleIdentifier == AppState.codexAppBundleId
}
switch Self.codexAppServerExitStrategy(
hostRunning: hostRunning,
watcherActive: codexAppServerObservers != nil
) {
case .reconnectPreservingSessions:
// Preserve state-backed cards, refresh immediately, and restore the
// live JSON-RPC channel with capped exponential backoff.
invalidateCodexAppServerQuestionsAfterDisconnect()
requestCodexDesktopDiscoveryScan()
scheduleCodexAppServerReconnect()
case .stopAndRemoveSessions:
codexAppServerReconnectTask?.cancel()
codexAppServerReconnectTask = nil
removeCodexAppServerSessions()
}
}

/// A server->client request belongs to the exact client process that issued
/// it and is not replayed onto a replacement connection. Remove only those
/// dead-channel questions; hook-backed queues retain their continuations.
func invalidateCodexAppServerQuestionsAfterDisconnect() {
let affectedSessionIds = Set(questionQueue.compactMap { request in
request.isCodexAppServer ? request.event.sessionId : nil
})
guard !affectedSessionIds.isEmpty else { return }
questionQueue.removeAll { $0.isCodexAppServer }
for sessionId in affectedSessionIds {
guard sessions[sessionId]?.status == .waitingQuestion else { continue }
sessions[sessionId]?.status = .processing
sessions[sessionId]?.currentTool = nil
sessions[sessionId]?.toolDescription = nil
}
showNextPending()
refreshDerivedState()
}

private func scheduleCodexAppServerReconnect() {
guard codexAppServerObservers != nil,
codexAppServerReconnectTask == nil else { return }

codexAppServerReconnectTask = Task { @MainActor [weak self] in
var delay: UInt64 = 500_000_000
while !Task.isCancelled {
try? await Task.sleep(nanoseconds: delay)
guard !Task.isCancelled, let self else { return }
guard self.codexAppServerObservers != nil else {
self.codexAppServerReconnectTask = nil
return
}
guard NSWorkspace.shared.runningApplications.contains(where: {
$0.bundleIdentifier == AppState.codexAppBundleId
}) else {
self.codexAppServerReconnectTask = nil
self.removeCodexAppServerSessions()
return
}

self.startCodexAppServerClientIfPossible()
if self.codexAppServerClient != nil {
self.codexAppServerReconnectTask = nil
return
}
delay = min(delay * 2, 10_000_000_000)
}
}
}

static func codexAppServerExecutableURL(
runningBundleURLs: [URL] = NSWorkspace.shared.runningApplications.compactMap { app in
app.bundleIdentifier == AppState.codexAppBundleId ? app.bundleURL : nil
Expand Down Expand Up @@ -118,17 +210,19 @@ extension AppState {
}

private func stopCodexAppServerClient() {
codexAppServerClient?.stop()
codexAppServerReconnectTask?.cancel()
codexAppServerReconnectTask = nil
let client = codexAppServerClient
codexAppServerClient = nil
client?.stop()
removeCodexAppServerSessions()
}

private func removeCodexAppServerSessions() {
let stale = sessions.keys.filter { $0.hasPrefix(AppState.codexAppSessionPrefix) }
for id in stale {
sessions.removeValue(forKey: id)
removeSession(id)
}
refreshDerivedState()
}

// MARK: - Notification dispatch
Expand Down Expand Up @@ -315,6 +409,7 @@ extension AppState {
guard let thread = params["thread"]?.asObject else { return }
guard let threadId = thread["id"]?.asString else { return }
let sessionId = AppState.codexAppSessionPrefix + threadId
closedCodexAppThreads.removeValue(forKey: threadId)

var snapshot = sessions[sessionId] ?? SessionSnapshot(startTime: Date())
snapshot.source = "codex"
Expand Down Expand Up @@ -354,9 +449,8 @@ extension AppState {
private func applyCodexThreadClosedNotification(params: [String: AnyCodableLike]) {
guard let threadId = params["threadId"]?.asString else { return }
let sessionId = AppState.codexAppSessionPrefix + threadId
sessions.removeValue(forKey: sessionId)
detachTranscriptTailer(sessionId: sessionId)
refreshDerivedState()
closedCodexAppThreads[threadId] = Date()
removeSession(sessionId)
}

/// Map a ThreadStatus union onto our flat AgentStatus enum. Shared between the
Expand Down
39 changes: 11 additions & 28 deletions Sources/CodeIsland/AppState+TranscriptTailer.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,10 @@ extension AppState {
}
}

transcriptTailer.attach(sessionId: sessionId, filePath: path)
attachedTranscriptTokens[sessionId] = transcriptTailer.attach(
sessionId: sessionId,
filePath: path
)
}

/// Backfill freshness bound for flipping a session into the display-only
Expand Down Expand Up @@ -160,42 +163,22 @@ extension AppState {
}
}

private nonisolated static func latestCodexTurnStatus(path: String) -> ConversationTurnStatus? {
guard let handle = FileHandle(forReadingAtPath: path) else { return nil }
defer { handle.closeFile() }

// A long Codex turn can place its task_started event well before the
// final 128 KB after emitting large reasoning/tool rows. Scan in chunks
// so startup state recovery remains bounded in memory without missing
// that event.
handle.seek(toFileOffset: 0)
let chunkSize = 64 * 1024
var pendingFragment = Data()
var latestStatus: ConversationTurnStatus?

while true {
let chunk = handle.readData(ofLength: chunkSize)
if chunk.isEmpty { break }

let result = JSONLTailer.scanLines(pendingFragment + chunk)
pendingFragment = result.trailingFragment
if let turnStatus = result.delta.turnStatus {
latestStatus = turnStatus
}
}

return latestStatus
}

/// Stop watching a session's transcript. Called when the session is removed or
/// when a new transcript path supersedes an older one.
func detachTranscriptTailer(sessionId: String) {
attachedTranscriptPaths.removeValue(forKey: sessionId)
attachedTranscriptTokens.removeValue(forKey: sessionId)
transcriptTailer.detach(sessionId: sessionId)
}

/// Apply an incremental update produced by the tailer. Runs on the main actor.
func applyTranscriptDelta(_ delta: ConversationTailDelta) {
if let attachmentToken = delta.attachmentToken {
guard attachedTranscriptTokens[delta.sessionId] == attachmentToken,
attachedTranscriptPaths[delta.sessionId] == delta.filePath else {
return
}
}
guard var session = sessions[delta.sessionId] else { return }
var mutated = false

Expand Down
Loading