fix(lifecycle): clear stale dispatch gate after cancel (#2825)
This commit is contained in:
@@ -195,6 +195,9 @@ func (m *Manager) CancelAgent(ctx context.Context, executionID string) error {
|
||||
execution.promptFinishedMu.Unlock()
|
||||
|
||||
if ch == nil {
|
||||
if execution.dispatchedPromptPending.Load() {
|
||||
return m.escalateStuckCancel(ctx, execution, nil)
|
||||
}
|
||||
if streamDisconnected {
|
||||
m.logger.Info("agent stream already disconnected; cancel is complete",
|
||||
zap.String("execution_id", executionID))
|
||||
@@ -230,6 +233,9 @@ func (m *Manager) CancelAgent(ctx context.Context, executionID string) error {
|
||||
// Without this, a follow-up PromptAgent races on promptDoneCh with two readers.
|
||||
select {
|
||||
case <-ch:
|
||||
if execution.dispatchedPromptPending.Load() {
|
||||
return m.escalateStuckCancel(ctx, execution, ch)
|
||||
}
|
||||
m.logger.Debug("in-flight prompt finished after cancel",
|
||||
zap.String("execution_id", executionID))
|
||||
return nil
|
||||
@@ -263,6 +269,10 @@ func (m *Manager) escalateStuckCancel(ctx context.Context, execution *AgentExecu
|
||||
zap.String("execution_id", execution.ID),
|
||||
zap.String("session_id", execution.SessionID))
|
||||
|
||||
// Clear the cancelled dispatch-only prompt before releasing its waiter. If
|
||||
// the waiter immediately admits a successor, a later cleanup must not clear
|
||||
// the successor's gate.
|
||||
execution.dispatchedPromptPending.Store(false)
|
||||
execution.signalPromptCompletionForStartupGeneration(
|
||||
execution.startupAttemptSnapshot(),
|
||||
PromptCompletionSignal{
|
||||
@@ -272,18 +282,20 @@ func (m *Manager) escalateStuckCancel(ctx context.Context, execution *AgentExecu
|
||||
},
|
||||
)
|
||||
|
||||
select {
|
||||
case <-ch:
|
||||
m.logger.Info("in-flight prompt released after cancel escalation",
|
||||
zap.String("execution_id", execution.ID))
|
||||
case <-time.After(cancelEscalationTimeout):
|
||||
m.logger.Warn("in-flight prompt did not release after cancel escalation",
|
||||
zap.String("execution_id", execution.ID))
|
||||
case <-ctx.Done():
|
||||
// Fall through to MarkReady/drain below — once the synthetic signal is
|
||||
// queued, the cleanup must survive the caller's context cancellation
|
||||
// or the execution leaks in the Running state and the stale signal
|
||||
// breaks the next PromptAgent call.
|
||||
if ch != nil {
|
||||
select {
|
||||
case <-ch:
|
||||
m.logger.Info("in-flight prompt released after cancel escalation",
|
||||
zap.String("execution_id", execution.ID))
|
||||
case <-time.After(cancelEscalationTimeout):
|
||||
m.logger.Warn("in-flight prompt did not release after cancel escalation",
|
||||
zap.String("execution_id", execution.ID))
|
||||
case <-ctx.Done():
|
||||
// Fall through to MarkReady/drain below — once the synthetic signal is
|
||||
// queued, the cleanup must survive the caller's context cancellation
|
||||
// or the execution leaks in the Running state and the stale signal
|
||||
// breaks the next PromptAgent call.
|
||||
}
|
||||
}
|
||||
|
||||
if err := m.markReadyEventWithContext(context.Background(), execution.ID, events.AgentReady, true); err != nil {
|
||||
|
||||
@@ -50,11 +50,7 @@ func TestManager_CancelAgent_EscalatesWhenAgentHangs(t *testing.T) {
|
||||
streamCtx, streamCancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(streamCancel)
|
||||
require.NoError(t, client.StreamUpdates(streamCtx, func(_ agentctlClient.AgentEvent) {}, nil, nil))
|
||||
select {
|
||||
case <-mock.wsConnected:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("mock server did not see WS connection")
|
||||
}
|
||||
waitForWSConnected(t, mock)
|
||||
|
||||
mgr := newTestManager(t)
|
||||
mockBus, ok := mgr.eventBus.(*MockEventBus)
|
||||
@@ -128,6 +124,87 @@ func TestManager_CancelAgent_EscalatesWhenAgentHangs(t *testing.T) {
|
||||
require.True(t, sawReady, "expected AgentReady event after cancel escalation")
|
||||
}
|
||||
|
||||
// TestManager_CancelAgent_ClearsPendingDispatchGate reproduces the lost-turn
|
||||
// completion path: a dispatch-only prompt leaves a completion gate behind,
|
||||
// cancellation escalates through the manager, and the next prompt must still
|
||||
// reach the existing agentctl stream.
|
||||
func TestManager_CancelAgent_ClearsPendingDispatchGate(t *testing.T) {
|
||||
prevWait := cancelWaitTimeout
|
||||
prevEsc := cancelEscalationTimeout
|
||||
cancelWaitTimeout = 20 * time.Millisecond
|
||||
cancelEscalationTimeout = 20 * time.Millisecond
|
||||
t.Cleanup(func() {
|
||||
cancelWaitTimeout = prevWait
|
||||
cancelEscalationTimeout = prevEsc
|
||||
})
|
||||
|
||||
mock := newMockAgentServer(t)
|
||||
t.Cleanup(func() { mock.server.Close() })
|
||||
promptSeen := make(chan struct{}, 2)
|
||||
mock.handler = func(msg ws.Message) *ws.Message {
|
||||
if msg.Action == "agent.prompt" {
|
||||
promptSeen <- struct{}{}
|
||||
}
|
||||
if msg.Action == "agent.cancel" {
|
||||
resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
|
||||
"success": true,
|
||||
})
|
||||
return resp
|
||||
}
|
||||
return mock.defaultHandler(msg)
|
||||
}
|
||||
|
||||
client := createTestClient(t, mock.server.URL)
|
||||
t.Cleanup(client.Close)
|
||||
streamCtx, streamCancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(streamCancel)
|
||||
require.NoError(t, client.StreamUpdates(streamCtx, func(_ agentctlClient.AgentEvent) {}, nil, nil))
|
||||
select {
|
||||
case <-mock.wsConnected:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("mock server did not see WS connection")
|
||||
}
|
||||
|
||||
mgr := newTestManager(t)
|
||||
exec := &AgentExecution{
|
||||
ID: "exec-cancel-dispatch-gate",
|
||||
TaskID: "task-1",
|
||||
SessionID: "session-1",
|
||||
Status: v1.AgentStatusRunning,
|
||||
WorkspacePath: "/workspace",
|
||||
agentctl: client,
|
||||
promptDoneCh: make(chan PromptCompletionSignal, 1),
|
||||
}
|
||||
require.NoError(t, mgr.executionStore.Add(exec))
|
||||
|
||||
_, err := mgr.PromptAgent(context.Background(), exec.ID, "dispatch-only", nil, true)
|
||||
require.NoError(t, err)
|
||||
select {
|
||||
case <-promptSeen:
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
t.Fatal("dispatch-only prompt did not reach agentctl")
|
||||
}
|
||||
require.True(t, exec.dispatchedPromptPending.Load(),
|
||||
"dispatch-only prompt must leave its completion gate pending")
|
||||
|
||||
cancelCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
defer cancel()
|
||||
require.ErrorIs(t, mgr.CancelAgent(cancelCtx, exec.ID), ErrCancelEscalated)
|
||||
require.False(t, exec.dispatchedPromptPending.Load(),
|
||||
"escalation must clear dispatch gate before releasing the waiter")
|
||||
|
||||
followUpCtx, followUpCancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
defer followUpCancel()
|
||||
_, err = mgr.PromptAgent(followUpCtx, exec.ID, "follow-up", nil, true)
|
||||
require.NoError(t, err,
|
||||
"follow-up prompt should not wait on the cancelled dispatch-only gate")
|
||||
select {
|
||||
case <-promptSeen:
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
t.Fatal("follow-up prompt did not reach agentctl after cancel escalation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_CancelAgent_DisconnectedStreamWithoutPromptIsAlreadyCancelled(t *testing.T) {
|
||||
mgr := newTestManager(t)
|
||||
exec := &AgentExecution{
|
||||
|
||||
@@ -1527,12 +1527,12 @@ func (sm *SessionManager) triggerPrompt(
|
||||
}
|
||||
|
||||
func waitForPendingDispatchedPrompt(ctx context.Context, execution *AgentExecution) error {
|
||||
if !execution.dispatchedPromptPending {
|
||||
if !execution.dispatchedPromptPending.Load() {
|
||||
return nil
|
||||
}
|
||||
select {
|
||||
case <-execution.promptDoneCh:
|
||||
execution.dispatchedPromptPending = false
|
||||
execution.dispatchedPromptPending.Store(false)
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
@@ -1547,7 +1547,7 @@ func (sm *SessionManager) finishAcceptedPrompt(
|
||||
promptGeneration uint64,
|
||||
) (*PromptResult, error) {
|
||||
if dispatchOnly {
|
||||
execution.dispatchedPromptPending = true
|
||||
execution.dispatchedPromptPending.Store(true)
|
||||
}
|
||||
if onDispatched != nil {
|
||||
onDispatched()
|
||||
|
||||
@@ -203,7 +203,7 @@ type AgentExecution struct {
|
||||
// response buffers active for an execution. Agentctl accepts prompt requests
|
||||
// asynchronously, so its transport-level gate alone cannot provide this.
|
||||
promptMu sync.Mutex
|
||||
dispatchedPromptPending bool
|
||||
dispatchedPromptPending atomic.Bool
|
||||
|
||||
// Closed when the current SendPrompt returns, so CancelAgent can wait
|
||||
// for the in-flight prompt to finish before the caller retries.
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
---
|
||||
spec: docs/specs/agent-stall-recovery/spec.md
|
||||
created: 2026-08-19
|
||||
status: done
|
||||
---
|
||||
|
||||
# Implementation Plan: Lost Turn-Completion Cancel Recovery
|
||||
|
||||
## Overview
|
||||
|
||||
Issue [#2823](https://github.com/kdlbs/kandev/issues/2823) reports a session
|
||||
that cannot accept prompts after one completion event is lost. A focused Go
|
||||
reproduction confirms the root cause. Cancel escalation drains its synthetic
|
||||
completion but leaves the dispatch-only prompt gate set.
|
||||
|
||||
This plan clears that gate at the cancellation boundary. It also makes the gate
|
||||
state safe for access from prompt and cancellation goroutines.
|
||||
|
||||
## Backend
|
||||
|
||||
### Prompt gate state
|
||||
|
||||
- In `apps/backend/internal/agent/runtime/lifecycle/types.go`, change
|
||||
`dispatchedPromptPending` to `atomic.Bool`.
|
||||
- In `apps/backend/internal/agent/runtime/lifecycle/session.go`, use atomic load
|
||||
and store operations in `waitForPendingDispatchedPrompt` and
|
||||
`finishAcceptedPrompt`.
|
||||
- In `apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go`,
|
||||
route cancellation through the same escalation cleanup when a pending
|
||||
dispatch gate has no live prompt barrier, then clear the gate before
|
||||
`escalateStuckCancel` sends its synthetic completion signal. Escalation skips
|
||||
the waiter timeout when no prompt barrier exists.
|
||||
|
||||
The cancellation path must clear the old gate before it releases a waiter.
|
||||
This order prevents a stale cleanup from clearing a successor dispatch.
|
||||
|
||||
## Tests
|
||||
|
||||
- **What:** A dispatch-only prompt has no completion. Cancellation escalates.
|
||||
The next prompt reaches agentctl without a backend restart.
|
||||
- **File:**
|
||||
`apps/backend/internal/agent/runtime/lifecycle/manager_interaction_cancel_test.go`
|
||||
- **How:** Use the existing agentctl WebSocket test server and the real
|
||||
`Manager.CancelAgent` path. The regression test must fail before the code
|
||||
change because the follow-up prompt waits until its context expires.
|
||||
- **Concurrency:** Run the focused lifecycle tests with the Go race detector.
|
||||
|
||||
## Verification Results
|
||||
|
||||
- RED: `TestManager_CancelAgent_ClearsPendingDispatchGate` failed because the
|
||||
follow-up prompt returned `context deadline exceeded` after cancellation.
|
||||
- GREEN: `cd apps/backend && go test -race ./internal/agent/runtime/lifecycle -run
|
||||
'TestManager_CancelAgent_(EscalatesWhenAgentHangs|ClearsPendingDispatchGate)'
|
||||
-count=1` passed with 2 tests.
|
||||
- GREEN: `cd apps/backend && go test ./internal/agent/runtime/lifecycle -run
|
||||
'TestSendPrompt_DispatchOnly' -count=1` passed with 2 tests.
|
||||
- GREEN: `git diff --check` passed.
|
||||
- Review remediation: cancellation now escalates when the real dispatch-only
|
||||
path has a nil or stale closed prompt barrier, and the regression asserts the
|
||||
gate is clear before the follow-up prompt.
|
||||
- Review remediation checks: the cancel race pair passed with 2 tests, the
|
||||
dispatch-only regression passed with 2 tests, `gofmt -l` reported no files,
|
||||
and `git diff --check` passed.
|
||||
|
||||
## Implementation Waves And Parallel Candidates
|
||||
|
||||
Wave 1:
|
||||
|
||||
- [x] [task-01-clear-stale-dispatch-gate](task-01-clear-stale-dispatch-gate.md)
|
||||
|
||||
Execution is sequential. The task changes one shared lifecycle state machine
|
||||
and is not safe for parallel implementation.
|
||||
|
||||
## Risks
|
||||
|
||||
- Cancellation and prompt dispatch run on different goroutines. Non-atomic gate
|
||||
access can introduce a data race.
|
||||
- If cancellation clears the gate after it releases a prompt waiter, it can
|
||||
clear state that belongs to a successor dispatch.
|
||||
|
||||
## Out of Scope
|
||||
|
||||
- Automatic turn cancellation based only on elapsed time.
|
||||
- A fixed timeout for legitimate long-running dispatch-only turns.
|
||||
- New conversation messages for dispatch errors.
|
||||
- Changes to the stall detector or ACP adapter completion fallback.
|
||||
@@ -0,0 +1,89 @@
|
||||
---
|
||||
id: "01-clear-stale-dispatch-gate"
|
||||
title: "Clear stale dispatch gate"
|
||||
status: done
|
||||
wave: 1
|
||||
depends_on: []
|
||||
plan: "plan.md"
|
||||
spec: "../../specs/agent-stall-recovery/spec.md"
|
||||
---
|
||||
|
||||
# Task 01: Clear Stale Dispatch Gate
|
||||
|
||||
## Acceptance
|
||||
|
||||
1. Cancel escalation clears a pending dispatch-only completion gate.
|
||||
2. A prompt that follows the completed cancellation reaches agentctl without a
|
||||
backend restart.
|
||||
3. Prompt and cancellation goroutines access the gate without a data race.
|
||||
|
||||
## Verification
|
||||
|
||||
```bash
|
||||
cd apps/backend && go test -race ./internal/agent/runtime/lifecycle -run 'TestManager_CancelAgent_(EscalatesWhenAgentHangs|ClearsPendingDispatchGate)' -count=1
|
||||
cd apps/backend && go test ./internal/agent/runtime/lifecycle -run 'TestSendPrompt_DispatchOnly' -count=1
|
||||
git diff --check
|
||||
```
|
||||
|
||||
## Files Likely Touched
|
||||
|
||||
- `apps/backend/internal/agent/runtime/lifecycle/types.go`
|
||||
- `apps/backend/internal/agent/runtime/lifecycle/session.go`
|
||||
- `apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go`
|
||||
- `apps/backend/internal/agent/runtime/lifecycle/manager_interaction_cancel_test.go`
|
||||
- `docs/plans/lost-turn-completion-cancel-recovery/plan.md`
|
||||
- `docs/plans/lost-turn-completion-cancel-recovery/task-01-clear-stale-dispatch-gate.md`
|
||||
|
||||
## Dependencies
|
||||
|
||||
None.
|
||||
|
||||
## Parallelism
|
||||
|
||||
`sequential`. The files share one lifecycle state machine.
|
||||
|
||||
## Inputs
|
||||
|
||||
- The cancel recovery scenario in
|
||||
`docs/specs/agent-stall-recovery/spec.md`.
|
||||
- The root-cause analysis in
|
||||
`docs/plans/lost-turn-completion-cancel-recovery/plan.md`.
|
||||
- Existing cancel escalation tests in
|
||||
`manager_interaction_cancel_test.go`.
|
||||
- Existing dispatch-only prompt tests in `session_test.go`.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
1. Write a regression test that uses the real `Manager.CancelAgent` path.
|
||||
2. Make sure that the test fails because the follow-up prompt does not reach
|
||||
agentctl.
|
||||
3. Change `dispatchedPromptPending` to `atomic.Bool`.
|
||||
4. Replace direct reads and writes with `Load` and `Store`.
|
||||
5. Route nil or stale closed prompt barriers through escalation when a dispatch
|
||||
gate is still pending.
|
||||
6. Clear the old gate before `escalateStuckCancel` sends its synthetic signal.
|
||||
7. Do not add an elapsed-time timeout to the gate.
|
||||
|
||||
## Output Contract
|
||||
|
||||
Report the changed files, exact test commands, test results, and remaining
|
||||
risks. Update this task and `plan.md` in the same conversation.
|
||||
|
||||
## Results
|
||||
|
||||
- RED: `TestManager_CancelAgent_ClearsPendingDispatchGate` failed with
|
||||
`context deadline exceeded` because escalation left the dispatch gate set.
|
||||
- GREEN: `cd apps/backend && go test -race ./internal/agent/runtime/lifecycle
|
||||
-run 'TestManager_CancelAgent_(EscalatesWhenAgentHangs|ClearsPendingDispatchGate)'
|
||||
-count=1` passed with 2 tests.
|
||||
- GREEN: `cd apps/backend && go test ./internal/agent/runtime/lifecycle -run
|
||||
'TestSendPrompt_DispatchOnly' -count=1` passed with 2 tests.
|
||||
- GREEN: `git diff --check` passed.
|
||||
- Changed files: `types.go`, `session.go`, `manager_interaction.go`, and
|
||||
`manager_interaction_cancel_test.go`, plus this plan package.
|
||||
- Review remediation: the regression now uses the real dispatch-only state with
|
||||
no fabricated prompt barrier, asserts the gate is cleared before the
|
||||
follow-up, and bounds the initial prompt observation.
|
||||
- The dispatch gate is atomic. Cancellation handles nil and stale closed
|
||||
barriers, clears the gate before releasing the synthetic completion signal,
|
||||
and marks the execution ready. No known blockers remain.
|
||||
@@ -18,6 +18,7 @@ Decisions:
|
||||
Implementation plans:
|
||||
|
||||
- [Agent stall recovery](../../plans/agent-stall-recovery/plan.md)
|
||||
- [Lost turn-completion cancel recovery](../../plans/lost-turn-completion-cancel-recovery/plan.md)
|
||||
- [OpenCode terminal error surfacing](../../plans/opencode-terminal-error-surfacing/plan.md)
|
||||
- [OpenCode actionable error links](../../plans/opencode-actionable-error-links/plan.md)
|
||||
|
||||
@@ -137,6 +138,9 @@ sanitized diagnostic message for the collapsed technical-details surface.
|
||||
makes the session input-ready.
|
||||
- If the agent does not acknowledge cancellation, the existing bounded
|
||||
cancel-escalation path releases the prompt and reconciles the session.
|
||||
- Cancel escalation clears every prompt-admission barrier for the cancelled
|
||||
turn, including a pending dispatch-only completion. A later prompt can
|
||||
dispatch without a backend restart.
|
||||
- If notice-message persistence fails, the failure is logged without changing
|
||||
or terminating the running session.
|
||||
- If OpenCode does not support error-only stderr emission, changes its log
|
||||
@@ -178,6 +182,9 @@ sanitized diagnostic message for the collapsed technical-details surface.
|
||||
- **GIVEN** a stall notice is visible, **WHEN** the user activates
|
||||
**Cancel turn**, **THEN** the existing cancellation path settles the turn and
|
||||
the session becomes available for new input without a backend restart.
|
||||
- **GIVEN** a dispatch-only prompt never reports completion and cancellation
|
||||
escalates, **WHEN** the user sends a later prompt, **THEN** Kandev dispatches
|
||||
it without a backend restart.
|
||||
- **GIVEN** a stall notice is visible on a phone viewport, **WHEN** the user taps
|
||||
**Cancel turn**, **THEN** the same cancellation outcome is reachable through
|
||||
an inline, content-width touch target of at least 44px.
|
||||
|
||||
Reference in New Issue
Block a user