diff --git a/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go b/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go index 8b580885bf..deb74efafe 100644 --- a/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go +++ b/apps/backend/internal/agent/runtime/agentctl/launcher/launcher.go @@ -15,6 +15,7 @@ import ( "os/exec" "path/filepath" "runtime" + "strings" "sync" "time" @@ -456,16 +457,72 @@ func (l *Launcher) waitForHealthy(ctx context.Context) error { return fmt.Errorf("timeout waiting for agentctl to become healthy") } -// pipeOutput reads from a scanner and logs each line. -// stderr is logged at WARN level for visibility; stdout remains at DEBUG. +// pipeOutput reads from a scanner and logs each line. stdout is diagnostic +// noise (ACP travels over the socket, not stdout) so it stays at DEBUG. +// stderr carries the child's own structured logs; rather than flatten every +// line to WARN — which turns routine child INFO/DEBUG into shutdown noise in +// the parent log — it is forwarded at the level the child already tagged it +// with. Lines without a recognizable level fall back to WARN so unstructured +// panics/tracebacks stay visible. func (l *Launcher) pipeOutput(name string, scanner *bufio.Scanner) { for scanner.Scan() { line := scanner.Text() - if name == "stderr" { - l.logger.Warn(line, zap.String("stream", name)) - } else { + if name != "stderr" { l.logger.Debug(line, zap.String("stream", name)) + continue } + switch childLogLevel(line) { + case "DEBUG": + l.logger.Debug(line, zap.String("stream", name)) + case "INFO": + l.logger.Info(line, zap.String("stream", name)) + case "ERROR", "FATAL", "PANIC", "DPANIC": + l.logger.Error(line, zap.String("stream", name)) + default: + l.logger.Warn(line, zap.String("stream", name)) + } + } +} + +// childLogLevel extracts the level token from a line emitted by the agentctl +// child's console-format zap logger: "\t\t\t", where +// LEVEL is capitalized and may be wrapped in ANSI color codes. It returns the +// uppercased level ("INFO"/"WARN"/…) or "" when the line does not match that +// shape. +func childLogLevel(line string) string { + // A well-formed console record is "\t\t\t", so the + // level token must be bounded by at least a following caller field. Requiring + // three segments rejects truncated lines (e.g. "\t") whose second + // field is not actually a level, so they fall back to WARN. + fields := strings.SplitN(line, "\t", 4) + if len(fields) < 3 { + return "" + } + level := strings.ToUpper(stripANSI(fields[1])) + switch level { + case "DEBUG", "INFO", "WARN", "ERROR", "FATAL", "PANIC", "DPANIC": + return level + default: + return "" + } +} + +// stripANSI removes ANSI SGR escape sequences (e.g. the color codes the +// console encoder wraps the level token in) so the bare token can be matched. +func stripANSI(s string) string { + for { + start := strings.IndexByte(s, 0x1b) + if start < 0 { + return s + } + end := strings.IndexByte(s[start:], 'm') + if end < 0 { + // Unterminated escape: leave the raw ESC byte in place rather than + // dropping the tail, so a truncated token (e.g. "INFO\x1b[34") does + // not collapse into a valid level and instead falls back to WARN. + return s + } + s = s[:start] + s[start+end+1:] } } diff --git a/apps/backend/internal/agent/runtime/agentctl/launcher/launcher_loglevel_test.go b/apps/backend/internal/agent/runtime/agentctl/launcher/launcher_loglevel_test.go new file mode 100644 index 0000000000..abec2d454e --- /dev/null +++ b/apps/backend/internal/agent/runtime/agentctl/launcher/launcher_loglevel_test.go @@ -0,0 +1,99 @@ +package launcher + +import "testing" + +func TestChildLogLevel(t *testing.T) { + tests := []struct { + name string + line string + want string + }{ + { + name: "plain info line", + line: "2026-08-12T10:00:00.000Z\tINFO\tmain.go:97\tstarting agentctl", + want: "INFO", + }, + { + name: "plain warn line", + line: "2026-08-12T10:00:00.000Z\tWARN\tserver.go:12\tslow request", + want: "WARN", + }, + { + name: "plain error line", + line: "2026-08-12T10:00:00.000Z\tERROR\tmain.go:213\tHTTP server failed to bind", + want: "ERROR", + }, + { + name: "plain debug line", + line: "2026-08-12T10:00:00.000Z\tDEBUG\tx.go:1\tnoise", + want: "DEBUG", + }, + { + name: "ansi-wrapped level token", + line: "2026-08-12T10:00:00.000Z\t\x1b[34mINFO\x1b[0m\tmain.go:97\tstarting agentctl", + want: "INFO", + }, + { + name: "lowercase level normalizes to upper", + line: "2026-08-12T10:00:00.000Z\tinfo\tmain.go:97\tstarting agentctl", + want: "INFO", + }, + { + name: "unstructured line has no level", + line: "panic: runtime error: invalid memory address", + want: "", + }, + { + name: "single field line has no level", + line: "just-one-field", + want: "", + }, + { + name: "unknown token is not a level", + line: "2026-08-12T10:00:00.000Z\tTRACE\tx.go:1\tsomething", + want: "", + }, + { + name: "two-field record is too short to be a level", + line: "2026-08-12T10:00:00.000Z\tINFO", + want: "", + }, + { + name: "three-field record with empty message still has level", + line: "2026-08-12T10:00:00.000Z\tINFO\tmain.go:97", + want: "INFO", + }, + { + name: "unterminated ansi escape falls back to no level", + line: "2026-08-12T10:00:00.000Z\tINFO\x1b[34\tmain.go:97\tstarting agentctl", + want: "", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := childLogLevel(tt.line); got != tt.want { + t.Fatalf("childLogLevel(%q) = %q, want %q", tt.line, got, tt.want) + } + }) + } +} + +func TestStripANSI(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {name: "no escapes", in: "INFO", want: "INFO"}, + {name: "color wrapped", in: "\x1b[34mINFO\x1b[0m", want: "INFO"}, + {name: "multiple sequences", in: "\x1b[1m\x1b[31mERROR\x1b[0m", want: "ERROR"}, + {name: "unterminated escape keeps raw byte", in: "INFO\x1b[34", want: "INFO\x1b[34"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := stripANSI(tt.in); got != tt.want { + t.Fatalf("stripANSI(%q) = %q, want %q", tt.in, got, tt.want) + } + }) + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go index 0e47cf937e..436f28005b 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go @@ -541,11 +541,21 @@ func (m *Manager) handleStreamDisconnect( err error, promptGeneration uint64, ) { - m.logger.Warn("agent updates stream disconnected", + disconnectFields := []zap.Field{ zap.String("execution_id", execution.ID), zap.String("session_id", execution.SessionID), zap.Uint64("prompt_generation", promptGeneration), - zap.Error(err)) + zap.Error(err), + } + // A disconnect during graceful shutdown is expected: StopAllAgents stops the + // agent process, which drops the agentctl WebSocket. Logging it at WARN turns + // routine teardown into noise, so downgrade to DEBUG while shutting down. The + // failed-status/error-event handling below is unchanged either way. + if m.IsShuttingDown() { + m.logger.Debug("agent updates stream disconnected during shutdown", disconnectFields...) + } else { + m.logger.Warn("agent updates stream disconnected", disconnectFields...) + } if promptGeneration != 0 { execution.promptLifecycleMu.Lock() diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_stream_disconnect_log_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_stream_disconnect_log_test.go new file mode 100644 index 0000000000..e6d516bf64 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_stream_disconnect_log_test.go @@ -0,0 +1,94 @@ +package lifecycle + +import ( + "errors" + "testing" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/kandev/kandev/internal/common/logger" +) + +// disconnectTestManager builds a Manager wired with just enough to run +// handleStreamDisconnect for the promptGeneration==0 path: an execution store, +// an event publisher over a tracking bus, and a logger backed by an observer +// core capturing DEBUG and above so the test can read back the exact level. +func disconnectTestManager(t *testing.T) (*Manager, *observer.ObservedLogs) { + t.Helper() + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + mgr := &Manager{ + logger: log, + executionStore: NewExecutionStore(), + eventPublisher: NewEventPublisher(&MockEventBusWithTracking{}, log), + } + return mgr, logs +} + +// disconnectLogEntry returns the single entry whose message names the stream +// disconnect, ignoring the incidental "updated execution status" debug lines +// that UpdateStatus emits on the same logger. +func disconnectLogEntry(t *testing.T, logs *observer.ObservedLogs) observer.LoggedEntry { + t.Helper() + var found []observer.LoggedEntry + for _, e := range logs.All() { + if e.Message == "agent updates stream disconnected" || + e.Message == "agent updates stream disconnected during shutdown" { + found = append(found, e) + } + } + if len(found) != 1 { + t.Fatalf("expected 1 disconnect log entry, got %d (all: %v)", len(found), logs.All()) + } + return found[0] +} + +func TestHandleStreamDisconnectLogLevel(t *testing.T) { + tests := []struct { + name string + shuttingDown bool + wantLevel zapcore.Level + wantMsg string + }{ + { + name: "active manager logs warn", + shuttingDown: false, + wantLevel: zapcore.WarnLevel, + wantMsg: "agent updates stream disconnected", + }, + { + name: "shutting down logs debug", + shuttingDown: true, + wantLevel: zapcore.DebugLevel, + wantMsg: "agent updates stream disconnected during shutdown", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + mgr, logs := disconnectTestManager(t) + if tt.shuttingDown { + mgr.shuttingDown.Store(true) + } + execution := &AgentExecution{ID: "exec-1", SessionID: "session-1"} + if err := mgr.executionStore.Add(execution); err != nil { + t.Fatalf("add execution: %v", err) + } + + // promptGeneration==0 exercises the simple disconnect path. + mgr.handleStreamDisconnect(execution, errors.New("use of closed network connection"), 0) + + entry := disconnectLogEntry(t, logs) + if entry.Level != tt.wantLevel { + t.Fatalf("level = %v, want %v", entry.Level, tt.wantLevel) + } + if entry.Message != tt.wantMsg { + t.Fatalf("message = %q, want %q", entry.Message, tt.wantMsg) + } + }) + } +} diff --git a/apps/backend/internal/github/poller.go b/apps/backend/internal/github/poller.go index 4dc0136592..be65b82128 100644 --- a/apps/backend/internal/github/poller.go +++ b/apps/backend/internal/github/poller.go @@ -2,6 +2,7 @@ package github import ( "context" + "errors" "sync" "time" @@ -521,8 +522,8 @@ func (p *Poller) checkReviewWatches(ctx context.Context) { } // Clean up tasks for merged/closed PRs that the user hasn't opened. if cleaned, err := p.service.CleanupMergedReviewTasks(ctx, watch); err != nil { - p.logger.Warn("failed to cleanup merged review tasks", - zap.String("watch_id", watch.ID), zap.Error(err)) + p.logCleanupError("failed to cleanup merged review tasks", err, + zap.String("watch_id", watch.ID)) } else if cleaned > 0 { p.logger.Info("cleaned up merged review tasks", zap.String("watch_id", watch.ID), zap.Int("deleted", cleaned)) @@ -533,12 +534,25 @@ func (p *Poller) checkReviewWatches(ctx context.Context) { // would never be re-examined, since the per-watch loop only iterates // enabled watches. if cleaned, err := p.service.CleanupAllOrphanedReviewTasks(ctx); err != nil { - p.logger.Warn("failed to sweep orphaned review tasks", zap.Error(err)) + p.logCleanupError("failed to sweep orphaned review tasks", err) } else if cleaned > 0 { p.logger.Info("swept orphaned review tasks", zap.Int("deleted", cleaned)) } } +// logCleanupError logs a poller cleanup failure at WARN, except when the error +// is a context cancellation. That happens when the poll cycle is interrupted by +// backend shutdown, which is expected teardown rather than a fault, so it is +// downgraded to DEBUG to keep shutdown logs quiet. +func (p *Poller) logCleanupError(msg string, err error, fields ...zap.Field) { + fields = append(fields, zap.Error(err)) + if errors.Is(err, context.Canceled) { + p.logger.Debug(msg+" (context canceled during shutdown)", fields...) + return + } + p.logger.Warn(msg, fields...) +} + // issueWatchLoop polls issue watches for new GitHub issues. func (p *Poller) issueWatchLoop(ctx context.Context) { defer p.wg.Done() diff --git a/apps/backend/internal/github/poller_cleanup_log_test.go b/apps/backend/internal/github/poller_cleanup_log_test.go new file mode 100644 index 0000000000..024c17cd8b --- /dev/null +++ b/apps/backend/internal/github/poller_cleanup_log_test.go @@ -0,0 +1,95 @@ +package github + +import ( + "context" + "errors" + "fmt" + "testing" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/kandev/kandev/internal/common/logger" +) + +func TestPollerLogCleanupError(t *testing.T) { + tests := []struct { + name string + err error + wantLevel zapcore.Level + wantMsg string + }{ + { + name: "generic error logs warn", + err: errors.New("db is down"), + wantLevel: zapcore.WarnLevel, + wantMsg: "failed to sweep orphaned review tasks", + }, + { + name: "context canceled logs debug", + err: context.Canceled, + wantLevel: zapcore.DebugLevel, + wantMsg: "failed to sweep orphaned review tasks (context canceled during shutdown)", + }, + { + name: "wrapped context canceled logs debug", + err: fmt.Errorf("sweep: %w", context.Canceled), + wantLevel: zapcore.DebugLevel, + wantMsg: "failed to sweep orphaned review tasks (context canceled during shutdown)", + }, + { + name: "deadline exceeded is not treated as shutdown", + err: context.DeadlineExceeded, + wantLevel: zapcore.WarnLevel, + wantMsg: "failed to sweep orphaned review tasks", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + p := &Poller{logger: log} + + p.logCleanupError("failed to sweep orphaned review tasks", tt.err) + + entries := logs.All() + if len(entries) != 1 { + t.Fatalf("expected 1 log entry, got %d", len(entries)) + } + if entries[0].Level != tt.wantLevel { + t.Fatalf("level = %v, want %v", entries[0].Level, tt.wantLevel) + } + if entries[0].Message != tt.wantMsg { + t.Fatalf("message = %q, want %q", entries[0].Message, tt.wantMsg) + } + }) + } +} + +func TestPollerLogCleanupErrorForwardsFields(t *testing.T) { + core, logs := observer.New(zapcore.DebugLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + p := &Poller{logger: log} + + p.logCleanupError("failed to cleanup merged review tasks", errors.New("boom"), + zap.String("watch_id", "rw-1")) + + entries := logs.All() + if len(entries) != 1 { + t.Fatalf("expected 1 log entry, got %d", len(entries)) + } + fields := entries[0].ContextMap() + if fields["watch_id"] != "rw-1" { + t.Fatalf("watch_id field = %v, want rw-1", fields["watch_id"]) + } + if fields["error"] != "boom" { + t.Fatalf("error field = %v, want boom", fields["error"]) + } +}