fix(core): log compression turn completion before queued turns

core/compress_completion_log_test.go

@@ -0,0 +1,147 @@
+package core
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+type compressLogBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *compressLogBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *compressLogBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
+func captureCompressLogs(t *testing.T) *compressLogBuffer {
+	t.Helper()
+	var buf compressLogBuffer
+	previous := slog.Default()
+	slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, nil)))
+	t.Cleanup(func() { slog.SetDefault(previous) })
+	return &buf
+}
+
+func compressLogRecords(t *testing.T, buf *compressLogBuffer) []map[string]any {
+	t.Helper()
+	var records []map[string]any
+	for _, line := range strings.Split(strings.TrimSpace(buf.String()), "\n") {
+		if line == "" {
+			continue
+		}
+		var record map[string]any
+		if err := json.Unmarshal([]byte(line), &record); err != nil {
+			t.Fatal(err)
+		}
+		records = append(records, record)
+	}
+	return records
+}
+
+func TestProcessCompressEvents_LogsCompletionOnEveryExit(t *testing.T) {
+	for _, auto := range []bool{false, true} {
+		for _, outcome := range []string{"result", "error", "eof", "idle_timeout", "stopped", "shutdown"} {
+			t.Run(outcome+map[bool]string{false: "/manual", true: "/auto"}[auto], func(t *testing.T) {
+				logs := captureCompressLogs(t)
+				e := newTestEngine()
+				t.Cleanup(e.cancel)
+				p := &stubPlatformEngine{n: "test"}
+				agentSession := newQueuingSession("compress")
+				state := &interactiveState{agentSession: agentSession, platform: p}
+				session := e.sessions.GetOrCreateActive("test:user")
+				session.SetAgentSessionID("agent-compress", "stub")
+				lockGen, _ := session.TryLock()
+				defer session.Unlock(lockGen)
+				switch outcome {
+				case "result":
+					agentSession.events <- Event{Type: EventResult, Done: true}
+				case "error":
+					agentSession.events <- Event{Type: EventError, Error: errors.New("failed")}
+				case "eof":
+					close(agentSession.events)
+				case "idle_timeout":
+					e.eventIdleTimeout = time.Millisecond
+				case "stopped":
+					state.markStopped()
+				case "shutdown":
+					e.cancel()
+				}
+				unlocked := false
+				e.processCompressEvents(state, session, e.sessions, "test:user", p, nil, &unlocked, auto, lockGen)
+				var completions []map[string]any
+				for _, record := range compressLogRecords(t, logs) {
+					if record["msg"] == "turn complete" && record["compress"] == true {
+						completions = append(completions, record)
+					}
+				}
+				if len(completions) != 1 {
+					t.Fatalf("got %d compression completion logs, want 1: %s", len(completions), logs)
+				}
+				r := completions[0]
+				if r["outcome"] != outcome || r["auto"] != auto || r["session"] != session.ID || r["agent_session"] != "agent-compress" || r["level"] != "INFO" {
+					t.Fatalf("incorrect completion fields: %#v", r)
+				}
+				if duration, ok := r["turn_duration"].(float64); !ok || duration < 0 {
+					t.Fatalf("missing/negative duration: %#v", r)
+				}
+			})
+		}
+	}
+}
+
+type compressLogQueueSession struct{ *queuingAgentSession }
+
+func (s *compressLogQueueSession) Send(_ string, _ string, _ []ImageAttachment, _ []FileAttachment) error {
+	slog.Info("test queued turn sent")
+	s.events <- Event{Type: EventResult, Content: "queued reply", Done: true}
+	return nil
+}
+
+func TestProcessCompressEvents_LogsCompletionBeforeQueuedTurn(t *testing.T) {
+	for _, eventType := range []EventType{EventResult, EventError} {
+		t.Run(string(eventType), func(t *testing.T) {
+			logs := captureCompressLogs(t)
+			e := newTestEngine()
+			t.Cleanup(e.cancel)
+			p := &stubPlatformEngine{n: "test"}
+			agentSession := &compressLogQueueSession{newQueuingSession("compress-queue")}
+			state := &interactiveState{agentSession: agentSession, platform: p, pendingMessages: []queuedMessage{{content: "next", platform: p}}}
+			session := e.sessions.GetOrCreateActive("test:user")
+			lockGen, _ := session.TryLock()
+			defer session.Unlock(lockGen)
+			agentSession.events <- Event{Type: eventType, Done: true, Error: errors.New("failed")}
+			unlocked := false
+			e.processCompressEvents(state, session, e.sessions, "test:user", p, nil, &unlocked, false, lockGen)
+			completionIndex, queueIndex := -1, -1
+			for i, r := range compressLogRecords(t, logs) {
+				if r["msg"] == "turn complete" && r["compress"] == true {
+					if completionIndex != -1 {
+						t.Fatal("duplicate compression completion")
+					}
+					completionIndex = i
+				}
+				if r["msg"] == "test queued turn sent" {
+					queueIndex = i
+				}
+			}
+			if completionIndex < 0 || queueIndex < 0 || completionIndex >= queueIndex {
+				t.Fatalf("compression must finish before queued turn starts: %s", logs)
+			}
+		})
+	}
+}

