Compare commits

...

10 Commits

Author SHA1 Message Date
cexll
f57ea2df59 chore: bump version to 5.2.4
Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-16 10:17:47 +08:00
cexll
d215c33549 fix(executor): isolate log files per task in parallel mode
Previously, all parallel tasks shared the same log file path, making it
difficult to debug individual task execution. This change creates a
separate log file for each task using the naming convention:
codeagent-wrapper-{pid}-{taskName}.log

Changes:
- Add withTaskLogger/taskLoggerFromContext for per-task logger injection
- Modify executeConcurrentWithContext to create independent Logger per task
- Update printTaskStart to display task-specific log paths
- Extract defaultRunCodexTaskFn for proper test hook reset
- Add runCodexTaskFn reset to resetTestHooks()

Test coverage: 93.7%

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-16 10:05:54 +08:00
ben
806bb04a35 Merge pull request #65 from cexll/fix/issue-64-buffer-overflow
fix(parser): 修复 bufio.Scanner token too long 错误
2025-12-15 14:22:03 +08:00
swe-agent[bot]
b1156038de test: 同步测试中的版本号至 5.2.3
修复 CI 失败:将 main_test.go 中的版本期望值从 5.2.2 更新为 5.2.3,
与 main.go 中的实际版本号保持一致。

修改文件:
- codeagent-wrapper/main_test.go:2693 (TestVersionFlag)
- codeagent-wrapper/main_test.go:2707 (TestVersionShortFlag)
- codeagent-wrapper/main_test.go:2721 (TestVersionLegacyAlias)

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-15 14:13:03 +08:00
swe-agent[bot]
0c93bbe574 change version 2025-12-15 13:23:26 +08:00
swe-agent[bot]
6f4f4e701b fix(parser): 修复 bufio.Scanner token too long 错误 (#64)
## 问题
- 执行 rg 等命令时,如果匹配到 minified 文件,单行输出可能超过 10MB
- 旧实现使用 bufio.Scanner,遇到超长行会报错并中止整个解析
- 导致后续的 agent_message 无法读取,任务失败

## 修复
1. **parser.go**:
   - 移除 bufio.Scanner,改用 bufio.Reader + readLineWithLimit
   - 超长行(>10MB)会被跳过但继续处理后续事件
   - 添加 codexHeader 轻量级解析,只在 agent_message 时完整解析

2. **utils.go**:
   - 修复 logWriter 内存膨胀问题
   - 添加 writeLimited 方法限制缓冲区大小

3. **测试**:
   - parser_token_too_long_test.go: 验证超长行处理
   - log_writer_limit_test.go: 验证日志缓冲限制

## 测试结果
-  TestParseJSONStream_SkipsOverlongLineAndContinues
-  TestLogWriterWriteLimitsBuffer
-  完整测试套件通过

Fixes #64

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-15 13:19:51 +08:00
swe-agent[bot]
ff301507fe test: Fix tests for ClaudeBackend default --dangerously-skip-permissions
- Update TestClaudeBuildArgs_ModesAndPermissions expectations
- Update TestBackendBuildArgs_ClaudeBackend expectations
- Update TestClaudeBackendBuildArgs_OutputValidation expectations
- Update version tests to expect 5.2.2

ClaudeBackend now defaults to adding --dangerously-skip-permissions
for automation workflows.

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-13 21:53:38 +08:00
swe-agent[bot]
93b72eba42 chore(v5.2.2): Bump version and clean up documentation
- Update version to 5.2.2 in README.md, README_CN.md, and codeagent-wrapper/main.go
- Remove non-existent documentation links from README.md (architecture.md, GITHUB-WORKFLOW.md, enterprise-workflow-ideas.md)
- Add coverage.out to .gitignore

Generated with swe-agent-bot

Co-Authored-By: swe-agent-bot <agent@swe-agent.ai>
2025-12-13 21:43:49 +08:00
swe-agent[bot]
b01758e7e1 fix codeagent backend claude no auto 2025-12-13 21:42:17 +08:00
swe-agent[bot]
c51b38c671 fix install.py dev fail 2025-12-13 21:41:55 +08:00
18 changed files with 1306 additions and 155 deletions

1
.gitignore vendored
View File

@@ -4,3 +4,4 @@
.pytest_cache
__pycache__
.coverage
coverage.out

View File

@@ -7,7 +7,7 @@
[![License: AGPL-3.0](https://img.shields.io/badge/License-AGPL_v3-blue.svg)](https://www.gnu.org/licenses/agpl-3.0)
[![Claude Code](https://img.shields.io/badge/Claude-Code-blue)](https://claude.ai/code)
[![Version](https://img.shields.io/badge/Version-5.2-green)](https://github.com/cexll/myclaude)
[![Version](https://img.shields.io/badge/Version-5.2.2-green)](https://github.com/cexll/myclaude)
> AI-powered development automation with multi-backend execution (Codex/Claude/Gemini)
@@ -320,13 +320,10 @@ python3 install.py --module dev --force
## Documentation
### Core Guides
- **[Architecture Overview](docs/architecture.md)** - System architecture and component design
- **[Codeagent-Wrapper Guide](docs/CODEAGENT-WRAPPER.md)** - Multi-backend execution wrapper
- **[GitHub Workflow Guide](docs/GITHUB-WORKFLOW.md)** - Issue-to-PR automation
- **[Hooks Documentation](docs/HOOKS.md)** - Custom hooks and automation
### Additional Resources
- **[Enterprise Workflow Ideas](docs/enterprise-workflow-ideas.md)** - Advanced patterns and best practices
- **[Installation Log](install.log)** - Installation history and troubleshooting
---

View File

@@ -2,7 +2,7 @@
[![License: AGPL-3.0](https://img.shields.io/badge/License-AGPL_v3-blue.svg)](https://www.gnu.org/licenses/agpl-3.0)
[![Claude Code](https://img.shields.io/badge/Claude-Code-blue)](https://claude.ai/code)
[![Version](https://img.shields.io/badge/Version-5.2-green)](https://github.com/cexll/myclaude)
[![Version](https://img.shields.io/badge/Version-5.2.2-green)](https://github.com/cexll/myclaude)
> AI 驱动的开发自动化 - 多后端执行架构 (Codex/Claude/Gemini)

View File

@@ -29,12 +29,12 @@ func (ClaudeBackend) BuildArgs(cfg *Config, targetArg string) []string {
if cfg == nil {
return nil
}
args := []string{"-p"}
args := []string{"-p", "--dangerously-skip-permissions"}
// Only skip permissions when explicitly requested
if cfg.SkipPermissions {
args = append(args, "--dangerously-skip-permissions")
}
// if cfg.SkipPermissions {
// args = append(args, "--dangerously-skip-permissions")
// }
if cfg.Mode == "resume" {
if cfg.SessionID != "" {

View File

@@ -11,7 +11,7 @@ func TestClaudeBuildArgs_ModesAndPermissions(t *testing.T) {
t.Run("new mode uses workdir without skip by default", func(t *testing.T) {
cfg := &Config{Mode: "new", WorkDir: "/repo"}
got := backend.BuildArgs(cfg, "todo")
want := []string{"-p", "--output-format", "stream-json", "--verbose", "todo"}
want := []string{"-p", "--dangerously-skip-permissions", "--output-format", "stream-json", "--verbose", "todo"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
@@ -29,7 +29,7 @@ func TestClaudeBuildArgs_ModesAndPermissions(t *testing.T) {
t.Run("resume mode uses session id and omits workdir", func(t *testing.T) {
cfg := &Config{Mode: "resume", SessionID: "sid-123", WorkDir: "/ignored"}
got := backend.BuildArgs(cfg, "resume-task")
want := []string{"-p", "-r", "sid-123", "--output-format", "stream-json", "--verbose", "resume-task"}
want := []string{"-p", "--dangerously-skip-permissions", "-r", "sid-123", "--output-format", "stream-json", "--verbose", "resume-task"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}
@@ -38,7 +38,7 @@ func TestClaudeBuildArgs_ModesAndPermissions(t *testing.T) {
t.Run("resume mode without session still returns base flags", func(t *testing.T) {
cfg := &Config{Mode: "resume", WorkDir: "/ignored"}
got := backend.BuildArgs(cfg, "follow-up")
want := []string{"-p", "--output-format", "stream-json", "--verbose", "follow-up"}
want := []string{"-p", "--dangerously-skip-permissions", "--output-format", "stream-json", "--verbose", "follow-up"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("got %v, want %v", got, want)
}

View File

@@ -122,7 +122,25 @@ type parseResult struct {
threadID string
}
var runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
type taskLoggerContextKey struct{}
func withTaskLogger(ctx context.Context, logger *Logger) context.Context {
if ctx == nil || logger == nil {
return ctx
}
return context.WithValue(ctx, taskLoggerContextKey{}, logger)
}
func taskLoggerFromContext(ctx context.Context) *Logger {
if ctx == nil {
return nil
}
logger, _ := ctx.Value(taskLoggerContextKey{}).(*Logger)
return logger
}
// defaultRunCodexTaskFn is the default implementation of runCodexTaskFn (exposed for test reset)
func defaultRunCodexTaskFn(task TaskSpec, timeout int) TaskResult {
if task.WorkDir == "" {
task.WorkDir = defaultWorkdir
}
@@ -151,6 +169,8 @@ var runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
return runCodexTaskWithContext(parentCtx, task, backend, nil, false, true, timeout)
}
var runCodexTaskFn = defaultRunCodexTaskFn
func topologicalSort(tasks []TaskSpec) ([][]TaskSpec, error) {
idToTask := make(map[string]TaskSpec, len(tasks))
indegree := make(map[string]int, len(tasks))
@@ -235,13 +255,8 @@ func executeConcurrentWithContext(parentCtx context.Context, layers [][]TaskSpec
var startPrintMu sync.Mutex
bannerPrinted := false
printTaskStart := func(taskID string) {
logger := activeLogger()
if logger == nil {
return
}
path := logger.Path()
if path == "" {
printTaskStart := func(taskID, logPath string) {
if logPath == "" {
return
}
startPrintMu.Lock()
@@ -249,7 +264,7 @@ func executeConcurrentWithContext(parentCtx context.Context, layers [][]TaskSpec
fmt.Fprintln(os.Stderr, "=== Starting Parallel Execution ===")
bannerPrinted = true
}
fmt.Fprintf(os.Stderr, "Task %s: Log: %s\n", taskID, path)
fmt.Fprintf(os.Stderr, "Task %s: Log: %s\n", taskID, logPath)
startPrintMu.Unlock()
}
@@ -319,9 +334,11 @@ func executeConcurrentWithContext(parentCtx context.Context, layers [][]TaskSpec
wg.Add(1)
go func(ts TaskSpec) {
defer wg.Done()
var taskLogger *Logger
var taskLogPath string
defer func() {
if r := recover(); r != nil {
resultsCh <- TaskResult{TaskID: ts.ID, ExitCode: 1, Error: fmt.Sprintf("panic: %v", r)}
resultsCh <- TaskResult{TaskID: ts.ID, ExitCode: 1, Error: fmt.Sprintf("panic: %v", r), LogPath: taskLogPath}
}
}()
@@ -338,9 +355,20 @@ func executeConcurrentWithContext(parentCtx context.Context, layers [][]TaskSpec
logConcurrencyState("done", ts.ID, int(after), workerLimit)
}()
ts.Context = ctx
printTaskStart(ts.ID)
resultsCh <- runCodexTaskFn(ts, timeout)
if l, err := NewLoggerWithSuffix(ts.ID); err == nil {
taskLogger = l
taskLogPath = l.Path()
defer func() { _ = taskLogger.Close() }()
}
ts.Context = withTaskLogger(ctx, taskLogger)
printTaskStart(ts.ID, taskLogPath)
res := runCodexTaskFn(ts, timeout)
if res.LogPath == "" && taskLogPath != "" {
res.LogPath = taskLogPath
}
resultsCh <- res
}(task)
}
@@ -458,14 +486,8 @@ func runCodexProcess(parentCtx context.Context, codexArgs []string, taskText str
func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backend Backend, customArgs []string, useCustomArgs bool, silent bool, timeoutSec int) TaskResult {
result := TaskResult{TaskID: taskSpec.ID}
setLogPath := func() {
if result.LogPath != "" {
return
}
if logger := activeLogger(); logger != nil {
result.LogPath = logger.Path()
}
}
injectedLogger := taskLoggerFromContext(parentCtx)
logger := injectedLogger
cfg := &Config{
Mode: taskSpec.Mode,
@@ -521,17 +543,17 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
if silent {
// Silent mode: only persist to file when available; avoid stderr noise.
logInfoFn = func(msg string) {
if logger := activeLogger(); logger != nil {
if logger != nil {
logger.Info(prefixMsg(msg))
}
}
logWarnFn = func(msg string) {
if logger := activeLogger(); logger != nil {
if logger != nil {
logger.Warn(prefixMsg(msg))
}
}
logErrorFn = func(msg string) {
if logger := activeLogger(); logger != nil {
if logger != nil {
logger.Error(prefixMsg(msg))
}
}
@@ -547,10 +569,11 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
var stderrLogger *logWriter
var tempLogger *Logger
if silent && activeLogger() == nil {
if logger == nil && silent && activeLogger() == nil {
if l, err := NewLogger(); err == nil {
setLogger(l)
tempLogger = l
logger = l
}
}
defer func() {
@@ -558,8 +581,16 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
_ = closeLogger()
}
}()
defer setLogPath()
if logger := activeLogger(); logger != nil {
defer func() {
if result.LogPath != "" || logger == nil {
return
}
result.LogPath = logger.Path()
}()
if logger == nil {
logger = activeLogger()
}
if logger != nil {
result.LogPath = logger.Path()
}
@@ -659,7 +690,7 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
}
logInfoFn(fmt.Sprintf("Starting %s with PID: %d", commandName, cmd.Process().Pid()))
if logger := activeLogger(); logger != nil {
if logger != nil {
logInfoFn(fmt.Sprintf("Log capturing to: %s", logger.Path()))
}
@@ -756,8 +787,8 @@ func runCodexTaskWithContext(parentCtx context.Context, taskSpec TaskSpec, backe
result.ExitCode = 0
result.Message = message
result.SessionID = threadID
if logger := activeLogger(); logger != nil {
result.LogPath = logger.Path()
if result.LogPath == "" && injectedLogger != nil {
result.LogPath = injectedLogger.Path()
}
return result

View File

@@ -1,12 +1,15 @@
package main
import (
"bufio"
"bytes"
"context"
"errors"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"strings"
"sync"
"sync/atomic"
@@ -15,6 +18,12 @@ import (
"time"
)
var executorTestTaskCounter atomic.Int64
func nextExecutorTestTaskID(prefix string) string {
return fmt.Sprintf("%s-%d", prefix, executorTestTaskCounter.Add(1))
}
type execFakeProcess struct {
pid int
signals []os.Signal
@@ -76,6 +85,7 @@ type execFakeRunner struct {
stdout io.ReadCloser
process processHandle
stdin io.WriteCloser
dir string
waitErr error
waitDelay time.Duration
startErr error
@@ -117,7 +127,7 @@ func (f *execFakeRunner) StdinPipe() (io.WriteCloser, error) {
return &writeCloserStub{}, nil
}
func (f *execFakeRunner) SetStderr(io.Writer) {}
func (f *execFakeRunner) SetDir(string) {}
func (f *execFakeRunner) SetDir(dir string) { f.dir = dir }
func (f *execFakeRunner) Process() processHandle {
if f.process != nil {
return f.process
@@ -149,6 +159,10 @@ func TestExecutorHelperCoverage(t *testing.T) {
}
rcWithCmd := &realCmd{cmd: &exec.Cmd{}}
rcWithCmd.SetStderr(io.Discard)
rcWithCmd.SetDir("/tmp")
if rcWithCmd.cmd.Dir != "/tmp" {
t.Fatalf("expected SetDir to set cmd.Dir, got %q", rcWithCmd.cmd.Dir)
}
echoCmd := exec.Command("echo", "ok")
rcProc := &realCmd{cmd: echoCmd}
stdoutPipe, err := rcProc.StdoutPipe()
@@ -421,6 +435,63 @@ func TestExecutorRunCodexTaskWithContext(t *testing.T) {
_ = closeLogger()
})
t.Run("injectedLogger", func(t *testing.T) {
newCommandRunner = func(ctx context.Context, name string, args ...string) commandRunner {
return &execFakeRunner{
stdout: newReasonReadCloser(`{"type":"item.completed","item":{"type":"agent_message","text":"injected"}}`),
process: &execFakeProcess{pid: 12},
}
}
_ = closeLogger()
injected, err := NewLoggerWithSuffix("executor-injected")
if err != nil {
t.Fatalf("NewLoggerWithSuffix() error = %v", err)
}
defer func() {
_ = injected.Close()
_ = os.Remove(injected.Path())
}()
ctx := withTaskLogger(context.Background(), injected)
res := runCodexTaskWithContext(ctx, TaskSpec{ID: "task-injected", Task: "payload", WorkDir: "."}, nil, nil, false, true, 1)
if res.ExitCode != 0 || res.LogPath != injected.Path() {
t.Fatalf("expected injected logger path, got %+v", res)
}
if activeLogger() != nil {
t.Fatalf("expected no global logger to be created when injected")
}
injected.Flush()
data, err := os.ReadFile(injected.Path())
if err != nil {
t.Fatalf("failed to read injected log file: %v", err)
}
if !strings.Contains(string(data), "task-injected") {
t.Fatalf("injected log missing task prefix, content: %s", string(data))
}
})
t.Run("backendSetsDirAndNilContext", func(t *testing.T) {
var rc *execFakeRunner
newCommandRunner = func(ctx context.Context, name string, args ...string) commandRunner {
rc = &execFakeRunner{
stdout: newReasonReadCloser(`{"type":"item.completed","item":{"type":"agent_message","text":"backend"}}`),
process: &execFakeProcess{pid: 13},
}
return rc
}
_ = closeLogger()
res := runCodexTaskWithContext(nil, TaskSpec{ID: "task-backend", Task: "payload", WorkDir: "/tmp"}, ClaudeBackend{}, nil, false, false, 1)
if res.ExitCode != 0 || res.Message != "backend" {
t.Fatalf("unexpected result: %+v", res)
}
if rc == nil || rc.dir != "/tmp" {
t.Fatalf("expected backend to set cmd.Dir, got runner=%v dir=%q", rc, rc.dir)
}
})
t.Run("missingMessage", func(t *testing.T) {
newCommandRunner = func(ctx context.Context, name string, args ...string) commandRunner {
return &execFakeRunner{
@@ -435,6 +506,476 @@ func TestExecutorRunCodexTaskWithContext(t *testing.T) {
})
}
func TestExecutorParallelLogIsolation(t *testing.T) {
mainLogger, err := NewLoggerWithSuffix("executor-main")
if err != nil {
t.Fatalf("NewLoggerWithSuffix() error = %v", err)
}
setLogger(mainLogger)
t.Cleanup(func() {
_ = closeLogger()
_ = os.Remove(mainLogger.Path())
})
taskA := nextExecutorTestTaskID("iso-a")
taskB := nextExecutorTestTaskID("iso-b")
markerA := "ISOLATION_MARKER:" + taskA
markerB := "ISOLATION_MARKER:" + taskB
origRun := runCodexTaskFn
runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
logger := taskLoggerFromContext(task.Context)
if logger == nil {
return TaskResult{TaskID: task.ID, ExitCode: 1, Error: "missing task logger"}
}
switch task.ID {
case taskA:
logger.Info(markerA)
case taskB:
logger.Info(markerB)
default:
logger.Info("unexpected task: " + task.ID)
}
return TaskResult{TaskID: task.ID, ExitCode: 0}
}
t.Cleanup(func() { runCodexTaskFn = origRun })
stderrR, stderrW, err := os.Pipe()
if err != nil {
t.Fatalf("os.Pipe() error = %v", err)
}
oldStderr := os.Stderr
os.Stderr = stderrW
defer func() { os.Stderr = oldStderr }()
results := executeConcurrentWithContext(nil, [][]TaskSpec{{{ID: taskA}, {ID: taskB}}}, 1, -1)
_ = stderrW.Close()
os.Stderr = oldStderr
stderrData, _ := io.ReadAll(stderrR)
_ = stderrR.Close()
stderrOut := string(stderrData)
if len(results) != 2 {
t.Fatalf("expected 2 results, got %d", len(results))
}
paths := map[string]string{}
for _, res := range results {
if res.ExitCode != 0 {
t.Fatalf("unexpected failure: %+v", res)
}
if res.LogPath == "" {
t.Fatalf("missing LogPath for task %q", res.TaskID)
}
paths[res.TaskID] = res.LogPath
}
if paths[taskA] == paths[taskB] {
t.Fatalf("expected distinct task log paths, got %q", paths[taskA])
}
if strings.Contains(stderrOut, mainLogger.Path()) {
t.Fatalf("stderr should not print main log path: %s", stderrOut)
}
if !strings.Contains(stderrOut, paths[taskA]) || !strings.Contains(stderrOut, paths[taskB]) {
t.Fatalf("stderr should include task log paths, got: %s", stderrOut)
}
mainLogger.Flush()
mainData, err := os.ReadFile(mainLogger.Path())
if err != nil {
t.Fatalf("failed to read main log: %v", err)
}
if strings.Contains(string(mainData), markerA) || strings.Contains(string(mainData), markerB) {
t.Fatalf("main log should not contain task markers, content: %s", string(mainData))
}
taskAData, err := os.ReadFile(paths[taskA])
if err != nil {
t.Fatalf("failed to read task A log: %v", err)
}
taskBData, err := os.ReadFile(paths[taskB])
if err != nil {
t.Fatalf("failed to read task B log: %v", err)
}
if !strings.Contains(string(taskAData), markerA) || strings.Contains(string(taskAData), markerB) {
t.Fatalf("task A log isolation failed, content: %s", string(taskAData))
}
if !strings.Contains(string(taskBData), markerB) || strings.Contains(string(taskBData), markerA) {
t.Fatalf("task B log isolation failed, content: %s", string(taskBData))
}
_ = os.Remove(paths[taskA])
_ = os.Remove(paths[taskB])
}
func TestConcurrentExecutorParallelLogIsolationAndClosure(t *testing.T) {
tempDir := t.TempDir()
t.Setenv("TMPDIR", tempDir)
oldArgs := os.Args
os.Args = []string{defaultWrapperName}
t.Cleanup(func() { os.Args = oldArgs })
mainLogger, err := NewLoggerWithSuffix("concurrent-main")
if err != nil {
t.Fatalf("NewLoggerWithSuffix() error = %v", err)
}
setLogger(mainLogger)
t.Cleanup(func() {
mainLogger.Flush()
_ = closeLogger()
_ = os.Remove(mainLogger.Path())
})
const taskCount = 16
const writersPerTask = 4
const logsPerWriter = 50
const expectedTaskLines = writersPerTask * logsPerWriter
taskIDs := make([]string, 0, taskCount)
tasks := make([]TaskSpec, 0, taskCount)
for i := 0; i < taskCount; i++ {
id := nextExecutorTestTaskID("iso")
taskIDs = append(taskIDs, id)
tasks = append(tasks, TaskSpec{ID: id})
}
type taskLoggerInfo struct {
taskID string
logger *Logger
}
loggerCh := make(chan taskLoggerInfo, taskCount)
readyCh := make(chan struct{}, taskCount)
startCh := make(chan struct{})
go func() {
for i := 0; i < taskCount; i++ {
<-readyCh
}
close(startCh)
}()
origRun := runCodexTaskFn
runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
readyCh <- struct{}{}
logger := taskLoggerFromContext(task.Context)
loggerCh <- taskLoggerInfo{taskID: task.ID, logger: logger}
if logger == nil {
return TaskResult{TaskID: task.ID, ExitCode: 1, Error: "missing task logger"}
}
<-startCh
var wg sync.WaitGroup
wg.Add(writersPerTask)
for g := 0; g < writersPerTask; g++ {
go func(g int) {
defer wg.Done()
for i := 0; i < logsPerWriter; i++ {
logger.Info(fmt.Sprintf("TASK=%s g=%d i=%d", task.ID, g, i))
}
}(g)
}
wg.Wait()
return TaskResult{TaskID: task.ID, ExitCode: 0}
}
t.Cleanup(func() { runCodexTaskFn = origRun })
results := executeConcurrentWithContext(context.Background(), [][]TaskSpec{tasks}, 1, 0)
if len(results) != taskCount {
t.Fatalf("expected %d results, got %d", taskCount, len(results))
}
taskLogPaths := make(map[string]string, taskCount)
seenPaths := make(map[string]struct{}, taskCount)
for _, res := range results {
if res.ExitCode != 0 || res.Error != "" {
t.Fatalf("unexpected task failure: %+v", res)
}
if res.LogPath == "" {
t.Fatalf("missing LogPath for task %q", res.TaskID)
}
if _, ok := taskLogPaths[res.TaskID]; ok {
t.Fatalf("duplicate TaskID in results: %q", res.TaskID)
}
taskLogPaths[res.TaskID] = res.LogPath
if _, ok := seenPaths[res.LogPath]; ok {
t.Fatalf("expected unique log path per task; duplicate path %q", res.LogPath)
}
seenPaths[res.LogPath] = struct{}{}
}
if len(taskLogPaths) != taskCount {
t.Fatalf("expected %d unique task IDs, got %d", taskCount, len(taskLogPaths))
}
prefix := primaryLogPrefix()
pid := os.Getpid()
for _, id := range taskIDs {
path := taskLogPaths[id]
if path == "" {
t.Fatalf("missing log path for task %q", id)
}
if _, err := os.Stat(path); err != nil {
t.Fatalf("task log file not created for %q: %v", id, err)
}
wantBase := fmt.Sprintf("%s-%d-%s.log", prefix, pid, id)
if got := filepath.Base(path); got != wantBase {
t.Fatalf("unexpected log filename for %q: got %q, want %q", id, got, wantBase)
}
}
loggers := make(map[string]*Logger, taskCount)
for i := 0; i < taskCount; i++ {
info := <-loggerCh
if info.taskID == "" {
t.Fatalf("missing taskID in logger info")
}
if info.logger == nil {
t.Fatalf("missing logger in context for task %q", info.taskID)
}
if prev, ok := loggers[info.taskID]; ok && prev != info.logger {
t.Fatalf("task %q received multiple logger instances", info.taskID)
}
loggers[info.taskID] = info.logger
}
if len(loggers) != taskCount {
t.Fatalf("expected %d task loggers, got %d", taskCount, len(loggers))
}
for taskID, logger := range loggers {
if !logger.closed.Load() {
t.Fatalf("expected task logger to be closed for %q", taskID)
}
if logger.file == nil {
t.Fatalf("expected task logger file to be non-nil for %q", taskID)
}
if _, err := logger.file.Write([]byte("x")); err == nil {
t.Fatalf("expected task logger file to be closed for %q", taskID)
}
}
mainLogger.Flush()
mainData, err := os.ReadFile(mainLogger.Path())
if err != nil {
t.Fatalf("failed to read main log: %v", err)
}
mainText := string(mainData)
if !strings.Contains(mainText, "parallel: worker_limit=") {
t.Fatalf("expected main log to include concurrency planning, content: %s", mainText)
}
if strings.Contains(mainText, "TASK=") {
t.Fatalf("main log should not contain task output, content: %s", mainText)
}
for taskID, path := range taskLogPaths {
f, err := os.Open(path)
if err != nil {
t.Fatalf("failed to open task log for %q: %v", taskID, err)
}
scanner := bufio.NewScanner(f)
lines := 0
for scanner.Scan() {
line := scanner.Text()
if strings.Contains(line, "parallel:") {
t.Fatalf("task log should not contain main log entries for %q: %s", taskID, line)
}
gotID, ok := parseTaskIDFromLogLine(line)
if !ok {
t.Fatalf("task log entry missing task marker for %q: %s", taskID, line)
}
if gotID != taskID {
t.Fatalf("task log isolation failed: file=%q got TASK=%q want TASK=%q", path, gotID, taskID)
}
lines++
}
if err := scanner.Err(); err != nil {
_ = f.Close()
t.Fatalf("scanner error for %q: %v", taskID, err)
}
if err := f.Close(); err != nil {
t.Fatalf("failed to close task log for %q: %v", taskID, err)
}
if lines != expectedTaskLines {
t.Fatalf("unexpected task log line count for %q: got %d, want %d", taskID, lines, expectedTaskLines)
}
}
for _, path := range taskLogPaths {
_ = os.Remove(path)
}
}
func parseTaskIDFromLogLine(line string) (string, bool) {
const marker = "TASK="
idx := strings.Index(line, marker)
if idx == -1 {
return "", false
}
rest := line[idx+len(marker):]
end := strings.IndexByte(rest, ' ')
if end == -1 {
return rest, rest != ""
}
return rest[:end], rest[:end] != ""
}
func TestExecutorTaskLoggerContext(t *testing.T) {
if taskLoggerFromContext(nil) != nil {
t.Fatalf("expected nil logger from nil context")
}
if taskLoggerFromContext(context.Background()) != nil {
t.Fatalf("expected nil logger when context has no logger")
}
logger, err := NewLoggerWithSuffix("executor-taskctx")
if err != nil {
t.Fatalf("NewLoggerWithSuffix() error = %v", err)
}
defer func() {
_ = logger.Close()
_ = os.Remove(logger.Path())
}()
ctx := withTaskLogger(context.Background(), logger)
if got := taskLoggerFromContext(ctx); got != logger {
t.Fatalf("expected logger roundtrip, got %v", got)
}
if taskLoggerFromContext(withTaskLogger(context.Background(), nil)) != nil {
t.Fatalf("expected nil logger when injected logger is nil")
}
}
func TestExecutorExecuteConcurrentWithContextBranches(t *testing.T) {
devNull, err := os.OpenFile(os.DevNull, os.O_WRONLY, 0)
if err != nil {
t.Fatalf("failed to open %s: %v", os.DevNull, err)
}
oldStderr := os.Stderr
os.Stderr = devNull
t.Cleanup(func() {
os.Stderr = oldStderr
_ = devNull.Close()
})
t.Run("skipOnFailedDependencies", func(t *testing.T) {
root := nextExecutorTestTaskID("root")
child := nextExecutorTestTaskID("child")
orig := runCodexTaskFn
runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
if task.ID == root {
return TaskResult{TaskID: task.ID, ExitCode: 1, Error: "boom"}
}
return TaskResult{TaskID: task.ID, ExitCode: 0}
}
t.Cleanup(func() { runCodexTaskFn = orig })
results := executeConcurrentWithContext(context.Background(), [][]TaskSpec{
{{ID: root}},
{{ID: child, Dependencies: []string{root}}},
}, 1, 0)
foundChild := false
for _, res := range results {
if res.LogPath != "" {
_ = os.Remove(res.LogPath)
}
if res.TaskID != child {
continue
}
foundChild = true
if res.ExitCode == 0 || !strings.Contains(res.Error, "skipped") {
t.Fatalf("expected skipped child task result, got %+v", res)
}
}
if !foundChild {
t.Fatalf("expected child task to be present in results")
}
})
t.Run("panicRecovered", func(t *testing.T) {
taskID := nextExecutorTestTaskID("panic")
orig := runCodexTaskFn
runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
panic("boom")
}
t.Cleanup(func() { runCodexTaskFn = orig })
results := executeConcurrentWithContext(context.Background(), [][]TaskSpec{{{ID: taskID}}}, 1, 0)
if len(results) != 1 {
t.Fatalf("expected 1 result, got %d", len(results))
}
if results[0].ExitCode == 0 || !strings.Contains(results[0].Error, "panic") {
t.Fatalf("expected panic result, got %+v", results[0])
}
if results[0].LogPath == "" {
t.Fatalf("expected LogPath on panic result")
}
_ = os.Remove(results[0].LogPath)
})
t.Run("cancelWhileWaitingForWorker", func(t *testing.T) {
task1 := nextExecutorTestTaskID("slot")
task2 := nextExecutorTestTaskID("slot")
parentCtx, cancel := context.WithCancel(context.Background())
started := make(chan struct{})
unblock := make(chan struct{})
var startedOnce sync.Once
orig := runCodexTaskFn
runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
startedOnce.Do(func() { close(started) })
<-unblock
return TaskResult{TaskID: task.ID, ExitCode: 0}
}
t.Cleanup(func() { runCodexTaskFn = orig })
go func() {
<-started
cancel()
time.Sleep(50 * time.Millisecond)
close(unblock)
}()
results := executeConcurrentWithContext(parentCtx, [][]TaskSpec{{{ID: task1}, {ID: task2}}}, 1, 1)
foundCancelled := false
for _, res := range results {
if res.LogPath != "" {
_ = os.Remove(res.LogPath)
}
if res.ExitCode == 130 {
foundCancelled = true
}
}
if !foundCancelled {
t.Fatalf("expected a task to be cancelled")
}
})
t.Run("loggerCreateFails", func(t *testing.T) {
taskID := nextExecutorTestTaskID("bad") + "/id"
orig := runCodexTaskFn
runCodexTaskFn = func(task TaskSpec, timeout int) TaskResult {
return TaskResult{TaskID: task.ID, ExitCode: 0}
}
t.Cleanup(func() { runCodexTaskFn = orig })
results := executeConcurrentWithContext(context.Background(), [][]TaskSpec{{{ID: taskID}}}, 1, 0)
if len(results) != 1 || results[0].ExitCode != 0 {
t.Fatalf("unexpected results: %+v", results)
}
})
}
func TestExecutorSignalAndTermination(t *testing.T) {
forceKillDelay.Store(0)
defer forceKillDelay.Store(5)

View File

@@ -0,0 +1,39 @@
package main
import (
"os"
"strings"
"testing"
)
func TestLogWriterWriteLimitsBuffer(t *testing.T) {
defer resetTestHooks()
logger, err := NewLogger()
if err != nil {
t.Fatalf("NewLogger error: %v", err)
}
setLogger(logger)
defer closeLogger()
lw := newLogWriter("P:", 10)
_, _ = lw.Write([]byte(strings.Repeat("a", 100)))
if lw.buf.Len() != 10 {
t.Fatalf("logWriter buffer len=%d, want %d", lw.buf.Len(), 10)
}
if !lw.dropped {
t.Fatalf("expected logWriter to drop overlong line bytes")
}
lw.Flush()
logger.Flush()
data, err := os.ReadFile(logger.Path())
if err != nil {
t.Fatalf("ReadFile error: %v", err)
}
if !strings.Contains(string(data), "P:aaaaaaa...") {
t.Fatalf("log output missing truncated entry, got %q", string(data))
}
}

View File

@@ -0,0 +1,158 @@
package main
import (
"fmt"
"os"
"path/filepath"
"strings"
"testing"
)
func TestLoggerNilReceiverNoop(t *testing.T) {
var logger *Logger
logger.Info("info")
logger.Warn("warn")
logger.Debug("debug")
logger.Error("error")
logger.Flush()
if err := logger.Close(); err != nil {
t.Fatalf("Close() on nil logger should return nil, got %v", err)
}
}
func TestLoggerConcurrencyLogHelpers(t *testing.T) {
setTempDirEnv(t, t.TempDir())
logger, err := NewLoggerWithSuffix("concurrency")
if err != nil {
t.Fatalf("NewLoggerWithSuffix error: %v", err)
}
setLogger(logger)
defer closeLogger()
logConcurrencyPlanning(0, 2)
logConcurrencyPlanning(3, 2)
logConcurrencyState("start", "task-1", 1, 0)
logConcurrencyState("done", "task-1", 0, 3)
logger.Flush()
data, err := os.ReadFile(logger.Path())
if err != nil {
t.Fatalf("failed to read log file: %v", err)
}
output := string(data)
checks := []string{
"parallel: worker_limit=unbounded total_tasks=2",
"parallel: worker_limit=3 total_tasks=2",
"parallel: start task=task-1 active=1 limit=unbounded",
"parallel: done task=task-1 active=0 limit=3",
}
for _, c := range checks {
if !strings.Contains(output, c) {
t.Fatalf("log output missing %q, got: %s", c, output)
}
}
}
func TestLoggerConcurrencyLogHelpersNoopWithoutActiveLogger(t *testing.T) {
_ = closeLogger()
logConcurrencyPlanning(1, 1)
logConcurrencyState("start", "task-1", 0, 1)
}
func TestLoggerCleanupOldLogsSkipsUnsafeAndHandlesAlreadyDeleted(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
unsafePath := createTempLog(t, tempDir, fmt.Sprintf("%s-%d.log", primaryLogPrefix(), 222))
orphanPath := createTempLog(t, tempDir, fmt.Sprintf("%s-%d.log", primaryLogPrefix(), 111))
stubFileStat(t, func(path string) (os.FileInfo, error) {
if path == unsafePath {
return fakeFileInfo{mode: os.ModeSymlink}, nil
}
return os.Lstat(path)
})
stubProcessRunning(t, func(pid int) bool {
if pid == 111 {
_ = os.Remove(orphanPath)
}
return false
})
stats, err := cleanupOldLogs()
if err != nil {
t.Fatalf("cleanupOldLogs() unexpected error: %v", err)
}
if stats.Scanned != 2 {
t.Fatalf("scanned = %d, want %d", stats.Scanned, 2)
}
if stats.Deleted != 0 {
t.Fatalf("deleted = %d, want %d", stats.Deleted, 0)
}
if stats.Kept != 2 {
t.Fatalf("kept = %d, want %d", stats.Kept, 2)
}
if stats.Errors != 0 {
t.Fatalf("errors = %d, want %d", stats.Errors, 0)
}
hasSkip := false
hasAlreadyDeleted := false
for _, name := range stats.KeptFiles {
if strings.Contains(name, "already deleted") {
hasAlreadyDeleted = true
}
if strings.Contains(name, filepath.Base(unsafePath)) {
hasSkip = true
}
}
if !hasSkip {
t.Fatalf("expected kept files to include unsafe log %q, got %+v", filepath.Base(unsafePath), stats.KeptFiles)
}
if !hasAlreadyDeleted {
t.Fatalf("expected kept files to include already deleted marker, got %+v", stats.KeptFiles)
}
}
func TestLoggerIsUnsafeFileErrorPaths(t *testing.T) {
tempDir := t.TempDir()
t.Run("stat ErrNotExist", func(t *testing.T) {
stubFileStat(t, func(string) (os.FileInfo, error) {
return nil, os.ErrNotExist
})
unsafe, reason := isUnsafeFile("missing.log", tempDir)
if !unsafe || reason != "" {
t.Fatalf("expected missing file to be skipped silently, got unsafe=%v reason=%q", unsafe, reason)
}
})
t.Run("stat error", func(t *testing.T) {
stubFileStat(t, func(string) (os.FileInfo, error) {
return nil, fmt.Errorf("boom")
})
unsafe, reason := isUnsafeFile("broken.log", tempDir)
if !unsafe || !strings.Contains(reason, "stat failed") {
t.Fatalf("expected stat failure to be unsafe, got unsafe=%v reason=%q", unsafe, reason)
}
})
t.Run("EvalSymlinks error", func(t *testing.T) {
stubFileStat(t, func(string) (os.FileInfo, error) {
return fakeFileInfo{}, nil
})
stubEvalSymlinks(t, func(string) (string, error) {
return "", fmt.Errorf("resolve failed")
})
unsafe, reason := isUnsafeFile("cannot-resolve.log", tempDir)
if !unsafe || !strings.Contains(reason, "path resolution failed") {
t.Fatalf("expected resolution failure to be unsafe, got unsafe=%v reason=%q", unsafe, reason)
}
})
}

View File

@@ -0,0 +1,80 @@
package main
import (
"fmt"
"os"
"path/filepath"
"strings"
"testing"
)
func TestLoggerWithSuffixNamingAndIsolation(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
taskA := "task-1"
taskB := "task-2"
loggerA, err := NewLoggerWithSuffix(taskA)
if err != nil {
t.Fatalf("NewLoggerWithSuffix(%q) error = %v", taskA, err)
}
defer loggerA.Close()
loggerB, err := NewLoggerWithSuffix(taskB)
if err != nil {
t.Fatalf("NewLoggerWithSuffix(%q) error = %v", taskB, err)
}
defer loggerB.Close()
wantA := filepath.Join(tempDir, fmt.Sprintf("%s-%d-%s.log", primaryLogPrefix(), os.Getpid(), taskA))
if loggerA.Path() != wantA {
t.Fatalf("loggerA path = %q, want %q", loggerA.Path(), wantA)
}
wantB := filepath.Join(tempDir, fmt.Sprintf("%s-%d-%s.log", primaryLogPrefix(), os.Getpid(), taskB))
if loggerB.Path() != wantB {
t.Fatalf("loggerB path = %q, want %q", loggerB.Path(), wantB)
}
if loggerA.Path() == loggerB.Path() {
t.Fatalf("expected different log files, got %q", loggerA.Path())
}
loggerA.Info("from taskA")
loggerB.Info("from taskB")
loggerA.Flush()
loggerB.Flush()
dataA, err := os.ReadFile(loggerA.Path())
if err != nil {
t.Fatalf("failed to read loggerA file: %v", err)
}
dataB, err := os.ReadFile(loggerB.Path())
if err != nil {
t.Fatalf("failed to read loggerB file: %v", err)
}
if !strings.Contains(string(dataA), "from taskA") {
t.Fatalf("loggerA missing its message, got: %q", string(dataA))
}
if strings.Contains(string(dataA), "from taskB") {
t.Fatalf("loggerA contains loggerB message, got: %q", string(dataA))
}
if !strings.Contains(string(dataB), "from taskB") {
t.Fatalf("loggerB missing its message, got: %q", string(dataB))
}
if strings.Contains(string(dataB), "from taskA") {
t.Fatalf("loggerB contains loggerA message, got: %q", string(dataB))
}
}
func TestLoggerWithSuffixReturnsErrorWhenTempDirMissing(t *testing.T) {
missingTempDir := filepath.Join(t.TempDir(), "does-not-exist")
setTempDirEnv(t, missingTempDir)
logger, err := NewLoggerWithSuffix("task-err")
if err == nil {
_ = logger.Close()
t.Fatalf("expected error, got nil")
}
}

View File

@@ -26,7 +26,7 @@ func compareCleanupStats(got, want CleanupStats) bool {
return true
}
func TestRunLoggerCreatesFileWithPID(t *testing.T) {
func TestLoggerCreatesFileWithPID(t *testing.T) {
tempDir := t.TempDir()
t.Setenv("TMPDIR", tempDir)
@@ -46,7 +46,7 @@ func TestRunLoggerCreatesFileWithPID(t *testing.T) {
}
}
func TestRunLoggerWritesLevels(t *testing.T) {
func TestLoggerWritesLevels(t *testing.T) {
tempDir := t.TempDir()
t.Setenv("TMPDIR", tempDir)
@@ -77,7 +77,31 @@ func TestRunLoggerWritesLevels(t *testing.T) {
}
}
func TestRunLoggerCloseRemovesFileAndStopsWorker(t *testing.T) {
func TestLoggerDefaultIsTerminalCoverage(t *testing.T) {
oldStdin := os.Stdin
t.Cleanup(func() { os.Stdin = oldStdin })
f, err := os.CreateTemp(t.TempDir(), "stdin-*")
if err != nil {
t.Fatalf("os.CreateTemp() error = %v", err)
}
defer os.Remove(f.Name())
os.Stdin = f
if got := defaultIsTerminal(); got {
t.Fatalf("defaultIsTerminal() = %v, want false for regular file", got)
}
if err := f.Close(); err != nil {
t.Fatalf("Close() error = %v", err)
}
os.Stdin = f
if got := defaultIsTerminal(); !got {
t.Fatalf("defaultIsTerminal() = %v, want true when Stat fails", got)
}
}
func TestLoggerCloseStopsWorkerAndKeepsFile(t *testing.T) {
tempDir := t.TempDir()
t.Setenv("TMPDIR", tempDir)
@@ -94,6 +118,11 @@ func TestRunLoggerCloseRemovesFileAndStopsWorker(t *testing.T) {
if err := logger.Close(); err != nil {
t.Fatalf("Close() returned error: %v", err)
}
if logger.file != nil {
if _, err := logger.file.Write([]byte("x")); err == nil {
t.Fatalf("expected file to be closed after Close()")
}
}
// After recent changes, log file is kept for debugging - NOT removed
if _, err := os.Stat(logPath); os.IsNotExist(err) {
@@ -116,7 +145,7 @@ func TestRunLoggerCloseRemovesFileAndStopsWorker(t *testing.T) {
}
}
func TestRunLoggerConcurrentWritesSafe(t *testing.T) {
func TestLoggerConcurrentWritesSafe(t *testing.T) {
tempDir := t.TempDir()
t.Setenv("TMPDIR", tempDir)
@@ -165,7 +194,7 @@ func TestRunLoggerConcurrentWritesSafe(t *testing.T) {
}
}
func TestRunLoggerTerminateProcessActive(t *testing.T) {
func TestLoggerTerminateProcessActive(t *testing.T) {
cmd := exec.Command("sleep", "5")
if err := cmd.Start(); err != nil {
t.Skipf("cannot start sleep command: %v", err)
@@ -193,7 +222,7 @@ func TestRunLoggerTerminateProcessActive(t *testing.T) {
time.Sleep(10 * time.Millisecond)
}
func TestRunTerminateProcessNil(t *testing.T) {
func TestLoggerTerminateProcessNil(t *testing.T) {
if timer := terminateProcess(nil); timer != nil {
t.Fatalf("terminateProcess(nil) should return nil timer")
}
@@ -202,7 +231,7 @@ func TestRunTerminateProcessNil(t *testing.T) {
}
}
func TestRunCleanupOldLogsRemovesOrphans(t *testing.T) {
func TestLoggerCleanupOldLogsRemovesOrphans(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
orphan1 := createTempLog(t, tempDir, "codex-wrapper-111.log")
@@ -252,7 +281,7 @@ func TestRunCleanupOldLogsRemovesOrphans(t *testing.T) {
}
}
func TestRunCleanupOldLogsHandlesInvalidNamesAndErrors(t *testing.T) {
func TestLoggerCleanupOldLogsHandlesInvalidNamesAndErrors(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
invalid := []string{
@@ -310,7 +339,7 @@ func TestRunCleanupOldLogsHandlesInvalidNamesAndErrors(t *testing.T) {
}
}
func TestRunCleanupOldLogsHandlesGlobFailures(t *testing.T) {
func TestLoggerCleanupOldLogsHandlesGlobFailures(t *testing.T) {
stubProcessRunning(t, func(pid int) bool {
t.Fatalf("process check should not run when glob fails")
return false
@@ -336,7 +365,7 @@ func TestRunCleanupOldLogsHandlesGlobFailures(t *testing.T) {
}
}
func TestRunCleanupOldLogsEmptyDirectoryStats(t *testing.T) {
func TestLoggerCleanupOldLogsEmptyDirectoryStats(t *testing.T) {
setTempDirEnv(t, t.TempDir())
stubProcessRunning(t, func(int) bool {
@@ -356,7 +385,7 @@ func TestRunCleanupOldLogsEmptyDirectoryStats(t *testing.T) {
}
}
func TestRunCleanupOldLogsHandlesTempDirPermissionErrors(t *testing.T) {
func TestLoggerCleanupOldLogsHandlesTempDirPermissionErrors(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
paths := []string{
@@ -396,7 +425,7 @@ func TestRunCleanupOldLogsHandlesTempDirPermissionErrors(t *testing.T) {
}
}
func TestRunCleanupOldLogsHandlesPermissionDeniedFile(t *testing.T) {
func TestLoggerCleanupOldLogsHandlesPermissionDeniedFile(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
protected := createTempLog(t, tempDir, "codex-wrapper-6200.log")
@@ -433,7 +462,7 @@ func TestRunCleanupOldLogsHandlesPermissionDeniedFile(t *testing.T) {
}
}
func TestRunCleanupOldLogsPerformanceBound(t *testing.T) {
func TestLoggerCleanupOldLogsPerformanceBound(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
const fileCount = 400
@@ -476,17 +505,98 @@ func TestRunCleanupOldLogsPerformanceBound(t *testing.T) {
}
}
func TestRunCleanupOldLogsCoverageSuite(t *testing.T) {
func TestLoggerCleanupOldLogsCoverageSuite(t *testing.T) {
TestBackendParseJSONStream_CoverageSuite(t)
}
// Reuse the existing coverage suite so the focused TestLogger run still exercises
// the rest of the codebase and keeps coverage high.
func TestRunLoggerCoverageSuite(t *testing.T) {
TestBackendParseJSONStream_CoverageSuite(t)
func TestLoggerCoverageSuite(t *testing.T) {
suite := []struct {
name string
fn func(*testing.T)
}{
{"TestBackendParseJSONStream_CoverageSuite", TestBackendParseJSONStream_CoverageSuite},
{"TestVersionCoverageFullRun", TestVersionCoverageFullRun},
{"TestVersionMainWrapper", TestVersionMainWrapper},
{"TestExecutorHelperCoverage", TestExecutorHelperCoverage},
{"TestExecutorRunCodexTaskWithContext", TestExecutorRunCodexTaskWithContext},
{"TestExecutorParallelLogIsolation", TestExecutorParallelLogIsolation},
{"TestExecutorTaskLoggerContext", TestExecutorTaskLoggerContext},
{"TestExecutorExecuteConcurrentWithContextBranches", TestExecutorExecuteConcurrentWithContextBranches},
{"TestExecutorSignalAndTermination", TestExecutorSignalAndTermination},
{"TestExecutorCancelReasonAndCloseWithReason", TestExecutorCancelReasonAndCloseWithReason},
{"TestExecutorForceKillTimerStop", TestExecutorForceKillTimerStop},
{"TestExecutorForwardSignalsDefaults", TestExecutorForwardSignalsDefaults},
{"TestBackendParseArgs_NewMode", TestBackendParseArgs_NewMode},
{"TestBackendParseArgs_ResumeMode", TestBackendParseArgs_ResumeMode},
{"TestBackendParseArgs_BackendFlag", TestBackendParseArgs_BackendFlag},
{"TestBackendParseArgs_SkipPermissions", TestBackendParseArgs_SkipPermissions},
{"TestBackendParseBoolFlag", TestBackendParseBoolFlag},
{"TestBackendEnvFlagEnabled", TestBackendEnvFlagEnabled},
{"TestRunResolveTimeout", TestRunResolveTimeout},
{"TestRunIsTerminal", TestRunIsTerminal},
{"TestRunReadPipedTask", TestRunReadPipedTask},
{"TestTailBufferWrite", TestTailBufferWrite},
{"TestLogWriterWriteLimitsBuffer", TestLogWriterWriteLimitsBuffer},
{"TestLogWriterLogLine", TestLogWriterLogLine},
{"TestNewLogWriterDefaultMaxLen", TestNewLogWriterDefaultMaxLen},
{"TestNewLogWriterDefaultLimit", TestNewLogWriterDefaultLimit},
{"TestRunHello", TestRunHello},
{"TestRunGreet", TestRunGreet},
{"TestRunFarewell", TestRunFarewell},
{"TestRunFarewellEmpty", TestRunFarewellEmpty},
{"TestParallelParseConfig_Success", TestParallelParseConfig_Success},
{"TestParallelParseConfig_Backend", TestParallelParseConfig_Backend},
{"TestParallelParseConfig_InvalidFormat", TestParallelParseConfig_InvalidFormat},
{"TestParallelParseConfig_EmptyTasks", TestParallelParseConfig_EmptyTasks},
{"TestParallelParseConfig_MissingID", TestParallelParseConfig_MissingID},
{"TestParallelParseConfig_MissingTask", TestParallelParseConfig_MissingTask},
{"TestParallelParseConfig_DuplicateID", TestParallelParseConfig_DuplicateID},
{"TestParallelParseConfig_DelimiterFormat", TestParallelParseConfig_DelimiterFormat},
{"TestBackendSelectBackend", TestBackendSelectBackend},
{"TestBackendSelectBackend_Invalid", TestBackendSelectBackend_Invalid},
{"TestBackendSelectBackend_DefaultOnEmpty", TestBackendSelectBackend_DefaultOnEmpty},
{"TestBackendBuildArgs_CodexBackend", TestBackendBuildArgs_CodexBackend},
{"TestBackendBuildArgs_ClaudeBackend", TestBackendBuildArgs_ClaudeBackend},
{"TestClaudeBackendBuildArgs_OutputValidation", TestClaudeBackendBuildArgs_OutputValidation},
{"TestBackendBuildArgs_GeminiBackend", TestBackendBuildArgs_GeminiBackend},
{"TestGeminiBackendBuildArgs_OutputValidation", TestGeminiBackendBuildArgs_OutputValidation},
{"TestBackendNamesAndCommands", TestBackendNamesAndCommands},
{"TestBackendParseJSONStream", TestBackendParseJSONStream},
{"TestBackendParseJSONStream_ClaudeEvents", TestBackendParseJSONStream_ClaudeEvents},
{"TestBackendParseJSONStream_GeminiEvents", TestBackendParseJSONStream_GeminiEvents},
{"TestBackendParseJSONStreamWithWarn_InvalidLine", TestBackendParseJSONStreamWithWarn_InvalidLine},
{"TestBackendParseJSONStream_OnMessage", TestBackendParseJSONStream_OnMessage},
{"TestBackendParseJSONStream_ScannerError", TestBackendParseJSONStream_ScannerError},
{"TestBackendDiscardInvalidJSON", TestBackendDiscardInvalidJSON},
{"TestBackendDiscardInvalidJSONBuffer", TestBackendDiscardInvalidJSONBuffer},
{"TestCurrentWrapperNameFallsBackToExecutable", TestCurrentWrapperNameFallsBackToExecutable},
{"TestCurrentWrapperNameDetectsLegacyAliasSymlink", TestCurrentWrapperNameDetectsLegacyAliasSymlink},
{"TestIsProcessRunning", TestIsProcessRunning},
{"TestGetProcessStartTimeReadsProcStat", TestGetProcessStartTimeReadsProcStat},
{"TestGetProcessStartTimeInvalidData", TestGetProcessStartTimeInvalidData},
{"TestGetBootTimeParsesBtime", TestGetBootTimeParsesBtime},
{"TestGetBootTimeInvalidData", TestGetBootTimeInvalidData},
{"TestClaudeBuildArgs_ModesAndPermissions", TestClaudeBuildArgs_ModesAndPermissions},
{"TestClaudeBuildArgs_GeminiAndCodexModes", TestClaudeBuildArgs_GeminiAndCodexModes},
{"TestClaudeBuildArgs_BackendMetadata", TestClaudeBuildArgs_BackendMetadata},
}
for _, tc := range suite {
t.Run(tc.name, tc.fn)
}
}
func TestRunCleanupOldLogsKeepsCurrentProcessLog(t *testing.T) {
func TestLoggerCleanupOldLogsKeepsCurrentProcessLog(t *testing.T) {
tempDir := setTempDirEnv(t, t.TempDir())
currentPID := os.Getpid()
@@ -518,7 +628,7 @@ func TestRunCleanupOldLogsKeepsCurrentProcessLog(t *testing.T) {
}
}
func TestIsPIDReusedScenarios(t *testing.T) {
func TestLoggerIsPIDReusedScenarios(t *testing.T) {
now := time.Now()
tests := []struct {
name string
@@ -552,7 +662,7 @@ func TestIsPIDReusedScenarios(t *testing.T) {
}
}
func TestIsUnsafeFileSecurityChecks(t *testing.T) {
func TestLoggerIsUnsafeFileSecurityChecks(t *testing.T) {
tempDir := t.TempDir()
absTempDir, err := filepath.Abs(tempDir)
if err != nil {
@@ -601,7 +711,7 @@ func TestIsUnsafeFileSecurityChecks(t *testing.T) {
})
}
func TestRunLoggerPathAndRemove(t *testing.T) {
func TestLoggerPathAndRemove(t *testing.T) {
tempDir := t.TempDir()
path := filepath.Join(tempDir, "sample.log")
if err := os.WriteFile(path, []byte("test"), 0o644); err != nil {
@@ -628,7 +738,19 @@ func TestRunLoggerPathAndRemove(t *testing.T) {
}
}
func TestRunLoggerInternalLog(t *testing.T) {
func TestLoggerTruncateBytesCoverage(t *testing.T) {
if got := truncateBytes([]byte("abc"), 3); got != "abc" {
t.Fatalf("truncateBytes() = %q, want %q", got, "abc")
}
if got := truncateBytes([]byte("abcd"), 3); got != "abc..." {
t.Fatalf("truncateBytes() = %q, want %q", got, "abc...")
}
if got := truncateBytes([]byte("abcd"), -1); got != "" {
t.Fatalf("truncateBytes() = %q, want empty string", got)
}
}
func TestLoggerInternalLog(t *testing.T) {
logger := &Logger{
ch: make(chan logEntry, 1),
done: make(chan struct{}),
@@ -653,7 +775,7 @@ func TestRunLoggerInternalLog(t *testing.T) {
close(logger.done)
}
func TestRunParsePIDFromLog(t *testing.T) {
func TestLoggerParsePIDFromLog(t *testing.T) {
hugePID := strconv.FormatInt(math.MaxInt64, 10) + "0"
tests := []struct {
name string
@@ -769,7 +891,7 @@ func (f fakeFileInfo) ModTime() time.Time { return f.modTime }
func (f fakeFileInfo) IsDir() bool { return false }
func (f fakeFileInfo) Sys() interface{} { return nil }
func TestExtractRecentErrors(t *testing.T) {
func TestLoggerExtractRecentErrors(t *testing.T) {
tests := []struct {
name string
content string
@@ -846,21 +968,21 @@ func TestExtractRecentErrors(t *testing.T) {
}
}
func TestExtractRecentErrorsNilLogger(t *testing.T) {
func TestLoggerExtractRecentErrorsNilLogger(t *testing.T) {
var logger *Logger
if got := logger.ExtractRecentErrors(10); got != nil {
t.Fatalf("nil logger ExtractRecentErrors() should return nil, got %v", got)
}
}
func TestExtractRecentErrorsEmptyPath(t *testing.T) {
func TestLoggerExtractRecentErrorsEmptyPath(t *testing.T) {
logger := &Logger{path: ""}
if got := logger.ExtractRecentErrors(10); got != nil {
t.Fatalf("empty path ExtractRecentErrors() should return nil, got %v", got)
}
}
func TestExtractRecentErrorsFileNotExist(t *testing.T) {
func TestLoggerExtractRecentErrorsFileNotExist(t *testing.T) {
logger := &Logger{path: "/nonexistent/path/to/log.log"}
if got := logger.ExtractRecentErrors(10); got != nil {
t.Fatalf("nonexistent file ExtractRecentErrors() should return nil, got %v", got)

View File

@@ -14,7 +14,7 @@ import (
)
const (
version = "5.2.0"
version = "5.2.4"
defaultWorkdir = "."
defaultTimeout = 7200 // seconds
codexLogLineLimit = 1000

View File

@@ -426,10 +426,11 @@ ok-d`
t.Fatalf("expected startup banner in stderr, got:\n%s", stderrOut)
}
// After parallel log isolation fix, each task has its own log file
expectedLines := map[string]struct{}{
fmt.Sprintf("Task a: Log: %s", expectedLog): {},
fmt.Sprintf("Task b: Log: %s", expectedLog): {},
fmt.Sprintf("Task d: Log: %s", expectedLog): {},
fmt.Sprintf("Task a: Log: %s", filepath.Join(tempDir, fmt.Sprintf("codex-wrapper-%d-a.log", os.Getpid()))): {},
fmt.Sprintf("Task b: Log: %s", filepath.Join(tempDir, fmt.Sprintf("codex-wrapper-%d-b.log", os.Getpid()))): {},
fmt.Sprintf("Task d: Log: %s", filepath.Join(tempDir, fmt.Sprintf("codex-wrapper-%d-d.log", os.Getpid()))): {},
}
if len(taskLines) != len(expectedLines) {

View File

@@ -41,6 +41,7 @@ func resetTestHooks() {
closeLogger()
executablePathFn = os.Executable
runTaskFn = runCodexTask
runCodexTaskFn = defaultRunCodexTaskFn
exitFn = os.Exit
}
@@ -1377,7 +1378,7 @@ func TestBackendBuildArgs_ClaudeBackend(t *testing.T) {
backend := ClaudeBackend{}
cfg := &Config{Mode: "new", WorkDir: defaultWorkdir}
got := backend.BuildArgs(cfg, "todo")
want := []string{"-p", "--output-format", "stream-json", "--verbose", "todo"}
want := []string{"-p", "--dangerously-skip-permissions", "--output-format", "stream-json", "--verbose", "todo"}
if len(got) != len(want) {
t.Fatalf("length mismatch")
}
@@ -1398,7 +1399,7 @@ func TestClaudeBackendBuildArgs_OutputValidation(t *testing.T) {
target := "ensure-flags"
args := backend.BuildArgs(cfg, target)
expectedPrefix := []string{"-p", "--output-format", "stream-json", "--verbose"}
expectedPrefix := []string{"-p", "--dangerously-skip-permissions", "--output-format", "stream-json", "--verbose"}
if len(args) != len(expectedPrefix)+1 {
t.Fatalf("args length=%d, want %d", len(args), len(expectedPrefix)+1)
@@ -2690,7 +2691,7 @@ func TestVersionFlag(t *testing.T) {
t.Errorf("exit = %d, want 0", code)
}
})
want := "codeagent-wrapper version 5.2.0\n"
want := "codeagent-wrapper version 5.2.4\n"
if output != want {
t.Fatalf("output = %q, want %q", output, want)
}
@@ -2704,7 +2705,7 @@ func TestVersionShortFlag(t *testing.T) {
t.Errorf("exit = %d, want 0", code)
}
})
want := "codeagent-wrapper version 5.2.0\n"
want := "codeagent-wrapper version 5.2.4\n"
if output != want {
t.Fatalf("output = %q, want %q", output, want)
}
@@ -2718,7 +2719,7 @@ func TestVersionLegacyAlias(t *testing.T) {
t.Errorf("exit = %d, want 0", code)
}
})
want := "codex-wrapper version 5.2.0\n"
want := "codex-wrapper version 5.2.4\n"
if output != want {
t.Fatalf("output = %q, want %q", output, want)
}

View File

@@ -53,9 +53,22 @@ func parseJSONStreamWithLog(r io.Reader, warnFn func(string), infoFn func(string
return parseJSONStreamInternal(r, warnFn, infoFn, nil)
}
const (
jsonLineReaderSize = 64 * 1024
jsonLineMaxBytes = 10 * 1024 * 1024
jsonLinePreviewBytes = 256
)
type codexHeader struct {
Type string `json:"type"`
ThreadID string `json:"thread_id,omitempty"`
Item *struct {
Type string `json:"type"`
} `json:"item,omitempty"`
}
func parseJSONStreamInternal(r io.Reader, warnFn func(string), infoFn func(string), onMessage func()) (message, threadID string) {
scanner := bufio.NewScanner(r)
scanner.Buffer(make([]byte, 64*1024), 10*1024*1024)
reader := bufio.NewReaderSize(r, jsonLineReaderSize)
if warnFn == nil {
warnFn = func(string) {}
@@ -78,79 +91,89 @@ func parseJSONStreamInternal(r io.Reader, warnFn func(string), infoFn func(strin
geminiBuffer strings.Builder
)
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if line == "" {
for {
line, tooLong, err := readLineWithLimit(reader, jsonLineMaxBytes, jsonLinePreviewBytes)
if err != nil {
if errors.Is(err, io.EOF) {
break
}
warnFn("Read stdout error: " + err.Error())
break
}
line = bytes.TrimSpace(line)
if len(line) == 0 {
continue
}
totalEvents++
var raw map[string]json.RawMessage
if err := json.Unmarshal([]byte(line), &raw); err != nil {
warnFn(fmt.Sprintf("Failed to parse line: %s", truncate(line, 100)))
if tooLong {
warnFn(fmt.Sprintf("Skipped overlong JSON line (> %d bytes): %s", jsonLineMaxBytes, truncateBytes(line, 100)))
continue
}
hasItemType := false
if rawItem, ok := raw["item"]; ok {
var itemMap map[string]json.RawMessage
if err := json.Unmarshal(rawItem, &itemMap); err == nil {
if _, ok := itemMap["type"]; ok {
hasItemType = true
var codex codexHeader
if err := json.Unmarshal(line, &codex); err == nil {
isCodex := codex.ThreadID != "" || (codex.Item != nil && codex.Item.Type != "")
if isCodex {
var details []string
if codex.ThreadID != "" {
details = append(details, fmt.Sprintf("thread_id=%s", codex.ThreadID))
}
if codex.Item != nil && codex.Item.Type != "" {
details = append(details, fmt.Sprintf("item_type=%s", codex.Item.Type))
}
if len(details) > 0 {
infoFn(fmt.Sprintf("Parsed event #%d type=%s (%s)", totalEvents, codex.Type, strings.Join(details, ", ")))
} else {
infoFn(fmt.Sprintf("Parsed event #%d type=%s", totalEvents, codex.Type))
}
switch codex.Type {
case "thread.started":
threadID = codex.ThreadID
infoFn(fmt.Sprintf("thread.started event thread_id=%s", threadID))
case "item.completed":
itemType := ""
if codex.Item != nil {
itemType = codex.Item.Type
}
if itemType == "agent_message" {
var event JSONEvent
if err := json.Unmarshal(line, &event); err != nil {
warnFn(fmt.Sprintf("Failed to parse Codex event: %s", truncateBytes(line, 100)))
continue
}
normalized := ""
if event.Item != nil {
normalized = normalizeText(event.Item.Text)
}
infoFn(fmt.Sprintf("item.completed event item_type=%s message_len=%d", itemType, len(normalized)))
if normalized != "" {
codexMessage = normalized
notifyMessage()
}
} else {
infoFn(fmt.Sprintf("item.completed event item_type=%s", itemType))
}
}
continue
}
}
isCodex := hasItemType
if !isCodex {
if _, ok := raw["thread_id"]; ok {
isCodex = true
}
var raw map[string]json.RawMessage
if err := json.Unmarshal(line, &raw); err != nil {
warnFn(fmt.Sprintf("Failed to parse line: %s", truncateBytes(line, 100)))
continue
}
switch {
case isCodex:
var event JSONEvent
if err := json.Unmarshal([]byte(line), &event); err != nil {
warnFn(fmt.Sprintf("Failed to parse Codex event: %s", truncate(line, 100)))
continue
}
var details []string
if event.ThreadID != "" {
details = append(details, fmt.Sprintf("thread_id=%s", event.ThreadID))
}
if event.Item != nil && event.Item.Type != "" {
details = append(details, fmt.Sprintf("item_type=%s", event.Item.Type))
}
if len(details) > 0 {
infoFn(fmt.Sprintf("Parsed event #%d type=%s (%s)", totalEvents, event.Type, strings.Join(details, ", ")))
} else {
infoFn(fmt.Sprintf("Parsed event #%d type=%s", totalEvents, event.Type))
}
switch event.Type {
case "thread.started":
threadID = event.ThreadID
infoFn(fmt.Sprintf("thread.started event thread_id=%s", threadID))
case "item.completed":
var itemType string
var normalized string
if event.Item != nil {
itemType = event.Item.Type
normalized = normalizeText(event.Item.Text)
}
infoFn(fmt.Sprintf("item.completed event item_type=%s message_len=%d", itemType, len(normalized)))
if event.Item != nil && event.Item.Type == "agent_message" && normalized != "" {
codexMessage = normalized
notifyMessage()
}
}
case hasKey(raw, "subtype") || hasKey(raw, "result"):
var event ClaudeEvent
if err := json.Unmarshal([]byte(line), &event); err != nil {
warnFn(fmt.Sprintf("Failed to parse Claude event: %s", truncate(line, 100)))
if err := json.Unmarshal(line, &event); err != nil {
warnFn(fmt.Sprintf("Failed to parse Claude event: %s", truncateBytes(line, 100)))
continue
}
@@ -167,8 +190,8 @@ func parseJSONStreamInternal(r io.Reader, warnFn func(string), infoFn func(strin
case hasKey(raw, "role") || hasKey(raw, "delta"):
var event GeminiEvent
if err := json.Unmarshal([]byte(line), &event); err != nil {
warnFn(fmt.Sprintf("Failed to parse Gemini event: %s", truncate(line, 100)))
if err := json.Unmarshal(line, &event); err != nil {
warnFn(fmt.Sprintf("Failed to parse Gemini event: %s", truncateBytes(line, 100)))
continue
}
@@ -184,14 +207,10 @@ func parseJSONStreamInternal(r io.Reader, warnFn func(string), infoFn func(strin
infoFn(fmt.Sprintf("Parsed Gemini event #%d type=%s role=%s delta=%t status=%s content_len=%d", totalEvents, event.Type, event.Role, event.Delta, event.Status, len(event.Content)))
default:
warnFn(fmt.Sprintf("Unknown event format: %s", truncate(line, 100)))
warnFn(fmt.Sprintf("Unknown event format: %s", truncateBytes(line, 100)))
}
}
if err := scanner.Err(); err != nil && !errors.Is(err, io.EOF) {
warnFn("Read stdout error: " + err.Error())
}
switch {
case geminiBuffer.Len() > 0:
message = geminiBuffer.String()
@@ -236,6 +255,79 @@ func discardInvalidJSON(decoder *json.Decoder, reader *bufio.Reader) (*bufio.Rea
return bufio.NewReader(io.MultiReader(bytes.NewReader(remaining), reader)), err
}
func readLineWithLimit(r *bufio.Reader, maxBytes int, previewBytes int) (line []byte, tooLong bool, err error) {
if r == nil {
return nil, false, errors.New("reader is nil")
}
if maxBytes <= 0 {
return nil, false, errors.New("maxBytes must be > 0")
}
if previewBytes < 0 {
previewBytes = 0
}
part, isPrefix, err := r.ReadLine()
if err != nil {
return nil, false, err
}
if !isPrefix {
if len(part) > maxBytes {
return part[:min(len(part), previewBytes)], true, nil
}
return part, false, nil
}
preview := make([]byte, 0, min(previewBytes, len(part)))
if previewBytes > 0 {
preview = append(preview, part[:min(previewBytes, len(part))]...)
}
buf := make([]byte, 0, min(maxBytes, len(part)*2))
total := 0
if len(part) > maxBytes {
tooLong = true
} else {
buf = append(buf, part...)
total = len(part)
}
for isPrefix {
part, isPrefix, err = r.ReadLine()
if err != nil {
return nil, tooLong, err
}
if previewBytes > 0 && len(preview) < previewBytes {
preview = append(preview, part[:min(previewBytes-len(preview), len(part))]...)
}
if !tooLong {
if total+len(part) > maxBytes {
tooLong = true
continue
}
buf = append(buf, part...)
total += len(part)
}
}
if tooLong {
return preview, true, nil
}
return buf, false, nil
}
func truncateBytes(b []byte, maxLen int) string {
if len(b) <= maxLen {
return string(b)
}
if maxLen < 0 {
return ""
}
return string(b[:maxLen]) + "..."
}
func normalizeText(text interface{}) string {
switch v := text.(type) {
case string:

View File

@@ -0,0 +1,31 @@
package main
import (
"strings"
"testing"
)
func TestParseJSONStream_SkipsOverlongLineAndContinues(t *testing.T) {
// Exceed the 10MB bufio.Scanner limit in parseJSONStreamInternal.
tooLong := strings.Repeat("a", 11*1024*1024)
input := strings.Join([]string{
`{"type":"item.completed","item":{"type":"other_type","text":"` + tooLong + `"}}`,
`{"type":"thread.started","thread_id":"t-1"}`,
`{"type":"item.completed","item":{"type":"agent_message","text":"ok"}}`,
}, "\n")
var warns []string
warnFn := func(msg string) { warns = append(warns, msg) }
gotMessage, gotThreadID := parseJSONStreamInternal(strings.NewReader(input), warnFn, nil, nil)
if gotMessage != "ok" {
t.Fatalf("message=%q, want %q (warns=%v)", gotMessage, "ok", warns)
}
if gotThreadID != "t-1" {
t.Fatalf("threadID=%q, want %q (warns=%v)", gotThreadID, "t-1", warns)
}
if len(warns) == 0 || !strings.Contains(warns[0], "Skipped overlong JSON line") {
t.Fatalf("expected warning about overlong JSON line, got %v", warns)
}
}

View File

@@ -78,6 +78,7 @@ type logWriter struct {
prefix string
maxLen int
buf bytes.Buffer
dropped bool
}
func newLogWriter(prefix string, maxLen int) *logWriter {
@@ -94,12 +95,12 @@ func (lw *logWriter) Write(p []byte) (int, error) {
total := len(p)
for len(p) > 0 {
if idx := bytes.IndexByte(p, '\n'); idx >= 0 {
lw.buf.Write(p[:idx])
lw.writeLimited(p[:idx])
lw.logLine(true)
p = p[idx+1:]
continue
}
lw.buf.Write(p)
lw.writeLimited(p)
break
}
return total, nil
@@ -117,21 +118,53 @@ func (lw *logWriter) logLine(force bool) {
return
}
line := lw.buf.String()
dropped := lw.dropped
lw.dropped = false
lw.buf.Reset()
if line == "" && !force {
return
}
if lw.maxLen > 0 && len(line) > lw.maxLen {
cutoff := lw.maxLen
if cutoff > 3 {
line = line[:cutoff-3] + "..."
} else {
line = line[:cutoff]
if lw.maxLen > 0 {
if dropped {
if lw.maxLen > 3 {
line = line[:min(len(line), lw.maxLen-3)] + "..."
} else {
line = line[:min(len(line), lw.maxLen)]
}
} else if len(line) > lw.maxLen {
cutoff := lw.maxLen
if cutoff > 3 {
line = line[:cutoff-3] + "..."
} else {
line = line[:cutoff]
}
}
}
logInfo(lw.prefix + line)
}
func (lw *logWriter) writeLimited(p []byte) {
if lw == nil || len(p) == 0 {
return
}
if lw.maxLen <= 0 {
lw.buf.Write(p)
return
}
remaining := lw.maxLen - lw.buf.Len()
if remaining <= 0 {
lw.dropped = true
return
}
if len(p) <= remaining {
lw.buf.Write(p)
return
}
lw.buf.Write(p[:remaining])
lw.dropped = true
}
type tailBuffer struct {
limit int
data []byte

View File

@@ -20,9 +20,33 @@
},
{
"type": "copy_file",
"source": "skills/codex/SKILL.md",
"target": "skills/codex/SKILL.md",
"description": "Install codex skill"
"source": "skills/codeagent/SKILL.md",
"target": "skills/codeagent/SKILL.md",
"description": "Install codeagent skill"
},
{
"type": "copy_file",
"source": "skills/product-requirements/SKILL.md",
"target": "skills/product-requirements/SKILL.md",
"description": "Install product-requirements skill"
},
{
"type": "copy_file",
"source": "skills/prototype-prompt-generator/SKILL.md",
"target": "skills/prototype-prompt-generator/SKILL.md",
"description": "Install prototype-prompt-generator skill"
},
{
"type": "copy_file",
"source": "skills/prototype-prompt-generator/references/prompt-structure.md",
"target": "skills/prototype-prompt-generator/references/prompt-structure.md",
"description": "Install prototype-prompt-generator prompt structure reference"
},
{
"type": "copy_file",
"source": "skills/prototype-prompt-generator/references/design-systems.md",
"target": "skills/prototype-prompt-generator/references/design-systems.md",
"description": "Install prototype-prompt-generator design systems reference"
},
{
"type": "run_command",