#!/usr/bin/env bash
# Build dependency-first Beagle Core modules through the frozen native program.
#
# Usage:
#   beagle build --materializer c17|qbe|wasm [...] --out DIR
#     [--abi lp64|wasm32] [--entry NS/NAME]... [--simd]
#     [--emit-workers N]
#     [--module-root LOGICAL_PREFIX=PHYSICAL_DIRECTORY]... SOURCE.bgl...
#
# A successful run always writes the authoritative frozen native program encoding
# and digest, plus every explicitly selected materializer projection. The
# command accepts canonical `.bgl` / bare `#lang beagle` sources only.
# `BEAGLE_DEV_FACT_REUSE=1` opts iterative development builds into exact
# per-unit Store reuse; the default and every cold-authority caller keep it off.

set -euo pipefail

BEAGLE_DIR="$(cd "$(dirname "$0")/.." && pwd)"
BIN="$BEAGLE_DIR/bin"
PROJECTOR="$BEAGLE_DIR/native-core/bin/source-facts.clj"
SEMANTIC_READ_STORE="$BEAGLE_DIR/native-core/bin/semantic_read_store.clj"
SEMANTIC_READ_BLOB_STORE="$BEAGLE_DIR/native-core/bin/source_fact_store.clj"
SEMANTIC_READ_DATABASE="$BEAGLE_DIR/store/database.clj"
SEMANTIC_READ_WRITER_AUTHORITY="$BEAGLE_DIR/store/writer_authority.clj"
AST_VERIFIER="$BEAGLE_DIR/native-core/bin/verify-checked-ast.rkt"
FINALIZER="$BEAGLE_DIR/native-core/validation/build-finalize.clj"
CORE_COMPILER_PROJECTION="$BIN/beagle-core-compiler-projection"
EMIT_MANAGER="$BEAGLE_DIR/native-core/bin/emit-workers"
MODULE_SOURCE_ROOT_CLI="$BEAGLE_DIR/beagle-lib/private/module-source-root-cli.rkt"
original_args=("$@")
source "$BIN/_beagle-racket"
source "$BIN/_beagle-source-id"
source "$BIN/_beagle-compiler-provenance"

core_kill_grace="${BEAGLE_CORE_KILL_GRACE_SECONDS:-5}"
[[ "$core_kill_grace" =~ ^[1-9][0-9]*$ ]] || {
    echo "beagle build: BEAGLE_CORE_KILL_GRACE_SECONDS must be a positive integer" >&2
    exit 2
}
[[ -x "$EMIT_MANAGER" ]] || {
    echo "beagle build: emission manager is unavailable: $EMIT_MANAGER" >&2
    exit 2
}

# The native supervisor is cached outside launch-critical checkouts.
source "$BIN/_beagle-rust-supervisor"
NATIVE_SUPERVISOR="$(beagle_resolve_rust_supervisor 'beagle build')" || {
    echo "beagle build: native bounded supervisor is unavailable" >&2
    exit 2
}
export BEAGLE_RUST_SUPERVISOR="$NATIVE_SUPERVISOR"
# One bounded supervisor owns the entire build tree. It prefers a private PID
# namespace and falls back to a scoped process group when the host forbids
# unprivileged user namespaces. Individual phases use the same supervisor with
# tighter deadlines.
if [[ "${BEAGLE_CORE_SUPERVISED:-0}" != "1" ]]; then
    overall_timeout="${BEAGLE_CORE_OVERALL_TIMEOUT_SECONDS:-600}"
    [[ "$overall_timeout" =~ ^[1-9][0-9]*$ ]] || {
        echo "beagle build: BEAGLE_CORE_OVERALL_TIMEOUT_SECONDS must be a positive integer" >&2
        exit 2
    }
    exec "$NATIVE_SUPERVISOR" \
        "$overall_timeout" "$core_kill_grace" -- \
        env BEAGLE_CORE_SUPERVISED=1 "$0" "$@"
fi

source "$BEAGLE_DIR/share/targets.sh"

usage() {
    sed -n '2,10p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'
}

die() {
    echo "beagle build: $*" >&2
    exit 2
}

dev_fact_reuse=0
case "${BEAGLE_DEV_FACT_REUSE:-0}" in
    0) ;;
    1)
        if [[ "${BEAGLE_FACT_REUSE_FORBIDDEN:-0}" == 1 ]]; then
            echo "beagle build: dev-facts FORBIDDEN cold-authority" >&2
        else
            dev_fact_reuse=1
        fi
        ;;
    *) die "BEAGLE_DEV_FACT_REUSE must be 0 or 1" ;;
