A start failure after admission leaves a session running with no turn

This commit is contained in:
Ian Gulliver
2026-10-16 12:36:05 -07:00
parent 9e039712b1
commit acbbd8747e
2 changed files with 30 additions and 11 deletions
-2
View File
@@ -8,8 +8,6 @@
A `CommitAll` error is only logged (`internal/agents/manager.go:838–841`) and `FinishStage` still runs (`:885–889`), so `RecordStage` stores the old `git.Head` (`internal/agents/workflows.go:349–356`), the next stage branches without the turn's work, and `discardStages` later deletes the worktree that held it. On commit failure, treat the turn as unfinished: skip FinishStage, the handoff and the summary, and file the error in the task's inbox. A `CommitAll` error is only logged (`internal/agents/manager.go:838–841`) and `FinishStage` still runs (`:885–889`), so `RecordStage` stores the old `git.Head` (`internal/agents/workflows.go:349–356`), the next stage branches without the turn's work, and `discardStages` later deletes the worktree that held it. On commit failure, treat the turn as unfinished: skip FinishStage, the handoff and the summary, and file the error in the task's inbox.
- One bad session at startup stops byre from starting - One bad session at startup stops byre from starting
`Recover` returns on a `taskFor` or `message` error (`internal/agents/manager.go:300–303`, `:338–341`) and `server.New` treats it as fatal (`internal/server/server.go:72–76`), so one unreadable worktree keeps the service down. Handle a failed `message` like a failed `ensureLive` (logUndelivered, DropStageOutput, EndTurn with the error, discard and merge follow-up) and continue; a `taskFor` error skips only that session. `Recover` returns on a `taskFor` or `message` error (`internal/agents/manager.go:300–303`, `:338–341`) and `server.New` treats it as fatal (`internal/server/server.go:72–76`), so one unreadable worktree keeps the service down. Handle a failed `message` like a failed `ensureLive` (logUndelivered, DropStageOutput, EndTurn with the error, discard and merge follow-up) and continue; a `taskFor` error skips only that session.
- A start failure after admission leaves a session `running` with no turn
`launch` resets to idle only on `ensureLive`/`syncModel` errors (`internal/agents/manager.go:439–466`); a local `message` error with a live process (missing blob, failed `rev-parse HEAD`, failed event write) leaves it running, so later messages queue forever and Merge skips the task. When `message` fails before `send`, set the session idle with the error; a `send` failure is already handled by pump.
- Codex mid-turn delivery deletes input before the steer is accepted - 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. 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 - Inbox answer validation isn't atomic with resolving the question
+30 -9
View File
@@ -450,14 +450,17 @@ func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Ta
err = m.syncModel(ctx, l, sess.Model) err = m.syncModel(ctx, l, sess.Model)
} }
if err != nil { if err != nil {
serr := m.store.SetSessionState(ctx, sess.Id, api.AgentSessionStateIdle, actor, map[string]any{"error": err.Error()}) m.failStart(ctx, sess.Id, actor, err)
if serr != nil {
slog.Error("record failed start", "session", sess.Id, "err", serr)
}
return api.AgentSession{}, err return api.AgentSession{}, err
} }
err = m.message(ctx, l, actor, text, images) imgs, err := m.prepareMessage(ctx, l, actor, text, images)
if err != nil {
m.failStart(ctx, sess.Id, actor, err)
return api.AgentSession{}, err
}
err = l.sess.send(ctx, text, imgs)
if err != nil { if err != nil {
return api.AgentSession{}, err return api.AgentSession{}, err
} }
@@ -465,6 +468,16 @@ func (m *Manager) launch(ctx context.Context, sess api.AgentSession, task api.Ta
return m.store.GetSession(ctx, sess.Id) return m.store.GetSession(ctx, sess.Id)
} }
func (m *Manager) failStart(ctx context.Context, sessionID, actor string, err error) {
cctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer cancel()
serr := m.store.SetSessionState(cctx, sessionID, api.AgentSessionStateIdle, actor, map[string]any{"error": err.Error()})
if serr != nil {
slog.Error("record failed start", "session", sessionID, "err", serr)
}
}
func (m *Manager) syncModel(ctx context.Context, l *live, model string) error { func (m *Manager) syncModel(ctx context.Context, l *live, model string) error {
if model == l.model { if model == l.model {
return nil return nil
@@ -490,11 +503,19 @@ func transition(ctx context.Context, st *store.Store, id string, from, to api.Ag
} }
func (m *Manager) message(ctx context.Context, l *live, actor, text string, images []string) error { func (m *Manager) message(ctx context.Context, l *live, actor, text string, images []string) error {
imgs, err := m.prepareMessage(ctx, l, actor, text, images)
if err != nil {
return err
}
return l.sess.send(ctx, text, imgs)
}
func (m *Manager) prepareMessage(ctx context.Context, l *live, actor, text string, images []string) ([]image, error) {
imgs := []image{} imgs := []image{}
for _, id := range images { for _, id := range images {
b, data, err := m.store.ReadBlob(ctx, id) b, data, err := m.store.ReadBlob(ctx, id)
if err != nil { if err != nil {
return err return nil, err
} }
imgs = append(imgs, image{mediaType: b.ContentType, data: data}) imgs = append(imgs, image{mediaType: b.ContentType, data: data})
} }
@@ -503,7 +524,7 @@ func (m *Manager) message(ctx context.Context, l *live, actor, text string, imag
if l.implementer() { if l.implementer() {
head, err := git.Head(l.dir) head, err := git.Head(l.dir)
if err != nil { if err != nil {
return err return nil, err
} }
payload["head"] = head payload["head"] = head
} }
@@ -519,7 +540,7 @@ func (m *Manager) message(ctx context.Context, l *live, actor, text string, imag
} }
err := m.emit(ctx, l, actor, "user.message", payload) err := m.emit(ctx, l, actor, "user.message", payload)
if err != nil { if err != nil {
return err return nil, err
} }
m.mu.Lock() m.mu.Lock()
@@ -532,7 +553,7 @@ func (m *Manager) message(ctx context.Context, l *live, actor, text string, imag
l.lastText = "" l.lastText = ""
m.mu.Unlock() m.mu.Unlock()
return l.sess.send(ctx, text, imgs) return imgs, nil
} }
func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) { func (m *Manager) ensureLive(ctx context.Context, sess api.AgentSession, task api.Task) (*live, error) {