Retry ladder: time budget with forecast-based cap skipping

CHANGELOG.md

@@ -11,6 +11,12 @@ brute-force reference or a closed-form value and has a regression test.
   outside the stationary support could leave others with incomplete outgoing rows, so builds from
   truncated reverse machines failed with "joint masses from ... sum to 0.86". Restriction to the
   stationary support and to states with complete rows now alternate until neither removes anything.
+- The retry ladder takes a `time_budget` (`build_bidirectional_with_ladder`, `try_build_bidirectional`,
+  `resolve_bidirectional`), and `generative_model` / `minimal_wyner_model` charge the build to their
+  own `time_budget`. Mixed-state construction is not interruptible, so the ladder skips a cap when the
+  previous cap's time times the squared cap ratio would overrun, and sets `budget_exhausted`. On a
+  4-state survey process the 10k and 50k caps took 9 s and 86 s and the 200k cap did not finish in
+  15 minutes, so the ladder ran past the survey's hard timeout before any bounds were reported.
 
 ### Generator caps
 

sofic/generators/bidirectional_build.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import time
 from dataclasses import dataclass
 from typing import Literal
 
@@ -44,6 +45,7 @@ class BidirBuildResult:
     preflight: BidirPreflightResult
     pe_drift: float | None = None
     edge_stabilized: bool | None = None
+    budget_exhausted: bool = False
 
 
 def _edge_tensor(bidir: BidirectionalEpsilonMachine) -> np.ndarray:
@@ -102,20 +104,39 @@ def build_bidirectional_with_ladder(
     *,
     caps: tuple[int, ...] = _DEFAULT_CAPS,
     edge_tol: float = 1e-8,
+    time_budget: float | None = None,
 ) -> BidirBuildResult:
-    """Build a bidirectional machine, increasing reverse ``max_states`` until ``Pe`` stabilizes."""
+    """Build a bidirectional machine, increasing reverse ``max_states`` until ``Pe`` stabilizes.
+
+    With ``time_budget`` (seconds) the ladder stops before a cap whose forecast cost would overrun
+    the budget, and ``budget_exhausted`` is set on the result. A single mixed-state construction
+    cannot be interrupted, so the forecast scales the previous cap's time by the square of the cap
+    ratio (construction time grows faster than linearly in the cap); the first cap always runs.
+    """
+    start = time.monotonic()
     last_pe: np.ndarray | None = None
     last_bidir: BidirectionalEpsilonMachine | None = None
     last_preflight: BidirPreflightResult | None = None
+    previous: tuple[int, float] | None = None
+    exhausted = False
 
     for cap in caps:
+        if time_budget is not None and previous is not None:
+            prev_cap, prev_seconds = previous
+            forecast = prev_seconds * (cap / prev_cap) ** 2
+            if time.monotonic() - start + forecast > time_budget:
+                exhausted = True
+                break
+        attempt = time.monotonic()
         pre = preflight_bidir(forward, reverse_max_states=cap)
         last_preflight = pre
         if pre.bidir_class in ("explosion", "failed_build"):
+            previous = (cap, time.monotonic() - attempt)
             continue
         reverse = infer_reverse_epsilon_machine(forward, max_states=cap)
         bidir = build_bidirectional_epsilon_machine(forward, reverse)
         pe = _edge_tensor(bidir)
+        previous = (cap, time.monotonic() - attempt)
         if last_pe is not None:
             drift = _pe_max_drift(last_pe, pe)
             if drift <= edge_tol:
@@ -134,10 +155,11 @@ def build_bidirectional_with_ladder(
             last_preflight,
             pe_drift=drift,
             edge_stabilized=False,
+            budget_exhausted=exhausted,
         )
 
     if last_preflight is not None:
-        return BidirBuildResult(None, last_preflight)
+        return BidirBuildResult(None, last_preflight, budget_exhausted=exhausted)
     pre = preflight_bidir(forward, reverse_max_states=caps[-1])
     return BidirBuildResult(None, pre)
 
@@ -149,11 +171,18 @@ def try_build_bidirectional(
     use_ladder: bool = False,
     caps: tuple[int, ...] = _DEFAULT_CAPS,
     edge_tol: float = 1e-8,
+    time_budget: float | None = None,
 ) -> BidirBuildResult:
-    """Single-cap build, or ladder when ``use_ladder`` and the default cap explodes."""
+    """Single-cap build, or ladder when ``use_ladder`` and the default cap explodes.
+
+    ``time_budget`` (seconds, including the first preflight) is passed on to
+    :func:`build_bidirectional_with_ladder`.
+    """
+    start = time.monotonic()
     pre = preflight_bidir(forward, reverse_max_states=reverse_max_states)
     if pre.bidir_class == "explosion" and use_ladder:
-        return build_bidirectional_with_ladder(forward, caps=caps, edge_tol=edge_tol)
+        remaining = None if time_budget is None else max(0.0, time_budget - (time.monotonic() - start))
+        return build_bidirectional_with_ladder(forward, caps=caps, edge_tol=edge_tol, time_budget=remaining)
     if pre.bidir_class in ("explosion", "failed_build"):
         return BidirBuildResult(None, pre)
     reverse = infer_reverse_epsilon_machine(forward, max_states=reverse_max_states)

sofic/generators/generative_search.py

@@ -190,20 +190,24 @@ def resolve_bidirectional(
     *,
     reverse_max_states: int = 10_000,
     use_ladder: bool = True,
+    time_budget: float | None = None,
 ) -> tuple[BidirectionalEpsilonMachine, dict[str, Any]]:
     """Build or accept a bidirectional ε-machine for generator search."""
     from sofic.generators.bidirectional_build import try_build_bidirectional
 
     if isinstance(machine, BidirectionalEpsilonMachine):
         return machine, {"bidir_class": "finite", "source": "bidirectional input"}
     forward = machine if isinstance(machine, EpsilonMachine) else EpsilonMachine.from_hmm(machine)
