quixos-protocol/src/capability-language/authoring-converge.ts

import { readFile, writeFile, rename, rm, realpath } from "node:fs/promises";
import path from "node:path";
import { execFile as callback } from "node:child_process";
import { promisify } from "node:util";
import { randomUUID } from "node:crypto";
import { authoringContext } from "./authoring-context.js";
import { managedResourcePath } from "./managed-source-path.js";
import { snapshotCommit } from "./checked-build.js";
import { profile } from "./profile.js";
import {
  loadQuixosLock,
  parseQuixosLockDocument,
  formatQuixosLockDocument,
  retentionTagForCommit,
  type GitSource,
} from "../resource-lock/index.js";

const execFile = promisify(callback);
// Bound independent repository operations. One visit owns each repository;
// all visits finish before the final concurrent-edit validation pass.
const concurrency = 4;
async function mapConcurrent<T, R>(values: readonly T[], action: (value: T) => Promise<R>): Promise<R[]> {
  const results = new Array<R>(values.length);
  let next = 0;
  await Promise.all(
    Array.from({ length: Math.min(concurrency, values.length) }, async () => {
      while (next < values.length) {
        const index = next++;
        results[index] = await action(values[index]);
      }
    }),
  );
  return results;
}
const limiter = () => {
  let running = 0;
  const waiting: Array<() => void> = [];
  return async <T>(action: () => Promise<T>): Promise<T> => {
    await new Promise<void>((resolve) => {
      if (running < concurrency) {
        running++;
        resolve();
      } else waiting.push(resolve);
    });
    try {
      return await action();
    } finally {
      const next = waiting.shift();
      if (next) next();
      else running--;
    }
  };
};

const command = async (cwd: string, executable: string, args: string[]) => {
  const start = performance.now();
  // Do not log arguments: transports may contain credentials. Source identities
  // remain in the normal checked result, not in this timing channel.
  const label = `${path.basename(cwd)} ${executable} ${args[0]}`;
  try {
    return (
      await execFile(executable, args, {
        cwd,
        maxBuffer: 4 * 1024 * 1024,
        env: { ...process.env, QUIXOS_JJ_NO_CHECKPOINT: "1", QUIXOS_SUBTREE_PUBLISH: "0", GIT_TERMINAL_PROMPT: "0" },
      })
    ).stdout.trim();
  } finally {
    if (process.env.QUIXOS_TRACE_CAPTURE === "1")
      console.error(`[${new Date().toISOString()}] Capture: ${label}: ${Math.round(performance.now() - start)}ms`);
  }
};
const identity = (kind: string, repository: string) => `${kind}\0${repository}`;

export type AuthoringBlocker = {
  directory: string;
  phase: "resolution" | "dependency" | "source" | "publication" | "concurrent-edit";
  message: string;
};

/** Source retention only. Neither successful convergence nor an empty source
 * worklist grants typechecking, semantic review or activation approval.
 * Caller serializes coordinators; package authors may still be editing. */
