Fix nested subagent hierarchy: re-parent depth>=2 subagents to their spawner (#1320)

Fixes #1246. Supersedes #1247.

Claude Code stores nested subagents in one flat directory, so path-based parsing can attach deeper children to the top-level session instead of the session that actually spawned them. This change reconciles effective parents from recorded spawn edges, which also puts nested usage rollups under the right session.

Conflicting edges resolve deterministically using normalized start times and immutable parser provenance. Before exclusions or message replacement can erase an edge, sync persists the affected child IDs in SQLite; repair and queue cleanup then commit together. Failed writes or repairs remain retryable across later syncs, process restarts, and archive rebuilds.

SQLite migration and provenance backfill are atomic. PostgreSQL now mirrors parser provenance through schema compatibility, reads, writes, conflict checks, and fingerprints, while a v2 backfill marker makes existing mirrors republish unchanged legacy rows once. Global work stays bounded by spawn edges, and incremental work stays scoped to affected sessions.

One conservative limit remains: if an edge disappears while the recorded parent still exists and no competing edge remains, the historical parent is retained because the archive cannot prove whether it came from the removed edge or parser provenance. Broader copy and fork provenance remains tracked in #1250.

Reviewers should start with `internal/db/sessions.go`, then follow sync orchestration in `internal/sync/engine.go` and PostgreSQL propagation in `internal/postgres/`.

Co-authored-by: Wes McKinney <wesm@users.noreply.github.com>

Author: wesm

internal/db/db.go

@@ -1763,6 +1763,14 @@ func schemaColumnMigrations() []schemaColumnMigration {
 			"sessions", "display_name",
 			"ALTER TABLE sessions ADD COLUMN display_name TEXT",
 		},
+		{
+			// Preserve the current parent exactly once when the private parser
+			// provenance column is introduced. Running the UPDATE on every open
+			// would let a later linker-derived effective parent overwrite it.
+			"sessions", "parser_parent_session_id",
+			"ALTER TABLE sessions ADD COLUMN parser_parent_session_id TEXT;" +
+				" UPDATE sessions SET parser_parent_session_id = parent_session_id",
+		},
 		{
 			"sessions", "session_name",
 			"ALTER TABLE sessions ADD COLUMN session_name TEXT",
@@ -2160,11 +2168,26 @@ func schemaColumnMigrations() []schemaColumnMigration {
 	}
 }
 
-func applySchemaColumnMigrations(
-	queryRow func(string, ...any) rowScanner,
-	exec func(string, ...any) (sql.Result, error),
-) error {
-	return applyColumnMigrations(schemaColumnMigrations(), queryRow, exec)
+func applySchemaColumnMigrations(w *writerHandle) error {
+	tx, err := w.BeginTx(context.Background(), nil)
+	if err != nil {
+		return fmt.Errorf("starting column migration transaction: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	if err := applyColumnMigrations(
+		schemaColumnMigrations(),
+		func(query string, args ...any) rowScanner {
+			return tx.QueryRow(query, args...)
+		},
+		tx.Exec,
+	); err != nil {
+		return err
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("committing column migrations: %w", err)
+	}
+	return nil
 }
 
 func applyColumnMigrations(
@@ -2411,7 +2434,7 @@ func (db *DB) migrateColumns() error {
 	if _, err := w.Exec(artifactSessionQueueTriggerDropsSQL); err != nil {
 		return fmt.Errorf("dropping artifact session queue triggers: %w", err)
 	}
-	if err := applySchemaColumnMigrations(w.QueryRow, w.Exec); err != nil {
+	if err := applySchemaColumnMigrations(w); err != nil {
 		return err
 	}
 	if _, err := w.Exec(artifactSessionQueueTriggerCreatesSQL); err != nil {

internal/db/db_test.go

@@ -1285,6 +1285,35 @@ func TestSessionParentSessionID(t *testing.T) {
 	})
 }
 
+func TestUpsertSessionRefreshesParserParentSessionID(t *testing.T) {
+	d := testDB(t)
+	s := Session{
+		ID:              "kid",
+		Project:         "proj",
+		Machine:         defaultMachine,
+		Agent:           defaultAgent,
+		ParentSessionID: Ptr("first-parent"),
+	}
+	require.NoError(t, d.UpsertSession(s), "insert session")
+
+	assertParserParent := func(want string) {
+		t.Helper()
+		var got sql.NullString
+		err := d.getReader().QueryRow(
+			`SELECT parser_parent_session_id FROM sessions WHERE id = ?`,
+			"kid",
+		).Scan(&got)
+		require.NoError(t, err, "query parser parent")
+		require.True(t, got.Valid, "parser parent must be set")
+		assert.Equal(t, want, got.String, "parser parent")
+	}
+
+	assertParserParent("first-parent")
+	s.ParentSessionID = Ptr("second-parent")
+	require.NoError(t, d.UpsertSession(s), "update session")
+	assertParserParent("second-parent")
+}
+
 func TestGetChildSessions(t *testing.T) {
 	d := testDB(t)
 
@@ -5588,7 +5617,7 @@ func TestCopySyncStateFrom_NoSourceTable(t *testing.T) {
 	assert.Equal(t, "marker-123", got)
 }
 
-func TestCopySyncStateFrom_OnlyCopiesDurablePGKeys(t *testing.T) {
+func TestCopySyncStateFrom_OnlyCopiesDurableKeys(t *testing.T) {
 	dir := t.TempDir()
 
 	srcPath := filepath.Join(dir, "src.db")
@@ -5601,6 +5630,11 @@ func TestCopySyncStateFrom_OnlyCopiesDurablePGKeys(t *testing.T) {
 		"seed source started")
 	require.NoError(t, srcDB.SetSyncState("last_sync_finished_at", "old-finish"),
 		"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")
@@ -5629,6 +5663,21 @@ func TestCopySyncStateFrom_OnlyCopiesDurablePGKeys(t *testing.T) {
 	assert.Contains(t, artifactExportQueueIDs(t, dstDB), "queued-session",
 		"artifact export queue rows must survive the copy")
 
+	var queuedRepairs int
+	require.NoError(t, dstDB.Reader().QueryRow(`
+		SELECT count(*) FROM subagent_parent_repair_queue
+		WHERE session_id = 'queued-child'`,
+	).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")
 	assert.Equal(t, "new-start", gotStarted)

internal/db/legacy_schema_test.go

@@ -227,6 +227,96 @@ func TestOpenLegacySchemasPreservesArchiveAndRequestsResync(t *testing.T) {
 	}
 }
 
+func TestParserParentSessionIDMigrationBackfillsCurrentParent(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "legacy.db")
+	conn, err := sql.Open("sqlite3", makeDSN(path, false))
+	require.NoError(t, err)
+	conn.SetMaxOpenConns(1)
+
+	_, err = conn.Exec(v06LegacySchema)
+	require.NoError(t, err, "create legacy schema")
+	_, err = conn.Exec(`
+		INSERT INTO sessions (
+			id, project, machine, agent, parent_session_id
+		) VALUES (
+			'kid', 'project-a', 'local', 'claude', 'parsed-parent'
+		)`)
+	require.NoError(t, err, "insert legacy child")
+	_, err = conn.Exec(fmt.Sprintf(
+		"PRAGMA user_version = %d", dataVersion,
+	))
+	require.NoError(t, err, "set current data version")
+	require.NoError(t, conn.Close(), "close legacy database")
+
+	d, err := Open(path)
+	require.NoError(t, err, "open migrated database")
+	defer d.Close()
+
+	var got sql.NullString
+	err = d.getReader().QueryRow(`
+		SELECT parser_parent_session_id FROM sessions WHERE id = 'kid'
+	`).Scan(&got)
+	require.NoError(t, err, "query migrated parser parent")
+	require.True(t, got.Valid, "migrated parser parent must be set")
+	assert.Equal(t, "parsed-parent", got.String, "migrated parser parent")
+}
+
+func TestParserParentSessionIDMigrationRollsBackWhenBackfillFails(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "legacy.db")
+	conn, err := sql.Open("sqlite3", makeDSN(path, false))
+	require.NoError(t, err)
+	conn.SetMaxOpenConns(1)
+
+	_, err = conn.Exec(v06LegacySchema)
+	require.NoError(t, err, "create legacy schema")
+	_, err = conn.Exec(`
+		INSERT INTO sessions (
+			id, project, machine, agent, parent_session_id
+		) VALUES (
+			'kid', 'project-a', 'local', 'claude', 'parsed-parent'
+		);
+		CREATE TRIGGER fail_parser_parent_backfill
+		BEFORE UPDATE OF parser_parent_session_id ON sessions BEGIN
+			SELECT RAISE(ABORT, 'injected parser parent backfill failure');
+		END;`)
+	require.NoError(t, err, "prepare failing legacy migration")
+	_, err = conn.Exec(fmt.Sprintf(
+		"PRAGMA user_version = %d", dataVersion,
+	))
+	require.NoError(t, err, "set current data version")
+	require.NoError(t, conn.Close(), "close legacy database")
+
+	d, err := Open(path)
+	require.ErrorContains(t, err, "injected parser parent backfill failure")
+	require.Nil(t, d)
+
+	conn, err = sql.Open("sqlite3", makeDSN(path, false))
+	require.NoError(t, err, "reopen failed migration")
+	conn.SetMaxOpenConns(1)
+	var columnCount int
+	err = conn.QueryRow(`
+		SELECT count(*) FROM pragma_table_info('sessions')
+		WHERE name = 'parser_parent_session_id'
+	`).Scan(&columnCount)
+	require.NoError(t, err, "inspect schema after failed migration")
+	assert.Zero(t, columnCount, "failed migration must roll back added column")
+	_, err = conn.Exec(`DROP TRIGGER fail_parser_parent_backfill`)
+	require.NoError(t, err, "remove injected migration failure")
+	require.NoError(t, conn.Close(), "close failed migration database")
+
+	d, err = Open(path)
+	require.NoError(t, err, "retry migration")
+	defer d.Close()
+
+	var got sql.NullString
+	err = d.getReader().QueryRow(`
+		SELECT parser_parent_session_id FROM sessions WHERE id = 'kid'
+	`).Scan(&got)
+	require.NoError(t, err, "query retried parser parent")
+	require.True(t, got.Valid, "retried parser parent must be set")
+	assert.Equal(t, "parsed-parent", got.String, "retried parser parent")
+}
+
 func TestLegacySchemaAddsArtifactImportAuthorityNonDestructively(t *testing.T) {
 	path := filepath.Join(t.TempDir(), "legacy.db")
 	conn, err := sql.Open("sqlite3", makeDSN(path, false))