ferroplan 0.9.0

//! PDDL2.1 temporal planning — durative actions (EPIC-Temporal).
//!
//! T2 (this module's [`compile`]): each `:durative-action` is split into two
//! instantaneous CLASSICAL actions so the existing grounder/heuristic can be
//! reused. `A-START` takes the action's `at start` conditions as its
//! precondition and applies its `at start` effects plus a `(RUNNING-A ?params)`
//! token; `A-END` requires the `at end` conditions and that token, applies the
//! `at end` effects, and deletes the token.
//!
//! The `over all` invariant and the duration are not expressible in classical
//! STRIPS, so they are kept in a side table ([`SnapInfo`]) that the decision-epoch
//! temporal search (T3) consumes: it only lets `A-END` fire `duration` after the
//! matching `A-START`, and checks the invariant holds across the interval.

use crate::types::{
    eval_numpre, Action, AssignOp, CompOp, Domain, Duration, Effect, Expr, Formula, NExpr, NumEff,
    NumPre, Problem, Sym, Term, TimeSpec,
};

/// Temporal metadata for one durative action, paired with its snap-actions.
#[derive(Clone, Debug)]
pub struct SnapInfo {
    /// Name of the generated start snap-action (e.g. `MOVE-START`).
    pub start_action: Sym,
    /// Name of the generated end snap-action (e.g. `MOVE-END`).
    pub end_action: Sym,
    /// `RUNNING-…` token predicate that pairs a start with its end.
    pub running_pred: Sym,
    /// Duration constraint (fixed `=` or an inequality range) over the action's
    /// parameters / fluents.
    pub duration: Duration,
    /// `over all` invariant that must hold across the action's execution.
    pub invariant: Formula,
    /// The action's typed parameters (for grounding the duration/invariant).
    pub params: Vec<(Sym, Sym)>,
}

/// The result of compiling durative actions to classical snap-actions.
pub struct TemporalCompiled {
    /// Domain with `durative_actions` replaced by classical start/end actions.
    pub domain: Domain,
    pub problem: Problem,
    /// One entry per original durative action.
    pub snaps: Vec<SnapInfo>,
    /// Timed initial literals as `(absolute time, synthetic applier action name)`.
    /// Each name is a 0-arg classical action added to `domain` whose effect asserts /
    /// retracts the literal; the search fires it from the agenda at `time`.
    pub til_ops: Vec<(f64, Sym)>,
}

/// Does this domain use durative actions (i.e. is it a temporal problem)?
pub fn is_temporal(domain: &Domain) -> bool {
    !domain.durative_actions.is_empty()
}

fn and_formulas(parts: Vec<Formula>) -> Formula {
    match parts.len() {
        0 => Formula::True,
        1 => parts.into_iter().next().unwrap(),
        _ => Formula::And(parts),
    }
}

fn and_effects(mut parts: Vec<Effect>) -> Effect {
    if parts.len() == 1 {
        parts.pop().unwrap()
    } else {
        Effect::And(parts)
    }
}

fn pick_conditions(da: &crate::types::DurativeAction, when: TimeSpec) -> Formula {
    and_formulas(
        da.conditions
            .iter()
            .filter(|(t, _)| *t == when)
            .map(|(_, f)| f.clone())
            .collect(),
    )
}

fn pick_effects(da: &crate::types::DurativeAction, when: TimeSpec) -> Vec<Effect> {
    da.effects
        .iter()
        .filter(|(t, _)| *t == when)
        .map(|(_, e)| e.clone())
        .collect()
}

/// Compile a temporal domain (durative actions) into a classical domain of
/// snap-actions plus the [`SnapInfo`] side table.
pub fn compile(domain: &Domain, problem: &Problem) -> TemporalCompiled {
    let mut d = domain.clone();
    let mut snaps = Vec::new();

    for da in &domain.durative_actions {
        let running = format!("RUNNING-{}", da.name);
        let start_name = format!("{}-START", da.name);
        let end_name = format!("{}-END", da.name);
        let run_args: Vec<Term> = da
            .params
            .iter()
            .map(|(p, _)| Term::Var(p.clone()))
            .collect();
        let run_types: Vec<Sym> = da.params.iter().map(|(_, t)| t.clone()).collect();

        d.predicates.push((running.clone(), run_types));
        let invariant = pick_conditions(da, TimeSpec::All);

        // start snap: (at-start conditions + invariant) -> at-start effects + token.
        // The invariant is also checked at both endpoints (a sound approximation
        // of `over all`: an interval violation surfaces when the END precondition
        // fails, e.g. a concurrent action removing the invariant fact).
        let start_pre = and_formulas(vec![
            pick_conditions(da, TimeSpec::Start),
            invariant.clone(),
        ]);
        let mut start_eff = pick_effects(da, TimeSpec::Start);
        start_eff.push(Effect::Add(running.clone(), run_args.clone()));
        d.actions.push(Action {
            name: start_name.clone(),
            params: da.params.clone(),
            precond: start_pre,
            effect: and_effects(start_eff),
            monitored: false,
        });

        // end snap: (at-end conditions + invariant + token) -> at-end effects, drop token
        let end_pre = and_formulas(vec![
            pick_conditions(da, TimeSpec::End),
            invariant.clone(),
            Formula::Atom(running.clone(), run_args.clone()),
        ]);
        let mut end_eff = pick_effects(da, TimeSpec::End);
        end_eff.push(Effect::Del(running.clone(), run_args.clone()));
        d.actions.push(Action {
            name: end_name.clone(),
            params: da.params.clone(),
            precond: end_pre,
            effect: and_effects(end_eff),
            monitored: false,
        });

        snaps.push(SnapInfo {
            start_action: start_name,
            end_action: end_name,
            running_pred: running,
            duration: da.duration.clone(),
            invariant,
            params: da.params.clone(),
        });
    }

    d.durative_actions.clear(); // now expressed as classical snap-actions

    // Timed initial literals → one synthetic 0-arg classical action each (precond
    // True, effect asserts/retracts the literal). This (a) registers the literal's
    // fact with the grounder and (b) makes a positive TIL relaxed-reachable, so a goal
    // achievable only via a TIL isn't pruned as a dead end. The search never *starts*
    // these (classified `Kind::Til`); it fires them from a pre-seeded agenda at `time`.
    let mut til_ops = Vec::new();
    for (k, t) in problem.til.iter().enumerate() {
        let name = format!("TIL-{k}");
        let args: Vec<Term> = t.args.iter().map(|a| Term::Const(a.clone())).collect();
        let eff = if t.add {
            Effect::Add(t.pred.clone(), args)
        } else {
            Effect::Del(t.pred.clone(), args)
        };
        d.actions.push(Action {
            name: name.clone(),
            params: Vec::new(),
            precond: Formula::True,
            effect: eff,
            monitored: false,
        });
        til_ops.push((t.time, name));
    }

    TemporalCompiled {
        domain: d,
        problem: problem.clone(),
        snaps,
        til_ops,
    }
}

// ---------------------------------------------------------------------------
// T3: decision-epoch temporal search.
// ---------------------------------------------------------------------------

use crate::features::DemandMode;
use crate::ground::{ground, Outcome};
use crate::hash::FxHashMap;
use crate::heuristic::{relaxed_helpful, relaxed_to, Scratch};
use crate::packed::{PackedTask, State, StateKey};
use std::cmp::Reverse;
use std::collections::{BinaryHeap, HashMap, HashSet};

/// One action in a timed plan (a durative action is one step with its duration;
/// the end snap is implied).
#[derive(Clone, Debug)]
pub struct TimedStep {
    pub time: f64,
    pub action: String,
    pub duration: Option<f64>,
}

/// A timed (temporal) plan.
#[derive(Clone, Debug)]
pub struct TimedPlan {
    pub steps: Vec<TimedStep>,
    pub makespan: f64,
}

impl TimedPlan {
    /// Render in the IPC temporal plan format: `t: (action args) [duration]`.
    pub fn to_ipc(&self) -> String {
        let mut s = String::new();
        for step in &self.steps {
            s.push_str(&format!(
                "{:.3}: ({}) [{:.3}]\n",
                step.time,
                step.action.to_lowercase(),
                step.duration.unwrap_or(0.001),
            ));
        }
        s
    }
}

#[derive(Clone, Copy)]
pub(crate) enum Kind {
    /// durative start: resolved duration (constant or parameter-dependent) + the
    /// matching end op index
    Start {
        dur: f64,
        end_op: usize,
    },
    End,
    Classical,
    /// a synthetic timed-initial-literal applier: never started by the search (block
    /// (a) skips it), only fired from the pre-seeded agenda at its absolute time.
    Til,
    /// a start whose duration/end can't be resolved (undefined duration fluent,
    /// non-positive value, or missing end op); never applied
    Skip,
}

