fix(concurrency): stop /new orphaning per-chat mutex; raise sub-agent stdout scan cap (#29)

Two concurrency/correctness bugs from the audit (FIX_PLAN.md item 5):

1. Telegram /new deleted the per-chat mutex (chatMu.Delete) while an
   in-flight handleChatMessage goroutine still held it. The next message
   LoadOrStored a fresh mutex and TryLocked it successfully, so two agent
   runs executed concurrently for the same chat — corrupting interleaved
   sessionManager.Save writes and clobbering each other's approver. The
   /new reset is extracted into resetChatForNew, which archives the session
   and resets approver trust but retains the mutex (naturally bounded, one
   per chat ID, so not a meaningful leak).

2. delegate_tasks read sub-agent stdout with a default bufio.Scanner (64KB
   token cap). A streamed tool_call event embedding large tool args (e.g. a
   big write_file) exceeded the cap -> ErrTooLong -> reader stopped -> child
   blocked on a full pipe -> cmd.Wait hung until the 120s timeout killed a
   task that had actually succeeded. The scan loop is extracted into
   scanSubagentStream with the buffer cap raised to 10 MiB.

Both paths now have regression tests (verified to fail without the fix):
TestResetChatForNew_KeepsMutex and TestScanSubagentStream_LargeLineNotTruncated.

https://claude.ai/code/session_01PGZHe9fYB8p7eb6Rs8VLHt

Co-authored-by: Claude <noreply@anthropic.com>

Author: jkyberneees

cmd/odek/subagent_scan_test.go

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"strings"
+	"testing"
+)
+
+// TestScanSubagentStream_LargeLineNotTruncated is the regression test for the
+// >64KB NDJSON line bug. A streamed tool_call event embedding large tool
+// arguments exceeds bufio.Scanner's default 64KB token cap. With the default
+// cap the scanner returns ErrTooLong, the result that follows is never read,
+// and (in production) the child blocks on a full pipe until the timeout. This
+// test feeds an oversized progress line followed by the real result and asserts
+// the result is still parsed and no scanner error occurs.
+func TestScanSubagentStream_LargeLineNotTruncated(t *testing.T) {
+	// A tool_call progress line whose embedded args are ~256KB — well past the
+	// 64KB default scanner cap, but under maxSubagentLine.
+	bigArg := strings.Repeat("a", 256*1024)
+	progress := `{"type":"tool_call","name":"write_file","args":{"content":"` + bigArg + `"}}`
+	result := `{"status":"success","summary":"wrote the file"}`
+	input := progress + "\n" + result + "\n"
+
+	var logged int
+	res, lastLine, err := scanSubagentStream(strings.NewReader(input), func(string) { logged++ })
+	if err != nil {
+		t.Fatalf("scanSubagentStream returned error on large line: %v", err)
+	}
+	if res == nil {
+		t.Fatal("result is nil — the final result line was lost after the oversized progress line")
+	}
+	if res["status"] != "success" {
+		t.Errorf("result status = %v, want success", res["status"])
+	}
+	if logged != 1 {
+		t.Errorf("onLog called %d times, want 1 (the big progress line)", logged)
+	}
+	if lastLine != result {
+		t.Errorf("lastLine = %q, want the result line", lastLine)
+	}
+}
+
+// TestScanSubagentStream_NormalFlow verifies ordinary progress + result parsing
+// and that progress lines are forwarded to onLog while the result is returned.
+func TestScanSubagentStream_NormalFlow(t *testing.T) {
+	input := strings.Join([]string{
+		`{"type":"tool_call","name":"shell"}`,
+		`{"type":"tool_result","name":"shell"}`,
+		`{"status":"success","summary":"done"}`,
+	}, "\n") + "\n"
+
+	var logs []string
+	res, _, err := scanSubagentStream(strings.NewReader(input), func(l string) { logs = append(logs, l) })
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res == nil || res["summary"] != "done" {
+		t.Fatalf("result = %v, want summary=done", res)
+	}
+	if len(logs) != 2 {
+		t.Errorf("onLog called %d times, want 2 progress lines", len(logs))
+	}
+}
+
+// TestScanSubagentStream_NilOnLog ensures a nil onLog callback is tolerated
+// (the production path passes nil when no streaming sink is attached).
+func TestScanSubagentStream_NilOnLog(t *testing.T) {
+	input := `{"type":"tool_call"}` + "\n" + `{"status":"ok"}` + "\n"
+	res, _, err := scanSubagentStream(strings.NewReader(input), nil)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if res == nil || res["status"] != "ok" {
+		t.Fatalf("result = %v, want status=ok", res)
+	}
+}

cmd/odek/subagent_tool.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"strings"
@@ -241,34 +242,15 @@ func (t *delegateTasksTool) runTask(taskIdx int, goal, taskContext, guidance, tr
 		return fmt.Sprintf(`{"error":"start: %v"}`, err)
 	}
 
