fix(duckdb): preserve source machine attribution (#1302)

DuckDB mirrors now preserve each session's source-machine label, falling back to the publishing machine only for legacy empty or `local` values. The resolved label also participates in the session fingerprint, so attribution changes re-push the row and match PostgreSQL's existing behavior.

Mirror status, residency checks, and curation replacement now operate across every session in the single source archive instead of filtering by the last publishing machine. These dropped machine filters are the most behaviorally significant part of the change: the publisher remains mirror metadata, while resident sessions may carry labels from multiple filesystem sources.

This bumps the disposable mirror schema from 6 to 7. Existing mirrors rebuild once into a fresh validated file and atomically swap on upgrade, which corrects older rows that could sit outside the incremental cutoff.

The change was extracted from #1170, where it was bundled with separate session-source work, so that PR can shed this pre-existing DuckDB/PostgreSQL parity fix and its schema-wide rebuild.

Co-authored-by: Matthew Jacobs <mjacobs@users.noreply.github.com>

Author: mjacobs

internal/duckdb/connect.go

@@ -154,20 +154,10 @@ func readLocalMirrorStatus(
 	)
 }
 
-// readMachineStatus reads the target's push metadata and machine-scoped row
-// counts. It tolerates a mirror with no sync_metadata rows yet, or missing
-// tables entirely (a fresh, foreign, or pre-v3 file): both degrade to zero
-// values instead of erroring, since status must not crash on an old mirror.
-//
-// The row counts are filtered by the TARGET's own recorded LastPushMachine
-// when it is set, not by the caller's configured machine name: a remote
-// Quack client's configured machine name is normally its own hostname,
-// which almost never matches the hostname of whatever machine actually
-// pushed the mirror it is reading, so filtering by the configured name
-// there would report zero rows even though the display line above already
-// shows a real LastPushMachine and non-zero mirror content. The configured
-// machine name is used only as the fallback for a mirror with no recorded
-// push yet (LastPushMachine == ""), where it is the best available guess.
+// readMachineStatus reads the target's push metadata and total mirror row
+// counts. Sessions retain their per-source machine attribution, so status
+// cannot use LastPushMachine as a row filter. Missing metadata or tables
+// degrade to zero values so status remains usable for fresh and old mirrors.
 func readMachineStatus(
 	ctx context.Context,
 	duck *sql.DB,
@@ -186,14 +176,8 @@ func readMachineStatus(
 	status.DataVersion = meta.DataVersion
 	status.Scope = meta.Scope
 
-	countMachine := meta.LastPushMachine
-	if countMachine == "" {
-		countMachine = machine
-	}
-
 	if err := queryDuckDBRowContext(ctx, duck, connectionKind, quack,
-		`SELECT COUNT(*) FROM sessions WHERE machine = ?`,
-		countMachine,
+		`SELECT COUNT(*) FROM sessions`,
 	).Scan(&status.DuckDBSessions); err != nil {
 		if isMissingDuckDBTable(err) {
 			return status, nil
@@ -204,9 +188,8 @@ func readMachineStatus(
 		`SELECT COUNT(*)
 		 FROM messages
 		 WHERE session_id IN (
-			SELECT id FROM sessions WHERE machine = ?
+			SELECT id FROM sessions
 		 )`,
-		countMachine,
 	).Scan(&status.DuckDBMessages); err != nil {
 		if isMissingDuckDBTable(err) {
 			return status, nil

internal/duckdb/probe.go

@@ -278,15 +278,10 @@ func (p MirrorProbe) NeedsRebuild(scope string, sourceDataVersion int) bool {
 // rebuilds: identity-less mirrors only come from earlier builds of this
 // unreleased branch, so they simply rebuild once and record the id.
 //
-// The machine-change check exists because mirror rows are machine-stamped
-// (see the sessions.machine column and duckSessionFingerprintFields): an
-// incremental push only rewrites sessions whose LOCAL content changed
-// within the current mirror window, so a session that has not changed
-// since the mirror's last push stays permanently labeled with the OLD
-// machine name even after the push metadata's LastPushMachine flips to the
-// new one — silently stranding it under a machine filter (see
-// readMachineStatus) that will never again select it. A full rebuild
-// re-pushes every session under the new machine name instead.
+// The machine-change check remains necessary for legacy sessions whose local
+// machine is empty or "local": mirroring substitutes the current push machine
+// for those values, so a full rebuild must restamp every such row when that
+// configured name changes. Explicit per-source machine labels remain unchanged.
 //
 // localDeletionRevision is the caller's local.SessionDeletionPublicationRevision
 // read, and localDatabaseID the caller's local.GetDatabaseID read; both are

internal/duckdb/push.go

@@ -600,7 +600,7 @@ func (snap curationSnapshot) fingerprint() (string, error) {
 // All work is bounded by curation size, not mirror size: mirror membership
 // is validated for exactly the snapshot's session IDs (one batched lookup,
 // see mirrorResidentSessionIDs), pin notes are preserved, and the delete
-// side stays the machine-scoped clear of both tables, so removed
+// side clears both tables for sessions in this source archive, so removed
 // stars/pins disappear without enumerating them.
 func (s *Sync) replaceCuration(
 	ctx context.Context, snap curationSnapshot,
@@ -641,8 +641,8 @@ func (s *Sync) replaceCuration(
 			if err := s.execMutation(ctx, tx, `
 				DELETE FROM `+table+`
 				WHERE session_id IN (
-					SELECT id FROM sessions WHERE machine = ?
-				)`, s.machine); err != nil {
+					SELECT id FROM sessions
+				)`); err != nil {
 				return fmt.Errorf("clearing duckdb %s: %w", table, err)
 			}
 		}
@@ -899,7 +899,9 @@ func (s *Sync) upsertSession(
 			secrets_rules_version = excluded.secrets_rules_version,
 			agentsview_push_fingerprint = excluded.agentsview_push_fingerprint`
 
-	args := sessionInsertArgs(sess, s.machine, fingerprint)
+	args := sessionInsertArgs(
+		sess, mirroredSessionMachine(sess, s.machine), fingerprint,
+	)
 	if err := s.execMutation(ctx, exec, query, args...); err != nil {
 		return fmt.Errorf("writing duckdb session %s: %w", sess.ID, err)
 	}
@@ -944,6 +946,13 @@ func sessionInsertArgs(sess db.Session, machine, fingerprint string) []any {
 	}
 }
 
+func mirroredSessionMachine(sess db.Session, pushMachine string) string {
+	if sess.Machine == "" || sess.Machine == "local" {
+		return pushMachine
+	}
+	return sess.Machine
+}
+
 func insertMessages(
 	ctx context.Context, exec duckMutationExecutor, msgs []db.Message,
 ) error {

internal/duckdb/rebuild.go

@@ -81,7 +81,7 @@ func ensureMirrorWorkDir(path string) (string, error) {
 // rebuildMirror builds a fresh DuckDB mirror file from scratch in a
 // temporary file inside the mirror's work directory, then atomically swaps
 // it over path. It is
-// the only way a schema v6 mirror is created or repaired: unlike Sync.Push,
+// the only way a schema v7 mirror is created or repaired: unlike Sync.Push,
 // it never touches an existing mirror file in place, so a rebuild that
 // fails at any point leaves the previous mirror (if any) fully intact.
 func rebuildMirror(

internal/duckdb/schema.go

@@ -11,10 +11,10 @@ import (
 )
 
 // SchemaVersion is the version of the DuckDB mirror schema created by
-// createSchema. Mirror schema v6 is create-only: there are no in-place
+// createSchema. Mirror schema v7 is create-only: there are no in-place
 // migrations between versions. A version mismatch means the mirror file
 // must be rebuilt with 'agentsview duckdb push --full'.
-const SchemaVersion = 6
+const SchemaVersion = 7
 
 const schemaVersionMetadataKey = "agentsview_schema_version"
 

internal/duckdb/sync.go

@@ -597,7 +597,7 @@ func (s *Sync) readMirrorFingerprintBatch(
 }
 
 // mirrorResidentSessionIDs reports which of ids currently exist in the
-// mirror under this Sync's machine name. IDs are deduplicated and queried
+// mirror for this source archive. IDs are deduplicated and queried
 // in batches of 500, so cost tracks the caller's ID list (a candidate
 // window, a tombstone delta, the curation set), never total mirror size.
 func (s *Sync) mirrorResidentSessionIDs(
@@ -627,14 +627,13 @@ func (s *Sync) readMirrorResidentBatch(
 	ctx context.Context, batch []string, out map[string]bool,
 ) error {
 	placeholders := make([]string, len(batch))
-	args := make([]any, 0, len(batch)+1)
-	args = append(args, s.machine)
+	args := make([]any, 0, len(batch))
 	for i, id := range batch {
 		placeholders[i] = "?"
 		args = append(args, id)
 	}
 	rows, err := s.duck.QueryContext(ctx,
-		`SELECT id FROM sessions WHERE machine = ? AND id IN (`+
+		`SELECT id FROM sessions WHERE id IN (`+
 			strings.Join(placeholders, ",")+`)`, args...,
 	)
 	if err != nil {
@@ -938,7 +937,9 @@ func (s *Sync) sessionFingerprints(
 			SecretFindings []db.SecretFinding
 			Pins           []db.PinnedMessage
 		}{
-			SessionFields:  duckSessionFingerprintFields(sess, s.machine),
+			SessionFields: duckSessionFingerprintFields(
+				sess, mirroredSessionMachine(sess, s.machine),
+			),
 			Messages:       msgs,
 			Usage:          usage[sess.ID],
 			ToolCalls:      toolCalls,