Reconcile protected library merges into TreeDX

src/operations-runner/entrypoint-support/operations/operation-poll.ts

@@ -4,6 +4,7 @@ import { ContextQueryCheckMaintenanceScheduler } from '../../../api/capacity/ser
 import { ContextQueryCheckService } from '../../../api/capacity/services/capacity/agents/context-query-check-service.js';
 import { FeedbackRetentionScheduler } from '../../feedback/retention-scheduler.js';
 import { TreeDxCommitReplicationScheduler } from '../../treedx/commit-replication-scheduler.js';
+import { TreeDxRemoteHeadReconciliationScheduler } from '../../treedx/remote-head-reconciliation-scheduler.js';
 import { createClient,createControlPlaneStore,createExecutorsForOptions,loadConfig,packageVersion,registerAndHeartbeat } from '../index.js';
 import { runPlatformOperationOnce } from './operation-execution.js';
 
@@ -45,6 +46,8 @@ export async function runOnce(options: any = {}) {
 			await feedbackRetention.runIfDue();
 			const treeDxCommitReplication = new TreeDxCommitReplicationScheduler(controlPlaneStore);
 			await treeDxCommitReplication.runIfDue();
+			const treeDxRemoteHeadReconciliation = new TreeDxRemoteHeadReconciliationScheduler(controlPlaneStore);
+			await treeDxRemoteHeadReconciliation.runIfDue();
         }
         return result;
     }

src/operations-runner/entrypoint-support/support/executor-registry.ts

@@ -4,6 +4,7 @@ import { createKnowledgePackCleanupExecutor, createKnowledgePackExecutor } from
 import { createGitHubWorkflowExecutor } from '../../workflows/github-workflow-executor.ts';
 import { createGitHubConfigurationExecutor } from '../../workflows/github-configuration-executor.ts';
 import { createTreeDxCommitReplicationExecutor } from '../../treedx/commit-replication-executor.ts';
