Coverage for src/keel/runcontrols.py: 100%
135 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-02 20:26 +0000
1"""Deterministic run budgets, step caps, and oscillation detection."""
3from __future__ import annotations
5from dataclasses import dataclass
6from typing import Any
8from . import artifacts, model
10SCHEMA_VERSION = "keel.run-controls.v1"
12#: Schema of :func:`fix_attribution` — who implemented, and who fixed which round (#1016).
13FIX_ATTRIBUTION_SCHEMA_VERSION = "keel.fix-attribution.v1"
15#: Slot (or step id) an s4 implementation event carries.
16IMPLEMENT_SLOT = "implement"
17IMPLEMENT_STEP = "s4"
19#: Slot (or step id) an s9 fix-round event carries.
20FIX_SLOT = "fixloop"
21FIX_STEP = "s9"
23DEFAULT_RUN_BUDGET = 250
24DEFAULT_STEP_CAP = 1
25DEFAULT_FIXLOOP_CAP = 3
26DEFAULT_REVIEWER_CAP = 3
27DEFAULT_TEST_CAP = 2
28DEFAULT_IDENTICAL_THRESHOLD = 3
29DEFAULT_ALTERNATION_WINDOW = 4
32@dataclass(frozen=True)
33class HaltReason:
34 """One deterministic hard-halt reason."""
36 control: str
37 reason: str
38 scope: str
39 observed: int | str
40 limit: int | str
41 action: str = "halt"
43 def as_dict(self) -> dict[str, Any]:
44 return {
45 "control": self.control,
46 "reason": self.reason,
47 "scope": self.scope,
48 "observed": self.observed,
49 "limit": self.limit,
50 "action": self.action,
51 "rendered": artifacts.render_run_control_halt(
52 control=self.control,
53 reason=self.reason,
54 scope=self.scope,
55 observed=self.observed,
56 limit=self.limit,
57 action=self.action,
58 ),
59 }
62def contract_as_dict() -> dict[str, Any]:
63 """Return the pure-core run-control contract for agentic commands."""
64 return {
65 "schema_version": SCHEMA_VERSION,
66 "consumer_neutral": True,
67 "deterministic": True,
68 "stdlib_only": True,
69 "fail_closed": True,
70 "wall_clock_timeouts": False,
71 "hard_halts": [
72 "run-budget-exceeded",
73 "step-cap-exceeded",
74 "oscillation-detected",
75 ],
76 "fail_soft_preserved": (
77 "soft failures are recorded by their owning step; only budget, cap, or "
78 "oscillation breaches produce a hard halt"
79 ),
80 "budget": {
81 "unit": "work_unit",
82 "default_max_work_units": DEFAULT_RUN_BUDGET,
83 "breach_policy": "halt-fail-closed",
84 },
85 "step_caps": {
86 "applies_to_slots": list(model.SLOTS),
87 "default_max_iterations": DEFAULT_STEP_CAP,
88 "overrides": _default_step_caps(),
89 "breach_policy": "halt-fail-closed",
90 },
91 "oscillation": {
92 "identical_action_threshold": DEFAULT_IDENTICAL_THRESHOLD,
93 "alternating_diff_window": DEFAULT_ALTERNATION_WINDOW,
94 "breach_policy": "halt-fail-closed",
95 },
96 "renderer": {
97 "halt_reason": "keel.artifacts.render_run_control_halt",
98 "marker": artifacts.RUN_CONTROL_HALT_MARKER,
99 },
100 "attribution": {
101 "schema_version": FIX_ATTRIBUTION_SCHEMA_VERSION,
102 "reader": "keel.runcontrols.fix_attribution",
103 "event_fields": ["provider", "attribution", "stage", "round"],
104 "implement_event": {"slot": IMPLEMENT_SLOT, "step_id": IMPLEMENT_STEP},
105 "fix_event": {"slot": FIX_SLOT, "step_id": FIX_STEP},
106 },
107 }
110def evaluate_run_controls(
111 events: list[dict[str, Any]] | tuple[dict[str, Any], ...],
112 *,
113 max_work_units: int = DEFAULT_RUN_BUDGET,
114 default_step_cap: int = DEFAULT_STEP_CAP,
115 step_caps: dict[str, int] | None = None,
116 identical_action_threshold: int = DEFAULT_IDENTICAL_THRESHOLD,
117 alternating_diff_window: int = DEFAULT_ALTERNATION_WINDOW,
118) -> dict[str, Any]:
119 """Evaluate run controls and return pass or a structured hard halt."""
120 normalized = [_normalize_event(event) for event in events if isinstance(event, dict)]
121 reason = (
122 _budget_halt(normalized, max_work_units)
123 or _step_cap_halt(normalized, default_step_cap, step_caps or _default_step_caps())
124 or _oscillation_halt(
125 normalized,
126 identical_action_threshold=identical_action_threshold,
127 alternating_diff_window=alternating_diff_window,
128 )
129 )
130 return {
131 "schema_version": SCHEMA_VERSION,
132 "status": "halt" if reason else "pass",
133 "hard_halt": reason is not None,
134 "fail_closed": reason is not None,
135 "reason": reason.as_dict() if reason else None,
136 "summary": {
137 "event_count": len(normalized),
138 "work_units": sum(event["work_units"] for event in normalized),
139 },
140 }
143def _is_fix_event(event: dict[str, Any]) -> bool:
144 return event["slot"] == FIX_SLOT or event["step_id"] == FIX_STEP
147def _is_implement_event(event: dict[str, Any]) -> bool:
148 return event["slot"] == IMPLEMENT_SLOT or event["step_id"] == IMPLEMENT_STEP
151def _actor(event: dict[str, Any]) -> str:
152 """What to call whoever ran this event: the attribution label, else the provider."""
153 return event["attribution"] or event["provider"]
156def fix_attribution(
157 events: list[dict[str, Any]] | tuple[dict[str, Any], ...],
158) -> dict[str, Any]:
159 """Who implemented, and who fixed which round (#1016).
161 s9 could escalate a failed round to a different provider — implementer, then the gate
162 seat, then the host — and nothing wrote down which of them actually produced the
163 change that merged. The closure comment then attributed the whole run to the
164 implementer, which is exactly the sentence a reader trusts and exactly the one that
165 was wrong.
167 So the run-events file carries ``provider`` and ``attribution`` on the events it
168 already appends, and this reads them back: the implementation actor from the s4
169 event, one record per s9 round, and the deterministic ``sentence`` the s11 closure
170 comment embeds. A round without an explicit ``round`` is numbered by its position
171 among the fix events, so an adapter that only appends events still gets stable
172 numbering.
173 """
174 normalized = [_normalize_event(event) for event in events if isinstance(event, dict)]
175 implementer = None
176 for event in normalized:
177 if _is_implement_event(event) and _actor(event):
178 implementer = {
179 "provider": event["provider"] or _actor(event),
180 "attribution": event["attribution"] or None,
181 "actor": _actor(event),
182 }
183 break
184 rounds: list[dict[str, Any]] = []
185 for event in normalized:
186 if not _is_fix_event(event) or not _actor(event):
187 continue
188 rounds.append(
189 {
190 "round": event["round"] if event["round"] is not None else len(rounds) + 1,
191 "provider": event["provider"] or _actor(event),
192 "attribution": event["attribution"] or None,
193 "stage": event["stage"] or None,
194 "actor": _actor(event),
195 }
196 )
197 return {
198 "schema_version": FIX_ATTRIBUTION_SCHEMA_VERSION,
199 "implementer": implementer,
200 "rounds": rounds,
201 "sentence": attribution_sentence(implementer, rounds),
202 }
205def attribution_sentence(
206 implementer: dict[str, Any] | None,
207 rounds: list[dict[str, Any]] | tuple[dict[str, Any], ...],
208) -> str:
209 """``implemented by agy, fixed by opus in round 2`` — deterministic, never a guess."""
210 parts = []
211 if implementer is not None:
212 parts.append(f"implemented by {implementer['actor']}")
213 clauses = [f"{item['actor']} in round {item['round']}" for item in rounds]
214 if clauses:
215 if len(clauses) == 1:
216 fixed = clauses[0]
217 else:
218 fixed = f"{', '.join(clauses[:-1])} and {clauses[-1]}"
219 parts.append(f"fixed by {fixed}")
220 if not parts:
221 return "no implementer or fixer recorded"
222 return ", ".join(parts)
225def _default_step_caps() -> dict[str, int]:
226 return {
227 "fixloop": DEFAULT_FIXLOOP_CAP,
228 "reviewers": DEFAULT_REVIEWER_CAP,
229 "tester": DEFAULT_TEST_CAP,
230 "test": DEFAULT_TEST_CAP,
231 }
234def _normalize_event(event: dict[str, Any]) -> dict[str, Any]:
235 return {
236 "step_id": _string(event.get("step_id")),
237 "slot": _string(event.get("slot")),
238 "action": _string(event.get("action")),
239 "output_fingerprint": _string(event.get("output_fingerprint")),
240 "diff_fingerprint": _string(event.get("diff_fingerprint")),
241 "work_units": _positive_int(event.get("work_units"), default=1),
242 "soft_failure": bool(event.get("soft_failure")),
243 "provider": _string(event.get("provider")),
244 "attribution": _attribution_label(event.get("attribution")),
245 "stage": _string(event.get("stage")),
246 "round": event.get("round") if _is_round(event.get("round")) else None,
247 }
250def _is_round(value: Any) -> bool:
251 return isinstance(value, int) and not isinstance(value, bool) and value > 0
254def _attribution_label(value: Any) -> str:
255 """The human label for who ran a step.
257 ``keel delegate run`` returns ``attribution`` as the object core computed
258 (``agent_label`` / ``model_label`` / ``system``); an operator appending an event by
259 hand passes a bare string. Both are accepted so the adapter can pass the delegate's
260 own document through without reshaping it — the whole point of computing attribution
261 in core was that the label written down and the label that ran cannot drift.
262 """
263 if isinstance(value, dict):
264 return _string(value.get("agent_label")) or _string(value.get("model_label"))
265 return _string(value)
268def _budget_halt(events: list[dict[str, Any]], max_work_units: int) -> HaltReason | None:
269 limit = _positive_int(max_work_units, default=DEFAULT_RUN_BUDGET)
270 observed = sum(event["work_units"] for event in events)
271 if observed > limit:
272 return HaltReason(
273 control="run-budget",
274 reason="run-budget-exceeded",
275 scope="run",
276 observed=observed,
277 limit=limit,
278 )
279 return None
282def _step_cap_halt(
283 events: list[dict[str, Any]],
284 default_step_cap: int,
285 step_caps: dict[str, int],
286) -> HaltReason | None:
287 default_limit = _positive_int(default_step_cap, default=DEFAULT_STEP_CAP)
288 counts: dict[str, int] = {}
289 for event in events:
290 scope = event["slot"] or event["step_id"]
291 if not scope:
292 continue
293 counts[scope] = counts.get(scope, 0) + 1
294 limit = _step_limit(scope, step_caps, default_limit)
295 if counts[scope] > limit:
296 return HaltReason(
297 control="step-cap",
298 reason="step-cap-exceeded",
299 scope=scope,
300 observed=counts[scope],
301 limit=limit,
302 )
303 return None
306def _oscillation_halt(
307 events: list[dict[str, Any]],
308 *,
309 identical_action_threshold: int,
310 alternating_diff_window: int,
311) -> HaltReason | None:
312 repeated = _repeated_identical_action(
313 events,
314 threshold=_positive_int(
315 identical_action_threshold,
316 default=DEFAULT_IDENTICAL_THRESHOLD,
317 ),
318 )
319 if repeated:
320 return repeated
321 return _alternating_diff(
322 events,
323 window=_positive_int(
324 alternating_diff_window,
325 default=DEFAULT_ALTERNATION_WINDOW,
326 ),
327 )
330def _repeated_identical_action(
331 events: list[dict[str, Any]],
332 *,
333 threshold: int,
334) -> HaltReason | None:
335 if threshold <= 1:
336 threshold = DEFAULT_IDENTICAL_THRESHOLD
337 streak = 0
338 previous: tuple[str, str, str, str] | None = None
339 for event in events:
340 if not event["action"] and not event["output_fingerprint"]:
341 previous = None
342 streak = 0
343 continue
344 current = (
345 event["step_id"],
346 event["slot"],
347 event["action"],
348 event["output_fingerprint"],
349 )
350 streak = streak + 1 if current == previous else 1
351 previous = current
352 if streak >= threshold:
353 return HaltReason(
354 control="oscillation",
355 reason="repeated-identical-action",
356 scope=event["slot"] or event["step_id"] or "run",
357 observed=streak,
358 limit=threshold,
359 )
360 return None
363def _alternating_diff(events: list[dict[str, Any]], *, window: int) -> HaltReason | None:
364 if window < 4 or window % 2:
365 window = DEFAULT_ALTERNATION_WINDOW
366 diffs = [event["diff_fingerprint"] for event in events if event["diff_fingerprint"]]
367 if len(diffs) < window:
368 return None
369 tail = diffs[-window:]
370 left = tail[: window // 2]
371 right = tail[window // 2 :]
372 if left == right and len(set(left)) > 1:
373 return HaltReason(
374 control="oscillation",
375 reason="alternating-diff-fingerprint",
376 scope="diff",
377 observed=",".join(tail),
378 limit=f"no {window}-round alternation",
379 )
380 return None
383def _string(value: Any) -> str:
384 return value.strip() if isinstance(value, str) and value.strip() else ""
387def _positive_int(value: Any, *, default: int) -> int:
388 return value if isinstance(value, int) and value > 0 else default
391def _step_limit(scope: str, step_caps: dict[str, int], default_limit: int) -> int:
392 default_caps = _default_step_caps()
393 scope_default = default_caps.get(scope, default_limit)
394 configured = step_caps.get(scope)
395 return configured if isinstance(configured, int) and configured > 0 else scope_default