fix: make clarification watchdog recovery race-free (#2982)

apps/backend/internal/backendapp/helpers.go

@@ -604,7 +604,14 @@ func registerRoutes(p routeParams) {
 	// external answer_question_kandev/list_pending_questions_kandev MCP tools
 	// (R3: both entry points must race through the same claim).
 	clarificationResolver := clarification.NewResolver(
-		clarificationStore, p.taskRepo, p.msgCreator, p.taskSvc, p.orchestratorSvc, p.eventBus, p.log,
+		clarificationStore,
+		p.taskRepo,
+		p.msgCreator,
+		p.taskSvc,
+		p.orchestratorSvc,
+		p.eventBus,
+		p.orchestratorSvc.HandleClarificationPrimaryAnswered,
+		p.log,
 	)
 
 	// Wire pending clarification requests into the office inbox.

apps/backend/internal/clarification/handlers.go

@@ -96,6 +96,27 @@ type EventBus interface {
 	Publish(ctx context.Context, topic string, event *bus.Event) error
 }
 
+// PrimaryAnswered is the local ordering notification for a primary-path
+// clarification response. The resolver invokes its notifier synchronously after
+// durable live delivery confirmation and before the live waiter is released.
+// The event bus remains a fan-out projection for other consumers.
+type PrimaryAnswered struct {
+	SessionID           string
+	TaskID              string
+	PendingID           string
+	ClarificationTurnID string
+	Question            string
+	AnswerText          string
+	Rejected            bool
+	RejectReason        string
+}
+
+// PrimaryAnsweredNotifier arms the orchestrator's primary-answer watchdog at
+// the live delivery boundary. It must return only after the local watchdog has
+// been registered. It is deliberately separate from EventBus: NATS Publish is
+// fire-and-forget and cannot acknowledge local watchdog registration.
+type PrimaryAnsweredNotifier func(context.Context, PrimaryAnswered)
+
 // DetachedClarificationResume contains the durable context required to resume
 // a session after its original clarification waiter has gone away.
 type DetachedClarificationResume struct {

apps/backend/internal/clarification/handlers_authz_test.go

@@ -40,7 +40,7 @@ func setupAuthzHandler(t *testing.T, msgs map[string][]*taskmodels.Message, auth
 	repo := &stubMessageStore{messages: msgs}
 	eventBus := &stubEventBus{}
 	messageCreator := &stubMessageCreator{}
-	resolver := NewResolver(store, repo, messageCreator, authorizer, eventBus, eventBus, logger.Default())
+	resolver := NewResolver(store, repo, messageCreator, authorizer, eventBus, eventBus, nil, logger.Default())
 	h := NewHandlers(store, nil, messageCreator, repo, eventBus, resolver, logger.Default())
 	return h, store
 }
@@ -295,7 +295,7 @@ func TestHttpGetRequest_RepositoryFailure_500NotSilent404(t *testing.T) {
 	store := NewStore(time.Minute)
 	eventBus := &stubEventBus{}
 	messageCreator := &stubMessageCreator{}
-	resolver := NewResolver(store, repo, messageCreator, &stubAuthorizer{}, eventBus, eventBus, logger.Default())
+	resolver := NewResolver(store, repo, messageCreator, &stubAuthorizer{}, eventBus, eventBus, nil, logger.Default())
 	h := NewHandlers(store, nil, messageCreator, repo, eventBus, resolver, logger.Default())
 
 	rec := runGet(t, h, "pending-repo-failure")
@@ -313,7 +313,7 @@ func TestHttpWaitForResponse_RepositoryFailure_500NotSilent404(t *testing.T) {
 	store := NewStore(time.Minute)
 	eventBus := &stubEventBus{}
 	messageCreator := &stubMessageCreator{}
-	resolver := NewResolver(store, repo, messageCreator, &stubAuthorizer{}, eventBus, eventBus, logger.Default())
+	resolver := NewResolver(store, repo, messageCreator, &stubAuthorizer{}, eventBus, eventBus, nil, logger.Default())
 	h := NewHandlers(store, nil, messageCreator, repo, eventBus, resolver, logger.Default())
 
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
@@ -333,7 +333,7 @@ func TestHttpCancelRequest_RepositoryFailure_500NotSilent404(t *testing.T) {
 	store := NewStore(time.Minute)
 	eventBus := &stubEventBus{}
 	messageCreator := &stubMessageCreator{}
-	resolver := NewResolver(store, repo, messageCreator, &stubAuthorizer{}, eventBus, eventBus, logger.Default())
+	resolver := NewResolver(store, repo, messageCreator, &stubAuthorizer{}, eventBus, eventBus, nil, logger.Default())
 	h := NewHandlers(store, nil, messageCreator, repo, eventBus, resolver, logger.Default())
 
 	rec := runCancel(t, h, "pending-repo-failure")

apps/backend/internal/clarification/resolver.go

@@ -70,14 +70,15 @@ type clarificationResponseClaim struct {
 // on a win delivers through upstream's existing live-waiter/detached-resume
 // path (R7) unchanged; on a loss it reconstructs the winner's outcome (R2).
 type Resolver struct {
-	store           *Store
-	repo            handlerMessageStore
-	messages        MessageCreator
-	authorizer      taskAccessAuthorizer
-	detachedResumer DetachedClarificationResumer
-	eventBus        EventBus
-	logger          *logger.Logger
-	now             func() time.Time // seam for deterministic tests
+	store                   *Store
+	repo                    handlerMessageStore
+	messages                MessageCreator
+	authorizer              taskAccessAuthorizer
+	detachedResumer         DetachedClarificationResumer
+	eventBus                EventBus
+	primaryAnsweredNotifier PrimaryAnsweredNotifier
+	logger                  *logger.Logger
+	now                     func() time.Time // seam for deterministic tests
 }
 
 // NewResolver creates a Resolver.
@@ -88,17 +89,19 @@ func NewResolver(
 	authorizer taskAccessAuthorizer,
 	detachedResumer DetachedClarificationResumer,
 	eventBus EventBus,
+	primaryAnsweredNotifier PrimaryAnsweredNotifier,
 	log *logger.Logger,
 ) *Resolver {
 	return &Resolver{
-		store:           store,
-		repo:            repo,
-		messages:        messages,
-		authorizer:      authorizer,
-		detachedResumer: detachedResumer,
-		eventBus:        eventBus,
-		logger:          log.WithFields(zap.String("component", "clarification-resolver")),
-		now:             time.Now,
+		store:                   store,
+		repo:                    repo,
+		messages:                messages,
+		authorizer:              authorizer,
+		detachedResumer:         detachedResumer,
+		eventBus:                eventBus,
+		primaryAnsweredNotifier: primaryAnsweredNotifier,
+		logger:                  log.WithFields(zap.String("component", "clarification-resolver")),
+		now:                     time.Now,
 	}
 }
 
@@ -385,6 +388,14 @@ func (r *Resolver) deliverClaimedResolution(
 		func() error {
 			finalized, confirmErr := r.confirmLiveClarificationResponseDelivery(ctx, pendingID, claim)
 			if confirmErr == nil {
+				r.notifyPrimaryAnsweredBeforeWaiter(
+					ctx,
+					pendingID,
+					response.Answers,
+					response.Rejected,
+					response.RejectReason,
+					r.clarificationClaimTurnID(pendingID, finalized),
+				)
 				finalizedMessages <- finalized
 			}
 			return confirmErr
@@ -393,14 +404,6 @@ func (r *Resolver) deliverClaimedResolution(
 	if deliveryErr == nil {
 		finalized := <-finalizedMessages
 		r.publishClarificationBundleUpdates(ctx, pendingID, finalized)
-		r.publishPrimaryAnsweredEvent(
-			ctx,
-			pendingID,
-			response.Answers,
-			response.Rejected,
-			response.RejectReason,
-			r.clarificationClaimTurnID(pendingID, finalized),
-		)
 		r.logger.Info("clarification answered via primary path (same turn)",
 			zap.String("pending_id", pendingID),
 			zap.Int("answers", len(response.Answers)),
@@ -570,14 +573,18 @@ func clarificationClaimTurnIdentity(messages []*taskmodels.Message) (string, err
 	return turnID, nil
 }
 
-func (r *Resolver) publishPrimaryAnsweredEvent(
+// notifyPrimaryAnsweredBeforeWaiter is the synchronous local ordering
+// boundary. It must complete before the delivery-confirmation callback returns
+// to Store, because Store then releases the live waiter. Event-bus fan-out is
+// deliberately kept after this notifier and is never used as its acknowledgement.
+func (r *Resolver) notifyPrimaryAnsweredBeforeWaiter(
 	ctx context.Context,
 	pendingID string,
 	answers []Answer,
 	rejected bool,
 	rejectReason, clarificationTurnID string,
 ) {
-	if r.eventBus == nil {
+	if r.eventBus == nil && r.primaryAnsweredNotifier == nil {
 		return
 	}
 	persistenceCtx, cancel := clarificationPersistenceContext(ctx)
@@ -597,24 +604,50 @@ func (r *Resolver) publishPrimaryAnsweredEvent(
 	}
 
 	answerText := buildAnswerSummary(clarificationCtx.Questions, answers, rejected, rejectReason)
-	eventData := map[string]any{
-		metaSessionIDKey:        clarificationCtx.SessionID,
-		metaTaskIDKey:           clarificationCtx.TaskID,
-		metaPendingIDKey:        pendingID,
-		"clarification_turn_id": clarificationTurnID,
-		metaQuestionKey:         clarificationCtx.QuestionSummary,
-		"answer_text":           answerText,
-		metaRejectedKey:         rejected,
-		"reject_reason":         rejectReason,
+	answered := PrimaryAnswered{
+		SessionID:           clarificationCtx.SessionID,
+		TaskID:              clarificationCtx.TaskID,
+		PendingID:           pendingID,
+		ClarificationTurnID: clarificationTurnID,
+		Question:            clarificationCtx.QuestionSummary,
+		AnswerText:          answerText,
+		Rejected:            rejected,
+		RejectReason:        rejectReason,
 	}
-	if err := r.eventBus.Publish(persistenceCtx, events.ClarificationPrimaryAnswered, bus.NewEvent(
+	if r.primaryAnsweredNotifier != nil {
+		r.primaryAnsweredNotifier(persistenceCtx, answered)
+	}
+	r.publishPrimaryAnsweredEvent(persistenceCtx, answered)
+}
+
+// publishPrimaryAnsweredEvent is asynchronous fan-out only. The local
+// watchdog notifier is intentionally not called from this method: NATS
+// publication returns before its subscribers process the event.
+func (r *Resolver) publishPrimaryAnsweredEvent(ctx context.Context, answered PrimaryAnswered) {
+	if r.eventBus == nil {
+		return
+	}
+	eventData := map[string]any{
+		metaSessionIDKey:        answered.SessionID,
+		metaTaskIDKey:           answered.TaskID,
+		metaPendingIDKey:        answered.PendingID,
+		"clarification_turn_id": answered.ClarificationTurnID,
+		metaQuestionKey:         answered.Question,
+		"answer_text":           answered.AnswerText,
+		metaRejectedKey:         answered.Rejected,
+		"reject_reason":         answered.RejectReason,
+		// The orchestrator handled this event synchronously in the resolver's
+		// process. Its event-bus subscription must not arm a second watchdog.
+		"handled_inline": r.primaryAnsweredNotifier != nil,
+	}
+	if err := r.eventBus.Publish(ctx, events.ClarificationPrimaryAnswered, bus.NewEvent(
 		events.ClarificationPrimaryAnswered,
 		"clarification-resolver",
 		eventData,
 	)); err != nil {
 		r.logger.Warn("failed to publish primary clarification event",
-			zap.String("pending_id", pendingID),
-			zap.String("session_id", clarificationCtx.SessionID),
+			zap.String("pending_id", answered.PendingID),
+			zap.String("session_id", answered.SessionID),
 			zap.Error(err))
 	}
 }

apps/backend/internal/clarification/resolver_delivery_test.go

@@ -4,10 +4,13 @@ import (
 	"context"
 	"errors"
 	"maps"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/kandev/kandev/internal/common/logger"
+	"github.com/kandev/kandev/internal/events"
+	"github.com/kandev/kandev/internal/events/bus"
 	taskmodels "github.com/kandev/kandev/internal/task/models"
 )
 
@@ -186,11 +189,40 @@ func newResolverDeliveryFixture(
 		&stubAuthorizer{},
 		eventBus,
 		eventBus,
+		nil,
 		logger.Default(),
 	)
 	return resolver, store, repo, creator, eventBus
 }
 
+type resolverOrderingEventBus struct {
+	*stubEventBus
+	onPrimaryPublish func()
+}
+
+func (b *resolverOrderingEventBus) Publish(ctx context.Context, subject string, event *bus.Event) error {
+	if subject == events.ClarificationPrimaryAnswered && b.onPrimaryPublish != nil {
+		b.onPrimaryPublish()
+	}
+	return b.stubEventBus.Publish(ctx, subject, event)
+}
+
+type resolverAsyncFanoutEventBus struct {
+	*stubEventBus
+	primaryPublishReturned chan struct{}
+	releaseFanout          chan struct{}
+}
+
+func (b *resolverAsyncFanoutEventBus) Publish(ctx context.Context, subject string, event *bus.Event) error {
+	if subject == events.ClarificationPrimaryAnswered {
+		close(b.primaryPublishReturned)
+		go func() {
+			<-b.releaseFanout
+		}()
+	}
+	return b.stubEventBus.Publish(ctx, subject, event)
+}
+
 func resolverAnswer() Outcome {
 	return Outcome{Answers: []Answer{{
 		QuestionID:      "q1",
@@ -241,6 +273,192 @@ func TestResolverLiveDeliveryRequiresDurableConfirmation(t *testing.T) {
 	}
 }
 
+func TestResolverLiveDeliveryPublishesPrimaryAnswerBeforeWaiterReturns(t *testing.T) {
+	const pendingID = "pending-live-ordering"
+	message := resolverDeliveryMessage(pendingID, "message-live-ordering", "turn-1")
+	resolver, store, _, _, eventBus := newResolverDeliveryFixture(
+		t, pendingID, []*taskmodels.Message{message},
+	)
+	orderingBus := &resolverOrderingEventBus{stubEventBus: eventBus}
+	resolver.eventBus = orderingBus
+	store.CreateRequest(&Request{
+		PendingID: pendingID,
+		SessionID: "session-1",
+		TaskID:    "task-1",
+		Questions: []Question{{
+			ID: "q1", Prompt: "Continue?",
+			Options: []Option{{ID: "yes", Label: "Yes"}, {ID: "no", Label: "No"}},
+		}},
+	})
+
+	waitEntered := make(chan struct{}, 1)
+	store.SetOnWaitEntered(func(string) { waitEntered <- struct{}{} })
+	waitReturned := make(chan struct{})
+	waitDone := make(chan error, 1)
+	go func() {
+		_, err := store.WaitForResponse(context.Background(), pendingID)
+		close(waitReturned)
+		waitDone <- err
+	}()
+	<-waitEntered
+	store.SetOnWaitEntered(nil)
+
+	eventBeforeWaiterReturn := true
+	eventPublished := make(chan struct{})
+	orderingBus.onPrimaryPublish = func() {
+		select {
+		case <-waitReturned:
+			eventBeforeWaiterReturn = false
+		default:
+		}
+		close(eventPublished)
+	}
+
+	_, claimed, err := resolver.ResolveBundle(context.Background(), pendingID, resolverAnswer())
+	if err != nil || !claimed {
+		t.Fatalf("ResolveBundle = claimed %v, err %v; want live delivery", claimed, err)
+	}
+	<-eventPublished
+	if err := <-waitDone; err != nil {
+		t.Fatalf("WaitForResponse: %v", err)
+	}
+	if !eventBeforeWaiterReturn {
+		t.Fatal("primary-answer event was published after the live waiter returned")
+	}
+	if len(eventBus.events) != 1 || eventBus.events[0].Type != events.ClarificationPrimaryAnswered {
+		t.Fatalf("primary-answer events = %d, want exactly 1", len(eventBus.events))
+	}
+}
+
+func TestResolverLiveDeliveryNotifiesWatchdogBeforeWaiterReturns(t *testing.T) {
+	const pendingID = "pending-live-watchdog-notifier"
+	message := resolverDeliveryMessage(pendingID, "message-live-watchdog-notifier", "turn-1")
+	resolver, store, _, _, _ := newResolverDeliveryFixture(t, pendingID, []*taskmodels.Message{message})
+
+	notified := make(chan struct{})
+	resolver.primaryAnsweredNotifier = func(context.Context, PrimaryAnswered) {
+		close(notified)
+	}
+
+	store.CreateRequest(&Request{
+		PendingID: pendingID,
+		SessionID: "session-1",
+		TaskID:    "task-1",
+		Questions: []Question{{
+			ID: "q1", Prompt: "Continue?",
+			Options: []Option{{ID: "yes", Label: "Yes"}, {ID: "no", Label: "No"}},
+		}},
+	})
+	waitEntered := make(chan struct{}, 1)
+	store.SetOnWaitEntered(func(string) { waitEntered <- struct{}{} })
+	waitDone := make(chan error, 1)
+	go func() {
+		_, err := store.WaitForResponse(context.Background(), pendingID)
+		waitDone <- err
+	}()
+	<-waitEntered
+	store.SetOnWaitEntered(nil)
+
+	_, claimed, err := resolver.ResolveBundle(context.Background(), pendingID, resolverAnswer())
+	if err != nil || !claimed {
+		t.Fatalf("ResolveBundle = claimed %v, err %v; want live delivery", claimed, err)
+	}
+	var waitErr error
+	select {
+	case waitErr = <-waitDone:
+	case <-time.After(time.Second):
+		t.Fatal("WaitForResponse did not return")
+	}
+	select {
+	case <-notified:
+	default:
+		t.Fatal("live waiter returned before the primary-answer watchdog was notified")
+	}
+	if waitErr != nil {
+		t.Fatalf("WaitForResponse: %v", waitErr)
+	}
+}
+
+// This is reviewer-requested contract coverage for transports whose Publish
+// method returns before subscribers run, such as NATS. The local notifier is
+// the ordering acknowledgement; fan-out delivery is not.
+func TestResolverLiveDeliveryUsesSynchronousNotifierBeforeAsyncFanout(t *testing.T) {
+	const pendingID = "pending-live-watchdog-async-fanout"
+	message := resolverDeliveryMessage(pendingID, "message-live-watchdog-async-fanout", "turn-1")
+	resolver, store, _, _, eventBus := newResolverDeliveryFixture(t, pendingID, []*taskmodels.Message{message})
+	asyncBus := &resolverAsyncFanoutEventBus{
+		stubEventBus:           eventBus,
+		primaryPublishReturned: make(chan struct{}),
+		releaseFanout:          make(chan struct{}),
+	}
+	resolver.eventBus = asyncBus
+	t.Cleanup(func() { close(asyncBus.releaseFanout) })
+
+	notifierEntered := make(chan struct{})
+	releaseNotifier := make(chan struct{})
+	var notifierReturned atomic.Bool
+	resolver.primaryAnsweredNotifier = func(context.Context, PrimaryAnswered) {
+		close(notifierEntered)
+		<-releaseNotifier
+		notifierReturned.Store(true)
+	}
+
+	store.CreateRequest(&Request{
+		PendingID: pendingID,
+		SessionID: "session-1",
+		TaskID:    "task-1",
+		Questions: []Question{{
+			ID: "q1", Prompt: "Continue?",
+			Options: []Option{{ID: "yes", Label: "Yes"}, {ID: "no", Label: "No"}},
+		}},
+	})
+	waitEntered := make(chan struct{}, 1)
+	store.SetOnWaitEntered(func(string) { waitEntered <- struct{}{} })
+	waitDone := make(chan error, 1)
+	go func() {
+		_, err := store.WaitForResponse(context.Background(), pendingID)
+		if !notifierReturned.Load() {
+			t.Errorf("live waiter returned before notifier acknowledgement")
+		}
+		waitDone <- err
+	}()
+	<-waitEntered
+	store.SetOnWaitEntered(nil)
+
+	resolveDone := make(chan error, 1)
+	go func() {
+		_, _, err := resolver.ResolveBundle(context.Background(), pendingID, resolverAnswer())
+		if !notifierReturned.Load() {
+			t.Errorf("resolver returned before notifier acknowledgement")
+		}
+		resolveDone <- err
+	}()
+
+	select {
+	case <-notifierEntered:
+	case <-time.After(time.Second):
+		t.Fatal("synchronous primary-answer notifier was not reached")
+	}
+	select {
+	case <-asyncBus.primaryPublishReturned:
+		t.Fatal("event-bus fan-out started before the synchronous notifier returned")
+	default:
+	}
+
+	close(releaseNotifier)
+	if err := <-resolveDone; err != nil {
+		t.Fatalf("ResolveBundle: %v", err)
+	}
+	if err := <-waitDone; err != nil {
+		t.Fatalf("WaitForResponse: %v", err)
+	}
+	select {
+	case <-asyncBus.primaryPublishReturned:
+	default:
+		t.Fatal("primary-answer fan-out was not published after notifier acknowledgement")
+	}
+}
+
 func TestResolverDetachedResumeFailureRestoresRetryableBundle(t *testing.T) {
 	const pendingID = "pending-detached-retry"
 	message := resolverDeliveryMessage(pendingID, "message-detached-retry", "turn-1")

apps/backend/internal/mcp/handlers/question_handlers_test.go

@@ -339,7 +339,7 @@ func (stubDetachedResumer) ResumeDetachedClarification(context.Context, clarific
 func newTestResolver(t *testing.T, store *clarification.Store, repo *sqliterepo.Repository, svc *service.Service) *clarification.Resolver {
 	t.Helper()
 	resumer := stubDetachedResumer{}
-	return clarification.NewResolver(store, repo, &svcMessageUpdater{Service: svc}, svc, resumer, resumer, testLogger(t))
+	return clarification.NewResolver(store, repo, &svcMessageUpdater{Service: svc}, svc, resumer, resumer, nil, testLogger(t))
 }
 
 // seedBundle creates a task/session/turn and a two-question clarification

apps/backend/internal/orchestrator/clarification_turn_authority.go

@@ -51,3 +51,81 @@ func (s *Service) clarificationTurnStillCurrent(ctx context.Context, data clarif
 	}
 	return true
 }
+
+// clarificationTurnStillCurrentAfterRecovery repeats the turn-authority check
+// after silent cancellation has settled. Owned silent cancellation normally
+// completes the captured clarification turn, so an absent active turn is valid
+// only when that exact turn is durably completed. A different active turn is a
+// successor and must prevent the stale watchdog from queueing over it.
+func (s *Service) clarificationTurnStillCurrentAfterRecovery(
+	ctx context.Context,
+	data clarificationAnsweredData,
+) bool {
+	if s.turnService == nil {
+		s.logger.Warn("skipping clarification fallback after recovery: turn service unavailable",
+			zap.String("session_id", data.SessionID),
+			zap.String("pending_id", data.PendingID))
+		return false
+	}
+	if data.ClarificationTurnID == "" {
+		s.logger.Debug("skipping clarification fallback after recovery: event carries no turn ID",
+			zap.String("session_id", data.SessionID),
+			zap.String("pending_id", data.PendingID))
+		return false
+	}
+
+	activeTurn, err := s.turnService.GetActiveTurn(ctx, data.SessionID)
+	if err != nil {
+		if ctx.Err() != nil || errors.Is(err, context.Canceled) {
+			s.logger.Debug("clarification fallback post-recovery authority check canceled",
+				zap.String("session_id", data.SessionID),
+				zap.String("pending_id", data.PendingID),
+				zap.String("clarification_turn_id", data.ClarificationTurnID),
+				zap.Error(err))
+			return false
+		}
+		s.logger.Warn("failed to verify clarification fallback turn authority after recovery",
+			zap.String("session_id", data.SessionID),
+			zap.String("pending_id", data.PendingID),
+			zap.String("clarification_turn_id", data.ClarificationTurnID),
+			zap.Error(err))
+		return false
+	}
+	if activeTurn != nil {
+		if activeTurn.ID == data.ClarificationTurnID {
+			return true
+		}
+		s.logger.Info("skipping superseded clarification fallback after recovery",
+			zap.String("session_id", data.SessionID),
+			zap.String("pending_id", data.PendingID),
+			zap.String("clarification_turn_id", data.ClarificationTurnID),
+			zap.String("current_turn_id", activeTurn.ID))
+		return false
+	}
+
+	completedTurn, err := s.turnService.GetTurn(ctx, data.ClarificationTurnID)
+	if err != nil {
+		if ctx.Err() != nil || errors.Is(err, context.Canceled) {
+			s.logger.Debug("clarification fallback completed-turn lookup canceled",
+				zap.String("session_id", data.SessionID),
+				zap.String("pending_id", data.PendingID),
+				zap.String("clarification_turn_id", data.ClarificationTurnID),
+				zap.Error(err))
+			return false
+		}
+		s.logger.Warn("failed to verify completed clarification turn after recovery",
+			zap.String("session_id", data.SessionID),
+			zap.String("pending_id", data.PendingID),
+			zap.String("clarification_turn_id", data.ClarificationTurnID),
+			zap.Error(err))
+		return false
+	}
+	if completedTurn == nil || completedTurn.CompletedAt == nil {
+		s.logger.Info("skipping clarification fallback after recovery: captured turn is not settled",
+			zap.String("session_id", data.SessionID),
+			zap.String("pending_id", data.PendingID),
+			zap.String("clarification_turn_id", data.ClarificationTurnID))
+		return false
+	}
+	return true
+}

apps/backend/internal/orchestrator/clarification_turn_authority_test.go

@@ -72,6 +72,47 @@ func TestClarificationTurnAuthorityLogsCanceledLookupAtDebug(t *testing.T) {
 	}
 }
 
+func TestClarificationTurnAuthorityAfterRecoveryAllowsOwnedCompletionButRejectsSuccessor(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, "task-recovery-authority", "session-recovery-authority", models.TaskSessionStateRunning)
+	turnService := &repoTurnService{repo: repo}
+	base := time.Now().UTC()
+	completedAt := base.Add(time.Second)
+	if err := repo.CreateTurn(ctx, &models.Turn{
+		ID:            "turn-recovery-authority",
+		TaskID:        "task-recovery-authority",
+		TaskSessionID: "session-recovery-authority",
+		StartedAt:     base,
+		CompletedAt:   &completedAt,
+	}); err != nil {
+		t.Fatalf("create completed clarification turn: %v", err)
+	}
+
+	svc := &Service{logger: testLogger(), turnService: turnService}
+	data := clarificationAnsweredData{
+		SessionID:           "session-recovery-authority",
+		PendingID:           "pending-recovery-authority",
+		ClarificationTurnID: "turn-recovery-authority",
+	}
+	if !svc.clarificationTurnStillCurrentAfterRecovery(ctx, data) {
+		t.Fatal("owned cancellation completion was rejected before queueing the replacement")
+	}
+
+	successor := &models.Turn{
+		ID:            "turn-recovery-successor",
+		TaskID:        "task-recovery-authority",
+		TaskSessionID: "session-recovery-authority",
+		StartedAt:     base.Add(time.Minute),
+	}
+	if err := repo.CreateTurn(ctx, successor); err != nil {
+		t.Fatalf("create successor turn: %v", err)
+	}
+	if svc.clarificationTurnStillCurrentAfterRecovery(ctx, data) {
+		t.Fatal("clarification recovery accepted a successor active turn")
+	}
+}
+
 func TestClarificationWatchdogDoesNotDispatchAfterTurnIsSuperseded(t *testing.T) {
 	svc, agentMgr := setupSupersededClarificationTurn(t)
 	svc.clarificationWatchdogTimeout = time.Millisecond

apps/backend/internal/orchestrator/event_handlers_clarification.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"go.uber.org/zap"
@@ -119,10 +120,30 @@ type clarificationAnsweredData struct {
 	AnswerText          string `json:"answer_text"`
 	Rejected            bool   `json:"rejected"`
 	RejectReason        string `json:"reject_reason"`
+	HandledInline       bool   `json:"handled_inline"`
 }
 
 type clarificationWatchdogEntry struct {
 	cancel func()
+	// Recovery's silent cancellation can synchronously emit stream activity.
+	// Keep that activity from cancelling this entry's own recovery context.
+	recoveryCancellationActive atomic.Bool
+}
+
+func (e *clarificationWatchdogEntry) beginRecoveryCancellation() {
+	if e != nil {
+		e.recoveryCancellationActive.Store(true)
+	}
+}
+
+func (e *clarificationWatchdogEntry) endRecoveryCancellation() {
+	if e != nil {
+		e.recoveryCancellationActive.Store(false)
+	}
+}
+
+func (e *clarificationWatchdogEntry) isRecoveryCancellationActive() bool {
+	return e != nil && e.recoveryCancellationActive.Load()
 }
 
 // handleClarificationAnswered handles user responses to agent clarification questions.
@@ -252,6 +273,16 @@ func (s *Service) handleClarificationPrimaryAnswered(ctx context.Context, event
 			zap.String("pending_id", data.PendingID))
 		return nil
 	}
+	if data.HandledInline {
+		return nil
+	}
+	return s.handleClarificationPrimaryAnsweredData(ctx, data)
+}
+
+func (s *Service) handleClarificationPrimaryAnsweredData(ctx context.Context, data clarificationAnsweredData) error {
+	if data.SessionID == "" || data.TaskID == "" || data.PendingID == "" {
+		return nil
+	}
 
 	// A directly answered MCP request does not transition the session through
 	// WAITING_FOR_INPUT, so no ordinary turn-start event exists to move a task
@@ -261,10 +292,41 @@ func (s *Service) handleClarificationPrimaryAnswered(ctx context.Context, event
 	return nil
 }
 
+// HandleClarificationPrimaryAnswered is the synchronous local notification
+// used by clarification.Resolver to arm the watchdog before releasing a live
+// clarification waiter. The bus event remains available for projections.
+func (s *Service) HandleClarificationPrimaryAnswered(ctx context.Context, answered clarification.PrimaryAnswered) {
+	_ = s.handleClarificationPrimaryAnsweredData(ctx, clarificationAnsweredData{
+		SessionID:           answered.SessionID,
+		TaskID:              answered.TaskID,
+		PendingID:           answered.PendingID,
+		ClarificationTurnID: answered.ClarificationTurnID,
+		Question:            answered.Question,
+		AnswerText:          answered.AnswerText,
+		Rejected:            answered.Rejected,
+		RejectReason:        answered.RejectReason,
+	})
+}
+
 func (s *Service) clarificationWatchdogKey(sessionID, pendingID string) string {
 	return sessionID + "::" + pendingID
 }
 
+func (s *Service) loadClarificationWatchdogEntry(sessionID, pendingID string) *clarificationWatchdogEntry {
+	if sessionID == "" || pendingID == "" {
+		return nil
+	}
+	value, ok := s.clarificationWatchdogs.Load(s.clarificationWatchdogKey(sessionID, pendingID))
+	if !ok {
+		return nil
+	}
+	entry, ok := value.(*clarificationWatchdogEntry)
+	if !ok {
+		return nil
+	}
+	return entry
+}
+
 func (s *Service) getClarificationWatchdogTimeout() time.Duration {
 	if s.clarificationWatchdogTimeout > 0 {
 		return s.clarificationWatchdogTimeout
@@ -410,10 +472,26 @@ func (s *Service) retryClarificationAfterCancel(ctx context.Context, data clarif
 		return false
 	}
 
-	if err := s.cancelAgentSilentWithGuard(ctx, data.TaskID, data.SessionID, guard.unlock, guard.relock); err != nil {
+	watchdogEntry := s.loadClarificationWatchdogEntry(data.SessionID, data.PendingID)
+	if watchdogEntry != nil {
+		watchdogEntry.beginRecoveryCancellation()
+	}
+	expectedTurnID := data.ClarificationTurnID
+	cancelErr := s.cancelAgentSilentExpectedWithGuard(
+		ctx,
+		data.TaskID,
+		data.SessionID,
+		&expectedTurnID,
+		guard.unlock,
+		guard.relock,
+	)
+	if watchdogEntry != nil {
+		watchdogEntry.endRecoveryCancellation()
+	}
+	if cancelErr != nil {
 		s.logger.Warn("cancel failed (agent likely dead), force-transitioning session state",
 			zap.String("session_id", data.SessionID),
-			zap.Error(err))
+			zap.Error(cancelErr))
 		// Revert through the terminal-safe state writer. Production uses a
 		// compare-and-set, and the shared guard keeps coordinator cancellation
 		// outside this narrow mutation.
@@ -450,6 +528,9 @@ func (s *Service) retryClarificationAfterCancel(ctx context.Context, data clarif
 			zap.String("session_state", string(session.State)))
 		return false
 	}
+	if !s.clarificationTurnStillCurrentAfterRecovery(ctx, data) {
+		return false
+	}
 
 	if err := s.dispatchClarificationResumeLocked(ctx, data, prompt); err != nil {
 		if errors.Is(err, errClarificationResumeQueuedForDrain) {
@@ -912,7 +993,10 @@ func (s *Service) logSilentCancelReconciled(taskID, sessionID string, err error)
 		zap.Error(err))
 }
 
-func (s *Service) cancelClarificationWatchdogsForSession(sessionID, reason string) {
+func (s *Service) cancelClarificationWatchdogsForSession(
+	sessionID, reason string,
+	payload *lifecycle.AgentStreamEventPayload,
+) {
 	if sessionID == "" {
 		return
 	}
@@ -924,6 +1008,15 @@ func (s *Service) cancelClarificationWatchdogsForSession(sessionID, reason strin
 		if !ok || !strings.HasPrefix(keyStr, prefix) {
 			return true
 		}
+		// Silent cancellation may synchronously emit a frame for the captured
+		// execution/prompt. Ignore only that operation-owned identity. A newer
+		// execution or prompt generation remains authoritative and cancels the
+		// watchdog even while the fallback cancellation is blocked.
+		if entry, ok := value.(*clarificationWatchdogEntry); ok &&
+			entry.isRecoveryCancellationActive() &&
+			s.clarificationRecoveryOwnsStreamEvent(sessionID, payload) {
+			return true
+		}
 		s.clarificationWatchdogs.Delete(keyStr)
 		if entry, ok := value.(*clarificationWatchdogEntry); ok && entry.cancel != nil {
 			entry.cancel()
@@ -940,6 +1033,53 @@ func (s *Service) cancelClarificationWatchdogsForSession(sessionID, reason strin
 	}
 }
 
+func (s *Service) clarificationRecoveryOwnsStreamEvent(
+	sessionID string,
+	payload *lifecycle.AgentStreamEventPayload,
+) bool {
+	if payload == nil || payload.Data == nil {
+		return false
+	}
+	operation := s.currentCancellation(sessionID)
+	if operation == nil || operation.kind != cancellationKindSilent {
+		return false
+	}
+	if !clarificationRecoveryCancellationFrame(payload) {
+		// Message, thinking, and tool frames are always live activity. Even an
+		// exact execution/generation match cannot prove that those frames came
+		// from the cancellation rather than the agent's normal work.
+		return false
+	}
+	identity, ready := s.cancellationIdentitySnapshot(operation)
+	if !ready || identity.executionID == "" || identity.promptGeneration == 0 {
+		// Recovery may ignore a frame only when both immutable parts of the
+		// cancellation identity are present on the frame. Missing identity is
+		// independent activity by definition, not evidence of cancellation.
+		return false
+	}
+	eventExecutionID := payload.ExecutionID
+	if eventExecutionID == "" {
+		eventExecutionID = payload.AgentID
+	}
+	return eventExecutionID != "" &&
+		eventExecutionID == identity.executionID &&
+		payload.Data.PromptGeneration == identity.promptGeneration
+}
+
+func clarificationRecoveryCancellationFrame(payload *lifecycle.AgentStreamEventPayload) bool {
+	if payload == nil || payload.Data == nil {
+		return false
+	}
+	switch payload.Data.Type {
+	case "session_info", "session_status":
+		return true
+	case agentEventComplete:
+		return extractStopReason(payload) == stopReasonCancelled
+	default:
+		return false
+	}
+}
+
 func (s *Service) cancelAllClarificationWatchdogs() {
 	s.clarificationWatchdogs.Range(func(key, value interface{}) bool {
 		keyStr, ok := key.(string)

apps/backend/internal/orchestrator/event_handlers_clarification_test.go

@@ -1471,6 +1471,279 @@ func TestClarificationWatchdogCancellationInterruptsFallbackLookup(t *testing.T)
 	}
 }
 
+func TestClarificationWatchdogRecoveryIgnoresOwnCancelActivity(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-watchdog-own-cancel"
+		sessionID = "session-watchdog-own-cancel"
+		execution = "exec-watchdog-own-cancel"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning)
+	seedExecutorRunning(t, repo, sessionID, taskID, execution)
+
+	promptDone := make(chan struct{})
+	var svc *Service
+	agentMgr := &mockAgentManager{
+		isAgentRunning: true,
+		promptDone:     promptDone,
+	}
+	agentMgr.cancelAgentFunc = func(cancelCtx context.Context, _ string) error {
+		svc.handleAgentStreamEvent(cancelCtx, &lifecycle.AgentStreamEventPayload{
+			TaskID:      taskID,
+			SessionID:   sessionID,
+			ExecutionID: execution,
+			Data: &lifecycle.AgentStreamEventData{
+				Type:             "session_info",
+				PromptGeneration: 1,
+			},
+		})
+		return nil
+	}
+	agentMgr.currentPromptExecutionID = execution
+	agentMgr.currentPromptGeneration.Store(1)
+	svc = createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
+	svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+	svc.turnService = &repoTurnService{repo: repo}
+	clarificationTurn, err := svc.turnService.StartTurn(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("start clarification turn: %v", err)
+	}
+
+	watchCtx, cancelWatchdog := context.WithCancel(context.Background())
+	defer cancelWatchdog()
+	pendingID := "pending-watchdog-own-cancel"
+	entry := &clarificationWatchdogEntry{cancel: cancelWatchdog}
+	key := svc.clarificationWatchdogKey(sessionID, pendingID)
+	svc.clarificationWatchdogs.Store(key, entry)
+	t.Cleanup(func() { svc.cancelAllClarificationWatchdogs() })
+
+	recovered := svc.retryClarificationAfterCancel(
+		watchCtx,
+		clarificationAnsweredData{
+			TaskID: taskID, SessionID: sessionID, PendingID: pendingID,
+			ClarificationTurnID: clarificationTurn.ID,
+		},
+		"the clarification answer",
+		fmt.Errorf("wrapped: %w", ErrAgentPromptInProgress),
+	)
+	if !recovered {
+		t.Fatal("clarification watchdog recovery did not complete after its own cancellation activity")
+	}
+	select {
+	case <-promptDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("clarification answer was not handed off after recovery")
+	}
+	if err := watchCtx.Err(); err != nil {
+		t.Fatalf("watchdog recovery context was cancelled by its own stream activity: %v", err)
+	}
+	if got := len(agentMgr.capturedPrompts); got != 1 {
+		t.Fatalf("clarification replacement prompts = %d, want exactly 1", got)
+	}
+}
+
+func TestClarificationWatchdogRecoveryCancelsOnIndependentActivity(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-watchdog-independent-activity"
+		sessionID = "session-watchdog-independent-activity"
+		execution = "exec-watchdog-independent-activity"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning)
+	seedExecutorRunning(t, repo, sessionID, taskID, execution)
+
+	cancelEntered := make(chan struct{}, 1)
+	releaseCancel := make(chan struct{})
+	var releaseCancelOnce sync.Once
+	release := func() {
+		releaseCancelOnce.Do(func() { close(releaseCancel) })
+	}
+	promptDone := make(chan struct{})
+	agentMgr := &mockAgentManager{
+		isAgentRunning:           true,
+		promptDone:               promptDone,
+		cancelAgentBlock:         releaseCancel,
+		cancelAgentEntered:       cancelEntered,
+		currentPromptExecutionID: execution,
+	}
+	agentMgr.currentPromptGeneration.Store(1)
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
+	svc.messageCreator = &mockMessageCreator{}
+	svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+	svc.turnService = &repoTurnService{repo: repo}
+	clarificationTurn, err := svc.turnService.StartTurn(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("start clarification turn: %v", err)
+	}
+
+	watchCtx, cancelWatchdog := context.WithCancel(context.Background())
+	defer cancelWatchdog()
+	pendingID := "pending-watchdog-independent-activity"
+	entry := &clarificationWatchdogEntry{cancel: cancelWatchdog}
+	key := svc.clarificationWatchdogKey(sessionID, pendingID)
+	svc.clarificationWatchdogs.Store(key, entry)
+	t.Cleanup(func() {
+		release()
+		svc.cancelAllClarificationWatchdogs()
+	})
+
+	recoveryDone := make(chan bool, 1)
+	go func() {
+		recoveryDone <- svc.retryClarificationAfterCancel(
+			watchCtx,
+			clarificationAnsweredData{
+				TaskID: taskID, SessionID: sessionID, PendingID: pendingID,
+				ClarificationTurnID: clarificationTurn.ID,
+			},
+			"the clarification answer",
+			fmt.Errorf("wrapped: %w", ErrAgentPromptInProgress),
+		)
+	}()
+
+	select {
+	case <-cancelEntered:
+	case <-time.After(time.Second):
+		t.Fatal("silent cancellation did not reach the agent")
+	}
+
+	// This frame is from a newer prompt generation on the same execution. It
+	// must remain authoritative even while the fallback cancellation is blocked.
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      taskID,
+		SessionID:   sessionID,
+		ExecutionID: execution,
+		Data: &lifecycle.AgentStreamEventData{
+			Type:             "session_info",
+			PromptGeneration: 2,
+		},
+	})
+	if watchCtx.Err() == nil {
+		t.Fatal("independent stream activity did not cancel the watchdog")
+	}
+
+	release()
+	select {
+	case <-recoveryDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("silent cancellation did not settle")
+	}
+}
+
+func TestClarificationWatchdogRecoveryCancelsOnSameExecutionMessageActivity(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-watchdog-same-execution-message"
+		sessionID = "session-watchdog-same-execution-message"
+		execution = "exec-watchdog-same-execution-message"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning)
+	seedExecutorRunning(t, repo, sessionID, taskID, execution)
+
+	cancelEntered := make(chan struct{}, 1)
+	releaseCancel := make(chan struct{})
+	var releaseCancelOnce sync.Once
+	release := func() {
+		releaseCancelOnce.Do(func() { close(releaseCancel) })
+	}
+	agentMgr := &mockAgentManager{
+		isAgentRunning:           true,
+		cancelAgentBlock:         releaseCancel,
+		cancelAgentEntered:       cancelEntered,
+		currentPromptExecutionID: execution,
+	}
+	agentMgr.currentPromptGeneration.Store(1)
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
+	svc.messageCreator = &mockMessageCreator{}
+	svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+	svc.turnService = &repoTurnService{repo: repo}
+	clarificationTurn, err := svc.turnService.StartTurn(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("start clarification turn: %v", err)
+	}
+
+	watchCtx, cancelWatchdog := context.WithCancel(context.Background())
+	defer cancelWatchdog()
+	pendingID := "pending-watchdog-same-execution-message"
+	entry := &clarificationWatchdogEntry{cancel: cancelWatchdog}
+	key := svc.clarificationWatchdogKey(sessionID, pendingID)
+	svc.clarificationWatchdogs.Store(key, entry)
+	t.Cleanup(func() {
+		release()
+		svc.cancelAllClarificationWatchdogs()
+	})
+
+	recoveryDone := make(chan bool, 1)
+	go func() {
+		recoveryDone <- svc.retryClarificationAfterCancel(
+			watchCtx,
+			clarificationAnsweredData{
+				TaskID: taskID, SessionID: sessionID, PendingID: pendingID,
+				ClarificationTurnID: clarificationTurn.ID,
+			},
+			"the clarification answer",
+			fmt.Errorf("wrapped: %w", ErrAgentPromptInProgress),
+		)
+	}()
+
+	select {
+	case <-cancelEntered:
+	case <-time.After(time.Second):
+		t.Fatal("silent cancellation did not reach the agent")
+	}
+
+	// A same-execution message frame is still independent activity. Matching the
+	// cancellation's execution and prompt generation does not prove that a
+	// message frame came from the cancellation, so it must cancel recovery.
+	svc.handleAgentStreamEvent(ctx, &lifecycle.AgentStreamEventPayload{
+		TaskID:      taskID,
+		SessionID:   sessionID,
+		ExecutionID: execution,
+		Data: &lifecycle.AgentStreamEventData{
+			Type:             "message_streaming",
+			PromptGeneration: 1,
+		},
+	})
+	if watchCtx.Err() == nil {
+		t.Fatal("same-execution message activity did not cancel the watchdog")
+	}
+
+	release()
+	select {
+	case <-recoveryDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("silent cancellation did not settle")
+	}
+}
+
+func TestClarificationRecoveryCancellationFrameRejectsNormalAgentActivity(t *testing.T) {
+	for _, eventType := range []string{"message_streaming", "thinking_streaming", agentEventToolCall, agentEventToolUpdate} {
+		t.Run(eventType, func(t *testing.T) {
+			if clarificationRecoveryCancellationFrame(&lifecycle.AgentStreamEventPayload{
+				Data: &lifecycle.AgentStreamEventData{Type: eventType},
+			}) {
+				t.Fatalf("%q was classified as recovery cancellation activity", eventType)
+			}
+		})
+	}
+
+	if !clarificationRecoveryCancellationFrame(&lifecycle.AgentStreamEventPayload{
+		Data: &lifecycle.AgentStreamEventData{Type: "session_info"},
+	}) {
+		t.Fatal("session_info was not classified as recovery cancellation activity")
+	}
+	if !clarificationRecoveryCancellationFrame(&lifecycle.AgentStreamEventPayload{
+		Data: &lifecycle.AgentStreamEventData{
+			Type: agentEventComplete,
+			Data: map[string]interface{}{"stop_reason": "cancelled"},
+		},
+	}) {
+		t.Fatal("cancelled complete was not classified as recovery cancellation activity")
+	}
+}
+
 // TestRetryClarificationAfterCancel_DoesNotStarveUserCancel is the regression
 // test for the production hang where a clarification-timeout recovery left a
 // session permanently unstoppable. retryClarificationAfterCancel used to send

apps/backend/internal/orchestrator/event_handlers_streaming.go

@@ -54,11 +54,13 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle
 	if eventExecutionID == "" {
 		eventExecutionID = payload.AgentID
 	}
+	eventType := payload.Data.Type
 	if !s.cancellationOwnsStreamEvent(
 		payload.SessionID,
 		eventExecutionID,
 		payload.Data.PromptGeneration,
 	) {
+		s.cancelClarificationWatchdogsForSession(payload.SessionID, eventType, payload)
 		s.logger.Debug("ignoring stream event for execution outside cancellation identity",
 			zap.String("task_id", payload.TaskID),
 			zap.String("session_id", payload.SessionID),
@@ -68,7 +70,6 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle
 	}
 	taskID := payload.TaskID
 	sessionID := payload.SessionID
-	eventType := payload.Data.Type
 	terminalCompleteStream := false
 
 	if eventType == agentEventComplete {
@@ -111,7 +112,7 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle
 		// Any live agent stream activity means the agent resumed after clarification.
 		// Cancel primary-path clarification watchdogs for this session. Late terminal
 		// completes are excluded because they belong to an already-finished execution.
-		s.cancelClarificationWatchdogsForSession(sessionID, eventType)
+		s.cancelClarificationWatchdogsForSession(sessionID, eventType, payload)
 	}
 
 	s.logger.Debug("handling agent stream event",

docs/plans/clarification-watchdog-race-recovery/plan.md

@@ -0,0 +1,89 @@
+---
+created: 2026-08-24
+status: completed
+requirements:
+  - REQ-TASKS-CLARIFICATION-LIFECYCLE-001
+  - REQ-TASKS-CLARIFICATION-SCENARIOS-001
+system_design:
+  - ../../specs/tasks/system-design/clarification-active-lifecycle.md
+legacy_specs: []
+---
+
+# Implementation Plan: Clarification Watchdog Race Recovery
+
+## Overview
+
+Make live clarification answer delivery and watchdog recovery one ordered operation. The fix first
+arms recovery before ACP can acknowledge the answer, then prevents recovery-owned cancellation frames
+from cancelling the fallback context that must hand off that answer.
+
+## Scope
+
+### In scope
+
+- Arm the primary-answer watchdog within the durable live-delivery confirmation boundary.
+- Preserve the fallback context across activity caused by its own silent cancellation.
+- Keep independent session activity and service shutdown able to interrupt fallback work.
+- Add deterministic regressions for both observed races.
+
+### Out of scope
+
+- Changing the 15-second watchdog timeout.
+- Changing detached clarification delivery or persistence schemas.
+- Changing ACP, MCP, or clarification response envelopes.
+- Resuming or modifying the affected historical session.
+
+## Technical approach
+
+### Live delivery ordering
+
+Update `clarification.Resolver` so its construction-supplied synchronous local orchestrator notifier
+arms the watchdog from the successful delivery-confirmation callback after durable finalization and
+before `Store.WaitForResponse` can return the response to the agent. Keep event-bus publication as
+fan-out, mark it as handled by the local notifier to avoid a duplicate watchdog, and keep terminal
+bundle-message publication after confirmed delivery. NATS publication is never the watchdog
+acknowledgement.
+
+### Recovery-owned cancellation activity
+
+Extend `clarificationWatchdogEntry` with concurrency-safe recovery-cancellation phase state. Mark that
+phase only around expected-turn silent cancellation. `cancelClarificationWatchdogsForSession` must
+preserve the matching entry only when a live stream frame has an allowed cancellation-acknowledgement
+type and matches the cancellation operation's captured execution and prompt-generation identity.
+Message, thinking, and tool activity, newer or independent activity, and
+`cancelAllClarificationWatchdogs`, retain their cancellation authority. After cancellation, recheck
+the captured turn and allow queue handoff only when it remains current or is the exact turn durably
+completed by the owned cancellation.
+
+### Regression coverage
+
+Add a resolver delivery test proving that the synchronous local notifier runs before the live waiter
+receives the response. Add an orchestrator test whose silent cancellation synchronously emits the same
+`session_info` activity observed in production and prove the answer still reaches exactly one queued or
+dispatched replacement. Add a concurrent newer-prompt activity regression and a post-cancellation
+turn-authority regression.
+
+## Tests
+
+| Acceptance criterion | Evidence |
+| --- | --- |
+| `AC-TASKS-CLARIFICATION-LIFECYCLE-001.3` | Resolver ordering and orchestrator recovery regressions |
+| `AC-TASKS-CLARIFICATION-SCENARIOS-001.1` | Immediate acknowledgement and recovery-owned cancellation scenarios |
+
+## Work orders
+
+- [x] [Task 01: Make clarification watchdog recovery race-free](task-01-race-free-watchdog-recovery.md)
+
+## Verification results
+
+- `go test ./internal/clarification ./internal/orchestrator -run 'TestResolverLiveDeliveryPublishesPrimaryAnswerBeforeWaiterReturns|TestClarificationWatchdogRecoveryIgnoresOwnCancelActivity|TestClarificationWatchdogCancellationInterruptsFallbackLookup|TestHandleAgentStreamEvent_CancelsClarificationWatchdogs' -count=1` passed.
+- `go test -race ./internal/clarification ./internal/orchestrator` passed.
+
+## Risks
+
+- Publishing the primary-answer event earlier must not publish terminal clarification messages before
+  durable delivery succeeds.
+- The recovery-owned phase must be narrow; a broad exemption could ignore genuine agent activity that
+  should stop fallback work.
+- Cancellation and replacement prompt ownership must retain the prompt-generation guarantees in ADR
+  0035.

docs/plans/clarification-watchdog-race-recovery/task-01-race-free-watchdog-recovery.md

@@ -0,0 +1,111 @@
+---
+id: "01-race-free-watchdog-recovery"
+title: "Make clarification watchdog recovery race-free"
+status: done
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-TASKS-CLARIFICATION-LIFECYCLE-001
+  - REQ-TASKS-CLARIFICATION-SCENARIOS-001
+acceptance_criteria:
+  - AC-TASKS-CLARIFICATION-LIFECYCLE-001.3
+  - AC-TASKS-CLARIFICATION-SCENARIOS-001.1
+system_design:
+  - ../../specs/tasks/system-design/clarification-active-lifecycle.md
+---
+
+# Task 01: Make Clarification Watchdog Recovery Race-Free
+
+## Summary
+
+Order live-answer watchdog registration before ACP response release. Preserve fallback recovery across
+the stream frames produced by its own silent cancellation while keeping all independent cancellation
+paths effective.
+
+## In scope
+
+- Write the two deterministic regression tests before changing production code.
+- Supply a synchronous local primary-answer notifier to the resolver and invoke it inside the
+  successful live delivery-confirmation boundary; retain the event bus as fan-out without duplicate
+  local watchdog registration.
+- Track the narrow recovery-owned silent-cancellation phase on the watchdog entry and classify only
+  exact cancellation acknowledgement frames by the cancellation operation's execution/prompt-generation
+  identity. Message, thinking, and tool frames remain authoritative activity.
+- Bind cancellation to the captured clarification turn and recheck turn authority after cancellation
+  before queueing the replacement; keep current-turn, queue, and shutdown behavior unchanged.
+
+## Out of scope
+
+- Timeout tuning.
+- Schema, API, frontend, or executor changes.
+- Detached-response flow changes.
+
+## Acceptance
+
+- A live waiter cannot receive the clarification answer before the synchronous local notifier has armed
+  its watchdog, and the fan-out event is published exactly once without a duplicate local watchdog.
+- Stream activity emitted synchronously by fallback's silent cancellation does not cancel its recovery
+  context; the answer reaches exactly one replacement queue or dispatch.
+- Independent live activity, successor-turn authority, and service shutdown still cancel or reject
+  in-flight watchdog fallback work.
+
+## Verification
+
+```bash
+go test ./internal/clarification ./internal/orchestrator -run 'TestResolverLiveDeliveryPublishesPrimaryAnswerBeforeWaiterReturns|TestClarificationWatchdogRecoveryIgnoresOwnCancelActivity|TestClarificationWatchdogCancellationInterruptsFallbackLookup|TestHandleAgentStreamEvent_CancelsClarificationWatchdogs'
+go test -race ./internal/clarification ./internal/orchestrator
+```
+
+## Files likely touched
+
+- `apps/backend/internal/clarification/resolver.go`
+- `apps/backend/internal/clarification/resolver_delivery_test.go`
+- `apps/backend/internal/orchestrator/event_handlers_clarification.go`
+- `apps/backend/internal/orchestrator/event_handlers_clarification_test.go`
+- `apps/backend/internal/orchestrator/event_handlers_streaming.go`
+
+## Dependencies
+
+None.
+
+## Risks
+
+- Event ordering is part of durable live-delivery confirmation and must not be moved outside that
+  boundary.
+- Recovery-owned activity classification must not suppress explicit user or coordinator cancellation.
+
+## Parallelism
+
+`sequential`
+
+## Inputs
+
+- `docs/specs/tasks/requirements/clarification-active-lifecycle.md`
+- `docs/specs/tasks/requirements/clarification-active-lifecycle-scenarios.md`
+- `docs/specs/tasks/system-design/clarification-active-lifecycle.md`
+- `docs/decisions/2026-08-14-current-turn-clarification-ownership.md`
+- `docs/decisions/0035-version-agent-ready-events-by-prompt-generation.md`
+- Production ACP and backend timeline from task `24e6e498-a816-45dc-926b-e8b32c8bc5e9`.
+
+## Results
+
+- Added `TestResolverLiveDeliveryPublishesPrimaryAnswerBeforeWaiterReturns`, which proves the primary-answer event is published before the live waiter returns and is published exactly once.
+- Added `TestResolverLiveDeliveryNotifiesWatchdogBeforeWaiterReturns`, which proves the synchronous local
+  watchdog notifier runs before the live waiter returns.
+- Added `TestResolverLiveDeliveryUsesSynchronousNotifierBeforeAsyncFanout`, which proves an async/NATS-like
+  event-bus publication cannot substitute for the construction-supplied local notifier.
+- Added `TestClarificationWatchdogRecoveryIgnoresOwnCancelActivity`, which synchronously emits `session_info` from silent cancellation and proves the watchdog context survives to one replacement prompt.
+- Kept primary-answer fan-out publication in the successful durable delivery-confirmation callback and
+  added the synchronous local watchdog notifier at that boundary.
+- Added a synchronous local watchdog notifier and an inline-handled bus marker so NATS fan-out cannot
+  release the waiter before local watchdog registration or arm a duplicate watchdog.
+- Added an atomic recovery-cancellation phase marker keyed to exact cancellation acknowledgement frames
+  and the cancellation operation's execution/prompt-generation identity; message, thinking, tool,
+  independent, and service-wide activity remain authoritative.
+- Bound silent cancellation to the captured clarification turn and added a post-cancel authority check
+  that accepts owned completion but rejects a successor turn.
+- Added regressions for newer prompt activity, same-execution message activity, post-cancel authority,
+  normal-agent-frame classification, and idempotent blocked-cancel cleanup.
+- Targeted regression command passed.
+- Race-enabled clarification and orchestrator suites passed.

docs/specs/tasks/requirements/clarification-active-lifecycle-scenarios.md

@@ -95,9 +95,18 @@ The lifecycle contract covers timeout, supersession, response, cleanup, and reco
 - **GIVEN** a primary-answer watchdog survives until another turn supersedes its clarification turn,
   **WHEN** its fallback timer expires, **THEN** both the preflight and serialized prompt-admission checks
   reject the stale answer without prompting or cancelling the successor.
+- **GIVEN** a live clarification answer is durably finalized, **WHEN** the agent acknowledges the answer
+  before the response endpoint finishes, **THEN** the primary-answer watchdog observes that activity and
+  does not cancel the current turn or dispatch the answer again.
 - **GIVEN** a primary-answer watchdog has entered fallback authority or prompt work, **WHEN** session
-  activity or service shutdown cancels the watchdog, **THEN** that cancellation reaches the in-flight
-  repository and prompt calls.
+  activity independent of fallback recovery or service shutdown cancels the watchdog, **THEN** that
+  cancellation reaches the in-flight repository and prompt calls.
+- **GIVEN** primary-answer fallback silently cancels the stuck turn, **WHEN** that cancellation emits
+  session or completion activity, **THEN** the fallback keeps its bounded recovery context through state
+  reconciliation and exactly one replacement-answer handoff.
+- **GIVEN** primary-answer fallback is silently cancelling a stuck turn, **WHEN** a message, thinking,
+  or tool frame arrives, **THEN** that frame cancels the watchdog unless it is an explicit cancellation
+  acknowledgement; matching execution identity alone does not suppress normal agent activity.
 - **GIVEN** a reserved successor is marked dispatch-attempted but remains unpublished, **WHEN** a client
   loads turn history before agentctl accepts or rejects it, **THEN** the successor is omitted until
   publication or durable message evidence prevents rollback from leaving stale client-only history.
@@ -180,4 +189,4 @@ The lifecycle contract covers timeout, supersession, response, cleanup, and reco
 - Changing permission-request lifecycle semantics; pending-session navigation reuses the shared
   `pending_action` type for permission parity.
 - Changing notification history for a clarification that was valid when its notification fired.
-- Automatically choosing a non-primary session when the task has no current pending action.
+- Automatically choosing a non-primary session when the task has no current pending action.

docs/specs/tasks/requirements/clarification-active-lifecycle.md

@@ -111,11 +111,19 @@ hiding the action the icon represents.
   endpoint still returns the acknowledgement error. A publication or summary-convergence error after
   the database restore does not make that retry unsafe; durable pending state remains authoritative.
   Once agentctl accepts the prompt, later publication or completion errors cannot roll back the
-  successor turn or reopen the answer. A primary-answer watchdog carries the clarification turn ID
-  and revalidates that ID both before fallback and inside serialized prompt admission, so it cannot
-  dispatch a stale answer into a successor turn. Its fallback keeps the watchdog cancellation context
-  through authority reads and prompt admission so session activity or service shutdown interrupts
-  in-flight recovery work.
+  successor turn or reopen the answer. A primary-answer watchdog is observable before a confirmed live
+  waiter returns the response to the agent, so acknowledgement activity cannot occur before the
+  watchdog can observe it. The resolver receives a synchronous local notifier at construction time for
+  this ordering boundary; event-bus fan-out, including NATS publication, is not an acknowledgement. The
+  watchdog carries the clarification turn ID and revalidates that ID both
+  before fallback and inside serialized prompt admission, so it cannot dispatch a stale answer into a
+  successor turn. Its fallback keeps the watchdog cancellation context through authority reads and
+  prompt admission so independent session activity or service shutdown interrupts in-flight recovery
+  work. Activity emitted by the fallback's own silent cancellation remains part of that recovery and
+  cannot cancel its context before the answer reaches the replacement handoff. Recovery may exempt only
+  a frame with the captured execution and prompt-generation identity and a cancellation-acknowledgement
+  type. Message, thinking, and tool frames remain authoritative activity even when their identity
+  matches the cancelled prompt.
 - A current-turn bundle remains answerable while any sibling question is pending. Recovery claims only
   those pending rows, preserves siblings already made terminal by an earlier partial write, and restores
   only the claimed rows if detached delivery fails. Primary delivery events and detached recovery derive
@@ -168,4 +176,4 @@ hiding the action the icon represents.
 
 
 
-Clarification presentation details belong to the UI system; agent identity and provider runtime behavior belong to the agent system.
+Clarification presentation details belong to the UI system; agent identity and provider runtime behavior belong to the agent system.

docs/specs/tasks/system-design/clarification-active-lifecycle.md

@@ -121,6 +121,36 @@ Transitions:
   `delivery_claimed` state is recoverable. A new request creates a new
   bundle identity; message deletion cannot reverse this transition.
 
+## Primary-answer watchdog recovery
+
+The primary path uses the live waiter's durable delivery-confirmation callback as the ordering
+boundary between response persistence and ACP delivery. The resolver is constructed with a synchronous
+local orchestrator notifier. After the claim is finalized, that notifier registers the watchdog for the
+pending ID and clarification turn before the callback returns the response to the agent. The event-bus
+publication remains fan-out for other consumers and carries an inline-handled marker so the local
+subscription does not arm a duplicate watchdog. NATS publication is not used as the acknowledgement:
+its publish call can return before subscribers run. The agent therefore cannot emit a tool-completion
+acknowledgement before an armed watchdog can observe and cancel itself. Terminal clarification message
+publication remains after successful delivery; watchdog registration does not publish those messages
+early.
+
+Each watchdog has an armed phase and a fallback-recovery phase. Independent live stream activity
+cancels either phase, and service shutdown cancels all phases. Once fallback owns the per-session
+cancel-and-handoff sequence, it marks the silent cancellation as recovery-owned. The cancellation
+operation's captured execution and prompt-generation identity plus a cancellation-acknowledgement
+frame type classifies frames caused by that cancellation. Only those exact frames are ignored; message,
+thinking, and tool frames always cancel the recovery context, and a newer execution or prompt
+generation cancels it even while cancellation is blocked. The marker is cleared when cancellation
+settles, so later independent activity retains its normal cancellation authority.
+
+Fallback uses one bounded context through turn-authority reads, silent cancellation, terminal-safe
+session reconciliation, and replacement-answer queue handoff. The recovery-owned activity exception
+does not weaken current-turn checks, prompt-generation checks, explicit user cancellation, coordinator
+stop, or service-shutdown cancellation. Recovery checks turn authority before cancellation and again
+after it settles. Because owned cancellation normally closes the captured turn, the post-cancel check
+accepts only that exact turn's durable completion when no active turn remains; an active successor fails
+closed. No watchdog or recovery phase is persisted.
+
 ## Permissions
 
 No authorization change. A user can see, answer, or dismiss only clarification data for a task and
@@ -168,6 +198,15 @@ session they can already access. Session selection does not broaden task visibil
   report whether retry state was recovered, and never restore it after a successor turn is accepted. A
   started confirmation may finish after the responder's bounded wait, but it cannot mutate the
   responder's claim snapshot and its durable finalization serializes against that restore.
+- Live delivery acknowledgement races response finalization: invoke the resolver's construction-supplied
+  local primary-answer watchdog notifier synchronously inside the finalized delivery-confirmation
+  boundary before returning the response to the agent. Keep the event-bus publication as fan-out and
+  mark the event as handled inline so the local subscriber does not register a duplicate watchdog. Do
+  not rely on NATS or another asynchronous bus delivery to establish watchdog ordering.
+- Primary-answer fallback cancellation produces its own stream activity: retain the bounded fallback
+  context while that recovery-owned cancellation settles, then revalidate the captured turn's durable
+  authority before the single replacement-answer handoff. Independent activity and service shutdown
+  still cancel recovery; a successor turn prevents the stale handoff.
 - Historical partial terminalization leaves pending and terminal siblings in one current-turn bundle:
   complete the pending siblings without rewriting terminal history or returning a permanent conflict.
 - A malformed persisted pending row has no matching durable turn: drain any live in-memory waiter, but