export async function convergeAuthoring(start: string, target = "root") {
  const started = performance.now();
  const context = await authoringContext(start);
  const graphFile = path.join(context.workbench, ".quixos/resource-graph.json");
  const graphBefore = await readFile(graphFile, "utf8");
  const graph = JSON.parse(graphBefore);
  // Discovery resolves outer guest aliases. Persist those same physical
  // selections, without accepting symlinks within the managed workbench.
  for (const resource of graph.resources)
    resource.directory = await managedResourcePath(context.workbench, resource.directory);
  const blockers: AuthoringBlocker[] = [];
  const nodes = new Map<string, { directory: string; kind: string; source: GitSource; dependencies: string[] }>();
  const selected = new Map<string, string>();
  const inventory = await mapConcurrent(context.resources, async (entry) => {
    const root = path.join(context.workbench, entry.directory);
    let repository = entry.source?.repository;
    try {
      if ((await realpath(root)) !== root) throw new Error(`Managed checkout crosses a symlink: ${entry.directory}`);
      // Transport rewrites must not become committed source identities.
      const origin = await command(root, "git", ["config", "--get", "remote.origin.url"]);
      if (repository && origin !== repository)
        throw new Error(`Origin differs from registered source for ${entry.directory}`);
      repository ??= origin;
    } catch (error) {
      blockers.push({ directory: entry.directory, phase: "source", message: String(error).slice(0, 2000) });
      if (!repository) throw error; // The root has no separate registered source.
    }
    return { ...entry, repository };
  });
  for (const { repository, ...entry } of inventory) {
    const key = identity(entry.kind, repository);
    if (selected.has(key)) throw new Error(`More than one editable checkout for ${repository}`);
    selected.set(key, entry.directory);
    nodes.set(entry.directory, {
      ...entry,
      source: { resolver: "git", repository, commit: entry.source?.commit ?? "" },
      dependencies: [],
    });
  }
  await mapConcurrent([...nodes.values()], async (node) => {
    try {
      const lock = await loadQuixosLock(path.join(context.workbench, node.directory, "quixos.lock"));
      if (!lock.ok) throw new Error(lock.diagnostics.map((d) => `${d.fileName}: ${d.message}`).join("\n"));
      node.dependencies = [
        ...new Set(
          lock.lock.resources.flatMap((entry) => {
            const directory = selected.get(identity(entry.kind, entry.source.repository));
            return directory ? [directory] : [];
          }),
        ),
      ];
    } catch (error) {
      blockers.push({ directory: node.directory, phase: "resolution", message: String(error) });
    }
  });
  // Detect cycles before starting concurrent futures: two roots awaiting each
  // other must fail clearly rather than deadlocking on their cached promises.
  const scanned = new Set<string>();
  const detectCycles = (directory: string, parents: string[] = []) => {
    if (parents.includes(directory)) {
      blockers.push({
        directory,
        phase: "dependency",
        message: `Source dependency cycle: ${[...parents, directory].join(" -> ")}`,
      });
      return;
    }
    if (scanned.has(directory)) return;
    for (const dependency of nodes.get(directory)!.dependencies) detectCycles(dependency, [...parents, directory]);
    scanned.add(directory);
  };
  for (const directory of nodes.keys()) detectCycles(directory);
  const complete = new Map<string, GitSource>();
  const pending = new Map<string, Promise<boolean>>();
  const capture = limiter();
  const visited = new Set<string>();
  if (!nodes.has(target)) throw new Error(`Not a registered repository: ${target}`);
  const visit = (directory: string): Promise<boolean> => {
    visited.add(directory);
    const existing = pending.get(directory);
    if (existing) return existing;
    const result = (async () => {
      if (blockers.some((entry) => entry.directory === directory)) return false;
      const node = nodes.get(directory)!;
      const ready = await Promise.all(node.dependencies.map(visit));
      const unavailable = node.dependencies.find((_, index) => !ready[index]);
      if (unavailable) {
        blockers.push({ directory, phase: "dependency", message: `Waiting for ${unavailable}` });
        return false;
      }
      return await capture(async () => {
        const root = path.join(context.workbench, directory);
        let phase: AuthoringBlocker["phase"] = "source";
        try {
          const lock = await loadQuixosLock(path.join(root, "quixos.lock"));
          if (!lock.ok) throw new Error("Lock changed during convergence; retry after joining writers");
          for (const file of lock.lock.sourceFiles ?? ["quixos.lock"]) {
            const filename = path.join(root, file),
              before = await readFile(filename, "utf8");
            const parsed = parseQuixosLockDocument(before, file);
            if (!parsed.ok) throw new Error(`Invalid lock ${file}`);
            let changed = false;
            for (const dependency of parsed.document.resources) {
              const target = selected.get(identity(dependency.kind, dependency.source.repository));
              const source = target ? complete.get(target) : undefined;
              if (target && !source)
                throw new Error(
                  `Dependencies changed during convergence (${dependency.binding}); join writers and retry`,
                );
              if (source && source.commit !== dependency.source.commit) {
                dependency.source = source;
                changed = true;
              }
            }
            if (changed) {
              const temporary = `${filename}.${randomUUID()}.tmp`;
              try {
                await writeFile(temporary, formatQuixosLockDocument(parsed.document), { flag: "wx" });
                if ((await readFile(filename, "utf8")) !== before)
                  throw new Error(`Concurrent edit to ${file}; retry after joining writers`);
                await rename(temporary, filename);
              } finally {
                await rm(temporary, { force: true });
              }
            }
          }
          const snapshotStarted = performance.now();
          const commit = await snapshotCommit(root);
          if (process.env.QUIXOS_TRACE_CAPTURE === "1")
            console.error(
              `[${new Date().toISOString()}] Capture: ${directory} snapshot: ${Math.round(performance.now() - snapshotStarted)}ms`,
            );
          phase = "publication";
          const ref = retentionTagForCommit(commit);
          const remote = await command(root, "git", ["ls-remote", "--refs", node.source.repository, ref]);
          if (remote && remote.split(/\s+/)[0] !== commit)
            throw new Error(`Conflicting immutable retention ref ${ref}`);
          if (!remote) {
            await command(root, "git", ["push", node.source.repository, `${commit}:${ref}`]);
            if (
              (await command(root, "git", ["ls-remote", "--refs", node.source.repository, ref])).split(/\s+/)[0] !==
              commit
            )
              throw new Error("Published source retention was not observed");
          }
          complete.set(directory, { ...node.source, commit });
        } catch (error) {
          blockers.push({ directory, phase, message: String(error).slice(0, 4000) });
        }
        return complete.has(directory);
      });
    })();
    pending.set(directory, result);
    return result;
  };
  // Include newly created, not-yet-imported resources, then the root.
  if (target === "root") await Promise.all([...nodes.keys()].filter((d) => d !== "root").map(visit));
  await visit(target);
  for (let index = blockers.length - 1; index >= 0; index--)
    if (!visited.has(blockers[index].directory)) blockers.splice(index, 1);
  const recheckStarted = performance.now();
  await mapConcurrent([...complete], async ([directory, source]) => {
    try {
      if ((await snapshotCommit(path.join(context.workbench, directory))) !== source.commit)
        throw new Error("Source advanced while converging; join writers and retry");
    } catch (error) {
      blockers.push({ directory, phase: "concurrent-edit", message: String(error) });
    }
  });
  profile("capture-recheck", recheckStarted, { repositories: complete.size });
  // Persist successful selections even if another repository is still broken.
  // Recovery must not depend on all parents succeeding in the same invocation.
  const replacements = new Map<string, string>();
  for (const resource of graph.resources) {
    const source = complete.get(resource.directory);
    if (!source) continue;
    const key = `${resource.kind}\0${source.repository}\0${source.commit}`;
    replacements.set(resource.key, key);
    if (resource.source.commit !== source.commit) delete resource.revisionId;
    resource.source = source;
    resource.key = key;
  }
  for (const resource of graph.resources)
    for (const dependency of resource.dependencies ?? []) {
      dependency.resourceKey = replacements.get(dependency.resourceKey) ?? dependency.resourceKey;
    }
  for (const direct of graph.directResources ?? [])
    direct.resourceKey = replacements.get(direct.resourceKey) ?? direct.resourceKey;
  // Inventory is a projection of actual locks, including newly added/removed
  // imports. Never require a successful parent compilation to repair it.
  for (const [directory] of complete) {
    const lock = await loadQuixosLock(path.join(context.workbench, directory, "quixos.lock"));
    if (!lock.ok) continue;
    const dependencies = lock.lock.resources.map((dependency) => ({
      binding: `${dependency.kind}\0${dependency.binding}`,
      resourceKey: `${dependency.kind}\0${dependency.source.repository}\0${dependency.source.commit}`,
    }));
    if (directory === "root") {
      graph.quixos = lock.lock.quixos;
      graph.directResources = lock.lock.resources.map((dependency, index) => ({
        kind: dependency.kind,
        binding: dependency.binding,
        resourceKey: dependencies[index].resourceKey,
        ...(selected.has(identity(dependency.kind, dependency.source.repository))
          ? { directory: selected.get(identity(dependency.kind, dependency.source.repository)) }
          : {}),
      }));
    } else {
      const resource = graph.resources.find((entry: { directory: string }) => entry.directory === directory);
      if (resource) resource.dependencies = dependencies;
    }
  }
  const graphAfter = JSON.stringify(graph, null, 2) + "\n";
  if (graphBefore !== graphAfter) {
    const temporary = `${graphFile}.${randomUUID()}.tmp`;
    try {
      await writeFile(temporary, graphAfter, { flag: "wx", mode: 0o600 });
      if ((await readFile(graphFile, "utf8")) !== graphBefore)
        throw new Error(
          "Managed inventory changed during convergence; source is retained, retry after joining writers",
        );
      await rename(temporary, graphFile);
    } finally {
      await rm(temporary, { force: true });
    }
  }
  profile("capture", started, { repositories: nodes.size, snapshots: complete.size * 2, target });
  return {
    workbench: context.workbench,
    converged: blockers.length === 0,
    candidate: blockers.length ? null : (complete.get(target) ?? null),
    retained: [...complete]
      .sort(([left], [right]) => left.localeCompare(right))
      .map(([directory, source]) => ({ directory, source })),
    worklist: blockers,
    verificationEvidence: false,
    activated: false,
  };
}