diff --git a/src/operations-runner/entrypoint-support/operations/operation-poll.ts b/src/operations-runner/entrypoint-support/operations/operation-poll.ts index cc44a5a2..61f16333 100644 --- a/src/operations-runner/entrypoint-support/operations/operation-poll.ts +++ b/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; } diff --git a/src/operations-runner/entrypoint-support/support/executor-registry.ts b/src/operations-runner/entrypoint-support/support/executor-registry.ts index 722d6829..ce886090 100644 --- a/src/operations-runner/entrypoint-support/support/executor-registry.ts +++ b/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); diff --git a/src/operations-runner/entrypoint-support/support/run-loop.ts b/src/operations-runner/entrypoint-support/support/run-loop.ts index 352e4148..21192a32 100644 --- a/src/operations-runner/entrypoint-support/support/run-loop.ts +++ b/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)); } diff --git a/src/operations-runner/treedx/remote-head-reconciliation-executor.ts b/src/operations-runner/treedx/remote-head-reconciliation-executor.ts new file mode 100644 index 00000000..cd7cd0f7 --- /dev/null +++ b/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; + 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 }; + }, + }; +} diff --git a/src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts b/src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts new file mode 100644 index 00000000..acf491d9 --- /dev/null +++ b/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 { + if (!value || typeof value !== 'object' || Array.isArray(value)) return {}; + return value as Record; +} + +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 }; + } +} diff --git a/tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts b/tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts new file mode 100644 index 00000000..eccc837c --- /dev/null +++ b/tests/unit/control-plane/treedx/remote-head-reconciliation.test.ts @@ -0,0 +1,116 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const { resolveCredential, resolveConnection, createDelivery, enqueueReplication } = vi.hoisted(() => ({ + resolveCredential: vi.fn(), resolveConnection: vi.fn(), createDelivery: vi.fn(), enqueueReplication: vi.fn(), +})); + +vi.mock('../../../../src/security/provider-credential-authority.ts', () => ({ + resolveGitHubCredentialAuthority: resolveCredential, +})); +vi.mock('../../../../src/api/knowledge/gateway-treedx-connection.ts', () => ({ + resolveKnowledgeGatewayConnection: resolveConnection, +})); +vi.mock('../../../../src/security/remote-git-credential-delivery.ts', () => ({ + createRemoteGitCredentialDelivery: createDelivery, +})); +vi.mock('../../../../src/api/capacity/services/treedx/repositories/treedx-commit-replication.ts', () => ({ + enqueueTreeDxCommitReplication: enqueueReplication, +})); + +import { TreeDxRemoteHeadReconciliationScheduler } from '../../../../src/operations-runner/treedx/remote-head-reconciliation-scheduler.ts'; +import { createTreeDxRemoteHeadReconciliationExecutor } from '../../../../src/operations-runner/treedx/remote-head-reconciliation-executor.ts'; + +const oldHead = 'a'.repeat(40); +const newHead = 'b'.repeat(40); +const publicationRef = 'refs/heads/staging'; + +function githubFetch(head = newHead) { + return vi.fn(async () => new Response(JSON.stringify({ object: { sha: head } }), { + status: 200, headers: { 'content-type': 'application/json' }, + })) as unknown as typeof fetch; +} + +beforeEach(() => { + vi.clearAllMocks(); + resolveCredential.mockResolvedValue({ token: 'token' }); + createDelivery.mockResolvedValue({ deliveryId: 'delivery' }); + enqueueReplication.mockResolvedValue({ id: 'replication' }); +}); + +describe('TreeDX protected branch reconciliation', () => { + it('queues one idempotent operation when GitHub is newer than the current TreeDX view', async () => { + 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: oldHead, + metadata_json: JSON.stringify({ resolvedRef: oldHead }) }]; }, + async createPlatformOperation(value: any) { operations.push(value); }, + }; + const scheduler = new TreeDxRemoteHeadReconciliationScheduler(store, 1, githubFetch()); + const first = await scheduler.runIfDue(Date.parse('2026-08-31T12:00:00.000Z')); + expect(first).toMatchObject({ scheduled: true, observed: 1, queued: 1, failed: 0 }); + expect(operations[0]).toMatchObject({ namespace: 'treedx', operation: 'reconcile_remote_head', + input: { teamId: 'team', projectId: 'project', publicationRef, remoteHead: newHead } }); + expect(operations[0].idempotencyKey).toMatch(/^treedx-remote-head:[a-f0-9]{64}$/u); + }); + + it('does not queue when the logical view, binding, and remote are already converged', async () => { + 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, + metadata_json: JSON.stringify({ resolvedRef: newHead }) }]; }, + async createPlatformOperation(value: any) { operations.push(value); } }; + 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 () => { + const runs: Array<{ query: string; params: unknown[] }> = []; + const upserts: any[] = []; + let refs = [{ name: publicationRef, target: oldHead }, { name: 'refs/remotes/origin/staging', 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' })), + }; + resolveConnection.mockResolvedValue({ client, repositoryId: 'repository', nodeId: 'node' }); + const library = { contentPath: '.', contentRepositoryUrl: 'https://github.com/treeseed-ai/sdk-library.git', + contentRepositoryDefaultBranch: 'main', metadata: { retained: true, upstreamHeads: {} } }; + const store: any = { + async first() { return { id: 'binding', provider_id: 'github', grant_status: 'ready', + publication_ref: publicationRef, authority_id: 'authority', owner: 'treeseed-ai', name: 'sdk-library', + clone_url: library.contentRepositoryUrl }; }, + async getProjectTreeDxLibrary() { return library; }, + async upsertProjectTreeDxLibrary(_projectId: string, value: any) { upserts.push(value); return value; }, + async run(query: string, params: unknown[]) { runs.push({ query, params }); }, + }; + const checkpoints: any[] = []; + 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, + 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, + })); + expect(checkpoints).toHaveLength(1); + expect(result).toMatchObject({ projectId: 'project', remoteHead: newHead, + replication: { id: 'replication' } }); + }); +});