-	// Read stdout line-by-line — NDJSON progress lines followed by final JSON result
-	var result map[string]any
-	var lastLine string
-	scanner := bufio.NewScanner(stdout)
-	for scanner.Scan() {
-		line := scanner.Text()
-		lastLine = line
-
-		// Check if this is a progress line (NDJSON with "type":"tool_call" or "type":"tool_result")
-		var event struct {
-			Type string `json:"type"`
-		}
-		if err := json.Unmarshal([]byte(line), &event); err == nil {
-			if event.Type == "tool_call" || event.Type == "tool_result" {
-				if t.OnSubagentLog != nil {
-					t.OnSubagentLog(taskIdx, line)
-				}
-				continue
-			}
-		}
-
-		// Not a progress line — parse as result JSON
-		var r map[string]any
-		if err := json.Unmarshal([]byte(line), &r); err == nil {
-			result = r
-		}
+	// Read stdout line-by-line — NDJSON progress lines followed by final JSON
+	// result. A streamed tool_call event can embed full tool arguments (e.g. a
+	// large write_file), so lines routinely exceed bufio.Scanner's default 64KB
+	// token cap; scanSubagentStream raises the cap to avoid losing the result.
+	var onLog func(line string)
+	if t.OnSubagentLog != nil {
+		onLog = func(line string) { t.OnSubagentLog(taskIdx, line) }
 	}
-	scannerErr := scanner.Err()
+	result, lastLine, scannerErr := scanSubagentStream(stdout, onLog)
 
 	if err := cmd.Wait(); err != nil {
 		// Process exited with error — result may still be valid
@@ -302,5 +284,49 @@ func (t *delegateTasksTool) runTask(taskIdx int, goal, taskContext, guidance, tr
 	return `{"error":"no result from sub-agent"}`
 }
 
+// maxSubagentLine caps a single NDJSON line read from a sub-agent's stdout.
+// Streamed tool_call events embed full tool arguments (e.g. a large write_file
+// or patch), which routinely exceed bufio.Scanner's default 64KB token cap.
+// Without a raised cap the scanner returns bufio.ErrTooLong, the reader stops,
+// the child blocks writing to a full stdout pipe, and cmd.Wait hangs until the
+// timeout kills a task that actually completed successfully.
+const maxSubagentLine = 10 << 20 // 10 MiB
+
+// scanSubagentStream reads a sub-agent's NDJSON stdout: zero or more progress
+// lines (objects with "type":"tool_call" or "type":"tool_result") followed by
+// the final JSON result object. Progress lines are forwarded to onLog (when
+// non-nil); the last line that parses as a JSON object is returned as result.
+// It returns the result map (nil if none parsed), the last raw line seen, and
+// any scanner error. The scan buffer is sized to maxSubagentLine so large
+// streamed events do not truncate the stream.
+func scanSubagentStream(r io.Reader, onLog func(line string)) (result map[string]any, lastLine string, err error) {
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 64*1024), maxSubagentLine)
+	for scanner.Scan() {
+		line := scanner.Text()
+		lastLine = line
+
+		// Check if this is a progress line (NDJSON with "type":"tool_call" or "type":"tool_result")
+		var event struct {
+			Type string `json:"type"`
+		}
+		if uerr := json.Unmarshal([]byte(line), &event); uerr == nil {
+			if event.Type == "tool_call" || event.Type == "tool_result" {
+				if onLog != nil {
+					onLog(line)
+				}
+				continue
+			}
+		}
+
+		// Not a progress line — parse as result JSON
+		var rmap map[string]any
+		if uerr := json.Unmarshal([]byte(line), &rmap); uerr == nil {
+			result = rmap
+		}
+	}
+	return result, lastLine, scanner.Err()
+}
+
 // Ensure delegateTasksTool implements odek.Tool
 var _ odek.Tool = (*delegateTasksTool)(nil)

cmd/odek/telegram.go

@@ -82,6 +82,25 @@ func getChatMutex(chatID int64) *sync.Mutex {
 	return v.(*sync.Mutex)
 }
 