struct TNode {
    state: State,
    time: f64,
    /// pending ends as (absolute_end_time, end_op), kept sorted ascending.
    agenda: Vec<(f64, usize)>,
    father: usize,
    /// (op applied, time) that produced this node; None for the root.
    ev: Option<(usize, f64)>,
    /// number of happenings to reach this node (depth `g`, for the heap key).
    g: u32,
    /// FF helpful start/classical ops for this state under pruning (empty = no
    /// restriction / fall back to a full scan). Only populated in the pruned pass.
    helpful: Vec<u32>,
    /// Cumulative-availability per demand resource (init stock + everything produced
    /// along this path, clamped to the demand). Empty unless FF_TDEMAND is on. Tracks
    /// production rather than current stock so the gradient survives consumption.
    met: Vec<i32>,
}

fn tkey(task: &PackedTask, n: &TNode) -> (StateKey, Vec<(i64, usize)>) {
    let ag = n
        .agenda
        .iter()
        .map(|&(t, o)| ((t * 1000.0).round() as i64, o))
        .collect();
    (task.state_key(&n.state), ag)
}

/// Evaluate a (possibly parameter-dependent) duration for one grounded
/// snap-action. The action's parameters are bound positionally to the grounded
/// args; fluents are read from the INITIAL state — IPC temporal durations depend
/// on static fluents like `(= ?duration (/ (distance ?a ?b) (speed ?v)))`, which
/// keep their init value. Returns None for a non-positive duration, an undefined
/// fluent, or division by zero (the caller then skips the action).
fn eval_duration(snap: &SnapInfo, args: &[&str], task: &PackedTask, init: &State) -> Option<f64> {
    let bind = duration_bind(snap, args);
    // Commit to the shortest feasible duration (the lower bound; the upper bound only
    // for a sole `<=`). Inequality slack is given up here in exchange for a single
    // resolved duration the decision-epoch search can schedule — see `validate`, which
    // accepts the whole `[min, max]` range.
    let d = eval_expr(snap.duration.chosen()?, &bind, task, init)?;
    if d.is_finite() && d > 0.0 {
        Some(d)
    } else {
        None
    }
}

/// Evaluate the `[min, max]` duration bounds against the initial state (for the
/// validator). An open side stays `None` (unbounded). A bound that fails to evaluate
/// (undefined fluent, div-by-zero) also yields `None` for that side.
fn eval_duration_bounds(
    snap: &SnapInfo,
    args: &[&str],
    task: &PackedTask,
    init: &State,
) -> (Option<f64>, Option<f64>) {
    let bind = duration_bind(snap, args);
    let ev = |o: &Option<Expr>| o.as_ref().and_then(|e| eval_expr(e, &bind, task, init));
    (ev(&snap.duration.min), ev(&snap.duration.max))
}

/// Bind a snap-action's parameters positionally to the grounded args.
fn duration_bind<'a>(snap: &'a SnapInfo, args: &[&'a str]) -> HashMap<&'a str, &'a str> {
    snap.params
        .iter()
        .map(|(p, _)| p.as_str())
        .zip(args.iter().copied())
        .collect()
}

fn eval_expr(e: &Expr, bind: &HashMap<&str, &str>, task: &PackedTask, init: &State) -> Option<f64> {
    match e {
        Expr::Num(n) => Some(*n),
        Expr::Fluent(name, terms) => {
            let mut disp = String::from("(");
            disp.push_str(name);
            for t in terms {
                disp.push(' ');
                match t {
                    Term::Const(c) => disp.push_str(c),
                    Term::Var(v) => disp.push_str(bind.get(v.as_str())?),
                }
            }
            disp.push(')');
            let id = task.fluent_id(&disp)?;
            init.fdef[id].then(|| init.fv[id])
        }
        Expr::Add(a, b) => Some(eval_expr(a, bind, task, init)? + eval_expr(b, bind, task, init)?),
        Expr::Sub(a, b) => Some(eval_expr(a, bind, task, init)? - eval_expr(b, bind, task, init)?),
        Expr::Mul(a, b) => Some(eval_expr(a, bind, task, init)? * eval_expr(b, bind, task, init)?),
        Expr::Div(a, b) => {
            let d = eval_expr(b, bind, task, init)?;
            if d == 0.0 {
                return None;
            }
            Some(eval_expr(a, bind, task, init)? / d)
        }
        Expr::Neg(a) => Some(-eval_expr(a, bind, task, init)?),
    }
}

/// Solve a temporal (durative-action) problem by decision-epoch forward search.
/// Returns a timed plan, or None if unsolved within the node budget. Durations
/// may be constants or parameter-dependent (evaluated against the initial state);
/// the `over all` invariant is enforced at the start and end happenings via the
/// snap preconditions.
pub fn solve(domain: &Domain, problem: &Problem, threads: usize) -> Option<TimedPlan> {
    let ambient = crate::features::demand_mode();
    if let Some(plan) = solve_monolithic(domain, problem, threads, ambient) {
        return Some(plan);
    }
    // On-failure escalation ladder (see `features::escalate`). Each rung runs only
    // after the previous failed, so nothing that solves above can change — the
    // ladder converts failures into solves at the cost of extra time on failures.
    // Gated off by FF_NO_ESCALATE, and by FF_NO_TDEMAND (the master pre-v0.2
    // opt-out — escalating from `Off` would contradict it). Measured (cabin):
    // crew-solo/pair + skilled-specialists solve at the Full rung; order-8/12
    // solve at the decomposer rung.
    if ambient == DemandMode::Off || !crate::features::escalate() {
        return None;
    }
    if ambient != DemandMode::Full {
        if let Some(plan) = solve_monolithic(domain, problem, threads, DemandMode::Full) {
            return Some(plan);
        }
    }
    // Decomposer rung — the ladder variant, which skips the decomposer's own
    // monolithic fallbacks (this ladder already ran that exact search at both
    // tiers) and thus also cannot recurse. Passes this ladder's own rung-0 tier
    // so the premise can't drift with the process-global override.
    crate::tresolve::solve_after_ladder(domain, problem, threads, ambient)
}

/// The monolithic temporal search at an explicit demand `tier` — the scheduling
/// phase + the plain decision-epoch search, WITHOUT the escalation ladder. This is
/// the primitive `solve` builds its ladder from and the decomposer's single-group
/// fallback (`tresolve`) terminates on.
pub(crate) fn solve_monolithic(
    domain: &Domain,
    problem: &Problem,
    threads: usize,
    tier: DemandMode,
) -> Option<TimedPlan> {
    // Concurrent scheduling phase (gated). The multi-actor search is flaky, so we
    // search a SINGLE-actor reduction (tractable) and then repack that plan onto the
    // full crew — one job per worker, resources permitting — to minimise makespan.
    // Validated + only-if-shorter inside `reschedule`, so it can only improve things;
    // if the reduction finds nothing we fall through to a normal solve.
    if crate::features::tconc() {
        // ≥2 actors ⇒ the reduction is a *super-worker* (all skills), so its plan is
        // only valid for `problem` once reassigned to real skilled workers; <2 ⇒ the
        // reduction is `problem` itself, so its plan is valid as-is.
        let reduced = crate::tsched::n_actors(domain, problem) >= 2;
        let solo = crate::tsched::single_actor_problem(domain, problem);
        if let Some(plan) = solve_inner(domain, &solo, threads, tier) {
            if let Some(rp) = crate::tsched::reschedule(domain, problem, &plan) {
                return Some(rp);
            }
            if !reduced {
                return Some(plan);
            }
            // reduced but couldn't validly reschedule (e.g. a task needs a skill no
            // single worker has) — fall through to an honest full-problem search.
        }
    }
    solve_inner(domain, problem, threads, tier)
}

/// Search a temporal plan for `problem` as-is (no scheduling phase).
fn solve_inner(
    domain: &Domain,
    problem: &Problem,
    threads: usize,
    tier: DemandMode,
) -> Option<TimedPlan> {
    let c = compile(domain, problem);
    let task = match ground(&c.domain, &c.problem, threads) {
        Outcome::Task(t) => t,
        Outcome::GoalTrue => {
            return Some(TimedPlan {
                steps: Vec::new(),
                makespan: 0.0,
            })
        }
        _ => return None,
    };

    let kind = build_kind(&task, &c);
    // Resolve each TIL's synthetic applier to its grounded op id (0-arg ⇒ op display
    // is the action name). A TIL whose op didn't ground is silently dropped.
    let by_display: HashMap<&str, usize> = task
        .op_display
        .iter()
        .enumerate()
        .map(|(i, d)| (d.as_str(), i))
        .collect();
    let til_events: Vec<(f64, usize)> = c
        .til_ops
        .iter()
        .filter_map(|(t, name)| by_display.get(name.as_str()).map(|&oi| (*t, oi)))
        .collect();

    solve_from(
        &task,
        &kind,
        &task.initial(),
        &task.goal_pos,
        &task.goal_num,
        &[],
        &til_events,
        threads,
        tier,
    )
}

