A Codex RPC can deadlock the reader it waits on

TODO.md

@@ -2,8 +2,6 @@
 
 - Inbox answer validation isn't atomic with resolving the question
   `CheckResponse` reads open state outside the write transaction and `Respond` updates by id alone (`internal/agents/respond.go:26–34`, `:93`; `internal/store/inbox.go:140`), so two concurrent answers both land, or a withdrawal is overwritten. Condition the update on `state='open'`, check RowsAffected, roll back the batch on conflict and return 409.
-- A Codex RPC can deadlock the reader it waits on
-  `drive` calls `Steer` synchronously while `readLoop` pushes notifications into a 256-entry channel (`internal/codex/session.go:186`, `:440–476`, `:598`); if it fills before the response is read, each waits on the other. Route RPC responses separately from notification consumption and give RPCs bounded deadlines, without replacing backpressure with an unbounded queue.
 - A malformed file image URL breaks transcript rendering
   `decodeURI` throws on e.g. `file:///tmp/100%.png` (`web/src/markdown.ts:347`), and `renderAll` appends events before mounting the composer (`web/src/task.ts:608–609`), so the rest of the transcript and the composer don't render. Catch the decode error and render the plain fallback link.
 - Stale reload and focus responses overwrite newer state

docs/operations.md

@@ -9,7 +9,7 @@
 - `internal/git`: worktrees, the task changes view (see "Changes"), snapshots (see "Snapshots"), unmerged-change summaries, squash merges, merge notes and pushes.
 - `internal/agents`: the agent session manager. It owns the running agent processes, task agents, lead agents (see "Lead agent") and reviewers (see "Reviews") alike, and records their activity as events.
 - `internal/claude`: the Claude Code stream-json client (see `docs/claude-protocol.md`).
