fix(sync): preserve hierarchy repair intent

Generic relink seeds do not prove that a missing parent was deleted, so treating every queued session as cleanup work could erase valid parser-derived hierarchy while ingestion was incomplete. Persist destructive cleanup intent separately from ordinary edge reconciliation while retaining the row-keyed queue cost shape and legacy queue semantics.

Archive rebuilds must also consume copied repair state after orphan restoration. Abort before swap when that transactional repair fails so a rebuilt archive cannot install stale hierarchy.

Author: wesm

internal/db/db_test.go

@@ -5632,6 +5632,9 @@ func TestCopySyncStateFrom_OnlyCopiesDurableKeys(t *testing.T) {
 		"seed source finished")
 	require.NoError(t, srcDB.QueueSubagentParentRepairs([]string{"queued-child"}),
 		"seed durable hierarchy repair")
+	require.NoError(t, srcDB.QueueSubagentParentCleanupRepairs(
+		[]string{"queued-former-child"},
+	), "seed durable hierarchy cleanup")
 	require.NoError(t, srcDB.UpsertSession(Session{
 		ID: "queued-session", Project: "p", Machine: "local", Agent: "claude",
 	}), "seed source queued session")
@@ -5667,6 +5670,13 @@ func TestCopySyncStateFrom_OnlyCopiesDurableKeys(t *testing.T) {
 	).Scan(&queuedRepairs), "query copied subagent repair queue")
 	assert.Equal(t, 1, queuedRepairs,
 		"pending hierarchy repairs must survive an archive rebuild")
+	var queuedCleanups int
+	require.NoError(t, dstDB.Reader().QueryRow(`
+		SELECT count(*) FROM subagent_parent_cleanup_queue
+		WHERE session_id = 'queued-former-child'`,
+	).Scan(&queuedCleanups), "query copied subagent cleanup queue")
+	assert.Equal(t, 1, queuedCleanups,
+		"pending destructive cleanup intent must survive an archive rebuild")
 
 	gotStarted, err := dstDB.GetSyncState("last_sync_started_at")
 	require.NoError(t, err, "GetSyncState last_sync_started_at")

internal/db/link_subagent_nested_test.go

@@ -775,8 +775,9 @@ func TestLinkSubagentSessionsForSessionsClearsDanglingParent(t *testing.T) {
 	require.Equal(t, 1, n, "spawner must be deleted")
 
 	// The engine captures the kid as a pre-write child of the deleted
-	// spawner and carries it into the scoped batch.
-	require.NoError(t, d.LinkSubagentSessionsForSessions([]string{"kid"}))
+	// spawner and persists cleanup intent before the destructive write.
+	require.NoError(t, d.QueueSubagentParentCleanupRepairs([]string{"kid"}))
+	require.NoError(t, d.RepairQueuedSubagentParents())
 
 	kid, err := d.GetSession(context.Background(), "kid")
 	requireNoError(t, err, "GetSession kid")
@@ -788,6 +789,39 @@ func TestLinkSubagentSessionsForSessionsClearsDanglingParent(t *testing.T) {
 			"re-links the child")
 }
 