esac
dev_fact_store="${BEAGLE_DEV_FACT_STORE:-$BEAGLE_DIR/.beagle/dev-compile-facts.storelog}"
if [[ "$dev_fact_reuse" == 1 && "$dev_fact_store" != /* ]]; then
    die "BEAGLE_DEV_FACT_STORE must be an absolute path"
fi

run_bounded() {
    local seconds="$1"
    local kill_grace="$2"
    shift 2
    "$NATIVE_SUPERVISOR" "$seconds" "$kill_grace" -- "$@"
}

run_phase_with_grace() {
    local name="$1"
    local seconds="$2"
    local kill_grace="$3"
    shift 3
    echo "beagle build: phase $name START" >&2
    local rc=0
    run_bounded "$seconds" "$kill_grace" "$@" || rc=$?
    if [[ $rc -eq 0 ]]; then
        echo "beagle build: phase $name END" >&2
    else
        echo "beagle build: phase $name ERROR ($rc)" >&2
    fi
    return "$rc"
}

run_phase() {
    local name="$1"
    local seconds="$2"
    shift 2
    run_phase_with_grace "$name" "$seconds" "$core_kill_grace" "$@"
}

# A LEAF phase: a command that is one process and can never acquire a
# descendant. The Racket supervisor exists to contain a process TREE — it
# prefers a private PID namespace and re-execs itself into it, so it costs two
# Racket VMs — and a command with no tree to contain buys nothing for that.
# `timeout` enforces the same contract at the same boundary: SIGTERM at the
# deadline, SIGKILL after the same kill grace, exit 124 when the deadline is
# what ended it. Containment is not weakened, it is trivially satisfied: the
# signal reaches the whole of what the phase ever created.
#
# The leaf claim is the caller's obligation and is NOT a guess about the
# command's name. Use this only where the command is known to be a single
# execve that spawns nothing (flock taking an already-open fd is the case this
# exists for); anything that might fork stays on the supervisor.
#
# The claim is discharged by evidence, not by reading: the build was traced at
# clone/clone3/vfork return values (not just execve, which misses a fork that
# never execs) and the tree reconstructed. module-source-closure, source-facts,
# native-receipts, staged-manifest, and locked-staged-verifier spawned ZERO
# descendants, and module-source-root-cli.rkt, source-facts.clj, and
# build-finalize.clj hold no spawn call site (the two apparent hits in
# build-finalize.clj are string literals naming wasm.materializer.sh, not
# calls). Checked-AST verification is deliberately excluded: its per-source
# cohort needs directly owned supervisors so a failed parallel worker can stop
# and reap its siblings before staging is removed.
#
# ast-bundle, core-compiler, core-lowering, core-checkpoint-wire, and
# wasm-materializer genuinely fork and MUST keep real tree containment.
run_phase_leaf() {
    local name="$1"
    local seconds="$2"
    shift 2
    echo "beagle build: phase $name START (leaf, deadline=${seconds}s kill-grace=${core_kill_grace}s)" >&2
    local rc=0
    timeout -k "$core_kill_grace" "$seconds" "$@" || rc=$?
    if [[ $rc -eq 0 ]]; then
        echo "beagle build: phase $name END" >&2
    elif [[ $rc -eq 124 ]]; then
        echo "beagle build: phase $name TIMEOUT ($rc)" >&2
    else
        echo "beagle build: phase $name ERROR ($rc)" >&2
    fi
    return "$rc"
}

# Direct supervisor PIDs, never shell-function subshells. The EXIT/signal path
# must be able to stop the exact owners that guarantee their verifier subtree is
# gone before cleanup removes the metadata paths those children write.
ast_verify_live_pids=()
ast_verified_logical_paths=()
ast_verified_interface_sha256s=()
stop_ast_verify_workers() {
    local pid
    for pid in "${ast_verify_live_pids[@]:-}"; do
        [[ -n "$pid" ]] || continue
        kill -TERM "$pid" 2>/dev/null || true
    done
    for pid in "${ast_verify_live_pids[@]:-}"; do
        [[ -n "$pid" ]] || continue
        wait "$pid" 2>/dev/null || true
    done
    ast_verify_live_pids=()
}

run_ast_verifier_pool() {
    local source_count="${#normalized_sources[@]}"
    local worker_count="$emit_workers"
    local deadline="${BEAGLE_CORE_VALIDATION_TIMEOUT_SECONDS:-30}"
    local next_index=0 failure_seen=0 failure_cutoff=-1
    local finished_pid="" finished_index="" status=0 expected_receipt=""
    local receipt_value="" source="" ast="" metadata="" log="" receipt=""
    local source_digest_after="" pid="" live_index="" selected_index=-1
    local selected_status=0 index=0 interface_sha256_path=""
    local interface_sha256=""
    local -a live=() next_live=() checked_fields=()
    local -A index_by_pid=() status_by_index=() canceled_by_index=()
    local -A log_by_index=() receipt_by_index=() started_by_index=()

    ((worker_count <= source_count)) || worker_count="$source_count"
    ast_verified_logical_paths=()
    ast_verified_interface_sha256s=()

    while ((next_index < source_count || ${#live[@]} > 0)); do
        while ((failure_seen == 0 && next_index < source_count &&
                ${#live[@]} < worker_count)); do
            source="${normalized_sources[$next_index]}"
            ast="$work/source_${next_index}.ast.json"
            metadata="$work/source_${next_index}.metadata"
            log="$work/source_${next_index}.verify.log"
            receipt="$work/source_${next_index}.verify.receipt"
            printf 'beagle build: phase ast-verify-%s START (workers=%s, deadline=%ss kill-grace=%ss)\n' \
                "$next_index" "$worker_count" "$deadline" "$core_kill_grace" >"$log"
            log_by_index["$next_index"]="$log"
            receipt_by_index["$next_index"]="$receipt"
            started_by_index["$next_index"]=1
            source_digest_after=""
            if ! source_digest_after="$(sha256sum "$source" 2>/dev/null | awk '{print $1}')"; then
                source_digest_after=unavailable
            fi
            if [[ "${source_digest_befores[$next_index]}" != "$source_digest_after" ]]; then
                printf 'beagle build: source changed while its checked projection was captured: %s\n' \
                    "$source" >>"$log"
                printf 'beagle build: phase ast-verify-%s ERROR (2)\n' \
                    "$next_index" >>"$log"
                status_by_index["$next_index"]=2
                failure_seen=1
                failure_cutoff="$next_index"
                next_index=$((next_index + 1))
                break
            fi
            BEAGLE_BOUNDED_COMPLETION_RECEIPT="$receipt" \
                "$NATIVE_SUPERVISOR" "$deadline" "$core_kill_grace" -- \
                "$RACKET" "$AST_VERIFIER" "$ast" "$source" \
                >"$metadata" 2>>"$log" &
            pid=$!
            live+=("$pid")
            ast_verify_live_pids+=("$pid")
            index_by_pid["$pid"]="$next_index"
            next_index=$((next_index + 1))
        done

        ((${#live[@]} > 0)) || break
        finished_pid=""
        if wait -n -p finished_pid "${live[@]}"; then
            status=0
        else
            status=$?
        fi
        if [[ -z "$finished_pid" ||
              -z "${index_by_pid[$finished_pid]+known}" ]]; then
            echo "beagle build: checked-AST supervisor wait contract failed" >&2
            stop_ast_verify_workers
            return 2
        fi
        finished_index="${index_by_pid[$finished_pid]}"
        unset 'index_by_pid[$finished_pid]'
        next_live=()
        for pid in "${live[@]}"; do
            [[ "$pid" == "$finished_pid" ]] || next_live+=("$pid")
        done
        live=("${next_live[@]}")
        ast_verify_live_pids=("${live[@]}")

        receipt_value=""
        [[ -f "${receipt_by_index[$finished_index]}" ]] &&
            receipt_value="$(<"${receipt_by_index[$finished_index]}")"
        if [[ "$status" -eq 124 ]]; then
            expected_receipt='subtree-reaped-v0 timeout status=124'
        else
            expected_receipt="subtree-reaped-v0 exit status=$status"
        fi
        if [[ "$receipt_value" != "$expected_receipt" ]]; then
            printf 'beagle build: ast-verify-%s supervisor receipt mismatch: status=%s receipt=%s\n' \
                "$finished_index" "$status" "${receipt_value:-missing}" \
                >>"${log_by_index[$finished_index]}"
            status=2
            unset 'canceled_by_index[$finished_index]'
        fi
        if [[ "$status" -eq 0 &&
              -z "${canceled_by_index[$finished_index]+canceled}" ]]; then
            metadata="$work/source_${finished_index}.metadata"
            source="${normalized_sources[$finished_index]}"
            checked_fields=()
            if ! mapfile -d '' -t checked_fields <"$metadata"; then
                printf 'beagle build: checked AST verifier metadata is unreadable for %s\n' \
                    "$source" >>"${log_by_index[$finished_index]}"
                status=2
            elif [[ ${#checked_fields[@]} -ne 4 ]]; then
                printf 'beagle build: checked AST verifier returned malformed metadata for %s\n' \
                    "$source" >>"${log_by_index[$finished_index]}"
                status=2
            fi
            if [[ "$status" -eq 0 ]]; then
                interface_sha256_path="$work/source_${finished_index}.interface.sha256"
                if [[ -f "$interface_sha256_path" ]] &&
                   interface_sha256="$(tr -d '\r\n' <"$interface_sha256_path")"; then
                    :
                else
                    interface_sha256=""
                fi
                if [[ ! "$interface_sha256" =~ ^sha256:[0-9a-f]{64}$ ]]; then
                    printf 'beagle build: checked AST bundle returned a malformed interface digest for %s\n' \
                        "$source" >>"${log_by_index[$finished_index]}"
                    status=2
                else
                    ast_verified_logical_paths["$finished_index"]="${checked_fields[0]}"
                    ast_verified_interface_sha256s["$finished_index"]="$interface_sha256"
                fi
            fi
        fi
        status_by_index["$finished_index"]="$status"
        if [[ -n "${canceled_by_index[$finished_index]+canceled}" ]]; then
            printf 'beagle build: phase ast-verify-%s CANCELLED (%s)\n' \
                "$finished_index" "$status" >>"${log_by_index[$finished_index]}"
        elif [[ "$status" -eq 0 ]]; then
            printf 'beagle build: phase ast-verify-%s END\n' \
                "$finished_index" >>"${log_by_index[$finished_index]}"
        elif [[ "$status" -eq 124 ]]; then
            printf 'beagle build: phase ast-verify-%s TIMEOUT (%s)\n' \
                "$finished_index" "$status" >>"${log_by_index[$finished_index]}"
        else
            printf 'beagle build: phase ast-verify-%s ERROR (%s)\n' \
                "$finished_index" "$status" >>"${log_by_index[$finished_index]}"
        fi

        if [[ -z "${canceled_by_index[$finished_index]+canceled}" &&
              "$status" -ne 0 ]]; then
            failure_seen=1
            if [[ "$status" -ne 124 &&
                  ("$failure_cutoff" -lt 0 ||
                   "$finished_index" -lt "$failure_cutoff") ]]; then
                failure_cutoff="$finished_index"
                for pid in "${live[@]}"; do
                    live_index="${index_by_pid[$pid]}"
                    if ((live_index > failure_cutoff)) &&
                       [[ -z "${canceled_by_index[$live_index]+canceled}" ]]; then
                        canceled_by_index["$live_index"]=1
                        kill -TERM "$pid" 2>/dev/null || true
                    fi
                done
            fi
        fi
    done

    ast_verify_live_pids=()
    for ((index = 0; index < next_index; index++)); do
        [[ -n "${started_by_index[$index]+started}" ]] || continue
        cat "${log_by_index[$index]}" >&2
    done

    # A completed verifier defect outranks a deadline breach, and source order
    # breaks ties. Workers canceled only because a lower-index defect decided
    # the cohort are not independent failures.
    for ((index = 0; index < next_index; index++)); do
        [[ -z "${canceled_by_index[$index]+canceled}" ]] || continue
        status="${status_by_index[$index]:-2}"
        if [[ "$status" -ne 0 && "$status" -ne 124 ]]; then
            selected_index="$index"
            selected_status="$status"
            break
        fi
    done
    if ((selected_index < 0)); then
        for ((index = 0; index < next_index; index++)); do
            [[ -z "${canceled_by_index[$index]+canceled}" ]] || continue
            status="${status_by_index[$index]:-2}"
            if [[ "$status" -eq 124 ]]; then
                selected_index="$index"
                selected_status=124
                break
            fi
        done
    fi
    if ((selected_index >= 0)); then
        echo "beagle build: checked AST verification failed for ${normalized_sources[$selected_index]} (ast-verify-$selected_index status=$selected_status)" >&2
        return "$selected_status"
    fi
    return 0
}

# The Core result cache lock guards two SEPARATE critical sections, not one:
# the pre-build lookup (which retires a corrupt entry in place) and each
# publication (which renames a validated staging tree into its final name).
# It used to be taken for the lookup and never released on any path, so one
# `exec {fd}>` held it until the process exited. A Wasm build always reaches
# the lookup, because `wasm_selected` bypasses the alias fast path above it, so
# a cold Wasm build held the lock through materialization and any same-key
# sibling waited out the whole `-w` window and then died. Each section now
# takes it and gives it back, and a publication that finds a peer's identical
# entry already in place validates and adopts it instead of failing.
core_result_lock_fd=""
core_result_lock_held=0
acquire_core_result_lock() {
    local phase="$1"
    [[ -n "$core_result_lock_fd" ]] || return 0
    [[ "$core_result_lock_held" == 0 ]] || return 0
    run_phase_leaf "$phase" "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        flock -x -w "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        "$core_result_lock_fd" ||
        die "timed out acquiring the Core result cache lock ($phase)"
    core_result_lock_held=1
}
release_core_result_lock() {
    [[ -n "$core_result_lock_fd" ]] || return 0
    [[ "$core_result_lock_held" == 1 ]] || return 0
    core_result_lock_held=0
    flock -u "$core_result_lock_fd"
}

out=""
abi="lp64"
materializers=()
entries=()
module_roots=()
sources=()
simd_required=0
emit_workers=""
emit_workers_explicit=0

while [[ $# -gt 0 ]]; do
    case "$1" in
        --out)
            [[ $# -ge 2 ]] || die "--out needs a directory"
            [[ -z "$out" ]] || die "--out may be specified only once"
            out="$2"
            shift 2
            ;;
        --out=*)
            [[ -z "$out" ]] || die "--out may be specified only once"
            out="${1#*=}"
            shift
            ;;
        --abi)
            [[ $# -ge 2 ]] || die "--abi needs a profile id"
            abi="$2"
            shift 2
            ;;
        --abi=*)
            abi="${1#*=}"
            shift
            ;;
        --materializer)
            [[ $# -ge 2 ]] || die "--materializer needs one of: $BEAGLE_MATERIALIZER_IDS_LIST"
            materializers+=("$2")
            shift 2
            ;;
        --materializer=*)
            materializers+=("${1#*=}")
            shift
            ;;
        --entry)
            [[ $# -ge 2 ]] || die "--entry needs a qualified NS/NAME"
            entries+=("$2")
            shift 2
            ;;
        --entry=*)
            entries+=("${1#*=}")
            shift
            ;;
        --module-root)
            [[ $# -ge 2 ]] ||
                die "--module-root needs LOGICAL_PREFIX=PHYSICAL_DIRECTORY"
            module_roots+=("$2")
            shift 2
            ;;
        --module-root=*)
            module_roots+=("${1#*=}")
            shift
            ;;
        --simd)
            simd_required=1
            shift
            ;;
        --emit-workers)
            [[ $# -ge 2 ]] || die "--emit-workers needs a positive integer"
            [[ -z "$emit_workers" ]] || die "--emit-workers may be specified only once"
            emit_workers="$2"
            emit_workers_explicit=1
            shift 2
            ;;
        --emit-workers=*)
            [[ -z "$emit_workers" ]] || die "--emit-workers may be specified only once"
            emit_workers="${1#*=}"
            emit_workers_explicit=1
            shift
            ;;
        --help|-h)
            usage
            exit 0
            ;;
        --)
            shift
            sources+=("$@")
            break
            ;;
        -*)
            die "unknown option: $1 (try --help)"
            ;;
        *)
            sources+=("$1")
            shift
            ;;
    esac
done

[[ -n "$out" ]] || die "--out DIR is required"
[[ ${#materializers[@]} -gt 0 ]] || die "at least one --materializer is required"
if [[ -z "$emit_workers" ]]; then
    if command -v nproc >/dev/null 2>&1; then
        emit_workers="$(nproc)"
    elif command -v getconf >/dev/null 2>&1; then
        emit_workers="$(getconf _NPROCESSORS_ONLN 2>/dev/null || printf '1')"
    else
        emit_workers=1
    fi
fi
[[ "$emit_workers" =~ ^[1-9][0-9]*$ ]] ||
    die "--emit-workers expects a positive integer, got: $emit_workers"
declare -A seen_materializers=()
wasm_selected=false
for materializer in "${materializers[@]}"; do
    beagle_known_materializer "$materializer" ||
        die "unknown materializer '$materializer' (expected $BEAGLE_MATERIALIZER_IDS_LIST)"
    [[ -z "${seen_materializers[$materializer]:-}" ]] ||
        die "duplicate materializer: $materializer"
    seen_materializers["$materializer"]=1
    [[ "$materializer" == "wasm" ]] && wasm_selected=true
done
core_checkpoint_eligible=1
for materializer in "${materializers[@]}"; do
    [[ "$materializer" != "wasm" ]] || core_checkpoint_eligible=0
done
$wasm_selected && [[ "$abi" != "wasm32" ]] &&
    die "materializer 'wasm' requires --abi wasm32 (got '$abi')"
declare -A seen_entries=()
declare -A seen_entry_exports=()
for entry in "${entries[@]}"; do
    [[ "$entry" =~ ^[^[:space:]/]+/[^[:space:]/]+$ ]] ||
        die "--entry must be one whitespace-free NS/NAME: $entry"
    entry_namespace="${entry%/*}"
    entry_name="${entry##*/}"
    [[ -n "$entry_namespace" && -n "$entry_name" && "$entry_namespace" != */* ]] ||
        die "--entry must contain exactly one namespace separator: $entry"
    [[ -z "${seen_entries[$entry]:-}" ]] || die "duplicate --entry: $entry"
    seen_entries["$entry"]=1
    if $wasm_selected; then
        # The Wasm export name flattens NS/NAME into one C identifier, so two
        # distinct entries may only build together when their flattened names
        # stay distinct.
        entry_export="beagle_wasm_entry_v1__$(printf '%s' "$entry_namespace" |
            LC_ALL=C sed 's/[^A-Za-z0-9]/_/g')__$(printf '%s' "$entry_name" |
            LC_ALL=C sed 's/[^A-Za-z0-9]/_/g')"
        [[ -z "${seen_entry_exports[$entry_export]:-}" ]] ||
            die "entries ${seen_entry_exports[$entry_export]} and $entry" \
                "flatten to one Wasm export name ($entry_export)"
        seen_entry_exports["$entry_export"]="$entry"
    fi
done
[[ ${#sources[@]} -gt 0 ]] || die "provide at least one canonical .bgl source"

for source in "${sources[@]}"; do
    [[ -f "$source" ]] || die "source file not found: $source"
    [[ "$source" == *.bgl ]] ||
        die "Core accepts .bgl only; hosted Clojure stays .bclj with #lang beagle/clj: $source"
    first_line="$(sed -n '1p' "$source")"
    [[ "$first_line" =~ ^#lang[[:space:]]+beagle[[:space:]]*$ ]] ||
        die "$source must begin with bare #lang beagle"
done

for command in bb flock python3 sha256sum sync timeout; do
    command -v "$command" >/dev/null 2>&1 || die "required command is unavailable: $command"
done
[[ -f "$PROJECTOR" ]] || die "source-fact projector is unavailable: $PROJECTOR"
[[ -f "$SEMANTIC_READ_STORE" ]] ||
    die "semantic-read Store adapter is unavailable: $SEMANTIC_READ_STORE"
[[ -f "$SEMANTIC_READ_BLOB_STORE" ]] ||
    die "semantic-read blob Store adapter is unavailable: $SEMANTIC_READ_BLOB_STORE"
[[ -d "$BEAGLE_DIR/store/out" && ! -L "$BEAGLE_DIR/store/out" ]] ||
    die "semantic-read Store projection is unavailable: $BEAGLE_DIR/store/out"
[[ -f "$SEMANTIC_READ_DATABASE" ]] ||
    die "semantic-read Store database is unavailable: $SEMANTIC_READ_DATABASE"
[[ -f "$SEMANTIC_READ_WRITER_AUTHORITY" ]] ||
    die "semantic-read Store writer authority is unavailable: $SEMANTIC_READ_WRITER_AUTHORITY"
[[ -f "$AST_VERIFIER" ]] || die "checked-AST verifier is unavailable: $AST_VERIFIER"
[[ -f "$FINALIZER" ]] || die "build finalizer is unavailable: $FINALIZER"
[[ -x "$CORE_COMPILER_PROJECTION" ]] ||
    die "Core compiler projection builder is unavailable: $CORE_COMPILER_PROJECTION"
[[ -f "$MODULE_SOURCE_ROOT_CLI" ]] ||
    die "module source root resolver is unavailable: $MODULE_SOURCE_ROOT_CLI"

mkdir -p "$out"
out="$(cd "$out" && pwd)"
[[ "$out" != "/" ]] || die "--out may not be the filesystem root"
managed_artifacts=(
    source.facts
    report.txt
    module.native-program
    module.native-program.sha256
    module.simd-plan-v0
    module.simd-plan-v0.sha256
    native.receipts
    native.entry-map
    c17.receipt
    wasm.receipt
    module_0.h
    module_0.c
    module_0.ssa
    module_0.wasm
    module_0.wasm.sha256
    module_0.wasm.seams
    wasm-report.txt
    wasm-audit.txt
    wasm.retention.c
    wasm.adapter.c
    wasm.entry-contract.clj
    wasm.seams.clj
    wasm.ast-verifier.rkt
    wasm.receipt-finalizer.clj
    wasm.materializer.sh
    wasm.supervisor.rkt
    wasm.cc-identity.txt
    wasm.ld-identity.txt
    wasm.runtime-identity.txt
    build.manifest
    build.manifest.sha256
    native_shim.h
    native_shim.c
    native_parallel.h
    native_parallel.c
    native_unicode15_data.h
    UNICODE-LICENSE.txt
)
work="$(mktemp -d "$out/.beagle-stage.XXXXXX")"
marker_pending="$out/.build.manifest.sha256.pending.$$"
build_committed=0
commit_started=0
work_cleaned=0
core_result_staging=""
core_result_alias_staging=""
core_checkpoint_staging=""
core_checkpoint_alias_staging=""
cleanup() {
    local rc=$?
    stop_ast_verify_workers
    if [[ "$commit_started" == "1" && "$build_committed" != "1" ]]; then
        rm -f -- "$out/build.manifest.sha256"
        for artifact in "${managed_artifacts[@]}"; do
            rm -f -- "$out/$artifact"
        done
        sync -f "$out" 2>/dev/null || true
    fi
    if [[ "$build_committed" != "1" ]]; then
        rm -f -- "$marker_pending"
        if [[ -n "$core_result_staging" && -d "$core_result_staging" ]]; then
            rm -rf -- "${core_result_staging:?}"
        fi
        if [[ -n "$core_result_alias_staging" &&
              -f "$core_result_alias_staging" ]]; then
            rm -f -- "$core_result_alias_staging"
        fi
        if [[ -n "$core_checkpoint_staging" &&
              -d "$core_checkpoint_staging" ]]; then
            rm -rf -- "${core_checkpoint_staging:?}"
        fi
        if [[ -n "$core_checkpoint_alias_staging" &&
              -f "$core_checkpoint_alias_staging" ]]; then
            rm -f -- "$core_checkpoint_alias_staging"
        fi
        if [[ "$work_cleaned" != "1" ]]; then
            rm -rf "${work:?}"
        fi
    fi
    return "$rc"
}
publish_interrupted_progress() {
    local progress_path="${BEAGLE_CORE_REPORT:-}"
    if [[ -n "$progress_path" && -f "$progress_path" ]]; then
        echo "beagle build: interrupted Core progress" >&2
        cat -- "$progress_path" >&2
    fi
}
interrupted() { publish_interrupted_progress; exit 143; }
trap cleanup EXIT
trap interrupted HUP INT TERM
mkdir -p "$work/artifacts"

# Core lowering can reuse a content-addressed compiler projection after the
# checkout that selected it is gone. Snapshot the hosted semantic-read closure
# before the long checked-AST phases so that reuse does not retain a physical
# checkout dependency. Its bytes join compiler_source_digest below; absolute
# snapshot and worktree paths never enter the semantic identity.
semantic_read_classpath_root="$work/semantic-read-classpath"
mkdir -p "$semantic_read_classpath_root/native-core/bin" \
    "$semantic_read_classpath_root/store"
cp -- "$SEMANTIC_READ_STORE" "$SEMANTIC_READ_BLOB_STORE" \
    "$semantic_read_classpath_root/native-core/bin/"
cp -- "$SEMANTIC_READ_DATABASE" "$SEMANTIC_READ_WRITER_AUTHORITY" \
    "$semantic_read_classpath_root/store/"
cp -a --no-preserve=mode -- "$BEAGLE_DIR/store/out" "$semantic_read_classpath_root/store/out"
semantic_read_classpath="$semantic_read_classpath_root/store/out:$semantic_read_classpath_root/native-core/bin"

facts="$work/source-facts.manifest"
projector_args=()
wasm_checked_sources=()
source_snapshot_paths=()
source_snapshot_digests=()
explicit_normalized_sources=()
explicit_source_ids=()
normalized_sources=()
source_ids=()
source_digest_befores=()
for source in "${sources[@]}"; do
    source="$(realpath "$source")"
    explicit_normalized_sources+=("$source")
    explicit_source_ids+=("$(beagle_source_id "$source" "${module_roots[@]}")")
done

module_root_args=()
for module_root in "${module_roots[@]}"; do
    module_root_args+=(--module-root "$module_root")
done
closure_args=("${module_root_args[@]}")
for source_index in "${!explicit_normalized_sources[@]}"; do
    closure_args+=(
        --source
        "${explicit_normalized_sources[$source_index]}"
        "${explicit_source_ids[$source_index]}"
    )
done
closure_fields_path="$work/module-source-closure.fields"
run_phase_leaf module-source-closure "${BEAGLE_CORE_AST_TIMEOUT_SECONDS:-60}" \
    "$RACKET" "$MODULE_SOURCE_ROOT_CLI" "${closure_args[@]}" \
    >"$closure_fields_path"
mapfile -d '' -t closure_fields <"$closure_fields_path"
(( ${#closure_fields[@]} > 0 && ${#closure_fields[@]} % 2 == 0 )) ||
    die "module source root resolver returned malformed path/id pairs"
for ((closure_index = 0; closure_index < ${#closure_fields[@]}; closure_index += 2)); do
    source="${closure_fields[$closure_index]}"
    source_id="${closure_fields[$((closure_index + 1))]}"
    [[ -n "$source" && -n "$source_id" ]] ||
        die "module source root resolver returned an empty path or source id"
    [[ -f "$source" ]] ||
        die "module source root resolver returned a missing source: $source"
    normalized_sources+=("$source")
    source_ids+=("$source_id")
    source_digest="$(sha256sum "$source" | awk '{print $1}')"
    source_digest_befores+=("$source_digest")
    source_snapshot_paths+=("$source")
    source_snapshot_digests+=("$source_digest")
done

core_modules=()
mapfile -t core_modules < <("$CORE_COMPILER_PROJECTION" --list-sources)
[[ ${#core_modules[@]} -gt 0 ]] || die "Core compiler projection source list is empty"

# One `sha256sum` for a whole list of files, one digest per line in ARGUMENT
# ORDER, so a caller can zip the digests back onto its own list. Hashing a file
# at a time through a command substitution costs three processes each; a build
# identity covers 26 such files and is re-derived on every revalidation round,
# which is where most of a warm rebuild's process count came from. The digests
# are identical either way — this changes how many processes read the files,
# never which files are read.
sha256_digests_in_order() {
    sha256sum "$@" | awk '{ print $1 }'
}

# Every tracked file that can change emitted bytes is hashed here by content,
# so a modified worktree misses on the file it modified instead of being locked
# out of the cache wholesale. `native-core/shim/*` is deliberately absent: those
# bytes are restaged from the worktree on every run including a cache hit, and
# nothing reads them during emission, so they can never be served stale.
cache_key_material() {
    local relative digest_index input_index
    local -a key_inputs present_inputs present_flags digests
    key_inputs=("${core_modules[@]}"
                bin/beagle-build-core
                bin/_beagle-dev-unit-rule-identity
                bin/beagle-core-compiler-projection
                bin/beagle-build-all
                bin/beagle-build
                bin/_beagle-racket
                bin/_beagle-source-id
                bin/_beagle-compiler-provenance
                bin/_beagle-rust-supervisor
                native-core/bin/emit-workers
                native-core/validation/build-finalize.clj
                share/targets.sh
                flake.nix
                flake.lock)
    # Presence is decided ONCE and the hash list is built from that same
    # decision, so a file appearing or vanishing mid-scan cannot slide the
    # digests out of step with the names they label.
    present_inputs=()
    present_flags=()
    for relative in "${key_inputs[@]}"; do
        if [[ -f "$BEAGLE_DIR/$relative" ]]; then
            present_flags+=(1)
            present_inputs+=("$BEAGLE_DIR/$relative")
        else
            present_flags+=(0)
        fi
    done
    digests=()
    if ((${#present_inputs[@]} > 0)); then
        mapfile -t digests < <(sha256_digests_in_order "${present_inputs[@]}")
        ((${#digests[@]} == ${#present_inputs[@]})) ||
            die "cache key hashing returned ${#digests[@]} digests for ${#present_inputs[@]} inputs"
    fi
    digest_index=0
    for input_index in "${!key_inputs[@]}"; do
        # An absent input is its own key value, so deleting one moves the key
        # rather than failing the build.
        if [[ "${present_flags[$input_index]}" == 1 ]]; then
            printf '%s %s\n' "${key_inputs[$input_index]}" "${digests[$digest_index]}"
            digest_index=$((digest_index + 1))
        else
            printf '%s absent\n' "${key_inputs[$input_index]}"
        fi
    done
    find "$BEAGLE_DIR/beagle-lib" -name compiled -prune -o -name '*.rkt' -print0 |
        LC_ALL=C sort -z |
        xargs -0 sha256sum |
        awk -v prefix="${#BEAGLE_DIR}" '{ print substr($0, 67 + prefix + 1), substr($0, 1, 64) }'
    find "$semantic_read_classpath_root" -type f -print0 |
        LC_ALL=C sort -z |
        xargs -0 sha256sum |
        awk -v prefix="${#semantic_read_classpath_root}" \
            '{ print "semantic-read-classpath" substr($0, 67 + prefix), substr($0, 1, 64) }'
    printf 'native-supervisor %s\n' \
        "$(sha256sum "$NATIVE_SUPERVISOR" | awk '{print $1}')"
    if [[ -n "$native_compiler_digest" ]]; then
        printf 'compiler-artifact-sha256 %s\n' "$native_compiler_digest"
    fi
}

cache_root="${BEAGLE_CORE_BUILD_CACHE:-${XDG_CACHE_HOME:-$HOME/.cache}/beagle/build-core}"
compiled_override="${BEAGLE_CORE_COMPILED_OVERRIDE:-}"
native_compiler_bin=""
native_compiler_digest=""
if [[ "${BEAGLE_NATIVE_COMPILER_BIN+x}" == x ]]; then
    [[ -z "$compiled_override" ]] ||
        die "BEAGLE_NATIVE_COMPILER_BIN cannot be combined with BEAGLE_CORE_COMPILED_OVERRIDE"
    [[ -n "$BEAGLE_NATIVE_COMPILER_BIN" ]] ||
        die "BEAGLE_NATIVE_COMPILER_BIN must name an executable native compiler"
    [[ -x "$BEAGLE_NATIVE_COMPILER_BIN" ]] ||
        die "native compiler is unavailable or not executable: $BEAGLE_NATIVE_COMPILER_BIN"
    native_compiler_bin="$(realpath "$BEAGLE_NATIVE_COMPILER_BIN")"
    native_compiler_digest="$(sha256sum "$native_compiler_bin" | awk '{print $1}')"
fi
compiler_source_digest="$(cache_key_material | sha256sum | awk '{print $1}')"
semantic_read_store_cohort="$(
    find "$semantic_read_classpath_root" -type f -print0 |
        LC_ALL=C sort -z |
        xargs -0 sha256sum |
        awk -v prefix="${#semantic_read_classpath_root}" \
            '{ print substr($0, 67 + prefix), substr($0, 1, 64) }' |
        sha256sum |
        awk '{print $1}'
)"
[[ "$semantic_read_store_cohort" =~ ^[0-9a-f]{64}$ ]] ||
    die "semantic-read Store cohort identity is malformed"
compiler_commit="$(beagle_resolve_compiler_commit "$BEAGLE_DIR")" || exit $?
if [[ -n "$compiled_override" ]]; then
    [[ -d "$compiled_override" ]] ||
        die "native Core compiler projection is unavailable: $compiled_override"
    [[ -f "$compiled_override/native/core.clj" ]] ||
        die "native Core compiler projection omitted native/core.clj: $compiled_override"
fi
core_result_cache_enabled=1
core_result_cache_bypass_reason="untrusted-compiler"
# A dirty worktree is cacheable: `cache_key_material` closes over every input
# by content, so a modified file misses on its own digest. The commit still
# gates because it is embedded in the artifact through BEAGLE_CORE_COMMIT.
if [[ "$dev_fact_reuse" == 1 ]]; then
    # A development fact run must reach the per-unit read path instead of
    # disappearing behind the older whole-result cache.
    core_result_cache_enabled=0
    core_result_cache_bypass_reason="dev-fact-reuse"
fi

# The interpreter half of a build identity. It is re-derived on every
# revalidation round, and booting two interpreters plus hashing a 50 MB Racket
# binary costs ~176 ms a round. The memo is keyed on the RESOLVED interpreter
# paths, and a `/nix/store` path is itself the content address of what it names:
# an unchanged path there is proof of unchanged bytes, so the guard is intact.
# An interpreter resolved outside the store carries no such proof and is re-read
# every round.
toolchain_identity_memo=""
toolchain_identity_memo_key=""
write_toolchain_identity() {
    local racket_path bb_path key
    racket_path="$(realpath "$RACKET")"
    bb_path="$(realpath "$(command -v bb)")"
    key="$racket_path::$bb_path"
    if [[ -z "$toolchain_identity_memo" ||
          "$key" != "$toolchain_identity_memo_key" ||
          "$racket_path" != /nix/store/* || "$bb_path" != /nix/store/* ]]; then
        toolchain_identity_memo="$(
            printf 'racket %q\n' "$($RACKET --version 2>&1 | sed -n '1p')"
            printf 'racket-bin-sha256 %s\n' \
                "$(sha256sum "$racket_path" | awk '{print $1}')"
            printf 'bb %q\n' "$(bb --version 2>&1 | sed -n '1p')"
            printf 'bb-bin-sha256 %s\n' "$(sha256sum "$bb_path" | awk '{print $1}')"
        )"$'\n'
        toolchain_identity_memo_key="$key"
    fi
    printf '%s' "$toolchain_identity_memo"
}

# The tools every build identity names, hashed in one process instead of one
# command substitution each. Both identity manifests emit this same block.
identity_tools=("$PROJECTOR" "$AST_VERIFIER" "$MODULE_SOURCE_ROOT_CLI"
                "$BIN/beagle-ast")
write_identity_tool_lines() {
    local tool_index
    local -a digests
    mapfile -t digests < <(sha256_digests_in_order "${identity_tools[@]}")
    ((${#digests[@]} == ${#identity_tools[@]})) ||
        die "tool hashing returned ${#digests[@]} digests for ${#identity_tools[@]} tools"
    for tool_index in "${!identity_tools[@]}"; do
        printf 'tool %q %s\n' "${identity_tools[$tool_index]#"$BEAGLE_DIR/"}" \
            "${digests[$tool_index]}"
    done
}

# The two build identities -- the early one, which names what is known before
# the compiler projection exists, and the whole-result one -- differ only in
# their header. Everything from the compiler digests down to the materializer
# list was written out twice, in two functions a thousand lines apart, and a key
# input added to one of them and missed in the other is exactly the
# under-inclusion that makes a cache hit unsound. There is now one copy of each
# shared half, so an input cannot land in one identity and not the other.
write_identity_compiler_lines() {
    printf 'compiler-source-sha256 %s\n' "$compiler_source_digest"
    if [[ -n "$native_compiler_digest" ]]; then
        printf 'compiler-artifact-sha256 %s\n' "$native_compiler_digest"
    fi
}

write_identity_selection_lines() {
    local index
    printf 'abi %q\nsimd %s\n' "$abi" "$simd_required"
    write_identity_tool_lines
    write_toolchain_identity
    for index in "${!source_ids[@]}"; do
        printf 'source %06d %q %s\n' "$index" "${source_ids[$index]}" \
            "${source_digest_befores[$index]}"
    done
    for index in "${!entries[@]}"; do
        printf 'entry %06d %q\n' "$index" "${entries[$index]}"
    done
    for index in "${!materializers[@]}"; do
        printf 'materializer %06d %q\n' "$index" "${materializers[$index]}"
    done
}

write_core_result_early_manifest() {
    local destination="$1"
    {
        printf 'beagle-core-result-early-input/v1\n'
        write_identity_compiler_lines
        printf 'compiler-commit %s\n' "$compiler_commit"
        write_identity_selection_lines
    } >"$destination"
}

# A checkpoint freezes the program before any materializer runs. Its identity
# therefore contains the compiler, sources, entries, ABI, and SIMD selection,
# but never the downstream projection list carried by a whole-result identity.
write_core_checkpoint_identity() {
    local result_identity="$1" destination="$2"
    grep -v '^materializer [0-9]' "$result_identity" >"$destination"
}

# What both closure revalidations prove before they re-derive their manifest:
# every snapshotted source still hashes to what it hashed to, HEAD has not
# moved, and the compiler tree still hashes to the key this build committed to.
# This is re-derived, never memoized: a revalidation that trusts an earlier
# answer proves nothing, and each caller runs it on the far side of a lock wait
# that another build could have used to change the tree.
validate_identity_closure_prefix() {
    local source_index current_digest current_compiler_commit
    for source_index in "${!source_snapshot_paths[@]}"; do
        current_digest="$(sha256sum "${source_snapshot_paths[$source_index]}" |
            awk '{print $1}')"
        [[ "$current_digest" == "${source_snapshot_digests[$source_index]}" ]] ||
            return 1
    done
    current_compiler_commit="$(beagle_resolve_compiler_commit "$BEAGLE_DIR")" ||
        return 1
    [[ "$current_compiler_commit" == "$compiler_commit" ]] || return 1
    [[ "$(cache_key_material | sha256sum | awk '{print $1}')" == \
       "$compiler_source_digest" ]]
}

validate_core_result_early_closure() {
    local expected_input="$1" current_input="$2"
    validate_identity_closure_prefix || return 1
    write_core_result_early_manifest "$current_input"
    cmp -s "$expected_input" "$current_input"
}

validate_core_result_entry_from_alias() {
    local entry="$1" expected_key="$2" expected_input_digest="$3"
    local artifact_manifest_digest expected_names actual_names required
    [[ -d "$entry" && ! -L "$entry" &&
       -f "$entry/input.manifest" && ! -L "$entry/input.manifest" &&
       -f "$entry/artifacts.sha256" && ! -L "$entry/artifacts.sha256" &&
       -f "$entry/READY" && ! -L "$entry/READY" &&
       -d "$entry/artifacts" && ! -L "$entry/artifacts" ]] || return 1
    [[ "$(sha256sum "$entry/input.manifest" | awk '{print $1}')" == \
       "$expected_input_digest" ]] || return 1
    if find "$entry/artifacts" -mindepth 1 -maxdepth 1 ! -type f -print -quit |
        grep -q .; then
        return 1
    fi
    expected_names="$(awk '{ print $2 }' "$entry/artifacts.sha256" | LC_ALL=C sort)"
    actual_names="$(find "$entry/artifacts" -mindepth 1 -maxdepth 1 -type f \
        -printf '%f\n' | LC_ALL=C sort)"
    [[ -n "$expected_names" && "$expected_names" == "$actual_names" ]] || return 1
    (cd "$entry/artifacts" &&
        sha256sum --check --strict ../artifacts.sha256 >/dev/null 2>&1) || return 1
    for required in source.facts report.txt module.native-program \
                    native.receipts native.entry-map; do
        [[ -f "$entry/artifacts/$required" ]] || return 1
    done
    if [[ -n "${seen_materializers[c17]:-}" ]]; then
        for required in module_0.h module_0.c c17.receipt; do
            [[ -f "$entry/artifacts/$required" ]] || return 1
        done
    fi
    if [[ -n "${seen_materializers[qbe]:-}" ]]; then
        [[ -f "$entry/artifacts/module_0.ssa" ]] || return 1
    fi
    artifact_manifest_digest="$(sha256sum "$entry/artifacts.sha256" | awk '{print $1}')"
    [[ "$(<"$entry/READY")" == \
       "beagle-core-result/v1 $expected_key $artifact_manifest_digest" ]]
}

validate_core_checkpoint_entry_from_alias() {
    local entry="$1" expected_key="$2" expected_input_digest="$3"
    local artifact_manifest_digest expected_names actual_names allowed_names required
    local expected_root_names actual_root_names
    [[ -d "$entry" && ! -L "$entry" &&
       -f "$entry/input.manifest" && ! -L "$entry/input.manifest" &&
       -f "$entry/artifacts.sha256" && ! -L "$entry/artifacts.sha256" &&
       -f "$entry/READY" && ! -L "$entry/READY" &&
       -d "$entry/artifacts" && ! -L "$entry/artifacts" ]] || return 1
    expected_root_names="$(printf '%s\n' READY artifacts artifacts.sha256 input.manifest |
        LC_ALL=C sort)"
    actual_root_names="$(find "$entry" -mindepth 1 -maxdepth 1 -printf '%f\n' |
        LC_ALL=C sort)"
    [[ "$expected_root_names" == "$actual_root_names" ]] || return 1
    [[ "$(sha256sum "$entry/input.manifest" | awk '{print $1}')" == \
       "$expected_input_digest" ]] || return 1
    if find "$entry/artifacts" -mindepth 1 -maxdepth 1 ! -type f -print -quit |
        grep -q .; then
        return 1
    fi
    expected_names="$(awk '{ print $2 }' "$entry/artifacts.sha256" | LC_ALL=C sort)"
    actual_names="$(find "$entry/artifacts" -mindepth 1 -maxdepth 1 -type f \
        -printf '%f\n' | LC_ALL=C sort)"
    [[ -n "$expected_names" && "$expected_names" == "$actual_names" ]] || return 1
    allowed_names="$(printf '%s\n' frozen-native-stage.wire-v1 \
        qbe-frozen-native-stage.wire-v1 \
        module.native-program native.entry-map native.receipts report.head \
        source.facts stage-progress | LC_ALL=C sort)"
    if [[ "$simd_required" == 1 ]]; then
        allowed_names="$(printf '%s\n%s\n' "$allowed_names" module.simd-plan-v0 |
            LC_ALL=C sort)"
    fi
    [[ "$allowed_names" == "$actual_names" ]] || return 1
    (cd "$entry/artifacts" &&
        sha256sum --check --strict ../artifacts.sha256 >/dev/null 2>&1) || return 1
    for required in source.facts frozen-native-stage.wire-v1 \
                    qbe-frozen-native-stage.wire-v1 \
                    module.native-program native.receipts native.entry-map \
                    report.head stage-progress; do
        [[ -f "$entry/artifacts/$required" &&
           ! -L "$entry/artifacts/$required" ]] || return 1
    done
    if [[ "$simd_required" == 1 ]]; then
        [[ -f "$entry/artifacts/module.simd-plan-v0" &&
           ! -L "$entry/artifacts/module.simd-plan-v0" ]] || return 1
    fi
    artifact_manifest_digest="$(sha256sum "$entry/artifacts.sha256" | awk '{print $1}')"
    [[ "$(<"$entry/READY")" == \
       "beagle-core-pre-materializer/v1 $expected_key $artifact_manifest_digest" ]]
}

core_result_early_input="$work/core-result-early-input.manifest"
write_core_result_early_manifest "$core_result_early_input"
core_result_early_key="$(sha256sum "$core_result_early_input" | awk '{print $1}')"
core_checkpoint_early_input="$work/core-checkpoint-early-input.manifest"
write_core_checkpoint_identity "$core_result_early_input" \
    "$core_checkpoint_early_input"
core_checkpoint_early_key="$(sha256sum "$core_checkpoint_early_input" | awk '{print $1}')"
core_result_alias_root="$cache_root/aliases"
core_result_root="$cache_root/results"
core_c17_attestation_root="$cache_root/c17-attestations"
core_result_alias="$core_result_alias_root/$core_result_early_key"
core_checkpoint_alias_root="$cache_root/checkpoint-aliases"
core_checkpoint_root="$cache_root/checkpoints"
core_checkpoint_wire_attestation_root=\
"$cache_root/checkpoint-wire-attestations"
core_checkpoint_alias="$core_checkpoint_alias_root/$core_checkpoint_early_key"
# Writer authority is local to one physical compiler/source universe. Source
# bytes stay out of this scope key so the Store can reuse its independently
# authenticated per-module shards after an edit; different checkouts or source
# sets must never inherit one process-global writer authority.
source_fact_store_scope_key="$(
    {
        printf 'beagle-source-fact-store-scope/v1\0%s\0' "$BEAGLE_DIR"
        printf '%s\0' "${normalized_sources[@]}" | LC_ALL=C sort -z
    } | sha256sum | awk '{print $1}'
)"
source_fact_store_root="$cache_root/source-facts-$semantic_read_store_cohort"
source_fact_store="$source_fact_store_root/$source_fact_store_scope_key.storelog"
core_result_early_hit=0
core_checkpoint_early_hit=0
if [[ "$core_result_cache_enabled" == 1 && "$wasm_selected" == false ]]; then
    mkdir -p "$core_result_alias_root" "$cache_root/.alias-locks" \
        "$cache_root/.result-tmp"
    exec {core_result_alias_lock_fd}>"$cache_root/.alias-locks/$core_result_early_key.lock"
    run_phase_leaf core-result-alias-lock \
        "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        flock -x -w "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        "$core_result_alias_lock_fd" || die "timed out acquiring the Core result alias lock"
    validate_core_result_early_closure "$core_result_early_input" \
        "$work/core-result-early-input.lookup" ||
        die "Core result early input closure changed before alias lookup"
    if [[ -e "$core_result_alias" ]]; then
        alias_format=""
        alias_early_key=""
        alias_full_key=""
        alias_input_digest=""
        alias_extra=""
        if [[ -f "$core_result_alias" && ! -L "$core_result_alias" ]]; then
            read -r alias_format alias_early_key alias_full_key alias_input_digest \
                alias_extra <"$core_result_alias" || true
        fi
        alias_entry="$core_result_root/$alias_full_key"
        if [[ "$alias_format" == "beagle-core-result-alias/v1" &&
              "$alias_early_key" == "$core_result_early_key" &&
              "$alias_full_key" =~ ^[0-9a-f]{64}$ &&
              "$alias_input_digest" =~ ^[0-9a-f]{64}$ &&
              -z "${alias_extra:-}" ]] &&
           validate_core_result_entry_from_alias "$alias_entry" "$alias_full_key" \
               "$alias_input_digest"; then
            cp -a "$alias_entry/artifacts/." "$work/artifacts/"
            core_result_early_hit=1
            echo "beagle build: core-result-alias HIT $core_result_early_key -> $alias_full_key" >&2
        else
            corrupt_alias="$cache_root/.result-tmp/corrupt-alias.$core_result_early_key.$$"
            echo "beagle build: core-result-alias CORRUPT $core_result_early_key; retiring" >&2
            mv "$core_result_alias" "$corrupt_alias"
            rm -rf -- "${corrupt_alias:?}"
        fi
    fi
    if [[ "$core_result_early_hit" != 1 ]]; then
        echo "beagle build: core-result-alias MISS $core_result_early_key" >&2
    fi
    flock -u "$core_result_alias_lock_fd"
elif [[ "$core_result_cache_enabled" != 1 ]]; then
    echo "beagle build: core-result-alias BYPASS $core_result_cache_bypass_reason" >&2
else
    echo "beagle build: core-result-alias BYPASS wasm-materializer" >&2
fi

if [[ "$core_result_cache_enabled" == 1 && "$core_checkpoint_eligible" == 1 &&
      "$core_result_early_hit" != 1 ]]; then
    mkdir -p "$core_checkpoint_alias_root" "$core_checkpoint_root" \
        "$cache_root/.checkpoint-alias-locks" "$cache_root/.checkpoint-tmp"
    exec {core_checkpoint_alias_lock_fd}>\
        "$cache_root/.checkpoint-alias-locks/$core_checkpoint_early_key.lock"
    run_phase_leaf core-checkpoint-alias-lock \
        "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        flock -x -w "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        "$core_checkpoint_alias_lock_fd" ||
        die "timed out acquiring the Core checkpoint alias lock"
    validate_core_result_early_closure "$core_result_early_input" \
        "$work/core-result-early-input.checkpoint-lookup" &&
        write_core_checkpoint_identity \
            "$work/core-result-early-input.checkpoint-lookup" \
            "$work/core-checkpoint-early-input.lookup" &&
        cmp -s "$core_checkpoint_early_input" \
            "$work/core-checkpoint-early-input.lookup" ||
        die "Core checkpoint early input closure changed before alias lookup"
    if [[ -e "$core_checkpoint_alias" ]]; then
        checkpoint_alias_format=""
        checkpoint_alias_early_key=""
        checkpoint_alias_full_key=""
        checkpoint_alias_input_digest=""
        checkpoint_alias_extra=""
        if [[ -f "$core_checkpoint_alias" && ! -L "$core_checkpoint_alias" ]]; then
            read -r checkpoint_alias_format checkpoint_alias_early_key \
                checkpoint_alias_full_key checkpoint_alias_input_digest \
                checkpoint_alias_extra <"$core_checkpoint_alias" || true
        fi
        checkpoint_alias_entry="$core_checkpoint_root/$checkpoint_alias_full_key"
        if [[ "$checkpoint_alias_format" == \
                  "beagle-core-pre-materializer-alias/v1" &&
              "$checkpoint_alias_early_key" == "$core_checkpoint_early_key" &&
              "$checkpoint_alias_full_key" =~ ^[0-9a-f]{64}$ &&
              "$checkpoint_alias_input_digest" =~ ^[0-9a-f]{64}$ &&
              -z "${checkpoint_alias_extra:-}" ]] &&
           validate_core_checkpoint_entry_from_alias "$checkpoint_alias_entry" \
               "$checkpoint_alias_full_key" "$checkpoint_alias_input_digest"; then
            cp -- "$checkpoint_alias_entry/artifacts/source.facts" \
                "$work/artifacts/source.facts"
            core_checkpoint_early_hit=1
            echo "beagle build: core-checkpoint-alias HIT $core_checkpoint_early_key -> $checkpoint_alias_full_key" >&2
        else
            corrupt_checkpoint_alias=\
"$cache_root/.checkpoint-tmp/corrupt-alias.$core_checkpoint_early_key.$$"
            echo "beagle build: core-checkpoint-alias CORRUPT $core_checkpoint_early_key; retiring" >&2
            mv "$core_checkpoint_alias" "$corrupt_checkpoint_alias"
            rm -rf -- "${corrupt_checkpoint_alias:?}"
        fi
    fi
    if [[ "$core_checkpoint_early_hit" != 1 ]]; then
        echo "beagle build: core-checkpoint-alias MISS $core_checkpoint_early_key" >&2
    fi
    flock -u "$core_checkpoint_alias_lock_fd"
elif [[ "$core_result_early_hit" == 1 ]]; then
    echo "beagle build: core-checkpoint-alias SERVED-BY-RESULT-ALIAS $core_result_early_key" >&2
elif [[ "$core_result_cache_enabled" != 1 ]]; then
    echo "beagle build: core-checkpoint-alias BYPASS $core_result_cache_bypass_reason" >&2
else
    echo "beagle build: core-checkpoint-alias BYPASS checkpoint-ineligible" >&2
fi

if [[ "$core_result_early_hit" != 1 && "$core_checkpoint_early_hit" != 1 ]]; then
if [[ -n "${BEAGLE_CORE_FRONTEND_DIR:-}" ]]; then
    frontend_dir="$BEAGLE_CORE_FRONTEND_DIR"
    frontend_facts="${BEAGLE_CORE_FRONTEND_FACTS_MANIFEST:-}"
    [[ -d "$frontend_dir" ]] ||
        die "native Core frontend directory is unavailable: $frontend_dir"
    [[ -f "$frontend_facts" ]] ||
        die "native Core frontend facts manifest is unavailable: $frontend_facts"
    facts="$frontend_facts"
    source_index=0
    for source in "${normalized_sources[@]}"; do
        ast="$frontend_dir/source_${source_index}.ast.json"
        [[ -f "$ast" ]] ||
            die "native Core frontend omitted checked AST for source: $source"
        source_digest_after="$(sha256sum "$source" | awk '{print $1}')"
        [[ "${source_digest_befores[$source_index]}" == "$source_digest_after" ]] ||
            die "source changed while its native checked projection was captured: $source"
        python3 - "$ast" "${source_ids[$source_index]}" <<'PY'
import json
import pathlib
import sys

projection = json.loads(pathlib.Path(sys.argv[1]).read_text())
expected_source_id = sys.argv[2]
if projection.get("kind") != "beagle.checked-program":
    raise SystemExit("native Core frontend returned a non-checked projection")
if projection.get("phase") != "checked" or projection.get("target") != "core":
    raise SystemExit("native Core frontend returned a non-Core checked projection")
if projection.get("sourceId") != expected_source_id:
    raise SystemExit(
        "native Core frontend source identity mismatch: "
        f"expected {expected_source_id}, got {projection.get('sourceId')}"
    )
PY
        if $wasm_selected; then
            wasm_checked_sources+=(--checked-source "$source" "$ast")
        fi
        source_index=$((source_index + 1))
    done
else
bundle_ast="$work/source_bundle.ast.json"
run_phase "ast-bundle" "${BEAGLE_CORE_AST_TIMEOUT_SECONDS:-60}" \
    env BEAGLE_DEV_FACT_STORE="$dev_fact_store" \
    "$BIN/beagle-ast" --bundle "${module_root_args[@]}" \
    "${explicit_normalized_sources[@]}" >"$bundle_ast"

python3 - "$bundle_ast" "$work" "${#normalized_sources[@]}" \
    "${normalized_sources[@]}" "${source_ids[@]}" <<'PY'
import json
import pathlib
import sys

bundle_path = pathlib.Path(sys.argv[1])
work = pathlib.Path(sys.argv[2])
source_count = int(sys.argv[3])
sources = [
    str(pathlib.Path(source).resolve())
    for source in sys.argv[4:4 + source_count]
]
source_ids = sys.argv[4 + source_count:]
if len(source_ids) != source_count:
    raise SystemExit("beagle build: internal source path/id pairing is malformed")
bundle = json.loads(bundle_path.read_text())
if bundle.get("schemaVersion") != 2:
    raise SystemExit(
        f"beagle build: expected bundle AST schemaVersion 2, got {bundle.get('schemaVersion')}"
    )
module_list = bundle["modules"]
modules = {module["source"]: module for module in module_list}
if len(modules) != len(module_list):
    raise SystemExit("beagle build: bundle AST contains duplicate source ids")
if set(modules) != set(source_ids):
    missing = sorted(set(source_ids) - set(modules))
    extra = sorted(set(modules) - set(source_ids))
    raise SystemExit(
        f"beagle build: bundle AST source mismatch: missing={missing} extra={extra}"
    )
for index, (source, source_id) in enumerate(zip(sources, source_ids)):
    module = modules[source_id]
    destination = work / f"source_{index}.ast.json"
    destination.write_text(
        json.dumps(module["program"], sort_keys=True, separators=(",", ":")) + "\n"
    )
    interface_destination = work / f"source_{index}.interface.sha256"
    interface_destination.write_text(module["interfaceSha256"] + "\n")
PY

run_ast_verifier_pool
mkdir -p "$source_fact_store_root"
source_index=0
for source in "${normalized_sources[@]}"; do
    ast="$work/source_${source_index}.ast.json"
    logical_path="${ast_verified_logical_paths[$source_index]}"
    interface_sha256="${ast_verified_interface_sha256s[$source_index]}"
    projector_args+=(--input "$ast=$logical_path"
        --interface-sha256 "$logical_path=$interface_sha256")
    if $wasm_selected; then
        wasm_checked_sources+=(--checked-source "$source" "$ast")
    fi
    source_index=$((source_index + 1))
done
run_phase_leaf source-facts "${BEAGLE_CORE_FACTS_TIMEOUT_SECONDS:-60}" \
    bb "$PROJECTOR" "${projector_args[@]}" --output "$facts" --include-defs \
    --store "$source_fact_store"
fi
else
    facts="$work/artifacts/source.facts"
fi

# The hosted Core compiler projection is content-addressed separately from a
# program result. A full result key below names the exact generated projection
# bytes in addition to the early input closure.
#
# The projection is stored under the key of ITS OWN inputs, not under the whole
# build key. `compiler_source_digest` additionally covers this driver, the
# emission manager, the finalizer and the bounded-run supervisors — every one of
# which consumes the projection rather than producing it, so folding them into
# the projection's storage key discarded a multi-minute artifact on edits that
# provably cannot change a byte of it. The projection owns the definition of its
# own closure (`--print-key`), so there is exactly one such definition.
#
# The native route keeps `compiler_source_digest`: there the projection is
# emitted by the inline loop below, so this driver's own bytes are a real input
# to it, and the compiler artifact digest is already folded into that key.
if [[ -n "$native_compiler_bin" ]]; then
    projection_cache_key="$compiler_source_digest"
else
    projection_cache_key="$("$CORE_COMPILER_PROJECTION" --print-key)"
    [[ "$projection_cache_key" =~ ^[0-9a-f]{64}$ ]] ||
        die "Core compiler projection did not report a usable cache key"
fi
cache_entry="$cache_root/$projection_cache_key"
compiled="${compiled_override:-$cache_entry/compiled}"
: >"$work/build.log"

if [[ -z "$compiled_override" && ! -f "$cache_entry/.complete" ]]; then
    compiled="$work/compiled"
    sorted_build_args=()
    mapfile -t sorted_build_args < <(
        printf '%s\n' "${core_modules[@]}" |
            sed "s|^|$BEAGLE_DIR/|" | LC_ALL=C sort
    )
    # The Racket bundle remains the unset-selector shadow. Native selection is
    # explicit and content-addresses the executable before it can populate the
    # compiler projection cache.
    if [[ -n "$native_compiler_bin" ]]; then
        mkdir -p "$compiled"
        # Cold projection construction is a multi-minute phase, not a rebuild:
        # any edit under bin/ invalidates compiler_source_digest and forces it.
        run_phase core-compiler "${BEAGLE_CORE_COMPILER_TIMEOUT_SECONDS:-900}" \
            bash -c '
                set -euo pipefail
                out="$1"
                native="$2"
                module_root="$3"
                build_log="$4"
                shift 4
                sources=("$@")
                bundle_args=()
                for source in "${sources[@]}"; do
                    bundle_args+=(--source "$source")
                done
                for source in "${sources[@]}"; do
                    module="$(basename "$source" .bclj)"
                    destination="$out/native/$module.clj"
                    mkdir -p "$(dirname "$destination")"
                    "$native" emit --target clj \
                        --module-root "native-core/src=$module_root" \
                        "${bundle_args[@]}" "$source" \
                        >"$destination" 2>>"$build_log"
                    printf '  %s -> %s\n' "$source" "$destination" >>"$build_log"
                done
                printf "\n%d built, 0 error(s)\n" "${#sources[@]}" >>"$build_log"
            ' _ "$compiled" "$native_compiler_bin" \
            "$BEAGLE_DIR/native-core/src" "$work/build.log" \
            "${sorted_build_args[@]}" || {
                sed -n '1,200p' "$work/build.log" >&2
                exit 1
            }
        echo "beagle build: core-compiler-cache MISS $compiler_source_digest" >&2
    else
        projection_path_file="$work/core-compiler-projection.path"
        run_phase core-compiler "${BEAGLE_CORE_COMPILER_TIMEOUT_SECONDS:-900}" \
            "$CORE_COMPILER_PROJECTION" --cache >"$projection_path_file" || {
                sed -n '1,200p' "$work/build.log" >&2
                exit 1
            }
        mapfile -t cached_projection_paths <"$projection_path_file"
        [[ ${#cached_projection_paths[@]} -eq 1 &&
           -n "${cached_projection_paths[0]}" ]] ||
            die "Core compiler projection cache returned a malformed path"
        compiled="${cached_projection_paths[0]}"
        [[ -d "$compiled" && ! -L "$compiled" &&
           -f "$compiled/native/core.clj" &&
           "$(basename -- "$(dirname -- "$compiled")")" == \
               "$projection_cache_key" ]] ||
            die "Core compiler projection cache returned the wrong keyed entry"
    fi

    # Generated Clojure needs provider records referred and imported until the
    # emitter qualifies cross-module union patterns itself.
    if [[ -n "$native_compiler_bin" ]]; then
        core_records="$(sed -nE 's/.*\(defrecord ([^ ]+).*/\1/p' \
            "$compiled/native/core.clj" | tr '\n' ' ')"
        for module in stages simd lower obligations c11 slice fold_c17 body_c17 body_slice qbe; do
            generated="$compiled/native/$module.clj"
            [[ -f "$generated" ]] || continue
            sed -i 's/\[native\.core :as core\]/[native.core :as core :refer :all]/' "$generated"
            awk -v imp="(import '[native.core $core_records])" \
                '!seen && /^$/ { print imp; seen = 1 } { print }' \
                "$generated" >"$generated.tmp"
            mv "$generated.tmp" "$generated"
        done
    fi

    # Publish by renaming a fully-populated staging dir, so a reader never sees
    # a half-written entry and a lost race just keeps this run's own copy.
    if mkdir -p "$cache_root" 2>/dev/null &&
       staging="$(mktemp -d "$cache_root/.staging.XXXXXX" 2>/dev/null)"; then
        if cp -a "$compiled" "$staging/compiled" 2>/dev/null &&
           : >"$staging/.complete" 2>/dev/null &&
           mv -T "$staging" "$cache_entry" 2>/dev/null; then
            compiled="$cache_entry/compiled"
            find "$cache_root" -regextype posix-extended -mindepth 1 -maxdepth 1 \
                -type d -regex '.*/[0-9a-f]{64}' -mtime +14 \
                -exec rm -rf -- '{}' + 2>/dev/null || true
        else
            rm -rf -- "${staging:?}"
        fi
    fi
else
    touch -c "$cache_entry" 2>/dev/null || true
    if [[ -n "$native_compiler_bin" && -z "$compiled_override" ]]; then
        echo "beagle build: core-compiler-cache HIT $compiler_source_digest" >&2
    fi
fi

compiled_digest="$(
    cd "$compiled"
    find . -type f -print0 | LC_ALL=C sort -z | xargs -0 sha256sum |
        sha256sum | awk '{print $1}'
)"

if [[ "${BEAGLE_CORE_PROJECTION_ONLY:-0}" == 1 ]]; then
    projection_out="${BEAGLE_CORE_PROJECTION_ONLY_OUT:-}"
    [[ -n "$projection_out" ]] ||
        die "BEAGLE_CORE_PROJECTION_ONLY_OUT is required for projection-only builds"
    [[ "$projection_out" != "/" ]] ||
        die "BEAGLE_CORE_PROJECTION_ONLY_OUT may not be the filesystem root"
    mkdir -p "$projection_out"
    cp -a "$compiled/." "$projection_out/"
    if [[ -n "${BEAGLE_CORE_PROJECTION_ONLY_LOG:-}" ]]; then
        cp -- "$work/build.log" "$BEAGLE_CORE_PROJECTION_ONLY_LOG"
    fi
    exit 0
fi

# Seam 9 authenticates and caches the compiler-module projection above.  Once
# that contract is established, the full native compiler owns Core lowering,
# obligations, parallel emission, materializer reports, and atomic publication.
# The public arguments are identical, so the handoff introduces no second CLI
# surface.  This phase is separately supervised and timed; the outer build
# deadline remains containment, never a substitute for this deadline.
if [[ -n "$native_compiler_bin" ]]; then
    native_phase_start_ns="$(date +%s%N)"
    set +e
    run_phase core-lowering "${BEAGLE_CORE_LOWERING_TIMEOUT_SECONDS:-180}" \
        env -u BEAGLE_NATIVE_COMPILER_BIN \
        "$native_compiler_bin" "${original_args[@]}"
    native_runner_rc=$?
    set -e
    native_phase_end_ns="$(date +%s%N)"
    native_phase_seconds=$((
        (native_phase_end_ns - native_phase_start_ns + 999999999) / 1000000000
    ))
    echo "beagle build: phase core-lowering SECONDS $native_phase_seconds" >&2
    [[ "$native_runner_rc" == 0 ]] || exit "$native_runner_rc"
    rm -rf "${work:?}"
    work_cleaned=1
    build_committed=1
    exit 0
fi

write_core_result_input_manifest() {
    local destination="$1"
    {
        printf 'beagle-core-result-input/v1\n'
        write_identity_compiler_lines
        printf 'compiler-projection-sha256 %s\n' "$compiled_digest"
        printf 'compiler-commit %s\n' "$compiler_commit"
        printf 'source-facts-sha256 %s\n' \
            "$(sha256sum "$facts" | awk '{print $1}')"
        write_identity_selection_lines
    } >"$destination"
}

validate_core_result_input_closure() {
    local expected_input="$1" current_input="$2" current_digest
    validate_identity_closure_prefix || return 1
    current_digest="$(
        cd "$compiled"
        find . -type f -print0 | LC_ALL=C sort -z | xargs -0 sha256sum |
            sha256sum | awk '{print $1}'
    )"
    [[ "$current_digest" == "$compiled_digest" ]] || return 1
    write_core_result_input_manifest "$current_input"
    cmp -s "$expected_input" "$current_input"
}

# A content stamp over an entire staged directory: every file's path and bytes,
# in one deterministic order, folded to one digest. Cheap enough to take twice
# around a lock that a full re-verification cannot be.
staged_tree_stamp() {
    local root="$1"
    (
        cd "$root"
        find . -mindepth 1 -type f -print0 |
            LC_ALL=C sort -z |
            xargs -0 -r sha256sum
    ) | sha256sum | awk '{print $1}'
}

write_core_result_artifact_manifest() {
    local artifacts_dir="$1" destination="$2"
    (
        cd "$artifacts_dir"
        find . -mindepth 1 -maxdepth 1 -type f -printf '%P\0' |
            LC_ALL=C sort -z |
            xargs -0 -r sha256sum
    ) >"$destination"
}

validate_core_result_entry() {
    local entry="$1" expected_input="$2" expected_key="$3"
    local artifact_manifest_digest expected_names actual_names required
    [[ -d "$entry" && ! -L "$entry" &&
       -f "$entry/input.manifest" && ! -L "$entry/input.manifest" &&
       -f "$entry/artifacts.sha256" && ! -L "$entry/artifacts.sha256" &&
       -f "$entry/READY" && ! -L "$entry/READY" &&
       -d "$entry/artifacts" && ! -L "$entry/artifacts" ]] || return 1
    cmp -s "$expected_input" "$entry/input.manifest" || return 1
    if find "$entry/artifacts" -mindepth 1 -maxdepth 1 ! -type f -print -quit |
        grep -q .; then
        return 1
    fi
    expected_names="$(awk '{ print $2 }' "$entry/artifacts.sha256" | LC_ALL=C sort)"
    actual_names="$(find "$entry/artifacts" -mindepth 1 -maxdepth 1 -type f \
        -printf '%f\n' | LC_ALL=C sort)"
    [[ -n "$expected_names" && "$expected_names" == "$actual_names" ]] || return 1
    (
        cd "$entry/artifacts"
        sha256sum --check --strict ../artifacts.sha256 >/dev/null 2>&1
    ) || return 1
    for required in source.facts report.txt module.native-program \
                    native.receipts native.entry-map; do
        [[ -f "$entry/artifacts/$required" ]] || return 1
    done
    if [[ -n "${seen_materializers[c17]:-}" ||
          -n "${seen_materializers[wasm]:-}" ]]; then
        for required in module_0.h module_0.c c17.receipt; do
            [[ -f "$entry/artifacts/$required" ]] || return 1
        done
    fi
    if [[ -n "${seen_materializers[qbe]:-}" ]]; then
        [[ -f "$entry/artifacts/module_0.ssa" ]] || return 1
    fi
    if [[ "$simd_required" == 1 ]]; then
        [[ -f "$entry/artifacts/module.simd-plan-v0" ]] || return 1
    fi
    artifact_manifest_digest="$(sha256sum "$entry/artifacts.sha256" | awk '{print $1}')"
    [[ "$(<"$entry/READY")" == \
       "beagle-core-result/v1 $expected_key $artifact_manifest_digest" ]]
}

validate_recorded_core_result_pass() {
    local entry="$1" key artifact_manifest_digest
    local expected_names actual_names expected_root_names actual_root_names
    [[ -d "$entry" && ! -L "$entry" ]] || return 1
    key="$(basename "$entry")"
    [[ "$key" =~ ^[0-9a-f]{64}$ &&
       -f "$entry/input.manifest" && ! -L "$entry/input.manifest" &&
       "$(sha256sum "$entry/input.manifest" | awk '{print $1}')" == "$key" &&
       -f "$entry/artifacts.sha256" && ! -L "$entry/artifacts.sha256" &&
       -f "$entry/READY" && ! -L "$entry/READY" &&
       -d "$entry/artifacts" && ! -L "$entry/artifacts" ]] || return 1
    expected_root_names="$(printf '%s\n' READY artifacts artifacts.sha256 input.manifest |
        LC_ALL=C sort)"
    actual_root_names="$(find "$entry" -mindepth 1 -maxdepth 1 -printf '%f\n' |
        LC_ALL=C sort)"
    [[ "$expected_root_names" == "$actual_root_names" ]] || return 1
    if find "$entry/artifacts" -mindepth 1 -maxdepth 1 ! -type f -print -quit |
        grep -q .; then
        return 1
    fi
    expected_names="$(awk '{ print $2 }' "$entry/artifacts.sha256" | LC_ALL=C sort)"
    actual_names="$(find "$entry/artifacts" -mindepth 1 -maxdepth 1 -type f \
        -printf '%f\n' | LC_ALL=C sort)"
    [[ -n "$expected_names" && "$expected_names" == "$actual_names" ]] || return 1
    (cd "$entry/artifacts" &&
        sha256sum --check --strict ../artifacts.sha256 >/dev/null 2>&1) || return 1
    artifact_manifest_digest="$(sha256sum "$entry/artifacts.sha256" |
        awk '{print $1}')"
    [[ "$(<"$entry/READY")" == \
       "beagle-core-result/v1 $key $artifact_manifest_digest" ]]
}

