From eaea6f4725fd302c5b5b17818e83812b8325155a Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 23:32:41 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=87=AA=E5=8A=A8=E7=89=A9=E5=8C=96=20?= =?UTF-8?q?AgentRun=20managed=20repositories?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/agentrun.yaml | 36 ++ .../agentrun/git-mirror-proxy-connect.cjs | 55 ++ .../managed-repository-reconciler.mjs | 593 ++++++++++++++++++ scripts/src/agentrun-lanes.ts | 165 ++++- ...trun-managed-repository-reconciler.test.ts | 204 ++++++ .../agentrun-managed-repository-reconciler.ts | 258 ++++++++ scripts/src/agentrun/git-mirror.ts | 131 +++- scripts/src/agentrun/secrets.ts | 228 ++++++- 8 files changed, 1613 insertions(+), 57 deletions(-) create mode 100644 scripts/native/agentrun/git-mirror-proxy-connect.cjs create mode 100644 scripts/native/agentrun/managed-repository-reconciler.mjs create mode 100644 scripts/src/agentrun-managed-repository-reconciler.test.ts create mode 100644 scripts/src/agentrun-managed-repository-reconciler.ts diff --git a/config/agentrun.yaml b/config/agentrun.yaml index ae60ce2a..be845349 100644 --- a/config/agentrun.yaml +++ b/config/agentrun.yaml @@ -241,16 +241,52 @@ controlPlane: toolsImage: 127.0.0.1:5000/hwlab/hwlab-ci-node-tools:node22-alpine-bun-v1 syncJobPrefix: git-mirror-agentrun-nc01-v02-sync-manual flushJobPrefix: git-mirror-agentrun-nc01-v02-flush-manual + repositoryReconciler: + enabled: true + deploymentName: agentrun-nc01-v02-managed-repository-reconciler + configMapName: agentrun-nc01-v02-managed-repository-reconciler + serviceAccountName: agentrun-nc01-v02-managed-repository-reconciler + imagePullPolicy: IfNotPresent + cacheMountPath: /cache + stateDirectory: .agentrun-managed-repositories/nc01-v02 + reconcileIntervalMs: 30000 + fetchTimeoutMs: 240000 + shutdownGraceMs: 30000 + maxConcurrentRepositories: 2 + retry: + maxAttempts: 4 + initialDelayMs: 1000 + maxDelayMs: 15000 + freshness: + maxAgeMs: 180000 + readiness: + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 6 + lifecycle: + undeclaredRepositoryPolicy: retain + managedRefs: source-branch-only + resources: + requests: + cpu: 50m + memory: 96Mi + limits: + cpu: 500m + memory: 256Mi repositories: - key: agentrun repository: pikasTech/agentrun + remote: ssh://git@ssh.github.com:443/pikasTech/agentrun.git sourceBranch: v0.2 gitopsBranch: nc01-v0.2-gitops - key: unidesk repository: pikasTech/unidesk + remote: ssh://git@ssh.github.com:443/pikasTech/unidesk.git sourceBranch: master - key: agent_skills repository: pikasTech/agent_skills + remote: ssh://git@ssh.github.com:443/pikasTech/agent_skills.git sourceBranch: master database: mode: external-postgres diff --git a/scripts/native/agentrun/git-mirror-proxy-connect.cjs b/scripts/native/agentrun/git-mirror-proxy-connect.cjs new file mode 100644 index 00000000..3b6052f9 --- /dev/null +++ b/scripts/native/agentrun/git-mirror-proxy-connect.cjs @@ -0,0 +1,55 @@ +#!/usr/bin/env node +const net = require("node:net"); + +const [proxyHost, proxyPortRaw, targetHost, targetPortRaw] = process.argv.slice(2); +const proxyPort = Number.parseInt(proxyPortRaw || "", 10); +const targetPort = Number.parseInt(targetPortRaw || "", 10); +if (!proxyHost || !Number.isInteger(proxyPort) || !targetHost || !Number.isInteger(targetPort)) { + console.error("agentrun managed repository proxy-connect: invalid arguments"); + process.exit(64); +} + +let settled = false; +let tunnelEstablished = false; +function finish(code, message) { + if (settled) return; + settled = true; + if (message) console.error(`agentrun managed repository proxy-connect: ${message}`); + process.exit(code); +} + +const socket = net.createConnection({ host: proxyHost, port: proxyPort }); +let buffer = Buffer.alloc(0); +socket.setTimeout(15000, () => { + socket.destroy(); + finish(65, "proxy connection timeout"); +}); +socket.on("connect", () => socket.write(`CONNECT ${targetHost}:${targetPort} HTTP/1.1\r\nHost: ${targetHost}:${targetPort}\r\nProxy-Connection: Keep-Alive\r\n\r\n`)); +socket.on("error", (error) => finish(tunnelEstablished ? 69 : 66, error?.message || String(error))); +socket.on("close", () => finish(tunnelEstablished ? 0 : 68, tunnelEstablished ? null : "proxy closed before CONNECT completed")); +socket.on("data", function onData(chunk) { + buffer = Buffer.concat([buffer, chunk]); + const headerEnd = buffer.indexOf("\r\n\r\n"); + if (headerEnd === -1 && buffer.length < 8192) return; + if (headerEnd === -1) { + socket.destroy(); + finish(68, "proxy response header exceeded 8192 bytes"); + return; + } + const statusLine = buffer.subarray(0, headerEnd).toString("latin1").split("\r\n", 1)[0] || ""; + const statusCode = Number.parseInt(statusLine.split(" ")[1] || "", 10); + if (!statusLine.startsWith("HTTP/1.") || !Number.isInteger(statusCode) || statusCode < 200 || statusCode > 299) { + socket.destroy(); + finish(67, `proxy CONNECT failed: ${statusLine.replace(/[^\x20-\x7e]/g, "?").slice(0, 160)}`); + return; + } + socket.removeListener("data", onData); + socket.setTimeout(0); + tunnelEstablished = true; + const rest = buffer.subarray(headerEnd + 4); + if (rest.length > 0) process.stdout.write(rest); + process.stdin.on("error", () => {}); + process.stdout.on("error", () => {}); + process.stdin.pipe(socket); + socket.pipe(process.stdout); +}); diff --git a/scripts/native/agentrun/managed-repository-reconciler.mjs b/scripts/native/agentrun/managed-repository-reconciler.mjs new file mode 100644 index 00000000..81306db1 --- /dev/null +++ b/scripts/native/agentrun/managed-repository-reconciler.mjs @@ -0,0 +1,593 @@ +#!/usr/bin/env node +import { createHash, randomBytes } from "node:crypto"; +import { spawn } from "node:child_process"; +import { + existsSync, + mkdirSync, + readFileSync, + renameSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { dirname, join, resolve, sep } from "node:path"; + +const args = process.argv.slice(2); +const configPath = optionValue("--config"); +const mode = args.includes("--health") + ? "health" + : args.includes("--status") + ? "status" + : args.includes("--once") + ? "once" + : "run"; + +if (!configPath) failUsage("--config is required"); + +const config = readConfig(configPath); +const statePath = join(config.cacheMountPath, config.stateDirectory, "status.json"); +const activeChildren = new Set(); +let stopping = false; + +process.on("SIGTERM", beginShutdown); +process.on("SIGINT", beginShutdown); + +if (mode === "health" || mode === "status") { + const view = await currentStatus(config, statePath); + process.stdout.write(`${JSON.stringify(view)}\n`); + process.exit(mode === "health" && view.ready !== true ? 1 : 0); +} + +if (mode === "once") { + const state = await reconcileOnce(config, statePath); + process.stdout.write(`${JSON.stringify(compactStatus(config, state))}\n`); + process.exit(state.ready === true ? 0 : 1); +} + +log({ event: "managed-repository-reconciler-started", node: config.nodeId, lane: config.lane, desiredHash: config.desiredHash, repositoryCount: config.repositories.length }); +while (!stopping) { + const state = await reconcileOnce(config, statePath); + log({ + event: "managed-repository-reconcile-completed", + node: config.nodeId, + lane: config.lane, + ready: state.ready, + lastCycleOk: state.lastCycleOk, + repositoryCount: state.repositories.length, + desiredHash: config.desiredHash, + }); + if (!stopping) await sleep(config.reconcileIntervalMs); +} +log({ event: "managed-repository-reconciler-stopped", node: config.nodeId, lane: config.lane }); + +function optionValue(name) { + const index = args.indexOf(name); + if (index < 0) return null; + const value = args[index + 1]; + if (!value || value.startsWith("--")) failUsage(`${name} requires a value`); + return value; +} + +function failUsage(message) { + process.stderr.write(`${JSON.stringify({ ok: false, errorType: "invalid-arguments", message })}\n`); + process.exit(64); +} + +function readConfig(path) { + let raw; + try { + raw = JSON.parse(readFileSync(path, "utf8")); + } catch { + throw new Error("managed repository config is not readable JSON"); + } + requireObject(raw, "config"); + const desiredHash = requireString(raw.desiredHash, "desiredHash"); + const { desiredHash: _omitted, ...desired } = raw; + const computedHash = sha256(JSON.stringify(desired)); + if (computedHash !== desiredHash) throw new Error("managed repository config desiredHash does not match rendered content"); + if (raw.schemaVersion !== 1 || raw.kind !== "AgentRunManagedRepositoryReconciler") throw new Error("unsupported managed repository config envelope"); + const cacheMountPath = requireAbsolutePath(raw.cacheMountPath, "cacheMountPath"); + const stateDirectory = requireRelativePath(raw.stateDirectory, "stateDirectory"); + const repositories = requireArray(raw.repositories, "repositories").map((item, index) => { + requireObject(item, `repositories[${index}]`); + const key = requireSimpleId(item.key, `repositories[${index}].key`); + const repository = requireRepository(item.repository, `repositories[${index}].repository`); + const remote = requireString(item.remote, `repositories[${index}].remote`); + const sourceBranch = requireGitRef(item.sourceBranch, `repositories[${index}].sourceBranch`); + const desiredFingerprint = requireHex(item.desiredFingerprint, `repositories[${index}].desiredFingerprint`); + const expectedFingerprint = sha256([key, repository, remote, sourceBranch].join("\0")); + if (desiredFingerprint !== expectedFingerprint) throw new Error(`repositories[${index}].desiredFingerprint does not match repository desired state`); + return { key, repository, remote, sourceBranch, desiredFingerprint }; + }); + if (repositories.length === 0) throw new Error("repositories must not be empty"); + ensureUnique(repositories.map((item) => item.key), "repository keys"); + ensureUnique(repositories.map((item) => item.repository), "repository paths"); + ensureUnique(repositories.map((item) => item.remote), "repository remotes"); + requireObject(raw.retry, "retry"); + requireObject(raw.freshness, "freshness"); + requireObject(raw.lifecycle, "lifecycle"); + if (raw.lifecycle.undeclaredRepositoryPolicy !== "retain" || raw.lifecycle.managedRefs !== "source-branch-only") { + throw new Error("unsupported managed repository lifecycle policy"); + } + return { + schemaVersion: 1, + kind: raw.kind, + desiredHash, + nodeId: requireString(raw.nodeId, "nodeId"), + lane: requireString(raw.lane, "lane"), + cacheMountPath, + stateDirectory, + reconcileIntervalMs: requirePositiveInteger(raw.reconcileIntervalMs, "reconcileIntervalMs"), + fetchTimeoutMs: requirePositiveInteger(raw.fetchTimeoutMs, "fetchTimeoutMs"), + shutdownGraceMs: requirePositiveInteger(raw.shutdownGraceMs, "shutdownGraceMs"), + maxConcurrentRepositories: requirePositiveInteger(raw.maxConcurrentRepositories, "maxConcurrentRepositories"), + retry: { + maxAttempts: requirePositiveInteger(raw.retry.maxAttempts, "retry.maxAttempts"), + initialDelayMs: requirePositiveInteger(raw.retry.initialDelayMs, "retry.initialDelayMs"), + maxDelayMs: requirePositiveInteger(raw.retry.maxDelayMs, "retry.maxDelayMs"), + }, + freshness: { + maxAgeMs: requirePositiveInteger(raw.freshness.maxAgeMs, "freshness.maxAgeMs"), + }, + lifecycle: { + undeclaredRepositoryPolicy: "retain", + managedRefs: "source-branch-only", + }, + repositories, + ssh: raw.ssh && typeof raw.ssh === "object" ? { + identityFile: requireAbsolutePath(raw.ssh.identityFile, "ssh.identityFile"), + knownHostsFile: requireAbsolutePath(raw.ssh.knownHostsFile, "ssh.knownHostsFile"), + proxyConnectScript: requireAbsolutePath(raw.ssh.proxyConnectScript, "ssh.proxyConnectScript"), + proxyHost: requireString(raw.ssh.proxyHost, "ssh.proxyHost"), + proxyPort: requirePositiveInteger(raw.ssh.proxyPort, "ssh.proxyPort"), + } : null, + }; +} + +async function reconcileOnce(cfg, path) { + mkdirSync(dirname(path), { recursive: true }); + const previous = readState(path); + const previousByFingerprint = new Map( + Array.isArray(previous?.repositories) + ? previous.repositories.map((item) => [item?.desiredFingerprint, item]) + : [], + ); + const results = await mapConcurrent(cfg.repositories, cfg.maxConcurrentRepositories, async (repository) => { + const prior = previousByFingerprint.get(repository.desiredFingerprint) ?? null; + return reconcileWithRetry(cfg, repository, prior); + }); + const now = new Date().toISOString(); + const repositories = results.map((item) => withFreshness(item, cfg.freshness.maxAgeMs)); + const state = { + schemaVersion: 1, + kind: "AgentRunManagedRepositoryReconcilerStatus", + desiredHash: cfg.desiredHash, + nodeId: cfg.nodeId, + lane: cfg.lane, + observedAt: now, + heartbeatAt: now, + ready: repositories.every((item) => item.fresh === true && item.refPresent === true), + lastCycleOk: repositories.every((item) => item.lastAttemptOk === true), + lifecycle: cfg.lifecycle, + repositories, + valuesPrinted: false, + }; + writeState(path, state); + return state; +} + +async function reconcileWithRetry(cfg, repository, prior) { + let failure = null; + for (let attempt = 1; attempt <= cfg.retry.maxAttempts && !stopping; attempt += 1) { + try { + const result = await materializeRepository(cfg, repository); + return { + ...repositoryIdentity(repository), + refPresent: true, + commit: result.commit, + lastAttemptAt: new Date().toISOString(), + lastAttemptOk: true, + lastSuccessAt: new Date().toISOString(), + attempts: attempt, + errorCode: null, + errorFingerprint: null, + }; + } catch (error) { + failure = safeError(error); + log({ event: "managed-repository-reconcile-attempt-failed", repository: repository.repository, sourceBranch: repository.sourceBranch, attempt, errorCode: failure.code, errorFingerprint: failure.fingerprint }); + if (attempt < cfg.retry.maxAttempts && !stopping) { + const delayMs = Math.min(cfg.retry.maxDelayMs, cfg.retry.initialDelayMs * (2 ** (attempt - 1))); + await sleep(delayMs); + } + } + } + const inspected = await inspectRepository(cfg, repository); + const preserveSuccess = prior + && prior.desiredFingerprint === repository.desiredFingerprint + && prior.commit === inspected.commit + && typeof prior.lastSuccessAt === "string"; + return { + ...repositoryIdentity(repository), + refPresent: inspected.refPresent, + commit: inspected.commit, + lastAttemptAt: new Date().toISOString(), + lastAttemptOk: false, + lastSuccessAt: preserveSuccess ? prior.lastSuccessAt : null, + attempts: cfg.retry.maxAttempts, + errorCode: failure?.code ?? (stopping ? "shutdown" : "reconcile-failed"), + errorFingerprint: failure?.fingerprint ?? null, + }; +} + +async function materializeRepository(cfg, repository) { + const finalPath = repositoryPath(cfg, repository); + mkdirSync(dirname(finalPath), { recursive: true }); + const existing = existsSync(finalPath); + if (existing && !(await isBareRepository(finalPath, cfg.fetchTimeoutMs))) { + throw codedError("invalid-repository-path", "declared repository path exists but is not a bare Git repository"); + } + const workPath = existing + ? finalPath + : join(dirname(finalPath), `.${repository.key}.managed-tmp-${process.pid}-${randomBytes(6).toString("hex")}`); + const stageRef = `refs/agentrun-managed/stage/${repository.key}`; + try { + if (!existing) await git(cfg, ["init", "--bare", workPath]); + await git(cfg, ["--git-dir", workPath, "config", "uploadpack.allowReachableSHA1InWant", "true"]); + await git(cfg, ["--git-dir", workPath, "config", "uploadpack.allowAnySHA1InWant", "true"]); + await git(cfg, ["--git-dir", workPath, "config", "http.uploadpack", "true"]); + await git(cfg, ["--git-dir", workPath, "fetch", "--no-tags", repository.remote, `+refs/heads/${repository.sourceBranch}:${stageRef}`]); + const commit = (await git(cfg, ["--git-dir", workPath, "rev-parse", "--verify", `${stageRef}^{commit}`])).stdout.trim(); + if (!/^[0-9a-f]{40}$/u.test(commit)) throw codedError("invalid-source-commit", "fetched source ref did not resolve to a commit"); + await git(cfg, ["--git-dir", workPath, "update-ref", `refs/heads/${repository.sourceBranch}`, commit]); + await git(cfg, ["--git-dir", workPath, "update-ref", "-d", stageRef]); + await git(cfg, ["--git-dir", workPath, "update-server-info"]); + if (!existing) { + await git(cfg, ["--git-dir", workPath, "symbolic-ref", "HEAD", `refs/heads/${repository.sourceBranch}`]); + renameSync(workPath, finalPath); + } + return { commit }; + } catch (error) { + if (existing) { + try { await git(cfg, ["--git-dir", workPath, "update-ref", "-d", stageRef]); } catch {} + } else { + rmSync(workPath, { recursive: true, force: true }); + } + throw error; + } +} + +async function inspectRepository(cfg, repository) { + const path = repositoryPath(cfg, repository); + if (!existsSync(path) || !(await isBareRepository(path, Math.min(cfg.fetchTimeoutMs, 10000)))) return { refPresent: false, commit: null }; + try { + const result = await run("git", ["--git-dir", path, "rev-parse", "--verify", `refs/heads/${repository.sourceBranch}^{commit}`], { timeoutMs: Math.min(cfg.fetchTimeoutMs, 10000), env: process.env }); + const commit = result.stdout.trim(); + return { refPresent: /^[0-9a-f]{40}$/u.test(commit), commit: /^[0-9a-f]{40}$/u.test(commit) ? commit : null }; + } catch { + return { refPresent: false, commit: null }; + } +} + +async function isBareRepository(path, timeoutMs) { + try { + const result = await run("git", ["--git-dir", path, "rev-parse", "--is-bare-repository"], { timeoutMs, env: process.env }); + return result.stdout.trim() === "true"; + } catch { + return false; + } +} + +async function git(cfg, gitArgs) { + const env = { ...process.env, GIT_TERMINAL_PROMPT: "0" }; + if (cfg.ssh) { + mkdirSync(dirname(cfg.ssh.knownHostsFile), { recursive: true }); + const proxyCommand = `node ${shellWord(cfg.ssh.proxyConnectScript)} ${shellWord(cfg.ssh.proxyHost)} ${cfg.ssh.proxyPort} %h %p`; + env.GIT_SSH_COMMAND = [ + "ssh", + "-i", shellWord(cfg.ssh.identityFile), + "-o", "IdentitiesOnly=yes", + "-o", "BatchMode=yes", + "-o", "StrictHostKeyChecking=accept-new", + "-o", `UserKnownHostsFile=${shellWord(cfg.ssh.knownHostsFile)}`, + "-o", "ConnectTimeout=15", + "-o", `ProxyCommand=${shellWord(proxyCommand)}`, + ].join(" "); + } + return run("git", gitArgs, { timeoutMs: cfg.fetchTimeoutMs, env }); +} + +async function currentStatus(cfg, path) { + const state = readState(path); + if (!state || state.desiredHash !== cfg.desiredHash || !Array.isArray(state.repositories)) { + return { + ok: false, + ready: false, + desiredHash: cfg.desiredHash, + observedDesiredHash: state?.desiredHash ?? null, + reason: state ? "desired-state-mismatch" : "status-missing", + repositoryCount: cfg.repositories.length, + valuesPrinted: false, + }; + } + const stateByFingerprint = new Map(state.repositories.map((item) => [item?.desiredFingerprint, item])); + const repositories = []; + for (const desired of cfg.repositories) { + const observed = stateByFingerprint.get(desired.desiredFingerprint) ?? null; + const inspected = await inspectRepository(cfg, desired); + const desiredIdentity = repositoryIdentity(desired); + const item = withFreshness({ + ...desiredIdentity, + remoteFingerprint: observed?.remoteFingerprint ?? null, + remoteFingerprintAligned: observed?.remoteFingerprint === desiredIdentity.remoteFingerprint, + refPresent: inspected.refPresent, + commit: inspected.commit, + lastAttemptAt: observed?.lastAttemptAt ?? null, + lastAttemptOk: observed?.lastAttemptOk === true, + lastSuccessAt: observed?.commit === inspected.commit ? observed?.lastSuccessAt ?? null : null, + attempts: Number.isInteger(observed?.attempts) ? observed.attempts : null, + errorCode: observed?.errorCode ?? null, + errorFingerprint: observed?.errorFingerprint ?? null, + }, cfg.freshness.maxAgeMs); + repositories.push(item); + } + const heartbeatAgeMs = isoAgeMs(state.heartbeatAt); + const heartbeatFresh = heartbeatAgeMs !== null && heartbeatAgeMs <= cfg.freshness.maxAgeMs; + const ready = heartbeatFresh && repositories.every((item) => item.refPresent === true && item.fresh === true && item.remoteFingerprintAligned === true); + return { + ok: ready, + ready, + desiredHash: cfg.desiredHash, + observedDesiredHash: state.desiredHash, + nodeId: cfg.nodeId, + lane: cfg.lane, + heartbeatAt: state.heartbeatAt ?? null, + heartbeatAgeMs, + heartbeatFresh, + lifecycle: cfg.lifecycle, + repositories, + valuesPrinted: false, + }; +} + +function compactStatus(cfg, state) { + return { + ok: state.ready === true, + ready: state.ready === true, + desiredHash: cfg.desiredHash, + nodeId: cfg.nodeId, + lane: cfg.lane, + heartbeatAt: state.heartbeatAt, + lifecycle: cfg.lifecycle, + repositories: state.repositories, + valuesPrinted: false, + }; +} + +function repositoryIdentity(repository) { + return { + key: repository.key, + repository: repository.repository, + sourceBranch: repository.sourceBranch, + desiredFingerprint: repository.desiredFingerprint, + remoteFingerprint: `sha256:${sha256(repository.remote)}`, + }; +} + +function withFreshness(item, maxAgeMs) { + const ageMs = isoAgeMs(item.lastSuccessAt); + return { + ...item, + lastSuccessAgeMs: ageMs, + fresh: item.refPresent === true && ageMs !== null && ageMs <= maxAgeMs, + }; +} + +function repositoryPath(cfg, repository) { + const root = resolve(cfg.cacheMountPath); + const path = resolve(root, `${repository.repository}.git`); + if (path !== root && !path.startsWith(`${root}${sep}`)) throw codedError("unsafe-repository-path", "repository path escapes cache root"); + return path; +} + +function readState(path) { + try { + return JSON.parse(readFileSync(path, "utf8")); + } catch { + return null; + } +} + +function writeState(path, state) { + mkdirSync(dirname(path), { recursive: true }); + const temporary = `${path}.tmp-${process.pid}-${randomBytes(4).toString("hex")}`; + writeFileSync(temporary, `${JSON.stringify(state)}\n`, { mode: 0o600 }); + renameSync(temporary, path); +} + +async function mapConcurrent(items, concurrency, worker) { + const results = new Array(items.length); + let index = 0; + async function consume() { + while (!stopping) { + const current = index; + index += 1; + if (current >= items.length) return; + results[current] = await worker(items[current], current); + } + } + await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, () => consume())); + for (let current = 0; current < results.length; current += 1) { + if (results[current] === undefined) { + const item = items[current]; + results[current] = { + ...repositoryIdentity(item), + refPresent: false, + commit: null, + lastAttemptAt: new Date().toISOString(), + lastAttemptOk: false, + lastSuccessAt: null, + attempts: 0, + errorCode: "shutdown", + errorFingerprint: null, + }; + } + } + return results; +} + +function run(command, commandArgs, options) { + return new Promise((resolvePromise, rejectPromise) => { + const child = spawn(command, commandArgs, { env: options.env, stdio: ["ignore", "pipe", "pipe"] }); + activeChildren.add(child); + let stdout = ""; + let stderr = ""; + let timedOut = false; + const timer = setTimeout(() => { + timedOut = true; + child.kill("SIGKILL"); + }, options.timeoutMs); + child.stdout.on("data", (chunk) => { if (stdout.length < 32768) stdout += chunk.toString("utf8"); }); + child.stderr.on("data", (chunk) => { if (stderr.length < 32768) stderr += chunk.toString("utf8"); }); + child.on("error", (error) => { + clearTimeout(timer); + activeChildren.delete(child); + rejectPromise(codedError("command-start-failed", error.message)); + }); + child.on("close", (code, signal) => { + clearTimeout(timer); + activeChildren.delete(child); + if (timedOut) return rejectPromise(codedError("fetch-timeout", `${command} exceeded timeout`)); + if (code !== 0) return rejectPromise(codedError("command-failed", `${command} exited ${code ?? signal}: ${stderr || stdout}`)); + resolvePromise({ stdout, stderr }); + }); + }); +} + +function beginShutdown() { + if (stopping) return; + stopping = true; + const timer = setTimeout(() => { + for (const child of activeChildren) child.kill("SIGKILL"); + }, config.shutdownGraceMs); + timer.unref(); +} + +function sleep(ms) { + if (stopping) return Promise.resolve(); + return new Promise((resolvePromise) => { + let settled = false; + const finish = () => { + if (settled) return; + settled = true; + clearTimeout(timer); + clearInterval(check); + resolvePromise(); + }; + const timer = setTimeout(finish, ms); + const check = setInterval(() => { + if (stopping) finish(); + }, Math.min(ms, 100)); + }); +} + +function safeError(error) { + const message = error instanceof Error ? error.message : String(error); + const lower = message.toLowerCase(); + const code = error?.code && typeof error.code === "string" && error.code !== "command-failed" + ? error.code + : lower.includes("couldn't find remote ref") || lower.includes("remote ref does not exist") + ? "source-ref-missing" + : lower.includes("repository not found") || lower.includes("not found") + ? "repository-not-found" + : lower.includes("permission denied") || lower.includes("publickey") || lower.includes("authentication") + ? "authentication-failed" + : lower.includes("timeout") || lower.includes("timed out") + ? "fetch-timeout" + : "fetch-failed"; + return { code, fingerprint: sha256(message).slice(0, 16) }; +} + +function codedError(code, message) { + const error = new Error(message); + error.code = code; + return error; +} + +function isoAgeMs(value) { + if (typeof value !== "string") return null; + const time = Date.parse(value); + return Number.isFinite(time) ? Math.max(0, Date.now() - time) : null; +} + +function shellWord(value) { + return `'${String(value).replaceAll("'", `'"'"'`)}'`; +} + +function sha256(value) { + return createHash("sha256").update(value).digest("hex"); +} + +function log(value) { + process.stdout.write(`${JSON.stringify({ ...value, valuesPrinted: false })}\n`); +} + +function requireObject(value, path) { + if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error(`${path} must be an object`); + return value; +} + +function requireArray(value, path) { + if (!Array.isArray(value)) throw new Error(`${path} must be an array`); + return value; +} + +function requireString(value, path) { + if (typeof value !== "string" || value.length === 0) throw new Error(`${path} must be a non-empty string`); + return value; +} + +function requirePositiveInteger(value, path) { + if (!Number.isInteger(value) || value <= 0) throw new Error(`${path} must be a positive integer`); + return value; +} + +function requireAbsolutePath(value, path) { + const text = requireString(value, path); + if (!text.startsWith("/") || text.includes("..")) throw new Error(`${path} must be an absolute path without ..`); + return text; +} + +function requireRelativePath(value, path) { + const text = requireString(value, path); + if (text.startsWith("/") || text.includes("..") || text === ".") throw new Error(`${path} must be a dedicated relative path without ..`); + return text; +} + +function requireSimpleId(value, path) { + const text = requireString(value, path); + if (!/^[A-Za-z0-9._-]+$/u.test(text)) throw new Error(`${path} must use a simple id`); + return text; +} + +function requireRepository(value, path) { + const text = requireString(value, path); + if (!/^[A-Za-z0-9._-]+\/[A-Za-z0-9._-]+$/u.test(text) || text.includes("..")) throw new Error(`${path} must be owner/name without ..`); + return text; +} + +function requireGitRef(value, path) { + const text = requireString(value, path); + if (text.startsWith("-") || text.startsWith("/") || text.endsWith("/") || text.endsWith(".") || text.includes("..") || text.includes("@{") || /[\u0000-\u0020~^:?*[\\]/u.test(text)) { + throw new Error(`${path} must be a safe Git ref name`); + } + return text; +} + +function requireHex(value, path) { + const text = requireString(value, path); + if (!/^[0-9a-f]{64}$/u.test(text)) throw new Error(`${path} must be a sha256 fingerprint`); + return text; +} + +function ensureUnique(values, path) { + if (new Set(values).size !== values.length) throw new Error(`${path} must be unique`); +} diff --git a/scripts/src/agentrun-lanes.ts b/scripts/src/agentrun-lanes.ts index 05615f30..fd839deb 100644 --- a/scripts/src/agentrun-lanes.ts +++ b/scripts/src/agentrun-lanes.ts @@ -20,10 +20,44 @@ export const AGENTRUN_CONFIG_PATH = "config/agentrun.yaml"; export interface AgentRunGitMirrorRepositorySpec { readonly key: string; readonly repository: string; + readonly remote: string; readonly sourceBranch: string; readonly gitopsBranch?: string; } +export interface AgentRunManagedRepositoryReconcilerSpec { + readonly enabled: boolean; + readonly deploymentName: string; + readonly configMapName: string; + readonly serviceAccountName: string; + readonly imagePullPolicy: "Always" | "IfNotPresent" | "Never"; + readonly cacheMountPath: string; + readonly stateDirectory: string; + readonly reconcileIntervalMs: number; + readonly fetchTimeoutMs: number; + readonly shutdownGraceMs: number; + readonly maxConcurrentRepositories: number; + readonly retry: { + readonly maxAttempts: number; + readonly initialDelayMs: number; + readonly maxDelayMs: number; + }; + readonly freshness: { + readonly maxAgeMs: number; + }; + readonly readiness: { + readonly initialDelaySeconds: number; + readonly periodSeconds: number; + readonly timeoutSeconds: number; + readonly failureThreshold: number; + }; + readonly lifecycle: { + readonly undeclaredRepositoryPolicy: "retain"; + readonly managedRefs: "source-branch-only"; + }; + readonly resources: AgentRunContainerResources; +} + export interface AgentRunSourceAuthoritySpec { readonly mode: "gitMirrorSnapshot"; readonly resolver: "k8s-git-mirror"; @@ -205,6 +239,7 @@ export interface AgentRunLaneSpec { readonly toolsImage: string; readonly syncJobPrefix: string; readonly flushJobPrefix: string; + readonly repositoryReconciler: AgentRunManagedRepositoryReconcilerSpec; readonly repositories: readonly AgentRunGitMirrorRepositorySpec[]; }; readonly database: { @@ -327,6 +362,16 @@ export function resolveAgentRunLaneTarget(options: { node?: string | null; lane? return { configPath: config.sourcePath, spec }; } +export function resolveAgentRunLaneTargetsForNode(nodeId: string, env: NodeJS.ProcessEnv = process.env): readonly AgentRunLaneTarget[] { + const config = readAgentRunControlPlaneConfig(env); + const targets = Object.values(config.lanes) + .filter((lane) => lane.nodeId === nodeId) + .sort((left, right) => left.lane.localeCompare(right.lane)) + .map((spec) => ({ configPath: config.sourcePath, spec })); + if (targets.length === 0) throw new Error(`${config.sourcePath}: no AgentRun lane is declared for node ${nodeId}`); + return targets; +} + export function agentRunLaneSummary(spec: AgentRunLaneSpec): Record { return { node: { @@ -425,9 +470,11 @@ export function agentRunLaneSummary(spec: AgentRunLaneSpec): Record ({ key: repo.key, repository: repo.repository, + remoteConfigured: repo.remote.length > 0, sourceBranch: repo.sourceBranch, gitopsBranch: repo.gitopsBranch ?? null, })), @@ -646,7 +693,8 @@ function parseLane(lane: string, node: AgentRunNodeSpec, input: Record parseGitMirrorRepository(repo, `${path}.gitMirror.repositories[${index}]`)), + repositoryReconciler: parseManagedRepositoryReconciler(recordField(gitMirror, "repositoryReconciler", `${path}.gitMirror`), `${path}.gitMirror.repositoryReconciler`), + repositories: parseGitMirrorRepositories(arrayField(gitMirror, "repositories", `${path}.gitMirror`), `${path}.gitMirror.repositories`), }, database: parseDatabase(database, `${path}.database`), secrets: parseLaneSecrets(input, path), @@ -1030,16 +1078,118 @@ function parseProviderCredentialBinding(input: Record, path: st return profile; } +function parseGitMirrorRepositories(input: readonly Record[], path: string): readonly AgentRunGitMirrorRepositorySpec[] { + if (input.length === 0) throw new Error(`${path} must declare at least one repository`); + const repositories = input.map((item, index) => parseGitMirrorRepository(item, `${path}[${index}]`)); + const keys = new Set(); + const names = new Set(); + const remotes = new Set(); + for (const item of repositories) { + if (keys.has(item.key)) throw new Error(`${path} contains duplicate key ${item.key}`); + if (names.has(item.repository)) throw new Error(`${path} contains duplicate repository ${item.repository}`); + if (remotes.has(item.remote)) throw new Error(`${path} contains duplicate remote ownership for ${item.remote}`); + keys.add(item.key); + names.add(item.repository); + remotes.add(item.remote); + } + return repositories; +} + function parseGitMirrorRepository(input: Record, path: string): AgentRunGitMirrorRepositorySpec { + const key = stringField(input, "key", path); + const repository = stringField(input, "repository", path); + const remote = stringField(input, "remote", path); + const sourceBranch = stringField(input, "sourceBranch", path); const gitopsBranch = optionalStringField(input, "gitopsBranch", path); + validateSimpleId(key, `${path}.key`); + validateRepositoryPath(repository, `${path}.repository`); + validateRepositoryRemote(remote, repository, `${path}.remote`); + validateGitRefName(sourceBranch, `${path}.sourceBranch`); + if (gitopsBranch !== undefined) validateGitRefName(gitopsBranch, `${path}.gitopsBranch`); return { - key: stringField(input, "key", path), - repository: stringField(input, "repository", path), - sourceBranch: stringField(input, "sourceBranch", path), + key, + repository, + remote, + sourceBranch, ...(gitopsBranch === undefined ? {} : { gitopsBranch }), }; } +function parseManagedRepositoryReconciler(input: Record, path: string): AgentRunManagedRepositoryReconcilerSpec { + const retry = recordField(input, "retry", path); + const freshness = recordField(input, "freshness", path); + const readiness = recordField(input, "readiness", path); + const lifecycle = recordField(input, "lifecycle", path); + const stateDirectory = relativePathField(input, "stateDirectory", path); + if (stateDirectory === "." || stateDirectory.length === 0) throw new Error(`${path}.stateDirectory must name a dedicated relative directory`); + return { + enabled: booleanField(input, "enabled", path), + deploymentName: kubernetesNameField(input, "deploymentName", path), + configMapName: kubernetesNameField(input, "configMapName", path), + serviceAccountName: kubernetesNameField(input, "serviceAccountName", path), + imagePullPolicy: enumField(input, "imagePullPolicy", path, ["Always", "IfNotPresent", "Never"]), + cacheMountPath: absolutePathField(input, "cacheMountPath", path), + stateDirectory, + reconcileIntervalMs: positiveIntegerField(input, "reconcileIntervalMs", path), + fetchTimeoutMs: positiveIntegerField(input, "fetchTimeoutMs", path), + shutdownGraceMs: positiveIntegerField(input, "shutdownGraceMs", path), + maxConcurrentRepositories: positiveIntegerField(input, "maxConcurrentRepositories", path), + retry: { + maxAttempts: positiveIntegerField(retry, "maxAttempts", `${path}.retry`), + initialDelayMs: positiveIntegerField(retry, "initialDelayMs", `${path}.retry`), + maxDelayMs: positiveIntegerField(retry, "maxDelayMs", `${path}.retry`), + }, + freshness: { + maxAgeMs: positiveIntegerField(freshness, "maxAgeMs", `${path}.freshness`), + }, + readiness: { + initialDelaySeconds: positiveIntegerField(readiness, "initialDelaySeconds", `${path}.readiness`), + periodSeconds: positiveIntegerField(readiness, "periodSeconds", `${path}.readiness`), + timeoutSeconds: positiveIntegerField(readiness, "timeoutSeconds", `${path}.readiness`), + failureThreshold: positiveIntegerField(readiness, "failureThreshold", `${path}.readiness`), + }, + lifecycle: { + undeclaredRepositoryPolicy: enumField(lifecycle, "undeclaredRepositoryPolicy", `${path}.lifecycle`, ["retain"]), + managedRefs: enumField(lifecycle, "managedRefs", `${path}.lifecycle`, ["source-branch-only"]), + }, + resources: parseContainerResources(recordField(input, "resources", path), `${path}.resources`), + }; +} + +function validateRepositoryPath(value: string, path: string): void { + if (!/^[A-Za-z0-9._-]+\/[A-Za-z0-9._-]+$/u.test(value) || value.includes("..")) { + throw new Error(`${path} must be an owner/name repository path without ..`); + } +} + +function validateRepositoryRemote(value: string, repository: string, path: string): void { + try { + const remote = new URL(value); + const remoteRepository = decodeURIComponent(remote.pathname).replace(/^\/+|\.git$/gu, ""); + if (remote.protocol !== "ssh:" || remote.username.length === 0 || remote.password.length > 0 || remote.search.length > 0 || remote.hash.length > 0) { + throw new Error("unsupported remote"); + } + if (remoteRepository !== repository) throw new Error("repository mismatch"); + } catch { + throw new Error(`${path} must be an ssh URL owned by the same repository entry (${repository})`); + } +} + +function validateGitRefName(value: string, path: string): void { + if ( + value.length === 0 + || value.startsWith("-") + || value.startsWith("/") + || value.endsWith("/") + || value.endsWith(".") + || value.includes("..") + || value.includes("@{") + || /[\u0000-\u0020~^:?*[\\]/u.test(value) + ) { + throw new Error(`${path} must be a safe Git branch name`); + } +} + function parseDatabase(input: Record, path: string): AgentRunLaneSpec["database"] { const mode = enumField(input, "mode", path, ["local-postgres", "external-postgres"]); const sslmode = optionalStringField(input, "sslmode", path); @@ -1177,6 +1327,13 @@ function validateKubernetesNamePrefix(value: string, path: string): void { if (!/^[a-z0-9]([-a-z0-9]*[a-z0-9])?$/u.test(value)) throw new Error(`${path} must be a Kubernetes name prefix`); } +function kubernetesNameField(obj: Record, key: string, path: string): string { + const value = stringField(obj, key, path); + validateKubernetesNamePrefix(value, `${path}.${key}`); + if (value.length > 63) throw new Error(`${path}.${key} must be at most 63 characters`); + return value; +} + function validateKubernetesLabelKey(value: string, path: string): void { const parts = value.split("/"); const name = parts.length === 2 ? parts[1] : parts[0]; diff --git a/scripts/src/agentrun-managed-repository-reconciler.test.ts b/scripts/src/agentrun-managed-repository-reconciler.test.ts new file mode 100644 index 00000000..b208722f --- /dev/null +++ b/scripts/src/agentrun-managed-repository-reconciler.test.ts @@ -0,0 +1,204 @@ +import { createHash } from "node:crypto"; +import { + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { join } from "node:path"; +import { describe, expect, test } from "bun:test"; +import { + agentRunManagedRepositoryDesiredConfig, + renderAgentRunManagedRepositoryReconcilerFragment, +} from "./agentrun-managed-repository-reconciler"; +import { resolveAgentRunLaneTarget } from "./agentrun-lanes"; +import { yamlLaneGitMirrorStatusScript } from "./agentrun/secrets"; + +const controllerPath = new URL("../native/agentrun/managed-repository-reconciler.mjs", import.meta.url).pathname; + +describe("AgentRun YAML-owned managed repository reconciler", () => { + test("renders one narrow controller from the lane repository list", () => { + const spec = resolveAgentRunLaneTarget({ node: "NC01", lane: "nc01-v02" }).spec; + const desired = agentRunManagedRepositoryDesiredConfig(spec); + expect(desired.repositories.map((item) => item.repository)).toEqual([ + "pikasTech/agentrun", + "pikasTech/unidesk", + "pikasTech/agent_skills", + ]); + expect(desired.repositories.every((item) => item.remote.startsWith("ssh://"))).toBe(true); + expect(desired.lifecycle).toEqual({ + undeclaredRepositoryPolicy: "retain", + managedRefs: "source-branch-only", + }); + + const documents = renderAgentRunManagedRepositoryReconcilerFragment("NC01") + .split(/^---\s*$/mu) + .map((item) => Bun.YAML.parse(item) as any); + expect(documents.map((item) => item.kind)).toEqual(["ServiceAccount", "ConfigMap", "Deployment"]); + expect(documents[0].automountServiceAccountToken).toBe(false); + expect(documents[1].metadata.annotations["unidesk.ai/undeclared-repository-policy"]).toBe("retain"); + expect(JSON.parse(documents[1].data["config.json"]).repositories).toEqual(desired.repositories); + expect(documents[2].spec.strategy.type).toBe("Recreate"); + expect(documents[2].spec.template.spec.automountServiceAccountToken).toBe(false); + expect(documents[2].spec.template.spec.containers[0].readinessProbe.exec.command).toContain("--health"); + + const controllerSource = readFileSync(controllerPath, "utf8"); + expect(controllerSource).not.toContain("agent_skills"); + expect(controllerSource).not.toContain("pikasTech/"); + expect(controllerSource).not.toContain("github.com"); + }); + + test("requires every repository entry to own a matching remote", () => { + const temporary = mkdtempSync("/tmp/unidesk-agentrun-repository-config-"); + try { + const source = readFileSync(new URL("../../config/agentrun.yaml", import.meta.url), "utf8"); + const invalid = source.replace( + "remote: ssh://git@ssh.github.com:443/pikasTech/agent_skills.git", + "remote: ssh://git@ssh.github.com:443/pikasTech/unidesk.git", + ); + const configPath = join(temporary, "agentrun.yaml"); + writeFileSync(configPath, invalid); + expect(() => resolveAgentRunLaneTarget( + { node: "NC01", lane: "nc01-v02" }, + { ...process.env, AGENTRUN_CONTROL_PLANE_CONFIG: configPath }, + )).toThrow("same repository entry"); + } finally { + rmSync(temporary, { recursive: true, force: true }); + } + }); + + test("materializes source refs, exposes health, and retains undeclared cache content", () => { + const temporary = mkdtempSync("/tmp/unidesk-agentrun-managed-repository-"); + try { + const work = join(temporary, "work"); + const remote = join(temporary, "remote.git"); + const cache = join(temporary, "cache"); + mkdirSync(work, { recursive: true }); + mkdirSync(cache, { recursive: true }); + git(["init", work]); + git(["-C", work, "config", "user.name", "UniDesk fixture"]); + git(["-C", work, "config", "user.email", "fixture@unidesk.invalid"]); + writeFileSync(join(work, "README.md"), "managed repository fixture\n"); + git(["-C", work, "add", "README.md"]); + git(["-C", work, "commit", "-m", "fixture"]); + git(["-C", work, "branch", "-M", "main"]); + git(["clone", "--bare", work, remote]); + + const retainedPath = join(cache, "undeclared", "retained.git"); + mkdirSync(retainedPath, { recursive: true }); + writeFileSync(join(retainedPath, "marker"), "retain\n"); + + const remoteUrl = `file://${remote}`; + const repository = { + key: "fixture", + repository: "acme/example", + remote: remoteUrl, + sourceBranch: "main", + desiredFingerprint: sha256(["fixture", "acme/example", remoteUrl, "main"].join("\0")), + }; + const desired = { + schemaVersion: 1, + kind: "AgentRunManagedRepositoryReconciler", + nodeId: "fixture-node", + lane: "fixture-lane", + cacheMountPath: cache, + stateDirectory: ".managed-state", + reconcileIntervalMs: 1000, + fetchTimeoutMs: 10000, + shutdownGraceMs: 1000, + maxConcurrentRepositories: 1, + retry: { maxAttempts: 1, initialDelayMs: 10, maxDelayMs: 10 }, + freshness: { maxAgeMs: 60000 }, + lifecycle: { undeclaredRepositoryPolicy: "retain", managedRefs: "source-branch-only" }, + repositories: [repository], + }; + const config = { ...desired, desiredHash: sha256(JSON.stringify(desired)) }; + const configPath = join(temporary, "config.json"); + writeFileSync(configPath, `${JSON.stringify(config, null, 2)}\n`); + + const once = Bun.spawnSync(["node", controllerPath, "--once", "--config", configPath], { + stdout: "pipe", + stderr: "pipe", + }); + expect(once.exitCode, once.stderr.toString()).toBe(0); + const onceStatus = JSON.parse(once.stdout.toString()); + expect(onceStatus.ready).toBe(true); + expect(onceStatus.repositories[0].remoteFingerprint).toBe(`sha256:${sha256(remoteUrl)}`); + expect(existsSync(join(retainedPath, "marker"))).toBe(true); + + const mirroredCommit = git([ + "--git-dir", + join(cache, "acme", "example.git"), + "rev-parse", + "--verify", + "refs/heads/main^{commit}", + ]).trim(); + expect(mirroredCommit).toMatch(/^[0-9a-f]{40}$/u); + git([ + "--git-dir", + join(cache, "acme", "example.git"), + "update-ref", + "refs/heads/retained-gitops", + mirroredCommit, + ]); + const secondOnce = Bun.spawnSync(["node", controllerPath, "--once", "--config", configPath], { + stdout: "pipe", + stderr: "pipe", + }); + expect(secondOnce.exitCode, secondOnce.stderr.toString()).toBe(0); + expect(git([ + "--git-dir", + join(cache, "acme", "example.git"), + "rev-parse", + "--verify", + "refs/heads/retained-gitops^{commit}", + ])).toBe(mirroredCommit); + expect(git([ + "--git-dir", + join(cache, "acme", "example.git"), + "for-each-ref", + "--format=%(refname)", + "refs/agentrun-managed", + ])).toBe(""); + + const health = Bun.spawnSync(["node", controllerPath, "--health", "--config", configPath], { + stdout: "pipe", + stderr: "pipe", + }); + expect(health.exitCode, health.stderr.toString()).toBe(0); + const healthStatus = JSON.parse(health.stdout.toString()); + expect(healthStatus.ready).toBe(true); + expect(healthStatus.repositories[0]).toMatchObject({ + refPresent: true, + fresh: true, + commit: mirroredCommit, + remoteFingerprintAligned: true, + }); + } finally { + rmSync(temporary, { recursive: true, force: true }); + } + }); + + test("keeps legacy main-repository fields and bounds per-repository probes", () => { + const spec = resolveAgentRunLaneTarget({ node: "NC01", lane: "nc01-v02" }).spec; + const script = yamlLaneGitMirrorStatusScript(spec); + expect(script).toContain("'repository': os.environ.get('REPOSITORY')"); + expect(script).toContain("'sourceCommit': legacy_ref("); + expect(script).toContain("'gitopsCommit': legacy_ref("); + expect(script.match(/timeout 20 kubectl/gu)?.length).toBe(spec.gitMirror.repositories.length); + expect(script.match(/timeout 20 kubectl[^\n]+ &/gu)?.length).toBe(spec.gitMirror.repositories.length); + expect(script).toContain("remoteFingerprintAligned"); + }); +}); + +function git(args: string[]): string { + const result = Bun.spawnSync(["git", ...args], { stdout: "pipe", stderr: "pipe" }); + if (result.exitCode !== 0) throw new Error(`git ${args[0]} failed: ${result.stderr.toString()}`); + return result.stdout.toString().trim(); +} + +function sha256(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} diff --git a/scripts/src/agentrun-managed-repository-reconciler.ts b/scripts/src/agentrun-managed-repository-reconciler.ts new file mode 100644 index 00000000..92e73980 --- /dev/null +++ b/scripts/src/agentrun-managed-repository-reconciler.ts @@ -0,0 +1,258 @@ +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { rootPath } from "./config"; +import { + resolveAgentRunLaneTargetsForNode, + type AgentRunLaneSpec, +} from "./agentrun-lanes"; + +const controllerSourcePath = rootPath("scripts", "native", "agentrun", "managed-repository-reconciler.mjs"); +const proxySourcePath = rootPath("scripts", "native", "agentrun", "git-mirror-proxy-connect.cjs"); +const controllerMountPath = "/etc/agentrun-managed-repository"; +const sshMountPath = "/var/run/secrets/agentrun-managed-repository"; +const knownHostsPath = "/tmp/agentrun-managed-repository/.ssh/known_hosts"; + +export interface AgentRunManagedRepositoryDesiredConfig { + readonly schemaVersion: 1; + readonly kind: "AgentRunManagedRepositoryReconciler"; + readonly nodeId: string; + readonly lane: string; + readonly cacheMountPath: string; + readonly stateDirectory: string; + readonly reconcileIntervalMs: number; + readonly fetchTimeoutMs: number; + readonly shutdownGraceMs: number; + readonly maxConcurrentRepositories: number; + readonly retry: { + readonly maxAttempts: number; + readonly initialDelayMs: number; + readonly maxDelayMs: number; + }; + readonly freshness: { + readonly maxAgeMs: number; + }; + readonly lifecycle: { + readonly undeclaredRepositoryPolicy: "retain"; + readonly managedRefs: "source-branch-only"; + }; + readonly repositories: readonly { + readonly key: string; + readonly repository: string; + readonly remote: string; + readonly sourceBranch: string; + readonly desiredFingerprint: string; + }[]; + readonly ssh: { + readonly identityFile: string; + readonly knownHostsFile: string; + readonly proxyConnectScript: string; + readonly proxyHost: string; + readonly proxyPort: number; + }; + readonly desiredHash: string; +} + +export function agentRunManagedRepositoryDesiredConfig(spec: AgentRunLaneSpec): AgentRunManagedRepositoryDesiredConfig { + const reconciler = spec.gitMirror.repositoryReconciler; + const desired = { + schemaVersion: 1 as const, + kind: "AgentRunManagedRepositoryReconciler" as const, + nodeId: spec.nodeId, + lane: spec.lane, + cacheMountPath: reconciler.cacheMountPath, + stateDirectory: reconciler.stateDirectory, + reconcileIntervalMs: reconciler.reconcileIntervalMs, + fetchTimeoutMs: reconciler.fetchTimeoutMs, + shutdownGraceMs: reconciler.shutdownGraceMs, + maxConcurrentRepositories: reconciler.maxConcurrentRepositories, + retry: reconciler.retry, + freshness: reconciler.freshness, + lifecycle: reconciler.lifecycle, + repositories: spec.gitMirror.repositories.map((repository) => ({ + key: repository.key, + repository: repository.repository, + remote: repository.remote, + sourceBranch: repository.sourceBranch, + desiredFingerprint: sha256([repository.key, repository.repository, repository.remote, repository.sourceBranch].join("\0")), + })), + ssh: { + identityFile: `${sshMountPath}/ssh-privatekey`, + knownHostsFile: knownHostsPath, + proxyConnectScript: `${controllerMountPath}/proxy-connect.cjs`, + proxyHost: spec.gitMirror.githubProxy.host, + proxyPort: spec.gitMirror.githubProxy.port, + }, + }; + return { ...desired, desiredHash: sha256(JSON.stringify(desired)) }; +} + +export function agentRunManagedRepositoryRenderHash(spec: AgentRunLaneSpec): string { + const config = agentRunManagedRepositoryDesiredConfig(spec); + return sha256([ + JSON.stringify(config), + readFileSync(controllerSourcePath, "utf8"), + readFileSync(proxySourcePath, "utf8"), + ].join("\0")); +} + +export function renderAgentRunManagedRepositoryReconcilerFragment(nodeId: string, env: NodeJS.ProcessEnv = process.env): string { + const targets = resolveAgentRunLaneTargetsForNode(nodeId, env) + .filter(({ spec }) => spec.gitMirror.repositoryReconciler.enabled); + assertUniqueManagedRepositoryOwnership(targets.map(({ spec }) => spec)); + const objects = targets + .flatMap(({ spec }) => managedRepositoryObjects(spec)); + if (objects.length === 0) return ""; + return `${objects.map((object) => Bun.YAML.stringify(object).trim()).join("\n---\n")}\n`; +} + +function assertUniqueManagedRepositoryOwnership(specs: readonly AgentRunLaneSpec[]): void { + const objectNames = new Set(); + const cacheRepositories = new Map(); + for (const spec of specs) { + const reconciler = spec.gitMirror.repositoryReconciler; + for (const [kind, name] of [["Deployment", reconciler.deploymentName], ["ConfigMap", reconciler.configMapName], ["ServiceAccount", reconciler.serviceAccountName]]) { + const identity = `${kind}:${spec.gitMirror.namespace}/${name}`; + if (objectNames.has(identity)) throw new Error(`AgentRun managed repository renderer has duplicate object ownership: ${identity}`); + objectNames.add(identity); + } + const cacheIdentity = spec.gitMirror.cacheHostPath === null + ? `pvc:${spec.gitMirror.namespace}/${spec.gitMirror.cachePvc}` + : `hostPath:${spec.nodeId}/${spec.gitMirror.cacheHostPath}`; + for (const repository of spec.gitMirror.repositories) { + const identity = `${cacheIdentity}/${repository.repository}`; + const owner = cacheRepositories.get(identity); + if (owner !== undefined) throw new Error(`AgentRun managed repository ${identity} is owned by both ${owner} and ${spec.lane}`); + cacheRepositories.set(identity, spec.lane); + } + } +} + +function managedRepositoryObjects(spec: AgentRunLaneSpec): Record[] { + const reconciler = spec.gitMirror.repositoryReconciler; + const desired = agentRunManagedRepositoryDesiredConfig(spec); + const desiredJson = JSON.stringify(desired, null, 2); + const controllerSource = readFileSync(controllerSourcePath, "utf8"); + const proxySource = readFileSync(proxySourcePath, "utf8"); + const renderHash = agentRunManagedRepositoryRenderHash(spec); + const labels = { + "app.kubernetes.io/name": reconciler.deploymentName, + "app.kubernetes.io/component": "managed-repository-reconciler", + "app.kubernetes.io/part-of": "agentrun", + "app.kubernetes.io/managed-by": "unidesk", + "agentrun.pikastech.local/lane": spec.lane, + "agentrun.pikastech.local/node": spec.nodeId, + }; + const annotations = { + "unidesk.ai/managed-repository-desired-sha": desired.desiredHash, + "unidesk.ai/managed-repository-render-sha": renderHash, + "unidesk.ai/undeclared-repository-policy": reconciler.lifecycle.undeclaredRepositoryPolicy, + "unidesk.ai/managed-refs": reconciler.lifecycle.managedRefs, + }; + const cacheVolume = spec.gitMirror.cacheHostPath === null + ? { name: "cache", persistentVolumeClaim: { claimName: spec.gitMirror.cachePvc } } + : { name: "cache", hostPath: { path: spec.gitMirror.cacheHostPath, type: "DirectoryOrCreate" } }; + const loopbackProxy = spec.gitMirror.githubProxy.host === "127.0.0.1" || spec.gitMirror.githubProxy.host === "localhost"; + return [ + { + apiVersion: "v1", + kind: "ServiceAccount", + metadata: { + name: reconciler.serviceAccountName, + namespace: spec.gitMirror.namespace, + labels, + }, + automountServiceAccountToken: false, + }, + { + apiVersion: "v1", + kind: "ConfigMap", + metadata: { + name: reconciler.configMapName, + namespace: spec.gitMirror.namespace, + labels, + annotations, + }, + data: { + "config.json": `${desiredJson}\n`, + "controller.mjs": controllerSource, + "proxy-connect.cjs": proxySource, + }, + }, + { + apiVersion: "apps/v1", + kind: "Deployment", + metadata: { + name: reconciler.deploymentName, + namespace: spec.gitMirror.namespace, + labels, + annotations, + }, + spec: { + replicas: 1, + strategy: { type: "Recreate" }, + selector: { + matchLabels: { + "app.kubernetes.io/name": reconciler.deploymentName, + "app.kubernetes.io/component": "managed-repository-reconciler", + }, + }, + template: { + metadata: { + labels, + annotations: { + "unidesk.ai/managed-repository-desired-sha": desired.desiredHash, + "unidesk.ai/managed-repository-render-sha": renderHash, + }, + }, + spec: { + serviceAccountName: reconciler.serviceAccountName, + automountServiceAccountToken: false, + terminationGracePeriodSeconds: Math.ceil(reconciler.shutdownGraceMs / 1000), + ...(loopbackProxy ? { hostNetwork: true, dnsPolicy: "ClusterFirstWithHostNet" } : {}), + containers: [{ + name: "managed-repository-reconciler", + image: spec.gitMirror.toolsImage, + imagePullPolicy: reconciler.imagePullPolicy, + command: [ + "node", + `${controllerMountPath}/controller.mjs`, + "--config", + `${controllerMountPath}/config.json`, + ], + readinessProbe: { + exec: { + command: [ + "node", + `${controllerMountPath}/controller.mjs`, + "--health", + "--config", + `${controllerMountPath}/config.json`, + ], + }, + initialDelaySeconds: reconciler.readiness.initialDelaySeconds, + periodSeconds: reconciler.readiness.periodSeconds, + timeoutSeconds: reconciler.readiness.timeoutSeconds, + failureThreshold: reconciler.readiness.failureThreshold, + }, + resources: reconciler.resources, + volumeMounts: [ + { name: "config", mountPath: controllerMountPath, readOnly: true }, + { name: "cache", mountPath: reconciler.cacheMountPath }, + { name: "git-ssh", mountPath: sshMountPath, readOnly: true }, + ], + }], + volumes: [ + { name: "config", configMap: { name: reconciler.configMapName, defaultMode: 0o555 } }, + cacheVolume, + { name: "git-ssh", secret: { secretName: spec.gitMirror.sshSecretName, defaultMode: 0o400 } }, + ], + }, + }, + }, + }, + ]; +} + +function sha256(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} diff --git a/scripts/src/agentrun/git-mirror.ts b/scripts/src/agentrun/git-mirror.ts index c371b292..0148cb2d 100644 --- a/scripts/src/agentrun/git-mirror.ts +++ b/scripts/src/agentrun/git-mirror.ts @@ -15,7 +15,6 @@ import { runRemoteSshCommandCapture } from "../remote"; import { startJob } from "../jobs"; import { AGENTRUN_CONFIG_PATH, - agentRunLaneSummary, agentRunPipelineRunName, agentRunProviderCredentialRefs, resolveAgentRunLaneTarget, @@ -32,11 +31,11 @@ import { type AgentRunArtifactService, } from "../agentrun-manifests"; import { sha256Fingerprint } from "../platform-infra-ops-library"; -import { pacAutomaticDeliveryFix, pacReadOnlyNext, resolveCicdDeliveryAuthority } from "../cicd-delivery-authority"; +import { pacAutomaticDeliveryFix, resolveCicdDeliveryAuthority } from "../cicd-delivery-authority"; import type { GitMirrorStatusOptions } from "./options"; import { readGitMirrorStatus } from "./rest-bridge"; -import { compactCapture, shQuote } from "./utils"; +import { compactCapture, record, shQuote } from "./utils"; export function cleanupRunnersFinalizeNodeScript(): string { return String.raw` @@ -484,19 +483,67 @@ export async function gitMirrorStatus(config: UniDeskConfig, options: GitMirrorS }); const observation = await readGitMirrorStatus(config, target); const summary = observation.summary; + const outputSummary = options.full || options.raw ? summary : compactManagedRepositoryStatus(summary); + const expanded = options.full || options.raw; + const compactTarget = { + node: spec.nodeId, + lane: spec.lane, + namespace: spec.gitMirror.namespace, + controller: spec.gitMirror.repositoryReconciler.deploymentName, + repositories: spec.gitMirror.repositories.map((repository) => ({ key: repository.key, sourceBranch: repository.sourceBranch })), + }; + const fullTarget = { + ...compactTarget, + version: spec.version, + sourceAuthority: spec.source.sourceAuthority, + gitMirror: { + readService: spec.gitMirror.readService, + readDeployment: spec.gitMirror.readDeployment, + writeService: spec.gitMirror.writeService, + writeDeployment: spec.gitMirror.writeDeployment, + resourceBundleBaseUrl: spec.gitMirror.resourceBundleBaseUrl, + cachePvc: spec.gitMirror.cachePvc, + cacheHostPath: spec.gitMirror.cacheHostPath, + sshSecretName: spec.gitMirror.sshSecretName, + githubProxy: spec.gitMirror.githubProxy, + repositoryReconciler: spec.gitMirror.repositoryReconciler, + repositories: spec.gitMirror.repositories.map((repository) => ({ + key: repository.key, + repository: repository.repository, + sourceBranch: repository.sourceBranch, + gitopsBranch: repository.gitopsBranch ?? null, + remoteConfigured: true, + })), + }, + }; + const outputNext = deliveryAuthority.kind === "pac-pr-merge" + ? { + status: `bun scripts/cli.ts platform-infra pipelines-as-code status --target ${deliveryAuthority.consumer.node} --consumer ${deliveryAuthority.consumer.consumerId}`, + history: `bun scripts/cli.ts platform-infra pipelines-as-code history --target ${deliveryAuthority.consumer.node} --consumer ${deliveryAuthority.consumer.consumerId}`, + valuesPrinted: false, + } + : deliveryAuthority.kind === "unknown" + ? { + status: `bun scripts/cli.ts agentrun control-plane status --node ${spec.nodeId} --lane ${spec.lane}`, + ...(expanded ? { fixAutomaticDelivery: pacAutomaticDeliveryFix(deliveryAuthority) } : {}), + valuesPrinted: false, + } + : { + sync: `bun scripts/cli.ts agentrun git-mirror sync --node ${spec.nodeId} --lane ${spec.lane} --confirm`, + flush: summary.pendingFlush === true ? `bun scripts/cli.ts agentrun git-mirror flush --node ${spec.nodeId} --lane ${spec.lane} --confirm` : null, + }; return { ok: observation.ok, command: "agentrun git-mirror status", mode: "yaml-declared-node-lane", configPath: target.configPath, - deliveryAuthority, - target: agentRunLaneSummary(spec), + deliveryAuthority: compactDeliveryAuthority(deliveryAuthority), + target: expanded ? fullTarget : compactTarget, namespace: spec.gitMirror.namespace, - readUrl: spec.gitMirror.readUrl, - writeUrl: spec.gitMirror.writeUrl, - summary, + ...(expanded ? { readUrl: spec.gitMirror.readUrl, writeUrl: spec.gitMirror.writeUrl } : {}), + summary: outputSummary, ...(options.raw ? { raw: observation.raw } : {}), - probe: compactCapture(observation.result, { full: options.full || options.raw, stdoutTailChars: 6000, stderrTailChars: 3000 }), + probe: compactCapture(observation.result, { full: options.raw, stdoutTailChars: 6000, stderrTailChars: 3000 }), disclosure: { defaultView: "compact-low-noise", full: options.full, @@ -506,17 +553,59 @@ export async function gitMirrorStatus(config: UniDeskConfig, options: GitMirrorS expandWith: `bun scripts/cli.ts agentrun git-mirror status --node ${spec.nodeId} --lane ${spec.lane} --full`, rawWith: `bun scripts/cli.ts agentrun git-mirror status --node ${spec.nodeId} --lane ${spec.lane} --raw`, }, - next: deliveryAuthority.kind === "pac-pr-merge" - ? pacReadOnlyNext(deliveryAuthority.consumer.node, deliveryAuthority.consumer.consumerId) - : deliveryAuthority.kind === "unknown" - ? { - status: `bun scripts/cli.ts agentrun control-plane status --node ${spec.nodeId} --lane ${spec.lane}`, - fixAutomaticDelivery: pacAutomaticDeliveryFix(deliveryAuthority), - valuesPrinted: false, - } - : { - sync: `bun scripts/cli.ts agentrun git-mirror sync --node ${spec.nodeId} --lane ${spec.lane} --confirm`, - flush: summary.pendingFlush === true ? `bun scripts/cli.ts agentrun git-mirror flush --node ${spec.nodeId} --lane ${spec.lane} --confirm` : null, - }, + next: outputNext, }; } + +function compactDeliveryAuthority(authority: ReturnType): Record { + return authority.kind === "pac-pr-merge" + ? { + kind: authority.kind, + migrated: authority.migrated, + mutationHintsAllowed: authority.mutationHintsAllowed, + consumer: { + consumerId: authority.consumer.consumerId, + node: authority.consumer.node, + lane: authority.consumer.lane, + }, + valuesPrinted: false, + } + : { + kind: authority.kind, + migrated: authority.migrated, + mutationHintsAllowed: authority.mutationHintsAllowed, + valuesPrinted: false, + }; +} + +function compactManagedRepositoryStatus(summary: Record): Record { + const controller = record(summary.controller); + const repositories = Array.isArray(summary.repositories) + ? summary.repositories.map((value) => record(value)).map((repository) => ({ + key: repository.key ?? null, + sourceBranch: repository.sourceBranch ?? null, + ref: shortCommit(repository.cacheCommit), + refPresent: repository.cacheRefPresent === true, + fresh: repository.fresh === true, + served: shortCommit(repository.servedCommit), + servedRefPresent: repository.servedRefPresent === true, + aligned: repository.refAligned === true && repository.remoteFingerprintAligned === true, + })) + : []; + return { + ok: summary.ok === true, + controller: { + deploymentReady: controller.deploymentReady === true, + desiredRendered: controller.configRendered === true && controller.workloadRendered === true, + statusAligned: controller.statusAligned === true, + heartbeatFresh: controller.heartbeatFresh === true, + }, + repositories, + firstDrift: summary.firstDrift ?? null, + valuesPrinted: false, + }; +} + +function shortCommit(value: unknown): string | null { + return typeof value === "string" && /^[0-9a-f]{40}$/u.test(value) ? value.slice(0, 12) : null; +} diff --git a/scripts/src/agentrun/secrets.ts b/scripts/src/agentrun/secrets.ts index d8de8451..b548dc17 100644 --- a/scripts/src/agentrun/secrets.ts +++ b/scripts/src/agentrun/secrets.ts @@ -34,6 +34,10 @@ import { renderAgentRunGitopsFiles, type AgentRunArtifactService, } from "../agentrun-manifests"; +import { + agentRunManagedRepositoryDesiredConfig, + agentRunManagedRepositoryRenderHash, +} from "../agentrun-managed-repository-reconciler"; import { sha256Fingerprint } from "../platform-infra-ops-library"; import { capture, captureJsonPayload, compactCapture, progressEvent, record, shQuote, sleep, stringOrNull } from "./utils"; @@ -629,6 +633,28 @@ export function yamlLaneGitMirrorFlushShell(spec: AgentRunLaneSpec): string { } export function yamlLaneGitMirrorStatusScript(spec: AgentRunLaneSpec): string { + const reconciler = spec.gitMirror.repositoryReconciler; + const desired = agentRunManagedRepositoryDesiredConfig(spec); + const expectedRepositories = desired.repositories.map((repository) => ({ + key: repository.key, + repository: repository.repository, + sourceBranch: repository.sourceBranch, + desiredFingerprint: repository.desiredFingerprint, + remoteFingerprint: sha256Fingerprint(repository.remote), + })); + const repositoryReadProbes = expectedRepositories.flatMap((repository) => { + const url = `${spec.gitMirror.resourceBundleBaseUrl.replace(/\/+$/u, "")}/${repository.repository}.git`; + const ref = `refs/heads/${repository.sourceBranch}`; + const inner = [ + "url=\"$1\"", + "ref=\"$2\"", + "commit=$(timeout 15 git ls-remote \"$url\" \"$ref\" 2>/dev/null | awk 'NR == 1 { print $1 }')", + "printf '%s\\t%s\\n' \"$3\" \"$commit\"", + ].join("; "); + return [ + `( timeout 20 kubectl -n "$namespace" exec deploy/"$controller_deployment" -- sh -lc ${shQuote(inner)} sh ${shQuote(url)} ${shQuote(ref)} ${shQuote(repository.key)} 2>/dev/null || printf '%s\\t\\n' ${shQuote(repository.key)} ) >${shQuote(`/tmp/agentrun-managed-repository-read-refs/${repository.key}.tsv`)} &`, + ]; + }); return [ "set +e", `namespace=${shQuote(spec.gitMirror.namespace)}`, @@ -637,52 +663,176 @@ export function yamlLaneGitMirrorStatusScript(spec: AgentRunLaneSpec): string { `write_service=${shQuote(spec.gitMirror.writeService)}`, `cache_pvc=${shQuote(spec.gitMirror.cachePvc)}`, `cache_host_path=${spec.gitMirror.cacheHostPath === null ? "''" : shQuote(spec.gitMirror.cacheHostPath)}`, + `controller_deployment=${shQuote(reconciler.deploymentName)}`, + `controller_configmap=${shQuote(reconciler.configMapName)}`, + `expected_desired_hash=${shQuote(desired.desiredHash)}`, + `expected_render_hash=${shQuote(agentRunManagedRepositoryRenderHash(spec))}`, + `repositories_json=${shQuote(JSON.stringify(expectedRepositories))}`, `repository=${shQuote(spec.source.repository)}`, `source_branch=${shQuote(spec.source.branch)}`, `gitops_branch=${shQuote(spec.gitops.branch)}`, - `repositories_json=${shQuote(JSON.stringify(spec.gitMirror.repositories))}`, - "kubectl -n \"$namespace\" get svc \"$read_service\" -o json >/tmp/agentrun-gitmirror-read.json 2>/dev/null", - "read_exit=$?", - "kubectl -n \"$namespace\" get svc \"$write_service\" -o json >/tmp/agentrun-gitmirror-write.json 2>/dev/null", - "write_exit=$?", + "kubectl -n \"$namespace\" get deploy \"$read_deployment\" -o json >/tmp/agentrun-gitmirror-read-deploy.json 2>/dev/null", + "read_deploy_exit=$?", + "kubectl -n \"$namespace\" get svc \"$read_service\" -o json >/tmp/agentrun-gitmirror-read-service.json 2>/dev/null", + "read_service_exit=$?", + "kubectl -n \"$namespace\" get endpoints \"$read_service\" -o json >/tmp/agentrun-gitmirror-read-endpoints.json 2>/dev/null", + "read_endpoints_exit=$?", + "kubectl -n \"$namespace\" get svc \"$write_service\" -o json >/tmp/agentrun-gitmirror-write-service.json 2>/dev/null", + "write_service_exit=$?", + "kubectl -n \"$namespace\" get endpoints \"$write_service\" -o json >/tmp/agentrun-gitmirror-write-endpoints.json 2>/dev/null", + "write_endpoints_exit=$?", "kubectl -n \"$namespace\" get pvc \"$cache_pvc\" -o json >/tmp/agentrun-gitmirror-cache.json 2>/dev/null", "cache_exit=$?", + "kubectl -n \"$namespace\" get deploy \"$controller_deployment\" -o json >/tmp/agentrun-managed-repository-controller.json 2>/dev/null", + "controller_deploy_exit=$?", + "kubectl -n \"$namespace\" get configmap \"$controller_configmap\" -o json >/tmp/agentrun-managed-repository-configmap.json 2>/dev/null", + "controller_configmap_exit=$?", "cache_mode=pvc", "if [ -n \"$cache_host_path\" ]; then cache_mode=hostPath; fi", - "kubectl -n \"$namespace\" get deploy,svc,pvc -o name > /tmp/agentrun-gitmirror-names.txt 2>/dev/null", - "repo_path=\"/cache/${repository}.git\"", - "rm -f /tmp/agentrun-gitmirror-refs.txt", - "if [ \"$read_exit\" -eq 0 ]; then", - " kubectl -n \"$namespace\" exec deploy/\"$read_deployment\" -- sh -lc 'repo_path=\"$1\"; source_branch=\"$2\"; gitops_branch=\"$3\"; source_commit=$(git --git-dir=\"$repo_path\" rev-parse --verify \"refs/heads/$source_branch^{commit}\" 2>/dev/null || true); gitops_commit=$(git --git-dir=\"$repo_path\" rev-parse --verify \"refs/heads/$gitops_branch^{commit}\" 2>/dev/null || true); printf \"sourceCommit=%s\\n\" \"$source_commit\"; printf \"gitopsCommit=%s\\n\" \"$gitops_commit\"' sh \"$repo_path\" \"$source_branch\" \"$gitops_branch\" > /tmp/agentrun-gitmirror-refs.txt 2>/dev/null", - "fi", - "NAMESPACE=\"$namespace\" READ_EXIT=\"$read_exit\" WRITE_EXIT=\"$write_exit\" CACHE_EXIT=\"$cache_exit\" CACHE_MODE=\"$cache_mode\" CACHE_HOST_PATH=\"$cache_host_path\" REPOSITORY=\"$repository\" SOURCE_BRANCH=\"$source_branch\" GITOPS_BRANCH=\"$gitops_branch\" REPOSITORIES_JSON=\"$repositories_json\" python3 - <<'PY'", - "import json, os, re", - "def read_text(path):", + "timeout 10 kubectl -n \"$namespace\" exec deploy/\"$controller_deployment\" -- node /etc/agentrun-managed-repository/controller.mjs --status --config /etc/agentrun-managed-repository/config.json >/tmp/agentrun-managed-repository-status.json 2>/dev/null", + "controller_status_exit=$?", + "rm -f /tmp/agentrun-gitmirror-legacy-refs.txt", + "timeout 10 kubectl -n \"$namespace\" exec deploy/\"$read_deployment\" -- sh -lc 'repo_path=\"$1\"; source_branch=\"$2\"; gitops_branch=\"$3\"; source_commit=$(git --git-dir=\"$repo_path\" rev-parse --verify \"refs/heads/$source_branch^{commit}\" 2>/dev/null || true); gitops_commit=$(git --git-dir=\"$repo_path\" rev-parse --verify \"refs/heads/$gitops_branch^{commit}\" 2>/dev/null || true); printf \"sourceCommit=%s\\n\" \"$source_commit\"; printf \"gitopsCommit=%s\\n\" \"$gitops_commit\"' sh \"/cache/${repository}.git\" \"$source_branch\" \"$gitops_branch\" >/tmp/agentrun-gitmirror-legacy-refs.txt 2>/dev/null", + "rm -rf /tmp/agentrun-managed-repository-read-refs", + "mkdir -p /tmp/agentrun-managed-repository-read-refs", + ...repositoryReadProbes, + "wait", + "cat /tmp/agentrun-managed-repository-read-refs/*.tsv > /tmp/agentrun-managed-repository-read-refs.tsv 2>/dev/null || true", + "NAMESPACE=\"$namespace\" READ_DEPLOY_EXIT=\"$read_deploy_exit\" READ_SERVICE_EXIT=\"$read_service_exit\" READ_ENDPOINTS_EXIT=\"$read_endpoints_exit\" WRITE_SERVICE_EXIT=\"$write_service_exit\" WRITE_ENDPOINTS_EXIT=\"$write_endpoints_exit\" CACHE_EXIT=\"$cache_exit\" CONTROLLER_DEPLOY_EXIT=\"$controller_deploy_exit\" CONTROLLER_CONFIGMAP_EXIT=\"$controller_configmap_exit\" CONTROLLER_STATUS_EXIT=\"$controller_status_exit\" CACHE_MODE=\"$cache_mode\" CACHE_HOST_PATH=\"$cache_host_path\" EXPECTED_DESIRED_HASH=\"$expected_desired_hash\" EXPECTED_RENDER_HASH=\"$expected_render_hash\" REPOSITORIES_JSON=\"$repositories_json\" REPOSITORY=\"$repository\" SOURCE_BRANCH=\"$source_branch\" GITOPS_BRANCH=\"$gitops_branch\" python3 - <<'PY'", + "import json, os", + "def read_json(path):", " try:", " with open(path, 'r', encoding='utf-8') as fh:", - " return fh.read()", + " value = json.load(fh)", + " return value if isinstance(value, dict) else {}", " except Exception:", - " return ''", - "def ref_value(text, key):", + " return {}", + "def exit_ok(name):", + " return os.environ.get(name) == '0'", + "def deployment_ready(value):", + " status = value.get('status') or {}", + " spec = value.get('spec') or {}", + " replicas = int(spec.get('replicas') or 0)", + " available = int(status.get('availableReplicas') or 0)", + " return replicas > 0 and available >= replicas and int(status.get('observedGeneration') or 0) >= int((value.get('metadata') or {}).get('generation') or 0)", + "def endpoint_ready(value):", + " for subset in value.get('subsets') or []:", + " if len(subset.get('addresses') or []) > 0:", + " return True", + " return False", + "def annotation(value, key):", + " return ((value.get('metadata') or {}).get('annotations') or {}).get(key)", + "def pod_annotation(value, key):", + " return (((((value.get('spec') or {}).get('template') or {}).get('metadata') or {}).get('annotations') or {}).get(key))", + "def read_refs(path):", + " result = {}", + " try:", + " with open(path, 'r', encoding='utf-8') as fh:", + " for line in fh:", + " parts = line.rstrip('\\n').split('\\t', 1)", + " if len(parts) == 2:", + " result[parts[0]] = parts[1] or None", + " except Exception:", + " pass", + " return result", + "def legacy_ref(path, key):", " prefix = key + '='", - " for line in re.split(r'\\r?\\n', text):", - " if line.startswith(prefix):", - " value = line[len(prefix):].strip()", - " return value or None", + " try:", + " with open(path, 'r', encoding='utf-8') as fh:", + " for line in fh:", + " if line.startswith(prefix):", + " return line[len(prefix):].strip() or None", + " except Exception:", + " pass", " return None", "try:", " repositories = json.loads(os.environ.get('REPOSITORIES_JSON') or '[]')", "except Exception:", " repositories = []", - "names = read_text('/tmp/agentrun-gitmirror-names.txt')", - "refs = read_text('/tmp/agentrun-gitmirror-refs.txt')", - "read_ready = os.environ.get('READ_EXIT') == '0'", - "write_ready = os.environ.get('WRITE_EXIT') == '0'", - "cache_pvc_exists = os.environ.get('CACHE_EXIT') == '0'", + "read_deploy = read_json('/tmp/agentrun-gitmirror-read-deploy.json')", + "read_endpoints = read_json('/tmp/agentrun-gitmirror-read-endpoints.json')", + "write_endpoints = read_json('/tmp/agentrun-gitmirror-write-endpoints.json')", + "cache = read_json('/tmp/agentrun-gitmirror-cache.json')", + "controller_deploy = read_json('/tmp/agentrun-managed-repository-controller.json')", + "controller_configmap = read_json('/tmp/agentrun-managed-repository-configmap.json')", + "controller_status = read_json('/tmp/agentrun-managed-repository-status.json')", + "served_refs = read_refs('/tmp/agentrun-managed-repository-read-refs.tsv')", + "expected_desired_hash = os.environ.get('EXPECTED_DESIRED_HASH')", + "expected_render_hash = os.environ.get('EXPECTED_RENDER_HASH')", + "read_ready = exit_ok('READ_DEPLOY_EXIT') and deployment_ready(read_deploy) and exit_ok('READ_SERVICE_EXIT') and exit_ok('READ_ENDPOINTS_EXIT') and endpoint_ready(read_endpoints)", + "write_ready = exit_ok('WRITE_SERVICE_EXIT') and exit_ok('WRITE_ENDPOINTS_EXIT') and endpoint_ready(write_endpoints)", + "cache_pvc_exists = exit_ok('CACHE_EXIT')", "cache_mode = os.environ.get('CACHE_MODE') or 'pvc'", - "cache_ready = True if cache_mode == 'hostPath' else cache_pvc_exists", + "cache_ready = True if cache_mode == 'hostPath' else cache_pvc_exists and (cache.get('status') or {}).get('phase') == 'Bound'", + "controller_deployment_ready = exit_ok('CONTROLLER_DEPLOY_EXIT') and deployment_ready(controller_deploy)", + "controller_config_rendered = exit_ok('CONTROLLER_CONFIGMAP_EXIT') and annotation(controller_configmap, 'unidesk.ai/managed-repository-desired-sha') == expected_desired_hash and annotation(controller_configmap, 'unidesk.ai/managed-repository-render-sha') == expected_render_hash", + "controller_workload_rendered = annotation(controller_deploy, 'unidesk.ai/managed-repository-desired-sha') == expected_desired_hash and annotation(controller_deploy, 'unidesk.ai/managed-repository-render-sha') == expected_render_hash and pod_annotation(controller_deploy, 'unidesk.ai/managed-repository-desired-sha') == expected_desired_hash and pod_annotation(controller_deploy, 'unidesk.ai/managed-repository-render-sha') == expected_render_hash", + "controller_status_aligned = exit_ok('CONTROLLER_STATUS_EXIT') and controller_status.get('desiredHash') == expected_desired_hash", + "status_repositories = {item.get('desiredFingerprint'): item for item in controller_status.get('repositories') or [] if isinstance(item, dict)}", + "repository_status = []", + "for desired in repositories:", + " observed = status_repositories.get(desired.get('desiredFingerprint')) or {}", + " served_commit = served_refs.get(desired.get('key'))", + " cache_commit = observed.get('commit')", + " remote_fingerprint_aligned = observed.get('remoteFingerprint') == desired.get('remoteFingerprint')", + " cache_ref_present = observed.get('refPresent') is True and isinstance(cache_commit, str) and len(cache_commit) == 40", + " served_ref_present = isinstance(served_commit, str) and len(served_commit) == 40", + " repository_status.append({", + " 'key': desired.get('key'),", + " 'repository': desired.get('repository'),", + " 'sourceBranch': desired.get('sourceBranch'),", + " 'desiredFingerprint': desired.get('desiredFingerprint'),", + " 'remoteFingerprint': desired.get('remoteFingerprint'),", + " 'observedRemoteFingerprint': observed.get('remoteFingerprint'),", + " 'remoteFingerprintAligned': remote_fingerprint_aligned,", + " 'cacheRefPresent': cache_ref_present,", + " 'cacheCommit': cache_commit if cache_ref_present else None,", + " 'servedRefPresent': served_ref_present,", + " 'servedCommit': served_commit if served_ref_present else None,", + " 'refAligned': cache_ref_present and served_ref_present and cache_commit == served_commit,", + " 'fresh': observed.get('fresh') is True,", + " 'lastSuccessAt': observed.get('lastSuccessAt'),", + " 'lastSuccessAgeMs': observed.get('lastSuccessAgeMs'),", + " 'lastAttemptOk': observed.get('lastAttemptOk') is True,", + " 'errorCode': observed.get('errorCode'),", + " 'errorFingerprint': observed.get('errorFingerprint'),", + " })", + "first_drift = None", + "if not controller_config_rendered:", + " first_drift = 'rendered-desired-not-applied'", + "elif not controller_workload_rendered:", + " first_drift = 'controller-workload-render-mismatch'", + "elif not controller_deployment_ready:", + " first_drift = 'controller-not-ready'", + "elif not cache_ready:", + " first_drift = 'cache-not-ready'", + "elif not controller_status_aligned:", + " first_drift = 'controller-status-not-aligned'", + "elif not read_ready:", + " first_drift = 'read-service-not-ready'", + "elif not write_ready:", + " first_drift = 'write-service-not-ready'", + "else:", + " for item in repository_status:", + " if not item.get('cacheRefPresent'):", + " first_drift = 'cache-ref-missing:' + str(item.get('key'))", + " break", + " if not item.get('remoteFingerprintAligned'):", + " first_drift = 'remote-fingerprint-mismatch:' + str(item.get('key'))", + " break", + " if not item.get('fresh'):", + " first_drift = 'repository-stale:' + str(item.get('key'))", + " break", + " if not item.get('servedRefPresent'):", + " first_drift = 'served-ref-missing:' + str(item.get('key'))", + " break", + " if not item.get('refAligned'):", + " first_drift = 'served-ref-mismatch:' + str(item.get('key'))", + " break", + "repositories_ready = len(repository_status) == len(repositories) and all(item.get('cacheRefPresent') and item.get('fresh') and item.get('refAligned') and item.get('remoteFingerprintAligned') for item in repository_status)", + "ok = controller_config_rendered and controller_workload_rendered and controller_deployment_ready and controller_status_aligned and cache_ready and read_ready and write_ready and repositories_ready", "print(json.dumps({", - " 'ok': read_ready and write_ready and cache_ready,", + " 'ok': ok,", " 'namespace': os.environ.get('NAMESPACE'),", " 'readReady': read_ready,", " 'writeReady': write_ready,", @@ -690,13 +840,27 @@ export function yamlLaneGitMirrorStatusScript(spec: AgentRunLaneSpec): string { " 'cacheReady': cache_ready,", " 'cachePvcExists': cache_pvc_exists,", " 'cacheHostPath': os.environ.get('CACHE_HOST_PATH') or None,", - " 'resources': [line for line in re.split(r'\\r?\\n', names) if line][:40],", " 'repository': os.environ.get('REPOSITORY'),", " 'sourceBranch': os.environ.get('SOURCE_BRANCH'),", " 'gitopsBranch': os.environ.get('GITOPS_BRANCH'),", - " 'sourceCommit': ref_value(refs, 'sourceCommit'),", - " 'gitopsCommit': ref_value(refs, 'gitopsCommit'),", - " 'repositories': repositories,", + " 'sourceCommit': legacy_ref('/tmp/agentrun-gitmirror-legacy-refs.txt', 'sourceCommit'),", + " 'gitopsCommit': legacy_ref('/tmp/agentrun-gitmirror-legacy-refs.txt', 'gitopsCommit'),", + " 'controller': {", + " 'deploymentReady': controller_deployment_ready,", + " 'configRendered': controller_config_rendered,", + " 'workloadRendered': controller_workload_rendered,", + " 'statusAligned': controller_status_aligned,", + " 'desiredHash': expected_desired_hash,", + " 'renderHash': expected_render_hash,", + " 'observedDesiredHash': controller_status.get('desiredHash'),", + " 'heartbeatAt': controller_status.get('heartbeatAt'),", + " 'heartbeatAgeMs': controller_status.get('heartbeatAgeMs'),", + " 'heartbeatFresh': controller_status.get('heartbeatFresh') is True,", + " },", + " 'repositoriesReady': repositories_ready,", + " 'repositories': repository_status,", + " 'firstDrift': first_drift,", + " 'pendingFlush': None,", " 'valuesPrinted': False", "}, ensure_ascii=False))", "PY",