fix: bug-hunt wave — transport, persistence, and approval teardown

internal/client/client.go

@@ -13,7 +13,6 @@ import (
 	"net"
 	"net/http"
 	"net/url"
-	"strings"
 	"sync"
 	"time"
 
@@ -323,11 +322,13 @@ func dialWS(cfg *ws.Config) (*ws.Conn, error) {
 // scheme's standard port when absent (odek serve always prints one, but a
 // hand-typed ws://host URL should still dial).
 func hostPortAddr(u *url.URL) string {
-	if u.Host != "" && !strings.Contains(u.Host, ":") {
+	if u.Host != "" && u.Port() == "" {
+		// u.Port() is empty for bare hosts AND bracketed IPv6 literals
+		// ("[::1]" contains ':' but no port) — both get the scheme default.
 		if u.Scheme == "wss" || u.Scheme == "https" {
-			return u.Host + ":443"
+			return net.JoinHostPort(u.Hostname(), "443")
 		}
-		return u.Host + ":80"
+		return net.JoinHostPort(u.Hostname(), "80")
 	}
 	return u.Host
 }

internal/client/client_addr_test.go

@@ -0,0 +1,39 @@
+package client
+
+import (
+	"net/url"
+	"testing"
+)
+
+func parseURL(t *testing.T, raw string) *url.URL {
+	t.Helper()
+	u, err := url.Parse(raw)
+	if err != nil {
+		t.Fatalf("parse %s: %v", raw, err)
+	}
+	return u
+}
+
+// Regression: hostPortAddr treated every IPv6 literal as "already has a
+// port" because the address contains ':', so ws://[::1]/ws dialled an
+// address with no port and failed with "missing port in address".
+func TestHostPortAddrIPv6DefaultPort(t *testing.T) {
+	tests := []struct {
+		name, raw, want string
+	}{
+		{"ipv6 ws default", "ws://[::1]/ws", "[::1]:80"},
+		{"ipv6 wss default", "wss://[2001:db8::1]/ws", "[2001:db8::1]:443"},
+		{"ipv6 explicit port", "ws://[::1]:8080/ws", "[::1]:8080"},
+		{"ipv4 default", "ws://127.0.0.1/ws", "127.0.0.1:80"},
+		{"hostname default", "ws://localhost/ws", "localhost:80"},
+		{"explicit port", "ws://127.0.0.1:9000/ws", "127.0.0.1:9000"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			u := parseURL(t, tc.raw)
+			if got := hostPortAddr(u); got != tc.want {
+				t.Fatalf("hostPortAddr(%s) = %q, want %q", tc.raw, got, tc.want)
+			}
+		})
+	}
+}

internal/client/rest.go

@@ -63,6 +63,11 @@ type ModelInfo struct {
 	Current     bool   `json:"current"`
 }
 
+// maxJSONBytes bounds any single REST JSON decode (sessions, models, jobs…).
+// Transcripts can be large, but a broken server streaming unbounded bytes
+// must not OOM the client.
+const maxJSONBytes = 256 << 20
+
 // Sessions lists recent saved sessions (auth tokens are not included).
 func (c *Client) Sessions() ([]Session, error) {
 	var out []Session
@@ -391,5 +396,7 @@ func (c *Client) getJSON(u, sessionToken string, dst interface{}) error {
 	if resp.StatusCode != http.StatusOK {
 		return fmt.Errorf("status %s", resp.Status)
 	}
-	return json.NewDecoder(resp.Body).Decode(dst)
+	// Bound the decode: a broken or hostile server streaming unbounded
+	// bytes must not OOM bodek (ExportSession already does this).
+	return json.NewDecoder(io.LimitReader(resp.Body, maxJSONBytes)).Decode(dst)
 }

internal/server/server.go

@@ -294,6 +294,11 @@ func (c *Conn) Stop() {
 	if !c.stopping.CompareAndSwap(false, true) {
 		return
 	}
+	// An already-reaped child must not be signalled: the PID may have been
+	// recycled and the group kill would hit an innocent process.
+	if c.reaped.Load() {
+		return
+	}
 	// Graceful shutdown owns the exit — retire the orphan watchdog first.
 	c.watchMu.Lock()
 	if c.watch != nil {
@@ -375,8 +380,10 @@ func splitTokenURL(raw string) (base, token string) {
 	if err != nil {
 		return raw, ""
 	}
-	token = u.Query().Get("token")
-	u.RawQuery = ""
+	q := u.Query()
+	token = q.Get("token")
+	q.Del("token") // strip only the token; other params must survive
+	u.RawQuery = q.Encode()
 	u.Fragment = ""
 	return u.String(), token
 }
@@ -452,6 +459,13 @@ func (s *tokenScanWriter) scan(p []byte) {
 // appendTail splits p into complete lines and keeps the last maxTailLines
 // of them in the diagnostics tail. Callers hold s.mu.
 func (s *tokenScanWriter) appendTail(p []byte) {
+	// A chunk may start with the tail of a line whose head was buffered by
+	// a previous partialTail call — merge before splitting, or the line
+	// lands severed in the diagnostics tail.
+	if len(s.buf) > 0 {
+		p = append(s.buf, p...)
+		s.buf = nil
+	}
 	rest := p
 	for {
 		i := bytes.IndexByte(rest, '\n')

internal/server/server_split_test.go

@@ -0,0 +1,27 @@
+package server
+
+import (
+	"testing"
+)
+
+// Regression: splitTokenURL cleared the entire RawQuery, so an attach URL
+// with extra parameters (?token=x&profile=y) silently lost them.
+func TestSplitTokenURLKeepsOtherParams(t *testing.T) {
+	base, token := splitTokenURL("http://127.0.0.1:8080/?token=abc&profile=dev")
+	if token != "abc" {
+		t.Fatalf("token = %q, want abc", token)
+	}
+	if want := "http://127.0.0.1:8080/?profile=dev"; base != want {
+		t.Fatalf("base = %q, want %q", base, want)
+	}
+	// A token-only URL still strips cleanly.
+	base, token = splitTokenURL("http://127.0.0.1:8080/?token=abc")
+	if token != "abc" || base != "http://127.0.0.1:8080/" {
+		t.Fatalf("token-only URL: base=%q token=%q", base, token)
+	}
+	// Fragments are stripped too.
+	base, token = splitTokenURL("http://127.0.0.1:8080/?token=abc#frag")
+	if token != "abc" || base != "http://127.0.0.1:8080/" {
+		t.Fatalf("fragment not stripped: base=%q token=%q", base, token)
+	}
+}

internal/server/server_tail_test.go

@@ -0,0 +1,19 @@
+package server
+
+import (
+	"io"
+	"testing"
+)
+
+// Regression: appendTail never merged the buffered partial line (s.buf) into
+// the incoming chunk, so a stderr line straddling two pipe writes lost its
+// head half — the tail recorded "ror: boom" instead of "error: boom".
+func TestTailSplitsAcrossWritesStayWhole(t *testing.T) {
+	s := &tokenScanWriter{w: io.Discard, tok: "found", tail: []string{}}
+	s.Write([]byte("er")) // no newline: buffered
+	s.Write([]byte("ror: bind: address already in use\n"))
+	got := s.Tail(4)
+	if got != "error: bind: address already in use" {
+		t.Fatalf("split line not reassembled whole in tail: %q", got)
+	}
+}

internal/tokens/tokens.go

@@ -45,7 +45,7 @@ func openAt(path string) *Store {
 		return s // missing store: fresh start
 	}
 	if err := json.Unmarshal(data, &s.m); err != nil {
-		if qErr := os.Rename(path, path+".corrupt"); qErr == nil {
+		if qErr := quarantine(path); qErr == nil {
 			warnPersist(fmt.Errorf("corrupt store quarantined as %s.corrupt: %w", path, err))
 		} else {
 			warnPersist(fmt.Errorf("corrupt store kept in place: %w", err))
@@ -80,6 +80,7 @@ func (s *Store) Set(id, token string) {
 	if s.m[id] == token {
 		return // no change; skip the disk write
 	}
+	s.mergeLocked()
 	s.m[id] = token
 	s.persistLocked()
 }
@@ -94,10 +95,34 @@ func (s *Store) Delete(id string) {
 	if _, ok := s.m[id]; !ok {
 		return
 	}
+	s.mergeLocked()
 	delete(s.m, id)
 	s.persistLocked()
 }
 
+// mergeLocked adopts the on-disk map wholesale so deletions by a peer
+// converge (a plain add-only merge let a stale peer rewrite every token
+// another instance had deleted on its next unrelated persist). The caller
+// re-applies its own mutation right after — that id wins. Mirrors
+// workspace's reloadLocked.
+func (s *Store) mergeLocked() {
+	if s.path == "" {
+		return
+	}
+	data, err := os.ReadFile(s.path)
+	if err != nil {
+		return // missing or unreadable: keep what we have
+	}
+	var disk map[string]string
+	if json.Unmarshal(data, &disk) != nil {
+		return // corrupt: Open's quarantine owns the diagnosis
+	}
+	if disk == nil {
+		disk = map[string]string{}
+	}
+	s.m = disk
+}
+
 // persistLocked writes the store while the mutex is held. Snapshot-then-
 // persist-outside-the-lock let interleaved Set/Delete writes reorder on
 // disk: an older snapshot landing last resurrected deleted tokens and
@@ -154,6 +179,19 @@ func persist(path string, m map[string]string) error {
 	return nil
 }
 
+// quarantine sets a corrupt store aside as <path>.corrupt, rotating any
+// earlier backup to .corrupt.1 so repeat corruption never destroys the
+// previous quarantined evidence (POSIX rename replaces its destination).
+func quarantine(path string) error {
+	dst := path + ".corrupt"
+	if _, err := os.Stat(dst); err == nil {
+		if err := os.Rename(dst, dst+".1"); err != nil {
+			return err
+		}
+	}
+	return os.Rename(path, dst)
+}
+
 // warnPersist reports a failed best-effort save without aborting the
 // operation: the store stays a working in-memory cache, but a silent failure
 // would break session resume with no diagnostic.

internal/tokens/tokens_merge_test.go

@@ -0,0 +1,78 @@
+package tokens
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// Regression: persistLocked rewrote the whole map from the in-memory
+// snapshot without re-reading disk, so two concurrent bodek instances
+// erased each other's minted session tokens (B's token vanished when A's
+// next Set persisted its stale snapshot).
+func TestSetMergesForeignTokens(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "sessions.json")
+	a := openAt(path)
+	b := openAt(path)
+
+	a.Set("sess-a", "tok-a") // instance A mints and persists
+	b.Set("sess-b", "tok-b") // instance B, opened before A's write
+
+	// A fresh store must see both tokens.
+	c := openAt(path)
+	if got := c.Get("sess-a"); got != "tok-a" {
+		t.Fatalf("foreign token lost: Get(sess-a) = %q", got)
+	}
+	if got := c.Get("sess-b"); got != "tok-b" {
+		t.Fatalf("own token lost: Get(sess-b) = %q", got)
+	}
+}
+
+// Regression: a stale peer must not resurrect tokens another instance
+// deleted. B (opened before A's delete) writes an unrelated token — the
+// wholesale disk adoption must keep the deletion converged.
+func TestStalePeerKeepsPeerDeletions(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "sessions.json")
+	a := openAt(path)
+	b := openAt(path)
+
+	a.Set("sess-x", "tok-x") // both instances now know sess-x
+	a.Delete("sess-x")       // A deletes it (persists the deletion)
+
+	b.Set("sess-y", "tok-y") // stale B persists an unrelated write
+
+	c := openAt(path)
+	if got := c.Get("sess-x"); got != "" {
+		t.Fatalf("deleted token resurrected by stale peer: %q", got)
+	}
+	if got := c.Get("sess-y"); got != "tok-y" {
+		t.Fatalf("unrelated token lost: %q", got)
+	}
+}
+
+// Regression: the .corrupt quarantine used a fixed name, so a second
+// corrupting write replaced the first quarantined evidence (POSIX rename
+// replaces its destination) — the earlier snapshot became undiagnosable.
+func TestQuarantineRotatesBackups(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "sessions.json")
+
+	if err := os.WriteFile(path, []byte("{first"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	openAt(path)
+	first, err := os.ReadFile(path + ".corrupt")
+	if err != nil || string(first) != "{first" {
+		t.Fatalf("first quarantine missing: %q err=%v", first, err)
+	}
+
+	if err := os.WriteFile(path, []byte("{second"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	openAt(path)
+	if got, _ := os.ReadFile(path + ".corrupt"); string(got) != "{second" {
+		t.Fatalf("latest quarantine wrong: %q", got)
+	}
+	if got, err := os.ReadFile(path + ".corrupt.1"); err != nil || string(got) != "{first" {
+		t.Fatalf("first quarantine was clobbered: %q err=%v", got, err)
+	}
+}

internal/tui/clear_approval_teardown_test.go

@@ -0,0 +1,50 @@
+package tui
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/BackendStack21/bodek/internal/client"
+)
+
+// Regression: the errMsg teardown cleared approvals and deadlines but left
+// apprBells armed, breaking the documented lockstep invariant — the next
+// approval popped a stale bell latch.
+func TestErrMsgTeardownKeepsBellLockstep(t *testing.T) {
+	m := newTestModel()
+	feedApproval(t, m, client.Event{Type: "approval_request", ID: "apr", Risk: "shell_exec", Command: "rm x"})
+	if len(m.apprBells) != len(m.apprDeadlines) {
+		t.Fatalf("precondition: bells/deadlines out of lockstep: %d/%d",
+			len(m.apprBells), len(m.apprDeadlines))
+	}
+
+	m.Update(errMsg{err: errors.New("write failed")})
+
+	if len(m.approvals) != 0 || len(m.apprDeadlines) != 0 {
+		t.Fatalf("teardown left approvals armed: %d/%d", len(m.approvals), len(m.apprDeadlines))
+	}
+	if len(m.apprBells) != 0 {
+		t.Fatalf("teardown left bells armed (lockstep broken): %d", len(m.apprBells))
+	}
+}
+
+// Regression: /clear mid-turn wiped the transcript but left the approval
+// queue, deadlines, bells, and clarify armed over a dead request — the
+// keyboard stayed gated by a request that no longer had a turn behind it.
+// Every other teardown path (done/error/disconnect) clears them explicitly.
+func TestClearConversationDropsPendingApprovals(t *testing.T) {
+	m := newTestModel()
+	feedApproval(t, m, client.Event{Type: "approval_request", ID: "apr", Risk: "shell_exec", Command: "rm x"})
+	m.apprDeadlines[0] = time.Now().Add(time.Minute)
+
+	m.clearConversation()
+
+	if len(m.approvals) != 0 || len(m.apprDeadlines) != 0 || len(m.apprBells) != 0 {
+		t.Fatalf("/clear left approvals armed: appr=%d dl=%d bells=%d",
+			len(m.approvals), len(m.apprDeadlines), len(m.apprBells))
+	}
+	if m.clarify != nil || m.clarifyBuf != "" {
+		t.Fatalf("/clear left clarify armed: q=%v buf=%q", m.clarify != nil, m.clarifyBuf)
+	}
+}

internal/tui/model.go

@@ -578,6 +578,7 @@ func (m *Model) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 		// the keyboard after busy is already false.
 		m.approvals = nil
 		m.apprDeadlines = nil
+		m.apprBells = nil // lockstep with apprDeadlines — a stale latch must not leak
 		m.resetApprovalInput()
 		m.clearClarify()
 		m.relayout() // the busy status line releases its row
@@ -1189,6 +1190,14 @@ func (m *Model) handleKey(msg tea.KeyMsg) (tea.Model, tea.Cmd) {
 func (m *Model) clearConversation() tea.Cmd {
 	m.inspect = nil
 	m.focusIdx = -1 // stale anchor would copy/move against the regrown transcript
+	// Pending approvals and clarify die with the conversation — leaving them
+	// armed captures the keyboard over a request with no turn behind it
+	// (the same contract done/error/disconnect document).
+	m.approvals = nil
+	m.apprDeadlines = nil
+	m.apprBells = nil
+	m.resetApprovalInput()
+	m.clearClarify()
 	captureHome(m)
 	m.msgs = nil
 	m.curIdx = -1