+// resetChatForNew implements the /new command's session reset: it archives the
+// current session and clears the approver's trust state for a fresh start.
+//
+// It deliberately does NOT remove the per-chat mutex from chatMu. Deleting the
+// mutex while an in-flight handleChatMessage goroutine still holds it lets the
+// next message LoadOrStore a *fresh* mutex and TryLock it successfully, so two
+// agent runs execute concurrently for the same chat — corrupting interleaved
+// sessionManager.Save writes and clobbering each other's approver. The per-chat
+// mutex is naturally bounded (one entry per chat ID ever seen), so retaining it
+// is not a meaningful leak.
+func resetChatForNew(chatID int64, sessionManager *telegram.SessionManager, handler *telegram.Handler, log telegram.Logger) {
+	if err := sessionManager.ArchiveAndDelete(chatID); err != nil {
+		log.Warn("archive session", "chat_id", chatID, "error", err)
+	}
+	if a := handler.GetApprover(chatID); a != nil {
+		a.ResetTrust()
+	}
+}
+
 // telegramCmd is the entry point for "odek telegram".
 func telegramCmd(args []string) error {
 	// 0. Acquire singleton lock — kill any stale previous instance.
@@ -266,13 +285,7 @@ func telegramCmd(args []string) error {
 		// The current session is preserved as a timestamped archive file
 		// so it can be revisited via `odek session list`.
 		if cmdName == "new" {
-			if err := sessionManager.ArchiveAndDelete(chatID); err != nil {
-				handlerLog.Warn("archive session", "chat_id", chatID, "error", err)
-			}
-			chatMu.Delete(chatID) // prevent mutex leak on session reset
-			if a := handler.GetApprover(chatID); a != nil {
-				a.ResetTrust()
-			}
+			resetChatForNew(chatID, sessionManager, handler, handlerLog)
 			var b strings.Builder
 			b.WriteString("🔄 *Session archived, starting fresh*\n\n")
 			fmt.Fprintf(&b, "• Model: `%s`\n", resolved.Model)

cmd/odek/telegram_new_reset_test.go

@@ -0,0 +1,85 @@
+package main
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/BackendStack21/odek/internal/session"
+	"github.com/BackendStack21/odek/internal/telegram"
+)
+
+// TestResetChatForNew_KeepsMutex is the regression test for the "/new deletes
+// the per-chat mutex while a run holds it" bug. The scenario:
+//
+//  1. An in-flight handleChatMessage goroutine holds the per-chat mutex (run A).
+//  2. The user sends /new, which calls resetChatForNew.
+//  3. The next message calls getChatMutex again (run B).
+//
+// If resetChatForNew removes the mutex from chatMu (the old behavior), run B's
+// getChatMutex LoadOrStores a *fresh* mutex and TryLocks it successfully — two
+// runs execute concurrently for the same chat. This test asserts the mutex is
+// preserved across the reset so run B still blocks on run A.
+func TestResetChatForNew_KeepsMutex(t *testing.T) {
+	withTempHome(t) // session.NewStore writes under a sandbox HOME
+
+	chatID := int64(770001)
+	chatMu = sync.Map{} // isolate from other tests
+
+	store, err := session.NewStore()
+	if err != nil {
+		t.Fatalf("session.NewStore: %v", err)
+	}
+	sm := telegram.NewSessionManager(store, time.Hour)
+	bot := telegram.NewBot("test:token")
+	handler := telegram.NewHandler(bot)
+
+	// Run A acquires the per-chat mutex and holds it for the duration.
+	muA := getChatMutex(chatID)
+	if !muA.TryLock() {
+		t.Fatal("run A could not acquire a fresh mutex")
+	}
+	defer muA.Unlock()
+
+	// User sends /new while run A is still in flight.
+	resetChatForNew(chatID, sm, handler, telegram.NewNopLogger())
+
+	// Run B fetches the per-chat mutex. It MUST be the same instance run A
+	// holds, so B cannot lock it — serialization is preserved.
+	muB := getChatMutex(chatID)
+	if muB != muA {
+		t.Fatal("resetChatForNew orphaned the per-chat mutex: run B got a different mutex and would run concurrently with run A")
+	}
+	if muB.TryLock() {
+		muB.Unlock()
+		t.Fatal("run B was able to lock the per-chat mutex while run A holds it — concurrent same-chat runs possible")
+	}
+}
+
+// TestResetChatForNew_ResetsApproverTrust verifies the reset still clears the
+// approver's trust state (the legitimate purpose of /new's reset), so removing
+// the mutex delete did not drop the approver reset.
+func TestResetChatForNew_ResetsApproverTrust(t *testing.T) {
+	withTempHome(t)
+
+	chatID := int64(770002)
+	chatMu = sync.Map{}
+
+	store, err := session.NewStore()
+	if err != nil {
+		t.Fatalf("session.NewStore: %v", err)
+	}
+	sm := telegram.NewSessionManager(store, time.Hour)
+	bot := telegram.NewBot("test:token")
+	handler := telegram.NewHandler(bot)
+
+	approver := telegram.NewTelegramApprover(bot, chatID)
+	handler.SetApprover(chatID, approver)
+
+	// Should not panic and should reach the approver reset path.
+	resetChatForNew(chatID, sm, handler, telegram.NewNopLogger())
+
+	if handler.GetApprover(chatID) != approver {
+		t.Fatal("approver unexpectedly replaced by resetChatForNew")
+	}
+}