Resolve a Codex loop's real thread before reading its goal or resuming it (#346)

GraphcodeKit/Sources/Sessions/CodexThreadResolver.swift

@@ -0,0 +1,80 @@
+import Foundation
+
+#if canImport(SQLite3)
+  import SQLite3
+#endif
+
+/// The Codex thread a loop's session really runs on.
+///
+/// The `notify` hook banks the `thread-id` of the event it is handed, and once a session
+/// opens on `/goal` that id names a thread Codex never persists: no rollout, no row in its
+/// own `threads` table, no goal. Resuming it fails, and its goal verdict cannot be read
+/// (#346). Codex does persist the real thread, and its first message is the launch line,
+/// which carries the node's own memory path — so the node id finds it.
+public enum CodexThreadResolver {
+  public static var codexDirectory: URL = FileManager.default.homeDirectoryForCurrentUser
+    .appendingPathComponent(".codex", isDirectory: true)
+
+  /// The banked id when Codex knows it as a thread; otherwise the newest thread whose first
+  /// message names the node; otherwise the banked id as it was.
+  public static func threadID(forNodeID nodeID: UUID, banked: String?) -> String? {
+    guard let database = stateDatabase() else { return banked }
+    return threadID(forNodeID: nodeID, banked: banked, database: database)
+  }
+
+  static func threadID(forNodeID nodeID: UUID, banked: String?, database: URL) -> String? {
+    #if canImport(SQLite3)
+      var handle: OpaquePointer?
+      guard sqlite3_open_v2(database.path, &handle, SQLITE_OPEN_READONLY, nil) == SQLITE_OK
+      else {
+        sqlite3_close(handle)
+        return banked
+      }
+      defer { sqlite3_close(handle) }
+      sqlite3_busy_timeout(handle, 500)
+      if let banked, firstText(handle, "SELECT id FROM threads WHERE id = ?", banked) != nil {
+        return banked
+      }
+      let named = firstText(
+        handle,
+        "SELECT id FROM threads WHERE first_user_message LIKE ? "
+          + "ORDER BY created_at_ms DESC LIMIT 1",
+        "%\(nodeID.uuidString)%")
+      return named ?? banked
+    #else
+      return banked
+    #endif
+  }
+
+  /// Codex versions its state database in the file name (`state_5.sqlite`), so the newest
+  /// version present is the one in use.
+  static func stateDatabase() -> URL? {
+    guard
+      let names = try? FileManager.default.contentsOfDirectory(atPath: codexDirectory.path)
+    else { return nil }
+    let versions = names.compactMap { name -> (Int, String)? in
+      guard name.hasPrefix("state_"), name.hasSuffix(".sqlite"),
+        let version = Int(name.dropFirst("state_".count).dropLast(".sqlite".count))
+      else { return nil }
+      return (version, name)
+    }
+    return versions.max { $0.0 < $1.0 }.map { codexDirectory.appendingPathComponent($0.1) }
+  }
+
+  #if canImport(SQLite3)
+    private static func firstText(_ handle: OpaquePointer?, _ sql: String, _ value: String)
+      -> String?
+    {
+      var statement: OpaquePointer?
+      guard sqlite3_prepare_v2(handle, sql, -1, &statement, nil) == SQLITE_OK else {
+        return nil
+      }
+      defer { sqlite3_finalize(statement) }
+      let transient = unsafeBitCast(-1, to: sqlite3_destructor_type.self)
+      sqlite3_bind_text(statement, 1, value, -1, transient)
+      guard sqlite3_step(statement) == SQLITE_ROW, let text = sqlite3_column_text(statement, 0)
+      else { return nil }
+      return String(cString: text)
+    }
+  #endif
+}

GraphcodeKit/Sources/Sessions/GoalVerdictReader.swift

