fix(tui): stop cancel-then-resend from racing the daemon's session lock (#2912)

Closes #2901

Stacked on #2899 (base branch: `tmi/issue-2899-tui-stats`) — not `main`.

## Why this matters

Cancelling a turn (Esc/Ctrl+C) used to free the composer immediately,
letting a follow-up query go out on the same session before the daemon
had actually finished the previous one — the daemon only releases its
per-session lock when the sidecar's worker thread notices the cancel and
unwinds (cooperative, checked once per agent-loop step), not when the
client drops its connection. A fast enough retype could beat that
release and land on a bare, confusing 409.

Live testing on real hardware then found the fix itself had a worse
problem: a **second** cancel-then-resend in the same session quit the
app outright (3/3, clean exit, no panic). Root cause: a cancelled turn
on the daemon transport never actually settles via `doneMsg` — it
settles via its own terminal event (the server's cooperative cancel just
produces an ordinary "stopped" answer), and that code path never rearms
the listener `doneMsg` depends on. The turn's cancel bookkeeping was
left stuck "pending" for the rest of the session, so the *next*
Esc/Ctrl+C failed its own guard and fell through to quit. Fixed —
settlement now happens on every real terminal signal for the live turn,
not only the one that could never arrive.

**Scope note:** live testing also measured input staying blocked ~18–25s
per cancel while the server's cooperative cancellation finishes whatever
step is already in flight. That is server-side latency, not something
this client-side fix can shorten, so it is split out to #2917 with a
recommended direction (queue the resend instead of blocking on it)
rather than folded in here.

When a conflict does still happen (a genuinely concurrent request), the
client reports it with an actionable message naming the busy session,
instead of the generic "check daemon status" copy meant for an
unreachable daemon.

The server-side 409 policy itself is intentional and untouched
(`hub/agents/email/python/gaia_agent_email/query_routes.py`) — it is
documented, tested behavior that prevents two turns from stepping on one
session's agent state. This PR only closes the client-side windows that
let a resend either race that lock or crash the app.

## Live evidence

- **409s:** 0 of 8 cancel-then-resend attempts (down from 5/5 before
this PR).
- **Second-cycle crash:** 3/3 reproducible before this fix, closed by a
failing-test-first regression
(`TestSecondCancelThenResendDoesNotQuitTheApp`,
`TestSecondCtrlCThenResendDoesNotQuitTheApp`) that fails against the
pre-fix code and passes after.
- **Input-blocking (~18–25s per cancel):** confirmed real, not fixable
client-side alone — tracked in #2917.

## Test plan

- [x] `cd tui && go test ./...` — all packages pass
- [x] `go test -race ./internal/ui/chat/... ./internal/client/...` —
clean
- [x] New regression tests reproduce the live two-cycle crash end to end
(cancel turn 1 → settle via its own terminal event → resend turn 2 →
cancel again) and fail against the pre-fix code, pass after
- [ ] Live-mailbox validation (cancel-then-resend, including the
two-cycle case) — scheduled separately

Author: itomek

tui/internal/client/client.go

@@ -36,6 +36,26 @@ type TranscriptResetter interface {
 	ResetTranscript()
 }
 
+// AgentCanceler is implemented by transports where the server, not this
+// client dropping its connection, decides when a cancelled run has actually
+// settled (#2901) — e.g. a daemon-relay session guarded by a server-side
+// lock that a worker thread releases on its own cooperative schedule.
+//
+// Cancel asks the server to stop the active run out of band. It deliberately
+// does NOT tear down the caller's own read of the run's event channel: that
+// read has to keep going until the channel closes on its own, because THAT
+// closure — not this call returning — is the one signal proven to follow the
+// server's cleanup. A transport with no such server-side lock (e.g. a local
+// subprocess) does not implement this; for it, tearing down the local
+// connection/process IS the whole cancellation, and the caller's own
+// context.CancelFunc already does that.
+type AgentCanceler interface {
+	// Cancel asks the server to stop the currently active run. It returns an
+	// actionable error if the request could not be delivered; a run that has
+	// already ended is not an error (there is nothing left to cancel).
+	Cancel(ctx context.Context) error
+}
+
 // AgentConfirmer is implemented by transports that can resolve a
 // needs_confirmation pause under the resume model (spec §5: the event carries
 // a non-empty confirm_url and the run stays paused server-side awaiting it).

tui/internal/client/sse.go

