Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 36 additions & 13 deletions src/v1/compile.dag
Original file line number Diff line number Diff line change
Expand Up @@ -214,14 +214,20 @@ type DagCollectPending {

// Provisional location key until v2 content_hash (see DagNodeId 🟡 gate). Collection
// fail-closes when two distinct nodes share this key (dag_node_fingerprint mismatch).
// Synthetic 0..0 scaffold nodes key by recursive fingerprint; span-bearing nodes
// stay span-keyed so distinct same-shape occurrences remain distinct.
fn dag_node_key(node: Node) -> String {
let anchor = dag_node_collection_anchor(node: node)
concat(
anchor.span.file, ":", to_string(i: anchor.span.start), "..", to_string(i: anchor.span.end),
match anchor.ident {
Present { value: id } => concat(":", to_string(i: id))
Absent => ""
})
if anchor.span.start == 0 && anchor.span.end == 0 {
concat(":0..0:", dag_node_fingerprint(node: anchor))
} else {
concat(
anchor.span.file, ":", to_string(i: anchor.span.start), "..", to_string(i: anchor.span.end),
match anchor.ident {
Present { value: id } => concat(":", to_string(i: id))
Absent => ""
})
}
}

// Resolved on expression nodes is ordinary type inference (ExprVar, calls, etc.).
Expand Down Expand Up @@ -291,16 +297,33 @@ fn expr_data_variant(data: ExprData) -> String {
}
}

