From 8c7c3bee2ea95d346088bc66aa286cad703daa9f Mon Sep 17 00:00:00 2001 From: Ian Gulliver Date: Fri, 16 Oct 2026 14:18:40 -0700 Subject: [PATCH] Codex mid-turn delivery deletes input before the steer is accepted --- TODO.md | 2 - docs/operations.md | 2 +- internal/agents/codex.go | 7 ++-- internal/agents/inbox.go | 76 +++++++++++++++++++++++++++--------- internal/agents/queue.go | 12 +++--- internal/agents/snapshots.go | 6 +-- internal/store/inbox.go | 12 ++++++ internal/store/lead.go | 17 ++++++++ internal/store/taskinputs.go | 17 ++++++++ 9 files changed, 118 insertions(+), 33 deletions(-) diff --git a/TODO.md b/TODO.md index 5b23591..856912f 100644 --- a/TODO.md +++ b/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 diff --git a/docs/operations.md b/docs/operations.md index a9cb839..91e6fee 100644 --- a/docs/operations.md +++ b/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 diff --git a/internal/agents/codex.go b/internal/agents/codex.go index 0bc9ba2..79fb695 100644 --- a/internal/agents/codex.go +++ b/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) - if text == "" { + mt := m.deliverMidTurn(ctx, l) + 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 } diff --git a/internal/agents/inbox.go b/internal/agents/inbox.go index e3c0507..45b02fc 100644 --- a/internal/agents/inbox.go +++ b/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) - if len(msgs) == 0 { - return "" + inputs := m.takeLeadQueuedMidTurn(ctx, l) + if len(inputs) == 0 { + 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) - if len(msgs) > 0 { + mt.taskInputs = m.takeQueuedMidTurn(ctx, l) + 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 { diff --git a/internal/agents/queue.go b/internal/agents/queue.go index df7996c..248c136 100644 --- a/internal/agents/queue.go +++ b/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 { diff --git a/internal/agents/snapshots.go b/internal/agents/snapshots.go index 2397ccf..9a31f96 100644 --- a/internal/agents/snapshots.go +++ b/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) - if text == "" { + mt := m.deliverMidTurn(ctx, l) + if mt.text == "" { return map[string]any{}, nil } return map[string]any{ "hookSpecificOutput": map[string]any{ "hookEventName": "PostToolUse", - "additionalContext": text, + "additionalContext": mt.text, }, }, nil } diff --git a/internal/store/inbox.go b/internal/store/inbox.go index 09c33df..17b78df 100644 --- a/internal/store/inbox.go +++ b/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) diff --git a/internal/store/lead.go b/internal/store/lead.go index 3111516..fcbd7fc 100644 --- a/internal/store/lead.go +++ b/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 { diff --git a/internal/store/taskinputs.go b/internal/store/taskinputs.go index 05cb7d8..881af97 100644 --- a/internal/store/taskinputs.go +++ b/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 {