diff --git a/lope/cli.py b/lope/cli.py index 566ad0c..6721c56 100644 --- a/lope/cli.py +++ b/lope/cli.py @@ -222,6 +222,10 @@ def _add_pool_flags(p): help="Name of the primary validator (must be in --validators)") p.add_argument("--timeout", type=_positive_int, default=None, help="Per-validator timeout in seconds") + p.add_argument("--respect-provider-timeout", dest="respect_provider_timeout", + action="store_true", default=False, + help="Let a provider's configured timeout outrank --timeout " + "(default: the stricter of the two wins)") run_group = p.add_mutually_exclusive_group() run_group.add_argument( "--run-timeout", dest="run_timeout", type=_positive_int, default=None, @@ -785,6 +789,11 @@ def _add_update_flags(p): format="%(name)s | %(message)s", ) + # Timeout precedence is resolved deep inside each provider call, far from + # argv; carry the operator's choice there through the environment. + if getattr(args, "respect_provider_timeout", False): + os.environ["LOPE_RESPECT_PROVIDER_TIMEOUT"] = "1" + if args.command == "version": from . import __version__ as _v from .logo import box @@ -2122,6 +2131,13 @@ def _plan_external_request( f"nominal ceiling {plan.nominal_wall_ceiling_seconds:g}s", file=sys.stderr, ) + try: + from .latency import fit_warnings + + for line in fit_warnings(validators, cfg.timeout): + print(line, file=sys.stderr) + except Exception: + pass return plan diff --git a/lope/generic_validators.py b/lope/generic_validators.py index a323cfe..a6a071b 100644 --- a/lope/generic_validators.py +++ b/lope/generic_validators.py @@ -61,6 +61,21 @@ class ConfigError(ValueError): """Raised when a provider config entry is invalid.""" +def respect_provider_timeout() -> bool: + """True when a provider's configured timeout outranks the call ceiling. + + Off by default: the stricter-wins rule below is deliberate and protects + bounded probes like `team test --timeout 10`. This escape hatch exists for + the opposite case — a provider configured slow *on purpose* (a + high-reasoning-effort model) being clamped by a generic ceiling that some + caller guessed. Set by `--respect-provider-timeout`. + """ + + return os.environ.get("LOPE_RESPECT_PROVIDER_TIMEOUT", "").strip().lower() in { + "1", "true", "yes", "on", + } + + def _effective_timeout(provider_timeout: Optional[int], call_timeout: Optional[int]) -> Optional[int]: """Return the timeout Lope should actually enforce for one provider call. @@ -68,11 +83,17 @@ def _effective_timeout(provider_timeout: Optional[int], call_timeout: Optional[i cap than the global Lope default. That provider value must never silently extend an explicit per-call timeout (`lope ask --timeout 30`, `team test --timeout 10`, etc.). Use the stricter value when both are present. + + The clamp is reported before launch by `latency.fit_warnings`, and can be + inverted per-invocation with `--respect-provider-timeout` when the provider + is the one that knows better. """ if provider_timeout is None: return call_timeout if call_timeout is None: return provider_timeout + if respect_provider_timeout(): + return provider_timeout return min(provider_timeout, call_timeout) diff --git a/lope/latency.py b/lope/latency.py new file mode 100644 index 0000000..eaede24 --- /dev/null +++ b/lope/latency.py @@ -0,0 +1,277 @@ +"""Per-validator latency ledger and pre-launch budget fit check. + +Lope has always measured how long each validator call takes — it prints the +number at the end of every run — but it never remembered it. Every run +re-derived its timeout from a flag or a config default, so a budget that could +not possibly fit was discovered only by spending the whole budget and failing. +A caller (increasingly, another agent) picks a number like 240 with no way to +know the validator's own p90 is 178s, and the run dies at the ceiling with all +of its work discarded. + +This module remembers, and answers one question before launch: given what this +validator has actually done here before, is the ceiling we are about to enforce +big enough for it to finish? The answer is advisory — it prints, it does not +block — because a first-ever call has no history and must still be allowed to +run. + +Storage is a bounded sample of recent call durations per validator at +``$LOPE_HOME/latency.json``. Timed-out calls are recorded too, as censored +lower bounds: a call killed at 240s proves the true duration exceeds 240s, so +folding it in as 240 biases the estimate upward, which is the safe direction +for a budget check. + +Nothing here may ever break a run. Every public entry point swallows its own +errors: a corrupt or unwritable ledger degrades to "no advice", never to a +failed validation. +""" + +from __future__ import annotations + +import json +import math +import os +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence + +SCHEMA_VERSION = 1 + +#: Keep the ledger small and recent; provider performance drifts with model +#: and machine changes, and an unbounded file would grow without value. +MAX_SAMPLES_PER_VALIDATOR = 50 + +#: Below this many observations a p90 is noise, so the fit check stays quiet. +MIN_SAMPLES_FOR_ADVICE = 3 + +#: A ceiling should clear the observed p90 with room for run-to-run variance. +#: Calibrated against the incident this module exists to prevent: the same +#: prompt on the same validator took 136s and 178s on two consecutive runs +#: (+31%), then exceeded 240s twice under load. A factor of 1.25 would have +#: called that 240s ceiling a good fit, which is precisely the wrong answer. +FIT_SAFETY_FACTOR = 1.5 + +#: Only these two outcomes say anything about how long the provider needs to +#: think. A launch error, a parse error or a parent-side cancellation produces +#: a duration that is an artefact of the failure, not of the provider's speed, +#: and folding those into the estimate would corrupt it. +_RECORDED_OUTCOMES = {"ok", "provider_timeout"} + +#: The outcome that means "the provider hit our wall", not "it answered". +_TIMEOUT_OUTCOME = "provider_timeout" + + +def enabled() -> bool: + """Return False when the operator has switched the ledger off.""" + + return os.environ.get("LOPE_LATENCY", "").strip().lower() not in { + "off", "0", "false", "no", + } + + +def ledger_path() -> Path: + """Return the ledger location, honouring LOPE_HOME like the config does.""" + + home = os.environ.get("LOPE_HOME", os.path.expanduser("~/.lope")) + return Path(home) / "latency.json" + + +@dataclass(frozen=True) +class Stats: + """Summary of one validator's recent observed call durations.""" + + validator: str + count: int + p50: float + p90: float + maximum: float + timeouts: int + + def recommended_ceiling(self) -> int: + """Smallest whole-second budget that clears p90 with variance room.""" + + return int(self.p90 * FIT_SAFETY_FACTOR) + 1 + + +def _percentile(values: Sequence[float], fraction: float) -> float: + """Nearest-rank percentile; `values` must be sorted and non-empty.""" + + if not values: + return 0.0 + rank = max(1, min(len(values), math.ceil(fraction * len(values)))) + return float(values[rank - 1]) + + +def _load() -> Dict[str, Any]: + try: + with open(ledger_path(), "r", encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, ValueError): + return {"schema_version": SCHEMA_VERSION, "validators": {}} + if not isinstance(data, dict) or not isinstance(data.get("validators"), dict): + return {"schema_version": SCHEMA_VERSION, "validators": {}} + return data + + +def _atomic_write(path: Path, payload: Dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd, tmp_name = tempfile.mkstemp(prefix=".latency-", suffix=".tmp", dir=str(path.parent)) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, sort_keys=True, separators=(",", ":")) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_name, path) + except Exception: + try: + os.unlink(tmp_name) + except OSError: + pass + raise + + +def record(validator: str, duration_seconds: float, *, outcome: str = "ok") -> None: + """Append one observation. Never raises.""" + + if not enabled(): + return + try: + name = str(validator or "").strip() + outcome_name = str(outcome or "").strip() + if not name or outcome_name not in _RECORDED_OUTCOMES: + return + # Round before validating, so a sub-millisecond duration is discarded + # rather than stored as a meaningless 0.0 sample. + duration = round(float(duration_seconds), 3) + if duration <= 0.0: + return + timed_out = outcome_name == _TIMEOUT_OUTCOME + data = _load() + validators = data.setdefault("validators", {}) + row = validators.setdefault(name, {"samples": []}) + samples = row.get("samples") + if not isinstance(samples, list): + samples = [] + samples.append({"d": duration, "t": timed_out}) + row["samples"] = samples[-MAX_SAMPLES_PER_VALIDATOR:] + data["schema_version"] = SCHEMA_VERSION + _atomic_write(ledger_path(), data) + except Exception: + return + + +def stats(validator: str) -> Optional[Stats]: + """Return summary stats for one validator, or None when unknown.""" + + if not enabled(): + return None + try: + row = (_load().get("validators") or {}).get(str(validator)) + if not isinstance(row, dict): + return None + raw = row.get("samples") + if not isinstance(raw, list) or not raw: + return None + durations: List[float] = [] + timeouts = 0 + for sample in raw: + if not isinstance(sample, dict): + continue + try: + value = float(sample.get("d")) + except (TypeError, ValueError): + continue + if value <= 0.0: + continue + durations.append(value) + if bool(sample.get("t")): + timeouts += 1 + if not durations: + return None + durations.sort() + return Stats( + validator=str(validator), + count=len(durations), + p50=_percentile(durations, 0.50), + p90=_percentile(durations, 0.90), + maximum=durations[-1], + timeouts=timeouts, + ) + except Exception: + return None + + +def _respects_provider_timeout() -> bool: + try: + from .generic_validators import respect_provider_timeout + + return respect_provider_timeout() + except Exception: + return False + + +def effective_ceiling(validator: Any, call_timeout: Optional[int]) -> Optional[int]: + """Mirror the timeout a call against *validator* would actually enforce.""" + + provider_timeout = getattr(validator, "_timeout_override", None) + if not isinstance(provider_timeout, int): + return call_timeout + if call_timeout is None: + return provider_timeout + if _respects_provider_timeout(): + return provider_timeout + return min(provider_timeout, call_timeout) + + +def fit_warnings(validators: Sequence[Any], call_timeout: Optional[int]) -> List[str]: + """Return advisory lines about budgets that history says will not fit. + + Two distinct problems get reported, because they have different fixes: + + * a **clamp** — the call ceiling is silently cutting a provider's own + configured budget, which is invisible in every other output; and + * a **misfit** — the enforced ceiling is below this validator's observed + p90 plus variance room, so the call is predicted to be killed. + """ + + lines: List[str] = [] + if not enabled() or call_timeout is None: + return lines + for validator in validators or []: + try: + name = str(getattr(validator, "name", "") or "") + if not name: + continue + provider_timeout = getattr(validator, "_timeout_override", None) + enforced = effective_ceiling(validator, call_timeout) + if enforced is None: + continue + clamped = ( + isinstance(provider_timeout, int) + and provider_timeout > enforced + ) + summary = stats(name) + misfit = ( + summary is not None + and summary.count >= MIN_SAMPLES_FOR_ADVICE + and enforced < summary.p90 * FIT_SAFETY_FACTOR + ) + if not clamped and not misfit: + continue + detail = [f"budget advice: {name} ceiling {enforced}s"] + if clamped: + detail.append( + f"clamps its configured {provider_timeout}s " + f"(pass --respect-provider-timeout to honour the config)" + ) + if misfit and summary is not None: + detail.append( + f"is below observed p90 {summary.p90:.0f}s over " + f"{summary.count} call(s) — predicted timeout; " + f"raise --timeout to >= {summary.recommended_ceiling()}s" + ) + lines.append(" " + "; ".join(detail)) + except Exception: + continue + return lines diff --git a/lope/runtime.py b/lope/runtime.py index 8db1d74..268a5e2 100644 --- a/lope/runtime.py +++ b/lope/runtime.py @@ -372,6 +372,21 @@ def _finish_call( record.outcome, record.reason, ) + observed_validator = record.validator + observed_duration = record.duration_seconds + observed_outcome = record.outcome + # Outside the lock: the ledger touches the filesystem, and it is + # advisory telemetry that must never delay or break a call. + try: + from .latency import record as record_latency + + record_latency( + observed_validator, + observed_duration, + outcome=observed_outcome, + ) + except Exception: + pass def records(self) -> List[CallRecord]: with self._lock: diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..bb0c622 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,18 @@ +"""Suite-wide safety net: keep tests out of the operator's real ~/.lope. + +The latency ledger is written from the runtime call-accounting path, which a +great many tests exercise with synthetic validator names ("stub", "a", "b"). +Without this fixture, running the suite silently writes that fixture data into +the user's real ledger and skews the very budget advice the ledger exists to +provide. Tests that genuinely exercise the ledger opt back in by pointing +LOPE_HOME at tmp_path and clearing this variable — see tests/test_latency.py. +""" + +from __future__ import annotations + +import pytest + + +@pytest.fixture(autouse=True) +def _isolate_latency_ledger(monkeypatch): + monkeypatch.setenv("LOPE_LATENCY", "off") diff --git a/tests/test_latency.py b/tests/test_latency.py new file mode 100644 index 0000000..83fba4c --- /dev/null +++ b/tests/test_latency.py @@ -0,0 +1,290 @@ +"""Tests for the per-validator latency ledger and pre-launch fit check. + +Covers: + - record()/stats() round-trip, percentiles, and per-validator isolation + - the sample window stays bounded and keeps the most recent observations + - timed-out calls are stored as censored samples that bias estimates upward + - LOPE_LATENCY=off disables recording, reading, and advice + - a corrupt or unwritable ledger degrades to "no advice", never an exception + - timeout precedence: stricter-wins by default, provider-wins under the flag + - fit_warnings reports a clamp, reports a predicted timeout, and stays quiet + when the budget fits or when there is not enough history to judge +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from lope import latency + + +class _FakeValidator: + """Minimal stand-in exposing the two attributes the ledger reads.""" + + def __init__(self, name, timeout_override=None): + self.name = name + self._timeout_override = timeout_override + + +def _use_ledger(tmp_path, monkeypatch): + monkeypatch.setenv("LOPE_HOME", str(tmp_path)) + monkeypatch.delenv("LOPE_LATENCY", raising=False) + monkeypatch.delenv("LOPE_RESPECT_PROVIDER_TIMEOUT", raising=False) + return tmp_path / "latency.json" + + +# --------------------------------------------------------------------------- +# Recording and summarising +# --------------------------------------------------------------------------- + +def test_record_then_stats_round_trip(tmp_path, monkeypatch): + path = _use_ledger(tmp_path, monkeypatch) + for value in (10.0, 20.0, 30.0): + latency.record("alpha", value) + + assert path.exists() + summary = latency.stats("alpha") + assert summary is not None + assert summary.count == 3 + assert summary.maximum == 30.0 + assert summary.timeouts == 0 + assert 10.0 <= summary.p50 <= 30.0 + + +def test_stats_unknown_validator_is_none(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + latency.record("alpha", 5.0) + assert latency.stats("never-run") is None + + +def test_validators_are_isolated(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + latency.record("alpha", 100.0) + latency.record("beta", 1.0) + + alpha = latency.stats("alpha") + beta = latency.stats("beta") + assert alpha is not None and beta is not None + assert alpha.maximum == 100.0 + assert beta.maximum == 1.0 + + +def test_p90_tracks_the_slow_tail(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + for _ in range(9): + latency.record("alpha", 10.0) + latency.record("alpha", 200.0) + + summary = latency.stats("alpha") + assert summary is not None + assert summary.p50 == 10.0 + assert summary.p90 >= 10.0 + assert summary.maximum == 200.0 + # The recommendation must clear the tail it just observed. + assert summary.recommended_ceiling() > summary.p90 + + +def test_sample_window_is_bounded_and_keeps_recent(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + total = latency.MAX_SAMPLES_PER_VALIDATOR + 25 + for index in range(total): + latency.record("alpha", float(index + 1)) + + summary = latency.stats("alpha") + assert summary is not None + assert summary.count == latency.MAX_SAMPLES_PER_VALIDATOR + # The oldest (smallest) observations were evicted, the newest kept. + assert summary.maximum == float(total) + + +def test_timeouts_are_recorded_as_censored_samples(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + latency.record("alpha", 240.0, outcome="provider_timeout") + latency.record("alpha", 30.0, outcome="ok") + + summary = latency.stats("alpha") + assert summary is not None + assert summary.count == 2 + assert summary.timeouts == 1 + # A kill at 240s is a lower bound, so it must drag the estimate upward. + assert summary.maximum == 240.0 + + +def test_non_positive_and_empty_observations_are_ignored(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + latency.record("alpha", 0.0) + latency.record("alpha", -5.0) + latency.record("alpha", 0.0004) # rounds to 0.0 — meaningless as a sample + latency.record("", 10.0) + assert latency.stats("alpha") is None + + +def test_only_informative_outcomes_are_recorded(tmp_path, monkeypatch): + """A crash or cancellation says nothing about how long the model needs.""" + _use_ledger(tmp_path, monkeypatch) + for bad in ("launch_error", "nonzero_exit", "parse_error", "cancelled", "circuit_open"): + latency.record("alpha", 5.0, outcome=bad) + assert latency.stats("alpha") is None + + latency.record("alpha", 5.0, outcome="ok") + latency.record("alpha", 240.0, outcome="provider_timeout") + summary = latency.stats("alpha") + assert summary is not None and summary.count == 2 + + +# --------------------------------------------------------------------------- +# Failure containment — advisory telemetry must never break a run +# --------------------------------------------------------------------------- + +def test_disabled_by_env(tmp_path, monkeypatch): + path = _use_ledger(tmp_path, monkeypatch) + monkeypatch.setenv("LOPE_LATENCY", "off") + + latency.record("alpha", 10.0) + assert not path.exists() + assert latency.stats("alpha") is None + assert latency.fit_warnings([_FakeValidator("alpha", 600)], 10) == [] + + +def test_corrupt_ledger_is_survivable(tmp_path, monkeypatch): + path = _use_ledger(tmp_path, monkeypatch) + path.write_text("{not json at all", encoding="utf-8") + + assert latency.stats("alpha") is None + latency.record("alpha", 12.0) # must not raise; rewrites the file + summary = latency.stats("alpha") + assert summary is not None and summary.count == 1 + + +def test_ledger_with_wrong_shape_is_survivable(tmp_path, monkeypatch): + path = _use_ledger(tmp_path, monkeypatch) + path.write_text(json.dumps({"validators": {"alpha": "not-a-dict"}}), encoding="utf-8") + assert latency.stats("alpha") is None + + +def test_record_never_raises_when_unwritable(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + + def _boom(*_args, **_kwargs): + raise OSError("read-only filesystem") + + monkeypatch.setattr(latency, "_atomic_write", _boom) + latency.record("alpha", 10.0) # swallowed + + +# --------------------------------------------------------------------------- +# Timeout precedence +# --------------------------------------------------------------------------- + +def test_effective_ceiling_is_stricter_of_the_two_by_default(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + slow = _FakeValidator("slow", 600) + assert latency.effective_ceiling(slow, 240) == 240 + assert latency.effective_ceiling(slow, 900) == 600 + + +def test_effective_ceiling_honours_provider_under_flag(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + monkeypatch.setenv("LOPE_RESPECT_PROVIDER_TIMEOUT", "1") + slow = _FakeValidator("slow", 600) + assert latency.effective_ceiling(slow, 240) == 600 + + +def test_effective_ceiling_without_provider_timeout(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + plain = _FakeValidator("plain", None) + assert latency.effective_ceiling(plain, 240) == 240 + assert latency.effective_ceiling(plain, None) is None + + +def test_generic_validators_effective_timeout_matches(monkeypatch): + from lope.generic_validators import _effective_timeout + + monkeypatch.delenv("LOPE_RESPECT_PROVIDER_TIMEOUT", raising=False) + assert _effective_timeout(600, 240) == 240 + assert _effective_timeout(240, 600) == 240 + assert _effective_timeout(None, 240) == 240 + assert _effective_timeout(600, None) == 600 + + monkeypatch.setenv("LOPE_RESPECT_PROVIDER_TIMEOUT", "1") + assert _effective_timeout(600, 240) == 600 + # The escape hatch still cannot invent a budget out of nothing. + assert _effective_timeout(None, 240) == 240 + + +# --------------------------------------------------------------------------- +# Fit check +# --------------------------------------------------------------------------- + +def test_fit_warning_reports_a_predicted_timeout(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + for _ in range(5): + latency.record("slow", 178.0) + + lines = latency.fit_warnings([_FakeValidator("slow", None)], 240) + assert len(lines) == 1 + assert "slow" in lines[0] + assert "predicted timeout" in lines[0] + assert "p90" in lines[0] + + +def test_fit_warning_reports_a_clamp(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + lines = latency.fit_warnings([_FakeValidator("slow", 600)], 240) + assert len(lines) == 1 + assert "clamps its configured 600s" in lines[0] + assert "--respect-provider-timeout" in lines[0] + + +def test_fit_check_is_quiet_when_the_budget_fits(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + for _ in range(5): + latency.record("quick", 10.0) + assert latency.fit_warnings([_FakeValidator("quick", None)], 600) == [] + + +def test_fit_check_is_quiet_without_enough_history(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + latency.record("new", 500.0) # one sample only + assert latency.stats("new").count < latency.MIN_SAMPLES_FOR_ADVICE + assert latency.fit_warnings([_FakeValidator("new", None)], 10) == [] + + +def test_fit_check_quiet_when_no_call_timeout(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + for _ in range(5): + latency.record("slow", 178.0) + assert latency.fit_warnings([_FakeValidator("slow", None)], None) == [] + + +def test_fit_check_uses_the_enforced_ceiling_not_the_flag(tmp_path, monkeypatch): + """A provider clamped to 240 is judged against 240, not its configured 600.""" + _use_ledger(tmp_path, monkeypatch) + for _ in range(5): + latency.record("slow", 178.0) + + lines = latency.fit_warnings([_FakeValidator("slow", 600)], 240) + assert len(lines) == 1 + assert "ceiling 240s" in lines[0] + assert "clamps its configured 600s" in lines[0] + assert "predicted timeout" in lines[0] + + # Honouring the provider's 600s budget resolves both complaints. + monkeypatch.setenv("LOPE_RESPECT_PROVIDER_TIMEOUT", "1") + assert latency.fit_warnings([_FakeValidator("slow", 600)], 240) == [] + + +def test_fit_check_tolerates_broken_validator_objects(tmp_path, monkeypatch): + _use_ledger(tmp_path, monkeypatch) + + class _Hostile: + @property + def name(self): + raise RuntimeError("boom") + + assert latency.fit_warnings([_Hostile(), _FakeValidator("", None)], 240) == []