/// Classify every grounded op as a durative Start (with resolved duration + paired
/// end op), End, Classical, or Skip (unresolvable). Shared by `solve` and the
/// decomposer (`tresolve`), built once per grounded task.
pub(crate) fn build_kind(task: &PackedTask, c: &TemporalCompiled) -> Vec<Kind> {
    // Durations are constant or parameter-dependent (evaluated against the initial
    // state); pair each start snap with its matching end op.
    let init = task.initial();
    let snap_by_start: HashMap<&str, &SnapInfo> = c
        .snaps
        .iter()
        .map(|s| (s.start_action.as_str(), s))
        .collect();
    let end_names: HashSet<&str> = c.snaps.iter().map(|s| s.end_action.as_str()).collect();
    let til_names: HashSet<&str> = c.til_ops.iter().map(|(_, n)| n.as_str()).collect();
    let by_display: HashMap<&str, usize> = task
        .op_display
        .iter()
        .enumerate()
        .map(|(i, d)| (d.as_str(), i))
        .collect();
    (0..task.n_ops)
        .map(|oi| {
            let disp = &task.op_display[oi];
            let head = disp.split_whitespace().next().unwrap_or("");
            if let Some(snap) = snap_by_start.get(head) {
                let args: Vec<&str> = disp.split_whitespace().skip(1).collect();
                let end_disp = disp.replacen("-START", "-END", 1);
                match (
                    eval_duration(snap, &args, task, &init),
                    by_display.get(end_disp.as_str()),
                ) {
                    (Some(dur), Some(&end_op)) => Kind::Start { dur, end_op },
                    _ => Kind::Skip,
                }
            } else if end_names.contains(head) {
                Kind::End
            } else if til_names.contains(head) {
                Kind::Til
            } else {
                Kind::Classical
            }
        })
        .collect()
}

/// Goal-relevance op mask (`true` = keep). Backward closure from the goal: an op is
/// relevant if it ADDS or DELETES a relevant fact, or INCREASES a relevant resource
/// (incl. via a conditional effect); marking it pulls its preconditions (positive
/// facts, numeric `>=` thresholds) and consumed resources into the relevant set,
/// transitively. Ops that cannot contribute — e.g. `forage-food`/`gather-herbs` when
/// food/herbs are in no recipe the goal needs: unbounded accumulators that otherwise
/// explode the complete search with food=1,2,3,… states — are dropped. Applied to
/// BOTH phases: phase 1 (helpful) is usually stuck under delete-relaxation anyway;
/// the win is letting the COMPLETE phase 2 solve within the relevant subspace instead
/// of drowning in irrelevant accumulation. Sound (completeness-preserving) because a
/// pruned op neither produces nor consumes nor toggles anything any solution needs —
/// the `del`-of-relevant clause conservatively keeps re-enablers of negative
/// preconditions. Necessary travel is kept: a relevant op's `(at a l)` precond makes
/// the travel achieving it relevant, transitively along the route.
fn relevant_op_mask(
    task: &PackedTask,
    goal_pos: &[u32],
    goal_num: &[NumPre],
    tight: bool,
) -> Vec<bool> {
    let mut rel_fact: crate::hash::FxHashSet<u32> = goal_pos.iter().copied().collect();
    let mut rel_res: crate::hash::FxHashSet<u32> = goal_num
        .iter()
        .filter_map(|np| as_threshold(np).map(|(t, _)| t))
        .collect();
    // TIGHT mode: a resource is "produced" only by its single best-yield producer, so
    // marking (say) `planks` relevant pulls in `saw-planks` but NOT the alternative
    // producer `haul-cargo` — which would otherwise drag the whole logistics subsystem
    // into the relevant set and re-explode. best_end[r] = that producer's op id.
    let best_end: Vec<Option<usize>> = if tight {
        (0..task.fv0.len())
            .map(|r| best_producer(task, r as u32).map(|(o, _)| o))
            .collect()
    } else {
        Vec::new()
    };
    let produces = |oi: usize, t: u32| -> bool {
        !tight || best_end.get(t as usize).copied().flatten() == Some(oi)
    };
    let mut relevant = vec![false; task.n_ops];
    loop {
        let mut changed = false;
        // range loop: the body both reads task slices by `oi` and writes `relevant[oi]`.
        #[allow(clippy::needless_range_loop)]
        for oi in 0..task.n_ops {
            if relevant[oi] {
                continue;
            }
            let touches_fact = task.add.slice(oi).iter().any(|f| rel_fact.contains(f))
                || task.del.slice(oi).iter().any(|f| rel_fact.contains(f));
            let inc_res = task.num_eff.slice(oi).iter().any(|ne| {
                matches!(ne.op, AssignOp::Increase)
                    && rel_res.contains(&ne.target)
                    && produces(oi, ne.target)
            });
            let cond_rel = task.cond_effs(oi).any(|ce| {
                ce.add.iter().any(|f| rel_fact.contains(f))
                    || ce.del.iter().any(|f| rel_fact.contains(f))
                    || ce.num.iter().any(|ne| {
                        matches!(ne.op, AssignOp::Increase) && rel_res.contains(&ne.target)
                    })
            });
            if touches_fact || inc_res || cond_rel {
                relevant[oi] = true;
                changed = true;
                for &f in task.pre_pos.slice(oi) {
                    rel_fact.insert(f);
                }
                for np in task.pre_num.slice(oi) {
                    if let Some((t, _)) = as_threshold(np) {
                        rel_res.insert(t);
                    }
                }
                for ne in task.num_eff.slice(oi) {
                    if matches!(ne.op, AssignOp::Decrease) {
                        rel_res.insert(ne.target);
                    }
                }
                for ce in task.cond_effs(oi) {
                    for &f in &ce.cond_pos {
                        rel_fact.insert(f);
                    }
                    for ne in &ce.num {
                        if matches!(ne.op, AssignOp::Decrease) {
                            rel_res.insert(ne.target);
                        }
                    }
                }
            }
        }
        if !changed {
            break;
        }
    }
    relevant
}