# A C17 attestation hands back a RECEIPT and the emitted C17 itself, and the
# build finalizer requires that receipt's compiler commit to equal the commit
# this build stamps into its own native receipts (validate-c17!,
# native-core/validation/build-finalize.clj). The commit is therefore part of
# what a C17 PASS is, not incidental to it, and it belongs in the attestation's
# identity: a receipt minted at another commit is not slower to reuse, it is
# impossible to reuse.
#
# The compiler tree is keyed here too, and must be. The native program digest
# names the program FROZEN BEFORE ANY MATERIALIZER RUNS, so it does not move
# when a C17 materializer stage changes -- native/body_c17, native/fold_c17 and
# native/c11 all emit downstream of that freeze. The commit cannot stand in for
# the tree either, because the result cache no longer requires a clean checkout.
# Without compiler-source-sha256 an edited C17 emitter would hand this build the
# previous emitter's module_0.c under an unchanged key.
core_c17_attestation_key() {
    local native_digest="$1"
    printf 'beagle-core-c17-identity/v3\nnative-program-sha256 %s\ncompiler-commit %s\ncompiler-source-sha256 %s\n' \
        "$native_digest" "$compiler_commit" "$compiler_source_digest" |
        sha256sum | awk '{print $1}'
}

use_core_c17_attestation_source() {
    local source_entry="$1" native_digest="$2" facts_digest="$3"
    local entry_map_digest="$4" report receipt header_digest source_digest
    [[ "$source_entry" != "$core_result_entry" ]] || return 1
    validate_recorded_core_result_pass "$source_entry" || return 1
    # The keyed lookup below cannot reach a foreign commit or a foreign compiler
    # tree, but the linear scan can, so both are proven here too. The manifest is
    # bound to the entry's key by validate_recorded_core_result_pass.
    grep -Fqx "compiler-commit $compiler_commit" \
        "$source_entry/input.manifest" || return 1
    grep -Fqx "compiler-source-sha256 $compiler_source_digest" \
        "$source_entry/input.manifest" || return 1
    report="$source_entry/artifacts/report.txt"
    receipt="$source_entry/artifacts/c17.receipt"
    [[ -f "$report" && -f "$receipt" &&
       -f "$source_entry/artifacts/module.native-program" &&
       -f "$source_entry/artifacts/module_0.h" &&
       -f "$source_entry/artifacts/module_0.c" &&
       "$native_digest" == "$(sha256sum \
           "$source_entry/artifacts/module.native-program" | awk '{print $1}')" ]] ||
        return 1
    cmp -s "$work/artifacts/native.entry-map" \
        "$source_entry/artifacts/native.entry-map" || return 1
    header_digest="$(sha256sum "$source_entry/artifacts/module_0.h" |
        awk '{print $1}')"
    source_digest="$(sha256sum "$source_entry/artifacts/module_0.c" |
        awk '{print $1}')"
    [[ "$(grep -Fxc "native-provenance-v0 epoch sha256:$native_digest" \
            "$report" || true)" == 1 &&
       "$(grep -Fxc 'materialize-c17 OK module_0.h module_0.c' \
            "$report" || true)" == 1 &&
       "$(grep -Fxc "materialize-c17-artifact module_0.h sha256:$header_digest" \
            "$report" || true)" == 1 &&
       "$(grep -Fxc "materialize-c17-artifact module_0.c sha256:$source_digest" \
            "$report" || true)" == 1 &&
       "$(grep -Fxc "materialize-c17-artifact native.entry-map sha256:$entry_map_digest" \
            "$report" || true)" == 1 &&
       "$(grep -Fxc 'result PASS' "$report" || true)" == 1 &&
       "$(grep -c '^obligation-projection PASS ' "$report" || true)" == 10 ]] ||
        return 1
    ! grep -q '^pending ' "$report" || return 1
    # The receipt is the artifact the finalizer validates against this build's
    # commit; it encodes that commit as one canonical 52-byte string field.
    grep -Fq "52:6:string:40:$compiler_commit" "$receipt" || return 1
    grep -Fq "sha256:$native_digest" "$receipt" || return 1
    grep -Fq "source-facts-sha256=$facts_digest" "$receipt" || return 1
    grep -Fq 'abi=lp64' "$receipt" || return 1
    grep -Fq 'profile=3' "$receipt" || return 1
    grep -Fq 'native-to-c17' "$receipt" || return 1
    grep -Fq 'restricted-c17-v0' "$receipt" || return 1
    grep -Fq "sha256:$header_digest" "$receipt" || return 1
    grep -Fq "sha256:$source_digest" "$receipt" || return 1
    grep -Fq "sha256:$entry_map_digest" "$receipt" || return 1
    cp -- "$source_entry/artifacts/module_0.h" "$work/artifacts/module_0.h"
    cp -- "$source_entry/artifacts/module_0.c" "$work/artifacts/module_0.c"
    cp -- "$receipt" "$work/artifacts/c17.receipt"
    grep '^materialize-c17' "$report" >"$work/core-c17-attestation.report"
    core_c17_attestation_report="$work/core-c17-attestation.report"
}

record_core_c17_attestation() {
    local source_entry="$1" native_digest="$2" source_key ready_digest
    local attestation staging
    source_key="$(basename "$source_entry")"
    ready_digest="$(sha256sum "$source_entry/READY" | awk '{print $1}')"
    mkdir -p "$core_c17_attestation_root" "$cache_root/.result-tmp"
    attestation="$core_c17_attestation_root/$(core_c17_attestation_key "$native_digest")"
    staging="$(mktemp "$cache_root/.result-tmp/$native_digest.c17.XXXXXX")"
    printf 'beagle-core-c17-pass/v2 %s %s %s %s\n' \
        "$native_digest" "$compiler_commit" "$source_key" "$ready_digest" >"$staging"
    sync -f "$staging"
    [[ ! -e "$attestation" && ! -L "$attestation" ]] || rm -f -- "$attestation"
    ln "$staging" "$attestation"
    rm -f -- "$staging"
    sync -f "$core_c17_attestation_root"
}

