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 <carlosmflorencio@gmail.com>

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
+}

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

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 {

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

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)
+	}
+}

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 = ?

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
+}

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
+}

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
+}

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)
+}

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()

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)

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