From 99c7484f52f57b5dcbff595c2c18db2929bf9c25 Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Mon, 31 Aug 2026 06:11:15 -0400 Subject: [PATCH 1/8] sysid: honest fit report (like-for-like SSE, no-op banner), offline diagnosis script Stage 1 of the inert-fitter work from the 2026-08-30 bridge run review: - run_rollout_sysid computes the init-params SSE on the SAME survivor set post_sse is computed on (pre_sse_survivors); pre_sse still covers all segments. The sim.fit report's "% SSE reduction" is now measured on that like-for-like baseline - the old ratio compared all-segment init SSE against survivor-only fitted SSE, so with 21/27 segments trimmed it reported "74% SSE reduction" from a fit whose every parameter delta was +0.0000. - The report opens with an explicit banner when the optimizer moved no parameter: the values below are declared inits, not estimates. - The identifiability report gives contraction > 1 its own label ("posterior Nx WIDER than the prior - the data carries ~no information") instead of calling 4.4x or 33x "posterior ~= prior". Stage 2 tooling: scripts/sysid_fit_diagnosis.py replays a run's fit offline from the persisted fit_data pickle + the run's simulator.py (same prep, scaling, and trimming), then probes the survivor-set SSE with each parameter alone at its range extremes, and the same probe on the trimmed-away segments - separating "objective genuinely flat" from "the optimizer missed a real slope" and from "the information lives in the segments the trimming dropped". Verified: mypy clean, pylint clean, 243 tests green across code_sim_learning + the sim-learning approach suites (all domains). Claude-Session: https://claude.ai/code/session_017Acb4dHk2Ryju5XFvcjdNU --- predicators/agent_sdk/tools/synthesis.py | 38 ++- .../code_sim_learning/identifiability.py | 12 +- predicators/code_sim_learning/orchestrator.py | 48 +++- scripts/sysid_fit_diagnosis.py | 258 ++++++++++++++++++ 4 files changed, 338 insertions(+), 18 deletions(-) create mode 100644 scripts/sysid_fit_diagnosis.py diff --git a/predicators/agent_sdk/tools/synthesis.py b/predicators/agent_sdk/tools/synthesis.py index 93ffeb357..4dafae547 100644 --- a/predicators/agent_sdk/tools/synthesis.py +++ b/predicators/agent_sdk/tools/synthesis.py @@ -447,11 +447,23 @@ def _evaluate_rollout_fit(rules: list, len(rollouts), outcome.traj_rms) kept_at_init = sorted(n for n in physical_names if applied[n] != fitted[n]) - if pre_sse > 0: - pct_str = (f"({(pre_sse - post_sse) / pre_sse * 100:.1f}% " - "SSE reduction vs init)") + # Like-for-like SSE headline: the % reduction is measured on the + # SAME segment set post_sse was computed on (the trimming + # survivors). pre_sse covers all segments, so the old ratio + # counted the trimming as fit improvement - run_20260830 + # reported "74% SSE reduction" from a fit that moved nothing. + pre_surv = getattr(outcome, "pre_sse_survivors", float("nan")) + if not np.isfinite(pre_surv): + pre_surv = pre_sse + if pre_surv > 0: + pct = (pre_surv - post_sse) / pre_surv * 100 + pct_str = (f"({pct:.1f}% SSE reduction vs init on the SAME " + "segments)") else: - pct_str = "(init SSE was 0)" + pct_str = "(init SSE on these segments was 0)" + n_surv = outcome.num_survivors + surv_note = (f" on the {n_surv} surviving segments" + if n_surv < len(rollouts) else "") mode_note = ( f"EXPLORATORY, trajectories {sorted(traj_idxs or [])} only" if exploratory else "canonical") @@ -465,11 +477,25 @@ def _evaluate_rollout_fit(rules: list, "per-feature normalized (angles wrapped), so SSE/RMS are " "dimensionless fractions of typical motion.", "", - f"At init params: rollout SSE = {pre_sse:.6f}", - f"After joint fit: rollout SSE = {post_sse:.6f} {pct_str}", + f"At init params: rollout SSE = {pre_sse:.6f} " + f"(all {len(rollouts)} segments)" + + (f", {pre_surv:.6f}{surv_note}" if surv_note else ""), + f"After joint fit: rollout SSE = {post_sse:.6f}{surv_note} " + f"{pct_str}", "", "Fitted parameters:", ] + # An inert fit must announce itself: every value below is then + # the declared init, not an estimate, and "fitted" language + # would launder hand-set constants as calibration. + if fitted and all(fitted[n] == init_params[n] for n in fitted): + lines.insert( + 1, "NOTE: the optimizer moved NO parameter (every delta " + "is +0.0000) - this fit is a no-op and the values below " + "are exactly the declared init_values, not estimates. " + "The data as trimmed does not pull any parameter away " + "from its starting point; treat the model as running on " + "hand-set constants.") if outcome.num_survivors < len(rollouts): rms_str = ", ".join(f"{r:.4g}" for r in outcome.traj_rms) dropped = len(rollouts) - outcome.num_survivors diff --git a/predicators/code_sim_learning/identifiability.py b/predicators/code_sim_learning/identifiability.py index e55029c31..49a911f7e 100644 --- a/predicators/code_sim_learning/identifiability.py +++ b/predicators/code_sim_learning/identifiability.py @@ -187,9 +187,19 @@ def identifiability_report( verdict = Verdict.IDENTIFIED elif contraction < _WEAK_CONTRACTION: verdict = Verdict.WEAKLY_IDENTIFIED - else: + elif contraction <= 1.0 + 1e-9: verdict = Verdict.NOT_IDENTIFIED note = "posterior ~= prior; MAP arbitrary" + else: + # Contraction > 1 is not "~= prior": the reported width + # EXCEEDS the prior (the curvature probe read near-zero + # information and the width floor / flat-likelihood spread + # inflated it). Calling 4.4x or 33x "~= prior" hid exactly + # this in run_20260830's fit reports. + verdict = Verdict.NOT_IDENTIFIED + note = (f"posterior {contraction:.2g}x WIDER than the prior - " + "the data carries ~no information on this parameter " + "(width floor / flat-likelihood spread); MAP arbitrary") if (verdict is Verdict.IDENTIFIED and num_explainable is not None and num_explainable < 2): verdict = Verdict.WEAKLY_IDENTIFIED diff --git a/predicators/code_sim_learning/orchestrator.py b/predicators/code_sim_learning/orchestrator.py index 80d184e1d..77cfa95b3 100644 --- a/predicators/code_sim_learning/orchestrator.py +++ b/predicators/code_sim_learning/orchestrator.py @@ -72,6 +72,12 @@ class SysIdOutcome: traj_rms: List[float] pre_sse: float post_sse: float + # Init-params SSE on the SAME survivor set ``post_sse`` is computed + # on - the like-for-like baseline for any "reduction" claim. + # ``pre_sse`` covers all segments, so pre/post ratios across the two + # sets measure the trimming, not the fit (run_20260830: "74% SSE + # reduction" reported while the optimizer moved no parameter). + pre_sse_survivors: float = float("nan") hull_candidates: List[Dict[str, float]] = field(default_factory=list) from_cache: bool = False @@ -87,6 +93,8 @@ class _FitComputation: hull_candidates: List[Dict[str, float]] = field(default_factory=list) pre_sse: float = float("nan") post_sse: float = float("nan") + # See SysIdOutcome.pre_sse_survivors. + pre_sse_survivors: float = float("nan") # SSE of an arbitrary joint theta on the fit's surviving segments # with the fit's own scaling (the closure identifiability_report # consumed); None when no fit ran. Cache-safe: it closes over the @@ -183,17 +191,21 @@ def run_rollout_sysid( num_rollouts_run() - n0, time.monotonic() - t0, " (fit cache hit)" if from_cache else "") - return SysIdOutcome(fit_result=core.fit_result, - report=report, - fitted=fitted, - applied=applied, - num_segments=len(rollouts), - num_survivors=core.num_survivors, - traj_rms=list(core.traj_rms), - pre_sse=core.pre_sse, - post_sse=core.post_sse, - hull_candidates=list(core.hull_candidates), - from_cache=from_cache) + return SysIdOutcome( + fit_result=core.fit_result, + report=report, + fitted=fitted, + applied=applied, + num_segments=len(rollouts), + num_survivors=core.num_survivors, + traj_rms=list(core.traj_rms), + pre_sse=core.pre_sse, + post_sse=core.post_sse, + # getattr: a _FitComputation unpickled from a pre-field + # checkpoint restores its saved __dict__ without the default. + pre_sse_survivors=getattr(core, "pre_sse_survivors", float("nan")), + hull_candidates=list(core.hull_candidates), + from_cache=from_cache) def _log_data_health(report: Dict[str, Dict[str, Any]], @@ -289,6 +301,19 @@ def _compute_fit( residual_features, physical_names, rules, latent_init, scaling) logger.info("Rollout sysID - post-SSE: %.6f", post_sse) + # Init-params SSE on the SAME survivor set: the only baseline a + # "reduction from the fit" claim may be measured against (see + # SysIdOutcome.pre_sse_survivors). Costs len(survivors) extra + # rollouts once per uncached fit; skipped when nothing was trimmed. + if len(survivors) == len(rollouts): + pre_sse_survivors = pre_sse + else: + pre_sse_survivors = compute_rollout_sse(fit_env, survivors, + init_params, residual_features, + physical_names, rules, + latent_init, scaling) + logger.info("Rollout sysID - pre-SSE on the %d survivors: %.6f", + len(survivors), pre_sse_survivors) def rollout_sse_fn(params: Dict[str, float]) -> float: return compute_rollout_sse(fit_env, survivors, params, @@ -317,4 +342,5 @@ def rollout_sse_fn(params: Dict[str, float]) -> float: hull_candidates=list(hull_candidates), pre_sse=pre_sse, post_sse=post_sse, + pre_sse_survivors=pre_sse_survivors, sse_fn=rollout_sse_fn) diff --git a/scripts/sysid_fit_diagnosis.py b/scripts/sysid_fit_diagnosis.py new file mode 100644 index 000000000..9960848e0 --- /dev/null +++ b/scripts/sysid_fit_diagnosis.py @@ -0,0 +1,258 @@ +"""Offline diagnosis of a run's rollout system-ID fit. + +Answers ONE question about a recorded fit that moved no parameter +("SSE X -> X in 1 fn-evals"): is the objective genuinely flat in every +parameter on the data the fit was given (unworkable input), or does the +data respond and the optimizer failed to see the slope (a fitting bug)? + +Replays the fit exactly as ``sim.fit`` ran it - same persisted +trajectories (``/fit_data/*.pkl``), same agent artifact +(``/sandbox/simulator.py``), same prep (settled-tail +truncation, rest-point segmentation), same scaling and trimming - then +probes the survivor-set SSE (the objective the optimizer minimized) +with each parameter moved alone to the extremes of its declared range. +The same probe is repeated on the trimmed-away segments, so a flat +survivor objective can be told apart from "the information lives in the +segments the trimming dropped". + +Usage (compute node; ~10 min of fresh-env rollouts for a bridge-scale +artifact): + + python scripts/sysid_fit_diagnosis.py \ + --run_dir logs/.../seed1/run_20260830_145216 --env pybullet_bridge + +``--smoke`` loads, preps, and reports segment statistics without +running any rollout (login-node safe). +""" +import argparse +import pickle +import sys +import time +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +# pylint: disable=wrong-import-position +from predicators import utils +from predicators.code_sim_learning.fit_space import ParamSpec +from predicators.code_sim_learning.physical_sysid import \ + fit_params_rollout_trimmed +from predicators.code_sim_learning.rollout_env import RolloutTrajectory, \ + physical_param_anchors +from predicators.code_sim_learning.rollout_objective import compute_rollout_sse +from predicators.code_sim_learning.trajectory_prep import \ + compute_residual_scaling, split_at_rest_points, truncate_settled_tail +from predicators.code_sim_learning.utils import read_latent_init, \ + read_physical_param_specs, read_simulator_components +from predicators.envs import create_new_env +from predicators.settings import CFG + +# A parameter "responds" when moving it alone shifts the probed SSE by +# more than this fraction of the baseline (plus a tiny absolute floor +# for near-zero baselines). +_RESPONSE_REL = 1e-3 +_RESPONSE_ABS = 1e-6 + + +def _load_fit_data(run_dir: Path, pickle_name: Optional[str]) -> Dict: + fit_dir = run_dir / "fit_data" + if pickle_name: + path = fit_dir / pickle_name + else: + pickles = sorted(fit_dir.glob("*.pkl")) + if not pickles: + sys.exit(f"No fit_data pickles under {fit_dir}") + path = pickles[-1] + print(f"Loading fit data: {path}") + with open(path, "rb") as f: + return pickle.load(f) + + +def _load_artifact(run_dir: Path) -> Tuple[list, list, Dict, Any, list]: + """Exec the agent's simulator.py the way ``sim.fit`` does.""" + path = run_dir / "sandbox" / "simulator.py" + if not path.is_file(): + sys.exit(f"No agent artifact at {path}") + print(f"Loading artifact: {path}") + ns: Dict[str, Any] = {"np": np, "ParamSpec": ParamSpec} + exec(path.read_text(encoding="utf-8"), ns) # pylint: disable=exec-used + rules, specs, features = read_simulator_components(ns) + latent_init = read_latent_init(ns) + physical_specs = read_physical_param_specs(ns) or [] + if features is None: + sys.exit("Artifact declares no RESIDUAL_FEATURES.") + return rules or [], specs or [], features, latent_init, physical_specs + + +def _prep_rollouts(trajectories: list, + residual_features: Dict) -> List[RolloutTrajectory]: + """Mirror ``_rollout_fit_trajectories``: whole trajs -> truncate -> + segment.""" + rollouts: List[RolloutTrajectory] = [] + for traj in trajectories: + if traj.actions and len(traj.states) == len(traj.actions) + 1: + rollouts.append((list(traj.states), list(traj.actions))) + if CFG.code_sim_learning_rollout_truncate_settled: + rollouts = [ + truncate_settled_tail(r, residual_features) for r in rollouts + ] + if CFG.code_sim_learning_rollout_segment_on_rest: + segments: List[RolloutTrajectory] = [] + for r in rollouts: + segments.extend(split_at_rest_points(r, residual_features)) + if segments: + rollouts = segments + return rollouts + + +def _probe_values(spec: ParamSpec) -> List[float]: + """The extreme values a parameter is probed at (besides its init).""" + lo, hi = spec.lo, spec.hi + if lo is None or hi is None: + # No declared box: probe a wide multiplicative/additive spread. + if spec.scale == "log": + return [spec.init_value / 4.0, spec.init_value * 4.0] + span = max(abs(spec.init_value), 1e-3) + return [spec.init_value - span, spec.init_value + span] + return [float(lo), float(hi)] + + +def _probe_set(label: str, segments: List[RolloutTrajectory], + specs: List[ParamSpec], init_params: Dict[str, float], + sse_at: Any) -> None: + """Print, per parameter, the SSE with that parameter alone at its range + extremes, and a FLAT/RESPONDS verdict.""" + base = sse_at(segments, init_params) + print(f"\n== {label}: {len(segments)} segment(s), " + f"SSE at init = {base:.6f}") + n_flat = 0 + for spec in specs: + deltas = [] + for val in _probe_values(spec): + probed = dict(init_params) + probed[spec.name] = val + deltas.append(sse_at(segments, probed) - base) + thresh = _RESPONSE_ABS + _RESPONSE_REL * abs(base) + flat = all(abs(d) < thresh for d in deltas) + n_flat += int(flat) + verdict = "FLAT " if flat else "RESPONDS" + delta_str = ", ".join(f"{d:+.6f}" for d in deltas) + vals_str = ", ".join(f"{v:g}" for v in _probe_values(spec)) + print(f" {verdict} {spec.name:<22} at [{vals_str}]: " + f"dSSE [{delta_str}]") + print(f" -> {n_flat}/{len(specs)} parameters FLAT across their " + f"whole range on this segment set.") + + +def _main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run_dir", required=True, type=Path) + parser.add_argument("--env", required=True, type=str) + parser.add_argument("--seed", type=int, default=0) + parser.add_argument("--pickle", + type=str, + default=None, + help="fit_data pickle name (default: latest)") + parser.add_argument("--max_dropped", + type=int, + default=4, + help="how many trimmed-away segments to probe") + parser.add_argument("--smoke", + action="store_true", + help="load + prep only; run no rollouts") + args = parser.parse_args() + + utils.reset_config({"env": args.env, "seed": args.seed}) + + payload = _load_fit_data(args.run_dir, args.pickle) + trajectories = payload["trajectories"] + pickled_physical = payload.get("physical_param_specs") or [] + identified = payload.get("identified_physical_params") or {} + print(f"{len(trajectories)} recorded trajectories; " + f"{len(pickled_physical)} pickled physical specs; " + f"identified at record time: {identified}") + + rules, rule_specs, features, latent_init, physical_specs = \ + _load_artifact(args.run_dir) + if not physical_specs: + physical_specs = list(pickled_physical) + physical_names = [s.name for s in physical_specs] + all_specs = list(physical_specs) + list(rule_specs) + init_params = {s.name: s.init_value for s in all_specs} + print(f"Artifact: {len(rules)} rules, {len(rule_specs)} rule specs, " + f"{len(physical_specs)} physical specs, " + f"latent_init={'yes' if latent_init is not None else 'no'}") + + rollouts = _prep_rollouts(trajectories, features) + print(f"Prep: {len(rollouts)} motion segments " + f"(lengths {[len(a) for _s, a in rollouts]})") + if args.smoke: + print("--smoke: stopping before any rollout.") + return + + def fit_env() -> Any: + env = create_new_env(CFG.env, + do_cache=False, + use_gui=False, + skip_residual_dynamics=True) + if identified: + env.apply_physical_param_overrides(dict(identified)) + return env + + scaling = compute_residual_scaling(rollouts, features) + anchors_env = fit_env() + anchors = physical_param_anchors(anchors_env, physical_specs) + + def sse_at(segments: List[RolloutTrajectory], + params: Dict[str, float]) -> float: + return compute_rollout_sse(fit_env, segments, params, features, + physical_names, rules, latent_init, scaling) + + t0 = time.monotonic() + print("\nReplaying the trimming + fit exactly as sim.fit ran it...") + result, survivors, rms, _hull = fit_params_rollout_trimmed( + fit_env, + rollouts, + physical_specs, + features, + rules=rules, + rule_specs=rule_specs, + latent_init=latent_init, + scaling=scaling, + anchors=anchors) + fitted = result.point_estimate + moved = { + n: (init_params[n], fitted[n]) + for n in fitted if fitted[n] != init_params[n] + } + print(f"Fit replay: {len(survivors)}/{len(rollouts)} segments " + f"survived trimming (per-segment best RMS: " + f"{[f'{r:.4g}' for r in rms]}); " + f"parameters moved by the fit: {moved or 'NONE'} " + f"[{time.monotonic() - t0:.1f}s]") + + # The verdict probes. (1) The optimizer's own objective: SSE over + # the survivors. FLAT everywhere = the fit had nothing to work + # with; any RESPONDS row = the optimizer missed a real slope. + _probe_set("SURVIVORS (the fit's objective)", survivors, all_specs, + init_params, sse_at) + + # (2) The trimmed-away segments, most-informative first: if these + # respond where the survivors are flat, the trimming discarded the + # only segments that carried parameter information. + dropped_idx = [ + i for i, r in enumerate(rollouts) if not any(r is s for s in survivors) + ] + dropped_idx.sort(key=lambda i: -rms[i]) + dropped = [rollouts[i] for i in dropped_idx[:args.max_dropped]] + if dropped: + _probe_set(f"DROPPED (top {len(dropped)} by best-achievable RMS)", + dropped, all_specs, init_params, sse_at) + print(f"\nTotal wall time: {time.monotonic() - t0:.1f}s") + + +if __name__ == "__main__": + _main() From 8b7c408947da48126ec4f692d93cb2e99d3a9d57 Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Mon, 31 Aug 2026 06:18:02 -0400 Subject: [PATCH 2/8] sysid_fit_diagnosis: replay the run's logged flags to rebuild its env With defaults-only CFG the fresh fit env could not reconstruct the recorded states (env construction depends on weld pinning, contact margins, and more), so the script now parses the exact main.py invocation from the run's info.log and replays those flags, with --env/--seed as optional overrides. Claude-Session: https://claude.ai/code/session_017Acb4dHk2Ryju5XFvcjdNU --- scripts/sysid_fit_diagnosis.py | 56 ++++++++++++++++++++++++++++++++-- 1 file changed, 53 insertions(+), 3 deletions(-) diff --git a/scripts/sysid_fit_diagnosis.py b/scripts/sysid_fit_diagnosis.py index 9960848e0..de14d06ab 100644 --- a/scripts/sysid_fit_diagnosis.py +++ b/scripts/sysid_fit_diagnosis.py @@ -26,6 +26,7 @@ """ import argparse import pickle +import re import sys import time from pathlib import Path @@ -57,6 +58,51 @@ _RESPONSE_ABS = 1e-6 +def _apply_run_config(run_dir: Path, env: Optional[str], + seed: Optional[int]) -> None: + """Reproduce the run's CFG by replaying its logged command line. + + Env construction depends on far more than the env name (weld + pinning, contact margins, wait caps, ...): with defaults-only CFG + the fresh env could not even reconstruct the recorded states (blocks + parked at their creation poses). The run's info.log logs the exact + ``main.py`` invocation; replay its flags verbatim, with ``--env`` / + ``--seed`` as explicit overrides. + """ + argv: List[str] = [] + info = run_dir / "info.log" + if info.is_file(): + with open(info, encoding="utf-8") as f: + for line in f: + if "Running command:" not in line: + continue + tokens = re.sub(r"\x1b\[[0-9;]*m", "", line).split() + mains = [ + i for i, t in enumerate(tokens) if t.endswith("main.py") + ] + if mains: + argv = tokens[mains[0] + 1:] + break + if not argv: + if env is None: + sys.exit(f"No logged command in {info} - pass --env explicitly.") + utils.reset_config({"env": env, "seed": seed or 0}) + return + old_argv = sys.argv + try: + sys.argv = ["sysid_fit_diagnosis"] + argv + parsed = utils.parse_args() + finally: + sys.argv = old_argv + if env is not None: + parsed["env"] = env + if seed is not None: + parsed["seed"] = seed + utils.update_config(parsed) + print(f"Replayed {len(argv) // 2} flags from the run's logged command " + f"(env={CFG.env}, seed={CFG.seed}).") + + def _load_fit_data(run_dir: Path, pickle_name: Optional[str]) -> Dict: fit_dir = run_dir / "fit_data" if pickle_name: @@ -150,8 +196,12 @@ def _probe_set(label: str, segments: List[RolloutTrajectory], def _main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run_dir", required=True, type=Path) - parser.add_argument("--env", required=True, type=str) - parser.add_argument("--seed", type=int, default=0) + parser.add_argument("--env", + type=str, + default=None, + help="override the run's own env (default: replay " + "the run's logged flags)") + parser.add_argument("--seed", type=int, default=None) parser.add_argument("--pickle", type=str, default=None, @@ -165,7 +215,7 @@ def _main() -> None: help="load + prep only; run no rollouts") args = parser.parse_args() - utils.reset_config({"env": args.env, "seed": args.seed}) + _apply_run_config(args.run_dir, args.env, args.seed) payload = _load_fit_data(args.run_dir, args.pickle) trajectories = payload["trajectories"] From 8590f9f98a0084ca17c8f5ff7d8b2bba4f73b780 Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Mon, 31 Aug 2026 12:57:39 -0400 Subject: [PATCH 3/8] sysid: fork-parallel rollout objective, revive the MAP-path bracket search Two coupled fixes to the rollout fit, from the 2026-08-31 diagnosis of the bridge runs (98 and 34 fit-minutes per run for zero parameter movement, with one real improvement missed): Parallel objective. _iter_rollout_residual_terms fans whole trajectories out to the existing prefetch_parallel fork pool when the env is a per-rollout factory - the case where a child's term list is bit-identical to the serial path's, as the finite-difference Jacobian requires. Every stage (LM Jacobian, grid seed, sensitivity, ablation) funnels through it, so all inherit the speedup; child rollouts are credited to the parent's counter (add_rollouts_run) and the pool's per-call log line drops to DEBUG here (hundreds of calls per fit). Shared env instances keep the serial path untouched. Bracket-search revival. The zero-gradient bracket search for threshold/gate params was dead code on the rollout MAP path: the Gaussian prior rows have structurally nonzero derivative in every column, so the exact-0.0 all-rows column test could never fire. zero_jacobian_columns now tests the DATA rows only, with a tolerance (abs 1e-8 / rel 1e-6) instead of exact zero. The search itself now takes the residual vector: flat verdicts are judged on data-rows SSE (at a box edge the prior rows alone add orders of magnitude more than the flat tolerance), while the argmin and move acceptance stay on the full MAP objective. Box edges are probed first, settling a flat parameter in 2 evaluations instead of 9; params measured flat across their box feed FitResult.sensitivity as INSENSITIVE evidence, sparing the identifiability probe their 2 rollout evals each, and the bracket notes ride FitResult.lm_notes into the sim.fit report. sysid_fit_diagnosis gains --parallel_workers to override the replayed flag (0 forces the serial path for parity checks). Verification (logs/sysid_diagnosis/, vs old-code baselines 21643951/21643952): - seed1 workers=6 vs workers=0: logs byte-identical except elapsed time; fit stage 398 s vs 1246 s (3.1x). Outcome unchanged - init at the SSE floor, moved NONE, same 25 verdicts. - seed0: trimming byte-identical, and the fit now recovers cure_steps 24 -> 23.49 and glue_lat_radius 0.011 -> 0.0148, both corroborated by the baseline edge probes (dSSE -1.0 and -0.04); the old fit moved neither. - Suites: 617 passed (code_sim_learning + agent_sdk); full-tree pylint, mypy, formatters clean. Claude-Session: https://claude.ai/code/session_01FkveUgLNgKQywdAdjJD1iE --- predicators/agent_sdk/parallel_rollouts.py | 12 +- predicators/code_sim_learning/lm.py | 170 +++++++++++++----- .../code_sim_learning/physical_sysid.py | 33 +++- predicators/code_sim_learning/rollout_env.py | 11 ++ .../code_sim_learning/rollout_objective.py | 150 ++++++++++++---- predicators/settings.py | 26 +-- scripts/sysid_fit_diagnosis.py | 9 + .../code_sim_learning/test_physical_sysid.py | 161 +++++++++++++++++ .../test_zero_gradient_search.py | 126 ++++++++++++- 9 files changed, 606 insertions(+), 92 deletions(-) diff --git a/predicators/agent_sdk/parallel_rollouts.py b/predicators/agent_sdk/parallel_rollouts.py index 8815993b8..79b8d6c1d 100644 --- a/predicators/agent_sdk/parallel_rollouts.py +++ b/predicators/agent_sdk/parallel_rollouts.py @@ -48,7 +48,8 @@ def parallel_rollouts_available() -> bool: def prefetch_parallel(jobs: Sequence[Callable[[], Any]], - label: str) -> List[Optional[Any]]: + label: str, + quiet: bool = False) -> List[Optional[Any]]: """Pre-run independent rollout jobs as forked children when enabled. Returns an index-aligned result list, or all-``None`` when parallel @@ -58,6 +59,10 @@ def prefetch_parallel(jobs: Sequence[Callable[[], Any]], path, so verdict semantics (seeds, scopes, bookkeeping, early breaks) are identical with the flag on or off - the parallel pass only prepays the rollouts. + + ``quiet`` demotes the per-call INFO line to DEBUG: the sysID + objective calls this once per candidate theta, hundreds of times + per fit, and one log line each would drown the run's info.log. """ # Deferred: settings must stay import-cycle-free from tool modules. # pylint: disable-next=import-outside-toplevel @@ -65,8 +70,9 @@ def prefetch_parallel(jobs: Sequence[Callable[[], Any]], workers = min(int(CFG.agent_validation_parallel_workers), len(jobs)) if workers <= 1 or len(jobs) <= 1 or not parallel_rollouts_available(): return [None] * len(jobs) - logger.info("[%s] prefetching %d rollouts across %d forked children.", - label, len(jobs), workers) + logger.log(logging.DEBUG if quiet else logging.INFO, + "[%s] prefetching %d rollouts across %d forked children.", + label, len(jobs), workers) return run_forked_rollouts(jobs, workers, label) diff --git a/predicators/code_sim_learning/lm.py b/predicators/code_sim_learning/lm.py index 19752504b..134e376cc 100644 --- a/predicators/code_sim_learning/lm.py +++ b/predicators/code_sim_learning/lm.py @@ -147,22 +147,53 @@ def lm_point_fit_result( _GATE_REFINE_ITERS = 6 _GATE_MIN_REL_IMPROVEMENT = 1e-6 - -def zero_jacobian_columns(jac: np.ndarray) -> List[int]: - """Indices of parameters whose Jacobian column is identically zero.""" +# Zero-gradient column detection tolerances. Exact ``!= 0.0`` is the +# wrong test on simulation residuals: solver jitter under the coarse +# finite-difference step leaves ~1e-10-scale junk in columns that carry +# no signal (jitter ~1e-12 over a 2e-2 relative step), while a genuinely +# responsive column of the dimensionless scaled residuals is O(1). A +# column counts as zero-gradient when its largest |entry| on the DATA +# rows is at or below max(abs tol, rel tol * the largest entry in any +# data column). +_ZERO_COL_ABS_TOL = 1e-8 +_ZERO_COL_REL_TOL = 1e-6 + + +def zero_jacobian_columns(jac: np.ndarray, n_prior_rows: int = 0) -> List[int]: + """Indices of parameters whose DATA-row Jacobian column carries no usable + gradient. + + ``n_prior_rows`` trailing rows (the MAP objective's Gaussian prior + rows) are excluded from the test: a prior row's derivative with + respect to its own parameter is the nonzero constant + ``noise_sigma / prior_sigma``, so on a prior-folded objective no + column of the FULL Jacobian is ever zero and an all-rows test can + never fire - which left the bracket search dead on the rollout MAP + path (the 2026-08-30 bridge runs logged zero bracket searches + across every fit while 20 of 25 parameters were data-flat). + Detection uses the ``_ZERO_COL_*_TOL`` tolerances above rather than + exact zero. + """ if jac.ndim != 2 or jac.size == 0: return [] - return [j for j in range(jac.shape[1]) if not np.any(jac[:, j] != 0.0)] + data = jac[:jac.shape[0] - n_prior_rows] if n_prior_rows > 0 else jac + if data.size == 0: + return [] + col_max = np.max(np.abs(data), axis=0) + tol = max(_ZERO_COL_ABS_TOL, _ZERO_COL_REL_TOL * float(np.max(col_max))) + return [j for j in range(data.shape[1]) if col_max[j] <= tol] def bracket_search_zero_gradient_params( - sse_fn: Callable[[np.ndarray], float], + residuals_fn: Callable[[np.ndarray], np.ndarray], z: np.ndarray, lo: np.ndarray, hi: np.ndarray, cols: List[int], param_specs: List[ParamSpec], label: str, + n_prior_rows: int = 0, + flat_out: Optional[List[str]] = None, ) -> Tuple[np.ndarray, float, List[str]]: """Coordinate-wise bracket search (fit space) for the parameters in ``cols``, holding every other parameter fixed. @@ -176,12 +207,42 @@ def bracket_search_zero_gradient_params( to hand-bracketing by the agent. This search evaluates the SSE on a grid across the parameter's box, refines around the best grid point by golden section, and keeps a move only when it lowers the - SSE. Returns ``(z, sse, notes)`` with one note per parameter - saying what happened, in external units, for the agent-facing - report. + SSE. Returns ``(z, sse, notes)`` (``sse`` is the TOTAL objective, + prior rows included) with one note per parameter saying what + happened, in external units, for the agent-facing report. + + Takes the residual VECTOR function rather than an SSE function + because the two verdicts need different rows of it (one evaluation + serves both): flat verdicts are judged on the DATA rows only - at + a box edge the ``n_prior_rows`` Gaussian prior rows alone add + ``(noise_sigma * (z - c) / sigma)**2`` to the total, orders of + magnitude above the flat tolerance, so testing the MAP total would + read every data-flat parameter as responsive - while the argmin + and move acceptance use the TOTAL objective, so a move trades data + improvement against distance from the anchor exactly as LM does. + + Cost control: the box EDGES are evaluated first, and a parameter + whose data SSE is flat at both edges is declared flat for 2 + evaluations instead of 9 - for the piecewise-constant thresholds + this search exists for, a response anywhere in the box almost + always shows at an edge. (An interior-only dip whose edges match + the current SSE is the accepted blind spot; the full grid only + ever ran for parameters the LM gradient already called flat.) + ``flat_out``, when given, collects the names of parameters found + flat across their box - box-wide insensitivity evidence the + identifiability report can consume instead of re-probing them. """ z = np.array(z, dtype=float) - sse = float(sse_fn(z)) + + def _sses(zz: np.ndarray) -> Tuple[float, float]: + """``(data_sse, total_sse)`` from one residual evaluation.""" + res = np.asarray(residuals_fn(zz), dtype=float) + total = float(np.sum(res**2)) + if n_prior_rows <= 0: + return total, total + return float(np.sum(res[:res.size - n_prior_rows]**2)), total + + data_sse, sse = _sses(z) notes: List[str] = [] for j in cols: spec = param_specs[j] @@ -192,18 +253,36 @@ def bracket_search_zero_gradient_params( f"{init_ext:.4g} (NOT fit from data).") continue grid = np.linspace(lo[j], hi[j], _GATE_GRID_POINTS) - vals = [] - for g in grid: + + def _eval_at(g: float, col: int = j) -> Tuple[float, float]: zz = z.copy() - zz[j] = g - vals.append(float(sse_fn(zz))) - span = max(vals) - min(vals) - if span <= _GATE_MIN_REL_IMPROVEMENT * max(sse, 1e-12): - notes.append(f"{spec.name}: SSE is flat across its whole box " - f"({_GATE_GRID_POINTS} points), so the data do not " - f"constrain it; kept at {init_ext:.4g} (NOT fit " + zz[col] = g + return _sses(zz) + + flat_tol = _GATE_MIN_REL_IMPROVEMENT * max(data_sse, 1e-12) + lo_data, lo_total = _eval_at(float(grid[0])) + hi_data, hi_total = _eval_at(float(grid[-1])) + if (abs(lo_data - data_sse) <= flat_tol + and abs(hi_data - data_sse) <= flat_tol): + notes.append(f"{spec.name}: data SSE is flat at both box edges, " + "so the data do not constrain it; kept at " + f"{init_ext:.4g} (NOT fit from data).") + if flat_out is not None: + flat_out.append(spec.name) + continue + pairs = [(lo_data, lo_total)] + pairs += [_eval_at(float(g)) for g in grid[1:-1]] + pairs.append((hi_data, hi_total)) + data_vals = [p[0] for p in pairs] + if max(data_vals) - min(data_vals) <= flat_tol: + notes.append(f"{spec.name}: data SSE is flat across its whole " + f"box ({_GATE_GRID_POINTS} points), so the data do " + f"not constrain it; kept at {init_ext:.4g} (NOT fit " "from data).") + if flat_out is not None: + flat_out.append(spec.name) continue + vals = [p[1] for p in pairs] best = int(np.argmin(vals)) best_z, best_sse = float(grid[best]), vals[best] a = float(grid[max(best - 1, 0)]) @@ -211,22 +290,16 @@ def bracket_search_zero_gradient_params( phi = (np.sqrt(5.0) - 1.0) / 2.0 x1 = b - phi * (b - a) x2 = a + phi * (b - a) - - def _eval(x: float, col: int = j) -> float: - zz = z.copy() - zz[col] = x - return float(sse_fn(zz)) - - f1, f2 = _eval(x1), _eval(x2) + f1, f2 = _eval_at(x1)[1], _eval_at(x2)[1] for _ in range(_GATE_REFINE_ITERS): if f1 < f2: b, x2, f2 = x2, x1, f1 x1 = b - phi * (b - a) - f1 = _eval(x1) + f1 = _eval_at(x1)[1] else: a, x1, f1 = x1, x2, f2 x2 = a + phi * (b - a) - f2 = _eval(x2) + f2 = _eval_at(x2)[1] for x, f in ((x1, f1), (x2, f2)): if f < best_sse: best_z, best_sse = x, f @@ -237,7 +310,9 @@ def _eval(x: float, col: int = j) -> float: f"{init_ext:.4g} -> {new_ext:.4g} (SSE {sse:.4g} -> " f"{best_sse:.4g}).") z[j] = best_z - sse = best_sse + # Refresh both SSEs at the moved point: later parameters' + # flat tests compare against the CURRENT data SSE. + data_sse, sse = _sses(z) else: notes.append(f"{spec.name}: LM gradient is zero; bracket search " f"over its box found nothing better than " @@ -254,6 +329,8 @@ def solve_lm( label: str, diff_step: Optional[float] = None, notes_out: Optional[List[str]] = None, + n_prior_rows: int = 0, + flat_params_out: Optional[List[str]] = None, ) -> Tuple[np.ndarray, Optional[np.ndarray]]: """Shared Levenberg-Marquardt core for the per-transition, recurrent, and rollout (``physical_sysid``) MAP fits. @@ -285,13 +362,20 @@ def solve_lm( for log params, so the finite-difference gradient stays equally informative across decades instead of vanishing at the low end. - Parameters whose Jacobian column at the LM optimum is identically - zero (threshold/gate parameters, whose finite-difference gradient - exists only where a data point is crossed) get a coordinate-wise - bracket search over their box (:func:`bracket_search_zero_gradient_ - params`); if any moves, LM is re-run from the new point so the - smooth parameters re-adapt. ``notes_out`` collects one line per - such parameter for the agent-facing fit report. + Parameters whose DATA-row Jacobian column at the LM optimum carries + no gradient (threshold/gate parameters, whose finite-difference + gradient exists only where a data point is crossed) get a + coordinate-wise bracket search over their box + (:func:`bracket_search_zero_gradient_params`); if any moves, LM is + re-run from the new point so the smooth parameters re-adapt. + ``n_prior_rows`` is how many trailing residual rows are Gaussian + prior rows: they must be excluded from the zero-gradient test (see + :func:`zero_jacobian_columns`) or the bracket search never runs on + a MAP objective. ``notes_out`` collects one line per searched + parameter for the agent-facing fit report; ``flat_params_out`` + collects the names of parameters the search measured flat across + their whole box (box-wide insensitivity evidence for the + identifiability report). """ from scipy.optimize import \ least_squares # pylint: disable=import-outside-toplevel @@ -334,14 +418,18 @@ def internal_residuals(z: np.ndarray) -> np.ndarray: "converged" if result.success else "max-evals") jac = np.asarray(result.jac, dtype=float) - zero_cols = zero_jacobian_columns(jac) + zero_cols = zero_jacobian_columns(jac, n_prior_rows) if zero_cols: - - def _sse(z: np.ndarray) -> float: - return float(np.sum(internal_residuals(z)**2)) - z_new, sse_new, notes = bracket_search_zero_gradient_params( - _sse, result.x, lo, hi, zero_cols, param_specs, label) + internal_residuals, + result.x, + lo, + hi, + zero_cols, + param_specs, + label, + n_prior_rows=n_prior_rows, + flat_out=flat_params_out) if notes_out is not None: notes_out.extend(notes) if sse_new < sse_lm: diff --git a/predicators/code_sim_learning/physical_sysid.py b/predicators/code_sim_learning/physical_sysid.py index b7422ef39..490e39bb1 100644 --- a/predicators/code_sim_learning/physical_sysid.py +++ b/predicators/code_sim_learning/physical_sysid.py @@ -261,6 +261,8 @@ def fit_params_rollout( # The grid-seeded, prior-folded LM MAP IS the fit. The Jacobian at # the MAP is kept on the result as the Laplace bundle for the # info-seeking explorer's calibrated ensemble. + lm_notes: List[str] = [] + lm_flat: List[str] = [] lm_theta, lm_jac = fit_map_lm_rollout(base_env, trajectories, lm_physical_specs, @@ -271,10 +273,36 @@ def fit_params_rollout( scaling=scaling, prior_centers=center_int, prior_sigmas=prior_sigma, - noise_sigma=noise_sigma) + noise_sigma=noise_sigma, + notes_out=lm_notes, + flat_params_out=lm_flat) if (config.log_hessian_identifiability and lm_jac is not None and lm_jac.size > 0): log_hessian_identifiability(lm_jac, names, noise_sigma, prior_sigma) + # Box-flat verdicts from the LM bracket grid: for params without + # grid-sweep coverage (rule params, or fits run with the grid + # disabled), a measured flat-across-the-box SSE at the MAP is the + # same box-wide insensitivity evidence the sweep supplies, so it + # feeds the same INSENSITIVE verdict and whole-box interval - and + # spares the identifiability probe its 2 rollout evals per such + # param. A param the grid DID sweep keeps the sweep's verdict (the + # sweep held the others at their anchors, the bracket at the MAP; + # when they disagree the anchored measurement stands). + if lm_flat: + if sensitivity is None: + sensitivity = {} + spec_by_name = {s.name: s for s in all_specs} + for flat_name in lm_flat: + spec = spec_by_name.get(flat_name) + if flat_name in sensitivity or spec is None: + continue + if spec.lo is None or spec.hi is None: + continue + sensitivity[flat_name] = { + "flat_interval": [float(spec.lo), + float(spec.hi)], + "sensitive": False, + } result = FitResult(names=names, samples=np.asarray(lm_theta, dtype=float)[None, :], log_probs=np.zeros(1), @@ -282,7 +310,8 @@ def fit_params_rollout( noise_sigma=noise_sigma, prior_sigma=prior_sigma, scales=scales, - sensitivity=sensitivity) + sensitivity=sensitivity, + lm_notes=lm_notes) n_lm = num_rollouts_run() - n_start - n_grid if (config.anchor_ablation and config.grid_flat_frac > 0 and trajectories): result = _anchor_backward_elimination( diff --git a/predicators/code_sim_learning/rollout_env.py b/predicators/code_sim_learning/rollout_env.py index 8348ac46c..fa1771e00 100644 --- a/predicators/code_sim_learning/rollout_env.py +++ b/predicators/code_sim_learning/rollout_env.py @@ -31,6 +31,17 @@ def num_rollouts_run() -> int: return _NUM_ROLLOUTS +def add_rollouts_run(n: int) -> None: + """Credit ``n`` rollouts executed OUTSIDE this process to the counter. + + The parallel objective path runs its rollouts in forked children + whose counter copies die with them; the parent credits each + successful child here so the per-stage budget logs stay honest. + """ + global _NUM_ROLLOUTS # pylint: disable=global-statement + _NUM_ROLLOUTS += n + + def _zero_all_velocities(base_env: Any) -> None: """Zero every velocity in the env's client: base velocities of all bodies AND joint velocities of articulated bodies (the robot arm). diff --git a/predicators/code_sim_learning/rollout_objective.py b/predicators/code_sim_learning/rollout_objective.py index 1ccca5c14..125b611ee 100644 --- a/predicators/code_sim_learning/rollout_objective.py +++ b/predicators/code_sim_learning/rollout_objective.py @@ -10,8 +10,10 @@ from __future__ import annotations import dataclasses +import functools import logging -from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple +from typing import Any, Callable, Dict, Iterator, List, Optional, Sequence, \ + Tuple import numpy as np @@ -19,7 +21,7 @@ from predicators.code_sim_learning.fit_space import ParamSpec, to_fit_space from predicators.code_sim_learning.lm import solve_lm from predicators.code_sim_learning.rollout_env import RolloutTrajectory, \ - rollout_states + add_rollouts_run, rollout_states from predicators.code_sim_learning.trajectory_prep import ResidualScaling from predicators.settings import CFG from predicators.structs import Action, State @@ -176,6 +178,11 @@ def _iter_rollout_residual_terms( rolled-out state, and score every in-scope feature against its observation. Deterministic iteration order. Without ``scaling`` the residual is the raw ``pred - obs`` difference (legacy objective). + With a factory ``base_env`` and fork-parallel workers enabled + (``agent_validation_parallel_workers``), per-step scoring fans + whole trajectories out to forked children; the terms and their + order are identical to the serial path's (see + :func:`_prefetch_trajectory_terms`). Each per-step residual is Huber-capped (``huber_delta``), and two kinds of per-trajectory SUMMARY residuals are appended with weight @@ -224,12 +231,14 @@ def _iter_rollout_residual_terms( # scored segment: the positions only line up at the episode's start. id_maps = (_episode_id_maps(tracks, trajectories, config, paired_tracks) if score_intervals else []) - episode_rollouts: List[List[State]] = [] physical = {n: params[n] for n in physical_names if n in params} rules_list = list(rules) latent_mode = bool(rules_list) and has_latent_rules(rules_list) - for traj_index, (states, actions) in enumerate(trajectories): + def _rollout_with_rules( + states: List[State], actions: List[Action] + ) -> Tuple[List[State], List[Dict[Any, Dict[str, Any]]]]: + """Free-run ONE trajectory with the rules in-the-loop.""" latent: Dict[str, Any] = (init_latent(latent_init, params) if latent_mode else {}) history: List[Tuple[State, Optional[Action]]] = [] @@ -240,8 +249,6 @@ def _iter_rollout_residual_terms( # state (never written back into the physics world). updates_per_step: List[Dict[Any, Dict[str, Any]]] = [] - # pylint: disable=cell-var-from-loop - # (Consumed within this same loop iteration, before rebinding.) def _run_rules_post_step(env: Any, sim_state: State, i: int) -> None: cmds = CommandBuffer() if latent_mode: @@ -262,31 +269,25 @@ def _run_rules_post_step(env: Any, sim_state: State, i: int) -> None: if cmds: env.queue_residual_commands(cmds.commands) - # pylint: enable=cell-var-from-loop sim_states = rollout_states( base_env, states[0], actions, physical, post_step=_run_rules_post_step if rules_list else None) - if score_intervals: - # The per-step loop below is skipped entirely rather than added - # to. Under open-loop nothing corrects the twin, so those steps - # are the twin's own simulation and including them would let the - # defect this flag exists to fix outvote the real evidence by - # thousands of terms to a handful. - if paired_tracks: - # One track per trajectory: each trajectory IS an episode, so - # scoring it on its own already is per-episode. - yield from _interval_residual_terms(sim_states, states, - tracks[traj_index], - id_maps[traj_index], - config, summary_w) - else: - # Segments of ONE episode. Held, not scored: see the episode - # -level yield after this loop. - episode_rollouts.append(sim_states) - continue + return sim_states, updates_per_step + + def _per_step_terms(states: List[State], + actions: List[Action]) -> List[float]: + """All residual terms of ONE trajectory under per-step scoring. + + Self-contained per trajectory (rollout, per-step residuals, + endpoint and onset summaries), which is what lets the parallel + path hand whole trajectories to forked children and get back + exactly the serial path's terms. + """ + sim_states, updates_per_step = _rollout_with_rules(states, actions) + terms: List[float] = [] endpoint_residuals: List[float] = [] for i, sim_state in enumerate(sim_states): obs_state = states[i + 1] @@ -314,20 +315,93 @@ def _run_rules_post_step(env: Any, sim_state: State, i: int) -> None: res = pred_val - obs_val if is_last and summary_w > 0: endpoint_residuals.append(res) - yield _huberize(res, delta) + terms.append(_huberize(res, delta)) if summary_w > 0 and sim_states: for res in endpoint_residuals: - yield summary_w * _huberize(res, delta) - yield from _onset_residuals([states[0]] + sim_states, states, - residual_features, config.settle_tol, - summary_w) - if score_intervals and not paired_tracks and episode_rollouts: + terms.append(summary_w * _huberize(res, delta)) + terms.extend( + _onset_residuals([states[0]] + sim_states, states, + residual_features, config.settle_tol, + summary_w)) + return terms + + if not score_intervals: + prefetched = _prefetch_trajectory_terms(base_env, trajectories, + _per_step_terms) + for idx, (states, actions) in enumerate(trajectories): + terms = None if prefetched is None else prefetched[idx] + if terms is None: + terms = _per_step_terms(states, actions) + yield from terms + return + + # Track-interval scoring. The per-step terms above are skipped + # entirely rather than added to. Under open-loop nothing corrects + # the twin, so those steps are the twin's own simulation and + # including them would let the defect this flag exists to fix + # outvote the real evidence by thousands of terms to a handful. + episode_rollouts: List[List[State]] = [] + for traj_index, (states, actions) in enumerate(trajectories): + sim_states, _ = _rollout_with_rules(states, actions) + if paired_tracks: + # One track per trajectory: each trajectory IS an episode, so + # scoring it on its own already is per-episode. + yield from _interval_residual_terms(sim_states, states, + tracks[traj_index], + id_maps[traj_index], config, + summary_w) + else: + # Segments of ONE episode. Held, not scored: see the episode + # -level yield after this loop. + episode_rollouts.append(sim_states) + if not paired_tracks and episode_rollouts: yield from _episode_interval_terms(episode_rollouts, [s for s, _ in trajectories], tracks[-1], id_maps[0], config, summary_w) +def _prefetch_trajectory_terms( + base_env: Any, + trajectories: List[RolloutTrajectory], + score_fn: Callable[[List[State], List[Action]], List[float]], +) -> Optional[List[Optional[List[float]]]]: + """Fan per-trajectory scoring out to forked children when enabled. + + Factory envs only: each rollout builds (and disposes) its own fresh + world whichever process runs it, so a child's term list is + bit-identical to what the serial path would compute - which is what + the LM finite-difference Jacobian requires of repeated same-theta + evaluations. A shared env instance keeps the serial path untouched + (its rollouts mutate the caller's env, which must happen in this + process). + + Returns the index-aligned per-trajectory term lists - the caller + recomputes any ``None`` entry serially, per the + :func:`~predicators.agent_sdk.parallel_rollouts.prefetch_parallel` + contract - or ``None`` when parallelism is disabled, unavailable, + or pointless. Successful child rollouts are credited to this + process's rollout counter, which the children's exits would + otherwise lose from the per-stage budget logs. + """ + if not callable(base_env) or len(trajectories) <= 1: + return None + # Deferred: agent_sdk imports this module's package; importing the + # (dependency-free) pool module lazily keeps the layering acyclic. + # pylint: disable-next=import-outside-toplevel + from predicators.agent_sdk.parallel_rollouts import prefetch_parallel + jobs: List[Callable[[], List[float]]] = [ + functools.partial(score_fn, states, actions) + for states, actions in trajectories + ] + results = prefetch_parallel(jobs, "sysid objective", quiet=True) + done = sum(1 for r in results if r is not None) + if done == 0: + return None + add_rollouts_run(done) + return results + + def _load_scored_track(config: SysIdConfig) -> Optional[Any]: """The observation track to score against, or None to fall back. @@ -795,6 +869,8 @@ def fit_map_lm_rollout( prior_sigmas: Optional[np.ndarray] = None, noise_sigma: float = 0.05, fixed_physical: Optional[Dict[str, float]] = None, + notes_out: Optional[List[str]] = None, + flat_params_out: Optional[List[str]] = None, ) -> Tuple[np.ndarray, Optional[np.ndarray]]: """MAP estimate of the joint physical+rule theta via Levenberg-Marquardt. @@ -819,7 +895,12 @@ def fit_map_lm_rollout( noise (a flat likelihood's finite-difference gradient is pure noise). The prior rows are stripped from the returned Jacobian - its consumers (Laplace ensemble, Hessian diagnostic) add the prior - term themselves and would otherwise double-count it. + term themselves and would otherwise double-count it - and excluded + from the zero-gradient gate detection (``n_prior_rows``), without + which the bracket search for threshold-like parameters could never + fire on this MAP path (see :func:`lm.zero_jacobian_columns`). + ``notes_out``/``flat_params_out`` pass through to + :func:`lm.solve_lm`. """ all_specs = list(physical_specs) + list(rule_specs) names = [s.name for s in all_specs] @@ -843,7 +924,10 @@ def residuals_fn(theta: np.ndarray) -> np.ndarray: all_specs, max_nfev, "rollout", - diff_step=_ROLLOUT_LM_DIFF_STEP) + diff_step=_ROLLOUT_LM_DIFF_STEP, + notes_out=notes_out, + n_prior_rows=len(all_specs) if use_prior else 0, + flat_params_out=flat_params_out) if use_prior and jac is not None and jac.shape[0] > len(all_specs): jac = jac[:-len(all_specs)] return theta_map, jac diff --git a/predicators/settings.py b/predicators/settings.py index 72c8f98e1..aad47452e 100644 --- a/predicators/settings.py +++ b/predicators/settings.py @@ -1783,18 +1783,20 @@ class GlobalSettings: # installs the ensemble providers (see rule_param_margin_provider), # which requires agent_explorer_info_seeking's ensemble. agent_plan_validation_rule_param_margin = False - # Fork-parallel validation rollouts: the capture gate's repeat - # rollouts, its physics/rule-param margin sweeps, and the belief - # probe's trials/physics_sweep modes each run N INDEPENDENT - # fresh-env rollouts; with a value W > 1, up to W run concurrently - # as forked children (see agent_sdk/parallel_rollouts.py). Verdict - # semantics are unchanged: each rollout runs under the exact seed / - # override scope it would run under sequentially, and a failed - # child is transparently re-run in-process. Benchmark (job - # 21336169, 8-CPU node, fresh bridge env per rollout): 1.89x at - # W=2, 3.74x at W=4, 4.81x at W=8. 0 (the default) keeps every - # rollout sequential; enable in experiment configs sized to the - # job's CPU allocation (e.g. 6 with --cpus-per-task=8). + # Fork-parallel rollouts: the capture gate's repeat rollouts, its + # physics/rule-param margin sweeps, the belief probe's + # trials/physics_sweep modes, and the rollout-sysID objective (each + # candidate theta scores N trajectory segments) all run N + # INDEPENDENT fresh-env rollouts; with a value W > 1, up to W run + # concurrently as forked children (see + # agent_sdk/parallel_rollouts.py). Verdict/fit semantics are + # unchanged: each rollout runs under the exact seed / override + # scope it would run under sequentially, and a failed child is + # transparently re-run in-process. Benchmark (job 21336169, 8-CPU + # node, fresh bridge env per rollout): 1.89x at W=2, 3.74x at W=4, + # 4.81x at W=8. 0 (the default) keeps every rollout sequential; + # enable in experiment configs sized to the job's CPU allocation + # (e.g. 6 with --cpus-per-task=8). agent_validation_parallel_workers = 0 # Agent bilevel explorer settings. Separate from the solve-path budget # above because the explorer runs full backtracking while looking for diff --git a/scripts/sysid_fit_diagnosis.py b/scripts/sysid_fit_diagnosis.py index de14d06ab..7ba8d39ef 100644 --- a/scripts/sysid_fit_diagnosis.py +++ b/scripts/sysid_fit_diagnosis.py @@ -213,9 +213,18 @@ def _main() -> None: parser.add_argument("--smoke", action="store_true", help="load + prep only; run no rollouts") + parser.add_argument("--parallel_workers", + type=int, + default=None, + help="override the run's " + "agent_validation_parallel_workers (0 forces the " + "serial objective path, for parallel-parity checks)") args = parser.parse_args() _apply_run_config(args.run_dir, args.env, args.seed) + if args.parallel_workers is not None: + utils.update_config( + {"agent_validation_parallel_workers": args.parallel_workers}) payload = _load_fit_data(args.run_dir, args.pickle) trajectories = payload["trajectories"] diff --git a/tests/code_sim_learning/test_physical_sysid.py b/tests/code_sim_learning/test_physical_sysid.py index 667cb4f12..adfaa7516 100644 --- a/tests/code_sim_learning/test_physical_sysid.py +++ b/tests/code_sim_learning/test_physical_sysid.py @@ -1517,3 +1517,164 @@ def sse(delta): assert sse(0.0) == pytest.approx(100.0**2) assert sse(1.0) == pytest.approx(2 * 100.0 - 1.0) + + +# ── Fork-parallel rollout objective ──────────────────────────────── + + +class _LinearParamEnv: + """Fresh-per-rollout env whose domino advances by the physical param ``k`` + each step, so residuals depend on theta and on the trajectory.""" + + def __init__(self, log): + self._physics_client_id = p.connect(p.DIRECT) + log.append(self) + self._k = 0.0 + self._x = 0.0 + self._domino = Object("d0", _DOMINO_TYPE) + self._robot = Object("r0", _ROBOT_TYPE) + + def apply_physical_param_overrides(self, params): + """Record the candidate ``k`` the rollout advances by.""" + self._k = float(params.get("k", 0.0)) + + def _set_state(self, state): + domino = next(o for o in state if o.name == "d0") + self._x = float(state.get(domino, "x")) + + def step(self, action): + """Advance the domino by ``k`` and return the post-step state.""" + del action + self._x += self._k + return State({ + self._domino: np.array([self._x], dtype=float), + self._robot: np.array([0.0], dtype=float), + }) + + @property + def connected(self): + """Whether this stub's PyBullet client is still connected.""" + return bool( + p.getConnectionInfo(self._physics_client_id)["isConnected"]) + + +def test_rollout_residuals_parallel_matches_serial(monkeypatch): + """The fork-parallel objective path returns exactly the serial terms. + + The parallel path exists purely as a wall-clock optimization of the + sysID objective (LM Jacobian columns, sensitivity/identifiability + sweeps, grid seeding all funnel through it); any value or ordering + difference from the serial path would silently change fits. + """ + from predicators.agent_sdk.parallel_rollouts import \ + parallel_rollouts_available + from predicators.settings import CFG + if not parallel_rollouts_available(): + pytest.skip("fork not available on this platform") + trajectories = [ + _trajectory([0.0, 0.1, 0.2]), + _trajectory([0.0, 0.2, 0.4]), + _trajectory([0.0, 0.05, 0.1]), + ] + built = [] + + def factory(): + return _LinearParamEnv(built) + + params = {"k": 0.1} + monkeypatch.setattr(CFG, "agent_validation_parallel_workers", 0) + serial = rollout_objective.compute_rollout_residuals( + factory, trajectories, params, _RESIDUAL_FEATURES, ["k"]) + assert serial.size > 0 + + monkeypatch.setattr(CFG, "agent_validation_parallel_workers", 2) + n0 = rollout_env.num_rollouts_run() + parallel = rollout_objective.compute_rollout_residuals( + factory, trajectories, params, _RESIDUAL_FEATURES, ["k"]) + assert np.array_equal(serial, parallel) + # The parent's rollout counter stays honest: every trajectory is + # credited exactly once whether its rollout ran in a child or (on a + # child failure) was recomputed serially here. + assert rollout_env.num_rollouts_run() - n0 == len(trajectories) + # Every parent-built env was disposed (children build their own). + assert all(not env.connected for env in built) + + +def test_rollout_residuals_shared_env_instance_stays_serial(monkeypatch): + """A caller-owned env instance never takes the parallel path: its rollouts + mutate the caller's env, which must happen in-process.""" + from predicators.settings import CFG + monkeypatch.setattr(CFG, "agent_validation_parallel_workers", 4) + trajectories = [ + _trajectory([0.0, 0.1, 0.2]), + _trajectory([0.0, 0.2, 0.4]), + ] + built = [] + env = _LinearParamEnv(built) + res = rollout_objective.compute_rollout_residuals(env, trajectories, + {"k": 0.1}, + _RESIDUAL_FEATURES, + ["k"]) + assert res.size > 0 + assert len(built) == 1 # the caller's env, used for every rollout + assert env.connected # and never disposed + p.disconnect(env._physics_client_id) + + +# ── Bracket-grid flat verdicts feeding the identifiability report ── + + +def test_fit_params_rollout_folds_bracket_flat_into_sensitivity(monkeypatch): + """Box-flat LM bracket verdicts become INSENSITIVE evidence for params the + grid sweep does not cover, and the notes ride the FitResult.""" + phys_specs = [ParamSpec("mu", 0.5, lo=0.1, hi=1.0)] + rule_specs = [ParamSpec("gate", 2.0, lo=1.0, hi=3.0)] + + def fake_lm(_env, + _trajs, + physical_specs, + _features, + _rules=(), + rule_specs=(), + _latent=None, + **kwargs): + notes = kwargs.get("notes_out") + flat = kwargs.get("flat_params_out") + if notes is not None: + notes.append("gate: data SSE is flat at both box edges, so the " + "data do not constrain it; kept at 2 (NOT fit from " + "data).") + if flat is not None: + flat.append("gate") + all_specs = list(physical_specs) + list(rule_specs) + theta = np.array([s.init_value for s in all_specs], dtype=float) + return theta, None + + monkeypatch.setattr(physical_sysid, "fit_map_lm_rollout", fake_lm) + result = physical_sysid.fit_params_rollout(None, [], + phys_specs, {"domino": ["x"]}, + rule_specs=rule_specs) + assert result.sensitivity is not None + entry = result.sensitivity["gate"] + assert entry["sensitive"] is False + assert entry["flat_interval"] == [1.0, 3.0] + # Only MEASURED flats are folded in; the physical param (grid sweep + # skipped: no trajectories) stays uncovered. + assert "mu" not in result.sensitivity + assert result.lm_notes and result.lm_notes[0].startswith("gate:") + + # Downstream: the report renders the flat param INSENSITIVE and the + # curvature probe never spends rollouts on it - its two ±sigma + # evals would only re-measure what the bracket grid already did. + calls = [] + + def counting_sse(params): + calls.append(dict(params)) + return 1.0 + + report = identifiability_report(result, counting_sse, + phys_specs + rule_specs) + assert report["gate"]["verdict"] is Verdict.INSENSITIVE + # 3 MAP evals (noise floor) + 2 perturbations of mu only. + assert len(calls) == 5 + assert all(p["gate"] == 2.0 for p in calls) diff --git a/tests/code_sim_learning/test_zero_gradient_search.py b/tests/code_sim_learning/test_zero_gradient_search.py index 1c818232e..c7021351c 100644 --- a/tests/code_sim_learning/test_zero_gradient_search.py +++ b/tests/code_sim_learning/test_zero_gradient_search.py @@ -6,7 +6,8 @@ import numpy as np from predicators.code_sim_learning.fit_space import ParamSpec -from predicators.code_sim_learning.lm import solve_lm, zero_jacobian_columns +from predicators.code_sim_learning.lm import \ + bracket_search_zero_gradient_params, solve_lm, zero_jacobian_columns def test_zero_jacobian_columns() -> None: @@ -57,3 +58,126 @@ def residuals(theta: np.ndarray) -> np.ndarray: assert abs(theta[0] - 3.0) < 1e-3 assert theta[1] == 0.7 assert len(notes) == 1 and "NOT fit from data" in notes[0] + + +def test_zero_jacobian_columns_tolerance_and_prior_rows() -> None: + """Junk-scale columns are flagged, and prior rows cannot mask them.""" + # Column 1 carries only finite-difference junk (~1e-10) next to an + # O(1) column: the exact-zero test missed it, the tolerance flags it. + jac = np.array([[1.0, 1e-10], [0.5, -1e-10]]) + assert zero_jacobian_columns(jac) == [1] + # A MAP objective appends one prior row per parameter, and a prior + # row's own-column entry is a nonzero constant - so on the full + # Jacobian no column can ever read zero (the dead-bracket bug of + # the 2026-08-30 bridge runs). Excluding the prior rows finds the + # data-flat column again. + data_rows = np.array([[1.0, 0.0], [0.5, 0.0]]) + prior_rows = np.array([[0.2, 0.0], [0.0, 0.2]]) + map_jac = np.vstack([data_rows, prior_rows]) + assert zero_jacobian_columns(map_jac) == [] + assert zero_jacobian_columns(map_jac, n_prior_rows=2) == [1] + + +def test_solve_lm_map_prior_rows_bracket_still_fires() -> None: + """The bracket search fires on a MAP objective (prior rows folded). + + Regression for the 2026-08-30 bridge runs: with the Gaussian prior + folded in as residual rows, the all-rows exact-zero column test + could never fire and the bracket search was dead code on the rollout + MAP path (seed0's cure_steps improvement went unfound). + """ + xs = np.linspace(0.0, 1.0, 41) + true_gain, true_thresh = 2.0, 0.6 + observed = true_gain * (xs > true_thresh) + specs = [ + ParamSpec("gain", 1.0, lo=0.0, hi=5.0), + ParamSpec("thresh", 0.2, lo=0.0, hi=1.0), + ] + centers = np.array([1.0, 0.2]) + sigmas = np.array([5.0, 5.0]) # wide prior: the data should win + + def residuals(theta: np.ndarray) -> np.ndarray: + gain, thresh = float(theta[0]), float(theta[1]) + data = gain * (xs > thresh) - observed + prior = 0.05 * (np.asarray(theta) - centers) / sigmas + return np.concatenate([data, prior]) + + notes: List[str] = [] + flat: List[str] = [] + theta, _ = solve_lm(residuals, + specs, + 200, + "test", + notes_out=notes, + n_prior_rows=2, + flat_params_out=flat) + assert abs(theta[1] - true_thresh) < 0.03 + assert abs(theta[0] - true_gain) < 1e-2 + assert any(n.startswith("thresh:") and "moved it" in n for n in notes) + assert not flat + + +def test_bracket_search_flat_param_settled_from_edge_evals() -> None: + """A box-flat parameter is settled from the two edge evaluations, is + reported in ``flat_out``, and never pays the full 9-point grid.""" + calls: List[np.ndarray] = [] + + def residuals(z: np.ndarray) -> np.ndarray: + calls.append(np.array(z)) + return np.array([3.0 - z[0]]) # depends only on param 0 + + specs = [ + ParamSpec("slope", 3.0, lo=0.0, hi=5.0), + ParamSpec("gate", 0.7, lo=0.5, hi=1.0), + ] + lo = np.array([0.0, 0.5]) + hi = np.array([5.0, 1.0]) + z = np.array([3.0, 0.7]) + flat: List[str] = [] + z_new, _, notes = bracket_search_zero_gradient_params(residuals, + z, + lo, + hi, [1], + specs, + "test", + flat_out=flat) + assert z_new[1] == 0.7 + assert flat == ["gate"] + assert len(notes) == 1 and "NOT fit from data" in notes[0] + # One baseline evaluation plus the two box edges. + assert len(calls) == 3 + + +def test_bracket_search_flat_test_ignores_prior_rows() -> None: + """Prior rows must not make a data-flat parameter look responsive. + + At a box edge the prior rows alone raise the TOTAL SSE well above + the flat tolerance; the flat verdict must therefore be judged on the + data rows, or every data-flat parameter pays the full search and + loses its honest 'NOT fit from data' note. + """ + center = 0.75 + + def residuals(z: np.ndarray) -> np.ndarray: + data = np.array([3.0 - z[0]]) + prior = np.array([0.0, (z[1] - center) / 0.5]) + return np.concatenate([data, prior]) + + specs = [ + ParamSpec("slope", 3.0, lo=0.0, hi=5.0), + ParamSpec("gate", center, lo=0.5, hi=1.0), + ] + lo = np.array([0.0, 0.5]) + hi = np.array([5.0, 1.0]) + z = np.array([3.0, center]) + flat: List[str] = [] + _, _, notes = bracket_search_zero_gradient_params(residuals, + z, + lo, + hi, [1], + specs, + "test", + n_prior_rows=2, + flat_out=flat) + assert flat == ["gate"] + assert len(notes) == 1 and "NOT fit from data" in notes[0] From af0186ede1f23a4144e550c001cd359e4b873452 Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Mon, 31 Aug 2026 15:10:47 -0400 Subject: [PATCH 4/8] review fixes: traceless renders, live Wait notices, honest fit-report edges An independent high-effort review of today's five commits confirmed eight defects, and the full test tree (1806 tests) caught a ninth. All fixed: - scene.py: a stateful render_pybullet_image never restored the env after ctx.env._set_state(state), so the submit gate's per-step renders teleported the SHARED session env through the plan's states (wiping its residual-command queue and re-syncing welds) while the gate's rollouts ran on a fresh env - and planner simulate() SKIPS its own reset when the incoming state allclose-matches, inheriting the residue. Renders now snapshot and restore the env state and command queue. A plausible root cause of the open belief session-state leak. - utils.wait_rollout_step_cap(): the stalled-Wait advisories compared against max_num_steps_option_rollout (1000) while the option model's wait_option_max_steps backstop (bridge: 120) terminated the Wait first, so they could never fire in exactly the configured envs they were written for; both sites now key on whichever cap binds. - physical_sysid: the anchor-ablation rebuild dropped lm_notes, so the bracket search's "NOT fit from data" warnings vanished from fit reports whenever any parameter got pinned. - synthesis + sysid_fit_diagnosis: exact float equality broke the no-op banner and reported phantom moves - log-scale params round-trip through exp(log(x)) and trf nudges bound-sitting inits by 1e-9; both now use np.isclose. - lm.py: a failed or regressing post-bracket polish returned the bracket-moved theta paired with the PRE-bracket Jacobian; the pair now degrades to jacobian=None (all consumers handle it). The zero-gradient tolerance scales by the MEDIAN column max so one badly-scaled residual block cannot inflate every other parameter's threshold. Edge-screened flat params stay OUT of flat_out - two edge evals cannot rule out an interior-only response, so the identifiability probe stays armed as their backstop. - belief_probe: sim.state()'s full dict resolves bare-name lookups via __missing__, so helpers persisted from before the name:type keying keep working across a resume. - python_exec: the sandbox chdir moved inside the try - the budget watchdog's async exception could land between chdir and try-entry, skip the finally, and strand the whole process in the sandbox cwd. - config/settings: relative CFG track paths anchor at LAUNCH_CWD (captured at process start), so a sim.fit issued inside run_python's sandbox-cwd exec window cannot silently miss the fan domain's track manifest and fall back to per-step scoring. - skill_factories (full-tree pytest catch): ERROR level restored inside the log_errors gate - that flag is the method's final-failure emission contract (every in-run caller passes False), and the log spam the morning demotion targeted came only from the ungated per-candidate site, which stays debug. Verified: 681 tests across the affected suites on a compute node, new unit tests for each regression path (wait-cap binding, edge-screen probe backstop, launch-cwd anchoring, bare-name state lookups), pylint, mypy, and formatters clean. Claude-Session: https://claude.ai/code/session_01X3v2SxFDcKFoehfyzpvHXr --- predicators/agent_sdk/belief_probe.py | 35 ++++++++++++++--- predicators/agent_sdk/tools/python_exec.py | 12 ++++-- predicators/agent_sdk/tools/scene.py | 29 ++++++++++++-- predicators/agent_sdk/tools/synthesis.py | 10 ++++- predicators/agent_sdk/tools/testing.py | 6 +-- predicators/code_sim_learning/config.py | 21 +++++++++- predicators/code_sim_learning/lm.py | 39 +++++++++++++------ .../code_sim_learning/physical_sysid.py | 1 + predicators/settings.py | 11 ++++++ predicators/utils.py | 19 +++++++++ scripts/sysid_fit_diagnosis.py | 7 +++- .../agent_sdk/test_bilevel_sketch_regions.py | 22 +++++++++++ .../test_observation_track.py | 29 ++++++++++++++ .../test_zero_gradient_search.py | 13 +++++-- 14 files changed, 217 insertions(+), 37 deletions(-) diff --git a/predicators/agent_sdk/belief_probe.py b/predicators/agent_sdk/belief_probe.py index 0ddab1a7a..57b29320e 100644 --- a/predicators/agent_sdk/belief_probe.py +++ b/predicators/agent_sdk/belief_probe.py @@ -173,6 +173,26 @@ def _attach_step_contacts(events: List[Dict[str, Any]], start += s["num_actions"] +class _BareNameFallbackDict(Dict[str, Dict[str, float]]): + """``state()``'s full dict: keyed ``name:type``, resolving BARE object + names via ``__missing__``. + + Agent code persisted before the ``name:type`` keying (sandbox + helpers, journal recipes replayed after a resume/requeue) indexes + the full dict by bare name; resolving that here costs new code + nothing and closes the same gap the single-object form's bare-name + path already closes. Ambiguous bare names (two objects sharing a + name across types - not constructible in current envs) fall through + to the ordinary KeyError. + """ + + def __missing__(self, key: str) -> Dict[str, float]: + matches = [k for k in self if k.split(":", 1)[0] == key] + if len(matches) == 1: + return self[matches[0]] + raise KeyError(key) + + class _StrLikeResult: """String conveniences shared by the probe result types. @@ -853,7 +873,10 @@ def state( - keys use the same ``name:type`` form as atoms, plan lines, and the prompt, so a key read anywhere else indexes here directly; ``state("domino_1")`` and ``state("domino_1:domino")`` both -> - that object's ``{feat: value}``. + that object's ``{feat: value}``. The full dict also resolves + BARE-name lookups (``state()["domino_1"]``) via a fallback, so + agent helpers persisted from before the ``name:type`` keying + keep working across a resume. """ cur = self._require_state() @@ -875,7 +898,9 @@ def _features(obj: Any) -> Dict[str, float]: return _features(obj) raise ValueError(f"Unknown object '{obj_name}'. Available: " f"{sorted(str(o) for o in cur)}") - return {str(obj): _features(obj) for obj in sorted(cur, key=str)} + return _BareNameFallbackDict( + {str(obj): _features(obj) + for obj in sorted(cur, key=str)}) def atoms(self) -> List[str]: """Sorted ground atoms true in the current state.""" @@ -1496,10 +1521,10 @@ def _on_step(i: int, outcome: Any) -> None: img = render_scene_image( ctx, f"probe_step_{i}_{outcome.option.name}") if render else None - if (outcome.option.name == "Wait" and failure is None and - outcome.num_actions >= CFG.max_num_steps_option_rollout): + if (outcome.option.name == "Wait" and failure is None + and outcome.num_actions >= utils.wait_rollout_step_cap()): notices.append( - f"step {i} (Wait) ran to the option-rollout cap " + f"step {i} (Wait) ran to its step cap " f"({outcome.num_actions} actions): its wait-target " "atoms never became true in the belief (and no other " "atom changed). Check whether the awaited change is " diff --git a/predicators/agent_sdk/tools/python_exec.py b/predicators/agent_sdk/tools/python_exec.py index 8b05ae59f..d315a49b7 100644 --- a/predicators/agent_sdk/tools/python_exec.py +++ b/predicators/agent_sdk/tools/python_exec.py @@ -204,10 +204,15 @@ def _footer() -> str: # event loop, so no other tool call can observe the changed # cwd; harness paths reachable from tool code are absolute # (see _get_log_dir). + # The chdir happens INSIDE the try: the budget watchdog armed + # above delivers an ASYNC exception, and one landing between a + # pre-try chdir and try-entry would skip the finally and leave + # the whole process in the sandbox cwd. Restoring prev_cwd when + # the chdir never ran is a harmless no-op. prev_cwd = os.getcwd() - if sandbox_dir is not None: - os.chdir(sandbox_dir) try: + if sandbox_dir is not None: + os.chdir(sandbox_dir) exec(compile(code, label, "exec"), exec_ns) # pylint: disable=exec-used except ProbeBudgetExceeded as e: partial = captured.getvalue() @@ -226,8 +231,7 @@ def _footer() -> str: prefix = f"{partial}\n" if partial else "" return text_result(f"{prefix}Error:\n{tb}{_footer()}") finally: - if sandbox_dir is not None: - os.chdir(prev_cwd) + os.chdir(prev_cwd) if watchdog_disarm is not None: watchdog_disarm() sys.stdout = old_stdout diff --git a/predicators/agent_sdk/tools/scene.py b/predicators/agent_sdk/tools/scene.py index 3c3861650..476988760 100644 --- a/predicators/agent_sdk/tools/scene.py +++ b/predicators/agent_sdk/tools/scene.py @@ -79,11 +79,32 @@ def render_pybullet_image( from PIL import Image as PILImage + # pylint: disable=protected-access + prev: Optional[Tuple[State, List[Any]]] = None if state is not None: - ctx.env._set_state(state) # pylint: disable=protected-access - - with agent_render_resolution(): - video = ctx.env.render() + # A stateful render must leave NO trace on the shared + # session env: the submit gate renders per-step states while + # its rollouts run on a fresh env, and leaving the env + # teleported to the plan's states (with the welds and + # residual-command queue _set_state syncs to them) corrupts + # the substrate later consumers trust - planner simulate() + # even SKIPS its own reset when the incoming state + # allclose-matches the env's current one, inheriting + # whatever the render left behind. + if ctx.env._current_observation is not None: + prev = (ctx.env._current_state.copy(), + ctx.env._pending_residual_commands) + try: + if state is not None: + ctx.env._set_state(state) + with agent_render_resolution(): + video = ctx.env.render() + finally: + if prev is not None: + prev_state, prev_commands = prev + ctx.env._set_state(prev_state) + ctx.env._pending_residual_commands = prev_commands + # pylint: enable=protected-access if not video: return None rgb_array = np.asarray(video[0], dtype=np.uint8) diff --git a/predicators/agent_sdk/tools/synthesis.py b/predicators/agent_sdk/tools/synthesis.py index 4dafae547..53b384e07 100644 --- a/predicators/agent_sdk/tools/synthesis.py +++ b/predicators/agent_sdk/tools/synthesis.py @@ -487,8 +487,14 @@ def _evaluate_rollout_fit(rules: list, ] # An inert fit must announce itself: every value below is then # the declared init, not an estimate, and "fitted" language - # would launder hand-set constants as calibration. - if fitted and all(fitted[n] == init_params[n] for n in fitted): + # would launder hand-set constants as calibration. Tolerance, + # not exact equality: log-scale params round-trip through + # exp(log(x)) (inexact for e.g. 0.1 or 3.0) and solve_lm nudges + # bound-sitting inits into the interior by 1e-9, so an exact + # test never fires on exactly the fits it exists to call out. + if fitted and all( + np.isclose(fitted[n], init_params[n], rtol=1e-8, atol=1e-12) + for n in fitted): lines.insert( 1, "NOTE: the optimizer moved NO parameter (every delta " "is +0.0000) - this fit is a no-op and the values below " diff --git a/predicators/agent_sdk/tools/testing.py b/predicators/agent_sdk/tools/testing.py index 5910735c4..67e044838 100644 --- a/predicators/agent_sdk/tools/testing.py +++ b/predicators/agent_sdk/tools/testing.py @@ -368,10 +368,10 @@ def _report_step(i: int, outcome: Any) -> None: f"{format_object_poses(outcome.pre_state)}") return step_line = f"Step {i}: {sig} ({outcome.num_actions} actions)" - if (opt.name == "Wait" and outcome.failure_reason is None and - outcome.num_actions >= CFG.max_num_steps_option_rollout): + if (opt.name == "Wait" and outcome.failure_reason is None + and outcome.num_actions >= utils.wait_rollout_step_cap()): step_line += ( - "\n NOTE: this Wait ran to the option-rollout cap - " + "\n NOTE: this Wait ran to its step cap - " "its wait-target atoms never became true in the " "belief (and no other atom changed). Check whether " "the awaited change is modeled, or drop the Wait.") diff --git a/predicators/code_sim_learning/config.py b/predicators/code_sim_learning/config.py index dc2f34c39..b135f14ac 100644 --- a/predicators/code_sim_learning/config.py +++ b/predicators/code_sim_learning/config.py @@ -17,10 +17,26 @@ from __future__ import annotations +import os from dataclasses import dataclass from typing import Tuple -from predicators.settings import CFG +from predicators.settings import CFG, LAUNCH_CWD + + +def _anchor_at_launch_cwd(path: str) -> str: + """Resolve a relative CFG path against the process launch directory. + + Experiment yamls declare track manifests relative to the launch + (repo) directory, but ``from_cfg`` may run inside ``run_python``'s + exec window, whose working directory is the agent sandbox - a + bare relative open() there misses the file and the track scorer + silently falls back to per-step scoring. Absolute and empty paths + pass through untouched. + """ + if not path or os.path.isabs(path): + return path + return os.path.join(LAUNCH_CWD, path) @dataclass(frozen=True) @@ -100,7 +116,8 @@ def from_cfg(cls) -> SysIdConfig: CFG.code_sim_learning_log_hessian_identifiability), score_observed_only=( CFG.code_sim_learning_rollout_score_observed_only), - track_path=CFG.code_sim_learning_rollout_track_path, + track_path=_anchor_at_launch_cwd( + CFG.code_sim_learning_rollout_track_path), onset_confirm_deg=CFG.code_sim_learning_onset_confirm_deg, onset_deg=CFG.code_sim_learning_onset_deg, onset_min_persist=CFG.code_sim_learning_onset_min_persist, diff --git a/predicators/code_sim_learning/lm.py b/predicators/code_sim_learning/lm.py index 134e376cc..2800516cc 100644 --- a/predicators/code_sim_learning/lm.py +++ b/predicators/code_sim_learning/lm.py @@ -153,8 +153,11 @@ def lm_point_fit_result( # no signal (jitter ~1e-12 over a 2e-2 relative step), while a genuinely # responsive column of the dimensionless scaled residuals is O(1). A # column counts as zero-gradient when its largest |entry| on the DATA -# rows is at or below max(abs tol, rel tol * the largest entry in any -# data column). +# rows is at or below max(abs tol, rel tol * the MEDIAN column max) - +# median, not max, so one badly-scaled residual block cannot inflate +# the flatness threshold applied to every other parameter's column; +# when most columns are flat the median sits at the junk scale and the +# absolute floor governs, which errs conservative (LM keeps the param). _ZERO_COL_ABS_TOL = 1e-8 _ZERO_COL_REL_TOL = 1e-6 @@ -180,7 +183,7 @@ def zero_jacobian_columns(jac: np.ndarray, n_prior_rows: int = 0) -> List[int]: if data.size == 0: return [] col_max = np.max(np.abs(data), axis=0) - tol = max(_ZERO_COL_ABS_TOL, _ZERO_COL_REL_TOL * float(np.max(col_max))) + tol = max(_ZERO_COL_ABS_TOL, _ZERO_COL_REL_TOL * float(np.median(col_max))) return [j for j in range(data.shape[1]) if col_max[j] <= tol] @@ -225,12 +228,14 @@ def bracket_search_zero_gradient_params( whose data SSE is flat at both edges is declared flat for 2 evaluations instead of 9 - for the piecewise-constant thresholds this search exists for, a response anywhere in the box almost - always shows at an edge. (An interior-only dip whose edges match - the current SSE is the accepted blind spot; the full grid only - ever ran for parameters the LM gradient already called flat.) - ``flat_out``, when given, collects the names of parameters found - flat across their box - box-wide insensitivity evidence the - identifiability report can consume instead of re-probing them. + always shows at an edge. An interior-only dip whose edges match + the current SSE is the accepted blind spot of the SEARCH, but not + of the verdicts: edge-screened params are deliberately kept out of + ``flat_out`` so the identifiability probe (interior +-sigma evals) + stays armed as their backstop. ``flat_out``, when given, collects + only the params the FULL grid measured flat across their box - + box-wide insensitivity evidence the identifiability report can + consume instead of re-probing them. """ z = np.array(z, dtype=float) @@ -264,11 +269,14 @@ def _eval_at(g: float, col: int = j) -> Tuple[float, float]: hi_data, hi_total = _eval_at(float(grid[-1])) if (abs(lo_data - data_sse) <= flat_tol and abs(hi_data - data_sse) <= flat_tol): + # NOT added to ``flat_out``: 2 edge evaluations cannot rule + # out an interior-only response, so the identifiability + # probe (whose +-sigma evals are interior) must stay armed + # as the backstop for these params - only the full-grid + # verdict below may suppress it. notes.append(f"{spec.name}: data SSE is flat at both box edges, " "so the data do not constrain it; kept at " f"{init_ext:.4g} (NOT fit from data).") - if flat_out is not None: - flat_out.append(spec.name) continue pairs = [(lo_data, lo_total)] pairs += [_eval_at(float(g)) for g in grid[1:-1]] @@ -435,7 +443,12 @@ def internal_residuals(z: np.ndarray) -> np.ndarray: if sse_new < sse_lm: # The moved gates may have given the smooth parameters a # gradient: polish from the new point. A failure here keeps - # the searched point (better than the LM one by construction). + # the searched point (better than the LM one by construction) + # and DROPS the Jacobian: the pre-bracket jac was evaluated + # at the old theta, and returning it alongside the moved + # theta would feed a wrong-point curvature to the Laplace + # ensemble and the Hessian diagnostic (consumers already + # handle jacobian=None). try: polished = least_squares(internal_residuals, z_new, @@ -448,11 +461,13 @@ def internal_residuals(z: np.ndarray) -> np.ndarray: jac = np.asarray(result.jac, dtype=float) else: result.x = z_new + jac = np.zeros((0, 0)) except Exception as exc: # pylint: disable=broad-except logger.warning( "%s LM polish after bracket search raised " "%s; keeping the searched point.", label, exc) result.x = z_new + jac = np.zeros((0, 0)) sse_lm = min(float(2.0 * result.cost), sse_new) logger.info("%s LM fit after bracket search: SSE %.4f.", label, sse_lm) diff --git a/predicators/code_sim_learning/physical_sysid.py b/predicators/code_sim_learning/physical_sysid.py index 490e39bb1..84a0ec420 100644 --- a/predicators/code_sim_learning/physical_sysid.py +++ b/predicators/code_sim_learning/physical_sysid.py @@ -583,6 +583,7 @@ def refit_pinned(surviving: List[ParamSpec], prior_sigma=result.prior_sigma, scales=result.scales, sensitivity=result.sensitivity, + lm_notes=list(result.lm_notes), anchor_ablation=pinned) diff --git a/predicators/settings.py b/predicators/settings.py index aad47452e..b5a8ac337 100644 --- a/predicators/settings.py +++ b/predicators/settings.py @@ -4,12 +4,23 @@ (args.py). """ +import os from collections import defaultdict from types import SimpleNamespace from typing import Any, Dict, List, Optional, Set import numpy as np +# Working directory at process start. run_python's exec window chdirs +# into the agent sandbox (see agent_sdk/tools/python_exec.py), so any +# consumer that resolves a RELATIVE CFG path at call time must anchor +# it here rather than on os.getcwd() - first observed on the fan +# domain's relative rollout track path, which a sim.fit invoked inside +# run_python would fail to find, silently falling back to per-step +# scoring. This module is imported at process start, before any agent +# session (and therefore any sandbox chdir) can exist. +LAUNCH_CWD = os.path.abspath(os.getcwd()) + class GlobalSettings: """Unchanging settings.""" diff --git a/predicators/utils.py b/predicators/utils.py index f0384c81c..14c9b6a35 100644 --- a/predicators/utils.py +++ b/predicators/utils.py @@ -1846,6 +1846,25 @@ def _format_wait_target_debug( return "; ".join(details) +def wait_rollout_step_cap() -> int: + """The step count at which a belief-rollout Wait is force-terminated. + + Wait termination has two ceilings: the option model's + ``wait_option_max_steps`` backstop (active with + ``wait_option_terminate_on_atom_change``, mirroring the real + executor's branches in :func:`option_policy_to_policy`) and the + generic ``max_num_steps_option_rollout`` cap. Report code asking + "did this Wait stall?" must compare against whichever fires FIRST: + the bridge configures the backstop at 120 against a 1000-step + rollout cap, so a notice keyed on the rollout cap alone can never + fire in exactly the runs it was written for. + """ + cap = int(CFG.max_num_steps_option_rollout) + if CFG.wait_option_terminate_on_atom_change: + cap = min(cap, int(CFG.wait_option_max_steps)) + return cap + + def option_policy_to_policy( option_policy: Callable[[State], _Option], max_option_steps: Optional[int] = None, diff --git a/scripts/sysid_fit_diagnosis.py b/scripts/sysid_fit_diagnosis.py index 7ba8d39ef..ab971a6cc 100644 --- a/scripts/sysid_fit_diagnosis.py +++ b/scripts/sysid_fit_diagnosis.py @@ -283,9 +283,14 @@ def sse_at(segments: List[RolloutTrajectory], scaling=scaling, anchors=anchors) fitted = result.point_estimate + # Tolerance, not exact !=: log-scale params round-trip through + # exp(log(x)) and solve_lm nudges bound-sitting inits by 1e-9, so an + # exact test reports phantom "moved" parameters (observed: + # glue_dose_steps 3.0 -> 3.000000001 on run_20260830 seed0). moved = { n: (init_params[n], fitted[n]) - for n in fitted if fitted[n] != init_params[n] + for n in fitted + if not np.isclose(fitted[n], init_params[n], rtol=1e-8, atol=1e-12) } print(f"Fit replay: {len(survivors)}/{len(rollouts)} segments " f"survived trimming (per-segment best RMS: " diff --git a/tests/agent_sdk/test_bilevel_sketch_regions.py b/tests/agent_sdk/test_bilevel_sketch_regions.py index 6fce702f5..0c5bb4893 100644 --- a/tests/agent_sdk/test_bilevel_sketch_regions.py +++ b/tests/agent_sdk/test_bilevel_sketch_regions.py @@ -625,3 +625,25 @@ def test_format_step_line_renders_negative_subgoals(): atom = GroundAtom(_ReachedHi, [_block]) line = format_step_line(0, "Wait0", [_block], subgoal_neg_atoms={atom}) assert "-> {NOT ReachedHi(block0:block)}" in line + + +def test_wait_rollout_step_cap_tracks_the_binding_ceiling(): + """The stalled-Wait notice threshold is whichever cap fires first. + + Regression: the notice compared against max_num_steps_option_rollout + (1000) while the option model's wait_option_max_steps backstop (120 + on the bridge) terminated the Wait first, so the notice could never + fire in exactly the configured envs it was written for. + """ + utils.reset_config({ + "wait_option_terminate_on_atom_change": True, + "wait_option_max_steps": 120, + "max_num_steps_option_rollout": 1000, + }) + assert utils.wait_rollout_step_cap() == 120 + utils.reset_config({ + "wait_option_terminate_on_atom_change": False, + "wait_option_max_steps": 120, + "max_num_steps_option_rollout": 1000, + }) + assert utils.wait_rollout_step_cap() == 1000 diff --git a/tests/code_sim_learning/test_observation_track.py b/tests/code_sim_learning/test_observation_track.py index 94d2e1b0f..dc0662068 100644 --- a/tests/code_sim_learning/test_observation_track.py +++ b/tests/code_sim_learning/test_observation_track.py @@ -1581,3 +1581,32 @@ def _fake(_env, trajectories, *_a, **kwargs): def pytest_approx(value, abs=1e-9): # pylint: disable=redefined-builtin """Local approx so the comparisons above read as equations.""" return pytest.approx(value, abs=abs) + + +def test_sysid_config_track_path_anchored_at_launch_cwd(monkeypatch, tmp_path): + """A relative track path resolves against the launch directory even when + ``from_cfg`` runs under a different working directory. + + Regression: run_python's exec window chdirs into the agent sandbox, + so a sim.fit issued from agent code would open the fan domain's + relative track path against the sandbox, miss it, and silently fall + back to per-step scoring. + """ + # pylint: disable-next=import-outside-toplevel + from predicators.code_sim_learning.config import SysIdConfig + # pylint: disable-next=import-outside-toplevel + from predicators.settings import LAUNCH_CWD + utils.reset_config({ + "code_sim_learning_rollout_track_path": + "logs/zed_tracks/tracks.json", + }) + monkeypatch.chdir(tmp_path) # simulate the sandbox exec window + cfg = SysIdConfig.from_cfg() + assert cfg.track_path == os.path.join(LAUNCH_CWD, + "logs/zed_tracks/tracks.json") + # Absolute and empty paths pass through untouched. + abs_path = str(tmp_path / "tracks.json") + utils.reset_config({"code_sim_learning_rollout_track_path": abs_path}) + assert SysIdConfig.from_cfg().track_path == abs_path + utils.reset_config({"code_sim_learning_rollout_track_path": ""}) + assert SysIdConfig.from_cfg().track_path == "" diff --git a/tests/code_sim_learning/test_zero_gradient_search.py b/tests/code_sim_learning/test_zero_gradient_search.py index c7021351c..e95a888ce 100644 --- a/tests/code_sim_learning/test_zero_gradient_search.py +++ b/tests/code_sim_learning/test_zero_gradient_search.py @@ -118,8 +118,12 @@ def residuals(theta: np.ndarray) -> np.ndarray: def test_bracket_search_flat_param_settled_from_edge_evals() -> None: - """A box-flat parameter is settled from the two edge evaluations, is - reported in ``flat_out``, and never pays the full 9-point grid.""" + """A box-flat parameter is settled from the two edge evaluations and. + + never pays the full 9-point grid - but is NOT reported in + ``flat_out``: 2 edge evals cannot rule out an interior-only + response, so the identifiability probe must stay armed for it. + """ calls: List[np.ndarray] = [] def residuals(z: np.ndarray) -> np.ndarray: @@ -142,7 +146,7 @@ def residuals(z: np.ndarray) -> np.ndarray: "test", flat_out=flat) assert z_new[1] == 0.7 - assert flat == ["gate"] + assert not flat # edge screen alone never suppresses the probe assert len(notes) == 1 and "NOT fit from data" in notes[0] # One baseline evaluation plus the two box edges. assert len(calls) == 3 @@ -179,5 +183,6 @@ def residuals(z: np.ndarray) -> np.ndarray: "test", n_prior_rows=2, flat_out=flat) - assert flat == ["gate"] + assert not flat # edge screen alone never suppresses the probe assert len(notes) == 1 and "NOT fit from data" in notes[0] + assert "flat at both box edges" in notes[0] From 908411f44691e54f69ef7f8703d777229ac694c9 Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Tue, 1 Sep 2026 05:53:45 -0400 Subject: [PATCH 5/8] execution monitor: resume open-loop when a divergence has no refinable suffix From the 2026-09-01 seed 2-4 regression diagnosis: 21d21641 made invented-predicate annotations actually reach the subgoal monitor (the 2026-08-30 runs dropped them all at re-parse, so monitoring was inert), and the abort behind it - divergence + no refinable suffix + fallback disabled = ApproachFailure - killed both of seed 4's validated test episodes on millimeter-scale claim misses whose remaining settle/cure steps might still have delivered. An annotation is the agent's prediction, not proof the goal is out of reach, so a divergence with no refinable suffix (or a spent replan budget) now resumes the remaining not-yet-executed options open-loop: the divergence stays in the log, monitoring re-arms over the resumed suffix, and the goal check decides the episode. The grounded plan rides _exec_plan next to _exec_status so the handler can slice off the remaining options (the dispensed policy holds them only in its closure). agent_bilevel_replan_agent_fallback still opts into a fresh agent sketch instead. Claude-Session: https://claude.ai/code/session_017NQYkVi9etnpoMJBQobvGw --- .../approaches/agent_model_based_approach.py | 76 +++++++++++-------- predicators/settings.py | 16 ++-- .../test_agent_model_based_approach.py | 64 ++++++++++++---- 3 files changed, 105 insertions(+), 51 deletions(-) diff --git a/predicators/approaches/agent_model_based_approach.py b/predicators/approaches/agent_model_based_approach.py index 7fa957ece..dad592855 100644 --- a/predicators/approaches/agent_model_based_approach.py +++ b/predicators/approaches/agent_model_based_approach.py @@ -134,6 +134,11 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: # to the subgoal_annotations execution monitor. None whenever no # monitored plan is active (exploration, replanning disabled). self._exec_status: Optional[SubgoalExecutionStatus] = None + # The grounded option plan behind _exec_status, kept so a + # divergence with no refinable suffix can resume the remaining + # not-yet-executed options open-loop (the dispensed policy holds + # them only in its closure). Set/cleared alongside _exec_status. + self._exec_plan: Optional[List[_Option]] = None # Per-episode replan budget, refreshed by reset_for_new_episode. self._exec_replans_left = 0 # Whether the most recent sketch query ended because the agent hit @@ -168,6 +173,7 @@ def get_name(cls) -> str: def reset_for_new_episode(self) -> None: super().reset_for_new_episode() self._exec_status = None + self._exec_plan = None self._exec_replans_left = CFG.agent_bilevel_max_execution_replans # Optionally give each test solve a fresh agent conversation. reset() # fires once per test task (not on mid-episode replans, which go @@ -838,45 +844,52 @@ def _maybe_replan_from_divergence( problem (a sampled parameter whose real outcome differed from the option-model rollout), not a wrong skeleton, so we first try to resume a suffix of the executed sketch (cheap, no agent - query; see :meth:`_replan_suffix`). When that fails, the default - is to fail the episode: a fresh agent sketch query would re-open - a turn budget the attempt already spent (set - ``CFG.agent_bilevel_replan_agent_fallback`` to return None and - fall through to one instead). Also raises ApproachFailure when - the episode's replan budget is exhausted so the episode fails - fast instead of running the horizon open-loop. + query; see :meth:`_replan_suffix`). When no suffix refines - or + the episode's replan budget is spent - the remaining + not-yet-executed options resume OPEN-LOOP instead of failing the + episode: an annotation is the agent's prediction, not proof the + goal is out of reach, and aborting a plan whose remaining + settle/cure steps might still deliver turns a maybe-fail into a + certain fail. The divergence stays in the log and the goal check + decides the episode. Set + ``CFG.agent_bilevel_replan_agent_fallback`` to instead fall + through to a fresh agent sketch query when no suffix refines. """ status = self._exec_status if status is None or status.steps_initiated == 0: return None self._exec_status = None + exec_plan = self._exec_plan or [] + self._exec_plan = None failed_idx = status.steps_initiated - 1 steps = list(status.sketch) failed_name = steps[failed_idx].option.name - if self._exec_replans_left <= 0: - raise ApproachFailure( - f"Subgoal divergence after step {failed_idx} " - f"({failed_name}). No execution replans left.") - self._exec_replans_left -= 1 - logging.info( - "Subgoal divergence after step %d (%s). Replanning from the " - "current state (%d execution replans left).", failed_idx, - failed_name, self._exec_replans_left) - policy = self._replan_suffix(task.init, task, steps, failed_idx, - timeout) - if policy is None: - if not CFG.agent_bilevel_replan_agent_fallback: - raise ApproachFailure( - f"Subgoal divergence after step {failed_idx} " - f"({failed_name}): no suffix of the executed sketch " - "refines from here, and the fresh-agent-sketch " - "fallback is disabled " - "(agent_bilevel_replan_agent_fallback).") - # No suffix of the executed skeleton refines from here; fall - # through to pay for a fresh agent sketch. - logging.info("Suffix replan failed; querying the agent for a " - "fresh sketch.") - return policy + if self._exec_replans_left > 0: + self._exec_replans_left -= 1 + logging.info( + "Subgoal divergence after step %d (%s). Replanning from the " + "current state (%d execution replans left).", failed_idx, + failed_name, self._exec_replans_left) + policy = self._replan_suffix(task.init, task, steps, failed_idx, + timeout) + if policy is not None: + return policy + if CFG.agent_bilevel_replan_agent_fallback: + # No suffix of the executed skeleton refines from here; + # fall through to pay for a fresh agent sketch. + logging.info("Suffix replan failed; querying the agent for " + "a fresh sketch.") + return None + reason = "no suffix of the executed sketch refines from here" + else: + reason = "no execution replans left" + remaining = list(exec_plan[failed_idx + 1:]) + logging.warning( + "Subgoal divergence after step %d (%s): %s. Resuming the " + "remaining %d step(s) open-loop; the divergence stands " + "recorded and the goal check decides the episode.", failed_idx, + failed_name, reason, len(remaining)) + return self._plan_to_policy(remaining, sketch=steps[failed_idx + 1:]) def _nudge_final_submission(self) -> Optional[Callable[[State], Action]]: """One short follow-up query on the LAST attempt, after its query ended @@ -1187,6 +1200,7 @@ def _abstract(s: State) -> Set[GroundAtom]: assert sketch is not None status = SubgoalExecutionStatus(sketch=list(sketch)) self._exec_status = status + self._exec_plan = list(plan) def _option_policy(state: State) -> _Option: del state # unused diff --git a/predicators/settings.py b/predicators/settings.py index b5a8ac337..83b62fb84 100644 --- a/predicators/settings.py +++ b/predicators/settings.py @@ -1693,17 +1693,21 @@ class GlobalSettings: # that settled off-target), CogMan re-invokes solve(), which resumes a # re-refined suffix of the executed sketch from the current state, # instead of running the rest of the stale plan open-loop. Value = - # recoveries per test episode, shared across chained replans; 0 - # disables (legacy open-loop execution). Requires --execution_monitor + # recoveries per test episode, shared across chained replans; when no + # suffix refines (or the budget is spent) the remaining plan resumes + # open-loop rather than failing the episode. 0 disables (legacy + # open-loop execution). Requires --execution_monitor # subgoal_annotations (enforced at approach construction). agent_bilevel_max_execution_replans = 0 # When an execution replan's suffix refinement fails, whether to fall # back to querying the agent for a fresh sketch - a brand-new # full-turn-budget session. Default False: the cheap suffix replan is - # the only recovery, and the episode fails when no suffix of the - # executed sketch refines from the diverged state. Re-opening the - # agent budget is especially wasteful after a best-effort (non-solve) - # capture, whose execution diverges by construction. + # the only recovery, and when no suffix of the executed sketch + # refines from the diverged state the remaining plan resumes + # open-loop (the divergence is logged; the goal check decides the + # episode). Re-opening the agent budget is especially wasteful after + # a best-effort (non-solve) capture, whose execution diverges by + # construction. agent_bilevel_replan_agent_fallback = False # log state pretty_str before/after each step agent_bilevel_log_state = False diff --git a/tests/approaches/test_agent_model_based_approach.py b/tests/approaches/test_agent_model_based_approach.py index a79c812c4..d0c7683aa 100644 --- a/tests/approaches/test_agent_model_based_approach.py +++ b/tests/approaches/test_agent_model_based_approach.py @@ -1197,14 +1197,15 @@ def sentinel_policy(s): assert args[0] is state # replans from the real current state assert args[3] == 0 # the failed step is the annotated first step - def test_episode_fails_when_no_suffix_validates(self): - """Suffix path exhausted: by default the episode fails. + def test_openloop_resume_when_no_suffix_validates(self): + """Suffix path exhausted: the remaining plan resumes open-loop. - A fresh sketch query would re-open the agent turn budget the - attempt already spent, so it is opt-in + An annotation is the agent's prediction, not proof the goal is + out of reach, so by default the episode keeps executing and the + goal check decides. A fresh sketch query would re-open the agent + turn budget the attempt already spent, so it stays opt-in (agent_bilevel_replan_agent_fallback). """ - from predicators.approaches import ApproachFailure approach, _, task = _make_approach() _enable_replanning(approach, 2) holding = {GroundAtom(_Holding, [_block0])} @@ -1213,10 +1214,41 @@ def test_episode_fails_when_no_suffix_validates(self): state = _make_state() policy(state) approach._replan_suffix = MagicMock(return_value=None) - with pytest.raises(ApproachFailure, - match="agent_bilevel_replan_agent_fallback"): - approach._solve(Task(state, task.goal), timeout=10) + approach._query_agent_for_plan_sketch = MagicMock() + new_policy = approach._solve(Task(state, task.goal), timeout=10) approach._replan_suffix.assert_called_once() + approach._query_agent_for_plan_sketch.assert_not_called() + # The resumed policy executes the remaining step (Place), and + # monitoring re-arms over exactly that suffix. + new_policy(state) + status = approach.get_execution_monitoring_info()[0] + assert status.steps_initiated == 1 + assert status.current_option.name == "Place" + + def test_openloop_resume_at_last_step_ends_plan(self): + """Divergence at the final step leaves nothing to resume: the returned + policy ends through the normal plan-exhausted path (so a goal-reached + terminator still gets its chance), not a divergence abort.""" + from predicators.approaches import ApproachFailure + approach, _, task = _make_approach() + _enable_replanning(approach, 2) + holding = {GroundAtom(_Holding, [_block0])} + plan, _ = _make_two_step_plan(holding) + # Annotate the LAST step instead of the first. + sketch = [ + _SketchStep(_PickDone, [_block0], None), + _SketchStep(_PlaceDone, [_block0, _block1], holding), + ] + policy = approach._plan_to_policy(plan, sketch=sketch) + state = _make_state() # block0 not held: Place's subgoal fails + policy(state) # starts Pick + policy(state) # Pick terminal -> starts Place + monitor = _make_monitor(approach) + assert monitor.step(state) + approach._replan_suffix = MagicMock(return_value=None) + new_policy = approach._solve(Task(state, task.goal), timeout=10) + with pytest.raises(ApproachFailure, match="exhausted"): + new_policy(state) def test_full_resolve_when_no_suffix_validates_with_fallback(self): """With agent_bilevel_replan_agent_fallback, a failed suffix replan @@ -1243,9 +1275,9 @@ def test_full_resolve_when_no_suffix_validates_with_fallback(self): approach._replan_suffix.assert_called_once() def test_budget_shared_across_chained_replans(self): - """Chained replans share one per-episode budget and fail fast once it - is exhausted.""" - from predicators.approaches import ApproachFailure + """Chained replans share one per-episode budget; once it is exhausted, + a divergence resumes the remaining plan open-loop without paying for + further refinement.""" approach, _, task = _make_approach() _enable_replanning(approach, 1) holding = {GroundAtom(_Holding, [_block0])} @@ -1268,10 +1300,14 @@ def _suffix_replan(s, tsk, steps, k, t): new_policy(state) _sync(monitor, approach) assert monitor.step(state) - # Second divergence: no budget left. - with pytest.raises(ApproachFailure, match="No execution replans"): - approach._solve(Task(state, task.goal), timeout=10) + # Second divergence: no budget left - the remaining plan resumes + # open-loop, with no further refinement attempt. + resumed = approach._solve(Task(state, task.goal), timeout=10) + approach._replan_suffix.assert_called_once() approach._query_agent_for_plan_sketch.assert_not_called() + resumed(state) + status = approach.get_execution_monitoring_info()[0] + assert status.current_option.name == "Place" def test_reset_for_new_episode_clears_state(self): """A new episode refreshes the budget and clears the live status.""" From 338f746b944c37171bad783158cae2b271f9305a Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Tue, 1 Sep 2026 05:53:52 -0400 Subject: [PATCH 6/8] sim.fit report: margin-aware trimming advice, threshold in the banner From the 2026-09-01 seed 2-4 regression diagnosis: every fit of three independent bridge runs was refused with all segments at 1.00-1.35x the trimming cutoff - byte-identical under the pre-21d21641 code and the serial/fork objectives (job 21736393), i.e. a replay-fidelity floor, not chaotic data - yet the NO-FIT-RAN banner blamed unrepeatable recordings and steered the agents into re-collection loops that pegged their 45-minute solve budgets. Both trimming reports (the NO-FIT-RAN banner and the partial-trim note) now print the numeric threshold and split the dropped segments by margin: within 1.5x of the cutoff is reported as a model-fidelity limit (re-collecting the same experiments will score the same; improve the simulator's rules or trust the measured inits), and only far-over segments keep the chaotic-recording advice. DEFAULT_NOISE_SIGMA moves to physical_sysid so the reported cutoff and the trimmer's can never drift apart. Claude-Session: https://claude.ai/code/session_017NQYkVi9etnpoMJBQobvGw --- predicators/agent_sdk/tools/synthesis.py | 73 ++++++++++++++---- .../code_sim_learning/physical_sysid.py | 11 ++- tests/agent_sdk/test_trim_cause_note.py | 74 +++++++++++++++++++ 3 files changed, 142 insertions(+), 16 deletions(-) create mode 100644 tests/agent_sdk/test_trim_cause_note.py diff --git a/predicators/agent_sdk/tools/synthesis.py b/predicators/agent_sdk/tools/synthesis.py index 53b384e07..bb7bb9167 100644 --- a/predicators/agent_sdk/tools/synthesis.py +++ b/predicators/agent_sdk/tools/synthesis.py @@ -1,7 +1,7 @@ """Synthesis-session tools for sim learning (create_synthesis_tools).""" import dataclasses import os -from typing import Any, Callable, Dict, List, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union import numpy as np @@ -13,6 +13,52 @@ from predicators.agent_sdk.tools.sandbox_guard import _scrub_host_paths from predicators.agent_sdk.tools.snapshots import _ArtifactSnapshotter +# A trimmed segment scoring within this factor of the trimming +# threshold is reported as a model-fidelity limit rather than a chaotic +# recording. Calibrated on the 2026-08-31 bridge runs: every segment of +# three independent runs scored 1.00-1.35x the cutoff (a replay-fidelity +# floor no re-collection can move), while genuinely chaotic recordings +# score ~2x and beyond (seed4's scraping segments: 3-4x). +_TRIM_BORDERLINE_FACTOR = 1.5 + + +def _trim_cause_note(traj_rms: Sequence[float], threshold: float) -> List[str]: + """Advice for trimmed segments, split by how far past the cutoff they + scored. + + A segment within ``_TRIM_BORDERLINE_FACTOR`` of the threshold is the + closest the simulator can track that recording at ANY candidate + parameters - a model-fidelity floor, not a chaotic recording - so + re-collecting equivalent experiments cannot help and the advice says + so; only far-over segments get the chaotic-recording advice. + """ + dropped = [r for r in traj_rms if r > threshold] + close = [r for r in dropped if r <= _TRIM_BORDERLINE_FACTOR * threshold] + far = [r for r in dropped if r > _TRIM_BORDERLINE_FACTOR * threshold] + notes: List[str] = [] + if close: + pct = int(round((_TRIM_BORDERLINE_FACTOR - 1) * 100)) + notes.append( + f"{len(close)} dropped segment(s) score within {pct}% of the " + f"threshold (closest {min(close):.4g} vs {threshold:.4g}). A " + "margin that small is a model-fidelity limit, not chaotic " + "data: the simulator cannot replay those recordings any " + "closer at ANY candidate parameter values, so re-collecting " + "the same experiments will score the same. To make them " + "explainable, improve the simulator's dynamics rules where " + "replay deviates from the recording; until then the declared " + "init values (your own measurements) stand in as the model.") + if far: + notes.append( + f"{len(far)} dropped segment(s) score well past the threshold " + f"(worst {max(far):.4g}): such recordings are usually not " + "repeatable under replay (prolonged scraping/jamming " + "robot-object contact is chaotic). Collect experiments whose " + "outcome is dominated by object dynamics: actuate one or two " + "objects cleanly, then let the scene evolve and settle on " + "its own.") + return notes + @dataclasses.dataclass(frozen=True) class SynthesisToolkit: @@ -190,6 +236,8 @@ def create_synthesis_tools( from predicators.code_sim_learning.identifiability import \ format_identifiability, physics_sigma_points from predicators.code_sim_learning.orchestrator import run_rollout_sysid + from predicators.code_sim_learning.physical_sysid import \ + DEFAULT_NOISE_SIGMA from predicators.code_sim_learning.rollout_env import \ physical_param_anchors from predicators.code_sim_learning.rollout_objective import \ @@ -392,12 +440,14 @@ def _evaluate_rollout_fit(rules: list, approach._record_sysid_diagnostics( # pylint: disable=protected-access {}, physical_names, 0, len(rollouts), outcome.traj_rms) rms_str = ", ".join(f"{r:.4g}" for r in outcome.traj_rms) + trim_threshold = (CFG.code_sim_learning_rollout_trim_rms_factor * + DEFAULT_NOISE_SIGMA) return "\n".join([ f"[{version_tag}] NO FIT RAN: all {len(rollouts)} " "recorded motion segments were unexplainable at ANY " "candidate physical parameters (per-segment " f"best-achievable RMS [{rms_str}] all above the " - "trimming threshold).", + f"trimming threshold {trim_threshold:.4g}).", "", "Parameters were left at their baselines; nothing was " "applied to the planning base env.", @@ -406,13 +456,7 @@ def _evaluate_rollout_fit(rules: list, "values without a harness refit." if not exploratory else "Exploratory call: nothing recorded."), "", - "This usually means the recorded interactions are not " - "repeatable under replay (prolonged scraping/jamming " - "robot-object contact is chaotic). Collect experiments " - "whose outcome is dominated by object dynamics: actuate " - "one or two objects cleanly, then let the scene evolve " - "and settle on its own.", - ]) + ] + _trim_cause_note(outcome.traj_rms, trim_threshold)) fitted = outcome.fitted applied = outcome.applied ident_report = outcome.report @@ -505,15 +549,16 @@ def _evaluate_rollout_fit(rules: list, if outcome.num_survivors < len(rollouts): rms_str = ", ".join(f"{r:.4g}" for r in outcome.traj_rms) dropped = len(rollouts) - outcome.num_survivors + trim_threshold = (CFG.code_sim_learning_rollout_trim_rms_factor * + DEFAULT_NOISE_SIGMA) lines.insert( 1, f"Goodness-of-fit trimming: {dropped}" f" of {len(rollouts)} motion segments were unexplainable at " "ANY candidate params (per-segment best-achievable RMS: " - f"[{rms_str}]) and were dropped before fitting; the fit " - "below used only the explainable ones. Unexplainable " - "segments are not repeatable under replay - prefer " - "experiments whose outcome is dominated by object dynamics " - "rather than prolonged robot-object contact.") + f"[{rms_str}], trimming threshold {trim_threshold:.4g}) and " + "were dropped before fitting; the fit below used only the " + "explainable ones. " + + " ".join(_trim_cause_note(outcome.traj_rms, trim_threshold))) for name in sorted(fitted): init_val = init_params[name] fit_val = fitted[name] diff --git a/predicators/code_sim_learning/physical_sysid.py b/predicators/code_sim_learning/physical_sysid.py index 84a0ec420..b6159a62f 100644 --- a/predicators/code_sim_learning/physical_sysid.py +++ b/predicators/code_sim_learning/physical_sysid.py @@ -113,6 +113,13 @@ "truncate_settled_tail", ] +# The fit's assumed measurement noise on dimensionless scaled +# residuals. The signature defaults below and the sim.fit report's +# trimming-threshold math (agent_sdk/tools/synthesis.py) share this one +# constant so the reported cutoff can never drift from the one the +# trimmer applied. +DEFAULT_NOISE_SIGMA = 0.05 + # Prior width as a fraction of each param's box; shared by the rollout # fit default and the pinned-at-init fallback result so the two cannot # silently diverge. @@ -131,7 +138,7 @@ def fit_params_rollout( rules: Sequence[Any] = (), rule_specs: Sequence[ParamSpec] = (), latent_init: Any = None, - noise_sigma: float = 0.05, + noise_sigma: float = DEFAULT_NOISE_SIGMA, prior_sigma_scale: float = _ROLLOUT_PRIOR_SIGMA_SCALE, scaling: Optional[ResidualScaling] = None, anchors: Optional[Dict[str, float]] = None, @@ -650,7 +657,7 @@ def fit_params_rollout_trimmed( rules: Sequence[Any] = (), rule_specs: Sequence[ParamSpec] = (), latent_init: Any = None, - noise_sigma: float = 0.05, + noise_sigma: float = DEFAULT_NOISE_SIGMA, scaling: Optional[ResidualScaling] = None, anchors: Optional[Dict[str, float]] = None, rms_cache: Optional[Dict[Tuple, Tuple[List[float], diff --git a/tests/agent_sdk/test_trim_cause_note.py b/tests/agent_sdk/test_trim_cause_note.py new file mode 100644 index 000000000..b61106f38 --- /dev/null +++ b/tests/agent_sdk/test_trim_cause_note.py @@ -0,0 +1,74 @@ +"""Tests for the sim.fit trimming-advice helper. + +Regression for the 2026-08-31 bridge runs: every recorded segment of +three independent runs scored 0.1001-0.134 against the 0.100 trimming +threshold, and the report's only advice ("the recordings are chaotic, +collect different experiments") sent the agents into pointless +re-collection loops that burned their solve budgets. A segment a few +percent over the cutoff is a model-fidelity floor - no re-collection can +move it - and the advice must say so; only far-over segments earn the +chaotic-recording advice. +""" + +from predicators.agent_sdk.tools.synthesis import _TRIM_BORDERLINE_FACTOR, \ + _trim_cause_note + +_THRESHOLD = 0.1 + + +def test_borderline_segments_get_model_fidelity_advice() -> None: + """The 2026-08-31 pattern: everything a hair over the cutoff.""" + rms = [0.1001, 0.1025, 0.1126, 0.134] + notes = _trim_cause_note(rms, _THRESHOLD) + assert len(notes) == 1 + assert "model-fidelity limit" in notes[0] + assert "re-collecting the same experiments will score the same" \ + in notes[0] + assert "0.1001" in notes[0] # the closest miss is named + assert "0.1" in notes[0] # so is the threshold + assert "chaotic" not in notes[0].replace("not chaotic data", "") + + +def test_far_segments_get_chaos_advice() -> None: + """Only far-over segments keep the chaotic-recording advice.""" + rms = [0.3037, 0.4216] + notes = _trim_cause_note(rms, _THRESHOLD) + assert len(notes) == 1 + assert "not repeatable under replay" in notes[0] + assert "0.4216" in notes[0] # the worst offender is named + + +def test_mixed_segments_get_both_notes_with_correct_counts() -> None: + """Borderline and far segments each get their own counted note.""" + rms = [0.1001, 0.1126, 0.3037, 0.4216, 0.3037] + notes = _trim_cause_note(rms, _THRESHOLD) + assert len(notes) == 2 + assert notes[0].startswith("2 dropped segment(s)") + assert notes[1].startswith("3 dropped segment(s)") + + +def test_survivors_are_ignored() -> None: + """Values at or under the threshold were kept, not dropped: they contribute + to neither note.""" + rms = [0.05, 0.1, 0.1126] + notes = _trim_cause_note(rms, _THRESHOLD) + assert len(notes) == 1 + assert notes[0].startswith("1 dropped segment(s)") + + +def test_no_dropped_segments_means_no_notes() -> None: + """With nothing over the threshold there is nothing to advise on.""" + assert not _trim_cause_note([0.01, 0.02], _THRESHOLD) + assert not _trim_cause_note([], _THRESHOLD) + + +def test_boundary_lands_on_the_borderline_side() -> None: + """A segment exactly at factor x threshold is still borderline; just past + it is chaos.""" + at_edge = _TRIM_BORDERLINE_FACTOR * _THRESHOLD + notes = _trim_cause_note([at_edge], _THRESHOLD) + assert len(notes) == 1 + assert "model-fidelity limit" in notes[0] + notes = _trim_cause_note([at_edge * 1.01], _THRESHOLD) + assert len(notes) == 1 + assert "not repeatable under replay" in notes[0] From 1dfbf7d8057be0c02f8d330401f83448144a1940 Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Tue, 1 Sep 2026 06:00:41 -0400 Subject: [PATCH 7/8] config: disable execution replans for the plan arm Suffix replanning never succeeded in any bridge run: 0 of 4 divergence events recovered across seeds 0-4 (2026-08-29 through 2026-09-01), at ~8 futile refinement attempts of up to 600 s each, and every event then aborted the episode (pre-8cccfe42). Eval execution goes back to open-loop, matching the 2026-08-30 generation that solved 2/2. The flag also disarms the divergence monitor; re-raising it re-enables advisory monitoring, now with the open-loop resume from 8cccfe42 instead of the abort. Claude-Session: https://claude.ai/code/session_017NQYkVi9etnpoMJBQobvGw --- scripts/configs/predicatorv3/approaches/all.yaml | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/scripts/configs/predicatorv3/approaches/all.yaml b/scripts/configs/predicatorv3/approaches/all.yaml index f7d6eceaa..aa5118222 100644 --- a/scripts/configs/predicatorv3/approaches/all.yaml +++ b/scripts/configs/predicatorv3/approaches/all.yaml @@ -48,7 +48,14 @@ APPROACHES: partially_observable: True agent_explorer_info_seeking: True execution_monitor: "subgoal_annotations" - agent_bilevel_max_execution_replans: 2 + # Suffix replanning never succeeded in any bridge run (0/4 + # divergence events recovered, ~8 futile refinement attempts of up + # to 600 s each; measured 2026-09-01 across seeds 0-4), so eval + # execution runs open-loop like the 2026-08-30 generation that + # solved 2/2. 0 also disarms the divergence monitor - re-raise to + # re-enable it (divergences then resume open-loop after the failed + # refinement, 8cccfe42). + agent_bilevel_max_execution_replans: 0 agent_bilevel_use_llm_initial_params: True # LLM proposes params agent_sdk_max_agent_turns_per_iteration: 200 agent_sdk_image_max_px: 900 From 7651fde13cd5227f0bd559f6c6cc8d9042997433 Mon Sep 17 00:00:00 2001 From: Yichao Liang Date: Tue, 1 Sep 2026 09:21:44 -0400 Subject: [PATCH 8/8] utils: wait_rollout_step_cap tolerates the default inf backstop wait_rollout_step_cap() ran int(CFG.wait_option_max_steps) whenever wait_option_terminate_on_atom_change was set, but the setting's default is float('inf') (no backstop configured), so every probe/report path under a default config crashed with OverflowError: cannot convert float infinity to integer. The backstop now only participates in the min when it is finite, and the binding-ceiling regression test pins the inf case to the rollout cap. Claude-Session: https://claude.ai/code/session_01VQkk8ycwwwiSKraDw88FVm --- predicators/utils.py | 4 +++- tests/agent_sdk/test_bilevel_sketch_regions.py | 6 ++++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/predicators/utils.py b/predicators/utils.py index 14c9b6a35..ce76522e5 100644 --- a/predicators/utils.py +++ b/predicators/utils.py @@ -13,6 +13,7 @@ import io import itertools import logging +import math import os import pkgutil import re @@ -1860,7 +1861,8 @@ def wait_rollout_step_cap() -> int: fire in exactly the runs it was written for. """ cap = int(CFG.max_num_steps_option_rollout) - if CFG.wait_option_terminate_on_atom_change: + if CFG.wait_option_terminate_on_atom_change and \ + math.isfinite(CFG.wait_option_max_steps): cap = min(cap, int(CFG.wait_option_max_steps)) return cap diff --git a/tests/agent_sdk/test_bilevel_sketch_regions.py b/tests/agent_sdk/test_bilevel_sketch_regions.py index 0c5bb4893..4c8807093 100644 --- a/tests/agent_sdk/test_bilevel_sketch_regions.py +++ b/tests/agent_sdk/test_bilevel_sketch_regions.py @@ -647,3 +647,9 @@ def test_wait_rollout_step_cap_tracks_the_binding_ceiling(): "max_num_steps_option_rollout": 1000, }) assert utils.wait_rollout_step_cap() == 1000 + utils.reset_config({ + "wait_option_terminate_on_atom_change": True, + "wait_option_max_steps": float("inf"), + "max_num_steps_option_rollout": 1000, + }) + assert utils.wait_rollout_step_cap() == 1000