reuse_core_c17_attestation() {
    local native_digest facts_digest entry_map_digest attestation
    local format recorded_digest recorded_commit source_key ready_digest extra
    local source_entry
    [[ "$core_checkpoint_cache_hit" == 1 &&
       -n "${seen_materializers[c17]:-}" &&
       "$abi" == "lp64" && "$simd_required" == 0 ]] || return 1
    native_digest="$(sha256sum "$work/artifacts/module.native-program" |
        awk '{print $1}')"
    facts_digest="$(sha256sum "$work/artifacts/source.facts" | awk '{print $1}')"
    entry_map_digest="$(sha256sum "$work/artifacts/native.entry-map" |
        awk '{print $1}')"
    attestation="$core_c17_attestation_root/$(core_c17_attestation_key "$native_digest")"
    if [[ -f "$attestation" && ! -L "$attestation" ]]; then
        read -r format recorded_digest recorded_commit source_key ready_digest \
            extra <"$attestation" || true
        source_entry="$core_result_root/$source_key"
        if [[ "$format" == "beagle-core-c17-pass/v2" &&
              "$recorded_digest" == "$native_digest" &&
              "$recorded_commit" == "$compiler_commit" &&
              "$source_key" =~ ^[0-9a-f]{64}$ &&
              "$ready_digest" =~ ^[0-9a-f]{64}$ &&
              -z "${extra:-}" &&
              -f "$source_entry/READY" &&
              "$(sha256sum "$source_entry/READY" | awk '{print $1}')" == \
                  "$ready_digest" ]] &&
           use_core_c17_attestation_source "$source_entry" "$native_digest" \
               "$facts_digest" "$entry_map_digest"; then
            echo "beagle build: core-c17-attestation HIT $native_digest -> $source_key" >&2
            return 0
        fi
        rm -f -- "$attestation"
    fi
    while IFS= read -r -d '' source_entry; do
        if use_core_c17_attestation_source "$source_entry" "$native_digest" \
            "$facts_digest" "$entry_map_digest"; then
            record_core_c17_attestation "$source_entry" "$native_digest"
            source_key="$(basename "$source_entry")"
            echo "beagle build: core-c17-attestation HIT $native_digest -> $source_key" >&2
            return 0
        fi
    done < <(find "$core_result_root" -mindepth 1 -maxdepth 1 \
        -type d -print0 2>/dev/null)
    echo "beagle build: core-c17-attestation MISS $native_digest" >&2
    return 1
}

validate_core_checkpoint_entry() {
    local entry="$1" expected_input="$2" expected_key="$3"
    cmp -s "$expected_input" "$entry/input.manifest" || return 1
    validate_core_checkpoint_entry_from_alias "$entry" "$expected_key" \
        "$(sha256sum "$expected_input" | awk '{print $1}')"
}

validate_core_checkpoint_wire() {
    local entry="$1" log="$2" index path rc
    local -a labels=(native qbe)
    local -a paths=(
        "$entry/artifacts/frozen-native-stage.wire-v1"
        "$entry/artifacts/qbe-frozen-native-stage.wire-v1"
    )
    : >"$log"
    for index in "${!paths[@]}"; do
        path="${paths[$index]}"
        rc=0
        run_phase "core-checkpoint-wire-${labels[$index]}" \
            "${BEAGLE_CORE_VALIDATION_TIMEOUT_SECONDS:-${BEAGLE_CORE_LOWERING_TIMEOUT_SECONDS:-180}}" \
            bb -cp "$compiled" -e '
              (require (quote [native.stages :as stages]))
              (let [path (first *command-line-args*)
                    encoding (slurp path)
                    frozen (stages/decode-frozen-native-stage-wire-v1 encoding)]
                (when (nil? frozen)
                  (System/exit 1)))' \
            "$path" >>"$log" 2>&1 || rc=$?
        if [[ "$rc" -ne 0 ]]; then
            sed -n '1,120p' "$log" >&2
            return "$rc"
        fi
    done
    sed -n '1,120p' "$log" >&2
}

validate_core_checkpoint_wire_attestation() {
    local attestation="$1" expected_wire_digest="$2"
    local format wire_digest qbe_wire_digest source_key source_ready_digest extra
    local source_entry source_input_digest
    [[ -f "$attestation" && ! -L "$attestation" ]] || return 1
    read -r format wire_digest qbe_wire_digest source_key source_ready_digest extra \
        <"$attestation" || return 1
    [[ "$format" == "beagle-core-checkpoint-wire-pass/v2" &&
       "$wire_digest" == "$expected_wire_digest" &&
       "$qbe_wire_digest" =~ ^[0-9a-f]{64}$ &&
       "$source_key" =~ ^[0-9a-f]{64}$ &&
       "$source_ready_digest" =~ ^[0-9a-f]{64}$ &&
       -z "${extra:-}" ]] || return 1
    source_entry="$core_checkpoint_root/$source_key"
    [[ -f "$source_entry/READY" && ! -L "$source_entry/READY" &&
       "$(sha256sum "$source_entry/READY" | awk '{print $1}')" == \
           "$source_ready_digest" ]] || return 1
    source_input_digest="$(sha256sum "$source_entry/input.manifest" 2>/dev/null |
        awk '{print $1}')"
    [[ "$source_input_digest" == "$source_key" ]] || return 1
    validate_core_checkpoint_entry_from_alias "$source_entry" "$source_key" \
        "$source_input_digest" || return 1
    grep -Fqx "$wire_digest  frozen-native-stage.wire-v1" \
        "$source_entry/artifacts.sha256" &&
        grep -Fqx "$qbe_wire_digest  qbe-frozen-native-stage.wire-v1" \
            "$source_entry/artifacts.sha256"
}

record_core_checkpoint_wire_attestation() {
    local source_entry="$1" wire_digest qbe_wire_digest source_key source_input_digest
    local source_ready_digest attestation staging
    source_key="$(basename "$source_entry")"
    source_input_digest="$(sha256sum "$source_entry/input.manifest" | awk '{print $1}')"
    [[ "$source_key" == "$source_input_digest" ]] || return 1
    validate_core_checkpoint_entry_from_alias "$source_entry" "$source_key" \
        "$source_input_digest" || return 1
    wire_digest="$(sha256sum \
        "$source_entry/artifacts/frozen-native-stage.wire-v1" | awk '{print $1}')"
    qbe_wire_digest="$(sha256sum \
        "$source_entry/artifacts/qbe-frozen-native-stage.wire-v1" | awk '{print $1}')"
    source_ready_digest="$(sha256sum "$source_entry/READY" | awk '{print $1}')"
    mkdir -p "$core_checkpoint_wire_attestation_root" \
        "$cache_root/.checkpoint-tmp"
    attestation="$core_checkpoint_wire_attestation_root/$wire_digest"
    if ! validate_core_checkpoint_wire_attestation \
        "$attestation" "$wire_digest"; then
        [[ ! -e "$attestation" && ! -L "$attestation" ]] ||
            rm -f -- "$attestation"
    fi
    staging="$(mktemp \
        "$cache_root/.checkpoint-tmp/$wire_digest.attestation.XXXXXX")"
    printf 'beagle-core-checkpoint-wire-pass/v2 %s %s %s %s\n' \
        "$wire_digest" "$qbe_wire_digest" "$source_key" \
        "$source_ready_digest" >"$staging"
    sync -f "$staging"
    if [[ ! -e "$attestation" ]]; then
        ln "$staging" "$attestation" 2>/dev/null || true
        sync -f "$core_checkpoint_wire_attestation_root"
    fi
    rm -f -- "$staging"
    validate_core_checkpoint_wire_attestation "$attestation" "$wire_digest"
}

reuse_core_checkpoint_wire_attestation() {
    local entry="$1" wire_digest qbe_wire_digest attestation source_entry source_key
    local source_input_digest
    wire_digest="$(sha256sum \
        "$entry/artifacts/frozen-native-stage.wire-v1" | awk '{print $1}')"
    qbe_wire_digest="$(sha256sum \
        "$entry/artifacts/qbe-frozen-native-stage.wire-v1" | awk '{print $1}')"
    attestation="$core_checkpoint_wire_attestation_root/$wire_digest"
    if validate_core_checkpoint_wire_attestation "$attestation" "$wire_digest"; then
        read -r _ _ _ source_key _ <"$attestation"
        echo "beagle build: core-checkpoint-wire-attestation HIT $wire_digest -> $source_key" >&2
        return 0
    fi
    [[ ! -e "$attestation" && ! -L "$attestation" ]] ||
        rm -f -- "$attestation"

    while IFS= read -r -d '' source_entry; do
        [[ "$source_entry" != "$entry" ]] || continue
        source_key="$(basename "$source_entry")"
        [[ "$source_key" =~ ^[0-9a-f]{64}$ ]] || continue
        source_input_digest="$(sha256sum \
            "$source_entry/input.manifest" 2>/dev/null | awk '{print $1}')"
        [[ "$source_input_digest" == "$source_key" ]] || continue
        validate_core_checkpoint_entry_from_alias "$source_entry" "$source_key" \
            "$source_input_digest" || continue
        grep -Fqx "$wire_digest  frozen-native-stage.wire-v1" \
            "$source_entry/artifacts.sha256" || continue
        grep -Fqx "$qbe_wire_digest  qbe-frozen-native-stage.wire-v1" \
            "$source_entry/artifacts.sha256" || continue
        record_core_checkpoint_wire_attestation "$source_entry" || continue
        echo "beagle build: core-checkpoint-wire-attestation HIT $wire_digest -> $source_key" >&2
        return 0
    done < <(find "$core_checkpoint_root" -mindepth 1 -maxdepth 1 \
        -type d -print0 2>/dev/null)
    return 1
}

retire_core_checkpoint_alias() {
    local expected_key="$1" alias_format alias_early alias_full alias_digest alias_extra
    exec {core_checkpoint_retire_alias_lock_fd}>\
        "$cache_root/.checkpoint-alias-locks/$core_checkpoint_early_key.lock"
    run_phase_leaf core-checkpoint-alias-retire-lock \
        "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        flock -x -w "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        "$core_checkpoint_retire_alias_lock_fd" ||
        die "timed out acquiring the Core checkpoint alias retirement lock"
    if [[ -e "$core_checkpoint_alias" ]]; then
        alias_format=""
        alias_early=""
        alias_full=""
        alias_digest=""
        alias_extra=""
        if [[ -f "$core_checkpoint_alias" && ! -L "$core_checkpoint_alias" ]]; then
            read -r alias_format alias_early alias_full alias_digest alias_extra \
                <"$core_checkpoint_alias" || true
        fi
        if [[ "$alias_format" != "beagle-core-pre-materializer-alias/v1" ||
              "$alias_early" != "$core_checkpoint_early_key" ||
              ! "$alias_digest" =~ ^[0-9a-f]{64}$ ||
              -n "${alias_extra:-}" ||
              "$alias_full" == "$expected_key" ]]; then
            corrupt_checkpoint_alias=\
"$cache_root/.checkpoint-tmp/retired-alias.$core_checkpoint_early_key.$$"
            mv "$core_checkpoint_alias" "$corrupt_checkpoint_alias"
            rm -rf -- "${corrupt_checkpoint_alias:?}"
        fi
    fi
    flock -u "$core_checkpoint_retire_alias_lock_fd"
}

core_result_input="$work/core-result-input.manifest"
core_result_cache_hit=0
core_result_publish_pending=0
core_checkpoint_cache_hit=0
if [[ "$core_result_early_hit" == 1 ]]; then
    cp -- "$alias_entry/input.manifest" "$core_result_input"
    core_result_key="$alias_full_key"
    core_result_entry="$alias_entry"
    core_result_cache_hit=1
    echo "beagle build: core-result-cache HIT $core_result_key" >&2
elif [[ "$core_checkpoint_early_hit" == 1 ]]; then
    write_core_result_input_manifest "$core_result_input"
    core_result_key="$(sha256sum "$core_result_input" | awk '{print $1}')"
    core_result_entry="$core_result_root/$core_result_key"
else
    write_core_result_input_manifest "$core_result_input"
    core_result_key="$(sha256sum "$core_result_input" | awk '{print $1}')"
    core_result_entry="$core_result_root/$core_result_key"
fi
core_checkpoint_input="$work/core-checkpoint-input.manifest"
if [[ "$core_checkpoint_early_hit" == 1 ]]; then
    # The checkpoint carries post-stage source facts, not the projected facts
    # whose digest sealed its input manifest. Preserve that validated identity.
    cp -- "$checkpoint_alias_entry/input.manifest" "$core_checkpoint_input"
    [[ "$(grep -c '^compiler-projection-sha256 ' "$core_checkpoint_input" || true)" == 1 &&
       "$(grep '^compiler-projection-sha256 ' "$core_checkpoint_input")" == \
           "compiler-projection-sha256 $compiled_digest" ]] ||
        die "Core checkpoint compiler projection disagrees with the current compiler"
    core_checkpoint_key="$checkpoint_alias_full_key"
    core_checkpoint_entry="$checkpoint_alias_entry"
else
    write_core_checkpoint_identity "$core_result_input" "$core_checkpoint_input"
    core_checkpoint_key="$(sha256sum "$core_checkpoint_input" | awk '{print $1}')"
    core_checkpoint_entry="$core_checkpoint_root/$core_checkpoint_key"
fi
if [[ "$core_result_cache_enabled" == 1 &&
      "$core_result_early_hit" != 1 ]]; then
    mkdir -p "$core_result_root" "$core_checkpoint_root" \
        "$cache_root/.result-locks" "$cache_root/.result-tmp" \
        "$cache_root/.checkpoint-tmp"
    exec {core_result_lock_fd}>"$cache_root/.result-locks/$core_result_key.lock"
    acquire_core_result_lock core-result-cache-lock
    if [[ "$core_checkpoint_early_hit" == 1 ]]; then
        validate_core_result_early_closure "$core_result_early_input" \
            "$work/core-checkpoint-early-input.before-result-lookup" ||
            die "Core checkpoint early input closure changed before result lookup"
    else
        validate_core_result_input_closure "$core_result_input" \
            "$work/core-result-input.lookup" ||
            die "Core result input closure changed before cache lookup"
    fi
    if [[ -e "$core_result_entry" ]]; then
        if validate_core_result_entry "$core_result_entry" "$core_result_input" \
            "$core_result_key"; then
            cp -a "$core_result_entry/artifacts/." "$work/artifacts/"
            core_result_cache_hit=1
            echo "beagle build: core-result-cache HIT $core_result_key" >&2
        else
            corrupt_entry="$cache_root/.result-tmp/corrupt.$core_result_key.$$"
            echo "beagle build: core-result-cache CORRUPT $core_result_key; retiring" >&2
            mv "$core_result_entry" "$corrupt_entry"
            rm -rf -- "${corrupt_entry:?}"
        fi
    fi
    if [[ "$core_result_cache_hit" != 1 ]]; then
        echo "beagle build: core-result-cache MISS $core_result_key" >&2
    fi
elif [[ "$core_result_early_hit" == 1 ]]; then
    # The alias already served this result; the keyed lookup is redundant, not
    # bypassed. Saying BYPASS here would report the disabled-cache reason on a
    # run whose cache demonstrably hit.
    echo "beagle build: core-result-cache SERVED-BY-ALIAS $core_result_key" >&2
else
    echo "beagle build: core-result-cache BYPASS $core_result_cache_bypass_reason" >&2
fi

if [[ "$core_result_cache_hit" != 1 && "$core_result_cache_enabled" == 1 &&
      "$core_checkpoint_eligible" == 1 ]]; then
    core_checkpoint_structural_ok=0
    core_checkpoint_semantic_ok=0
    if [[ -e "$core_checkpoint_entry" ]]; then
        if validate_core_checkpoint_entry "$core_checkpoint_entry" \
            "$core_checkpoint_input" "$core_checkpoint_key"; then
            core_checkpoint_structural_ok=1
            if reuse_core_checkpoint_wire_attestation "$core_checkpoint_entry" ||
               validate_core_checkpoint_wire "$core_checkpoint_entry" \
                   "$work/core-checkpoint-wire.log"; then
                core_checkpoint_semantic_ok=1
                record_core_checkpoint_wire_attestation \
                    "$core_checkpoint_entry" ||
                    die "failed to record the Core checkpoint wire PASS"
            fi
        fi
        if [[ "$core_checkpoint_structural_ok" != 1 ||
              "$core_checkpoint_semantic_ok" != 1 ]]; then
            echo "beagle build: core-checkpoint CORRUPT $core_checkpoint_key; retiring" >&2
            corrupt_checkpoint=\
"$cache_root/.checkpoint-tmp/corrupt.$core_checkpoint_key.$$"
            mv "$core_checkpoint_entry" "$corrupt_checkpoint"
            rm -rf -- "${corrupt_checkpoint:?}"
            retire_core_checkpoint_alias "$core_checkpoint_key"
            if [[ "$core_checkpoint_early_hit" == 1 &&
                  "$core_checkpoint_structural_ok" == 1 ]]; then
                die "Core checkpoint wire was not a canonical resumable frozen stage; rerun after retirement"
            fi
        fi
    fi
    if [[ "$core_checkpoint_semantic_ok" == 1 ]]; then
        mkdir "$work/core-checkpoint-hit"
        cp -- "$core_checkpoint_entry/artifacts/frozen-native-stage.wire-v1" \
            "$work/core-checkpoint-hit/frozen-native-stage.wire-v1"
        cp -- "$core_checkpoint_entry/artifacts/qbe-frozen-native-stage.wire-v1" \
            "$work/core-checkpoint-hit/qbe-frozen-native-stage.wire-v1"
        cp -- "$core_checkpoint_entry/artifacts/report.head" \
            "$work/core-checkpoint-hit/report.head"
        cp -- "$core_checkpoint_entry/artifacts/stage-progress" \
            "$work/core-checkpoint-hit/stage-progress"
        for checkpoint_artifact in source.facts module.native-program \
                                   native.receipts native.entry-map; do
            cp -- "$core_checkpoint_entry/artifacts/$checkpoint_artifact" \
                "$work/artifacts/$checkpoint_artifact"
        done
        if [[ "$simd_required" == 1 ]]; then
            cp -- "$core_checkpoint_entry/artifacts/module.simd-plan-v0" \
                "$work/artifacts/module.simd-plan-v0"
        fi
        core_checkpoint_cache_hit=1
        echo "beagle build: core-checkpoint HIT $core_checkpoint_key" >&2
    else
        echo "beagle build: core-checkpoint MISS $core_checkpoint_key" >&2
    fi
elif [[ "$core_result_cache_hit" != 1 ]]; then
    if [[ "$core_result_cache_enabled" != 1 ]]; then
        echo "beagle build: core-checkpoint BYPASS $core_result_cache_bypass_reason" >&2
    else
        echo "beagle build: core-checkpoint BYPASS checkpoint-ineligible" >&2
    fi
fi

# End of the lookup critical section: every read, every corruption retirement,
# and every alias retirement the lock exists to serialize has happened. What
# follows is compilation, which is exactly what a sibling must not be made to
# wait behind. Each publication takes the lock again for its own rename.
release_core_result_lock

if [[ "$core_result_cache_hit" != 1 ]]; then
core_c17_attestation_report=""
reuse_core_c17_attestation || true
export BEAGLE_CORE_FACTS_MANIFEST="$facts"
export BEAGLE_CORE_ARTIFACTS="$work/artifacts"
export BEAGLE_CORE_COMMIT="$compiler_commit"
export BEAGLE_CORE_ABI="$abi"
export BEAGLE_CORE_SIMD_REQUIRED="$simd_required"
export BEAGLE_CORE_EMIT_WORKERS="$emit_workers"
export BEAGLE_CORE_EMIT_WORKERS_EXPLICIT="$emit_workers_explicit"
export BEAGLE_CORE_EMIT_MANAGER="$EMIT_MANAGER"
export BEAGLE_CORE_EMIT_COMPILED="$compiled"
export BEAGLE_CORE_EMIT_SUPERVISOR="$NATIVE_SUPERVISOR"
export BEAGLE_CORE_EMIT_BB="$(command -v bb)"
export BEAGLE_CORE_EMIT_WORKER_TIMEOUT_SECONDS=\
"${BEAGLE_CORE_EMIT_WORKER_TIMEOUT_SECONDS:-${BEAGLE_CORE_LOWERING_TIMEOUT_SECONDS:-180}}"
export BEAGLE_CORE_EMIT_KILL_GRACE_SECONDS="$core_kill_grace"
export BEAGLE_CORE_EMIT_WORK_ROOT="$work/emission-workers"
mkdir -p "$BEAGLE_CORE_EMIT_WORK_ROOT"
export BEAGLE_CORE_REPORT="$work/progress.txt"
export BEAGLE_CORE_C17_ATTESTATION_REPORT="$core_c17_attestation_report"
printf '%s\n' "${entries[@]}" >"$work/entries"
export BEAGLE_CORE_ENTRIES="$work/entries"
printf '%s\n' "${materializers[@]}" >"$work/materializers"
export BEAGLE_CORE_MATERIALIZERS="$work/materializers"
export BEAGLE_CORE_PROJECTED_FACTS="$work/artifacts/source.facts"
export BEAGLE_CORE_CHECKPOINT_WIRE=""
export BEAGLE_CORE_CHECKPOINT_QBE_WIRE=""
export BEAGLE_CORE_CHECKPOINT_REPORT_HEAD=""
export BEAGLE_CORE_CHECKPOINT_STAGE_PROGRESS=""
export BEAGLE_CORE_CHECKPOINT_STAGE=""
export BEAGLE_CORE_DEV_FACT_REUSE="$dev_fact_reuse"
export BEAGLE_CORE_DEV_FACT_STORE="$dev_fact_store"
export BEAGLE_CORE_REPO_ROOT="$BEAGLE_DIR"
export BEAGLE_CORE_COMPILER_SOURCE_DIGEST="$compiler_source_digest"
export BEAGLE_CORE_UNIT_RULE_EPOCH=""
if [[ "$dev_fact_reuse" == 1 ]]; then
    BEAGLE_CORE_UNIT_RULE_EPOCH="$(
        "$BEAGLE_DIR/bin/_beagle-dev-unit-rule-identity" \
            --compiled "$compiled" \
            --driver "$BEAGLE_DIR/bin/beagle-build-core" \
            --store-adapter "$BEAGLE_DIR/store/out/store/dev_compile_facts.clj" \
            --profile "profile=3" \
            --abi "$abi"
    )"
    [[ "$BEAGLE_CORE_UNIT_RULE_EPOCH" =~ ^sha256:[0-9a-f]{64}$ ]] ||
        die "development unit rule identity is malformed"
fi
export BEAGLE_CORE_COMPILER_PROJECTION_ID="sha256:$projection_cache_key"
export BEAGLE_CORE_SEMANTIC_READ_RULES_ID="sha256:$compiler_source_digest"
export BEAGLE_CORE_SEMANTIC_READ_STORE=\
"$cache_root/semantic-read-$semantic_read_store_cohort.storelog"
if [[ "$core_checkpoint_cache_hit" == 1 ]]; then
    export BEAGLE_CORE_CHECKPOINT_WIRE=\
"$work/core-checkpoint-hit/frozen-native-stage.wire-v1"
    export BEAGLE_CORE_CHECKPOINT_QBE_WIRE=\
"$work/core-checkpoint-hit/qbe-frozen-native-stage.wire-v1"
    export BEAGLE_CORE_CHECKPOINT_REPORT_HEAD="$work/core-checkpoint-hit/report.head"
    export BEAGLE_CORE_CHECKPOINT_STAGE_PROGRESS=\
"$work/core-checkpoint-hit/stage-progress"
elif [[ "$core_result_cache_enabled" == 1 &&
        "$core_checkpoint_eligible" == 1 ]]; then
    mkdir "$work/core-checkpoint-stage"
    export BEAGLE_CORE_CHECKPOINT_STAGE="$work/core-checkpoint-stage"
fi

set +e
run_phase core-lowering "${BEAGLE_CORE_LOWERING_TIMEOUT_SECONDS:-180}" \
bb -cp "$compiled:$semantic_read_classpath" -e '
(require (quote [native.core :as core])
         (quote [native.stages :as stages])
         (quote [native.simd :as simd])
         (quote [native.lower :as lower])
         (quote [native.obligations :as obligations])
         (quote [native.slice :as slice])
         (quote [native.unit-reuse :as unit])
         (quote [native.unit-compile :as singleton])
         (quote [native.c11 :as c11])
         (quote [native.body-c17 :as body])
         (quote [native.body-slice :as body-slice])
         (quote [native.qbe :as qbe])
         (quote [semantic-read-store :as semantic-cache])
         (quote [clojure.edn :as edn])
         (quote [clojure.java.shell :as shell])
         (quote [clojure.string :as str]))

(import (quote [java.io File])
        (quote [java.lang ProcessBuilder ProcessBuilder$Redirect ProcessHandle])
        (quote [java.nio ByteBuffer])
        (quote [java.nio.channels FileChannel])
        (quote [java.nio.charset StandardCharsets])
        (quote [java.nio.file CopyOption Files OpenOption Paths
                 StandardCopyOption StandardOpenOption])
        (quote [java.util.concurrent Callable Executors TimeUnit]))

(def report-path (System/getenv "BEAGLE_CORE_REPORT"))
(def native-report-format "beagle-native-report/v1")
(def stage-progress (atom []))
(def stage-progress-prefix (atom ""))
(def active-stage (atom nil))
(def stage-work (atom nil))

(defn durable-write! [path text]
  (let [destination (Paths/get path (make-array String 0))
        directory (.getParent destination)
        temporary (.resolve directory
                            (str "." (.getFileName destination) "."
                                 (.pid (ProcessHandle/current)) "."
                                 (System/nanoTime) ".tmp"))]
    (try
      (let [buffer (ByteBuffer/wrap (.getBytes text StandardCharsets/UTF_8))]
        (with-open [channel (FileChannel/open temporary
                                              (into-array OpenOption
                                                          [StandardOpenOption/CREATE_NEW
                                                           StandardOpenOption/WRITE]))]
          (while (.hasRemaining buffer)
            (.write channel buffer))
          (.force channel true)))
      (try
        (Files/move temporary destination
                    (into-array CopyOption
                                [StandardCopyOption/ATOMIC_MOVE
                                 StandardCopyOption/REPLACE_EXISTING]))
        ;; AtomicMoveNotSupportedException is outside the interpreter class
        ;; allowlist, so the cross-filesystem fallback keys on its supertype.
        (catch java.io.IOException _
          (Files/move temporary destination
                      (into-array CopyOption
                                  [StandardCopyOption/REPLACE_EXISTING]))))
      (with-open [channel (FileChannel/open directory
                                            (into-array OpenOption
                                                        [StandardOpenOption/READ]))]
        (.force channel true))
      (finally
        (Files/deleteIfExists temporary)))))

(defn stage-progress-text []
  (str @stage-progress-prefix
       (apply str
              (for [{:keys [name status elapsed-ms]} @stage-progress]
                (str "stage-progress " name " " status "\n")))))

(defn stage-elapsed-ms [name elapsed-ms]
  (let [active @active-stage]
    (if (and (some? active) (= name (:name active)))
      (quot (- (System/nanoTime) (:started active)) 1000000)
      elapsed-ms)))

(defn progress-snapshot-text []
  (str @stage-progress-prefix
       (apply str
              (for [{:keys [name status elapsed-ms]} @stage-progress]
                (str "stage-progress " name " " status "\n"
                     "stage-elapsed " name " "
                     (stage-elapsed-ms name elapsed-ms) "\n")))
       (when-let [{:keys [stage work status completed total]} @stage-work]
         (str "stage-work " stage " " work " " status " "
              completed " " total "\n"))))

(defn publish-progress! []
  (durable-write! report-path
                  (str (progress-snapshot-text) "result RUNNING\n")))

