refactor: codex checker를 위한 다중 패턴 대기 지원 및 edge config 업데이트

- CodexChecker에 waitForAny 패턴 지원 추가 (업데이트 프롬프트 처리)
- 'Skip' 자동 응답 추가하여 slash command 소비 방지
- 테스트 케이스 추가 및 edge config 업데이트
- 더 이상 사용되지 않는 run_codex.py 삭제
This commit is contained in:
toki 2026-05-15 07:25:32 +09:00
parent 1a72d9caa6
commit 5d4cacbdaf
4 changed files with 86 additions and 65 deletions

View file

@ -55,25 +55,41 @@ func (c *CodexChecker) Check(ctx context.Context) (*UsageStatus, error) {
}()
var fullOutput string
waitFor := func(pattern *regexp.Regexp, timeout time.Duration) error {
type waitPattern struct {
name string
pattern *regexp.Regexp
}
waitForAny := func(patterns []waitPattern, timeout time.Duration) (string, error) {
timeoutCh := time.After(timeout)
for {
if pattern.MatchString(cleanANSI(fullOutput)) {
return nil
cleanOutput := cleanANSI(fullOutput)
for _, candidate := range patterns {
if candidate.pattern.MatchString(cleanOutput) {
return candidate.name, nil
}
}
select {
case <-ctx.Done():
return ctx.Err()
return "", ctx.Err()
case <-timeoutCh:
return fmt.Errorf("timeout waiting for %s", pattern.String())
names := make([]string, 0, len(patterns))
for _, candidate := range patterns {
names = append(names, candidate.pattern.String())
}
return "", fmt.Errorf("timeout waiting for %s", strings.Join(names, " or "))
case chunk, ok := <-chunks:
if !ok {
return io.EOF
return "", io.EOF
}
fullOutput += chunk
}
}
}
waitFor := func(pattern *regexp.Regexp, timeout time.Duration) error {
_, err := waitForAny([]waitPattern{{name: pattern.String(), pattern: pattern}}, timeout)
return err
}
sendText := func(text string) {
for _, char := range text {
@ -82,14 +98,26 @@ func (c *CodexChecker) Check(ctx context.Context) (*UsageStatus, error) {
}
}
// 1. Wait for startup ready prompt
// 1. Wait for startup ready prompt. Codex may show an update prompt before
// the composer; choose "Skip" so the following slash command is not consumed.
readyRegex := regexp.MustCompile(`(Tip:|model:)`)
if err := waitFor(readyRegex, 15*time.Second); err != nil {
updatePromptRegex := regexp.MustCompile(`Update available!|Press enter to continue`)
startupState, err := waitForAny([]waitPattern{
{name: "ready", pattern: readyRegex},
{name: "update", pattern: updatePromptRegex},
}, 15*time.Second)
if err != nil {
return nil, fmt.Errorf("codex startup failed: %w", err)
}
if startupState == "update" {
sendText("2\r")
if err := waitFor(readyRegex, 15*time.Second); err != nil {
return nil, fmt.Errorf("codex startup failed after skipping update prompt: %w", err)
}
}
// 2. Request status
sendText("/status\r")
sendText("\x15/status\r")
// 3. Wait for either refresh message or actual limits
statusOrRefreshRegex := regexp.MustCompile(`(?:refresh requested.*run /status again|Weekly limit:)`)

View file

@ -3,10 +3,58 @@ package status
import (
"context"
"fmt"
"os"
"path/filepath"
"testing"
"time"
)
func TestCodexCheckerSkipsUpdatePromptBeforeStatus(t *testing.T) {
dir := t.TempDir()
fakeCodex := filepath.Join(dir, "codex")
script := `#!/usr/bin/env sh
printf 'Update available! 0.128.0 -> 0.130.0\n'
printf 'Press enter to continue\n'
IFS= read -r choice
if [ "$choice" != "2" ]; then
printf 'unexpected choice: %s\n' "$choice"
exit 2
fi
printf 'model: test\n'
printf 'Tip: fake\n'
IFS= read -r command
if [ "$command" != "/status" ]; then
printf 'unexpected command: %s\n' "$command"
exit 3
fi
printf '5h limit: [####################] 98%% left (resets 18:38)\n'
printf 'Weekly limit: [####----------------] 22%% left (resets 10:20 on 8 May)\n'
IFS= read -r exit_command
exit 0
`
if err := os.WriteFile(fakeCodex, []byte(script), 0o755); err != nil {
t.Fatalf("write fake codex: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
checker := NewCodexChecker(fakeCodex)
status, err := checker.Check(ctx)
if err != nil {
t.Fatalf("Check failed: %v", err)
}
if status.DailyLimit != "98%" {
t.Fatalf("DailyLimit: got %q want %q", status.DailyLimit, "98%")
}
if status.WeeklyLimit != "22%" {
t.Fatalf("WeeklyLimit: got %q want %q", status.WeeklyLimit, "22%")
}
if status.WeeklyResetTime != "10:20 on 8 May" {
t.Fatalf("WeeklyResetTime: got %q want %q", status.WeeklyResetTime, "10:20 on 8 May")
}
}
func TestCodexChecker(t *testing.T) {
t.Skip("Skipping interactive TUI test by default")

View file

@ -17,7 +17,7 @@ metrics:
console:
adapter: "cli"
target: "gemini"
target: "codex"
session_id: "default"
background: false
timeout_sec: 300

View file

@ -1,55 +0,0 @@
import pty
import os
import sys
import time
import select
import termios
import struct
import fcntl
def set_winsize(fd, row, col):
winsize = struct.pack("HHHH", row, col, 0, 0)
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
def send_text(fd, text):
for char in text:
os.write(fd, char.encode('utf-8'))
time.sleep(0.1)
pid, fd = pty.fork()
if pid == 0:
os.environ['TERM'] = 'xterm-256color'
os.execvp('codex', ['codex', '--no-alt-screen'])
else:
set_winsize(fd, 40, 120)
time.sleep(6)
# First /status
send_text(fd, "/status\r")
time.sleep(8)
# Clear line by pressing ESC, or Backspace
send_text(fd, "\x1b") # ESC
time.sleep(1)
# Second /status
send_text(fd, "/status\r")
time.sleep(5)
output = b""
while True:
r, _, _ = select.select([fd], [], [], 2)
if r:
try:
data = os.read(fd, 4096)
if not data: break
output += data
except OSError:
break
else:
break
print(output.decode('utf-8', errors='ignore'))
send_text(fd, "/exit\r")
time.sleep(1)