fix(ui): preserve cancel progress across task switches (#2199)

This commit is contained in:
Carlos Florêncio
2026-08-03 22:03:23 +01:00
committed by GitHub
parent ea07ac7863
commit b2518b9b8b
53 changed files with 2467 additions and 137 deletions
+10
View File
@@ -101,6 +101,16 @@ apps/backend/
- Handles event-driven state transitions via workflow engine
- Located in `internal/orchestrator/`
**Cancellation progress projection:** `orchestrator.Service.CancellationPending(sessionID)` is a
runtime-only, session-scoped view of accepted cancellation work. Serialization that carries the
boolean with ordering identity uses the atomic `CancellationPendingSnapshot(sessionID)` provider,
whose process-local revision increments on first-begin and last-end transitions. The task DTO
package exposes both the compatibility boolean provider and snapshot seam; boot state, task-session
HTTP/WS lists and detail responses, and the session-scoped WebSocket notification must project
explicit `true`/`false` values plus the revision. Keep count, revision, and publication queue updates
in one critical section, drain event-bus sends outside it, and never persist this transient marker or
turn it into a coarse session lifecycle state.
**Watcher Dispatch Coordinator** (`internal/orchestrator/watcher_dispatch.go`) is the single pipeline that turns a freshly-observed external issue (Linear, Jira, future) into a Kandev task. Bus subscribers for each integration forward the event to `WatcherDispatchCoordinator.Dispatch` with a per-integration `WatcherSource` implementation (`source_linear.go`, `source_jira.go`). Source methods carry the integration-specific bits (reserve dedup, build task request, attach task ID, release, auto-start params); the coordinator owns the cross-cutting pipeline (create task, decide auto-start, error/release handling). Add a new watcher = implement `WatcherSource` + register a one-line bus subscriber. Do NOT add another `createXIssueTask` mirror.
**GitHub App registration catalog** (`internal/github/`) stores zero or more managed/imported App
@@ -413,7 +413,11 @@ func (b bootStateBuilder) quickChatSessions(ctx context.Context, workspaceID str
taskSessions := make(map[string]taskdto.TaskSessionDTO, len(items))
for _, item := range items {
sessions = append(sessions, mapQuickChatSessionState(item))
taskSessions[item.SessionID] = taskdto.FromTaskSession(item.Session)
sessionDTO := taskdto.FromTaskSession(item.Session)
if b.p.orchestratorSvc != nil {
taskdto.EnrichCancellationPending(&sessionDTO, b.p.orchestratorSvc)
}
taskSessions[item.SessionID] = sessionDTO
}
return quickChatBootState{sessions: sessions, taskSessions: taskSessions}, nil
}
@@ -977,6 +981,7 @@ func (b bootStateBuilder) addTaskDetailSessionsState(
// (ADR-0049). No-op for non-RUNNING sessions.
if b.p.orchestratorSvc != nil {
taskdto.EnrichForegroundActivity(&dto, b.p.orchestratorSvc)
taskdto.EnrichCancellationPending(&dto, b.p.orchestratorSvc)
}
dto.PendingAction = bootPendingActionPtr(&session.ID, pendingActionsBySession)
sessionItems[session.ID] = dto
+34 -2
View File
@@ -76,6 +76,7 @@ import (
spriteshandlers "github.com/kandev/kandev/internal/sprites"
sshhandlers "github.com/kandev/kandev/internal/ssh"
systemsvc "github.com/kandev/kandev/internal/system"
taskdto "github.com/kandev/kandev/internal/task/dto"
taskhandlers "github.com/kandev/kandev/internal/task/handlers"
"github.com/kandev/kandev/internal/task/models"
sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
@@ -105,7 +106,12 @@ const (
// buildSessionDataProvider constructs the session data provider function used by the WebSocket hub
// to send initial data (git status, context window, available commands) when a client subscribes.
func buildSessionDataProvider(taskRepo *sqliterepo.Repository, lifecycleMgr *lifecycle.Manager, log *logger.Logger) func(context.Context, string) ([]*ws.Message, error) {
func buildSessionDataProvider(
taskRepo *sqliterepo.Repository,
lifecycleMgr *lifecycle.Manager,
cancellationProvider taskdto.CancellationPendingProvider,
log *logger.Logger,
) func(context.Context, string) ([]*ws.Message, error) {
return func(ctx context.Context, sessionID string) ([]*ws.Message, error) {
session, err := taskRepo.GetTaskSession(ctx, sessionID)
if err != nil {
@@ -113,7 +119,7 @@ func buildSessionDataProvider(taskRepo *sqliterepo.Repository, lifecycleMgr *lif
}
var result []*ws.Message
result = appendSessionStateMessage(sessionID, session, result)
result = appendSessionStateMessageWithCancellation(sessionID, session, cancellationProvider, result)
result = appendAgentctlStatusMessage(ctx, lifecycleMgr, sessionID, result, log)
result = appendLiveGitStatusMessage(ctx, taskRepo, lifecycleMgr, sessionID, session, result, log)
result = appendContextWindowMessage(sessionID, session, result)
@@ -201,12 +207,38 @@ func appendAgentctlStatusMessage(
// (page reload, task switch, WS reconnect) can seed environmentIdBySessionId
// — without it, env-routed shell terminals stall on "Connecting terminal...".
func appendSessionStateMessage(sessionID string, session *models.TaskSession, result []*ws.Message) []*ws.Message {
return appendSessionStateMessageWithCancellation(sessionID, session, nil, result)
}
func cancellationPendingSnapshot(
provider taskdto.CancellationPendingProvider,
sessionID string,
) (bool, uint64) {
if snapshotProvider, ok := provider.(taskdto.CancellationPendingSnapshotProvider); ok {
return snapshotProvider.CancellationPendingSnapshot(sessionID)
}
return provider.CancellationPending(sessionID), 0
}
func appendSessionStateMessageWithCancellation(
sessionID string,
session *models.TaskSession,
cancellationProvider taskdto.CancellationPendingProvider,
result []*ws.Message,
) []*ws.Message {
payload := map[string]interface{}{
sessionIDPayloadKey: sessionID,
taskIDPayloadKey: session.TaskID,
newStatePayloadKey: string(session.State),
sessionUpdatedAtPayloadKey: session.UpdatedAt.UTC().Format(time.RFC3339Nano),
"name": session.Name,
"cancellation_pending": false,
"cancellation_revision": uint64(0),
}
if cancellationProvider != nil {
pending, revision := cancellationPendingSnapshot(cancellationProvider, sessionID)
payload["cancellation_pending"] = pending
payload["cancellation_revision"] = revision
}
if session.ReviewStatus != models.ReviewStatusNone {
payload["review_status"] = string(session.ReviewStatus)
@@ -146,6 +146,39 @@ func TestAppendSessionStateMessage_IncludesTaskEnvironmentID(t *testing.T) {
}
}
type fakeCancellationPendingProvider struct {
pending bool
revision uint64
}
func (p fakeCancellationPendingProvider) CancellationPending(string) bool {
return p.pending
}
func (p fakeCancellationPendingProvider) CancellationPendingSnapshot(string) (bool, uint64) {
return p.pending, p.revision
}
func TestAppendSessionStateMessage_IncludesCancellationPending(t *testing.T) {
session := &models.TaskSession{ID: "sess-cancel", TaskID: "task-1"}
msgs := appendSessionStateMessageWithCancellation(
session.ID,
session,
fakeCancellationPendingProvider{pending: true, revision: 7},
nil,
)
if len(msgs) != 1 {
t.Fatalf("expected 1 message, got %d", len(msgs))
}
payload := decodePayload(t, msgs[0].Payload)
if got, ok := payload["cancellation_pending"]; !ok || got != true {
t.Fatalf("cancellation_pending = %#v, want explicit true", payload["cancellation_pending"])
}
if got, ok := payload["cancellation_revision"]; !ok || got != float64(7) {
t.Fatalf("cancellation_revision = %#v, want 7", payload["cancellation_revision"])
}
}
func TestAppendAgentctlStatusMessage_IncludesWorkspacePathForReload(t *testing.T) {
log, err := logger.NewLogger(logger.LoggingConfig{
Level: "error",
+1 -1
View File
@@ -696,7 +696,7 @@ func startGatewayAndServe(
}
gateways.RegisterSessionStreamNotifications(ctx, eventBus, gateway.Hub, log)
gateway.Hub.SetSessionDataProvider(buildSessionDataProvider(repos.Task, lifecycleMgr, log))
gateway.Hub.SetSessionDataProvider(buildSessionDataProvider(repos.Task, lifecycleMgr, orchestratorSvc, log))
gateway.Hub.SetSessionGitDataProvider(buildSessionGitDataProvider(repos.Task, lifecycleMgr, log))
log.Info("Session data provider configured for session subscriptions (git status from snapshots)")
+3
View File
@@ -68,6 +68,9 @@ const (
// operator-facing composer and status indicator can distinguish
// "generating" from "waiting on spawned background work".
TaskSessionActivityChanged = "task_session.activity_changed"
// TaskSessionCancellationChanged fires when a cancellation request starts
// or the last overlapping request finishes for a session.
TaskSessionCancellationChanged = "task_session.cancellation_changed"
// TaskSessionErrorChanged is emitted when a recoverable agent error is
// created, replaced, or dismissed. It is a bounded status source; the
// error message itself is projected into the task summary rather than
@@ -2,6 +2,7 @@ package websocket
import (
"context"
"sync"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/events"
@@ -14,6 +15,8 @@ type TaskEventBroadcaster struct {
hub *Hub
subscriptions []bus.Subscription
logger *logger.Logger
closeMu sync.Mutex
closed bool
}
func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub *Hub, log *logger.Logger) *TaskEventBroadcaster {
@@ -72,6 +75,7 @@ func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub *
b.subscribe(eventBus, events.EnvironmentUpdated, ws.ActionEnvironmentUpdated)
b.subscribe(eventBus, events.EnvironmentDeleted, ws.ActionEnvironmentDeleted)
b.subscribe(eventBus, events.TaskSessionActivityChanged, ws.ActionSessionActivityChanged)
b.subscribe(eventBus, events.TaskSessionCancellationChanged, ws.ActionSessionCancellationChanged)
b.subscribe(eventBus, events.TaskStatusSummaryUpdated, ws.ActionTaskStatusSummaryUpdated)
b.subscribe(eventBus, events.MessageAdded, ws.ActionSessionMessageAdded)
b.subscribe(eventBus, events.MessageUpdated, ws.ActionSessionMessageUpdated)
@@ -97,12 +101,21 @@ func RegisterTaskNotifications(ctx context.Context, eventBus bus.EventBus, hub *
}
func (b *TaskEventBroadcaster) Close() {
for _, sub := range b.subscriptions {
b.closeMu.Lock()
if b.closed {
b.closeMu.Unlock()
return
}
b.closed = true
subscriptions := b.subscriptions
b.subscriptions = nil
b.closeMu.Unlock()
for _, sub := range subscriptions {
if sub != nil && sub.IsValid() {
_ = sub.Unsubscribe()
}
}
b.subscriptions = nil
}
func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action string) {
@@ -140,7 +153,16 @@ func (b *TaskEventBroadcaster) subscribeWithResolver(
b.logger.Error("failed to subscribe to events", zap.String("subject", subject), zap.Error(err))
return
}
b.closeMu.Lock()
if b.closed {
b.closeMu.Unlock()
if sub.IsValid() {
_ = sub.Unsubscribe()
}
return
}
b.subscriptions = append(b.subscriptions, sub)
b.closeMu.Unlock()
}
func (b *TaskEventBroadcaster) broadcastEvent(ctx context.Context, event *bus.Event, action string) error {
@@ -215,6 +237,11 @@ func (b *TaskEventBroadcaster) routeBroadcast(
b.hub.BroadcastToSession(sessionID, msg)
return nil
}
case ws.ActionSessionCancellationChanged:
if sessionID != "" {
b.hub.BroadcastToSession(sessionID, msg)
return nil
}
case ws.ActionMessageQueueStatusChanged:
if sessionID != "" {
b.hub.BroadcastToSession(sessionID, msg)
@@ -9,6 +9,7 @@ import (
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/events/bus"
ws "github.com/kandev/kandev/pkg/websocket"
"github.com/stretchr/testify/require"
)
func testLogger() *logger.Logger {
@@ -107,7 +108,7 @@ func TestTaskEventBroadcaster_NoDuplicateSubscriptions(t *testing.T) {
//
// Update this number when adding or removing event subscriptions in
// RegisterTaskNotifications — it is intentionally exact.
const wantSubscriptions = 62
const wantSubscriptions = 63
if got := len(b.subscriptions); got != wantSubscriptions {
t.Errorf("RegisterTaskNotifications created %d subscriptions, want %d — "+
"did an event get subscribed twice?", got, wantSubscriptions)
@@ -216,3 +217,59 @@ func TestTaskEventBroadcaster_PreservesAllFields(t *testing.T) {
t.Errorf("event data was modified\noriginal: %s\ncaptured: %s", origJSON, capturedJSON)
}
}
func TestTaskEventBroadcaster_CancellationIsSessionScoped(t *testing.T) {
h := newTestHub(t)
first := newTestClient("first")
second := newTestClient("second")
registerTestClient(h, first)
registerTestClient(h, second)
h.SubscribeToSession(first, "session-1")
h.SubscribeToSession(second, "session-2")
msg, err := ws.NewNotification("session.cancellation_changed", map[string]any{
"session_id": "session-1",
"cancellation_pending": true,
})
require.NoError(t, err)
b := &TaskEventBroadcaster{hub: h, logger: testLogger()}
require.NoError(t, b.routeBroadcast("session.cancellation_changed", msg.Payload, "session-1", "", msg))
if !clientReceived(first) {
t.Fatal("session subscriber did not receive cancellation notification")
}
if clientReceived(second) {
t.Fatal("cancellation notification crossed the session boundary")
}
}
func TestTaskEventBroadcaster_CancellationSubscriptionIsSessionScoped(t *testing.T) {
log := testLogger()
eventBus := bus.NewMemoryEventBus(log)
hub := newTestHub(t)
first := newTestClient("first")
second := newTestClient("second")
registerTestClient(hub, first)
registerTestClient(hub, second)
hub.SubscribeToSession(first, "session-1")
hub.SubscribeToSession(second, "session-2")
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
b := RegisterTaskNotifications(ctx, eventBus, hub, log)
t.Cleanup(b.Close)
require.NoError(t, eventBus.Publish(ctx, events.TaskSessionCancellationChanged, bus.NewEvent(
events.TaskSessionCancellationChanged,
"test",
map[string]any{"session_id": "session-1", "cancellation_pending": true},
)))
var notification ws.Message
received := <-first.send
require.NoError(t, json.Unmarshal(received, &notification))
require.Equal(t, ws.ActionSessionCancellationChanged, notification.Action)
if clientReceived(second) {
t.Fatal("cancellation subscription crossed the session boundary")
}
}
@@ -281,6 +281,10 @@ type mockAgentManager struct {
cancelAgentCalls atomic.Int32
cancelAgentBlock chan struct{}
cancelAgentEntered chan struct{}
// cancelAgentContextErr is returned after an optional block when the
// supplied context has been cancelled. It lets cancellation tests verify
// that accepted work uses a detached context.
cancelAgentContextErr error
// cancelAgentErr, when set, is returned by CancelAgent instead of nil —
// lets tests exercise callers that must react to a genuine cancel
// failure (as opposed to the tolerated ErrNoExecutionForSession /
@@ -399,7 +403,7 @@ func (m *mockAgentManager) PromptAgentWithDispatchCallback(ctx context.Context,
}
return result, err
}
func (m *mockAgentManager) CancelAgent(_ context.Context, _ string) error {
func (m *mockAgentManager) CancelAgent(ctx context.Context, _ string) error {
m.cancelAgentCalls.Add(1)
if m.cancelAgentEntered != nil {
select {
@@ -410,6 +414,9 @@ func (m *mockAgentManager) CancelAgent(_ context.Context, _ string) error {
if m.cancelAgentBlock != nil {
<-m.cancelAgentBlock
}
if m.cancelAgentContextErr != nil && ctx.Err() != nil {
return m.cancelAgentContextErr
}
return m.cancelAgentErr
}
func (m *mockAgentManager) RespondToPermissionBySessionID(_ context.Context, _, _, _ string, _ bool) error {
@@ -598,10 +598,13 @@ type Service struct {
// serialize their side effects, but they are not themselves cancellations;
// treating every mutex holder as a cancellation would make agent.boot_ready
// discard a legitimate boot signal while a stream frame is being persisted.
// Values are reference counts so duplicate cancellation requests waiting on
// the shared guard keep cancellation priority until the last request exits.
// Values hold the reference count, process-local transition revision, and
// publication queue for each session. Entries remain after the count reaches
// zero so a later operation cannot reuse an older revision. The single mutex
// makes each count transition and its queued publication one atomic state change; event-bus
// sends happen after releasing it so one slow bus cannot delay other sessions.
cancelOperationsMu sync.Mutex
cancelOperations map[string]int
cancelOperations map[string]*cancellationOperationState
// transientRetries tracks in-progress transient-provider-error (529
// Overloaded) retry loops. key: sessionID, value: *transientRetryEntry.
@@ -21,6 +21,8 @@ import (
"github.com/kandev/kandev/internal/agent/runtime/routingerr"
"github.com/kandev/kandev/internal/common/constants"
"github.com/kandev/kandev/internal/editors/capabilities"
"github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/orchestrator/dto"
"github.com/kandev/kandev/internal/orchestrator/executor"
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
@@ -4000,26 +4002,148 @@ func (s *Service) beginCancelInFlight(sessionID string) func() {
s.cancelOperationsMu.Lock()
if s.cancelOperations == nil {
s.cancelOperations = make(map[string]int)
s.cancelOperations = make(map[string]*cancellationOperationState)
}
state := s.cancelOperations[sessionID]
if state == nil {
state = &cancellationOperationState{}
s.cancelOperations[sessionID] = state
}
state.count++
startDrain := false
if state.count == 1 {
state.revision++
startDrain = s.enqueueCancellationPendingLocked(sessionID, state, true)
}
s.cancelOperations[sessionID]++
s.cancelOperationsMu.Unlock()
if startDrain {
s.drainCancellationPublicationQueue(sessionID, state)
}
var once sync.Once
return func() {
once.Do(func() {
startDrain := false
s.cancelOperationsMu.Lock()
count := s.cancelOperations[sessionID]
if count <= 1 {
delete(s.cancelOperations, sessionID)
} else {
s.cancelOperations[sessionID] = count - 1
state := s.cancelOperations[sessionID]
if state != nil && state.count > 0 {
state.count--
if state.count == 0 {
state.revision++
startDrain = s.enqueueCancellationPendingLocked(sessionID, state, false)
}
}
s.cancelOperationsMu.Unlock()
if startDrain {
s.drainCancellationPublicationQueue(sessionID, state)
}
})
}
}
// CancellationPending reports whether at least one accepted cancellation
// operation is active for sessionID. The projection is intentionally runtime
// scoped: a backend restart clears it because no cancellation work survives
// the process restart.
func (s *Service) CancellationPending(sessionID string) bool {
pending, _ := s.CancellationPendingSnapshot(sessionID)
return pending
}
// CancellationPendingSnapshot returns the pending projection and its
// process-local transition revision from one critical section. Keeping these
// values together prevents REST/boot hydration from observing a boolean from
// one generation with the revision from another.
func (s *Service) CancellationPendingSnapshot(sessionID string) (bool, uint64) {
if sessionID == "" {
return false, 0
}
s.cancelOperationsMu.Lock()
defer s.cancelOperationsMu.Unlock()
state := s.cancelOperations[sessionID]
if state == nil {
return false, 0
}
return state.count > 0, state.revision
}
func (s *Service) publishCancellationPending(sessionID string, pending bool, revision uint64) {
if s.eventBus == nil || sessionID == "" {
return
}
payload := map[string]interface{}{
"session_id": sessionID,
"cancellation_pending": pending,
"cancellation_revision": revision,
}
if err := s.eventBus.Publish(context.Background(), events.TaskSessionCancellationChanged,
bus.NewEvent(events.TaskSessionCancellationChanged, "orchestrator", payload)); err != nil {
if s.logger != nil {
s.logger.Warn("failed to publish cancellation state",
zap.String("session_id", sessionID),
zap.Bool("cancellation_pending", pending),
zap.Uint64("cancellation_revision", revision),
zap.Error(err))
}
}
}
type cancellationOperationState struct {
count int
revision uint64
publications cancellationPublicationQueue
}
type cancellationPublication struct {
pending bool
revision uint64
}
type cancellationPublicationQueue struct {
pending []cancellationPublication
publishing bool
}
// enqueueCancellationPendingLocked appends a transition while the count and
// revision are still protected by cancelOperationsMu. The caller must drain
// the queue after releasing that mutex when this returns true.
func (s *Service) enqueueCancellationPendingLocked(
sessionID string,
state *cancellationOperationState,
pending bool,
) bool {
if s.eventBus == nil || sessionID == "" {
return false
}
queue := &state.publications
queue.pending = append(queue.pending, cancellationPublication{pending: pending, revision: state.revision})
if queue.publishing {
return false
}
queue.publishing = true
return true
}
func (s *Service) drainCancellationPublicationQueue(
sessionID string,
state *cancellationOperationState,
) {
for {
s.cancelOperationsMu.Lock()
queue := &state.publications
if len(queue.pending) == 0 {
queue.publishing = false
s.cancelOperationsMu.Unlock()
return
}
publication := queue.pending[0]
queue.pending = queue.pending[1:]
s.cancelOperationsMu.Unlock()
s.publishCancellationPending(sessionID, publication.pending, publication.revision)
}
}
// isCancelInFlight reports whether an actual cancellation operation for
// sessionID is active or waiting on the shared guard. It deliberately does
// not inspect the guard mutex itself because stream/lifecycle handlers also
@@ -4030,7 +4154,8 @@ func (s *Service) isCancelInFlight(sessionID string) bool {
}
s.cancelOperationsMu.Lock()
defer s.cancelOperationsMu.Unlock()
return s.cancelOperations[sessionID] > 0
state := s.cancelOperations[sessionID]
return state != nil && state.count > 0
}
// reconcileCancelledTurn durably settles the cancelled session and its turn.
@@ -4163,6 +4288,10 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
s.logger.Debug("cancelling agent turn", zap.String("session_id", sessionID))
endCancel := s.beginCancelInFlight(sessionID)
defer endCancel()
// Once authorization and admission succeed, cancellation is accepted work.
// Keep it independent of the request context so a WebSocket disconnect does
// not leave the session stuck in RUNNING after the agent has been interrupted.
operationCtx := context.WithoutCancel(ctx)
// Deduplicate concurrent retries. The UI's cancel button has no in-flight
// disable, so impatient users click it multiple times while the agent is
@@ -4188,7 +4317,7 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
defer unlockGuard()
// Fetch session for state updates and message creation
session, err := s.repo.GetTaskSession(ctx, sessionID)
session, err := s.repo.GetTaskSession(operationCtx, sessionID)
if err != nil {
s.logger.Warn("failed to get session for cancel",
zap.String("session_id", sessionID),
@@ -4204,7 +4333,7 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
// transient turn-close failure. Keep that retry eligible only while
// an active turn is still present; a settled second cancel must not
// re-run the source step's completion actions.
activeTurnID, turnErr := s.peekActiveTurnID(ctx, sessionID)
activeTurnID, turnErr := s.peekActiveTurnID(operationCtx, sessionID)
if turnErr != nil {
return fmt.Errorf("inspect active turn before cancel retry: %w", turnErr)
}
@@ -4222,7 +4351,7 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
// The agent manager routes the cancel to the right signal: ACP cancel for
// regular sessions, Ctrl-C via PTY stdin for passthrough sessions. Service
// stays protocol-agnostic; the seam is in lifecycle.Manager.CancelAgentBySessionID.
if err := s.agentManager.CancelAgent(ctx, sessionID); err != nil {
if err := s.agentManager.CancelAgent(operationCtx, sessionID); err != nil {
switch {
case errors.Is(err, lifecycle.ErrNoExecutionForSession):
// The session was live but there is no execution to cancel — the agent process
@@ -4249,14 +4378,14 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
// workflow completion is evaluated; a failure leaves the task on its
// current step so a retry can reconcile it safely.
if session != nil {
reconciled, err := s.reconcileCancelledTurn(ctx, session.TaskID, sessionID, session, cancelTurnCompletionEligible)
reconciled, err := s.reconcileCancelledTurn(operationCtx, session.TaskID, sessionID, session, cancelTurnCompletionEligible)
if err != nil {
return fmt.Errorf("reconcile cancelled turn: %w", err)
}
if reconciled != nil {
session = reconciled
}
} else if _, err := s.reconcileCancelledTurn(ctx, "", sessionID, nil, false); err != nil {
} else if _, err := s.reconcileCancelledTurn(operationCtx, "", sessionID, nil, false); err != nil {
return fmt.Errorf("reconcile cancelled turn: %w", err)
}
@@ -4267,7 +4396,7 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
"variant": "warning",
}
if err := s.messageCreator.CreateSessionMessage(
ctx,
operationCtx,
session.TaskID,
"Turn cancelled by user",
sessionID,
@@ -4287,17 +4416,17 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
// durable, while the per-session guard still excludes a late agent.ready
// event from observing a successor turn and transitioning a second time.
if session != nil && cancelTurnCompletionEligible {
s.processOnTurnCompleteViaEngineWithCause(ctx, session.TaskID, session, turnCompletionCauseUserCancellation)
s.processOnTurnCompleteViaEngineWithCause(operationCtx, session.TaskID, session, turnCompletionCauseUserCancellation)
// Disabled/no-transition outcomes and successful nonterminal transitions
// both reconcile the source task to REVIEW after session and turn
// settlement. The guarded state write is a no-op for terminal tasks or a
// destination that has already restarted work.
s.reconcileCancelledTaskReview(ctx, session.TaskID, session.ID)
s.reconcileCancelledTaskReview(operationCtx, session.TaskID, session.ID)
} else if session != nil {
// Preserve the existing review reconciliation for cancellations that
// are not eligible to run workflow completion (already waiting or a
// terminal/ineligible session).
s.reconcileCancelledTaskReview(ctx, session.TaskID, session.ID)
s.reconcileCancelledTaskReview(operationCtx, session.TaskID, session.ID)
}
// Release the cancel/interrupt guard now that this session's cancel is
@@ -915,15 +915,13 @@ func (m *cancelContextAgentManager) CancelAgent(ctx context.Context, sessionID s
return err
}
func TestCancelAgent_DoesNotTransitionWhenTurnClosureFails(t *testing.T) {
func newCancelTurnFailureFixture(t *testing.T, repo *sqliterepo.Repository, taskID, sessionID string) (*Service, *cancelTurnFailureService) {
t.Helper()
ctx := context.Background()
repo := setupTestRepo(t)
taskID := "task-cancel-turn-failure"
sessionID := "session-cancel-turn-failure"
seedSession(t, repo, taskID, sessionID, "step1")
now := time.Now().UTC()
require.NoError(t, repo.CreateTurn(ctx, &models.Turn{
ID: "turn-cancel-turn-failure",
ID: "turn-" + sessionID,
TaskID: taskID,
TaskSessionID: sessionID,
StartedAt: now,
@@ -932,11 +930,18 @@ func TestCancelAgent_DoesNotTransitionWhenTurnClosureFails(t *testing.T) {
}))
svc := createEngineService(t, repo, cancelCompletionStepGetter(true, false), &mockAgentManager{})
turnService := &cancelTurnFailureService{
repoTurnService: &repoTurnService{repo: repo},
completeErr: errors.New("turn close failed"),
}
turnService := &cancelTurnFailureService{repoTurnService: &repoTurnService{repo: repo}}
svc.turnService = turnService
return svc, turnService
}
func TestCancelAgent_DoesNotTransitionWhenTurnClosureFails(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
taskID := "task-cancel-turn-failure"
sessionID := "session-cancel-turn-failure"
svc, turnService := newCancelTurnFailureFixture(t, repo, taskID, sessionID)
turnService.completeErr = errors.New("turn close failed")
err := svc.CancelAgent(ctx, sessionID)
require.Error(t, err)
@@ -964,24 +969,9 @@ func TestCancelAgent_DoesNotTransitionWhenActiveTurnInspectionFails(t *testing.T
repo := setupTestRepo(t)
taskID := "task-cancel-active-turn-inspection-failure"
sessionID := "session-cancel-active-turn-inspection-failure"
seedSession(t, repo, taskID, sessionID, "step1")
svc, turnService := newCancelTurnFailureFixture(t, repo, taskID, sessionID)
require.NoError(t, repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateWaitingForInput, ""))
now := time.Now().UTC()
require.NoError(t, repo.CreateTurn(ctx, &models.Turn{
ID: "turn-cancel-active-turn-inspection-failure",
TaskID: taskID,
TaskSessionID: sessionID,
StartedAt: now,
CreatedAt: now,
UpdatedAt: now,
}))
svc := createEngineService(t, repo, cancelCompletionStepGetter(true, false), &mockAgentManager{})
turnService := &cancelTurnFailureService{
repoTurnService: &repoTurnService{repo: repo},
activeErr: errors.New("active turn lookup failed"),
}
svc.turnService = turnService
turnService.activeErr = errors.New("active turn lookup failed")
err := svc.CancelAgent(ctx, sessionID)
require.Error(t, err)
@@ -1000,6 +990,31 @@ func TestCancelAgent_DoesNotTransitionWhenActiveTurnInspectionFails(t *testing.T
assert.Equal(t, "step2", task.WorkflowStepID, "a later retry should evaluate the settled cancellation")
}
func TestCancelAgent_SurvivesCallerCancellationDuringWaitingRetry(t *testing.T) {
repo := setupTestRepo(t)
taskID := "task-cancel-waiting-retry"
sessionID := "session-cancel-waiting-retry"
svc, turnService := newCancelTurnFailureFixture(t, repo, taskID, sessionID)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
require.NoError(t, repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateWaitingForInput, ""))
svc.agentManager = &cancelContextAgentManager{
mockAgentManager: &mockAgentManager{},
cancel: cancel,
}
require.NoError(t, svc.CancelAgent(ctx, sessionID))
task, err := repo.GetTask(context.Background(), taskID)
require.NoError(t, err)
assert.Equal(t, "step2", task.WorkflowStepID)
session, err := repo.GetTaskSession(context.Background(), sessionID)
require.NoError(t, err)
assert.Equal(t, models.TaskSessionStateWaitingForInput, session.State)
activeTurn, err := turnService.GetActiveTurn(context.Background(), sessionID)
require.NoError(t, err)
assert.Nil(t, activeTurn)
}
func TestCancelAgent_NonterminalTransitionReconcilesReviewState(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
@@ -1053,7 +1068,7 @@ func TestReconcileCancelledTurn_DoesNotCloseTurnWhenSessionStateWriteFails(t *te
assert.NotNil(t, activeTurn, "a failed session-state write must not close the active turn")
}
func TestCancelAgent_DoesNotTransitionWhenSessionStateWriteFails(t *testing.T) {
func TestCancelAgent_SurvivesCallerCancellationAfterAdmission(t *testing.T) {
repo := setupTestRepo(t)
taskID := "task-cancel-session-write"
sessionID := "session-cancel-session-write"
@@ -1067,13 +1082,13 @@ func TestCancelAgent_DoesNotTransitionWhenSessionStateWriteFails(t *testing.T) {
svc := createTestServiceWithAgent(repo, cancelCompletionStepGetter(true, false), newMockTaskRepo(), manager)
err := svc.CancelAgent(ctx, sessionID)
require.Error(t, err)
require.NoError(t, err)
task, err := repo.GetTask(context.Background(), taskID)
require.NoError(t, err)
assert.Equal(t, "step1", task.WorkflowStepID)
assert.Equal(t, "step2", task.WorkflowStepID)
session, err := repo.GetTaskSession(context.Background(), sessionID)
require.NoError(t, err)
assert.Equal(t, models.TaskSessionStateRunning, session.State)
assert.Equal(t, models.TaskSessionStateWaitingForInput, session.State)
}
func TestCancelAgent_TerminalTransitionSkipsIntermediateReview(t *testing.T) {
@@ -1188,6 +1203,123 @@ func TestCancelAgent_DeduplicatesConcurrentCalls(t *testing.T) {
}
}
func TestCancellationPendingTracksReferencesAndPublishesTransitions(t *testing.T) {
recorded := &recordingEventBus{}
svc := &Service{eventBus: recorded}
endFirst := svc.beginCancelInFlight("session1")
require.True(t, svc.CancellationPending("session1"))
require.False(t, svc.CancellationPending("session2"))
pending, revision := svc.CancellationPendingSnapshot("session1")
require.True(t, pending)
require.Equal(t, uint64(1), revision)
endSecond := svc.beginCancelInFlight("session1")
require.True(t, svc.CancellationPending("session1"))
require.Len(t, cancellationPendingEvents(recorded), 1)
endFirst()
require.True(t, svc.CancellationPending("session1"))
endSecond()
require.False(t, svc.CancellationPending("session1"))
pending, revision = svc.CancellationPendingSnapshot("session1")
require.False(t, pending)
require.Equal(t, uint64(2), revision)
events := cancellationPendingEvents(recorded)
require.Len(t, events, 2)
require.Equal(t, true, events[0].event.Data.(map[string]interface{})["cancellation_pending"])
require.Equal(t, uint64(1), events[0].event.Data.(map[string]interface{})["cancellation_revision"])
require.Equal(t, false, events[1].event.Data.(map[string]interface{})["cancellation_pending"])
require.Equal(t, uint64(2), events[1].event.Data.(map[string]interface{})["cancellation_revision"])
}
func TestCancellationPendingSuccessorGenerationPreservesTransitionOrder(t *testing.T) {
recorded := &recordingEventBus{}
svc := &Service{
eventBus: recorded,
cancelOperations: map[string]*cancellationOperationState{
"session1": {
count: 1,
revision: 1,
publications: cancellationPublicationQueue{
publishing: true,
},
},
},
}
state := svc.cancelOperations["session1"]
// Model the final release and successor admission in one critical section:
// the false transition must be queued before the successor's true transition
// while an earlier publication is still draining.
svc.cancelOperationsMu.Lock()
state.count--
state.revision++
require.False(t, svc.enqueueCancellationPendingLocked("session1", state, false))
state.count++
state.revision++
require.False(t, svc.enqueueCancellationPendingLocked("session1", state, true))
state.publications.publishing = false
svc.cancelOperationsMu.Unlock()
svc.drainCancellationPublicationQueue("session1", state)
events := cancellationPendingEvents(recorded)
require.Len(t, events, 2)
require.Equal(t, false, events[0].event.Data.(map[string]interface{})["cancellation_pending"])
require.Equal(t, uint64(2), events[0].event.Data.(map[string]interface{})["cancellation_revision"])
require.Equal(t, true, events[1].event.Data.(map[string]interface{})["cancellation_pending"])
require.Equal(t, uint64(3), events[1].event.Data.(map[string]interface{})["cancellation_revision"])
}
func TestCancelAgent_ClearsCancellationPendingOnError(t *testing.T) {
recorded := &recordingEventBus{}
repo := setupTestRepo(t)
agentMgr := &mockAgentManager{cancelAgentErr: errors.New("cancel failed")}
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
svc.eventBus = recorded
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateRunning)
require.Error(t, svc.CancelAgent(context.Background(), "session1"))
require.False(t, svc.CancellationPending("session1"))
require.Len(t, cancellationPendingEvents(recorded), 2)
}
func TestCancelAgent_SurvivesCallerCancellation(t *testing.T) {
repo := setupTestRepo(t)
agentMgr := &mockAgentManager{
isAgentRunning: true,
cancelAgentBlock: make(chan struct{}),
cancelAgentEntered: make(chan struct{}, 1),
cancelAgentContextErr: errors.New("cancel used caller context"),
}
svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateRunning)
ctx, cancel := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() {
done <- svc.CancelAgent(ctx, "session1")
}()
<-agentMgr.cancelAgentEntered
cancel()
close(agentMgr.cancelAgentBlock)
require.NoError(t, <-done)
}
func cancellationPendingEvents(recorded *recordingEventBus) []recordedEvent {
const subject = "task_session.cancellation_changed"
filtered := make([]recordedEvent, 0, len(recorded.events))
for _, event := range recorded.events {
if event.subject == subject {
filtered = append(filtered, event)
}
}
return filtered
}
// TestCancelAgent_TaskStateReconcile ensures cancel lands actively-working
// kanban tasks in REVIEW (treated as finished work the user may want to
// review). Office tasks and tasks already out of IN_PROGRESS / SCHEDULING
@@ -0,0 +1,36 @@
package dto
// EnrichCancellationPending stamps the current runtime cancellation state
// onto a full session DTO. The provider is authoritative for both true and
// false values, so a settled cancellation clears stale client state.
func EnrichCancellationPending(session *TaskSessionDTO, provider CancellationPendingProvider) {
if session == nil || provider == nil {
return
}
session.CancellationPending, session.CancellationRevision = cancellationPendingSnapshot(
provider,
session.ID,
)
}
// EnrichCancellationPendingSummary is the summary DTO equivalent of
// EnrichCancellationPending.
func EnrichCancellationPendingSummary(session *TaskSessionSummaryDTO, provider CancellationPendingProvider) {
if session == nil || provider == nil {
return
}
session.CancellationPending, session.CancellationRevision = cancellationPendingSnapshot(
provider,
session.ID,
)
}
func cancellationPendingSnapshot(
provider CancellationPendingProvider,
sessionID string,
) (bool, uint64) {
if snapshotProvider, ok := provider.(CancellationPendingSnapshotProvider); ok {
return snapshotProvider.CancellationPendingSnapshot(sessionID)
}
return provider.CancellationPending(sessionID), 0
}
@@ -0,0 +1,76 @@
package dto
import (
"encoding/json"
"testing"
"github.com/stretchr/testify/require"
)
type fakeCancellationPendingProvider struct {
pending bool
}
func (p fakeCancellationPendingProvider) CancellationPending(string) bool {
return p.pending
}
type fakeCancellationSnapshotProvider struct {
pending bool
revision uint64
}
func (p fakeCancellationSnapshotProvider) CancellationPending(string) bool {
return p.pending
}
func (p fakeCancellationSnapshotProvider) CancellationPendingSnapshot(string) (bool, uint64) {
return p.pending, p.revision
}
func TestTaskSessionCancellationPendingSerializesExplicitFalse(t *testing.T) {
for name, value := range map[string]any{
"full": TaskSessionDTO{},
"summary": TaskSessionSummaryDTO{},
} {
t.Run(name, func(t *testing.T) {
encoded, err := json.Marshal(value)
require.NoError(t, err)
var body map[string]any
require.NoError(t, json.Unmarshal(encoded, &body))
require.Contains(t, body, "cancellation_pending")
require.Equal(t, false, body["cancellation_pending"])
require.Contains(t, body, "cancellation_revision")
require.Equal(t, float64(0), body["cancellation_revision"])
})
}
}
func TestEnrichCancellationPendingWritesProviderValue(t *testing.T) {
for _, pending := range []bool{true, false} {
t.Run(map[bool]string{true: "pending", false: "settled"}[pending], func(t *testing.T) {
full := &TaskSessionDTO{ID: "s1", CancellationPending: !pending}
EnrichCancellationPending(full, fakeCancellationPendingProvider{pending: pending})
require.Equal(t, pending, full.CancellationPending)
summary := &TaskSessionSummaryDTO{ID: "s1", CancellationPending: !pending}
EnrichCancellationPendingSummary(summary, fakeCancellationPendingProvider{pending: pending})
require.Equal(t, pending, summary.CancellationPending)
})
}
}
func TestEnrichCancellationPendingWritesRevisionWithSnapshot(t *testing.T) {
full := &TaskSessionDTO{ID: "s1"}
summary := &TaskSessionSummaryDTO{ID: "s1"}
provider := fakeCancellationSnapshotProvider{pending: true, revision: 9}
EnrichCancellationPending(full, provider)
EnrichCancellationPendingSummary(summary, provider)
require.True(t, full.CancellationPending)
require.Equal(t, uint64(9), full.CancellationRevision)
require.True(t, summary.CancellationPending)
require.Equal(t, uint64(9), summary.CancellationRevision)
}
+26
View File
@@ -272,6 +272,13 @@ type TaskSessionDTO struct {
// Not persisted — populated at the serialization boundary by
// EnrichForegroundActivity, never by FromTaskSession.
ForegroundActivity v1.ForegroundActivity `json:"foreground_activity,omitempty"`
// CancellationPending mirrors the orchestrator's runtime cancellation
// projection. It is always serialized so false clears stale client state.
CancellationPending bool `json:"cancellation_pending"`
// CancellationRevision identifies the process-local cancellation transition
// generation that produced CancellationPending. It is always serialized so
// clients can reject delayed snapshots from older generations.
CancellationRevision uint64 `json:"cancellation_revision"`
// PendingAction is the compact per-session projection used when the
// session transcript is not loaded in the client.
PendingAction *string `json:"pending_action,omitempty"`
@@ -322,6 +329,12 @@ type TaskSessionSummaryDTO struct {
// ForegroundActivity mirrors the in-memory fine-grained busy substate
// (ADR-0049); see TaskSessionDTO.
ForegroundActivity v1.ForegroundActivity `json:"foreground_activity,omitempty"`
// CancellationPending mirrors the runtime cancellation projection and is
// always serialized so false clears stale client state.
CancellationPending bool `json:"cancellation_pending"`
// CancellationRevision identifies the process-local cancellation transition
// generation represented by CancellationPending.
CancellationRevision uint64 `json:"cancellation_revision"`
// PendingAction is the compact per-session projection used when the
// session transcript is not loaded in the client.
PendingAction *string `json:"pending_action"`
@@ -817,6 +830,19 @@ type ForegroundActivityProvider interface {
ForegroundActivity(sessionID string) v1.ForegroundActivity
}
// CancellationPendingProvider surfaces the orchestrator's runtime cancellation
// projection without coupling task serialization to the orchestrator package.
type CancellationPendingProvider interface {
CancellationPending(sessionID string) bool
}
// CancellationPendingSnapshotProvider is the atomic form of the cancellation
// projection used by serialization boundaries that need ordering identity.
// Implementations must read the boolean and revision from one critical section.
type CancellationPendingSnapshotProvider interface {
CancellationPendingSnapshot(sessionID string) (pending bool, revision uint64)
}
type ActiveSubagentCountProvider interface {
ActiveSubagentCount(sessionID string) int
}
@@ -0,0 +1,231 @@
package handlers
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/stretchr/testify/require"
"github.com/kandev/kandev/internal/task/dto"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/service"
ws "github.com/kandev/kandev/pkg/websocket"
)
type fakeCancellationPendingProvider struct {
pending bool
}
func (p fakeCancellationPendingProvider) CancellationPending(string) bool {
return p.pending
}
type orchestratorWithCancellation struct {
captureOrchestrator
pending bool
}
type messageOrchestratorWithCancellation struct {
firstTurnCaptureOrchestrator
pendingBySession map[string]bool
}
type cancellationListRepo struct {
mockRepository
sessionsByTask []*models.TaskSession
}
func (r *cancellationListRepo) ListTaskSessions(context.Context, string) ([]*models.TaskSession, error) {
return r.sessionsByTask, nil
}
func (r *cancellationListRepo) CountToolCallMessagesBySession(context.Context, []string) (map[string]int, error) {
return nil, nil
}
func (o *orchestratorWithCancellation) CancellationPending(string) bool {
return o.pending
}
func (o messageOrchestratorWithCancellation) CancellationPending(sessionID string) bool {
return o.pendingBySession[sessionID]
}
func TestNewTaskHandlers_DerivesCancellationPendingProvider(t *testing.T) {
withCancellation := &orchestratorWithCancellation{pending: true}
h := NewTaskHandlers(nil, withCancellation, nil, nil, newTestLogger(t))
require.NotNil(t, h.cancellationPending)
require.True(t, h.cancellationPending.CancellationPending("session-1"))
plain := &captureOrchestrator{}
h2 := NewTaskHandlers(nil, plain, nil, nil, newTestLogger(t))
require.Nil(t, h2.cancellationPending)
}
func TestNewMessageHandlers_DerivesCancellationPendingProvider(t *testing.T) {
withCancellation := &messageOrchestratorWithCancellation{
pendingBySession: map[string]bool{"session-1": true},
}
h := NewMessageHandlers(nil, withCancellation, newTestLogger(t))
require.NotNil(t, h.cancellationPending)
require.True(t, h.cancellationPending.CancellationPending("session-1"))
plain := &firstTurnCaptureOrchestrator{}
h2 := NewMessageHandlers(nil, plain, newTestLogger(t))
require.Nil(t, h2.cancellationPending)
}
func TestHTTPGetTaskSession_StampsCancellationPending(t *testing.T) {
svc, _ := newSessionHandlerService(t, &models.TaskSession{
ID: "sess-cancel",
State: models.TaskSessionStateRunning,
})
h := &TaskHandlers{
service: svc,
cancellationPending: fakeCancellationPendingProvider{pending: true},
logger: newTestLogger(t),
}
resp := doGetTaskSession(t, h, "sess-cancel")
require.True(t, resp.Session.CancellationPending)
}
func TestHTTPListTaskSessions_StampsCancellationPending(t *testing.T) {
gin.SetMode(gin.TestMode)
session := &models.TaskSession{ID: "sess-cancel", TaskID: "task-1", State: models.TaskSessionStateRunning}
repo := &cancellationListRepo{
mockRepository: mockRepository{sessions: map[string]*models.TaskSession{session.ID: session}},
sessionsByTask: []*models.TaskSession{session},
}
svc := service.NewService(service.Repos{
Workspaces: repo, Tasks: repo, TaskRepos: repo, Workflows: repo,
Messages: repo, Turns: repo, Sessions: repo, GitSnapshots: repo,
RepoEntities: repo, Executors: repo, Environments: repo,
TaskEnvironments: repo, Reviews: repo,
}, nil, newTestLogger(t), service.RepositoryDiscoveryConfig{})
h := &TaskHandlers{
service: svc,
repo: repo,
cancellationPending: fakeCancellationPendingProvider{pending: true},
logger: newTestLogger(t),
}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodGet, "/tasks/task-1/sessions", nil).WithContext(context.Background())
c.Params = gin.Params{{Key: "id", Value: "task-1"}}
h.httpListTaskSessions(c)
require.Equal(t, http.StatusOK, rec.Code, rec.Body.String())
var response dto.ListTaskSessionSummariesResponse
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &response))
require.Len(t, response.Sessions, 1)
require.True(t, response.Sessions[0].CancellationPending)
}
func TestWSListTaskSessions_StampsCancellationPending(t *testing.T) {
session := &models.TaskSession{ID: "sess-cancel", TaskID: "task-1", State: models.TaskSessionStateRunning}
repo := &cancellationListRepo{
mockRepository: mockRepository{sessions: map[string]*models.TaskSession{session.ID: session}},
sessionsByTask: []*models.TaskSession{session},
}
svc := service.NewService(service.Repos{
Workspaces: repo, Tasks: repo, TaskRepos: repo, Workflows: repo,
Messages: repo, Turns: repo, Sessions: repo, GitSnapshots: repo,
RepoEntities: repo, Executors: repo, Environments: repo,
TaskEnvironments: repo, Reviews: repo,
}, nil, newTestLogger(t), service.RepositoryDiscoveryConfig{})
h := &TaskHandlers{
service: svc,
cancellationPending: fakeCancellationPendingProvider{pending: true},
logger: newTestLogger(t),
}
request, err := ws.NewRequest("list", ws.ActionTaskSessionList, map[string]string{"task_id": "task-1"})
require.NoError(t, err)
response, err := h.doListTaskSessions(context.Background(), request, "task-1")
require.NoError(t, err)
var body dto.ListTaskSessionSummariesResponse
require.NoError(t, response.ParsePayload(&body))
require.Len(t, body.Sessions, 1)
require.True(t, body.Sessions[0].CancellationPending)
}
func newMessageCancellationService(t *testing.T, repo *messageAddSwitchRepo) *service.Service {
t.Helper()
return service.NewService(service.Repos{
Workspaces: repo, Tasks: repo, TaskRepos: repo,
Workflows: repo, Messages: repo, Turns: repo,
Sessions: repo, GitSnapshots: repo, RepoEntities: repo,
Executors: repo, Environments: repo, TaskEnvironments: repo,
Reviews: repo,
}, nil, newTestLogger(t), service.RepositoryDiscoveryConfig{})
}
func TestResolveSessionAfterTurnStart_EnrichesSubmittedSessionCancellation(t *testing.T) {
now := time.Now().UTC()
repo := &messageAddSwitchRepo{
tasks: map[string]*models.Task{"task-1": {ID: "task-1"}},
sessions: map[string]*models.TaskSession{
"session-1": {ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateWaitingForInput, UpdatedAt: now},
},
primaryID: "session-1",
}
h := NewMessageHandlers(newMessageCancellationService(t, repo), &messageOrchestratorWithCancellation{
firstTurnCaptureOrchestrator: firstTurnCaptureOrchestrator{},
pendingBySession: map[string]bool{"session-1": true},
}, newTestLogger(t))
current := &dto.GetTaskSessionResponse{Session: dto.FromTaskSession(repo.sessions["session-1"])}
response, err := h.resolveSessionAfterTurnStart(context.Background(), "task-1", "session-1", current)
require.NoError(t, err)
require.Equal(t, "session-1", response.Session.ID)
require.True(t, response.Session.CancellationPending)
}
func TestResolveSessionAfterTurnStart_EnrichesReplacementSessionCancellation(t *testing.T) {
now := time.Now().UTC()
repo := &messageAddSwitchRepo{
tasks: map[string]*models.Task{"task-1": {ID: "task-1"}},
sessions: map[string]*models.TaskSession{
"session-1": {ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateCompleted, UpdatedAt: now},
"session-2": {ID: "session-2", TaskID: "task-1", State: models.TaskSessionStateWaitingForInput, UpdatedAt: now},
},
primaryID: "session-2",
}
h := NewMessageHandlers(newMessageCancellationService(t, repo), &messageOrchestratorWithCancellation{
firstTurnCaptureOrchestrator: firstTurnCaptureOrchestrator{},
pendingBySession: map[string]bool{"session-1": true, "session-2": false},
}, newTestLogger(t))
current := &dto.GetTaskSessionResponse{Session: dto.FromTaskSession(repo.sessions["session-1"])}
response, err := h.resolveSessionAfterTurnStart(context.Background(), "task-1", "session-1", current)
require.NoError(t, err)
require.Equal(t, "session-2", response.Session.ID)
require.False(t, response.Session.CancellationPending)
}
func TestCheckSessionStateForMessage_EnrichesCancellation(t *testing.T) {
repo := &messageAddSwitchRepo{
tasks: map[string]*models.Task{"task-1": {ID: "task-1"}},
sessions: map[string]*models.TaskSession{
"session-1": {ID: "session-1", TaskID: "task-1", State: models.TaskSessionStateWaitingForInput},
},
}
h := NewMessageHandlers(newMessageCancellationService(t, repo), &messageOrchestratorWithCancellation{
firstTurnCaptureOrchestrator: firstTurnCaptureOrchestrator{},
pendingBySession: map[string]bool{"session-1": true},
}, newTestLogger(t))
request, err := ws.NewRequest("check", ws.ActionMessageAdd, nil)
require.NoError(t, err)
response, wsErr := h.checkSessionStateForMessage(context.Background(), request, "session-1")
require.Nil(t, wsErr)
require.True(t, response.Session.CancellationPending)
}
var _ dto.CancellationPendingProvider = fakeCancellationPendingProvider{}
@@ -45,12 +45,13 @@ type taskTitleSessionClaimer interface {
// MessageHandlers handles WebSocket requests for messages
type MessageHandlers struct {
service *service.Service
orchestrator OrchestratorService
logger *logger.Logger
referenceValidator entityrefs.SubmissionValidator
messageIDMu sync.Mutex
messageIDGates map[string]*messageIDGate
service *service.Service
orchestrator OrchestratorService
cancellationPending dto.CancellationPendingProvider
logger *logger.Logger
referenceValidator entityrefs.SubmissionValidator
messageIDMu sync.Mutex
messageIDGates map[string]*messageIDGate
}
type messageIDGate struct {
@@ -73,6 +74,9 @@ func NewMessageHandlers(
if len(validators) > 0 {
handlers.referenceValidator = validators[0]
}
if cancellation, ok := orchestrator.(dto.CancellationPendingProvider); ok {
handlers.cancellationPending = cancellation
}
return handlers
}
@@ -512,7 +516,9 @@ func (h *MessageHandlers) resolveSessionAfterTurnStart(
return nil, errors.New("failed to reload submitted session after on_turn_start")
}
if reloaded.State != models.TaskSessionStateCompleted {
return &dto.GetTaskSessionResponse{Session: dto.FromTaskSession(reloaded)}, nil
sessionDTO := dto.FromTaskSession(reloaded)
dto.EnrichCancellationPending(&sessionDTO, h.cancellationPending)
return &dto.GetTaskSessionResponse{Session: sessionDTO}, nil
}
primary, err := h.service.GetPrimarySession(ctx, taskID)
if err != nil || primary == nil {
@@ -527,7 +533,9 @@ func (h *MessageHandlers) resolveSessionAfterTurnStart(
if primary.ID == submittedSessionID {
return nil, errors.New("submitted session completed during on_turn_start but remains primary")
}
return &dto.GetTaskSessionResponse{Session: dto.FromTaskSession(primary)}, nil
sessionDTO := dto.FromTaskSession(primary)
dto.EnrichCancellationPending(&sessionDTO, h.cancellationPending)
return &dto.GetTaskSessionResponse{Session: sessionDTO}, nil
}
func (h *MessageHandlers) errorForBlockedMessageSession(msg *ws.Message, sessionID string, state models.TaskSessionState) *ws.Message {
@@ -578,6 +586,7 @@ func (h *MessageHandlers) checkSessionStateForMessage(ctx context.Context, msg *
return nil, wsErr
}
sessionDTO := dto.FromTaskSession(session)
dto.EnrichCancellationPending(&sessionDTO, h.cancellationPending)
resp := &dto.GetTaskSessionResponse{Session: sessionDTO}
if wsErr := h.errorForBlockedMessageSession(msg, sessionID, sessionDTO.State); wsErr != nil {
if sessionDTO.State == models.TaskSessionStateRunning {
@@ -62,7 +62,11 @@ func TestHTTPListQuickChatSessions(t *testing.T) {
Executors: repo, Environments: repo, TaskEnvironments: repo,
Reviews: repo,
}, nil, log, service.RepositoryDiscoveryConfig{})
h := &TaskHandlers{service: svc, logger: log}
h := &TaskHandlers{
service: svc,
logger: log,
cancellationPending: fakeCancellationPendingProvider{pending: true},
}
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
@@ -85,6 +89,7 @@ func TestHTTPListQuickChatSessions(t *testing.T) {
}, body.Sessions[0])
require.Len(t, body.TaskSessions, 1)
assert.Equal(t, "task-chat", body.TaskSessions[0].TaskID)
assert.True(t, body.TaskSessions[0].CancellationPending)
}
// TestHTTPListQuickChatSessionsEmptyIsArray keeps the empty response an array
@@ -34,6 +34,7 @@ type TaskHandlers struct {
service *service.Service
orchestrator OrchestratorStarter
foregroundActivity dto.ForegroundActivityProvider
cancellationPending dto.CancellationPendingProvider
repo handlerRepo
planService *service.PlanService
handoffSvc *service.HandoffService
@@ -108,6 +109,9 @@ func NewTaskHandlers(svc *service.Service, orchestrator OrchestratorStarter, rep
if fa, ok := orchestrator.(dto.ForegroundActivityProvider); ok {
h.foregroundActivity = fa
}
if cancellation, ok := orchestrator.(dto.CancellationPendingProvider); ok {
h.cancellationPending = cancellation
}
return h
}
@@ -384,6 +384,7 @@ func pendingActionPtr(
func (h *TaskHandlers) taskSessionDTO(ctx context.Context, session *models.TaskSession) dto.TaskSessionDTO {
result := dto.FromTaskSession(session)
dto.EnrichCancellationPending(&result, h.cancellationPending)
if !isInputCapableSession(session) {
return result
}
@@ -437,6 +438,7 @@ func (h *TaskHandlers) httpListTaskSessions(c *gin.Context) {
for _, session := range sessions {
summary := dto.FromTaskSessionSummary(session)
dto.EnrichForegroundActivitySummary(&summary, h.foregroundActivity)
dto.EnrichCancellationPendingSummary(&summary, h.cancellationPending)
summary.PendingAction = pendingActionPtr(&session.ID, pendingActionsBySession)
sessionDTOs = append(sessionDTOs, summary)
ids = append(ids, session.ID)
@@ -1764,7 +1766,9 @@ func (h *TaskHandlers) httpListQuickChatSessions(c *gin.Context) {
Name: item.Name,
AgentProfileID: item.AgentProfileID,
})
response.TaskSessions = append(response.TaskSessions, dto.FromTaskSession(item.Session))
sessionDTO := dto.FromTaskSession(item.Session)
dto.EnrichCancellationPending(&sessionDTO, h.cancellationPending)
response.TaskSessions = append(response.TaskSessions, sessionDTO)
}
c.JSON(http.StatusOK, response)
}
@@ -48,6 +48,7 @@ func (h *TaskHandlers) doListTaskSessions(ctx context.Context, msg *ws.Message,
for _, session := range sessions {
summary := dto.FromTaskSessionSummary(session)
dto.EnrichForegroundActivitySummary(&summary, h.foregroundActivity)
dto.EnrichCancellationPendingSummary(&summary, h.cancellationPending)
summary.PendingAction = pendingActionPtr(&session.ID, pendingActionsBySession)
sessionDTOs = append(sessionDTOs, summary)
}
+1
View File
@@ -208,6 +208,7 @@ const (
ActionSessionMessageDeleted = "session.message.deleted"
ActionSessionStateChanged = "session.state_changed"
ActionSessionActivityChanged = "session.activity_changed"
ActionSessionCancellationChanged = "session.cancellation_changed"
ActionTaskStatusSummaryUpdated = "task.status_summary.updated"
ActionSessionAgentctlStarting = "session.agentctl_starting"
ActionSessionAgentctlReady = "session.agentctl_ready"
@@ -23,6 +23,9 @@ const appState = {
renameQuickChatSession,
setQuickChatInitialPrompt,
setTaskSession,
taskSessions: {
items: {} as Record<string, { cancellation_pending?: boolean }>,
},
agentProfiles: {
items: [
{ id: CONFIG_PROFILE_ID, cli_passthrough: false },
@@ -60,6 +63,10 @@ beforeEach(() => {
{ id: CONFIG_PROFILE_ID, cli_passthrough: false },
{ id: PASSTHROUGH_PROFILE_ID, cli_passthrough: true },
];
appState.taskSessions.items = {};
setTaskSession.mockImplementation((session: { id: string; cancellation_pending?: boolean }) => {
appState.taskSessions.items[session.id] = session;
});
startConfigChat.mockResolvedValue({ task_id: TASK_ID, session_id: SESSION_ID });
});
@@ -81,6 +88,9 @@ describe("useConfigChat unified launch", () => {
expect(setTaskSession).toHaveBeenCalledWith(
expect.objectContaining({ id: SESSION_ID, task_id: TASK_ID }),
);
expect(appState.taskSessions.items[SESSION_ID]).toEqual(
expect.objectContaining({ cancellation_pending: false }),
);
expect(closeQuickChatSession).toHaveBeenCalledWith(
getQuickChatSetupSessionId(WORKSPACE_ID, "config"),
);
@@ -79,6 +79,7 @@ function registerStartedSession({
id: toSessionId(response.session_id),
task_id: toTaskId(response.task_id),
state: "CREATED",
cancellation_pending: false,
started_at: now,
updated_at: now,
agent_profile_id: toAgentProfileId(agentProfileId),
@@ -185,6 +185,7 @@ function DesktopRightSection(props: {
<SubmitButton
isAgentBusy={props.isAgentBusy}
canCancelAgent={props.canCancelAgent}
sessionId={props.sessionId}
hasContent={props.hasContent}
isDisabled={props.isDisabled}
submitDisabledReason={props.submitDisabledReason}
@@ -188,6 +188,7 @@ export function MobileChatInputToolbar(props: MobileToolbarProps) {
<SubmitButton
isAgentBusy={props.isAgentBusy}
canCancelAgent={props.canCancelAgent}
sessionId={props.sessionId}
hasContent={props.hasContent}
isDisabled={props.isDisabled}
submitDisabledReason={props.submitDisabledReason}
@@ -1,6 +1,6 @@
"use client";
import { useCallback, useRef, useState } from "react";
import { useCallback } from "react";
import {
IconArrowUp,
IconFileTextSpark,
@@ -12,7 +12,7 @@ import {
import { GridSpinner } from "@/components/grid-spinner";
import { KeyboardShortcutTooltip } from "@/components/keyboard-shortcut-tooltip";
import { useAppStore } from "@/components/state-provider";
import { useAppStore, useAppStoreApi } from "@/components/state-provider";
import { useTouchDrawer } from "@/hooks/use-compact-task-chrome";
import { Button } from "@kandev/ui/button";
import { Drawer, DrawerContent, DrawerHeader, DrawerTitle, DrawerTrigger } from "@kandev/ui/drawer";
@@ -26,6 +26,7 @@ import type { MCPAttachmentHistory } from "@/lib/state/slices/session-runtime/ty
type SubmitButtonProps = {
isAgentBusy: boolean;
canCancelAgent?: boolean;
sessionId: string | null;
hasContent: boolean;
isDisabled: boolean;
submitDisabledReason?: string;
@@ -97,6 +98,7 @@ function SendSubmitButton({
export function SubmitButton({
isAgentBusy,
canCancelAgent = isAgentBusy,
sessionId,
hasContent,
isDisabled,
submitDisabledReason,
@@ -107,26 +109,31 @@ export function SubmitButton({
submitShortcut,
}: SubmitButtonProps) {
const showSendButton = !isAgentBusy || hasContent;
const [isCancelling, setIsCancelling] = useState(false);
const storeApi = useAppStoreApi();
const isCancelling = useAppStore((state) => {
if (!sessionId) return false;
return (
state.taskSessions.items[sessionId]?.cancellation_pending === true ||
state.chatInput.cancellingBySessionId[sessionId] === true
);
});
const tooltipDescription = submitTooltipDescription(
isAgentBusy,
planModeEnabled,
submitDisabledReason,
);
const cancellingRef = useRef(false);
const handleCancelClick = useCallback(async () => {
if (cancellingRef.current) return;
cancellingRef.current = true;
setIsCancelling(true);
if (!sessionId) return;
if (storeApi.getState().chatInput.cancellingBySessionId[sessionId]) return;
storeApi.getState().setCancelTurnPending(sessionId, true);
try {
await onCancel();
} catch (error) {
console.error("Failed to cancel agent turn:", error);
} finally {
cancellingRef.current = false;
setIsCancelling(false);
storeApi.getState().setCancelTurnPending(sessionId, false);
}
}, [onCancel]);
}, [onCancel, sessionId, storeApi]);
return (
<div className="flex items-center gap-1">
@@ -1,5 +1,7 @@
import { describe, it, expect, vi, afterEach, beforeEach } from "vitest";
import { render, screen, fireEvent, act, cleanup } from "@testing-library/react";
import { StateProvider } from "@/components/state-provider";
import type { TaskSession } from "@/lib/types/http";
const responsiveMock = vi.hoisted(() => ({
breakpoint: "desktop" as "mobile" | "tablet" | "compactDesktop" | "desktop",
@@ -94,6 +96,10 @@ vi.mock("./reset-context-button", () => ({
ResetContextButton: () => <button type="button">Reset context</button>,
}));
vi.mock("./chat-input-plugin-actions", () => ({
ChatInputPluginActions: () => null,
}));
vi.mock("./voice-input-button", () => ({
VoiceInputButton: () => <button type="button">Voice</button>,
}));
@@ -102,6 +108,8 @@ import { ChatInputToolbar } from "./chat-input-toolbar";
import type { ChatInputToolbarProps } from "./chat-input-toolbar";
const MOBILE_TOOLBAR_TEST_ID = "mobile-chat-input-toolbar";
const CANCEL_AGENT_BUTTON_TEST_ID = "cancel-agent-button";
const SESSION_TIMESTAMP = "2026-01-01T00:00:00Z";
function deferred<T>() {
let resolve!: (v: T) => void;
@@ -113,56 +121,175 @@ function deferred<T>() {
function renderToolbar(onCancel: () => void | Promise<void>) {
return render(
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy
isDisabled={false}
isSending={false}
onCancel={onCancel}
onSubmit={() => {}}
minimalToolbar
/>,
<StateProvider>
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy
isDisabled={false}
isSending={false}
onCancel={onCancel}
onSubmit={() => {}}
minimalToolbar
/>
</StateProvider>,
);
}
function renderFullToolbar(overrides: Partial<ChatInputToolbarProps> = {}) {
return render(
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy={false}
hasContent
isDisabled={false}
isSending={false}
onCancel={() => {}}
onSubmit={() => {}}
hidePlanMode
mcpServers={["filesystem"]}
contextCount={2}
onAttachFiles={() => {}}
onEnhancePrompt={() => {}}
isUtilityConfigured
{...overrides}
/>,
<StateProvider>
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy={false}
hasContent
isDisabled={false}
isSending={false}
onCancel={() => {}}
onSubmit={() => {}}
hidePlanMode
mcpServers={["filesystem"]}
contextCount={2}
onAttachFiles={() => {}}
onEnhancePrompt={() => {}}
isUtilityConfigured
{...overrides}
/>
</StateProvider>,
);
}
function makeRunningSession(cancellationPending: boolean): TaskSession {
return {
id: "s1",
task_id: "t1",
state: "RUNNING",
cancellation_pending: cancellationPending,
started_at: SESSION_TIMESTAMP,
updated_at: SESSION_TIMESTAMP,
} as TaskSession;
}
describe("ChatInputToolbar backend cancellation state", () => {
it("renders backend-owned pending state after store hydration", () => {
render(
<StateProvider initialState={{ taskSessions: { items: { s1: makeRunningSession(true) } } }}>
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy
isDisabled={false}
isSending={false}
onCancel={() => {}}
onSubmit={() => {}}
hasContent={false}
/>
</StateProvider>,
);
const button = screen.getByTestId(CANCEL_AGENT_BUTTON_TEST_ID) as HTMLButtonElement;
expect(button.disabled).toBe(true);
expect(button.querySelector('[role="status"]')).toBeTruthy();
});
it("uses backend pending state as the duplicate-click guard", () => {
const onCancel = vi.fn();
render(
<StateProvider initialState={{ taskSessions: { items: { s1: makeRunningSession(true) } } }}>
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy
isDisabled={false}
isSending={false}
onCancel={onCancel}
onSubmit={() => {}}
hasContent={false}
/>
</StateProvider>,
);
const button = screen.getByTestId(CANCEL_AGENT_BUTTON_TEST_ID) as HTMLButtonElement;
fireEvent.click(button);
expect(onCancel).not.toHaveBeenCalled();
expect(button.disabled).toBe(true);
expect(button.querySelector('[role="status"]')).toBeTruthy();
});
});
// The cancel button must disable itself while a cancel request is in flight.
// Without this guard, an impatient user clicking it repeatedly while the agent
// tears down a long-running tool (Claude Monitor, etc.) sends N cancel requests
// to the backend, each producing a duplicate "Turn cancelled by user" message.
describe("ChatInputToolbar cancel button", () => {
it.each(["desktop", "mobile"] as const)(
"keeps cancellation progress after a %s toolbar remount",
async (breakpoint) => {
responsiveMock.breakpoint = breakpoint;
const { promise, resolve } = deferred<void>();
const onCancel = vi.fn(() => promise);
let mounted = true;
const renderToolbarTree = () => (
<StateProvider>
{mounted ? (
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy
isDisabled={false}
isSending={false}
onCancel={onCancel}
onSubmit={() => {}}
hasContent={false}
/>
) : null}
</StateProvider>
);
const view = render(renderToolbarTree());
fireEvent.click(screen.getByTestId(CANCEL_AGENT_BUTTON_TEST_ID));
await act(async () => {});
mounted = false;
view.rerender(renderToolbarTree());
mounted = true;
view.rerender(renderToolbarTree());
const remountedButton = screen.getByTestId(CANCEL_AGENT_BUTTON_TEST_ID) as HTMLButtonElement;
expect(remountedButton.disabled).toBe(true);
expect(remountedButton.querySelector('[role="status"]')).toBeTruthy();
expect(onCancel).toHaveBeenCalledTimes(1);
await act(async () => {
resolve();
await promise;
});
expect(remountedButton.disabled).toBe(false);
},
);
it("keeps the queue affordance without a cancel control after clarification detaches", () => {
renderFullToolbar({ isAgentBusy: true, canCancelAgent: false });
expect(screen.queryByTestId("cancel-agent-button")).toBeNull();
expect(screen.queryByTestId(CANCEL_AGENT_BUTTON_TEST_ID)).toBeNull();
expect(screen.getByTestId("submit-message-button")).toBeTruthy();
expect(screen.getByText("Queue message")).toBeTruthy();
});
@@ -173,7 +300,7 @@ describe("ChatInputToolbar cancel button", () => {
renderToolbar(onCancel);
const button = screen.getByTestId("cancel-agent-button") as HTMLButtonElement;
const button = screen.getByTestId(CANCEL_AGENT_BUTTON_TEST_ID) as HTMLButtonElement;
expect(button.disabled).toBe(false);
fireEvent.click(button);
@@ -203,7 +330,7 @@ describe("ChatInputToolbar cancel button", () => {
const onCancel = vi.fn(() => promise.then(() => Promise.reject(new Error("network"))));
renderToolbar(onCancel);
const button = screen.getByTestId("cancel-agent-button") as HTMLButtonElement;
const button = screen.getByTestId(CANCEL_AGENT_BUTTON_TEST_ID) as HTMLButtonElement;
fireEvent.click(button);
await act(async () => {});
@@ -221,21 +348,23 @@ describe("ChatInputToolbar cancel button", () => {
describe("ChatInputToolbar submit button", () => {
it("shows the setup-disabled reason while keeping the submit button disabled", () => {
render(
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy={false}
hasContent
isDisabled
submitDisabledReason="The agent is still being set up."
isSending={false}
onCancel={() => {}}
onSubmit={() => {}}
minimalToolbar
/>,
<StateProvider>
<ChatInputToolbar
planModeEnabled={false}
onPlanModeChange={() => {}}
sessionId="s1"
taskId="t1"
taskDescription=""
isAgentBusy={false}
hasContent
isDisabled
submitDisabledReason="The agent is still being set up."
isSending={false}
onCancel={() => {}}
onSubmit={() => {}}
minimalToolbar
/>
</StateProvider>,
);
expect((screen.getByTestId("submit-message-button") as HTMLButtonElement).disabled).toBe(true);
@@ -66,6 +66,7 @@ export type ChatInputToolbarProps = {
function MinimalToolbar({
isAgentBusy,
canCancelAgent,
sessionId,
hasContent,
isDisabled,
submitDisabledReason,
@@ -77,6 +78,7 @@ function MinimalToolbar({
ChatInputToolbarProps,
| "isAgentBusy"
| "canCancelAgent"
| "sessionId"
| "hasContent"
| "isDisabled"
| "submitDisabledReason"
@@ -91,6 +93,7 @@ function MinimalToolbar({
<SubmitButton
isAgentBusy={isAgentBusy}
canCancelAgent={canCancelAgent}
sessionId={sessionId}
hasContent={hasContent ?? false}
isDisabled={isDisabled}
submitDisabledReason={submitDisabledReason}
@@ -129,6 +132,7 @@ export const ChatInputToolbar = memo(function ChatInputToolbar(rawProps: ChatInp
<MinimalToolbar
isAgentBusy={props.isAgentBusy}
canCancelAgent={props.canCancelAgent}
sessionId={props.sessionId}
hasContent={props.hasContent}
isDisabled={props.isDisabled}
submitDisabledReason={props.submitDisabledReason}
+18
View File
@@ -40,6 +40,24 @@ export async function waitForActiveSessionForegroundActivity(
);
}
/** Wait for the backend-owned cancellation projection on the active session. */
export async function waitForActiveSessionCancellationPending(
page: Page,
pending: boolean,
): Promise<void> {
await page.waitForFunction(
(expected) => {
const store = (window as E2EStoreWindow).__KANDEV_E2E_STORE__;
if (!store) return false;
const sessionId = store.getState().tasks.activeSessionId;
if (!sessionId) return false;
return store.getState().taskSessions.items[sessionId]?.cancellation_pending === expected;
},
pending,
{ timeout: 20_000 },
);
}
/**
* Simulate a lean session-list / partial WS update: preserve `is_passthrough`
* but drop `agent_profile_snapshot` from the client store.
@@ -0,0 +1,70 @@
import { test, expect } from "../../fixtures/test-base";
import { waitForSessionDone, seedIdleSession } from "../../helpers/session";
import { waitForActiveSessionCancellationPending } from "../../helpers/session-store";
test.describe("Cancel progress across task switches", () => {
test("keeps backend-owned cancel progress across task switches and reloads", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(120_000);
const taskB = await apiClient.createTaskWithAgent(
seedData.workspaceId,
"Cancel progress B",
seedData.agentProfileId,
{
description: "/e2e:simple-message",
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
},
);
if (!taskB.session_id) throw new Error("Task B did not return a session_id");
await waitForSessionDone(
apiClient,
taskB.id,
taskB.session_id,
"Task B should finish its seed turn before switching to it",
);
const session = await seedIdleSession(testPage, apiClient, seedData, "Cancel progress A");
await session.sendMessage("/slow 8s");
const activeCancel = session.activeChat().getByTestId("cancel-agent-button");
await expect(activeCancel).toBeVisible({ timeout: 15_000 });
await activeCancel.click();
await waitForActiveSessionCancellationPending(testPage, true);
await expect(activeCancel).toBeDisabled();
await expect(activeCancel.getByRole("status", { name: "Loading" })).toBeVisible();
await expect(session.sidebar.getByText("Cancel progress B", { exact: true })).toBeVisible({
timeout: 15_000,
});
await session.clickTaskInSidebar("Cancel progress B");
await expect(testPage).toHaveURL(new RegExp(`/t/${taskB.id}$`));
await session.showSessionContext();
await expect(session.idleInput()).toBeVisible({ timeout: 15_000 });
await session.clickTaskInSidebar("Cancel progress A");
await expect(testPage).not.toHaveURL(new RegExp(`/t/${taskB.id}$`));
await session.showSessionContext();
const remountedCancel = session.activeChat().getByTestId("cancel-agent-button");
await expect(remountedCancel).toBeVisible({ timeout: 15_000 });
await expect(remountedCancel).toBeDisabled();
await expect(remountedCancel.getByRole("status", { name: "Loading" })).toBeVisible();
await testPage.reload();
await session.waitForLoad();
const reloadedCancel = session.activeChat().getByTestId("cancel-agent-button");
await expect(reloadedCancel).toBeVisible({ timeout: 15_000 });
await expect(reloadedCancel).toBeDisabled();
await expect(reloadedCancel.getByRole("status", { name: "Loading" })).toBeVisible();
await expect(session.idleInput()).toBeVisible({ timeout: 30_000 });
await waitForActiveSessionCancellationPending(testPage, false);
await expect(reloadedCancel).not.toBeVisible({ timeout: 15_000 });
});
});
@@ -0,0 +1,35 @@
import { test, expect } from "../../fixtures/test-base";
import { seedIdleSession } from "../../helpers/session";
import { waitForActiveSessionCancellationPending } from "../../helpers/session-store";
test.describe("Mobile cancel progress across reloads", () => {
test("keeps backend-owned cancel progress after a reload", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(120_000);
const session = await seedIdleSession(testPage, apiClient, seedData, "Mobile cancel progress");
await session.sendMessageViaButton("/slow 8s");
const cancel = session.activeChat().getByTestId("cancel-agent-button");
await expect(cancel).toBeVisible({ timeout: 15_000 });
await cancel.tap();
await waitForActiveSessionCancellationPending(testPage, true);
await expect(cancel).toBeDisabled();
await expect(cancel.getByRole("status", { name: "Loading" })).toBeVisible();
await testPage.reload();
await session.waitForLoad();
const reloadedCancel = session.activeChat().getByTestId("cancel-agent-button");
await expect(reloadedCancel).toBeVisible({ timeout: 15_000 });
await expect(reloadedCancel).toBeDisabled();
await expect(reloadedCancel.getByRole("status", { name: "Loading" })).toBeVisible();
await expect(session.idleInput()).toBeVisible({ timeout: 30_000 });
await waitForActiveSessionCancellationPending(testPage, false);
await expect(reloadedCancel).not.toBeVisible({ timeout: 15_000 });
});
});
@@ -101,11 +101,44 @@ function syncPrepareProgress(draft: any, session: TaskSession) {
if (prepareState) draft.prepareProgress.bySessionId[session.id] = prepareState;
}
/** Merge the runtime cancellation projection using its process-local revision. */
function mergeCancellationProjection(
existing: TaskSession,
incoming: TaskSession,
): Pick<TaskSession, "cancellation_pending" | "cancellation_revision"> {
const incomingRevision = incoming.cancellation_revision;
const existingRevision = existing.cancellation_revision;
const incomingIsCurrent =
incomingRevision !== undefined &&
(existingRevision === undefined || incomingRevision >= existingRevision);
if (incomingIsCurrent) {
return {
cancellation_pending: incoming.cancellation_pending ?? existing.cancellation_pending,
cancellation_revision: incomingRevision,
};
}
if (incomingRevision === undefined && existingRevision === undefined) {
return {
cancellation_pending: incoming.cancellation_pending ?? existing.cancellation_pending,
cancellation_revision: existingRevision,
};
}
return {
cancellation_pending: existing.cancellation_pending,
cancellation_revision: existingRevision,
};
}
/** Merge an incoming session update with an existing session, preserving nullable fields. */
function mergeTaskSession(existing: TaskSession, incoming: TaskSession): TaskSession {
const cancellation = mergeCancellationProjection(existing, incoming);
return {
...existing,
...incoming,
...cancellation,
agent_profile_snapshot: incoming.agent_profile_snapshot ?? existing.agent_profile_snapshot,
worktree_id: incoming.worktree_id ?? existing.worktree_id,
worktree_path: incoming.worktree_path ?? existing.worktree_path,
@@ -79,6 +79,17 @@ describe("upsertTaskSessionFromEvent", () => {
expect(session.repository_id).toBe("repo-1");
});
it("preserves backend cancellation state across unrelated partial events", () => {
const store = makeStore();
store
.getState()
.upsertTaskSessionFromEvent(TASK_ID, makeSession({ cancellation_pending: true }));
store.getState().upsertTaskSessionFromEvent(TASK_ID, makeSession({ state: "RUNNING" }));
expect(store.getState().taskSessions.items[SESSION_ID].cancellation_pending).toBe(true);
});
it("seeds environmentIdBySessionId when task_environment_id is present", () => {
const store = makeStore();
@@ -149,6 +160,37 @@ describe("upsertTaskSessionFromEvent", () => {
});
});
describe("cancellation revision ordering", () => {
it("rejects a stale REST cancellation snapshot after a newer live event", () => {
const store = makeStore();
store
.getState()
.upsertTaskSessionFromEvent(
TASK_ID,
makeSession({ cancellation_pending: true, cancellation_revision: 1 }),
);
store
.getState()
.upsertTaskSessionFromEvent(
TASK_ID,
makeSession({ cancellation_pending: false, cancellation_revision: 2 }),
);
// A delayed REST response captured during the older pending generation must
// not restore true after the newer live false event has settled.
store
.getState()
.setTaskSessionsForTask(TASK_ID, [
makeSession({ cancellation_pending: true, cancellation_revision: 1 }),
]);
const session = store.getState().taskSessions.items[SESSION_ID];
expect(session.cancellation_pending).toBe(false);
expect(session.cancellation_revision).toBe(2);
});
});
describe("setTaskSessionsForTask preserves WS-seeded fields", () => {
it("merges incoming sessions with existing rows so task_environment_id is not clobbered", () => {
const store = makeStore();
+3
View File
@@ -54,6 +54,8 @@ export type MobileSessionState = {
export type ChatInputState = {
planModeBySessionId: Record<string, boolean>;
/** True while a session's agent.cancel request is awaiting settlement. */
cancellingBySessionId: Record<string, boolean>;
};
export type TranscriptAutoScrollState = {
@@ -223,6 +225,7 @@ export type UISliceActions = {
setMobileSessionReview: (sessionId: string, mrKey: string | null) => void;
setMobileSessionTaskSwitcherOpen: (open: boolean) => void;
setPlanMode: (sessionId: string, enabled: boolean) => void;
setCancelTurnPending: (sessionId: string, pending: boolean) => void;
setTranscriptAutoScrollEnabled: (sessionId: string, enabled: boolean) => void;
setTranscriptScrollTop: (sessionId: string, scrollTop: number) => void;
setTranscriptVirtuosoState: (sessionId: string, state: StateSnapshot) => void;
@@ -44,6 +44,24 @@ function makeSidebarView(id: string, name: string): SidebarView {
};
}
describe("cancel-turn progress", () => {
it("tracks pending cancellation independently per session and clears one entry", () => {
const store = makeStore();
store.getState().setCancelTurnPending("session-a", true);
expect(store.getState().chatInput.cancellingBySessionId).toEqual({ "session-a": true });
expect(store.getState().chatInput.cancellingBySessionId["session-b"]).toBeUndefined();
store.getState().setCancelTurnPending("session-b", true);
store.getState().setCancelTurnPending("session-a", false);
expect(store.getState().chatInput.cancellingBySessionId).toEqual({ "session-b": true });
store.getState().setCancelTurnPending("session-b", false);
expect(store.getState().chatInput.cancellingBySessionId).toEqual({});
});
});
describe("migrateView", () => {
it("retains all supported boolean filter dimensions", () => {
const view = makeSidebarView("view-a", "View A");
+9 -1
View File
@@ -87,7 +87,7 @@ export const defaultUIState: UISliceState = {
reviewMRKeyBySessionId: {},
isTaskSwitcherOpen: false,
},
chatInput: { planModeBySessionId: {} },
chatInput: { planModeBySessionId: {}, cancellingBySessionId: {} },
transcriptAutoScroll: {
enabledBySessionId: {},
scrollTopBySessionId: {},
@@ -464,6 +464,14 @@ export const createUISlice: StateCreator<UISlice, [["zustand/immer", never]], []
set((draft) => {
draft.chatInput.planModeBySessionId[sessionId] = enabled;
}),
setCancelTurnPending: (sessionId, pending) =>
set((draft) => {
if (pending) {
draft.chatInput.cancellingBySessionId[sessionId] = true;
} else {
delete draft.chatInput.cancellingBySessionId[sessionId];
}
}),
setTranscriptAutoScrollEnabled: (sessionId, enabled) =>
set((draft) => {
draft.transcriptAutoScroll.enabledBySessionId[sessionId] = enabled;
+1
View File
@@ -328,6 +328,7 @@ export type AppState = KanbanSlice & {
setMobileSessionReview: (sessionId: string, mrKey: string | null) => void;
setMobileSessionTaskSwitcherOpen: (open: boolean) => void;
setPlanMode: (sessionId: string, enabled: boolean) => void;
setCancelTurnPending: UIA["setCancelTurnPending"];
setTranscriptAutoScrollEnabled: UIA["setTranscriptAutoScrollEnabled"];
setTranscriptScrollTop: UIA["setTranscriptScrollTop"];
setTranscriptVirtuosoState: UIA["setTranscriptVirtuosoState"];
+14
View File
@@ -285,6 +285,10 @@ export type TaskSessionStateChangedPayload = {
// live flips arrive on session.activity_changed.
foreground_activity?: ForegroundActivity | null;
active_subagent_count?: number;
/** Backend-owned cancellation projection carried by state snapshots. */
cancellation_pending?: boolean;
/** Process-local cancellation transition generation carried by state snapshots. */
cancellation_revision?: number;
};
/**
@@ -299,6 +303,12 @@ export type TaskSessionActivityChangedPayload = {
active_subagent_count: number;
};
export type TaskSessionCancellationChangedPayload = {
session_id: string;
cancellation_pending: boolean;
cancellation_revision: number;
};
export type TaskSessionNotificationPayload = {
task_id: string;
session_id: string;
@@ -553,6 +563,10 @@ export type BackendMessageMap = OfficeBackendMessageMap &
"session.activity_changed",
TaskSessionActivityChangedPayload
>;
"session.cancellation_changed": BackendMessage<
"session.cancellation_changed",
TaskSessionCancellationChangedPayload
>;
"session.clarification_requested": BackendMessage<
"session.clarification_requested",
TaskSessionNotificationPayload
+4
View File
@@ -436,6 +436,10 @@ export type TaskSession = ActiveSubagentCountFields & {
worktrees?: TaskSessionWorktree[];
task_environment_id?: string;
state: TaskSessionState;
/** Backend-owned runtime cancellation projection; API responses include it explicitly. */
cancellation_pending?: boolean;
/** Process-local cancellation transition generation used to reject stale snapshots. */
cancellation_revision?: number;
/** Fine-grained busy substate; background may outlive the foreground turn (ADR-0049). */
foreground_activity?: ForegroundActivity | null;
/** Compact pending-input projection used when this session's messages are unloaded. */
+115 -1
View File
@@ -5,9 +5,10 @@ import type { StoreApi } from "zustand";
import type { AppState } from "@/lib/state/store";
import { createAppStore } from "@/lib/state/store";
import { deriveSessionInputMode } from "@/hooks/domains/session/session-input-mode";
import type { TaskSession } from "@/lib/types/http";
import { sessionId as toSessionId, type TaskSession } from "@/lib/types/http";
import type {
TaskSessionActivityChangedPayload,
TaskSessionCancellationChangedPayload,
TaskSessionStateChangedPayload,
} from "@/lib/types/backend";
@@ -47,6 +48,7 @@ function makeStore(overrides: Record<string, unknown> = {}) {
const STATE_CHANGED_EVENT = "session.state_changed";
const ACTIVITY_EVENT = "session.activity_changed";
const CANCELLATION_EVENT = "session.cancellation_changed";
const RECOVERABLE_ERROR_MESSAGE = "peer disconnected before response";
const RECOVERABLE_ERROR_AT = "2026-06-14T14:06:40Z";
const TASK_ROOT = "/task-root";
@@ -73,6 +75,15 @@ function makeActivityMessage(
};
}
function makeCancellationMessage(payload: TaskSessionCancellationChangedPayload) {
return {
id: "m-cancel",
type: "notification" as const,
action: "session.cancellation_changed" as const,
payload,
};
}
function makeRealActivityStore(state: TaskSession["state"] = "WAITING_FOR_INPUT") {
const selected = {
id: "s-1",
@@ -214,6 +225,109 @@ describe("session.state_changed handler", () => {
});
});
describe("session.state_changed cancellation snapshot", () => {
it("merges an explicit cancellation false from the authoritative snapshot", () => {
const store = makeStore({
taskSessions: {
items: {
"s-1": {
id: "s-1",
task_id: "t-1",
state: "RUNNING",
cancellation_pending: true,
},
},
},
});
const handler = registerTaskSessionHandlers(store)[STATE_CHANGED_EVENT]!;
handler(
makeMessage({
task_id: "t-1",
session_id: "s-1",
new_state: "RUNNING",
cancellation_pending: false,
cancellation_revision: 2,
}),
);
expect(store.getState().upsertTaskSessionFromEvent).toHaveBeenCalledWith(
"t-1",
expect.objectContaining({ cancellation_pending: false, cancellation_revision: 2 }),
);
});
});
describe("session.cancellation_changed handler", () => {
it("updates only the addressed session in the real store", () => {
const store = createAppStore();
const selected = {
id: "s-1",
task_id: "t-1",
state: "RUNNING",
cancellation_pending: false,
started_at: RECOVERABLE_ERROR_AT,
updated_at: RECOVERABLE_ERROR_AT,
} as TaskSession;
const peer = { ...selected, id: toSessionId("s-2") };
store.getState().setTaskSession(selected);
store.getState().setTaskSession(peer);
const handler = registerTaskSessionHandlers(store)[CANCELLATION_EVENT]!;
handler(
makeCancellationMessage({
session_id: "s-1",
cancellation_pending: true,
cancellation_revision: 1,
}),
);
expect(store.getState().taskSessions.items["s-1"].cancellation_pending).toBe(true);
expect(store.getState().taskSessions.items["s-1"].cancellation_revision).toBe(1);
expect(store.getState().taskSessions.items["s-2"].cancellation_pending).toBe(false);
});
it("ignores an event for a session that has not been loaded", () => {
const store = makeStore();
const handler = registerTaskSessionHandlers(store)[CANCELLATION_EVENT]!;
handler(
makeCancellationMessage({
session_id: "unknown",
cancellation_pending: true,
cancellation_revision: 1,
}),
);
expect(store.getState().upsertTaskSessionFromEvent).not.toHaveBeenCalled();
});
it("rejects a lower-revision live event", () => {
const store = createAppStore();
store.getState().setTaskSession({
id: "s-1",
task_id: "t-1",
state: "RUNNING",
cancellation_pending: false,
cancellation_revision: 2,
started_at: RECOVERABLE_ERROR_AT,
updated_at: RECOVERABLE_ERROR_AT,
} as TaskSession);
const handler = registerTaskSessionHandlers(store)[CANCELLATION_EVENT]!;
handler(
makeCancellationMessage({
session_id: "s-1",
cancellation_pending: true,
cancellation_revision: 1,
}),
);
expect(store.getState().taskSessions.items["s-1"].cancellation_pending).toBe(false);
expect(store.getState().taskSessions.items["s-1"].cancellation_revision).toBe(2);
});
});
describe("session.workspace_sources.updated handler", () => {
it("adopts the workspace root and bumps the Files refresh key", () => {
const setTaskSession = vi.fn();
+72 -6
View File
@@ -7,6 +7,7 @@ import {
taskId as toTaskId,
type SessionId,
type TaskId,
type TaskSession,
type TaskSessionState,
} from "@/lib/types/http";
import type { QueuedMessage } from "@/lib/state/slices/session/types";
@@ -236,6 +237,10 @@ function buildSessionUpdate(payload: any): Record<string, unknown> {
update.foreground_activity = payload.foreground_activity;
if (payload.active_subagent_count !== undefined)
update.active_subagent_count = payload.active_subagent_count;
if (payload.cancellation_pending !== undefined)
update.cancellation_pending = payload.cancellation_pending;
if (payload.cancellation_revision !== undefined)
update.cancellation_revision = payload.cancellation_revision;
return update;
}
@@ -259,6 +264,8 @@ function upsertTaskSessionList(
id: sessionId,
task_id: taskId,
state: (newState ?? existing?.state) as TaskSessionState,
cancellation_pending: existing?.cancellation_pending ?? false,
cancellation_revision: existing?.cancellation_revision ?? 0,
started_at: existing?.started_at ?? "",
updated_at: (sessionUpdate.updated_at as string | undefined) ?? existing?.updated_at ?? "",
...(payload.agent_profile_id ? { agent_profile_id: payload.agent_profile_id } : {}),
@@ -379,6 +386,22 @@ function maybeAdoptSessionOnTransition(
* otherwise stay empty and env-routed shell terminals stall on
* "Connecting terminal...". Routes through `upsertTaskSessionFromEvent`
* which calls `syncEnvironmentMapping` and merges with any existing row. */
function sessionSeedFields(existing: TaskSession | undefined): {
state: TaskSessionState;
cancellation_pending: boolean;
cancellation_revision: number;
started_at: string;
updated_at: string;
} {
return {
state: existing?.state ?? "CREATED",
cancellation_pending: existing?.cancellation_pending ?? false,
cancellation_revision: existing?.cancellation_revision ?? 0,
started_at: existing?.started_at ?? "",
updated_at: existing?.updated_at ?? "",
};
}
function syncEnvFromAgentctlPayload(
store: StoreApi<AppState>,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -395,9 +418,7 @@ function syncEnvFromAgentctlPayload(
store.getState().upsertTaskSessionFromEvent(taskId, {
id: sessionId,
task_id: taskId,
state: existing?.state ?? "CREATED",
started_at: existing?.started_at ?? "",
updated_at: existing?.updated_at ?? "",
...sessionSeedFields(existing),
task_environment_id: envId,
...getAgentctlWorktreeFields(payload, isSibling),
workspace_path: payload.workspace_path ?? payload.task_workspace_path ?? payload.worktree_path,
@@ -544,6 +565,8 @@ function applyForegroundActivity(
id: sessionId,
task_id: taskId,
state: existing.state,
cancellation_pending: existing.cancellation_pending,
cancellation_revision: existing.cancellation_revision,
started_at: existing.started_at ?? "",
updated_at: existing.updated_at ?? "",
foreground_activity: payload.foreground_activity ?? null,
@@ -554,6 +577,32 @@ function applyForegroundActivity(
});
}
/** Apply the backend-owned cancellation projection to the addressed session. */
function applyCancellationPending(
store: StoreApi<AppState>,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
payload: any,
): void {
if (
!payload?.session_id ||
typeof payload.cancellation_pending !== "boolean" ||
typeof payload.cancellation_revision !== "number"
)
return;
const sessionId = toSessionId(payload.session_id);
const existing = store.getState().taskSessions.items[sessionId];
if (!existing) return;
store.getState().upsertTaskSessionFromEvent(existing.task_id, {
id: sessionId,
task_id: existing.task_id,
state: existing.state,
started_at: existing.started_at ?? "",
updated_at: existing.updated_at ?? "",
cancellation_pending: payload.cancellation_pending,
cancellation_revision: payload.cancellation_revision,
});
}
function handleWorkspaceSourcesUpdated(
store: StoreApi<AppState>,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -574,6 +623,22 @@ function handleWorkspaceSourcesUpdated(
store.getState().bumpSessionCommitsRefetch(sessionId);
}
function handleForegroundActivityMessage(
store: StoreApi<AppState>,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
payload: any,
): void {
applyForegroundActivity(store, payload);
}
function handleCancellationPendingMessage(
store: StoreApi<AppState>,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
payload: any,
): void {
applyCancellationPending(store, payload);
}
export function registerTaskSessionHandlers(store: StoreApi<AppState>): WsHandlers {
return {
"message.queue.status_changed": (message) => {
@@ -647,9 +712,10 @@ export function registerTaskSessionHandlers(store: StoreApi<AppState>): WsHandle
maybeFanOutOfficeRefetch(store, newState, existingSession?.state);
},
"session.activity_changed": (message) => {
applyForegroundActivity(store, message.payload);
},
"session.activity_changed": (message) =>
handleForegroundActivityMessage(store, message.payload),
"session.cancellation_changed": (message) =>
handleCancellationPendingMessage(store, message.payload),
"session.agentctl_starting": (message) => {
const payload = message.payload;
if (!payload?.session_id) return;
@@ -0,0 +1,79 @@
# ADR-2026-08-03-backend-owned-cancellation-progress: Keep Cancellation Progress Backend Owned
**Status:** accepted
**Date:** 2026-08-03
**Area:** backend, frontend, protocol
## Context
An accepted turn cancellation is a backend operation that can outlive the React component,
application route, WebSocket connection, or browser page that initiated it. A frontend-only pending
flag can preserve progress across component remounts, but it disappears on reload, is invisible to
other clients, and can disagree with the orchestrator's existing per-session cancellation guard.
At the same time, the durable `TaskSession.state=RUNNING` value controls prompt admission and
workflow behavior, so cancellation progress cannot safely become a new coarse lifecycle state.
## Decision
The orchestrator owns a session-scoped `cancellation_pending` runtime projection. It is derived from
the existing `cancelOperations` registry and exposed through a narrow
`CancellationPending(sessionID string) bool` provider plus an atomic
`CancellationPendingSnapshot(sessionID string) (bool, uint64)` form for serialization. The first
accepted cancellation reference sets the projection; the last reference clears it. A process-local
per-session revision increments on each 0→1 or 1→0 transition. The count, revision, and ordered
publication queue are mutated in one critical section; event-bus sends drain outside that section.
The existing per-session guard remains the owner of deduplicating lifecycle cancellation work.
After session authorization succeeds, `CancelAgent` continues the accepted operation independently
of cancellation of the initiating WebSocket request, while retaining the lifecycle manager's
existing bounded cancellation and escalation timeouts.
The backend publishes first-begin and last-end transitions as the semantic event
`task_session.cancellation_changed`, mapped to the session-scoped WebSocket action
`session.cancellation_changed`. Each event and every full/summary session DTO, task-detail boot
hydration, REST session read, and initial `session.state_changed` subscription snapshot carries both
the explicit `cancellation_pending` boolean and its `cancellation_revision`. The frontend rejects a
snapshot or live value with a lower revision than the session's current value, so delayed REST/boot
responses cannot restore an older boolean after a newer live transition. Missed live notifications
are therefore repaired by the next authoritative snapshot. In accordance with
[ADR-2026-08-01-separate-task-summary-session-stream-traffic](2026-08-01-separate-task-summary-session-stream-traffic.md),
the operation state stays on the opened session stream and is not copied into task summaries.
The projection remains orthogonal to `TaskSession.state` and is not stored in `task_sessions` or
session metadata. A backend restart clears it; startup session/execution recovery remains
authoritative, and a session still reported as `RUNNING` may be cancelled again. The frontend may
show a bounded optimistic flag before backend acceptance, but backend hydration and live updates
own progress after that boundary and across page lifecycles.
## Consequences
Task switches, reloads, replacement tabs, and second clients render the same active cancellation as
long as the backend operation is alive. Closing the initiating page no longer aborts an accepted
cancellation. The session lifecycle and prompt-admission rules stay unchanged, and no schema
migration or restart reconciliation for a synthetic operation marker is required.
The API and WebSocket session contracts gain one explicit boolean, one process-local revision, and
one semantic notification. Every cancellation exit path must clear the reference count, and
serialization tests must cover both `true` and explicit `false` plus revision ordering so stale
client state cannot survive hydration. A backend restart does not claim that an unrecoverable
process operation is still pending; users may see the cancel control become retryable when the
recovered coarse session remains running.
## Alternatives Considered
- **Keep cancellation progress only in the frontend application store.** Rejected because it fixes
route remounts but not reloads, replacement tabs, second clients, or disagreement with the real
backend operation.
- **Persist a cancellation marker in `task_sessions` or session metadata.** Rejected because the
process cancellation has no durable continuation token. A backend crash could strand a stale
marker that claims work is pending when no operation can finish it.
- **Add `CANCELLING` to `TaskSession.state`.** Rejected because the coarse state participates in
prompt admission, task projection, workflow transitions, and recovery. Cancellation is a
temporary operation overlay, not a new durable lifecycle phase.
- **Persist the flag in browser storage.** Rejected because it would remain client-owned, diverge
across tabs, and risk showing stale progress after the backend operation settled.
- **Order snapshots with database timestamps or transport arrival time.** Rejected because the
cancellation projection is runtime-only: session `updated_at` does not change for every runtime
transition, and independent REST/WebSocket delivery can arrive in either order. A process-local
per-session revision is the smallest identity that orders the boolean without pretending the
operation is durable.
+1
View File
@@ -109,3 +109,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or
| 2026-08-02-agent-terminal-diagnostics-over-stderr | [Capture Agent Terminal Diagnostics From Managed Stderr](2026-08-02-agent-terminal-diagnostics-over-stderr.md) | accepted | backend, frontend, protocol, security | 2026-08-02 |
| 2026-08-02-class-aware-git-subprocess-admission | [Class-Aware Git Subprocess Admission](2026-08-02-class-aware-git-subprocess-admission.md) | accepted | backend, agentctl, observability | 2026-08-02 |
| 2026-08-02-isolate-replaceable-session-stream-traffic | [Isolate Replaceable Session Stream Traffic](2026-08-02-isolate-replaceable-session-stream-traffic.md) | accepted | backend, frontend, protocol | 2026-08-02 |
| 2026-08-03-backend-owned-cancellation-progress | [Keep Cancellation Progress Backend Owned](2026-08-03-backend-owned-cancellation-progress.md) | accepted | backend, frontend, protocol | 2026-08-03 |
+216
View File
@@ -0,0 +1,216 @@
---
spec: docs/specs/ui/cancel-turn-progress.md
created: 2026-08-03
status: completed
---
# Implementation Plan: Backend-owned cancel-turn progress
## Overview
Promote the orchestrator's existing per-session cancellation registry into the authoritative
runtime projection described by
[ADR-2026-08-03](../../decisions/2026-08-03-backend-owned-cancellation-progress.md). Publish the
projection through live session events and every session hydration path, then make the shared cancel
control combine that backend value with its existing short-lived optimistic request state. Replace
the request-held browser regression with backend-accepted task-switch and reload coverage on desktop
and mobile.
## Root cause and superseded direction
`SubmitButton` originally owned cancellation progress in component-local state, so a task-route
remount lost it. Tasks 01 and 02 moved that state into the application UI store and proved SPA
navigation, but the store is still scoped to one loaded browser page. It cannot hydrate a reload or
second tab and it treats the request promise, rather than the orchestrator operation, as the source
of truth.
The existing `Service.cancelOperations` registry already tracks actual cancellation intent and the
existing `cancelInFlight` guard already deduplicates lifecycle work. The revised implementation
exposes that fact instead of adding another lifecycle state or a durable database marker. The Task
01 store remains useful only as immediate optimistic feedback before backend acceptance. Runtime
transitions carry a process-local per-session revision so independent REST, boot, and WebSocket
delivery can be ordered without relying on the database timestamp.
## Backend
### Cancellation operation ownership
- `apps/backend/internal/orchestrator/service.go`: document the cancellation registry as the source
of the public runtime projection; keep its reference-counted, per-session ownership separate from
the broader `cancelInFlight` mutex.
- `apps/backend/internal/orchestrator/task_operations.go`: expose
`CancellationPending(sessionID string) bool` and the atomic
`CancellationPendingSnapshot(sessionID string) (bool, uint64)` form; make
`beginCancelInFlight` publish only the `0 -> 1` and `1 -> 0` transitions; increment the
process-local revision and append the transition under the same critical section; drain each
session's queue outside the lock; retain idempotent release and duplicate-request reference
counting.
- In `Service.CancelAgent`, authorize with the caller context and then use
`context.WithoutCancel(ctx)` for the accepted cancellation, reconciliation, message, turn, and
transition publications. This preserves page-disconnect behavior without removing the lifecycle
manager's own ten-second cancel wait and escalation bounds.
- Publish `{session_id, cancellation_pending, cancellation_revision}` on
`events.TaskSessionCancellationChanged`. Event-publication failure is logged but never changes the
cancellation result; the final deferred release must clear pending on every error path.
### Session wire contract and routing
- `apps/backend/internal/events/types.go` and `apps/backend/pkg/websocket/actions.go`: add the
semantic event/action pair `task_session.cancellation_changed` /
`session.cancellation_changed`.
- `apps/backend/internal/gateway/websocket/task_notifications.go`: subscribe to the event and route
it only through `Hub.BroadcastToSession`. It is semantic session traffic, not replaceable stream
traffic and not a task-summary update.
- `apps/backend/internal/task/dto/dto.go`: add non-omitempty `CancellationPending bool` and
`CancellationRevision uint64` fields to both `TaskSessionDTO` and `TaskSessionSummaryDTO`; add a
narrow atomic snapshot provider alongside `CancellationPendingProvider` and enrichers that always
write the provider result.
- `apps/backend/internal/task/handlers/task_handlers.go` and
`task_http_handlers.go`: discover the provider from the orchestrator and enrich list/get session
responses alongside foreground activity.
### Boot, REST, and subscription reconciliation
- `apps/backend/internal/backendapp/boot_state.go`: enrich every task-detail and quick-chat session
DTO before inserting it into `taskSessions.items` and `taskSessionsByTask`.
- `apps/backend/internal/backendapp/helpers.go`: pass a cancellation provider into
`buildSessionDataProvider` and add explicit `cancellation_pending` to the authoritative
`session.state_changed` subscription snapshot.
- `apps/backend/internal/backendapp/main.go`: wire `orchestratorSvc` into the session data provider.
A fresh page or a client that missed a live transition must therefore reconcile without waiting
for another cancel event.
- No repository, schema, migration, or session-metadata change is planned. A backend restart resets
the runtime projection and leaves existing coarse session/execution recovery authoritative.
## Frontend
### Session contracts and live updates
- `apps/web/lib/types/http.ts`: add the backend-owned `cancellation_pending` and optional
`cancellation_revision` fields to `TaskSession`; backend wire boundaries emit both explicitly.
- `apps/web/lib/types/backend.ts`: add both cancellation fields to the state-snapshot payload,
define the revision-bearing `TaskSessionCancellationChangedPayload`, and register
`session.cancellation_changed` in the backend message map.
- `apps/web/lib/ws/handlers/agent-session.ts`: merge explicit true and false values carried by
`session.state_changed`; handle revision-bearing `session.cancellation_changed` by updating the
existing session row and its per-task list without changing coarse state or triggering task/Office
lifecycle work; reject lower-revision snapshots/events.
- `apps/web/lib/ws/handlers/agent-session.test.ts`: cover explicit-false clearing, session isolation,
an event arriving for an unloaded session, and state-snapshot/live-event convergence.
### Shared cancel control
- `apps/web/components/task/chat/chat-input-toolbar-primitives.tsx`: derive effective progress from
`taskSessions.items[sessionId].cancellation_pending ||
chatInput.cancellingBySessionId[sessionId]`. Keep the current UI-slice value as immediate
optimistic feedback and a same-page duplicate-click guard; it must not override an authoritative
backend `true` after the request promise settles.
- `apps/web/components/task/chat/chat-input-toolbar.test.tsx`: retain desktop/mobile remount coverage
and add a new-store hydration case proving a backend-pending session renders the spinner after a
page boundary. Prove explicit backend false plus cleared optimistic state returns to idle.
- The existing `chatInput` store action remains transient and is not added to boot state or browser
persistence.
### Mobile design contract
Desktop and mobile continue using the same inline `SubmitButton`; there is no new surface, copy,
hierarchy, safe-area behavior, scroll owner, or gesture. The backend session field drives both
layouts. Because reload persistence changes user-visible behavior on compact screens, a dedicated
`mobile-chrome` Playwright regression is required in addition to the shared component coverage.
## Tests
- **Backend operation lifecycle:** `apps/backend/internal/orchestrator/task_operations_test.go`
blocks the mock manager, observes pending true, cancels the initiating context, and proves the
accepted operation remains pending until release and clears on success/error. Concurrent requests
must still invoke the manager once and publish one true/false pair.
- **DTO projection:** add focused tests under `apps/backend/internal/task/dto/` for full and summary
serialization of explicit true and false values.
- **REST and boot hydration:** extend
`apps/backend/internal/task/handlers/task_http_handlers_test.go` and
`apps/backend/internal/backendapp/helpers_test.go` to assert current cancellation state in list,
get, task-detail boot, and initial session-subscription payloads.
- **WebSocket routing:** extend
`apps/backend/internal/gateway/websocket/task_notifications_test.go` to prove cancellation events
reach only the subscribed session and preserve both transition values.
- **Frontend reconciliation:** extend
`apps/web/lib/ws/handlers/agent-session.test.ts` and
`apps/web/components/task/chat/chat-input-toolbar.test.tsx` for hydration, live updates,
revision-aware delayed hydration, optimistic/backend union semantics, session isolation, and
desktop/mobile rendering.
## E2E Tests
- **Task switch and reload:** revise
`apps/web/e2e/tests/chat/cancel-progress-task-switch.spec.ts`. Start the existing slow mock-agent
turn, send cancellation through to the backend, wait until the exposed session store reports
`cancellation_pending=true`, switch away and back, reload the page, and assert the same disabled
animated control until the backend settles. This replaces the current helper that holds the
request before backend acceptance.
- **Compact reload parity:** add
`apps/web/e2e/tests/chat/mobile-cancel-progress-reload.spec.ts`, limited to `mobile-chrome`. Tap
cancel during the slow mock turn, wait for backend-owned pending state, reload, and assert the
shared control remains disabled and animated without desktop-only navigation assumptions.
- **Test support:** add a cancellation-pending reader/waiter to
`apps/web/e2e/helpers/session-store.ts`; remove
`routeMainWebSocketWithHeldCancelRequest` from `apps/web/e2e/helpers/ws-drop.ts` once no test uses
it. The existing `/slow` mock scenario is cancellation-insensitive long enough to keep the
lifecycle operation deterministically pending; no production delay or new test-only endpoint is
required.
## Verification Results
The backend-owned package is complete. Superseded Tasks 01 and 02 below remain as historical
records; revised Tasks 03-06 are the authoritative implementation evidence.
- Backend lifecycle-focused tests: 3 passed; cancellation regression set: 16 passed; race set: 4
passed.
- Backend projection packages (`internal/task/dto`, `internal/task/handlers`, `internal/backendapp`,
`internal/gateway/websocket`): 681 passed.
- Frontend-focused contract/control/store tests: 104 passed; typecheck, lint, and i18n ratchet passed.
- `make -C apps/backend test` and `make -C apps/backend build` completed without failures.
- Desktop backend-owned switch/reload E2E: 1 passed; mobile-chrome reload E2E: 1 passed.
- `git diff --check`: passed.
- No schema migration, durable cancellation marker, cross-workspace broadcast, held WebSocket frame,
or external side effect was added.
## Review remediation results
- RED: deterministic successor-generation ordering, waiting-state caller-cancellation, and delayed
REST hydration tests failed against the pre-fix implementation.
- Green backend: `go test ./internal/orchestrator ./internal/task/dto ./internal/task/handlers
./internal/backendapp ./internal/gateway/websocket -count=1` — 2,138 passed.
- Race: full `go test -race ./internal/orchestrator -count=1` — 1,453 passed; cancellation-focused
race set — 5 passed.
- Frontend: the five focused session/control suites — 134 passed; `pnpm run typecheck`, web lint,
and the i18n ratchet passed.
- `git diff --check` — passed.
The remediation adds a process-local per-session revision to every cancellation snapshot/event,
atomically queues transitions with count changes, rejects lower-revision client merges, and routes
the waiting-state retry through the detached accepted-operation context. No new mobile composition
was introduced; the existing mobile reload E2E covers the shared control, so a new mobile E2E was not
needed for this data-ordering-only change.
## Implementation Waves And Parallel Candidates
Previous implementation record (completed, partially retained as optimistic behavior):
- [x] [task-01-session-scoped-cancel-state](task-01-session-scoped-cancel-state.md) — frontend
navigation state; superseded as the authoritative owner by Tasks 03-05.
- [x] [task-02-task-switch-regression-e2e](task-02-task-switch-regression-e2e.md) — request-held
regression; replaced by Task 06's backend-accepted navigation/reload coverage.
Revised implementation, sequential dependency chain:
- [x] [task-03-backend-cancellation-lifecycle](task-03-backend-cancellation-lifecycle.md) — completed;
establishes runtime ownership and operation transitions.
- [x] [task-04-cancellation-projection-contract](task-04-cancellation-projection-contract.md) —
completed; depends on Task 03 and exposes hydration/live contracts.
- [x] [task-05-backend-owned-cancel-control](task-05-backend-owned-cancel-control.md) — completed;
depends on Task 04 and consumes the backend projection.
- [x] [task-06-cancel-reload-regression-e2e](task-06-cancel-reload-regression-e2e.md) — completed;
depends on Task 05 and proves desktop/mobile navigation and reload behavior.
These tasks are not parallel-safe: each changes or validates the contract established by the
preceding task. The wave list does not authorize subagents.
@@ -0,0 +1,82 @@
---
id: "01-session-scoped-cancel-state"
title: "Session-scoped cancel state"
status: done
wave: 1
depends_on: []
superseded_by: "05-backend-owned-cancel-control"
plan: "plan.md"
spec: "../../specs/ui/cancel-turn-progress.md"
---
# Task 01: Session-scoped cancel state
> Historical implementation record. The UI-slice flag remains only as immediate optimistic
> feedback; Task 05 replaces it as the authoritative owner with backend-hydrated session state.
## Acceptance
- A pending cancellation is represented in the `chatInput` UI slice by session ID and is cleared
when the request succeeds, rejects, times out, or cannot be sent.
- Unmounting and remounting the desktop or compact mobile toolbar under the same `StateProvider`
preserves the disabled spinner state for that session, while another session remains unaffected.
- Repeated activation during the pending request invokes `onCancel` only once; after failure, a
still-visible control can be retried.
## Verification
```bash
(cd apps && pnpm install --frozen-lockfile)
(cd apps && pnpm --filter @kandev/web test -- --run lib/state/slices/ui/ui-slice.test.ts components/task/chat/chat-input-toolbar.test.tsx)
(cd apps/web && pnpm run typecheck)
(cd apps && pnpm --filter @kandev/web run i18n:ratchet)
```
Follow TDD: first add the remount regression and confirm it fails because `SubmitButton` resets its
local state, then move the state into the shared slice and make the focused tests pass.
## Files likely touched
- `apps/web/lib/state/slices/ui/types.ts`
- `apps/web/lib/state/slices/ui/ui-slice.ts`
- `apps/web/lib/state/slices/ui/ui-slice.test.ts`
- `apps/web/components/task/chat/chat-input-toolbar-primitives.tsx`
- `apps/web/components/task/chat/chat-input-toolbar.tsx`
- `apps/web/components/task/chat/chat-input-toolbar-desktop.tsx`
- `apps/web/components/task/chat/chat-input-toolbar-mobile.tsx`
- `apps/web/components/task/chat/chat-input-toolbar.test.tsx`
## Dependencies
None.
## Parallelism
Sequential. The state action, request guard, responsive prop wiring, and regression test form one
shared control contract.
## Inputs
- Spec: `What`, `Persistence guarantees`, `Failure modes`, and `Scenarios`.
- Plan: `Decision`, `Transient chat-input state`, `Shared cancel control`, and `Tests`.
- Existing patterns: `chatInput.planModeBySessionId` in the UI slice and the shared
`SubmitButton` used by desktop, compact mobile, and minimal toolbars.
## Output contract
Report the behavioral change, exact files changed, Red/Green test evidence, typecheck and i18n
results, blockers or risks, and update this task plus `plan.md` status/results in the same primary
conversation.
## Results
- RED: `cd apps && pnpm --filter @kandev/web test -- --run components/task/chat/chat-input-toolbar.test.tsx` —
2 expected remount-regression failures (desktop and mobile) against the local-state implementation.
- `cd apps && pnpm install --frozen-lockfile` — passed.
- `cd apps && pnpm --filter @kandev/web test -- --run lib/state/slices/ui/ui-slice.test.ts components/task/chat/chat-input-toolbar.test.tsx` —
passed, 2 files / 52 tests.
- `cd apps/web && pnpm run typecheck` — first run caught the missing explicit `AppState` action export;
after adding it, rerun passed.
- `cd apps && pnpm --filter @kandev/web run i18n:ratchet` — passed; 0 added and 7 modified files clean.
- `git diff --check` — passed.
- Security/trust or external side effects: none.
@@ -0,0 +1,78 @@
---
id: "02-task-switch-regression-e2e"
title: "Task-switch cancellation regression"
status: done
wave: 2
depends_on: ["01-session-scoped-cancel-state"]
superseded_by: "06-cancel-reload-regression-e2e"
plan: "plan.md"
spec: "../../specs/ui/cancel-turn-progress.md"
---
# Task 02: Task-switch cancellation regression
> Historical implementation record. Holding the browser request proves only frontend-local state;
> Task 06 replaces this flow with backend-accepted task-switch, reload, and mobile regressions.
## Acceptance
- A WebSocket test controller can hold exactly one `agent.cancel` request, forward every unrelated
frame, expose the held-request count, and release the request before teardown.
- The browser regression starts cancellation on task A, switches to task B, returns to task A, and
observes the same disabled animated cancel state until release.
- Releasing the held request completes normal cancellation and leaves no held WebSocket request or
stale progress entry at test end.
## Verification
```bash
(cd apps && pnpm install --frozen-lockfile)
(cd apps/web && pnpm e2e:run --host --project chromium -- tests/chat/cancel-progress-task-switch.spec.ts)
```
Follow TDD: add and run the browser regression against the pre-fix control to demonstrate that the
spinner disappears after the return navigation, then run it against Task 01's implementation.
## Files likely touched
- `apps/web/e2e/helpers/ws-drop.ts`
- `apps/web/e2e/tests/chat/cancel-progress-task-switch.spec.ts`
## Dependencies
Task 01.
## Parallelism
Sequential. This task validates Task 01's state ownership through the complete task-navigation flow.
## Inputs
- Spec: the pending-cancellation navigation scenario and persistence boundary.
- Plan: `E2E Tests` and `Mobile design contract`.
- Existing patterns: `routeMainWebSocketWithMessageAddResponseDrop` in `ws-drop.ts`,
`SessionPage.clickTaskInSidebar`, and the task-switch flow in
`apps/web/e2e/tests/session/multi-session-ux.spec.ts`.
## Output contract
Report the initial failing assertion, final Playwright result and discovered test count, held-frame
cleanup evidence, files changed, blockers or risks, and update this task plus `plan.md`
status/results in the same primary conversation.
## Results
- Initial regression evidence: the Task 01 remount test failed for both desktop and mobile before
the store-backed state was implemented (`disabled` was `false` after remount instead of `true`).
- `cd apps && pnpm install --frozen-lockfile` — passed.
- `cd apps && pnpm --filter @kandev/web test -- --run lib/state/slices/ui/ui-slice.test.ts components/task/chat/chat-input-toolbar.test.tsx` — passed, 2 files / 52 tests.
- `cd apps/web && pnpm run typecheck` — passed.
- `cd apps/web && pnpm e2e:run --host --project chromium -- tests/chat/cancel-progress-task-switch.spec.ts` — passed, 1 test in 13.7s.
- The browser test observed one held `agent.cancel` request, verified the disabled loading control
after returning to task A, released the request, and asserted the held count returned to zero and
the idle composer resumed.
- Files changed: `apps/web/e2e/helpers/ws-drop.ts` and
`apps/web/e2e/tests/chat/cancel-progress-task-switch.spec.ts`.
- No blockers or external side effects found.
- PR fixup commit `1bb1b2539` refined the held-count contract and isolated the component test's
plugin-actions dependency; the focused unit, typecheck, lint, diff, and E2E checks passed again.
@@ -0,0 +1,78 @@
---
id: "03-backend-cancellation-lifecycle"
title: "Backend cancellation lifecycle"
status: completed
wave: 1
depends_on: []
plan: "plan.md"
spec: "../../specs/ui/cancel-turn-progress.md"
---
# Task 03: Backend cancellation lifecycle
## Acceptance
- `orchestrator.Service.CancellationPending(sessionID)` is true from the first accepted cancellation
reference until the final reference settles, isolated by session and false outside that window.
- `CancellationPendingSnapshot(sessionID)` returns the boolean and process-local revision from one
critical section; revisions advance only on first-begin and last-end transitions.
- The transition publication queue is appended while the count/revision are mutated, so a successor
generation cannot publish `true` before its predecessor's `false`.
- An accepted cancellation continues after the initiating request context is cancelled, while the
existing lifecycle timeout/escalation and per-session deduplication remain intact.
- Overlapping requests invoke the agent manager once and publish exactly one pending transition and
one idle transition; success and every error return clear the registry.
## Verification
```bash
(cd apps/backend && go test ./internal/orchestrator -run 'TestCancellationPendingTracksReferencesAndPublishesTransitions|TestCancelAgent_(SurvivesCallerCancellation|DeduplicatesConcurrentCalls|ClearsCancellationPendingOnError)')
make -C apps/backend test
```
Follow TDD: first block the mock agent manager and assert the missing public state/transitions, then
add transition-aware registry publication and detach the accepted work from caller cancellation.
## Files likely touched
- `apps/backend/internal/orchestrator/service.go`
- `apps/backend/internal/orchestrator/task_operations.go`
- `apps/backend/internal/orchestrator/task_operations_test.go`
- `apps/backend/internal/events/types.go`
## Dependencies
None.
## Parallelism
Sequential. Task 04 consumes the provider and event established here.
## Inputs
- Spec: `Data model`, `State machine`, `Failure modes`, and `Persistence guarantees`.
- Plan: `Cancellation operation ownership`.
- Decision: `ADR-2026-08-03-backend-owned-cancellation-progress`.
- Existing patterns: `cancelOperations`, `beginCancelInFlight`, `isCancelInFlight`, the
`cancelInFlight` guard, and `PromptTask`'s use of `context.WithoutCancel` after admission.
## Output contract
Report the state-transition API, context ownership, exact Red/Green tests, duplicate and error-path
evidence, files changed, blockers/risks, and synchronize this task plus `plan.md` in the same primary
conversation.
## Results
Implemented the runtime projection and accepted-operation context boundary.
- Red: `cd apps/backend && go test ./internal/orchestrator -run 'TestCancellationPendingTracksReferencesAndPublishesTransitions|TestCancelAgent_ClearsCancellationPendingOnError|TestCancelAgent_SurvivesCallerCancellation' -count=1` initially failed because the public provider and transition implementation were absent.
- Green: `cd apps/backend && go test ./internal/orchestrator -run 'TestCancellationPendingTracksReferencesAndPublishesTransitions|TestCancelAgent_ClearsCancellationPendingOnError|TestCancelAgent_SurvivesCallerCancellation' -count=1` — 3 passed.
- Regression: `cd apps/backend && go test ./internal/orchestrator -run 'TestCancelAgent|TestHandleAgentBootReady_DoesNotDrainWhileCancelInFlight' -count=1` — 15 passed.
- Race: `cd apps/backend && go test -race ./internal/orchestrator -run 'TestCancellationPending|TestCancelAgent_SurvivesCallerCancellation' -count=1` — 5 passed; full orchestrator race suite — 1,453 passed.
- `git diff --check` — passed.
The registry remains process-local and reference-counted; revisions are retained for the lifetime of
the backend process so delayed snapshots/events cannot reuse an older identity. No schema, repository,
metadata, external side effect, or new trust boundary was added. Event publication failures are logged
and cannot change the cancellation result.
@@ -0,0 +1,83 @@
---
id: "04-cancellation-projection-contract"
title: "Cancellation projection contract"
status: completed
wave: 2
depends_on: ["03-backend-cancellation-lifecycle"]
plan: "plan.md"
spec: "../../specs/ui/cancel-turn-progress.md"
---
# Task 04: Cancellation projection contract
## Acceptance
- Full and summary task-session DTOs always serialize `cancellation_pending` and
`cancellation_revision`, including explicit false/zero, from the narrow orchestrator provider
without a repository/schema write.
- Task-detail boot state, task-session REST list/get responses, and the initial session-subscription
snapshot expose the provider's current value, so a fresh client is correct before another live
transition occurs.
- `task_session.cancellation_changed` maps to `session.cancellation_changed` and is delivered only
to clients subscribed to that session; publishing failure cannot fail cancellation.
## Verification
```bash
(cd apps/backend && go test ./internal/task/dto ./internal/task/handlers ./internal/backendapp ./internal/gateway/websocket -run 'Cancellation|CancelPending|SessionDataProvider')
make -C apps/backend test
```
Follow TDD: add explicit-false DTO and fresh-load assertions first, then wire provider enrichment and
the session-scoped semantic notification until each boundary passes.
## Files likely touched
- `apps/backend/internal/task/dto/dto.go`
- `apps/backend/internal/task/dto/cancellation_pending_test.go`
- `apps/backend/internal/task/handlers/task_handlers.go`
- `apps/backend/internal/task/handlers/task_http_handlers.go`
- `apps/backend/internal/task/handlers/task_http_handlers_test.go`
- `apps/backend/internal/backendapp/boot_state.go`
- `apps/backend/internal/backendapp/helpers.go`
- `apps/backend/internal/backendapp/helpers_test.go`
- `apps/backend/internal/backendapp/main.go`
- `apps/backend/pkg/websocket/actions.go`
- `apps/backend/internal/gateway/websocket/task_notifications.go`
- `apps/backend/internal/gateway/websocket/task_notifications_test.go`
## Dependencies
Task 03.
## Parallelism
Sequential. It exposes Task 03's state and defines the contract Task 05 consumes.
## Inputs
- Spec: `API surface`, `Permissions`, `Failure modes`, and reload/second-client scenarios.
- Plan: `Session wire contract and routing` and `Boot, REST, and subscription reconciliation`.
- Existing patterns: `ForegroundActivityProvider`, `EnrichForegroundActivity`, task-handler provider
discovery, `addTaskDetailSessionsState`, `buildSessionDataProvider`, and
`TaskSessionActivityChanged` routing.
## Output contract
Report every contract shape and enrichment boundary, explicit-false and routing test evidence,
exact files/commands, blockers/risks, and synchronize this task plus `plan.md` in the same primary
conversation.
## Results
Implemented the runtime projection contract across DTO, REST, boot, subscription, and WebSocket
boundaries.
- `cd apps/backend && go test ./internal/task/dto ./internal/task/handlers ./internal/backendapp ./internal/gateway/websocket -count=1` — 685 passed.
- `cd apps/backend && go test ./internal/task/dto ./internal/task/handlers ./internal/backendapp ./internal/gateway/websocket -run 'Cancellation|CancelPending|SessionDataProvider|TaskEventBroadcaster_CancellationIsSessionScoped|AppendSessionStateMessage_IncludesCancellationPending' -count=1` — 11 passed.
- `git diff --check` — passed.
Full and summary DTOs serialize explicit `false`/`0`; task-detail/quick-chat boot state, REST list/get,
and the initial session subscription all enrich the atomic boolean/revision snapshot from the
orchestrator provider. Live transitions are session-scoped and publication errors are non-fatal. No
schema migration, durable marker, cross-workspace broadcast, or external side effect was added.
@@ -0,0 +1,84 @@
---
id: "05-backend-owned-cancel-control"
title: "Backend-owned cancel control"
status: completed
wave: 3
depends_on: ["04-cancellation-projection-contract"]
plan: "plan.md"
spec: "../../specs/ui/cancel-turn-progress.md"
---
# Task 05: Backend-owned cancel control
## Acceptance
- The frontend session contract hydrates and live-updates explicit true/false backend cancellation
state plus its revision without changing coarse session state, task selection, or Office refetch
behavior.
- Delayed REST/boot hydration with a lower revision cannot overwrite a newer live cancellation event.
- The shared cancel control renders progress when either the short-lived optimistic request flag or
backend `cancellation_pending` is true; backend true survives a fresh `StateProvider`, while both
false render the retryable idle control.
- Desktop, compact mobile, and minimal toolbars remain session-isolated and keep the current
duplicate-click, tooltip, error, and spinner behavior.
## Verification
```bash
(cd apps && pnpm install --frozen-lockfile)
(cd apps && pnpm --filter @kandev/web test -- --run lib/ws/handlers/agent-session.test.ts components/task/chat/chat-input-toolbar.test.tsx lib/state/slices/ui/ui-slice.test.ts)
(cd apps/web && pnpm run typecheck)
(cd apps && pnpm --filter @kandev/web run i18n:ratchet)
```
Follow TDD: first hydrate a new store with `cancellation_pending=true` and deliver revisioned
true/false live events against the existing code, then add the ordered merge contract and
effective-state union.
## Files likely touched
- `apps/web/lib/types/http.ts`
- `apps/web/lib/types/backend.ts`
- `apps/web/lib/ws/handlers/agent-session.ts`
- `apps/web/lib/ws/handlers/agent-session.test.ts`
- `apps/web/components/task/chat/chat-input-toolbar-primitives.tsx`
- `apps/web/components/task/chat/chat-input-toolbar.test.tsx`
## Dependencies
Task 04.
## Parallelism
Sequential. The frontend must compile against Task 04's exact payload and hydration contract.
## Inputs
- Spec: `What`, `Data model`, `API surface`, and desktop/mobile scenarios.
- Plan: `Session contracts and live updates`, `Shared cancel control`, and `Mobile design contract`.
- Existing patterns: `buildSessionUpdate`, `applyForegroundActivity`,
`upsertTaskSessionFromEvent`, `chatInput.cancellingBySessionId`, and the shared `SubmitButton`.
## Output contract
Report the authoritative/optimistic merge rule, Red/Green component and handler evidence, exact
files/commands, responsive coverage, blockers/risks, and synchronize this task plus `plan.md` in the
same primary conversation.
## Results
Implemented the backend-owned session contract and shared desktop/mobile control merge.
- `cd apps && pnpm --filter @kandev/web test -- --run lib/ws/handlers/agent-session.test.ts components/task/chat/chat-input-toolbar.test.tsx lib/state/slices/ui/ui-slice.test.ts` — 3 files, 104 passed.
- `cd apps/web && pnpm run typecheck` — passed.
- `cd apps && pnpm --filter @kandev/web lint` — passed.
- `cd apps && pnpm --filter @kandev/web run i18n:ratchet` — passed (0 added, 11 modified files clean).
- `git diff --check` — passed.
The shared cancel control uses backend `true` OR the transient optimistic request flag, so backend
state survives a fresh store while both cleared values remain retryable. Session merges accept only
the newest `cancellation_revision`, preventing delayed REST/boot snapshots from restoring stale
progress. The frontend wire fields remain optional for partial in-memory/test rows, while backend DTO
and hydration boundaries always serialize them explicitly. No new copy, browser persistence, security
boundary, or external side effect was introduced; desktop and compact mobile continue using the same
`SubmitButton`.
@@ -0,0 +1,78 @@
---
id: "06-cancel-reload-regression-e2e"
title: "Cancel reload regression"
status: completed
wave: 4
depends_on: ["05-backend-owned-cancel-control"]
plan: "plan.md"
spec: "../../specs/ui/cancel-turn-progress.md"
---
# Task 06: Cancel reload regression
## Acceptance
- The desktop regression sends cancellation to the backend during an existing slow mock turn,
observes backend `cancellation_pending=true`, switches tasks, returns, reloads, and sees the same
disabled animated control until cancellation settles.
- A `mobile-chrome` regression taps cancel, waits for backend-owned pending state, reloads, and
observes the same shared-control behavior without relying on desktop-only navigation.
- The old outbound-request hold helper and assertions are removed; teardown leaves no held frames,
stale pending state, or running seeded turn.
## Verification
```bash
(cd apps && pnpm install --frozen-lockfile)
(cd apps/web && pnpm e2e:run --host --project chromium -- tests/chat/cancel-progress-task-switch.spec.ts)
(cd apps/web && pnpm e2e:run --host --project mobile-chrome -- tests/chat/mobile-cancel-progress-reload.spec.ts)
(cd apps/web && pnpm run typecheck)
```
Follow TDD: first forward the cancel request and prove the reloaded control loses progress without
backend hydration, then enable Tasks 03-05 and make both viewport flows pass. Wait on the exposed
backend-derived store field rather than a timeout before reloading.
## Files likely touched
- `apps/web/e2e/helpers/session-store.ts`
- `apps/web/e2e/helpers/ws-drop.ts`
- `apps/web/e2e/tests/chat/cancel-progress-task-switch.spec.ts`
- `apps/web/e2e/tests/chat/mobile-cancel-progress-reload.spec.ts`
## Dependencies
Task 05.
## Parallelism
Sequential. This task validates the complete backend-to-hydration-to-control path.
## Inputs
- Spec: task navigation, reload, session-isolation, failure, and compact-mobile scenarios.
- Plan: `E2E Tests` and `Mobile design contract`.
- Existing patterns: the `/slow` mock-agent command, `SessionPage.clickTaskInSidebar`,
`page.reload()`, `waitForActiveSessionForegroundActivity`, and mobile-only spec naming.
## Output contract
Report the initial failing reload assertion, final desktop/mobile Playwright counts and durations,
seeded-turn settlement/cleanup evidence, exact files/commands, blockers/risks, and synchronize this
task plus `plan.md` in the same primary conversation.
## Results
Replaced request-held coverage with backend-accepted task navigation and reload coverage.
- `cd apps/web && pnpm exec playwright test --config e2e/playwright.config.ts --project=chromium e2e/tests/chat/cancel-progress-task-switch.spec.ts` — 1 passed (12.1s) after rebuilding the backend binary and Vite assets.
- `cd apps/web && pnpm exec playwright test --config e2e/playwright.config.ts --project=mobile-chrome e2e/tests/chat/mobile-cancel-progress-reload.spec.ts` — 1 passed (10.4s).
- `cd apps/web && pnpm run typecheck` — passed.
- `git diff --check` — passed.
The first desktop attempt exposed that the E2E fixture was launching a stale backend binary; rebuilding
`apps/backend/bin/kandev` and `apps/web/dist` made the new live event available. The first mobile
attempt used the desktop keyboard helper; it was corrected to submit through the mobile button. Both
final tests observe backend `cancellation_pending=true`, verify the disabled spinner after
navigation/reload, and wait for the explicit `false` settle. The old held-frame helper was removed,
and each run used only the isolated mock backend/workspace.
+1
View File
@@ -166,6 +166,7 @@ Subscription quota tracking and per-agent cheap-model profile routing.
| [ws-connectivity-warning](ui/ws-connectivity-warning.md) | approved |
| [context-compaction-count](context-compaction-count/spec.md) | approved |
| [context-window reset freshness](context-window-reset-freshness/spec.md) | shipped |
| [cancel-turn-progress](ui/cancel-turn-progress.md) | approved |
## system-page/ — operational diagnostics & maintenance UI
+140
View File
@@ -0,0 +1,140 @@
---
status: approved
created: 2026-08-03
owner: kandev
---
# Backend-owned cancel-turn progress
## Why
Cancelling an agent turn can outlive the chat component or browser page that started it. Users
need the cancel control to keep showing the real operation state after task navigation, a page
reload, or opening the session in another tab, rather than treating one browser component as the
source of truth.
Decision: [ADR-2026-08-03-backend-owned-cancellation-progress](../../decisions/2026-08-03-backend-owned-cancellation-progress.md)
## What
- Activating turn cancellation immediately disables that session's cancel control and shows its
existing progress animation.
- Once the backend accepts the request, the backend owns whether cancellation is pending. Every
client renders the same session-scoped state from backend hydration and live updates.
- Pending progress survives task and session navigation, a full page reload, replacement of the
browser tab, and a second tab while the same backend cancellation operation remains active.
- Cancellation progress is isolated by session; cancelling one session does not animate or disable
another session's control.
- Repeated activation may send a duplicate transport request during a race, but the backend runs at
most one cancellation operation for that session and keeps progress pending until the owning
operation settles.
- Success, reconciliation of a missing or unresponsive execution, rejection, or timeout clears the
backend progress state. A still-running session becomes retryable after a failed attempt.
- Desktop and mobile chat composers expose the same behavior through their shared cancel control.
## Data model
Cancellation progress is a backend runtime projection keyed by task-session ID:
- `cancellation_pending: boolean` is `true` while the orchestrator has an accepted cancellation
operation for the session and `false` otherwise.
- `cancellation_revision: uint64` is a process-local per-session generation. It increments only on
the first accepted reference (`0 → 1`) and the final release (`1 → 0`); overlapping references do
not create extra generations. The backend reads the boolean and revision atomically.
- The field is orthogonal to `TaskSession.state`; cancellation does not introduce a `CANCELLING`
lifecycle state and the session remains `RUNNING` until the existing lifecycle reconciliation
changes it.
- The projection is not written to `task_sessions`, session metadata, browser storage, or another
durable store. It is derived from the orchestrator's live cancellation registry and is serialized
explicitly as both `true` and `false` so hydration can clear stale client state.
- The frontend may retain a session-keyed optimistic request flag between the click and the first
backend signal. The effective control state is optimistic pending OR backend pending; only the
backend field is authoritative after request acceptance and across page boundaries.
## API surface
- The existing `agent.cancel` WebSocket request remains
`{ "session_id": string }`. Its success/error response continues to describe the completed
cancellation attempt; request authorization and error codes do not change.
- Full and summary task-session DTOs add the non-optional wire fields
`cancellation_pending: boolean` and `cancellation_revision: uint64`. The frontend's in-memory
`TaskSession` type may keep both properties optional for partial event/test rows, but backend DTO,
boot, REST, and subscription boundaries always serialize explicit values.
- Task-detail boot state and task-session REST responses include the field on every session row.
- The initial session-subscription `session.state_changed` snapshot includes
`cancellation_pending` and `cancellation_revision`, including explicit `false` and `0` values.
- Live transitions use the semantic WebSocket notification `session.cancellation_changed` with
payload `{ "session_id": string, "cancellation_pending": boolean, "cancellation_revision": number }`.
It is delivered only to clients subscribed to that session; task-summary and task-switcher traffic
do not carry it. Clients reject values whose revision is lower than the session's current revision.
## State machine
| State | Trigger | Result |
|---|---|---|
| Idle | An authorized `agent.cancel` request is accepted | Backend publishes pending and starts or joins the guarded cancellation operation. |
| Pending | The initiating page disconnects or reloads | No transition; the backend operation continues. |
| Pending | A duplicate request arrives | The existing operation remains the sole lifecycle cancellation; progress stays pending. |
| Pending | Cancellation succeeds or the backend reconciles a missing/unresponsive execution | Backend publishes idle after lifecycle, session, message, and turn reconciliation settle. |
| Pending | Cancellation fails or times out | Backend publishes idle and the existing request error behavior remains visible to the initiating client. |
## Permissions
Invoking cancellation continues to use the existing session authorization check. Boot state, REST
session reads, session subscriptions, and live cancellation notifications retain their existing
workspace/session visibility rules.
## Failure modes
- If the WebSocket is unavailable before the backend accepts the request, the optimistic browser
flag clears on request failure and no backend pending state is created.
- If the initiating page disconnects after acceptance, cancellation continues independently of the
request connection. A replacement page hydrates the backend state.
- If a live cancellation notification is missed, the next boot payload, REST session read, or
session-subscription snapshot repairs the client.
- If cancellation fails or reaches its lifecycle timeout, the backend clears pending in all exit
paths. A session still reported as running exposes the cancel control for retry.
- If publishing a live transition fails, cancellation itself still proceeds; an authoritative
snapshot repairs the display on the next read or subscription.
## Persistence guarantees
The state survives React remounts, SPA navigation, full page reloads, browser-tab replacement, and
other clients for as long as the same backend process owns the active cancellation operation. The
accepted operation also ignores cancellation of the initiating WebSocket request.
The runtime projection intentionally does not survive a Kandev backend restart. There is no safe
continuation token for an in-flight process cancellation, so startup exposes no stale pending flag
and existing session/execution recovery determines the durable coarse state. If the session remains
`RUNNING`, the user can issue another idempotent cancellation request.
## Scenarios
- **GIVEN** an agent turn is running, **WHEN** the user activates the cancel control, **THEN** the
control is immediately disabled, shows progress, and the backend runs one guarded cancellation.
- **GIVEN** the backend reports cancellation pending for task A, **WHEN** the user opens task B and
returns to task A, **THEN** task A's cancel control still shows progress and remains disabled.
- **GIVEN** the backend reports cancellation pending, **WHEN** the user reloads or replaces the
page, **THEN** the first hydrated task-session state shows the disabled animated control without
waiting for a new transition.
- **GIVEN** one tab has an accepted cancellation pending, **WHEN** a second authorized tab opens the
same session, **THEN** the second tab shows the same pending control state.
- **GIVEN** one session has a pending cancellation, **WHEN** the user views another running session,
**THEN** the other session's cancel control remains independent and enabled.
- **GIVEN** a cancellation is pending, **WHEN** another cancel request arrives for that session,
**THEN** the backend does not invoke a second lifecycle cancellation and does not clear progress
until the owning operation settles.
- **GIVEN** cancellation rejects or times out while the session remains running, **WHEN** pending is
cleared, **THEN** the control becomes retryable.
- **GIVEN** the compact mobile chat composer is active, **WHEN** cancellation is accepted and the
page reloads, **THEN** its shared cancel control hydrates the same backend-owned progress state.
- **GIVEN** cancellation was pending before a backend restart, **WHEN** Kandev starts again, **THEN**
no stale pending flag is exposed and the recovered session state determines whether cancel can be
retried.
## Out of scope
- Persisting or replaying an in-flight cancellation across a Kandev backend restart.
- Adding `CANCELLING` to the durable session lifecycle or changing prompt-admission semantics.
- Changing the lifecycle cancellation timeout, escalation policy, or terminal task/session result.
- Adding task-list badges, new cancel copy, notifications, navigation, layout, or touch gestures.