+// TestLinkSubagentSessionsForSessionsKeepsUnresolvedPathParent proves that a
+// normal changed-session seed is not evidence that its parent was deleted.
+// Providers can ingest a path-derived child before its parent, with no spawn
+// edge yet available; scoped edge linking must preserve that parser claim so
+// the later parent row completes the hierarchy instead of leaving the child
+// permanently un-parented.
+func TestLinkSubagentSessionsForSessionsKeepsUnresolvedPathParent(t *testing.T) {
+	d := testDB(t)
+
+	insertSession(t, d, "kid", "p", func(s *Session) {
+		s.MessageCount = 1
+		s.ParentSessionID = Ptr("parent-not-ingested-yet")
+		s.RelationshipType = "subagent"
+	})
+
+	require.NoError(t, d.LinkSubagentSessionsForSessions([]string{"kid"}))
+	assert.Equal(t, "parent-not-ingested-yet", parentOfSession(t, d, "kid"),
+		"a generic changed-session seed must preserve parser-derived parentage")
+	require.NoError(t, d.QueueSubagentParentRepairs([]string{"kid"}))
+	require.NoError(t, d.RepairQueuedSubagentParents())
+	assert.Equal(t, "parent-not-ingested-yet", parentOfSession(t, d, "kid"),
+		"a durable generic repair must remain relink-only")
+
+	insertSession(t, d, "parent-not-ingested-yet", "p", func(s *Session) {
+		s.MessageCount = 1
+	})
+	require.NoError(t, d.LinkSubagentSessionsForSessions(
+		[]string{"parent-not-ingested-yet"},
+	))
+	assert.Equal(t, "parent-not-ingested-yet", parentOfSession(t, d, "kid"),
+		"ingesting the parent later must leave the valid path hierarchy intact")
+}
+
 // TestLinkSubagentSessionsForSessionsKeepsParentWhenSpawnerRemains pins the
 // deliberate limit of the dangling-parent repair: when only the EDGE is
 // gone but the stored parent session still exists, nothing distinguishes an
@@ -904,6 +938,32 @@ func TestRepairQueuedSubagentParentsMigratesLegacyJSONQueue(t *testing.T) {
 	assert.Zero(t, queued, "successful repair must clear migrated rows")
 }
 
+func TestRepairQueuedSubagentParentsMigratesLegacyCleanupIntent(t *testing.T) {
+	d := testDB(t)
+	insertSession(t, d, "spawner", "p", func(s *Session) {
+		s.MessageCount = 1
+	})
+	insertSession(t, d, "kid", "p", func(s *Session) {
+		s.MessageCount = 1
+		s.ParentSessionID = Ptr("spawner")
+		s.RelationshipType = "subagent"
+	})
+	insertMessages(t, d, spawnEdgeTo("spawner", "kid", "spawn"))
+	require.NoError(t, d.SetSyncState(
+		subagentParentRepairQueueStateKey, `["kid"]`,
+	))
+	_, err := d.DeleteParserExcludedSessions([]string{"spawner"})
+	require.NoError(t, err)
+
+	require.NoError(t, d.RepairQueuedSubagentParents())
+
+	kid, err := d.GetSession(context.Background(), "kid")
+	require.NoError(t, err)
+	require.NotNil(t, kid)
+	assert.Nil(t, kid.ParentSessionID,
+		"the legacy queue contained pre-write children and must retain cleanup intent")
+}
+
 // TestLinkSubagentSessionsForSessionsPlanIsBatchBounded pins the cost shape
 // of the scoped statement, mirroring TestLinkSubagentSessionsPlanScalesWith-
 // SpawnEdges: the watcher calls this once per changed file, so neither

internal/db/orphaned.go

@@ -374,6 +374,13 @@ func (d *DB) CopySyncStateFrom(sourcePath string) error {
 			return fmt.Errorf("copying subagent parent repair queue: %w", err)
 		}
 	}
