From 70ae23cc826b44f399c0a3c59d01b68206dd66ec Mon Sep 17 00:00:00 2001 From: onevcat Date: Tue, 29 Sep 2026 01:21:32 +0900 Subject: [PATCH] Read daemon-owned rollouts for hand-typed Codex A Codex 0.157+ TUI attached to the shared daemon holds no rollout, so the log provider found nothing and detection relied on the screen alone. - When the TUI's open files are complete and hold no rollout, read the rollouts the pane drives through the daemon: the root thread of the pane's newest indexed submit and its descendants. The session log must belong to the current TUI process. - Start a newly bound rollout at the live task_started of that submit. Binding happens after the user item lands, which is after the turn started, so starting at the file end would miss the open turn. - Honor the live boundary of forked rollouts in the thread index. - Record the change in docs-ai 073.002. --- .../000-plan.md | 6 +- .../002-daemon-log-detection.md | 47 ++++++++ docs-ai/README.md | 2 +- docs/components/agent-detection.md | 9 ++ supacode/CLIService/CLICommandContext.swift | 6 +- .../CLIService/CodexDaemonThreadMapper.swift | 107 ++++++++++++++++-- .../AgentDetectionCoordinator.swift | 11 +- ...WorktreeTerminalState+AgentDetection.swift | 2 +- .../AgentDetection/CodexLogProvider.swift | 72 ++++++++++-- .../AgentDetection/CodexTUISessionLog.swift | 10 ++ .../CodexDaemonThreadMapperTests.swift | 68 +++++++++++ supacodeTests/CodexLogProviderTests.swift | 69 +++++++++++ 12 files changed, 377 insertions(+), 32 deletions(-) create mode 100644 docs-ai/073-codex-daemon-caller-identity/002-daemon-log-detection.md diff --git a/docs-ai/073-codex-daemon-caller-identity/000-plan.md b/docs-ai/073-codex-daemon-caller-identity/000-plan.md index 44896c052..214a46a61 100644 --- a/docs-ai/073-codex-daemon-caller-identity/000-plan.md +++ b/docs-ai/073-codex-daemon-caller-identity/000-plan.md @@ -47,8 +47,8 @@ Observed impact for a Codex TUI typed by hand (verified live with Codex 0.158.0) ### Non-goals -- Log-based working/idle detection for daemon-mode Codex. The screen fallback already - covers it (#838). The mapper built here can back that later. +- Log-based working/idle detection for daemon-mode Codex. (Corrected: delivered as a + stacked follow-up that reuses the mapper; see 002.) - Fixing the other inherited variables (`PROWL_WORKTREE_PATH`, `SSH_AUTH_SOCK`, …). - Forcing `--no-daemon` on user-typed commands. @@ -110,3 +110,5 @@ The spike that validated the session-log mapping lives on branch `spike/codex-daemon-log-detection` (`scripts/spikes/codex-daemon-map.py`). ## Amendments + +- Updated 2026-09-29: daemon-mode Codex log detection reuses the thread mapper — see [002-daemon-log-detection.md](002-daemon-log-detection.md) diff --git a/docs-ai/073-codex-daemon-caller-identity/002-daemon-log-detection.md b/docs-ai/073-codex-daemon-caller-identity/002-daemon-log-detection.md new file mode 100644 index 000000000..4b215e3ef --- /dev/null +++ b/docs-ai/073-codex-daemon-caller-identity/002-daemon-log-detection.md @@ -0,0 +1,47 @@ +# 073.002 — Daemon-Mode Codex Log Detection + +## Context + +`CodexLogProvider` found rollouts through the open files of the pane's Codex TUI. A TUI +attached to Codex's shared daemon holds none, so the state machine reported +`screen.noLiveTurn` and relied on the screen alone. That is how the 2026-09-28 symptom +(#838) became visible: the screen rules missed a Codex 0.158 footer and nothing else knew +the turn was open. + +## Change + +- `CodexDaemonThreadMapper.binding(surfaceID:daemonPID:tuiStartedAt:)` returns the rollouts + a pane drives: the root thread of its newest indexed submit plus every descendant the + daemon holds open. The session log must belong to the current TUI process (its header + within 120 s after the process start), so a log left by an earlier TUI cannot bind. +- The rollout index now honors the live boundary of forked rollouts (client ids and + `task_started` count only after the rollout's own `thread_settings_applied`) and records + the byte offset of the live `task_started` that precedes each client id. +- `CodexLogProvider` falls back to that binding when the TUI's inventory is complete and + holds no rollout. It finds the daemon through `daemon.pid` under the TUI's `CODEX_HOME` + and checks that the pid is still the managed daemon. A newly discovered bound rollout + starts at that turn offset instead of its end, because binding happens only after the + turn's user item is written, which is after `task_started`. +- `AgentDetectionCoordinator` passes its pane's surface id to the provider. + +Measured before the change: `/fork` in Codex 0.158 writes only the fork's own items, and a +forked subagent copies `task_started` but not user items before its live boundary. + +## Refs + +PR #840 (stacked on #839). + +## Current state + +Live acceptance in a Debug instance with an isolated `CODEX_HOME` daemon, two hand-typed +Codex panes in one cwd: + +| Scenario | `detection_reason` | +| --- | --- | +| Before the first submit | `screen.noLiveTurn` (unbound) | +| Concurrent turns in both panes | each `log.openWork`, then `log.turnEnded` at its own end | +| Subagent running while the parent waits | `log.openWork` throughout | +| `/new` then a new turn | `log.openWork` on the new thread | +| Steered message during a turn | `log.openWork` throughout | +| Restart with `codex resume` of an older thread, then a turn | `log.openWork` (live offset) | +| Esc interrupt | `log.turnEnded` | diff --git a/docs-ai/README.md b/docs-ai/README.md index ef1589c11..6ed92f83b 100644 --- a/docs-ai/README.md +++ b/docs-ai/README.md @@ -132,4 +132,4 @@ agent-facing manual for that). | 070 | [app-localization](070-app-localization/000-plan.md) | 2026-09-18 | Simplified Chinese UI from a String Catalog; translations sync at release time through the `sync-l10n` skill (nothing blocks everyday work); compiler-based coverage audit plus a baseline for unlocalized copy; living [glossary](070-app-localization/glossary.md); language choice stored only in the per-app `AppleLanguages` default | | 071 | [git-environment-recovery](071-git-environment-recovery/000-plan.md) | 2026-09-20 | Working Git selection, safe repository classification, recovery guidance, and plain-folder Shelf entry | | 072 | [workspace-editing](072-workspace-editing/000-plan.md) | 2026-09-25 | Edit a workspace after creation: metadata, member repositories (add, remove, reorder, roles) through one editor sheet reached from the sidebar, detail view, Settings, palette, and menu | -| 073 | [codex-daemon-caller-identity](073-codex-daemon-caller-identity/000-plan.md) | 2026-09-29 | CLI callers under Codex's shared app-server daemon: cut the ancestry at the daemon and map `CODEX_THREAD_ID` to the driving pane through the TUI session log; `prowl list` reports the caller | +| 073 | [codex-daemon-caller-identity](073-codex-daemon-caller-identity/000-plan.md) | 2026-09-29 | CLI callers under Codex's shared app-server daemon: cut the ancestry at the daemon and map `CODEX_THREAD_ID` to the driving pane through the TUI session log; `prowl list` reports the caller; the same mapping restores daemon-mode Codex log detection | diff --git a/docs/components/agent-detection.md b/docs/components/agent-detection.md index e42fe3499..b08e87a0e 100644 --- a/docs/components/agent-detection.md +++ b/docs/components/agent-detection.md @@ -111,6 +111,15 @@ at a `~/.grok/` install (so Cursor's own `agent` entrypoint stays Cursor). screen. Attribution is heuristic: switching to an old quiet chat before entering a new prompt can temporarily retain the previous log candidate. + A `codex` typed by hand in Codex 0.157+ attaches to a shared background daemon + that owns every session log, so the TUI process holds none. Prowl then reads the + daemon's logs for the thread this pane's Codex last submitted to, identified + through the pane's Codex session log (see + [cli identity](cli.md#identity-which-pane-am-i)), plus that thread's subagents. + Until the pane's Codex submits its first message, and for a moment after each + new submit, detection falls back to the screen. A Codex launched from an Agent + Profile or with `--no-daemon` owns its logs and is read directly. + Claude reads the detected PID's native registry under its configured root (`~/.claude/sessions` by default). `busy` and `shell` mean **Working**, including assigned child and background shell work; `waiting` means **Blocked**, and `idle` diff --git a/supacode/CLIService/CLICommandContext.swift b/supacode/CLIService/CLICommandContext.swift index 437037693..a1d941ed2 100644 --- a/supacode/CLIService/CLICommandContext.swift +++ b/supacode/CLIService/CLICommandContext.swift @@ -76,12 +76,8 @@ nonisolated struct CallerPane: Sendable, Equatable { func belongs(to generation: AgentProcessGeneration) -> Bool { if processAncestry.contains(generation) { return true } guard let codexSessionStartedAt else { return false } - let delay = codexSessionStartedAt.timeIntervalSince(generation.startedAt) - return delay >= -1 && delay <= Self.codexSessionStartWindow + return CodexTUISessionLog.belongs(sessionStartedAt: codexSessionStartedAt, toProcessStartedAt: generation.startedAt) } - - /// A TUI writes its session header during startup; a later header belongs to a later TUI. - static let codexSessionStartWindow: TimeInterval = 120 } /// The caller's ancestry up to, not including, a Codex managed daemon. diff --git a/supacode/CLIService/CodexDaemonThreadMapper.swift b/supacode/CLIService/CodexDaemonThreadMapper.swift index 6002b3873..aeb4faab5 100644 --- a/supacode/CLIService/CodexDaemonThreadMapper.swift +++ b/supacode/CLIService/CodexDaemonThreadMapper.swift @@ -6,6 +6,20 @@ nonisolated struct CodexDaemonRollout: Sendable, Equatable { let id: String let parentID: String? var clientIDs: [String] + var path = "" + /// A fork copies history before its own `thread_settings_applied`; that part is not live. + var isForked = false + /// Byte offset of the live `task_started` line that precedes each client message id. + var turnStartOffsets: [String: UInt64] = [:] +} + +/// The rollouts a pane's Codex TUI currently drives through the daemon (docs-ai 073.002). +nonisolated struct CodexDaemonBinding: Sendable, Equatable { + let rootID: String + /// The root thread and every descendant the daemon holds open. + let paths: [String] + /// Where reading the bound rollout starts: the turn that holds the pane's newest submit. + let liveOffsets: [String: UInt64] } /// The submits of the TUI that last started a pane's session log. @@ -40,6 +54,8 @@ actor CodexDaemonThreadMapper { var offset: UInt64 var pending = Data() var headerRead = false + var live = true + var lastTurnStart: UInt64? var rollout: Rollout? } @@ -72,22 +88,59 @@ actor CodexDaemonThreadMapper { return Self.resolve(threadID: threadID, rollouts: rollouts, logs: sessionLogs()) } + /// What the pane's current Codex TUI drives, or nil before its first indexed submit. The + /// session log must belong to the TUI process that started at `tuiStartedAt`. + func binding(surfaceID: UUID, daemonPID: pid_t, tuiStartedAt: Date) -> CodexDaemonBinding? { + let url = CodexTUISessionLog.url(for: surfaceID, in: sessionLogDirectory) + guard let size = (try? url.resourceValues(forKeys: [.fileSizeKey]))?.fileSize, size <= sessionLogLimit, + let data = try? Data(contentsOf: url), + let log = Self.parseSessionLog(data, surfaceID: surfaceID), + CodexTUISessionLog.belongs(sessionStartedAt: log.startedAt, toProcessStartedAt: tuiStartedAt), + !log.submits.isEmpty, + let paths = openFilePaths(daemonPID), + let rollouts = refreshRollouts(paths.filter(Self.isRolloutPath)) + else { return nil } + return Self.binding(for: log, rollouts: rollouts) + } + // MARK: - Resolution - static func resolve(threadID: String, rollouts: [Rollout], logs: [SessionLog]) -> CodexThreadPane? { + static func binding(for log: SessionLog, rollouts: [Rollout]) -> CodexDaemonBinding? { let byID = Dictionary(rollouts.map { ($0.id, $0) }, uniquingKeysWith: { first, _ in first }) + let owner = owners(rollouts) + guard let submit = log.submits.last(where: { owner[$0.clientID] != nil }), + let bound = owner[submit.clientID].flatMap({ byID[$0] }) + else { return nil } + let rootID = root(of: bound.id, in: byID) + let family = rollouts.filter { root(of: $0.id, in: byID) == rootID } + return CodexDaemonBinding( + rootID: rootID, + paths: family.map(\.path).sorted(), + liveOffsets: bound.turnStartOffsets[submit.clientID].map { [bound.path: $0] } ?? [:] + ) + } + + private static func owners(_ rollouts: [Rollout]) -> [String: String] { var owner: [String: String] = [:] for rollout in rollouts { for clientID in rollout.clientIDs { owner[clientID] = rollout.id } } - func root(_ id: String) -> String { - var current = id - var visited: Set = [] - while let parent = byID[current]?.parentID, visited.insert(current).inserted { - current = parent - } - return current + return owner + } + + private static func root(of id: String, in byID: [String: Rollout]) -> String { + var current = id + var visited: Set = [] + while let parent = byID[current]?.parentID, visited.insert(current).inserted { + current = parent } + return current + } + + static func resolve(threadID: String, rollouts: [Rollout], logs: [SessionLog]) -> CodexThreadPane? { + let byID = Dictionary(rollouts.map { ($0.id, $0) }, uniquingKeysWith: { first, _ in first }) + let owner = owners(rollouts) + func root(_ id: String) -> String { Self.root(of: id, in: byID) } let target = root(threadID) let bound = logs.compactMap { log -> (log: SessionLog, submittedAt: Date)? in guard let submit = log.submits.last(where: { owner[$0.clientID] != nil }), @@ -182,25 +235,35 @@ actor CodexDaemonThreadMapper { } catch { return nil } - Self.consume(&cursor) + Self.consume(&cursor, path: path) next[path] = cursor } cursors = next return next.values.compactMap(\.rollout) } - private static func consume(_ cursor: inout Cursor) { + private static func consume(_ cursor: inout Cursor, path: String) { + let base = cursor.offset - UInt64(cursor.pending.count) var start = cursor.pending.startIndex while let end = cursor.pending[start...].firstIndex(of: 10) { let line = cursor.pending[start.. [String: Any]? { + guard line.range(of: Data("\"\(type)\"".utf8)) != nil, + let record = try? JSONSerialization.jsonObject(with: line) as? [String: Any], + record["type"] as? String == "event_msg", + let payload = record["payload"] as? [String: Any], + payload["type"] as? String == type + else { return nil } + return payload + } + + static func isTaskStarted(_ line: Data) -> Bool { + eventPayload(line, type: "task_started") != nil + } + + static func isLiveBoundary(_ line: Data, threadID: String) -> Bool { + eventPayload(line, type: "thread_settings_applied")?["thread_id"] as? String == threadID } static func userMessageClientID(_ line: Data) -> String? { diff --git a/supacode/Features/Terminal/BusinessLogic/AgentDetectionCoordinator.swift b/supacode/Features/Terminal/BusinessLogic/AgentDetectionCoordinator.swift index 514a33f33..3d8533074 100644 --- a/supacode/Features/Terminal/BusinessLogic/AgentDetectionCoordinator.swift +++ b/supacode/Features/Terminal/BusinessLogic/AgentDetectionCoordinator.swift @@ -20,7 +20,14 @@ final class AgentDetectionCoordinator { private var interactionRevision: UInt64 = 0 private var now: TimeInterval { time() } - init(sample: Sample? = nil, time: @escaping () -> TimeInterval = { ProcessInfo.processInfo.systemUptime }) { + private let surfaceID: UUID? + + init( + surfaceID: UUID? = nil, + sample: Sample? = nil, + time: @escaping () -> TimeInterval = { ProcessInfo.processInfo.systemUptime } + ) { + self.surfaceID = surfaceID sampleOverride = sample self.time = time } @@ -82,7 +89,7 @@ final class AgentDetectionCoordinator { self.agent = agent self.process = process self.configRoot = configRoot - if agent == .codex, process != nil { logProvider = CodexLogProvider() } + if agent == .codex, process != nil { logProvider = CodexLogProvider(surfaceID: surfaceID) } if agent == .claude, process != nil { nativeProvider = ClaudeRuntimeProvider() } } let expectedRevision = revision diff --git a/supacode/Features/Terminal/Models/WorktreeTerminalState+AgentDetection.swift b/supacode/Features/Terminal/Models/WorktreeTerminalState+AgentDetection.swift index c32e67696..41e830d87 100644 --- a/supacode/Features/Terminal/Models/WorktreeTerminalState+AgentDetection.swift +++ b/supacode/Features/Terminal/Models/WorktreeTerminalState+AgentDetection.swift @@ -130,7 +130,7 @@ extension WorktreeTerminalState { previous.seen = current.seen previous.lastChangedAt = current.lastChangedAt guard previous == current else { return true } - let coordinator = agentDetectionCoordinators[surfaceID] ?? AgentDetectionCoordinator() + let coordinator = agentDetectionCoordinators[surfaceID] ?? AgentDetectionCoordinator(surfaceID: surfaceID) agentDetectionCoordinators[surfaceID] = coordinator let process = Self.processGeneration(identified) diff --git a/supacode/Infrastructure/AgentDetection/CodexLogProvider.swift b/supacode/Infrastructure/AgentDetection/CodexLogProvider.swift index f0999a6bb..d66c96a47 100644 --- a/supacode/Infrastructure/AgentDetection/CodexLogProvider.swift +++ b/supacode/Infrastructure/AgentDetection/CodexLogProvider.swift @@ -29,36 +29,56 @@ actor CodexLogProvider { private var cursors: [URL: Cursor] = [:] private var needsBaseline = false private let byteLimit = 8 * 1_024 * 1_024 + /// Codex 0.157+ TUIs attached to the shared daemon hold no rollout; the daemon does + /// (docs-ai 073.002). Nil for a caller that has no pane. + private let daemonBinding: DaemonBinding? + + typealias DaemonBinding = @Sendable (AgentProcessGeneration, URL?) async -> CodexDaemonBinding? init( + surfaceID: UUID? = nil, startedAt: Date = Date(), time: @escaping @Sendable () -> TimeInterval = { ProcessInfo.processInfo.systemUptime }, - diagnostic: @escaping @Sendable (String) -> Void = { SupaLogger("AgentDetection").warning($0) } + diagnostic: @escaping @Sendable (String) -> Void = { SupaLogger("AgentDetection").warning($0) }, + daemonBinding: DaemonBinding? = nil ) { self.startedAt = startedAt self.time = time self.diagnostic = diagnostic + self.daemonBinding = daemonBinding ?? surfaceID.map(Self.daemonBinding(surfaceID:)) } - func sample(process: AgentProcessGeneration, configRoot: URL?) -> [AgentDetectionEvent] { + func sample(process: AgentProcessGeneration, configRoot: URL?) async -> [AgentDetectionEvent] { processID = process.pid guard ProcessDetection.processStartDate(pid: process.pid) == process.startedAt else { return [.unavailable] } let parse = AgentSessionResolver.pathParser(profile: .profile(for: .codex), configRoot: configRoot) var complete = false - let paths = ProcessDetection.openFilePaths(pid: process.pid, complete: &complete) + var paths = ProcessDetection.openFilePaths(pid: process.pid, complete: &complete) .compactMap { parse($0)?.transcriptPath } - let events = sample(paths: paths, inventoryComplete: complete) + var liveOffsets: [URL: UInt64] = [:] + // An embedded TUI owns its rollouts. A TUI that owns none may drive daemon threads. + if complete, paths.isEmpty, let daemonBinding, let binding = await daemonBinding(process, configRoot) { + paths = binding.paths.map { URL(filePath: $0, directoryHint: .notDirectory) } + for (path, offset) in binding.liveOffsets { + liveOffsets[URL(filePath: path, directoryHint: .notDirectory)] = offset + } + } + let events = sample(paths: paths, inventoryComplete: complete, liveOffsets: liveOffsets) guard ProcessDetection.processStartDate(pid: process.pid) == process.startedAt else { return [.unavailable] } return events } - func sample(paths: [URL], inventoryComplete: Bool = true) -> [AgentDetectionEvent] { + /// `liveOffsets` start a newly discovered rollout at a turn boundary instead of its end, so + /// a daemon thread bound after its turn began still reports that turn. + func sample( + paths: [URL], inventoryComplete: Bool = true, liveOffsets: [URL: UInt64] = [:] + ) -> [AgentDetectionEvent] { guard inventoryComplete else { report("incompleteInventory") return [.suspended] } do { - let events = try read(paths: Array(Set(paths))) + let events = try read(paths: Array(Set(paths)), liveOffsets: liveOffsets) report(nil) return events } catch Failure.notReady { @@ -93,7 +113,7 @@ actor CodexLogProvider { "Log provider pid=\(pid) status=\(failure ?? "recovered") cursors=\(cursors.count) detail=\(detail ?? "none")") } - private func read(paths: [URL]) throws -> [AgentDetectionEvent] { + private func read(paths: [URL], liveOffsets: [URL: UInt64]) throws -> [AgentDetectionEvent] { guard paths.count <= 32 else { throw Failure.incomplete } var next = cursors.filter { paths.contains($0.key) } var remaining = byteLimit @@ -112,9 +132,12 @@ actor CodexLogProvider { let createdAt = attributes[.creationDate] as? Date else { throw Failure.incomplete } let baseline = needsBaseline || createdAt < startedAt + // A live offset points at a `task_started` after any inherited history. + let liveOffset = liveOffsets[path].flatMap { $0 > UInt64(end) && $0 <= size ? $0 : nil } next[path] = Cursor( - metadata: metadata, inode: inode, offset: baseline ? size : UInt64(end + 1), - decoder: CodexLogDecoder(sessionID: metadata.id, isLive: baseline || !metadata.hasInheritedHistory)) + metadata: metadata, inode: inode, offset: liveOffset ?? (baseline ? size : UInt64(end + 1)), + decoder: CodexLogDecoder( + sessionID: metadata.id, isLive: liveOffset != nil || baseline || !metadata.hasInheritedHistory)) } // Cache discovery separately from event consumption. Unsupported lineage // suspends authority without rereading headers or discarding observed work. @@ -176,6 +199,37 @@ actor CodexLogProvider { return events } + private static func daemonBinding(surfaceID: UUID) -> DaemonBinding { + { process, configRoot in + let home = configRoot ?? codexHome(forTUI: process.pid) + guard let daemonPID = managedDaemonPID(codexHome: home) else { return nil } + return await CodexDaemonThreadMapper.shared.binding( + surfaceID: surfaceID, daemonPID: daemonPID, tuiStartedAt: process.startedAt) + } + } + + /// The TUI's own `CODEX_HOME` selects its daemon socket, so it also selects the daemon. + nonisolated static func codexHome(forTUI pid: pid_t) -> URL { + if let home = ProcessDetection.processEnvironmentValue(pid: pid, name: "CODEX_HOME"), home.hasPrefix("/") { + return URL(filePath: home, directoryHint: .isDirectory) + } + return FileManager.default.homeDirectoryForCurrentUser.appending(path: ".codex", directoryHint: .isDirectory) + } + + /// `daemon.pid` names the managed daemon; the pid must still be that daemon. + nonisolated static func managedDaemonPID( + codexHome: URL, + isManagedDaemon: (pid_t) -> Bool = ProcessDetection.isCodexManagedDaemon + ) -> pid_t? { + let url = codexHome.appending(path: "app-server-daemon/daemon.pid", directoryHint: .notDirectory) + guard let data = try? Data(contentsOf: url), + let record = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let pid = (record["pid"] as? NSNumber).map({ pid_t(truncating: $0) }), pid > 1, + isManagedDaemon(pid) + else { return nil } + return pid + } + private func metadata(from data: Data) throws -> Metadata { guard let record = try JSONSerialization.jsonObject(with: data) as? [String: Any], record["type"] as? String == "session_meta", diff --git a/supacode/Infrastructure/AgentDetection/CodexTUISessionLog.swift b/supacode/Infrastructure/AgentDetection/CodexTUISessionLog.swift index 97464a054..e6d00b223 100644 --- a/supacode/Infrastructure/AgentDetection/CodexTUISessionLog.swift +++ b/supacode/Infrastructure/AgentDetection/CodexTUISessionLog.swift @@ -14,6 +14,16 @@ nonisolated enum CodexTUISessionLog { /// Debug and Release share the directory, so launch only removes stale files. static let staleAge: TimeInterval = 7 * 24 * 60 * 60 + /// A TUI writes its session header during startup; a later header belongs to a later TUI. + static let processStartWindow: TimeInterval = 120 + + /// Whether a log whose header was written at `sessionStartedAt` came from the process that + /// started at `processStartedAt`. Each launch truncates the log, so a stale log fails this. + static func belongs(sessionStartedAt: Date, toProcessStartedAt processStartedAt: Date) -> Bool { + let delay = sessionStartedAt.timeIntervalSince(processStartedAt) + return delay >= -1 && delay <= processStartWindow + } + static func url(for surfaceID: UUID, in directory: URL = SupacodePaths.codexTUISessionLogDirectory) -> URL { directory.appending(path: "\(surfaceID.uuidString).\(fileExtension)", directoryHint: .notDirectory) } diff --git a/supacodeTests/CodexDaemonThreadMapperTests.swift b/supacodeTests/CodexDaemonThreadMapperTests.swift index 36cc80f39..5fd1db385 100644 --- a/supacodeTests/CodexDaemonThreadMapperTests.swift +++ b/supacodeTests/CodexDaemonThreadMapperTests.swift @@ -162,6 +162,74 @@ struct CodexDaemonThreadMapperTests { #expect(await incomplete.pane(threadID: "t2", daemonPID: 1) == nil) } + @Test func bindingCoversTheRootFamilyAndStartsAtTheBoundTurn() { + var root = Self.rollout("root", ["a1"]) + root.path = "/r" + root.turnStartOffsets = ["a1": 100] + var child = Self.rollout("child", parent: "root", []) + child.path = "/c" + var other = Self.rollout("other", ["b1"]) + other.path = "/o" + let rollouts = [root, child, other] + + let binding = Mapper.binding(for: Self.log(Self.paneA, [("a1", 1), ("a2-pending", 2)]), rollouts: rollouts) + + #expect(binding == CodexDaemonBinding(rootID: "root", paths: ["/c", "/r"], liveOffsets: ["/r": 100])) + #expect(Mapper.binding(for: Self.log(Self.paneA, [("pending", 1)]), rollouts: rollouts) == nil) + } + + @Test func indexRecordsTurnStartsAndSkipsInheritedHistory() async throws { + let directory = FileManager.default.temporaryDirectory.appending(path: "codex-bind-\(UUID().uuidString)") + let logs = directory.appending(path: "logs") + try FileManager.default.createDirectory(at: logs, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + func event(_ type: String, _ extra: String = "") -> String { + #"{"type":"event_msg","payload":{"type":""# + type + #"""# + extra + "}}" + } + func user(_ id: String) -> String { + event("item_completed", #","item":{"type":"UserMessage","client_id":""# + id + #""}"#) + } + let main = directory.appending(path: "rollout-main.jsonl") + let mainHead = [ + #"{"type":"session_meta","payload":{"id":"main","source":"vscode"}}"#, + event("task_started", #","turn_id":"1""#), user("a1"), event("task_complete", #","turn_id":"1""#), "", + ].joined(separator: "\n") + let mainTail = [event("task_started", #","turn_id":"2""#), user("a2"), ""].joined(separator: "\n") + try Data((mainHead + mainTail).utf8).write(to: main) + let fork = directory.appending(path: "rollout-fork.jsonl") + let forkHead = [ + #"{"type":"session_meta","payload":{"id":"fork","forked_from_id":"main","source":"vscode"}}"#, + event("task_started", #","turn_id":"old""#), user("copied"), + event("thread_settings_applied", #","thread_id":"fork""#), "", + ].joined(separator: "\n") + let forkTail = [event("task_started", #","turn_id":"3""#), user("f1"), ""].joined(separator: "\n") + try Data((forkHead + forkTail).utf8).write(to: fork) + let tuiStart = Date(timeIntervalSince1970: 1_790_605_490) + let start = #"{"ts":"2026-09-28T14:25:00.000Z","kind":"session_start"}"# + let submit = + #"{"ts":"2026-09-28T14:25:05.000Z","kind":"op","payload":{"UserTurn":{"client_user_message_id":"f1"}}}"# + try Data([start, submit, ""].joined(separator: "\n").utf8).write( + to: CodexTUISessionLog.url(for: Self.paneA, in: logs)) + let open = [main, fork].map { $0.path(percentEncoded: false) } + let mapper = CodexDaemonThreadMapper(sessionLogDirectory: logs, openFilePaths: { _ in open }) + + let binding = await mapper.binding(surfaceID: Self.paneA, daemonPID: 1, tuiStartedAt: tuiStart) + #expect(binding?.rootID == "fork") + #expect(binding?.paths == [open[1]]) + #expect(binding?.liveOffsets == [open[1]: UInt64(forkHead.utf8.count)]) + #expect(await mapper.pane(threadID: "main", daemonPID: 1) == nil) + let late = await mapper.binding( + surfaceID: Self.paneA, daemonPID: 1, tuiStartedAt: tuiStart.addingTimeInterval(-3_600)) + #expect(late == nil) + + let submitB = submit.replacing("f1", with: "a2") + try Data([start, submitB, ""].joined(separator: "\n").utf8).write( + to: CodexTUISessionLog.url(for: Self.paneB, in: logs)) + let mainBinding = await mapper.binding(surfaceID: Self.paneB, daemonPID: 1, tuiStartedAt: tuiStart) + #expect(mainBinding?.rootID == "main") + #expect(mainBinding?.liveOffsets == [open[0]: UInt64(mainHead.utf8.count)]) + } + @Test func preparingTheLogDirectoryRemovesOnlyStaleLogs() throws { let directory = FileManager.default.temporaryDirectory.appending(path: "codex-logs-\(UUID().uuidString)") defer { try? FileManager.default.removeItem(at: directory) } diff --git a/supacodeTests/CodexLogProviderTests.swift b/supacodeTests/CodexLogProviderTests.swift index d83e39dcf..70c42449a 100644 --- a/supacodeTests/CodexLogProviderTests.swift +++ b/supacodeTests/CodexLogProviderTests.swift @@ -231,4 +231,73 @@ struct CodexLogProviderTests { let malformed = await provider.sample(paths: [path]) if case .unavailable = malformed.first {} else { Issue.record("Malformed append must invalidate") } } + + private func startedTurns(_ events: [AgentDetectionEvent]) -> [String] { + events.compactMap { event in + if case .turnStarted(_, let turn) = event { return turn } + return nil + } + } + + private func complete(_ turn: String) -> String { + #"{"type":"event_msg","payload":{"type":"task_complete","turn_id":"\#(turn)"}}"# + "\n" + } + + // A daemon thread is bound only after its user item lands, which is after `task_started`. + // Starting a pre-existing rollout at its end would miss that open turn. + @Test func liveOffsetStartsAPreexistingRolloutAtTheBoundTurn() async throws { + let directory = try fixture() + defer { try? FileManager.default.removeItem(at: directory) } + let path = directory.appending(path: "a.jsonl") + let prefix = header("a") + start("t1") + complete("t1") + try (prefix + start("t2")).write(to: path, atomically: false, encoding: .utf8) + let offset = UInt64(prefix.utf8.count) + + let baselined = CodexLogProvider(startedAt: .distantFuture) + #expect(startedTurns(await baselined.sample(paths: [path])).isEmpty) + let bound = CodexLogProvider(startedAt: .distantFuture) + #expect(startedTurns(await bound.sample(paths: [path], liveOffsets: [path: offset])) == ["t2"]) + let invalid = CodexLogProvider(startedAt: .distantFuture) + #expect(startedTurns(await invalid.sample(paths: [path], liveOffsets: [path: 1 << 40])).isEmpty) + } + + @Test func aTUIWithoutOwnRolloutsReadsItsDaemonBinding() async throws { + let directory = try fixture() + defer { try? FileManager.default.removeItem(at: directory) } + let path = directory.appending(path: "rollout-daemon.jsonl") + let prefix = header("root") + try (prefix + start("t1")).write(to: path, atomically: false, encoding: .utf8) + let started = try #require(ProcessDetection.processStartDate(pid: getpid())) + let process = AgentProcessGeneration(pid: getpid(), startedAt: started) + let requested = Mutex<[URL?]>([]) + let configRoot = directory.appending(path: "home") + let provider = CodexLogProvider( + startedAt: .distantFuture, + daemonBinding: { generation, root in + #expect(generation == process) + requested.withLock { $0.append(root) } + return CodexDaemonBinding( + rootID: "root", paths: [path.path(percentEncoded: false)], + liveOffsets: [path.path(percentEncoded: false): UInt64(prefix.utf8.count)]) + }) + + let events = await provider.sample(process: process, configRoot: configRoot) + + #expect(startedTurns(events) == ["t1"]) + #expect(requested.withLock { $0 } == [configRoot as URL?]) + } + + @Test func managedDaemonPIDRequiresALiveDaemonProcess() throws { + let home = try fixture() + defer { try? FileManager.default.removeItem(at: home) } + let directory = home.appending(path: "app-server-daemon") + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + try Data(#"{"pid":4242,"processStartTime":"x"}"#.utf8).write(to: directory.appending(path: "daemon.pid")) + + #expect(CodexLogProvider.managedDaemonPID(codexHome: home, isManagedDaemon: { $0 == 4242 }) == 4242) + #expect(CodexLogProvider.managedDaemonPID(codexHome: home, isManagedDaemon: { _ in false }) == nil) + #expect( + CodexLogProvider.managedDaemonPID(codexHome: home.appending(path: "missing"), isManagedDaemon: { _ in true }) + == nil) + } }