-    built = try_build_bidirectional(forward, reverse_max_states=reverse_max_states, use_ladder=use_ladder)
+    built = try_build_bidirectional(
+        forward, reverse_max_states=reverse_max_states, use_ladder=use_ladder, time_budget=time_budget
+    )
     meta: dict[str, Any] = {
         "bidir_class": built.preflight.bidir_class,
         "reverse_max_states": built.preflight.reverse_max_states,
         "reverse_recurrence_finite": built.preflight.reverse_recurrence_finite,
         "pe_drift": built.pe_drift,
         "edge_stabilized": built.edge_stabilized,
+        "budget_exhausted": built.budget_exhausted,
     }
     if built.preflight.error:
         meta["build_error"] = built.preflight.error
@@ -872,7 +876,12 @@ def minimal_wyner_model(
     word-verified. The state-entropy lower bounds do not apply, so ``lower`` is the excess entropy and
     the bracket is certified only when a candidate attains it.
     """
-    bidir, build_meta = resolve_bidirectional(machine, reverse_max_states=reverse_max_states, use_ladder=use_ladder)
+    start = time.monotonic()
+    bidir, build_meta = resolve_bidirectional(
+        machine, reverse_max_states=reverse_max_states, use_ladder=use_ladder, time_budget=time_budget
+    )
+    if time_budget is not None:
+        time_budget = max(0.0, time_budget - (time.monotonic() - start))
     result = _bracket_search(
         bidir,
         objective="wyner",
@@ -938,9 +947,15 @@ def generative_model(
     With ``time_budget`` (seconds) the search becomes anytime: once the budget is spent, remaining
     stages are skipped and the best verified bracket so far is returned with ``budget_exhausted`` set.
     The check runs between stages (and between line-split seeds and coupled-search sizes), so a single
-    long stage can overrun it.
+    long stage can overrun it. The bidirectional build's retry ladder counts against the budget
+    (see :func:`~sofic.generators.bidirectional_build.build_bidirectional_with_ladder`).
     """
-    bidir, build_meta = resolve_bidirectional(machine, reverse_max_states=reverse_max_states, use_ladder=use_ladder)
+    start = time.monotonic()
+    bidir, build_meta = resolve_bidirectional(
+        machine, reverse_max_states=reverse_max_states, use_ladder=use_ladder, time_budget=time_budget
+    )
+    if time_budget is not None:
+        time_budget = max(0.0, time_budget - (time.monotonic() - start))
     result = generative_model_from_bidirectional(
         bidir,
         max_states=max_states,

tests/test_bidirectional_build.py

@@ -60,6 +60,33 @@ def test_survey_explosion_code_may_build_with_ladder() -> None:
         assert built.bidir is not None
 
 
+def test_ladder_budget_skips_caps_forecast_to_overrun(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A cap whose forecast (previous time x cap ratio squared) overruns the budget is not attempted."""
+    import sofic.generators.bidirectional_build as bb
+
+    clock = {"now": 0.0}
+    attempted: list[int] = []
+
+    def fake_preflight(forward, *, reverse_max_states):
+        attempted.append(reverse_max_states)
+        clock["now"] += 10.0 * (reverse_max_states / 10_000) ** 2
+        return bb.BidirPreflightResult("explosion", reverse_max_states=reverse_max_states)
+
+    monkeypatch.setattr(bb, "preflight_bidir", fake_preflight)
+    monkeypatch.setattr(bb.time, "monotonic", lambda: clock["now"])
+    built = bb.build_bidirectional_with_ladder(None, caps=(10_000, 20_000, 40_000), time_budget=100.0)
+    assert attempted == [10_000, 20_000]
+    assert built.bidir is None
+    assert built.budget_exhausted
+    assert built.preflight.reverse_max_states == 20_000
+
+    attempted.clear()
+    clock["now"] = 0.0
+    built = bb.build_bidirectional_with_ladder(None, caps=(10_000, 20_000, 40_000))
+    assert attempted == [10_000, 20_000, 40_000]
+    assert not built.budget_exhausted
+
+
 def test_truncated_reverse_machine_prunes_to_complete_rows() -> None:
     """Regression: dropping sub-tolerance joint states left (2, 'CE') with 86% of its outgoing mass."""
     mats = {x: np.zeros((4, 4)) for x in (0, 1)}

tests/test_coupled_generators.py

@@ -393,13 +393,14 @@ def test_coupled_memory_and_targeted_searches_stop_at_an_expired_deadline() -> N
     import time
 
     bidir = _rll()
+    problem = cg._GeneratorProblem(bidir)
+    # separating_optima takes no deadline; keep it out of the timed region (slow CI runners)
+    _best, optima = cg.separating_optima(problem, starts=10, rng=np.random.default_rng(0))
     start = time.monotonic()
     results = cg.coupled_search(bidir, 3, starts=50, deadline=time.monotonic(), rng=np.random.default_rng(0))
     assert results and all(r.residual <= 1e-6 for r in results)
     lifts = cg.memory_lift_search(bidir, 2, starts=50, deadline=time.monotonic(), rng=np.random.default_rng(0))
     assert len(lifts) == 1
-    problem = cg._GeneratorProblem(bidir)
-    _best, optima = cg.separating_optima(problem, starts=10, rng=np.random.default_rng(0))
     seeds = cg.targeted_line_split_seeds(problem, optima, deadline=time.monotonic(), rng=np.random.default_rng(0))
     assert len(seeds) <= 1
     assert time.monotonic() - start < 120