-- `internal/codex`: the `codex app-server` JSON-RPC client, with types generated from the CLI's schema by `go generate ./internal/codex` (see `docs/codex-protocol.md`).
+- `internal/codex`: the `codex app-server` JSON-RPC client, with types generated from the CLI's schema by `go generate ./internal/codex` (see `docs/codex-protocol.md`). One goroutine reads the CLI's stdout and blocks on the bounded `Events` channel when its consumer falls behind, so responses queued behind those notifications wait too: the goroutine draining `Events` (`codexProcess.drive`) therefore never waits on a reply itself. It writes `turn/steer` in order and waits for the reply on another goroutine, and starts the prose-question nudge and the turn-end `Deliver` on their own goroutines. Every request gives up after `rpcTimeout`, or the caller's earlier deadline, and that deadline covers waiting for the write lock, writing the request to stdin and waiting for the reply. Stdin is an `os.Pipe`, so writes take a deadline. A request whose deadline passes before any of it is written fails and leaves the CLI running. One that times out after part of its line is written has broken the framing, so byre kills the CLI, logging `codex stopped reading its stdin mid-line`, and the session ends as any exited process does.
 - `internal/tailscale`: `tailscale serve` for task previews (see "Previews").
 - `internal/workflow`: validating workflow definitions and reading and writing their YAML files (see "Workflows").
 - `web/`: the SPA, frameworkless TypeScript compiled file-by-file to ES modules (no bundler). `web/src/*.ts` is the source; `web/src/api.gen.ts` holds types generated from `api/openapi.yaml`, and `web/src/api.ts` is the typed client over them. `go generate ./web` runs `npm ci`, regenerates the types and compiles `web/dist/`, which is committed so `go install` needs no Node. The binary embeds `index.html`, `style.css`, `favicon.svg` (a lead dot in the accent colour with spokes to five task-coloured dots; its colours are saturated versions of the `web/style.css` accent and task hues, not copies of the palette, since the palette's pastels blur together at 16px; they switch with the light/dark scheme), `favicon.png` (the same icon in its dark colours on an opaque dark `--surface` square at 180×180, rendered from `web/favicon-tile.svg`, which is not embedded; opaque because Orion puts a transparent icon on a white tile; linked as a second icon and as the `apple-touch-icon`, which Orion prefers over the SVG for its tab icon; Orion's icon store (`~/Library/Application Support/Orion/Defaults/website_icons`, kept apart from its cache) keys icons by hostname without the port, so another server on the same hostname can supply byre's tab icon; `favicon-tile.svg` repeats the shapes and dark colours of `favicon.svg`, so change both together, then regenerate the PNG with `rsvg-convert --width 180 --height 180 --output web/favicon.png web/favicon-tile.svg`), `dist/`, `src/` (for source maps), `fonts/` and `vendor/` (Mermaid, see "Inbox"; xterm.js, see "Terminals"). `byre serve --dev` serves `./web` from disk instead (run from the repo root); pair it with `npm --prefix web exec tsc -- --watch`.

internal/agents/codex.go

@@ -168,11 +168,20 @@ func (m *Manager) codexItemCompleted(ctx context.Context, l *live, s *codex.Sess
 	if mt.text == "" {
 		return nil
 	}
-	err = s.Steer(ctx, mt.text)
+	reply, err := s.Steer(mt.text)
 	if err != nil {
 		slog.Error("steer mid-turn delivery", "session", l.id, "err", err)
 		m.restoreMidTurn(ctx, l, mt)
+		return nil
 	}
+	go func() {
+		ctx := context.Background()
+		err := reply.Wait(ctx, nil)
+		if err != nil {
+			slog.Error("steer mid-turn delivery", "session", l.id, "err", err)
+			m.restoreMidTurn(ctx, l, mt)
+		}
+	}()
 	return nil
 }
 
@@ -258,8 +267,13 @@ func (m *Manager) codexTurnCompleted(ctx context.Context, l *live, s *codex.Sess
 		m.mu.Lock()
 		l.lastText = ""
 		m.mu.Unlock()
-		_, err = s.Send(ctx, proseQuestionNudge)
-		return err
+		go func() {
+			_, err := s.Send(context.Background(), proseQuestionNudge)
+			if err != nil {
+				slog.Error("send prose question nudge", "session", l.id, "err", err)
+			}
+		}()
+		return nil
 	}
 
 	res := endResult{

internal/agents/manager.go

@@ -1056,7 +1056,7 @@ func (m *Manager) turnEnded(ctx context.Context, l *live, res endResult) error {
 		go m.advanceStage(l)
 		return nil
 	}
-	m.afterTurn(ctx, l)
+	go m.afterTurn(context.WithoutCancel(ctx), l)
 	return nil
 }
 

internal/codex/session.go

@@ -90,11 +90,11 @@ type rpcResult struct {
 
 type Session struct {
 	cmd    *exec.Cmd
-	stdin  io.WriteCloser
+	stdin  *os.File
 	events chan Event
 	done   chan struct{}
 
-	writeMu    sync.Mutex
+	writeLock  chan struct{}
 	recordMu   sync.Mutex
 	transcript io.Writer
 
@@ -160,18 +160,23 @@ func spawn(opts Options) (*Session, error) {
 	cmd.Stderr = os.Stderr
 	cmd.Env = buildEnv(opts)
 
-	stdin, err := cmd.StdinPipe()
+	stdinR, stdin, err := os.Pipe()
 	if err != nil {
 		return nil, err
 	}
+	cmd.Stdin = stdinR
 
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
+		stdinR.Close()
+		stdin.Close()
 		return nil, err
 	}
 
 	err = cmd.Start()
+	stdinR.Close()
 	if err != nil {
+		stdin.Close()
 		return nil, fmt.Errorf("start codex: %w", err)
 	}
 
@@ -180,10 +185,11 @@ func spawn(opts Options) (*Session, error) {
 	return s, nil
 }
 
-func newSession(opts Options, stdin io.WriteCloser, stdout io.Reader) *Session {
+func newSession(opts Options, stdin *os.File, stdout io.Reader) *Session {
 	s := &Session{
 		stdin:      stdin,
 		events:     make(chan Event, 256),
+		writeLock:  make(chan struct{}, 1),
 		done:       make(chan struct{}),
 		transcript: opts.Transcript,
 		pending:    map[int64]chan rpcResult{},
@@ -421,47 +427,81 @@ func (s *Session) Interrupt(ctx context.Context) error {
 	return s.Call(ctx, MethodTurnInterrupt, params, nil)
 }
 
-func (s *Session) Steer(ctx context.Context, text string) error {
+func (s *Session) Steer(text string) (*Reply, error) {
 	s.stateMu.Lock()
 	params := TurnSteerParams{ThreadID: s.threadID, ExpectedTurnID: s.turnID, Input: []UserInput{{Type: UserInputTypeText, Text: text}}}
 	s.stateMu.Unlock()
 
 	if params.ExpectedTurnID == "" {
-		return errors.New("no turn in progress")
+		return nil, errors.New("no turn in progress")
 	}
-	return s.Call(ctx, MethodTurnSteer, params, nil)
+	return s.request(context.Background(), MethodTurnSteer, params)
 }
 
 func (s *Session) Compact(ctx context.Context) error {
 	return s.Call(ctx, MethodThreadCompactStart, ThreadCompactStartParams{ThreadID: s.ID()}, nil)
 }
 
+const rpcTimeout = time.Minute
+
+type Reply struct {
+	s        *Session
+	id       int64
+	ch       chan rpcResult
+	deadline time.Time
+}
+
 func (s *Session) Call(ctx context.Context, method string, params any, result any) error {
-	id := s.nextID.Add(1)
-	ch := make(chan rpcResult, 1)
+	r, err := s.request(ctx, method, params)
+	if err != nil {
+		return err
+	}
+	return r.Wait(ctx, result)
+}
 
-	s.pendingMu.Lock()
-	s.pending[id] = ch
-	s.pendingMu.Unlock()
-
-	defer func() {
-		s.pendingMu.Lock()
-		delete(s.pending, id)
-		s.pendingMu.Unlock()
-	}()
+func rpcDeadline(ctx context.Context) time.Time {
+	deadline := time.Now().Add(rpcTimeout)
+	d, ok := ctx.Deadline()
+	if ok && d.Before(deadline) {
+		return d
+	}
+	return deadline
+}
 
+func (s *Session) request(ctx context.Context, method string, params any) (*Reply, error) {
 	raw, err := json.Marshal(params)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	err = s.write(message{ID: json.RawMessage(strconv.FormatInt(id, 10)), Method: method, Params: raw})
+	r := &Reply{s: s, id: s.nextID.Add(1), ch: make(chan rpcResult, 1), deadline: rpcDeadline(ctx)}
+
+	s.pendingMu.Lock()
+	s.pending[r.id] = r.ch
+	s.pendingMu.Unlock()
+
+	err = s.write(message{ID: json.RawMessage(strconv.FormatInt(r.id, 10)), Method: method, Params: raw}, r.deadline)
 	if err != nil {
-		return err
+		r.forget()
+		return nil, err
 	}
+	return r, nil
+}
+
+func (r *Reply) forget() {
+	r.s.pendingMu.Lock()
+	delete(r.s.pending, r.id)
+	r.s.pendingMu.Unlock()
+}
+
+func (r *Reply) Wait(ctx context.Context, result any) error {
+	defer r.forget()
+
+	ctx, cancel := context.WithDeadline(ctx, r.deadline)
+	defer cancel()
 
 	select {
-	case res := <-ch:
+	case res := <-r.ch:
 		if res.err != nil {
 			return res.err
 		}
@@ -469,7 +509,7 @@ func (s *Session) Call(ctx context.Context, method string, params any, result an
 			return nil
 		}
 		return json.Unmarshal(res.result, result)
-	case <-s.done:
+	case <-r.s.done:
 		return errors.New("codex exited")
 	case <-ctx.Done():
 		return ctx.Err()
@@ -485,13 +525,13 @@ func (s *Session) notify(method string, params any) error {
 		}
 		m.Params = raw
 	}
-	return s.write(m)
+	return s.write(m, time.Now().Add(rpcTimeout))
 }
 
 func (s *Session) Close() error {
-	s.writeMu.Lock()
+	s.writeLock <- struct{}{}
 	s.stdin.Close()
-	s.writeMu.Unlock()
+	<-s.writeLock
 
 	select {
 	case <-s.done:
@@ -509,18 +549,36 @@ func (s *Session) Kill() {
 	s.cmd.Wait()
 }
 
-func (s *Session) write(m message) error {
+func (s *Session) write(m message, deadline time.Time) error {
 	b, err := json.Marshal(m)
 	if err != nil {
 		return err
 	}
 
-	s.writeMu.Lock()
-	defer s.writeMu.Unlock()
+	wait := time.NewTimer(time.Until(deadline))
+	defer wait.Stop()
+	select {
+	case s.writeLock <- struct{}{}:
+	case <-wait.C:
+		return os.ErrDeadlineExceeded
+	}
+	defer func() { <-s.writeLock }()
+
+	if !time.Now().Before(deadline) {
+		return os.ErrDeadlineExceeded
+	}
 
 	s.record("send", b)
 
-	_, err = s.stdin.Write(append(b, '\n'))
+	err = s.stdin.SetWriteDeadline(deadline)
+	if err != nil {
+		return err
+	}
+	n, err := s.stdin.Write(append(b, '\n'))
+	if n > 0 && errors.Is(err, os.ErrDeadlineExceeded) {
+		slog.Error("codex stopped reading its stdin mid-line; killing it", "pid", s.cmd.Process.Pid, "written", n, "len", len(b)+1)
+		s.cmd.Process.Kill()
+	}
 	return err
 }
 
@@ -646,7 +704,7 @@ func (s *Session) handleServerRequest(m message) {
 	result, err := s.answerServerRequest(m)
 	if err != nil {
 		slog.Error("codex server request failed", "method", m.Method, "err", err)
-		s.write(message{ID: m.ID, Error: &RPCError{Code: -32601, Message: err.Error()}})
+		s.write(message{ID: m.ID, Error: &RPCError{Code: -32601, Message: err.Error()}}, time.Now().Add(rpcTimeout))
 		return
 	}
 
@@ -655,7 +713,7 @@ func (s *Session) handleServerRequest(m message) {
 		slog.Error("encode codex server response", "method", m.Method, "err", err)
 		return
 	}
-	s.write(message{ID: m.ID, Result: raw})
+	s.write(message{ID: m.ID, Result: raw}, time.Now().Add(rpcTimeout))
 }
 
 func (s *Session) answerServerRequest(m message) (any, error) {