Complete live protected library reconciliation

src/api/knowledge/gateway-treedx-connection.ts

@@ -95,13 +95,15 @@ export async function resolveKnowledgeGatewayConnection(store: any, input: {
 		scope: { repositoryIds: [repositoryId], capabilities: input.replicationRefs?.length
 			? ['repos:read', 'files:read', 'git:read', 'git:fetch', 'git:push', 'registry:read', 'snapshot:build', 'artifact:export']
 			: input.maintenanceRefs?.length
-			? ['repos:read', 'files:read', 'git:read', 'git:diff', 'git:fetch', 'git:push', 'registry:read', 'policy:write']
+			? ['repos:read', 'files:read', 'files:search', 'git:read', 'git:diff', 'git:fetch', 'git:push',
+				'registry:read', 'graph:query', 'graph:refresh', 'policy:write']
 			: input.publishRefs?.length
 			? ['repos:read', 'files:read', 'files:search', 'git:read', 'git:fetch', 'git:push', 'registry:read', 'graph:query', 'graph:refresh']
 			: input.write
 			? ['repos:read', 'repos:write', 'workspace:create', 'files:read', 'files:search', 'files:write', 'files:delete', 'git:read', 'git:diff', 'git:commit', 'graph:query', 'graph:refresh']
 			: ['repos:read', 'files:read', 'files:search', 'git:read', 'git:diff', 'graph:query'],
 		refs: [...new Set([text(library.contentRepositoryRef, library.contentRepositoryDefaultBranch, 'main'),
+			canonicalTreeDxBranchRef(library.contentRepositoryDefaultBranch ?? 'main'),
 			...(input.write || input.communicationPaths || input.authoringPaths ? [canonicalAuthoringRef] : []),
 			...(input.readRefs ?? []), ...(input.publishRefs ?? []), ...(input.maintenanceRefs ?? []), ...(input.replicationRefs ?? []),
 			...(input.workspaceRefs ?? [])])],

src/api/store/operations/lifecycle/claims/claim-platform-operation.ts

@@ -17,15 +17,17 @@ export async function claimPlatformOperationMethod(this: ControlPlaneStore, inpu
 				    OR (status IN ('leased', 'running') AND lease_expires_at IS NOT NULL AND lease_expires_at < ?)
 				 )
 				 ${capabilityWhere}
-				 ORDER BY CASE namespace WHEN 'knowledge' THEN 0 WHEN 'feedback' THEN 1 ELSE 2 END,
+				 ORDER BY CASE WHEN namespace='knowledge' THEN 0 WHEN namespace='feedback' THEN 1
+				               WHEN namespace='treedx' AND operation='reconcile_remote_head' THEN 2 ELSE 3 END,
 				          created_at ASC LIMIT ?`, [input.operationId, now, ...capabilities, limit])
         : await this.all(`SELECT * FROM platform_operations
 				 WHERE (
 				    status = 'queued'
 				    OR (status IN ('leased', 'running') AND lease_expires_at IS NOT NULL AND lease_expires_at < ?)
 				 )
 				 ${capabilityWhere}
-				 ORDER BY CASE namespace WHEN 'knowledge' THEN 0 WHEN 'feedback' THEN 1 ELSE 2 END,
+				 ORDER BY CASE WHEN namespace='knowledge' THEN 0 WHEN namespace='feedback' THEN 1
+				               WHEN namespace='treedx' AND operation='reconcile_remote_head' THEN 2 ELSE 3 END,
 				          created_at ASC LIMIT ?`, [now, ...capabilities, limit]);
     const row = rows[0];
     if (!row)

src/operations-runner/treedx/remote-head-reconciliation-executor.ts

@@ -68,23 +68,14 @@ export function createTreeDxRemoteHeadReconciliationExecutor(options: any) {
 			await connection.client.fetchRemote({ repoId: connection.repositoryId, remoteName: 'origin',
 				remoteUrl: binding.clone_url, credentialId: delivery.deliveryId,
 				refspecs: [`+${publicationRef}:${remoteRef}`] });
-			let repositoryRefs = refs(await connection.client.upstream.repositories.refs(connection.repositoryId));
+			const repositoryRefs = refs(await connection.client.upstream.repositories.refs(connection.repositoryId));
 			if (head(repositoryRefs, remoteRef) !== remoteHead) throw new Error('TreeDX did not fetch the protected branch head exactly.');
-			const beforeHead = head(repositoryRefs, publicationRef);
-			let promotion: any = { status: 'already_current', beforeHead, afterHead: beforeHead };
-			if (beforeHead !== remoteHead) {
-				promotion = await connection.client.promoteRef({ repoId: connection.repositoryId, sourceRef: remoteRef,
-					destinationRef: publicationRef, expectedDestinationHead: beforeHead });
-				if (promotion.afterHead !== remoteHead) throw new Error('TreeDX did not advance its current logical view to the protected branch head.');
-			}
-			repositoryRefs = refs(await connection.client.upstream.repositories.refs(connection.repositoryId));
-			if (head(repositoryRefs, publicationRef) !== remoteHead) throw new Error('TreeDX current logical view failed exact read-back verification.');
 			const graph = await completedGraphRefresh(connection.client, { repoId: connection.repositoryId,
-				ref: publicationRef, paths: ['**'], forceFull: true });
+				ref: remoteRef, paths: ['**'], forceFull: true });
 			await connection.client.refreshSearchIndex({ repoId: connection.repositoryId,
-				ref: publicationRef, paths: ['**'], incremental: false });
+				ref: remoteRef, paths: ['**'], incremental: false });
 			const search = result(await connection.client.upstream.searchIndex.status(connection.repositoryId,
-				{ ref: publicationRef }), 'index');
+				{ ref: remoteRef }), 'index');
 			if (String(graph.resolvedRef ?? remoteHead) !== remoteHead || String(search.resolvedRef ?? '') !== remoteHead
 				|| search.ready !== true || search.stale === true || !(Number(search.segmentCount) > 0)) {
 				throw new Error('TreeDX graph/search did not converge on the protected branch head.');
@@ -98,16 +89,16 @@ export function createTreeDxRemoteHeadReconciliationExecutor(options: any) {
 				repositoryId: connection.repositoryId, contentPath: library.contentPath,
 				contentRepositoryUrl: library.contentRepositoryUrl,
 				contentRepositoryDefaultBranch: library.contentRepositoryDefaultBranch,
-				contentRepositoryRef: publicationRef, metadata,
+				contentRepositoryRef: remoteRef, metadata,
 			});
 			if (!updated) throw new Error('TreeDX library binding could not be updated after reconciliation.');
 			await store.run(`UPDATE project_remote_repository_bindings SET expected_head=?,observed_head=?,drift='none',
 				version=version+1,updated_at=? WHERE id=?`, [remoteHead, remoteHead, now, binding.id]);
 			const replication = await enqueueTreeDxCommitReplication(store, { teamId, projectId, commitSha: remoteHead,
-				sourceRef: publicationRef, createdAt: now });
+				sourceRef: remoteRef, createdAt: now });
 			await context.checkpoint({ phase: 'treedx.remote-head.reconciled', projectId, remoteHead },
 				{ kind: 'treedx.remote-head.reconciled', data: { projectId, remoteHead, publicationRef } });
-			return { projectId, publicationRef, remoteHead, promotion, graph, search, replication };
+			return { projectId, publicationRef, currentLogicalRef: remoteRef, remoteHead, graph, search, replication };
 		},
 	};
 }

src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts

@@ -44,13 +44,17 @@ export class TreeDxRemoteHeadReconciliationScheduler {
 				const metadata = parse(binding.metadata_json);
 				const currentResolvedRef = String(metadata.resolvedRef ?? '');
 				const canonicalRef = String(binding.content_repository_ref ?? '');
+				const remoteRef = `refs/remotes/origin/${publicationRef.slice('refs/heads/'.length)}`;
 				if (binding.expected_head === remoteHead && binding.observed_head === remoteHead
-					&& currentResolvedRef === remoteHead && canonicalRef === publicationRef) continue;
+					&& currentResolvedRef === remoteHead && canonicalRef === remoteRef) continue;
 				const idempotencyKey = operationKey(String(binding.project_id), publicationRef, remoteHead);
-				await this.store.createPlatformOperation({ namespace: 'treedx', operation: 'reconcile_remote_head',
+				const operation = await this.store.createPlatformOperation({ namespace: 'treedx', operation: 'reconcile_remote_head',
 					target: 'control_plane_operations_runner', idempotencyKey,
 					input: { teamId: binding.team_id, projectId: binding.project_id, publicationRef, remoteHead },
 					requestedByType: 'service', requestedById: 'treedx-remote-head-reconciliation-scheduler' });
+				if (['failed', 'cancelled'].includes(String(operation?.status ?? ''))) {
+					await this.store.retryPlatformOperation(operation.id);
+				}
 				queued += 1;
 			} catch {
 				// A broken provider binding must not block reconciliation for other projects.

tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts

@@ -46,7 +46,7 @@ describe('TreeDX protected branch reconciliation', () => {
 				owner: 'treeseed-ai', name: 'sdk-library', publication_ref: publicationRef,
 				expected_head: newHead, observed_head: newHead, content_repository_ref: oldHead,
 				metadata_json: JSON.stringify({ resolvedRef: oldHead }) }]; },
-			async createPlatformOperation(value: any) { operations.push(value); },
+			async createPlatformOperation(value: any) { operations.push(value); return { id: 'operation', status: 'queued' }; },
 		};
 		const scheduler = new TreeDxRemoteHeadReconciliationScheduler(store, 1, githubFetch());
 		const first = await scheduler.runIfDue(Date.parse('2026-08-31T12:00:00.000Z'));
@@ -60,26 +60,39 @@ describe('TreeDX protected branch reconciliation', () => {
 		const operations: any[] = [];
 		const store: any = { config: {}, async all() { return [{ id: 'binding', team_id: 'team', project_id: 'project',
 			authority_id: 'authority', owner: 'treeseed-ai', name: 'sdk-library', publication_ref: publicationRef,
-			expected_head: newHead, observed_head: newHead, content_repository_ref: publicationRef,
+			expected_head: newHead, observed_head: newHead, content_repository_ref: 'refs/remotes/origin/staging',
 			metadata_json: JSON.stringify({ resolvedRef: newHead }) }]; },
-			async createPlatformOperation(value: any) { operations.push(value); } };
+			async createPlatformOperation(value: any) { operations.push(value); return { id: 'operation', status: 'queued' }; } };
 		const result = await new TreeDxRemoteHeadReconciliationScheduler(store, 1, githubFetch())
 			.runIfDue(Date.parse('2026-08-31T12:00:00.000Z'));
 		expect(result).toMatchObject({ queued: 0, failed: 0 });
 		expect(operations).toHaveLength(0);
 	});
 
-	it('fetches, promotes, indexes, advances the logical binding, and queues the exact R2 mirror', async () => {
+	it('retries the same exact-head operation after a recoverable failure', async () => {
+		const retried: string[] = [];
+		const store: any = { config: {}, async all() { return [{ id: 'binding', team_id: 'team', project_id: 'project',
+			authority_id: 'authority', owner: 'treeseed-ai', name: 'sdk-library', publication_ref: publicationRef,
+			expected_head: oldHead, observed_head: oldHead, content_repository_ref: publicationRef,
+			metadata_json: JSON.stringify({ resolvedRef: oldHead }) }]; },
+			async createPlatformOperation() { return { id: 'failed-operation', status: 'failed' }; },
+			async retryPlatformOperation(id: string) { retried.push(id); } };
+		const result = await new TreeDxRemoteHeadReconciliationScheduler(store, 1, githubFetch())
+			.runIfDue(Date.parse('2026-08-31T12:00:00.000Z'));
+		expect(result).toMatchObject({ queued: 1, failed: 0 });
+		expect(retried).toEqual(['failed-operation']);
+	});
+
+	it('fetches, indexes, advances the logical binding, and queues the exact R2 mirror', async () => {
 		const runs: Array<{ query: string; params: unknown[] }> = [];
 		const upserts: any[] = [];
-		let refs = [{ name: publicationRef, target: oldHead }, { name: 'refs/remotes/origin/staging', target: newHead }];
+		const remoteRef = 'refs/remotes/origin/staging';
+		const refs = [{ name: publicationRef, target: oldHead }, { name: remoteRef, target: newHead }];
 		const client: any = {
 			fetchRemote: vi.fn(async () => ({})),
 			upstream: { repositories: { refs: vi.fn(async () => ({ refs })) },
 				searchIndex: { status: vi.fn(async () => ({ index: { ready: true, stale: false,
 					resolvedRef: newHead, segmentCount: 4 } })) } },
-			promoteRef: vi.fn(async () => { refs = [{ name: publicationRef, target: newHead },
-				{ name: 'refs/remotes/origin/staging', target: newHead }]; return { beforeHead: oldHead, afterHead: newHead }; }),
 			refreshGraph: vi.fn(async () => ({ graph: { status: 'completed', resolvedRef: newHead, graphVersion: 'graph-1' } })),
 			refreshSearchIndex: vi.fn(async () => ({ index: { status: 'completed' } })),
 			getPlacement: vi.fn(async () => ({ primaryNodeId: 'node' })),
@@ -99,15 +112,14 @@ describe('TreeDX protected branch reconciliation', () => {
 		const executor = createTreeDxRemoteHeadReconciliationExecutor({ controlPlaneStore: store, fetchImpl: githubFetch() });
 		const result = await executor.run({ teamId: 'team', projectId: 'project', publicationRef, remoteHead: newHead },
 			{ operation: { id: 'operation' }, checkpoint: async (...values: any[]) => checkpoints.push(values) });
-		expect(client.fetchRemote).toHaveBeenCalledWith(expect.objectContaining({ refspecs: [`+${publicationRef}:refs/remotes/origin/staging`] }));
-		expect(client.promoteRef).toHaveBeenCalledWith(expect.objectContaining({ expectedDestinationHead: oldHead }));
-		expect(client.refreshGraph).toHaveBeenCalledWith(expect.objectContaining({ ref: publicationRef, forceFull: true }));
-		expect(client.refreshSearchIndex).toHaveBeenCalledWith(expect.objectContaining({ ref: publicationRef, incremental: false }));
-		expect(upserts[0]).toMatchObject({ contentRepositoryRef: publicationRef,
+		expect(client.fetchRemote).toHaveBeenCalledWith(expect.objectContaining({ refspecs: [`+${publicationRef}:${remoteRef}`] }));
+		expect(client.refreshGraph).toHaveBeenCalledWith(expect.objectContaining({ ref: remoteRef, forceFull: true }));
+		expect(client.refreshSearchIndex).toHaveBeenCalledWith(expect.objectContaining({ ref: remoteRef, incremental: false }));
+		expect(upserts[0]).toMatchObject({ contentRepositoryRef: remoteRef,
 			metadata: { retained: true, resolvedRef: newHead } });
 		expect(runs.some((entry) => entry.query.includes('UPDATE project_remote_repository_bindings'))).toBe(true);
 		expect(enqueueReplication).toHaveBeenCalledWith(store, expect.objectContaining({
-			teamId: 'team', projectId: 'project', commitSha: newHead, sourceRef: publicationRef,
+			teamId: 'team', projectId: 'project', commitSha: newHead, sourceRef: remoteRef,
 		}));
 		expect(checkpoints).toHaveLength(1);
 		expect(result).toMatchObject({ projectId: 'project', remoteHead: newHead,