From 677118db7f86985e44bcb4e59ae133e14dee8dba Mon Sep 17 00:00:00 2001 From: Yashika Jain Date: Fri, 11 Sep 2026 09:32:22 -0400 Subject: [PATCH 1/3] feat: add time_budget policy --- docs/policies/time_budget.md | 50 +++++++++++++ src/tokenops/control/config.py | 2 + src/tokenops/control/policies/__init__.py | 2 + src/tokenops/control/policies/time_budget.py | 58 +++++++++++++++ tests/test_policies_wrap_integration.py | 20 ++++++ tests/test_time_budget.py | 75 ++++++++++++++++++++ 6 files changed, 207 insertions(+) create mode 100644 docs/policies/time_budget.md create mode 100644 src/tokenops/control/policies/time_budget.py create mode 100644 tests/test_time_budget.py diff --git a/docs/policies/time_budget.md b/docs/policies/time_budget.md new file mode 100644 index 0000000..59671af --- /dev/null +++ b/docs/policies/time_budget.md @@ -0,0 +1,50 @@ +# 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 - window[0].ts` (the current step's timestamp minus the first +recorded step's timestamp). If the window is empty the detector returns None. + +## 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 + +* Returns None on the first step (empty window) -- no elapsed time to measure yet. +* 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) | +| empty window | `None` (ALLOW) | +| e2e: two tool crossings, `max_seconds=0.0` | halts on the second | + +## Status + +Implemented and tested (unit + e2e). \ No newline at end of file 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..4337a94 --- /dev/null +++ b/src/tokenops/control/policies/time_budget.py @@ -0,0 +1,58 @@ +"""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 + + def observe(self, attr: Attribution, step: BoundaryStep, view: LedgerView) -> Signal | None: + window = view.window(attr.run_id) + if not window: + return None + elapsed = step.ts - window[0].ts + 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..b3feb37 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, ) @@ -204,6 +205,25 @@ def run(): len(calls) == 2 ) # 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() diff --git a/tests/test_time_budget.py b/tests/test_time_budget.py new file mode 100644 index 0000000..b71d81c --- /dev/null +++ b/tests/test_time_budget.py @@ -0,0 +1,75 @@ +"""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 _view_with_window(steps): + v = FakeView() + v._window = steps + return v + + +def test_detector_trips_at_ceiling(): + # make_step sets ts=float(step); use step numbers to control timestamps. + # window[0].ts = float(0) = 0.0; current step ts = float(61) = 61.0 → elapsed 61s >= 60s + det, _ = time_budget.build(max_seconds=60.0) + window = [make_step(step=0)] + view = _view_with_window(window) + sig = det.observe(make_attr(), make_step(step=61), view) + assert sig is not None + assert sig.severity.value == "trip" + + +def test_detector_allows_below_ceiling(): + # window[0].ts = 0.0; current step ts = 59.0 → elapsed 59s < 60s + det, _ = time_budget.build(max_seconds=60.0) + window = [make_step(step=0)] + view = _view_with_window(window) + assert det.observe(make_attr(), make_step(step=59), view) is None + + +def test_detector_empty_window_allows(): + det, _ = time_budget.build(max_seconds=60.0) + assert det.observe(make_attr(), make_step(step=1), _view_with_window([])) is None + + +def test_policy_halts(): + det, pol = time_budget.build(max_seconds=60.0) + window = [make_step(step=0)] + sig = det.observe(make_attr(), make_step(step=61), _view_with_window(window)) + 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) From 2c5dcd52d22a6e905bcdfa922285507b3feeef37 Mon Sep 17 00:00:00 2001 From: Yashika Jain Date: Sat, 12 Sep 2026 22:49:43 -0400 Subject: [PATCH 2/3] style: fix formatting --- tests/test_policies_wrap_integration.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_policies_wrap_integration.py b/tests/test_policies_wrap_integration.py index b3feb37..02d1c80 100644 --- a/tests/test_policies_wrap_integration.py +++ b/tests/test_policies_wrap_integration.py @@ -205,6 +205,7 @@ def run(): len(calls) == 2 ) # 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) From 1dd5a18c112bd518114b507cf3738dd618b9a776 Mon Sep 17 00:00:00 2001 From: Yashika Jain Date: Sat, 12 Sep 2026 23:03:33 -0400 Subject: [PATCH 3/3] Refactor: Store start time per run, fix docs, add policies-index row --- docs/policies/time_budget.md | 11 +++--- docs/product/policies-index.md | 1 + src/tokenops/control/policies/time_budget.py | 7 ++-- tests/test_time_budget.py | 36 +++++++++----------- 4 files changed, 26 insertions(+), 29 deletions(-) diff --git a/docs/policies/time_budget.md b/docs/policies/time_budget.md index 59671af..caeb024 100644 --- a/docs/policies/time_budget.md +++ b/docs/policies/time_budget.md @@ -15,8 +15,8 @@ not depend on pricing -- useful when a workflow's cost is latency rather than to elapsed(run) >= max_seconds -`elapsed` is `step.ts - window[0].ts` (the current step's timestamp minus the first -recorded step's timestamp). If the window is empty the detector returns None. +`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 @@ -32,7 +32,7 @@ runtime you actually know and want to bound. ## Edge cases -* Returns None on the first step (empty window) -- no elapsed time to measure yet. +* 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. @@ -42,9 +42,8 @@ runtime you actually know and want to bound. |---|---| | `elapsed >= max_seconds` | `Signal(TRIP)` -> `Action(HALT)` | | `elapsed < max_seconds` | `None` (ALLOW) | -| empty window | `None` (ALLOW) | -| e2e: two tool crossings, `max_seconds=0.0` | halts on the second | +| e2e: two tool crossings, `max_seconds=0.5` | halts on the second | ## Status -Implemented and tested (unit + e2e). \ No newline at end of file +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/policies/time_budget.py b/src/tokenops/control/policies/time_budget.py index 4337a94..7223092 100644 --- a/src/tokenops/control/policies/time_budget.py +++ b/src/tokenops/control/policies/time_budget.py @@ -30,12 +30,11 @@ class TimeBudgetDetector(Detector): 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: - window = view.window(attr.run_id) - if not window: - return None - elapsed = step.ts - window[0].ts + start = self._started.setdefault(attr.run_id, step.ts) + elapsed = step.ts - start if elapsed >= self.max_seconds: return Signal( detector=self.name, diff --git a/tests/test_time_budget.py b/tests/test_time_budget.py index b71d81c..5249c21 100644 --- a/tests/test_time_budget.py +++ b/tests/test_time_budget.py @@ -9,40 +9,38 @@ from tokenops.control.policies import time_budget -def _view_with_window(steps): - v = FakeView() - v._window = steps - return v - - def test_detector_trips_at_ceiling(): - # make_step sets ts=float(step); use step numbers to control timestamps. - # window[0].ts = float(0) = 0.0; current step ts = float(61) = 61.0 → elapsed 61s >= 60s + # First observe sets start=0.0; second observe with ts=61.0 → elapsed 61s >= 60s det, _ = time_budget.build(max_seconds=60.0) - window = [make_step(step=0)] - view = _view_with_window(window) - sig = det.observe(make_attr(), make_step(step=61), view) + 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(): - # window[0].ts = 0.0; current step ts = 59.0 → elapsed 59s < 60s + # First observe sets start=0.0; second observe with ts=59.0 → elapsed 59s < 60s det, _ = time_budget.build(max_seconds=60.0) - window = [make_step(step=0)] - view = _view_with_window(window) - assert det.observe(make_attr(), make_step(step=59), view) is None + 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_empty_window_allows(): +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), _view_with_window([])) is None + 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) - window = [make_step(step=0)] - sig = det.observe(make_attr(), make_step(step=61), _view_with_window(window)) + 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