fix: handle deleted tasks and attribute runtime failures (#3910)

apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go

@@ -23,6 +23,57 @@ const (
 	containerStateRunning = "running"
 )
 
+type recoveryOutcomeSummary struct {
+	candidateCount          int
+	candidateCountKnown     bool
+	retrackedCount          int
+	notRetrackedDeadline    int
+	notRetrackedCanceled    int
+	notRetrackedTaskID      int
+	notRetrackedEnvironment int
+	notRetrackedAgent       int
+	notRetrackedTurnStatus  int
+	notRetrackedDuplicate   int
+}
+
+func (s recoveryOutcomeSummary) notRetrackedCount() int {
+	// retrackedCount should never exceed candidateCount; clamp to 0 defensively.
+	if !s.candidateCountKnown || s.candidateCount <= s.retrackedCount {
+		return 0
+	}
+	return s.candidateCount - s.retrackedCount
+}
+
+func (s recoveryOutcomeSummary) knownNotRetrackedCount() int {
+	return s.notRetrackedDeadline + s.notRetrackedCanceled + s.notRetrackedTaskID + s.notRetrackedEnvironment +
+		s.notRetrackedAgent + s.notRetrackedTurnStatus + s.notRetrackedDuplicate
+}
+
+func (s recoveryOutcomeSummary) unknownNotRetrackedCount() int {
+	unknown := s.notRetrackedCount() - s.knownNotRetrackedCount()
+	if unknown < 0 {
+		return 0
+	}
+	return unknown
+}
+
+func (s recoveryOutcomeSummary) logFields() []zap.Field {
+	return []zap.Field{
+		zap.Int("candidate_count", s.candidateCount),
+		zap.Bool("candidate_count_known", s.candidateCountKnown),
+		zap.Int("retracked_count", s.retrackedCount),
+		zap.Int("not_retracked_count", s.notRetrackedCount()),
+		zap.Int("not_retracked_deadline", s.notRetrackedDeadline),
+		zap.Int("not_retracked_canceled", s.notRetrackedCanceled),
+		zap.Int("not_retracked_task_identity", s.notRetrackedTaskID),
+		zap.Int("not_retracked_task_environment", s.notRetrackedEnvironment),
+		zap.Int("not_retracked_agent_identity", s.notRetrackedAgent),
+		zap.Int("not_retracked_turn_status", s.notRetrackedTurnStatus),
+		zap.Int("not_retracked_duplicate_execution", s.notRetrackedDuplicate),
+		zap.Int("not_retracked_unknown", s.unknownNotRetrackedCount()),
+	}
+}
+
 // Start starts the lifecycle manager background tasks
 func (m *Manager) Start(ctx context.Context) error {
 	if m.executorRegistry == nil {
@@ -55,6 +106,7 @@ func (m *Manager) Start(ctx context.Context) error {
 	// server, and hand them to every runtime's RecoverInstances unchanged.
 	startup.BeginStep(ctx, startup.StepSessionsRecovery)
 	records, listErr := m.ListLiveStandaloneExecutorsRunning(ctx)
+	recoveryOutcome := recoveryOutcomeSummary{candidateCountKnown: listErr == nil}
 	m.runRecoveryErr = listErr
 	if listErr != nil {
 		// A failed read leaves the record set unknown, which is not the same
@@ -103,6 +155,7 @@ func (m *Manager) Start(ctx context.Context) error {
 	// only backend that reads them, so this narrows nothing else.
 	records = recoverableRecords(records, guardedSessions)
 	if listErr == nil {
+		recoveryOutcome.candidateCount = len(records)
 		startup.SetTotal(ctx, startup.StepSessionsRecovery, int64(len(records)))
 	}
 
@@ -141,6 +194,11 @@ func (m *Manager) Start(ctx context.Context) error {
 				m.logger.Warn("recovery deadline elapsed before this record could be reconstructed; treating as not re-tracked",
 					zap.String("instance_id", ri.InstanceID),
 					zap.String("session_id", ri.SessionID))
+				if errors.Is(recoveryCtx.Err(), context.DeadlineExceeded) {
+					recoveryOutcome.notRetrackedDeadline++
+				} else if errors.Is(recoveryCtx.Err(), context.Canceled) {
+					recoveryOutcome.notRetrackedCanceled++
+				}
 				m.dispatchUnreconstructableStop(&stopWG, ri)
 				startup.Advance(ctx, startup.StepSessionsRecovery, 1)
 				continue
@@ -193,6 +251,7 @@ func (m *Manager) Start(ctx context.Context) error {
 				m.logger.Error("refusing to re-track recovered execution: task identity was not present in the recovery-inventory record",
 					zap.String("instance_id", execution.ID),
 					zap.String("session_id", execution.SessionID))
+				recoveryOutcome.notRetrackedTaskID++
 				m.dispatchUnreconstructableStop(&stopWG, ri)
 				startup.Advance(ctx, startup.StepSessionsRecovery, 1)
 				continue
@@ -211,6 +270,7 @@ func (m *Manager) Start(ctx context.Context) error {
 					zap.String("instance_id", execution.ID),
 					zap.String("session_id", execution.SessionID),
 					zap.Error(err))
+				recoveryOutcome.notRetrackedEnvironment++
 				m.dispatchUnreconstructableStop(&stopWG, ri)
 				startup.Advance(ctx, startup.StepSessionsRecovery, 1)
 				continue
@@ -237,6 +297,7 @@ func (m *Manager) Start(ctx context.Context) error {
 					zap.String("session_id", execution.SessionID),
 					zap.String("agent_profile_id", execution.AgentProfileID),
 					zap.Error(err))
+				recoveryOutcome.notRetrackedAgent++
 				m.dispatchUnreconstructableStop(&stopWG, ri)
 				startup.Advance(ctx, startup.StepSessionsRecovery, 1)
 				continue
@@ -252,6 +313,7 @@ func (m *Manager) Start(ctx context.Context) error {
 				m.logger.Error("refusing to re-track recovered execution: turn status could not be retrieved",
 					zap.String("instance_id", execution.ID),
 					zap.String("session_id", execution.SessionID))
+				recoveryOutcome.notRetrackedTurnStatus++
 				m.dispatchUnreconstructableStop(&stopWG, ri)
 				startup.Advance(ctx, startup.StepSessionsRecovery, 1)
 				continue
@@ -278,6 +340,7 @@ func (m *Manager) Start(ctx context.Context) error {
 					zap.String("execution_id", execution.ID),
 					zap.String("session_id", execution.SessionID),
 					zap.Error(err))
+				recoveryOutcome.notRetrackedDuplicate++
 				if ri.Client != nil {
 					ri.Client.Close()
 				}
@@ -287,6 +350,7 @@ func (m *Manager) Start(ctx context.Context) error {
 				continue
 			}
 			m.setRuntimeInterest(execution.SessionID, true)
+			recoveryOutcome.retrackedCount++
 			// AC-EXECUTORS-SURVIVAL-003.1: this execution is durably in the
 			// store as of the Add above, so its session is re-tracked from
 			// this point on regardless of which branch below applies the
@@ -361,6 +425,7 @@ func (m *Manager) Start(ctx context.Context) error {
 	// retained-unstoppable or stop-in-flight (AC-EXECUTORS-SURVIVAL-002.16,
 	// AC-EXECUTORS-SURVIVAL-003.7), which ReleaseAllExceptRetained leaves held.
 	m.recoveryGuard.ReleaseAllExceptRetained()
+	m.logger.Info("startup session recovery summary", recoveryOutcome.logFields()...)
 
 	// AC-EXECUTORS-SURVIVAL-003.6: this pass's recovery work (adoption,
 	// enumeration, and every recovered instance's reconstruction above) is

apps/backend/internal/agent/runtime/lifecycle/manager_recovery_sessions_step_test.go

@@ -3,6 +3,7 @@ package lifecycle
 import (
 	"context"
 	"errors"
+	"fmt"
 	"testing"
 	"time"
 
@@ -51,6 +52,145 @@ func completedStepFields(t *testing.T, logs *observer.ObservedLogs, wantStep sta
 	return fields
 }
 
+func newObservedRecoveryManagerLogger(t *testing.T) (*logger.Logger, *observer.ObservedLogs) {
+	t.Helper()
+	core, logs := observer.New(zap.DebugLevel)
+	log, err := logger.NewFromZap(zap.New(core))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return log, logs
+}
+
+func recoveryOutcomeFields(t *testing.T, logs *observer.ObservedLogs) map[string]interface{} {
+	t.Helper()
+	entries := logs.FilterMessage("startup session recovery summary").All()
+	if len(entries) != 1 {
+		t.Fatalf("recovery summary entries = %d, want one", len(entries))
+	}
+	return entries[0].ContextMap()
+}
+
+func assertRecoveryCount(t *testing.T, fields map[string]interface{}, name string, want int) {
+	t.Helper()
+	if got := fmt.Sprint(fields[name]); got != fmt.Sprint(want) {
+		t.Fatalf("%s = %v, want %d", name, fields[name], want)
+	}
+}
+
+func newRecoverySummaryManager(t *testing.T, log *logger.Logger, backend *MockExecutor) *Manager {
+	t.Helper()
+	registry := NewExecutorRegistry(log)
+	registry.Register(backend)
+	mgr := NewManager(newTestRegistry(), &MockEventBus{}, registry, &MockCredentialsManager{}, &MockProfileResolver{}, nil,
+		ExecutorFallbackWarn, "", log)
+	cleanupManagerStopCh(t, mgr)
+	t.Cleanup(func() { _ = mgr.Stop() })
+	return mgr
+}
+
+// @covers AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.4
+func TestManagerStartLogsRecoveryOutcomeSummaryForZeroCandidates(t *testing.T) {
+	log, logs := newObservedRecoveryManagerLogger(t)
+	mgr := newRecoverySummaryManager(t, log, &MockExecutor{name: executor.NameStandalone})
+	mgr.SetExecutorRunningWriter(&listingWriter{rows: []*models.ExecutorRunning{}})
+
+	if err := mgr.Start(context.Background()); err != nil {
+		t.Fatalf("Start() error = %v", err)
+	}
+	fields := recoveryOutcomeFields(t, logs)
+	if fields["candidate_count_known"] != true {
+		t.Fatalf("candidate_count_known = %v, want true for a successful empty inventory", fields["candidate_count_known"])
+	}
+	for name, want := range map[string]int{
+		"candidate_count": 0, "retracked_count": 0, "not_retracked_count": 0,
+		"not_retracked_deadline": 0, "not_retracked_task_identity": 0,
+		"not_retracked_task_environment": 0, "not_retracked_agent_identity": 0,
+		"not_retracked_turn_status": 0, "not_retracked_duplicate_execution": 0,
+		"not_retracked_unknown": 0,
+	} {
+		assertRecoveryCount(t, fields, name, want)
+	}
+}
+
+// @covers AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.4
+func TestManagerStartMarksCandidateCountUnknownWhenInventoryReadFails(t *testing.T) {
+	log, logs := newObservedRecoveryManagerLogger(t)
+	mgr := newRecoverySummaryManager(t, log, &MockExecutor{name: executor.NameStandalone})
+	mgr.SetExecutorRunningWriter(&listingWriter{err: errors.New("inventory read failed")})
+
+	if err := mgr.Start(context.Background()); err != nil {
+		t.Fatalf("Start() error = %v", err)
+	}
+	fields := recoveryOutcomeFields(t, logs)
+	if fields["candidate_count_known"] != false {
+		t.Fatalf("candidate_count_known = %v, want false after inventory read failure", fields["candidate_count_known"])
+	}
+	assertRecoveryCount(t, fields, "candidate_count", 0)
+	assertRecoveryCount(t, fields, "not_retracked_count", 0)
+	assertRecoveryCount(t, fields, "not_retracked_unknown", 0)
+}
+
+// @covers AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.4
+func TestManagerStartCountsBackendOmissionsAsUnknownRecoveryOutcomes(t *testing.T) {
+	log, logs := newObservedRecoveryManagerLogger(t)
+	mgr := newRecoverySummaryManager(t, log, &MockExecutor{name: executor.NameStandalone})
+	mgr.SetExecutorRunningWriter(&listingWriter{rows: []*models.ExecutorRunning{{SessionID: "session-1"}}})
+	reporter, stepLogs := newObservedSessionsRecoveryReporter(t)
+	reporter.Set(startup.RecoveringSessions)
+
+	if err := mgr.Start(startup.WithReporter(context.Background(), reporter)); err != nil {
+		t.Fatalf("Start() error = %v", err)
+	}
+	fields := recoveryOutcomeFields(t, logs)
+	if fields["candidate_count_known"] != true {
+		t.Fatalf("candidate_count_known = %v, want true", fields["candidate_count_known"])
+	}
+	assertRecoveryCount(t, fields, "candidate_count", 1)
+	assertRecoveryCount(t, fields, "retracked_count", 0)
+	assertRecoveryCount(t, fields, "not_retracked_count", 1)
+	assertRecoveryCount(t, fields, "not_retracked_unknown", 1)
+	stepFields := completedStepFields(t, stepLogs, startup.StepSessionsRecovery)
+	assertRecoveryCount(t, stepFields, "done", 0)
+	assertRecoveryCount(t, stepFields, "total", 1)
+}
+
+// @covers AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.4
+func TestManagerStartSeparatesRetrackedAndKnownRefusalOutcomes(t *testing.T) {
+	log, logs := newObservedRecoveryManagerLogger(t)
+	backend := &MockExecutor{
+		name: executor.NameStandalone,
+		recoverInstances: []*ExecutorInstance{
+			newRecoveryTurnOutcomeExecutorInstance(),
+			{InstanceID: "instance-missing-task", SessionID: "session-2", RuntimeName: executor.NameStandalone},
+		},
+	}
+	mgr := newRecoverySummaryManager(t, log, backend)
+	registerRecoveryTestAgentProfile(t, mgr)
+	mgr.SetExecutorRunningWriter(&listingWriter{rows: []*models.ExecutorRunning{
+		{TaskID: "task-1", SessionID: "session-1"},
+		{TaskID: "task-2", SessionID: "session-2"},
+	}})
+	reporter, stepLogs := newObservedSessionsRecoveryReporter(t)
+	reporter.Set(startup.RecoveringSessions)
+
+	if err := mgr.Start(startup.WithReporter(context.Background(), reporter)); err != nil {
+		t.Fatalf("Start() error = %v", err)
+	}
+	fields := recoveryOutcomeFields(t, logs)
+	assertRecoveryCount(t, fields, "candidate_count", 2)
+	assertRecoveryCount(t, fields, "retracked_count", 1)
+	assertRecoveryCount(t, fields, "not_retracked_count", 1)
+	assertRecoveryCount(t, fields, "not_retracked_task_identity", 1)
+	assertRecoveryCount(t, fields, "not_retracked_unknown", 0)
+	if _, ok := mgr.GetExecutionBySessionID("session-1"); !ok {
+		t.Fatal("the valid recovered session was not re-tracked")
+	}
+	stepFields := completedStepFields(t, stepLogs, startup.StepSessionsRecovery)
+	assertRecoveryCount(t, stepFields, "done", 2)
+	assertRecoveryCount(t, stepFields, "total", 2)
+}
+
 // TestManagerStartRecoverySessionsStepReportsCountedProgress covers the
 // sessions.recovery half of AC-PLATFORM-STARTUP-PROGRESS-005.2/.8 on the
 // ordinary (successful inventory read) path: the step totals from the

apps/backend/internal/backendapp/gateway.go

@@ -27,6 +27,7 @@ import (
 	"github.com/kandev/kandev/internal/orchestrator"
 	orchestratorhandlers "github.com/kandev/kandev/internal/orchestrator/handlers"
 	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/repository/repoerrors"
 	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
 	taskservice "github.com/kandev/kandev/internal/task/service"
 	"github.com/kandev/kandev/internal/task/statussummary"
@@ -350,7 +351,7 @@ func provideGateway(
 					return nil, err
 				}
 				if task == nil {
-					return nil, fmt.Errorf("task %q not found", taskID)
+					return nil, fmt.Errorf("%w: %s", repoerrors.ErrTaskNotFound, taskID)
 				}
 				observation, observationErr := orchestratorSvc.CurrentSessionCeilingObservation(ctx)
 				return statussummary.LaunchQueueSummaryFromTaskWithCapacity(task, &statussummary.LaunchQueueCapacityObservation{
@@ -366,7 +367,7 @@ func provideGateway(
 					return "", err
 				}
 				if task == nil {
-					return "", fmt.Errorf("task %q not found", taskID)
+					return "", fmt.Errorf("%w: %s", repoerrors.ErrTaskNotFound, taskID)
 				}
 				return task.WorkspaceID, nil
 			},

apps/backend/internal/orchestrator/executor/executor_pr_base_identity_test.go

@@ -3,8 +3,15 @@ package executor
 import (
 	"context"
 	"errors"
+	"fmt"
+	"strings"
 	"testing"
 
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
+
+	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/task/models"
 )
 
@@ -55,6 +62,164 @@ func TestResolveTaskRepoInfo_PRBaseUsesTargetRepository(t *testing.T) {
 	}
 }
 
+// @covers AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.3
+func TestPRBaseIdentityReturnsBoundedMismatchReasons(t *testing.T) {
+	type identityCase struct {
+		name           string
+		target         models.ComparisonTarget
+		number         int
+		checkoutBranch string
+		expected       *models.ComparisonTarget
+		attached       models.ComparisonTargetRepository
+		hasAttached    bool
+		head           models.ComparisonTargetRepository
+		hasHead        bool
+		contribution   bool
+		wantValid      bool
+		wantMismatch   prBaseIdentityMismatchReason
+	}
+	base := *forkPRComparisonTarget()
+	makeCase := func(name string, configure func(*identityCase)) identityCase {
+		result := identityCase{
+			name: name, target: base, number: base.Number, checkoutBranch: base.HeadBranch,
+			attached: base.TargetRepository, hasAttached: true,
+			head: base.HeadRepository, hasHead: true,
+		}
+		if configure != nil {
+			configure(&result)
+		}
+		return result
+	}
+
+	tests := []identityCase{
+		makeCase("PR number", func(c *identityCase) {
+			c.target.Number++
+			c.wantMismatch = prBaseMismatchPRNumber
+		}),
+		makeCase("provider", func(c *identityCase) {
+			c.target.Provider = "gitlab"
+			c.wantMismatch = prBaseMismatchProvider
+		}),
+		makeCase("kind", func(c *identityCase) {
+			c.target.Kind = "issue"
+			c.wantMismatch = prBaseMismatchKind
+		}),
+		makeCase("head branch", func(c *identityCase) {
+			c.checkoutBranch = "other"
+			c.wantMismatch = prBaseMismatchHeadBranch
+		}),
+		makeCase("repository identity missing", func(c *identityCase) {
+			c.hasAttached = false
+			c.wantMismatch = prBaseMismatchRepositoryIdentity
+		}),
+		makeCase("head repository", func(c *identityCase) {
+			c.target.TargetRepository.Path = "other/widgets"
+			c.head.Path = "another/widgets"
+			c.wantMismatch = prBaseMismatchHeadRepository
+		}),
+		makeCase("target repository", func(c *identityCase) {
+			c.contribution = true
+			c.target.TargetRepository.Path = "other/widgets"
+			c.wantMismatch = prBaseMismatchTargetRepository
+		}),
+		makeCase("checkout branch required", func(c *identityCase) {
+			c.checkoutBranch = ""
+			c.wantMismatch = prBaseMismatchCheckoutBranch
+		}),
+		makeCase("comparison target", func(c *identityCase) {
+			expected := base
+			c.expected = &expected
+			c.target.TargetRepository.Path = "other/widgets"
+			c.wantMismatch = prBaseMismatchComparisonTarget
+		}),
+		makeCase("valid unbound fork", func(c *identityCase) { c.wantValid = true }),
+		makeCase("valid contribution fork", func(c *identityCase) {
+			c.contribution = true
+			c.wantValid = true
+		}),
+		makeCase("valid expected binding", func(c *identityCase) {
+			expected := base
+			c.expected = &expected
+			c.wantValid = true
+		}),
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			valid, reason := validPRBaseIdentity(models.PRBase{Target: test.target}, test.number, test.checkoutBranch,
+				test.expected, test.attached, test.hasAttached, test.head, test.hasHead, test.contribution)
+			if valid != test.wantValid {
+				t.Fatalf("validPRBaseIdentity() valid = %v, want %v", valid, test.wantValid)
+			}
+			if reason != test.wantMismatch {
+				t.Fatalf("validPRBaseIdentity() reason = %q, want %q", reason, test.wantMismatch)
+			}
+		})
+	}
+}
+
+// @covers AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.3
+func TestResolvePRBaseForLaunchLogsInvalidResolvedBaseWithoutProviderData(t *testing.T) {
+	const rawProvider = "provider-value-must-not-be-logged"
+	tests := []struct {
+		name           string
+		withBinding    bool
+		wantInvalidErr bool
+	}{
+		{name: "unbound target", wantInvalidErr: true},
+		{name: "bound target", withBinding: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			repository := targetAttachedPRRepository(t)
+			base := forkPRComparisonTarget()
+			invalid := *base
+			invalid.Provider = rawProvider
+
+			core, observed := observer.New(zapcore.DebugLevel)
+			observedLogger, err := logger.NewFromZap(zap.New(core))
+			if err != nil {
+				t.Fatalf("NewFromZap: %v", err)
+			}
+			exec := newTestExecutor(t, &mockAgentManager{}, newMockRepository())
+			exec.logger = observedLogger
+			exec.SetPRBaseResolver(&recordingPRBaseResolver{result: &models.PRBase{Target: invalid}})
+
+			info := &repoInfo{PRNumber: 42, CheckoutBranch: "feature"}
+			if test.withBinding {
+				info.ComparisonTarget = base
+			}
+			tr := &models.TaskRepository{ID: "task-repo-safe-log", TaskID: "task-safe-log", BaseBranch: "main"}
+			resolveErr := exec.resolvePRBaseForLaunch(context.Background(), tr, repository, info)
+			switch {
+			case test.wantInvalidErr:
+				var resolutionErr *PRBaseResolutionError
+				if !errors.As(resolveErr, &resolutionErr) || !resolutionErr.InvalidAssociation() {
+					t.Fatalf("resolvePRBaseForLaunch() error = %v, want invalid-association error", resolveErr)
+				}
+			case resolveErr != nil:
+				t.Fatalf("resolvePRBaseForLaunch() error = %v, want safe fallback", resolveErr)
+			case info.BaseBranch != base.TargetBranch:
+				t.Fatalf("fallback base branch = %q, want bound branch %q", info.BaseBranch, base.TargetBranch)
+			}
+
+			entries := observed.FilterMessage("live pull request base identity mismatch").All()
+			if len(entries) != 1 {
+				t.Fatalf("identity mismatch logs = %d, want one", len(entries))
+			}
+			fields := entries[0].ContextMap()
+			if fields["task_repository_id"] != tr.ID || fmt.Sprint(fields["pr_number"]) != "42" ||
+				fields["mismatch_reason"] != "invalid_resolved_pr_base" {
+				t.Fatalf("identity mismatch fields = %#v, want safe identity and fixed invalid-base reason", fields)
+			}
+			for _, entry := range observed.All() {
+				if strings.Contains(fmt.Sprint(entry.ContextMap()), rawProvider) {
+					t.Fatalf("identity mismatch log exposed raw provider data: %#v", entry)
+				}
+			}
+		})
+	}
+}
+
 func TestResolveTaskRepoInfo_PRBaseRejectsDifferentHeadForSameNumber(t *testing.T) {
 	target := forkPRComparisonTarget()
 	metadata := map[string]interface{}{"pr_number": 42}
@@ -254,6 +419,12 @@ func TestResolveAllRepoInfoRejectsLaunchWhenOnePRBindingIsInvalid(t *testing.T)
 	target.TargetBranch = "main"
 	resolver := &recordingPRBaseResolver{result: &models.PRBase{Target: *target}}
 	exec := newTestExecutor(t, &mockAgentManager{}, repo)
+	core, observed := observer.New(zapcore.DebugLevel)
+	observedLogger, err := logger.NewFromZap(zap.New(core))
+	if err != nil {
+		t.Fatalf("NewFromZap: %v", err)
+	}
+	exec.logger = observedLogger
 	exec.SetPRBaseResolver(resolver)
 	infos, err := exec.resolveAllRepoInfo(context.Background(), "task-1")
 	if err == nil || infos != nil {
@@ -262,6 +433,18 @@ func TestResolveAllRepoInfoRejectsLaunchWhenOnePRBindingIsInvalid(t *testing.T)
 	if len(resolver.calls) != 2 {
 		t.Fatalf("resolver calls = %d, want both repository bindings checked", len(resolver.calls))
 	}
+	entries := observed.FilterMessage("live pull request base identity mismatch").All()
+	if len(entries) != 1 {
+		t.Fatalf("identity mismatch logs = %d, want one", len(entries))
+	}
+	fields := entries[0].ContextMap()
+	if fields["task_repository_id"] != "task-repo-repo-2" || fmt.Sprint(fields["pr_number"]) != "42" ||
+		fields["mismatch_reason"] != "head_repository_mismatch" {
+		t.Fatalf("identity mismatch fields = %#v, want task repository, PR number and fixed head mismatch reason", fields)
+	}
+	if strings.Contains(fmt.Sprint(fields), "https://github.com") || strings.Contains(fmt.Sprint(fields), "upstream/widgets") {
+		t.Fatalf("identity mismatch log exposed repository identity: %#v", fields)
+	}
 }
 
 func targetAttachedPRRepository(t *testing.T) *models.Repository {

apps/backend/internal/orchestrator/executor/executor_resume.go

@@ -318,13 +318,13 @@ func (e *Executor) resolvePRBaseForLaunch(
 		return nil
 	}
 	if err := resolved.Validate(); err != nil {
+		e.logger.Warn("live pull request base identity mismatch",
+			zap.String("task_repository_id", tr.ID),
+			zap.Int("pr_number", info.PRNumber),
+			zap.String("mismatch_reason", string(prBaseMismatchInvalidResolvedBase)))
 		if expected == nil {
 			return NewPRBaseResolutionError(err, false, true)
 		}
-		e.logger.Debug("live pull request base identity did not match the task repository",
-			zap.String("task_id", tr.TaskID),
-			zap.Int("pr_number", info.PRNumber),
-			zap.Error(err))
 		return nil
 	}
 	attachedRepository, hasAttachedRepository := githubComparisonRepositoryFromRepository(repo)
@@ -333,14 +333,16 @@ func (e *Executor) resolvePRBaseForLaunch(
 	if info.RemoteContribution != nil {
 		headRepository, hasHeadRepository = normalizeGitHubComparisonRepository(info.RemoteContribution.SourceRepository)
 	}
-	if !validPRBaseIdentity(resolved, info.PRNumber, info.CheckoutBranch, expected,
-		attachedRepository, hasAttachedRepository, headRepository, hasHeadRepository, info.RemoteContribution != nil) {
+	validIdentity, mismatchReason := validPRBaseIdentity(resolved, info.PRNumber, info.CheckoutBranch, expected,
+		attachedRepository, hasAttachedRepository, headRepository, hasHeadRepository, info.RemoteContribution != nil)
+	if !validIdentity {
+		e.logger.Warn("live pull request base identity mismatch",
+			zap.String("task_repository_id", tr.ID),
+			zap.Int("pr_number", info.PRNumber),
+			zap.String("mismatch_reason", string(mismatchReason)))
 		if expected == nil {
 			return NewPRBaseResolutionError(errors.New("pull request identity did not match the task repository binding"), false, true)
 		}
-		e.logger.Debug("live pull request base identity did not match the task repository",
-			zap.String("task_id", tr.TaskID),
-			zap.Int("pr_number", info.PRNumber))
 		return nil
 	}
 	if resolved.Target.TargetBranch != tr.BaseBranch {
@@ -403,20 +405,40 @@ func isUnusablePRBaseAssociation(err error) bool {
 	return errors.As(err, &invalidAssociation) && invalidAssociation.InvalidAssociation()
 }
 
+type prBaseIdentityMismatchReason string
+
+const (
+	prBaseMismatchPRNumber            prBaseIdentityMismatchReason = "pr_number_mismatch"
+	prBaseMismatchProvider            prBaseIdentityMismatchReason = "provider_mismatch"
+	prBaseMismatchKind                prBaseIdentityMismatchReason = "kind_mismatch"
+	prBaseMismatchHeadBranch          prBaseIdentityMismatchReason = "head_branch_mismatch"
+	prBaseMismatchRepositoryIdentity  prBaseIdentityMismatchReason = "repository_identity_missing"
+	prBaseMismatchHeadRepository      prBaseIdentityMismatchReason = "head_repository_mismatch"
+	prBaseMismatchTargetRepository    prBaseIdentityMismatchReason = "target_repository_mismatch"
+	prBaseMismatchCheckoutBranch      prBaseIdentityMismatchReason = "checkout_branch_required"
+	prBaseMismatchComparisonTarget    prBaseIdentityMismatchReason = "comparison_target_mismatch"
+	prBaseMismatchInvalidResolvedBase prBaseIdentityMismatchReason = "invalid_resolved_pr_base"
+)
+
 func validPRBaseIdentity(
 	base models.PRBase, number int, checkoutBranch string,
 	expected *models.ComparisonTarget,
 	attachedRepository models.ComparisonTargetRepository, hasAttachedRepository bool,
 	headRepository models.ComparisonTargetRepository, hasHeadRepository bool,
 	contribution bool,
-) bool {
+) (bool, prBaseIdentityMismatchReason) {
 	target := base.Target
-	if target.Number != number || target.Provider != models.ComparisonTargetProviderGitHub ||
-		target.Kind != models.ComparisonTargetKindPullRequest {
-		return false
+	if target.Number != number {
+		return false, prBaseMismatchPRNumber
+	}
+	if target.Provider != models.ComparisonTargetProviderGitHub {
+		return false, prBaseMismatchProvider
+	}
+	if target.Kind != models.ComparisonTargetKindPullRequest {
+		return false, prBaseMismatchKind
 	}
 	if checkoutBranch != "" && target.HeadBranch != checkoutBranch {
-		return false
+		return false, prBaseMismatchHeadBranch
 	}
 	if expected != nil || contribution {
 		return validBoundPRBaseIdentity(target, expected, attachedRepository, hasAttachedRepository,
@@ -430,38 +452,49 @@ func validBoundPRBaseIdentity(
 	target models.ComparisonTarget, expected *models.ComparisonTarget,
 	attachedRepository models.ComparisonTargetRepository, hasAttachedRepository bool,
 	headRepository models.ComparisonTargetRepository, hasHeadRepository, contribution bool,
-) bool {
+) (bool, prBaseIdentityMismatchReason) {
 	if !hasAttachedRepository || !hasHeadRepository ||
 		!models.ComparisonTargetRepositoriesEqual(target.HeadRepository, headRepository) {
-		return false
+		if !hasAttachedRepository || !hasHeadRepository {
+			return false, prBaseMismatchRepositoryIdentity
+		}
+		return false, prBaseMismatchHeadRepository
 	}
 	if contribution && !models.ComparisonTargetRepositoriesEqual(target.TargetRepository, attachedRepository) {
-		return false
+		return false, prBaseMismatchTargetRepository
 	}
 	if expected == nil {
-		return true
+		return true, ""
 	}
-	return expected.ChangeIdentityEqual(target) &&
-		expected.HeadBranch == target.HeadBranch &&
-		models.ComparisonTargetRepositoriesEqual(expected.HeadRepository, target.HeadRepository)
+	if !expected.ChangeIdentityEqual(target) || expected.HeadBranch != target.HeadBranch ||
+		!models.ComparisonTargetRepositoriesEqual(expected.HeadRepository, target.HeadRepository) {
+		return false, prBaseMismatchComparisonTarget
+	}
+	return true, ""
 }
 
 func validUnboundPRBaseIdentity(
 	target models.ComparisonTarget, checkoutBranch string,
 	attachedRepository models.ComparisonTargetRepository, hasAttachedRepository bool,
 	headRepository models.ComparisonTargetRepository, hasHeadRepository bool,
-) bool {
+) (bool, prBaseIdentityMismatchReason) {
 	if !hasAttachedRepository || !hasHeadRepository {
-		return false
+		return false, prBaseMismatchRepositoryIdentity
 	}
 	if models.ComparisonTargetRepositoriesEqual(target.TargetRepository, attachedRepository) {
 		if models.ComparisonTargetRepositoriesEqual(target.HeadRepository, attachedRepository) {
-			return true
+			return true, ""
 		}
-		return checkoutBranch != ""
+		if checkoutBranch != "" {
+			return true, ""
+		}
+		return false, prBaseMismatchCheckoutBranch
 	}
 	// Fork-attached legacy tasks bind the attached repository to the PR head.
-	return models.ComparisonTargetRepositoriesEqual(target.HeadRepository, headRepository)
+	if models.ComparisonTargetRepositoriesEqual(target.HeadRepository, headRepository) {
+		return true, ""
+	}
+	return false, prBaseMismatchHeadRepository
 }
 
 func githubComparisonRepositoryFromRepository(repo *models.Repository) (models.ComparisonTargetRepository, bool) {

apps/backend/internal/persistence/requiredstores/health.go

@@ -2,6 +2,7 @@ package requiredstores
 
 import (
 	"context"
+	"database/sql"
 	"errors"
 	"fmt"
 	"strings"
@@ -34,6 +35,15 @@ type Health struct {
 	done     chan struct{}
 }
 
+type probeFailureDiagnostic struct {
+	stage       string
+	elapsed     time.Duration
+	errorClass  string
+	storeID     string
+	writerStats sql.DBStats
+	readerStats sql.DBStats
+}
+
 // NewHealth creates a runtime health probe for a completed store tracker.
 func NewHealth(tracker *Tracker, pool *db.Pool, log *logger.Logger) *Health {
 	return &Health{tracker: tracker, pool: pool, log: log, interval: defaultProbeInterval}
@@ -53,18 +63,29 @@ func (h *Health) SetInterval(interval time.Duration) {
 // more stores are unavailable, while recording an independent result for each
 // catalog entry.
 func (h *Health) Check(ctx context.Context) error {
+	_, err := h.check(ctx)
+	return err
+}
+
+func (h *Health) check(ctx context.Context) (*probeFailureDiagnostic, error) {
 	if h == nil || h.tracker == nil {
-		return errors.New("required-store health tracker is unavailable")
+		return nil, errors.New("required-store health tracker is unavailable")
 	}
 	if h.pool == nil || h.pool.Writer() == nil || h.pool.Reader() == nil {
-		return h.recordUnavailable(errors.New("database pool is unavailable"))
+		err := errors.New("database pool is unavailable")
+		diagnostic := h.failureDiagnostic("pool", 0, err, "")
+		return diagnostic, h.recordUnavailable(err)
 	}
-	probeErr := h.ping(ctx)
+	diagnostic, probeErr := h.pingObserved(ctx)
 	results := make([]error, len(h.tracker.catalog))
 	for index, descriptor := range h.tracker.catalog {
 		results[index] = probeErr
 		if probeErr == nil {
-			results[index] = h.probeTables(ctx, descriptor)
+			tableDiagnostic, tableErr := h.probeTablesObserved(ctx, descriptor)
+			results[index] = tableErr
+			if diagnostic == nil {
+				diagnostic = tableDiagnostic
+			}
 		}
 	}
 	var failures []error
@@ -79,10 +100,10 @@ func (h *Health) Check(ctx context.Context) error {
 	}
 	if len(failures) == 0 {
 		h.logTransition()
-		return nil
+		return nil, nil
 	}
 	h.logTransition()
-	return errors.Join(failures...)
+	return diagnostic, errors.Join(failures...)
 }
 
 // MarkUnavailable records a destructive database transition before the
@@ -100,31 +121,41 @@ func (h *Health) MarkUnavailable() {
 // the probe instead of turning bounded writer contention into an unhealthy
 // state. Startup callers continue to use Check, which remains strict.
 func (h *Health) checkRuntime(ctx context.Context) (deferred bool, err error) {
+	deferred, _, err = h.checkRuntimeDetailed(ctx)
+	return deferred, err
+}
+
+func (h *Health) checkRuntimeDetailed(
+	ctx context.Context,
+) (deferred bool, diagnostic *probeFailureDiagnostic, err error) {
 	if h.isSQLite() {
 		release, ok := maintenance.ForPool(h.pool).TryAcquire()
 		if !ok {
 			if h.log != nil {
 				h.log.Debug("required persistence probe deferred during database maintenance")
 			}
-			return true, nil
+			return true, nil, nil
 		}
 		defer release()
 	}
-	return false, h.Check(ctx)
+	diagnostic, err = h.check(ctx)
+	return false, diagnostic, err
 }
 
 func (h *Health) isSQLite() bool {
 	return h.pool != nil && h.pool.Writer() != nil && h.pool.Writer().DriverName() == dialect.SQLite3
 }
 
-func (h *Health) ping(ctx context.Context) error {
+func (h *Health) pingObserved(ctx context.Context) (*probeFailureDiagnostic, error) {
+	started := time.Now()
 	if err := h.pool.Writer().PingContext(ctx); err != nil {
-		return fmt.Errorf("writer ping failed: %w", err)
+		return h.failureDiagnostic("writer_ping", time.Since(started), err, ""), fmt.Errorf("writer ping failed: %w", err)
 	}
+	started = time.Now()
 	if err := h.pool.Reader().PingContext(ctx); err != nil {
-		return fmt.Errorf("reader ping failed: %w", err)
+		return h.failureDiagnostic("reader_ping", time.Since(started), err, ""), fmt.Errorf("reader ping failed: %w", err)
 	}
-	return nil
+	return nil, nil
 }
 
 func (h *Health) probeTables(ctx context.Context, descriptor Descriptor) error {
@@ -140,6 +171,84 @@ func (h *Health) probeTables(ctx context.Context, descriptor Descriptor) error {
 	return nil
 }
 
+func (h *Health) probeTablesObserved(
+	ctx context.Context,
+	descriptor Descriptor,
+) (*probeFailureDiagnostic, error) {
+	started := time.Now()
+	err := h.probeTables(ctx, descriptor)
+	if err == nil {
+		return nil, nil
+	}
+	return h.failureDiagnostic("table_probe", time.Since(started), err, descriptor.ID), err
+}
+
+func (h *Health) failureDiagnostic(stage string, elapsed time.Duration, err error, storeID string) *probeFailureDiagnostic {
+	diagnostic := &probeFailureDiagnostic{
+		stage:      stage,
+		elapsed:    elapsed,
+		errorClass: classifyProbeError(err),
+		storeID:    storeID,
+	}
+	if h.pool == nil {
+		return diagnostic
+	}
+	if writer := h.pool.Writer(); writer != nil {
+		diagnostic.writerStats = writer.Stats()
+	}
+	if reader := h.pool.Reader(); reader != nil {
+		diagnostic.readerStats = reader.Stats()
+	}
+	return diagnostic
+}
+
+func classifyProbeError(err error) string {
+	message := ""
+	if err != nil {
+		message = strings.ToLower(err.Error())
+	}
+	switch {
+	case errors.Is(err, context.DeadlineExceeded):
+		return "deadline_exceeded"
+	case errors.Is(err, context.Canceled):
+		return "canceled"
+	case strings.Contains(message, "required table") && strings.Contains(message, "missing"):
+		return "required_table_missing"
+	default:
+		return "probe_error"
+	}
+}
+
+func (d *probeFailureDiagnostic) logFields() []zap.Field {
+	if d == nil {
+		return []zap.Field{zap.String("stage", "unknown"), zap.String("error_class", "probe_error")}
+	}
+	fields := []zap.Field{
+		zap.String("stage", d.stage),
+		zap.Float64("elapsed_ms", float64(d.elapsed)/float64(time.Millisecond)),
+		zap.String("error_class", d.errorClass),
+		zap.Int("writer_open_connections", d.writerStats.OpenConnections),
+		zap.Int("writer_in_use", d.writerStats.InUse),
+		zap.Int64("writer_wait_count", d.writerStats.WaitCount),
+		zap.Float64("writer_wait_duration_ms", float64(d.writerStats.WaitDuration)/float64(time.Millisecond)),
+		zap.Int("reader_open_connections", d.readerStats.OpenConnections),
+		zap.Int("reader_in_use", d.readerStats.InUse),
+		zap.Int64("reader_wait_count", d.readerStats.WaitCount),
+		zap.Float64("reader_wait_duration_ms", float64(d.readerStats.WaitDuration)/float64(time.Millisecond)),
+	}
+	if d.storeID != "" {
+		fields = append(fields, zap.String("store_id", d.storeID))
+	}
+	return fields
+}
+
+func (h *Health) logProbeFailure(diagnostic *probeFailureDiagnostic) {
+	if h == nil || h.log == nil {
+		return
+	}
+	h.log.Warn("required persistence probe failed", diagnostic.logFields()...)
+}
+
 func (h *Health) recordUnavailable(err error) error {
 	for _, descriptor := range h.tracker.catalog {
 		if recordErr := h.tracker.RecordProbe(descriptor.ID, err); recordErr != nil {
@@ -179,10 +288,13 @@ func (h *Health) run(ctx context.Context, interval time.Duration, done chan stru
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
+			if ctx.Err() != nil {
+				return
+			}
 			checkCtx, cancel := context.WithTimeout(ctx, probeTimeout)
-			_, err := h.checkRuntime(checkCtx)
-			if err != nil && h.log != nil {
-				h.log.Warn("required persistence probe failed", zap.Error(err))
+			_, diagnostic, err := h.checkRuntimeDetailed(checkCtx)
+			if err != nil {
+				h.logProbeFailure(diagnostic)
 			}
 			cancel()
 		}

apps/backend/internal/persistence/requiredstores/health_test.go

@@ -6,7 +6,12 @@ import (
 	"testing"
 	"time"
 
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
+
 	"github.com/jmoiron/sqlx"
+	"github.com/kandev/kandev/internal/common/logger"
 	_ "github.com/mattn/go-sqlite3"
 
 	"github.com/kandev/kandev/internal/db"
@@ -219,6 +224,177 @@ func TestRuntimeHealthResumesAfterMaintenance(t *testing.T) {
 	successLease()
 }
 
+func TestRuntimeHealthProbeFailureLogsBoundedStageAndRecovers(t *testing.T) {
+	for _, test := range []struct {
+		name      string
+		stage     string
+		blockedDB string
+	}{
+		{name: "writer ping", stage: "writer_ping", blockedDB: "writer"},
+		{name: "reader ping", stage: "reader_ping", blockedDB: "reader"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			writer, err := sqlx.Open("sqlite3", ":memory:")
+			if err != nil {
+				t.Fatalf("open writer: %v", err)
+			}
+			writer.SetMaxOpenConns(1)
+			reader, err := sqlx.Open("sqlite3", ":memory:")
+			if err != nil {
+				_ = writer.Close()
+				t.Fatalf("open reader: %v", err)
+			}
+			reader.SetMaxOpenConns(1)
+			pool := db.NewPool(writer, reader)
+			t.Cleanup(func() { _ = pool.Close() })
+			for _, table := range []string{"first", "second"} {
+				if _, err := writer.Exec("CREATE TABLE " + table + " (id TEXT PRIMARY KEY)"); err != nil {
+					t.Fatalf("create %s table: %v", table, err)
+				}
+			}
+
+			descriptors := []Descriptor{
+				{ID: "first", OwnerPackage: "owner/first", RequiredTables: []string{"first"}, Sweep: startup.StepStoresRepositories},
+				{ID: "second", OwnerPackage: "owner/second", RequiredTables: []string{"second"}, Sweep: startup.StepStoresRepositories},
+			}
+			tracker, err := NewTracker(descriptors)
+			if err != nil {
+				t.Fatalf("NewTracker: %v", err)
+			}
+			for _, descriptor := range descriptors {
+				if err := tracker.RecordSuccess(descriptor.ID); err != nil {
+					t.Fatalf("RecordSuccess(%s): %v", descriptor.ID, err)
+				}
+			}
+
+			core, observed := observer.New(zapcore.WarnLevel)
+			warningObserved := make(chan struct{}, 1)
+			releaseWarning := make(chan struct{})
+			core = zapcore.RegisterHooks(core, func(entry zapcore.Entry) error {
+				if entry.Message == "required persistence probe failed" {
+					select {
+					case warningObserved <- struct{}{}:
+					default:
+					}
+					<-releaseWarning
+				}
+				return nil
+			})
+			log, err := logger.NewFromZap(zap.New(core))
+			if err != nil {
+				t.Fatalf("NewFromZap: %v", err)
+			}
+			health := NewHealth(tracker, pool, log)
+			if err := health.Check(context.Background()); err != nil {
+				t.Fatalf("initial Check: %v", err)
+			}
+
+			blocked := writer
+			if test.blockedDB == "reader" {
+				blocked = reader
+			}
+			tx, err := blocked.Beginx()
+			if err != nil {
+				t.Fatalf("begin blocking transaction: %v", err)
+			}
+			t.Cleanup(func() { _ = tx.Rollback() })
+
+			ctx, cancel := context.WithCancel(context.Background())
+			done := make(chan struct{})
+			go health.run(ctx, 200*time.Millisecond, done)
+			select {
+			case <-warningObserved:
+				cancel()
+				close(releaseWarning)
+			case <-time.After(5 * time.Second):
+				cancel()
+				close(releaseWarning)
+				<-done
+				t.Fatal("health probe loop did not emit its first failure warning")
+			}
+			select {
+			case <-done:
+			case <-time.After(2 * time.Second):
+				t.Fatal("health probe loop did not stop after its context was canceled")
+			}
+
+			entries := observed.FilterMessage("required persistence probe failed").All()
+			if len(entries) != 1 {
+				t.Fatalf("failure warnings = %d, want one bounded warning: %+v", len(entries), entries)
+			}
+			fields := entries[0].ContextMap()
+			if got := fields["stage"]; got != test.stage {
+				t.Fatalf("failure stage = %v, want %q", got, test.stage)
+			}
+			if got := fields["error_class"]; got != "deadline_exceeded" {
+				t.Fatalf("error_class = %v, want deadline_exceeded", got)
+			}
+			if got, ok := fields["elapsed_ms"].(float64); !ok || got <= 0 {
+				t.Fatalf("elapsed_ms = %v, want a positive number", fields["elapsed_ms"])
+			}
+			for _, field := range []string{
+				"writer_open_connections", "writer_in_use", "writer_wait_count", "writer_wait_duration_ms",
+				"reader_open_connections", "reader_in_use", "reader_wait_count", "reader_wait_duration_ms",
+			} {
+				if _, ok := fields[field]; !ok {
+					t.Fatalf("failure warning omitted %q: %#v", field, fields)
+				}
+			}
+			waitField := test.blockedDB + "_wait_count"
+			if got, ok := fields[waitField].(int64); !ok || got == 0 {
+				t.Fatalf("%s = %v, want evidence of a pool wait", waitField, fields[waitField])
+			}
+			if _, ok := fields["error"]; ok {
+				t.Fatalf("failure warning exposed raw error: %#v", fields["error"])
+			}
+			if health.Healthy() {
+				t.Fatal("health remained healthy after a timed-out probe")
+			}
+
+			if err := tx.Rollback(); err != nil {
+				t.Fatalf("release blocked connection: %v", err)
+			}
+			if err := health.Check(context.Background()); err != nil {
+				t.Fatalf("recovery Check: %v", err)
+			}
+			if !health.Healthy() {
+				t.Fatal("health did not recover after the blocked pool became available")
+			}
+		})
+	}
+}
+
+func TestHealthCheckCapturesTableProbeFailure(t *testing.T) {
+	_, _, _, health := newSQLiteHealthFixture(t, []Descriptor{{
+		ID: "first", OwnerPackage: "owner/first", RequiredTables: []string{"missing"}, Sweep: startup.StepStoresRepositories,
+	}})
+
+	diagnostic, err := health.check(context.Background())
+	if err == nil {
+		t.Fatal("health check returned nil with a missing required table")
+	}
+	if diagnostic == nil {
+		t.Fatal("health check returned no probe diagnostic")
+	}
+	if diagnostic.stage != "table_probe" {
+		t.Fatalf("diagnostic stage = %q, want table_probe", diagnostic.stage)
+	}
+	if diagnostic.errorClass != "required_table_missing" {
+		t.Fatalf("diagnostic error class = %q, want required_table_missing", diagnostic.errorClass)
+	}
+	if diagnostic.storeID != "first" {
+		t.Fatalf("diagnostic store ID = %q, want first", diagnostic.storeID)
+	}
+	if diagnostic.elapsed < 0 {
+		t.Fatalf("diagnostic elapsed = %s, want a non-negative duration", diagnostic.elapsed)
+	}
+	for _, field := range diagnostic.logFields() {
+		if field.Key == "error" || field.Key == "table" {
+			t.Fatalf("table probe diagnostic contains an unsafe field %q", field.Key)
+		}
+	}
+}
+
 func TestHealthCheckMarksMissingTableUnhealthyAndRecovers(t *testing.T) {
 	conn, err := sqlx.Open("sqlite3", ":memory:")
 	if err != nil {

apps/backend/internal/plugins/handlers.go

@@ -40,6 +40,9 @@ const (
 	maxPluginActionEnvelopeBytes = manifest.MaxActionBodyBytes + 4096
 	maxPluginActionResponseBytes = 1 << 20 // 1 MiB
 	contentTypeHeader            = "Content-Type"
+	webhookOriginHostLifecycle   = "host_lifecycle"
+	webhookOriginHostRPC         = "host_rpc"
+	webhookOriginPluginResponse  = "plugin_response"
 )
 
 var pluginActionTimeout = 15 * time.Second
@@ -454,6 +457,7 @@ func (c *Controller) webhook(ctx *gin.Context) {
 
 	leasedRecord, release, err := c.svc.beginPluginDispatch(id, dispatchGeneration(record))
 	if err != nil {
+		c.logWebhookFailure(id, http.StatusServiceUnavailable, webhookOriginHostLifecycle, err)
 		ctx.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()})
 		return
 	}
@@ -461,12 +465,39 @@ func (c *Controller) webhook(ctx *gin.Context) {
 
 	resp, err := c.webhookInvoker.InvokeWebhook(ctx.Request.Context(), id, req)
 	if err != nil {
+		c.logWebhookFailure(id, http.StatusServiceUnavailable, webhookOriginHostRPC, err)
 		ctx.JSON(http.StatusServiceUnavailable, gin.H{"error": err.Error()})
 		return
 	}
 	c.writeWebhookResponse(ctx, leasedRecord, resp)
 }
 
+func (c *Controller) logWebhookFailure(pluginID string, status int, origin string, err error) {
+	if c == nil || c.log == nil {
+		return
+	}
+	fields := []zap.Field{
+		zap.String("plugin_id", pluginID),
+		zap.Int("status", status),
+		zap.String("origin", origin),
+	}
+	if err != nil {
+		fields = append(fields, zap.String("error_class", webhookHostErrorClass(err)))
+	}
+	c.log.Warn("plugin webhook failed", fields...)
+}
+
+func webhookHostErrorClass(err error) string {
+	switch {
+	case errors.Is(err, context.DeadlineExceeded):
+		return "deadline_exceeded"
+	case errors.Is(err, context.Canceled):
+		return "canceled"
+	default:
+		return "host_error"
+	}
+}
+
 // webhookCallerAuthorized requires a caller identity unless the declaration
 // explicitly opts into public access. It runs before lookup errors to avoid
 // revealing installed plugin IDs or declared webhook keys to anonymous callers.
@@ -577,6 +608,7 @@ func webhookStatusForResponse(status int32) (int, bool) {
 func (c *Controller) writeWebhookResponse(ctx *gin.Context, record *store.Record, resp *pluginsdk.WebhookResponse) {
 	status, ok := webhookStatusForResponse(resp.Status)
 	if !ok {
+		c.logWebhookFailure(record.ID, http.StatusBadGateway, webhookOriginPluginResponse, nil)
 		ctx.JSON(http.StatusBadGateway, gin.H{
 			"error": fmt.Sprintf("plugin returned invalid webhook status %d", resp.Status),
 		})
@@ -597,6 +629,9 @@ func (c *Controller) writeWebhookResponse(ctx *gin.Context, record *store.Record
 			return
 		}
 	}
+	if status >= http.StatusInternalServerError {
+		c.logWebhookFailure(record.ID, status, webhookOriginPluginResponse, nil)
+	}
 	for k, v := range resp.Headers {
 		if http.CanonicalHeaderKey(k) == "Set-Cookie" {
 			continue

apps/backend/internal/plugins/handlers_webhook_lifecycle_test.go

@@ -8,12 +8,17 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"runtime"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+	"go.uber.org/zap/zaptest/observer"
 
+	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/plugins/pkgtar/pkgtartest"
 	"github.com/kandev/kandev/pkg/pluginsdk"
 )
@@ -25,13 +30,18 @@ type blockingWebhookInvoker struct {
 }
 
 type recordingWebhookInvoker struct {
-	calls int
+	calls    int
+	response *pluginsdk.WebhookResponse
+	err      error
 }
 
 func (i *recordingWebhookInvoker) InvokeWebhook(
 	_ context.Context, _ string, _ *pluginsdk.WebhookRequest,
 ) (*pluginsdk.WebhookResponse, error) {
 	i.calls++
+	if i.response != nil || i.err != nil {
+		return i.response, i.err
+	}
 	return &pluginsdk.WebhookResponse{Status: http.StatusOK}, nil
 }
 
@@ -111,6 +121,155 @@ runtime:
 	return &buf
 }
 
+func observedWebhookLogger(t *testing.T) (*logger.Logger, *observer.ObservedLogs) {
+	t.Helper()
+	core, logs := observer.New(zapcore.DebugLevel)
+	zapLogger := zap.New(core)
+	wrapped, err := logger.NewFromZap(zapLogger)
+	if err != nil {
+		t.Fatalf("NewFromZap: %v", err)
+	}
+	t.Cleanup(func() { _ = zapLogger.Sync() })
+	return wrapped, logs
+}
+
+func TestWebhookFailureOriginLogsSafeFields(t *testing.T) {
+	for _, test := range []struct {
+		name       string
+		origin     string
+		class      string
+		response   *pluginsdk.WebhookResponse
+		invokeErr  error
+		disable    bool
+		wantStatus int
+		wantBody   string
+	}{
+		{
+			name:       "host lifecycle",
+			origin:     "host_lifecycle",
+			class:      "host_error",
+			disable:    true,
+			wantStatus: http.StatusServiceUnavailable,
+			wantBody:   `"error":"plugins: plugin \"kandev-plugin-lifecycle-log\" lifecycle generation is no longer active"`,
+		},
+		{
+			name:       "host RPC",
+			origin:     "host_rpc",
+			class:      "host_error",
+			invokeErr:  fmt.Errorf("rpc failed with rpc-secret-sentinel"),
+			wantStatus: http.StatusServiceUnavailable,
+			wantBody:   `"error":"rpc failed with rpc-secret-sentinel"`,
+		},
+		{
+			name:       "plugin response",
+			origin:     "plugin_response",
+			response:   &pluginsdk.WebhookResponse{Status: http.StatusServiceUnavailable, Body: []byte("plugin-response-secret-sentinel")},
+			wantStatus: http.StatusServiceUnavailable,
+			wantBody:   "plugin-response-secret-sentinel",
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			const id = "kandev-plugin-lifecycle-log"
+			svc, _, _ := newTestService(t)
+			if _, err := svc.Install(t.Context(), lifecycleWebhookPackage(t, id, "1.0.0", "public", false)); err != nil {
+				t.Fatalf("Install: %v", err)
+			}
+			if test.disable {
+				if err := svc.Disable(id); err != nil {
+					t.Fatalf("Disable: %v", err)
+				}
+			}
+
+			log, logs := observedWebhookLogger(t)
+			invoker := &recordingWebhookInvoker{response: test.response, err: test.invokeErr}
+			ctrl := &Controller{svc: svc, log: log, webhookInvoker: invoker}
+			gin.SetMode(gin.TestMode)
+			router := gin.New()
+			router.POST("/api/plugins/:id/webhooks/:key", ctrl.webhook)
+
+			requestBody := "request-body-secret-sentinel"
+			request := httptest.NewRequest(http.MethodPost,
+				"/api/plugins/"+id+"/webhooks/callback?token=query-secret-sentinel",
+				strings.NewReader(requestBody),
+			)
+			request.Header.Set("X-Request-Secret", "header-secret-sentinel")
+			recorder := httptest.NewRecorder()
+			router.ServeHTTP(recorder, request)
+			if recorder.Code != test.wantStatus {
+				t.Fatalf("status = %d, want %d, body=%s", recorder.Code, test.wantStatus, recorder.Body.String())
+			}
+			if !strings.Contains(recorder.Body.String(), test.wantBody) {
+				t.Fatalf("body = %q, want it to contain %q", recorder.Body.String(), test.wantBody)
+			}
+			if test.disable && invoker.calls != 0 {
+				t.Fatalf("invoker calls = %d, want 0 after lifecycle lease failure", invoker.calls)
+			}
+			if !test.disable && invoker.calls != 1 {
+				t.Fatalf("invoker calls = %d, want 1", invoker.calls)
+			}
+
+			entries := logs.FilterMessage("plugin webhook failed").All()
+			if len(entries) != 1 {
+				t.Fatalf("webhook failure logs = %d, want 1", len(entries))
+			}
+			fields := entries[0].ContextMap()
+			if got := fields["plugin_id"]; got != id {
+				t.Fatalf("plugin_id = %v, want %q", got, id)
+			}
+			if got := fmt.Sprint(fields["status"]); got != fmt.Sprint(test.wantStatus) {
+				t.Fatalf("status field = %v, want %d", got, test.wantStatus)
+			}
+			if got := fields["origin"]; got != test.origin {
+				t.Fatalf("origin = %v, want %q", got, test.origin)
+			}
+			if test.class == "" {
+				if _, ok := fields["error_class"]; ok {
+					t.Fatalf("plugin response was labeled as a host error: %#v", fields["error_class"])
+				}
+			} else if got := fields["error_class"]; got != test.class {
+				t.Fatalf("error_class = %v, want %q", got, test.class)
+			}
+			logged := fmt.Sprint(fields)
+			for _, secret := range []string{
+				"request-body-secret-sentinel", "query-secret-sentinel", "header-secret-sentinel",
+				"rpc-secret-sentinel", "plugin-response-secret-sentinel",
+			} {
+				if strings.Contains(logged, secret) {
+					t.Fatalf("webhook diagnostic exposed %q: %s", secret, logged)
+				}
+			}
+			if _, ok := fields["error"]; ok {
+				t.Fatalf("webhook diagnostic contained a raw error field: %s", logged)
+			}
+		})
+	}
+}
+
+func TestWebhookFailureLogsWaitForAuthorizationAndDeclaration(t *testing.T) {
+	const id = "kandev-plugin-private-log"
+	svc, _, _ := newTestService(t)
+	if _, err := svc.Install(t.Context(), lifecycleWebhookPackage(t, id, "1.0.0", "authenticated", false)); err != nil {
+		t.Fatalf("Install: %v", err)
+	}
+	log, logs := observedWebhookLogger(t)
+	ctrl := &Controller{svc: svc, log: log, webhookInvoker: &recordingWebhookInvoker{}}
+	gin.SetMode(gin.TestMode)
+	router := gin.New()
+	router.POST("/api/plugins/:id/webhooks/:key", ctrl.webhook)
+
+	unauthorized := doRequest(router, http.MethodPost, "/api/plugins/"+id+"/webhooks/callback", "{}", nil)
+	if unauthorized.Code != http.StatusUnauthorized {
+		t.Fatalf("unauthorized status = %d, want 401", unauthorized.Code)
+	}
+	undeclared := doAuthedRequest(router, http.MethodPost, "/api/plugins/"+id+"/webhooks/unknown", "{}", nil)
+	if undeclared.Code != http.StatusNotFound {
+		t.Fatalf("undeclared status = %d, want 404", undeclared.Code)
+	}
+	if entries := logs.FilterMessage("plugin webhook failed").All(); len(entries) != 0 {
+		t.Fatalf("pre-dispatch requests emitted webhook failure diagnostics: %+v", entries)
+	}
+}
+
 func TestWebhookDispatchLeaseCoversRPCAndAuthLoginResponse(t *testing.T) {
 	const id = "kandev-plugin-generation"
 	svc, _, _ := newTestService(t)

apps/backend/internal/task/statussummary/projector.go

@@ -384,6 +384,13 @@ func (p *Projector) handleEvent(ctx context.Context, event *bus.Event) error {
 
 	state, err := p.ensureState(ctx, taskID)
 	if err != nil {
+		if event.Type == events.MessageQueueStatusChanged && isMissingTaskLookupErr(err) {
+			p.dropProjectionState(taskID)
+			p.logger.Debug("skipping queue status projection for missing task",
+				zap.String("task_id", taskID),
+				zap.Error(err))
+			return nil
+		}
 		return err
 	}
 	if workspaceID := stringField(data, "workspace_id"); workspaceID != "" {
@@ -394,7 +401,7 @@ func (p *Projector) handleEvent(ctx context.Context, event *bus.Event) error {
 		if err != nil {
 			// Lifecycle purge publishes queue-status after DeleteTask commits.
 			// Only skip verified not-found; transient resolve errors must surface.
-			if event.Type == events.MessageQueueStatusChanged && isMissingTaskResolveErr(err) {
+			if event.Type == events.MessageQueueStatusChanged && isMissingTaskLookupErr(err) {
 				p.dropProjectionState(taskID)
 				p.logger.Debug("skipping queue status projection for missing task",
 					zap.String("task_id", taskID),
@@ -434,6 +441,13 @@ func (p *Projector) handleEvent(ctx context.Context, event *bus.Event) error {
 	if p.loadLaunchQueue != nil {
 		nextQueue, loadErr := p.loadLaunchQueue(ctx, taskID)
 		if loadErr != nil {
+			if event.Type == events.MessageQueueStatusChanged && isMissingTaskLookupErr(loadErr) {
+				p.dropProjectionState(taskID)
+				p.logger.Debug("skipping queue status projection for missing task",
+					zap.String("task_id", taskID),
+					zap.Error(loadErr))
+				return nil
+			}
 			return fmt.Errorf("load launch queue for task status summary %q: %w", taskID, loadErr)
 		}
 		launchQueueChanged = !state.launchQueueObserved || !equalLaunchQueue(state.launchQueue, nextQueue)

apps/backend/internal/task/statussummary/projector_derive.go

@@ -1,9 +1,11 @@
 package statussummary
 
 import (
+	"errors"
 	"strings"
 
 	"github.com/kandev/kandev/internal/task/models"
+	"github.com/kandev/kandev/internal/task/repository/repoerrors"
 )
 
 func deriveSummary(state *projectionState) TaskStatusSummary {
@@ -225,14 +227,8 @@ func isGoneTaskPersistErr(err error) bool {
 		strings.Contains(msg, "violates foreign key")
 }
 
-// isMissingTaskResolveErr reports whether workspace resolution failed because
-// the task no longer exists. Transient repository errors must not match.
-func isMissingTaskResolveErr(err error) bool {
-	if err == nil {
-		return false
-	}
-	msg := strings.ToLower(err.Error())
-	return strings.Contains(msg, "not found") ||
-		strings.Contains(msg, "no rows") ||
-		strings.Contains(msg, "sql: no rows in result set")
+// isMissingTaskLookupErr reports whether a task lookup returned the repository's
+// authoritative missing-task sentinel. Transient repository errors must not match.
+func isMissingTaskLookupErr(err error) bool {
+	return errors.Is(err, repoerrors.ErrTaskNotFound)
 }

apps/backend/internal/task/statussummary/projector_queued_test.go

@@ -2,6 +2,7 @@ package statussummary
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync/atomic"
 	"testing"
@@ -10,6 +11,7 @@ import (
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/events"
 	"github.com/kandev/kandev/internal/events/bus"
+	"github.com/kandev/kandev/internal/task/repository/repoerrors"
 )
 
 func TestProjectorQueueEventUpdatesQueuedPromptCount(t *testing.T) {
@@ -273,7 +275,7 @@ func TestProjectorQueueEventForMissingTaskIsNoop(t *testing.T) {
 		// Simulate DeleteTask: the task row is already gone when purge
 		// publishes message.queue.status_changed.
 		ResolveWorkspace: func(context.Context, string) (string, error) {
-			return "", fmt.Errorf("task %q not found", taskID)
+			return "", fmt.Errorf("%w: %s", repoerrors.ErrTaskNotFound, taskID)
 		},
 		CountQueuedPrompts: func(context.Context, string) (int, error) {
 			counterCalls.Add(1)
@@ -319,6 +321,128 @@ func TestProjectorQueueEventForMissingTaskIsNoop(t *testing.T) {
 	}
 }
 
+// @covers AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11
+func TestProjectorQueueEventForColdMissingTaskWithProductionLoadersIsNoop(t *testing.T) {
+	const taskID = "task-cold-deleted-queue"
+	store := newProjectorTestStore()
+	eventBus := bus.NewMemoryEventBus(logger.Default())
+	updates := new(atomic.Int64)
+	if _, err := eventBus.Subscribe(events.TaskStatusSummaryUpdated, func(_ context.Context, _ *bus.Event) error {
+		updates.Add(1)
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	defer eventBus.Close()
+
+	projector := NewProjector(ProjectorConfig{
+		Store:    store,
+		EventBus: eventBus,
+		ResolveWorkspace: func(context.Context, string) (string, error) {
+			return "", fmt.Errorf("%w: %s", repoerrors.ErrTaskNotFound, taskID)
+		},
+		LoadLaunchQueue: func(context.Context, string) (*LaunchQueueSummary, error) {
+			return nil, fmt.Errorf("%w: %s", repoerrors.ErrTaskNotFound, taskID)
+		},
+	})
+
+	err := projector.handleEvent(context.Background(), bus.NewEvent(events.MessageQueueStatusChanged, "test", map[string]interface{}{
+		"task_id": taskID,
+	}))
+	if err != nil {
+		t.Fatalf("queue status for a cold deleted task returned error: %v", err)
+	}
+	if got := updates.Load(); got != 0 {
+		t.Fatalf("summary publishes = %d, want 0", got)
+	}
+	if len(store.rows) != 0 {
+		t.Fatalf("cold missing-task event wrote summaries: %d rows", len(store.rows))
+	}
+	projector.mu.Lock()
+	_, retained := projector.state[taskID]
+	projector.mu.Unlock()
+	if retained {
+		t.Fatal("cold missing-task event retained projection state")
+	}
+}
+
+// @covers AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11
+func TestProjectorQueueEventForMissingTaskWithCachedWorkspaceIsNoop(t *testing.T) {
+	const taskID = "task-cached-deleted-queue"
+	store := newProjectorTestStore()
+	eventBus := bus.NewMemoryEventBus(logger.Default())
+	updates := new(atomic.Int64)
+	if _, err := eventBus.Subscribe(events.TaskStatusSummaryUpdated, func(_ context.Context, _ *bus.Event) error {
+		updates.Add(1)
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	ctx := context.Background()
+	projector := NewProjector(ProjectorConfig{
+		Store:    store,
+		EventBus: eventBus,
+		ResolveWorkspace: func(context.Context, string) (string, error) {
+			t.Fatal("cached workspace should skip task lookup")
+			return "", nil
+		},
+		LoadLaunchQueue: func(context.Context, string) (*LaunchQueueSummary, error) {
+			return nil, fmt.Errorf("%w: %s", repoerrors.ErrTaskNotFound, taskID)
+		},
+		CountQueuedPrompts: func(context.Context, string) (int, error) {
+			return 0, nil
+		},
+	})
+	t.Cleanup(func() { eventBus.Close() })
+
+	projector.mu.Lock()
+	projector.state[taskID] = &projectionState{workspaceID: "workspace-1"}
+	projector.mu.Unlock()
+
+	err := projector.handleEvent(ctx, bus.NewEvent(events.MessageQueueStatusChanged, "test", map[string]interface{}{
+		"task_id": taskID,
+	}))
+	if err != nil {
+		t.Fatalf("queue status for deleted task with cached workspace returned error: %v", err)
+	}
+	if got := updates.Load(); got != 0 {
+		t.Fatalf("summary publishes = %d, want 0", got)
+	}
+	projector.mu.Lock()
+	_, retained := projector.state[taskID]
+	projector.mu.Unlock()
+	if retained {
+		t.Fatal("deleted task queue event retained cached projection state")
+	}
+}
+
+// @covers AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11
+func TestProjectorQueueEventPropagatesTransientLaunchQueueFailureContainingNotFoundText(t *testing.T) {
+	const taskID = "task-queue-loader-transient"
+	queryErr := errors.New("driver query failed: task not found in diagnostic context")
+	projector := NewProjector(ProjectorConfig{
+		LoadLaunchQueue: func(context.Context, string) (*LaunchQueueSummary, error) {
+			return nil, queryErr
+		},
+	})
+	projector.mu.Lock()
+	projector.state[taskID] = &projectionState{workspaceID: "workspace-1"}
+	projector.mu.Unlock()
+
+	err := projector.handleEvent(context.Background(), bus.NewEvent(events.MessageQueueStatusChanged, "test", map[string]interface{}{
+		"task_id": taskID,
+	}))
+	if !errors.Is(err, queryErr) {
+		t.Fatalf("launch queue error = %v, want original transient error", err)
+	}
+	projector.mu.Lock()
+	_, retained := projector.state[taskID]
+	projector.mu.Unlock()
+	if !retained {
+		t.Fatal("transient launch queue failure dropped cached projection state")
+	}
+}
+
 func TestProjectorQueueEventPropagatesTransientResolveFailure(t *testing.T) {
 	const taskID = "task-resolve-transient"
 	store := newProjectorTestStore()

apps/web/e2e/tests/chat/busy-signal.spec.ts

@@ -177,7 +177,9 @@ test.describe.serial("Claude background prompt handoff experiment", () => {
     await waitForActiveSessionForegroundActivity(testPage, "background");
 
     await session.sendMessage("/slow 2s");
-    await expect(testPage.getByText("/slow 2s")).toBeVisible({ timeout: 15_000 });
+    await expect(
+      session.activeChat().getByTestId("user-message-bubble").filter({ hasText: "/slow 2s" }),
+    ).toBeVisible({ timeout: 15_000 });
     await expect(testPage.getByTestId("queue-chip")).not.toBeVisible();
     await waitForActiveSessionForegroundActivity(testPage, "generating");
     await expect(session.idleInput()).toBeVisible({ timeout: 15_000 });
@@ -211,7 +213,9 @@ test.describe.serial("Claude background prompt handoff experiment", () => {
     await waitForActiveSessionForegroundActivity(testPage, "background");
 
     await session.sendMessage("/slow 2s");
-    await expect(testPage.getByText("/slow 2s")).toBeVisible({ timeout: 15_000 });
+    await expect(
+      session.activeChat().getByTestId("user-message-bubble").filter({ hasText: "/slow 2s" }),
+    ).toBeVisible({ timeout: 15_000 });
     await expect(testPage.getByTestId("queue-chip")).not.toBeVisible();
   });
 });