diff --git a/docs/policies/time_budget.md b/docs/policies/time_budget.md new file mode 100644 index 0000000..caeb024 --- /dev/null +++ b/docs/policies/time_budget.md @@ -0,0 +1,49 @@ +# time_budget -- opt-in wall-clock ceiling + +Code: `src/tokenops/control/policies/time_budget.py` +Tests: `tests/test_time_budget.py` + +--- + +## TL;DR + +A cheap circuit breaker on **elapsed wall-clock time** per run. Trips at **observe** +when the run's age reaches the ceiling, and the action it takes is **HALT**. It does +not depend on pricing -- useful when a workflow's cost is latency rather than tokens. + +## Detect (formula) + + elapsed(run) >= max_seconds + +`elapsed` is `step.ts - start` where `start` is the timestamp of the first step +observed for that run (stored on the detector, O(1) per step). + +## Action it takes to govern -- HALT + +Identical mechanism to `cost_budget` and `step_cap`: `Signal(TRIP)` -> `Action(HALT)` +-> flag set before raise -> sticky kill switch. The difference is purely the trigger +(wall-clock age, not spend or step count). + +## Why opt-in, not default + +Elapsed time is task-dependent -- a legitimate research run may take 30 seconds or +30 minutes. Budget is the universal backstop; `time_budget` is for workflows whose +runtime you actually know and want to bound. + +## Edge cases + +* First step: elapsed is `0.0` (the ledger records the step before `observe` runs, so `step.ts - start == 0.0`). +* Trips at **exactly** `max_seconds` elapsed (>=). +* Counts wall-clock time across all node types -- llm, tool, and delegate alike. + +## I/O & success criteria (test contract) + +| Input | Expect | +|---|---| +| `elapsed >= max_seconds` | `Signal(TRIP)` -> `Action(HALT)` | +| `elapsed < max_seconds` | `None` (ALLOW) | +| e2e: two tool crossings, `max_seconds=0.5` | halts on the second | + +## Status + +Implemented and tested (unit + e2e). diff --git a/docs/product/policies-index.md b/docs/product/policies-index.md index 14f71be..2a6724a 100644 --- a/docs/product/policies-index.md +++ b/docs/product/policies-index.md @@ -15,6 +15,7 @@ Canonical per-policy docs live under [`docs/policies/`](../policies/). Product o | `context_compaction` | [context_compaction.md](../policies/context_compaction.md) | | `output_runaway` | [output_runaway.md](../policies/output_runaway.md) | | `trajectory_hint` | [trajectory_hint.md](../policies/trajectory_hint.md) | +| `time_budget` | [time_budget.md](../policies/time_budget.md) | Actuators (HALT · MUTATE · INJECT · REJECT/QUEUE): see [`docs/control-plane-status.md`](../control-plane-status.md) and [`docs/governance-policy.md`](../governance-policy.md). diff --git a/src/tokenops/control/config.py b/src/tokenops/control/config.py index 1dbb529..c912df8 100644 --- a/src/tokenops/control/config.py +++ b/src/tokenops/control/config.py @@ -45,6 +45,7 @@ pre_call_worst_case, progress_guard, step_cap, + time_budget, tool_fix, tool_output_cap, ) @@ -77,6 +78,7 @@ def budget(self, ref: str) -> Budget: c.budget(p["budget"]), c.price, default_max_output=p.get("default_max_output", 1024) ), "step_cap": lambda p, c: step_cap.build(p["max_steps"]), + "time_budget": lambda p, c: time_budget.build(p["max_seconds"]), "concurrency_cap": lambda p, c: concurrency_cap.build( p["max_concurrent"], dimension=p.get("dimension", "run"), diff --git a/src/tokenops/control/policies/__init__.py b/src/tokenops/control/policies/__init__.py index 25f2307..e531113 100644 --- a/src/tokenops/control/policies/__init__.py +++ b/src/tokenops/control/policies/__init__.py @@ -18,6 +18,7 @@ pre_call_worst_case, progress_guard, step_cap, + time_budget, tool_fix, tool_output_cap, ) @@ -31,6 +32,7 @@ "pre_call_worst_case", "progress_guard", "step_cap", + "time_budget", "tool_fix", "tool_output_cap", ] diff --git a/src/tokenops/control/policies/time_budget.py b/src/tokenops/control/policies/time_budget.py new file mode 100644 index 0000000..7223092 --- /dev/null +++ b/src/tokenops/control/policies/time_budget.py @@ -0,0 +1,57 @@ +"""time_budget -- optional / opt-in. A wall-clock ceiling per run. + +LLD row: + Detect: elapsed(run) >= max_seconds (observe moment; per run) + Fix: HALT. Good for workflows whose cost is latency rather than tokens. + +Not a default -- elapsed time is task-dependent. Opt in when a workflow has a +known time bound and you want a cheap circuit breaker independent of pricing. +""" + +from __future__ import annotations + +from tokenops.control.core import ( + Action, + ActionKind, + Attribution, + BoundaryStep, + Detector, + LedgerView, + Policy, + Severity, + Signal, +) + + +class TimeBudgetDetector(Detector): + """TRIP when the run's wall-clock age reaches the ceiling.""" + + name = "time_budget" + + def __init__(self, max_seconds: float) -> None: + self.max_seconds = max_seconds + self._started: dict[str, float] = {} + + def observe(self, attr: Attribution, step: BoundaryStep, view: LedgerView) -> Signal | None: + start = self._started.setdefault(attr.run_id, step.ts) + elapsed = step.ts - start + if elapsed >= self.max_seconds: + return Signal( + detector=self.name, + severity=Severity.TRIP, + run_id=attr.run_id, + reason=f"time budget exceeded: {elapsed:.1f}s >= {self.max_seconds}s", + evidence={"elapsed_s": elapsed, "max_seconds": self.max_seconds}, + ) + return None + + +class TimeBudgetPolicy(Policy): + name = "time_budget" + + def decide(self, signal: Signal, view: LedgerView) -> Action: + return Action(kind=ActionKind.HALT, run_id=signal.run_id, reason=signal.reason) + + +def build(max_seconds: float) -> tuple[Detector, Policy]: + return TimeBudgetDetector(max_seconds), TimeBudgetPolicy() diff --git a/tests/test_policies_wrap_integration.py b/tests/test_policies_wrap_integration.py index 9554635..02d1c80 100644 --- a/tests/test_policies_wrap_integration.py +++ b/tests/test_policies_wrap_integration.py @@ -37,6 +37,7 @@ pre_call_worst_case, progress_guard, step_cap, + time_budget, tool_fix, tool_output_cap, ) @@ -205,6 +206,26 @@ def run(): ) # second call dispatches then observe HALTs; or halt on observe of step 2 +def test_time_budget_halts_run(): + controls = ApplyControls() + gov = Governor(Ledger(budgets=[], price=toy_price), controls) + gov.register(*time_budget.build(max_seconds=0.0)) + attr = _attr("r-tb") + gov.ledger.open_run("r-tb") + dispatch, calls = _dispatch() + governed = _governed(gov, attr, dispatch, run_id="r-tb") + + def run(): + # The ledger records the step before observe fires, so elapsed=0.0 >= 0.0 + # trips on the first call. + with pytest.raises(Halt): + governed("openai", "gpt-4o-mini", [{"role": "user", "content": "1"}]) + + _with_scope(gov, attr, "r-tb", run) + assert gov.ledger.is_halted("r-tb") + assert len(calls) == 1 + + def test_it_concurrency_cap_rejects_when_inflight_saturated(): controls = ApplyControls() gov = Governor(Ledger(price=toy_price), controls) diff --git a/tests/test_time_budget.py b/tests/test_time_budget.py new file mode 100644 index 0000000..5249c21 --- /dev/null +++ b/tests/test_time_budget.py @@ -0,0 +1,73 @@ +"""time_budget -- wall-clock ceiling per run. HALT when elapsed >= max_seconds.""" + +from __future__ import annotations + +import pytest + +from conftest import FakeView, make_attr, make_step, toy_price +from tokenops.control import ActionKind, Budget, Governor, Halt, Ledger, Observation +from tokenops.control.policies import time_budget + + +def test_detector_trips_at_ceiling(): + # First observe sets start=0.0; second observe with ts=61.0 → elapsed 61s >= 60s + det, _ = time_budget.build(max_seconds=60.0) + attr = make_attr() + view = FakeView() + det.observe(attr, make_step(step=0), view) # sets start time + sig = det.observe(attr, make_step(step=61), view) + assert sig is not None + assert sig.severity.value == "trip" + + +def test_detector_allows_below_ceiling(): + # First observe sets start=0.0; second observe with ts=59.0 → elapsed 59s < 60s + det, _ = time_budget.build(max_seconds=60.0) + attr = make_attr() + view = FakeView() + det.observe(attr, make_step(step=0), view) # sets start time + assert det.observe(attr, make_step(step=59), view) is None + + +def test_detector_first_step_allows(): + # First step for a run: elapsed is 0.0, always below any positive ceiling + det, _ = time_budget.build(max_seconds=60.0) + assert det.observe(make_attr(), make_step(step=1), FakeView()) is None + + +def test_policy_halts(): + det, pol = time_budget.build(max_seconds=60.0) + attr = make_attr() + view = FakeView() + det.observe(attr, make_step(step=0), view) # sets start time + sig = det.observe(attr, make_step(step=61), view) + assert pol.decide(sig, FakeView()).kind is ActionKind.HALT + + +def test_e2e_halts_when_time_exceeded(): + # The ledger records the step before observe fires, so window includes the current + # step. With max_seconds=0.5: tool(0.0) → elapsed=0.0 < 0.5 (allow); + # tool(1.0) → elapsed=1.0 - 0.0 = 1.0 >= 0.5 (HALT). + ledger = Ledger( + budgets=[Budget(budget_id="c", limit_micros=10**9, dimension="run")], price=toy_price + ) + gov = Governor(ledger) + gov.register(*time_budget.build(max_seconds=0.5)) + attr = make_attr() + ledger.open_run("run-1") + + def tool(ts): + gov.observe( + Observation( + attr=attr, + node_type="tool", + boundary_id="search", + ts=ts, + signature=f"s{ts}", + result_hash=f"r{ts}", + ) + ) + + tool(0.0) + with pytest.raises(Halt): + tool(1.0)