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/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/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 93ffeb357..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 @@ -447,11 +491,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,23 +521,44 @@ 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. 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 " + "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 + 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/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/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/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/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/lm.py b/predicators/code_sim_learning/lm.py index 19752504b..2800516cc 100644 --- a/predicators/code_sim_learning/lm.py +++ b/predicators/code_sim_learning/lm.py @@ -147,22 +147,56 @@ 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 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 + + +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.median(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 +210,44 @@ 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 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) - 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 +258,39 @@ 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): + # 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).") + 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 +298,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 +318,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 +337,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 +370,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,20 +426,29 @@ 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: # 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, @@ -360,11 +461,13 @@ def _sse(z: np.ndarray) -> float: 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/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/predicators/code_sim_learning/physical_sysid.py b/predicators/code_sim_learning/physical_sysid.py index b7422ef39..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, @@ -261,6 +268,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 +280,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 +317,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( @@ -554,6 +590,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) @@ -620,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/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..83b62fb84 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.""" @@ -1682,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 @@ -1783,18 +1798,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/predicators/utils.py b/predicators/utils.py index f0384c81c..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 @@ -1846,6 +1847,26 @@ 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 and \ + math.isfinite(CFG.wait_option_max_steps): + 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/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 diff --git a/scripts/sysid_fit_diagnosis.py b/scripts/sysid_fit_diagnosis.py new file mode 100644 index 000000000..ab971a6cc --- /dev/null +++ b/scripts/sysid_fit_diagnosis.py @@ -0,0 +1,322 @@ +"""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 re +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 _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: + 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", + 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, + 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") + 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"] + 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 + # 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 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: " + 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() diff --git a/tests/agent_sdk/test_bilevel_sketch_regions.py b/tests/agent_sdk/test_bilevel_sketch_regions.py index 6fce702f5..4c8807093 100644 --- a/tests/agent_sdk/test_bilevel_sketch_regions.py +++ b/tests/agent_sdk/test_bilevel_sketch_regions.py @@ -625,3 +625,31 @@ 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 + 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 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] 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.""" 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_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..e95a888ce 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,131 @@ 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 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: + 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 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 + + +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 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]