Simplify codebase per audit of 7aff773

internal/codex/session.go

@@ -1,8 +1,6 @@
 package codex
 
 import (
-	"bufio"
-	"bytes"
 	"context"
 	"crypto/rand"
 	"encoding/base64"
@@ -21,6 +19,7 @@ import (
 
 	"g.fc.run/ai/byre/internal/agentinput"
 	"g.fc.run/ai/byre/internal/hookcmd"
+	"g.fc.run/ai/byre/internal/jsonlines"
 	"g.fc.run/ai/byre/internal/mcp"
 	"g.fc.run/ai/byre/internal/procgroup"
 )
@@ -93,8 +92,7 @@ type Session struct {
 	reminder string
 	hookSrv  *http.Server
 
-	onBadLine func(line []byte, reason string)
-	lastLine  atomic.Int64
+	*jsonlines.Stream
 }
 
 func Start(ctx context.Context, opts Options) (*Session, error) {
@@ -214,12 +212,11 @@ func newSession(opts Options, stdin *os.File, stdout io.Reader) *Session {
 		pending:   map[int64]chan rpcResult{},
 		servers:   map[string]*mcp.Server{},
 		reminder:  opts.Reminder,
-		onBadLine: opts.OnBadLine,
+		Stream:    jsonlines.New("codex", opts.OnBadLine),
 	}
 	for _, srv := range opts.MCPServers {
 		s.servers[srv.Name] = srv
 	}
-	s.lastLine.Store(time.Now().UnixNano())
 	go s.readLoop(stdout)
 	return s
 }
@@ -341,10 +338,6 @@ func (s *Session) Pid() int {
 	return s.cmd.Process.Pid
 }
 
-func (s *Session) LastLine() time.Time {
-	return time.Unix(0, s.lastLine.Load())
-}
-
 func (s *Session) Exited() bool {
 	select {
 	case <-s.done:
@@ -616,25 +609,7 @@ func (s *Session) readLoop(stdout io.Reader) {
 	defer close(s.done)
 	defer close(s.events)
 
-	r := bufio.NewReader(stdout)
-	for {
-		line, err := r.ReadBytes('\n')
-		line = bytes.TrimRight(line, "\r\n")
-		if len(line) > 0 {
-			s.lastLine.Store(time.Now().UnixNano())
-		}
-		if len(line) > 0 && line[0] == '{' {
-			s.handleLine(line)
-		} else if len(line) > 0 {
-			s.badLine(line, "not a JSON object")
-		}
-		if err != nil {
-			if err != io.EOF {
-				slog.Error("read codex stdout", "err", err)
-			}
-			return
-		}
-	}
+	s.Read(stdout, s.handleLine)
 }
 
 func (s *Session) handleLine(line []byte) {
@@ -642,7 +617,7 @@ func (s *Session) handleLine(line []byte) {
 	m := message{}
 	err := json.Unmarshal(line, &m)
 	if err != nil {
-		s.badLine(line, err.Error())
+		s.BadLine(line, err.Error())
 		return
 	}
 
@@ -657,7 +632,7 @@ func (s *Session) handleLine(line []byte) {
 	}
 
 	if _, ok := notificationMethods[m.Method]; !ok {
-		s.badLine(line, fmt.Sprintf("unrecognized method %q", m.Method))
+		s.BadLine(line, fmt.Sprintf("unrecognized method %q", m.Method))
 	}
 
 	s.trackTurn(m)
@@ -688,7 +663,7 @@ func (s *Session) trackTurn(m message) {
 func (s *Session) handleResponse(line []byte, m message) {
 	id, err := strconv.ParseInt(string(m.ID), 10, 64)
 	if err != nil {
-		s.badLine(line, "response with a non-numeric id")
+		s.BadLine(line, "response with a non-numeric id")
 		return
 	}
 
@@ -763,10 +738,3 @@ func (s *Session) callTool(raw json.RawMessage) (any, error) {
 	}
 	return DynamicToolCallResponse{Success: !res.IsError, ContentItems: items}, nil
 }
-
-func (s *Session) badLine(line []byte, reason string) {
-	slog.Warn("bad codex line", "reason", reason, "line", string(line))
-	if s.onBadLine != nil {
-		s.onBadLine(line, reason)
-	}
-}