@@ -236,7 +236,8 @@ func (s *SSEClient) Send(ctx context.Context, query string) (<-chan interface{},
 		status := resp.StatusCode
 		resp.Body.Close()
 		cancel()
-		if status == http.StatusNotFound {
+		switch status {
+		case http.StatusNotFound:
 			// A 404 on the query path is specifically "this route does not exist
 			// there" — most often a sidecar predating the canonical /query
 			// endpoint. Saying so beats making the user decode a bare 404.
@@ -246,10 +247,22 @@ func (s *SSEClient) Send(ctx context.Context, query string) (<-chan interface{},
 					"`gaia daemon agents` and reinstall/update the agent — or no agent with "+
 					"that id is registered with the daemon",
 				s.agentID, detail)
+		case http.StatusConflict:
+			// The session's run_lock is still held by a previous turn (#2901) —
+			// most often the tail of a just-cancelled one the daemon has not
+			// finished unwinding yet. Name the busy resource, like Respond and
+			// Confirm already do for their own 409s, instead of the generic
+			// relay-refused copy below (which points at a daemon that is down,
+			// not one that is merely still finishing).
+			return nil, fmt.Errorf(
+				"the '%s' agent is still finishing the previous turn on this session (%s). "+
+					"Wait a moment and try again",
+				s.agentID, detail)
+		default:
+			return nil, fmt.Errorf(
+				"the daemon relay refused the '%s' query (%s). Check `gaia daemon status`",
+				s.agentID, detail)
 		}
-		return nil, fmt.Errorf(
-			"the daemon relay refused the '%s' query (%s). Check `gaia daemon status`",
-			s.agentID, detail)
 	}
 
 	handle := &runHandle{runID: runID, cancel: cancel}
@@ -588,6 +601,21 @@ type confirmRequest struct {
 	Approved bool `json:"approved"`
 }
 
+// postCancel POSTs /v1/<agent>/query/{runID}/cancel and returns the response,
+// or an error if the request itself could not be delivered. Shared by
+// cancelRun (best-effort, fire-and-forget) and Cancel (the caller-facing,
+// error-reporting seam — #2901).
+func (s *SSEClient) postCancel(ctx context.Context, inst *daemon.Instance, runID string) (*http.Response, error) {
+	resp, _, err := s.daemon.Do(ctx, inst, daemon.Request{
+		Method: http.MethodPost,
+		Path: fmt.Sprintf("/v1/%s/query/%s/cancel",
+			url.PathEscape(s.agentID), url.PathEscape(runID)),
+		HTTPClient: s.cancelHTTP,
+		Op:         fmt.Sprintf("cancel the '%s' run", s.agentID),
+	})
+	return resp, err
+}
+
 // cancelRun asks the relay to drop a run we are abandoning.
 //
 // Best-effort: the sidecar may already be gone. It owns its own background
@@ -605,13 +633,7 @@ func (s *SSEClient) cancelRun(handle *runHandle) {
 	ctx, cancel := context.WithTimeout(context.Background(), cancelTimeout)
 	defer cancel()
 
-	resp, _, err := s.daemon.Do(ctx, inst, daemon.Request{
-		Method: http.MethodPost,
-		Path: fmt.Sprintf("/v1/%s/query/%s/cancel",
-			url.PathEscape(s.agentID), url.PathEscape(handle.runID)),
-		HTTPClient: s.cancelHTTP,
-		Op:         fmt.Sprintf("cancel the '%s' run", s.agentID),
-	})
+	resp, err := s.postCancel(ctx, inst, handle.runID)
 	if err != nil {
 		s.opts.Logf("sse: best-effort cancel for '%s' run_id=%s failed: %v",
 			s.agentID, handle.runID, err)
@@ -632,6 +654,54 @@ func (s *SSEClient) cancelRun(handle *runHandle) {
 	}
 }
 
+// Cancel implements client.AgentCanceler (#2901). It asks the server to stop
+// the active run WITHOUT touching this client's own read of the run's SSE
+// stream — unlike the caller's context.CancelFunc, which tears that read
+// down immediately and is exactly what let a resend beat the daemon's
+// session run_lock (cooperative cancellation, released by the sidecar's
+// worker thread's own `finally`, checked once per agent-loop step; see
+// hub/agents/email/python/gaia_agent_email/query_routes.py).
+//
+// Leaving the read running is what makes the eventual terminal signal a
+// near-certain settlement signal in practice — not a proven one. The
+// sidecar's `finally` runs signal_done() and only THEN run_lock.release(),
+// same thread, no I/O between the two statements, so by the time anything
+// downstream notices signal_done the lock is released in all but a
+// vanishingly rare preemption between those two lines (#2912 review). A
+// client that aborts its own read instead observes a "done" that is
+// guaranteed to race ahead of the release, which is the bug this method
+// exists to avoid reintroducing. Closing that last window for good is a
+// one-line server-side reorder (release, then signal_done) — tracked
+// separately, not required here since this is a Go-only change.
+func (s *SSEClient) Cancel(ctx context.Context) error {
+	s.mu.Lock()
+	inst := s.inst
+	active := s.active
+	s.mu.Unlock()
+	if inst == nil || active == nil {
+		// Nothing live to cancel — most often the run already reached its own
+		// terminal event between the keypress and this call. Not an error: the
+		// caller's read will observe that completion on its own.
+		return nil
+	}
+
+	resp, err := s.postCancel(ctx, inst, active.runID)
+	if err != nil {
+		return fmt.Errorf("could not deliver the cancel request for the '%s' agent: %w", s.agentID, err)
+	}
+	defer resp.Body.Close()
+
+	switch resp.StatusCode {
+	case http.StatusOK, http.StatusNotFound:
+		// 404 means the run had already ended by the time this landed — the
+		// caller's read will see that completion on its own; nothing failed.
+		return nil
+	default:
+		return fmt.Errorf("cancelling the '%s' run failed (%s)",
+			s.agentID, daemon.ErrorDetail(resp))
+	}
+}
+
 func (s *SSEClient) clearActive(handle *runHandle) {
 	s.mu.Lock()
 	if s.active == handle {
@@ -782,5 +852,6 @@ var (
 	_ AgentClient        = (*SSEClient)(nil)
 	_ AgentResponder     = (*SSEClient)(nil)
 	_ AgentConfirmer     = (*SSEClient)(nil)
+	_ AgentCanceler      = (*SSEClient)(nil)
 	_ TranscriptResetter = (*SSEClient)(nil)
 )