fix(remote): forward Codex usage snapshots over SSH

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. 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`

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

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)

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(),

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 {

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))
+    }
 }

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

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",