core/engine.go

@@ -10887,6 +10887,17 @@ func (e *Engine) runCompress(state *interactiveState, session *Session, sessions
 // Unlike processInteractiveEvents it does NOT record history and treats
 // an empty result as success rather than "(empty response)".
 func (e *Engine) processCompressEvents(state *interactiveState, session *Session, sessions *SessionManager, sessionKey string, p Platform, replyCtx any, unlocked *bool, auto bool, lockGen uint64) {
+	started := time.Now()
+	logCompletion := func(outcome string) {
+		slog.Info("turn complete",
+			"session", session.ID,
+			"agent_session", session.GetAgentSessionID(),
+			"compress", true,
+			"auto", auto,
+			"outcome", outcome,
+			"turn_duration", time.Since(started),
+		)
+	}
 
 	var textParts []string
 	events := state.agentSession.Events()
@@ -10906,9 +10917,11 @@ func (e *Engine) processCompressEvents(state *interactiveState, session *Session
 
 		select {
 		case <-stopCh:
+			logCompletion("stopped")
 			return
 		case event, ok = <-events:
 			if !ok {
+				logCompletion("eof")
 				e.cleanupInteractiveState(sessionKey, state)
 				if !auto {
 					if len(textParts) > 0 {
@@ -10921,6 +10934,7 @@ func (e *Engine) processCompressEvents(state *interactiveState, session *Session
 				return
 			}
 		case <-idleCh:
+			logCompletion("idle_timeout")
 			if !auto {
 				e.send(p, replyCtx, fmt.Sprintf(e.i18n.T(MsgError), "compress timed out"))
 			}
@@ -10928,10 +10942,12 @@ func (e *Engine) processCompressEvents(state *interactiveState, session *Session
 			e.notifyDroppedQueuedMessages(state, fmt.Errorf("compress timed out"))
 			return
 		case <-e.ctx.Done():
+			logCompletion("shutdown")
 			return
 		}
 
 		if state.isStopped() {
+			logCompletion("stopped")
 			return
 		}
 
@@ -10966,6 +10982,8 @@ func (e *Engine) processCompressEvents(state *interactiveState, session *Session
 				textParts = append(textParts, fmt.Sprintf(e.i18n.T(MsgToolResult), tn, out)+"\n")
 			}
 		case EventResult:
+			// Finish compression before draining queued turns, which may run for minutes.
+			logCompletion("result")
 			result := event.Content
 			if result == "" && len(textParts) > 0 {
 				result = strings.Join(textParts, "")
@@ -10982,6 +11000,7 @@ func (e *Engine) processCompressEvents(state *interactiveState, session *Session
 			e.drainQueuedMessagesAfterCompress(state, session, sessions, sessionKey, unlocked, lockGen)
 			return
 		case EventError:
+			logCompletion("error")
 			if !auto && event.Error != nil {
 				e.reply(p, replyCtx, fmt.Sprintf(e.i18n.T(MsgError), event.Error))
 			}