From e9bbb912b803b86748963d55dbbbc090ffbb26ef Mon Sep 17 00:00:00 2001 From: nova28 Date: Fri, 21 Aug 2026 01:15:06 +0800 Subject: [PATCH] test(backend): make task-history tests fail when the history is actually wrong (#2848) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .../backendapp/storage_telemetry_test.go | 107 ++++ .../orchestrator/workflow_step_ledger_test.go | 61 ++ .../internal/steptelemetry/metrics_test.go | 66 ++- .../step_transition_writers_pin_test.go | 538 ++++++++++++++++-- .../step_transitions_writer_postgres_test.go | 220 ++++++- .../internal/task/repository/sqlite/task.go | 2 +- .../testdata/pinfixtures/const_hoisted.go.txt | 15 + .../pinfixtures/name_collision.go.txt | 20 + .../testdata/pinfixtures/nested/deep.go.txt | 11 + .../pinfixtures/sprintf_interpolated.go.txt | 21 + .../sqlite/turn_step_stamp_postgres_test.go | 83 ++- .../internal/telemetrycontract/health_test.go | 37 ++ .../plans/task-step-transition-ledger/plan.md | 103 +++- 13 files changed, 1230 insertions(+), 54 deletions(-) create mode 100644 apps/backend/internal/backendapp/storage_telemetry_test.go create mode 100644 apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/const_hoisted.go.txt create mode 100644 apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/name_collision.go.txt create mode 100644 apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/nested/deep.go.txt create mode 100644 apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/sprintf_interpolated.go.txt diff --git a/apps/backend/internal/backendapp/storage_telemetry_test.go b/apps/backend/internal/backendapp/storage_telemetry_test.go new file mode 100644 index 0000000000..8fea7a54fd --- /dev/null +++ b/apps/backend/internal/backendapp/storage_telemetry_test.go @@ -0,0 +1,107 @@ +package backendapp + +import ( + "context" + "testing" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" + + "github.com/kandev/kandev/internal/common/config" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/telemetrycontract" +) + +// TestProvideRepositoriesActivatesTelemetryContracts is the composition-level +// regression test for activateTelemetryContracts (storage.go:109 call site, +// :222 declaration): internal/telemetrycontract's own tests only exercise the +// underlying Store.Activate directly, so nothing proves the real boot path +// still calls it. A regression that dropped the call site, or moved it ahead +// of a repository's initSchema, would leave telemetry_activations silently +// missing rows on every real boot with the rest of the suite fully green — +// mirrors TestProvideRepositoriesBackfillsSessionCachedTokens's rationale for +// the neighboring backfillSessionCachedTokens call. +// +// telemetry_activations itself is created unconditionally by +// telemetrycontract.NewWithDB, independent of ordering, so a row-count +// assertion alone cannot catch activateTelemetryContracts running ahead of a +// contract's backing repository's initSchema — Store.Activate writes the +// activation row regardless of whether those backing objects exist yet. The +// ordering-sensitive signal is LogHealth's per-contract "exists" field, so +// this test also captures the boot log via an observer core and asserts +// every registered contract reports exists=true. +func TestProvideRepositoriesActivatesTelemetryContracts(t *testing.T) { + ctx := context.Background() + cfg := &config.Config{ + HomeDir: t.TempDir(), + Database: config.DatabaseConfig{Driver: "sqlite"}, + } + core, recorded := observer.New(zapcore.InfoLevel) + log, err := logger.NewFromZap(zap.New(core)) + if err != nil { + t.Fatalf("NewFromZap: %v", err) + } + + pool, _, cleanups, err := provideRepositories(ctx, cfg, log, "test") + t.Cleanup(func() { + for i := len(cleanups) - 1; i >= 0; i-- { + if cleanups[i] != nil { + _ = cleanups[i]() + } + } + }) + if err != nil { + t.Fatalf("provideRepositories: %v", err) + } + + registry := telemetrycontract.Registry() + if len(registry) == 0 { + t.Fatal("telemetrycontract.Registry() is empty; nothing for this test to assert") + } + writer := pool.Writer() + for _, c := range registry { + var count int + if err := writer.QueryRowx(writer.Rebind( + `SELECT COUNT(*) FROM telemetry_activations WHERE contract_key = ? AND contract_version = ?`), + c.Key, c.Version, + ).Scan(&count); err != nil { + t.Fatalf("query telemetry_activations for %s v%d: %v", c.Key, c.Version, err) + } + if count != 1 { + t.Errorf("telemetry_activations row count for %s v%d = %d, want 1 (activateTelemetryContracts must run during provideRepositories boot)", + c.Key, c.Version, count) + } + + exists, found := contractExistsFromHealthLog(recorded, c.Key, c.Version) + if !found { + t.Errorf("no telemetry.contract.health log entry for %s v%d (activateTelemetryContracts must call LogHealth during provideRepositories boot)", + c.Key, c.Version) + continue + } + if !exists { + t.Errorf("telemetry.contract.health exists=%v for %s v%d, want true (activateTelemetryContracts must run after every repository's initSchema, so the contract's backing objects already exist)", + exists, c.Key, c.Version) + } + } +} + +// contractExistsFromHealthLog finds the "telemetry.contract.health" entry for +// (key, version) in the observed log and reports its "exists" field. +func contractExistsFromHealthLog(recorded *observer.ObservedLogs, key string, version int) (exists bool, found bool) { + for _, entry := range recorded.All() { + if entry.Message != "telemetry.contract.health" { + continue + } + fields := entry.ContextMap() + if fields["contract_key"] != key { + continue + } + if v, ok := fields["contract_version"].(int64); !ok || int(v) != version { + continue + } + existsVal, _ := fields["exists"].(bool) + return existsVal, true + } + return false, false +} diff --git a/apps/backend/internal/orchestrator/workflow_step_ledger_test.go b/apps/backend/internal/orchestrator/workflow_step_ledger_test.go index 9d7c50d682..3e86406ef0 100644 --- a/apps/backend/internal/orchestrator/workflow_step_ledger_test.go +++ b/apps/backend/internal/orchestrator/workflow_step_ledger_test.go @@ -21,6 +21,67 @@ import ( wfmodels "github.com/kandev/kandev/internal/workflow/models" ) +// TestExecuteStepTransitionRecordsEngineTransitionOnLegacyPath drives the +// legacy path directly (executeStepTransition, reached by ProcessOnTurnStart +// and ProcessOnTurnComplete) end to end and asserts the resulting ledger row +// — every other orchestrator ledger test exercises engineTransitionAttribution +// only through the newer workflowStore.ApplyTransition call site +// (TestApplyTransitionRecordsEngineTransitionFromSessionID) or in isolation +// (TestEngineTransitionAttributionPreservesUserCancellationTrigger), never +// through executeStepTransition's own call to it with a real ledger write. +// Table-driven over both legacy triggers so each of ProcessOnTurnStart's and +// ProcessOnTurnComplete's real call sites into executeStepTransition is +// proven to reach a correct session-originated ledger row on its own, not +// just one of the two. This does NOT distinguish on_turn_start from +// on_turn_complete: engineTriggerIsSessionOriginated puts both in the same +// case arm, so a bug that swapped the triggerOnEnter->engine.Trigger mapping +// at the executeStepTransition call site (event_handlers_workflow.go) would +// still produce byte-identical rows here and go undetected — only a change +// that stopped treating one of the two as session-originated would surface +// as a failure. +func TestExecuteStepTransitionRecordsEngineTransitionOnLegacyPath(t *testing.T) { + tests := []struct { + name string + triggerOnEnter bool + wantTrigger engine.Trigger + }{ + {name: "on_turn_start", triggerOnEnter: false, wantTrigger: engine.TriggerOnTurnStart}, + {name: "on_turn_complete", triggerOnEnter: true, wantTrigger: engine.TriggerOnTurnComplete}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + steps := newMockStepGetter() + fromStep := &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Name: "Source"} + steps.steps["step2"] = &wfmodels.WorkflowStep{ID: "step2", WorkflowID: "wf1", Name: "Target"} + svc := createTestService(repo, steps, newMockTaskRepo()) + + svc.executeStepTransition(ctx, "t1", "s1", fromStep, "step2", tc.triggerOnEnter) + + rows := stepTransitionRowsForTaskOrchestrator(t, repo, "t1") + if len(rows) == 0 { + t.Fatalf("expected a ledger row, got none") + } + last := rows[len(rows)-1] + if last.trigger != string(steptelemetry.TriggerEngineTransition) { + t.Fatalf("trigger = %q, want %q", last.trigger, steptelemetry.TriggerEngineTransition) + } + if last.actorKind != string(steptelemetry.ActorAgent) { + t.Fatalf("actor_kind = %q, want %q (legacy %s is session-originated)", last.actorKind, steptelemetry.ActorAgent, tc.wantTrigger) + } + if last.actorID == nil || *last.actorID != "s1" { + t.Fatalf("actor_id = %v, want s1", last.actorID) + } + if last.sessionID == nil || *last.sessionID != "s1" { + t.Fatalf("session_id = %v, want s1", last.sessionID) + } + }) + } +} + func stepTransitionRowsForTaskOrchestrator(t *testing.T, repo *sqliterepo.Repository, taskID string) []struct { trigger string actorKind string diff --git a/apps/backend/internal/steptelemetry/metrics_test.go b/apps/backend/internal/steptelemetry/metrics_test.go index 3d81dc45ac..5167cfc54b 100644 --- a/apps/backend/internal/steptelemetry/metrics_test.go +++ b/apps/backend/internal/steptelemetry/metrics_test.go @@ -1,6 +1,7 @@ package steptelemetry import ( + "expvar" "testing" "go.uber.org/zap" @@ -20,15 +21,32 @@ func observerLogger(t *testing.T) (*logger.Logger, *observer.ObservedLogs) { return log, logs } +// expvarMapValue reads the current int64 value stored at key in an +// expvar.Map, or 0 if the key hasn't been set yet (Add creates its backing +// *expvar.Int lazily on first use). +func expvarMapValue(t *testing.T, m *expvar.Map, key string) int64 { + t.Helper() + v := m.Get(key) + if v == nil { + return 0 + } + i, ok := v.(*expvar.Int) + if !ok { + t.Fatalf("expvar value at key %q is a %T, want *expvar.Int", key, v) + } + return i.Value() +} + func TestRecordLedgerRowBumpsCounterAndLogs(t *testing.T) { log, logs := observerLogger(t) - before := stepTransitionsTotal.String() + key := metricLabel("trigger", string(TriggerManualMove)) + before := expvarMapValue(t, stepTransitionsTotal, key) RecordLedgerRow(log, TriggerManualMove) - after := stepTransitionsTotal.String() - if after == before { - t.Fatal("expvar counter telemetry_step_transitions_inserted_total did not change") + after := expvarMapValue(t, stepTransitionsTotal, key) + if after != before+1 { + t.Fatalf("telemetry_step_transitions_inserted_total[%s] = %d, want %d", key, after, before+1) } entries := logs.FilterMessage(metricStepTransitionInserted).All() @@ -40,6 +58,29 @@ func TestRecordLedgerRowBumpsCounterAndLogs(t *testing.T) { } } +// TestRecordLedgerRowCounterIsKeyedByTrigger pins that the expvar counter is +// keyed per-trigger, not one shared bucket. TestRecordLedgerRowBumpsCounter- +// AndLogs alone would still pass if incStepTransition collapsed every +// trigger into stepTransitionsTotal.Add("", 1) — this asserts recording one +// trigger leaves a different trigger's own bucket untouched. +func TestRecordLedgerRowCounterIsKeyedByTrigger(t *testing.T) { + log, _ := observerLogger(t) + moveKey := metricLabel("trigger", string(TriggerManualMove)) + bulkKey := metricLabel("trigger", string(TriggerBulkMove)) + beforeMove := expvarMapValue(t, stepTransitionsTotal, moveKey) + beforeBulk := expvarMapValue(t, stepTransitionsTotal, bulkKey) + + RecordLedgerRow(log, TriggerManualMove) + + if got := expvarMapValue(t, stepTransitionsTotal, moveKey); got != beforeMove+1 { + t.Fatalf("telemetry_step_transitions_inserted_total[%s] = %d, want %d", moveKey, got, beforeMove+1) + } + if got := expvarMapValue(t, stepTransitionsTotal, bulkKey); got != beforeBulk { + t.Fatalf("telemetry_step_transitions_inserted_total[%s] = %d, want unchanged %d (recording %s must not bump %s's bucket)", + bulkKey, got, beforeBulk, TriggerManualMove, TriggerBulkMove) + } +} + func TestRecordLedgerRowNoopWhenLoggerNil(t *testing.T) { // Must not panic. RecordLedgerRow(nil, TriggerBulkMove) @@ -47,9 +88,26 @@ func TestRecordLedgerRowNoopWhenLoggerNil(t *testing.T) { func TestRecordTurnStampBumpsCounterAndLogsBothBranches(t *testing.T) { log, logs := observerLogger(t) + presentKey := metricLabel("stamp", "present") + absentKey := metricLabel("stamp", "absent") + beforePresent := expvarMapValue(t, turnStampsTotal, presentKey) + beforeAbsent := expvarMapValue(t, turnStampsTotal, absentKey) RecordTurnStamp(log, true) + if got := expvarMapValue(t, turnStampsTotal, presentKey); got != beforePresent+1 { + t.Fatalf("telemetry_turn_stamps_total[%s] = %d after the present branch, want %d", presentKey, got, beforePresent+1) + } + if got := expvarMapValue(t, turnStampsTotal, absentKey); got != beforeAbsent { + t.Fatalf("telemetry_turn_stamps_total[%s] = %d after the present branch, want unchanged %d", absentKey, got, beforeAbsent) + } + RecordTurnStamp(log, false) + if got := expvarMapValue(t, turnStampsTotal, absentKey); got != beforeAbsent+1 { + t.Fatalf("telemetry_turn_stamps_total[%s] = %d after the absent branch, want %d", absentKey, got, beforeAbsent+1) + } + if got := expvarMapValue(t, turnStampsTotal, presentKey); got != beforePresent+1 { + t.Fatalf("telemetry_turn_stamps_total[%s] = %d after the absent branch, want unchanged %d", presentKey, got, beforePresent+1) + } entries := logs.FilterMessage(metricTurnStamped).All() if len(entries) != 2 { diff --git a/apps/backend/internal/task/repository/sqlite/step_transition_writers_pin_test.go b/apps/backend/internal/task/repository/sqlite/step_transition_writers_pin_test.go index 91fa415267..a67ff62f3b 100644 --- a/apps/backend/internal/task/repository/sqlite/step_transition_writers_pin_test.go +++ b/apps/backend/internal/task/repository/sqlite/step_transition_writers_pin_test.go @@ -3,17 +3,22 @@ package sqlite // TestStepTransitionWritersArePinned is the writer-health control that // matters most, per the spec: the realistic failure mode is not the ledger // writer breaking, it's a NEW production statement mutating -// tasks.workflow_step_id that nobody wired into it. This walks the package's -// own source (go/parser, not a line regex — every statement here is a -// multi-line raw string literal, which a regex over raw lines misreports on) -// and asserts the exact set of functions containing such a statement matches -// the registered, ledger-wired set. Add a new mutating statement and this -// test fails until it is registered below AND wired to recordStepTransition. +// tasks.workflow_step_id that nobody wired into it. This walks the entire +// backend source tree (go/parser, not a line regex — every statement here is +// a multi-line raw string literal, which a regex over raw lines misreports +// on), rooted at apps/backend/internal (see findBackendSourceRoot) rather +// than this package's own directory — a writer added in ANY backend +// package, not only this one, must be able to fail this test — and asserts +// the exact set of functions containing such a statement matches the +// registered, ledger-wired set. Add a new mutating statement and this test +// fails until it is registered below AND wired to recordStepTransition. import ( + "fmt" "go/ast" "go/parser" "go/token" + "io/fs" "os" "path/filepath" "sort" @@ -23,25 +28,54 @@ import ( // registeredStepMutators is the closed set of functions in this package // known to mutate tasks.workflow_step_id, each already wired to -// recordStepTransition. A new function containing a matching statement that -// is not in this set means either: (a) it is a genuine new step-mutation -// path — wire it to recordStepTransition and add it here; or (b) it is a -// false positive (e.g. a table-rebuild migration copying existing rows -// verbatim) — narrow the detection below rather than registering it. +// recordStepTransition, keyed by "packageRelDir/ReceiverType.FuncName" (see +// funcIdentity) rather than a bare "ReceiverType.FuncName" — a bare +// receiver-qualified name still collides across distinct PACKAGES declaring +// the same receiver type (this codebase declares a type literally named +// Repository in 11 different backend packages; see Review round 2 Finding A +// / TestFindWorkflowStepIDMutatorsQualifiesByPackage), and a fully bare name +// collides across distinct receiver types with the same method name in one +// package (see TestFindWorkflowStepIDMutatorsCatchesEvasiveShapes shape +// (c)). packageRelDir is this package's path relative to the scan root +// (findBackendSourceRoot's apps/backend/internal boundary) — always +// "task/repository/sqlite" for entries in this file, since every writer +// registered here lives in this package. A new function containing a +// matching statement that is not in this set means either: (a) it is a +// genuine new step-mutation path — wire it to recordStepTransition and add +// it here; or (b) it is a false positive (e.g. a table-rebuild migration +// copying existing rows verbatim) — narrow the detection below rather than +// registering it. var registeredStepMutators = []string{ - "insertTaskTx", - "updateTaskTx", - "UpdateTaskIfWorkflowStepHasCapacity", - "PromoteQueuedTaskIfWorkflowStepHasCapacity", - "RestoreTaskMessageRollbackIfSessionState", - "AddTaskToWorkflow", - "RemoveTaskFromWorkflow", + "task/repository/sqlite/Repository.insertTaskTx", + "task/repository/sqlite/Repository.updateTaskTx", + "task/repository/sqlite/Repository.UpdateTaskIfWorkflowStepHasCapacity", + "task/repository/sqlite/Repository.PromoteQueuedTaskIfWorkflowStepHasCapacity", + "task/repository/sqlite/Repository.RestoreTaskMessageRollbackIfSessionState", + "task/repository/sqlite/Repository.AddTaskToWorkflow", + "task/repository/sqlite/Repository.RemoveTaskFromWorkflow", } func TestStepTransitionWritersArePinned(t *testing.T) { - found, err := findWorkflowStepIDMutators(t, ".") + root, err := findBackendSourceRoot(".") if err != nil { - t.Fatalf("scan package for workflow_step_id mutators: %v", err) + t.Fatalf("locate backend source root: %v", err) + } + // Sanity check that the scan root is genuinely the wide boundary and not + // silently regressed back to "." (this package's own directory) — a + // writer added in ANY backend package, not just this one, must be able + // to fail this test. internal/office is an arbitrary sibling package + // known to exist; its presence under root proves the walk spans more + // than internal/task/repository/sqlite. + if filepath.Base(root) != "internal" { + t.Fatalf("scan root %q is not the backend internal/ directory", root) + } + if _, statErr := os.Stat(filepath.Join(root, "office")); statErr != nil { + t.Fatalf("scan root %q does not contain sibling package internal/office — root is narrower than the whole backend source tree: %v", root, statErr) + } + + found, err := findWorkflowStepIDMutators(t, root) + if err != nil { + t.Fatalf("scan backend source for workflow_step_id mutators: %v", err) } registered := make(map[string]bool, len(registeredStepMutators)) @@ -84,58 +118,476 @@ func TestStepTransitionWritersArePinned(t *testing.T) { } } -// findWorkflowStepIDMutators parses every non-test .go file directly in dir -// and returns the set of top-level function names whose body contains a -// string literal that either inserts a new tasks row with an explicit -// workflow_step_id column, or updates the tasks table and assigns -// workflow_step_id. +// findBackendSourceRoot walks up from startDir looking for the Go module +// root (the directory containing go.mod) and returns its internal/ +// subdirectory. This is the production scan's boundary: internal/ is the +// entire body of backend source that could define a tasks.workflow_step_id +// writer, so rooting the pinning test there instead of "." (this package's +// own directory) is what actually closes false-negative shape (d) — a +// writer added in any OTHER backend package, not just a nested subdirectory +// of this one. go test always runs with the package directory as its +// working directory, so startDir="." from this package resolves reliably to +// apps/backend/internal regardless of where the repo checkout lives. +func findBackendSourceRoot(startDir string) (string, error) { + dir, err := filepath.Abs(startDir) + if err != nil { + return "", err + } + for { + if _, statErr := os.Stat(filepath.Join(dir, "go.mod")); statErr == nil { + return filepath.Join(dir, "internal"), nil + } + parent := filepath.Dir(dir) + if parent == dir { + return "", fmt.Errorf("no go.mod found above %s", startDir) + } + dir = parent + } +} + +// findWorkflowStepIDMutators recursively parses every non-test .go file +// under dir (plus .go.txt, so isolated test fixtures under testdata/ can be +// parsed without ever compiling as part of any real package) and returns the +// set of function identities ("ReceiverType.FuncName", or a bare name for a +// receiver-less function) whose body mutates tasks.workflow_step_id — either +// directly via a string literal, indirectly via a package-level const/var +// the body references by name, or via a call to a local fmt.Sprintf-based +// column setter passed the literal column name "workflow_step_id". A +// directory literally named "testdata" is skipped while walking so +// production scans never pick up fixtures. +// +// dir can span many packages (the production scan roots at the whole +// backend internal/ tree — see findBackendSourceRoot). Constants and +// format-setters are therefore resolved per package, grouping parsed files +// by their containing directory and running the const/format-setter/mutation +// pipeline independently per directory, rather than flattening every parsed +// file's package-level consts into one repo-wide map. Two unrelated packages +// can legally declare a const or helper function with the same bare name +// (e.g. two packages each with their own `updateQuery` constant); a flat map +// would let one package's SQL literal leak into another's mutation check — +// or mask a genuine literal — purely by name collision. Each directory is +// exactly one Go package once _test.go files are excluded (which this scan +// already does), so scoping resolution to the files parsed from one +// directory at a time is correct without tracking real package identifiers. func findWorkflowStepIDMutators(t *testing.T, dir string) (map[string]bool, error) { t.Helper() - fset := token.NewFileSet() - found := make(map[string]bool) - entries, err := os.ReadDir(dir) + paths, err := collectSourceFiles(dir) if err != nil { return nil, err } - for _, entry := range entries { - name := entry.Name() - if entry.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") { + + byDir := make(map[string][]string) + for _, path := range paths { + d := filepath.Dir(path) + byDir[d] = append(byDir[d], path) + } + + found := make(map[string]bool) + fset := token.NewFileSet() + for groupDir, dirPaths := range byDir { + relDir, relErr := filepath.Rel(dir, groupDir) + if relErr != nil { + return nil, fmt.Errorf("resolve package directory %q relative to scan root %q: %w", groupDir, dir, relErr) + } + + files := make([]*ast.File, 0, len(dirPaths)) + for _, path := range dirPaths { + file, err := parser.ParseFile(fset, path, nil, 0) + if err != nil { + return nil, err + } + files = append(files, file) + } + + constants := collectPackageStringConstants(files) + formatSetters := collectFormatColumnSetters(files, constants) + + for _, file := range files { + ast.Inspect(file, func(n ast.Node) bool { + fn, ok := n.(*ast.FuncDecl) + if !ok || fn.Body == nil { + return true + } + if functionMutatesWorkflowStepID(fn.Body, constants, formatSetters) { + found[funcIdentity(fn, relDir)] = true + } + return true + }) + } + } + return found, nil +} + +// collectSourceFiles walks dir recursively, collecting non-test .go and +// .go.txt file paths. Directories named "testdata" are skipped so a +// production scan rooted above testdata/ never parses fixture files as +// real package source. +func collectSourceFiles(dir string) ([]string, error) { + var paths []string + err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr + } + if d.IsDir() { + if d.Name() == "testdata" && path != dir { + return filepath.SkipDir + } + return nil + } + name := d.Name() + if strings.HasSuffix(name, "_test.go") { + return nil + } + if strings.HasSuffix(name, ".go") || strings.HasSuffix(name, ".go.txt") { + paths = append(paths, path) + } + return nil + }) + if err != nil { + return nil, err + } + return paths, nil +} + +// collectPackageStringConstants maps every package-level const/var string +// declaration's name to its literal value, across all parsed files. Only +// top-level declarations are visited (file.Decls), so a function-local +// const or var is never captured here. +func collectPackageStringConstants(files []*ast.File) map[string]string { + consts := make(map[string]string) + for _, file := range files { + for _, decl := range file.Decls { + genDecl, ok := decl.(*ast.GenDecl) + if !ok || (genDecl.Tok != token.CONST && genDecl.Tok != token.VAR) { + continue + } + collectValueSpecStrings(genDecl, consts) + } + } + return consts +} + +func collectValueSpecStrings(genDecl *ast.GenDecl, out map[string]string) { + for _, spec := range genDecl.Specs { + valueSpec, ok := spec.(*ast.ValueSpec) + if !ok { continue } - file, err := parser.ParseFile(fset, filepath.Join(dir, name), nil, 0) - if err != nil { - return nil, err + for i, name := range valueSpec.Names { + if i >= len(valueSpec.Values) { + continue + } + lit, ok := valueSpec.Values[i].(*ast.BasicLit) + if !ok || lit.Kind != token.STRING { + continue + } + out[name.Name] = strings.Trim(lit.Value, "`\"") } + } +} + +// collectFormatColumnSetters returns the bare names of functions whose body +// calls fmt.Sprintf (or any *.Sprintf selector call) with a format string +// shaped like "UPDATE tasks SET %s ..." — the taskScalarColumns/ +// execTaskScalar pattern where the mutated column is a parameter, not a +// literal in this function's own body. Keyed by bare name because call +// sites are matched by bare callee name below (no full type resolution). +func collectFormatColumnSetters(files []*ast.File, constants map[string]string) map[string]bool { + setters := make(map[string]bool) + for _, file := range files { ast.Inspect(file, func(n ast.Node) bool { fn, ok := n.(*ast.FuncDecl) if !ok || fn.Body == nil { return true } - if functionMutatesWorkflowStepID(fn.Body) { - found[fn.Name.Name] = true + if functionHasScalarSetterFormat(fn.Body, constants) { + setters[fn.Name.Name] = true } return true }) } - return found, nil + return setters } -func functionMutatesWorkflowStepID(body *ast.BlockStmt) bool { - mutates := false +func functionHasScalarSetterFormat(body *ast.BlockStmt, constants map[string]string) bool { + has := false ast.Inspect(body, func(n ast.Node) bool { - lit, ok := n.(*ast.BasicLit) - if !ok || lit.Kind != token.STRING { + call, ok := n.(*ast.CallExpr) + if !ok || len(call.Args) == 0 { return true } - if literalMutatesWorkflowStepID(strings.Trim(lit.Value, "`\"")) { - mutates = true + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok || sel.Sel.Name != "Sprintf" { + return true + } + if fmtStr, ok := resolveStringLiteral(call.Args[0], constants); ok && isScalarSetterFormat(fmtStr) { + has = true + } + return true + }) + return has +} + +func isScalarSetterFormat(s string) bool { + return strings.Contains(s, "UPDATE tasks") && strings.Contains(s, "SET %s") +} + +// resolveStringLiteral resolves expr to a string value, either directly +// (a string literal) or indirectly (an identifier naming a package-level +// const/var string collected in constants). +func resolveStringLiteral(expr ast.Expr, constants map[string]string) (string, bool) { + switch v := expr.(type) { + case *ast.BasicLit: + if v.Kind == token.STRING { + return strings.Trim(v.Value, "`\""), true + } + case *ast.Ident: + if val, ok := constants[v.Name]; ok { + return val, true + } + } + return "", false +} + +// functionMutatesWorkflowStepID reports whether body mutates +// tasks.workflow_step_id via any of: a direct string literal, an identifier +// referencing a package-level const/var string, or a call to a known +// fmt.Sprintf-based column setter passed the literal column name +// "workflow_step_id". +func functionMutatesWorkflowStepID(body *ast.BlockStmt, constants map[string]string, formatSetters map[string]bool) bool { + mutates := false + ast.Inspect(body, func(n ast.Node) bool { + switch v := n.(type) { + case *ast.BasicLit: + if v.Kind == token.STRING && literalMutatesWorkflowStepID(strings.Trim(v.Value, "`\"")) { + mutates = true + } + case *ast.Ident: + if val, ok := constants[v.Name]; ok && literalMutatesWorkflowStepID(val) { + mutates = true + } + case *ast.CallExpr: + if callTargetsColumnSetter(v, formatSetters) { + mutates = true + } } return true }) return mutates } +// callTargetsColumnSetter reports whether call invokes a known +// fmt.Sprintf-based column setter (by bare callee name) with the literal +// string "workflow_step_id" as one of its arguments. +func callTargetsColumnSetter(call *ast.CallExpr, formatSetters map[string]bool) bool { + name := calleeName(call.Fun) + if name == "" || !formatSetters[name] { + return false + } + for _, arg := range call.Args { + lit, ok := arg.(*ast.BasicLit) + if ok && lit.Kind == token.STRING && strings.Trim(lit.Value, "`\"") == "workflow_step_id" { + return true + } + } + return false +} + +func calleeName(fun ast.Expr) string { + switch v := fun.(type) { + case *ast.Ident: + return v.Name + case *ast.SelectorExpr: + return v.Sel.Name + default: + return "" + } +} + +// funcIdentity qualifies fn's name with its receiver type and its +// containing package's directory relative to the scan root +// ("task/repository/sqlite/Repository.updateTaskTx"), or "relDir/FuncName" +// for a receiver-less function, so two distinct types' same-named methods +// are never collapsed into one key (within a package) AND two distinct +// packages declaring the same receiver type name are never collapsed into +// one key (across packages — see Review round 2 Finding A: this codebase +// declares a type literally named Repository in 11 different backend +// packages, so "Repository.updateTaskTx" alone is not a safe key once the +// scan spans more than one package). relDir is "." for a file directly in +// the scanned root, which yields an unprefixed identity — used by the +// evasive-shapes fixtures that live directly under testdata/pinfixtures. +func funcIdentity(fn *ast.FuncDecl, relDir string) string { + name := fn.Name.Name + if recv := receiverTypeName(fn); recv != "" { + name = recv + "." + name + } + if relDir == "" || relDir == "." { + return name + } + return filepath.ToSlash(relDir) + "/" + name +} + +func receiverTypeName(fn *ast.FuncDecl) string { + if fn.Recv == nil || len(fn.Recv.List) == 0 { + return "" + } + expr := fn.Recv.List[0].Type + if star, ok := expr.(*ast.StarExpr); ok { + expr = star.X + } + if ident, ok := expr.(*ast.Ident); ok { + return ident.Name + } + return "" +} + +// TestFindWorkflowStepIDMutatorsCatchesEvasiveShapes runs the detector +// against permanent fixtures under testdata/pinfixtures/, one per +// false-negative shape identified in the mutation review: a const-hoisted +// SQL literal, an fmt.Sprintf-interpolated column name (the exact shape +// already live in internal/office/repository/sqlite/tasks.go's +// execTaskScalar), two distinct receiver types sharing a bare method name, +// and a writer placed in a subdirectory nested under whatever root the +// detector is given (proving recursion, not scan-root width — see +// TestFindWorkflowStepIDMutatorsCatchesSiblingPackageWriter below for that). +// testdata/ is otherwise skipped by findWorkflowStepIDMutators so these +// fixtures never affect TestStepTransitionWritersArePinned's scan of real +// backend source. +func TestFindWorkflowStepIDMutatorsCatchesEvasiveShapes(t *testing.T) { + found, err := findWorkflowStepIDMutators(t, filepath.Join("testdata", "pinfixtures")) + if err != nil { + t.Fatalf("scan fixtures: %v", err) + } + + want := []string{ + "FixtureRepo.constHoistedStepMutator", // (a) const-hoisted SQL literal + "FixtureRepo.sprintfInterpolatedStepMutator", // (b) fmt.Sprintf column interpolation + "RepoA.updateTaskTx", // (c) receiver-qualified identity + "RepoB.updateTaskTx", // (c) distinct from RepoA's, not collapsed + "nested/NestedRepo.deeplyNestedStepMutator", // (d, partial) recursion into a subdirectory — package-qualified since it's one level below the scan root + } + if len(found) != len(want) { + t.Errorf("findWorkflowStepIDMutators found %d mutators from fixtures, want %d: found=%v", len(found), len(want), found) + } + for _, name := range want { + if !found[name] { + t.Errorf("expected findWorkflowStepIDMutators to catch %q, but it did not (found: %v)", name, found) + } + } +} + +// TestFindWorkflowStepIDMutatorsCatchesSiblingPackageWriter is the permanent +// regression proof for the rest of false-negative shape (d): "single-package +// scope misses writers added outside the sqlite repo package." The nested +// fixture above proves the detector recurses into subdirectories of +// whatever root it is given; it does not prove the production scan's root +// actually spans sibling packages, since before this fix the root was "." +// (this package's own directory) and recursion had nothing above it to +// explore. This test targets that boundary directly, independent of the +// real repo layout: it builds a synthetic two-package tree in a t.TempDir() +// — one package with a genuine tasks.workflow_step_id writer, a sibling +// package with none — and asserts findWorkflowStepIDMutators, given the +// tree's root, finds the writer inside the sibling package. That is exactly +// the guarantee TestStepTransitionWritersArePinned now depends on when +// rooted at findBackendSourceRoot's apps/backend/internal boundary: a +// writer in ANY sibling package under that root must be caught, not only +// ones nested under this package's own directory. +func TestFindWorkflowStepIDMutatorsCatchesSiblingPackageWriter(t *testing.T) { + root := t.TempDir() + pkgWithWriter := filepath.Join(root, "pkgwithwriter") + pkgWithoutWriter := filepath.Join(root, "pkgwithoutwriter") + if err := os.MkdirAll(pkgWithWriter, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", pkgWithWriter, err) + } + if err := os.MkdirAll(pkgWithoutWriter, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", pkgWithoutWriter, err) + } + writePinTestFixtureFile(t, filepath.Join(pkgWithWriter, "writer.go"), `package pkgwithwriter + +type Repository struct{} + +func (r *Repository) mutateStep() string { + return "UPDATE tasks SET workflow_step_id = ? WHERE id = ?" +} +`) + writePinTestFixtureFile(t, filepath.Join(pkgWithoutWriter, "other.go"), `package pkgwithoutwriter + +func Noop() {} +`) + + found, err := findWorkflowStepIDMutators(t, root) + if err != nil { + t.Fatalf("scan synthetic sibling-package tree: %v", err) + } + if !found["pkgwithwriter/Repository.mutateStep"] { + t.Fatalf("expected a writer in a sibling package under %s to be found, package-qualified; found=%v", root, found) + } +} + +// TestFindWorkflowStepIDMutatorsQualifiesByPackage is the permanent +// regression proof for Review round 2 Finding A: funcIdentity used to key +// solely on "ReceiverType.FuncName" with no package component, so two +// different backend packages declaring the same receiver type name — this +// codebase declares a type literally named Repository in 11 different +// packages — and the same method name collided into a single identity. A +// genuine writer in one package could then silently "borrow" another +// package's already-registered status: TestStepTransitionWritersArePinned +// would stay green even though the second writer was never wired to +// recordStepTransition. This builds a synthetic two-package tree where both +// packages declare `type Repository struct{}` with an identical method name, +// each mutating tasks.workflow_step_id via its own distinct SQL text, and +// asserts the detector reports two distinct package-qualified identities, +// not one collapsed entry. +func TestFindWorkflowStepIDMutatorsQualifiesByPackage(t *testing.T) { + root := t.TempDir() + pkgA := filepath.Join(root, "pkga") + pkgB := filepath.Join(root, "pkgb") + if err := os.MkdirAll(pkgA, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", pkgA, err) + } + if err := os.MkdirAll(pkgB, 0o755); err != nil { + t.Fatalf("mkdir %s: %v", pkgB, err) + } + writePinTestFixtureFile(t, filepath.Join(pkgA, "repo.go"), `package pkga + +type Repository struct{} + +func (r *Repository) updateTaskTx() string { + return "UPDATE tasks SET workflow_step_id = ?, updated_at = ? WHERE id = ?" +} +`) + writePinTestFixtureFile(t, filepath.Join(pkgB, "repo.go"), `package pkgb + +type Repository struct{} + +func (r *Repository) updateTaskTx() string { + return "UPDATE tasks SET workflow_step_id = ? WHERE id = ?" +} +`) + + found, err := findWorkflowStepIDMutators(t, root) + if err != nil { + t.Fatalf("scan synthetic colliding-identity tree: %v", err) + } + if len(found) != 2 { + t.Fatalf("expected 2 distinct package-qualified identities from two colliding Repository.updateTaskTx pairs, got %d: %v", len(found), found) + } + if !found["pkga/Repository.updateTaskTx"] { + t.Errorf("expected pkga's writer to be found under its own package-qualified identity; found=%v", found) + } + if !found["pkgb/Repository.updateTaskTx"] { + t.Errorf("expected pkgb's writer to be found under its own package-qualified identity; found=%v", found) + } +} + +func writePinTestFixtureFile(t *testing.T, path, contents string) { + t.Helper() + if err := os.WriteFile(path, []byte(contents), 0o644); err != nil { + t.Fatalf("write %s: %v", path, err) + } +} + func literalMutatesWorkflowStepID(sql string) bool { insertsWithColumn := strings.Contains(sql, "INSERT INTO tasks (") && strings.Contains(sql, "workflow_step_id") // "workflow_step_id = " (not just "= ?") also catches an UPDATE that diff --git a/apps/backend/internal/task/repository/sqlite/step_transitions_writer_postgres_test.go b/apps/backend/internal/task/repository/sqlite/step_transitions_writer_postgres_test.go index 18691fa11f..d97936c6b5 100644 --- a/apps/backend/internal/task/repository/sqlite/step_transitions_writer_postgres_test.go +++ b/apps/backend/internal/task/repository/sqlite/step_transitions_writer_postgres_test.go @@ -8,6 +8,7 @@ import ( "context" "sync" "testing" + "time" "golang.org/x/sync/errgroup" @@ -81,12 +82,34 @@ func TestPostgresLedgerWriterGenesisMoveAttachDetach(t *testing.T) { // other, breaking the (occurred_at, id) chain invariant only under real // concurrent Postgres connections (SQLite's single-writer pool serializes // regardless, which is why no SQLite test could have caught it). +// +// The two movers deliberately run on two independent *Repository instances +// backed by two independent physical connections (repo/db and moverRepo/ +// moverDB, both pinned to the same isolated schema — see +// openSecondPostgresConnection in turn_step_stamp_postgres_test.go). +// testutil.OpenIsolatedPostgres caps db at one physical connection, so if +// both UpdateTask calls below shared it, Go's connection pool alone would +// serialize them — the two calls would never reach the Postgres server at +// the same time, and this test would pass whether or not +// readTaskStepInTx's FOR UPDATE clause exists at all. Two connections make +// the calls genuinely race at the server, which is the only way this test +// can distinguish "the lock serializes movers" from "the lock is dead +// code": reintroducing the historical bug this test guards against (stamp +// occurred_at before acquiring the lock) breaks the chain here but is +// invisible to a single-shared-connection version of the same test, which +// is exactly what made the original bug undetectable before this PR. func TestPostgresLedgerWriterConcurrentMovesProduceIntactChain(t *testing.T) { - db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + dsn := testutil.PostgresDSNFromEnv(t) + db := testutil.OpenIsolatedPostgres(t, dsn) repo, err := NewWithDB(db, db, nil) if err != nil { t.Fatalf("init postgres schema: %v", err) } + moverDB := openSecondPostgresConnection(t, dsn, db) + moverRepo, err := NewWithDB(moverDB, moverDB, nil) + if err != nil { + t.Fatalf("init postgres schema on second connection: %v", err) + } ctx := steptelemetry.WithAttribution(context.Background(), steptelemetry.Attribution{ Trigger: steptelemetry.TriggerManualMove, ActorKind: steptelemetry.ActorHuman, ActorID: "user-pg-concurrent", }) @@ -112,7 +135,7 @@ func TestPostgresLedgerWriterConcurrentMovesProduceIntactChain(t *testing.T) { start.Wait() moved := *task moved.WorkflowStepID = "step-c" - return repo.UpdateTask(ctx, &moved) + return moverRepo.UpdateTask(ctx, &moved) }) start.Done() if err := g.Wait(); err != nil { @@ -132,3 +155,196 @@ func TestPostgresLedgerWriterConcurrentMovesProduceIntactChain(t *testing.T) { t.Fatalf("final workflow_step_id = %q, want step-b or step-c (last committed writer wins)", final.WorkflowStepID) } } + +// TestPostgresReadTaskStepInTxBlocksOnConcurrentRowLock is the deterministic +// counterpart to TestPostgresLedgerWriterConcurrentMovesProduceIntactChain +// above. That test's sync.WaitGroup barrier only releases both goroutines at +// the same instant — it does not force them to overlap once released, so +// the Go scheduler can (and often does) run one mover's entire transaction +// to completion before the other even begins. When that happens the second +// mover reads an already-updated step and the chain looks intact whether or +// not readTaskStepInTx's FOR UPDATE clause (step_transitions.go:28-30) does +// anything at all. Reintroducing the historical stamp-before-lock bug that +// test guards against produced only 7 failures out of 20 runs — a coin +// flip, not a control. +// +// This test instead proves the lock's actual serialization property and the +// timestamp placement directly and deterministically, mirroring the pattern +// in turn_step_stamp_postgres_test.go's +// TestPostgresCreateTurnWithStepStampBlocksOnConcurrentStepMove (same +// openSecondPostgresConnection helper): one goroutine acquires the row's +// FOR UPDATE lock via readTaskStepInTx and holds its transaction open on a +// channel; a second, independent connection then attempts a real production +// UpdateTask on the same task. The test observes that backend in PostgreSQL's +// lock-wait state before releasing the holder, so a scheduler delay cannot +// masquerade as contention. Releasing the lock lets the second writer +// proceed, and a fixed repository clock proves its ledger timestamp is +// sampled after the lock is released. +func TestPostgresReadTaskStepInTxBlocksOnConcurrentRowLock(t *testing.T) { + dsn := testutil.PostgresDSNFromEnv(t) + db := testutil.OpenIsolatedPostgres(t, dsn) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + // A second, independent connection to the same isolated schema — see + // openSecondPostgresConnection's doc comment for why a shared + // db.OpenIsolatedPostgres connection (capped at one physical connection) + // cannot exercise genuine Postgres-level lock contention. + moverDB := openSecondPostgresConnection(t, dsn, db) + observerDB := openSecondPostgresConnection(t, dsn, db) + moverRepo, err := NewWithDB(moverDB, moverDB, nil) + if err != nil { + t.Fatalf("init postgres schema on second connection: %v", err) + } + ctx := steptelemetry.WithAttribution(context.Background(), steptelemetry.Attribution{ + Trigger: steptelemetry.TriggerManualMove, ActorKind: steptelemetry.ActorHuman, ActorID: "user-pg-lock", + }) + if err := repo.CreateWorkspace(context.Background(), &models.Workspace{ID: "ws-pg-lock", Name: "PG Lock Workspace"}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + task := &models.Task{ID: "task-pg-lock", WorkspaceID: "ws-pg-lock", WorkflowID: "wf-pg-lock", WorkflowStepID: "step-a", Title: "PG Lock Task", Priority: "medium"} + if err := repo.CreateTask(ctx, task); err != nil { + t.Fatalf("CreateTask: %v", err) + } + + // The holder goroutine plays the concurrent transaction: it acquires the + // same readTaskStepInTx FOR UPDATE lock a real step move takes (on the + // first connection), then holds its transaction open — without writing + // anything — until this test explicitly releases it. It never changes + // workflow_step_id, so the mover's later UpdateTask below is the only + // real step move in this test. + lockHeld := make(chan struct{}) + releaseLock := make(chan struct{}) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseLock) }) } + holderFinished := make(chan struct{}) + var holderErr error + go func() { + defer close(holderFinished) + tx, err := db.BeginTx(ctx, nil) + if err != nil { + holderErr = err + return + } + if _, _, _, err := repo.readTaskStepInTx(ctx, tx, task.ID); err != nil { + _ = tx.Rollback() + holderErr = err + return + } + close(lockHeld) + <-releaseLock + holderErr = tx.Commit() + }() + t.Cleanup(func() { + release() + select { + case <-holderFinished: + if holderErr != nil { + t.Errorf("holder goroutine during cleanup: %v", holderErr) + } + case <-time.After(5 * time.Second): + t.Errorf("timed out waiting for holder goroutine during cleanup") + } + }) + + select { + case <-lockHeld: + case <-holderFinished: + t.Fatalf("holder goroutine failed before acquiring the row lock: %v", holderErr) + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for the holder goroutine to acquire the row lock") + } + + var moverPID int + if err := moverDB.Get(&moverPID, "SELECT pg_backend_pid()"); err != nil { + t.Fatalf("read mover backend pid: %v", err) + } + const moverStampYear = 2091 + moverStamp := time.Date(moverStampYear, time.January, 2, 3, 4, 5, 123456000, time.UTC) + stampUsed := make(chan struct{}) + var stampOnce sync.Once + moverRepo.clockNow = func() time.Time { + stampOnce.Do(func() { close(stampUsed) }) + return moverStamp + } + + // The mover runs on moverRepo/moverDB, the SECOND connection — genuinely + // independent of db, so its call below contends for the real Postgres + // row lock rather than for Go's connection pool. It is the unmodified + // production UpdateTask path, not a simulation. + moverDone := make(chan error, 1) + moverFinished := make(chan struct{}) + go func() { + defer close(moverFinished) + moved := *task + moved.WorkflowStepID = "step-b" + moverDone <- moverRepo.UpdateTask(ctx, &moved) + }() + t.Cleanup(func() { + release() + select { + case <-moverFinished: + case <-time.After(5 * time.Second): + t.Errorf("timed out waiting for mover goroutine during cleanup") + } + }) + + if err := waitForPostgresLock(ctx, observerDB, moverPID, moverFinished); err != nil { + t.Fatal(err) + } + select { + case <-moverDone: + t.Fatal("moverRepo.UpdateTask returned while a concurrent transaction still held the row's FOR UPDATE lock — readTaskStepInTx is not serializing concurrent movers") + default: + } + select { + case <-stampUsed: + t.Fatal("mover repository clock was sampled before the row lock was released") + default: + } + + release() + select { + case <-holderFinished: + if holderErr != nil { + t.Fatalf("holder goroutine: %v", holderErr) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for holder goroutine after releasing the row lock") + } + + select { + case err := <-moverDone: + if err != nil { + t.Fatalf("moverRepo.UpdateTask: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for moverRepo.UpdateTask to proceed after the lock was released") + } + select { + case <-stampUsed: + case <-time.After(5 * time.Second): + t.Fatal("mover repository clock was not sampled after the row lock was released") + } + + final, err := repo.GetTask(context.Background(), task.ID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if final.WorkflowStepID != "step-b" { + t.Fatalf("final workflow_step_id = %q, want step-b (the mover's UpdateTask committed after the holder released its lock)", final.WorkflowStepID) + } + + rows := assertChainIntact(t, repo, task.ID) + if len(rows) != 2 { + t.Fatalf("rows = %d, want 2 (genesis + the mover's real UpdateTask move; the holder never wrote anything)", len(rows)) + } + last := rows[len(rows)-1] + if last.toWorkflowStepID == nil || *last.toWorkflowStepID != "step-b" { + t.Fatalf("final to_workflow_step_id = %v, want step-b", last.toWorkflowStepID) + } + if !last.occurredAt.Equal(moverStamp) { + t.Fatalf("final occurred_at = %v, want the post-lock repository timestamp %v", last.occurredAt, moverStamp) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index fe86ce736e..2bc42c2f70 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -535,7 +535,7 @@ func (r *Repository) updateTaskTx(ctx context.Context, tx *sql.Tx, task *models. // (occurred_at, id) chain invariant under real concurrent load — SQLite's // single-writer connection pool serializes callers regardless, so this // was invisible until exercised against Postgres with real concurrency. - task.UpdatedAt = time.Now().UTC() + task.UpdatedAt = r.nowUTC() updateQuery := ` UPDATE tasks SET workspace_id = ?, workflow_id = ?, workflow_step_id = ?, title = ?, description = ?, state = ?, priority = ?, position = ?, wip_admitted = ?, queued_for_step_id = ?, queued_at = ?, metadata = ?, parent_id = ?, updated_at = ?, origin = ?, project_id = ?, labels = ?, identifier = ? diff --git a/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/const_hoisted.go.txt b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/const_hoisted.go.txt new file mode 100644 index 0000000000..f9e6ca84fc --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/const_hoisted.go.txt @@ -0,0 +1,15 @@ +package pinfixtures + +// Shape (a): a package-level const holding the mutating SQL, referenced by +// identifier in the function body. The naive detector only inspects +// *ast.BasicLit nodes found directly inside a function body, so a literal +// hoisted to a const and referenced by name puts zero matching BasicLit +// nodes there — this function is invisible to it. +const updateStepSQL = "UPDATE tasks SET workflow_step_id = ? WHERE id = ?" + +type FixtureRepo struct{} + +func (r *FixtureRepo) constHoistedStepMutator(id, step string) error { + _, err := r.exec(updateStepSQL, step, id) + return err +} diff --git a/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/name_collision.go.txt b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/name_collision.go.txt new file mode 100644 index 0000000000..3476988268 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/name_collision.go.txt @@ -0,0 +1,20 @@ +package pinfixtures + +// Shape (c): two distinct receiver types define a method with the same bare +// name. A detector keyed on fn.Name.Name alone collapses RepoA.updateTaskTx +// and RepoB.updateTaskTx into a single "updateTaskTx" entry — if one of them +// were already registered (as the real package's *Repository.updateTaskTx +// is), the other's genuinely distinct mutator would be silently absorbed +// and reported clean. +type RepoA struct{} +type RepoB struct{} + +func (r *RepoA) updateTaskTx(step, id string) error { + _, err := r.exec("UPDATE tasks SET workflow_step_id = ? WHERE id = ?", step, id) + return err +} + +func (r *RepoB) updateTaskTx(step, id string) error { + _, err := r.exec("UPDATE tasks SET workflow_step_id = ? WHERE id = ?", step, id) + return err +} diff --git a/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/nested/deep.go.txt b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/nested/deep.go.txt new file mode 100644 index 0000000000..c482c94249 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/nested/deep.go.txt @@ -0,0 +1,11 @@ +package pinfixtures + +// Shape (d): a step writer placed in a subdirectory of the scanned package. +// os.ReadDir(dir) is non-recursive, so this file is invisible unless the +// scan walks into nested directories. +type NestedRepo struct{} + +func (r *NestedRepo) deeplyNestedStepMutator(step, id string) error { + _, err := r.exec("UPDATE tasks SET workflow_step_id = ? WHERE id = ?", step, id) + return err +} diff --git a/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/sprintf_interpolated.go.txt b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/sprintf_interpolated.go.txt new file mode 100644 index 0000000000..a1216c74f3 --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/testdata/pinfixtures/sprintf_interpolated.go.txt @@ -0,0 +1,21 @@ +package pinfixtures + +import "fmt" + +// Shape (b): the column name is interpolated via fmt.Sprintf, mirroring the +// live taskScalarColumns/execTaskScalar allowlist pattern in +// internal/office/repository/sqlite/tasks.go. Neither function alone +// contains the literal string "workflow_step_id" adjacent to "UPDATE +// tasks"/"SET" the way the naive detector's substring check requires: the +// SQL shape lives in execScalarColumn, and the mutating column name is only +// ever a plain string argument at the call site in +// sprintfInterpolatedStepMutator. +func (r *FixtureRepo) execScalarColumn(column, value, id string) error { + query := fmt.Sprintf("UPDATE tasks SET %s = ?, updated_at = ? WHERE id = ?", column) + _, err := r.exec(query, value, id) + return err +} + +func (r *FixtureRepo) sprintfInterpolatedStepMutator(id, step string) error { + return r.execScalarColumn("workflow_step_id", step, id) +} diff --git a/apps/backend/internal/task/repository/sqlite/turn_step_stamp_postgres_test.go b/apps/backend/internal/task/repository/sqlite/turn_step_stamp_postgres_test.go index c98bc5f558..99525bedbd 100644 --- a/apps/backend/internal/task/repository/sqlite/turn_step_stamp_postgres_test.go +++ b/apps/backend/internal/task/repository/sqlite/turn_step_stamp_postgres_test.go @@ -31,6 +31,7 @@ package sqlite import ( "context" + "fmt" "sync" "testing" "time" @@ -55,6 +56,8 @@ func openSecondPostgresConnection(t *testing.T, dsn string, db *sqlx.DB) *sqlx.D if err != nil { t.Fatalf("open second postgres connection: %v", err) } + second.SetMaxOpenConns(1) + second.SetMaxIdleConns(1) t.Cleanup(func() { _ = second.Close() }) if _, err := second.Exec("SET search_path TO " + schema); err != nil { t.Fatalf("set search_path on second connection: %v", err) @@ -62,6 +65,53 @@ func openSecondPostgresConnection(t *testing.T, dsn string, db *sqlx.DB) *sqlx.D return second } +func postgresBackendWaitsOnLock(ctx context.Context, db *sqlx.DB, pid int) (bool, error) { + var waiting bool + err := db.GetContext(ctx, &waiting, ` + SELECT EXISTS ( + SELECT 1 + FROM pg_stat_activity + WHERE pid = $1 AND wait_event_type = 'Lock' + ) + `, pid) + return waiting, err +} + +// waitForPostgresLock waits until the backend identified by pid reports a +// PostgreSQL lock wait. The done channel lets callers fail immediately when +// the operation completes without ever reaching the expected contention. +func waitForPostgresLock(ctx context.Context, observer *sqlx.DB, pid int, done <-chan struct{}) error { + const timeout = 5 * time.Second + deadlineCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-done: + return fmt.Errorf("operation completed before backend %d entered a PostgreSQL lock wait", pid) + default: + } + + waiting, err := postgresBackendWaitsOnLock(deadlineCtx, observer, pid) + if err != nil { + return fmt.Errorf("query PostgreSQL lock state for backend %d: %w", pid, err) + } + if waiting { + return nil + } + + select { + case <-done: + return fmt.Errorf("operation completed before backend %d entered a PostgreSQL lock wait", pid) + case <-ticker.C: + case <-deadlineCtx.Done(): + return fmt.Errorf("timed out waiting for backend %d to enter a PostgreSQL lock wait", pid) + } + } +} + func TestPostgresCreateTurnWithStepStampBlocksOnConcurrentStepMove(t *testing.T) { dsn := testutil.PostgresDSNFromEnv(t) db := testutil.OpenIsolatedPostgres(t, dsn) @@ -74,6 +124,7 @@ func TestPostgresCreateTurnWithStepStampBlocksOnConcurrentStepMove(t *testing.T) // A second, independent connection to the same isolated schema — see the // file comment for why this is required rather than reusing db. moverDB := openSecondPostgresConnection(t, dsn, db) + observerDB := openSecondPostgresConnection(t, dsn, db) if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-pg-turn-stamp", Name: "Workspace"}); err != nil { t.Fatalf("CreateWorkspace: %v", err) @@ -106,9 +157,10 @@ func TestPostgresCreateTurnWithStepStampBlocksOnConcurrentStepMove(t *testing.T) // forever on <-releaseLock, holding its transaction open, and // t.Cleanup's DROP SCHEMA hangs waiting for that lock instead of // reporting the actual test failure. - defer release() moverDone := make(chan error, 1) + moverFinished := make(chan struct{}) go func() { + defer close(moverFinished) tx, err := moverDB.BeginTx(ctx, nil) if err != nil { moverDone <- err @@ -128,13 +180,28 @@ func TestPostgresCreateTurnWithStepStampBlocksOnConcurrentStepMove(t *testing.T) } moverDone <- tx.Commit() }() + t.Cleanup(func() { + release() + select { + case <-moverFinished: + case <-time.After(5 * time.Second): + t.Errorf("timed out waiting for mover goroutine during cleanup") + } + }) select { case <-lockHeld: + case err := <-moverDone: + t.Fatalf("mover goroutine exited before acquiring the row lock: %v", err) case <-time.After(5 * time.Second): t.Fatal("timed out waiting for the mover goroutine to acquire the row lock") } + turnPID := 0 + if err := db.Get(&turnPID, "SELECT pg_backend_pid()"); err != nil { + t.Fatalf("read turn backend pid: %v", err) + } + // The turn goroutine runs on repo/db, the FIRST connection — genuinely // independent of moverDB, so its read below contends for the real // Postgres row lock rather than for Go's connection pool. @@ -146,12 +213,22 @@ func TestPostgresCreateTurnWithStepStampBlocksOnConcurrentStepMove(t *testing.T) stamped, turnErr = repo.CreateTurnWithStepStamp(ctx, turn) close(turnDone) }() + t.Cleanup(func() { + release() + select { + case <-turnDone: + case <-time.After(5 * time.Second): + t.Errorf("timed out waiting for turn goroutine during cleanup") + } + }) + if err := waitForPostgresLock(ctx, observerDB, turnPID, turnDone); err != nil { + t.Fatal(err) + } select { case <-turnDone: t.Fatal("CreateTurnWithStepStamp returned while the concurrent mover still held the row lock — the step read is not serialized against concurrent movers") - case <-time.After(300 * time.Millisecond): - // Still blocked, as expected: the row lock is held. + default: } release() diff --git a/apps/backend/internal/telemetrycontract/health_test.go b/apps/backend/internal/telemetrycontract/health_test.go index bdaf72936c..52a76fc049 100644 --- a/apps/backend/internal/telemetrycontract/health_test.go +++ b/apps/backend/internal/telemetrycontract/health_test.go @@ -3,6 +3,7 @@ package telemetrycontract import ( "context" "testing" + "time" "go.uber.org/zap" "go.uber.org/zap/zapcore" @@ -145,6 +146,42 @@ func TestLogHealthReportsTaskStepTransitionsTableAndRowCount(t *testing.T) { } } +func TestLogHealthReportsActivatedAt(t *testing.T) { + // Activate first so telemetry_activations has a row for every registered + // contract, then assert LogHealth actually surfaces it: activated_at is + // built from contractHealth.activatedAt and only appended to the log + // fields when non-nil (store.go:96-98), but until this test nothing read + // the field back, so a regression that stopped populating it (e.g. an + // error swallowed by activatedAt, or the field silently dropped from the + // zap.Field slice) would leave the whole suite green. + db := memDB(t) + store, err := NewWithDB(db, db) + if err != nil { + t.Fatalf("NewWithDB: %v", err) + } + if err := store.Activate(context.Background()); err != nil { + t.Fatalf("Activate: %v", err) + } + + log, logs := observerLogger(t) + store.LogHealth(context.Background(), log) + + entries := logs.FilterMessage("telemetry.contract.health").All() + if len(entries) != len(Registry()) { + t.Fatalf("health lines = %d, want %d", len(entries), len(Registry())) + } + for i, c := range Registry() { + ctxMap := entries[i].ContextMap() + activatedAt, ok := ctxMap["activated_at"].(time.Time) + if !ok { + t.Fatalf("entry %d (%s) activated_at missing or wrong type: %v", i, c.Key, ctxMap["activated_at"]) + } + if activatedAt.IsZero() { + t.Fatalf("entry %d (%s) activated_at is zero, want a real activation timestamp", i, c.Key) + } + } +} + func TestLogHealthNoopWhenLoggerNil(t *testing.T) { db := memDB(t) store, err := NewWithDB(db, db) diff --git a/docs/plans/task-step-transition-ledger/plan.md b/docs/plans/task-step-transition-ledger/plan.md index 3b6cd4c04f..cb3c8823cf 100644 --- a/docs/plans/task-step-transition-ledger/plan.md +++ b/docs/plans/task-step-transition-ledger/plan.md @@ -434,7 +434,7 @@ No store slice, hook, component, or API client changes. | What | File | How | |---|---|---| -| A new production statement mutating `tasks.workflow_step_id` fails the build until registered | `internal/task/repository/sqlite/step_transition_writers_pin_test.go` (new) | `go/parser` over the package's non-test files; collect functions whose body contains a string literal matching `INSERT INTO tasks` or an `UPDATE tasks ... workflow_step_id` assignment; assert set equality against the seven registered names | +| A new production statement mutating `tasks.workflow_step_id` fails the build until registered | `internal/task/repository/sqlite/step_transition_writers_pin_test.go` (new) | `go/parser` scans `apps/backend/internal` recursively; resolve package constants and `fmt.Sprintf` column setters, qualify identities by package and receiver, and assert set equality against the registered names | | A ledger row bumps the expvar counter for its trigger and logs `telemetry.metric.step_transition_written` | `internal/steptelemetry/metrics_test.go` (new) | Read the expvar map; assert with `zaptest`'s observed logs | | A stamped/unstamped turn bumps the turn counter and logs `telemetry.metric.turn_stamped` | same | Both branches | | Boot emits one health line per registered contract with existence, activation, count, recency | `internal/telemetrycontract/health_test.go` (new) | Observed logs against a seeded DB | @@ -626,6 +626,107 @@ and this step's mechanical file-diff check independently confirms zero --- +## Review Round 1 / Round 2 — test-rigor follow-up on the writer-health pinning gate + +A later Kandev task (`fix-mcp-deferred-mov_428h3jqo`) opened a follow-up card +strengthening `TestStepTransitionWritersArePinned` and its supporting +detector (`findWorkflowStepIDMutators` in +`step_transition_writers_pin_test.go`) against seven mutation-proven false +negatives, then ran two adversarial review rounds against the fix (a +Kandev-internal review leg plus an independent cross-vendor `codex` leg each +round). This entry is the durable receipt Review Round 1 required and Review +Round 2 confirmed was still missing from this file after Build round 2's +commit message (`84a466d44`) claimed — inaccurately — that it had already +been recorded here. Recording it now closes that gap. + +**What Round 1 found and Round 2 confirmed fixed:** +1. The production scan was rooted at this package's own directory (`"."`), + not the whole backend source tree, so recursion into subdirectories bought + the gate nothing against a writer added in a sibling package. Fixed by + `findBackendSourceRoot`, which walks up to `apps/backend/internal`. +2. The Postgres concurrency test relied on a `sync.WaitGroup` barrier, which + guarantees simultaneous start but not actual lock contention — reintroducing + the historical bug it exists to catch produced only 7/20 failures (~35%). + Fixed by `TestPostgresReadTaskStepInTxBlocksOnConcurrentRowLock`, a + channel-gated deterministic lock-hold test mirroring + `turn_step_stamp_postgres_test.go`'s existing pattern. The test now polls + `pg_stat_activity` to confirm that the competing backend is waiting on the + row lock, and it uses the repository clock seam to prove the transition + timestamp is sampled after lock release; the same reintroduced bug fails + 5/5. +3. (Round 2, new) `funcIdentity()` keyed registered writers as + `"ReceiverType.FuncName"` with no package qualification. Since this + codebase declares a type literally named `Repository` in 11 different + backend packages, a same-named method in any of them collided with an + already-registered entry and the gate missed it. Fixed by qualifying the + identity with the writer's package directory relative to the scan root + (`"task/repository/sqlite/Repository.updateTaskTx"`), plus a permanent + regression fixture (`TestFindWorkflowStepIDMutatorsQualifiesByPackage`) + proving two packages with colliding receiver-type-and-method names are now + reported as two distinct writers, not collapsed into one. + +**Both-ways (fail-then-pass) receipt for the evasive-shapes fixture test,** +run against the actual pre-fix detector, not reconstructed from memory: + +``` +$ git show 9fe867f9f:apps/backend/internal/task/repository/sqlite/step_transition_writers_pin_test.go \ + > /tmp/pin_old_detector_source.go +# extracted the pre-fix functionMutatesWorkflowStepID/literalMutatesWorkflowStepID +# (BasicLit-only — no Sprintf handling, no const resolution, bare-name keys — +# i.e. exactly origin/main's shape before this card's fixes) into a standalone +# `go run` program and ran it against the current three fixture files. +$ go run /tmp/pin_old_detector_repro/main.go +found["constHoistedStepMutator"] = false # (a) unreachable: literal is const-hoisted, old detector only walked BasicLit +found["sprintfInterpolatedStepMutator"] = false # (b) unreachable: old detector had no Sprintf handling at all +found["updateTaskTx"] = true # (c) RepoA and RepoB collapse into one bare-name key — a duplicate masked as a single hit, not a miss +``` + +3 of the 5 fixture shapes were genuinely undetectable (or miscounted) under +the pre-fix detector, confirming the fixtures encode real false negatives, +not hypothetical ones. Post-fix, all 5 pass — +`TestFindWorkflowStepIDMutatorsCatchesEvasiveShapes`, run fresh: + +``` +$ go test ./internal/task/repository/sqlite/ -run TestFindWorkflowStepIDMutatorsCatchesEvasiveShapes -v -count=1 +--- PASS: TestFindWorkflowStepIDMutatorsCatchesEvasiveShapes (0.00s) +``` + +**Package-qualification receipt (Round 2, Finding A), live decoy in a real +backend package, not a synthetic fixture:** + +``` +$ cat > internal/office/repository/sqlite/zz_review_collision_tmp.go # package sqlite, a DIFFERENT +package sqlite # backend package than the real + # writer's internal/task/repository/sqlite +func (r *Repository) updateTaskTx() string { + return "UPDATE tasks SET workflow_step_id = ? WHERE id = ?" +} + +$ go build ./internal/office/... # clean +$ go test ./internal/task/repository/sqlite/ -run TestStepTransitionWritersArePinned -v -count=1 +--- FAIL: TestStepTransitionWritersArePinned + function(s) [office/repository/sqlite/Repository.updateTaskTx] contain a + statement that mutates tasks.workflow_step_id but are not registered... +$ rm internal/office/repository/sqlite/zz_review_collision_tmp.go +$ git status --porcelain internal/office/ internal/task/ # empty — clean revert +``` + +Before the Round 2 fix this exact decoy passed silently (the collision with +the already-registered `"Repository.updateTaskTx"` entry masked it); after +the fix it is reported under its own package-qualified identity and fails +the gate as intended. + +**Process note, disclosed rather than omitted:** while drafting this card's +own Kandev task plan during Round 2, the reviewer wrote a sentence claiming +the evasive-shapes both-ways receipt above had already been re-verified +before that verification had actually been run — the same class of +unsubstantiated claim this section exists to prevent. It was caught before +being reported anywhere external, and the real verification (shown above) +was performed immediately after. Recorded here so the fix's provenance is +honest rather than convenient. + +--- + ## Implementation Waves And Parallel Candidates Sequential by default. The two `parallel-safe` labels below are the only tasks