/// Solve toward an ARBITRARY `(goal_pos, goal_num)` from an arbitrary `start` state
/// over a shared grounded temporal task — the reusable subplanner the decomposer
/// (`tresolve`) calls per contract. `forbidden` masks ops (sibling protection; empty
/// = none). `tier` is the demand tier this pass runs at — threaded explicitly so
/// the escalation ladder can retry at `Full` without touching process-global state.
/// `solve_monolithic` is the whole-task wrapper (start = init, goal = task goal, no
/// forbidden); `temporal::solve` is that plus the on-failure escalation ladder.
///
/// Multi-pass decision-epoch search: a fast pass restricting start/classical
/// expansion to FF helpful actions, then unrestricted complete passes on failure
/// (tight-masked → sound-masked → unmasked; see the pruning block below).
/// Phase-1 key = W_G*g + W_H*h + W_L*(unmet numeric landmarks) + the converging-
/// resource demand deficit; the complete passes use the original pure-h key.
#[allow(clippy::too_many_arguments)]
pub(crate) fn solve_from(
    task: &PackedTask,
    kind: &[Kind],
    start: &State,
    goal_pos: &[u32],
    goal_num: &[NumPre],
    forbidden: &[bool],
    til_events: &[(f64, usize)],
    threads: usize,
    tier: DemandMode,
) -> Option<TimedPlan> {
    // Fail fast on statically unproducible goals — nothing any pass could reach.
    if statically_unsolvable(task, start, goal_pos, goal_num) {
        return None;
    }
    // Landmarks are ALWAYS on (phase-1 key), so seed them from the numeric goal ONLY
    // — keeping the default path byte-identical. The predicate-goal thresholds (which
    // would change default ordering) ride the FF_TDEMAND-gated demand seed instead.
    let landmarks = extract_landmarks(task, goal_num);
    // Converging-resource demand guidance (FF_TDEMAND, default OFF → empty → the
    // phase-1 key is bit-identical to the prior temporal search). Phase 2 (the
    // complete pure-h pass) is unaffected regardless, so completeness is preserved.
    let demand = if tier != DemandMode::Off {
        let w = std::env::var("FF_TDEMAND_W")
            .ok()
            .and_then(|s| s.parse::<i64>().ok())
            .unwrap_or(3);
        // demand seed = numeric goal (always) + numeric thresholds implied by
        // PREDICATE goals' achievers (Full tier only — so `(built-wall)` drives the
        // blocks>=4 chain). The predicate half is gated off by default because it
        // reads a renewable-pool guard (e.g. `(>= (avail) 1)`) as accumulation demand
        // and serializes concurrency domains; the numeric half is the measured win.
        let mut seed: Vec<NumPre> = goal_num.to_vec();
        if tier == DemandMode::Full {
            seed.extend(predicate_goal_thresholds(task, kind, goal_pos));
        }
        let d = compute_demand(task, kind, &seed, w);
        if std::env::var("FF_RES_DEBUG").is_ok() {
            let pretty: Vec<(String, i32)> = d
                .res
                .iter()
                .map(|&(f, a)| (task.fluent_names[f as usize].clone(), a))
                .collect();
            eprintln!("[TDEMAND] w={w} total={} resources={:?}", d.total, pretty);
        }
        d
    } else {
        Demand::empty()
    };
    // Goal-relevance pruning (default-on with the demand tiers; `FF_NOREL` disables
    // pruning alone, `FF_NO_TDEMAND` restores the pristine pre-v0.2 path entirely).
    // Two masks: SOUND (every producer of each relevant resource) and TIGHT (only the
    // single best-yield producer — drops alternative-recipe subsystems like logistics
    // for `planks`). Four passes: helpful (sound) → full+tight → full+sound →
    // full+unmasked. The tight pass solves the conjunctive/structural builds without
    // exploding; the sound pass solves within the relevant subspace instead of
    // drowning in irrelevant unbounded accumulators (food=1,2,3,…); the final
    // unmasked pass makes completeness UNCONDITIONAL — even a hypothetical mask bug
    // cannot lose coverage, it can only cost time on unsolvable inputs.
    // Graduated from the Full tier in v0.3.0: `flour >= 2` on a fully-featured hub
    // (rpg-world bread-line) needs pruning to solve at all — the default search
    // exhausted its node budget in the irrelevant-accumulator swamp.
    let on = tier != DemandMode::Off && std::env::var("FF_NOREL").is_err();
    let sound = if on {
        relevant_op_mask(task, goal_pos, goal_num, false)
    } else {
        Vec::new()
    };
    let tight = if on {
        relevant_op_mask(task, goal_pos, goal_num, true)
    } else {
        Vec::new()
    };
    if on && std::env::var("FF_RES_DEBUG").is_ok() {
        eprintln!(
            "[TREL] sound {}/{}  tight {}/{}",
            sound.iter().filter(|&&b| b).count(),
            sound.len(),
            tight.iter().filter(|&&b| b).count(),
            tight.len()
        );
    }
    let go = |rel: &[bool], prune: bool| {
        temporal_search(
            task, kind, &landmarks, &demand, start, goal_pos, goal_num, forbidden, rel, til_events,
            prune, threads,
        )
    };
    go(&sound, true)
        .or_else(|| if on { go(&tight, false) } else { None })
        .or_else(|| go(&sound, false))
        // Unmasked complete backstop — only distinct from the previous pass when
        // pruning is on (off ⇒ `sound` is already empty ⇒ pass 3 was unmasked).
        .or_else(|| if on { go(&[], false) } else { None })
}

/// Static unproducibility: is some goal conjunct impossible to ever achieve because
/// NOTHING in the grounded task can move it? Two sound, instant checks:
/// - a positive goal fact not true in `start` that **no op adds** (plain or
///   conditional effect — TIL appliers are ops too, so exogenous adds count);
/// - a `>=`/`>` numeric threshold not met in `start` whose fluent **no op can
///   possibly raise** (an effect "can raise" unless it provably never does:
///   `increase` by a constant ≤ 0 or `decrease` by a constant ≥ 0; `assign`/
///   `scale-up`/`scale-down`/non-constant deltas all count as potential raisers).
///
/// A `true` here means every search pass would exhaust its budget and fail anyway —
/// e.g. rpg-world `bread-line` goals on `(bread) >= 2` but no action produces
/// `bread`, and the search burned ~45 s across passes proving it. This check makes
/// such instances (and decomposer contracts) fail in microseconds. It never changes
/// a *found* plan, only converts an exhaustive failure into an instant one.
pub(crate) fn statically_unsolvable(
    task: &PackedTask,
    start: &State,
    goal_pos: &[u32],
    goal_num: &[NumPre],
) -> bool {
    let fact_true = |f: u32| (start.bits[f as usize / 64] >> (f as usize % 64)) & 1 == 1;
    let never_raises = |ne: &NumEff| match (&ne.op, &ne.value) {
        (AssignOp::Increase, NExpr::Num(w)) => w <= &0.0,
        (AssignOp::Decrease, NExpr::Num(w)) => w >= &0.0,
        _ => false,
    };
    let some_op_adds = |g: u32| {
        (0..task.n_ops).any(|oi| {
            task.add.slice(oi).contains(&g) || task.cond_effs(oi).any(|ce| ce.add.contains(&g))
        })
    };
    let some_op_raises = |t: u32| {
        (0..task.n_ops).any(|oi| {
            task.num_eff
                .slice(oi)
                .iter()
                .any(|ne| ne.target == t && !never_raises(ne))
                || task
                    .cond
                    .slice(oi)
                    .iter()
                    .any(|ce| ce.num.iter().any(|ne| ne.target == t && !never_raises(ne)))
        })
    };
    for &g in goal_pos {
        if !fact_true(g) && !some_op_adds(g) {
            return true;
        }
    }
    for np in goal_num {
        if let Some((t, _)) = as_threshold(np) {
            let already = eval_numpre(np, &start.fv, &start.fdef) == Some(true);
            if !already && !some_op_raises(t) {
                return true;
            }
        }
    }
    false
}

/// A numeric `>=`/`>` threshold `(fluent, value)`, or `None` if `np` isn't of that
/// canonical recipe-gate shape.
fn as_threshold(np: &NumPre) -> Option<(u32, f64)> {
    match (&np.op, &np.lhs, &np.rhs) {
        (CompOp::Ge | CompOp::Gt, NExpr::Fluent(t), NExpr::Num(w)) => Some((*t, *w)),
        _ => None,
    }
}

/// Numeric `>=` thresholds implied by the achievers of the PREDICATE goal facts: for
/// each goal fact, the op that adds it (the END snap) gates on numeric preconditions
/// that live on the matching START snap — bridge END->START via the RUNNING token
/// exactly as `extract_landmarks` does, and collect those `>=` preconds. Lets a
/// predicate goal like `(built-wall)` seed the `blocks>=4` demand chain (Stage 0).
fn predicate_goal_thresholds(task: &PackedTask, kind: &[Kind], goal_pos: &[u32]) -> Vec<NumPre> {
    let mut out: Vec<NumPre> = Vec::new();
    let collect_thr = |oi: usize, out: &mut Vec<NumPre>| {
        for pre in task.pre_num.slice(oi) {
            if as_threshold(pre).is_some() {
                out.push(pre.clone());
            }
        }
    };
    for &gf in goal_pos {
        for &oi in task.add_by_fact.slice(gf as usize) {
            let oi = oi as usize;
            collect_thr(oi, &mut out); // classical / direct numeric precond
            for &f in task.pre_pos.slice(oi) {
                for &start in task.add_by_fact.slice(f as usize) {
                    if matches!(kind[start as usize], Kind::Start { .. }) {
                        collect_thr(start as usize, &mut out); // bridged START precond
                    }
                }
            }
        }
    }
    out
}

/// Numeric-threshold landmarks: the transitive closure of the `>=` preconditions of
/// the ops that *increase* each goal fluent. The delete-relaxed extraction drops
/// these (it never recurses on `pre_num`), so on a converging DAG — where two inputs
/// are separate numeric thresholds feeding one join — `h` goes flat. Counting how
/// many a state has NOT met gives each converging input its own descending term in
/// the phase-1 key, restoring the gradient the FF count lacks.
fn extract_landmarks(task: &PackedTask, seed: &[NumPre]) -> Vec<NumPre> {
    let mut out: Vec<NumPre> = Vec::new();
    let mut seen: HashSet<(u32, u64)> = HashSet::new();
    let mut work: Vec<NumPre> = seed.to_vec();
    let mut iters = 0usize;
    while let Some(np) = work.pop() {
        iters += 1;
        if iters > 8000 {
            break; // safety cap against accumulator cycles
        }
        let Some((t, w)) = as_threshold(&np) else {
            continue;
        };
        if !seen.insert((t, w.to_bits())) {
            continue;
        }
        out.push(np.clone());
        let add_pre_num = |oi: usize, work: &mut Vec<NumPre>| {
            for pre in task.pre_num.slice(oi) {
                if as_threshold(pre).is_some() {
                    work.push(pre.clone());
                }
            }
        };
        // recurse toward the recipe inputs of ops that INCREASE fluent `t`.
        for &oi in task.neff_by_fluent.slice(t as usize) {
            let oi = oi as usize;
            let increases = task
                .num_eff
                .slice(oi)
                .iter()
                .any(|ne| ne.target == t && matches!(ne.op, AssignOp::Increase));
            if !increases {
                continue;
            }
            // (a) classical case: numeric preconds sit on the increasing op itself.
            add_pre_num(oi, &mut work);
            // (b) snap-compiled case: the increase is on the END snap, but the
            // recipe's numeric inputs are on the matching START snap — bridge via the
            // RUNNING token (END requires it, START adds it).
            for &f in task.pre_pos.slice(oi) {
                for &start in task.add_by_fact.slice(f as usize) {
                    add_pre_num(start as usize, &mut work);
                }
            }
        }
    }
    out
}

