fix: bug-hunt round — batched events, wss, sanitization, layout bounds

AGENTS.md

@@ -131,7 +131,10 @@ feat(tui): compact tool steps with Ctrl+E details toggle
   cannot run View per fragment (that filled the 256-deep Events buffer
   and tripped odek's 30s write timeout — connection lost after any
   prompt). `ingestWireBatch` re-arms `listen()` as a top-level cmd, never
-  nested inside handleEvent's Batch. The client also merges consecutive
+  nested inside handleEvent's Batch. While a batch drains, `handleEvent` parks its
+  state-consuming follow-ups (`sendQueued`, `planFollowup`, attention, `flushKicks`)
+  in `batchCarry` instead of returning them, so the flat batch never drops a
+  queue pop, bell, or tab refresh. The client also merges consecutive
   thinking deltas (16 at a time) before enqueueing. Assistant text deltas
   enqueue immediately so short progress notes cannot wait for a later frame;
   their redraws still coalesce in the TUI. The header

cmd/bodek/bughunt_regress_test.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/BackendStack21/bodek/internal/settings"
+)
+
+// Another instance's saved preference must survive this instance's save.
+func TestRegressSettingsSaveKeepsOtherInstanceChange(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "config.json")
+	t.Setenv("BODEK_CONFIG", path)
+	cfg, err := parseConfig(nil, io.Discard)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Instance A saves a theme after this instance started.
+	if err := settings.Save(settings.Settings{Theme: "ember-light"}); err != nil {
+		t.Fatal(err)
+	}
+	// This instance changes only verbosity.
+	cfg.persist.Verbosity = "quiet"
+	if err := cfg.save(); err != nil {
+		t.Fatal(err)
+	}
+	got, err := settings.Load()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.Theme != "ember-light" {
+		t.Fatalf("theme saved by the other instance was lost: %+v", got)
+	}
+}
+
+// The log holds odek's `WS token:` line; a pre-existing permissive file must
+// not be reused as-is.
+func TestRegressServerLogNotReusedWhenPermissive(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("TMPDIR", dir)
+	p := filepath.Join(dir, serverLogName)
+	if err := os.WriteFile(p, []byte("x"), 0o666); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Chmod(p, 0o666); err != nil {
+		t.Fatal(err)
+	}
+	_, path, closeFn := openServerLog()
+	defer closeFn()
+	if path == "" {
+		return // refusing the file is acceptable
+	}
+	fi, err := os.Stat(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if fi.Mode().Perm()&0o077 != 0 {
+		t.Fatalf("token-bearing log is group/world accessible: %v", fi.Mode().Perm())
+	}
+}

cmd/bodek/log.go

@@ -14,11 +14,17 @@ import (
 // so it never grows without bound.
 const serverLogName = "bodek-odek-serve.log"
 
-// openServerLog opens (truncating) the server log file. On failure it falls
+// openServerLog (re)creates the server log file. On failure it falls
 // back to a discarding writer so a missing temp dir never breaks startup.
 func openServerLog() (w io.Writer, path string, closeFn func()) {
 	path = filepath.Join(os.TempDir(), serverLogName)
-	f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
+	// The log receives odek's "WS token:" line. Never reuse an existing
+	// file: another user may have pre-created it (or a symlink to a file of
+	// theirs) with open permissions. Remove it and create exclusively; if the
+	// remove is refused (not ours, sticky dir) the create fails and logging
+	// is skipped rather than leaking the token.
+	_ = os.Remove(path)
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
 	if err != nil {
 		return io.Discard, "", func() {}
 	}

cmd/bodek/main.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"os/signal"
+	"reflect"
 	"runtime"
 	"strings"
 	"syscall"
@@ -50,6 +51,10 @@ type config struct {
 	extraArgs []string
 
 	persist settings.Settings // the loaded file, re-saved when /theme switches
+	// persisted is the last state read from or written to disk; save writes
+	// only the fields that differ from it, merged over the file's current
+	// contents, so a change made by another running instance survives.
+	persisted settings.Settings
 	// persistDisabled is set when the settings file could not be read;
 	// a full-replace Save would then overwrite the (unparsed) file with
 	// a near-empty struct, so all persistence is refused for the run.
@@ -67,7 +72,21 @@ func (c *config) save() error {
 	if c.persistDisabled {
 		return fmt.Errorf("settings file could not be parsed; changes are not saved this run")
 	}
-	return settings.Save(c.persist)
+	cur, err := settings.Load()
+	if err != nil {
+		cur = c.persisted // file vanished or changed shape: fall back to our baseline
+	}
+	rb, rc, rd := reflect.ValueOf(c.persisted), reflect.ValueOf(c.persist), reflect.ValueOf(&cur).Elem()
+	for i := range rc.NumField() {
+		if !reflect.DeepEqual(rb.Field(i).Interface(), rc.Field(i).Interface()) {
+			rd.Field(i).Set(rc.Field(i))
+		}
+	}
+	if err := settings.Save(cur); err != nil {
+		return err
+	}
+	c.persisted = c.persist
+	return nil
 }
 
 func parseConfig(args []string, output io.Writer) (config, error) {
@@ -89,6 +108,7 @@ func parseConfig(args []string, output io.Writer) (config, error) {
 		_, _ = fmt.Fprintln(output, "bodek: settings changes will not be saved this run (file could not be parsed)")
 	}
 	cfg.persist = st
+	cfg.persisted = st
 	fs := flag.NewFlagSet("bodek", flag.ContinueOnError)
 	if output != nil {
 		fs.SetOutput(output)

internal/client/bughunt_regress_test.go

@@ -0,0 +1,100 @@
+package client
+
+import (
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	ws "golang.org/x/net/websocket"
+)
+
+// wss:// must start a TLS handshake (ClientHello record type 0x16).
+func TestRegressWSSStartsTLS(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = ln.Close() }()
+	first := make(chan byte, 1)
+	go func() {
+		c, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer func() { _ = c.Close() }()
+		_ = c.SetReadDeadline(time.Now().Add(3 * time.Second))
+		b := make([]byte, 1)
+		if _, err := c.Read(b); err == nil {
+			first <- b[0]
+		}
+	}()
+	addr := ln.Addr().String()
+	go func() { _, _ = Dial("wss://"+addr+"/ws", "https://"+addr, "https://"+addr, "tok") }()
+	select {
+	case b := <-first:
+		if b != 0x16 {
+			t.Fatalf("wss dial sent %q (0x%02x) instead of a TLS ClientHello", b, b)
+		}
+	case <-time.After(4 * time.Second):
+		t.Fatal("no bytes received")
+	}
+}
+
+func validateServer(t *testing.T, onWS func(*ws.Conn), mux func(*http.ServeMux)) string {
+	t.Helper()
+	m := http.NewServeMux()
+	m.Handle("/ws", ws.Server{
+		Handshake: func(*ws.Config, *http.Request) error { return nil },
+		Handler:   ws.Handler(onWS),
+	})
+	if mux != nil {
+		mux(m)
+	}
+	srv := httptest.NewServer(m)
+	t.Cleanup(srv.Close)
+	return srv.URL
+}
+
+// The trailing <16 thinking deltas must surface without a later frame.
+func TestRegressThinkingTailFlushesWhenIdle(t *testing.T) {
+	base := validateServer(t, func(c *ws.Conn) {
+		for range 3 {
+			_ = ws.Message.Send(c, `{"type":"thinking_delta","content":"x"}`)
+		}
+		time.Sleep(3 * time.Second)
+	}, nil)
+	c, err := Dial("ws"+strings.TrimPrefix(base, "http")+"/ws", base, base, "tok")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = c.Close() }()
+	select {
+	case ev := <-c.Events:
+		if ev.Type != "thinking_delta" {
+			t.Fatalf("got %v", ev.Type)
+		}
+	case <-time.After(1500 * time.Millisecond):
+		t.Fatal("pending thinking deltas were held back while the stream idled")
+	}
+}
+
+// Consolidation is LLM-backed; it must not be cut at the 3s interactive budget.
+func TestRegressConsolidateOutlivesInteractiveTimeout(t *testing.T) {
+	base := validateServer(t, func(c *ws.Conn) { time.Sleep(6 * time.Second) }, func(m *http.ServeMux) {
+		m.HandleFunc("/api/memory/consolidate", func(w http.ResponseWriter, r *http.Request) {
+			time.Sleep(3500 * time.Millisecond)
+			_, _ = w.Write([]byte(`{}`))
+		})
+	})
+	c, err := Dial("ws"+strings.TrimPrefix(base, "http")+"/ws", base, base, "tok")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = c.Close() }()
+	if err := c.ConsolidateMemory("user"); err != nil {
+		t.Fatalf("consolidate failed although the server completes it: %v", err)
+	}
+}

internal/client/client.go

@@ -8,6 +8,7 @@
 package client
 
 import (
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"net"
@@ -310,12 +311,32 @@
 		_ = raw.Close()
 		return nil, err
 	}
-	conn, err := ws.NewClient(cfg, raw)
+	// ws.NewClient speaks plain WS over whatever it is handed; a wss:// URL
+	// needs the TLS handshake done here (ws.DialConfig would, but has no
+	// dial timeout).
+	var nc net.Conn = raw
+	if cfg.Location.Scheme == "wss" {
+		tc := cfg.TlsConfig
+		if tc == nil {
+			tc = &tls.Config{MinVersion: tls.VersionTLS12}
+		}
+		if tc.ServerName == "" {
+			tc = tc.Clone()
+			tc.ServerName = cfg.Location.Hostname()
+		}
+		tlsConn := tls.Client(raw, tc)
+		if err := tlsConn.Handshake(); err != nil {
+			_ = raw.Close()
+			return nil, err
+		}
+		nc = tlsConn
+	}
+	conn, err := ws.NewClient(cfg, nc)
 	if err != nil {
-		_ = raw.Close()
+		_ = nc.Close()
 		return nil, err
 	}
-	_ = raw.SetDeadline(time.Time{}) // live stream: no deadline
+	_ = nc.SetDeadline(time.Time{}) // live stream: no deadline
 	return conn, nil
 }
 
@@ -363,50 +384,86 @@
 // only happens on a dead link. A test hook; do not shrink further.
 var readIdleTimeout = 45 * time.Second
 
+// thinkingIdleFlush bounds how long a merged thinking_delta run may sit
+// pending when the stream goes quiet: the coalescer only flushes on the next
+// frame, so the tail of a reasoning burst would otherwise stay invisible
+// until the server's next heartbeat.
+var thinkingIdleFlush = 50 * time.Millisecond
+
 func (c *Client) readLoop() {
 	defer close(c.Events)
 	defer func() { _ = c.conn.Close() }() // release the fd even when the sender never closes (reconnect swap)
-	var pending *Event
-	n := 0
-	flush := func() {
+	var (
+		mu      sync.Mutex // guards pending/n/closed and orders every emit
+		pending *Event
+		n       int
+		closed  bool
+		timer   *time.Timer
+	)
+	flush := func() { // caller holds mu
 		if pending == nil {
 			return
 		}
 		c.emit(*pending)
 		pending = nil
 		n = 0
 	}
+	defer func() { // runs before close(c.Events): the idle timer must not emit after it
+		mu.Lock()
+		closed = true
+		if timer != nil {
+			timer.Stop()
+		}
+		mu.Unlock()
+	}()
 	for {
 		var data []byte
 		_ = c.conn.SetReadDeadline(time.Now().Add(readIdleTimeout))
 		if err := ws.Message.Receive(c.conn, &data); err != nil {
+			mu.Lock()
 			flush()
 			c.emit(Event{Type: EventDisconnected})
+			mu.Unlock()
 			return
 		}
 		_ = c.conn.SetReadDeadline(time.Time{}) // received: drop the deadline while decoding
 		var ev Event
 		if err := json.Unmarshal(data, &ev); err != nil {
 			continue // ignore malformed frames
 		}
+		mu.Lock()
 		if ev.Type == "thinking_delta" {
 			if pending != nil && pending.Type == ev.Type && pending.TurnID == ev.TurnID {
 				pending.Content += ev.Content
 				n++
-				if n < deltaCoalesceMax {
-					continue
+				if n >= deltaCoalesceMax {
+					flush()
 				}
+			} else {
 				flush()
-				continue
+				e := ev
+				pending = &e
+				n = 1
 			}
-			flush()
-			e := ev
-			pending = &e
-			n = 1
+			if pending != nil {
+				if timer == nil {
+					timer = time.AfterFunc(thinkingIdleFlush, func() {
+						mu.Lock()
+						defer mu.Unlock()
+						if !closed {
+							flush()
+						}
+					})
+				} else {
+					timer.Reset(thinkingIdleFlush)
+				}
+			}
+			mu.Unlock()
 			continue
 		}
 		flush()
 		c.emit(ev)
+		mu.Unlock()
 	}
 }