diff --git a/Sources/CodeIsland/AppState+CodexAppServer.swift b/Sources/CodeIsland/AppState+CodexAppServer.swift index ab7b5e3e..2ee8b384 100644 --- a/Sources/CodeIsland/AppState+CodexAppServer.swift +++ b/Sources/CodeIsland/AppState+CodexAppServer.swift @@ -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. @@ -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) } @@ -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() } } @@ -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 @@ -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 @@ -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" @@ -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 diff --git a/Sources/CodeIsland/AppState+TranscriptTailer.swift b/Sources/CodeIsland/AppState+TranscriptTailer.swift index 4fd25f0e..1da7dd24 100644 --- a/Sources/CodeIsland/AppState+TranscriptTailer.swift +++ b/Sources/CodeIsland/AppState+TranscriptTailer.swift @@ -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 @@ -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 diff --git a/Sources/CodeIsland/AppState.swift b/Sources/CodeIsland/AppState.swift index d3715c7f..ecc4c818 100644 --- a/Sources/CodeIsland/AppState.swift +++ b/Sources/CodeIsland/AppState.swift @@ -36,11 +36,62 @@ struct CodexSubagentMetadata: Equatable, Sendable { let agentNickname: String? } +private struct CodexSpawnEdgeRecord { + let metadata: CodexSubagentMetadata + let status: String? + let transcriptPath: String? +} + +private enum CodexTranscriptSubagentInspection { + /// The first line could not be established as a valid `session_meta`. + case unavailable + /// A valid root `session_meta` explicitly contains no subagent relation. + case root + case subagent(CodexSubagentMetadata) +} + struct ProcessIdentity: Equatable { let pid: pid_t let startTime: Date? } +struct CodexDiscoveryIdentity: Equatable, Sendable { + let sessionId: String + let providerSessionId: String? + let termBundleId: String? +} + +struct CodexProcessDiscoveryCandidate: Sendable { + let pid: pid_t + let cwd: String? + let startTime: Date? + let isDesktop: Bool +} + +enum CodexTranscriptOrigin: Equatable, Sendable { + case desktop + case cli + case unknown +} + +/// A Codex Desktop thread recovered from Codex's local state database. +/// +/// Recent Codex Desktop builds run their app-server with `/` as the process CWD, +/// so the CLI discovery path cannot map that process back to a project. Keeping +/// this small value type separate lets discovery hydrate native-app sessions from +/// the authoritative state DB without exposing the private `DiscoveredSession`. +struct CodexDesktopThreadRecord: Sendable { + let sessionId: String + let cwd: String + let model: String? + let modifiedAt: Date + let recentMessages: [ChatMessage] + let transcriptPath: String + let status: AgentStatus + let subagentMetadata: CodexSubagentMetadata? + let subagentStatus: String? +} + @MainActor @Observable final class AppState { @@ -105,6 +156,11 @@ final class AppState { /// reattach when the path actually changes. See AppState+TranscriptTailer. @ObservationIgnored var attachedTranscriptPaths: [String: String] = [:] + /// Token for the exact JSONLTailer attachment currently owned by a session. + /// A queued delta from an older attachment must not mutate a same-id session + /// that was closed and subsequently re-created. + @ObservationIgnored + var attachedTranscriptTokens: [String: UUID] = [:] /// Watches active session transcripts for appended assistant lines. Lazily /// constructed so the delta handler can safely capture `self`. @ObservationIgnored @@ -120,6 +176,14 @@ final class AppState { /// NSWorkspace launch/terminate observers tracking Codex Desktop. @ObservationIgnored var codexAppServerObservers: [NSObjectProtocol]? + /// Backoff loop used when the auxiliary app-server exits while the Desktop + /// host remains open. + @ObservationIgnored + nonisolated(unsafe) var codexAppServerReconnectTask: Task? + /// Prevent an in-flight/stale state-DB scan from recreating a thread after + /// app-server delivered `thread/closed`. + @ObservationIgnored + var closedCodexAppThreads: [String: Date] = [:] /// Computed: first item in permission queue (backward compat for UI reads) var pendingPermission: PermissionRequest? { permissionQueue.first } @@ -188,6 +252,9 @@ final class AppState { @ObservationIgnored nonisolated(unsafe) private var discoveryScanTask: Task? private var pendingDiscoveryRescan = false + @ObservationIgnored + nonisolated(unsafe) private var codexDesktopDiscoveryScanTask: Task? + private var lastCodexDesktopDiscoveryPollAt: Date? private var isShowingCompletion: Bool { if case .completionCard = surface { return true } return false @@ -345,6 +412,15 @@ final class AppState { !runningBundleIds.contains(bundleId) else { continue } removeSession(key) } + let discoveryPollNow = Date() + if Self.shouldPollCodexDesktopDiscovery( + runningBundleIdentifiers: runningBundleIds, + lastPollAt: lastCodexDesktopDiscoveryPollAt, + now: discoveryPollNow + ) { + lastCodexDesktopDiscoveryPollAt = discoveryPollNow + requestCodexDesktopDiscoveryScan() + } // 4. Remove idle sessions past timeout (user setting, or 10 min default for no-monitor sessions) let userTimeout = SettingsManager.shared.sessionTimeout @@ -471,7 +547,8 @@ final class AppState { } private nonisolated static func isNativeAppProcess(_ pid: pid_t, source: String) -> Bool { - guard let path = executablePath(for: pid)?.lowercased() else { return false } + guard let executable = executablePath(for: pid) else { return false } + let path = executable.lowercased() switch source { case "cursor": return path.contains("/cursor.app/contents/") case "trae": return path.contains("/trae.app/contents/") @@ -485,7 +562,7 @@ final class AppState { case "codebuddy": return path.contains("/codebuddy.app/contents/") case "codybuddycn": return path.contains("/codebuddycn.app/contents/") || path.contains("/codebuddy.app/contents/") case "stepfun": return path.contains("/stepfun.app/contents/") - case "codex": return path.contains("/codex.app/contents/") + case "codex": return isCodexExecutablePath(executable) case "opencode": return path.contains("/opencode.app/contents/") case "antigravity": return path.contains("/antigravity.app/contents/") // Google Antigravity IDE — host app is Antigravity.app. Same .app path as @@ -617,7 +694,7 @@ final class AppState { /// Remove a session, clean up its monitor, and resume any pending continuations. /// Every removal path (cleanup timer, process exit, reducer effect) goes through here /// so leaked continuations / connections are impossible. - private func removeSession(_ sessionId: String) { + func removeSession(_ sessionId: String) { // Resume ALL pending continuations for this session drainPermissions(forSession: sessionId, reason: "removeSession") drainQuestions(forSession: sessionId, reason: "removeSession") @@ -728,6 +805,19 @@ final class AppState { /// Prefers the PID captured by the bridge (_ppid), falls back to source-aware process scans by CWD. private func tryMonitorSession(_ sessionId: String) { guard sessions[sessionId]?.isRemote != true else { return } + // Codex Desktop cards are owned by NSWorkspace/app-server lifecycle. A + // CWD lookup can only find an unrelated Codex CLI process (or the + // shared Desktop helper), and binding either PID would remove every + // Desktop thread when that one process exits. Other native providers + // keep their existing monitor behavior. + if sessions[sessionId]?.source == "codex", + sessions[sessionId]?.termBundleId == Self.codexAppBundleId { + stopMonitor(sessionId) + sessions[sessionId]?.cliPid = nil + sessions[sessionId]?.cliStartTime = nil + exitingSessions.removeValue(forKey: sessionId) + return + } let currentMonitor = processMonitors[sessionId]?.process // Primary: use PID from bridge (works for any CLI) @@ -1117,6 +1207,26 @@ final class AppState { } let sessionId = event.sessionId ?? "default" + let normalizedEventName = EventNormalizer.normalize(event.eventName) + + if source?.lowercased() == "codex", + event.rawJSON["_term_bundle"] as? String == Self.codexAppBundleId, + let rawProviderSessionId = event.rawJSON["session_id"] as? String { + let providerSessionId = rawProviderSessionId.hasPrefix(Self.codexAppSessionPrefix) + ? String(rawProviderSessionId.dropFirst(Self.codexAppSessionPrefix.count)) + : rawProviderSessionId + if closedCodexAppThreads[providerSessionId] != nil { + // Tool/notification/terminal hooks can trail app-server + // `thread/closed`. Only explicit generation-start activity may + // reopen; `thread/started` clears the same tombstone directly. + if normalizedEventName == "SessionStart" + || normalizedEventName == "UserPromptSubmit" { + closedCodexAppThreads.removeValue(forKey: providerSessionId) + } else { + return + } + } + } // Skip Codex APP internal sessions (title generation, etc.) — they have no transcript if (event.rawJSON["_source"] as? String) == "codex" @@ -1129,7 +1239,6 @@ final class AppState { sessions[sessionId] = SessionSnapshot() } - let normalizedEventName = EventNormalizer.normalize(event.eventName) let prevStatus = sessions[sessionId]?.status let wasWaiting = prevStatus == .waitingApproval || prevStatus == .waitingQuestion let cwdBeforeReduce = sessions[sessionId]?.cwd @@ -1520,6 +1629,29 @@ final class AppState { })?.key } + /// Resolve a Codex parent in the same Desktop/CLI namespace as its child. + /// Raw and `codexapp:` cards may legitimately share a provider thread id + /// when a user resumes the same thread in both surfaces. + func findCodexParentSessionId( + providerSessionId: String, + childTermBundleId: String? + ) -> String? { + let childIsDesktop = childTermBundleId == Self.codexAppBundleId + let canonicalId = childIsDesktop + ? Self.codexAppSessionPrefix + providerSessionId + : providerSessionId + if let session = sessions[canonicalId], + session.source == "codex", + session.isNativeAppMode == childIsDesktop { + return canonicalId + } + return sessions.first(where: { _, snapshot in + snapshot.source == "codex" + && snapshot.providerSessionId == providerSessionId + && snapshot.isNativeAppMode == childIsDesktop + })?.key + } + func denyPermission() { guard !permissionQueue.isEmpty else { return } let pending = permissionQueue.removeFirst() @@ -2295,7 +2427,7 @@ final class AppState { // MARK: - Session Discovery (FSEventStream + process scan) // MARK: - Session Persistence - private func scheduleSave() { + func scheduleSave() { saveTimer?.invalidate() saveTimer = Timer.scheduledTimer(withTimeInterval: 2, repeats: false) { [weak self] _ in Task { @MainActor in @@ -2312,8 +2444,14 @@ final class AppState { let persisted = SessionPersistence.load() let cutoff = Date().addingTimeInterval(-30 * 60) // 30 minutes for p in persisted where p.lastActivity > cutoff { - guard sessions[p.sessionId] == nil else { continue } guard let source = SessionSnapshot.normalizedSupportedSource(p.source) else { continue } + let restoredSessionId = Self.canonicalRestoredCodexSessionId( + sessionId: p.sessionId, + source: source, + providerSessionId: p.providerSessionId, + termBundleId: p.termBundleId + ) + guard sessions[restoredSessionId] == nil else { continue } var snapshot = SessionSnapshot(startTime: p.startTime) snapshot.cwd = p.cwd snapshot.source = source @@ -2362,19 +2500,23 @@ final class AppState { if snapshot.cliPid == nil && snapshot.status == .idle && snapshot.lastUserPrompt == nil, !Self.shouldKeepRestoredIdleCursorSession( source: source, - sessionId: p.sessionId, + sessionId: restoredSessionId, providerSessionId: snapshot.providerSessionId, transcriptPath: snapshot.transcriptPath, closedSubagentIds: snapshot.closedSubagentIds ) { continue } - sessions[p.sessionId] = snapshot - refreshProviderTitle(for: p.sessionId) + sessions[restoredSessionId] = snapshot + refreshProviderTitle(for: restoredSessionId) // Branch is re-read, not persisted — it may have changed between runs. - maybeRefreshGitBranch(for: p.sessionId, cwdBefore: nil, normalizedEventName: "SessionStart") + maybeRefreshGitBranch( + for: restoredSessionId, + cwdBefore: nil, + normalizedEventName: "SessionStart" + ) // Reattach exit monitoring without changing the restored idle/running snapshot. - tryMonitorSession(p.sessionId) + tryMonitorSession(restoredSessionId) } SessionPersistence.clear() _ = applyCodexSubsessionModeToKnownSessions() @@ -2496,6 +2638,40 @@ final class AppState { } } + /// FSEvents is not guaranteed to report every append below + /// `~/.codex/sessions` on all macOS/filesystem combinations. Reuse the + /// cleanup tick as a low-frequency fallback, but only while Codex Desktop + /// is running and only after the polling interval has elapsed. + nonisolated static func shouldPollCodexDesktopDiscovery( + runningBundleIdentifiers: Set, + lastPollAt: Date?, + now: Date, + interval: TimeInterval = 6 + ) -> Bool { + guard runningBundleIdentifiers.contains(codexAppBundleId) else { return false } + guard let lastPollAt else { return true } + return now.timeIntervalSince(lastPollAt) >= interval + } + + func requestCodexDesktopDiscoveryScan() { + guard codexDesktopDiscoveryScanTask == nil else { return } + codexDesktopDiscoveryScanTask = Task.detached { [weak self] in + let discovered = ConfigInstaller.isEnabled(source: "codex") + ? Self.findRecentCodexDesktopSessions() + : [] + guard !Task.isCancelled else { return } + await MainActor.run { [weak self] in + guard let self else { return } + guard !Task.isCancelled else { + self.codexDesktopDiscoveryScanTask = nil + return + } + self.integrateDiscovered(discovered) + self.codexDesktopDiscoveryScanTask = nil + } + } + } + func startSessionDiscovery() { startCleanupTimer() // Restore persisted sessions before process scan (deduped by scan) @@ -2533,7 +2709,11 @@ final class AppState { watchRoots as CFArray, FSEventStreamEventId(kFSEventStreamEventIdSinceNow), 2.0, // 2-second latency (coalesces rapid writes) - FSEventStreamCreateFlags(kFSEventStreamCreateFlagUseCFTypes) + FSEventStreamCreateFlags( + kFSEventStreamCreateFlagUseCFTypes + | kFSEventStreamCreateFlagNoDefer + | kFSEventStreamCreateFlagFileEvents + ) ) guard let stream = stream else { return } @@ -2548,16 +2728,57 @@ final class AppState { nonisolated fileprivate func handleProjectsDirChange() { Task { @MainActor [weak self] in guard let self = self else { return } - // Debounce: skip if scanned within the last 3 seconds - guard Date().timeIntervalSince(self.lastFSScanTime) > 3 else { return } - self.lastFSScanTime = Date() + // FSEvents already coalesces writes with a two-second latency, and + // requestDiscoveryScan() coalesces events that arrive during a scan. + // Dropping events in an additional time window can miss the final + // lifecycle write and leave the island showing stale state. self.requestDiscoveryScan() } } + /// Provider thread IDs are authoritative when both sides have one. Two + /// native Codex Desktop threads can legitimately share a source and CWD + /// while having no PID, so they must not be collapsed into one card merely + /// because the generic discovery heuristic sees the same project path. + nonisolated static func providerSessionIdentifiersMayRepresentSameSession( + existing: String?, + discovered: String? + ) -> Bool { + guard let discovered, !discovered.isEmpty else { return true } + guard let existing, !existing.isEmpty else { return false } + return existing == discovered + } + + /// Native-app and CLI discoveries are separate namespaces even when they + /// share the same provider and working directory. In particular, a Codex + /// CLI rollout has no native host bundle and must not be folded into a + /// Codex Desktop card left by state-database discovery. + nonisolated static func discoveryAppModesMayRepresentSameSession( + existingIsNativeAppMode: Bool, + discoveredTermBundleId: String? + ) -> Bool { + existingIsNativeAppMode == (discoveredTermBundleId != nil) + } + + /// Completed Codex subagents are transient workers, not idle top-level + /// sessions. Their spawn-edge row may remain `open`, so transcript lifecycle + /// is the reliable signal for removing them from the island. + nonisolated static func isCompletedCodexSubagentDiscovery( + source: String, + parentSessionId: String?, + status: AgentStatus? + ) -> Bool { + source == "codex" + && parentSessionId?.isEmpty == false + && status == .idle + } + /// Update existing session's messages from discovered transcript data. private func backfillSessionMessages(sessionId: String, from info: DiscoveredSession) -> Bool { guard var session = sessions[sessionId], !info.recentMessages.isEmpty else { return false } + guard info.modifiedAt >= session.lastActivity || session.recentMessages.isEmpty else { + return false + } var mutated = false let messagesChanged = session.recentMessages.count != info.recentMessages.count || zip(session.recentMessages, info.recentMessages).contains { $0.isUser != $1.isUser || $0.text != $1.text } @@ -2582,19 +2803,37 @@ final class AppState { } /// Merge discovered sessions into current state (skip already-known ones) - private func integrateDiscovered(_ discovered: [DiscoveredSession]) { + func integrateDiscovered(_ discovered: [DiscoveredSession]) { var didMutate = false for info in discovered { + if shouldSuppressClosedCodexDesktopDiscovery(info) { + continue + } if routeDiscoveredSubsessionIfNeeded(info) { didMutate = true continue } + if Self.isCompletedCodexSubagentDiscovery( + source: info.source, + parentSessionId: info.parentSessionId, + status: info.status + ) { + if sessions[info.sessionId] != nil { + removeSession(info.sessionId) + didMutate = true + } + continue + } + // Session already known — try to update PID and attach monitor. // Discovery PIDs are heuristic (matched by CWD), so when the session already // has a known-good alive PID that differs from discovery, we trust the existing // one for both cliPid and monitor to avoid cross-session contamination. if sessions[info.sessionId] != nil { + if applyDiscoveredRuntimeState(info, to: info.sessionId) { + didMutate = true + } if let pid = info.pid, pid > 0 { let existingPid = sessions[info.sessionId]?.cliPid ?? 0 let existingProcess = resolvedSessionProcessIdentity(for: info.sessionId) @@ -2612,17 +2851,15 @@ final class AppState { if backfillSessionMessages(sessionId: info.sessionId, from: info) { didMutate = true } - if sessions[info.sessionId]?.cwd != info.cwd { - sessions[info.sessionId]?.cwd = info.cwd - didMutate = true - } - if let path = info.transcriptPath, sessions[info.sessionId]?.transcriptPath != path { - sessions[info.sessionId]?.transcriptPath = path + if applyDiscoveredMetadata(info, to: info.sessionId) { didMutate = true } attachTranscriptTailerIfNeeded(sessionId: info.sessionId) tryMonitorSession(info.sessionId) - refreshProviderTitle(for: info.sessionId, providerSessionId: info.sessionId) + refreshProviderTitle( + for: info.sessionId, + providerSessionId: info.providerSessionId ?? info.sessionId + ) continue } @@ -2636,6 +2873,14 @@ final class AppState { let duplicateKey = sessions.first(where: { (_, existing) in guard existing.source == info.source, existing.cwd != nil, existing.cwd == info.cwd else { return false } + guard Self.discoveryAppModesMayRepresentSameSession( + existingIsNativeAppMode: existing.isNativeAppMode, + discoveredTermBundleId: info.termBundleId + ) else { return false } + guard Self.providerSessionIdentifiersMayRepresentSameSession( + existing: existing.providerSessionId, + discovered: info.providerSessionId + ) else { return false } // Don't merge CLI discovery into a stale native app session whose app has quit — // the PID was likely reattached incorrectly. If the native app IS running, allow merge. if existing.isNativeAppMode, @@ -2652,6 +2897,9 @@ final class AppState { })?.key if let existingKey = duplicateKey { + if applyDiscoveredRuntimeState(info, to: existingKey) { + didMutate = true + } // Same guard as above: don't let unreliable discovery PID contaminate // an existing session that has a known-good alive PID. if let pid = info.pid, pid > 0 { @@ -2669,17 +2917,15 @@ final class AppState { if backfillSessionMessages(sessionId: existingKey, from: info) { didMutate = true } - if sessions[existingKey]?.cwd != info.cwd { - sessions[existingKey]?.cwd = info.cwd - didMutate = true - } - if let path = info.transcriptPath, sessions[existingKey]?.transcriptPath != path { - sessions[existingKey]?.transcriptPath = path + if applyDiscoveredMetadata(info, to: existingKey) { didMutate = true } attachTranscriptTailerIfNeeded(sessionId: existingKey) tryMonitorSession(existingKey) - refreshProviderTitle(for: existingKey, providerSessionId: info.sessionId) + refreshProviderTitle( + for: existingKey, + providerSessionId: info.providerSessionId ?? info.sessionId + ) continue } @@ -2689,13 +2935,18 @@ final class AppState { session.ttyPath = info.tty session.recentMessages = info.recentMessages session.source = info.source + session.status = info.status ?? .idle + session.lastActivity = info.modifiedAt + session.termBundleId = info.termBundleId if let pid = info.pid, let process = Self.liveProcessIdentity(for: pid) { session.cliPid = process.pid session.cliStartTime = process.startTime } else { session.cliPid = info.pid } - session.providerSessionId = SessionTitleStore.supports(provider: info.source) ? info.sessionId : nil + session.providerSessionId = SessionTitleStore.supports(provider: info.source) + ? (info.providerSessionId ?? info.sessionId) + : nil if let last = info.recentMessages.last(where: { $0.isUser }) { session.lastUserPrompt = last.text } @@ -2704,7 +2955,10 @@ final class AppState { } session.transcriptPath = info.transcriptPath sessions[info.sessionId] = session - refreshProviderTitle(for: info.sessionId, providerSessionId: info.sessionId) + refreshProviderTitle( + for: info.sessionId, + providerSessionId: info.providerSessionId ?? info.sessionId + ) tryMonitorSession(info.sessionId) attachTranscriptTailerIfNeeded(sessionId: info.sessionId) didMutate = true @@ -2724,22 +2978,128 @@ final class AppState { refreshDerivedState() } + private func shouldSuppressClosedCodexDesktopDiscovery(_ info: DiscoveredSession) -> Bool { + guard info.source == "codex", + info.termBundleId == Self.codexAppBundleId, + let providerSessionId = info.providerSessionId, + let closedAt = closedCodexAppThreads[providerSessionId] else { + return false + } + if info.modifiedAt <= closedAt { + return true + } + // A terminal flush may land just after `thread/closed`; only a newer + // processing lifecycle is evidence that the thread genuinely resumed. + guard info.status == .processing else { return true } + closedCodexAppThreads.removeValue(forKey: providerSessionId) + return false + } + + /// Apply runtime fields that filesystem/state-DB discovery can know more + /// accurately than a stale restored snapshot. Interactive approval/question + /// states remain authoritative until their owning channel resolves them. + @discardableResult + private func applyDiscoveredRuntimeState(_ info: DiscoveredSession, to sessionId: String) -> Bool { + guard var session = sessions[sessionId] else { return false } + var mutated = false + let canApplyStatus = info.modifiedAt >= session.lastActivity + + if info.modifiedAt > session.lastActivity { + session.lastActivity = info.modifiedAt + mutated = true + } + if let bundleId = info.termBundleId, + (canApplyStatus || session.termBundleId == nil), + session.termBundleId != bundleId { + session.termBundleId = bundleId + mutated = true + } + if let discoveredStatus = info.status, + canApplyStatus, + session.status != .waitingApproval, + session.status != .waitingQuestion, + session.status != discoveredStatus { + session.status = discoveredStatus + if discoveredStatus == .idle { + session.currentTool = nil + session.toolDescription = nil + } + mutated = true + } + + if mutated { + sessions[sessionId] = session + } + return mutated + } + + /// CWD and transcript path participate in tailer/process routing, so an old + /// asynchronous scan must not rewind them after a newer DB result. Missing + /// restored fields may still be filled regardless of timestamp. + @discardableResult + private func applyDiscoveredMetadata(_ info: DiscoveredSession, to sessionId: String) -> Bool { + guard var session = sessions[sessionId] else { return false } + let canReplace = info.modifiedAt >= session.lastActivity + var mutated = false + + if (canReplace || session.cwd?.isEmpty != false), session.cwd != info.cwd { + session.cwd = info.cwd + mutated = true + } + if let path = info.transcriptPath, + (canReplace || session.transcriptPath == nil), + session.transcriptPath != path { + session.transcriptPath = path + mutated = true + } + if mutated { + sessions[sessionId] = session + } + return mutated + } + @discardableResult - func applyCodexSubsessionModeToKnownSessions() -> Bool { + func applyCodexSubsessionModeToKnownSessions(statePath: String? = nil) -> Bool { let mode = Self.currentPluginSessionMode() guard mode == "hide" || mode == "merge" else { return false } let candidates = sessions.map { (sessionId: $0.key, session: $0.value) } + var transcriptMetadata: [String: CodexSubagentMetadata] = [:] + var databaseLookupIds: Set = [] + + for candidate in candidates where candidate.session.source == "codex" { + let providerSessionId = candidate.session.providerSessionId ?? candidate.sessionId + if let transcriptPath = candidate.session.transcriptPath { + switch Self.inspectCodexSubagentMetadata(inTranscriptPath: transcriptPath) { + case .subagent(let metadata): + transcriptMetadata[candidate.sessionId] = metadata + databaseLookupIds.insert(providerSessionId) + case .root: + // A readable root session_meta is definitive. Falling back + // to SQLite here turns every root card into a separate DB + // open and can also resurrect stale spawn-edge relations. + continue + case .unavailable: + databaseLookupIds.insert(providerSessionId) + } + } else { + databaseLookupIds.insert(providerSessionId) + } + } + // One read-only DB connection for all candidates that genuinely need + // relation/status fallback. Root transcripts never enter this query. + let databaseRecords = Self.codexSpawnEdgeRecords( + threadIds: databaseLookupIds, + statePath: statePath + ) var didMutate = false for candidate in candidates where candidate.session.source == "codex" { let providerSessionId = candidate.session.providerSessionId ?? candidate.sessionId - guard let metadata = Self.codexSubagentMetadata( - threadId: providerSessionId, - transcriptPath: candidate.session.transcriptPath - ), + guard let metadata = transcriptMetadata[candidate.sessionId] + ?? databaseRecords[providerSessionId]?.metadata, metadata.parentThreadId != providerSessionId else { continue } @@ -2752,18 +3112,29 @@ final class AppState { continue } - guard let parentKey = findSessionId(providerSessionId: metadata.parentThreadId), + guard let parentKey = findCodexParentSessionId( + providerSessionId: metadata.parentThreadId, + childTermBundleId: candidate.session.termBundleId + ), parentKey != candidate.sessionId else { continue } + // Preserve the terminal snapshot before removeSession tears down the + // child card. Transcript lifecycle is authoritative even when the + // spawn edge is missing or remains "open". + let completedByTranscript = candidate.session.status == .idle if sessions[candidate.sessionId] != nil { removeSession(candidate.sessionId) didMutate = true } - if Self.codexThreadSpawnStatus(childThreadId: providerSessionId)?.lowercased() == "closed" { + if completedByTranscript + || databaseRecords[providerSessionId]?.status?.lowercased() == "closed" { if sessions[parentKey]?.subagents.removeValue(forKey: providerSessionId) != nil { + if sessions[parentKey]?.subagents.isEmpty == true { + clearSubagentProjection(fromParentSession: parentKey) + } didMutate = true } continue @@ -2772,7 +3143,7 @@ final class AppState { let agentType = metadata.agentType ?? metadata.agentNickname ?? "Agent" var subagent = sessions[parentKey]?.subagents[providerSessionId] ?? SubagentState(agentId: providerSessionId, agentType: agentType) - subagent.status = candidate.session.status == .idle ? .running : candidate.session.status + subagent.status = candidate.session.status subagent.currentTool = candidate.session.currentTool subagent.toolDescription = candidate.session.toolDescription ?? metadata.agentNickname if candidate.session.lastActivity > subagent.lastActivity { @@ -3092,11 +3463,22 @@ final class AppState { @discardableResult private func separateMergedCodexSubagents() -> Bool { let parentCandidates = sessions.map { (sessionId: $0.key, session: $0.value) } + let childIds = Set(parentCandidates.flatMap { $0.session.subagents.keys }) + // One DB connection for the whole mode transition; never open SQLite + // once per child (each open carries a busy timeout). + let databaseRecords = Self.codexSpawnEdgeRecords(threadIds: childIds) var didMutate = false for parent in parentCandidates where parent.session.source == "codex" && !parent.session.subagents.isEmpty { for (agentId, subagent) in parent.session.subagents { - let childKey = findSessionId(providerSessionId: agentId) ?? agentId + let childIsDesktop = parent.session.termBundleId == Self.codexAppBundleId + let childKey = findCodexParentSessionId( + providerSessionId: agentId, + childTermBundleId: parent.session.termBundleId + ) ?? Self.codexDiscoveryIdentity( + rawSessionId: agentId, + isDesktop: childIsDesktop + ).sessionId guard childKey != parent.sessionId else { continue } var child = sessions[childKey] ?? SessionSnapshot(startTime: subagent.startTime) @@ -3122,16 +3504,27 @@ final class AppState { child.toolDescription = subagent.toolDescription ?? subagent.agentType child.lastActivity = subagent.lastActivity - let metadata = Self.codexSubagentMetadata(threadId: agentId, transcriptPath: child.transcriptPath) + let transcriptMetadata = child.transcriptPath.flatMap { + Self.codexSubagentMetadata(inTranscriptPath: $0) + } + let metadata = transcriptMetadata ?? databaseRecords[agentId]?.metadata if child.sessionTitle == nil { child.sessionTitle = metadata?.agentNickname ?? subagent.toolDescription ?? subagent.agentType } if child.transcriptPath == nil { - child.transcriptPath = Self.codexTranscriptPath( - sessionId: agentId, - cwd: parent.session.cwd, - processStart: parent.session.cliStartTime - ) + if let statePath = databaseRecords[agentId]?.transcriptPath, + FileManager.default.fileExists(atPath: statePath) { + child.transcriptPath = statePath + } else if let cwd = parent.session.cwd { + let base = FileManager.default.homeDirectoryForCurrentUser + .appendingPathComponent(".codex/sessions").path + child.transcriptPath = Self.findRecentCodexSession( + base: base, + cwd: cwd, + after: parent.session.cliStartTime, + fm: .default + ) + } } if let transcriptPath = child.transcriptPath { let (model, messages) = Self.readRecentFromCodexTranscript(path: transcriptPath) @@ -3288,7 +3681,10 @@ final class AppState { return true } - guard let parentKey = findSessionId(providerSessionId: parentSessionId) else { + guard let parentKey = findCodexParentSessionId( + providerSessionId: parentSessionId, + childTermBundleId: info.termBundleId + ) else { return false } @@ -3296,25 +3692,33 @@ final class AppState { removeSession(info.sessionId) } - if info.subagentStatus?.lowercased() == "closed" { - sessions[parentKey]?.subagents.removeValue(forKey: info.sessionId) + let providerSessionId = info.providerSessionId ?? info.sessionId + + if info.subagentStatus?.lowercased() == "closed" || info.status == .idle { + if sessions[parentKey]?.subagents.removeValue(forKey: providerSessionId) != nil, + sessions[parentKey]?.subagents.isEmpty == true { + clearSubagentProjection(fromParentSession: parentKey) + } return true } let agentType = info.agentType ?? info.agentNickname ?? "Agent" - var subagent = sessions[parentKey]?.subagents[info.sessionId] - ?? SubagentState(agentId: info.sessionId, agentType: agentType) + var subagent = sessions[parentKey]?.subagents[providerSessionId] + ?? SubagentState(agentId: providerSessionId, agentType: agentType) subagent.status = .running subagent.toolDescription = info.agentNickname - subagent.lastActivity = Date() - sessions[parentKey]?.subagents[info.sessionId] = subagent + subagent.lastActivity = info.modifiedAt + sessions[parentKey]?.subagents[providerSessionId] = subagent if sessions[parentKey]?.status != .waitingApproval && sessions[parentKey]?.status != .waitingQuestion { sessions[parentKey]?.status = .running sessions[parentKey]?.currentTool = "Agent" sessions[parentKey]?.toolDescription = info.agentNickname ?? agentType } - sessions[parentKey]?.lastActivity = Date() + sessions[parentKey]?.lastActivity = max( + sessions[parentKey]?.lastActivity ?? .distantPast, + info.modifiedAt + ) activeSessionId = parentKey return true } @@ -3328,6 +3732,9 @@ final class AppState { discoveryScanTask?.cancel() discoveryScanTask = nil pendingDiscoveryRescan = false + codexDesktopDiscoveryScanTask?.cancel() + codexDesktopDiscoveryScanTask = nil + lastCodexDesktopDiscoveryPollAt = nil for key in Array(processMonitors.keys) { stopMonitor(key) } } @@ -3368,12 +3775,14 @@ final class AppState { saveTimer?.invalidate() tearDownProjectsWatcher() discoveryScanTask?.cancel() + codexDesktopDiscoveryScanTask?.cancel() + codexAppServerReconnectTask?.cancel() for (_, monitor) in processMonitors { monitor.source.cancel() } } - private struct DiscoveredSession { + struct DiscoveredSession { let sessionId: String let cwd: String let tty: String? @@ -3390,6 +3799,12 @@ final class AppState { var subagentStatus: String? = nil var agentType: String? = nil var agentNickname: String? = nil + /// Optional status inferred by a provider-specific discovery path. + var status: AgentStatus? = nil + /// Native host bundle when discovery represents an app session. + var termBundleId: String? = nil + /// Provider's raw thread ID when the tracking key uses a namespace prefix. + var providerSessionId: String? = nil } /// Find running `claude` processes, match to transcript files, extract recent messages @@ -4679,6 +5094,28 @@ final class AppState { return statement } + private nonisolated static func sqliteTableColumns( + db: OpaquePointer, + tableName: String + ) -> Set { + guard tableName.allSatisfy({ $0.isLetter || $0.isNumber || $0 == "_" }), + let statement = prepareSQLiteStatement( + db: db, + sql: "PRAGMA table_info(\(tableName));" + ) else { + return [] + } + defer { sqlite3_finalize(statement) } + + var columns = Set() + while sqlite3_step(statement) == SQLITE_ROW { + if let name = sqliteColumnString(statement, index: 1) { + columns.insert(name) + } + } + return columns + } + private nonisolated static func bindSQLiteText(_ text: String, to statement: OpaquePointer, index: Int32) { let transient = unsafeBitCast(-1, to: sqlite3_destructor_type.self) _ = text.withCString { pointer in @@ -4844,6 +5281,68 @@ final class AppState { return cwd == nil || cwd?.trimmingCharacters(in: .whitespacesAndNewlines) == "/" } + /// Keep all three Codex Desktop discovery channels on one stable tracking + /// key while leaving CLI sessions in the provider's raw namespace. + nonisolated static func codexDiscoveryIdentity( + rawSessionId: String, + isDesktop: Bool + ) -> CodexDiscoveryIdentity { + if isDesktop { + return CodexDiscoveryIdentity( + sessionId: codexAppSessionPrefix + rawSessionId, + providerSessionId: rawSessionId, + termBundleId: codexAppBundleId + ) + } + return CodexDiscoveryIdentity( + sessionId: rawSessionId, + providerSessionId: nil, + termBundleId: nil + ) + } + + /// Upgrade pre-namespace persisted Desktop hook cards without touching raw + /// Codex CLI ids. `termBundleId` is the reliable discriminator retained by + /// #267-era snapshots. + nonisolated static func canonicalRestoredCodexSessionId( + sessionId: String, + source: String, + providerSessionId: String?, + termBundleId: String? + ) -> String { + guard source == "codex", termBundleId == codexAppBundleId else { + return sessionId + } + if sessionId.hasPrefix(codexAppSessionPrefix) { + return sessionId + } + let rawId = providerSessionId?.isEmpty == false ? providerSessionId! : sessionId + return codexAppSessionPrefix + rawId + } + + /// Read only the rollout's first `session_meta` row to distinguish Desktop + /// rollouts from CLI rollouts when both exist under `~/.codex/sessions`. + /// Unknown future shapes remain eligible for the owning process as a + /// best-effort fallback instead of being dropped. + nonisolated static func codexTranscriptOrigin(path: String) -> CodexTranscriptOrigin { + guard let firstLine = readFirstLine(path: path), + let data = firstLine.data(using: .utf8), + let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let payload = json["payload"] as? [String: Any] else { + return .unknown + } + + let originator = (payload["originator"] as? String)?.lowercased() ?? "" + let source = (payload["source"] as? String)?.lowercased() ?? "" + if originator.contains("desktop") || source == "vscode" || source == "appserver" { + return .desktop + } + if originator.contains("cli") || source == "cli" || source == "exec" { + return .cli + } + return .unknown + } + private nonisolated static func findCodexPids(candidatePids: [pid_t]? = nil) -> [pid_t] { var codexPids: [pid_t] = [] @@ -4904,21 +5403,247 @@ final class AppState { return args } + /// Scan a rollout backwards in chunks until the latest lifecycle marker is + /// found. This avoids both a full-file read and a fixed-tail limit while + /// reusing `JSONLTailer`'s canonical Codex event interpretation. + nonisolated static func latestCodexTurnStatus( + path: String, + chunkSize: UInt64 = 256 * 1024, + maxBytesToScan: UInt64 = 64 * 1024 * 1024, + maxLineBytes: Int = 4 * 1024 * 1024 + ) -> ConversationTurnStatus? { + guard chunkSize > 0, maxBytesToScan > 0, maxLineBytes > 0 else { return nil } + guard let handle = FileHandle(forReadingAtPath: path) else { return nil } + defer { handle.closeFile() } + + let fileSize = handle.seekToEndOfFile() + let lowerBound = fileSize > maxBytesToScan ? fileSize - maxBytesToScan : 0 + var cursor = fileSize + var leadingFragment = Data() + var discardingOversizedLine = false + + while cursor > lowerBound { + let readSize = min(cursor - lowerBound, chunkSize) + cursor -= readSize + handle.seek(toFileOffset: cursor) + let chunk = handle.readData(ofLength: Int(readSize)) + var combined: Data + if discardingOversizedLine { + // We are walking backwards through one over-budget line. Drop + // its bytes until the previous newline, then resume with older + // complete lines instead of retaining/copying an unbounded blob. + guard let newline = chunk.lastIndex(of: 0x0A) else { continue } + combined = Data(chunk[...newline]) + discardingOversizedLine = false + } else { + combined = chunk + combined.append(leadingFragment) + } + + var lines: [Data] = [] + var lineStart = combined.startIndex + for index in combined.indices where combined[index] == 0x0A { + lines.append(Data(combined[lineStart.. lowerBound, !lines.isEmpty { + let fragment = lines.removeFirst() + if fragment.count > maxLineBytes { + leadingFragment.removeAll(keepingCapacity: true) + discardingOversizedLine = true + } else { + leadingFragment = fragment + } + } else { + leadingFragment.removeAll(keepingCapacity: true) + // A capped scan begins in the middle of an unknown line. Never + // parse that boundary fragment as if it were complete JSON. + if lowerBound > 0, !lines.isEmpty { + lines.removeFirst() + } + } + + for line in lines.reversed() + where !line.isEmpty && line.count <= maxLineBytes { + if let status = codexTurnStatus(fromCompleteLine: line) { + return status + } + } + } + + return nil + } + + private nonisolated static func codexTurnStatus(fromCompleteLine line: Data) -> ConversationTurnStatus? { + var newlineTerminated = line + newlineTerminated.append(0x0A) + return JSONLTailer.latestTurnStatus(in: newlineTerminated) + } + + /// Load recently touched Codex Desktop threads from Codex's own state DB. + /// Desktop app-server processes use `/` as CWD in current releases, making + /// the process-to-project matcher insufficient on its own. + nonisolated static func recentCodexDesktopThreadRecords( + statePath overrideStatePath: String? = nil, + now: Date = Date(), + freshnessWindow: TimeInterval = 600, + completionSettleWindow: TimeInterval = 30, + fileManager: FileManager = .default + ) -> [CodexDesktopThreadRecord] { + let statePath = overrideStatePath ?? FileManager.default.homeDirectoryForCurrentUser + .appendingPathComponent(".codex/state_5.sqlite").path + // `has_user_event` is not a reliable visibility bit for Desktop threads: + // current ChatGPT/Codex builds may keep it at 0 while a user turn is + // actively writing to the rollout. Pull a slightly wider DB candidate + // window, then enforce the tighter freshness window from DB/file mtimes. + let candidateWindow = max(freshnessWindow, 60 * 60) + let cutoff = now.addingTimeInterval(-candidateWindow).timeIntervalSince1970 + + return withSQLiteDatabase(at: statePath) { db in + let columns = sqliteTableColumns(db: db, tableName: "threads") + let requiredColumns: Set = [ + "id", "rollout_path", "cwd", "archived", "source", + ] + guard requiredColumns.isSubset(of: columns) else { return [] } + + let updatedAtExpression: String + if columns.contains("updated_at_ms"), columns.contains("updated_at") { + updatedAtExpression = "COALESCE(NULLIF(updated_at_ms, 0) / 1000.0, updated_at)" + } else if columns.contains("updated_at_ms") { + updatedAtExpression = "updated_at_ms / 1000.0" + } else if columns.contains("updated_at") { + updatedAtExpression = "updated_at" + } else { + return [] + } + let modelExpression = columns.contains("model") ? "model" : "NULL" + let spawnColumns = sqliteTableColumns(db: db, tableName: "thread_spawn_edges") + let hasSpawnStatus = Set(["child_thread_id", "status"]).isSubset(of: spawnColumns) + let spawnJoin = hasSpawnStatus + ? "LEFT JOIN thread_spawn_edges edge ON edge.child_thread_id = thread.id" + : "" + let spawnStatusExpression = hasSpawnStatus ? "edge.status" : "NULL" + + guard let statement = prepareSQLiteStatement( + db: db, + sql: """ + SELECT thread.id, thread.rollout_path, thread.cwd, + \(updatedAtExpression), \(modelExpression), + \(spawnStatusExpression), thread.source + FROM threads AS thread + \(spawnJoin) + WHERE thread.archived = 0 + AND \(updatedAtExpression) >= ? + AND ( + thread.source = 'vscode' + OR thread.source = 'appServer' + OR thread.source LIKE '{"subagent"%' + ) + ORDER BY \(updatedAtExpression) DESC + LIMIT 50; + """ + ) else { return [] } + defer { sqlite3_finalize(statement) } + sqlite3_bind_double(statement, 1, cutoff) + + var records: [CodexDesktopThreadRecord] = [] + while sqlite3_step(statement) == SQLITE_ROW { + guard let sessionId = sqliteColumnString(statement, index: 0), + let transcriptPath = sqliteColumnString(statement, index: 1), + let cwd = sqliteColumnString(statement, index: 2), + fileManager.fileExists(atPath: transcriptPath) else { continue } + // JSON `source` rows can also belong to a concurrent Codex CLI + // subagent. Never re-key an explicitly CLI-origin rollout into + // the Desktop namespace merely because the Desktop host is open. + let transcriptOrigin = codexTranscriptOrigin(path: transcriptPath) + let databaseSource = sqliteColumnString(statement, index: 6) ?? "" + if databaseSource.hasPrefix("{\"subagent") { + // JSON source alone does not identify the owning surface. + // Require positive Desktop provenance before namespacing it. + guard transcriptOrigin == .desktop else { continue } + } else { + guard transcriptOrigin != .cli else { continue } + } + + let dbUpdatedAt = Date(timeIntervalSince1970: sqlite3_column_double(statement, 3)) + let fileModifiedAt = (try? fileManager.attributesOfItem(atPath: transcriptPath))?[.modificationDate] as? Date + let modifiedAt = max(dbUpdatedAt, fileModifiedAt ?? dbUpdatedAt) + guard modifiedAt >= now.addingTimeInterval(-freshnessWindow) else { continue } + + let dbModel = sqliteColumnString(statement, index: 4) + let (transcriptModel, messages) = readRecentFromCodexTranscript(path: transcriptPath) + let turnStatus = latestCodexTurnStatus(path: transcriptPath) + let isFresh = now.timeIntervalSince(modifiedAt) <= 300 + let isActive = isFresh && turnStatus == .processing + if !isActive, + now.timeIntervalSince(modifiedAt) > completionSettleWindow { + continue + } + + records.append(CodexDesktopThreadRecord( + sessionId: sessionId, + cwd: cwd, + model: dbModel ?? transcriptModel, + modifiedAt: modifiedAt, + recentMessages: messages, + transcriptPath: transcriptPath, + status: isActive ? .processing : .idle, + subagentMetadata: codexSubagentMetadata(inTranscriptPath: transcriptPath), + subagentStatus: sqliteColumnString(statement, index: 5) + )) + } + return records + } ?? [] + } + /// Find active Codex sessions by matching running processes to session files private nonisolated static func findActiveCodexSessions(candidatePids: [pid_t]? = nil) -> [DiscoveredSession] { let codexPids = findCodexPids(candidatePids: candidatePids) guard !codexPids.isEmpty else { return [] } + let processes = codexPids.map { pid in + CodexProcessDiscoveryCandidate( + pid: pid, + cwd: getCwd(for: pid), + startTime: getProcessStartTime(pid), + isDesktop: executablePath(for: pid).map(isCodexExecutablePath) ?? false + ) + } + let home = FileManager.default.homeDirectoryForCurrentUser.path - let fm = FileManager.default let sessionsBase = "\(home)/.codex/sessions" - guard fm.fileExists(atPath: sessionsBase) else { return [] } + let statePath = "\(home)/.codex/state_5.sqlite" + return discoverCodexSessions( + processes: processes, + sessionsBase: sessionsBase, + statePath: statePath + ) + } + /// Filesystem/state-DB portion of Codex discovery, split from process + /// inspection so root-CWD fallback and DB degradation can be tested without + /// relying on live PIDs. + nonisolated static func discoverCodexSessions( + processes: [CodexProcessDiscoveryCandidate], + sessionsBase: String, + statePath: String, + now: Date = Date(), + fileManager fm: FileManager = .default + ) -> [DiscoveredSession] { var results: [DiscoveredSession] = [] - var seenSessionIds: Set = [] - - for pid in codexPids { - let processCwd = getCwd(for: pid) + var seenDiscoveryIds: Set = [] + var seenDesktopSessionIds: Set = [] + var claimedUnknownSessionIds: Set = [] + + // Prefer the native host for unknown-origin root rollouts. Explicit CLI + // and Desktop metadata is filtered below, so concurrent modes remain + // distinct even when they share a project directory. + let orderedProcesses = processes.sorted { + $0.isDesktop && !$1.isDesktop + } + for process in orderedProcesses { + let processCwd = process.cwd let useTranscriptCwd = codexDiscoveryUsesTranscriptCwd(processCwd: processCwd) if !useTranscriptCwd, let processCwd, @@ -4926,84 +5651,196 @@ final class AppState { continue } - let processStart = getProcessStartTime(pid) - // Codex stores sessions in date-based dirs: ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl. // A terminal process maps to one cwd; the shared Desktop app-server // may own several sessions, so inspect all fresh rollouts instead. + let candidates = fm.fileExists(atPath: sessionsBase) + ? findRecentCodexSessions(base: sessionsBase, after: process.startTime, fm: fm) + : [] let files: [String] if useTranscriptCwd { - files = findRecentCodexSessions(base: sessionsBase, after: processStart, fm: fm) - } else if let processCwd, - let bestFile = findRecentCodexSession( - base: sessionsBase, - cwd: processCwd, - after: processStart, - fm: fm - ) { - files = [bestFile] + files = candidates.filter { file in + switch codexTranscriptOrigin(path: file) { + case .desktop: return process.isDesktop + case .cli: return !process.isDesktop + case .unknown: return true + } + } + } else if let processCwd { + files = Array(candidates.filter { file in + guard codexSessionMatchesCwd(path: file, cwd: processCwd) else { return false } + switch codexTranscriptOrigin(path: file) { + case .desktop: return process.isDesktop + case .cli: return !process.isDesktop + case .unknown: return true + } + }.prefix(1)) } else { files = [] } for file in files { let fileName = (file as NSString).lastPathComponent - let sessionId = extractCodexSessionId(from: fileName) - guard !sessionId.isEmpty, !seenSessionIds.contains(sessionId) else { continue } - seenSessionIds.insert(sessionId) + let rawSessionId = extractCodexSessionId(from: fileName) + guard !rawSessionId.isEmpty else { continue } + let identity = codexDiscoveryIdentity( + rawSessionId: rawSessionId, + isDesktop: process.isDesktop + ) + guard !seenDiscoveryIds.contains(identity.sessionId) else { continue } + let transcriptOrigin = codexTranscriptOrigin(path: file) + if transcriptOrigin == .unknown, + claimedUnknownSessionIds.contains(rawSessionId) { + continue + } - let modifiedAt = (try? fm.attributesOfItem(atPath: file))?[.modificationDate] as? Date ?? Date() - let codexFreshnessLimit: TimeInterval = processStart != nil ? -300 : -30 - if modifiedAt.timeIntervalSinceNow < codexFreshnessLimit { continue } + let modifiedAt = (try? fm.attributesOfItem(atPath: file))?[.modificationDate] as? Date ?? now + let codexFreshnessLimit: TimeInterval = process.startTime != nil ? -300 : -30 + if modifiedAt.timeIntervalSince(now) < codexFreshnessLimit { continue } - let sessionCwd = codexSessionCwd(path: file) ?? processCwd + let transcriptCwd = codexSessionCwd(path: file) + let sessionCwd = useTranscriptCwd ? transcriptCwd : (transcriptCwd ?? processCwd) guard let sessionCwd, !sessionCwd.isEmpty, !isSubagentWorktree(sessionCwd) else { continue } let (model, messages) = readRecentFromCodexTranscript(path: file) let subagentMetadata = codexSubagentMetadata(inTranscriptPath: file) + let turnStatus = latestCodexTurnStatus(path: file) results.append(DiscoveredSession( - sessionId: sessionId, + sessionId: identity.sessionId, cwd: sessionCwd, tty: nil, model: model, - pid: pid, + pid: process.pid, modifiedAt: modifiedAt, recentMessages: messages, source: "codex", transcriptPath: file, parentSessionId: subagentMetadata?.parentThreadId, - subagentStatus: codexThreadSpawnStatus(childThreadId: sessionId), + subagentStatus: nil, agentType: subagentMetadata?.agentType, - agentNickname: subagentMetadata?.agentNickname + agentNickname: subagentMetadata?.agentNickname, + status: turnStatus.map { $0 == .processing ? .processing : .idle }, + termBundleId: identity.termBundleId, + providerSessionId: identity.providerSessionId )) + // Only accepted candidates exclude a matching DB record. A stale + // or malformed rollout must not hide a valid state-backed thread. + seenDiscoveryIds.insert(identity.sessionId) + if transcriptOrigin == .unknown { + claimedUnknownSessionIds.insert(rawSessionId) + } + if process.isDesktop { + seenDesktopSessionIds.insert(rawSessionId) + } } } + + // Only actual active subagents need the optional spawn-edge status. + // Resolve all of them through one read-only SQLite connection; terminal + // transcript state already wins without consulting the database. + let activeSubagentIds = Set(results.compactMap { result -> String? in + guard result.parentSessionId != nil, result.status != .idle else { return nil } + return result.providerSessionId ?? result.sessionId + }) + if !activeSubagentIds.isEmpty { + let spawnRecords = codexSpawnEdgeRecords( + threadIds: activeSubagentIds, + statePath: statePath + ) + for index in results.indices { + let providerSessionId = results[index].providerSessionId ?? results[index].sessionId + results[index].subagentStatus = spawnRecords[providerSessionId]?.status + } + } + + // Native Codex Desktop / ChatGPT sessions cannot be mapped by process + // CWD because its app-server runs from `/`. Hydrate recent threads from + // state_5.sqlite and let transcript activity drive running/idle state. + if processes.contains(where: { $0.isDesktop }) { + results.append(contentsOf: findRecentCodexDesktopSessions( + excluding: seenDesktopSessionIds, + statePath: statePath, + now: now, + fileManager: fm + )) + } return results } - nonisolated static func codexSubagentMetadata(inTranscriptPath path: String) -> CodexSubagentMetadata? { + private nonisolated static func findRecentCodexDesktopSessions( + excluding excludedSessionIds: Set = [], + statePath: String? = nil, + now: Date = Date(), + fileManager: FileManager = .default + ) -> [DiscoveredSession] { + let resolvedStatePath = statePath ?? FileManager.default.homeDirectoryForCurrentUser + .appendingPathComponent(".codex/state_5.sqlite").path + + return recentCodexDesktopThreadRecords( + statePath: resolvedStatePath, + now: now, + fileManager: fileManager + ).compactMap { record in + guard !excludedSessionIds.contains(record.sessionId) else { return nil } + let identity = codexDiscoveryIdentity(rawSessionId: record.sessionId, isDesktop: true) + return DiscoveredSession( + sessionId: identity.sessionId, + cwd: record.cwd, + tty: nil, + model: record.model, + pid: nil, + modifiedAt: record.modifiedAt, + recentMessages: record.recentMessages, + source: "codex", + transcriptPath: record.transcriptPath, + parentSessionId: record.subagentMetadata?.parentThreadId, + subagentStatus: record.subagentStatus, + agentType: record.subagentMetadata?.agentType, + agentNickname: record.subagentMetadata?.agentNickname, + status: record.status, + termBundleId: identity.termBundleId, + providerSessionId: identity.providerSessionId + ) + } + } + + private nonisolated static func inspectCodexSubagentMetadata( + inTranscriptPath path: String + ) -> CodexTranscriptSubagentInspection { guard let firstLine = readFirstLine(path: path), let data = firstLine.data(using: .utf8), let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any], - let payload = json["payload"] as? [String: Any], - let source = payload["source"] as? [String: Any], + json["type"] as? String == "session_meta", + let payload = json["payload"] as? [String: Any] else { + return .unavailable + } + + guard let source = payload["source"] as? [String: Any], let subagent = source["subagent"] as? [String: Any], !subagent.isEmpty else { - return nil + return .root } - let parent = firstStringRecursively(in: subagent, key: "parent_thread_id") - guard let parent, !parent.isEmpty else { return nil } + guard let parent, !parent.isEmpty else { return .unavailable } let agentType = firstStringRecursively(in: subagent, key: "agent_role") ?? subagent.keys.sorted().first let nickname = firstStringRecursively(in: subagent, key: "agent_nickname") - return CodexSubagentMetadata( + return .subagent(CodexSubagentMetadata( parentThreadId: parent, agentType: agentType, agentNickname: nickname - ) + )) + } + + nonisolated static func codexSubagentMetadata(inTranscriptPath path: String) -> CodexSubagentMetadata? { + guard case .subagent(let metadata) = inspectCodexSubagentMetadata( + inTranscriptPath: path + ) else { + return nil + } + return metadata } nonisolated static func codexSubagentMetadata( @@ -5011,46 +5848,99 @@ final class AppState { transcriptPath: String?, statePath overrideStatePath: String? = nil ) -> CodexSubagentMetadata? { - if let transcriptPath, - let metadata = codexSubagentMetadata(inTranscriptPath: transcriptPath) { - return metadata + if let transcriptPath { + switch inspectCodexSubagentMetadata(inTranscriptPath: transcriptPath) { + case .subagent(let metadata): + return metadata + case .root: + return nil + case .unavailable: + break + } } + return codexSpawnEdgeRecords( + threadIds: [threadId], + statePath: overrideStatePath + )[threadId]?.metadata + } + + /// Resolve spawn relations/statuses in one read-only connection. Queries are + /// chunked to keep SQLite variable counts bounded when restoring many cards. + private nonisolated static func codexSpawnEdgeRecords( + threadIds: Set, + statePath overrideStatePath: String? = nil + ) -> [String: CodexSpawnEdgeRecord] { + guard !threadIds.isEmpty else { return [:] } let statePath = overrideStatePath ?? { let home = FileManager.default.homeDirectoryForCurrentUser.path return "\(home)/.codex/state_5.sqlite" }() return withSQLiteDatabase(at: statePath) { db in - let sql = """ - SELECT e.parent_thread_id, t.agent_role, t.agent_nickname, t.source - FROM thread_spawn_edges e - LEFT JOIN threads t ON t.id = e.child_thread_id - WHERE e.child_thread_id = ? - LIMIT 1 - """ - guard let statement = prepareSQLiteStatement(db: db, sql: sql) else { return nil } - defer { sqlite3_finalize(statement) } - bindSQLiteText(threadId, to: statement, index: 1) - guard sqlite3_step(statement) == SQLITE_ROW, - let parent = nonEmptySQLiteColumnString(statement, index: 0) else { - return nil - } - - var agentType = nonEmptySQLiteColumnString(statement, index: 1) - var nickname = nonEmptySQLiteColumnString(statement, index: 2) - if let source = nonEmptySQLiteColumnString(statement, index: 3), - let data = source.data(using: .utf8), - let json = try? JSONSerialization.jsonObject(with: data) { - agentType = agentType ?? firstStringRecursively(in: json, key: "agent_role") - nickname = nickname ?? firstStringRecursively(in: json, key: "agent_nickname") + let edgeColumns = sqliteTableColumns(db: db, tableName: "thread_spawn_edges") + guard Set(["parent_thread_id", "child_thread_id", "status"]).isSubset(of: edgeColumns) else { + return [:] + } + let threadColumns = sqliteTableColumns(db: db, tableName: "threads") + let canJoinThreads = threadColumns.contains("id") + let roleExpression = canJoinThreads && threadColumns.contains("agent_role") + ? "thread.agent_role" : "NULL" + let nicknameExpression = canJoinThreads && threadColumns.contains("agent_nickname") + ? "thread.agent_nickname" : "NULL" + let sourceExpression = canJoinThreads && threadColumns.contains("source") + ? "thread.source" : "NULL" + let transcriptExpression = canJoinThreads && threadColumns.contains("rollout_path") + ? "thread.rollout_path" : "NULL" + let join = canJoinThreads + ? "LEFT JOIN threads AS thread ON thread.id = edge.child_thread_id" + : "" + + let orderedIds = threadIds.sorted() + let chunkSize = 200 + var records: [String: CodexSpawnEdgeRecord] = [:] + for start in stride(from: 0, to: orderedIds.count, by: chunkSize) { + let end = min(start + chunkSize, orderedIds.count) + let chunk = Array(orderedIds[start.. String? { @@ -5101,19 +5991,6 @@ final class AppState { return value } - private nonisolated static func codexThreadSpawnStatus(childThreadId: String) -> String? { - let home = FileManager.default.homeDirectoryForCurrentUser.path - let statePath = "\(home)/.codex/state_5.sqlite" - return withSQLiteDatabase(at: statePath) { db in - let sql = "SELECT status FROM thread_spawn_edges WHERE child_thread_id = ? LIMIT 1" - guard let statement = prepareSQLiteStatement(db: db, sql: sql) else { return nil } - defer { sqlite3_finalize(statement) } - bindSQLiteText(childThreadId, to: statement, index: 1) - guard sqlite3_step(statement) == SQLITE_ROW else { return nil } - return sqliteColumnString(statement, index: 0) - } - } - /// Find the most recent Codex session file matching a CWD /// Scans back up to 7 days to cover long-running sessions that span day boundaries private nonisolated static func findRecentCodexSession(base: String, cwd: String, after: Date?, fm: FileManager) -> String? { diff --git a/Sources/CodeIsland/HookServer.swift b/Sources/CodeIsland/HookServer.swift index 0f26f709..146ceba6 100644 --- a/Sources/CodeIsland/HookServer.swift +++ b/Sources/CodeIsland/HookServer.swift @@ -320,7 +320,7 @@ class HookServer { return AppState.codexSubagentMetadata(inTranscriptPath: path) } - private func codexNativeSubsessionParentId(from raw: [String: Any]) -> String? { + func codexNativeSubsessionParentId(from raw: [String: Any]) -> String? { guard (raw["_via_plugin"] as? Bool) != true, SessionSnapshot.normalizedSupportedSource(raw["_source"] as? String) == "codex", let childSessionId = Self.rawSessionId(from: raw) else { @@ -328,7 +328,10 @@ class HookServer { } if let metadata = Self.codexSubagentMetadata(from: raw), - let parentSessionId = appState.findSessionId(providerSessionId: metadata.parentThreadId) { + let parentSessionId = appState.findCodexParentSessionId( + providerSessionId: metadata.parentThreadId, + childTermBundleId: raw["_term_bundle"] as? String + ) { return parentSessionId } diff --git a/Sources/CodeIslandCore/JSONLTailer.swift b/Sources/CodeIslandCore/JSONLTailer.swift index 04252b67..2b71b8f7 100644 --- a/Sources/CodeIslandCore/JSONLTailer.swift +++ b/Sources/CodeIslandCore/JSONLTailer.swift @@ -31,6 +31,15 @@ public struct ConversationTailDelta: Equatable, Sendable { public let turnStatus: ConversationTurnStatus? public let hasActivity: Bool public let cursorQuestion: CursorQuestionSignal? + /// Identifies the exact tailer attachment that produced this delta. + /// + /// AppState uses this to reject a callback that crossed an actor hop after + /// the session was detached and re-created with the same provider id. + /// Optional for source compatibility with synthetic/test deltas. + public let attachmentToken: UUID? + /// File path paired with `attachmentToken`; an additional guard against + /// applying a delta after the session switched rollout files. + public let filePath: String? public init( sessionId: String, @@ -38,7 +47,9 @@ public struct ConversationTailDelta: Equatable, Sendable { lastAssistantMessage: String?, turnStatus: ConversationTurnStatus? = nil, hasActivity: Bool = false, - cursorQuestion: CursorQuestionSignal? = nil + cursorQuestion: CursorQuestionSignal? = nil, + attachmentToken: UUID? = nil, + filePath: String? = nil ) { self.sessionId = sessionId self.lastUserPrompt = lastUserPrompt @@ -46,6 +57,8 @@ public struct ConversationTailDelta: Equatable, Sendable { self.turnStatus = turnStatus self.hasActivity = hasActivity self.cursorQuestion = cursorQuestion + self.attachmentToken = attachmentToken + self.filePath = filePath } /// A delta only carries signal when at least one field is non-nil. @@ -77,8 +90,19 @@ public final class JSONLTailer: @unchecked Sendable { var inode: ino_t var pendingFragment: Data var source: DispatchSourceFileSystemObject - - init(sessionId: String, filePath: String, fd: Int32, offset: off_t, inode: ino_t, source: DispatchSourceFileSystemObject) { + let generation: UInt64 + let attachmentToken: UUID + + init( + sessionId: String, + filePath: String, + fd: Int32, + offset: off_t, + inode: ino_t, + source: DispatchSourceFileSystemObject, + generation: UInt64, + attachmentToken: UUID + ) { self.sessionId = sessionId self.filePath = filePath self.fd = fd @@ -86,18 +110,25 @@ public final class JSONLTailer: @unchecked Sendable { self.inode = inode self.pendingFragment = Data() self.source = source + self.generation = generation + self.attachmentToken = attachmentToken } } private let queue: DispatchQueue private let onDelta: DeltaHandler + private let replacementReattachDelay: DispatchTimeInterval private var watches: [String: Watch] = [:] + private var desiredFilePaths: [String: String] = [:] + private var generations: [String: UInt64] = [:] public init( queue: DispatchQueue = DispatchQueue(label: "com.codeisland.jsonl-tailer"), + replacementReattachDelay: DispatchTimeInterval = .milliseconds(50), onDelta: @escaping DeltaHandler ) { self.queue = queue + self.replacementReattachDelay = replacementReattachDelay self.onDelta = onDelta } @@ -109,23 +140,41 @@ public final class JSONLTailer: @unchecked Sendable { // MARK: - Public API - public func attach(sessionId: String, filePath: String) { + @discardableResult + public func attach(sessionId: String, filePath: String) -> UUID { + let attachmentToken = UUID() queue.async { [weak self] in - self?.detachOnQueue(sessionId: sessionId) - self?.attachOnQueue(sessionId: sessionId, filePath: filePath, initialOffset: nil) + guard let self else { return } + let generation = self.advanceGenerationOnQueue(sessionId: sessionId) + self.desiredFilePaths[sessionId] = filePath + self.detachOnQueue(sessionId: sessionId) + self.attachOnQueue( + sessionId: sessionId, + filePath: filePath, + initialOffset: nil, + generation: generation, + attachmentToken: attachmentToken + ) } + return attachmentToken } public func detach(sessionId: String) { queue.async { [weak self] in - self?.detachOnQueue(sessionId: sessionId) + guard let self else { return } + self.desiredFilePaths.removeValue(forKey: sessionId) + _ = self.advanceGenerationOnQueue(sessionId: sessionId) + self.detachOnQueue(sessionId: sessionId) } } public func detachAll() { queue.async { [weak self] in guard let self else { return } - for key in Array(self.watches.keys) { + let sessionIds = Set(self.watches.keys).union(self.desiredFilePaths.keys) + self.desiredFilePaths.removeAll() + for key in sessionIds { + _ = self.advanceGenerationOnQueue(sessionId: key) self.detachOnQueue(sessionId: key) } } @@ -137,7 +186,23 @@ public final class JSONLTailer: @unchecked Sendable { // MARK: - Watch lifecycle - private func attachOnQueue(sessionId: String, filePath: String, initialOffset: off_t?) { + private func advanceGenerationOnQueue(sessionId: String) -> UInt64 { + let generation = (generations[sessionId] ?? 0) &+ 1 + generations[sessionId] = generation + return generation + } + + private func attachOnQueue( + sessionId: String, + filePath: String, + initialOffset: off_t?, + generation: UInt64, + attachmentToken: UUID + ) { + guard desiredFilePaths[sessionId] == filePath, + generations[sessionId] == generation else { + return + } let fd = open(filePath, O_RDONLY | O_NONBLOCK) guard fd >= 0 else { return } var fileStat = stat() @@ -158,7 +223,9 @@ public final class JSONLTailer: @unchecked Sendable { fd: fd, offset: offset, inode: fileStat.st_ino, - source: source + source: source, + generation: generation, + attachmentToken: attachmentToken ) source.setEventHandler { [weak self] in @@ -182,6 +249,12 @@ public final class JSONLTailer: @unchecked Sendable { // MARK: - Event handling private func handleEvents(_ events: DispatchSource.FileSystemEvent, watch: Watch) { + guard watches[watch.sessionId] === watch, + desiredFilePaths[watch.sessionId] == watch.filePath, + generations[watch.sessionId] == watch.generation else { + return + } + // A rotate or delete means the file has been replaced underneath us (e.g. /clear). // Re-attach from a fresh fd so future writes reach our handler. if events.contains(.delete) || events.contains(.rename) || events.contains(.revoke) { @@ -189,8 +262,14 @@ public final class JSONLTailer: @unchecked Sendable { let sid = watch.sessionId detachOnQueue(sessionId: sid) // Give the writer a moment to finish writing the new file before we reopen. - queue.asyncAfter(deadline: .now() + .milliseconds(50)) { [weak self] in - self?.attachOnQueue(sessionId: sid, filePath: path, initialOffset: 0) + queue.asyncAfter(deadline: .now() + replacementReattachDelay) { [weak self] in + self?.attachOnQueue( + sessionId: sid, + filePath: path, + initialOffset: 0, + generation: watch.generation, + attachmentToken: watch.attachmentToken + ) } return } @@ -202,7 +281,13 @@ public final class JSONLTailer: @unchecked Sendable { let path = watch.filePath let sid = watch.sessionId detachOnQueue(sessionId: sid) - attachOnQueue(sessionId: sid, filePath: path, initialOffset: 0) + attachOnQueue( + sessionId: sid, + filePath: path, + initialOffset: 0, + generation: watch.generation, + attachmentToken: watch.attachmentToken + ) return } if fileStat.st_size < watch.offset { @@ -233,7 +318,9 @@ public final class JSONLTailer: @unchecked Sendable { lastAssistantMessage: scan.delta.lastAssistantMessage, turnStatus: scan.delta.turnStatus, hasActivity: scan.delta.hasActivity, - cursorQuestion: scan.delta.cursorQuestion + cursorQuestion: scan.delta.cursorQuestion, + attachmentToken: watch.attachmentToken, + filePath: watch.filePath ) onDelta(delta) } diff --git a/Sources/CodeIslandCore/Models.swift b/Sources/CodeIslandCore/Models.swift index b0a46241..94ae4d97 100644 --- a/Sources/CodeIslandCore/Models.swift +++ b/Sources/CodeIslandCore/Models.swift @@ -165,6 +165,15 @@ public struct HookEvent { let remoteHostId = json["_remote_host_id"] as? String, !remoteHostId.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { self.sessionId = "remote:\(remoteHostId):\(rawSessionId)" + } else if let rawSessionId, + (json["_source"] as? String)?.lowercased() == "codex", + json["_term_bundle"] as? String == "com.openai.codex" { + // Codex Desktop hooks, app-server notifications, rollout discovery, + // and state-DB polling must all address one card. Keep the provider's + // raw id in rawJSON for title lookup and persisted providerSessionId. + self.sessionId = rawSessionId.hasPrefix("codexapp:") + ? rawSessionId + : "codexapp:\(rawSessionId)" } else { self.sessionId = rawSessionId } diff --git a/Sources/CodeIslandCore/SessionSnapshot.swift b/Sources/CodeIslandCore/SessionSnapshot.swift index 789b2f23..0070faf6 100644 --- a/Sources/CodeIslandCore/SessionSnapshot.swift +++ b/Sources/CodeIslandCore/SessionSnapshot.swift @@ -1086,9 +1086,16 @@ public func reduceEvent( if let remoteHostName = event.rawJSON["_remote_host_name"] as? String, !remoteHostName.isEmpty { sessions[sessionId]?.remoteHostName = remoteHostName } - if let providerSessionId = event.rawJSON["session_id"] as? String, !providerSessionId.isEmpty, - sessions[sessionId]?.isRemote == true { - sessions[sessionId]?.providerSessionId = providerSessionId + if let providerSessionId = event.rawJSON["session_id"] as? String, + !providerSessionId.isEmpty, + sessions[sessionId]?.isRemote == true + || (sessions[sessionId]?.source == "codex" + && sessions[sessionId]?.termBundleId == "com.openai.codex") { + let isCodexDesktop = sessions[sessionId]?.source == "codex" + && sessions[sessionId]?.termBundleId == "com.openai.codex" + sessions[sessionId]?.providerSessionId = isCodexDesktop + ? normalizedCodexDesktopProviderSessionId(providerSessionId) + : providerSessionId } if let sessionTitle = event.rawJSON["session_title"] as? String, !sessionTitle.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { @@ -1376,10 +1383,16 @@ public func extractMetadata(into sessions: inout [String: SessionSnapshot], sess if let remoteHostName = event.rawJSON["_remote_host_name"] as? String, !remoteHostName.isEmpty { sessions[sessionId]?.remoteHostName = remoteHostName } - if sessions[sessionId]?.isRemote == true, - let providerSessionId = event.rawJSON["session_id"] as? String, - !providerSessionId.isEmpty { - sessions[sessionId]?.providerSessionId = providerSessionId + if let providerSessionId = event.rawJSON["session_id"] as? String, + !providerSessionId.isEmpty, + sessions[sessionId]?.isRemote == true + || (sessions[sessionId]?.source == "codex" + && sessions[sessionId]?.termBundleId == "com.openai.codex") { + let isCodexDesktop = sessions[sessionId]?.source == "codex" + && sessions[sessionId]?.termBundleId == "com.openai.codex" + sessions[sessionId]?.providerSessionId = isCodexDesktop + ? normalizedCodexDesktopProviderSessionId(providerSessionId) + : providerSessionId } if let sessionTitle = event.rawJSON["session_title"] as? String, !sessionTitle.trimmingCharacters(in: .whitespacesAndNewlines).isEmpty { @@ -1387,6 +1400,12 @@ public func extractMetadata(into sessions: inout [String: SessionSnapshot], sess } } +private func normalizedCodexDesktopProviderSessionId(_ sessionId: String) -> String { + let prefix = "codexapp:" + guard sessionId.hasPrefix(prefix) else { return sessionId } + return String(sessionId.dropFirst(prefix.count)) +} + /// Pull a non-empty string from a hook JSON value. /// Supports plain strings and content-part arrays such as Kimi Code CLI's /// `prompt: [{ "type": "text", "text": "..." }]`. diff --git a/Tests/CodeIslandCoreTests/JSONLTailerTests.swift b/Tests/CodeIslandCoreTests/JSONLTailerTests.swift index 08e17693..19bf14d3 100644 --- a/Tests/CodeIslandCoreTests/JSONLTailerTests.swift +++ b/Tests/CodeIslandCoreTests/JSONLTailerTests.swift @@ -251,6 +251,69 @@ final class JSONLTailerTests: XCTestCase { XCTAssertEqual(callCount.value, 1) } + func testDetachCancelsDelayedReattachAfterFileReplacement() throws { + let url = temporaryFileURL() + let backup = url.deletingLastPathComponent() + .appendingPathComponent(url.lastPathComponent + ".old") + try Data("".utf8).write(to: url) + defer { + try? FileManager.default.removeItem(at: url) + try? FileManager.default.removeItem(at: backup) + } + + let tailer = JSONLTailer( + queue: DispatchQueue(label: "tailer-test"), + replacementReattachDelay: .milliseconds(500), + onDelta: { _ in } + ) + tailer.attach(sessionId: "s1", filePath: url.path) + XCTAssertTrue(waitUntil { tailer.activeSessionCount == 1 }) + + try FileManager.default.moveItem(at: url, to: backup) + try Data("".utf8).write(to: url) + // Wait for the rename handler to remove the old watch. The injected + // delay leaves a deterministic window before its reopen block runs. + XCTAssertTrue(waitUntil { tailer.activeSessionCount == 0 }) + tailer.detach(sessionId: "s1") + Thread.sleep(forTimeInterval: 0.6) + + XCTAssertEqual(tailer.activeSessionCount, 0) + } + + func testFileReplacementReattachesWhenSessionRemainsDesired() throws { + let url = temporaryFileURL() + let backup = url.deletingLastPathComponent() + .appendingPathComponent(url.lastPathComponent + ".old") + try Data("".utf8).write(to: url) + defer { + try? FileManager.default.removeItem(at: url) + try? FileManager.default.removeItem(at: backup) + } + + let replacementDelta = expectation(description: "replacement file delta delivered") + let tailer = JSONLTailer( + queue: DispatchQueue(label: "tailer-test"), + replacementReattachDelay: .milliseconds(200), + onDelta: { delta in + if delta.lastAssistantMessage == "after-rotate" { + replacementDelta.fulfill() + } + } + ) + tailer.attach(sessionId: "s1", filePath: url.path) + XCTAssertTrue(waitUntil { tailer.activeSessionCount == 1 }) + + try FileManager.default.moveItem(at: url, to: backup) + try Data("".utf8).write(to: url) + XCTAssertTrue(waitUntil { tailer.activeSessionCount == 0 }) + XCTAssertTrue(waitUntil { tailer.activeSessionCount == 1 }) + try appendToFile(url: url, content: assistantLine(text: "after-rotate") + "\n") + + wait(for: [replacementDelta], timeout: 2) + XCTAssertEqual(tailer.activeSessionCount, 1) + tailer.detach(sessionId: "s1") + } + // MARK: - Integration: offset accounting across partial writes (#278) /// A JSONL line flushed in two chunks (no newline in the first) must be @@ -404,6 +467,19 @@ final class JSONLTailerTests: XCTestCase { try handle.write(contentsOf: Data(content.utf8)) try handle.close() } + + private func waitUntil( + timeout: TimeInterval = 2, + pollInterval: TimeInterval = 0.005, + _ predicate: () -> Bool + ) -> Bool { + let deadline = Date().addingTimeInterval(timeout) + repeat { + if predicate() { return true } + Thread.sleep(forTimeInterval: pollInterval) + } while Date() < deadline + return predicate() + } } private final class LockedValue: @unchecked Sendable { diff --git a/Tests/CodeIslandTests/AppStateCodexAppServerTests.swift b/Tests/CodeIslandTests/AppStateCodexAppServerTests.swift index eff62170..3b594f69 100644 --- a/Tests/CodeIslandTests/AppStateCodexAppServerTests.swift +++ b/Tests/CodeIslandTests/AppStateCodexAppServerTests.swift @@ -47,6 +47,94 @@ final class AppStateCodexAppServerTests: XCTestCase { )) } + func testDesktopHookRolloutAndAppServerUseOneCanonicalCard() throws { + let appState = AppState() + let rawId = "desktop-hook-thread" + let sessionId = AppState.codexAppSessionPrefix + rawId + let transcript = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-hook-\(UUID().uuidString).jsonl") + try Data((#"{"type":"event_msg","payload":{"type":"task_started"}}"# + "\n").utf8) + .write(to: transcript) + defer { try? FileManager.default.removeItem(at: transcript) } + + let hookPayload: [String: Any] = [ + "hook_event_name": "SessionStart", + "session_id": rawId, + "_source": "codex", + "_term_bundle": AppState.codexAppBundleId, + "cwd": "/desktop/repo", + "transcript_path": transcript.path, + ] + let hook = try XCTUnwrap(HookEvent( + from: JSONSerialization.data(withJSONObject: hookPayload) + )) + XCTAssertEqual(hook.sessionId, sessionId) + appState.handleEvent(hook) + XCTAssertEqual(Set(appState.sessions.keys), [sessionId]) + XCTAssertEqual(appState.sessions[sessionId]?.providerSessionId, rawId) + + appState.integrateDiscovered([AppState.DiscoveredSession( + sessionId: sessionId, + cwd: "/desktop/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: Date(), + recentMessages: [], + source: "codex", + transcriptPath: transcript.path, + status: .processing, + termBundleId: AppState.codexAppBundleId, + providerSessionId: rawId + )]) + + let started = try XCTUnwrap(CodexAppServerClient.parseMessage(Data( + """ + {"jsonrpc":"2.0","method":"thread/started","params":{"thread":{"id":"\(rawId)","cwd":"/desktop/repo","path":"\(transcript.path)","status":{"type":"active","activeFlags":[]}}}} + """.utf8 + ))) + appState.handleCodexAppServerMessage(started) + + XCTAssertEqual(Set(appState.sessions.keys), [sessionId]) + XCTAssertNil(appState.sessions[rawId]) + XCTAssertEqual(appState.sessions[sessionId]?.providerSessionId, rawId) + } + + func testCodexCLIHookKeepsRawSessionId() throws { + let payload: [String: Any] = [ + "hook_event_name": "SessionStart", + "session_id": "cli-thread", + "_source": "codex", + "_term_bundle": "com.apple.Terminal", + "cwd": "/repo", + ] + let event = try XCTUnwrap(HookEvent( + from: JSONSerialization.data(withJSONObject: payload) + )) + XCTAssertEqual(event.sessionId, "cli-thread") + } + + func testAlreadyPrefixedDesktopHookKeepsRawProviderIdentifier() throws { + let payload: [String: Any] = [ + "hook_event_name": "SessionStart", + "session_id": "codexapp:desktop-thread", + "_source": "codex", + "_term_bundle": AppState.codexAppBundleId, + "cwd": "/repo", + ] + let event = try XCTUnwrap(HookEvent( + from: JSONSerialization.data(withJSONObject: payload) + )) + let appState = AppState() + appState.handleEvent(event) + + XCTAssertEqual(event.sessionId, "codexapp:desktop-thread") + XCTAssertEqual( + appState.sessions["codexapp:desktop-thread"]?.providerSessionId, + "desktop-thread" + ) + } + func testCodexTranscriptCwdReadsLargeSessionMetaLine() throws { let url = FileManager.default.temporaryDirectory .appendingPathComponent("codeisland-codex-meta-\(UUID().uuidString).jsonl") @@ -107,6 +195,130 @@ final class AppStateCodexAppServerTests: XCTestCase { XCTAssertEqual(resolved?.path, fallbackExecutable.path) } + func testCodexAppServerExitReconnectsOnlyWhileHostAndWatcherRemainActive() { + XCTAssertEqual( + AppState.codexAppServerExitStrategy(hostRunning: true, watcherActive: true), + .reconnectPreservingSessions + ) + XCTAssertEqual( + AppState.codexAppServerExitStrategy(hostRunning: false, watcherActive: true), + .stopAndRemoveSessions + ) + XCTAssertEqual( + AppState.codexAppServerExitStrategy(hostRunning: true, watcherActive: false), + .stopAndRemoveSessions + ) + } + + func testAppServerDisconnectDropsDeadChannelQuestionAndResumesCard() throws { + let appState = AppState() + let request = try XCTUnwrap(CodexAppServerClient.parseMessage(Data( + """ + {"jsonrpc":"2.0","id":"request-1","method":"item/tool/requestUserInput","params":{"threadId":"disconnect-thread","questions":[{"id":"q1","question":"Continue?"}]}} + """.utf8 + ))) + appState.handleCodexAppServerMessage(request) + let sessionId = AppState.codexAppSessionPrefix + "disconnect-thread" + XCTAssertEqual(appState.questionQueue.count, 1) + XCTAssertEqual(appState.sessions[sessionId]?.status, .waitingQuestion) + + appState.invalidateCodexAppServerQuestionsAfterDisconnect() + + XCTAssertTrue(appState.questionQueue.isEmpty) + XCTAssertEqual(appState.sessions[sessionId]?.status, .processing) + } + + func testThreadClosedUsesUnifiedSessionCleanup() throws { + let appState = AppState() + let threadId = "closed-thread" + let sessionId = AppState.codexAppSessionPrefix + threadId + let request = try XCTUnwrap(CodexAppServerClient.parseMessage(Data( + """ + {"jsonrpc":"2.0","id":"request-1","method":"item/tool/requestUserInput","params":{"threadId":"\(threadId)","questions":[{"id":"q1","question":"Continue?"}]}} + """.utf8 + ))) + appState.handleCodexAppServerMessage(request) + + let transcript = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-close-\(UUID().uuidString).jsonl") + try Data().write(to: transcript) + defer { try? FileManager.default.removeItem(at: transcript) } + appState.sessions[sessionId]?.transcriptPath = transcript.path + appState.attachTranscriptTailerIfNeeded(sessionId: sessionId) + + XCTAssertNotNil(appState.sessions[sessionId]) + XCTAssertEqual(appState.questionQueue.count, 1) + XCTAssertEqual(appState.attachedTranscriptPaths[sessionId], transcript.path) + + let closed = try XCTUnwrap(CodexAppServerClient.parseMessage(Data( + #"{"jsonrpc":"2.0","method":"thread/closed","params":{"threadId":"closed-thread"}}"#.utf8 + ))) + appState.handleCodexAppServerMessage(closed) + + XCTAssertNil(appState.sessions[sessionId]) + XCTAssertTrue(appState.questionQueue.isEmpty) + XCTAssertNil(appState.attachedTranscriptPaths[sessionId]) + + // A result already captured by the independent DB poll must not recreate + // the just-closed card. + appState.integrateDiscovered([AppState.DiscoveredSession( + sessionId: sessionId, + cwd: "/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: Date().addingTimeInterval(-60), + recentMessages: [], + source: "codex", + transcriptPath: transcript.path, + status: .idle, + termBundleId: AppState.codexAppBundleId, + providerSessionId: threadId + )]) + XCTAssertNil(appState.sessions[sessionId]) + + // A final terminal DB/file flush can be newer than the close callback; + // idle is not evidence of a resumed generation. + appState.integrateDiscovered([AppState.DiscoveredSession( + sessionId: sessionId, + cwd: "/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: Date().addingTimeInterval(60), + recentMessages: [], + source: "codex", + transcriptPath: transcript.path, + status: .idle, + termBundleId: AppState.codexAppBundleId, + providerSessionId: threadId + )]) + XCTAssertNil(appState.sessions[sessionId]) + + let lateHookPayload: [String: Any] = [ + "hook_event_name": "PostToolUse", + "session_id": threadId, + "_source": "codex", + "_term_bundle": AppState.codexAppBundleId, + "cwd": "/repo", + "transcript_path": transcript.path, + ] + let lateHook = try XCTUnwrap(HookEvent( + from: JSONSerialization.data(withJSONObject: lateHookPayload) + )) + appState.handleEvent(lateHook) + XCTAssertNil(appState.sessions[sessionId]) + + // A real app-server start is a new generation and clears the tombstone. + let restarted = try XCTUnwrap(CodexAppServerClient.parseMessage(Data( + """ + {"jsonrpc":"2.0","method":"thread/started","params":{"thread":{"id":"\(threadId)","cwd":"/repo","path":"\(transcript.path)","status":{"type":"active","activeFlags":[]}}}} + """.utf8 + ))) + appState.handleCodexAppServerMessage(restarted) + XCTAssertNotNil(appState.sessions[sessionId]) + } + func testActiveWithApprovalFlagMapsToWaitingApproval() { var snapshot = SessionSnapshot() snapshot.status = .idle diff --git a/Tests/CodeIslandTests/AppStateCodexSubsessionTests.swift b/Tests/CodeIslandTests/AppStateCodexSubsessionTests.swift index 14db22dc..c94ad98e 100644 --- a/Tests/CodeIslandTests/AppStateCodexSubsessionTests.swift +++ b/Tests/CodeIslandTests/AppStateCodexSubsessionTests.swift @@ -2,9 +2,414 @@ import XCTest @testable import CodeIsland import CodeIslandCore import SQLite3 +import Darwin @MainActor final class AppStateCodexSubsessionTests: XCTestCase { + func testDistinctCodexDesktopProviderThreadsAreNotDeduplicated() { + XCTAssertFalse(AppState.providerSessionIdentifiersMayRepresentSameSession( + existing: "root-thread", + discovered: "child-thread" + )) + } + + func testDesktopProviderThreadRequiresAnExactExistingIdentifier() { + XCTAssertFalse(AppState.providerSessionIdentifiersMayRepresentSameSession( + existing: nil, + discovered: "desktop-thread" + )) + XCTAssertTrue(AppState.providerSessionIdentifiersMayRepresentSameSession( + existing: "desktop-thread", + discovered: "desktop-thread" + )) + } + + func testCodexDesktopAndCLIDiscoveriesAreNotDeduplicated() { + XCTAssertFalse(AppState.discoveryAppModesMayRepresentSameSession( + existingIsNativeAppMode: true, + discoveredTermBundleId: nil + )) + XCTAssertFalse(AppState.discoveryAppModesMayRepresentSameSession( + existingIsNativeAppMode: false, + discoveredTermBundleId: AppState.codexAppBundleId + )) + XCTAssertTrue(AppState.discoveryAppModesMayRepresentSameSession( + existingIsNativeAppMode: true, + discoveredTermBundleId: AppState.codexAppBundleId + )) + XCTAssertTrue(AppState.discoveryAppModesMayRepresentSameSession( + existingIsNativeAppMode: false, + discoveredTermBundleId: nil + )) + } + + func testCodexDesktopDiscoveryPollingRequiresRunningAppAndElapsedInterval() { + let now = Date(timeIntervalSince1970: 1_000) + XCTAssertFalse(AppState.shouldPollCodexDesktopDiscovery( + runningBundleIdentifiers: [], + lastPollAt: nil, + now: now + )) + XCTAssertTrue(AppState.shouldPollCodexDesktopDiscovery( + runningBundleIdentifiers: [AppState.codexAppBundleId], + lastPollAt: nil, + now: now + )) + XCTAssertFalse(AppState.shouldPollCodexDesktopDiscovery( + runningBundleIdentifiers: [AppState.codexAppBundleId], + lastPollAt: now.addingTimeInterval(-5), + now: now + )) + XCTAssertTrue(AppState.shouldPollCodexDesktopDiscovery( + runningBundleIdentifiers: [AppState.codexAppBundleId], + lastPollAt: now.addingTimeInterval(-6), + now: now + )) + } + + func testCodexDesktopDiscoveryChannelsShareOneIdentityWhileCLIStaysRaw() { + let rawId = "desktop-thread" + let desktop = AppState.codexDiscoveryIdentity(rawSessionId: rawId, isDesktop: true) + let cli = AppState.codexDiscoveryIdentity(rawSessionId: rawId, isDesktop: false) + + XCTAssertEqual(desktop.sessionId, "codexapp:\(rawId)") + XCTAssertEqual(desktop.providerSessionId, rawId) + XCTAssertEqual(desktop.termBundleId, AppState.codexAppBundleId) + XCTAssertEqual(cli.sessionId, rawId) + XCTAssertNil(cli.providerSessionId) + XCTAssertNil(cli.termBundleId) + } + + func testLegacyDesktopRestoreCanonicalizesWithoutChangingCLIKey() { + XCTAssertEqual( + AppState.canonicalRestoredCodexSessionId( + sessionId: "legacy-thread", + source: "codex", + providerSessionId: "legacy-thread", + termBundleId: AppState.codexAppBundleId + ), + "codexapp:legacy-thread" + ) + XCTAssertEqual( + AppState.canonicalRestoredCodexSessionId( + sessionId: "cli-thread", + source: "codex", + providerSessionId: "cli-thread", + termBundleId: nil + ), + "cli-thread" + ) + } + + func testSequentialDesktopRolloutAndDatabaseScansKeepOneCardAndRejectStaleStatus() { + let appState = AppState() + let rawId = "same-provider-thread" + let identity = AppState.codexDiscoveryIdentity(rawSessionId: rawId, isDesktop: true) + let firstActivity = Date() + let newerActivity = firstActivity.addingTimeInterval(2) + + appState.integrateDiscovered([AppState.DiscoveredSession( + sessionId: identity.sessionId, + cwd: "/fallback/repo", + tty: nil, + model: "gpt-test", + pid: nil, + modifiedAt: firstActivity, + recentMessages: [], + source: "codex", + transcriptPath: "/tmp/fallback-rollout.jsonl", + status: .processing, + termBundleId: identity.termBundleId, + providerSessionId: identity.providerSessionId + )]) + appState.integrateDiscovered([AppState.DiscoveredSession( + sessionId: identity.sessionId, + cwd: "/database/repo", + tty: nil, + model: "gpt-test", + pid: nil, + modifiedAt: newerActivity, + recentMessages: [], + source: "codex", + transcriptPath: "/tmp/database-rollout.jsonl", + status: .idle, + termBundleId: identity.termBundleId, + providerSessionId: identity.providerSessionId + )]) + + XCTAssertEqual(appState.sessions.count, 1) + XCTAssertEqual(appState.sessions[identity.sessionId]?.status, .idle) + XCTAssertEqual(appState.sessions[identity.sessionId]?.lastActivity, newerActivity) + + // Simulate the slower generic FSEvent scan arriving after the DB poll. + appState.integrateDiscovered([AppState.DiscoveredSession( + sessionId: identity.sessionId, + cwd: "/stale/repo", + tty: nil, + model: "gpt-test", + pid: nil, + modifiedAt: firstActivity.addingTimeInterval(1), + recentMessages: [], + source: "codex", + transcriptPath: "/tmp/stale-rollout.jsonl", + status: .processing, + termBundleId: "com.example.stale", + providerSessionId: identity.providerSessionId + )]) + + XCTAssertEqual(appState.sessions.count, 1) + XCTAssertEqual(appState.sessions[identity.sessionId]?.status, .idle) + XCTAssertEqual(appState.sessions[identity.sessionId]?.lastActivity, newerActivity) + XCTAssertEqual(appState.sessions[identity.sessionId]?.cwd, "/database/repo") + XCTAssertEqual( + appState.sessions[identity.sessionId]?.transcriptPath, + "/tmp/database-rollout.jsonl" + ) + XCTAssertEqual(appState.sessions[identity.sessionId]?.termBundleId, AppState.codexAppBundleId) + } + + func testDesktopAndCLIWithSameCwdRemainSeparateCards() { + let appState = AppState() + let rawId = "desktop-thread" + let desktop = AppState.codexDiscoveryIdentity(rawSessionId: rawId, isDesktop: true) + let now = Date() + + appState.integrateDiscovered([ + AppState.DiscoveredSession( + sessionId: desktop.sessionId, + cwd: "/same/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: now, + recentMessages: [], + source: "codex", + status: .processing, + termBundleId: desktop.termBundleId, + providerSessionId: desktop.providerSessionId + ), + AppState.DiscoveredSession( + sessionId: "cli-thread", + cwd: "/same/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: now, + recentMessages: [], + source: "codex", + status: .processing + ), + ]) + + XCTAssertEqual(Set(appState.sessions.keys), ["codexapp:\(rawId)", "cli-thread"]) + XCTAssertTrue(appState.sessions["codexapp:\(rawId)"]?.isNativeAppMode == true) + XCTAssertFalse(appState.sessions["cli-thread"]?.isNativeAppMode == true) + } + + func testCodexSubagentsChooseParentInTheirOwnDesktopOrCLINamespace() { + let previousMode = UserDefaults.standard.object(forKey: SettingsKey.pluginSessionMode) + UserDefaults.standard.set("merge", forKey: SettingsKey.pluginSessionMode) + defer { + if let previousMode { + UserDefaults.standard.set(previousMode, forKey: SettingsKey.pluginSessionMode) + } else { + UserDefaults.standard.removeObject(forKey: SettingsKey.pluginSessionMode) + } + } + + let appState = AppState() + var desktopParent = SessionSnapshot() + desktopParent.source = "codex" + desktopParent.providerSessionId = "parent-thread" + desktopParent.termBundleId = AppState.codexAppBundleId + var cliParent = SessionSnapshot() + cliParent.source = "codex" + cliParent.providerSessionId = "parent-thread" + appState.sessions["codexapp:parent-thread"] = desktopParent + appState.sessions["parent-thread"] = cliParent + + appState.integrateDiscovered([ + AppState.DiscoveredSession( + sessionId: "codexapp:desktop-child", + cwd: "/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: Date(), + recentMessages: [], + source: "codex", + parentSessionId: "parent-thread", + agentType: "desktop-worker", + status: .processing, + termBundleId: AppState.codexAppBundleId, + providerSessionId: "desktop-child" + ), + AppState.DiscoveredSession( + sessionId: "cli-child", + cwd: "/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: Date(), + recentMessages: [], + source: "codex", + parentSessionId: "parent-thread", + agentType: "cli-worker", + status: .processing, + providerSessionId: "cli-child" + ), + ]) + + XCTAssertNotNil(appState.sessions["codexapp:parent-thread"]?.subagents["desktop-child"]) + XCTAssertNil(appState.sessions["codexapp:parent-thread"]?.subagents["cli-child"]) + XCTAssertNotNil(appState.sessions["parent-thread"]?.subagents["cli-child"]) + XCTAssertNil(appState.sessions["parent-thread"]?.subagents["desktop-child"]) + } + + func testCodexHookSubagentChoosesParentInItsDesktopNamespace() throws { + let dir = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-hook-parent-\(UUID().uuidString)") + try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: dir) } + let transcript = dir.appendingPathComponent("child.jsonl") + try (#"{"type":"session_meta","payload":{"id":"child","cwd":"/repo","source":{"subagent":{"thread_spawn":{"parent_thread_id":"parent-thread"}}}}}"# + "\n") + .write(to: transcript, atomically: true, encoding: .utf8) + + let appState = AppState() + var desktopParent = SessionSnapshot() + desktopParent.source = "codex" + desktopParent.providerSessionId = "parent-thread" + desktopParent.termBundleId = AppState.codexAppBundleId + var cliParent = SessionSnapshot() + cliParent.source = "codex" + cliParent.providerSessionId = "parent-thread" + appState.sessions["codexapp:parent-thread"] = desktopParent + appState.sessions["parent-thread"] = cliParent + + let server = HookServer(appState: appState) + XCTAssertEqual(server.codexNativeSubsessionParentId(from: [ + "session_id": "child", + "_source": "codex", + "_term_bundle": AppState.codexAppBundleId, + "transcript_path": transcript.path, + ]), "codexapp:parent-thread") + XCTAssertEqual(server.codexNativeSubsessionParentId(from: [ + "session_id": "child", + "_source": "codex", + "transcript_path": transcript.path, + ]), "parent-thread") + } + + func testCodexDesktopDoesNotRetainSharedHelperPidButOtherSessionsDo() { + let appState = AppState() + let now = Date() + let currentPid = getpid() + + appState.integrateDiscovered([ + AppState.DiscoveredSession( + sessionId: "codexapp:desktop", + cwd: "/same/repo", + tty: nil, + model: nil, + pid: currentPid, + modifiedAt: now, + recentMessages: [], + source: "codex", + termBundleId: AppState.codexAppBundleId, + providerSessionId: "desktop" + ), + AppState.DiscoveredSession( + sessionId: "codex-cli", + cwd: "/same/repo", + tty: nil, + model: nil, + pid: currentPid, + modifiedAt: now, + recentMessages: [], + source: "codex" + ), + AppState.DiscoveredSession( + sessionId: "other-native", + cwd: "/same/repo", + tty: nil, + model: nil, + pid: currentPid, + modifiedAt: now, + recentMessages: [], + source: "cursor", + termBundleId: "com.todesktop.230313mzl4w4u92" + ), + ]) + + XCTAssertNil(appState.sessions["codexapp:desktop"]?.cliPid) + XCTAssertEqual(appState.sessions["codex-cli"]?.cliPid, currentPid) + XCTAssertEqual(appState.sessions["other-native"]?.cliPid, currentPid) + } + + func testCompletedLastCodexSubagentClearsParentProjection() { + let previousMode = UserDefaults.standard.object(forKey: SettingsKey.pluginSessionMode) + UserDefaults.standard.set("merge", forKey: SettingsKey.pluginSessionMode) + defer { + if let previousMode { + UserDefaults.standard.set(previousMode, forKey: SettingsKey.pluginSessionMode) + } else { + UserDefaults.standard.removeObject(forKey: SettingsKey.pluginSessionMode) + } + } + + let appState = AppState() + var parent = SessionSnapshot() + parent.source = "codex" + parent.providerSessionId = "parent-thread" + parent.status = .running + parent.currentTool = "Agent" + parent.toolDescription = "worker" + parent.subagents["child-thread"] = SubagentState( + agentId: "child-thread", + agentType: "worker" + ) + appState.sessions["codexapp:parent-thread"] = parent + + appState.integrateDiscovered([AppState.DiscoveredSession( + sessionId: "codexapp:child-thread", + cwd: "/same/repo", + tty: nil, + model: nil, + pid: nil, + modifiedAt: Date(), + recentMessages: [], + source: "codex", + parentSessionId: "parent-thread", + status: .idle, + termBundleId: AppState.codexAppBundleId, + providerSessionId: "child-thread" + )]) + + let updatedParent = appState.sessions["codexapp:parent-thread"] + XCTAssertTrue(updatedParent?.subagents.isEmpty == true) + XCTAssertEqual(updatedParent?.status, .processing) + XCTAssertNil(updatedParent?.currentTool) + XCTAssertNil(updatedParent?.toolDescription) + } + + func testCompletedCodexSubagentDiscoveryIsTransient() { + XCTAssertTrue(AppState.isCompletedCodexSubagentDiscovery( + source: "codex", + parentSessionId: "parent-thread", + status: .idle + )) + XCTAssertFalse(AppState.isCompletedCodexSubagentDiscovery( + source: "codex", + parentSessionId: nil, + status: .idle + )) + XCTAssertFalse(AppState.isCompletedCodexSubagentDiscovery( + source: "codex", + parentSessionId: "parent-thread", + status: .processing + )) + } + func testCodexSubagentMetadataParsesParentThreadFromRolloutSessionMeta() throws { let dir = FileManager.default.temporaryDirectory .appendingPathComponent("codeisland-codex-subagent-\(UUID().uuidString)") @@ -59,6 +464,35 @@ final class AppStateCodexSubsessionTests: XCTestCase { XCTAssertEqual(metadata.agentNickname, "Ohm") } + func testReadableRootTranscriptDoesNotFallBackToStaleSpawnEdge() throws { + let dir = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-root-metadata-\(UUID().uuidString)") + try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: dir) } + let transcript = dir.appendingPathComponent("root.jsonl") + try (#"{"type":"session_meta","payload":{"id":"root-thread","cwd":"/repo","source":"vscode"}}"# + "\n") + .write(to: transcript, atomically: true, encoding: .utf8) + + let statePath = dir.appendingPathComponent("state_5.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + XCTAssertEqual(sqlite3_exec(db, """ + CREATE TABLE thread_spawn_edges ( + parent_thread_id TEXT NOT NULL, + child_thread_id TEXT NOT NULL PRIMARY KEY, + status TEXT NOT NULL + ); + INSERT INTO thread_spawn_edges VALUES ('stale-parent', 'root-thread', 'open'); + """, nil, nil, nil), SQLITE_OK) + + XCTAssertNil(AppState.codexSubagentMetadata( + threadId: "root-thread", + transcriptPath: transcript.path, + statePath: statePath + )) + } + func testKnownCodexSubagentSessionMergesIntoParentFromTranscriptMetadata() throws { let previousMode = UserDefaults.standard.object(forKey: SettingsKey.pluginSessionMode) UserDefaults.standard.set("merge", forKey: SettingsKey.pluginSessionMode) @@ -104,6 +538,65 @@ final class AppStateCodexSubsessionTests: XCTestCase { XCTAssertEqual(appState.activeSessionId, "parent") } + func testKnownIdleCodexSubagentClearsParentEvenWhenSpawnEdgeRemainsOpen() throws { + let previousMode = UserDefaults.standard.object(forKey: SettingsKey.pluginSessionMode) + UserDefaults.standard.set("merge", forKey: SettingsKey.pluginSessionMode) + defer { + if let previousMode { + UserDefaults.standard.set(previousMode, forKey: SettingsKey.pluginSessionMode) + } else { + UserDefaults.standard.removeObject(forKey: SettingsKey.pluginSessionMode) + } + } + + let dir = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-idle-edge-\(UUID().uuidString)") + try FileManager.default.createDirectory(at: dir, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: dir) } + let childTranscript = dir.appendingPathComponent("child.jsonl") + try (#"{"type":"session_meta","payload":{"id":"child-thread","cwd":"/repo","source":{"subagent":{"thread_spawn":{"parent_thread_id":"parent-thread","agent_role":"worker"}}}}}"# + "\n") + .write(to: childTranscript, atomically: true, encoding: .utf8) + + let statePath = dir.appendingPathComponent("state_5.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + XCTAssertEqual(sqlite3_exec(db, """ + CREATE TABLE thread_spawn_edges ( + parent_thread_id TEXT NOT NULL, + child_thread_id TEXT NOT NULL PRIMARY KEY, + status TEXT NOT NULL + ); + INSERT INTO thread_spawn_edges VALUES ('parent-thread', 'child-thread', 'open'); + """, nil, nil, nil), SQLITE_OK) + + let appState = AppState() + var parent = SessionSnapshot() + parent.source = "codex" + parent.providerSessionId = "parent-thread" + parent.status = .running + parent.currentTool = "Agent" + parent.toolDescription = "worker" + parent.subagents["child-thread"] = SubagentState( + agentId: "child-thread", + agentType: "worker" + ) + var child = SessionSnapshot() + child.source = "codex" + child.providerSessionId = "child-thread" + child.transcriptPath = childTranscript.path + child.status = .idle + appState.sessions["parent"] = parent + appState.sessions["child"] = child + + XCTAssertTrue(appState.applyCodexSubsessionModeToKnownSessions(statePath: statePath)) + XCTAssertNil(appState.sessions["child"]) + XCTAssertTrue(appState.sessions["parent"]?.subagents.isEmpty == true) + XCTAssertEqual(appState.sessions["parent"]?.status, .processing) + XCTAssertNil(appState.sessions["parent"]?.currentTool) + XCTAssertNil(appState.sessions["parent"]?.toolDescription) + } + func testCurrentPluginSessionModeSeparateSplitsMergedCodexSubagent() { let previousMode = UserDefaults.standard.object(forKey: SettingsKey.pluginSessionMode) UserDefaults.standard.set("separate", forKey: SettingsKey.pluginSessionMode) @@ -146,6 +639,47 @@ final class AppStateCodexSubsessionTests: XCTestCase { XCTAssertEqual(appState.activeSessionId, "child-thread") } + func testSeparateDesktopSubagentDoesNotOverwriteSameProviderCLIChild() { + let previousMode = UserDefaults.standard.object(forKey: SettingsKey.pluginSessionMode) + UserDefaults.standard.set("separate", forKey: SettingsKey.pluginSessionMode) + defer { + if let previousMode { + UserDefaults.standard.set(previousMode, forKey: SettingsKey.pluginSessionMode) + } else { + UserDefaults.standard.removeObject(forKey: SettingsKey.pluginSessionMode) + } + } + + let appState = AppState() + var parent = SessionSnapshot() + parent.source = "codex" + parent.providerSessionId = "desktop-parent" + parent.termBundleId = AppState.codexAppBundleId + parent.subagents["shared-child"] = SubagentState( + agentId: "shared-child", + agentType: "desktop-worker" + ) + var cliChild = SessionSnapshot() + cliChild.source = "codex" + cliChild.providerSessionId = "shared-child" + cliChild.cwd = "/cli/repo" + appState.sessions["codexapp:desktop-parent"] = parent + appState.sessions["shared-child"] = cliChild + + appState.applyCurrentPluginSessionMode(persist: false) + + XCTAssertEqual(appState.sessions["shared-child"]?.cwd, "/cli/repo") + XCTAssertNil(appState.sessions["shared-child"]?.termBundleId) + XCTAssertEqual( + appState.sessions["codexapp:shared-child"]?.termBundleId, + AppState.codexAppBundleId + ) + XCTAssertEqual( + appState.sessions["codexapp:shared-child"]?.providerSessionId, + "shared-child" + ) + } + func testCurrentPluginSessionModeHideClearsMergedCodexSubagent() { let previousMode = UserDefaults.standard.object(forKey: SettingsKey.pluginSessionMode) UserDefaults.standard.set("hide", forKey: SettingsKey.pluginSessionMode) diff --git a/Tests/CodeIslandTests/AppStateCodexTranscriptTests.swift b/Tests/CodeIslandTests/AppStateCodexTranscriptTests.swift index 2d148471..3ddd6cbf 100644 --- a/Tests/CodeIslandTests/AppStateCodexTranscriptTests.swift +++ b/Tests/CodeIslandTests/AppStateCodexTranscriptTests.swift @@ -1,7 +1,609 @@ import XCTest @testable import CodeIsland +import SQLite3 final class AppStateCodexTranscriptTests: XCTestCase { + func testDelayedDeltaFromDetachedGenerationCannotMutateRecreatedSession() throws { + let appState = AppState() + let sessionId = "codexapp:reused-thread" + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-generation-\(UUID().uuidString)") + let oldTranscript = directory.appendingPathComponent("old.jsonl") + let newTranscript = directory.appendingPathComponent("new.jsonl") + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + try Data().write(to: oldTranscript) + try Data().write(to: newTranscript) + defer { + appState.detachTranscriptTailer(sessionId: sessionId) + try? FileManager.default.removeItem(at: directory) + } + + var oldSession = SessionSnapshot() + oldSession.source = "codex" + oldSession.termBundleId = AppState.codexAppBundleId + oldSession.transcriptPath = oldTranscript.path + oldSession.status = .processing + appState.sessions[sessionId] = oldSession + appState.attachTranscriptTailerIfNeeded(sessionId: sessionId) + let oldToken = try XCTUnwrap(appState.attachedTranscriptTokens[sessionId]) + + // Model a delta that has left JSONLTailer's serial queue but whose + // MainActor hop is delayed until after close + same-id recreation. + let delayedOldDelta = ConversationTailDelta( + sessionId: sessionId, + lastUserPrompt: nil, + lastAssistantMessage: "stale reply", + turnStatus: .idle, + hasActivity: true, + attachmentToken: oldToken, + filePath: oldTranscript.path + ) + + appState.removeSession(sessionId) + XCTAssertNil(appState.attachedTranscriptTokens[sessionId]) + + var recreated = SessionSnapshot() + recreated.source = "codex" + recreated.termBundleId = AppState.codexAppBundleId + recreated.transcriptPath = newTranscript.path + recreated.status = .processing + recreated.lastAssistantMessage = "fresh reply" + appState.sessions[sessionId] = recreated + appState.attachTranscriptTailerIfNeeded(sessionId: sessionId) + let newToken = try XCTUnwrap(appState.attachedTranscriptTokens[sessionId]) + XCTAssertNotEqual(newToken, oldToken) + + appState.applyTranscriptDelta(delayedOldDelta) + + XCTAssertEqual(appState.sessions[sessionId]?.status, .processing) + XCTAssertEqual(appState.sessions[sessionId]?.lastAssistantMessage, "fresh reply") + XCTAssertEqual(appState.attachedTranscriptPaths[sessionId], newTranscript.path) + XCTAssertEqual(appState.attachedTranscriptTokens[sessionId], newToken) + } + + func testRecentCodexDesktopRecordsFindSameCwdThreadsBeyondFixedTailWindow() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-desktop-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let now = Date() + let hugePayload = String(repeating: "x", count: 200_000) + let activeRoot = tempDir.appendingPathComponent("root.jsonl") + let activeSibling = tempDir.appendingPathComponent("sibling.jsonl") + let completed = tempDir.appendingPathComponent("completed.jsonl") + let started = #"{"type":"event_msg","payload":{"type":"task_started"}}"# + let toolOutput = #"{"type":"response_item","payload":{"type":"function_call_output","output":"\#(hugePayload)"}}"# + let finished = #"{"type":"event_msg","payload":{"type":"turn_aborted"}}"# + + try ([started, toolOutput].joined(separator: "\n") + "\n") + .write(to: activeRoot, atomically: true, encoding: .utf8) + try (started + "\n").write(to: activeSibling, atomically: true, encoding: .utf8) + try ([started, finished].joined(separator: "\n") + "\n") + .write(to: completed, atomically: true, encoding: .utf8) + try fm.setAttributes( + [.modificationDate: now.addingTimeInterval(-60)], + ofItemAtPath: completed.path + ) + + let statePath = tempDir.appendingPathComponent("state_5.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + + let current = Int64(now.timeIntervalSince1970) + let laggingDesktopUpdate = current - 1_800 + let old = current - 60 + let sql = """ + CREATE TABLE threads ( + id TEXT PRIMARY KEY, + rollout_path TEXT NOT NULL, + cwd TEXT NOT NULL, + updated_at INTEGER NOT NULL, + model TEXT, + archived INTEGER NOT NULL, + has_user_event INTEGER NOT NULL, + source TEXT NOT NULL + ); + INSERT INTO threads VALUES ('root-thread', '\(activeRoot.path)', '/same/repo', \(laggingDesktopUpdate), 'gpt-test', 0, 0, 'vscode'); + INSERT INTO threads VALUES ('sibling-thread', '\(activeSibling.path)', '/same/repo', \(current), 'gpt-test', 0, 1, 'appServer'); + INSERT INTO threads VALUES ('completed-thread', '\(completed.path)', '/same/repo', \(old), 'gpt-test', 0, 1, 'vscode'); + """ + XCTAssertEqual(sqlite3_exec(db, sql, nil, nil, nil), SQLITE_OK) + + let records = AppState.recentCodexDesktopThreadRecords( + statePath: statePath, + now: now, + freshnessWindow: 600, + completionSettleWindow: 30, + fileManager: fm + ) + + XCTAssertEqual(Set(records.map(\.sessionId)), ["root-thread", "sibling-thread"]) + XCTAssertTrue(records.allSatisfy { $0.cwd == "/same/repo" }) + XCTAssertTrue(records.allSatisfy { $0.status == .processing }) + } + + func testCodexDesktopStateScanFindsThreadCreatedAfterInitialEmptyScan() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-desktop-poll-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let now = Date() + let statePath = tempDir.appendingPathComponent("state_5.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + XCTAssertEqual(sqlite3_exec(db, """ + CREATE TABLE threads ( + id TEXT PRIMARY KEY, + rollout_path TEXT NOT NULL, + cwd TEXT NOT NULL, + updated_at INTEGER NOT NULL, + model TEXT, + archived INTEGER NOT NULL, + has_user_event INTEGER NOT NULL, + source TEXT NOT NULL + ); + """, nil, nil, nil), SQLITE_OK) + + XCTAssertTrue(AppState.recentCodexDesktopThreadRecords( + statePath: statePath, + now: now, + fileManager: fm + ).isEmpty) + + let rollout = tempDir.appendingPathComponent("new-thread.jsonl") + try (#"{"type":"event_msg","payload":{"type":"task_started"}}"# + "\n") + .write(to: rollout, atomically: true, encoding: .utf8) + let insert = """ + INSERT INTO threads VALUES ( + 'new-thread', '\(rollout.path)', '/new/repo', + \(Int64(now.timeIntervalSince1970)), 'gpt-test', 0, 0, 'appServer' + ); + """ + XCTAssertEqual(sqlite3_exec(db, insert, nil, nil, nil), SQLITE_OK) + + let records = AppState.recentCodexDesktopThreadRecords( + statePath: statePath, + now: now, + fileManager: fm + ) + XCTAssertEqual(records.map(\.sessionId), ["new-thread"]) + XCTAssertEqual(records.first?.status, .processing) + XCTAssertEqual(records.first?.cwd, "/new/repo") + } + + func testCodexDesktopStateScanTreatsMissingAndIncompatibleDatabasesAsEmpty() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-db-fallback-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let missingPath = tempDir.appendingPathComponent("missing.sqlite").path + XCTAssertTrue(AppState.recentCodexDesktopThreadRecords( + statePath: missingPath, + fileManager: fm + ).isEmpty) + XCTAssertFalse(fm.fileExists(atPath: missingPath)) + + let incompatiblePath = tempDir.appendingPathComponent("incompatible.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(incompatiblePath, &db), SQLITE_OK) + XCTAssertEqual(sqlite3_exec( + db, + "CREATE TABLE threads (id TEXT PRIMARY KEY, unexpected_column TEXT);", + nil, + nil, + nil + ), SQLITE_OK) + sqlite3_close_v2(db) + + XCTAssertTrue(AppState.recentCodexDesktopThreadRecords( + statePath: incompatiblePath, + fileManager: fm + ).isEmpty) + } + + func testCodexDesktopStateScanPrefersMillisecondTimestampWhenBothColumnsExist() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-db-ms-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let now = Date() + let rollout = tempDir.appendingPathComponent("millisecond-thread.jsonl") + try (#"{"type":"event_msg","payload":{"type":"task_started"}}"# + "\n") + .write(to: rollout, atomically: true, encoding: .utf8) + try fm.setAttributes( + [.modificationDate: now.addingTimeInterval(-7_200)], + ofItemAtPath: rollout.path + ) + + let statePath = tempDir.appendingPathComponent("state_5.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + let staleSeconds = Int64(now.addingTimeInterval(-7_200).timeIntervalSince1970) + let currentMilliseconds = Int64(now.timeIntervalSince1970 * 1_000) + let sql = """ + CREATE TABLE threads ( + id TEXT PRIMARY KEY, + rollout_path TEXT NOT NULL, + cwd TEXT NOT NULL, + updated_at INTEGER NOT NULL, + updated_at_ms INTEGER, + model TEXT, + archived INTEGER NOT NULL, + source TEXT NOT NULL + ); + INSERT INTO threads VALUES ( + 'millisecond-thread', '\(rollout.path)', '/ms/repo', + \(staleSeconds), \(currentMilliseconds), 'gpt-test', 0, 'vscode' + ); + """ + XCTAssertEqual(sqlite3_exec(db, sql, nil, nil, nil), SQLITE_OK) + + let records = AppState.recentCodexDesktopThreadRecords( + statePath: statePath, + now: now, + fileManager: fm + ) + XCTAssertEqual(records.map(\.sessionId), ["millisecond-thread"]) + XCTAssertEqual(records.first?.status, .processing) + } + + func testCodexDesktopStateScanRejectsCorruptNullAndMissingTranscriptRows() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-db-invalid-rows-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let corruptPath = tempDir.appendingPathComponent("corrupt.sqlite").path + try Data("not-a-sqlite-database".utf8).write(to: URL(fileURLWithPath: corruptPath)) + XCTAssertTrue(AppState.recentCodexDesktopThreadRecords( + statePath: corruptPath, + fileManager: fm + ).isEmpty) + + let statePath = tempDir.appendingPathComponent("nullable.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + let now = Int64(Date().timeIntervalSince1970) + let missingTranscript = tempDir.appendingPathComponent("missing.jsonl").path + XCTAssertEqual(sqlite3_exec(db, """ + CREATE TABLE threads ( + id TEXT PRIMARY KEY, + rollout_path TEXT, + cwd TEXT, + updated_at INTEGER, + model TEXT, + archived INTEGER, + source TEXT + ); + INSERT INTO threads VALUES ('null-path', NULL, '/repo', \(now), NULL, 0, 'vscode'); + INSERT INTO threads VALUES ('null-cwd', '\(missingTranscript)', NULL, \(now), NULL, 0, 'vscode'); + INSERT INTO threads VALUES ('missing-file', '\(missingTranscript)', '/repo', \(now), NULL, 0, 'vscode'); + """, nil, nil, nil), SQLITE_OK) + + XCTAssertTrue(AppState.recentCodexDesktopThreadRecords( + statePath: statePath, + fileManager: fm + ).isEmpty) + } + + func testCodexDesktopStateScanExcludesExplicitCLISubagentRollout() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-db-origin-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let cliRollout = tempDir.appendingPathComponent("cli-child.jsonl") + let desktopRollout = tempDir.appendingPathComponent("desktop-child.jsonl") + let unknownRollout = tempDir.appendingPathComponent("unknown-child.jsonl") + let started = #"{"type":"event_msg","payload":{"type":"task_started"}}"# + try ([ + #"{"type":"session_meta","payload":{"id":"cli-child","cwd":"/repo","originator":"Codex CLI","source":"cli"}}"#, + started, + ].joined(separator: "\n") + "\n").write( + to: cliRollout, + atomically: true, + encoding: .utf8 + ) + try ([ + #"{"type":"session_meta","payload":{"id":"desktop-child","cwd":"/repo","originator":"Codex Desktop","source":"vscode"}}"#, + started, + ].joined(separator: "\n") + "\n").write( + to: desktopRollout, + atomically: true, + encoding: .utf8 + ) + try (["truncated-session-meta", started].joined(separator: "\n") + "\n").write( + to: unknownRollout, + atomically: true, + encoding: .utf8 + ) + + let statePath = tempDir.appendingPathComponent("state_5.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + let now = Int64(Date().timeIntervalSince1970) + let subagentSource = #"{"subagent":{"thread_spawn":{"parent_thread_id":"parent"}}}"# + XCTAssertEqual(sqlite3_exec(db, """ + CREATE TABLE threads ( + id TEXT PRIMARY KEY, + rollout_path TEXT NOT NULL, + cwd TEXT NOT NULL, + updated_at INTEGER NOT NULL, + model TEXT, + archived INTEGER NOT NULL, + source TEXT NOT NULL + ); + INSERT INTO threads VALUES ( + 'cli-child', '\(cliRollout.path)', '/repo', \(now), NULL, 0, '\(subagentSource)' + ); + INSERT INTO threads VALUES ( + 'desktop-child', '\(desktopRollout.path)', '/repo', \(now), NULL, 0, '\(subagentSource)' + ); + INSERT INTO threads VALUES ( + 'unknown-child', '\(unknownRollout.path)', '/repo', \(now), NULL, 0, '\(subagentSource)' + ); + """, nil, nil, nil), SQLITE_OK) + + let records = AppState.recentCodexDesktopThreadRecords( + statePath: statePath, + fileManager: fm + ) + XCTAssertEqual(records.map(\.sessionId), ["desktop-child"]) + } + + func testReverseLifecycleScanSkipsOversizedUnterminatedTailLine() throws { + let file = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-oversized-tail-\(UUID().uuidString).jsonl") + defer { try? FileManager.default.removeItem(at: file) } + var data = Data( + (#"{"type":"event_msg","payload":{"type":"task_started"}}"# + "\n").utf8 + ) + data.append(Data(repeating: 0x78, count: 512 * 1024)) + try data.write(to: file) + + XCTAssertEqual( + AppState.latestCodexTurnStatus( + path: file.path, + chunkSize: 16 * 1024, + maxBytesToScan: 1024 * 1024, + maxLineBytes: 64 * 1024 + ), + .processing + ) + } + + func testReverseLifecycleScanHonorsExplicitTotalByteBudget() throws { + let file = FileManager.default.temporaryDirectory + .appendingPathComponent("codeisland-codex-scan-budget-\(UUID().uuidString).jsonl") + defer { try? FileManager.default.removeItem(at: file) } + var data = Data( + (#"{"type":"event_msg","payload":{"type":"task_started"}}"# + "\n").utf8 + ) + data.append(Data(repeating: 0x78, count: 256 * 1024)) + try data.write(to: file) + + XCTAssertNil(AppState.latestCodexTurnStatus( + path: file.path, + chunkSize: 16 * 1024, + maxBytesToScan: 64 * 1024, + maxLineBytes: 32 * 1024 + )) + } + + func testRootCwdDesktopRolloutSurvivesMissingStateDatabase() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-root-fallback-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let now = Date() + let rawSessionId = "11111111-2222-4333-8444-555555555555" + let sessionsBase = tempDir.appendingPathComponent("sessions") + _ = try makeCodexRollout( + sessionsBase: sessionsBase, + sessionId: rawSessionId, + cwd: "/fallback/repo", + now: now + ) + + let discovered = AppState.discoverCodexSessions( + processes: [CodexProcessDiscoveryCandidate( + pid: 123, + cwd: "/", + startTime: now.addingTimeInterval(-60), + isDesktop: true + )], + sessionsBase: sessionsBase.path, + statePath: tempDir.appendingPathComponent("missing.sqlite").path, + now: now, + fileManager: fm + ) + + XCTAssertEqual(discovered.map(\.sessionId), ["codexapp:\(rawSessionId)"]) + XCTAssertEqual(discovered.first?.providerSessionId, rawSessionId) + XCTAssertEqual(discovered.first?.termBundleId, AppState.codexAppBundleId) + XCTAssertEqual(discovered.first?.cwd, "/fallback/repo") + } + + func testSameRawThreadOpenInDesktopAndCLIRemainsTwoDiscoveries() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-dual-mode-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let now = Date() + let rawSessionId = "12345678-1234-4234-8234-123456789abc" + let sessionsBase = tempDir.appendingPathComponent("sessions") + _ = try makeCodexRollout( + sessionsBase: sessionsBase, + sessionId: rawSessionId, + cwd: "/shared/repo", + now: now, + originator: "Codex Desktop", + source: "vscode", + filenameTimestamp: "2000-01-01T00-00-00" + ) + _ = try makeCodexRollout( + sessionsBase: sessionsBase, + sessionId: rawSessionId, + cwd: "/shared/repo", + now: now, + originator: "Codex CLI", + source: "cli", + filenameTimestamp: "2000-01-01T00-00-01" + ) + + let discovered = AppState.discoverCodexSessions( + processes: [ + CodexProcessDiscoveryCandidate( + pid: 100, + cwd: "/", + startTime: now.addingTimeInterval(-60), + isDesktop: true + ), + CodexProcessDiscoveryCandidate( + pid: 101, + cwd: "/shared/repo", + startTime: now.addingTimeInterval(-60), + isDesktop: false + ), + ], + sessionsBase: sessionsBase.path, + statePath: tempDir.appendingPathComponent("missing.sqlite").path, + now: now, + fileManager: fm + ) + + XCTAssertEqual(Set(discovered.map(\.sessionId)), [ + rawSessionId, + "codexapp:\(rawSessionId)", + ]) + XCTAssertEqual( + discovered.first(where: { $0.sessionId == rawSessionId })?.termBundleId, + nil + ) + XCTAssertEqual( + discovered.first(where: { $0.sessionId.hasPrefix("codexapp:") })?.termBundleId, + AppState.codexAppBundleId + ) + } + + func testUnknownOriginRolloutIsClaimedOnlyByPreferredDesktopProcess() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-unknown-mode-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let now = Date() + let rawSessionId = "99999999-8888-4777-8666-555555555555" + let sessionsBase = tempDir.appendingPathComponent("sessions") + _ = try makeCodexRollout( + sessionsBase: sessionsBase, + sessionId: rawSessionId, + cwd: "/shared/repo", + now: now, + originator: "future-surface", + source: "future-source" + ) + + let discovered = AppState.discoverCodexSessions( + processes: [ + CodexProcessDiscoveryCandidate( + pid: 100, + cwd: "/", + startTime: now.addingTimeInterval(-60), + isDesktop: true + ), + CodexProcessDiscoveryCandidate( + pid: 101, + cwd: "/shared/repo", + startTime: now.addingTimeInterval(-60), + isDesktop: false + ), + ], + sessionsBase: sessionsBase.path, + statePath: tempDir.appendingPathComponent("missing.sqlite").path, + now: now, + fileManager: fm + ) + + XCTAssertEqual(discovered.map(\.sessionId), ["codexapp:\(rawSessionId)"]) + } + + func testRejectedRootRolloutDoesNotExcludeValidDatabaseRecord() throws { + let fm = FileManager.default + let tempDir = fm.temporaryDirectory + .appendingPathComponent("codeisland-codex-rejected-fallback-\(UUID().uuidString)") + try fm.createDirectory(at: tempDir, withIntermediateDirectories: true) + defer { try? fm.removeItem(at: tempDir) } + + let now = Date() + let rawSessionId = "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee" + let sessionsBase = tempDir.appendingPathComponent("sessions") + let rollout = try makeCodexRollout( + sessionsBase: sessionsBase, + sessionId: rawSessionId, + cwd: nil, + now: now + ) + + let statePath = tempDir.appendingPathComponent("state_5.sqlite").path + var db: OpaquePointer? + XCTAssertEqual(sqlite3_open(statePath, &db), SQLITE_OK) + defer { sqlite3_close_v2(db) } + let sql = """ + CREATE TABLE threads ( + id TEXT PRIMARY KEY, + rollout_path TEXT NOT NULL, + cwd TEXT NOT NULL, + updated_at INTEGER NOT NULL, + model TEXT, + archived INTEGER NOT NULL, + source TEXT NOT NULL + ); + INSERT INTO threads VALUES ( + '\(rawSessionId)', '\(rollout.path)', '/database/repo', + \(Int64(now.timeIntervalSince1970)), 'gpt-test', 0, 'vscode' + ); + """ + XCTAssertEqual(sqlite3_exec(db, sql, nil, nil, nil), SQLITE_OK) + + let discovered = AppState.discoverCodexSessions( + processes: [CodexProcessDiscoveryCandidate( + pid: 123, + cwd: "/", + startTime: now.addingTimeInterval(-60), + isDesktop: true + )], + sessionsBase: sessionsBase.path, + statePath: statePath, + now: now, + fileManager: fm + ) + + XCTAssertEqual(discovered.map(\.sessionId), ["codexapp:\(rawSessionId)"]) + XCTAssertEqual(discovered.first?.cwd, "/database/repo") + XCTAssertEqual(discovered.first?.status, .processing) + } + func testCodexLatestTerminalTurnTimestampPrefersNewestTerminalEvent() throws { let transcript = [ #"{"timestamp":"2026-04-09T03:17:16.000Z","type":"event_msg","payload":{"type":"task_complete","turn_id":"turn-1"}}"#, @@ -89,4 +691,48 @@ final class AppStateCodexTranscriptTests: XCTestCase { plain.formatOptions = [.withInternetDateTime] return try XCTUnwrap(plain.date(from: raw)) } + + private func makeCodexRollout( + sessionsBase: URL, + sessionId: String, + cwd: String?, + now: Date, + originator: String = "Codex Desktop", + source: String = "vscode", + filenameTimestamp: String = "2000-01-01T00-00-00" + ) throws -> URL { + let calendar = Calendar.current + let dayDirectory = sessionsBase + .appendingPathComponent(String(format: "%04d", calendar.component(.year, from: now))) + .appendingPathComponent(String(format: "%02d", calendar.component(.month, from: now))) + .appendingPathComponent(String(format: "%02d", calendar.component(.day, from: now))) + try FileManager.default.createDirectory(at: dayDirectory, withIntermediateDirectories: true) + + var payload: [String: Any] = [ + "id": sessionId, + "originator": originator, + "source": source, + ] + if let cwd { + payload["cwd"] = cwd + } + let sessionMeta = try JSONSerialization.data(withJSONObject: [ + "type": "session_meta", + "payload": payload, + ]) + let started = Data(#"{"type":"event_msg","payload":{"type":"task_started"}}"#.utf8) + var contents = sessionMeta + contents.append(0x0A) + contents.append(started) + contents.append(0x0A) + + let rollout = dayDirectory + .appendingPathComponent("rollout-\(filenameTimestamp)-\(sessionId).jsonl") + try contents.write(to: rollout) + try FileManager.default.setAttributes( + [.modificationDate: now], + ofItemAtPath: rollout.path + ) + return rollout + } }