/// Summed DEFICIT of the landmark thresholds in `(fv, fdef)` — for each `(fluent >=
/// want)` landmark, how far below `want` the fluent is. Unlike a binary met/unmet
/// count this gives a gradient across MULTIPLE rounds (e.g. steel>=2 descends 2→1→0),
/// so deep/wide converging accumulation gets guidance, not just single assembly.
fn landmark_deficit(landmarks: &[NumPre], fv: &[f64], fdef: &[bool]) -> i64 {
    landmarks
        .iter()
        .map(|np| match as_threshold(np) {
            Some((t, want)) => {
                let cur = if fdef[t as usize] {
                    fv[t as usize]
                } else {
                    0.0
                };
                (want - cur).max(0.0).ceil() as i64
            }
            None => 0,
        })
        .sum()
}

/// Total resource DEMAND implied by the numeric goal, regressed down the recipe
/// DAG. A `(fluent >= want)` goal needs `want` of that fluent; its best (highest-
/// yield) producer needs `ceil(want / yield)` applications, each consuming its
/// inputs — recurse. Unlike the per-recipe landmark thresholds (`ingots >= 1`), this
/// captures the FULL multi-round quantity (`steel >= 2` ⇒ ingots/coal/ore ≥ 2,
/// logs ≥ 4) that the delete-relaxed heuristic — which reuses each consumed unit —
/// never demands. `weight` 0 / empty `res` ⇒ the whole term is inert (heap key
/// bit-identical), so the default temporal path is unchanged.
struct Demand {
    res: Vec<(u32, i32)>,
    idx: FxHashMap<u32, usize>,
    total: i32,
    weight: i64,
}

impl Demand {
    fn empty() -> Self {
        Demand {
            res: Vec::new(),
            idx: FxHashMap::default(),
            total: 0,
            weight: 0,
        }
    }
}

/// The highest base-yield op that increases fluent `t` (raw resources have a gather
/// producer with no numeric inputs, so regression bottoms out there). Conditional
/// (role-bonus) increases are ignored — the base yield is the safe estimate.
fn best_producer(task: &PackedTask, t: u32) -> Option<(usize, i32)> {
    let mut best: Option<(usize, i32)> = None;
    for &oi in task.neff_by_fluent.slice(t as usize) {
        let oi = oi as usize;
        for ne in task.num_eff.slice(oi) {
            if ne.target == t && matches!(ne.op, AssignOp::Increase) {
                let y = ne.value.eval(&task.fv0, &task.fdef0).unwrap_or(0.0);
                if y > 0.0 {
                    let yi = y.ceil() as i32;
                    if best.map_or(true, |(_, by)| yi > by) {
                        best = Some((oi, yi));
                    }
                }
            }
        }
    }
    best
}

fn compute_demand(task: &PackedTask, kind: &[Kind], seed: &[NumPre], weight: i64) -> Demand {
    use crate::hash::FxHashSet;
    const MAX_ITERS: usize = 20_000;
    const CAP: i32 = 100_000; // guard against cyclic/regenerating recipe blowup
    let mut need: FxHashMap<u32, i32> = FxHashMap::default();
    let mut work: Vec<(u32, i32)> = seed
        .iter()
        .filter_map(|np| as_threshold(np).map(|(t, w)| (t, w.ceil().max(0.0) as i32)))
        .collect();
    let mut iters = 0usize;
    while let Some((t, amt)) = work.pop() {
        iters += 1;
        if iters > MAX_ITERS {
            break;
        }
        if amt <= 0 {
            continue;
        }
        let cur = need.entry(t).or_insert(0);
        let target = (*cur + amt).min(CAP);
        let delta = target - *cur; // only propagate the marginal new demand
        if delta <= 0 {
            continue;
        }
        *cur = target;
        let Some((oi, yield_t)) = best_producer(task, t) else {
            continue; // raw resource — bottoms out (stays in `need`)
        };
        let apps = (delta + yield_t - 1) / yield_t; // ceil
                                                    // Inputs = the producer's own decreases PLUS, for snap-compiled durative
                                                    // recipes, the matching START snap's decreases (the increase is on the END
                                                    // snap; the consume is on the START that adds a RUNNING token END requires).
                                                    // Bridge exactly as extract_landmarks does, filtering adders to START snaps.
        let mut consumers: FxHashSet<usize> = FxHashSet::default();
        consumers.insert(oi);
        for &f in task.pre_pos.slice(oi) {
            for &start in task.add_by_fact.slice(f as usize) {
                if matches!(kind[start as usize], Kind::Start { .. }) {
                    consumers.insert(start as usize);
                }
            }
        }
        for op in consumers {
            for ne in task.num_eff.slice(op) {
                if matches!(ne.op, AssignOp::Decrease) {
                    let c = ne.value.eval(&task.fv0, &task.fdef0).unwrap_or(0.0);
                    if c > 0.0 {
                        work.push((ne.target, apps.saturating_mul(c.ceil() as i32)));
                    }
                }
            }
        }
    }
    let mut res: Vec<(u32, i32)> = need.into_iter().collect();
    res.sort_unstable(); // deterministic order, independent of hashmap iteration
    let mut idx = FxHashMap::default();
    let mut total = 0i32;
    for (i, &(f, a)) in res.iter().enumerate() {
        idx.insert(f, i);
        total += a;
    }
    Demand {
        res,
        idx,
        total,
        weight,
    }
}

/// The set of resources in the demand-closure of `goal_num` (the recipe inputs that
/// producing the goal consumes, transitively) — used by the decomposer to order
/// contracts so a goal that is itself an input to another goal is produced LAST.
pub(crate) fn demand_resources(task: &PackedTask, kind: &[Kind], goal_num: &[NumPre]) -> Vec<u32> {
    compute_demand(task, kind, goal_num, 1)
        .res
        .into_iter()
        .map(|(f, _)| f)
        .collect()
}

/// Root availability: initial stock of each demand resource, clamped to its demand.
fn met_root(demand: &Demand, task: &PackedTask) -> Vec<i32> {
    demand
        .res
        .iter()
        .map(|&(f, a)| {
            let cur = if task.fdef0[f as usize] {
                task.fv0[f as usize]
            } else {
                0.0
            };
            (cur.max(0.0) as i32).min(a)
        })
        .collect()
}

/// Child availability: parent's, plus op `oi`'s (unconditional) increases on demand
/// resources, clamped. Production-only (consumption never lowers it) — so a consumed
/// intermediate still counts as delivered, the key to the multi-round gradient.
fn met_child(parent: &[i32], demand: &Demand, task: &PackedTask, oi: usize) -> Vec<i32> {
    if demand.res.is_empty() {
        return Vec::new();
    }
    let mut m = parent.to_vec();
    for ne in task.num_eff.slice(oi) {
        if matches!(ne.op, AssignOp::Increase) {
            if let Some(&i) = demand.idx.get(&ne.target) {
                let v = ne.value.eval(&task.fv0, &task.fdef0).unwrap_or(0.0);
                if v > 0.0 {
                    m[i] = (m[i] + v.ceil() as i32).min(demand.res[i].1);
                }
            }
        }
    }
    m
}

#[inline]
fn demand_deficit(met: &[i32], demand: &Demand) -> i64 {
    (demand.total - met.iter().sum::<i32>()) as i64
}

/// `h`, plus — under `prune` — the Skip-filtered helpful start/classical ops for
/// `s`. `None` iff `s` is a relaxed dead end (so this also gates dead ends).
#[allow(clippy::too_many_arguments)]
fn eval_node(
    task: &PackedTask,
    kind: &[Kind],
    sc: &mut Scratch,
    s: &State,
    goal_pos: &[u32],
    goal_num: &[NumPre],
    prune: bool,
) -> Option<(i32, Vec<u32>)> {
    if prune {
        let (h, helpful) = relaxed_helpful(task, sc, &s.bits, &s.fv, &s.fdef, goal_pos, goal_num)?;
        let hf = helpful
            .into_iter()
            .filter(|&oi| matches!(kind[oi as usize], Kind::Start { .. } | Kind::Classical))
            .collect();
        Some((h, hf))
    } else {
        // relaxed_to with the task goal == the old `relaxed`; with a subgoal it
        // targets the contract (used by solve_from). Byte-identical for the default.
        let h = relaxed_to(task, sc, &s.bits, &s.fv, &s.fdef, goal_pos, goal_num)?;
        Some((h, Vec::new()))
    }
}

