fix(data): close project governance consistency gaps

Filtered watch pushes must stay bounded by the changed batch, accepted reclassification previews must describe the exact write they authorize, and derived mirrors must retain the machine identity used by mapping rules. Without those guarantees, large archives regress over time and project governance can either apply unseen changes or report the wrong result across machines.

Author: wesm

internal/db/worktree_reclassification.go

@@ -14,7 +14,9 @@ import (
 
 const worktreeReclassificationSampleLimit = 10
 
-var ErrWorktreeMappingSetChanged = errors.New("worktree mapping set changed")
+var ErrWorktreeMappingSetChanged = errors.New(
+	"worktree reclassification preview changed",
+)
 
 type WorktreeReclassificationDraft struct {
 	Machine         string `json:"machine"`
@@ -49,10 +51,11 @@ type WorktreeReclassificationPreview struct {
 }
 
 type worktreeReclassificationEvaluation struct {
-	matched  int
-	updates  []worktreeMappingSessionUpdate
-	projects map[string]int
-	sessions []WorktreeReclassificationSessionSample
+	matched     int
+	updates     []worktreeMappingSessionUpdate
+	projects    map[string]int
+	sessions    []WorktreeReclassificationSessionSample
+	impactToken string
 }
 
 func (db *DB) PreviewWorktreeReclassification(
@@ -120,8 +123,18 @@ func (db *DB) ApplyWorktreeReclassification(
 	if err != nil {
 		return WorktreeProjectMapping{}, WorktreeReclassificationPreview{}, err
 	}
-	currentToken := worktreeMappingSetToken(stored)
 	collision := exactWorktreeMapping(stored, normalized.PathPrefix)
+	effective := overlayWorktreeMapping(stored, normalized, collision)
+	evaluation, err := evaluateWorktreeMappingsTx(
+		ctx, tx, normalized.Machine, enabledWorktreeMappings(effective),
+		&normalized, "",
+	)
+	if err != nil {
+		return WorktreeProjectMapping{}, WorktreeReclassificationPreview{}, err
+	}
+	currentToken := worktreeReclassificationToken(
+		stored, normalized, collision, evaluation,
+	)
 	if acceptedToken == "" || acceptedToken != currentToken ||
 		!sameOptionalMappingID(existingMappingID, mappingIDPointer(collision)) {
 		return WorktreeProjectMapping{}, WorktreeReclassificationPreview{},
@@ -134,18 +147,6 @@ func (db *DB) ApplyWorktreeReclassification(
 	if err != nil {
 		return WorktreeProjectMapping{}, WorktreeReclassificationPreview{}, err
 	}
-	active, err := loadActiveWorktreeMappingsTx(ctx, tx, normalized.Machine)
-	if err != nil {
-		return WorktreeProjectMapping{}, WorktreeReclassificationPreview{}, fmt.Errorf(
-			"loading active worktree mappings: %w", err,
-		)
-	}
-	evaluation, err := evaluateWorktreeMappingsTx(
-		ctx, tx, normalized.Machine, active, &mapping, "",
-	)
-	if err != nil {
-		return WorktreeProjectMapping{}, WorktreeReclassificationPreview{}, err
-	}
 	affected := map[string]struct{}{mapping.Project: {}}
 	updated := 0
 	for _, update := range evaluation.updates {
@@ -171,7 +172,7 @@ func (db *DB) ApplyWorktreeReclassification(
 	}
 
 	preview := worktreeReclassificationPreviewFromEvaluation(
-		worktreeMappingSetTokenWithReplacement(stored, mapping),
+		acceptedToken,
 		mapping.Project, mappingIDPointer(&mapping), evaluation,
 	)
 	preview.UpdatedSessions = updated
@@ -210,7 +211,8 @@ func previewWorktreeReclassificationTx(
 		return WorktreeReclassificationPreview{}, err
 	}
 	return worktreeReclassificationPreviewFromEvaluation(
-		worktreeMappingSetToken(stored), draft.Project,
+		worktreeReclassificationToken(stored, draft, collision, evaluation),
+		draft.Project,
 		mappingIDPointer(collision), evaluation,
 	), nil
 }
@@ -353,6 +355,8 @@ func evaluateWorktreeMappingsTx(
 		},
 	)
 
+	impactHash := sha256.New()
+	writeWorktreeTokenFields(impactHash, "impact-v1")
 	evaluation := worktreeReclassificationEvaluation{projects: map[string]int{}}
 	for _, row := range sessions {
 		if sessionID != "" && row.id != sessionID {
@@ -375,6 +379,18 @@ func evaluateWorktreeMappingsTx(
 			continue
 		}
 		evaluation.matched++
+		matchCwd := row.matchCwd
+		if matchCwd == "" {
+			matchCwd = row.cwd
+		}
+		nextProject := row.project
+		if shouldUpdate {
+			nextProject = update.nextProject
+		}
+		writeWorktreeTokenFields(
+			impactHash, row.id, row.project, nextProject, row.cwd,
+			matchCwd, row.filePath,
+		)
 		if !shouldUpdate {
 			continue
 		}
@@ -386,6 +402,7 @@ func evaluateWorktreeMappingsTx(
 				NextProject: update.nextProject, Cwd: update.cwd,
 			})
 	}
+	evaluation.impactToken = hex.EncodeToString(impactHash.Sum(nil))
 	return evaluation, nil
 }
 
@@ -428,22 +445,41 @@ func worktreeMappingSetToken(mappings []WorktreeProjectMapping) string {
 		fields := []string{
 			strconv.FormatInt(mapping.ID, 10), mapping.Machine,
 			normalizedMappingPath(mapping.PathPrefix), mapping.Layout,
-			mapping.Project, strconv.FormatBool(mapping.Enabled), mapping.UpdatedAt,
-		}
-		for _, field := range fields {
-			_, _ = fmt.Fprintf(hash, "%d:%s", len(field), field)
+			mapping.Project, mapping.OriginalProject,
+			strconv.FormatBool(mapping.Enabled), mapping.UpdatedAt,
 		}
+		writeWorktreeTokenFields(hash, fields...)
 	}
 	return hex.EncodeToString(hash.Sum(nil))
 }
 
-func worktreeMappingSetTokenWithReplacement(
+func worktreeReclassificationToken(
 	stored []WorktreeProjectMapping,
-	mapping WorktreeProjectMapping,
+	draft WorktreeProjectMapping,
+	collision *WorktreeProjectMapping,
+	evaluation worktreeReclassificationEvaluation,
 ) string {
-	return worktreeMappingSetToken(overlayWorktreeMapping(
-		stored, mapping, exactWorktreeMapping(stored, mapping.PathPrefix),
-	))
+	collisionID := ""
+	if collision != nil {
+		collisionID = strconv.FormatInt(collision.ID, 10)
+	}
+	hash := sha256.New()
+	writeWorktreeTokenFields(
+		hash, "reclassification-v2", worktreeMappingSetToken(stored),
+		draft.Machine, normalizedMappingPath(draft.PathPrefix), draft.Layout,
+		draft.Project, draft.OriginalProject, strconv.FormatBool(draft.Enabled),
+		collisionID, evaluation.impactToken,
+	)
+	return hex.EncodeToString(hash.Sum(nil))
+}
+
+func writeWorktreeTokenFields(
+	hash interface{ Write([]byte) (int, error) },
+	fields ...string,
+) {
+	for _, field := range fields {
+		_, _ = fmt.Fprintf(hash, "%d:%s", len(field), field)
+	}
 }
 
 func mappingIDPointer(mapping *WorktreeProjectMapping) *int64 {

internal/db/worktree_reclassification_test.go

@@ -68,7 +68,7 @@ func TestWorktreeReclassificationPreviewHonorsSpecificRuleAndBoundsSamples(t *te
 		"the specific mapping must remain authoritative")
 }
 
-func TestWorktreeReclassificationTokenTracksMappingsButNotSessions(t *testing.T) {
+func TestWorktreeReclassificationTokenBindsDraftAndAffectedSessions(t *testing.T) {
 	d := testDB(t)
 	ctx := context.Background()
 	seedReclassificationSession(t, d, "one", "archive.example", "/worktrees/service/one", "branch")
@@ -79,14 +79,29 @@ func TestWorktreeReclassificationTokenTracksMappingsButNotSessions(t *testing.T)
 
 	preview, err := d.PreviewWorktreeReclassification(ctx, draft)
 	require.NoError(t, err)
+	changedDraft := draft
+	changedDraft.Project = "different-service"
+	_, _, err = d.ApplyWorktreeReclassification(
+		ctx, changedDraft, preview.MappingToken, preview.ExistingMappingID,
+	)
+	require.ErrorIs(t, err, ErrWorktreeMappingSetChanged,
+		"a preview for one normalized draft must not authorize another")
+
 	seedReclassificationSession(t, d, "two", "archive.example", "/worktrees/service/two", "branch")
-	mapping, applied, err := d.ApplyWorktreeReclassification(
+	_, _, err = d.ApplyWorktreeReclassification(
 		ctx, draft, preview.MappingToken, preview.ExistingMappingID,
 	)
+	require.ErrorIs(t, err, ErrWorktreeMappingSetChanged,
+		"a newly affected session must invalidate the accepted preview")
+
+	current, err := d.PreviewWorktreeReclassification(ctx, draft)
+	require.NoError(t, err)
+	mapping, applied, err := d.ApplyWorktreeReclassification(
+		ctx, draft, current.MappingToken, current.ExistingMappingID,
+	)
 	require.NoError(t, err)
 	assert.Equal(t, "branch", mapping.OriginalProject)
-	assert.Equal(t, 2, applied.UpdatedSessions,
-		"sessions arriving after preview are included")
+	assert.Equal(t, 2, applied.UpdatedSessions)
 
 	stalePreview, err := d.PreviewWorktreeReclassification(ctx, WorktreeReclassificationDraft{
 		Machine: "other.example", PathPrefix: "/worktrees/service",

internal/duckdb/project_inventory_test.go

@@ -44,30 +44,17 @@ func seedInventorySession(
 	}}), "InsertMessages")
 }
 
-// duckPushMachine is the fixed machine name the DuckDB sync harness stamps
-// onto every pushed session (internal/duckdb/sync.go's newTestSync passes
-// this literal to New). Unlike PostgreSQL's push (which preserves each
-// session's own Machine field via pushedSessionMachine, falling back only
-// for "local"/empty sentinels), DuckDB's push.go unconditionally writes
-// s.machine onto every mirrored session row -- worktree mapping rows are
-// not affected, since mapping publication mirrors the Machine field
-// verbatim. Fixtures pushed through the sync harness must therefore set
-// every session's Machine (and every governing mapping's Machine) to this
-// value, or governance evaluation will silently never match after push.
+// duckPushMachine is the fallback machine name used by the DuckDB sync
+// harness for sessions whose source machine is empty or the "local" sentinel.
 const duckPushMachine = "test-machine"
 
 // buildInventoryFixture seeds the shared alpha/beta/gamma/misc aggregate
 // fixture used by the DuckDB inventory tests: distinct and empty cwds, a
 // trashed session excluded from every count, and three mapping rules that
 // together exercise every branch of annotateProjectInventoryRows. It
-// mirrors internal/postgres's buildInventoryFixture in spirit, but every
-// session and every governing mapping shares duckPushMachine (see above);
-// alpha-2 is excluded from governance by a non-matching cwd prefix instead
-// of a different machine, since DuckDB's push collapses per-session
-// machine identity and so cannot express a same-push cross-machine
-// exclusion the way PostgreSQL's fixture does (that scoping is instead
-// covered by TestDuckProjectInventoryCrossArchiveIsolation's hand-inserted
-// rows, which bypass the push collapse).
+// mirrors internal/postgres's buildInventoryFixture in spirit. Every session
+// and governing mapping happens to share duckPushMachine; multi-machine push
+// behavior is covered separately below.
 func buildInventoryFixture(t *testing.T, local *db.DB, ctx context.Context) {
 	t.Helper()
 	seedInventorySession(t, local, "alpha-1", "alpha", func(s *db.Session) {
@@ -297,16 +284,62 @@ func TestDuckProjectInventoryIgnoresUnattributedSessions(t *testing.T) {
 		"alpha's session count is unchanged")
 }
 
+func TestDuckPushPreservesSessionMachineForGovernance(t *testing.T) {
+	ctx := context.Background()
+	local := newLocalDB(t)
+	for _, fixture := range []struct {
+		id      string
+		project string
+		machine string
+		cwd     string
+	}{
+		{id: "host-a-session", project: "alpha", machine: "host-a", cwd: "/repos/alpha"},
+		{id: "host-b-session", project: "beta", machine: "host-b", cwd: "/repos/beta"},
+	} {
+		seedInventorySession(t, local, fixture.id, fixture.project, func(s *db.Session) {
+			s.Machine = fixture.machine
+			s.Cwd = fixture.cwd
+		})
+		_, err := local.CreateWorktreeProjectMapping(ctx, db.WorktreeProjectMapping{
+			Machine: fixture.machine, PathPrefix: fixture.cwd,
+			Layout:  db.WorktreeMappingLayoutExplicit,
+			Project: fixture.project, Enabled: true,
+		})
+		require.NoError(t, err)
+	}
+
+	syncer := newInMemoryTestSync(t, local, SyncOptions{})
+	pushDataReadMirror(t, ctx, syncer)
+
+	rows, err := syncer.DB().QueryContext(ctx,
+		`SELECT id, machine FROM sessions ORDER BY id`)
+	require.NoError(t, err)
+	defer rows.Close()
+	machines := map[string]string{}
+	for rows.Next() {
+		var id, machine string
+		require.NoError(t, rows.Scan(&id, &machine))
+		machines[id] = machine
+	}
+	require.NoError(t, rows.Err())
+	assert.Equal(t, map[string]string{
+		"host-a-session": "host-a",
+		"host-b-session": "host-b",
+	}, machines)
+
+	inventory, err := NewStoreFromDB(syncer.DB()).GetProjectInventory(ctx)
+	require.NoError(t, err)
+	assert.Equal(t, 2, inventory.GovernedSessions,
+		"each mirrored session must join the mapping from its source machine")
+}
+
 // TestDuckProjectInventoryCrossArchiveIsolation verifies that inventory
 // governance is scoped per source archive AND per machine when a DuckDB
 // mirror serves more than one archive. It pushes one archive through the
 // real sync path (archive A) and hand-inserts a second archive's mapping
 // and session rows directly into the DuckDB mirror (archive B), following
-// the pattern in TestDuckPushReplicatesWorktreeMappings. Archive A's
-// session ends up with machine=duckPushMachine after push (DuckDB's push
-// collapses every session's machine field to the syncing machine; see
-// duckPushMachine's doc comment), so archive B's hand-inserted rows
-// deliberately reuse that same machine name and path prefix, isolating
+// the pattern in TestDuckPushReplicatesWorktreeMappings. Both archives
+// deliberately use the same machine name and path prefix, isolating
 // source_archive_id as the only variable between archive A and B: a broken
 // (source_archive_id, machine) scope -- either in the candidate-row SQL or
 // in how projectInventoryMappings groups rules by archive -- would
@@ -324,6 +357,7 @@ func TestDuckProjectInventoryCrossArchiveIsolation(t *testing.T) {
 	// Archive A (the local push archive) has no worktree mapping of its
 	// own on duckPushMachine.
 	seedInventorySession(t, local, "a-session", "proj-a", func(s *db.Session) {
+		s.Machine = duckPushMachine
 		s.Cwd = "/repos/shared"
 		s.StartedAt = new("2024-01-01T00:00:00Z")
 	})
@@ -336,10 +370,10 @@ func TestDuckProjectInventoryCrossArchiveIsolation(t *testing.T) {
 		`SELECT machine FROM sessions WHERE id = 'a-session'`,
 	).Scan(&aSessionMachine), "read back a-session machine")
 	require.Equal(t, duckPushMachine, aSessionMachine,
-		"push must stamp every session with the syncing machine name")
+		"push must preserve an explicit source machine")
 
 	// Archive B: hand-inserted mirror rows for a second source archive,
-	// reusing the same (collapsed) machine name and path prefix as archive
+	// reusing the same machine name and path prefix as archive
 	// A's session, plus its own governed session.
 	const archiveB = "archive-b"
 	_, err := syncer.DB().ExecContext(ctx, `

internal/duckdb/push.go

@@ -915,9 +915,15 @@ func (s *Sync) upsertSession(
 	return nil
 }
 
-func sessionInsertArgs(sess db.Session, machine, archiveID, fingerprint string) []any {
+func sessionInsertArgs(
+	sess db.Session,
+	fallbackMachine string,
+	archiveID string,
+	fingerprint string,
+) []any {
 	return []any{
-		sess.ID, sess.Project, machine, sess.Agent,
+		sess.ID, sess.Project,
+		mirroredSessionMachine(sess, fallbackMachine), sess.Agent,
 		sess.AgentLabel, sess.Entrypoint,
 		nilString(sess.FirstMessage), nilString(sess.DisplayName),
 		nilString(sess.SessionName),
@@ -953,6 +959,16 @@ func sessionInsertArgs(sess db.Session, machine, archiveID, fingerprint string)
 	}
 }
 
+// mirroredSessionMachine preserves the source archive's machine identity.
+// "local" and empty are local-only sentinels, so only those use the machine
+// configured for this mirror push.
+func mirroredSessionMachine(sess db.Session, fallbackMachine string) string {
+	if sess.Machine != "" && sess.Machine != "local" {
+		return sess.Machine
+	}
+	return fallbackMachine
+}
+
 func insertMessages(
 	ctx context.Context, exec duckMutationExecutor, msgs []db.Message,
 ) error {

internal/duckdb/sync.go

@@ -1002,7 +1002,7 @@ func (s *Sync) sessionFingerprints(
 // quality analytics stale until the next full rebuild.
 func duckSessionFingerprintFields(sess db.Session, machine string) []any {
 	return []any{
-		sess.ID, sess.Project, machine, sess.Agent,
+		sess.ID, sess.Project, mirroredSessionMachine(sess, machine), sess.Agent,
 		sess.AgentLabel, sess.Entrypoint,
 		nilString(sess.FirstMessage), nilString(sess.DisplayName),
 		nilString(sess.SessionName),