(defn publish-stage-work! [stage work status completed total]
  (let [active @active-stage]
    (when-not (= stage (:name active))
      (throw (ex-info "Core build work does not match active stage"
                      {:active active :stage stage :work work})))
    (when (or (< completed 0) (< total completed))
      (throw (ex-info "Core build work count is invalid"
                      {:stage stage :work work
                       :completed completed :total total})))
    (reset! stage-work {:stage stage
                        :work work
                        :status status
                        :completed completed
                        :total total})
    (publish-progress!)))

(defn start-stage! [name]
  (when (some? @active-stage)
    (throw (ex-info "Core build stage overlap" {:active @active-stage
                                                 :next name})))
  (let [started (System/nanoTime)]
    (swap! stage-progress conj {:name name
                                :status "RUNNING"
                                :elapsed-ms 0})
    (reset! stage-work nil)
    (reset! active-stage {:name name :started started})
    (publish-progress!)))

(defn finish-stage! [name status]
  (let [active @active-stage]
    (when-not (= name (:name active))
      (throw (ex-info "Core build stage mismatch" {:active active
                                                    :finished name})))
    (let [elapsed (quot (- (System/nanoTime) (:started active)) 1000000)]
      (swap! stage-progress
             (fn [stages]
               (conj (vec (butlast stages))
                     (assoc (last stages)
                            :status status
                            :elapsed-ms elapsed))))
      (reset! active-stage nil)
      (publish-progress!)
      (binding [*out* *err*]
        (println (str "beagle build: core-stage " name " " status
                      " elapsed-ms=" elapsed))))))

(defn fail-active-stage! []
  (when-let [active @active-stage]
    (finish-stage! (:name active) "ERROR")))

(defn report-with-stage-progress [report]
  (let [lines (str/split-lines report)
        result (last lines)]
    (when-not (str/starts-with? result "result ")
      (throw (ex-info "Core build report omitted its final result" {:report report})))
    (str (str/join "\n" (butlast lines)) "\n"
         (stage-progress-text)
         result "\n")))

;; core-lowering-progress-fixture-begin
(def body-progress-cadence 32)

(defn observed-call [operation accepted]
  (let [result (operation)]
    (accepted)
    result))

(defn body-progress-boundary? [completed total]
  (or (= completed 1)
      (= completed total)
      (= 0 (mod completed body-progress-cadence))))

(defn observed-source-program [rows module-name relative-path]
  (publish-stage-work!
   "source-freeze" "source-reconstruction" "RUNNING" 0 1)
  (observed-call
   #(slice/source-program rows module-name relative-path)
   #(publish-stage-work!
     "source-freeze" "source-reconstruction" "ACCEPTED" 1 1)))

(defn observed-freeze-source-stage [source compiler-commit configuration]
  (let [original-encode stages/encode-source-stage
        original-valid? lower/source-stage-valid?]
    (publish-stage-work!
     "source-freeze" "source-encoding" "RUNNING" 0 1)
    (with-redefs
      [stages/encode-source-stage
       (fn [stage]
         (observed-call
          #(original-encode stage)
          #(publish-stage-work!
            "source-freeze" "source-encoding" "ACCEPTED" 1 1)))
       lower/source-stage-valid?
       (fn [stage]
         (observed-call
          #(original-valid? stage)
          #(publish-stage-work!
            "source-freeze" "source-validation" "ACCEPTED" 1 1)))]
      (lower/freeze-source-stage source compiler-commit configuration))))

(defn observed-lower-typed-stage [frozen-source compiler-commit configuration]
  (let [original-prelude lower/prepare-typing-prelude
        original-attach-bodies lower/attach-bodies
        original-attach-body lower/attach-body
        completed (atom 0)
        body-total (atom 0)]
    (publish-stage-work!
     "source-to-typed" "typing-prelude" "RUNNING" 0 1)
    (let [result
          (with-redefs
            [lower/prepare-typing-prelude
             (fn [source config]
               (observed-call
                #(original-prelude source config)
                #(publish-stage-work!
                  "source-to-typed" "typing-prelude" "ACCEPTED" 1 1)))
             lower/attach-bodies
             (fn [env resolutions sources]
               (let [total (min (count resolutions) (count sources))]
                 (reset! body-total total)
                 (publish-stage-work!
                  "source-to-typed" "function-bodies" "RUNNING" 0 total)
                 (observed-call
                  #(original-attach-bodies env resolutions sources)
                  #(publish-stage-work!
                    "source-to-typed" "typing-finalization" "RUNNING" 0 1))))
             lower/attach-body
             (fn [env resolution source]
               (observed-call
                #(original-attach-body env resolution source)
                #(let [done (swap! completed inc)
                       total @body-total]
                   (when (body-progress-boundary? done total)
                     (publish-stage-work!
                      "source-to-typed" "function-bodies" "RUNNING"
                      done total)))))]
            (lower/lower-typed-stage
             frozen-source compiler-commit configuration))]
      (publish-stage-work!
       "source-to-typed" "typing-finalization" "ACCEPTED" 1 1)
      result)))
;; core-lowering-progress-fixture-end

(defn id-value [id]
  (core/nativeid-value id))

(defn definition-nodes [index]
  (vec
   (mapcat #(lower/index-form-ids index %)
           ["record" "defunion" "deferror" "def" "defn" "defn-multi"
            "unsupported-defprotocol" "unsupported-extend-type"])))

(defn definition-child-items [index definition predicate]
  (if-let [items (lower/index-first-object index definition predicate)]
    (lower/index-sequence-items index items)
    []))

(defn definition-child-names [index definition predicate]
  (vec
   (remove str/blank?
           (map #(lower/fact-text index % "name")
                (definition-child-items index definition predicate)))))