/// Dedup and enqueue a candidate node whose heuristic `(h, helpful)` is already
/// computed. Both the serial path ([`push_node`]) and the batched parallel path
/// funnel through this on ONE thread, in input order — that serial funnel is what
/// keeps parallel evaluation byte-identical to the sequential search.
#[allow(clippy::too_many_arguments)]
fn enqueue_evaluated(
    task: &PackedTask,
    nodes: &mut Vec<TNode>,
    heap: &mut BinaryHeap<Reverse<(i64, usize)>>,
    visited: &mut HashSet<(StateKey, Vec<(i64, usize)>)>,
    landmarks: &[NumPre],
    demand: &Demand,
    prune: bool,
    mut n: TNode,
    h: i32,
    helpful: Vec<u32>,
) {
    // Gentle h-weight (1g + 3h, vs the classical 1g+5h) keeps required-concurrency
    // branches in contention; the unit g breaks the flat-h plateau on long chains;
    // W_L counts unmet numeric-threshold landmarks, restoring a gradient on
    // converging DAGs (where the FF count goes flat). AGENDA_W is 0: penalizing open
    // intervals suppresses the very parallelism we want — keep it off.
    const W_G: i64 = 1;
    const W_H: i64 = 3;
    const W_L: i64 = 3;
    const AGENDA_W: i64 = 0;
    let k = tkey(task, &n);
    if visited.insert(k) {
        n.helpful = helpful;
        // Cumulative-availability for the demand term: parent's plus this op's
        // production. Empty (no-op) unless FF_TDEMAND is on.
        let op = n.ev.map(|(o, _)| o).unwrap_or(usize::MAX);
        n.met = if op == usize::MAX {
            nodes[n.father].met.clone()
        } else {
            met_child(&nodes[n.father].met, demand, task, op)
        };
        // Phase 1 (prune): weighted g+h plus the unmet-landmark term AND the
        // total converging-resource demand deficit (the multi-round gradient the
        // relaxation is blind to), to break the flat-h plateau on long chains AND
        // converging DAGs. Phase 2 (full): the ORIGINAL pure-h key — byte-for-byte
        // the old complete search, so nothing it solved before can regress.
        let key = if prune {
            W_G * n.g as i64
                + W_H * h as i64
                + W_L * landmark_deficit(landmarks, &n.state.fv, &n.state.fdef)
                + demand.weight * demand_deficit(&n.met, demand)
                + AGENDA_W * n.agenda.len() as i64
        } else {
            h as i64
        };
        let idx = nodes.len();
        nodes.push(n);
        heap.push(Reverse((key, idx)));
    }
}

/// Evaluate, dedup, and enqueue a candidate node with the weighted heap key.
#[allow(clippy::too_many_arguments)]
fn push_node(
    task: &PackedTask,
    kind: &[Kind],
    sc: &mut Scratch,
    nodes: &mut Vec<TNode>,
    heap: &mut BinaryHeap<Reverse<(i64, usize)>>,
    visited: &mut HashSet<(StateKey, Vec<(i64, usize)>)>,
    landmarks: &[NumPre],
    demand: &Demand,
    goal_pos: &[u32],
    goal_num: &[NumPre],
    prune: bool,
    n: TNode,
) {
    if let Some((h, helpful)) = eval_node(task, kind, sc, &n.state, goal_pos, goal_num, prune) {
        enqueue_evaluated(
            task, nodes, heap, visited, landmarks, demand, prune, n, h, helpful,
        );
    }
}

/// One decision-epoch search pass. `prune` restricts block-(a) expansion to the
/// node's helpful ops (with a per-node full-scan fallback so no node with a legal
/// successor is stranded); `false` is the full, complete search.
#[allow(clippy::too_many_arguments)]
fn temporal_search(
    task: &PackedTask,
    kind: &[Kind],
    landmarks: &[NumPre],
    demand: &Demand,
    start: &State,
    goal_pos: &[u32],
    goal_num: &[NumPre],
    forbidden: &[bool],
    relevant: &[bool],
    til_events: &[(f64, usize)],
    prune: bool,
    threads: usize,
) -> Option<TimedPlan> {
    const MAX_NODES: usize = 400_000;
    // Worker count for batched successor evaluation (0 = auto, like the classical
    // search). Parallelism only changes evaluation COST — see the funnel comment in
    // the expansion block; plans are identical for any value.
    let workers = if threads == 0 {
        crate::par::num_threads()
    } else {
        threads
    };
    // Root from the (possibly mid-composition) START state, but always at clock 0
    // with an agenda holding only the timed initial literals (sorted ascending): a
    // contract is solved as a fresh interval and drains its agenda before returning,
    // so it never inherits a parent's running durations.
    let init = start.clone();
    let mut sc = Scratch::new(task);

    let (_h0, hf0) = eval_node(task, kind, &mut sc, &init, goal_pos, goal_num, prune)?; // dead-end gate
    let mut root_agenda: Vec<(f64, usize)> = til_events.to_vec();
    root_agenda.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal));
    let mut nodes = vec![TNode {
        state: init,
        time: 0.0,
        agenda: root_agenda,
        father: usize::MAX,
        ev: None,
        g: 0,
        helpful: hf0,
        met: met_root(demand, task),
    }];
    let mut heap: BinaryHeap<Reverse<(i64, usize)>> = BinaryHeap::new();
    heap.push(Reverse((0, 0)));
    let mut visited: HashSet<(StateKey, Vec<(i64, usize)>)> = HashSet::new();
    visited.insert(tkey(task, &nodes[0]));

    while let Some(Reverse((_k, ni))) = heap.pop() {
        // The goal is reached once no *action* end is still pending. Unfired future
        // TILs may remain on the agenda — they're exogenous and don't gate completion.
        let ends_pending = nodes[ni]
            .agenda
            .iter()
            .any(|&(_, op)| !matches!(kind[op], Kind::Til));
        if task.goal_met_with(&nodes[ni].state, goal_pos, goal_num) && !ends_pending {
            let plan = reconstruct(task, &nodes, ni, kind);
            return Some(epsilon_separate(task, plan, !til_events.is_empty()));
        }
        if nodes.len() > MAX_NODES {
            break;
        }
        let time = nodes[ni].time;
        let pg = nodes[ni].g;

        // (a) start a durative action / apply a classical action — restricted to
        // the node's helpful set under pruning (else a full scan), minus any
        // forbidden ops (sibling protection; forbidding a START suffices).
        // forbidden (sibling protection) + goal-relevance pruning, both phases.
        // Empty relevance mask = keep all (default path). Sound: a non-relevant op
        // can't be on any path to this goal, so phase 2 stays complete on the
        // relevant subspace.
        let allow = |oi: usize| {
            !forbidden.get(oi).copied().unwrap_or(false)
                && (relevant.is_empty() || relevant.get(oi).copied().unwrap_or(true))
        };
        let candidates: Vec<usize> = if prune && !nodes[ni].helpful.is_empty() {
            nodes[ni]
                .helpful
                .iter()
                .map(|&o| o as usize)
                .filter(|&oi| allow(oi))
                .collect()
        } else {
            (0..task.n_ops).filter(|&oi| allow(oi)).collect()
        };
        // Successor prototypes first (cheap state application), heuristics second —
        // batched across worker threads when the frontier is big enough, then
        // funneled through `enqueue_evaluated` serially IN INPUT ORDER. The funnel
        // order matches the old per-candidate loop exactly, so heap and visited-set
        // evolve identically and the plan is byte-identical for any thread count.
        let mut protos: Vec<TNode> = Vec::new();
        for oi in candidates {
            match kind[oi] {
                Kind::Start { dur, end_op } => {
                    if task.op_applicable(oi, &nodes[ni].state) {
                        let ns = task.apply(oi, &nodes[ni].state);
                        let mut ag = nodes[ni].agenda.clone();
                        let te = time + dur;
                        let pos = ag.partition_point(|x| x.0 <= te);
                        ag.insert(pos, (te, end_op));
                        protos.push(TNode {
                            state: ns,
                            time,
                            agenda: ag,
                            father: ni,
                            ev: Some((oi, time)),
                            g: pg + 1,
                            helpful: Vec::new(),
                            met: Vec::new(),
                        });
                    }
                }
                Kind::Classical => {
                    if task.op_applicable(oi, &nodes[ni].state) {
                        let ns = task.apply(oi, &nodes[ni].state);
                        let ag = nodes[ni].agenda.clone();
                        protos.push(TNode {
                            state: ns,
                            time,
                            agenda: ag,
                            father: ni,
                            ev: Some((oi, time)),
                            g: pg + 1,
                            helpful: Vec::new(),
                            met: Vec::new(),
                        });
                    }
                }
                Kind::End | Kind::Til | Kind::Skip => {}
            }
        }
        // Small frontiers evaluate serially on the persistent Scratch (no per-round
        // allocation); big ones fan out with one fresh Scratch per worker. The
        // threshold is deliberately higher than the classical `par::MIN_PAR` (32):
        // this fans out PER EXPANSION POP, so the scoped-spawn cost recurs every
        // round — at 32-item frontiers it measurably loses (trade-bazaar +39% at
        // t8); it has to amortize against a full unpruned op scan to win.
        const PAR_FRONTIER: usize = 128;
        if workers <= 1 || protos.len() < PAR_FRONTIER {
            for n in protos {
                push_node(
                    task,
                    kind,
                    &mut sc,
                    &mut nodes,
                    &mut heap,
                    &mut visited,
                    landmarks,
                    demand,
                    goal_pos,
                    goal_num,
                    prune,
                    n,
                );
            }
        } else {
            let evals: Vec<Option<(i32, Vec<u32>)>> = crate::par::par_map_with(
                &protos,
                workers,
                || Scratch::new(task),
                |wsc, n| eval_node(task, kind, wsc, &n.state, goal_pos, goal_num, prune),
            );
            for (n, ev) in protos.into_iter().zip(evals) {
                if let Some((h, helpful)) = ev {
                    enqueue_evaluated(
                        task,
                        &mut nodes,
                        &mut heap,
                        &mut visited,
                        landmarks,
                        demand,
                        prune,
                        n,
                        h,
                        helpful,
                    );
                }
            }
        }

        // (b) advance time: fire the earliest pending agenda event — an action END or
        // a timed initial literal. Both apply their grounded op's effect at its time.
        if let Some(&(te, end_op)) = nodes[ni].agenda.first() {
            if task.op_applicable(end_op, &nodes[ni].state) {
                let ns = task.apply(end_op, &nodes[ni].state);
                let ag = nodes[ni].agenda[1..].to_vec();
                push_node(
                    task,
                    kind,
                    &mut sc,
                    &mut nodes,
                    &mut heap,
                    &mut visited,
                    landmarks,
                    demand,
                    goal_pos,
                    goal_num,
                    prune,
                    TNode {
                        state: ns,
                        time: te,
                        agenda: ag,
                        father: ni,
                        ev: Some((end_op, te)),
                        g: pg + 1,
                        helpful: Vec::new(),
                        met: Vec::new(),
                    },
                );
            }
        }
    }
    None
}