+	if oldDBHasTable(ctx, tx, "subagent_parent_cleanup_queue") {
+		if _, err := tx.ExecContext(ctx, `
+			INSERT OR IGNORE INTO main.subagent_parent_cleanup_queue (session_id)
+			SELECT session_id FROM old_db.subagent_parent_cleanup_queue`); err != nil {
+			return fmt.Errorf("copying subagent parent cleanup queue: %w", err)
+		}
+	}
 
 	headRevisionExpr := "0"
 	if oldDBHasColumn(ctx, tx, "artifact_checkpoint_heads", "publication_revision") {

internal/db/schema.sql

@@ -1081,6 +1081,13 @@ CREATE TABLE IF NOT EXISTS subagent_parent_repair_queue (
     session_id TEXT PRIMARY KEY
 ) WITHOUT ROWID;
 
+-- Subset of queued hierarchy repairs whose pre-write spawn edge may have been
+-- removed. Only these captured former children are eligible for destructive
+-- dangling-parent cleanup; ordinary changed-session seeds are relink-only.
+CREATE TABLE IF NOT EXISTS subagent_parent_cleanup_queue (
+    session_id TEXT PRIMARY KEY
+) WITHOUT ROWID;
+
 -- Model pricing for cost calculation
 CREATE TABLE IF NOT EXISTS model_pricing (
     model_pattern    TEXT PRIMARY KEY,

internal/db/sessions.go

@@ -1823,7 +1823,7 @@ func linkSubagentSessionsForSessionsQuery(ph string) string {
 	)`
 }
 
-// clearDanglingSubagentParentQuery repairs a captured child whose LAST spawn
+// clearDanglingSubagentParentQuery repairs a captured former child whose LAST spawn
 // edge was removed together with its spawner: both UNION branches of the
 // linking statement select from remaining tool_calls, so an edge-less child
 // can never be re-resolved there, and its parent now points at a session
@@ -1853,8 +1853,11 @@ func clearDanglingSubagentParentQuery(ph string) string {
 // LinkSubagentSessionsForSessions is LinkSubagentSessions scoped to the
 // sessions written by one sync batch: only children reachable from a batch
 // member's spawn edges (or batch members that are themselves children) are
-// re-resolved. Children in ids whose edges are all gone and whose parent no
-// longer exists are un-parented (see clearDanglingSubagentParentQuery).
+// re-resolved. Generic changed-session IDs are deliberately ineligible for
+// dangling-parent cleanup: a parser-derived parent may simply not have been
+// ingested yet. Destructive cleanup is reserved for former children captured
+// before a write that can remove their spawn edges and persisted through
+// QueueSubagentParentCleanupRepairs.
 // Per-event paths — the session watcher re-syncs a single file
 // on every change — must use this form so their linking cost tracks the
 // changed batch; bulk paths (full sync, reconciliation, resync) keep the
@@ -1880,25 +1883,27 @@ func (db *DB) LinkSubagentSessionsForSessions(ids []string) error {
 				len(chunk), err,
 			)
 		}
-		_, err = db.getWriter().Exec(
-			clearDanglingSubagentParentQuery(ph), args...,
-		)
-		if err != nil {
-			return fmt.Errorf(
-				"clearing dangling subagent parents for %d changed "+
-					"sessions: %w",
-				len(chunk), err,
-			)
-		}
 		return nil
 	})
 }
 
-// QueueSubagentParentRepairs durably records sessions whose hierarchy may be
-// changed by an upcoming write. Callers must queue the IDs before deleting or
-// replacing messages because those writes can cascade away the only spawn
-// edge that identifies an affected child.
+// QueueSubagentParentRepairs durably records sessions whose hierarchy must be
+// re-evaluated from surviving spawn edges. These generic seeds are never used
+// for destructive dangling-parent cleanup; callers that captured a former
+// child before removing edges use QueueSubagentParentCleanupRepairs instead.
 func (db *DB) QueueSubagentParentRepairs(ids []string) error {
+	return db.queueSubagentParentRepairs(ids, false)
+}
+
+// QueueSubagentParentCleanupRepairs durably records former children captured
+// before an exclusion or message replacement can remove their spawn edges.
+// Cleanup intent is separate from ordinary relink work so a newly parsed child
+// whose parent has not arrived yet never loses valid parser-derived parentage.
+func (db *DB) QueueSubagentParentCleanupRepairs(ids []string) error {
+	return db.queueSubagentParentRepairs(ids, true)
+}
+
+func (db *DB) queueSubagentParentRepairs(ids []string, cleanup bool) error {
 	if len(ids) == 0 {
 		return nil
 	}
@@ -1911,20 +1916,39 @@ func (db *DB) QueueSubagentParentRepairs(ids []string) error {
 	}
 	defer func() { _ = tx.Rollback() }()
 
-	stmt, err := tx.Prepare(`
+	repairStmt, err := tx.Prepare(`
 		INSERT INTO subagent_parent_repair_queue (session_id) VALUES (?)
 		ON CONFLICT(session_id) DO NOTHING`)
 	if err != nil {
 		return fmt.Errorf("preparing subagent parent repair queue insert: %w", err)
 	}
-	defer stmt.Close()
+	defer repairStmt.Close()
+	var cleanupStmt *sql.Stmt
+	if cleanup {
+		cleanupStmt, err = tx.Prepare(`
+			INSERT INTO subagent_parent_cleanup_queue (session_id) VALUES (?)
+			ON CONFLICT(session_id) DO NOTHING`)
+		if err != nil {
+			return fmt.Errorf(
+				"preparing subagent parent cleanup queue insert: %w", err,
+			)
+		}
+		defer cleanupStmt.Close()
+	}
 	for _, id := range ids {
 		if id == "" {
 			continue
 		}
-		if _, err := stmt.Exec(id); err != nil {
+		if _, err := repairStmt.Exec(id); err != nil {
 			return fmt.Errorf("queueing subagent parent repair for %s: %w", id, err)
 		}
+		if cleanupStmt != nil {
+			if _, err := cleanupStmt.Exec(id); err != nil {
+				return fmt.Errorf(
+					"queueing subagent parent cleanup for %s: %w", id, err,
+				)
+			}
+		}
 	}
 	if err := tx.Commit(); err != nil {
 		return fmt.Errorf("committing subagent parent repair queue: %w", err)
@@ -1943,6 +1967,7 @@ func (db *DB) RepairQueuedSubagentParents() error {
 	var pending int
 	err := db.getWriter().QueryRow(`
 		SELECT EXISTS(SELECT 1 FROM subagent_parent_repair_queue)
+		    OR EXISTS(SELECT 1 FROM subagent_parent_cleanup_queue)
 		    OR EXISTS(SELECT 1 FROM pg_sync_state WHERE key = ?)`,
 		subagentParentRepairQueueStateKey,
 	).Scan(&pending)
@@ -1963,6 +1988,8 @@ func (db *DB) RepairQueuedSubagentParents() error {
 	for {
 		rows, err := tx.Query(`
 			SELECT session_id FROM subagent_parent_repair_queue
+			UNION
+			SELECT session_id FROM subagent_parent_cleanup_queue
 			ORDER BY session_id LIMIT ?`, maxSQLVars/2)
 		if err != nil {
 			return fmt.Errorf("listing queued subagent parent repairs: %w", err)
@@ -1996,15 +2023,26 @@ func (db *DB) RepairQueuedSubagentParents() error {
 				len(chunk), err,
 			)
 		}
+		cleanupSeeds := `(SELECT session_id
+			FROM subagent_parent_cleanup_queue WHERE session_id IN ` + ph + `)`
 		if _, err := tx.Exec(
-			clearDanglingSubagentParentQuery(ph), args...,
+			clearDanglingSubagentParentQuery(cleanupSeeds), args...,
 		); err != nil {
 			return fmt.Errorf(
 				"clearing queued dangling subagent parents for %d "+
 					"sessions: %w",
 				len(chunk), err,
 			)
 		}
+		if _, err := tx.Exec(
+			"DELETE FROM subagent_parent_cleanup_queue WHERE session_id IN "+ph,
+			args...,
+		); err != nil {
+			return fmt.Errorf(
+				"clearing %d queued subagent parent cleanups: %w",
+				len(chunk), err,
+			)
+		}
 		if _, err := tx.Exec(
 			"DELETE FROM subagent_parent_repair_queue WHERE session_id IN "+ph,
 			args...,
@@ -2037,20 +2075,35 @@ func migrateLegacySubagentParentRepairQueueTx(tx *sql.Tx) error {
 	if err := json.Unmarshal([]byte(encoded), &ids); err != nil {
 		return fmt.Errorf("decoding legacy subagent parent repair queue: %w", err)
 	}
-	stmt, err := tx.Prepare(`
+	repairStmt, err := tx.Prepare(`
 		INSERT INTO subagent_parent_repair_queue (session_id) VALUES (?)
 		ON CONFLICT(session_id) DO NOTHING`)
 	if err != nil {
 		return fmt.Errorf("preparing legacy subagent parent repair migration: %w", err)
 	}
-	defer stmt.Close()
+	defer repairStmt.Close()
+	cleanupStmt, err := tx.Prepare(`
+		INSERT INTO subagent_parent_cleanup_queue (session_id) VALUES (?)
+		ON CONFLICT(session_id) DO NOTHING`)
+	if err != nil {
+		return fmt.Errorf("preparing legacy subagent parent cleanup migration: %w", err)
+	}
+	defer cleanupStmt.Close()
 	for _, id := range ids {
 		if id == "" {
 			continue
 		}
-		if _, err := stmt.Exec(id); err != nil {
+		if _, err := repairStmt.Exec(id); err != nil {
 			return fmt.Errorf("migrating legacy subagent parent repair for %s: %w", id, err)
 		}
+		// The JSON queue predates generic post-write and attempted-session
+		// seeds; every legacy ID was captured before a destructive write and
+		// therefore carries cleanup intent.
+		if _, err := cleanupStmt.Exec(id); err != nil {
+			return fmt.Errorf(
+				"migrating legacy subagent parent cleanup for %s: %w", id, err,
+			)
+		}
 	}
 	if _, err := tx.Exec(
 		"DELETE FROM pg_sync_state WHERE key = ?",

internal/sync/engine.go

@@ -2343,6 +2343,26 @@ func (e *Engine) resyncBuildLocked(
 		}
 	}
 
+	// CopySyncStateFrom runs after the fresh archive's normal linking pass so
+	// pending hierarchy work from the original must be consumed explicitly.
+	// Wait until orphan restoration is complete so every queued session and
+	// copied spawn edge is present. A failed repair leaves hierarchy state
+	// uncertain and must abort before the replacement can be installed.
+	if err := newDB.RepairQueuedSubagentParents(); err != nil {
+		log.Printf("resync: repair copied subagent parents: %v", err)
+		stats.Aborted = true
+		stats.Warnings = append(stats.Warnings,
+			"hierarchy repair failed, aborting swap: "+err.Error(),
+		)
+		newDB.Close()
+		removeTempDB(tempPath)
+		restoreSkipCache()
+		e.mu.Lock()
+		e.lastSyncStats = stats
+		e.mu.Unlock()
+		return stats, err
+	}
+
 	// Copy recall entries and their evidence from the quiesced old DB.
 	// The fresh DB is built from source files, which never contain
 	// recall entries, so without this every accepted entry is lost on
@@ -6710,7 +6730,7 @@ func (e *Engine) collectAndBatch(
 		// Persist affected IDs before any exclusion or replacement can
 		// cascade their only spawn edge away. The queue is cleared only in
 		// the same transaction that successfully repairs the hierarchy.
-		if err := e.db.QueueSubagentParentRepairs(children); err != nil {
+		if err := e.db.QueueSubagentParentCleanupRepairs(children); err != nil {
 			log.Printf("queue subagent parent repairs: %v", err)
 			stats.RecordFailed()
 			e.noteSQLiteContainerResult(r.path, false)
@@ -13374,7 +13394,7 @@ func (e *Engine) SyncSingleSessionContext(
 	if err := e.db.RepairQueuedSubagentParents(); err != nil {
 		return fmt.Errorf("repair queued subagent parents: %w", err)
 	}
-	if err := e.db.QueueSubagentParentRepairs(priorChildren); err != nil {
+	if err := e.db.QueueSubagentParentCleanupRepairs(priorChildren); err != nil {
 		return fmt.Errorf("queue subagent parent repairs: %w", err)
 	}
 	// Always attempt queued work after mutations begin, including when a later