Codex mid-turn delivery deletes input before the steer is accepted

TODO.md

@@ -1,7 +1,5 @@
 # Bugs
 
-- Codex mid-turn delivery deletes input before the steer is accepted
-  Queued inputs are deleted or marked delivered and recorded as `user.message` (`internal/agents/queue.go:119`, `:158`; `internal/agents/inbox.go:404`) before `Steer`, whose error is only logged (`internal/agents/codex.go:167–174`). Return the consumed ids from `deliverMidTurn`; when Steer fails, re-insert the inputs in order and clear `delivered_at` so the turn-end Deliver sends them.
 - 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

docs/operations.md

@@ -227,7 +227,7 @@ Agents ask questions with the `ask_human` MCP tool. Each question carries a `con
 | `other` (free text in `text`: an answer, a question back, or new instructions) | questions, errors | `answered` |
 | `dismiss` | errors | `dismissed` |
 
-Answers reach the agent as one message holding every undelivered `option` and `other` response of the session (`deliveryMessage` in `internal/agents/inbox.go`), never while one of its questions is still open and unanswered (`AwaitingHuman` in `internal/store/inbox.go`); until then they stay held (`inbox_responses.delivered_at` empty). While a turn is running, the task session's PostToolUse hook (`afterTool` in `internal/agents/snapshots.go`, through `deliverMidTurn`) hands them over after the agent's next tool call as the hook's `additionalContext`, after any queued messages (see "Agent sessions"), so the agent reads them alongside that tool's result without being interrupted (for Codex, after each completed tool item, `deliverMidTurn` hands them over with `turn/steer` instead); the `user.message` event recording it carries `mid_turn: true` and no `head`, and the transcript doesn't treat it as a turn start. A turn that makes no further tool call leaves them to `Deliver`, which runs after each submission and when a turn ends, and sends them as a new turn once the session is `idle`. The log traces each answer: `inbox response recorded` when it is stored, `deliver held` with the reason, `deliver sent mid-turn`, `deliver sent` with the session's state and whether its agent process had exited or byre was closing, and `undelivered responses` with a reason wherever a session has held answers when its turn ends in a handoff or merge, its agent process ends, startup resumes it, or startup finds it idle. `sqlite3 byre.db "SELECT * FROM inbox_responses WHERE delivered_at IS NULL"` lists the answers held now.
+Answers reach the agent as one message holding every undelivered `option` and `other` response of the session (`deliveryMessage` in `internal/agents/inbox.go`), never while one of its questions is still open and unanswered (`AwaitingHuman` in `internal/store/inbox.go`); until then they stay held (`inbox_responses.delivered_at` empty). While a turn is running, the task session's PostToolUse hook (`afterTool` in `internal/agents/snapshots.go`, through `deliverMidTurn`) hands them over after the agent's next tool call as the hook's `additionalContext`, after any queued messages (see "Agent sessions"), so the agent reads them alongside that tool's result without being interrupted (for Codex, after each completed tool item, `deliverMidTurn` hands them over with `turn/steer` instead; when the steer fails, `restoreMidTurn` puts the taken queued messages back under their original ids and clears the answers' `delivered_at`, so `Deliver` sends them when the turn ends, and logs `mid-turn delivery restored`); the `user.message` event recording it carries `mid_turn: true` and no `head`, and the transcript doesn't treat it as a turn start. A turn that makes no further tool call leaves them to `Deliver`, which runs after each submission and when a turn ends, and sends them as a new turn once the session is `idle`. The log traces each answer: `inbox response recorded` when it is stored, `deliver held` with the reason, `deliver sent mid-turn`, `deliver sent` with the session's state and whether its agent process had exited or byre was closing, and `undelivered responses` with a reason wherever a session has held answers when its turn ends in a handoff or merge, its agent process ends, startup resumes it, or startup finds it idle. `sqlite3 byre.db "SELECT * FROM inbox_responses WHERE delivered_at IS NULL"` lists the answers held now.
 
 ## Notifications
 

internal/agents/codex.go

@@ -164,13 +164,14 @@ func (m *Manager) codexItemCompleted(ctx context.Context, l *live, s *codex.Sess
 	if l.reviewID != "" {
 		return nil
 	}
-	text := m.deliverMidTurn(ctx, l)
+	mt := m.deliverMidTurn(ctx, l)
-	if text == "" {
+	if mt.text == "" {
 		return nil
 	}
-	err = s.Steer(ctx, text)
+	err = s.Steer(ctx, mt.text)
 	if err != nil {
 		slog.Error("steer mid-turn delivery", "session", l.id, "err", err)
+		m.restoreMidTurn(ctx, l, mt)
 	}
 	return nil
 }

internal/agents/inbox.go

@@ -350,13 +350,24 @@ func (m *Manager) Deliver(ctx context.Context, sessionID string) error {
 	return m.store.MarkDelivered(ctx, ids)
 }
 
-func (m *Manager) deliverMidTurn(ctx context.Context, l *live) string {
+type midTurn struct {
+	text       string
+	taskInputs []store.TaskInput
+	leadInputs []store.LeadInput
+	responses  []string
+}
+
+func (m *Manager) deliverMidTurn(ctx context.Context, l *live) midTurn {
 	if l.owner != "" {
-		msgs := m.takeLeadQueuedMidTurn(ctx, l)
+		inputs := m.takeLeadQueuedMidTurn(ctx, l)
-		if len(msgs) == 0 {
+		if len(inputs) == 0 {
-			return ""
+			return midTurn{}
 		}
-		return m.recordQueued(ctx, l, msgs)
+		msgs := []queued{}
+		for _, in := range inputs {
+			msgs = append(msgs, queued{actor: in.Actor, text: in.Text})
+		}
+		return midTurn{text: m.recordQueued(ctx, l, msgs), leadInputs: inputs}
 	}
 
 	unlock := m.lockDelivery(l.id)
@@ -365,49 +376,78 @@ func (m *Manager) deliverMidTurn(ctx context.Context, l *live) string {
 	discard, err := m.store.GetDiscard(ctx, l.taskID)
 	if err != nil {
 		slog.Error("read discard state", "task", l.taskID, "err", err)
-		return ""
+		return midTurn{}
 	}
 	if discard.State != store.DiscardNone {
-		return ""
+		return midTurn{}
 	}
 
+	mt := midTurn{}
 	parts := []string{}
-	msgs := m.takeQueuedMidTurn(ctx, l)
+	mt.taskInputs = m.takeQueuedMidTurn(ctx, l)
-	if len(msgs) > 0 {
+	if len(mt.taskInputs) > 0 {
+		msgs := []queued{}
+		for _, in := range mt.taskInputs {
+			msgs = append(msgs, queued{actor: in.Actor, text: in.Text})
+		}
 		parts = append(parts, m.recordQueued(ctx, l, msgs))
 	}
-	responses := m.responsesMidTurn(ctx, l)
+	responses, ids := m.responsesMidTurn(ctx, l)
 	if responses != "" {
 		parts = append(parts, responses)
+		mt.responses = ids
 	}
-	return strings.Join(parts, "\n\n")
+	mt.text = strings.Join(parts, "\n\n")
+	return mt
 }
 
-func (m *Manager) responsesMidTurn(ctx context.Context, l *live) string {
+func (m *Manager) restoreMidTurn(ctx context.Context, l *live, mt midTurn) {
+	if len(mt.taskInputs) > 0 {
+		err := m.store.RestoreTaskInputs(ctx, mt.taskInputs)
+		if err != nil {
+			slog.Error("restore queued messages", "session", l.id, "err", err)
+		}
+	}
+	if len(mt.leadInputs) > 0 {
+		err := m.store.RestoreLeadInputs(ctx, mt.leadInputs)
+		if err != nil {
+			slog.Error("restore held lead messages", "session", l.id, "err", err)
+		}
+	}
+	if len(mt.responses) > 0 {
+		err := m.store.MarkUndelivered(ctx, mt.responses)
+		if err != nil {
+			slog.Error("mark responses undelivered", "session", l.id, "err", err, "responses", mt.responses)
+		}
+	}
+	slog.Info("mid-turn delivery restored", "session", l.id, "inputs", len(mt.taskInputs)+len(mt.leadInputs), "responses", mt.responses)
+}
+
+func (m *Manager) responsesMidTurn(ctx context.Context, l *live) (string, []string) {
 	pending, err := m.store.Undelivered(ctx, l.id)
 	if err != nil {
 		slog.Error("list undelivered responses", "session", l.id, "err", err)
-		return ""
+		return "", nil
 	}
 	if len(pending) == 0 {
-		return ""
+		return "", nil
 	}
 	ids := responseIDs(pending)
 
 	waiting, err := m.store.AwaitingHuman(ctx, l.id)
 	if err != nil {
 		slog.Error("count waiting questions", "session", l.id, "err", err)
-		return ""
+		return "", nil
 	}
 	if waiting > 0 {
 		slog.Info("deliver held mid-turn: question waiting", "session", l.id, "waiting", waiting, "responses", ids)
-		return ""
+		return "", nil
 	}
 
 	err = m.store.MarkDelivered(ctx, ids)
 	if err != nil {
 		slog.Error("mark responses delivered", "session", l.id, "err", err, "responses", ids)
-		return ""
+		return "", nil
 	}
 
 	text := deliveryMessage(pending)
@@ -416,7 +456,7 @@ func (m *Manager) responsesMidTurn(ctx context.Context, l *live) string {
 		slog.Error("record mid-turn delivery", "session", l.id, "err", err)
 	}
 	slog.Info("deliver sent mid-turn", "session", l.id, "responses", ids)
-	return text
+	return text, ids
 }
 
 func responseIDs(pending []store.Pending) []string {

internal/agents/queue.go

@@ -119,7 +119,7 @@ type queued struct {
 	text  string
 }
 
-func (m *Manager) takeQueuedMidTurn(ctx context.Context, l *live) []queued {
+func (m *Manager) takeQueuedMidTurn(ctx context.Context, l *live) []store.TaskInput {
 	m.mu.Lock()
 	compacting := l.compacting
 	m.mu.Unlock()
@@ -132,13 +132,13 @@ func (m *Manager) takeQueuedMidTurn(ctx context.Context, l *live) []queued {
 		slog.Error("list queued messages", "session", l.id, "err", err)
 		return nil
 	}
-	out := []queued{}
+	out := []store.TaskInput{}
 	ids := []int64{}
 	for _, in := range inputs {
 		if len(in.Images) > 0 {
 			break
 		}
-		out = append(out, queued{actor: in.Actor, text: in.Text})
+		out = append(out, in)
 		ids = append(ids, in.ID)
 	}
 	if len(ids) == 0 {
@@ -152,7 +152,7 @@ func (m *Manager) takeQueuedMidTurn(ctx context.Context, l *live) []queued {
 	return out
 }
 
-func (m *Manager) takeLeadQueuedMidTurn(ctx context.Context, l *live) []queued {
+func (m *Manager) takeLeadQueuedMidTurn(ctx context.Context, l *live) []store.LeadInput {
 	m.mu.Lock()
 	compacting := l.compacting
 	m.mu.Unlock()
@@ -168,7 +168,7 @@ func (m *Manager) takeLeadQueuedMidTurn(ctx context.Context, l *live) []queued {
 		slog.Error("list held lead messages", "session", l.id, "err", err)
 		return nil
 	}
-	out := []queued{}
+	out := []store.LeadInput{}
 	ids := []int64{}
 	for _, in := range inputs {
 		if in.Kind != store.LeadMessage {
@@ -177,7 +177,7 @@ func (m *Manager) takeLeadQueuedMidTurn(ctx context.Context, l *live) []queued {
 		if len(in.Images) > 0 {
 			break
 		}
-		out = append(out, queued{actor: in.Actor, text: in.Text})
+		out = append(out, in)
 		ids = append(ids, in.ID)
 	}
 	if len(ids) == 0 {

internal/agents/snapshots.go

@@ -168,14 +168,14 @@ func (m *Manager) dropSnapshots(repo, taskID string) error {
 
 func (m *Manager) afterTool(l *live) claude.HookFunc {
 	return func(ctx context.Context, input json.RawMessage) (map[string]any, error) {
-		text := m.deliverMidTurn(ctx, l)
+		mt := m.deliverMidTurn(ctx, l)
-		if text == "" {
+		if mt.text == "" {
 			return map[string]any{}, nil
 		}
 		return map[string]any{
 			"hookSpecificOutput": map[string]any{
 				"hookEventName":     "PostToolUse",
-				"additionalContext": text,
+				"additionalContext": mt.text,
 			},
 		}, nil
 	}

internal/store/inbox.go

@@ -274,6 +274,18 @@ func (s *Store) MarkDelivered(ctx context.Context, responseIDs []string) error {
 	})
 }
 
+func (s *Store) MarkUndelivered(ctx context.Context, responseIDs []string) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		for _, id := range responseIDs {
+			_, err := tx.ExecContext(ctx, `UPDATE inbox_responses SET delivered_at = NULL WHERE id = ?`, id)
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 type querier interface {
 	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
 	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)

internal/store/lead.go

@@ -70,6 +70,23 @@ func (s *Store) LeadInputs(ctx context.Context, owner string) ([]LeadInput, erro
 	return inputs, rows.Err()
 }
 
+func (s *Store) RestoreLeadInputs(ctx context.Context, inputs []LeadInput) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		for _, in := range inputs {
+			images, err := json.Marshal(append([]string{}, in.Images...))
+			if err != nil {
+				return err
+			}
+			_, err = tx.ExecContext(ctx, `INSERT INTO lead_inputs (id, owner, kind, actor, text, images, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
+				in.ID, in.Owner, string(in.Kind), in.Actor, in.Text, string(images), formatTime(in.CreatedAt))
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 func (s *Store) DeleteLeadInputs(ctx context.Context, ids []int64) error {
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		for _, id := range ids {

internal/store/taskinputs.go

@@ -57,6 +57,23 @@ func (s *Store) TaskInputs(ctx context.Context, taskID string) ([]TaskInput, err
 	return inputs, rows.Err()
 }
 
+func (s *Store) RestoreTaskInputs(ctx context.Context, inputs []TaskInput) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		for _, in := range inputs {
+			images, err := json.Marshal(append([]string{}, in.Images...))
+			if err != nil {
+				return err
+			}
+			_, err = tx.ExecContext(ctx, `INSERT INTO task_inputs (id, task_id, actor, text, images, created_at) VALUES (?, ?, ?, ?, ?, ?)`,
+				in.ID, in.TaskID, in.Actor, in.Text, string(images), formatTime(in.CreatedAt))
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 func (s *Store) DeleteTaskInputs(ctx context.Context, ids []int64) error {
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		for _, id := range ids {