From 416b2f7b69f443126eb98194ca2779a5c2b3ae7c Mon Sep 17 00:00:00 2001 From: "shuigao.fh" Date: Mon, 31 Aug 2026 21:01:45 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=90=9B=20fix(remote):=20forward=20Cod?= =?UTF-8?q?ex=20usage=20snapshots=20over=20SSH?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - emit cached Codex quota and token snapshots from the remote bridge\n- merge remote snapshots by freshness and isolate per-host token baselines\n- cover the wire protocol and document forwarded usage data --- AGENTS.md | 1 + .../Remote/RemoteConnectorManager.swift | 37 +++ .../Services/Session/SessionMonitor.swift | 45 ++- .../Services/Usage/AgentUsageAnalytics.swift | 4 +- PingIsland/Services/Usage/CodexUsage.swift | 15 + .../AgentUsageAnalyticsTests.swift | 37 +++ PingIslandTests/CodexUsageLoaderTests.swift | 20 ++ .../RemoteHookConfigurationTests.swift | 44 +++ Prototype/Sources/IslandBridge/main.swift | 311 ++++++++++++++++++ .../IslandTests/IslandBridgeE2ETests.swift | 98 +++++- README.md | 4 +- docs/privacy-policy.md | 5 +- 12 files changed, 609 insertions(+), 12 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index e924e06d..98411539 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -51,6 +51,7 @@ This file is a routing layer for coding agents working in this repo. Keep it sho - Remote SSH forwarding and remote-host management: `PingIsland/Services/Remote/` - Remote hosts can bootstrap a bridge on the SSH target, rewrite remote hooks, install managed plugin-directory integrations such as Hermes under the remote home directory, and attach a bidirectional forwarding channel back into PingIsland - The remote bridge forwards recent Codex app-server thread activity from the SSH target's `~/.codex/state_*.sqlite` through the existing remote hook-event channel + - The remote bridge also tails recent Codex rollout quota snapshots and emits `codex_usage` messages. Keep local and remote snapshots source-qualified, choose the newest `capturedAt` for the account-wide usage UI, and resend the current remote snapshot whenever a control client attaches. - Provider/client routing: bridge envelopes are normalized in `PingIsland/Services/Hooks/HookSocketServer.swift`, stored on `SessionState`, and launched via `PingIsland/Services/Window/SessionLauncher.swift` - Client profile registry: installable hook clients and runtime client branding / recognition are centralized in `PingIsland/Models/ClientProfile.swift` - VS Code-compatible IDE focus extension install / URI launch: `PingIsland/Services/Window/IDEExtensionInstaller.swift`, `PingIsland/Services/Window/TerminalSessionFocuser.swift` diff --git a/PingIsland/Services/Remote/RemoteConnectorManager.swift b/PingIsland/Services/Remote/RemoteConnectorManager.swift index d3aaa477..13ce5397 100644 --- a/PingIsland/Services/Remote/RemoteConnectorManager.swift +++ b/PingIsland/Services/Remote/RemoteConnectorManager.swift @@ -15,6 +15,7 @@ final class RemoteConnectorManager: ObservableObject { private let persistenceKey = "RemoteConnectorManager.endpoints.v1" private var eventHandler: (@Sendable (HookEvent) -> Void)? + private var codexUsageHandler: (@Sendable (CodexUsageSnapshot) -> Void)? private var permissionFailureHandler: (@Sendable (_ sessionId: String, _ toolUseId: String) -> Void)? private var connectors: [UUID: RemoteAttachConnector] = [:] private var pendingRequests = RemotePendingRequestStore() @@ -31,9 +32,11 @@ final class RemoteConnectorManager: ObservableObject { func start( onEvent: @escaping @Sendable (HookEvent) -> Void, + onCodexUsage: (@Sendable (CodexUsageSnapshot) -> Void)? = nil, onPermissionFailure: (@Sendable (_ sessionId: String, _ toolUseId: String) -> Void)? = nil ) { eventHandler = onEvent + codexUsageHandler = onCodexUsage permissionFailureHandler = onPermissionFailure guard !hasStarted else { return } @@ -577,6 +580,22 @@ final class RemoteConnectorManager: ObservableObject { } setState(for: endpointID, phase: .connected, detail: "远程转发已连接", agentVersion: hello.version) + case .codexUsage(let usageMessage): + let endpoint = endpoint(for: endpointID) + let remoteSource = Self.remoteUsageSourcePath( + usageMessage.payload.sourceFilePath, + endpoint: endpoint + ) + let snapshot = CodexUsageSnapshot( + sourceFilePath: remoteSource, + capturedAt: usageMessage.payload.capturedAt, + planType: usageMessage.payload.planType, + limitID: usageMessage.payload.limitID, + tokenUsage: usageMessage.payload.tokenUsage, + windows: usageMessage.payload.windows + ) + codexUsageHandler?(snapshot) + case .hookEvent(let eventMessage): let payload = eventMessage.payload guard let provider = SessionProvider(rawValue: payload.provider) else { @@ -646,6 +665,16 @@ final class RemoteConnectorManager: ObservableObject { } } + nonisolated static func remoteUsageSourcePath( + _ sourceFilePath: String, + endpoint: RemoteEndpoint? + ) -> String { + guard let endpoint else { return sourceFilePath } + let prefix = endpoint.sshURL?.absoluteString ?? "ssh://\(endpoint.sshDisplayTarget)" + let separator = sourceFilePath.hasPrefix("/") ? "" : "/" + return "\(prefix)\(separator)\(sourceFilePath)" + } + private func handleDisconnect(endpointID: UUID, error: Error?) { connectors.removeValue(forKey: endpointID) logger.error( @@ -1965,6 +1994,7 @@ private final class RemoteAttachConnector { private enum RemoteInboundMessage: Decodable { case hello(RemoteDaemonHello) case hookEvent(RemoteHookEventMessage) + case codexUsage(RemoteCodexUsageMessage) private enum CodingKeys: String, CodingKey { case type @@ -1978,12 +2008,19 @@ private enum RemoteInboundMessage: Decodable { self = .hello(try RemoteDaemonHello(from: decoder)) case "hook_event": self = .hookEvent(try RemoteHookEventMessage(from: decoder)) + case "codex_usage": + self = .codexUsage(try RemoteCodexUsageMessage(from: decoder)) default: throw RemoteConnectorError.invalidRemoteMessage } } } +struct RemoteCodexUsageMessage: Codable, Equatable, Sendable { + let type: String + let payload: CodexUsageSnapshot +} + private struct SSHExecutionResult { let stdout: String let stderr: String diff --git a/PingIsland/Services/Session/SessionMonitor.swift b/PingIsland/Services/Session/SessionMonitor.swift index ef35e7d9..8c495810 100644 --- a/PingIsland/Services/Session/SessionMonitor.swift +++ b/PingIsland/Services/Session/SessionMonitor.swift @@ -140,6 +140,11 @@ class SessionMonitor: ObservableObject { } RemoteConnectorManager.shared.start( onEvent: handleHookEvent, + onCodexUsage: { [weak self] snapshot in + Task { @MainActor in + await self?.applyRemoteCodexUsageSnapshot(snapshot) + } + }, onPermissionFailure: { sessionId, toolUseId in Task { await SessionStore.shared.process( @@ -336,7 +341,6 @@ class SessionMonitor: ObservableObject { UsageSnapshotCacheStore.saveClaude(claudeSnapshot) } if let codexSnapshot { - UsageSnapshotCacheStore.saveCodex(codexSnapshot) let sessionTitle: String? if let threadID = codexSnapshot.threadID { sessionTitle = await SessionStore.shared.session(for: threadID)?.displayTitle @@ -350,11 +354,44 @@ class SessionMonitor: ObservableObject { } self.claudeUsageSnapshot = claudeSnapshot ?? cachedClaudeSnapshot - self.codexUsageSnapshot = codexSnapshot ?? cachedCodexSnapshot + let preferredCodexSnapshot = CodexUsageSnapshot.newest([ + self.codexUsageSnapshot, + codexSnapshot, + cachedCodexSnapshot, + ]) + if let preferredCodexSnapshot { + UsageSnapshotCacheStore.saveCodex(preferredCodexSnapshot) + } + self.codexUsageSnapshot = preferredCodexSnapshot self.syncCodexThreadDiscovery(using: self.codexUsageSnapshot) } } + private func applyRemoteCodexUsageSnapshot(_ snapshot: CodexUsageSnapshot) async { + guard shouldRefreshUsage, AppSettings.showUsage else { return } + + let sessionTitle: String? + if let threadID = snapshot.threadID { + sessionTitle = await SessionStore.shared.session(for: threadID)?.displayTitle + } else { + sessionTitle = nil + } + await AgentUsageStore.shared.recordCodexUsageSnapshot( + snapshot, + sessionTitle: sessionTitle + ) + + let preferredSnapshot = CodexUsageSnapshot.newest([ + codexUsageSnapshot, + snapshot, + ]) + if let preferredSnapshot { + UsageSnapshotCacheStore.saveCodex(preferredSnapshot) + } + codexUsageSnapshot = preferredSnapshot + syncCodexThreadDiscovery(using: codexUsageSnapshot) + } + // MARK: - Native Runtime func startNativeSession(provider: SessionProvider, cwd: String, preferredSessionID: String? = nil) { @@ -1238,7 +1275,9 @@ class SessionMonitor: ObservableObject { } private func syncCodexThreadDiscovery(using snapshot: CodexUsageSnapshot?) { - guard let threadID = snapshot?.threadID else { return } + guard let snapshot, + !snapshot.isRemoteSource, + let threadID = snapshot.threadID else { return } Task { let alreadyTracked = await SessionStore.shared.containsSession(threadID) diff --git a/PingIsland/Services/Usage/AgentUsageAnalytics.swift b/PingIsland/Services/Usage/AgentUsageAnalytics.swift index 606c79ea..78d4f22c 100644 --- a/PingIsland/Services/Usage/AgentUsageAnalytics.swift +++ b/PingIsland/Services/Usage/AgentUsageAnalytics.swift @@ -900,7 +900,9 @@ actor AgentUsageStore { return } - let sourceKey = snapshot.threadID ?? snapshot.sourceFilePath + let sourceKey = snapshot.isRemoteSource + ? snapshot.sourceFilePath + : (snapshot.threadID ?? snapshot.sourceFilePath) await recordTokenUsage( provider: .codex, clientInfo: .codexCLI(), diff --git a/PingIsland/Services/Usage/CodexUsage.swift b/PingIsland/Services/Usage/CodexUsage.swift index 4fc6e6b8..6630c8d0 100644 --- a/PingIsland/Services/Usage/CodexUsage.swift +++ b/PingIsland/Services/Usage/CodexUsage.swift @@ -66,10 +66,25 @@ struct CodexUsageSnapshot: Equatable, Codable, Sendable { return UUID(uuidString: candidate) == nil ? nil : candidate } + nonisolated var isRemoteSource: Bool { + sourceFilePath.hasPrefix("ssh://") + } + nonisolated var isEmpty: Bool { windows.isEmpty } + + nonisolated static func newest(_ snapshots: [CodexUsageSnapshot?]) -> CodexUsageSnapshot? { + snapshots.compactMap { $0 }.max { lhs, rhs in + let lhsDate = lhs.capturedAt ?? .distantPast + let rhsDate = rhs.capturedAt ?? .distantPast + if lhsDate == rhsDate { + return lhs.sourceFilePath.localizedStandardCompare(rhs.sourceFilePath) == .orderedAscending + } + return lhsDate < rhsDate + } + } } enum CodexUsageLoader { diff --git a/PingIslandTests/AgentUsageAnalyticsTests.swift b/PingIslandTests/AgentUsageAnalyticsTests.swift index 6d469b50..81991bb9 100644 --- a/PingIslandTests/AgentUsageAnalyticsTests.swift +++ b/PingIslandTests/AgentUsageAnalyticsTests.swift @@ -442,4 +442,41 @@ final class AgentUsageAnalyticsTests: XCTestCase { XCTAssertEqual(snapshot.topSessionThisWeek?.tokenTotals, AgentUsageTokenTotals(input: 75, output: 30, total: 105)) XCTAssertEqual(snapshot.topSessionThisWeek?.title, "Display token usage by session title") } + + func testRemoteCodexUsageSnapshotsKeepIndependentHostBaselines() async throws { + let directoryURL = FileManager.default.temporaryDirectory + .appendingPathComponent("ping-island-remote-agent-usage-\(UUID().uuidString)", isDirectory: true) + let fileURL = directoryURL.appendingPathComponent("usage.json") + defer { + try? FileManager.default.removeItem(at: directoryURL) + } + + let store = AgentUsageStore(fileURL: fileURL) + let capturedAt = Date() + let rolloutName = "rollout-2026-08-31T00-00-00-019db9a7-336a-7b62-9288-7304c3d2d4b9.jsonl" + let firstHostPath = "ssh://root@first.example:3006/root/.codex/sessions/\(rolloutName)" + let secondHostPath = "ssh://root@second.example:3006/root/.codex/sessions/\(rolloutName)" + + for sourcePath in [firstHostPath, secondHostPath] { + await store.recordCodexUsageSnapshot(CodexUsageSnapshot( + sourceFilePath: sourcePath, + capturedAt: capturedAt, + planType: "pro", + limitID: "codex", + tokenUsage: CodexTokenUsage(inputTokens: 100, outputTokens: 50, totalTokens: 150), + windows: [] + )) + await store.recordCodexUsageSnapshot(CodexUsageSnapshot( + sourceFilePath: sourcePath, + capturedAt: capturedAt, + planType: "pro", + limitID: "codex", + tokenUsage: CodexTokenUsage(inputTokens: 125, outputTokens: 60, totalTokens: 185), + windows: [] + )) + } + + let snapshot = await store.snapshot(range: .today, now: capturedAt) + XCTAssertEqual(snapshot.tokenTotals, AgentUsageTokenTotals(input: 50, output: 20, total: 70)) + } } diff --git a/PingIslandTests/CodexUsageLoaderTests.swift b/PingIslandTests/CodexUsageLoaderTests.swift index 9f296f42..2759a9b1 100644 --- a/PingIslandTests/CodexUsageLoaderTests.swift +++ b/PingIslandTests/CodexUsageLoaderTests.swift @@ -2,6 +2,26 @@ import XCTest @testable import Ping_Island final class CodexUsageLoaderTests: XCTestCase { + func testNewestSnapshotPrefersMostRecentlyCapturedUsage() { + let older = CodexUsageSnapshot( + sourceFilePath: "/local/rollout.jsonl", + capturedAt: Date(timeIntervalSince1970: 100), + planType: "pro", + limitID: "codex", + windows: [] + ) + let newer = CodexUsageSnapshot( + sourceFilePath: "ssh://dev@example.test:2222/root/.codex/sessions/rollout.jsonl", + capturedAt: Date(timeIntervalSince1970: 200), + planType: "pro", + limitID: "codex", + windows: [] + ) + + XCTAssertEqual(CodexUsageSnapshot.newest([older, nil, newer]), newer) + XCTAssertEqual(CodexUsageSnapshot.newest([newer, older]), newer) + } + func testLoadParsesLastTokenCountRateLimits() throws { let rootURL = temporaryRootURL(named: "codex-usage") let rolloutURL = rootURL diff --git a/PingIslandTests/RemoteHookConfigurationTests.swift b/PingIslandTests/RemoteHookConfigurationTests.swift index c7240972..8f197021 100644 --- a/PingIslandTests/RemoteHookConfigurationTests.swift +++ b/PingIslandTests/RemoteHookConfigurationTests.swift @@ -2,6 +2,50 @@ import XCTest @testable import Ping_Island final class RemoteHookConfigurationTests: XCTestCase { + func testRemoteCodexUsageMessageRoundTripsSnapshot() throws { + let snapshot = CodexUsageSnapshot( + sourceFilePath: "/root/.codex/sessions/rollout.jsonl", + capturedAt: Date(timeIntervalSince1970: 1_788_183_296), + planType: "pro", + limitID: "codex", + tokenUsage: CodexTokenUsage(inputTokens: 1_200, outputTokens: 345, totalTokens: 1_545), + windows: [ + CodexUsageWindow( + key: "primary", + label: "5h", + usedPercentage: 17, + leftPercentage: 83, + windowMinutes: 300, + resetsAt: Date(timeIntervalSince1970: 1_788_183_296) + ) + ] + ) + let message = RemoteCodexUsageMessage(type: "codex_usage", payload: snapshot) + + let decoded = try JSONDecoder().decode( + RemoteCodexUsageMessage.self, + from: JSONEncoder().encode(message) + ) + + XCTAssertEqual(decoded, message) + } + + func testRemoteCodexUsageSourcePathIncludesSSHEndpoint() { + let endpoint = RemoteEndpoint( + displayName: "Development host", + sshTarget: "root@example.test", + sshPort: 3006 + ) + + XCTAssertEqual( + RemoteConnectorManager.remoteUsageSourcePath( + "/root/.codex/sessions/rollout.jsonl", + endpoint: endpoint + ), + "ssh://root@example.test:3006/root/.codex/sessions/rollout.jsonl" + ) + } + func testRemoteBootstrapPrepareCommandStopsRunningAgentBeforeReplacingBridge() { let command = RemoteConnectorManager.remoteBootstrapPrepareCommand( installRoot: "/root/.ping-island", diff --git a/Prototype/Sources/IslandBridge/main.swift b/Prototype/Sources/IslandBridge/main.swift index a87980de..f71d215b 100644 --- a/Prototype/Sources/IslandBridge/main.swift +++ b/Prototype/Sources/IslandBridge/main.swift @@ -846,7 +846,9 @@ private final class RemoteAgentService: @unchecked Sendable { private var pendingRequests: [UUID: PendingRemoteBridgeRequest] = [:] private var queuedMessages: [Data] = [] private var codexStateSource: DispatchSourceTimer? + private var codexUsageSource: DispatchSourceTimer? private var deliveredCodexThreadUpdates: [String: Int64] = [:] + private var deliveredCodexUsageSnapshot: RemoteCodexUsageSnapshot? private let encoder = JSONEncoder() private let decoder = JSONDecoder() @@ -882,6 +884,7 @@ private final class RemoteAgentService: @unchecked Sendable { controlAcceptSource?.resume() startCodexStatePolling() + startCodexUsagePolling() } private func makeListeningSocket(path: String) throws -> Int32 { @@ -939,6 +942,27 @@ private final class RemoteAgentService: @unchecked Sendable { } } + private func startCodexUsagePolling() { + let source = DispatchSource.makeTimerSource(queue: queue) + source.schedule(deadline: .now() + 1, repeating: 15) + source.setEventHandler { [weak self] in + self?.pollCodexUsage() + } + codexUsageSource = source + source.resume() + } + + private func pollCodexUsage(force: Bool = false) { + let codexHome = Self.homeDirectory.appendingPathComponent(".codex", isDirectory: true) + guard let snapshot = RemoteCodexUsagePoller.loadLatestSnapshot(codexHome: codexHome), + force || snapshot != deliveredCodexUsageSnapshot else { + return + } + + deliveredCodexUsageSnapshot = snapshot + enqueue(RemoteCodexUsageMessage(type: "codex_usage", payload: snapshot)) + } + private nonisolated static var homeDirectory: URL { if let home = ProcessInfo.processInfo.environment["HOME"]?.trimmingCharacters(in: .whitespacesAndNewlines), !home.isEmpty { @@ -1012,6 +1036,7 @@ private final class RemoteAgentService: @unchecked Sendable { controlClientSocket = clientSocket sendHello() + pollCodexUsage(force: true) flushQueuedMessages() controlClientReadSource = DispatchSource.makeReadSource(fileDescriptor: clientSocket, queue: queue) @@ -1188,6 +1213,35 @@ private struct RemoteCodexThread: Equatable { let updatedAtMs: Int64 } +private struct RemoteCodexUsageWindow: Codable, Equatable { + let key: String + let label: String + let usedPercentage: Double + let leftPercentage: Double + let windowMinutes: Int + let resetsAt: Date? +} + +private struct RemoteCodexTokenUsage: Codable, Equatable { + let inputTokens: Int + let outputTokens: Int + let totalTokens: Int +} + +private struct RemoteCodexUsageSnapshot: Codable, Equatable { + let sourceFilePath: String + let capturedAt: Date? + let planType: String? + let limitID: String? + let tokenUsage: RemoteCodexTokenUsage? + let windows: [RemoteCodexUsageWindow] +} + +private struct RemoteCodexUsageMessage: Codable { + let type: String + let payload: RemoteCodexUsageSnapshot +} + private enum RemoteCodexStatePoller { static func readRecentThreads(codexHome: URL, updatedSinceMs: Int64) -> [RemoteCodexThread] { guard let stateURL = newestStateDatabase(in: codexHome), @@ -1265,6 +1319,263 @@ private enum RemoteCodexStatePoller { } } +private enum RemoteCodexUsagePoller { + private static let candidateScanLimit = 24 + private static let maxBytesPerFile = 4 * 1024 * 1024 + private static let cacheLock = NSLock() + nonisolated(unsafe) private static var cachedResult: CachedResult? + + private struct Candidate { + let fileURL: URL + let modifiedAt: Date + let fileSize: UInt64 + } + + private struct CachedResult { + let fingerprint: String + let snapshot: RemoteCodexUsageSnapshot? + } + + static func loadLatestSnapshot(codexHome: URL) -> RemoteCodexUsageSnapshot? { + let sessionsURL = codexHome.appendingPathComponent("sessions", isDirectory: true) + guard let enumerator = FileManager.default.enumerator( + at: sessionsURL, + includingPropertiesForKeys: [.contentModificationDateKey, .fileSizeKey, .isRegularFileKey], + options: [.skipsHiddenFiles] + ) else { + return nil + } + + var candidates: [Candidate] = [] + for case let fileURL as URL in enumerator { + guard fileURL.lastPathComponent.hasPrefix("rollout-"), + fileURL.pathExtension == "jsonl", + let values = try? fileURL.resourceValues( + forKeys: [.contentModificationDateKey, .fileSizeKey, .isRegularFileKey] + ), + values.isRegularFile == true else { + continue + } + + candidates.append(Candidate( + fileURL: fileURL, + modifiedAt: values.contentModificationDate ?? .distantPast, + fileSize: UInt64(max(0, values.fileSize ?? 0)) + )) + } + + let recentCandidates = Array(candidates.sorted { lhs, rhs in + if lhs.modifiedAt == rhs.modifiedAt { + return lhs.fileURL.path.localizedStandardCompare(rhs.fileURL.path) == .orderedDescending + } + return lhs.modifiedAt > rhs.modifiedAt + }.prefix(candidateScanLimit)) + + let fingerprint = cacheFingerprint(codexHome: codexHome, candidates: recentCandidates) + if let cached = cachedSnapshot(for: fingerprint) { + return cached + } + + var bestSnapshot: RemoteCodexUsageSnapshot? + var bestCapturedAt = Date.distantPast + for candidate in recentCandidates { + guard let snapshot = loadLatestSnapshot(from: candidate) else { continue } + let capturedAt = snapshot.capturedAt ?? candidate.modifiedAt + if capturedAt > bestCapturedAt { + bestSnapshot = snapshot + bestCapturedAt = capturedAt + } + } + cache(snapshot: bestSnapshot, for: fingerprint) + return bestSnapshot + } + + private static func cachedSnapshot(for fingerprint: String) -> RemoteCodexUsageSnapshot?? { + cacheLock.lock() + defer { cacheLock.unlock() } + guard let cachedResult, cachedResult.fingerprint == fingerprint else { return nil } + return cachedResult.snapshot + } + + private static func cache(snapshot: RemoteCodexUsageSnapshot?, for fingerprint: String) { + cacheLock.lock() + cachedResult = CachedResult(fingerprint: fingerprint, snapshot: snapshot) + cacheLock.unlock() + } + + private static func cacheFingerprint(codexHome: URL, candidates: [Candidate]) -> String { + var parts = [codexHome.resolvingSymlinksInPath().path] + parts.reserveCapacity(candidates.count + 1) + for candidate in candidates { + parts.append([ + candidate.fileURL.resolvingSymlinksInPath().path, + String(candidate.modifiedAt.timeIntervalSinceReferenceDate), + String(candidate.fileSize), + ].joined(separator: "|")) + } + return parts.joined(separator: "\n") + } + + private static func loadLatestSnapshot(from candidate: Candidate) -> RemoteCodexUsageSnapshot? { + guard candidate.fileSize > 0, + let contents = readSuffixText( + from: candidate.fileURL, + fileSize: candidate.fileSize, + maxBytes: maxBytesPerFile + ) else { + return nil + } + + var legacySnapshot: RemoteCodexUsageSnapshot? + for line in contents.split(separator: "\n", omittingEmptySubsequences: false).reversed() { + guard line.contains("\"token_count\""), + line.contains("\"rate_limits\""), + let snapshot = snapshot( + from: String(line), + filePath: candidate.fileURL.path, + fallbackTimestamp: candidate.modifiedAt + ) else { + continue + } + if snapshot.limitID == "codex" { + return snapshot + } + if snapshot.limitID == nil, legacySnapshot == nil { + legacySnapshot = snapshot + } + } + return legacySnapshot + } + + private static func readSuffixText(from fileURL: URL, fileSize: UInt64, maxBytes: Int) -> String? { + guard let handle = try? FileHandle(forReadingFrom: fileURL) else { return nil } + defer { try? handle.close() } + + let readSize = min(fileSize, UInt64(maxBytes)) + let offset = fileSize - readSize + do { + try handle.seek(toOffset: offset) + guard let data = try handle.read(upToCount: Int(readSize)) else { return nil } + var text = String(decoding: data, as: UTF8.self) + if offset > 0, let newline = text.firstIndex(of: "\n") { + text = String(text[text.index(after: newline)...]) + } + return text + } catch { + return nil + } + } + + private static func snapshot( + from line: String, + filePath: String, + fallbackTimestamp: Date + ) -> RemoteCodexUsageSnapshot? { + guard let data = line.data(using: .utf8), + let object = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + object["type"] as? String == "event_msg" else { + return nil + } + + let payload = object["payload"] as? [String: Any] ?? [:] + guard payload["type"] as? String == "token_count", + let rateLimits = payload["rate_limits"] as? [String: Any] else { + return nil + } + + let windows = ["primary", "secondary"].compactMap { key in + usageWindow(for: key, in: rateLimits) + } + guard !windows.isEmpty else { return nil } + + return RemoteCodexUsageSnapshot( + sourceFilePath: filePath, + capturedAt: timestamp(from: object["timestamp"]) ?? fallbackTimestamp, + planType: string(from: rateLimits["plan_type"]), + limitID: string(from: rateLimits["limit_id"]), + tokenUsage: tokenUsage(from: payload["info"]), + windows: windows + ) + } + + private static func usageWindow( + for key: String, + in rateLimits: [String: Any] + ) -> RemoteCodexUsageWindow? { + guard let payload = rateLimits[key] as? [String: Any], + let usedPercentage = number(from: payload["used_percent"]), + let windowMinutes = integer(from: payload["window_minutes"]) else { + return nil + } + + return RemoteCodexUsageWindow( + key: key, + label: windowLabel(forMinutes: windowMinutes), + usedPercentage: usedPercentage, + leftPercentage: max(0, 100 - usedPercentage), + windowMinutes: windowMinutes, + resetsAt: date(from: payload["resets_at"]) + ) + } + + private static func windowLabel(forMinutes minutes: Int) -> String { + let days = minutes / 1_440 + let hours = (minutes % 1_440) / 60 + let remainingMinutes = minutes % 60 + if days > 0, hours == 0, remainingMinutes == 0 { return "\(days)d" } + if days > 0, hours > 0 { return "\(days)d \(hours)h" } + if hours > 0, remainingMinutes == 0 { return "\(hours)h" } + if hours > 0 { return "\(hours)h \(remainingMinutes)m" } + return "\(minutes)m" + } + + private static func tokenUsage(from value: Any?) -> RemoteCodexTokenUsage? { + guard let info = value as? [String: Any], + let usage = info["total_token_usage"] as? [String: Any] else { + return nil + } + let input = integer(from: usage["input_tokens"]) + ?? integer(from: usage["prompt_tokens"]) + ?? 0 + let output = integer(from: usage["output_tokens"]) + ?? integer(from: usage["completion_tokens"]) + ?? 0 + let total = integer(from: usage["total_tokens"]) ?? max(0, input + output) + guard input > 0 || output > 0 || total > 0 else { return nil } + return RemoteCodexTokenUsage(inputTokens: input, outputTokens: output, totalTokens: total) + } + + private static func timestamp(from value: Any?) -> Date? { + guard let value = value as? String else { return nil } + let formatter = ISO8601DateFormatter() + formatter.formatOptions = [.withInternetDateTime, .withFractionalSeconds] + return formatter.date(from: value) + } + + private static func number(from value: Any?) -> Double? { + if let number = value as? NSNumber { return number.doubleValue } + if let string = value as? String { return Double(string) } + return nil + } + + private static func integer(from value: Any?) -> Int? { + if let number = value as? NSNumber { return number.intValue } + if let string = value as? String { return Int(string) } + return nil + } + + private static func date(from value: Any?) -> Date? { + guard let seconds = number(from: value) else { return nil } + return Date(timeIntervalSince1970: seconds) + } + + private static func string(from value: Any?) -> String? { + if let string = value as? String { return string.isEmpty ? nil : string } + if let number = value as? NSNumber { return number.stringValue } + return nil + } +} + private final class RemoteSQLiteDatabase { fileprivate typealias SQLiteOpenV2 = @convention(c) (UnsafePointer?, UnsafeMutablePointer?, Int32, UnsafePointer?) -> Int32 fileprivate typealias SQLiteClose = @convention(c) (OpaquePointer?) -> Int32 diff --git a/Prototype/Tests/IslandTests/IslandBridgeE2ETests.swift b/Prototype/Tests/IslandTests/IslandBridgeE2ETests.swift index 11d09d89..da2e1bdb 100644 --- a/Prototype/Tests/IslandTests/IslandBridgeE2ETests.swift +++ b/Prototype/Tests/IslandTests/IslandBridgeE2ETests.swift @@ -427,6 +427,60 @@ func remoteAgentForwardsCodexAppServerStateUpdates() async throws { } } +@Test +func remoteAgentForwardsCodexUsageSnapshots() async throws { + try await withTemporaryDirectory { directory in + let executable = try TestRuntime.executableURL(named: "PingIslandBridge") + let rolloutURL = directory + .appending(path: ".codex/sessions/2026/08/31", directoryHint: .isDirectory) + .appending(path: "rollout-remote-codex-thread.jsonl") + try FileManager.default.createDirectory( + at: rolloutURL.deletingLastPathComponent(), + withIntermediateDirectories: true + ) + try Data(""" + {"timestamp":"2026-08-31T12:34:56.000Z","type":"event_msg","payload":{"type":"token_count","info":{"total_token_usage":{"input_tokens":1200,"output_tokens":345,"total_tokens":1545}},"rate_limits":{"limit_id":"codex","plan_type":"pro","primary":{"used_percent":17,"window_minutes":300,"resets_at":1788183296},"secondary":{"used_percent":29,"window_minutes":10080,"resets_at":1788788096}}}} + """.utf8).write(to: rolloutURL) + + let socketID = UUID().uuidString.prefix(8) + let hookSocketPath = "/tmp/pi-\(socketID)-h.sock" + let controlSocketPath = "/tmp/pi-\(socketID)-c.sock" + let service = try RunningProcess( + executableURL: executable, + arguments: [ + "--mode", "remote-agent-service", + "--hook-socket", hookSocketPath, + "--control-socket", controlSocketPath + ], + environment: ["HOME": directory.path()] + ) + defer { + service.terminate() + _ = service.waitForExit() + try? FileManager.default.removeItem(atPath: hookSocketPath) + try? FileManager.default.removeItem(atPath: controlSocketPath) + } + + try await waitUntil(description: "remote agent service should create control socket") { + FileManager.default.fileExists(atPath: controlSocketPath) + } + + let message = try await readRemoteMessage( + controlSocketPath: controlSocketPath, + as: TestRemoteCodexUsageMessage.self, + description: "remote Codex usage message", + matching: { $0.type == "codex_usage" } + ) + + #expect(message.payload.sourceFilePath == rolloutURL.path()) + #expect(message.payload.planType == "pro") + #expect(message.payload.limitID == "codex") + #expect(message.payload.tokenUsage?.totalTokens == 1_545) + #expect(message.payload.windows.map(\.label) == ["5h", "7d"]) + #expect(message.payload.windows.map(\.usedPercentage) == [17, 29]) + } +} + private func bridgeTestEnvironment(_ values: [String: String] = [:]) -> [String: String] { var environment = values environment[BridgeRuntimeConfig.configPathEnvironmentKey] = @@ -476,6 +530,20 @@ private func readRemoteHookEvent( controlSocketPath: String, matching predicate: @escaping (TestRemoteHookEventMessage) -> Bool ) async throws -> TestRemoteHookEventMessage { + try await readRemoteMessage( + controlSocketPath: controlSocketPath, + as: TestRemoteHookEventMessage.self, + description: "remote Codex hook event", + matching: predicate + ) +} + +private func readRemoteMessage( + controlSocketPath: String, + as type: Message.Type, + description: String, + matching predicate: @escaping (Message) -> Bool +) async throws -> Message { let fd = socket(AF_UNIX, SOCK_STREAM, 0) guard fd >= 0 else { throw POSIXError(.EIO) } defer { close(fd) } @@ -508,9 +576,9 @@ private func readRemoteHookEvent( while let newline = buffer.firstRange(of: Data([0x0A])) { let line = buffer.subdata(in: 0.. Date: Tue, 1 Sep 2026 10:17:35 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=90=9B=20fix(remote):=20keep=20usage?= =?UTF-8?q?=20scans=20off=20attach=20queue?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bound the initial rollout scan and coalesce overlapping polls so large Codex histories cannot block control connections. --- AGENTS.md | 2 +- Prototype/Sources/IslandBridge/main.swift | 36 ++++++++++++++++++----- 2 files changed, 29 insertions(+), 9 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 98411539..f90b680c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -51,7 +51,7 @@ This file is a routing layer for coding agents working in this repo. Keep it sho - Remote SSH forwarding and remote-host management: `PingIsland/Services/Remote/` - Remote hosts can bootstrap a bridge on the SSH target, rewrite remote hooks, install managed plugin-directory integrations such as Hermes under the remote home directory, and attach a bidirectional forwarding channel back into PingIsland - The remote bridge forwards recent Codex app-server thread activity from the SSH target's `~/.codex/state_*.sqlite` through the existing remote hook-event channel - - The remote bridge also tails recent Codex rollout quota snapshots and emits `codex_usage` messages. Keep local and remote snapshots source-qualified, choose the newest `capturedAt` for the account-wide usage UI, and resend the current remote snapshot whenever a control client attaches. + - The remote bridge also tails recent Codex rollout quota snapshots and emits `codex_usage` messages. Keep local and remote snapshots source-qualified, choose the newest `capturedAt` for the account-wide usage UI, and resend the current remote snapshot whenever a control client attaches. Run rollout parsing off the control queue, bound the candidate/tail scan, and prevent overlapping polls so large histories cannot stall SSH attach or hook delivery. - Provider/client routing: bridge envelopes are normalized in `PingIsland/Services/Hooks/HookSocketServer.swift`, stored on `SessionState`, and launched via `PingIsland/Services/Window/SessionLauncher.swift` - Client profile registry: installable hook clients and runtime client branding / recognition are centralized in `PingIsland/Models/ClientProfile.swift` - VS Code-compatible IDE focus extension install / URI launch: `PingIsland/Services/Window/IDEExtensionInstaller.swift`, `PingIsland/Services/Window/TerminalSessionFocuser.swift` diff --git a/Prototype/Sources/IslandBridge/main.swift b/Prototype/Sources/IslandBridge/main.swift index f71d215b..7cf3c04d 100644 --- a/Prototype/Sources/IslandBridge/main.swift +++ b/Prototype/Sources/IslandBridge/main.swift @@ -835,6 +835,10 @@ private final class RemoteAgentService: @unchecked Sendable { private let hookSocketPath: String private let controlSocketPath: String private let queue = DispatchQueue(label: "com.wudanwu.pingisland.remote-agent", qos: .userInitiated) + private let codexUsageQueue = DispatchQueue( + label: "com.wudanwu.pingisland.remote-agent.codex-usage", + qos: .utility + ) private var hookServerSocket: Int32 = -1 private var controlServerSocket: Int32 = -1 @@ -849,6 +853,8 @@ private final class RemoteAgentService: @unchecked Sendable { private var codexUsageSource: DispatchSourceTimer? private var deliveredCodexThreadUpdates: [String: Int64] = [:] private var deliveredCodexUsageSnapshot: RemoteCodexUsageSnapshot? + private var codexUsagePollInFlight = false + private var codexUsageForceDeliveryPending = false private let encoder = JSONEncoder() private let decoder = JSONDecoder() @@ -953,14 +959,28 @@ private final class RemoteAgentService: @unchecked Sendable { } private func pollCodexUsage(force: Bool = false) { + codexUsageForceDeliveryPending = codexUsageForceDeliveryPending || force + guard !codexUsagePollInFlight else { return } + + codexUsagePollInFlight = true let codexHome = Self.homeDirectory.appendingPathComponent(".codex", isDirectory: true) - guard let snapshot = RemoteCodexUsagePoller.loadLatestSnapshot(codexHome: codexHome), - force || snapshot != deliveredCodexUsageSnapshot else { - return - } + codexUsageQueue.async { [weak self] in + let snapshot = RemoteCodexUsagePoller.loadLatestSnapshot(codexHome: codexHome) + self?.queue.async { [weak self] in + guard let self else { return } + self.codexUsagePollInFlight = false + + let shouldForceDelivery = self.codexUsageForceDeliveryPending + self.codexUsageForceDeliveryPending = false + guard let snapshot, + shouldForceDelivery || snapshot != self.deliveredCodexUsageSnapshot else { + return + } - deliveredCodexUsageSnapshot = snapshot - enqueue(RemoteCodexUsageMessage(type: "codex_usage", payload: snapshot)) + self.deliveredCodexUsageSnapshot = snapshot + self.enqueue(RemoteCodexUsageMessage(type: "codex_usage", payload: snapshot)) + } + } } private nonisolated static var homeDirectory: URL { @@ -1320,8 +1340,8 @@ private enum RemoteCodexStatePoller { } private enum RemoteCodexUsagePoller { - private static let candidateScanLimit = 24 - private static let maxBytesPerFile = 4 * 1024 * 1024 + private static let candidateScanLimit = 8 + private static let maxBytesPerFile = 1 * 1024 * 1024 private static let cacheLock = NSLock() nonisolated(unsafe) private static var cachedResult: CachedResult?