fix(sync): key ownership baselines and rehydration on stored attribution

Two more paths still derived a machine instead of carrying the stored one.

Rehydrated reconciliation candidates dropped candidate.Machine when building
their DiscoveredFile, so workers recomputed attribution from the physical
source path. That is wrong for providers whose source can sit outside the
labeled root it was configured under, and it could retire a valid labeled
baseline and lose later deletions. Both rehydration branches now carry the
candidate's machine, matching what baselineReconciliationCandidates already
expects.

Ownership baselines keyed on the freshly parsed machine, but prepareSessionWrite
preserves an existing archive row's label, and it works on a copy that the
caller never sees. After a relabel the baseline could land under a machine no
session row holds, leaving the source untombstonable. Writes now record the
machine they persisted under and baselines key on it. The DB-backed streaming
path resolves the stored machine for the same reason; that lookup short-circuits
unless the agent actually has labeled roots, so unlabeled setups add no queries.

The added test pins that a relabel keeps both the session and its baseline on
the original machine. It documents the invariant but does not by itself
reproduce the reported pending-write mismatch; see the PR discussion.

Author: mjacobs

internal/sync/engine.go

@@ -689,6 +689,30 @@ func (e *Engine) reconciliationOwnershipMachines(
 	return machines
 }
 
+// storedSourceMachine returns the machine an already-ingested session at path
+// was admitted under, or "" when the archive holds none. Attribution is
+// immutable, so a stored row outranks the configured label whenever the two
+// disagree. It short-circuits unless the agent actually has labeled roots, so
+// the common unlabeled setup adds no per-source queries.
+func (e *Engine) storedSourceMachine(
+	agent parser.AgentType, path string,
+) string {
+	if len(e.sourceMachines[agent]) == 0 || path == "" {
+		return ""
+	}
+	ids, err := e.db.ListSessionIDsByFilePath(path, string(agent))
+	if err != nil {
+		return ""
+	}
+	for _, id := range ids {
+		session, err := e.db.GetSession(context.Background(), id)
+		if err == nil && session != nil && session.Machine != "" {
+			return session.Machine
+		}
+	}
+	return ""
+}
+
 func pathWithinRoot(path, root string) bool {
 	root = filepath.Clean(root)
 	rel, err := filepath.Rel(root, path)
@@ -4501,6 +4525,7 @@ func (e *Engine) rehydrateReconciliationPage(
 				files = append(files, parser.DiscoveredFile{
 					Path: candidate.Path, Project: source.ProjectHint,
 					Agent: candidate.Provider, ForceParse: forceCandidate,
+					Machine:        candidate.Machine,
 					ProviderSource: &source, ProviderProcess: true,
 				})
 				continue
@@ -4527,6 +4552,10 @@ func (e *Engine) rehydrateReconciliationPage(
 		files = append(files, parser.DiscoveredFile{
 			Path: candidate.Path, Project: source.ProjectHint,
 			Agent: candidate.Provider, ForceParse: forceCandidate,
+			// Carry the candidate's stored attribution: recomputing it from the
+			// physical path is wrong for providers whose source can sit outside
+			// the labeled root it was configured under.
+			Machine:        candidate.Machine,
 			ProviderSource: &source, ProviderProcess: true,
 		})
 	}
@@ -6656,10 +6685,17 @@ func (e *Engine) syncProviderDBBacked(
 		if path == "" {
 			return nil
 		}
+		storedPath := e.effectiveSourcePath(path)
+		machine := e.machineForProviderSource(agent, source, path)
+		// An already-ingested session keeps the label it was admitted under,
+		// so the baseline must follow the row rather than current config.
+		if stored := e.storedSourceMachine(agent, storedPath); stored != "" {
+			machine = stored
+		}
 		baselines = append(baselines, machineSessionSource{
-			Machine: e.machineForProviderSource(agent, source, path),
+			Machine: machine,
 			Source: db.SessionSourcePath{
-				Agent: string(agent), FilePath: e.effectiveSourcePath(path),
+				Agent: string(agent), FilePath: storedPath,
 			},
 		})
 		if len(baselines) == reconciliationPageSize {
@@ -7463,8 +7499,14 @@ func (e *Engine) baselinePendingWriteSources(
 	)
 	for i, write := range pending {
 		path := e.effectiveSourcePath(write.sess.File.Path)
+		// Key on what was persisted, not on what the parser proposed: an
+		// already-ingested session keeps its original label through a relabel.
+		machine := write.persistedMachine
+		if machine == "" {
+			machine = write.sess.Machine
+		}
 		source := machineSessionSource{
-			Machine: write.sess.Machine,
+			Machine: machine,
 			Source: db.SessionSourcePath{
 				Agent: string(write.sess.Agent), FilePath: path,
 			},
@@ -10663,6 +10705,12 @@ type pendingWrite struct {
 	// baselineEligible is set by collectAndBatch only when the complete source
 	// outcome is safe to make deletion-eligible after this write succeeds.
 	baselineEligible bool
+	// persistedMachine is the machine the session was actually written under.
+	// prepareSessionWrite preserves an existing archive row's attribution, so
+	// after a label edit this differs from sess.Machine. Ownership baselines
+	// must key on it or they land under a machine no session row holds, and
+	// the source can never be tombstoned when it later disappears.
+	persistedMachine string
 	// storageTrustPath/State/Snap promote the session's OpenCode
 	// storage-gate trust after its batch is confirmed fully written.
 	// Empty for everything else.
@@ -10912,6 +10960,7 @@ func (e *Engine) writeBatchWithOutcome(
 			}
 			continue
 		}
+		batch[i].persistedMachine = s.Machine
 
 		// Detect stale parser version BEFORE UpsertSession
 		// overwrites it. Existing message rows from an
@@ -11887,6 +11936,7 @@ func (e *Engine) writeBatchBulkWithOutcome(
 			}
 			continue
 		}
+		batch[pendingIndex].persistedMachine = s.Machine
 		replaceMessages := shouldReplaceFullParseMessages(
 			pw, forceReplace, false, false,
 		)

internal/sync/session_source_machine_test.go

@@ -520,3 +520,74 @@ func activeSessionMachines(t *testing.T, database *db.DB) map[string]string {
 	}
 	return out
 }
+
+// TestBaselineFollowsPersistedMachineAfterRelabel pins the ownership baseline
+// to the machine a session was actually written under. prepareSessionWrite
+// preserves the original label, so keying the baseline off the freshly parsed
+// (configured) machine strands it under a machine no session row holds, and the
+// source can never be tombstoned once it disappears.
+func TestBaselineFollowsPersistedMachineAfterRelabel(t *testing.T) {
+	archiveRoot := t.TempDir()
+	archivePath := writeSessionSourceClaudeFile(t, archiveRoot, "archive-session.jsonl")
+	database := openTestDB(t)
+
+	newEngine := func(machine string) *Engine {
+		return NewEngine(database, EngineConfig{
+			AgentDirs: map[parser.AgentType][]string{
+				parser.AgentClaude: {archiveRoot},
+			},
+			SourceMachines: map[parser.AgentType]map[string]string{
+				parser.AgentClaude: {archiveRoot: machine},
+			},
+			Machine: "localbox",
+		})
+	}
+
+	first := newEngine("archivebox")
+	t.Cleanup(first.Close)
+	require.False(t, first.SyncAll(context.Background(), nil).Aborted)
+
+	// Append to the source so the relabeled pass actually reparses and rewrites
+	// it. An unchanged file is skipped, which never exercises the write path.
+	appendSessionSourceClaudeMessage(t, archivePath)
+
+	// Relabel the root and resync. The session keeps "archivebox"; the baseline
+	// must land there too, not under the newly configured "renamedbox".
+	relabeled := newEngine("renamedbox")
+	t.Cleanup(relabeled.Close)
+	require.False(t, relabeled.SyncAll(context.Background(), nil).Aborted)
+
+	require.Equal(t, "archivebox",
+		activeSessionMachines(t, database)["archive-session"])
+
+	ownershipFor := func(machine string) []db.SessionSourceOwnership {
+		rows, err := database.ListActiveSessionSourceOwnershipScopesPage(
+			context.Background(), machine, string(parser.AgentClaude),
+			[]db.StoredSourcePathHintScope{{Path: archiveRoot}},
+			db.SessionSourceCursor{},
+		)
+		require.NoError(t, err)
+		return rows
+	}
+
+	stranded := ownershipFor("renamedbox")
+	assert.Empty(t, stranded,
+		"the baseline must not be keyed under a label no session row holds")
+
+	owned := ownershipFor("archivebox")
+	require.Len(t, owned, 1,
+		"the baseline must follow the persisted machine")
+	assert.Equal(t, archivePath, owned[0].FilePath)
+}
+
+// appendSessionSourceClaudeMessage grows an existing Claude transcript so the
+// next sync sees a changed source instead of skipping it.
+func appendSessionSourceClaudeMessage(t *testing.T, path string) {
+	t.Helper()
+	builder := testjsonl.NewSessionBuilder()
+	builder.AddClaudeUser("2026-07-01T10:00:00Z", "hello")
+	builder.AddClaudeAssistant("2026-07-01T10:00:01Z", "hi")
+	builder.AddClaudeUser("2026-07-01T10:00:02Z", "more")
+	builder.AddClaudeAssistant("2026-07-01T10:00:03Z", "sure")
+	require.NoError(t, os.WriteFile(path, []byte(builder.String()), 0o600))
+}