SysID honest fit reports, parallel objective, open-loop resume

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 "

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,15 +59,20 @@ 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
     from predicators.settings import CFG
     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)
 
 

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

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)

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]

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.")