/// Walk the father chain into a timed plan: each START becomes a durative step
/// with its duration (the END is implied); END events are dropped; classical
/// actions appear instantaneously.
fn reconstruct(task: &PackedTask, nodes: &[TNode], goal: usize, kind: &[Kind]) -> TimedPlan {
    let mut events: Vec<(usize, f64)> = Vec::new();
    let mut cur = goal;
    while let Some((op, t)) = nodes[cur].ev {
        events.push((op, t));
        cur = nodes[cur].father;
    }
    events.reverse();

    let mut steps = Vec::new();
    let mut makespan = 0.0f64;
    for (op, t) in events {
        let disp = &task.op_display[op];
        let head = disp.split_whitespace().next().unwrap_or("");
        let args = disp
            .split_whitespace()
            .skip(1)
            .collect::<Vec<_>>()
            .join(" ");
        // Use the durations resolved in `solve` so constant and parameter-dependent
        // durative actions render identically. END events are implied by their start.
        let (name, duration) = match kind[op] {
            Kind::End => {
                makespan = makespan.max(t);
                continue;
            }
            // exogenous TIL firings are not plan steps and don't define the makespan.
            Kind::Til => continue,
            Kind::Start { dur, .. } => {
                makespan = makespan.max(t + dur);
                (head.trim_end_matches("-START"), Some(dur))
            }
            _ => {
                makespan = makespan.max(t);
                (head, None)
            }
        };
        let action = if args.is_empty() {
            name.to_string()
        } else {
            format!("{} {}", name, args)
        };
        steps.push(TimedStep {
            time: t,
            action,
            duration,
        });
    }
    TimedPlan { steps, makespan }
}

// ---------------------------------------------------------------------------
// Temporal plan validation (independent of the search).
// ---------------------------------------------------------------------------

/// Validate a [`TimedPlan`] against the temporal semantics, independently of how
/// it was produced: expand each durative step into a START happening at `t` and
/// an END happening at `t + duration`, order all happenings by time (ends before
/// starts at equal time), and simulate over the same snap-action compilation —
/// checking each happening's precondition + `over all` invariant holds, applying
/// its effects, cross-checking each duration against the domain expression, and
/// finally that the goal holds. Returns `Ok(())` if executable and goal-reaching,
/// else a human-readable reason. A cross-check on the search (and on any
/// externally-supplied plan).
pub fn validate(domain: &Domain, problem: &Problem, plan: &TimedPlan) -> Result<(), String> {
    let c = compile(domain, problem);
    let task = match ground(&c.domain, &c.problem, 1) {
        Outcome::Task(t) => t,
        Outcome::GoalTrue => {
            return if plan.steps.is_empty() {
                Ok(())
            } else {
                Err("goal is already true but the plan is non-empty".into())
            }
        }
        _ => return Err("problem grounds to unsolvable".into()),
    };
    let init = task.initial();
    let snap_by_start: HashMap<&str, &SnapInfo> = c
        .snaps
        .iter()
        .map(|s| (s.start_action.as_str(), s))
        .collect();
    let find = |disp: &str| {
        task.op_display
            .iter()
            .position(|d| d == disp)
            .ok_or_else(|| format!("plan references unknown action `{disp}`"))
    };

    struct Happening {
        time: f64,
        op: usize,
        is_start: bool,
    }
    let mut happenings: Vec<Happening> = Vec::new();
    for step in &plan.steps {
        let mut it = step.action.splitn(2, ' ');
        let head = it.next().unwrap_or("");
        let rest = it.next();
        let with = |suffix: &str| match rest {
            Some(r) => format!("{head}{suffix} {r}"),
            None => format!("{head}{suffix}"),
        };
        match step.duration {
            Some(dur) => {
                let start_name = format!("{head}-START");
                let snap = snap_by_start
                    .get(start_name.as_str())
                    .ok_or_else(|| format!("`{head}` is not a durative action"))?;
                // cross-check the stated duration against the domain's constraint:
                // it must fall within the `[min, max]` range (a fixed `=` collapses the
                // range to a point, recovering exact-equality).
                let args: Vec<&str> = rest
                    .map(|r| r.split_whitespace().collect())
                    .unwrap_or_default();
                let (lo, hi) = eval_duration_bounds(snap, &args, &task, &init);
                if let Some(min) = lo {
                    if dur < min - 1e-6 {
                        return Err(format!(
                            "`{}` has duration {dur} below the domain minimum {min}",
                            step.action
                        ));
                    }
                }
                if let Some(max) = hi {
                    if dur > max + 1e-6 {
                        return Err(format!(
                            "`{}` has duration {dur} above the domain maximum {max}",
                            step.action
                        ));
                    }
                }
                happenings.push(Happening {
                    time: step.time,
                    op: find(&with("-START"))?,
                    is_start: true,
                });
                happenings.push(Happening {
                    time: step.time + dur,
                    op: find(&with("-END"))?,
                    is_start: false,
                });
            }
            None => happenings.push(Happening {
                time: step.time,
                op: find(&step.action)?,
                is_start: true,
            }),
        }
    }

    // Replay timed initial literals as exogenous happenings, up to the plan horizon
    // (the last action happening). A TIL strictly after the plan's end is beyond the
    // plan's interval and must not retroactively undo the end-state goal check.
    let horizon = happenings.iter().map(|h| h.time).fold(0.0f64, f64::max);
    for (t, name) in &c.til_ops {
        if *t <= horizon + EPS {
            happenings.push(Happening {
                time: *t,
                op: find(name)?,
                // fire with ends (before starts) at the same epoch, so a gate the TIL
                // opens is available to an action starting at that instant.
                is_start: false,
            });
        }
    }

    // Execute in time order; at the SAME decision epoch, ends (which free
    // tokens/resources) fire before starts (which consume them) — ferroplan's
    // decision-epoch semantics. Key on the ε-grid-rounded time, not the raw float,
    // so a producer-END and consumer-START at the same epoch order deterministically
    // even when composition offsets introduce sub-ε float noise.
    happenings.sort_by_key(|h| ((h.time / EPS).round() as i64, h.is_start));
    let mut state = init.clone();
    for h in &happenings {
        if !task.op_applicable(h.op, &state) {
            return Err(format!(
                "at t={:.3}, `{}` is not applicable (precondition or invariant violated)",
                h.time, task.op_display[h.op]
            ));
        }
        state = task.apply(h.op, &state);
    }
    if !task.goal_met(&state) {
        return Err("the plan does not achieve the goal".into());
    }
    Ok(())
}