@@ -31,7 +31,10 @@ public enum GoalVerdictReader {
       return claudeVerdict(
         lines: CopilotSessionLog.tailLines(ofLogAt: transcript), goalSummary: goal.summary)
     case .codex:
-      guard let threadID = SessionIDStore.load(forNodeID: node.id) else { return nil }
+      guard
+        let threadID = CodexThreadResolver.threadID(
+          forNodeID: node.id, banked: SessionIDStore.load(forNodeID: node.id))
+      else { return nil }
       return codexVerdict(threadID: threadID, database: codexGoalsDatabase)
     case .copilotCLI:
       let name = SurfaceRef(id: node.id, launchesClaudeCode: true).zmxSessionName

GraphcodeKit/Sources/Sessions/ZmxSessionLauncher.swift

@@ -1842,7 +1842,7 @@ public enum ZmxSessionLauncher {
     // no session a keystroke can reach, so the run branch relaunches it — this is
     // what lets an ensure, a send, or the sweep wake a loop that died unattended
     // (issue #215), which a `zmx get` check could never do.
-    let sessionID: String? =
+    let bankedID: String? =
       SessionIDStore.load(forNodeID: node.id)
       ?? {
         switch node.backend {
@@ -1853,6 +1853,11 @@ public enum ZmxSessionLauncher {
           return nil
         }
       }()
+    // Codex only: its notify hook can bank a thread Codex never persisted, and resuming
+    // that id fails over to a fresh launch (#346).
+    let sessionID =
+      node.backend == .codex
+      ? CodexThreadResolver.threadID(forNodeID: node.id, banked: bankedID) : bankedID
     guard let runArgs = arguments(forNode: node, projectPath: projectPath) else { return }
     let name = SurfaceRef(id: node.id, launchesClaudeCode: true).zmxSessionName
     if let sessionID,

graphcode/Tests/CodexThreadResolverTests.swift

@@ -0,0 +1,82 @@
+import Foundation
+import SQLite3
+import Testing
+
+@testable import GraphcodeKit
+
+/// Codex's `notify` can bank a thread id Codex never persists; the node's real thread is
+/// found from Codex's own `threads` table instead (#346).
+@Suite
+struct CodexThreadResolverTests {
+  private func database(_ rows: [(id: String, message: String, createdAt: Int)]) -> URL {
+    let url = FileManager.default.temporaryDirectory
+      .appendingPathComponent("state-\(UUID().uuidString).sqlite")
+    var handle: OpaquePointer?
+    sqlite3_open(url.path, &handle)
+    sqlite3_exec(
+      handle,
+      "CREATE TABLE threads (id TEXT PRIMARY KEY, first_user_message TEXT, created_at_ms INTEGER)",
+      nil, nil, nil)
+    for row in rows {
+      let message = row.message.replacingOccurrences(of: "'", with: "''")
+      sqlite3_exec(
+        handle, "INSERT INTO threads VALUES ('\(row.id)', '\(message)', \(row.createdAt))", nil,
+        nil, nil)
+    }
+    sqlite3_close(handle)
+    return url
+  }
+
+  @Test
+  func aBankedIdCodexKnowsIsKept() {
+    let node = UUID()
+    let url = database([("real", "/goal read /x/\(node.uuidString)/PROMPT.md", 1)])
+    defer { try? FileManager.default.removeItem(at: url) }
+
+    #expect(
+      CodexThreadResolver.threadID(forNodeID: node, banked: "real", database: url) == "real")
+  }
+
+  @Test
+  func aBankedIdCodexNeverPersistedResolvesToTheNodesNewestThread() {
+    let node = UUID()
+    let url = database([
+      ("older", "/goal read /x/\(node.uuidString)/PROMPT.md", 1),
+      ("newer", "/goal read /x/\(node.uuidString)/PROMPT.md", 2),
+      ("other", "/goal read /x/\(UUID().uuidString)/PROMPT.md", 3),
+    ])
+    defer { try? FileManager.default.removeItem(at: url) }
+
+    #expect(
+      CodexThreadResolver.threadID(forNodeID: node, banked: "ephemeral", database: url)
+        == "newer")
+    #expect(CodexThreadResolver.threadID(forNodeID: node, banked: nil, database: url) == "newer")
+  }
+
+  @Test
+  func withNothingToGoOnTheBankedIdStands() {
+    let url = database([("other", "no node here", 1)])
+    defer { try? FileManager.default.removeItem(at: url) }
+
+    #expect(
+      CodexThreadResolver.threadID(forNodeID: UUID(), banked: "ephemeral", database: url)
+        == "ephemeral")
+    #expect(CodexThreadResolver.threadID(forNodeID: UUID(), banked: nil, database: url) == nil)
+  }
+
+  @Test
+  func theNewestStateDatabaseVersionIsTheOneRead() throws {
+    let directory = FileManager.default.temporaryDirectory
+      .appendingPathComponent("codex-\(UUID().uuidString)", isDirectory: true)
+    try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true)
+    defer { try? FileManager.default.removeItem(at: directory) }
+    for name in ["state_4.sqlite", "state_12.sqlite", "goals_1.sqlite", "state_5.sqlite-wal"] {
+      FileManager.default.createFile(atPath: directory.appendingPathComponent(name).path, contents: nil)
+    }
+    let original = CodexThreadResolver.codexDirectory
+    CodexThreadResolver.codexDirectory = directory
+    defer { CodexThreadResolver.codexDirectory = original }
+
+    #expect(CodexThreadResolver.stateDatabase()?.lastPathComponent == "state_12.sqlite")
+  }
+}