From 8796d6948f6f1975aef3e57472f4bf382ecb1a8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 24 Sep 2026 19:06:03 +0100 Subject: [PATCH] fix: handle deleted tasks and attribute runtime failures (#3910) --- .../runtime/lifecycle/manager_lifecycle.go | 65 +++++++ .../manager_recovery_sessions_step_test.go | 140 ++++++++++++++ apps/backend/internal/backendapp/gateway.go | 5 +- .../executor_pr_base_identity_test.go | 183 ++++++++++++++++++ .../orchestrator/executor/executor_resume.go | 85 +++++--- .../persistence/requiredstores/health.go | 142 ++++++++++++-- .../persistence/requiredstores/health_test.go | 176 +++++++++++++++++ apps/backend/internal/plugins/handlers.go | 35 ++++ .../handlers_webhook_lifecycle_test.go | 161 ++++++++++++++- .../internal/task/statussummary/projector.go | 16 +- .../task/statussummary/projector_derive.go | 16 +- .../statussummary/projector_queued_test.go | 126 +++++++++++- apps/web/e2e/tests/chat/busy-signal.spec.ts | 8 +- .../recent-runtime-log-remediation/plan.md | 157 +++++++++++++++ .../task-01-drop-deleted-task-projection.md | 85 ++++++++ .../task-02-attribute-persistence-probes.md | 92 +++++++++ .../task-03-attribute-plugin-webhooks.md | 86 ++++++++ .../task-04-attribute-pr-identity.md | 86 ++++++++ .../task-05-summarize-recovery-outcomes.md | 90 +++++++++ docs/specs/platform/README.md | 2 + .../bounded-task-status-delivery.md | 4 + .../runtime-failure-attribution.md | 47 +++++ .../bounded-task-status-delivery.md | 12 +- .../runtime-failure-attribution.md | 57 ++++++ 24 files changed, 1817 insertions(+), 59 deletions(-) create mode 100644 docs/plans/recent-runtime-log-remediation/plan.md create mode 100644 docs/plans/recent-runtime-log-remediation/task-01-drop-deleted-task-projection.md create mode 100644 docs/plans/recent-runtime-log-remediation/task-02-attribute-persistence-probes.md create mode 100644 docs/plans/recent-runtime-log-remediation/task-03-attribute-plugin-webhooks.md create mode 100644 docs/plans/recent-runtime-log-remediation/task-04-attribute-pr-identity.md create mode 100644 docs/plans/recent-runtime-log-remediation/task-05-summarize-recovery-outcomes.md create mode 100644 docs/specs/platform/requirements/runtime-failure-attribution.md create mode 100644 docs/specs/platform/system-design/runtime-failure-attribution.md diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go b/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go index a2f530d487..157d0aa292 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go +++ b/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 diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_recovery_sessions_step_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_recovery_sessions_step_test.go index b9ce8ea2f6..1c097e5dd1 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_recovery_sessions_step_test.go +++ b/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 diff --git a/apps/backend/internal/backendapp/gateway.go b/apps/backend/internal/backendapp/gateway.go index 7656f1f5f4..143b4c1be8 100644 --- a/apps/backend/internal/backendapp/gateway.go +++ b/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 }, diff --git a/apps/backend/internal/orchestrator/executor/executor_pr_base_identity_test.go b/apps/backend/internal/orchestrator/executor/executor_pr_base_identity_test.go index 9634481089..41084a39fd 100644 --- a/apps/backend/internal/orchestrator/executor/executor_pr_base_identity_test.go +++ b/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 { diff --git a/apps/backend/internal/orchestrator/executor/executor_resume.go b/apps/backend/internal/orchestrator/executor/executor_resume.go index 4cc686854f..5365926d89 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume.go +++ b/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) { diff --git a/apps/backend/internal/persistence/requiredstores/health.go b/apps/backend/internal/persistence/requiredstores/health.go index 5f80bcca9b..2ef3ae643b 100644 --- a/apps/backend/internal/persistence/requiredstores/health.go +++ b/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() } diff --git a/apps/backend/internal/persistence/requiredstores/health_test.go b/apps/backend/internal/persistence/requiredstores/health_test.go index 7bc29cdda6..12e3966d20 100644 --- a/apps/backend/internal/persistence/requiredstores/health_test.go +++ b/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 { diff --git a/apps/backend/internal/plugins/handlers.go b/apps/backend/internal/plugins/handlers.go index 26f2940bca..258df45670 100644 --- a/apps/backend/internal/plugins/handlers.go +++ b/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 diff --git a/apps/backend/internal/plugins/handlers_webhook_lifecycle_test.go b/apps/backend/internal/plugins/handlers_webhook_lifecycle_test.go index 3c020456c9..2aab8f040c 100644 --- a/apps/backend/internal/plugins/handlers_webhook_lifecycle_test.go +++ b/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) diff --git a/apps/backend/internal/task/statussummary/projector.go b/apps/backend/internal/task/statussummary/projector.go index 8b7a603743..b65346d749 100644 --- a/apps/backend/internal/task/statussummary/projector.go +++ b/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) diff --git a/apps/backend/internal/task/statussummary/projector_derive.go b/apps/backend/internal/task/statussummary/projector_derive.go index 142a8c30ec..bbf17d44b5 100644 --- a/apps/backend/internal/task/statussummary/projector_derive.go +++ b/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) } diff --git a/apps/backend/internal/task/statussummary/projector_queued_test.go b/apps/backend/internal/task/statussummary/projector_queued_test.go index 10cc44576f..131cfba7e7 100644 --- a/apps/backend/internal/task/statussummary/projector_queued_test.go +++ b/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() diff --git a/apps/web/e2e/tests/chat/busy-signal.spec.ts b/apps/web/e2e/tests/chat/busy-signal.spec.ts index 025ed1ed20..8d74bc6f71 100644 --- a/apps/web/e2e/tests/chat/busy-signal.spec.ts +++ b/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(); }); }); diff --git a/docs/plans/recent-runtime-log-remediation/plan.md b/docs/plans/recent-runtime-log-remediation/plan.md new file mode 100644 index 0000000000..fc49837bfc --- /dev/null +++ b/docs/plans/recent-runtime-log-remediation/plan.md @@ -0,0 +1,157 @@ +--- +created: 2026-09-24 +status: implemented +requirements: + - REQ-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001 + - REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001 + - REQ-PLATFORM-POSTGRES-DOMAIN-STORE-PARITY-007 + - REQ-PLUGINS-PLUGINS-001 + - REQ-TASKS-TASK-LAUNCH-FAILURE-RECOVERY-001 + - REQ-PLATFORM-STARTUP-PROGRESS-001 +system_design: + - ../../specs/platform/system-design/bounded-task-status-delivery.md + - ../../specs/platform/system-design/runtime-failure-attribution.md + - ../../specs/platform/system-design/postgres-domain-store-parity.md + - ../../specs/plugins/system-design/plugins-02.md + - ../../specs/tasks/system-design/task-launch-failure-recovery.md + - ../../specs/platform/system-design/startup-progress-visibility.md +legacy_specs: [] +--- + +# Implementation Plan: Recent Runtime Log Remediation + +## Overview + +Fix the reproducible late queue-status event after task deletion. Add bounded +diagnostics at four other failure boundaries so a future occurrence identifies +the cause without changing established safety policy. The September 24 logs +showed a 13-second persistence outage, six earlier webhook 503s, one PR +identity rejection, and startup recovery at 0/39. They do not prove the +underlying cause of the first three. The 39 startup records were attached to +cancelled, failed, or missing sessions, so zero retracked is not evidence of a +lost live agent. Implement in the order below; each work order passes alone. + +## Scope + +### In scope + +- Drop cached task summary state on verified deletion after a queue-status event. +- Attribute failed persistence probes, plugin webhook 5xx responses, PR identity + mismatches, and startup non-recovery with safe structured diagnostics. +- Keep the current readiness, webhook response, PR identity, and recovery + decisions unchanged; use captured evidence to decide any later cause-specific fix. + +### Out of scope + +- Securing the live listener, per the user's explicit exclusion. +- Raising the two-second database probe timeout, bypassing fail-closed health, + suppressing the 0/39 warning, deleting unknown-liveness records, or relaxing + PR identity checks without evidence and a separate reviewed contract change. +- Changing plugin-owned manifest declarations in this monorepo. Installed + plugin packages live in dedicated repositories; any needed release is a + separate follow-up after webhook origin is known. +- Changing intentional `KANDEV_LOG_LEVEL=debug` volume or the isolated cancel + escalation that settled successfully. + +## Technical approach + +1. In `internal/task/statussummary/Projector.handleEvent`, handle verified + missing-task from `LoadLaunchQueue` for queue-status events as the same + terminal condition already handled at workspace resolution. Discard cached + state under the task lock and publish nothing. Preserve transient errors. +2. In `internal/persistence/requiredstores/Health`, measure the existing probe + stages and capture writer/reader `sql.DB.Stats` at failure. Emit one bounded + warning per failed periodic check. Keep the current 15-second schedule, + two-second timeout, maintenance lease, and all-store health transition. +3. In `internal/plugins/Controller.webhook`, classify 5xx outcomes at the host + dispatch lease, RPC, and plugin-response boundaries. Do not log body, key, + query, headers, or error strings that may contain remote data. +4. In `internal/orchestrator/executor`, return a fixed mismatch reason alongside + the current PR identity decision, then log it at the rejection boundary. + Keep exact existing bound/unbound and fork rules and launch behavior. +5. In `internal/agent/runtime/lifecycle`, report one aggregate recovery outcome + after the pass. Classify only proven reasons; treat missing backend returns + as unknown. Preserve the startup step's existing done/total and warning. + +After deployment, correlate a repeated persistence failure with pool wait and +stage data before proposing a contention or host-IO fix. Correlate a webhook +503 with its origin before changing host code or an external plugin. Reassess +the PR association only when a mismatch reason recurs with trustworthy task +binding evidence. These are evidence gates, not unimplemented promises in this +package. + +## Tests + +| Acceptance criterion | Evidence | +| --- | --- | +| `AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11` | `projector_queued_test.go`: cold and cached deleted tasks plus a transient queue-loader error; no publication or retained state on deletion, while transient failures propagate. | +| `AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.1` | `requiredstores/health_test.go`: blocked writer or reader, stage/elapsed/pool fields, one warning, unchanged health recovery. | +| `AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.2` | `plugins/handlers_webhook_lifecycle_test.go`: host lease failure, RPC error, and plugin-supplied 503 retain response semantics and log only safe fields. | +| `AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.3` | `executor_pr_base_identity_test.go`: all mismatch reasons and valid fork cases preserve the current decision. | +| `AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.4` | `manager_recovery_sessions_step_test.go`: 0/N stale/unknown outcomes, partial recovery, and progress warning semantics. | + +## Work orders + +- [x] [Task 01: Drop deleted task projection state](task-01-drop-deleted-task-projection.md) +- [x] [Task 02: Attribute persistence probe failures](task-02-attribute-persistence-probes.md) +- [x] [Task 03: Attribute plugin webhook failures](task-03-attribute-plugin-webhooks.md) +- [x] [Task 04: Attribute PR identity rejection](task-04-attribute-pr-identity.md) +- [x] [Task 05: Summarize startup recovery outcomes](task-05-summarize-recovery-outcomes.md) + +## Verification results + +All five work orders are implemented, and each focused check passes: + +- Task 01: `go test -tags fts5 ./internal/task/statussummary -run 'TestProjectorQueueEvent' -count=1` +- Task 02: `go test -tags fts5 ./internal/persistence/requiredstores -run 'Test(RuntimeHealth|StartupHealth|HealthCheck|ProbeTables)' -count=1` +- Task 03: `go test -tags fts5 ./internal/plugins -run 'TestWebhook' -count=1` +- Task 04: `go test -tags fts5 ./internal/orchestrator/executor -run 'Test.*PRBaseIdentity|Test.*PRBase' -count=1` +- Task 05: `go test -tags fts5 ./internal/agent/runtime/lifecycle -run 'Test.*Recovery' -count=1` and `go test -tags fts5 ./internal/startup -run 'Test.*Step' -count=1` + +The complete affected-package suite also passes: +`go test -tags fts5 ./internal/task/statussummary ./internal/persistence/requiredstores ./internal/plugins ./internal/orchestrator/executor ./internal/agent/runtime/lifecycle`. +`make -C apps/backend build` passes. `git diff --check` is clean. + +Review follow-up tightened deletion detection to `errors.Is` with the typed +task-not-found sentinel, updated gateway nil-task fallbacks, and added the +`invalid_resolved_pr_base` PR diagnostic. Verification after these changes: +`go test -tags fts5 ./internal/task/statussummary ./internal/orchestrator/executor ./internal/backendapp` +and `make -C apps/backend build` both pass; `git diff --check` remains clean. + +PR review follow-up added a cold projection regression for a deleted task and +kept the persistence-loop test alive until its first failure warning, then +canceled it deterministically. `Service.Get` uses the in-memory registry and +returns only a record or `store.ErrNotFound`, so the proposed webhook lookup +500 path cannot occur and requires no code change. Verification after these +follow-ups: + +- `go test -tags fts5 ./internal/task/statussummary -run 'TestProjectorQueueEvent' -count=1` +- `go test -tags fts5 ./internal/persistence/requiredstores -run 'TestRuntimeHealthProbeFailureLogsBoundedStageAndRecovers' -count=3` +- `go test -tags fts5 ./internal/task/statussummary ./internal/persistence/requiredstores ./internal/plugins ./internal/orchestrator/executor ./internal/agent/runtime/lifecycle ./internal/startup ./internal/backendapp` +- `make -C apps/backend build` + +The focused and complete affected-package checks passed. The backend build +passed; Darwin helper binaries were left unsigned because neither `codesign` +nor `rcodesign` is installed in this environment. `list-docs.py validate`, +`lint-spec-files.py --all`, and `git diff --check` also passed. + +PR E2E follow-up: shard 6/14 was canceled at its 35-minute job limit, which +caused the report-merge and aggregate E2E checks to fail. Replaying the exact +PR merge checkout and shard manifest exposed an intermittent strict-mode +failure: a page-wide `/slow 2s` locator matched both the sent bubble and the +composer draft. Both assertions now target the active chat's user-message +bubble. The two-test serial group passed in the CI runtime image with its +retry override temporarily disabled and Playwright retries set to zero. The +full local replay was stopped after 41 of 255 tests, so it is not recorded as +a full-shard pass. The PR workflow completed all 14 normal E2E shards and all +6 container E2E shards, merged the reports, and passed the aggregate E2E gate +on head `725b7d5dbf0f4e1932e859f5ae7c8e5835a32aca` (run `36022650892`). + +## Risks + +- A single pool snapshot cannot prove the cause of a timeout; compare repeated + stage and wait measurements before changing database policy. +- Plugin-generated 503s may require a change in a dedicated plugin repository. +- Recovery inventory rows with unknown liveness may still be live; summary + generation must never become a cleanup trigger. +- Diagnostic fields must remain bounded and exclude secrets and request data. diff --git a/docs/plans/recent-runtime-log-remediation/task-01-drop-deleted-task-projection.md b/docs/plans/recent-runtime-log-remediation/task-01-drop-deleted-task-projection.md new file mode 100644 index 0000000000..8d84e9fda1 --- /dev/null +++ b/docs/plans/recent-runtime-log-remediation/task-01-drop-deleted-task-projection.md @@ -0,0 +1,85 @@ +--- +id: "01-drop-deleted-task-projection" +title: "Drop deleted task projection state" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001 +acceptance_criteria: + - AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11 +system_design: + - ../../specs/platform/system-design/bounded-task-status-delivery.md +--- + +# Task 01: Drop deleted task projection state + +## Summary + +A late queue-status event can reach a projector with a cached workspace after +task deletion, or arrive first while the projection is cold. Treat +authoritative not-found during rehydration or from the launch-queue loader as +deletion and drop the cached state without publishing an update. + +## In scope + +- Add failing warm-cache and cold-rehydration regressions, then handle verified + not-found from `ensureState` or `LoadLaunchQueue` in `Projector.handleEvent` + using the repository's typed `ErrTaskNotFound` sentinel. +- Keep transient loader failures observable as errors. +- Return the same typed sentinel from gateway task lookup fallbacks that receive + a nil task with no error. + +## Out of scope + +- Changing task deletion order or generic projection error handling. + +## Acceptance + +- Cold and cached deleted-task queue events return nil, publish nothing, and + retain no state. +- Wrapped `ErrTaskNotFound` is treated as deletion; a different error whose + text contains "not found" still propagates and retains cached state. + +## Verification + +```bash +(cd apps/backend && go test -tags fts5 ./internal/task/statussummary -run 'TestProjectorQueueEvent' -count=1) +``` + +## Files likely touched + +- `apps/backend/internal/task/statussummary/projector.go` +- `apps/backend/internal/task/statussummary/projector_queued_test.go` + +## Dependencies + +None. + +## Risks + +- Text-based not-found checks can suppress transient failures. Use only + `errors.Is(err, repoerrors.ErrTaskNotFound)` for the deletion decision. + +## Parallelism + +`sequential` + +## Inputs + +- `AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11` and its system design. +- Existing cold-task and warm-projector regressions in `projector_queued_test.go`. + +## Results + +Implemented the cached-workspace deletion guard with typed sentinel matching, +and made gateway nil-task fallbacks return that sentinel. A cold-state test +with both production-shaped task loaders reproduced the rehydration error; +queue-status events now treat the wrapped sentinel from `ensureState` as +deletion. Wrapped sentinel errors are ignored as deletion while a driver error +containing "task not found" propagates and retains state. Verification passed: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/task/statussummary -run 'TestProjectorQueueEvent' -count=1) +``` diff --git a/docs/plans/recent-runtime-log-remediation/task-02-attribute-persistence-probes.md b/docs/plans/recent-runtime-log-remediation/task-02-attribute-persistence-probes.md new file mode 100644 index 0000000000..32fe4c57f6 --- /dev/null +++ b/docs/plans/recent-runtime-log-remediation/task-02-attribute-persistence-probes.md @@ -0,0 +1,92 @@ +--- +id: "02-attribute-persistence-probes" +title: "Attribute persistence probe failures" +status: done +wave: 2 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001 + - REQ-PLATFORM-POSTGRES-DOMAIN-STORE-PARITY-007 +acceptance_criteria: + - AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.1 + - AC-PLATFORM-POSTGRES-DOMAIN-STORE-PARITY-007.3 + - AC-PLATFORM-POSTGRES-DOMAIN-STORE-PARITY-007.4 +system_design: + - ../../specs/platform/system-design/runtime-failure-attribution.md + - ../../specs/platform/system-design/postgres-domain-store-parity.md +--- + +# Task 02: Attribute persistence probe failures + +## Summary + +The September 24 probe timed out at the writer and made stateful calls +temporarily unavailable. Add one safe failure diagnostic that distinguishes a +pool wait from other probe stages without changing the health policy. + +## In scope + +- Measure writer ping, reader ping, and table probe stages in `Health`. +- Capture bounded writer and reader pool statistics at a failed probe. +- Test timeout and subsequent recovery with unchanged readiness behavior. + +## Out of scope + +- Changing probe deadline, pool sizes, maintenance coordination, or 503 policy. + +## Acceptance + +- One failed sweep emits one stage, elapsed time, class, and pool snapshot. +- No SQL, DSN, or raw database error containing sensitive data enters new fields. +- Recovery continues on the next successful probe without restart. + +## Verification + +```bash +(cd apps/backend && go test -tags fts5 ./internal/persistence/requiredstores -run 'Test(RuntimeHealth|StartupHealth|HealthCheck|ProbeTables)' -count=1) +``` + +## Files likely touched + +- `apps/backend/internal/persistence/requiredstores/health.go` +- `apps/backend/internal/persistence/requiredstores/health_test.go` + +## Dependencies + +None. + +## Risks + +- Pool statistics are a snapshot, not a causal trace; do not infer a fix from + one event. + +## Parallelism + +`sequential` + +## Inputs + +- Runtime-failure-attribution and required-store health designs. +- Required-internal-persistence and maintenance-health-probe ADRs. + +## Results + +The periodic health warning now records the failed stage, elapsed time, a +fixed error class, and writer/reader pool snapshots without the raw database +error. Writer and reader timeouts emit one warning for a sweep, and the next +successful check restores health. Table-probe failures carry their store ID. +The periodic-loop test now waits for its first failure warning before +cancelling; the loop checks cancellation before starting another probe when a +tick is already queued. Verification passed: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/persistence/requiredstores -run 'Test(RuntimeHealth|StartupHealth|HealthCheck|ProbeTables)' -count=1) +``` + +The synchronized writer/reader warning regression also passed three consecutive +runs: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/persistence/requiredstores -run 'TestRuntimeHealthProbeFailureLogsBoundedStageAndRecovers' -count=3) +``` diff --git a/docs/plans/recent-runtime-log-remediation/task-03-attribute-plugin-webhooks.md b/docs/plans/recent-runtime-log-remediation/task-03-attribute-plugin-webhooks.md new file mode 100644 index 0000000000..8d08665a4b --- /dev/null +++ b/docs/plans/recent-runtime-log-remediation/task-03-attribute-plugin-webhooks.md @@ -0,0 +1,86 @@ +--- +id: "03-attribute-plugin-webhooks" +title: "Attribute plugin webhook failures" +status: done +wave: 3 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001 + - REQ-PLUGINS-PLUGINS-001 +acceptance_criteria: + - AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.2 + - AC-PLUGINS-PLUGINS-001.4 +system_design: + - ../../specs/platform/system-design/runtime-failure-attribution.md + - ../../specs/plugins/system-design/plugins-02.md +--- + +# Task 03: Attribute plugin webhook failures + +## Summary + +Earlier plugin webhook 503s cannot be assigned to the host or a plugin from +access logs. Classify each 5xx at the dispatch, RPC, or plugin-response +boundary while preserving the exact HTTP response. + +## In scope + +- Add safe structured origin and status logging in `Controller.webhook`. +- Cover host lease/RPC failure and plugin-supplied 503 with tests. + +## Out of scope + +- Changing webhook authentication, plugin lifecycle, or external manifests. + +## Acceptance + +- Logs identify origin and plugin ID for every authenticated/declaration-passed 5xx. +- Webhook key, body, query, headers, and credentials are absent from new logs. +- Response status and body remain unchanged. + +## Verification + +```bash +(cd apps/backend && go test -tags fts5 ./internal/plugins -run 'TestWebhook' -count=1) +``` + +PR review follow-up confirmed that `Service.Get` reads the in-memory plugin +registry and returns only the record or `store.ErrNotFound`; the webhook maps +that not-found result to 404 after authorization. The proposed lookup-driven +500 branch is unreachable in the current implementation, so no code change was +needed. + +## Files likely touched + +- `apps/backend/internal/plugins/handlers.go` +- `apps/backend/internal/plugins/handlers_webhook_lifecycle_test.go` + +## Dependencies + +None. + +## Risks + +- Logging before the authorization and declaration gates could reveal an + installed plugin ID; emit only after those gates. + +## Parallelism + +`sequential` + +## Inputs + +- Runtime-failure-attribution design and plugin webhook proxy contract. +- Existing webhook lifecycle and authentication tests. + +## Results + +Webhook 5xx outcomes now report plugin ID, status, and a fixed origin after the +authorization and declaration gates. Host lifecycle/RPC failures use bounded +error classes; plugin responses carry no host error class. Status and body +semantics are unchanged. Verification passed: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/plugins -run 'TestWebhook' -count=1) +``` diff --git a/docs/plans/recent-runtime-log-remediation/task-04-attribute-pr-identity.md b/docs/plans/recent-runtime-log-remediation/task-04-attribute-pr-identity.md new file mode 100644 index 0000000000..5216534a18 --- /dev/null +++ b/docs/plans/recent-runtime-log-remediation/task-04-attribute-pr-identity.md @@ -0,0 +1,86 @@ +--- +id: "04-attribute-pr-identity" +title: "Attribute PR identity rejection" +status: done +wave: 4 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001 + - REQ-TASKS-TASK-LAUNCH-FAILURE-RECOVERY-001 +acceptance_criteria: + - AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.3 + - AC-TASKS-TASK-LAUNCH-FAILURE-RECOVERY-001.2 +system_design: + - ../../specs/platform/system-design/runtime-failure-attribution.md + - ../../specs/tasks/system-design/task-launch-failure-recovery.md +--- + +# Task 04: Attribute PR identity rejection + +## Summary + +One task launch failed at the PR identity guard with no field-specific reason. +Return a fixed reason from the comparison and record it when the current +failure or safe fallback path is taken. + +## In scope + +- Refactor `validPRBaseIdentity` and its bound/unbound helpers to expose a + bounded reason while preserving their exact acceptance matrix. +- Emit a fixed `invalid_resolved_pr_base` reason when resolver data fails + structural validation, while preserving the existing block or safe fallback. +- Test attached and fork PRs, valid cases, and every mismatch branch. + +## Out of scope + +- Relaxing repository binding or changing launch error text and actions. + +## Acceptance + +- Each mismatch emits a fixed reason with task-repository ID and PR number. +- Invalid resolver output emits a fixed reason for both unbound and bound + targets without logging provider data. +- Valid fork PRs still launch; invalid identities still fail closed. +- Logs contain no repository URL, credential, or raw provider payload. + +## Verification + +```bash +(cd apps/backend && go test -tags fts5 ./internal/orchestrator/executor -run 'Test.*PRBaseIdentity|Test.*PRBase' -count=1) +``` + +## Files likely touched + +- `apps/backend/internal/orchestrator/executor/executor_resume.go` +- `apps/backend/internal/orchestrator/executor/executor_pr_base_identity_test.go` + +## Dependencies + +None. + +## Risks + +- A reason-bearing refactor must preserve short-circuit behavior and both + expected-target and unbound association rules. + +## Parallelism + +`sequential` + +## Inputs + +- Runtime-failure-attribution and task-launch-failure-recovery designs. +- Existing PR identity and fork regression tests. + +## Results + +PR base identity comparisons now return fixed mismatch reasons for the +comparison branch that failed. Structurally invalid resolver output also logs +`invalid_resolved_pr_base` with only task-repository ID and PR number. Tests +cover both unbound rejection and bound fallback without provider data in the +log. Launch behavior remains unchanged. Verification passed: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/orchestrator/executor -run 'Test.*PRBaseIdentity|Test.*PRBase' -count=1) +``` diff --git a/docs/plans/recent-runtime-log-remediation/task-05-summarize-recovery-outcomes.md b/docs/plans/recent-runtime-log-remediation/task-05-summarize-recovery-outcomes.md new file mode 100644 index 0000000000..ddd7ae76d7 --- /dev/null +++ b/docs/plans/recent-runtime-log-remediation/task-05-summarize-recovery-outcomes.md @@ -0,0 +1,90 @@ +--- +id: "05-summarize-recovery-outcomes" +title: "Summarize startup recovery outcomes" +status: done +wave: 5 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001 + - REQ-PLATFORM-STARTUP-PROGRESS-001 +acceptance_criteria: + - AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.4 + - AC-PLATFORM-STARTUP-PROGRESS-001.22 +system_design: + - ../../specs/platform/system-design/runtime-failure-attribution.md + - ../../specs/platform/system-design/startup-progress-visibility.md +--- + +# Task 05: Summarize startup recovery outcomes + +## Summary + +Startup reported 0/39 recovered sessions, but the inventory was stale and +existing safety rules correctly retained uncertain records. Emit one aggregate +outcome summary so future operators can interpret the warning. + +## In scope + +- Count candidates, retracked instances, and not-retracked records in the + lifecycle recovery pass; classify only reasons the pass proves. +- Cover zero-candidate, partial, and unknown-liveness cases in focused tests. + +## Out of scope + +- Counting skipped records as done, suppressing the progress warning, or + deleting/stopping uncertain runtime records. + +## Acceptance + +- One summary per pass reconciles candidate and outcome counts. +- Unmatched backend records remain unknown unless a specific reason is proven. +- Existing startup progress and guarded recovery behavior are unchanged. + +## Verification + +```bash +(cd apps/backend && go test -tags fts5 ./internal/agent/runtime/lifecycle -run 'Test.*Recovery' -count=1) +(cd apps/backend && go test -tags fts5 ./internal/startup -run 'Test.*Step' -count=1) +``` + +## Files likely touched + +- `apps/backend/internal/agent/runtime/lifecycle/manager_lifecycle.go` +- `apps/backend/internal/agent/runtime/lifecycle/manager_recovery_sessions_step_test.go` +- `apps/backend/internal/startup/step_lifecycle_test.go` + +## Dependencies + +None. + +## Risks + +- The recovery backend cannot report why every record was declined; use an + explicit unknown bucket rather than fabricating a reason. + +## Parallelism + +`sequential` + +## Inputs + +- Runtime-failure-attribution and startup-progress designs. +- Executor survival requirements for unknown-liveness records. + +## Results + +Startup recovery now emits one aggregate summary with candidate-count +certainty, re-tracked count, known refusal counts, and the unknown remainder. +Missing inventory data is marked unknown; it does not become an empty +candidate set. Existing progress totals, warnings, liveness guards, and stop +decisions remain unchanged. Verification passed: + +```bash +(cd apps/backend && go test -tags fts5 ./internal/agent/runtime/lifecycle -run 'Test.*Recovery' -count=1) +(cd apps/backend && go test -tags fts5 ./internal/startup -run 'Test.*Step' -count=1) +``` + +PR review follow-up added a code comment stating the candidate/retracked-count +invariant and defensive zero clamp. The recovery logic and decisions are +unchanged. diff --git a/docs/specs/platform/README.md b/docs/specs/platform/README.md index 7150478aaa..e0adcdd91f 100644 --- a/docs/specs/platform/README.md +++ b/docs/specs/platform/README.md @@ -62,6 +62,7 @@ retains ownership of its values, validation, authority, and persistence. - [Environment-specific browser tab title prefixes](requirements/dev-preview-title-prefixes.md) - [Browser console retention](requirements/browser-console-retention.md) - [Diagnostic logging](requirements/diagnostic-logging.md) +- [Runtime failure attribution](requirements/runtime-failure-attribution.md) - [Duration-aware E2E sharding and CI reliability](requirements/e2e-duration-aware-sharding.md) - [External CI runner capacity](requirements/external-e2e-runner-capacity.md) - [Expected runtime log severity](requirements/expected-runtime-log-severity.md) @@ -114,6 +115,7 @@ retains ownership of its values, validation, authority, and persistence. - [Viewport-bounded Session Delivery](system-design/viewport-bounded-session-delivery.md) - [Diagnostic logging System Design Part 1](system-design/diagnostic-logging-01.md) - [Diagnostic logging System Design Part 2](system-design/diagnostic-logging-02.md) +- [Runtime failure attribution](system-design/runtime-failure-attribution.md) - [Expected runtime log severity](system-design/expected-runtime-log-severity.md) - [Duration-aware E2E sharding and CI reliability](system-design/e2e-duration-aware-sharding.md) - [External CI runner capacity](system-design/external-e2e-runner-capacity.md) diff --git a/docs/specs/platform/requirements/bounded-task-status-delivery.md b/docs/specs/platform/requirements/bounded-task-status-delivery.md index f3e67a0cda..fdc9da73fd 100644 --- a/docs/specs/platform/requirements/bounded-task-status-delivery.md +++ b/docs/specs/platform/requirements/bounded-task-status-delivery.md @@ -37,6 +37,10 @@ Task rows currently obtain compact status indicators by observing large, session an idle agent after a restart, `status_summary.last_activity_at` shall remain unchanged. A later task mutation, user prompt, or conversational agent turn can advance it. +- **AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11:** When a queue-status event + arrives after its task is deleted, the projector shall discard the task's + cached summary state without publishing a replacement. A transient task read + failure shall remain an error. ## System design diff --git a/docs/specs/platform/requirements/runtime-failure-attribution.md b/docs/specs/platform/requirements/runtime-failure-attribution.md new file mode 100644 index 0000000000..743fc3b4ef --- /dev/null +++ b/docs/specs/platform/requirements/runtime-failure-attribution.md @@ -0,0 +1,47 @@ +--- +status: active +system: platform +created: 2026-09-24 +owners: + - kandev +--- + +# Runtime failure attribution requirements + +## Overview + +Several operational warnings identify a failed operation but not the boundary +that failed. Platform owns the cross-cutting diagnostic contract. The owning +systems retain their health, webhook, repository, and recovery behavior. + +## Requirements + +### REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001: Bounded failure attribution + +**Intent:** An operator can distinguish a host-side failure, a plugin response, +an identity rejection, and expected non-recovery without exposing request data. + +#### Acceptance criteria + +- **AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.1:** When a required persistence + probe fails, one bounded diagnostic shall identify the failing probe stage, + elapsed time, error class, and writer/reader connection-pool pressure. The + existing health transition and recovery behavior shall remain unchanged. +- **AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.2:** When a plugin webhook + returns a server error, a diagnostic shall distinguish host lifecycle failure, + host RPC failure, and a plugin-supplied response. It shall identify the plugin + and status without logging webhook keys, request contents, or credentials. +- **AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.3:** When a live pull request is + rejected by the task-repository identity guard, a bounded diagnostic shall + identify the comparison that failed. The same mismatch shall continue to + block launch or follow the existing safe fallback path. +- **AC-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001.4:** When startup session + recovery finishes below its inventory count, diagnostics shall distinguish + retracked records from records not retracked and classify known reasons. + Unknown liveness shall not be treated as absence or trigger cleanup. + +## Out of scope + +- Changing probe timeouts or readiness policy before the database stall is attributed. +- Changing plugin-provided webhook responses or external plugin manifests. +- Relaxing the pull-request identity guard or suppressing the startup progress warning. diff --git a/docs/specs/platform/system-design/bounded-task-status-delivery.md b/docs/specs/platform/system-design/bounded-task-status-delivery.md index 75781913c2..ab9c813685 100644 --- a/docs/specs/platform/system-design/bounded-task-status-delivery.md +++ b/docs/specs/platform/system-design/bounded-task-status-delivery.md @@ -1,5 +1,5 @@ --- -status: draft +status: current system: platform requirements: - REQ-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001 @@ -87,6 +87,14 @@ remain during migration, but switchers use the summary when present. ## Derivation rules +For a late `message.queue.status_changed` event, `Projector.handleEvent` treats +an authoritative task-not-found result from either workspace resolution or +`LoadLaunchQueue` as deletion. It drops cached projection state under the +task lock and publishes nothing. A different lookup error remains an error; +the read model cannot infer deletion from a transient database failure. This +extends AC-PLATFORM-BOUNDED-TASK-STATUS-DELIVERY-001.11 to projectors that +already cached the task's workspace before deletion. + - Pending permission outranks pending clarification for the row's primary state icon. Both outrank generating/background activity, which outranks coarse lifecycle state. This is the existing task-row precedence. @@ -419,3 +427,5 @@ intermediate replacement. [`../../plans/deleted-session-error-summary/plan.md`](../../../plans/deleted-session-error-summary/plan.md) - Semantic no-op equality repair: [Task-summary semantic equality](../../../plans/task-summary-semantic-equality/plan.md) +- Late deleted-task queue event repair: + [Recent runtime log remediation](../../../plans/recent-runtime-log-remediation/plan.md) diff --git a/docs/specs/platform/system-design/runtime-failure-attribution.md b/docs/specs/platform/system-design/runtime-failure-attribution.md new file mode 100644 index 0000000000..2c14283dc5 --- /dev/null +++ b/docs/specs/platform/system-design/runtime-failure-attribution.md @@ -0,0 +1,57 @@ +--- +status: current +system: platform +requirements: + - REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001 +created: 2026-09-24 +owners: + - kandev +--- + +# Runtime failure attribution system design + +## Purpose and boundaries + +This design adds structured, bounded attribution to existing failure paths. +It does not redefine [required-store health](postgres-domain-store-parity.md), +[task status projection](bounded-task-status-delivery.md), +[startup progress](startup-progress-visibility.md), plugin webhook behavior, or +task repository identity checks. + +## Requirement mapping + +| Requirement | Design section | +| --- | --- | +| `REQ-PLATFORM-RUNTIME-FAILURE-ATTRIBUTION-001` | [Diagnostic sites](#diagnostic-sites), [Safety](#safety) | + +## Diagnostic sites + +| Site | Bounded fields | Evidence boundary | +| --- | --- | --- | +| `internal/persistence/requiredstores/Health` | stage (`writer_ping`, `reader_ping`, `table_probe`), elapsed milliseconds, classified error, writer/reader `sql.DB.Stats` snapshots (`OpenConnections`, `InUse`, `WaitCount`, `WaitDuration`) | Capture on probe failure before cancellation; one diagnostic per failed sweep, not one per catalog store. Pool pressure is correlation evidence, not proof of the cause. | +| `internal/plugins/Controller.webhook` | plugin ID, HTTP status, origin (`host_lifecycle`, `host_rpc`, `plugin_response`), safe error class for host failures | Record only after the authenticated/declaration gate. A plugin response status is not recast as a host error. | +| `internal/orchestrator/executor` PR-base resolution | task-repository ID, PR number, mismatch reason enum | Have the comparison return a reason while retaining the current boolean decision. Test bound and unbound fork identities and every failure branch. | +| `internal/agent/runtime/lifecycle` recovery | candidate count, retracked count, not-retracked count, classified known reasons, unknown count | Summarize only outcomes the recovery code can prove. A missing backend return is `not_retracked_unknown` unless a specific reason is available; it is not evidence that the process is dead. | + +The recovery summary is separate from the counted startup progress step. The +step continues to advance only at its current reconstruction points and can +end below total with the existing warning. The summary is a single event for +one recovery pass, including zero-candidate passes; it does not log one row per +old inventory record. + +## Safety + +Do not log SQL, DSNs, webhook URLs or keys, request queries, headers, bodies, +tokens, local paths, repository URLs, or raw comparison identities. Use fixed +enum values and numeric measurements; keep identifiers to existing safe task, +plugin, and task-repository IDs. Diagnostics do not change any return value, +readiness state, retry, stop decision, or persisted record. + +## Related decisions + +- [Required internal persistence](../../../decisions/2026-09-05-required-internal-persistence.md) +- [Maintenance health-probe coordination](../../../decisions/2026-09-17-maintenance-health-probe-coordination.md) + +## Implementation plan + +- [Recent runtime log remediation](../../../plans/recent-runtime-log-remediation/plan.md)