(defn definition-provided-names [index definition]
  (let [kind (lower/fact-text index definition "form-kind")
        declared (lower/fact-text index definition "name")
        primary (if (str/blank? declared)
                  (lower/fact-text index definition "semantic-unit-name")
                  declared)
        own (if (str/blank? primary) [] [primary])]
    (vec
     (distinct
      (concat
       own
       (cond
         (= "record" kind) [(str "->" declared) (str "map->" declared)]
         (or (= "defunion" kind) (= "deferror" kind))
         (let [members (definition-child-names index definition "members")]
           (concat members (map #(str "->" %) members)))
         (= "unsupported-defprotocol" kind)
         (definition-child-names index definition "methods")
         :else []))))))

(defn definitions-by-key [index]
  (reduce
   (fn [definitions definition]
     (reduce
      (fn [indexed provided-name]
        (update indexed
                (str (lower/source-module-name index definition) "/" provided-name)
                (fnil conj [])
                definition))
      definitions
      (definition-provided-names index definition)))
   {}
   (definition-nodes index)))

(defn definition-kind [index definition]
  (lower/fact-text index definition "form-kind"))

(defn unique-nodes [nodes]
  (vals
   (reduce (fn [found node] (assoc found (id-value node) node)) {} nodes)))

(defn resolved-definition-nodes [definitions index module-name spelling]
  (let [resolved (lower/resolved-source-key
                  (lower/resolve-source-name index module-name spelling))]
    (unique-nodes (get definitions resolved []))))

(defn definition-spellings [index definition definition-ids]
  (loop [queue (conj clojure.lang.PersistentQueue/EMPTY definition)
         seen #{}
         references #{}
         types #{}]
    (if (empty? queue)
      {:references references :types types}
      (let [node (peek queue)
            remaining (pop queue)
            identity (id-value node)]
        (if (or (contains? seen identity)
                (and (not= identity (id-value definition))
                     (contains? definition-ids identity)))
          (recur remaining (conj seen identity) references types)
          (let [kind (lower/fact-text index node "form-kind")
                reference
                (when (and (= "ref" kind)
                           (str/blank?
                            (lower/fact-text index node "binding-id")))
                  (lower/source-reference-spelling index node))
                type-name
                (when (#{"type-prim" "type-app"} kind)
                  (lower/fact-text index node "name"))]
            (recur
             (into remaining (lower/index-child-nodes index node))
             (conj seen identity)
             (if (str/blank? reference) references (conj references reference))
             (if (str/blank? type-name) types (conj types type-name)))))))))

(defn dependency-nodes [definitions index definition definition-ids]
  (let [module-name (lower/source-module-name index definition)
        spellings (definition-spellings index definition definition-ids)
        references
        (vec
         (concat
          (:references spellings)
          (if (= "unsupported-extend-type" (definition-kind index definition))
            (remove str/blank?
                    (map #(lower/fact-text index % "protocol")
                         (definition-child-items index definition "impls")))
            [])))
        types (:types spellings)]
    (unique-nodes
     (concat
      (mapcat #(resolved-definition-nodes definitions index module-name %) references)
      (mapcat #(resolved-definition-nodes definitions index module-name %) types)))))

(def projection-parallelism
  (max 1
       (min 22
            (- (.availableProcessors (Runtime/getRuntime)) 2))))

(defn ordered-parallel-mapv [operation values]
  (let [items (vec values)]
    (if (or (= 1 projection-parallelism) (< (count items) 2))
      (mapv operation items)
      (let [pool (Executors/newFixedThreadPool projection-parallelism)
            futures
            (mapv
             (fn [item]
               (.submit
                pool
                (reify Callable
                  (call [_] (operation item)))))
             items)]
        (try
          (mapv #(.get %) futures)
          (finally
            (.shutdownNow pool)))))))

(defn module-definition-batches [index definitions]
  (->> definitions
       (group-by #(lower/source-module-name index %))
       (map (fn [[module-name module-definitions]]
              [module-name (vec (sort-by id-value module-definitions))]))
       (sort-by first)
       vec))

(defn flatten-module-entries [module-results]
  (vec
   (mapcat #(sort-by first (second %))
           (sort-by first module-results))))

(defn dependency-projection [definitions index]
  (let [all-definitions (vec (definition-nodes index))
        module-batches (module-definition-batches index all-definitions)
        definition-ids (set (map id-value all-definitions))
        ordered-definitions (vec (mapcat second module-batches))
        _worker-evidence
        (binding [*out* *err*]
          (println (str "beagle build: source-projection workers="
                        (min projection-parallelism (count ordered-definitions))
                        " definitions=" (count ordered-definitions))))
        entries
        (ordered-parallel-mapv
         (fn [definition]
           [(id-value definition)
            (vec (dependency-nodes
                  definitions index definition definition-ids))])
         ordered-definitions)]
    {:ordered ordered-definitions :by-id (into {} entries)}))

(defn source-node-name-index [rows]
  (reduce
   (fn [names row]
     (let [with-subject
           (assoc names
                  (id-value (slice/node-id (slice/slicefactv0-subject row)))
                  (slice/slicefactv0-subject row))]
       (if (= "n" (slice/slicefactv0-kind row))
         (assoc with-subject
                (id-value (slice/node-id (slice/slicefactv0-object row)))
                (slice/slicefactv0-object row))
         with-subject)))
   {}
   rows))

(defn semantic-read-set-rows
  [rows index selected ordered-definitions dependencies-by-id]
  (let [node-names (source-node-name-index rows)
        selected-module-batches
        (->> (module-definition-batches index ordered-definitions)
             (map (fn [[module-name module-definitions]]
                    [module-name
                     (filterv #(contains? selected (id-value %))
                              module-definitions)]))
             (remove #(empty? (second %)))
             vec)
        reads
        (vec
         (mapcat
          identity
          (map second
               (flatten-module-entries
                (ordered-parallel-mapv
                 (fn [[module-name module-definitions]]
                   [module-name
                    (mapv
                     (fn [definition]
                       (let [definition-id (id-value definition)
                             subject (get node-names definition-id)
                             dependencies
                             (sort-by id-value
                                      (remove #(= definition-id (id-value %))
                                              (get dependencies-by-id definition-id [])))]
                         (when-not subject
                           (throw (ex-info "semantic unit has no source-fact node name"
                                           {:definition definition-id})))
                         [definition-id
                          (mapv
                           (fn [dependency]
                             (let [object (get node-names (id-value dependency))]
                               (when-not object
                                 (throw
                                  (ex-info "semantic read has no source-fact node name"
                                           {:definition definition-id
                                            :dependency (id-value dependency)})))
                               (slice/->SliceFactV0 subject "semantic-read" "n" object)))
                           dependencies)]))
                     module-definitions)])
                 selected-module-batches)))))]
    (vec (concat rows reads))))

(defn reachable-definition-ids [dependencies-by-id seeds]
  (loop [queue (into clojure.lang.PersistentQueue/EMPTY seeds) selected #{}]
    (if (empty? queue)
      selected
      (let [definition (peek queue)
            identity (id-value definition)]
        (if (contains? selected identity)
          (recur (pop queue) selected)
          (recur (into (pop queue) (get dependencies-by-id identity []))
                 (conj selected identity)))))))

(defn definition-sequence-names [rows]
  (set
   (for [row rows
         :when (and (= "definitions" (slice/slicefactv0-predicate row))
                    (= "n" (slice/slicefactv0-kind row)))]
     (slice/slicefactv0-object row))))

(defn sequence-row? [definition-sequences row]
  (and (contains? definition-sequences (slice/slicefactv0-subject row))
       (some? (re-matches #"[fa][0-9]+" (slice/slicefactv0-predicate row)))))

(defn replacement-sequence-rows [rows definition-sequences selected]
  (let [rows-by-sequence
        (reduce
         (fn [buckets row]
           (let [sequence-name (slice/slicefactv0-subject row)]
             (if (and (contains? definition-sequences sequence-name)
                      (some? (re-matches #"[fa][0-9]+"
                                         (slice/slicefactv0-predicate row)))
                      (= "n" (slice/slicefactv0-kind row)))
               (update buckets sequence-name (fnil conj []) row)
               buckets)))
         {}
         rows)]
    (vec
     (mapcat
      (fn [sequence-name]
        (map-indexed
         (fn [position definition-name]
           (slice/->SliceFactV0 sequence-name (str "f" position) "n" definition-name))
         (for [row (get rows-by-sequence sequence-name [])
               :when (contains? selected
                                (id-value
                                 (slice/node-id (slice/slicefactv0-object row))))]
           (slice/slicefactv0-object row))))
      (sort definition-sequences)))))

(defn reachable-node-names [rows]
  (let [children
        (reduce
         (fn [edges row]
           (if (= "n" (slice/slicefactv0-kind row))
             (update edges (slice/slicefactv0-subject row) (fnil conj [])
                     (slice/slicefactv0-object row))
             edges))
         {}
         rows)]
    (loop [queue (conj clojure.lang.PersistentQueue/EMPTY "0") reached #{}]
      (if (empty? queue)
        reached
        (let [node (peek queue)]
          (if (contains? reached node)
            (recur (pop queue) reached)
            (recur (into (pop queue) (get children node []))
                   (conj reached node))))))))

(defn projected-rows [rows selected]
  (let [sequences (definition-sequence-names rows)
        without-definitions (remove #(sequence-row? sequences %) rows)
        candidate (vec (concat without-definitions
                               (replacement-sequence-rows rows sequences selected)))
        reached (reachable-node-names candidate)]
    (filterv #(contains? reached (slice/slicefactv0-subject %)) candidate)))

(defn rows-text [rows]
  (apply str
         (for [row rows]
           (str (slice/slicefactv0-subject row) "\t"
                (slice/slicefactv0-predicate row) "\t"
                (slice/slicefactv0-kind row) "\t"
                (slice/slicefactv0-object row) "\n"))))

(defn entry-seeds [definitions index entries strict-entry-abi?]
  (loop [remaining entries seeds []]
    (if (empty? remaining)
      {:ok? true :seeds seeds}
      (let [entry (first remaining)
            matches (get definitions entry [])]
        (cond
          (empty? matches)
          {:ok? false :detail (str "entry not found: " entry)}

          (not= 1 (count matches))
          {:ok? false :detail (str "entry is ambiguous: " entry)}

          (not= "defn" (definition-kind index (first matches)))
          {:ok? false :detail (str "entry is not a source function: " entry)}

          (= "true" (lower/fact-text index (first matches) "private"))
          {:ok? false
           :detail (str "entry " entry " must be a public source function")}

          (not= "false" (lower/fact-text index (first matches) "private"))
          {:ok? false :detail (str "entry export visibility is unavailable: " entry)}

          (and strict-entry-abi?
               (not= 0
                     (count
                      (lower/function-parameter-items index (first matches)))))
          {:ok? false
           :detail
           (str "entry " entry " must have zero source parameters (got "
                (count (lower/function-parameter-items index (first matches))) ")")}

          (and strict-entry-abi?
               (some? (lower/index-first-object index (first matches) "rest")))
          {:ok? false
           :detail (str "entry " entry " must not have a rest parameter")}

          (and strict-entry-abi?
               (let [return-type
                     (lower/index-first-object index (first matches) "ret")]
                 (or
                  (nil? return-type)
                  (not= "type-prim"
                        (lower/fact-text index return-type "form-kind"))
                  (not= "Int" (lower/fact-text index return-type "name")))))
          {:ok? false
           :detail (str "entry " entry " must have an explicit Int return")}

          :else
          (recur (rest remaining) (conj seeds (first matches))))))))

(defn project-entry-rows [rows entries strict-entry-abi?]
  (let [source (slice/source-program rows "native.module" "source.facts")
        index (lower/build-source-index
               (stages/sourcestagev1-terms source)
               (stages/sourcestagev1-modules source))
        definitions (definitions-by-key index)
        dependencies (dependency-projection definitions index)
        seeds (if (empty? entries)
                {:ok? true :seeds (:ordered dependencies)}
                (entry-seeds definitions index entries strict-entry-abi?))]
    (if-not (:ok? seeds)
      seeds
      (let [selected
            (reachable-definition-ids (:by-id dependencies) (:seeds seeds))]
        {:ok? true
         :rows (if (empty? entries) rows (projected-rows rows selected))
         :index index
         :selected selected
         :ordered-definitions (:ordered dependencies)
         :dependencies-by-id (:by-id dependencies)}))))

(defn source-definition-count [rows]
  (reduce + (map #(count (slice/form-node-names rows %))
                 ["record" "defunion" "deferror" "def" "defn"])))

(defn obligation-pass? [verdicts]
  (and (= 10 (count verdicts))
       (every? obligations/obligation-passed? verdicts)))

(defn epoch-pending-details [epoch-result]
  (if (instance? native.lower.EpochLoweringPendingV0 epoch-result)
    (mapv core/receiptobligationv0-detail
          (lower/epochloweringpendingv0-obligations epoch-result))
    []))

(defn epoch-pending-report [epoch-result]
  (apply str (map #(str "epoch-pending " % "\n")
                  (epoch-pending-details epoch-result))))

(defn parallel-instructions [program]
  (mapcat (fn [function]
            (mapcat core/basicblock-instructions
                    (core/functiondef-blocks function)))
          (core/nativecoreprogram-functions program)))

(defn program-has-parallel? [program]
  (some? (some #(or
                  (instance? native.core.TiledStepF64InstructionV0 %)
                  (instance? native.core.F64BufferSumInstructionV0 %))
               (parallel-instructions program))))

(defn c11-parallel-refusal-report [program]
  (if-not (program-has-parallel? program)
    ""
    (let [result (c11/materialize-program program 0)]
      (if (instance? native.c11.C11Failure result)
        (str "c11-parallel REFUSED " (c11/c11failure-detail result) "\n")
        "c11-parallel INVALID accepted\n"))))

(defn parallel-kernel-name [program target]
  (or (some (fn [function]
              (when (core/native-id= (core/functiondef-id function) target)
                (core/functiondef-name function)))
            (core/nativecoreprogram-functions program))
      "missing"))

(defn parallel-plan-report [program]
  (let [plans (sort-by #(core/nativeid-value
                         (core/instruction-result-id %))
                       (filter #(or
                                  (instance? native.core.TiledStepF64InstructionV0 %)
                                  (instance? native.core.F64BufferSumInstructionV0 %))
                               (parallel-instructions program)))]
    (apply str
      (map (fn [instruction]
             (let [id (core/nativeid-value (core/instruction-result-id instruction))]
               (if (instance? native.core.TiledStepF64InstructionV0 instruction)
                 (str "parallel-plan " id "\n"
                      "parallel-kernel "
                      (parallel-kernel-name
                        program
                        (core/tiledstepf64instructionv0-kernel-id instruction))
                      "\nparallel-partition-policy linear-contiguous-v0\n"
                      "parallel-tile-width "
                      (core/tiledstepf64instructionv0-tile-width instruction) "\n"
                      "parallel-halo "
                      (core/tiledstepf64instructionv0-left-halo instruction) " "
                      (core/tiledstepf64instructionv0-right-halo instruction) "\n"
                      "parallel-boundary "
                      (if (instance? native.core.ParallelPeriodicV0
                            (core/tiledstepf64instructionv0-boundary instruction))
                        "periodic" "bounded")
                      "\nparallel-reduction none\n"
                      "parallel-worker-count excluded-from-program-identity\n")
                 (str "parallel-plan " id "\n"
                      "parallel-kernel none\n"
                      "parallel-partition-policy linear-contiguous-v0\n"
                      "parallel-tile-width "
                      (core/f64buffersuminstructionv0-tile-width instruction) "\n"
                      "parallel-halo 0 0\n"
                      "parallel-boundary bounded\n"
                      "parallel-reduction f64-adjacent-pairwise-v0\n"
                      "parallel-worker-count excluded-from-program-identity\n"))))
           plans))))

(defn epoch-result-receipt [epoch-result]
  (if (instance? native.lower.EpochLoweringCompleteV0 epoch-result)
    (lower/epochloweringcompletev0-receipt epoch-result)
    (lower/epochloweringpendingv0-receipt epoch-result)))

(defn native-provenance-report
  [source-receipt frozen-source typing-result frozen-typed native-result native-frozen
   epoch-result frozen-native compiler-commit configuration]
  (let [source-digest (stages/frozensourcestagev1-digest frozen-source)
        typing-receipt (lower/typingacceptedv0-receipt typing-result)
        typed-digest (stages/frozentypedstagev0-digest frozen-typed)
        native-receipt (lower/nativeloweringcompletev0-receipt native-result)
        native-digest (stages/frozennativestagev0-digest native-frozen)
        epoch-receipt (epoch-result-receipt epoch-result)
        epoch-digest (stages/frozennativestagev0-digest frozen-native)
        receipts [source-receipt typing-receipt native-receipt epoch-receipt]
        receipt-context-ok?
        (every?
         (fn [receipt]
           (and (= compiler-commit (core/passreceiptv0-compiler-commit receipt))
                (= configuration (core/passreceiptv0-configuration receipt))
                (= (stages/configuration-digest configuration)
                   (core/passreceiptv0-configuration-digest receipt))))
         receipts)]
    (when-not (and receipt-context-ok?
                   (= source-digest
                      (core/passreceiptv0-input-digest source-receipt))
                   (= source-digest
                      (core/passreceiptv0-output-digest source-receipt))
                   (= (core/passreceiptv0-output-digest source-receipt)
                      (core/passreceiptv0-input-digest typing-receipt))
                   (= typed-digest
                      (core/passreceiptv0-output-digest typing-receipt))
                   (= typed-digest
                      (core/passreceiptv0-input-digest native-receipt))
                   (= native-digest
                      (core/passreceiptv0-output-digest native-receipt))
                   (= native-digest
                      (core/passreceiptv0-input-digest epoch-receipt))
                   (= epoch-digest
                      (core/passreceiptv0-output-digest epoch-receipt)))
      (throw (ex-info "Native Core provenance receipt chain is broken"
                      {:source source-digest
                       :typed typed-digest
                       :native native-digest
                       :epoch epoch-digest})))
    (str "native-provenance-v0 source " source-digest "\n"
         "native-provenance-v0 typed " typed-digest "\n"
         "native-provenance-v0 native " native-digest "\n"
         "native-provenance-v0 epoch " epoch-digest "\n")))

(defn write-native-receipts!
  [artifacts-dir source-receipt typing-receipt native-receipt epoch-receipt]
  (let [receipts [source-receipt typing-receipt native-receipt epoch-receipt]
        encoding (stages/encode-pass-receipts receipts)]
    (spit (str artifacts-dir "/native.receipts") encoding)))

(defn native-report-head
  [rows entries source program projected native-program epoch-result abi verdicts]
  (str (apply str (map #(str "source-entry " % "\n") entries))
       "stage source-freeze ACCEPTED\n"
       "stage source-to-typed ACCEPTED\n"
       "stage typed-to-native COMPLETE\n"
       "native-lowering-result NativeLoweringCompleteV0\n"
       "stage native-to-epoch "
       (if (lower/epoch-result-complete? epoch-result) "COMPLETE" "PENDING") "\n"
       "epoch-regions-minted "
       (- (count (core/nativecoreprogram-regions program))
          (count (core/nativecoreprogram-regions native-program))) "\n"
       (epoch-pending-report epoch-result)
       "source-modules " (count (stages/sourcestagev1-modules source)) "\n"
       "source-imports " (count (stages/sourcestagev1-imports source)) "\n"
       "source-definitions " (source-definition-count rows) "\n"
       "program-types " (count (core/nativecoreprogram-types program)) "\n"
       "program-functions " (count (core/nativecoreprogram-functions program)) "\n"
       "program-abis " (count (core/nativecoreprogram-abis program)) "\n"
       "native-program module.native-program\n"
       "native-program-sha256 module.native-program.sha256\n"
       "projected-types " (count (core/nativecoreprogram-types projected)) "\n"
       (body-slice/function-report (core/nativecoreprogram-functions projected))
       (parallel-plan-report projected)
       (c11-parallel-refusal-report projected)
       (body-slice/obligation-report "obligation-projection" verdicts)
       (body-slice/failing-obligation-verdicts verdicts)))

;; Function bodies are independent after the program and every canonical
;; whole-program table have frozen.  Workers receive absolute positions in the
;; canonical function order and write one isolated fragment per position.  The
;; coordinator alone concatenates those fragments, in absolute order.
(def emission-worker-code
  (str
   "(require (quote [clojure.string :as str])\n"
   "         (quote [native.body-c17 :as body])\n"
   "         (quote [native.qbe :as qbe])\n"
   "         (quote [native.stages :as stages]))\n"
   "(let [[backend wire-path worker-dir indexes-text abi-id worker-index]\n"
   "      *command-line-args*\n"
   "      result-path (str worker-dir \"/result\")\n"
   "      error-path (str worker-dir \"/error.txt\")]\n"
   "  (try\n"
   "    (when (= worker-index\n"
   "             (System/getenv \"BEAGLE_CORE_EMIT_FAIL_WORKER_INDEX\"))\n"
   "      (throw (ex-info \"injected emission worker failure\" {})))\n"
   "    (let [program (stages/decode-trusted-native-core-program-wire-v1\n"
   "                    (slurp wire-path))\n"
   "          indexes (mapv #(Long/parseLong %) (str/split indexes-text #\",\"))]\n"
   "      (when (nil? program)\n"
   "        (throw (ex-info \"worker could not decode canonical native program\" {})))\n"
   "      (let [batch (case backend\n"
   "                    \"c17\" (body/materialize-function-indexes\n"
   "                              program 0 abi-id [] indexes)\n"
   "                    \"qbe\" (qbe/materialize-function-indexes\n"
   "                              program 0 abi-id indexes)\n"
   "                    (throw (ex-info \"unknown emission backend\"\n"
   "                                    {:backend backend})))\n"
   "            ok? (case backend\n"
   "                  \"c17\" (body/function-batch-ok? batch)\n"
   "                  \"qbe\" (qbe/function-batch-ok? batch))]\n"
   "        (if-not ok?\n"
   "          (do\n"
   "            (spit error-path\n"
   "                  (case backend\n"
   "                    \"c17\" (body/function-batch-detail batch)\n"
   "                    \"qbe\" (qbe/function-batch-detail batch)))\n"
   "            (spit result-path \"REFUSED\\n\")\n"
   "            (System/exit 1))\n"
   "          (do\n"
   "            (doseq [fragment\n"
   "                    (case backend\n"
   "                      \"c17\" (body/function-batch-fragments batch)\n"
   "                      \"qbe\" (qbe/function-batch-fragments batch))]\n"
   "              (let [index (case backend\n"
   "                            \"c17\" (body/bodyfunctionfragmentv0-function-index fragment)\n"
   "                            \"qbe\" (qbe/qbefunctionfragmentv0-function-index fragment))\n"
   "                    text (case backend\n"
   "                           \"c17\" (body/bodyfunctionfragmentv0-text fragment)\n"
   "                           \"qbe\" (qbe/qbefunctionfragmentv0-text fragment))]\n"
   "                (spit (format \"%s/fragment-%012d.part\" worker-dir index) text)))\n"
   "            (spit result-path \"OK\\n\")))))\n"
   "    (catch Throwable error\n"
   "      (spit error-path\n"
   "            (str (.getName (class error)) \": \"\n"
   "                 (or (ex-message error) \"unknown worker error\") \"\\n\"))\n"
   "      (spit result-path \"ERROR\\n\")\n"
   "      (System/exit 2))))\n"))

;; The commander benchmark measured 5.49s serial for 118 functions versus
;; roughly 40s of fixed worker-process startup.  Linear extrapolation puts the
;; crossover near 118 * 40 / 5.49 = 860 functions.
(def emission-parallel-threshold 860)

(defn emission-parallel? [program]
  (let [function-count (count (core/nativecoreprogram-functions program))
        requested (Long/parseLong (System/getenv "BEAGLE_CORE_EMIT_WORKERS"))
        explicit? (= "1" (System/getenv "BEAGLE_CORE_EMIT_WORKERS_EXPLICIT"))]
    (and (> requested 1)
         (> function-count 1)
         (or explicit?
             (>= function-count emission-parallel-threshold)))))

(defn emission-manager-error [root]
  (let [path (str root "/manager.stderr")]
    (if (.isFile (File. path))
      (str/trim (slurp path))
      "manager produced no diagnostic")))

(defn launch-emission-manager!
  [backend wire-path function-count worker-count abi-id]
  (let [root (System/getenv "BEAGLE_CORE_EMIT_WORK_ROOT")
        stdout-path (str root "/manager.stdout")
        stderr-path (str root "/manager.stderr")
        command [(System/getenv "BEAGLE_CORE_EMIT_MANAGER")
                 (System/getenv "BEAGLE_CORE_EMIT_SUPERVISOR")
                 (System/getenv "BEAGLE_CORE_EMIT_WORKER_TIMEOUT_SECONDS")
                 (System/getenv "BEAGLE_CORE_EMIT_KILL_GRACE_SECONDS")
                 (System/getenv "BEAGLE_CORE_EMIT_BB")
                 (System/getenv "BEAGLE_CORE_EMIT_COMPILED")
                 backend wire-path root (str function-count) (str worker-count)
                 abi-id emission-worker-code]
        builder (ProcessBuilder. (into-array String command))]
    (.redirectOutput builder (ProcessBuilder$Redirect/to (File. stdout-path)))
    (.redirectError builder (ProcessBuilder$Redirect/to (File. stderr-path)))
    {:root root :process (.start builder)}))

(defn emission-manager-deadline-seconds [worker-timeout kill-grace]
  (let [raw (or (System/getenv "BEAGLE_DEADLINE_SCALE") "1")
        pieces (str/split raw #"/" -1)
        scale (try
                (case (count pieces)
                  1 (Double/parseDouble (first pieces))
                  2 (/ (Double/parseDouble (first pieces))
                       (Double/parseDouble (second pieces)))
                  Double/NaN)
                (catch Exception _ Double/NaN))]
    (when (or (Double/isNaN scale)
              (Double/isInfinite scale)
              (not (pos? scale)))
      (throw
       (ex-info
        (str "BEAGLE_DEADLINE_SCALE must be a positive rational: " raw)
        {:scale raw})))
    (+ (long (Math/ceil (* worker-timeout scale))) kill-grace 30)))

(defn await-emission-manager [stage-name manager]
  (let [process (:process manager)
        worker-timeout (Long/parseLong
                         (System/getenv
                          "BEAGLE_CORE_EMIT_WORKER_TIMEOUT_SECONDS"))
        kill-grace (Long/parseLong
                     (System/getenv "BEAGLE_CORE_EMIT_KILL_GRACE_SECONDS"))
        wait-seconds (emission-manager-deadline-seconds
                      worker-timeout kill-grace)
        exited? (.waitFor process wait-seconds TimeUnit/SECONDS)]
    (when-not exited?
      (.destroy process)
      (when-not (.waitFor process kill-grace TimeUnit/SECONDS)
        (.destroyForcibly process)
        (.waitFor process kill-grace TimeUnit/SECONDS))
      (throw
       (ex-info
        (str stage-name " emission manager exceeded its stage deadline")
        {:stage stage-name :status :timeout})))
    (let [status (.exitValue process)
          receipt-path (str (:root manager) "/manager.receipt")
          receipt (if (.isFile (File. receipt-path))
                    (str/trim (slurp receipt-path))
                    "")]
      (when (or (not= status 0) (str/blank? receipt))
        (throw
         (ex-info
          (str stage-name " emission manager failed: "
               (emission-manager-error (:root manager)))
          {:stage stage-name :status status :receipt receipt})))
      receipt)))

(defn read-emission-worker [backend worker-index]
  (let [root (System/getenv "BEAGLE_CORE_EMIT_WORK_ROOT")
        worker-dir (str root "/" backend "-" worker-index)
        status-path (str worker-dir "/manager.status")
        status (if (.isFile (File. status-path))
                 (Long/parseLong (str/trim (slurp status-path)))
                 -1)
        receipt-path (str worker-dir "/supervisor.receipt")
        receipt (if (.isFile (File. receipt-path))
                  (str/trim (slurp receipt-path))
                  "")
        expected (if (= status 124)
                   "subtree-reaped-v0 timeout status=124"
                   (str "subtree-reaped-v0 exit status=" status))
        result-path (str worker-dir "/result")
        result (if (.isFile (File. result-path))
                 (str/trim (slurp result-path))
                 "MISSING")]
    {:index worker-index
     :worker-dir worker-dir
     :status status
     :receipt receipt
     :expected expected
     :result result}))

(defn emission-worker-error [worker]
  (let [path (str (:worker-dir worker) "/error.txt")]
    (if (.isFile (File. path))
      (str/trim (slurp path))
      (let [stderr-path (str (:worker-dir worker) "/stderr.log")]
        (if (.isFile (File. stderr-path))
          (str/trim (slurp stderr-path))
          "worker produced no diagnostic")))))

(defn parallel-function-text [stage-name backend program abi-id]
  (let [function-count (count (core/nativecoreprogram-functions program))
        requested (Long/parseLong (System/getenv "BEAGLE_CORE_EMIT_WORKERS"))
        worker-count (min requested function-count)]
    (if (= function-count 0)
      {:ok? true :text ""}
      (let [root (System/getenv "BEAGLE_CORE_EMIT_WORK_ROOT")
            wire-path (str root "/" backend ".native-program-wire-v1")
            _wire (spit wire-path
                        (stages/encode-native-core-program-wire-v1 program))
            manager (launch-emission-manager!
                     backend wire-path function-count worker-count abi-id)
            receipt (await-emission-manager stage-name manager)
            expected-receipt (str "emission-manager-v0 reaped workers="
                                  worker-count)
            _receipt
            (when-not (= receipt expected-receipt)
              (throw
               (ex-info
                (str stage-name " emission manager receipt mismatch")
                {:stage stage-name
                 :receipt receipt
                 :expected-receipt expected-receipt})))
            completed (mapv #(read-emission-worker backend %)
                            (range worker-count))
            failure (first
                     (filter #(or (not= (:receipt %) (:expected %))
                                  (not= (:status %) 0)
                                  (not= (:result %) "OK"))
                             completed))]
        (if (some? failure)
          (if (and (= (:status failure) 1)
                   (= (:result failure) "REFUSED")
                   (= (:receipt failure) (:expected failure)))
            {:ok? false :detail (emission-worker-error failure)}
            (throw
             (ex-info
              (str stage-name " emission worker " (:index failure)
                   " failed: " (emission-worker-error failure))
              {:backend backend
               :stage stage-name
               :worker (:index failure)
               :status (:status failure)
               :receipt (:receipt failure)
               :expected-receipt (:expected failure)
               :result (:result failure)})))
          {:ok? true
           :text
           (apply str
                  (map
                   (fn [function-index]
                     (str (if (and (= backend "qbe")
                                   (> function-index 0))
                            "\n" "")
                          (slurp
                           (format "%s/%s-%d/fragment-%012d.part"
                                   root backend
                                   (mod function-index worker-count)
                                   function-index))))
                   (range function-count)))})))))

(defn emit-c17-materialization!
  [result artifacts-dir stage-name report-name report input-digest
   compiler-commit configuration]
  (if-not (body/materialization-ok? result)
    (do
      (finish-stage! stage-name "REFUSED")
      {:ok? false
       :report report})
    (let [artifact (body/materialization-artifact result)
          entry-map-path (str artifacts-dir "/native.entry-map")
          entry-map-text (slurp entry-map-path)
          hashes (conj (body/artifact-hashes artifact)
                   (core/->ArtifactHashV0 "native.entry-map"
                     (stages/content-digest entry-map-text)))
          receipt (stages/make-artifact-receipt
                    input-digest compiler-commit "native-to-c17" configuration
                    "c17" "restricted-c17-v0" hashes)
          artifact-report
          (apply str
                 (map (fn [hash]
                        (str report-name "-artifact "
                             (core/artifacthashv0-name hash) " "
                             (core/artifacthashv0-digest hash) "\n"))
                      hashes))]
      (spit (str artifacts-dir "/" (body/bodyartifactv0-header-name artifact))
            (body/bodyartifactv0-header-text artifact))
      (spit (str artifacts-dir "/" (body/bodyartifactv0-source-name artifact))
            (body/bodyartifactv0-source-text artifact))
      (spit (str artifacts-dir "/c17.receipt")
            (stages/encode-pass-receipt receipt))
      (finish-stage! stage-name "OK")
      {:ok? true
       :report (str report artifact-report)})))

(defn materialize-c17!
  [projected artifacts-dir plan simd-required? abi-id input-digest
   compiler-commit configuration]
  (start-stage! "materialization-c17")
  (let [attestation-report (System/getenv
                             "BEAGLE_CORE_C17_ATTESTATION_REPORT")]
    (if-not (str/blank? attestation-report)
      (do
        (finish-stage! "materialization-c17" "OK")
        {:ok? true :report (slurp attestation-report)})
      (let [parallel? (and (not simd-required?)
                           (emission-parallel? projected))
            result (cond
                 parallel?
                 (let [rejection
                       (body/materialization-preflight-rejection
                        projected 0 abi-id)]
                   (if (some? rejection)
                     (body/->BodyFailureV0 rejection)
                     (let [batch (parallel-function-text
                                  "materialization-c17" "c17" projected
                                  abi-id)]
                       (if (:ok? batch)
                         (body/materialize-program-plans-with-function-text
                          projected 0 abi-id [] (:text batch))
                         (throw
                          (ex-info
                           "C17 worker refused after coordinator preflight passed"
                           {:backend "c17"
                            :stage "materialization-c17"
                            :detail (:detail batch)}))))))
                 (not simd-required?)
                 (body/materialize-program-for-abi projected 0 abi-id)
                 (not= abi-id "lp64")
                 (body/->BodyFailureV0
                   (body/->TargetCapabilityRejectionV0
                    (body/->SimdAbiGapV0 abi-id)
                    (simd/demanded-backend-refusal "wasm")))
                 :else (body/materialize-program-with-simd projected 0 plan))
        ok? (body/materialization-ok? result)
        parallel-wasm-refusal? (and (= abi-id "wasm32")
                                    (program-has-parallel? projected)
                                    (not ok?))
        report (if parallel-wasm-refusal?
                 (str "wasm-parallel REFUSED "
                      (body/materialization-detail result) "\n")
                 (str "materialize-c17 "
                      (if ok?
                        (let [artifact (body/materialization-artifact result)]
                          (str "OK "
                               (body/bodyartifactv0-header-name artifact) " "
                               (body/bodyartifactv0-source-name artifact)
                               (if simd-required?
                                 " SIMD-ELIGIBLE actual-vectorization=pending-compiler-evidence"
                                 "")))
                        (str "REFUSED " (body/materialization-detail result)))
                      "\n"))]
        (emit-c17-materialization! result artifacts-dir "materialization-c17"
                                   "materialize-c17" report input-digest
                                   compiler-commit configuration)))))

;; The first live Wasm materializer deliberately reuses the Restricted C17
;; projection. The external wasi-clang step is a separate shell seam below, so
;; a future direct emitter replaces this wrapper without changing Core freeze.
;; SIMD never reaches here: the shell pins this materializer to wasm32, and the
;; SIMD backend refuses every ABI but lp64.
(defn materialize-wasm-bootstrap-c17!
  [projected artifacts-dir abi-id input-digest compiler-commit configuration]
  (start-stage! "materialization-wasm-bootstrap-c17")
  (let [parallel? (emission-parallel? projected)
        result (if parallel?
                 (let [rejection
                       (body/materialization-preflight-rejection
                        projected 0 abi-id)]
                   (if (some? rejection)
                     (body/->BodyFailureV0 rejection)
                     (let [batch (parallel-function-text
                                  "materialization-wasm-bootstrap-c17"
                                  "c17" projected abi-id)]
                       (if (:ok? batch)
                         (body/materialize-program-plans-with-function-text
                          projected 0 abi-id [] (:text batch))
                         (throw
                          (ex-info
                           "C17 worker refused after coordinator preflight passed"
                           {:backend "c17"
                            :stage "materialization-wasm-bootstrap-c17"
                            :detail (:detail batch)}))))))
                 (body/materialize-program-for-abi projected 0 abi-id))
        ok? (body/materialization-ok? result)
        report (str "materialize-wasm-bootstrap-c17 "
                    (if ok?
                      (let [artifact (body/materialization-artifact result)]
                        (str "OK "
                             (body/bodyartifactv0-header-name artifact) " "
                             (body/bodyartifactv0-source-name artifact)))
                      (str "REFUSED " (body/materialization-detail result)))
                    "\n")]
    (emit-c17-materialization! result artifacts-dir
                               "materialization-wasm-bootstrap-c17"
                               "materialize-wasm-bootstrap-c17" report
                               input-digest compiler-commit configuration)))

(defn materialize-qbe! [projected artifacts-dir abi-id plan simd-required?]
  (start-stage! "materialization-qbe")
  (let [parallel? (and (not simd-required?)
                       (emission-parallel? projected))
        result (cond
                 simd-required?
                 (qbe/materialize-program-with-simd projected 0 abi-id plan)
                 parallel?
                 (let [batch (parallel-function-text
                              "materialization-qbe" "qbe" projected abi-id)]
                   (if (:ok? batch)
                     (qbe/materialize-program-with-function-text
                      projected 0 abi-id (:text batch))
                     (qbe/->QbeFailure (:detail batch))))
                 :else
                 (qbe/materialize-program projected 0 abi-id))
        ok? (instance? native.qbe.QbeSuccess result)
        parallel-refusal? (and (program-has-parallel? projected) (not ok?))
        report (if parallel-refusal?
                 (str "qbe-parallel REFUSED "
                      (qbe/qbefailure-detail result) "\n")
                 (str "materialize-qbe "
                      (if ok?
                        (let [artifact (qbe/qbesuccess-artifact result)]
                          (str "OK " (qbe/qbeartifact-module-name artifact)))
                        (str "REFUSED " (qbe/qbefailure-detail result)))
                      "\n"))]
    (if-not ok?
      (do
        (finish-stage! "materialization-qbe" "REFUSED")
        {:ok? false
         :report report})
      (let [artifact (qbe/qbesuccess-artifact result)]
        (spit (str artifacts-dir "/" (qbe/qbeartifact-module-name artifact))
              (qbe/qbeartifact-module-text artifact))
        (finish-stage! "materialization-qbe" "OK")
        {:ok? true
         :report report}))))

(defn materialize-programs!
  [projected qbe-projected artifacts-dir materializers abi-id plan
   simd-required? input-digest compiler-commit configuration]
  (loop [remaining materializers reports []]
    (if (empty? remaining)
      {:ok? true :report (str (apply str reports) "result PASS\n")}
      (let [materializer (first remaining)
            result (case materializer
                     "c17" (materialize-c17! projected artifacts-dir plan
                             simd-required? abi-id input-digest
                             compiler-commit configuration)
                     "qbe" (materialize-qbe! qbe-projected artifacts-dir abi-id
                             plan simd-required?)
                     "wasm" (materialize-wasm-bootstrap-c17!
                              projected artifacts-dir abi-id input-digest
                              compiler-commit configuration))
            next-reports (conj reports (:report result))]
        (if (:ok? result)
          (recur (rest remaining) next-reports)
          {:ok? false
           :report (str (apply str next-reports)
                        "result FAIL materialization\n")})))))

(defn stage-core-checkpoint! [frozen qbe-frozen report-head]
  (let [stage-dir (System/getenv "BEAGLE_CORE_CHECKPOINT_STAGE")]
    (when-not (str/blank? stage-dir)
      (spit (str stage-dir "/frozen-native-stage.wire-v1")
            (stages/encode-frozen-native-stage-wire-v1 frozen))
      (spit (str stage-dir "/qbe-frozen-native-stage.wire-v1")
            (stages/encode-frozen-native-stage-wire-v1 qbe-frozen))
      (spit (str stage-dir "/report.head") report-head)
      (spit (str stage-dir "/stage-progress") (stage-progress-text))
      (when (= "after-stage"
               (System/getenv "BEAGLE_CORE_CHECKPOINT_FAILPOINT"))
        (throw (ex-info "Core checkpoint post-stage failpoint" {}))))))

(defn resume-core!
  [wire-path qbe-wire-path report-head-path stage-progress-path artifacts-dir
   compiler-commit materializers abi-id abi simd-required?]
  (let [wire (slurp wire-path)
        qbe-wire (slurp qbe-wire-path)
        ;; The shell admits this path only after a digest-keyed canonical-wire
        ;; PASS or a fresh full decode/re-encode validation.
        frozen (stages/decode-attested-frozen-native-stage-wire-v1 wire)
        qbe-frozen
        (stages/decode-attested-frozen-native-stage-wire-v1 qbe-wire)]
    (when (or (nil? frozen) (nil? qbe-frozen))
      (throw (ex-info "Core checkpoint wires failed attested decoding" {})))
    (let [stage (stages/frozennativestagev0-stage frozen)
          qbe-stage (stages/frozennativestagev0-stage qbe-frozen)
          native-encoding (stages/frozennativestagev0-encoding frozen)
          published-native (slurp (str artifacts-dir "/module.native-program"))
          _native-match
          (when-not (= native-encoding published-native)
            (throw (ex-info "Core checkpoint legacy identity disagrees with its wire"
                            {})))
          program (stages/nativestagev0-program stage)
          projected (body-slice/projected-program program)
          qbe-program (stages/nativestagev0-program qbe-stage)
          _qbe-identity
          (when-not (core/program-epoch-0? qbe-program)
            (throw (ex-info "Core checkpoint QBE identity carries a minted epoch"
                            {})))
          qbe-projected (body-slice/projected-program qbe-program)
          projected-facts (slurp (str artifacts-dir "/source.facts"))
          configuration ["profile=3" (str "abi=" abi-id)
                         (str "source-facts-sha256="
                              (subs (stages/content-digest projected-facts) 7))]
          epoch-digest (stages/frozennativestagev0-digest frozen)
          simd-plan (simd/derive-plan projected epoch-digest abi-id)
          _simd-match
          (when simd-required?
            (let [cached-plan (slurp (str artifacts-dir "/module.simd-plan-v0"))]
              (when-not (= cached-plan (simd/simdplanv0-encoding simd-plan))
                (throw (ex-info "Core checkpoint SIMD plan is inconsistent" {})))))
          _progress (reset! stage-progress-prefix (slurp stage-progress-path))
          materialization
          (materialize-programs! projected qbe-projected artifacts-dir materializers
                                 abi-id simd-plan simd-required? epoch-digest
                                 compiler-commit configuration)]
      (assoc materialization
             :report (str (slurp report-head-path)
                          (:report materialization))))))

(def dev-fact-kind "DevCompileUnitResultV1")
(def dev-fact-profile "profile=3")

(defn empty-dev-counters []
  {:hits 0 :misses 0 :divergences 0})

(defn add-dev-counter [counters status]
  (case status
    :hit (update counters :hits inc)
    :miss (update counters :misses inc)
    :divergence (update counters :divergences inc)
    counters))

(defn merge-dev-counters [left right]
  {:hits (+ (:hits left) (:hits right))
   :misses (+ (:misses left) (:misses right))
   :divergences (+ (:divergences left) (:divergences right))})

(defn dev-fact-enabled? []
  (= "1" (System/getenv "BEAGLE_CORE_DEV_FACT_REUSE")))

(defn dev-unit-id-text [identity]
  (core/nativeid-value identity))

(defn log-dev-fallback!
  [status stage result-key unit-id]
  (binding [*out* *err*]
    (println
     (str "beagle build: dev-facts "
          (if (= :divergence status) "DIVERGENCE" "MISS")
          " stage=" stage
          " key=" result-key
          " unit=" (dev-unit-id-text unit-id)
          " fallback=cold-unit"))))

(defn dev-fact-envelope
  [stage result-key compiler-context profile unit-id payload]
  [dev-fact-kind stage result-key compiler-context profile
   (dev-unit-id-text unit-id)
   (count (.getBytes ^String payload StandardCharsets/UTF_8))
   (stages/content-digest payload)
   payload])

(defn dev-fact-entry
  [stage result-key compiler-context profile unit-id payload]
  (let [envelope (dev-fact-envelope stage result-key compiler-context
                                    profile unit-id payload)
        encoding (pr-str envelope)]
    [(stages/content-digest encoding) encoding]))

(defn invoke-dev-fact-store [command request]
  (let [repo-root (System/getenv "BEAGLE_CORE_REPO_ROOT")
        classpath (str repo-root "/store/out")
        input (str (pr-str request) "\n")]
    (try
      (let [result
            (apply shell/sh
                   (concat
                    ["timeout" "--foreground" "-k" "1s" "2s"
                     "env" "-u" "BEAGLE_STORE_TELEMETRY_LOG"
                     "bb" "-cp" classpath "-m" "store.dev-compile-facts"
                     command]
                    [:in input]))]
        (if (zero? (:exit result))
          (edn/read-string (:out result))
          (do
            (binding [*out* *err*]
              (println
               (str "beagle build: dev-facts Store " command
                    " unavailable exit=" (:exit result))))
            nil)))
      (catch Exception error
        (binding [*out* *err*]
          (println
           (str "beagle build: dev-facts Store " command
                " unavailable: " (.getMessage error))))
        nil))))

(defn query-dev-facts [store requests]
  (let [response
        (invoke-dev-fact-store
         "query"
         ["store.dev-compile-facts/query-v1" store requests])]
    (if (and (vector? response)
             (= 5 (count response))
             (= "store.dev-compile-facts/query-response-v1" (nth response 0))
             (#{"ONLINE" "COLD"} (nth response 1))
             (vector? (nth response 4)))
      {:usable? true :mode (nth response 1) :rows (nth response 4)}
      {:usable? false :mode "DEGRADED" :rows []})))

(defn append-dev-facts [store entries]
  (if (empty? entries)
    "RETAINED"
    (let [response
          (invoke-dev-fact-store
           "append"
           ["store.dev-compile-facts/append-v1" store entries])]
      (if (and (vector? response)
               (= 5 (count response))
               (= "store.dev-compile-facts/append-response-v1"
                  (nth response 0))
               (= "ok" (nth response 1)))
        "PUBLISHED"
        "DEFERRED"))))

(defn parse-dev-fact-row [row]
  (try
    (when (and (vector? row)
               (= 4 (count row))
               (string? (nth row 0))
               (string? (nth row 1))
               (string? (nth row 2))
               (string? (nth row 3)))
      (let [requested-stage (nth row 0)
            requested-key (nth row 1)
            fact-id (nth row 2)
            encoding (nth row 3)
            envelope (edn/read-string encoding)]
        (when (and (vector? envelope)
                   (= 9 (count envelope))
                   (= dev-fact-kind (nth envelope 0))
                   (= encoding (pr-str envelope))
                   (= fact-id (stages/content-digest encoding)))
          {:requested-stage requested-stage
           :requested-key requested-key
           :id fact-id
           :stage (nth envelope 1)
           :result-key (nth envelope 2)
           :compiler-context (nth envelope 3)
           :profile (nth envelope 4)
           :unit-id (nth envelope 5)
           :byte-count (nth envelope 6)
           :digest (nth envelope 7)
           :payload (nth envelope 8)})))
    (catch Exception _ nil)))

(defn rows-for-dev-request [rows stage result-key]
  (filterv
   (fn [row]
     (and (vector? row)
          (= 4 (count row))
          (= stage (nth row 0))
          (= result-key (nth row 1))))
   rows))

(defn dev-fact-metadata-valid?
  [fact stage result-key compiler-context profile unit-id]
  (and (some? fact)
       (= stage (:requested-stage fact))
       (= result-key (:requested-key fact))
       (= stage (:stage fact))
       (= result-key (:result-key fact))
       (= compiler-context (:compiler-context fact))
       (= profile (:profile fact))
       (= (dev-unit-id-text unit-id) (:unit-id fact))
       (integer? (:byte-count fact))
       (<= 0 (:byte-count fact))
       (string? (:digest fact))
       (string? (:payload fact))))

(defn select-typed-dev-fact
  [rows result-key compiler-context profile unit-id]
  (try
    (let [candidates (rows-for-dev-request rows "typed" result-key)]
      (cond
        (empty? candidates) {:status :miss}
        (not= 1 (count candidates)) {:status :divergence}
        :else
        (let [fact (parse-dev-fact-row (first candidates))]
          (if-not
           (dev-fact-metadata-valid?
            fact "typed" result-key compiler-context profile unit-id)
            {:status :divergence}
            (let [decoded
                  (unit/decode-typed-unit-wire-v1
                   (:payload fact) (:byte-count fact) (:digest fact) unit-id)]
              (if (instance? native.unit_reuse.TypedUnitWireDecodedV1 decoded)
                {:status :hit
                 :unit (unit/typedunitwiredecodedv1-unit decoded)}
                {:status :divergence}))))))
    (catch Exception _ {:status :divergence})))

(defn select-native-dev-fact
  [rows result-key compiler-context profile unit-id]
  (try
    (let [candidates (rows-for-dev-request rows "native" result-key)]
      (cond
        (empty? candidates) {:status :miss}
        (not= 1 (count candidates)) {:status :divergence}
        :else
        (let [fact (parse-dev-fact-row (first candidates))]
          (if-not
           (dev-fact-metadata-valid?
            fact "native" result-key compiler-context profile unit-id)
            {:status :divergence}
            (let [decoded
                  (unit/decode-native-unit-wire-v1
                   (:payload fact) (:byte-count fact) (:digest fact) unit-id)]
              (if (instance? native.unit_reuse.NativeUnitWireDecodedV1 decoded)
                {:status :hit
                 :unit (unit/nativeunitwiredecodedv1-unit decoded)}
                {:status :divergence}))))))
    (catch Exception _ {:status :divergence})))

(defn dev-read-contracts [prepared source-unit]
  (let [wanted
        (set (map dev-unit-id-text
                  (stages/sourceunitv0-read-set source-unit)))]
    (filterv
     (fn [contract]
       (contains? wanted
                  (dev-unit-id-text
                   (unit/unitcontractv0-unit-id contract))))
     (singleton/preparedcandidatev0-contracts prepared))))

(defn dev-typing-environment-hash [prepared source-unit]
  (stages/content-digest
   (stages/canonical-set
    "beagle-dev-unit-typing-environment-v1"
    (mapv unit/unitcontractv0-digest
          (dev-read-contracts prepared source-unit)))))

(defn dev-unit-receipts
  [source-units prepared profile-identity compiler-context
   materialization-receipt-id]
  (mapv
   (fn [source-unit]
     (unit/make-unit-derivation-receipt
      profile-identity
      source-unit
      (dev-read-contracts prepared source-unit)
      (singleton/preparedcandidatev0-semantic-contracts prepared)
      (dev-typing-environment-hash prepared source-unit)
      compiler-context
      materialization-receipt-id
      []))
   source-units))

(defn dev-typed-requests
  [source-units receipts compiler-context profile]
  (mapv
   (fn [source-unit receipt]
     ["typed"
      (unit/unit-result-key receipt)
      compiler-context
      profile
      (dev-unit-id-text (stages/sourceunitv0-id source-unit))])
   source-units receipts))

(defn compile-dev-typed-units
  [source-units receipts prepared rows compiler-context profile]
  (loop [position 0
         compiled []
         pending []
         counters (empty-dev-counters)]
    (if (>= position (count source-units))
      {:ok? true
       :units compiled
       :pending pending
       :counters counters}
      (let [source-unit (nth source-units position)
            unit-id (stages/sourceunitv0-id source-unit)
            receipt (nth receipts position)
            result-key (unit/unit-result-key receipt)
            selection
            (select-typed-dev-fact
             rows result-key compiler-context profile unit-id)
            status (:status selection)
            next-counters (add-dev-counter counters status)]
        (if (= :hit status)
          (recur (inc position)
                 (conj compiled (:unit selection))
                 pending
                 next-counters)
          (let [result
                (do
                  (log-dev-fallback!
                   status "typed" result-key unit-id)
                  (singleton/compile-typed-unit
                   prepared
                   unit-id
                   (stages/sourceunitv0-semantic-digest source-unit)
                   (stages/sourceunitv0-read-set source-unit)
                   (dev-read-contracts prepared source-unit)
                   receipt
                   result-key))]
            (if-not
             (instance? native.unit_compile.TypedUnitCompiledV0 result)
              {:ok? false
               :reason
               (if
                (instance?
                 native.unit_compile.TypedUnitCompileRejectedV0 result)
                 (str
                  (singleton/typedunitcompilerejectedv0-code result)
                  ": "
                  (singleton/typedunitcompilerejectedv0-detail result))
                 "typed unit compiler returned an unknown result")
               :counters next-counters}
              (let [produced
                    (singleton/typedunitcompiledv0-unit result)
                    next-pending
                    (if (= :miss status)
                      (conj
                       pending
                       (dev-fact-entry
                        "typed" result-key compiler-context profile unit-id
                        (unit/typedunitv0-encoding produced)))
                      pending)]
                (recur (inc position)
                       (conj compiled produced)
                       next-pending
                       next-counters)))))))))

(defn dev-materialization-closure-digest [typed-units abi]
  (let [types (lower/append-type-def
               (unit/collect-types typed-units)
               (lower/union-tag-type))
        resolutions (lower/resolve-layouts types abi)
        layouts (lower/collect-layouts resolutions)]
    (lower/layout-digest layouts)))

(defn dev-native-result-key
  [receipt materialization-closure-digest]
  (stages/content-digest
   (stages/canonical-record
    "beagle-dev-native-result-v1"
    [(unit/unit-result-key receipt)
     materialization-closure-digest])))

(defn dev-native-requests
  [source-units receipts materialization-closure-digest compiler-context profile]
  (mapv
   (fn [source-unit receipt]
     ["native"
      (dev-native-result-key receipt materialization-closure-digest)
      compiler-context
      profile
      (dev-unit-id-text (stages/sourceunitv0-id source-unit))])
   source-units receipts))

(defn dev-typed-unit-for [typed-units unit-id]
  (some
   (fn [candidate]
     (when
      (core/native-id=
       unit-id (unit/typedunitv0-unit-id candidate))
       candidate))
   typed-units))

(defn compile-dev-native-units
  [source-units receipts prepared typed-units materialization-closure-digest
   rows compiler-context profile abi]
  (loop [position 0
         compiled []
         pending []
         counters (empty-dev-counters)]
    (if (>= position (count source-units))
      {:ok? true
       :units compiled
       :pending pending
       :counters counters}
      (let [source-unit (nth source-units position)
            unit-id (stages/sourceunitv0-id source-unit)
            receipt (nth receipts position)
            fact-key
            (dev-native-result-key receipt materialization-closure-digest)
            selection
            (select-native-dev-fact
             rows fact-key compiler-context profile unit-id)
            status (:status selection)
            next-counters (add-dev-counter counters status)]
        (if (= :hit status)
          (recur (inc position)
                 (conj compiled (:unit selection))
                 pending
                 next-counters)
          (let [base-key (unit/unit-result-key receipt)
                target (dev-typed-unit-for typed-units unit-id)
                result
                (do
                  (log-dev-fallback!
                   status "native" fact-key unit-id)
                  (singleton/compile-native-unit
                   prepared unit-id target typed-units
                   receipt base-key abi))]
            (if-not
             (instance? native.unit_compile.NativeUnitCompiledV0 result)
              {:ok? false
               :reason
               (if
                (instance?
                 native.unit_compile.NativeUnitCompileRejectedV0 result)
                 (str
                  (singleton/nativeunitcompilerejectedv0-code result)
                  ": "
                  (singleton/nativeunitcompilerejectedv0-detail result))
                 "native unit compiler returned an unknown result")
               :counters next-counters}
              (let [produced
                    (singleton/nativeunitcompiledv0-unit result)
                    next-pending
                    (if (= :miss status)
                      (conj
                       pending
                       (dev-fact-entry
                        "native" fact-key compiler-context profile unit-id
                        (unit/nativeunitv0-encoding produced)))
                      pending)]
                (recur (inc position)
                       (conj compiled produced)
                       next-pending
                       next-counters)))))))))

(defn dev-typing-result
  [assembly frozen-source compiler-commit configuration]
  (let [frozen (unit/unitassemblyv0-typed assembly)
        slice (unit/unitassemblyv0-typed-slice assembly)
        digest (stages/frozentypedstagev0-digest frozen)
        stage (stages/frozentypedstagev0-stage frozen)
        graph (stages/typedstagev0-terms stage)
        functions (lower/typedslicev0-functions slice)
        clean (lower/term-graph-clean? graph)
        obligations
        (vec
         (concat
          (lower/type-closure-obligations digest true clean)
          (lower/function-todo-obligations functions digest)))
        receipt
        (lower/make-receipt
         (stages/frozensourcestagev1-digest frozen-source)
         compiler-commit
         "source-to-typed"
         configuration
         []
         obligations
         digest)]
    (lower/->TypingAcceptedV0 frozen slice receipt)))

(defn dev-native-result
  [assembly compiler-commit configuration]
  (let [frozen (unit/unitassemblyv0-native assembly)
        digest (stages/frozennativestagev0-digest frozen)
        stage (stages/frozennativestagev0-stage frozen)
        clean (lower/term-graph-clean?
               (stages/nativestagev0-terms stage))
        receipt
        (lower/make-receipt
         (stages/nativestagev0-typed-digest stage)
         compiler-commit
         "typed-to-native"
         configuration
         []
         (lower/native-receipt-obligations
          digest true true true true clean true true true)
         digest)]
    (lower/->NativeLoweringCompleteV0 frozen receipt)))

(defn dev-compile-attempt
  [frozen-source compiler-commit configuration abi]
  (let [source
        (stages/frozensourcestagev1-stage frozen-source)
        source-units (stages/sourcestagev1-units source)
        compiler-rule-epoch
        (or (System/getenv "BEAGLE_CORE_UNIT_RULE_EPOCH")
            (throw
             (ex-info "development unit rule identity is unavailable" {})))
        profile-identity (unit/core-profile-identity-v1)
        semantic-contracts
        (unit/five-form-semantic-contracts profile-identity)
        materialization-receipt-id
        (stages/content-digest
         (stages/canonical-record
          "beagle-dev-unit-materialization-v1"
          [dev-fact-profile
           (str "abi=" (System/getenv "BEAGLE_CORE_ABI"))]))
        preparation
        (singleton/prepare-unit-compilation
         frozen-source compiler-rule-epoch semantic-contracts configuration)]
    (if-not
     (instance?
      native.unit_compile.UnitPreparationAcceptedV0 preparation)
      {:ok? false
       :reason "candidate is outside the admitted singleton unit subset"
       :counters (empty-dev-counters)}
      (let [prepared
            (singleton/unitpreparationacceptedv0-prepared preparation)
            receipts
            (dev-unit-receipts
             source-units prepared profile-identity compiler-rule-epoch
             materialization-receipt-id)
            store (System/getenv "BEAGLE_CORE_DEV_FACT_STORE")
            typed-query
            (query-dev-facts
             store
             (dev-typed-requests
              source-units receipts compiler-rule-epoch dev-fact-profile))
            typed-pass
            (compile-dev-typed-units
             source-units receipts prepared (:rows typed-query)
             compiler-rule-epoch dev-fact-profile)]
        (if-not (:ok? typed-pass)
          {:ok? false
           :reason (:reason typed-pass)
           :counters (:counters typed-pass)}
          (let [typed-units (:units typed-pass)
                materialization-closure-digest
                (dev-materialization-closure-digest typed-units abi)
                native-query
                (query-dev-facts
                 store
                 (dev-native-requests
                  source-units receipts materialization-closure-digest
                  compiler-rule-epoch dev-fact-profile))
                native-pass
                (compile-dev-native-units
                 source-units receipts prepared typed-units
                 materialization-closure-digest (:rows native-query)
                 compiler-rule-epoch dev-fact-profile abi)
                counters
                (merge-dev-counters
                 (:counters typed-pass)
                 (:counters native-pass))]
            (if-not (:ok? native-pass)
              {:ok? false
               :reason (:reason native-pass)
               :counters counters}
              (let [assembly-result
                    (unit/assemble-unit-payloads
                     frozen-source
                     (singleton/preparedcandidatev0-contracts prepared)
                     typed-units
                     (:units native-pass)
                     compiler-commit
                     configuration
                     abi)]
                (if-not
                 (instance?
                  native.unit_reuse.UnitAssemblyAcceptedV0 assembly-result)
                  {:ok? false
                   :reason
                   (if
                    (instance?
                     native.unit_reuse.UnitAssemblyRejectedV0 assembly-result)
                     (str
                      (unit/unitassemblyrejectedv0-code assembly-result)
                      ": "
                      (unit/unitassemblyrejectedv0-detail assembly-result))
                     "unit assembly returned an unknown result")
                   :counters (add-dev-counter counters :divergence)}
                  (let [assembly
                        (unit/unitassemblyacceptedv0-assembly assembly-result)
                        typed-stage-result
                        (dev-typing-result
                         assembly frozen-source compiler-commit configuration)
                        native-stage-result
                        (dev-native-result
                         assembly compiler-commit configuration)
                        epoch-result
                        (lower/epoch-derived-stage
                         (unit/unitassemblyv0-native assembly)
                         compiler-commit configuration abi)
                        epoch-match?
                        (=
                         (stages/frozennativestagev0-encoding
                          (unit/unitassemblyv0-epoch assembly))
                         (stages/frozennativestagev0-encoding
                          (lower/epoch-result-frozen epoch-result)))]
                    (if-not epoch-match?
                      {:ok? false
                       :reason "assembled epoch differs from cold epoch derivation"
                       :counters (add-dev-counter counters :divergence)}
                      (let [pending
                            (vec
                             (concat
                              (:pending typed-pass)
                              (:pending native-pass)))
                            maintenance
                            (if
                             (and (:usable? typed-query)
                                  (:usable? native-query))
                              (append-dev-facts store pending)
                              "UNAVAILABLE")]
                        {:ok? true
                         :typing-result typed-stage-result
                         :native-result native-stage-result
                         :epoch-result epoch-result
                         :counters counters
                         :maintenance maintenance}))))))))))))

(defn dev-fact-report [attempt]
  (let [counters (:counters attempt)
        mode
        (cond
          (not (:ok? attempt)) "COLD-FALLBACK"
          (pos? (:divergences counters)) "MIXED-DIVERGENCE"
          (pos? (:hits counters)) "FACT-REUSE"
          :else "COLD-POPULATE")]
    (str "dev-facts mode=" mode
         " hits=" (:hits counters)
         " misses=" (:misses counters)
         " divergences=" (:divergences counters)
         " maintenance=" (or (:maintenance attempt) "NONE")
         (if (:ok? attempt) "" (str " reason=" (:reason attempt)))
         "\n")))

(defn compile-core!
  [facts-manifest-path projected-facts-path artifacts-dir compiler-commit entries
   materializers abi-id abi simd-required?]
  (start-stage! "source-projection")
  (let [raw-rows (slice/read-fact-manifest facts-manifest-path)
        cache-source-rows
        (mapv (fn [row]
                [(slice/slicefactv0-subject row)
                 (slice/slicefactv0-predicate row)
                 (slice/slicefactv0-kind row)
                 (slice/slicefactv0-object row)])
              raw-rows)
        strict-entry-abi? (boolean (some #{"wasm"} materializers))
        cache-admission
        (semantic-cache/admit-and-identify-query
         (semantic-cache/->QueryAdmissionRequest
          cache-source-rows
          (System/getenv "BEAGLE_CORE_COMPILER_PROJECTION_ID")
          (System/getenv "BEAGLE_CORE_SEMANTIC_READ_RULES_ID")
          entries strict-entry-abi?))
        cache-result
        (if (semantic-cache/query-admitted? cache-admission)
          (semantic-cache/query!
           (System/getenv "BEAGLE_CORE_SEMANTIC_READ_STORE")
           (semantic-cache/admitted-query cache-admission))
          cache-admission)
        cache-rejection
        (cond
          (semantic-cache/query-rejected? cache-admission) cache-admission
          (semantic-cache/result-rejected? cache-result) cache-result
          :else nil)
        cached-projected-facts
        (when (semantic-cache/result-admitted? cache-result)
          (semantic-cache/admitted-result-payload cache-result))
        projection
        (when (semantic-cache/result-missing? cache-result)
          (project-entry-rows raw-rows entries strict-entry-abi?))
        projected-rows
        (when (and (semantic-cache/result-missing? cache-result)
                   (:ok? projection))
          (semantic-read-set-rows
           (:rows projection) (:index projection) (:selected projection)
           (:ordered-definitions projection)
           (:dependencies-by-id projection)))]
    (binding [*out* *err*]
      (println (str "beagle build: semantic-read-cache "
                    (cond
                      cache-rejection "REJECTED "
                      (semantic-cache/result-missing? cache-result) "MISS "
                      :else "HIT ")
                    (semantic-cache/admitted-query-digest cache-admission))))
    (cond
      cache-rejection
      (do
        (finish-stage! "source-projection" "REJECTED")
        {:ok? false
         :report (str "stage source-projection REJECTED\n"
                      "semantic-read-cache "
                      (semantic-cache/rejection-detail cache-rejection) "\n"
                      "result FAIL semantic-read cache\n")})
      (and (semantic-cache/result-missing? cache-result) (not (:ok? projection)))
      (do
        (finish-stage! "source-projection" "REJECTED")
        {:ok? false
         :report (str "stage source-projection REJECTED\n"
                      "entry-error " (:detail projection) "\n"
                      "result FAIL entry projection\n")})
      :else
      (let [projected-facts
            (or cached-projected-facts (rows-text projected-rows))
            cache-publish
            (when (semantic-cache/result-missing? cache-result)
              (semantic-cache/append!
               (System/getenv "BEAGLE_CORE_SEMANTIC_READ_STORE")
               (semantic-cache/admitted-query cache-admission) projected-facts))]
        (if (semantic-cache/result-rejected? cache-publish)
          (do
            (finish-stage! "source-projection" "REJECTED")
            {:ok? false
             :report (str "stage source-projection REJECTED\n"
                          "semantic-read-cache "
                          (semantic-cache/rejection-detail cache-publish) "\n"
                          "result FAIL semantic-read cache append\n")})
          (let [rows (or projected-rows (slice/parse-facts projected-facts))
                _ (spit projected-facts-path projected-facts)
                configuration ["profile=3" (str "abi=" abi-id)
                               (str "source-facts-sha256="
                                    (subs (stages/content-digest projected-facts) 7))]
            _finish-projection (finish-stage! "source-projection" "ACCEPTED")
            _start-freeze (start-stage! "source-freeze")
            source (observed-source-program
                    rows "beagle.core" "source.facts")
            freeze-result (observed-freeze-source-stage
                           source compiler-commit configuration)]
        (if-not (instance? native.lower.SourceFreezeAcceptedV0 freeze-result)
          (do
            (finish-stage! "source-freeze" "REJECTED")
            {:ok? false
             :report "stage source-freeze REJECTED\nresult FAIL source freeze\n"})
          (let [_finish-freeze (finish-stage! "source-freeze" "ACCEPTED")
                frozen-source (lower/sourcefreezeacceptedv0-frozen freeze-result)
                source-receipt (lower/sourcefreezeacceptedv0-receipt freeze-result)
                dev-attempt
                (when (dev-fact-enabled?)
                  (dev-compile-attempt
                   frozen-source compiler-commit configuration abi))
                dev-report
                (if (some? dev-attempt) (dev-fact-report dev-attempt) "")
                _write-dev-report
                (when (some? dev-attempt)
                  (binding [*out* *err*]
                    (print (str "beagle build: " dev-report))))
                _start-typing (start-stage! "source-to-typed")
                typing-result
                (if (and (some? dev-attempt) (:ok? dev-attempt))
                  (do
                    (publish-stage-work!
                     "source-to-typed" "dev-fact-reuse" "ACCEPTED" 1 1)
                    (:typing-result dev-attempt))
                  (observed-lower-typed-stage
                   frozen-source compiler-commit configuration))]
            (if-not (instance? native.lower.TypingAcceptedV0 typing-result)
              (do
                (finish-stage! "source-to-typed" "REJECTED")
                {:ok? false
                 :report (str (body-slice/typing-rejection-report typing-result)
                              "result FAIL source typing\n")})
              (let [_finish-typing (finish-stage! "source-to-typed" "ACCEPTED")
                    frozen-typed (lower/typingacceptedv0-frozen typing-result)
                    typed-slice (lower/typingacceptedv0-slice typing-result)
                    _start-lowering (start-stage! "typed-to-native")
                    native-result
                    (if (and (some? dev-attempt) (:ok? dev-attempt))
                      (:native-result dev-attempt)
                      (lower/lower-native-stage
                       frozen-typed typed-slice
                       compiler-commit configuration abi))]
                (if-not (instance? native.lower.NativeLoweringCompleteV0 native-result)
                  (do
                    (finish-stage! "typed-to-native" "PENDING")
                    {:ok? false
                     :report (str "stage source-freeze ACCEPTED\n"
                                  "stage source-to-typed ACCEPTED\n"
                                  "stage typed-to-native PENDING\n"
                                  (slice/pending-reports (slice/native-pending native-result))
                                  "result FAIL NativeLoweringCompleteV0 required\n")})
                  (let [_finish-lowering (finish-stage! "typed-to-native" "COMPLETE")
                        native-frozen (lower/nativeloweringcompletev0-frozen native-result)
                        _start-epoch (start-stage! "native-to-epoch")
                        epoch-result
                        (if (and (some? dev-attempt) (:ok? dev-attempt))
                          (:epoch-result dev-attempt)
                          (lower/epoch-derived-stage
                           native-frozen compiler-commit configuration abi))
                        epoch-complete? (lower/epoch-result-complete? epoch-result)
                        epoch-pendings (epoch-pending-details epoch-result)
                        _finish-epoch (finish-stage!
                                       "native-to-epoch"
                                       (if epoch-complete? "COMPLETE" "PENDING"))
                        ;; The seam every materializer crosses: what a
                        ;; materializer sees is what the epoch stage froze.
                        frozen-native (lower/epoch-result-frozen epoch-result)
                        typing-receipt (lower/typingacceptedv0-receipt typing-result)
                        native-receipt (lower/nativeloweringcompletev0-receipt
                                        native-result)
                        epoch-receipt (epoch-result-receipt epoch-result)
                        epoch-digest (stages/frozennativestagev0-digest frozen-native)
                        native-encoding (stages/frozennativestagev0-encoding frozen-native)
                        _write-native
                        (spit (str artifacts-dir "/module.native-program") native-encoding)
                        program (stages/nativestagev0-program
                                 (stages/frozennativestagev0-stage frozen-native))
                        native-program (stages/nativestagev0-program
                                        (stages/frozennativestagev0-stage native-frozen))
                        ;; QBE opens no minted epoch, so it is handed the
                        ;; identity program rather than the derived one it
                        ;; would refuse.
                        qbe-epoch-result
                        (lower/epoch-identity-stage native-frozen compiler-commit
                                                    configuration abi)
                        qbe-frozen (lower/epoch-result-frozen qbe-epoch-result)
                        qbe-program (stages/nativestagev0-program
                                      (stages/frozennativestagev0-stage qbe-frozen))
                        projected (body-slice/projected-program program)
                        qbe-projected (body-slice/projected-program qbe-program)
                        simd-plan (simd/derive-plan projected
                                    (stages/frozennativestagev0-digest frozen-native)
                                    abi-id)
                        _write-simd-plan
                        (when simd-required?
                          (spit (str artifacts-dir "/module.simd-plan-v0")
                            (simd/simdplanv0-encoding simd-plan)))
                        _start-obligations (start-stage! "native-obligations")
                        verdicts
                        (ordered-parallel-mapv
                          (fn [validate] (validate))
                          [(fn [] (obligations/valid-ssa projected))
                           (fn [] (obligations/exhaustive-matches projected))
                           (fn [] (obligations/closed-layouts projected abi))
                           (fn [] (obligations/checked-arithmetic projected))
                           (fn [] (obligations/legal-abi projected))
                           (fn [] (obligations/discharged-tokens projected))
                           (fn [] (obligations/bounded-effects projected))
                           (fn [] (obligations/epoch-soundness projected))
                           (fn [] (obligations/leak-freedom projected))
                           (fn [] (obligations/deterministic-parallelism projected))])
                        head (str
                              (native-report-head rows entries source program projected
                                                  native-program epoch-result abi verdicts)
                              dev-report
                              (native-provenance-report
                               source-receipt frozen-source typing-result frozen-typed
                               native-result native-frozen epoch-result frozen-native
                               compiler-commit configuration))
                        _write-receipts
                        (write-native-receipts!
                         artifacts-dir source-receipt typing-receipt native-receipt
                         epoch-receipt)
                        _write-entry-map
                        (spit (str artifacts-dir "/native.entry-map")
                          (str (apply str
                                 (map (fn [entry] (str "source-entry " entry "\n"))
                                   entries))
                               (body-slice/function-report
                                (core/nativecoreprogram-functions projected))))]
                    (cond
                      ;; Pending with no named pending row is the epoch stage
                      ;; refusing its own ten obligations, never a frontier.
                      (and (not epoch-complete?) (empty? epoch-pendings))
                      (do
                        (finish-stage! "native-obligations" "FAIL")
                        {:ok? false
                         :report (str head
                                      "result FAIL epoch stage obligations required\n")})

                      (not (obligation-pass? verdicts))
                      (do
                        (finish-stage! "native-obligations" "FAIL")
                        {:ok? false
                         :report (str head
                                      "result FAIL ten native obligations required\n")})

                      :else
                      (do
                        (finish-stage! "native-obligations" "PASS")
                        (let [checkpoint-head
                              (str head
                                   (if simd-required?
                                     (simd/plan-report simd-plan)
                                     ""))
                              _checkpoint
                              (stage-core-checkpoint! frozen-native qbe-frozen
                                                      checkpoint-head)
                              materialization
                              (materialize-programs! projected qbe-projected
                                                     artifacts-dir materializers
                                                     abi-id simd-plan
                                                     simd-required? epoch-digest
                                                     compiler-commit
                                                     configuration)]
                          (assoc materialization
                                 :report (str checkpoint-head
                                           (:report materialization))))))))))))))))))

(try
  (let [facts-manifest-path (System/getenv "BEAGLE_CORE_FACTS_MANIFEST")
        projected-facts-path (System/getenv "BEAGLE_CORE_PROJECTED_FACTS")
        artifacts-dir (System/getenv "BEAGLE_CORE_ARTIFACTS")
        compiler-commit (System/getenv "BEAGLE_CORE_COMMIT")
        materializers (vec (remove str/blank?
                                   (str/split-lines
                                    (slurp (System/getenv
                                            "BEAGLE_CORE_MATERIALIZERS")))))
        entries (vec (remove str/blank?
                             (str/split-lines
                              (slurp (System/getenv "BEAGLE_CORE_ENTRIES")))))
        abi-id (or (System/getenv "BEAGLE_CORE_ABI") "lp64")
        simd-required? (= "1" (System/getenv "BEAGLE_CORE_SIMD_REQUIRED"))
        checkpoint-wire-path (System/getenv "BEAGLE_CORE_CHECKPOINT_WIRE")
        checkpoint-qbe-wire-path
        (System/getenv "BEAGLE_CORE_CHECKPOINT_QBE_WIRE")
        abi (core/abi-profile-for abi-id)
        result (if (nil? abi)
                 {:ok? false
                  :report (str "abi-error unknown abi profile " abi-id "\n"
                               "result FAIL unknown abi profile\n")}
                 (if (str/blank? checkpoint-wire-path)
                   (compile-core! facts-manifest-path projected-facts-path artifacts-dir
                                  compiler-commit entries materializers abi-id abi
                                  simd-required?)
                   (resume-core!
                    checkpoint-wire-path
                    checkpoint-qbe-wire-path
                    (System/getenv "BEAGLE_CORE_CHECKPOINT_REPORT_HEAD")
                    (System/getenv "BEAGLE_CORE_CHECKPOINT_STAGE_PROGRESS")
                    artifacts-dir compiler-commit materializers abi-id abi
                    simd-required?)))
        final-report
        (report-with-stage-progress
         (str native-report-format "\n" (:report result)))]
    (durable-write! report-path final-report)
    (spit (str artifacts-dir "/report.txt") final-report)
    (print final-report)
    (flush)
    (when-not (:ok? result) (System/exit 1)))
  (catch Throwable error
    (fail-active-stage!)
    (throw error)))
' >"$work/run.log" 2>&1
runner_rc=$?
set -e

if [[ -f "$work/run.log" ]]; then
    rg '^beagle build: core-stage ' "$work/run.log" >&2 || true
fi

if [[ "$dev_fact_reuse" == 1 && -f "$work/run.log" ]]; then
    # The hosted Core runner is captured for deterministic artifact handling;
    # surface its bounded fact evidence on the outer build log as well.
    rg '^beagle build: dev-facts ' "$work/run.log" >&2 || true
fi

if [[ "$core_result_cache_enabled" == 1 && "$core_checkpoint_eligible" == 1 &&
      "$core_checkpoint_cache_hit" != 1 &&
      -f "$work/core-checkpoint-stage/frozen-native-stage.wire-v1" &&
      -f "$work/core-checkpoint-stage/qbe-frozen-native-stage.wire-v1" ]]; then
    acquire_core_result_lock core-checkpoint-publish-lock
    validate_core_result_input_closure "$core_result_input" \
        "$work/core-checkpoint-input.before-publish" ||
        die "Core checkpoint input closure changed before publication"
    write_core_checkpoint_identity \
        "$work/core-checkpoint-input.before-publish" \
        "$work/core-checkpoint-identity.before-publish"
    cmp -s "$core_checkpoint_input" \
        "$work/core-checkpoint-identity.before-publish" ||
        die "Core checkpoint identity changed before publication"
    for checkpoint_artifact in source.facts module.native-program \
                               native.receipts native.entry-map; do
        [[ -f "$work/artifacts/$checkpoint_artifact" ]] ||
            die "Core checkpoint stage omitted $checkpoint_artifact"
    done
    if [[ "$simd_required" == 1 ]]; then
        [[ -f "$work/artifacts/module.simd-plan-v0" ]] ||
            die "Core checkpoint stage omitted module.simd-plan-v0"
    fi
    for checkpoint_internal in frozen-native-stage.wire-v1 \
                               qbe-frozen-native-stage.wire-v1 report.head \
                               stage-progress; do
        [[ -f "$work/core-checkpoint-stage/$checkpoint_internal" ]] ||
            die "Core checkpoint stage omitted $checkpoint_internal"
    done

    core_checkpoint_staging="$(mktemp -d \
        "$cache_root/.checkpoint-tmp/$core_checkpoint_key.XXXXXX")"
    cp -- "$core_checkpoint_input" "$core_checkpoint_staging/input.manifest"
    mkdir "$core_checkpoint_staging/artifacts"
    cp -- "$work/core-checkpoint-stage/frozen-native-stage.wire-v1" \
        "$core_checkpoint_staging/artifacts/frozen-native-stage.wire-v1"
    cp -- "$work/core-checkpoint-stage/qbe-frozen-native-stage.wire-v1" \
        "$core_checkpoint_staging/artifacts/qbe-frozen-native-stage.wire-v1"
    cp -- "$work/core-checkpoint-stage/report.head" \
        "$core_checkpoint_staging/artifacts/report.head"
    cp -- "$work/core-checkpoint-stage/stage-progress" \
        "$core_checkpoint_staging/artifacts/stage-progress"
    for checkpoint_artifact in source.facts module.native-program \
                               native.receipts native.entry-map; do
        cp -- "$work/artifacts/$checkpoint_artifact" \
            "$core_checkpoint_staging/artifacts/$checkpoint_artifact"
    done
    if [[ "$simd_required" == 1 ]]; then
        cp -- "$work/artifacts/module.simd-plan-v0" \
            "$core_checkpoint_staging/artifacts/module.simd-plan-v0"
    fi
    write_core_result_artifact_manifest "$core_checkpoint_staging/artifacts" \
        "$core_checkpoint_staging/artifacts.sha256"
    checkpoint_artifact_manifest_digest="$(sha256sum \
        "$core_checkpoint_staging/artifacts.sha256" | awk '{print $1}')"
    printf 'beagle-core-pre-materializer/v1 %s %s\n' "$core_checkpoint_key" \
        "$checkpoint_artifact_manifest_digest" >"$core_checkpoint_staging/READY"
    validate_core_checkpoint_entry "$core_checkpoint_staging" \
        "$core_checkpoint_input" "$core_checkpoint_key" ||
        die "new Core checkpoint failed structural validation"
    if ! validate_core_checkpoint_wire "$core_checkpoint_staging" \
        "$work/core-checkpoint-new-wire.log"; then
        die "new Core checkpoint failed wire validation"
    fi
    sync -f "$core_checkpoint_staging"
    core_checkpoint_entry="$core_checkpoint_root/$core_checkpoint_key"
    # A peer that built the identical checkpoint while this build compiled has
    # already published it under this same content key. Adopting its entry is
    # exactly a cache hit: the key IS the digest of the input manifest, and the
    # entry is put through the same full structural and wire validation a
    # lookup applies before it is trusted.
    if [[ -e "$core_checkpoint_entry" ]]; then
        validate_core_checkpoint_entry "$core_checkpoint_entry" \
            "$core_checkpoint_input" "$core_checkpoint_key" ||
            die "Core checkpoint appeared under its publication lock and did not validate"
        if ! validate_core_checkpoint_wire "$core_checkpoint_entry" \
            "$work/core-checkpoint-peer-wire.log"; then
            die "Core checkpoint appeared under its publication lock and did not decode"
        fi
        rm -rf -- "${core_checkpoint_staging:?}"
        core_checkpoint_staging=""
        echo "beagle build: core-checkpoint PUBLISHED-BY-PEER $core_checkpoint_key" >&2
    else
        mv "$core_checkpoint_staging" "$core_checkpoint_entry"
        core_checkpoint_staging=""
        sync -f "$core_checkpoint_root"
        echo "beagle build: core-checkpoint PUBLISHED $core_checkpoint_key" >&2
    fi
    record_core_checkpoint_wire_attestation "$core_checkpoint_entry" ||
        die "failed to record the published Core checkpoint wire PASS"

    validate_core_result_early_closure "$core_result_early_input" \
        "$work/core-result-early-input.before-checkpoint-alias-publish" ||
        die "Core checkpoint early input closure changed before alias publication"
    write_core_checkpoint_identity \
        "$work/core-result-early-input.before-checkpoint-alias-publish" \
        "$work/core-checkpoint-early-input.before-alias-publish"
    cmp -s "$core_checkpoint_early_input" \
        "$work/core-checkpoint-early-input.before-alias-publish" ||
        die "Core checkpoint early identity changed before alias publication"
    checkpoint_input_digest="$(sha256sum "$core_checkpoint_input" | awk '{print $1}')"
    validate_core_checkpoint_entry "$core_checkpoint_entry" \
        "$core_checkpoint_input" "$core_checkpoint_key" ||
        die "published Core checkpoint failed alias validation"
    exec {core_checkpoint_alias_publish_lock_fd}>\
        "$cache_root/.checkpoint-alias-locks/$core_checkpoint_early_key.lock"
    run_phase_leaf core-checkpoint-alias-publish-lock \
        "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        flock -x -w "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        "$core_checkpoint_alias_publish_lock_fd" ||
        die "timed out acquiring the Core checkpoint alias publication lock"
    core_checkpoint_alias_staging="$(mktemp \
        "$cache_root/.checkpoint-tmp/$core_checkpoint_early_key.alias.XXXXXX")"
    printf 'beagle-core-pre-materializer-alias/v1 %s %s %s\n' \
        "$core_checkpoint_early_key" "$core_checkpoint_key" "$checkpoint_input_digest" \
        >"$core_checkpoint_alias_staging"
    sync -f "$core_checkpoint_alias_staging"
    if { [[ -e "$core_checkpoint_alias" ]] || [[ -L "$core_checkpoint_alias" ]]; } &&
       { [[ ! -f "$core_checkpoint_alias" ]] ||
         [[ -L "$core_checkpoint_alias" ]] ||
         ! cmp -s "$core_checkpoint_alias_staging" "$core_checkpoint_alias"; }; then
        stale_checkpoint_alias=\
"$cache_root/.checkpoint-tmp/replaced-alias.$core_checkpoint_early_key.$$"
        echo "beagle build: core-checkpoint-alias STALE $core_checkpoint_early_key; replacing" >&2
        mv "$core_checkpoint_alias" "$stale_checkpoint_alias"
        rm -rf -- "${stale_checkpoint_alias:?}"
    fi
    if [[ ! -e "$core_checkpoint_alias" ]]; then
        mv "$core_checkpoint_alias_staging" "$core_checkpoint_alias"
        core_checkpoint_alias_staging=""
        sync -f "$core_checkpoint_alias_root"
    else
        rm -f -- "$core_checkpoint_alias_staging"
        core_checkpoint_alias_staging=""
    fi
    flock -u "$core_checkpoint_alias_publish_lock_fd"
    release_core_result_lock
fi

if [[ $runner_rc -ne 0 ]]; then
    if [[ -f "$work/artifacts/report.txt" ]]; then
        cat "$work/artifacts/report.txt" >&2
    else
        sed -n '1,200p' "$work/run.log" >&2
        if [[ -f "$BEAGLE_CORE_REPORT" ]]; then
            cat -- "$BEAGLE_CORE_REPORT" >&2
        fi
    fi
    exit "$runner_rc"
fi

for required in source.facts report.txt module.native-program native.receipts \
                native.entry-map; do
    [[ -f "$work/artifacts/$required" ]] ||
        die "successful Core runner omitted $required"
done
if $wasm_selected || [[ -n "${seen_materializers[c17]:-}" ]]; then
    for required in module_0.h module_0.c; do
        [[ -f "$work/artifacts/$required" ]] ||
            die "successful Core runner omitted $required"
    done
fi

fi

if [[ "$core_result_cache_enabled" == 1 ]]; then
    if [[ "$core_result_early_hit" == 1 ||
          "$core_checkpoint_early_hit" == 1 ]]; then
        validate_core_result_early_closure "$core_result_early_input" \
            "$work/core-result-early-input.before-finalize" ||
            die "Core result early input closure changed before finalization"
    else
        validate_core_result_input_closure "$core_result_input" \
            "$work/core-result-input.before-finalize" ||
            die "Core result input closure changed before finalization"
    fi
    if [[ "$core_result_cache_hit" != 1 ]]; then
        mkdir "$work/core-result-artifacts"
        cp -a "$work/artifacts/." "$work/core-result-artifacts/"
        core_result_publish_pending=1
    fi
fi

# A served result entry carries exactly the bytes that passed this check when
# the entry was published, and the entry's own artifacts.sha256 was re-checked
# file by file during lookup. Re-deriving the receipt index from identical
# bytes cannot reach a different verdict, and its output is never read again.
if [[ "$core_result_cache_hit" == 1 ]]; then
    echo "beagle build: phase native-receipts SKIP (validated at cache publication)" >&2
else
    run_phase_leaf native-receipts "${BEAGLE_CORE_VALIDATION_TIMEOUT_SECONDS:-30}" \
        bb -cp "$compiled" "$FINALIZER" native-index "$work/native.receipts.index" \
        "$work/artifacts/native.receipts" "$work/artifacts/module.native-program" \
        >"$work/native-receipts.log" 2>&1 || {
            sed -n '1,120p' "$work/native-receipts.log" >&2
            die "canonical Native Core receipt validation failed"
        }
fi
native_digest="$(sha256sum "$work/artifacts/module.native-program" | awk '{print $1}')"
printf '%s\n' "$native_digest" \
    >"$work/artifacts/module.native-program.sha256"
if [[ $simd_required -eq 1 ]]; then
    [[ -f "$work/artifacts/module.simd-plan-v0" ]] ||
        die "successful SIMD runner omitted module.simd-plan-v0"
    sha256sum "$work/artifacts/module.simd-plan-v0" | awk '{print $1}' \
        >"$work/artifacts/module.simd-plan-v0.sha256"
fi

source_facts_digest="$(sha256sum "$work/artifacts/source.facts" | awk '{print $1}')"
report_with_files="$work/artifacts/report.with-file-provenance.txt"
sed '$d' "$work/artifacts/report.txt" >"$report_with_files"
printf '%s\n' \
    "source-facts-sha256 $source_facts_digest" \
    "native-program-file-sha256 $native_digest" \
    "native-receipts-file-sha256 $(sha256sum "$work/artifacts/native.receipts" | awk '{print $1}')" \
    >>"$report_with_files"
tail -n 1 "$work/artifacts/report.txt" >>"$report_with_files"
mv -f -- "$report_with_files" "$work/artifacts/report.txt"

merge_wasm_report() {
    local final_result="$1"
    local merged="$work/artifacts/report.with-wasm.txt"
    sed '$d' "$work/artifacts/report.txt" >"$merged"
    cat "$work/artifacts/wasm-report.txt" >>"$merged"
    printf 'result %s\n' "$final_result" >>"$merged"
    mv "$merged" "$work/artifacts/report.txt"
}

if $wasm_selected; then
    wasm_args=(--artifacts "$work/artifacts" --compiled "$compiled"
               "${wasm_checked_sources[@]}")
    for entry in "${entries[@]}"; do
        wasm_args+=(--entry "$entry")
    done
    materializer_timeout="${BEAGLE_WASM_MATERIALIZER_TIMEOUT_SECONDS:-180}"
    materializer_kill_grace="${BEAGLE_WASM_KILL_GRACE_SECONDS:-5}"
    [[ "$materializer_timeout" =~ ^[1-9][0-9]*$ ]] ||
        die "BEAGLE_WASM_MATERIALIZER_TIMEOUT_SECONDS must be a positive integer"
    [[ "$materializer_kill_grace" =~ ^[1-9][0-9]*$ ]] ||
        die "BEAGLE_WASM_KILL_GRACE_SECONDS must be a positive integer"
    set +e
    run_phase_with_grace wasm-materializer "$materializer_timeout" \
        "$materializer_kill_grace" \
        "$BIN/beagle-materialize-wasm" "${wasm_args[@]}"
    wasm_rc=$?
    set -e
    if [[ ! -f "$work/artifacts/wasm-report.txt" ]]; then
        printf '%s\n' \
            "wasm-materializer bootstrap-c17-wasi-clang" \
            "wasm-materializer-direct false" \
            "wasm-abi wasm32" \
            "wasm-result FAIL driver-error" \
            >"$work/artifacts/wasm-report.txt"
    fi
    if [[ $wasm_rc -ne 0 ]]; then
        merge_wasm_report "FAIL wasm materialization"
        cat "$work/artifacts/report.txt" >&2
        exit "$wasm_rc"
    fi
    merge_wasm_report "PASS"
fi

install="$work/install"
mkdir -p "$install"
declare -A staged_names=()
stage_file() {
    local source="$1"
    local name="$2"
    [[ "$name" =~ ^[A-Za-z0-9._-]+$ ]] || die "unsafe managed artifact name: $name"
    if [[ -n "${staged_names[$name]:-}" ]]; then
        cmp -s "$source" "$install/$name" ||
            die "two materializers produced different bytes for $name"
        return
    fi
    cp -- "$source" "$install/$name"
    staged_names["$name"]=1
}

for artifact in source.facts module.native-program module.native-program.sha256 \
                native.receipts native.entry-map; do
    stage_file "$work/artifacts/$artifact" "$artifact"
done
if [[ $simd_required -eq 1 ]]; then
    for artifact in module.simd-plan-v0 module.simd-plan-v0.sha256; do
        stage_file "$work/artifacts/$artifact" "$artifact"
    done
fi
receipt_names=(native.receipts)
for materializer in "${materializers[@]}"; do
    case "$materializer" in
      c17)
        for artifact in module_0.h module_0.c c17.receipt; do
            [[ -f "$work/artifacts/$artifact" ]] ||
                die "successful C17 projection omitted $artifact"
            stage_file "$work/artifacts/$artifact" "$artifact"
        done
        receipt_names+=(c17.receipt)
        stage_file "$BEAGLE_DIR/native-core/shim/native_shim.c" native_shim.c
        stage_file "$BEAGLE_DIR/native-core/shim/native_shim.h" native_shim.h
        if grep -q '^parallel-plan ' "$work/artifacts/report.txt"; then
            stage_file "$BEAGLE_DIR/native-core/shim/native_parallel.c" \
                native_parallel.c
            stage_file "$BEAGLE_DIR/native-core/shim/native_parallel.h" \
                native_parallel.h
        fi
        stage_file "$BEAGLE_DIR/native-core/shim/native_unicode15_data.h" \
            native_unicode15_data.h
        stage_file "$BEAGLE_DIR/native-core/shim/UNICODE-LICENSE.txt" \
            UNICODE-LICENSE.txt
        ;;
      qbe)
        [[ -f "$work/artifacts/module_0.ssa" ]] ||
            die "successful QBE projection omitted module_0.ssa"
        stage_file "$work/artifacts/module_0.ssa" module_0.ssa
        ;;
      wasm)
        for artifact in module_0.wasm module_0.wasm.sha256 \
                        module_0.wasm.seams wasm.receipt wasm-audit.txt \
                        wasm-report.txt c17.receipt module_0.h module_0.c \
                        native_shim.h native_shim.c native_unicode15_data.h \
                        wasm.retention.c wasm.adapter.c wasm.entry-contract.clj \
                        wasm.seams.clj wasm.ast-verifier.rkt \
                        wasm.receipt-finalizer.clj wasm.materializer.sh \
                        wasm.supervisor.rkt \
                        wasm.cc-identity.txt wasm.ld-identity.txt \
                        wasm.runtime-identity.txt; do
            [[ -f "$work/artifacts/$artifact" ]] ||
                die "successful Wasm bootstrap omitted $artifact"
            stage_file "$work/artifacts/$artifact" "$artifact"
        done
        published_wasm_digest="$(sed -n '1p' \
            "$work/artifacts/module_0.wasm.sha256")"
        actual_wasm_digest="$(sha256sum "$work/artifacts/module_0.wasm" | \
            awk '{print $1}')"
        [[ "$published_wasm_digest" == "$actual_wasm_digest" ]] ||
            die "successful Wasm artifact does not match its digest"
        receipt_names+=(c17.receipt wasm.receipt)
        stage_file "$BEAGLE_DIR/native-core/shim/UNICODE-LICENSE.txt" \
            UNICODE-LICENSE.txt
        ;;
    esac
done

for source_index in "${!source_snapshot_paths[@]}"; do
    [[ "$(sha256sum "${source_snapshot_paths[$source_index]}" | awk '{print $1}')" == \
       "${source_snapshot_digests[$source_index]}" ]] ||
        die "source changed before generation publication: ${source_snapshot_paths[$source_index]}"
done
stage_file "$work/artifacts/report.txt" report.txt

manifest_args=(manifest-verify-staged "$install/build.manifest" "$install")
declare -A receipt_seen=()
for receipt in "${receipt_names[@]}"; do
    [[ -z "${receipt_seen[$receipt]:-}" ]] || continue
    receipt_seen["$receipt"]=1
    manifest_args+=(--receipt "$receipt" "$install/$receipt")
done
while IFS= read -r staged_name; do
    [[ -n "${receipt_seen[$staged_name]:-}" ]] && continue
    manifest_args+=(--artifact "$staged_name" "$install/$staged_name")
done < <(printf '%s\n' "${!staged_names[@]}" | LC_ALL=C sort)
run_phase_leaf staged-manifest "${BEAGLE_CORE_VALIDATION_TIMEOUT_SECONDS:-30}" \
    bb -cp "$compiled" "$FINALIZER" "${manifest_args[@]}" \
    >"$work/staged-manifest.digest" 2>"$work/staged-verifier.log" || {
        sed -n '1,120p' "$work/staged-verifier.log" >&2
        die "staged generation did not verify against its canonical build manifest"
    }
mapfile -t staged_manifest_digests <"$work/staged-manifest.digest"
[[ ${#staged_manifest_digests[@]} -eq 1 &&
   "${staged_manifest_digests[0]}" =~ ^sha256:[0-9a-f]{64}$ ]] ||
    die "staged verifier returned malformed manifest identity"
marker_digest="${staged_manifest_digests[0]}"
printf '%s\n' "$marker_digest" >"$install/build.manifest.sha256"
staged_stamp_before="$(staged_tree_stamp "$install")"

exec {publish_lock_fd}>"$out/.beagle-publish.lock"
# flock on an already-open fd is one execve that forks nothing: a leaf phase.
run_phase_leaf publish-lock "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
    flock -x -w "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" "$publish_lock_fd" ||
    die "timed out acquiring the output commit lock"
for source_index in "${!source_snapshot_paths[@]}"; do
    [[ "$(sha256sum "${source_snapshot_paths[$source_index]}" | awk '{print $1}')" == \
       "${source_snapshot_digests[$source_index]}" ]] ||
        die "source changed while waiting for the output commit lock"
done
# The staged tree was fully verified above -- manifest name rules, every
# artifact digest, the native receipt chain, the C17 receipt. This check exists
# for one narrower question: did anything change while this build waited on the
# output commit lock? Re-running the whole verifier answered it by re-deriving
# a verdict that is a pure function of these bytes, at the cost of a second
# babashka boot loading the finalizer. Comparing the tree's own content stamp
# answers the same question directly and is strictly stronger: identical bytes
# force the identical verdict, and the stamp also covers staged files the
# manifest does not name.
[[ "$(staged_tree_stamp "$install")" == "$staged_stamp_before" ]] ||
    die "staged generation changed before publication"

if [[ "$core_result_cache_enabled" == 1 ]]; then
    if [[ "$core_result_early_hit" == 1 ||
          "$core_checkpoint_early_hit" == 1 ]]; then
        validate_core_result_early_closure "$core_result_early_input" \
            "$work/core-result-early-input.before-cache-publish" ||
            die "Core result early input closure changed before cache publication"
    else
        validate_core_result_input_closure "$core_result_input" \
            "$work/core-result-input.before-cache-publish" ||
            die "Core result input closure changed before cache publication"
    fi
fi
if [[ "$core_result_publish_pending" == 1 ]]; then
    acquire_core_result_lock core-result-publish-lock
    core_result_staging="$(mktemp -d \
        "$cache_root/.result-tmp/$core_result_key.XXXXXX")"
    cp -- "$core_result_input" "$core_result_staging/input.manifest"
    mkdir "$core_result_staging/artifacts"
    cp -a "$work/core-result-artifacts/." "$core_result_staging/artifacts/"
    write_core_result_artifact_manifest "$core_result_staging/artifacts" \
        "$core_result_staging/artifacts.sha256"
    artifact_manifest_digest="$(sha256sum \
        "$core_result_staging/artifacts.sha256" | awk '{print $1}')"
    printf 'beagle-core-result/v1 %s %s\n' "$core_result_key" \
        "$artifact_manifest_digest" >"$core_result_staging/READY"
    validate_core_result_entry "$core_result_staging" "$core_result_input" \
        "$core_result_key" || die "new Core result cache entry failed validation"
    # As for the checkpoint above: a peer holding the identical key may have
    # published first. Its entry is validated against THIS build's own input
    # manifest and key before it is adopted, which is the same proof a cache
    # hit rests on.
    if [[ -e "$core_result_entry" ]]; then
        validate_core_result_entry "$core_result_entry" "$core_result_input" \
            "$core_result_key" ||
            die "Core result cache entry appeared under its publication lock and did not validate"
        rm -rf -- "${core_result_staging:?}"
        echo "beagle build: core-result-cache PUBLISHED-BY-PEER $core_result_key" >&2
    else
        mv "$core_result_staging" "$core_result_entry"
    fi
    core_result_staging=""
    core_result_publish_pending=0
    release_core_result_lock
fi

if [[ "$core_result_cache_enabled" == 1 && "$wasm_selected" == false &&
      "$core_result_early_hit" != 1 ]]; then
    validate_core_result_early_closure "$core_result_early_input" \
        "$work/core-result-early-input.before-alias-publish" ||
        die "Core result early input closure changed before alias publication"
    core_result_input_digest="$(sha256sum "$core_result_input" | awk '{print $1}')"
    validate_core_result_entry "$core_result_entry" "$core_result_input" \
        "$core_result_key" || die "published Core result failed alias validation"
    exec {core_result_alias_publish_lock_fd}>\
        "$cache_root/.alias-locks/$core_result_early_key.lock"
    run_phase_leaf core-result-alias-publish-lock \
        "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        flock -x -w "${BEAGLE_CORE_LOCK_TIMEOUT_SECONDS:-30}" \
        "$core_result_alias_publish_lock_fd" ||
        die "timed out acquiring the Core result alias publication lock"
    core_result_alias_staging="$(mktemp \
        "$cache_root/.result-tmp/$core_result_early_key.alias.XXXXXX")"
    printf 'beagle-core-result-alias/v1 %s %s %s\n' \
        "$core_result_early_key" "$core_result_key" "$core_result_input_digest" \
        >"$core_result_alias_staging"
    sync -f "$core_result_alias_staging"
    if { [[ -e "$core_result_alias" ]] || [[ -L "$core_result_alias" ]]; } &&
       { [[ ! -f "$core_result_alias" ]] || [[ -L "$core_result_alias" ]] ||
         ! cmp -s "$core_result_alias_staging" "$core_result_alias"; }; then
        corrupt_alias="$cache_root/.result-tmp/replaced-alias.$core_result_early_key.$$"
        echo "beagle build: core-result-alias STALE $core_result_early_key; replacing" >&2
        mv "$core_result_alias" "$corrupt_alias"
        rm -rf -- "${corrupt_alias:?}"
    fi
    if [[ ! -e "$core_result_alias" ]]; then
        mv "$core_result_alias_staging" "$core_result_alias"
        core_result_alias_staging=""
        sync -f "$core_result_alias_root"
    else
        rm -f -- "$core_result_alias_staging"
        core_result_alias_staging=""
    fi
    flock -u "$core_result_alias_publish_lock_fd"
fi

commit_started=1
rm -f -- "$out/build.manifest.sha256"
sync -f "$out"
[[ "${BEAGLE_CORE_PUBLISH_FAILPOINT:-}" != "after-invalidation" ]] || kill -TERM "$$"
installed_count=0
while IFS= read -r staged_name; do
    [[ "$staged_name" == "build.manifest" ||
       "$staged_name" == "build.manifest.sha256" ||
       "$staged_name" == "report.txt" ||
       "$staged_name" == "wasm-report.txt" ]] && continue
    mv -f -- "$install/$staged_name" "$out/$staged_name"
    sync -f "$out/$staged_name"
    installed_count=$((installed_count + 1))
    if [[ $installed_count -eq 1 &&
          "${BEAGLE_CORE_PUBLISH_FAILPOINT:-}" == "after-one-artifact" ]]; then
        kill -TERM "$$"
    fi
done < <(printf '%s\n' "${!staged_names[@]}" | LC_ALL=C sort)
for artifact in "${managed_artifacts[@]}"; do
    [[ "$artifact" == "build.manifest" ||
       "$artifact" == "build.manifest.sha256" ]] && continue
    [[ -n "${staged_names[$artifact]:-}" ]] || rm -f -- "$out/$artifact"
done
if [[ -n "${staged_names[wasm-report.txt]:-}" ]]; then
    mv -f -- "$install/wasm-report.txt" "$out/wasm-report.txt"
    sync -f "$out/wasm-report.txt"
fi
mv -f -- "$install/report.txt" "$out/report.txt"
sync -f "$out/report.txt"
mv -f -- "$install/build.manifest" "$out/build.manifest"
sync -f "$out/build.manifest"
sync -f "$out"
[[ "${BEAGLE_CORE_PUBLISH_FAILPOINT:-}" != "before-marker" ]] || kill -TERM "$$"
mv -f -- "$install/build.manifest.sha256" "$marker_pending"
sync -f "$marker_pending"
sync -f "$out"
rm -rf "${work:?}"
work_cleaned=1
mv -f -- "$marker_pending" "$out/build.manifest.sha256"
build_committed=1
