TEST ONLY: concurrency cancel-in-progress (do not merge)

.github/workflows/ci.yml

@@ -9,6 +9,10 @@ on:
 permissions:
   contents: read
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
 jobs:
   test:
     runs-on: self-hosted

internal/llmloop/compression.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -41,6 +42,16 @@ type compressionJob struct {
 	snapshotLen int // message count when the snapshot was taken
 }
 
+// compressionState is the async-compression bookkeeping for a single
+// conversation (one RunPerFile call). The Runner is shared by concurrent
+// per-file goroutines, so this state must not live on the Runner: a shared
+// slot lets one file apply, cancel, or replace another file's compression
+// job (#384).
+type compressionState struct {
+	mu         sync.Mutex
+	pendingJob *compressionJob
+}
+
 // CountMessagesTokens returns the rough token count of msgs by summing the
 // per-message text token count. Exported because both review and scan top
 // layers may want it for pre-flight checks.
@@ -217,7 +228,6 @@ func (r *Runner) runCompression(ctx context.Context, msgs []llm.Message, filePat
 	rec := fs.AppendTaskRecord(session.MemoryCompressionTask, compressionMsgs)
 	if err != nil {
 		rec.SetError(err, duration)
-		fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
 		// Return msgs unchanged: truncating to frozenEnd would discard all
 		// conversation context, which is worse than staying over the token
 		// limit temporarily.
@@ -252,31 +262,40 @@ func (r *Runner) runCompression(ctx context.Context, msgs []llm.Message, filePat
 	return rebuilt, nil
 }
 
-// triggerAsyncCompression kicks off a background compression job.
-func (r *Runner) triggerAsyncCompression(ctx context.Context, messages []llm.Message, filePath string) {
+// triggerAsyncCompression kicks off a background compression job for the
+// conversation owning st. A no-op when a job is already pending — the
+// check-and-set happens under st.mu so concurrent callers cannot replace
+// (and thereby leak) an in-flight job.
+func (r *Runner) triggerAsyncCompression(ctx context.Context, st *compressionState, messages []llm.Message, filePath string) {
+	st.mu.Lock()
+	if st.pendingJob != nil {
+		st.mu.Unlock()
+		return
+	}
 	msgSnapshot := copyMessages(messages)
-
 	asyncCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Minute)
-
 	job := &compressionJob{done: make(chan struct{}), cancel: cancel, snapshotLen: len(messages)}
-	r.compressionMu.Lock()
-	r.pendingJob = job
-	r.compressionMu.Unlock()
+	st.pendingJob = job
+	st.mu.Unlock()
 
 	go func() {
 		defer cancel()
 		rebuilt, err := r.runCompression(asyncCtx, msgSnapshot, filePath)
 
-		r.compressionMu.Lock()
-		defer r.compressionMu.Unlock()
+		st.mu.Lock()
+		defer st.mu.Unlock()
 
-		if r.pendingJob != job {
+		if st.pendingJob != job {
 			return // cancelled or superseded
 		}
 		if err != nil {
-			// Compression failed — abandon the job rather than applying a
-			// truncated/unmodified snapshot over live messages.
-			r.pendingJob = nil
+			// Still the owner, so this is a genuine failure rather than a
+			// deliberate cancel (cancelPendingCompression cancels and clears
+			// pendingJob under the lock, so cancelled jobs fail the ownership
+			// check above and die silently). Abandon the job rather than
+			// applying a truncated/unmodified snapshot over live messages.
+			fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
+			st.pendingJob = nil
 			close(job.done)
 			return
 		}
@@ -288,10 +307,10 @@ func (r *Runner) triggerAsyncCompression(ctx context.Context, messages []llm.Mes
 // tryApplyPendingCompression checks whether a background compression has
 // completed and swaps the rebuilt messages into place. Returns true if
 // applied.
-func (r *Runner) tryApplyPendingCompression(messages *[]llm.Message) bool {
-	r.compressionMu.Lock()
-	job := r.pendingJob
-	r.compressionMu.Unlock()
+func (r *Runner) tryApplyPendingCompression(st *compressionState, messages *[]llm.Message) bool {
+	st.mu.Lock()
+	job := st.pendingJob
+	st.mu.Unlock()
 
 	if job == nil {
 		return false
@@ -300,8 +319,8 @@ func (r *Runner) tryApplyPendingCompression(messages *[]llm.Message) bool {
 	select {
 	case <-job.done:
 		applied := false
-		r.compressionMu.Lock()
-		if r.pendingJob == job && job.rebuilt != nil {
+		st.mu.Lock()
+		if st.pendingJob == job && job.rebuilt != nil {
 			rebuilt := job.rebuilt
 			// Preserve any messages appended after the snapshot was taken —
 			// the background job only compressed messages[:snapshotLen].
@@ -311,23 +330,24 @@ func (r *Runner) tryApplyPendingCompression(messages *[]llm.Message) bool {
 			*messages = rebuilt
 			applied = true
 		}
-		if r.pendingJob == job {
-			r.pendingJob = nil
+		if st.pendingJob == job {
+			st.pendingJob = nil
 		}
-		r.compressionMu.Unlock()
+		st.mu.Unlock()
 		return applied
 	default:
 		return false
 	}
 }
 
-// cancelPendingCompression aborts any in-flight background compression.
-func (r *Runner) cancelPendingCompression() {
-	r.compressionMu.Lock()
-	defer r.compressionMu.Unlock()
+// cancelPendingCompression aborts the conversation's in-flight background
+// compression, if any.
+func (r *Runner) cancelPendingCompression(st *compressionState) {
+	st.mu.Lock()
+	defer st.mu.Unlock()
 
-	if r.pendingJob != nil {
-		r.pendingJob.cancel()
-		r.pendingJob = nil
+	if st.pendingJob != nil {
+		st.pendingJob.cancel()
+		st.pendingJob = nil
 	}
 }

internal/llmloop/loop.go

@@ -39,8 +39,9 @@ type Deps struct {
 }
 
 // Runner is a per-session (across files) executor of the LLM tool-use
-// loop. Token counters, warnings, and the optional background compression
-// job are aggregated across every RunPerFile call.
+// loop. Token counters and warnings are aggregated across every RunPerFile
+// call; background memory compression is scoped to each RunPerFile
+// conversation (see compressionState).
 type Runner struct {
 	deps                  Deps
 	totalInputTokens      int64 // atomically updated
@@ -51,8 +52,6 @@ type Runner struct {
 	warnings              []AgentWarning
 	toolCallsMu           sync.Mutex
 	toolCalls             map[string]int64
-	compressionMu         sync.Mutex
-	pendingJob            *compressionJob
 }
 
 // NewRunner returns a Runner bound to the given dependencies.
@@ -153,6 +152,11 @@ func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath
 	consecutiveEmptyRounds := 0
 	sessionID := uuid.NewString()
 
+	// Async compression is owned by this conversation alone; the deferred
+	// cancel aborts any job still in flight when the conversation ends.
+	st := &compressionState{}
+	defer r.cancelPendingCompression(st)
+
 	for toolReqCount > 0 {
 		select {
 		case <-ctx.Done():
@@ -250,7 +254,7 @@ func (r *Runner) RunPerFile(ctx context.Context, messages []llm.Message, newPath
 			consecutiveEmptyRounds = 0
 		}
 
-		succeed := r.addNextMessage(ctx, content, calls, results, &messages, newPath)
+		succeed := r.addNextMessage(ctx, content, calls, results, &messages, newPath, st)
 		if !succeed {
 			fmt.Fprintf(stdout.Writer(), "[ocr] Context compression exceeded threshold for %s, stopping.\n", newPath)
 			break
@@ -430,23 +434,23 @@ func (r *Runner) executeToolCall(ctx context.Context, newPath string, call llm.T
 // warning (80%) MaxTokens thresholds. Returns false when even after
 // synchronous compression the conversation is still over the warning
 // threshold — caller should stop the loop in that case.
-func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, toolCalls []llm.ToolCall, results []tool.ToolCallResult, messages *[]llm.Message, filePath string) bool {
+func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, toolCalls []llm.ToolCall, results []tool.ToolCallResult, messages *[]llm.Message, filePath string, st *compressionState) bool {
 	maxAllowed := r.deps.Template.MaxTokens
 	softLimit := int(float64(maxAllowed) * tokenSoftThreshold)
 	warnLimit := int(float64(maxAllowed) * tokenWarningThreshold)
 
-	r.tryApplyPendingCompression(messages)
-
-	tokenCount := CountMessagesTokens(*messages)
-
-	if tokenCount > warnLimit {
-		r.cancelPendingCompression()
-		*messages, _ = r.runCompression(ctx, *messages, filePath)
-		tokenCount = CountMessagesTokens(*messages)
-	}
-
-	if tokenCount > softLimit && r.pendingJob == nil {
-		r.triggerAsyncCompression(ctx, *messages, filePath)
+	r.tryApplyPendingCompression(st, messages)
+
+	// A conversation can already be over the warning threshold before this
+	// round's messages are appended (e.g. an oversized initial prompt).
+	if CountMessagesTokens(*messages) > warnLimit {
+		r.cancelPendingCompression(st)
+		var err error
+		if *messages, err = r.runCompression(ctx, *messages, filePath); err != nil {
+			// Compression failed; continue with over-limit messages — the
+			// post-append check below will retry.
+			fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
+		}
 	}
 
 	if len(toolCalls) > 0 {
@@ -461,11 +465,22 @@ func (r *Runner) addNextMessage(ctx context.Context, assistantContent string, to
 
 	finalCount := CountMessagesTokens(*messages)
 	if finalCount > warnLimit {
-		r.cancelPendingCompression()
-		*messages, _ = r.runCompression(ctx, *messages, filePath)
+		r.cancelPendingCompression(st)
+		var err error
+		if *messages, err = r.runCompression(ctx, *messages, filePath); err != nil {
+			fmt.Fprintf(stdout.Writer(), "[ocr] Memory compression failed: %v\n", err)
+		}
+		finalCount = CountMessagesTokens(*messages)
+	}
+
+	// Trigger async compression only after all appends for this update, so
+	// a job is never started and then immediately cancelled by the same
+	// call (#384), and never started when we are about to return false.
+	if finalCount > softLimit && finalCount < warnLimit {
+		r.triggerAsyncCompression(ctx, st, *messages, filePath)
 	}
 
-	return CountMessagesTokens(*messages) < warnLimit
+	return finalCount < warnLimit
 }
 
 // parseToolArgs unmarshals a tool call's raw JSON arguments, always