fn dag_node_surface_fingerprint(node: Node) -> String {
let child_names = node.children |> map(c => c.name) |> join(separator: ",")
let param_names = node.params |> map(p => p.name) |> join(separator: ",")
concat(
fn dag_node_bag_hash(digests: List<String>) -> String {
fold(
sort_by(digests, d => d),
init: atom_identity_hash(s: "^dag_collect_bag_empty"),
f: fn(acc, digest) { hash_combine(a: acc, b: digest) }
)
}

fn dag_node_surface_leaf_mix(node: Node) -> String {
atom_identity_hash(s: concat(
node.name, "|",
connective_name(value: node.connective), "|",
inferred_fingerprint(value: node.inferred), "|",
expr_data_variant(data: node.expr_data), "|",
child_names, "|",
param_names)
expr_data_variant(data: node.expr_data)))
}

fn dag_node_surface_fingerprint_rec(node: Node) -> String {
let child_hashes = node.children |> map(c => dag_node_surface_fingerprint_rec(node: c))
let param_hashes = node.params |> map(p => dag_node_surface_fingerprint_rec(node: p))
let with_children = hash_combine(
a: dag_node_surface_leaf_mix(node: node),
b: dag_node_bag_hash(digests: child_hashes))
hash_combine(a: with_children, b: dag_node_bag_hash(digests: param_hashes))
}

fn dag_node_surface_fingerprint(node: Node) -> String {
dag_node_surface_fingerprint_rec(node: node)
}

fn dag_node_fingerprint(node: Node) -> String {
Expand Down
127 changes: 69 additions & 58 deletions src/v1/stage0/src/bin/claim_executor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,10 @@
//! dependency frontier through `v2.workflow.executor` and returns
//! `List<List<Runnable>>` — the outer list is batches in execution order, the
//! inner list is the runnables (a `SingleClaim` witness/gate, or the whole
//! `DiscoveryBatch` corpus) runnable in parallel within that batch. This binary
//! `DiscoveryBatch` corpus) runnable within that batch. This binary
//! evaluates that plan, walks the returned value, and RUNS it: batch by batch
//! (respecting the executor's ordering), nodes within a batch concurrently.
//! (respecting the executor's ordering), runnables within a batch sequentially
//! against one shared module index.
//!
//! It does NOT decide grouping or ordering — add a node or a dependency in the
//! `.dag` and the batches change with zero edit here. That is the dogfood: the
Expand All @@ -18,8 +19,9 @@
//!
//! Like `claim_batch`/`regen_stage0`, this is a hand-written CLI bin — NOT routed
//! through the generated `main.rs`/emit stage — reusing the same resolve/run
//! primitives as `gunbc run` (`cli_run::resolve_entry_graph`,
//! `cli_run::run_value`, `cli_run::run_claim`).
//! primitives as `gunbc run` (`cli_run::build_multi_entry_index`,
//! `cli_run::resolve_entry_with_index`, `cli_run::run_value`,
//! `cli_run::run_claim`).
//!
//! Usage:
//! claim_executor --source-root <dir> [--source-root <dir> ...] \
Expand All @@ -35,11 +37,10 @@
use std::fs;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::thread;

use v1_compiler::cli_run::{
make_eval_context, resolve_entry_graph, run_claim, run_discovery_corpus, run_value,
ClaimOutcome,
build_multi_entry_index, discovery_corpus_rows, make_eval_context, resolve_entry_with_index,
run_claim, run_discovery_rows_resolved, run_value, ClaimOutcome, MultiEntryIndex,
};
use v1_compiler::v1_interpreter::{ExecutionMode, InterpContext, Value};

Expand Down Expand Up @@ -244,29 +245,34 @@ fn batches_from_plan(plan: &Value, ctx: &InterpContext) -> Result<Vec<Vec<Runnab
Ok(batches)
}

/// Result of running one claim, in a thread-safe (Send) form. The resolved graph
/// is `Rc`-based (`!Send`), so each claim resolves and runs entirely within its
/// own thread and reports back only this plain summary.
/// Result of running one claim. The resolved graph is `Rc`-based (`!Send`), so
/// each runnable resolves and runs on the main thread against one shared
/// `MultiEntryIndex` (parse/typed-module caches stay warm across the walk).
struct ClaimResult {
function: String,
ok: bool,
detail: String,
}

fn run_one_runnable(source_roots: Vec<String>, runnable: Runnable) -> ClaimResult {
fn run_one_runnable(index: &MultiEntryIndex, runnable: &Runnable) -> ClaimResult {
match runnable {
Runnable::SingleClaim { entry, function } => {
run_single_claim(&source_roots, entry, function)
run_single_claim(index, entry.clone(), function.clone())
}
Runnable::DiscoveryBatch {
source_roots: roots,
source_roots,
scan_dirs,
explicit_entries,
} => run_discovery_batch_node(roots, scan_dirs, explicit_entries),
} => run_discovery_batch_node(
index,
source_roots.clone(),
scan_dirs.clone(),
explicit_entries.clone(),
),
}
}

fn run_single_claim(source_roots: &[String], entry: String, function: String) -> ClaimResult {
fn run_single_claim(index: &MultiEntryIndex, entry: String, function: String) -> ClaimResult {
// Fail-closed sentinel: the plan projects an empty-`entry` ClaimRef for any
// unmapped suite node or non-complete executor plan (see batch_runner.dag).
// It carries no resolvable witness, so it is a hard error — never a vacuous
Expand All @@ -279,14 +285,14 @@ fn run_single_claim(source_roots: &[String], entry: String, function: String) ->
.to_string(),
};
}
let (graph, source_indices) = match resolve_entry_graph(source_roots, &entry) {
let (graph, source_indices) = match resolve_entry_with_index(index, &entry) {
Ok(pair) => pair,
Err(msg) => {
return ClaimResult {
function,
ok: false,
detail: format!("resolve failed for {}: {}", entry, msg),
}
};
}
};
// Context scoped to this claim's graph: its `data` cache drops with it.
Expand Down Expand Up @@ -319,6 +325,7 @@ fn run_single_claim(source_roots: &[String], entry: String, function: String) ->
/// run loop (`cli_run::run_discovery_corpus`). Fail-closed: an empty roster, a
/// resolve failure, or any failing witness fails the node.
fn run_discovery_batch_node(
index: &MultiEntryIndex,
source_roots: Vec<String>,
scan_dirs: Vec<String>,
explicit_entries: Vec<(String, String)>,
Expand All @@ -328,10 +335,19 @@ fn run_discovery_batch_node(
source_roots.len(),
explicit_entries.len()
);
match run_discovery_corpus(
&source_roots,
&scan_dirs,
&explicit_entries,
let rows = match discovery_corpus_rows(&source_roots, &scan_dirs, &explicit_entries) {
Ok(rows) => rows,
Err(msg) => {
return ClaimResult {
function: label,
ok: false,
detail: format!("discovery corpus failed: {msg}"),
};
}
};
match run_discovery_rows_resolved(
|entry| resolve_entry_with_index(index, entry),
&rows,
ExecutionMode::Wet,
) {
Ok(summary) if summary.failures.is_empty() => ClaimResult {
Expand Down Expand Up @@ -361,11 +377,11 @@ fn run_discovery_batch_node(
/// batching authority: this reads the `List<List<Runnable>>` the plan function
/// returns and parses it — it never groups or orders anything itself.
fn eval_plan(
source_roots: &[String],
index: &MultiEntryIndex,
plan_entry: &str,
plan_function: &str,
) -> Result<Vec<Vec<Runnable>>, String> {
let (plan_graph, plan_indices) = resolve_entry_graph(source_roots, plan_entry)
let (plan_graph, plan_indices) = resolve_entry_with_index(index, plan_entry)
.map_err(|msg| format!("resolve failed for plan {}:\n{}", plan_entry, msg))?;
let plan_ctx = make_eval_context(&plan_graph, plan_indices, ExecutionMode::Wet);
let plan_value = run_value(&plan_ctx, plan_function).map_err(|msg| {
Expand All @@ -376,7 +392,6 @@ fn eval_plan(
})?;
let batches = batches_from_plan(&plan_value, &plan_ctx)
.map_err(|msg| format!("malformed plan value: {}", msg))?;
// Plan graph/value are `Rc`-based (`!Send`); drop before spawning claim threads.
drop(plan_value);
drop(plan_graph);
Ok(batches)
Expand All @@ -394,39 +409,27 @@ struct WalkOutcome {
/// batch boundary is a barrier: batch N+1 starts only after every claim in batch N
/// has reported. A failed batch halts the walk before its dependents. The batch
/// MEMBERSHIP and ORDER are the `.dag` plan's — this only walks them.
fn run_walk(source_roots: &[String], batches: &[Vec<Runnable>]) -> WalkOutcome {
/// Run batch by batch (executor ordering); runnables within a batch sequentially
/// against one shared module index. The batch boundary is a barrier: batch N+1
/// starts only after every runnable in batch N has reported.
fn run_walk(index: &MultiEntryIndex, batches: &[Vec<Runnable>]) -> WalkOutcome {
let mut any_failed = false;
let mut batches_run = 0usize;
for (bi, batch) in batches.iter().enumerate() {
batches_run = bi + 1;
eprintln!("claim_executor: batch {} — {} node(s)", bi + 1, batch.len());
let handles: Vec<_> = batch
.iter()
.map(|runnable| {
let roots = source_roots.to_vec();
let runnable = runnable.clone();
thread::spawn(move || run_one_runnable(roots, runnable))
})
.collect();
for handle in handles {
match handle.join() {
Ok(result) => {
if result.ok {
println!("PASS [batch {}] {}", bi + 1, result.function);
} else {
println!(
"FAIL [batch {}] {} ({})",
bi + 1,
result.function,
result.detail
);
any_failed = true;
}
}
Err(_) => {
println!("FAIL [batch {}] <claim thread panicked>", bi + 1);
any_failed = true;
}
for runnable in batch {
let result = run_one_runnable(index, runnable);
if result.ok {
println!("PASS [batch {}] {}", bi + 1, result.function);
} else {
println!(
"FAIL [batch {}] {} ({})",
bi + 1,
result.function,
result.detail
);
any_failed = true;
}
}
// Fail closed at the barrier: if a gating batch failed, do not run the
Expand Down Expand Up @@ -531,7 +534,8 @@ fn run_perturb_check(
plan_entry: &str,
plan_function: &str,
) -> Result<ExitCode, ExitCode> {
let batches = match eval_plan(source_roots, plan_entry, plan_function) {
let index = build_multi_entry_index(source_roots);
let batches = match eval_plan(&index, plan_entry, plan_function) {
Ok(b) => b,
Err(msg) => {
eprintln!("claim_executor: --perturb-check: {msg}");
Expand Down Expand Up @@ -626,7 +630,9 @@ fn run_perturb_check(
"claim_executor: --perturb-check: planted batch-1 gating witness `{}` -> false; re-walking",
gating_function
);
let outcome = run_walk(&[temp_root], &remapped);
// Perturb runs against a temp source tree — a fresh index is required.
let temp_index = build_multi_entry_index(&[temp_root.clone()]);
let outcome = run_walk(&temp_index, &remapped);
let _ = fs::remove_dir_all(&tmp);

// Receipt: the planted gating failure must fail the run closed AND halt the
Expand Down Expand Up @@ -704,8 +710,13 @@ fn run() -> Result<ExitCode, ExitCode> {
return run_perturb_check(&source_roots, &plan_entry, &plan_function);
}

// 1. Evaluate the executor-decided plan (the `.dag` is the batching authority).
let batches = match eval_plan(&source_roots, &plan_entry, &plan_function) {
// 1. Build the module index once for the whole executor walk (plan eval +
// every batch node share parse/typed-module caches).
eprintln!("claim_executor: building shared module index...");
let index = build_multi_entry_index(&source_roots);

// 2. Evaluate the executor-decided plan (the `.dag` is the batching authority).
let batches = match eval_plan(&index, &plan_entry, &plan_function) {
Ok(b) => b,
Err(msg) => {
eprintln!("claim_executor: {msg}");
Expand All @@ -729,8 +740,8 @@ fn run() -> Result<ExitCode, ExitCode> {
return Err(ExitCode::from(1));
}

// 2. Run batch by batch (executor ordering), claims within a batch in parallel.
let outcome = run_walk(&source_roots, &batches);
// 3. Run batch by batch (executor ordering), runnables within a batch sequentially.
let outcome = run_walk(&index, &batches);
if outcome.any_failed {
Ok(ExitCode::from(1))
} else {
Expand Down
Loading
Loading