FIX: keep session recovery controls available after errors

README.md

@@ -1244,6 +1244,15 @@ The other top-level views are:
   stale events from older identified turns cannot fail the new one. Errors from
   linked agents propagate to the parent row, below pending approvals and questions.
   No prompt is automatically retried and no model is automatically switched.
+  With a connected shared app-server, a stopped error state retains the follow-up
+  composer: type `continue` to try another turn, or use `/model` (and other
+  supported settings commands) to recover. This includes Codex's
+  [`systemError` runtime state](https://learn.chatgpt.com/docs/app-server#read-a-stored-thread-without-resuming);
+  it is not treated as a running turn. Recovery always checks the
+  live state before sending. Retrying turns, pending approvals/questions, unknown
+  or unloaded sessions, and uncertain writes keep their existing capability
+  guards. Local-only observations cannot grant control, and authentication,
+  network or quota problems may still require action in Codex or your account.
   With a shared app-server, working-command headings also show elapsed time or
   completed duration when Codex supplies valid item lifecycle timestamps.
   Codex 0.166.0's item fields and older notification-envelope fields are supported;

internal/codex/benchmark.go

@@ -629,18 +629,28 @@ func (s *appServerSession) nextEnvelopeOrHeartbeat(ctx context.Context, heartbea
 	case <-s.done:
 		return benchmarkEnvelope{}, false, errors.New("Codex app-server session closed")
 	case err := <-s.readErrors:
+		return s.nextEnvelopeAfterReadError(err)
+	case envelope := <-s.envelopes:
+		return envelope, false, nil
+	}
+}
+
+func (s *appServerSession) nextEnvelopeAfterReadError(err error) (benchmarkEnvelope, bool, error) {
+	select {
+	case envelope := <-s.envelopes:
+		// Draining a final response must not consume the terminal signal:
+		// the next read still needs to observe EOF/the decoder failure.
 		select {
-		case envelope := <-s.envelopes:
-			return envelope, false, nil
+		case s.readErrors <- err:
 		default:
 		}
-		if errors.Is(err, io.EOF) {
-			return benchmarkEnvelope{}, false, fmt.Errorf("Codex app-server closed unexpectedly: %s", strings.TrimSpace(s.stderr.String()))
-		}
-		return benchmarkEnvelope{}, false, err
-	case envelope := <-s.envelopes:
 		return envelope, false, nil
+	default:
+	}
+	if errors.Is(err, io.EOF) {
+		return benchmarkEnvelope{}, false, fmt.Errorf("Codex app-server closed unexpectedly: %s", strings.TrimSpace(s.stderr.String()))
 	}
+	return benchmarkEnvelope{}, false, err
 }
 
 func (s *appServerSession) close() {

internal/codex/benchmark_test.go

@@ -963,20 +963,52 @@ func TestStartAppServerUsesEphemeralIsolatedAPIKeyLogin(t *testing.T) {
 
 func TestAppServerReadLoopAndClose(t *testing.T) {
 	server, _ := newFakeBenchmarkServer()
+	t.Cleanup(server.close)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
 	input := strings.NewReader("{\"method\":\"first\"}\n{\"method\":\"second\"}\n")
 	go server.readLoop(json.NewDecoder(input))
 	for _, want := range []string{"first", "second"} {
-		envelope, err := server.nextEnvelope(context.Background())
+		envelope, err := server.nextEnvelope(ctx)
 		if err != nil || envelope.Method != want {
 			t.Fatalf("decoded envelope = %#v, %v; want %s", envelope, err, want)
 		}
 	}
-	if _, err := server.nextEnvelope(context.Background()); err == nil || !strings.Contains(err.Error(), "closed unexpectedly") {
+	if _, err := server.nextEnvelope(ctx); err == nil || !strings.Contains(err.Error(), "closed unexpectedly") {
 		t.Fatalf("decoder terminal error = %v, want unexpected-close diagnostic", err)
 	}
 	server.close()
 }
 
+func TestAppServerRetainsReadErrorWhileDrainingBufferedMessages(t *testing.T) {
+	for _, terminal := range []error{io.EOF, errors.New("decoder failed")} {
+		t.Run(terminal.Error(), func(t *testing.T) {
+			server, _ := newFakeBenchmarkServer(benchmarkEnvelope{Method: "first"}, benchmarkEnvelope{Method: "second"})
+			t.Cleanup(server.close)
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			// Reproduce the select race deterministically: a terminal error has
+			// just been received, but final decoded messages are also buffered.
+			first, heartbeat, err := server.nextEnvelopeAfterReadError(terminal)
+			if err != nil || heartbeat || first.Method != "first" {
+				t.Fatal("terminal error displaced the first buffered message", first, err)
+			}
+			second, err := server.nextEnvelope(ctx)
+			if err != nil || second.Method != "second" {
+				t.Fatal("terminal error displaced the second buffered message", second, err)
+			}
+			_, err = server.nextEnvelope(ctx)
+			if terminal == io.EOF {
+				if err == nil || !strings.Contains(err.Error(), "closed unexpectedly") {
+					t.Fatal("EOF diagnostic lost after draining messages", err)
+				}
+			} else if !errors.Is(err, terminal) {
+				t.Fatal("decoder error lost after draining messages", err)
+			}
+		})
+	}
+}
+
 func TestLockedBufferAndFatalErrorClassification(t *testing.T) {
 	buffer := &lockedBuffer{}
 	if _, err := buffer.Write([]byte("diagnostic")); err != nil || buffer.String() != "diagnostic" {

internal/codex/daemon_status.go

@@ -44,6 +44,9 @@ func parseSessionRuntimeStatus(kind string, flags []string) sessionRuntimeStatus
 	if kind == "idle" {
 		return sessionRuntimeIdle
 	}
+	if kind == "systemError" {
+		return sessionRuntimeFailed
+	}
 	if kind != "active" {
 		return sessionRuntimeUnknown
 	}
@@ -58,3 +61,13 @@ func parseSessionRuntimeStatus(kind string, flags []string) sessionRuntimeStatus
 	}
 	return status
 }
+
+// Both states describe a loaded thread with no running turn. A failed turn
+// must remain recoverable without treating unknown/unloaded threads as idle.
+func runtimeAcceptsFollowup(status sessionRuntimeStatus) bool {
+	return status == sessionRuntimeIdle || status == sessionRuntimeFailed
+}
+
+func threadAcceptsFollowup(kind string) bool {
+	return runtimeAcceptsFollowup(parseSessionRuntimeStatus(kind, nil))
+}

internal/codex/daemon_status_test.go

@@ -9,15 +9,26 @@ func TestParseSessionRuntimeStatus(t *testing.T) {
 		want  sessionRuntimeStatus
 	}{
 		{"idle", nil, sessionRuntimeIdle},
+		{"systemError", nil, sessionRuntimeFailed},
 		{"active", nil, sessionRuntimeWorking},
 		{"active", []string{"waitingOnUserInput"}, sessionRuntimeInput},
 		{"active", []string{"waitingOnApproval"}, sessionRuntimeApproval},
 		{"active", []string{"waitingOnUserInput", "waitingOnApproval"}, sessionRuntimeApproval},
 		{"notLoaded", nil, sessionRuntimeUnknown},
+		{"unrecognised", nil, sessionRuntimeUnknown},
 	}
 	for _, test := range tests {
 		if got := parseSessionRuntimeStatus(test.kind, test.flags); got != test.want {
 			t.Errorf("parseSessionRuntimeStatus(%q, %v) = %v; want %v", test.kind, test.flags, got, test.want)
 		}
 	}
 }
+
+func TestThreadAcceptsFollowup(t *testing.T) {
+	for _, kind := range []string{"idle", "systemError", "active", "notLoaded", "", "failed", "unknown"} {
+		want := kind == "idle" || kind == "systemError"
+		if got := threadAcceptsFollowup(kind); got != want {
+			t.Errorf("threadAcceptsFollowup(%q) = %v; want %v", kind, got, want)
+		}
+	}
+}

internal/codex/session_commands_unix.go

@@ -320,7 +320,7 @@ func (p *daemonStatusProvider) commandMenu(ctx context.Context, t commandThread,
 		m.Revision = commandDigest([]any{current, t.Cwd})
 		setting := func(label, help string, params map[string]any) {
 			params["threadId"] = t.ID
-			if t.Status.Type != "idle" && parts[0] != "tier" {
+			if !threadAcceptsFollowup(t.Status.Type) && parts[0] != "tier" {
 				m.Choices = append(m.Choices, SessionCommandChoice{Label: label, Help: help, Unavailable: "Waiting for idle. Refresh options when this turn finishes."})
 				return
 			}

internal/codex/session_context_daemon.go

@@ -23,6 +23,8 @@ type daemonContextState struct {
 	approvalsLimited                                bool
 	promptToken                                     string
 	promptSpent                                     bool
+	promptAttempt                                   string
+	promptDispatched                                bool
 	completed                                       bool
 	latest                                          SessionContext
 	requests                                        map[string]SessionContext
@@ -116,7 +118,8 @@ func daemonContextEvent(states map[string]*daemonContextState, method string, id
 		state.turnObserved = false
 		state.streamItem, state.streamText = "", ""
 		state.latest.Streaming = false
-		state.promptToken, state.promptSpent = "", false
+		state.promptToken, state.promptSpent, state.promptAttempt = "", false, ""
+		state.promptDispatched = false
 		state.completed = method == "turn/completed" && p.Turn.Status == "completed"
 		clear(state.requests)
 		state.approvalsLimited = false
@@ -145,6 +148,11 @@ func daemonContextEvent(states map[string]*daemonContextState, method string, id
 		state.latest = failure
 		state.ended, state.completed = true, false
 		state.streamItem, state.streamText = "", ""
+		// A definitive non-retrying failure closes this attempt even when no
+		// turn/completed follows. Do not carry its consumed editor into recovery.
+		// Offers still require a separately observed quiescent runtime status.
+		state.promptToken, state.promptSpent, state.promptAttempt, state.turnToken = "", false, "", ""
+		state.promptDispatched = false
 		clear(state.requests)
 		return
 	case "serverRequest/resolved":

internal/codex/session_directory_unix.go

@@ -102,7 +102,7 @@ func (p *daemonStatusProvider) commandDirectoryMenu(ctx context.Context, t comma
 	}
 	help := "Current directory: " + strconv.Quote(cwd) + "\nNew directory: " + strconv.Quote(target)
 	choice := SessionCommandChoice{Label: "Change working directory", Help: help}
-	if t.Status.Type == "idle" {
+	if threadAcceptsFollowup(t.Status.Type) {
 		choice.method, choice.params = "thread/settings/update", map[string]any{"threadId": t.ID, "cwd": target}
 	} else {
 		choice.Unavailable = "Waiting for idle. Refresh options when this turn finishes."

internal/codex/session_prompt_unix.go

@@ -61,9 +61,14 @@ func (p *daemonStatusProvider) promptLifecycleLocked(method string, raw json.Raw
 		p.statuses = map[string]sessionRuntimeStatus{}
 	}
 	p.statuses[event.ThreadID] = status
-	if status != sessionRuntimeIdle {
+	if !runtimeAcceptsFollowup(status) {
 		if s := p.contexts[event.ThreadID]; s != nil {
 			s.promptToken = ""
+			// A runtime change cancels a reservation that has sent no mutation,
+			// but cannot make an uncertain dispatched write reusable.
+			if !s.promptDispatched {
+				s.promptSpent, s.promptAttempt = false, ""
+			}
 		}
 	}
 }
@@ -88,7 +93,7 @@ func (p *daemonStatusProvider) SessionPrompt(thread string) SessionPromptOffer {
 			return inputOffer(c)
 		}
 	}
-	if p.statuses[thread] != sessionRuntimeIdle {
+	if !runtimeAcceptsFollowup(p.statuses[thread]) {
 		return SessionPromptOffer{}
 	}
 	if s == nil {
@@ -124,7 +129,7 @@ func (p *daemonStatusProvider) SendSessionPrompt(ctx context.Context, token stri
 			if _, ok := p.subscribed[thread]; !ok {
 				continue
 			}
-			if s.promptToken == token && len(s.requests) == 0 && p.statuses[thread] == sessionRuntimeIdle {
+			if s.promptToken == token && len(s.requests) == 0 && runtimeAcceptsFollowup(p.statuses[thread]) {
 				offer = SessionPromptOffer{Token: token, ThreadID: thread}
 				state = s
 				break
@@ -155,6 +160,8 @@ func (p *daemonStatusProvider) SendSessionPrompt(ctx context.Context, token stri
 	} else {
 		state.promptToken = ""
 		state.promptSpent = true
+		state.promptAttempt = token
+		state.promptDispatched = false
 		p.statuses[offer.ThreadID] = sessionRuntimeWorking
 	}
 	p.mu.Unlock()
@@ -165,6 +172,18 @@ func (p *daemonStatusProvider) SendSessionPrompt(ctx context.Context, token stri
 		}
 		return p.write(connection, map[string]any{"id": json.RawMessage(requestID), "result": map[string]any{"answers": mapped}})
 	}
+	dispatched := false
+	defer func() {
+		if !dispatched {
+			// No mutation was sent: a failed read-only preflight must not spend
+			// the editor forever. Fresh runtime observations still gate its return.
+			p.mu.Lock()
+			if p.connection == connection && p.contexts[offer.ThreadID] == state && state.promptAttempt == token {
+				state.promptSpent, state.promptAttempt = false, ""
+			}
+			p.mu.Unlock()
+		}
+	}()
 	// Recheck live runtime immediately before dispatch: idle telemetry may
 	// have aged while the user composed their prompt.
 	var read struct {
@@ -177,11 +196,17 @@ func (p *daemonStatusProvider) SendSessionPrompt(ctx context.Context, token stri
 	if err := p.requestOn(ctx, connection, "thread/read", map[string]any{"threadId": offer.ThreadID, "includeTurns": false}, &read); err != nil {
 		return err
 	}
-	if read.Thread.Status.Type != "idle" {
-		return errors.New("session is no longer idle; reply in Codex")
+	if !threadAcceptsFollowup(read.Thread.Status.Type) {
+		return errors.New("session is no longer ready for a follow-up; reply in Codex")
 	}
 	p.mu.Lock()
-	unchanged := p.connection == connection && p.contexts[offer.ThreadID] == state && len(state.requests) == 0
+	unchanged := p.connection == connection && p.contexts[offer.ThreadID] == state &&
+		state.promptSpent && state.promptAttempt == token && len(state.requests) == 0
+	if unchanged {
+		// Commit the phase while holding the same lock as runtime invalidation.
+		// Once this boundary is crossed, status notifications cannot release it.
+		state.promptDispatched, dispatched = true, true
+	}
 	p.mu.Unlock()
 	if !unchanged {
 		return errors.New("session changed; reply in Codex")

internal/codex/session_prompt_unix_test.go

@@ -14,7 +14,12 @@ import (
 	"github.com/gorilla/websocket"
 )
 
-func promptProvider(t *testing.T, runtime string) (*daemonStatusProvider, <-chan daemonEnvelope) {
+type promptTestFault struct {
+	method     string
+	beforeRead func()
+}
+
+func promptProvider(t *testing.T, runtime string, faults ...promptTestFault) (*daemonStatusProvider, <-chan daemonEnvelope) {
 	t.Helper()
 	received := make(chan daemonEnvelope, 16)
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -31,6 +36,15 @@ func promptProvider(t *testing.T, runtime string) (*daemonStatusProvider, <-chan
 			}
 			received <- e
 			if e.Method != "" {
+				if len(faults) > 0 && e.Method == "thread/read" && faults[0].beforeRead != nil {
+					faults[0].beforeRead()
+				}
+				if len(faults) > 0 && e.Method == faults[0].method {
+					if err := conn.WriteJSON(map[string]any{"id": e.ID, "error": map[string]any{"code": -32603, "message": "synthetic request failure"}}); err != nil {
+						return
+					}
+					continue
+				}
 				result := map[string]any{}
 				if e.Method == "thread/read" {
 					result = map[string]any{"thread": map[string]any{"status": map[string]any{"type": runtime}}}