feat: add time_budget policy

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

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

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

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",
 ]

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()

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)

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)