+import { createTreeDxRemoteHeadReconciliationExecutor } from '../../treedx/remote-head-reconciliation-executor.ts';
 
 export function createExecutors() {
 	return createExecutorsForOptions({});
@@ -28,6 +29,7 @@ export function createExecutorsForOptions(options: any = {}) {
 		createKnowledgePackExecutor(options),
 		createKnowledgePackCleanupExecutor(options),
 		createTreeDxCommitReplicationExecutor(options),
+		createTreeDxRemoteHeadReconciliationExecutor(options),
 		workflowExecutor,
 		workflowConfigurationExecutor,
 	].filter((executor) => !options.operationKey || `${executor.namespace}:${executor.operation}` === options.operationKey);

src/operations-runner/entrypoint-support/support/run-loop.ts

@@ -5,6 +5,7 @@ import { randomUUID } from 'node:crypto';
 import { drainNotificationEmailOutbox } from '../../../notifications/service.js';
 import { FeedbackRetentionScheduler } from '../../feedback/retention-scheduler.js';
 import { TreeDxCommitReplicationScheduler } from '../../treedx/commit-replication-scheduler.js';
+import { TreeDxRemoteHeadReconciliationScheduler } from '../../treedx/remote-head-reconciliation-scheduler.js';
 import { createClient,createControlPlaneStore,loadConfig,loadHealthConfig,packageVersion,parseRunnerOptions,registerAndHeartbeat,runOnceWithClient,startHealthServer } from '../index.js';
 
 export async function runLoop() {
@@ -22,6 +23,7 @@ export async function runLoop() {
 	let feedbackRetention = null;
 	let contextQueryCheckMaintenance = null;
 	let treeDxCommitReplication = null;
+	let treeDxRemoteHeadReconciliation = null;
 	let operationRunnerId = null;
     while (!stopping) {
         try {
@@ -40,6 +42,7 @@ export async function runLoop() {
 					: null;
 				feedbackRetention = controlPlaneStore ? new FeedbackRetentionScheduler(controlPlaneStore, config.feedbackRetentionIntervalMs) : null;
 				treeDxCommitReplication = controlPlaneStore ? new TreeDxCommitReplicationScheduler(controlPlaneStore) : null;
+				treeDxRemoteHeadReconciliation = controlPlaneStore ? new TreeDxRemoteHeadReconciliationScheduler(controlPlaneStore) : null;
                 await registerAndHeartbeat(client, config, version, { ...options, controlPlaneStore });
             }
             healthState.ready = true;
@@ -52,6 +55,7 @@ export async function runLoop() {
 			await contextQueryCheckMaintenance?.runIfDue();
 			await feedbackRetention?.runIfDue();
 			await treeDxCommitReplication?.runIfDue();
+			await treeDxRemoteHeadReconciliation?.runIfDue();
         }
         catch (error) {
             healthState.ready = false;
@@ -71,6 +75,7 @@ export async function runLoop() {
 			contextQueryCheckMaintenance = null;
 			feedbackRetention = null;
 			treeDxCommitReplication = null;
+			treeDxRemoteHeadReconciliation = null;
         }
         await new Promise((resolveSleep) => setTimeout(resolveSleep, options.pollIntervalMs));
     }

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

@@ -0,0 +1,113 @@
+import { enqueueTreeDxCommitReplication } from '../../api/capacity/services/treedx/repositories/treedx-commit-replication.ts';
+import { resolveKnowledgeGatewayConnection } from '../../api/knowledge/gateway-treedx-connection.ts';
+import { githubRepositoryHead } from '../../providers/github/repository-client.ts';
+import { resolveGitHubCredentialAuthority } from '../../security/provider-credential-authority.ts';
+import { createRemoteGitCredentialDelivery } from '../../security/remote-git-credential-delivery.ts';
+
+function refs(value: unknown): any[] {
+	const rows = value && typeof value === 'object' && !Array.isArray(value) ? (value as any).refs : value;
+	return Array.isArray(rows) ? rows : [];
+}
+
+function head(rows: any[], name: string) {
+	const row = rows.find((candidate) => String(candidate?.name ?? '') === name);
+	return String(row?.target ?? row?.sha ?? '');
+}
+
+function result(value: unknown, key: string) {
+	if (!value || typeof value !== 'object' || Array.isArray(value)) return {};
+	const record = value as Record<string, any>;
+	return record[key] && typeof record[key] === 'object' ? record[key] : record;
+}
+
+async function completedGraphRefresh(client: any, input: any) {
+	const started = result(await client.refreshGraph(input), 'graph');
+	if (!started.jobId || started.status === 'completed') return started;
+	for (let attempt = 0; attempt < 60; attempt += 1) {
+		const current = result(await client.getGraphRefreshJob({ ...input, jobId: started.jobId }), 'job');
+		if (current.status === 'completed') return current;
+		if (current.status === 'failed') throw new Error(`TreeDX graph refresh failed: ${current.errorCode ?? 'unknown error'}.`);
+		await new Promise((resolve) => setTimeout(resolve, 250));
+	}
+	throw new Error('TreeDX graph refresh did not complete before remote reconciliation timed out.');
+}
+
+export function createTreeDxRemoteHeadReconciliationExecutor(options: any) {
+	return {
+		namespace: 'treedx', operation: 'reconcile_remote_head',
+		async run(input: any, context: any) {
+			const store = options.controlPlaneStore;
+			if (!store) throw new Error('TreeDX remote reconciliation requires a control-plane store.');
+			const projectId = String(input?.projectId ?? '');
+			const teamId = String(input?.teamId ?? '');
+			const requestedHead = String(input?.remoteHead ?? '');
+			const publicationRef = String(input?.publicationRef ?? '');
+			if (!projectId || !teamId || !/^[a-f0-9]{40}$/u.test(requestedHead)
+				|| !publicationRef.startsWith('refs/heads/')) throw new Error('Remote reconciliation input is invalid.');
+			const binding: any = await store.first(`SELECT * FROM project_remote_repository_bindings
+				WHERE project_id=? AND team_id=? LIMIT 1`, [projectId, teamId]);
+			if (!binding || binding.provider_id !== 'github' || binding.grant_status !== 'ready'
+				|| binding.publication_ref !== publicationRef) throw new Error('The GitHub library binding is not ready for reconciliation.');
+			const credential = await resolveGitHubCredentialAuthority({ store, authorityId: binding.authority_id,
+				repositoryBindingId: binding.id, capability: 'repository-hosting', fetchImpl: options.fetchImpl });
+			const remoteHead = await githubRepositoryHead(options.fetchImpl ?? fetch, credential.token,
+				binding.owner, binding.name, publicationRef);
+			if (remoteHead !== requestedHead) throw new Error('The protected library branch advanced while reconciliation was queued.');
+			const remoteRef = `refs/remotes/origin/${publicationRef.slice('refs/heads/'.length)}`;
+			const connection = await resolveKnowledgeGatewayConnection(store, { projectId, write: false,
+				maintenanceRefs: [publicationRef, remoteRef, remoteHead] });
+			if (!connection) throw new Error('The project TreeDX repository is unavailable.');
+			const placement: any = await connection.client.getPlacement(connection.repositoryId);
+			const nodeId = String(placement?.primaryNodeId ?? placement?.placement?.primaryNodeId ?? connection.nodeId ?? '');
+			if (!nodeId) throw new Error('TreeDX did not resolve the repository primary node for credential delivery.');
+			const delivery = await createRemoteGitCredentialDelivery({ store, operationId: context.operation.id,
+				actorId: 'treedx-remote-head-reconciler', teamId, projectId, repositoryBindingId: binding.id,
+				credentialAuthorityId: binding.authority_id, nodeId, sourceRef: publicationRef,
+				destinationRef: remoteRef, reviewedCommit: remoteHead, expectedRemoteHead: remoteHead,
+				purpose: 'fetch', refspec: `+${publicationRef}:${remoteRef}` });
+			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));
+			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 });
+			await connection.client.refreshSearchIndex({ repoId: connection.repositoryId,
+				ref: publicationRef, paths: ['**'], incremental: false });
+			const search = result(await connection.client.upstream.searchIndex.status(connection.repositoryId,
+				{ ref: publicationRef }), '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.');
+			}
+			const library: any = await store.getProjectTreeDxLibrary(projectId);
+			const now = new Date().toISOString();
+			const metadata = { ...(library?.metadata ?? {}), resolvedRef: remoteHead,
+				upstreamHeads: { ...(library?.metadata?.upstreamHeads ?? {}), [remoteRef]: remoteHead },
+				searchIndex: { ready: true, segmentCount: search.segmentCount ?? null }, reconciledAt: now };
+			const updated = await store.upsertProjectTreeDxLibrary(projectId, {
+				repositoryId: connection.repositoryId, contentPath: library.contentPath,
+				contentRepositoryUrl: library.contentRepositoryUrl,
+				contentRepositoryDefaultBranch: library.contentRepositoryDefaultBranch,
+				contentRepositoryRef: publicationRef, 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 });
+			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 };
+		},
+	};
+}

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

