fix(data): scope published identity by session

Filtered mirrors must follow a session’s current project without rewriting its immutable source identity. Otherwise a project move can publish the session without its snapshot, retain evidence after the session leaves scope, or turn an uninspected cwd placeholder into authoritative project evidence.

Rebuild filtered identity from destination state and reconcile only changed session IDs between full publications. This keeps watcher work bounded while avoiding transmission of excluded-project tombstones.

Author: wesm

internal/db/project_identity.go

@@ -95,7 +95,8 @@ func (db *DB) LoadProjectIdentityPublicationDelta(
 	}
 
 	where, args := projectIdentityPublicationChangeWhere(
-		"c", afterRevision, throughRevision, projects, excludeProjects,
+		"c", "c.project", afterRevision, throughRevision,
+		projects, excludeProjects,
 	)
 	rows, err := db.getReader().QueryContext(ctx, `
 		SELECT o.source_archive_id, o.source_archive_salt,
@@ -168,6 +169,10 @@ func (db *DB) LoadProjectIdentityPublicationDelta(
 		return delta, fmt.Errorf("closing project identity observation tombstones: %w", err)
 	}
 
+	snapshotWhere, snapshotArgs := projectIdentityPublicationChangeWhere(
+		"c", "owner.project", afterRevision, throughRevision,
+		projects, excludeProjects,
+	)
 	rows, err = db.getReader().QueryContext(ctx, `
 		SELECT s.session_id, s.project, s.machine, s.root_path, s.git_remote,
 			s.git_remote_name, s.repository_path, s.worktree_name,
@@ -177,8 +182,11 @@ func (db *DB) LoadProjectIdentityPublicationDelta(
 		FROM session_project_identity_snapshot_changes c
 		JOIN session_project_identity_snapshots s
 		  ON s.session_id = c.session_id AND s.project = c.project
-		`+where+` AND c.deleted = 0
-		ORDER BY c.session_id, c.project`, args...)
+		JOIN sessions owner
+		  ON owner.id = s.session_id AND owner.deleted_at IS NULL
+		`+snapshotWhere+` AND c.deleted = 0
+		  AND (TRIM(s.key_source) != '' OR TRIM(s.worktree_root_path) != '')
+		ORDER BY c.session_id, c.project`, snapshotArgs...)
 	if err != nil {
 		return delta, fmt.Errorf("listing changed session project identity snapshots: %w", err)
 	}
@@ -213,11 +221,15 @@ func (db *DB) LoadProjectIdentityPublicationDelta(
 		return delta, fmt.Errorf("closing changed session project identity snapshots: %w", err)
 	}
 
+	tombstoneWhere, tombstoneArgs := projectIdentityPublicationChangeWhere(
+		"c", "c.project", afterRevision, throughRevision,
+		projects, excludeProjects,
+	)
 	rows, err = db.getReader().QueryContext(ctx, `
 		SELECT c.session_id, c.project
 		FROM session_project_identity_snapshot_changes c
-		`+where+` AND c.deleted = 1
-		ORDER BY c.session_id, c.project`, args...)
+		`+tombstoneWhere+` AND c.deleted = 1
+		ORDER BY c.session_id, c.project`, tombstoneArgs...)
 	if err != nil {
 		return delta, fmt.Errorf("listing session project identity snapshot tombstones: %w", err)
 	}
@@ -236,11 +248,13 @@ func (db *DB) LoadProjectIdentityPublicationDelta(
 }
 
 func projectIdentityPublicationChangeWhere(
-	alias string,
+	revisionAlias string,
+	projectExpression string,
 	afterRevision, throughRevision int64,
 	projects, excludeProjects []string,
 ) (string, []any) {
-	where := "WHERE " + alias + ".revision > ? AND " + alias + ".revision <= ?"
+	where := "WHERE " + revisionAlias + ".revision > ? AND " +
+		revisionAlias + ".revision <= ?"
 	args := []any{afterRevision, throughRevision}
 	appendProjects := func(values []string, negate bool) {
 		if len(values) == 0 {
@@ -255,7 +269,7 @@ func projectIdentityPublicationChangeWhere(
 		if negate {
 			op = " NOT IN "
 		}
-		where += " AND " + alias + ".project" + op +
+		where += " AND " + projectExpression + op +
 			"(" + strings.Join(placeholders, ",") + ")"
 	}
 	appendProjects(projects, false)
@@ -1735,6 +1749,123 @@ func (db *DB) ListSessionProjectIdentitySnapshots(
 	return out, nil
 }
 
+// ListPublishableSessionProjectIdentitySnapshots returns authoritative source
+// snapshots owned by sessions in the current project scope. sessionIDs limits
+// the result when non-nil; an empty non-nil slice returns no rows. Placeholder
+// snapshots created before a session is inspected are deliberately excluded.
+func (db *DB) ListPublishableSessionProjectIdentitySnapshots(
+	ctx context.Context,
+	sessionIDs, projects, excludeProjects []string,
+) ([]export.ProjectIdentityObservation, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if sessionIDs != nil && len(sessionIDs) == 0 {
+		return nil, nil
+	}
+
+	query := func(ids []string) ([]export.ProjectIdentityObservation, error) {
+		predicates := []string{
+			"owner.deleted_at IS NULL",
+			"(TRIM(snap.key_source) != '' OR TRIM(snap.worktree_root_path) != '')",
+		}
+		var args []any
+		appendSet := func(expression string, values []string, negate bool) {
+			if len(values) == 0 {
+				return
+			}
+			placeholders := make([]string, len(values))
+			for i, value := range values {
+				placeholders[i] = "?"
+				args = append(args, value)
+			}
+			op := " IN "
+			if negate {
+				op = " NOT IN "
+			}
+			predicates = append(
+				predicates,
+				expression+op+"("+strings.Join(placeholders, ",")+")",
+			)
+		}
+		appendSet("owner.project", projects, false)
+		appendSet("owner.project", excludeProjects, true)
+		appendSet("snap.session_id", ids, false)
+
+		rows, err := db.getReader().QueryContext(ctx, `
+			SELECT snap.session_id, snap.project, snap.machine, snap.root_path,
+				snap.git_remote, snap.git_remote_name, snap.repository_path,
+				snap.worktree_name, snap.worktree_root_path,
+				snap.worktree_relationship, snap.checkout_state,
+				snap.git_branch, snap.remote_resolution,
+				snap.remote_candidate_count, snap.observed_at,
+				snap.normalized_remote, snap.key_source, snap.key
+			FROM session_project_identity_snapshots snap
+			JOIN sessions owner ON owner.id = snap.session_id
+			WHERE `+strings.Join(predicates, " AND ")+`
+			ORDER BY snap.session_id`, args...)
+		if err != nil {
+			return nil, fmt.Errorf(
+				"listing publishable session project identity snapshots: %w",
+				err,
+			)
+		}
+		defer rows.Close()
+
+		var out []export.ProjectIdentityObservation
+		for rows.Next() {
+			var obs export.ProjectIdentityObservation
+			var observedAt string
+			if err := rows.Scan(
+				&obs.SessionID, &obs.Project, &obs.Machine, &obs.RootPath,
+				&obs.GitRemote, &obs.GitRemoteName, &obs.RepositoryPath,
+				&obs.WorktreeName, &obs.WorktreeRootPath,
+				&obs.WorktreeRelationship, &obs.CheckoutState,
+				&obs.GitBranch, &obs.RemoteResolution,
+				&obs.RemoteCandidateCount, &observedAt,
+				&obs.NormalizedRemote, &obs.KeySource, &obs.Key,
+			); err != nil {
+				return nil, fmt.Errorf(
+					"scanning publishable session project identity snapshot: %w",
+					err,
+				)
+			}
+			obs.ObservedAt, err = time.Parse(time.RFC3339Nano, observedAt)
+			if err != nil {
+				return nil, fmt.Errorf(
+					"parsing publishable session identity timestamp: %w", err,
+				)
+			}
+			out = append(out, obs)
+		}
+		if err := rows.Err(); err != nil {
+			return nil, fmt.Errorf(
+				"iterating publishable session project identity snapshots: %w",
+				err,
+			)
+		}
+		return out, nil
+	}
+
+	if sessionIDs == nil {
+		return query(nil)
+	}
+	chunkSize := maxSQLVars - len(projects) - len(excludeProjects)
+	if chunkSize < 1 {
+		chunkSize = 1
+	}
+	var out []export.ProjectIdentityObservation
+	err := queryChunkedSize(sessionIDs, chunkSize, func(ids []string) error {
+		rows, err := query(ids)
+		if err != nil {
+			return err
+		}
+		out = append(out, rows...)
+		return nil
+	})
+	return out, err
+}
+
 func (db *DB) BuildProjectIdentityMap(
 	ctx context.Context,
 	labels []string,

internal/duckdb/project_identity_upsert.go

@@ -56,38 +56,46 @@ func deleteProjectIdentityDelta(
 	return nil
 }
 
-func deleteProjectIdentityScope(
+func deleteProjectIdentityArchive(
 	exec duckProjectIdentityExec,
 	archiveID string,
-	projects, excludeProjects []string,
 ) error {
-	args := []any{archiveID}
-	predicates := []string{"source_archive_id = ?"}
-	appendSet := func(column string, values []string, negate bool) {
-		if len(values) == 0 {
-			return
-		}
-		placeholders := make([]string, len(values))
-		for i, value := range values {
-			placeholders[i] = "?"
-			args = append(args, value)
-		}
-		op := "IN"
-		if negate {
-			op = "NOT IN"
-		}
-		predicates = append(predicates,
-			column+" "+op+" ("+strings.Join(placeholders, ",")+")")
-	}
-	appendSet("project", projects, false)
-	appendSet("project", excludeProjects, true)
-	where := strings.Join(predicates, " AND ")
 	for _, table := range []string{
 		"source_project_identity_observations",
 		"source_session_project_identity_snapshots",
 	} {
-		if err := exec("DELETE FROM "+table+" WHERE "+where, args...); err != nil {
-			return fmt.Errorf("clearing duckdb %s publication scope: %w", table, err)
+		if err := exec(
+			"DELETE FROM "+table+" WHERE source_archive_id = ?",
+			archiveID,
+		); err != nil {
+			return fmt.Errorf("clearing duckdb %s archive: %w", table, err)
+		}
+	}
+	return nil
+}
+
+func deleteSessionProjectIdentitySnapshotsBySessionID(
+	exec duckProjectIdentityExec,
+	archiveID string,
+	sessionIDs []string,
+) error {
+	for start := 0; start < len(sessionIDs); start += projectIdentityDeleteBatchSize {
+		end := min(start+projectIdentityDeleteBatchSize, len(sessionIDs))
+		args := []any{archiveID}
+		placeholders := make([]string, 0, end-start)
+		for _, sessionID := range sessionIDs[start:end] {
+			args = append(args, sessionID)
+			placeholders = append(placeholders, "?")
+		}
+		if err := exec(`
+			DELETE FROM source_session_project_identity_snapshots
+			WHERE source_archive_id = ?
+			  AND source_session_id IN (`+
+			strings.Join(placeholders, ", ")+`)`, args...); err != nil {
+			return fmt.Errorf(
+				"deleting duckdb session identity snapshots by session id: %w",
+				err,
+			)
 		}
 	}
 	return nil

internal/duckdb/project_inventory_test.go

@@ -465,7 +465,7 @@ func pushDataReadMirror(t *testing.T, ctx context.Context, syncer *Sync) {
 	require.NoError(t, createSchema(ctx, syncer.DB()), "createSchema")
 	_, err := syncer.pushEverything(ctx, nil)
 	require.NoError(t, err, "pushEverything")
-	_, err = syncer.syncProjectIdentityObservations(ctx, 0, true)
+	_, err = syncer.syncProjectIdentityObservations(ctx, 0, true, nil)
 	require.NoError(t, err, "syncProjectIdentityObservations")
 	_, err = syncer.syncWorktreeMappings(ctx, 0, true)
 	require.NoError(t, err, "syncWorktreeMappings")