ScuttleBot
fix: codex-relay deterministic session discovery via file snapshot Replace the timestamp-closest heuristic in findLatestSessionPath with a pre-existing file snapshot approach. Before starting the Codex subprocess, the relay snapshots all .jsonl files under the sessions directory. During discovery, files in the snapshot are skipped so the relay only considers sessions created after its subprocess started. Among new files matching the target CWD, the oldest is selected. This fixes the race condition where multiple relays starting near-simultaneously in the same CWD would both latch onto the same session file because the "closest timestamp" heuristic couldn't distinguish them. The reconnect path passes nil for preExisting, preserving the existing behavior of finding any matching session regardless of age. Closes #66
Commit
3fc248398e62b410165317ebf869b7057a30122d85bbe188bfe0f916f6f15cbc
Parent
6091f6da05a68fd…
2 files changed
+45
-29
+117
+45
-29
| --- cmd/codex-relay/main.go | ||
| +++ cmd/codex-relay/main.go | ||
| @@ -207,10 +207,17 @@ | ||
| 207 | 207 | _ = relay.Close(closeCtx) |
| 208 | 208 | }() |
| 209 | 209 | } |
| 210 | 210 | |
| 211 | 211 | cmd := exec.Command(cfg.CodexBin, cfg.Args...) |
| 212 | + // Snapshot existing session files before starting the subprocess so | |
| 213 | + // discovery can distinguish our session from pre-existing ones. | |
| 214 | + var preExisting map[string]struct{} | |
| 215 | + if sessRoot, err := codexSessionsRoot(); err == nil { | |
| 216 | + preExisting = snapshotSessionFiles(sessRoot) | |
| 217 | + } | |
| 218 | + | |
| 212 | 219 | startedAt := time.Now() |
| 213 | 220 | cmd.Env = append(os.Environ(), |
| 214 | 221 | "SCUTTLEBOT_CONFIG_FILE="+cfg.ConfigFile, |
| 215 | 222 | "SCUTTLEBOT_URL="+cfg.URL, |
| 216 | 223 | "SCUTTLEBOT_TOKEN="+cfg.Token, |
| @@ -221,11 +228,11 @@ | ||
| 221 | 228 | "SCUTTLEBOT_SESSION_ID="+cfg.SessionID, |
| 222 | 229 | "SCUTTLEBOT_NICK="+cfg.Nick, |
| 223 | 230 | "SCUTTLEBOT_ACTIVITY_VIA_BROKER="+boolString(relayActive), |
| 224 | 231 | ) |
| 225 | 232 | if relayActive { |
| 226 | - go mirrorSessionLoop(ctx, relay, cfg, startedAt) | |
| 233 | + go mirrorSessionLoop(ctx, relay, cfg, startedAt, preExisting) | |
| 227 | 234 | go presenceLoopPtr(ctx, &relay, cfg.HeartbeatInterval) |
| 228 | 235 | } |
| 229 | 236 | |
| 230 | 237 | if !isInteractiveTTY() { |
| 231 | 238 | cmd.Stdin = os.Stdin |
| @@ -401,11 +408,11 @@ | ||
| 401 | 408 | fmt.Fprintf(os.Stderr, "codex-relay: reconnected, restarting mirror and input loops\n") |
| 402 | 409 | |
| 403 | 410 | // Restart mirror and input loops with the new connector. |
| 404 | 411 | // Use epoch time for mirror so it finds the existing session file |
| 405 | 412 | // regardless of when it was last modified. |
| 406 | - go mirrorSessionLoop(ctx, conn, cfg, time.Time{}) | |
| 413 | + go mirrorSessionLoop(ctx, conn, cfg, time.Time{}, nil) | |
| 407 | 414 | go relayInputLoop(ctx, conn, cfg, state, ptmx, now) |
| 408 | 415 | break |
| 409 | 416 | } |
| 410 | 417 | } |
| 411 | 418 | } |
| @@ -802,16 +809,16 @@ | ||
| 802 | 809 | func defaultSessionID(target string) string { |
| 803 | 810 | sum := crc32.ChecksumIEEE([]byte(fmt.Sprintf("%s|%d|%d|%d", target, os.Getpid(), os.Getppid(), time.Now().UnixNano()))) |
| 804 | 811 | return fmt.Sprintf("%08x", sum) |
| 805 | 812 | } |
| 806 | 813 | |
| 807 | -func mirrorSessionLoop(ctx context.Context, relay sessionrelay.Connector, cfg config, startedAt time.Time) { | |
| 814 | +func mirrorSessionLoop(ctx context.Context, relay sessionrelay.Connector, cfg config, startedAt time.Time, preExisting map[string]struct{}) { | |
| 808 | 815 | for { |
| 809 | 816 | if ctx.Err() != nil { |
| 810 | 817 | return |
| 811 | 818 | } |
| 812 | - sessionPath, err := discoverSessionPath(ctx, cfg, startedAt) | |
| 819 | + sessionPath, err := discoverSessionPath(ctx, cfg, startedAt, preExisting) | |
| 813 | 820 | if err != nil { |
| 814 | 821 | if ctx.Err() != nil { |
| 815 | 822 | return |
| 816 | 823 | } |
| 817 | 824 | time.Sleep(10 * time.Second) |
| @@ -834,11 +841,11 @@ | ||
| 834 | 841 | } |
| 835 | 842 | return |
| 836 | 843 | } |
| 837 | 844 | } |
| 838 | 845 | |
| 839 | -func discoverSessionPath(ctx context.Context, cfg config, startedAt time.Time) (string, error) { | |
| 846 | +func discoverSessionPath(ctx context.Context, cfg config, startedAt time.Time, preExisting map[string]struct{}) (string, error) { | |
| 840 | 847 | root, err := codexSessionsRoot() |
| 841 | 848 | if err != nil { |
| 842 | 849 | return "", err |
| 843 | 850 | } |
| 844 | 851 | |
| @@ -848,11 +855,11 @@ | ||
| 848 | 855 | }) |
| 849 | 856 | } |
| 850 | 857 | |
| 851 | 858 | target := filepath.Clean(cfg.TargetCWD) |
| 852 | 859 | return waitForSessionPath(ctx, func() (string, error) { |
| 853 | - return findLatestSessionPath(root, target, startedAt.Add(-2*time.Second)) | |
| 860 | + return findLatestSessionPath(root, target, startedAt.Add(-2*time.Second), preExisting) | |
| 854 | 861 | }) |
| 855 | 862 | } |
| 856 | 863 | |
| 857 | 864 | func waitForSessionPath(ctx context.Context, find func() (string, error)) (string, error) { |
| 858 | 865 | ctx, cancel := context.WithTimeout(ctx, defaultDiscoverWait) |
| @@ -1102,10 +1109,25 @@ | ||
| 1102 | 1109 | return strings.TrimSpace(args[i+1]) |
| 1103 | 1110 | } |
| 1104 | 1111 | } |
| 1105 | 1112 | return "" |
| 1106 | 1113 | } |
| 1114 | + | |
| 1115 | +// snapshotSessionFiles returns the set of .jsonl file paths currently under root. | |
| 1116 | +// Called before starting the Codex subprocess so discovery can skip pre-existing | |
| 1117 | +// sessions and deterministically find the one our subprocess creates. | |
| 1118 | +func snapshotSessionFiles(root string) map[string]struct{} { | |
| 1119 | + existing := make(map[string]struct{}) | |
| 1120 | + _ = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { | |
| 1121 | + if err != nil || d.IsDir() || !strings.HasSuffix(path, ".jsonl") { | |
| 1122 | + return nil | |
| 1123 | + } | |
| 1124 | + existing[path] = struct{}{} | |
| 1125 | + return nil | |
| 1126 | + }) | |
| 1127 | + return existing | |
| 1128 | +} | |
| 1107 | 1129 | |
| 1108 | 1130 | func codexSessionsRoot() (string, error) { |
| 1109 | 1131 | if value := os.Getenv("CODEX_HOME"); value != "" { |
| 1110 | 1132 | return filepath.Join(value, "sessions"), nil |
| 1111 | 1133 | } |
| @@ -1135,23 +1157,30 @@ | ||
| 1135 | 1157 | return "", os.ErrNotExist |
| 1136 | 1158 | } |
| 1137 | 1159 | return match, nil |
| 1138 | 1160 | } |
| 1139 | 1161 | |
| 1140 | -// findLatestSessionPath finds the .jsonl file in root whose first entry | |
| 1141 | -// timestamp is closest to notBefore — this ensures each relay latches onto | |
| 1142 | -// its own subprocess's session when multiple sessions share a CWD. | |
| 1143 | -func findLatestSessionPath(root, target string, notBefore time.Time) (string, error) { | |
| 1162 | +// findLatestSessionPath finds the .jsonl file in root that was created by our | |
| 1163 | +// subprocess. It uses a pre-existing file snapshot to skip sessions that | |
| 1164 | +// existed before the subprocess started, then filters by CWD and picks the | |
| 1165 | +// oldest new match. When preExisting is nil (reconnect), it falls back to | |
| 1166 | +// accepting any file whose timestamp is >= notBefore. | |
| 1167 | +func findLatestSessionPath(root, target string, notBefore time.Time, preExisting map[string]struct{}) (string, error) { | |
| 1144 | 1168 | type candidate struct { |
| 1145 | 1169 | path string |
| 1146 | 1170 | ts time.Time |
| 1147 | 1171 | } |
| 1148 | 1172 | var candidates []candidate |
| 1149 | 1173 | |
| 1150 | 1174 | err := filepath.WalkDir(root, func(path string, d os.DirEntry, walkErr error) error { |
| 1151 | 1175 | if walkErr != nil || d.IsDir() || !strings.HasSuffix(path, ".jsonl") { |
| 1152 | 1176 | return nil |
| 1177 | + } | |
| 1178 | + if preExisting != nil { | |
| 1179 | + if _, existed := preExisting[path]; existed { | |
| 1180 | + return nil | |
| 1181 | + } | |
| 1153 | 1182 | } |
| 1154 | 1183 | meta, ts, err := readSessionMeta(path) |
| 1155 | 1184 | if err != nil { |
| 1156 | 1185 | return nil |
| 1157 | 1186 | } |
| @@ -1168,29 +1197,16 @@ | ||
| 1168 | 1197 | return "", err |
| 1169 | 1198 | } |
| 1170 | 1199 | if len(candidates) == 0 { |
| 1171 | 1200 | return "", os.ErrNotExist |
| 1172 | 1201 | } |
| 1173 | - // Pick the session whose start time is closest to notBefore. | |
| 1174 | - // When multiple relays start near-simultaneously, each relay's | |
| 1175 | - // notBefore is slightly different, so each matches its own session. | |
| 1176 | - best := 0 | |
| 1177 | - bestDelta := candidates[0].ts.Sub(notBefore) | |
| 1178 | - for i := 1; i < len(candidates); i++ { | |
| 1179 | - delta := candidates[i].ts.Sub(notBefore) | |
| 1180 | - if delta < 0 { | |
| 1181 | - delta = -delta | |
| 1182 | - } | |
| 1183 | - if bestDelta < 0 { | |
| 1184 | - bestDelta = -bestDelta | |
| 1185 | - } | |
| 1186 | - if delta < bestDelta { | |
| 1187 | - best = i | |
| 1188 | - bestDelta = candidates[i].ts.Sub(notBefore) | |
| 1189 | - } | |
| 1190 | - } | |
| 1191 | - return candidates[best].path, nil | |
| 1202 | + // Pick the oldest new session — the first file created after our | |
| 1203 | + // subprocess started is most likely ours. | |
| 1204 | + sort.Slice(candidates, func(i, j int) bool { | |
| 1205 | + return candidates[i].ts.Before(candidates[j].ts) | |
| 1206 | + }) | |
| 1207 | + return candidates[0].path, nil | |
| 1192 | 1208 | } |
| 1193 | 1209 | |
| 1194 | 1210 | func readSessionMeta(path string) (sessionMetaPayload, time.Time, error) { |
| 1195 | 1211 | file, err := os.Open(path) |
| 1196 | 1212 | if err != nil { |
| 1197 | 1213 |
| --- cmd/codex-relay/main.go | |
| +++ cmd/codex-relay/main.go | |
| @@ -207,10 +207,17 @@ | |
| 207 | _ = relay.Close(closeCtx) |
| 208 | }() |
| 209 | } |
| 210 | |
| 211 | cmd := exec.Command(cfg.CodexBin, cfg.Args...) |
| 212 | startedAt := time.Now() |
| 213 | cmd.Env = append(os.Environ(), |
| 214 | "SCUTTLEBOT_CONFIG_FILE="+cfg.ConfigFile, |
| 215 | "SCUTTLEBOT_URL="+cfg.URL, |
| 216 | "SCUTTLEBOT_TOKEN="+cfg.Token, |
| @@ -221,11 +228,11 @@ | |
| 221 | "SCUTTLEBOT_SESSION_ID="+cfg.SessionID, |
| 222 | "SCUTTLEBOT_NICK="+cfg.Nick, |
| 223 | "SCUTTLEBOT_ACTIVITY_VIA_BROKER="+boolString(relayActive), |
| 224 | ) |
| 225 | if relayActive { |
| 226 | go mirrorSessionLoop(ctx, relay, cfg, startedAt) |
| 227 | go presenceLoopPtr(ctx, &relay, cfg.HeartbeatInterval) |
| 228 | } |
| 229 | |
| 230 | if !isInteractiveTTY() { |
| 231 | cmd.Stdin = os.Stdin |
| @@ -401,11 +408,11 @@ | |
| 401 | fmt.Fprintf(os.Stderr, "codex-relay: reconnected, restarting mirror and input loops\n") |
| 402 | |
| 403 | // Restart mirror and input loops with the new connector. |
| 404 | // Use epoch time for mirror so it finds the existing session file |
| 405 | // regardless of when it was last modified. |
| 406 | go mirrorSessionLoop(ctx, conn, cfg, time.Time{}) |
| 407 | go relayInputLoop(ctx, conn, cfg, state, ptmx, now) |
| 408 | break |
| 409 | } |
| 410 | } |
| 411 | } |
| @@ -802,16 +809,16 @@ | |
| 802 | func defaultSessionID(target string) string { |
| 803 | sum := crc32.ChecksumIEEE([]byte(fmt.Sprintf("%s|%d|%d|%d", target, os.Getpid(), os.Getppid(), time.Now().UnixNano()))) |
| 804 | return fmt.Sprintf("%08x", sum) |
| 805 | } |
| 806 | |
| 807 | func mirrorSessionLoop(ctx context.Context, relay sessionrelay.Connector, cfg config, startedAt time.Time) { |
| 808 | for { |
| 809 | if ctx.Err() != nil { |
| 810 | return |
| 811 | } |
| 812 | sessionPath, err := discoverSessionPath(ctx, cfg, startedAt) |
| 813 | if err != nil { |
| 814 | if ctx.Err() != nil { |
| 815 | return |
| 816 | } |
| 817 | time.Sleep(10 * time.Second) |
| @@ -834,11 +841,11 @@ | |
| 834 | } |
| 835 | return |
| 836 | } |
| 837 | } |
| 838 | |
| 839 | func discoverSessionPath(ctx context.Context, cfg config, startedAt time.Time) (string, error) { |
| 840 | root, err := codexSessionsRoot() |
| 841 | if err != nil { |
| 842 | return "", err |
| 843 | } |
| 844 | |
| @@ -848,11 +855,11 @@ | |
| 848 | }) |
| 849 | } |
| 850 | |
| 851 | target := filepath.Clean(cfg.TargetCWD) |
| 852 | return waitForSessionPath(ctx, func() (string, error) { |
| 853 | return findLatestSessionPath(root, target, startedAt.Add(-2*time.Second)) |
| 854 | }) |
| 855 | } |
| 856 | |
| 857 | func waitForSessionPath(ctx context.Context, find func() (string, error)) (string, error) { |
| 858 | ctx, cancel := context.WithTimeout(ctx, defaultDiscoverWait) |
| @@ -1102,10 +1109,25 @@ | |
| 1102 | return strings.TrimSpace(args[i+1]) |
| 1103 | } |
| 1104 | } |
| 1105 | return "" |
| 1106 | } |
| 1107 | |
| 1108 | func codexSessionsRoot() (string, error) { |
| 1109 | if value := os.Getenv("CODEX_HOME"); value != "" { |
| 1110 | return filepath.Join(value, "sessions"), nil |
| 1111 | } |
| @@ -1135,23 +1157,30 @@ | |
| 1135 | return "", os.ErrNotExist |
| 1136 | } |
| 1137 | return match, nil |
| 1138 | } |
| 1139 | |
| 1140 | // findLatestSessionPath finds the .jsonl file in root whose first entry |
| 1141 | // timestamp is closest to notBefore — this ensures each relay latches onto |
| 1142 | // its own subprocess's session when multiple sessions share a CWD. |
| 1143 | func findLatestSessionPath(root, target string, notBefore time.Time) (string, error) { |
| 1144 | type candidate struct { |
| 1145 | path string |
| 1146 | ts time.Time |
| 1147 | } |
| 1148 | var candidates []candidate |
| 1149 | |
| 1150 | err := filepath.WalkDir(root, func(path string, d os.DirEntry, walkErr error) error { |
| 1151 | if walkErr != nil || d.IsDir() || !strings.HasSuffix(path, ".jsonl") { |
| 1152 | return nil |
| 1153 | } |
| 1154 | meta, ts, err := readSessionMeta(path) |
| 1155 | if err != nil { |
| 1156 | return nil |
| 1157 | } |
| @@ -1168,29 +1197,16 @@ | |
| 1168 | return "", err |
| 1169 | } |
| 1170 | if len(candidates) == 0 { |
| 1171 | return "", os.ErrNotExist |
| 1172 | } |
| 1173 | // Pick the session whose start time is closest to notBefore. |
| 1174 | // When multiple relays start near-simultaneously, each relay's |
| 1175 | // notBefore is slightly different, so each matches its own session. |
| 1176 | best := 0 |
| 1177 | bestDelta := candidates[0].ts.Sub(notBefore) |
| 1178 | for i := 1; i < len(candidates); i++ { |
| 1179 | delta := candidates[i].ts.Sub(notBefore) |
| 1180 | if delta < 0 { |
| 1181 | delta = -delta |
| 1182 | } |
| 1183 | if bestDelta < 0 { |
| 1184 | bestDelta = -bestDelta |
| 1185 | } |
| 1186 | if delta < bestDelta { |
| 1187 | best = i |
| 1188 | bestDelta = candidates[i].ts.Sub(notBefore) |
| 1189 | } |
| 1190 | } |
| 1191 | return candidates[best].path, nil |
| 1192 | } |
| 1193 | |
| 1194 | func readSessionMeta(path string) (sessionMetaPayload, time.Time, error) { |
| 1195 | file, err := os.Open(path) |
| 1196 | if err != nil { |
| 1197 |
| --- cmd/codex-relay/main.go | |
| +++ cmd/codex-relay/main.go | |
| @@ -207,10 +207,17 @@ | |
| 207 | _ = relay.Close(closeCtx) |
| 208 | }() |
| 209 | } |
| 210 | |
| 211 | cmd := exec.Command(cfg.CodexBin, cfg.Args...) |
| 212 | // Snapshot existing session files before starting the subprocess so |
| 213 | // discovery can distinguish our session from pre-existing ones. |
| 214 | var preExisting map[string]struct{} |
| 215 | if sessRoot, err := codexSessionsRoot(); err == nil { |
| 216 | preExisting = snapshotSessionFiles(sessRoot) |
| 217 | } |
| 218 | |
| 219 | startedAt := time.Now() |
| 220 | cmd.Env = append(os.Environ(), |
| 221 | "SCUTTLEBOT_CONFIG_FILE="+cfg.ConfigFile, |
| 222 | "SCUTTLEBOT_URL="+cfg.URL, |
| 223 | "SCUTTLEBOT_TOKEN="+cfg.Token, |
| @@ -221,11 +228,11 @@ | |
| 228 | "SCUTTLEBOT_SESSION_ID="+cfg.SessionID, |
| 229 | "SCUTTLEBOT_NICK="+cfg.Nick, |
| 230 | "SCUTTLEBOT_ACTIVITY_VIA_BROKER="+boolString(relayActive), |
| 231 | ) |
| 232 | if relayActive { |
| 233 | go mirrorSessionLoop(ctx, relay, cfg, startedAt, preExisting) |
| 234 | go presenceLoopPtr(ctx, &relay, cfg.HeartbeatInterval) |
| 235 | } |
| 236 | |
| 237 | if !isInteractiveTTY() { |
| 238 | cmd.Stdin = os.Stdin |
| @@ -401,11 +408,11 @@ | |
| 408 | fmt.Fprintf(os.Stderr, "codex-relay: reconnected, restarting mirror and input loops\n") |
| 409 | |
| 410 | // Restart mirror and input loops with the new connector. |
| 411 | // Use epoch time for mirror so it finds the existing session file |
| 412 | // regardless of when it was last modified. |
| 413 | go mirrorSessionLoop(ctx, conn, cfg, time.Time{}, nil) |
| 414 | go relayInputLoop(ctx, conn, cfg, state, ptmx, now) |
| 415 | break |
| 416 | } |
| 417 | } |
| 418 | } |
| @@ -802,16 +809,16 @@ | |
| 809 | func defaultSessionID(target string) string { |
| 810 | sum := crc32.ChecksumIEEE([]byte(fmt.Sprintf("%s|%d|%d|%d", target, os.Getpid(), os.Getppid(), time.Now().UnixNano()))) |
| 811 | return fmt.Sprintf("%08x", sum) |
| 812 | } |
| 813 | |
| 814 | func mirrorSessionLoop(ctx context.Context, relay sessionrelay.Connector, cfg config, startedAt time.Time, preExisting map[string]struct{}) { |
| 815 | for { |
| 816 | if ctx.Err() != nil { |
| 817 | return |
| 818 | } |
| 819 | sessionPath, err := discoverSessionPath(ctx, cfg, startedAt, preExisting) |
| 820 | if err != nil { |
| 821 | if ctx.Err() != nil { |
| 822 | return |
| 823 | } |
| 824 | time.Sleep(10 * time.Second) |
| @@ -834,11 +841,11 @@ | |
| 841 | } |
| 842 | return |
| 843 | } |
| 844 | } |
| 845 | |
| 846 | func discoverSessionPath(ctx context.Context, cfg config, startedAt time.Time, preExisting map[string]struct{}) (string, error) { |
| 847 | root, err := codexSessionsRoot() |
| 848 | if err != nil { |
| 849 | return "", err |
| 850 | } |
| 851 | |
| @@ -848,11 +855,11 @@ | |
| 855 | }) |
| 856 | } |
| 857 | |
| 858 | target := filepath.Clean(cfg.TargetCWD) |
| 859 | return waitForSessionPath(ctx, func() (string, error) { |
| 860 | return findLatestSessionPath(root, target, startedAt.Add(-2*time.Second), preExisting) |
| 861 | }) |
| 862 | } |
| 863 | |
| 864 | func waitForSessionPath(ctx context.Context, find func() (string, error)) (string, error) { |
| 865 | ctx, cancel := context.WithTimeout(ctx, defaultDiscoverWait) |
| @@ -1102,10 +1109,25 @@ | |
| 1109 | return strings.TrimSpace(args[i+1]) |
| 1110 | } |
| 1111 | } |
| 1112 | return "" |
| 1113 | } |
| 1114 | |
| 1115 | // snapshotSessionFiles returns the set of .jsonl file paths currently under root. |
| 1116 | // Called before starting the Codex subprocess so discovery can skip pre-existing |
| 1117 | // sessions and deterministically find the one our subprocess creates. |
| 1118 | func snapshotSessionFiles(root string) map[string]struct{} { |
| 1119 | existing := make(map[string]struct{}) |
| 1120 | _ = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { |
| 1121 | if err != nil || d.IsDir() || !strings.HasSuffix(path, ".jsonl") { |
| 1122 | return nil |
| 1123 | } |
| 1124 | existing[path] = struct{}{} |
| 1125 | return nil |
| 1126 | }) |
| 1127 | return existing |
| 1128 | } |
| 1129 | |
| 1130 | func codexSessionsRoot() (string, error) { |
| 1131 | if value := os.Getenv("CODEX_HOME"); value != "" { |
| 1132 | return filepath.Join(value, "sessions"), nil |
| 1133 | } |
| @@ -1135,23 +1157,30 @@ | |
| 1157 | return "", os.ErrNotExist |
| 1158 | } |
| 1159 | return match, nil |
| 1160 | } |
| 1161 | |
| 1162 | // findLatestSessionPath finds the .jsonl file in root that was created by our |
| 1163 | // subprocess. It uses a pre-existing file snapshot to skip sessions that |
| 1164 | // existed before the subprocess started, then filters by CWD and picks the |
| 1165 | // oldest new match. When preExisting is nil (reconnect), it falls back to |
| 1166 | // accepting any file whose timestamp is >= notBefore. |
| 1167 | func findLatestSessionPath(root, target string, notBefore time.Time, preExisting map[string]struct{}) (string, error) { |
| 1168 | type candidate struct { |
| 1169 | path string |
| 1170 | ts time.Time |
| 1171 | } |
| 1172 | var candidates []candidate |
| 1173 | |
| 1174 | err := filepath.WalkDir(root, func(path string, d os.DirEntry, walkErr error) error { |
| 1175 | if walkErr != nil || d.IsDir() || !strings.HasSuffix(path, ".jsonl") { |
| 1176 | return nil |
| 1177 | } |
| 1178 | if preExisting != nil { |
| 1179 | if _, existed := preExisting[path]; existed { |
| 1180 | return nil |
| 1181 | } |
| 1182 | } |
| 1183 | meta, ts, err := readSessionMeta(path) |
| 1184 | if err != nil { |
| 1185 | return nil |
| 1186 | } |
| @@ -1168,29 +1197,16 @@ | |
| 1197 | return "", err |
| 1198 | } |
| 1199 | if len(candidates) == 0 { |
| 1200 | return "", os.ErrNotExist |
| 1201 | } |
| 1202 | // Pick the oldest new session — the first file created after our |
| 1203 | // subprocess started is most likely ours. |
| 1204 | sort.Slice(candidates, func(i, j int) bool { |
| 1205 | return candidates[i].ts.Before(candidates[j].ts) |
| 1206 | }) |
| 1207 | return candidates[0].path, nil |
| 1208 | } |
| 1209 | |
| 1210 | func readSessionMeta(path string) (sessionMetaPayload, time.Time, error) { |
| 1211 | file, err := os.Open(path) |
| 1212 | if err != nil { |
| 1213 |
+117
| --- cmd/codex-relay/main_test.go | ||
| +++ cmd/codex-relay/main_test.go | ||
| @@ -1,9 +1,11 @@ | ||
| 1 | 1 | package main |
| 2 | 2 | |
| 3 | 3 | import ( |
| 4 | 4 | "bytes" |
| 5 | + "fmt" | |
| 6 | + "os" | |
| 5 | 7 | "path/filepath" |
| 6 | 8 | "strings" |
| 7 | 9 | "testing" |
| 8 | 10 | "time" |
| 9 | 11 | ) |
| @@ -191,5 +193,120 @@ | ||
| 191 | 193 | want := "019d45e1-8328-7261-9a02-5c4304e07724" |
| 192 | 194 | if got != want { |
| 193 | 195 | t.Fatalf("explicitThreadID = %q, want %q", got, want) |
| 194 | 196 | } |
| 195 | 197 | } |
| 198 | + | |
| 199 | +func writeSessionFile(t *testing.T, dir, uuid, cwd, timestamp string) string { | |
| 200 | + t.Helper() | |
| 201 | + content := fmt.Sprintf(`{"type":"session_meta","payload":{"id":"%s","timestamp":"%s","cwd":"%s"}}`, uuid, timestamp, cwd) | |
| 202 | + name := fmt.Sprintf("rollout-%s-%s.jsonl", strings.ReplaceAll(timestamp[:19], ":", "-"), uuid) | |
| 203 | + path := filepath.Join(dir, name) | |
| 204 | + if err := os.WriteFile(path, []byte(content+"\n"), 0644); err != nil { | |
| 205 | + t.Fatal(err) | |
| 206 | + } | |
| 207 | + return path | |
| 208 | +} | |
| 209 | + | |
| 210 | +func TestFindLatestSessionPathSkipsPreExisting(t *testing.T) { | |
| 211 | + t.Helper() | |
| 212 | + | |
| 213 | + root := t.TempDir() | |
| 214 | + dateDir := filepath.Join(root, "2026", "04", "04") | |
| 215 | + if err := os.MkdirAll(dateDir, 0755); err != nil { | |
| 216 | + t.Fatal(err) | |
| 217 | + } | |
| 218 | + | |
| 219 | + cwd := "/home/user/project" | |
| 220 | + | |
| 221 | + // Create a pre-existing session file. | |
| 222 | + oldPath := writeSessionFile(t, dateDir, | |
| 223 | + "aaaa-aaaa-aaaa-aaaa", cwd, "2026-04-04T10:00:00Z") | |
| 224 | + | |
| 225 | + // Snapshot includes the old file. | |
| 226 | + preExisting := map[string]struct{}{oldPath: {}} | |
| 227 | + | |
| 228 | + // Create a new session file (not in snapshot). | |
| 229 | + newPath := writeSessionFile(t, dateDir, | |
| 230 | + "bbbb-bbbb-bbbb-bbbb", cwd, "2026-04-04T10:00:01Z") | |
| 231 | + | |
| 232 | + notBefore, _ := time.Parse(time.RFC3339, "2026-04-04T09:59:58Z") | |
| 233 | + got, err := findLatestSessionPath(root, cwd, notBefore, preExisting) | |
| 234 | + if err != nil { | |
| 235 | + t.Fatalf("findLatestSessionPath error: %v", err) | |
| 236 | + } | |
| 237 | + if got != newPath { | |
| 238 | + t.Fatalf("findLatestSessionPath = %q, want %q", got, newPath) | |
| 239 | + } | |
| 240 | +} | |
| 241 | + | |
| 242 | +func TestFindLatestSessionPathPicksOldestNew(t *testing.T) { | |
| 243 | + t.Helper() | |
| 244 | + | |
| 245 | + root := t.TempDir() | |
| 246 | + dateDir := filepath.Join(root, "2026", "04", "04") | |
| 247 | + if err := os.MkdirAll(dateDir, 0755); err != nil { | |
| 248 | + t.Fatal(err) | |
| 249 | + } | |
| 250 | + | |
| 251 | + cwd := "/home/user/project" | |
| 252 | + | |
| 253 | + // Two new sessions in the same CWD, no pre-existing. | |
| 254 | + earlyPath := writeSessionFile(t, dateDir, | |
| 255 | + "cccc-cccc-cccc-cccc", cwd, "2026-04-04T10:00:01Z") | |
| 256 | + _ = writeSessionFile(t, dateDir, | |
| 257 | + "dddd-dddd-dddd-dddd", cwd, "2026-04-04T10:00:02Z") | |
| 258 | + | |
| 259 | + notBefore, _ := time.Parse(time.RFC3339, "2026-04-04T10:00:00Z") | |
| 260 | + got, err := findLatestSessionPath(root, cwd, notBefore, map[string]struct{}{}) | |
| 261 | + if err != nil { | |
| 262 | + t.Fatalf("findLatestSessionPath error: %v", err) | |
| 263 | + } | |
| 264 | + if got != earlyPath { | |
| 265 | + t.Fatalf("findLatestSessionPath = %q, want oldest %q", got, earlyPath) | |
| 266 | + } | |
| 267 | +} | |
| 268 | + | |
| 269 | +func TestFindLatestSessionPathNilPreExistingAllowsAll(t *testing.T) { | |
| 270 | + t.Helper() | |
| 271 | + | |
| 272 | + root := t.TempDir() | |
| 273 | + dateDir := filepath.Join(root, "2026", "04", "04") | |
| 274 | + if err := os.MkdirAll(dateDir, 0755); err != nil { | |
| 275 | + t.Fatal(err) | |
| 276 | + } | |
| 277 | + | |
| 278 | + cwd := "/home/user/project" | |
| 279 | + | |
| 280 | + // Single file — nil preExisting (reconnect path) should find it. | |
| 281 | + path := writeSessionFile(t, dateDir, | |
| 282 | + "eeee-eeee-eeee-eeee", cwd, "2026-04-04T10:00:00Z") | |
| 283 | + | |
| 284 | + got, err := findLatestSessionPath(root, cwd, time.Time{}, nil) | |
| 285 | + if err != nil { | |
| 286 | + t.Fatalf("findLatestSessionPath error: %v", err) | |
| 287 | + } | |
| 288 | + if got != path { | |
| 289 | + t.Fatalf("findLatestSessionPath = %q, want %q", got, path) | |
| 290 | + } | |
| 291 | +} | |
| 292 | + | |
| 293 | +func TestSnapshotSessionFiles(t *testing.T) { | |
| 294 | + t.Helper() | |
| 295 | + | |
| 296 | + root := t.TempDir() | |
| 297 | + dateDir := filepath.Join(root, "2026", "04", "04") | |
| 298 | + if err := os.MkdirAll(dateDir, 0755); err != nil { | |
| 299 | + t.Fatal(err) | |
| 300 | + } | |
| 301 | + | |
| 302 | + path := writeSessionFile(t, dateDir, | |
| 303 | + "ffff-ffff-ffff-ffff", "/tmp", "2026-04-04T10:00:00Z") | |
| 304 | + | |
| 305 | + snap := snapshotSessionFiles(root) | |
| 306 | + if _, ok := snap[path]; !ok { | |
| 307 | + t.Fatalf("snapshotSessionFiles missing %q", path) | |
| 308 | + } | |
| 309 | + if len(snap) != 1 { | |
| 310 | + t.Fatalf("snapshotSessionFiles len = %d, want 1", len(snap)) | |
| 311 | + } | |
| 312 | +} | |
| 196 | 313 |
| --- cmd/codex-relay/main_test.go | |
| +++ cmd/codex-relay/main_test.go | |
| @@ -1,9 +1,11 @@ | |
| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "path/filepath" |
| 6 | "strings" |
| 7 | "testing" |
| 8 | "time" |
| 9 | ) |
| @@ -191,5 +193,120 @@ | |
| 191 | want := "019d45e1-8328-7261-9a02-5c4304e07724" |
| 192 | if got != want { |
| 193 | t.Fatalf("explicitThreadID = %q, want %q", got, want) |
| 194 | } |
| 195 | } |
| 196 |
| --- cmd/codex-relay/main_test.go | |
| +++ cmd/codex-relay/main_test.go | |
| @@ -1,9 +1,11 @@ | |
| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "fmt" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | "time" |
| 11 | ) |
| @@ -191,5 +193,120 @@ | |
| 193 | want := "019d45e1-8328-7261-9a02-5c4304e07724" |
| 194 | if got != want { |
| 195 | t.Fatalf("explicitThreadID = %q, want %q", got, want) |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | func writeSessionFile(t *testing.T, dir, uuid, cwd, timestamp string) string { |
| 200 | t.Helper() |
| 201 | content := fmt.Sprintf(`{"type":"session_meta","payload":{"id":"%s","timestamp":"%s","cwd":"%s"}}`, uuid, timestamp, cwd) |
| 202 | name := fmt.Sprintf("rollout-%s-%s.jsonl", strings.ReplaceAll(timestamp[:19], ":", "-"), uuid) |
| 203 | path := filepath.Join(dir, name) |
| 204 | if err := os.WriteFile(path, []byte(content+"\n"), 0644); err != nil { |
| 205 | t.Fatal(err) |
| 206 | } |
| 207 | return path |
| 208 | } |
| 209 | |
| 210 | func TestFindLatestSessionPathSkipsPreExisting(t *testing.T) { |
| 211 | t.Helper() |
| 212 | |
| 213 | root := t.TempDir() |
| 214 | dateDir := filepath.Join(root, "2026", "04", "04") |
| 215 | if err := os.MkdirAll(dateDir, 0755); err != nil { |
| 216 | t.Fatal(err) |
| 217 | } |
| 218 | |
| 219 | cwd := "/home/user/project" |
| 220 | |
| 221 | // Create a pre-existing session file. |
| 222 | oldPath := writeSessionFile(t, dateDir, |
| 223 | "aaaa-aaaa-aaaa-aaaa", cwd, "2026-04-04T10:00:00Z") |
| 224 | |
| 225 | // Snapshot includes the old file. |
| 226 | preExisting := map[string]struct{}{oldPath: {}} |
| 227 | |
| 228 | // Create a new session file (not in snapshot). |
| 229 | newPath := writeSessionFile(t, dateDir, |
| 230 | "bbbb-bbbb-bbbb-bbbb", cwd, "2026-04-04T10:00:01Z") |
| 231 | |
| 232 | notBefore, _ := time.Parse(time.RFC3339, "2026-04-04T09:59:58Z") |
| 233 | got, err := findLatestSessionPath(root, cwd, notBefore, preExisting) |
| 234 | if err != nil { |
| 235 | t.Fatalf("findLatestSessionPath error: %v", err) |
| 236 | } |
| 237 | if got != newPath { |
| 238 | t.Fatalf("findLatestSessionPath = %q, want %q", got, newPath) |
| 239 | } |
| 240 | } |
| 241 | |
| 242 | func TestFindLatestSessionPathPicksOldestNew(t *testing.T) { |
| 243 | t.Helper() |
| 244 | |
| 245 | root := t.TempDir() |
| 246 | dateDir := filepath.Join(root, "2026", "04", "04") |
| 247 | if err := os.MkdirAll(dateDir, 0755); err != nil { |
| 248 | t.Fatal(err) |
| 249 | } |
| 250 | |
| 251 | cwd := "/home/user/project" |
| 252 | |
| 253 | // Two new sessions in the same CWD, no pre-existing. |
| 254 | earlyPath := writeSessionFile(t, dateDir, |
| 255 | "cccc-cccc-cccc-cccc", cwd, "2026-04-04T10:00:01Z") |
| 256 | _ = writeSessionFile(t, dateDir, |
| 257 | "dddd-dddd-dddd-dddd", cwd, "2026-04-04T10:00:02Z") |
| 258 | |
| 259 | notBefore, _ := time.Parse(time.RFC3339, "2026-04-04T10:00:00Z") |
| 260 | got, err := findLatestSessionPath(root, cwd, notBefore, map[string]struct{}{}) |
| 261 | if err != nil { |
| 262 | t.Fatalf("findLatestSessionPath error: %v", err) |
| 263 | } |
| 264 | if got != earlyPath { |
| 265 | t.Fatalf("findLatestSessionPath = %q, want oldest %q", got, earlyPath) |
| 266 | } |
| 267 | } |
| 268 | |
| 269 | func TestFindLatestSessionPathNilPreExistingAllowsAll(t *testing.T) { |
| 270 | t.Helper() |
| 271 | |
| 272 | root := t.TempDir() |
| 273 | dateDir := filepath.Join(root, "2026", "04", "04") |
| 274 | if err := os.MkdirAll(dateDir, 0755); err != nil { |
| 275 | t.Fatal(err) |
| 276 | } |
| 277 | |
| 278 | cwd := "/home/user/project" |
| 279 | |
| 280 | // Single file — nil preExisting (reconnect path) should find it. |
| 281 | path := writeSessionFile(t, dateDir, |
| 282 | "eeee-eeee-eeee-eeee", cwd, "2026-04-04T10:00:00Z") |
| 283 | |
| 284 | got, err := findLatestSessionPath(root, cwd, time.Time{}, nil) |
| 285 | if err != nil { |
| 286 | t.Fatalf("findLatestSessionPath error: %v", err) |
| 287 | } |
| 288 | if got != path { |
| 289 | t.Fatalf("findLatestSessionPath = %q, want %q", got, path) |
| 290 | } |
| 291 | } |
| 292 | |
| 293 | func TestSnapshotSessionFiles(t *testing.T) { |
| 294 | t.Helper() |
| 295 | |
| 296 | root := t.TempDir() |
| 297 | dateDir := filepath.Join(root, "2026", "04", "04") |
| 298 | if err := os.MkdirAll(dateDir, 0755); err != nil { |
| 299 | t.Fatal(err) |
| 300 | } |
| 301 | |
| 302 | path := writeSessionFile(t, dateDir, |
| 303 | "ffff-ffff-ffff-ffff", "/tmp", "2026-04-04T10:00:00Z") |
| 304 | |
| 305 | snap := snapshotSessionFiles(root) |
| 306 | if _, ok := snap[path]; !ok { |
| 307 | t.Fatalf("snapshotSessionFiles missing %q", path) |
| 308 | } |
| 309 | if len(snap) != 1 { |
| 310 | t.Fatalf("snapshotSessionFiles len = %d, want 1", len(snap)) |
| 311 | } |
| 312 | } |
| 313 |