/// Replay a composed `TimedPlan` over `state` in global-time happening order (ends
/// before starts at equal time) and return the post-state, or `None` if any
/// happening is inapplicable on the running state (a shared-resource shortfall or
/// stale precondition — the decomposer's conflict signal). Mirrors `validate`'s
/// simulation loop, minus the duration cross-check and goal check, over the SAME
/// grounded `task` whose `op_display` the plan's steps name.
pub(crate) fn treplay(task: &PackedTask, state: &State, plan: &TimedPlan) -> Option<State> {
    let find = |disp: &str| task.op_display.iter().position(|d| d == disp);
    struct H {
        time: f64,
        op: usize,
        is_start: bool,
    }
    let mut hs: Vec<H> = Vec::new();
    for step in &plan.steps {
        let mut it = step.action.splitn(2, ' ');
        let head = it.next().unwrap_or("");
        let rest = it.next();
        let with = |suffix: &str| match rest {
            Some(r) => format!("{head}{suffix} {r}"),
            None => format!("{head}{suffix}"),
        };
        match step.duration {
            Some(dur) => {
                hs.push(H {
                    time: step.time,
                    op: find(&with("-START"))?,
                    is_start: true,
                });
                hs.push(H {
                    time: step.time + dur,
                    op: find(&with("-END"))?,
                    is_start: false,
                });
            }
            None => hs.push(H {
                time: step.time,
                op: find(&step.action)?,
                is_start: true,
            }),
        }
    }
    // Same ε-grid-rounded ordering as `validate` (ends before starts at one epoch),
    // so the decomposer's per-contract replay agrees with the global validator.
    hs.sort_by_key(|h| ((h.time / EPS).round() as i64, h.is_start));
    let mut s = state.clone();
    for h in &hs {
        if !task.op_applicable(h.op, &s) {
            return None;
        }
        s = task.apply(h.op, &s);
    }
    Some(s)
}

// ---------------------------------------------------------------------------
// ε-separation: make plans valid under PDDL2.1 continuous-time semantics.
// ---------------------------------------------------------------------------

/// PDDL2.1 separation between mutex happenings (the IPC convention).
const EPS: f64 = 0.001;

fn slices_intersect(a: &[u32], b: &[u32]) -> bool {
    // slices are tiny (a handful of facts), so the quadratic scan is fine
    a.iter().any(|x| b.contains(x))
}

/// Do two grounded happenings interfere (PDDL2.1 mutex)? True if one's add/del
/// clashes with the other's precondition, add, or delete on a shared fact —
/// requiring them to be ε-separated rather than simultaneous.
fn ops_mutex(task: &PackedTask, o1: usize, o2: usize) -> bool {
    let (p1, a1, d1) = (
        task.pre_pos.slice(o1),
        task.add.slice(o1),
        task.del.slice(o1),
    );
    let (p2, a2, d2) = (
        task.pre_pos.slice(o2),
        task.add.slice(o2),
        task.del.slice(o2),
    );
    slices_intersect(a1, d2)
        || slices_intersect(d1, a2)
        || slices_intersect(a1, p2)
        || slices_intersect(p1, a2)
        || slices_intersect(d1, p2)
        || slices_intersect(p1, d2)
}

/// Re-time a plan so mutex happenings are ε-separated (PDDL2.1 / VAL validity):
/// the decision-epoch search coincides dependent happenings (e.g. one action
/// starting the instant another's at-end effect lands), which VAL rejects. We
/// model the plan's happenings as a simple temporal network — preserve the
/// execution order, pin each end at start+duration, force ε between mutex pairs —
/// and solve the earliest-time schedule by longest paths (Bellman–Ford). On any
/// inconsistency or for very large plans the original plan is returned unchanged.
fn epsilon_separate(task: &PackedTask, plan: TimedPlan, floor_to_search: bool) -> TimedPlan {
    // happening: (op id, owning step index, is_start)
    struct H {
        op: usize,
        step: usize,
        is_start: bool,
        time: f64,
    }
    let find = |disp: &str| task.op_display.iter().position(|d| d == disp);
    let mut hs: Vec<H> = Vec::new();
    for (si, step) in plan.steps.iter().enumerate() {
        let mut it = step.action.splitn(2, ' ');
        let head = it.next().unwrap_or("");
        let rest = it.next();
        match step.duration {
            Some(dur) => {
                let sd = match rest {
                    Some(r) => format!("{head}-START {r}"),
                    None => format!("{head}-START"),
                };
                let ed = match rest {
                    Some(r) => format!("{head}-END {r}"),
                    None => format!("{head}-END"),
                };
                match (find(&sd), find(&ed)) {
                    (Some(so), Some(eo)) => {
                        hs.push(H {
                            op: so,
                            step: si,
                            is_start: true,
                            time: step.time,
                        });
                        hs.push(H {
                            op: eo,
                            step: si,
                            is_start: false,
                            time: step.time + dur,
                        });
                    }
                    _ => return plan, // can't map -> leave as-is
                }
            }
            None => match find(&step.action) {
                Some(o) => hs.push(H {
                    op: o,
                    step: si,
                    is_start: true,
                    time: step.time,
                }),
                None => return plan,
            },
        }
    }
    let n = hs.len();
    if n == 0 || n > 600 {
        return plan; // nothing to do, or too large to schedule cheaply
    }
    // execution order: by time, ends before starts at equal time
    let mut order: Vec<usize> = (0..n).collect();
    order.sort_by(|&a, &b| {
        hs[a]
            .time
            .partial_cmp(&hs[b].time)
            .unwrap_or(std::cmp::Ordering::Equal)
            .then(hs[a].is_start.cmp(&hs[b].is_start))
    });

    // STN edges: t[v] >= t[u] + w
    let mut edges: Vec<(usize, usize, f64)> = Vec::new();
    // preserve order (weak)
    for w in order.windows(2) {
        edges.push((w[0], w[1], 0.0));
    }
    // ε between mutex happenings (in execution order)
    for i in 0..n {
        for j in (i + 1)..n {
            let (u, v) = (order[i], order[j]);
            if ops_mutex(task, hs[u].op, hs[v].op) {
                edges.push((u, v, EPS));
            }
        }
    }
    // duration equality: end = start + dur  (two inequalities)
    for si in 0..plan.steps.len() {
        if let Some(dur) = plan.steps[si].duration {
            let (mut s, mut e) = (None, None);
            for (hi, h) in hs.iter().enumerate() {
                if h.step == si {
                    if h.is_start {
                        s = Some(hi)
                    } else {
                        e = Some(hi)
                    }
                }
            }
            if let (Some(s), Some(e)) = (s, e) {
                edges.push((s, e, dur));
                edges.push((e, s, -dur));
            }
        }
    }

    // longest-path (earliest feasible times) via Bellman–Ford. With timed initial
    // literals present, seed each happening at its SEARCH-assigned time as a lower
    // bound (the search already placed every happening at a TIL-feasible instant);
    // relaxation only pushes later, so a TIL-gated action can't be slid before its
    // gate. Without TILs, seed at 0 — byte-identical to the prior re-timing.
    let mut t: Vec<f64> = if floor_to_search {
        hs.iter().map(|h| h.time).collect()
    } else {
        vec![0.0f64; n]
    };
    for _ in 0..n {
        let mut changed = false;
        for &(u, v, w) in &edges {
            if t[v] < t[u] + w - 1e-12 {
                t[v] = t[u] + w;
                changed = true;
            }
        }
        if !changed {
            break;
        }
    }
    // positive-cycle check: another pass must not improve
    for &(u, v, w) in &edges {
        if t[v] < t[u] + w - 1e-12 {
            return plan; // inconsistent ordering -> keep original
        }
    }

    // re-time the steps from the scheduled start happenings
    let mut steps = plan.steps;
    for (hi, h) in hs.iter().enumerate() {
        if h.is_start {
            steps[h.step].time = t[hi];
        }
    }
    let makespan = steps
        .iter()
        .map(|s| s.time + s.duration.unwrap_or(0.0))
        .fold(0.0f64, f64::max);
    TimedPlan { steps, makespan }
}