@@ -0,0 +1,62 @@
+import { createHash } from 'node:crypto';
+import { githubRepositoryHead } from '../../providers/github/repository-client.ts';
+import { resolveGitHubCredentialAuthority } from '../../security/provider-credential-authority.ts';
+
+function object(value: unknown): Record<string, any> {
+	if (!value || typeof value !== 'object' || Array.isArray(value)) return {};
+	return value as Record<string, any>;
+}
+
+function parse(value: unknown) {
+	if (typeof value !== 'string') return object(value);
+	try { return object(JSON.parse(value)); } catch { return {}; }
+}
+
+function operationKey(projectId: string, publicationRef: string, remoteHead: string) {
+	const digest = createHash('sha256').update(`${projectId}:${publicationRef}:${remoteHead}`).digest('hex');
+	return `treedx-remote-head:${digest}`;
+}
+
+export class TreeDxRemoteHeadReconciliationScheduler {
+	private lastAttemptAt = 0;
+
+	constructor(private readonly store: any, private readonly intervalMs = 30_000,
+		private readonly fetchImpl: typeof fetch = store.config?.fetchImpl ?? fetch) {}
+
+	async runIfDue(time = Date.now()) {
+		if (time - this.lastAttemptAt < this.intervalMs) return { scheduled: false };
+		this.lastAttemptAt = time;
+		const bindings: any[] = await this.store.all(`SELECT b.*,l.repository_id,l.content_repository_ref,l.metadata_json
+			FROM project_remote_repository_bindings b JOIN treedx_project_libraries l ON l.project_id=b.project_id
+			WHERE b.provider_id='github' AND b.grant_status='ready' AND l.repository_id IS NOT NULL
+			ORDER BY b.project_id`);
+		let observed = 0, queued = 0, failed = 0;
+		for (const binding of bindings) {
+			try {
+				const credential = await resolveGitHubCredentialAuthority({ store: this.store,
+					authorityId: binding.authority_id, repositoryBindingId: binding.id,
+					capability: 'repository-hosting', fetchImpl: this.fetchImpl });
+				const publicationRef = String(binding.publication_ref ?? '');
+				const remoteHead = await githubRepositoryHead(this.fetchImpl, credential.token,
+					String(binding.owner), String(binding.name), publicationRef);
+				if (!remoteHead) continue;
+				observed += 1;
+				const metadata = parse(binding.metadata_json);
+				const currentResolvedRef = String(metadata.resolvedRef ?? '');
+				const canonicalRef = String(binding.content_repository_ref ?? '');
+				if (binding.expected_head === remoteHead && binding.observed_head === remoteHead
+					&& currentResolvedRef === remoteHead && canonicalRef === publicationRef) continue;
+				const idempotencyKey = operationKey(String(binding.project_id), publicationRef, remoteHead);
+				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' });
+				queued += 1;
+			} catch {
+				// A broken provider binding must not block reconciliation for other projects.
+				failed += 1;
+			}
+		}
+		return { scheduled: true, observed, queued, failed };
+	}
+}