Author overrides can give an author longer sessions

CHANGELOG.md

@@ -6,6 +6,15 @@ Versions follow [SemVer](https://semver.org).
 
 ## [Unreleased]
 
+- Author overrides accept operator-only `session_minutes` (10–240) and
+  `session_max_turns` (10–300). Limits bind with the author selection and survive
+  settings changes across wakes and review replies. Contracts can still lower
+  budgets; job walltime follows session duration and the operator job cap.
+  Judge budgets are unchanged.
+- Upgrading: the new override fields are optional; existing settings and legacy
+  run records need no migration. Finish runs using extended limits before
+  rolling back to a version without bound author limits.
+
 - Bad author overrides hold only affected fresh claims during ticks, with one log
   per entry per tick; malformed settings hold named targets, or all fresh claims
   when unreadable. Other tick services and bound runs continue; `outerloop start`

docs/install.md

@@ -626,6 +626,21 @@ appear twice:
 OUTERLOOP_AUTHOR_OVERRIDES='{"owner/repo":[{"backend":"claude","model":"served-model[endpoint=onprem]","slots":["agent-04"]},{"backend":"codex","model":"served-model[endpoint=onprem]","slots":["agent-03"]}]}'
 ```
 
+Either form also accepts optional integer `session_minutes` (10–240) and
+`session_max_turns` (10–300) on each entry, for authors that need longer sessions.
+For example, add `"session_minutes":180,"session_max_turns":250` to an entry.
+Values outside these ranges fail startup validation. Only operator settings can
+raise these limits; a contract's explicit session budget can still lower them.
+The session uses the smaller of the contract value and the override, subject to
+the existing floors. An overridden session duration gets a job budget of that
+duration plus 20 minutes of overhead; an explicit contract job budget can lower
+it. Panel work keeps its additional allowance. `OUTERLOOP_MAX_JOB_MINUTES` still
+caps the job and shortens the session when needed to leave overhead; at the cap,
+the panel allowance is what gets cut. A run keeps the limits it was claimed
+with, and its wake and review-reply jobs are sized from those limits too.
+Codex has no turn cap, so `session_max_turns` applies to Claude Code and Hermes
+authors only; Codex sessions are bounded by `session_minutes`.
+
 This is deployment configuration, not a contract setting. The setting is
 strictly validated by `outerloop start` and `outerloop init`; during ticks, an unusable
 entry holds fresh claims only for its slots, without falling back to the fleet author.
@@ -636,9 +651,11 @@ the selected backend's author credential. Normal author/judge credential
 separation still applies to the effective override credential.
 
 Queued climbs bind the selection when submitted (before an intake claim is
-launched); direct climbs bind at startup. Backend, model and key path are saved
-in the run record. Changing overrides cannot switch an existing run, even at a
-resume or wake. Panel inheritance and CI reviewers still use the fleet author,
+launched); direct climbs bind at startup. Backend, model, key path and resolved
+override limits are saved in the run record. Bound limits also apply to author-sleep wakes, resumed legs
+and author replies to reviews; judge budgets are unchanged. Entries without the
+new fields and older records retain their existing limits. Changing overrides
+cannot switch an existing run, even at a resume or wake. Panel inheritance and CI reviewers still use the fleet author,
 exactly as for a run without an override. Per-run board details show the author
 backend/model and mark overrides. Remove the setting (or use `{}`) to stop
 selecting overrides for new work.

src/outerloop/attempt.py

@@ -83,6 +83,7 @@
     observe_target,
     queue_pending,
 )
+from outerloop.limits import bound_limits, capped_limits, clamp_bound_limits, effective_limits
 from outerloop.markers import has_marker, marker
 from outerloop.measure import DispatchedMeasurer, DispatchSettings
 from outerloop.orchestrator import (
@@ -4391,6 +4392,7 @@ def live_attempt(
     author_model: str = "",
     author_key_file: str = "",
     author_overridden: bool = False,
+    author_limits: dict[str, int] | None = None,
     task_hypothesis: str = "",
     spec: RoleSpec | None = None,
     panel_lenses: tuple[PanelLens, ...] = (),
@@ -4421,6 +4423,7 @@ def live_attempt(
         author_backend=author_backend,
         author_model=author_model,
         author_overridden=author_overridden,
+        author_limits=author_limits,
         author_key_file=author_key_file,
         run_job_id=_os.environ.get("SLURM_JOB_ID", ""),
         stage={"hypothesis": redact(task_hypothesis, secrets)} if task_hypothesis else {},
@@ -4462,6 +4465,23 @@ def live_attempt(
         _exclude_merge_artifacts(workspace)
         contract_text = contract_text_in_tree(workspace)
         contract = load_contract(contract_text, config.target)
+        if (stored := bound_limits(record.author_limits)) is not None:
+            # Direct climbs learn the trusted contract only after cloning. Queued
+            # climbs already carry its clamp; neither path rereads operator settings.
+            bound = clamp_bound_limits(stored, contract.budgets)
+            record = dc_replace(record, author_limits=asdict(bound))
+            save_record(run_root, record, now)
+            spec = author_spec(
+                max_turns=bound.session_max_turns, walltime_s=bound.session_minutes * 60
+            )
+            from outerloop.harness import ClaudeCodeHarness, CodexHarness, HermesHarness
+
+            if isinstance(harness, (ClaudeCodeHarness, HermesHarness)):
+                harness = dc_replace(
+                    harness, max_turns=spec.budget.max_turns, timeout_s=spec.budget.walltime_s
+                )
+            elif isinstance(harness, CodexHarness):
+                harness = dc_replace(harness, timeout_s=spec.budget.walltime_s)
         # Load the brief budget from the contract and run state: callers do
         # not supply it (the dataclass default rendered "0.0 GPU-hours" and
         # honest agents refused to launch). Same weekly counting rule as the
@@ -5162,6 +5182,7 @@ def _run_id(value: str) -> str:
     )
     parser.add_argument("--author-bound", action="store_true", help=argparse.SUPPRESS)
     parser.add_argument("--author-overridden", action="store_true", help=argparse.SUPPRESS)
+    parser.add_argument("--author-limits", type=json.loads, default=None, help=argparse.SUPPRESS)
     args = parser.parse_args()
     from outerloop.author_overrides import select_override
     from outerloop.gpu_lanes import gpu_lanes_from_env
@@ -5173,6 +5194,23 @@ def _run_id(value: str) -> str:
             if selected:
                 args.author_backend, args.model = selected.backend, selected.resolved_model()
                 args.author_overridden = True
+                if selected.session_minutes is not None or selected.session_max_turns is not None:
+                    from outerloop.tick import _max_job_minutes_from_env
+
+                    limits = effective_limits(
+                        session_minutes=selected.session_minutes,
+                        session_max_turns=selected.session_max_turns,
+                    )
+                    if args.job_minutes:  # 0 keeps the self-deadline off
+                        args.job_minutes = min(args.job_minutes, _max_job_minutes_from_env())
+                        limits = capped_limits(limits, args.job_minutes)
+                    args.author_limits = asdict(limits)
+        if args.author_limits is not None and not args.resume:
+            passed = bound_limits(args.author_limits)
+            if passed is None:
+                raise ValueError("--author-limits must carry every session limit")
+            args.max_turns = passed.session_max_turns
+            args.session_minutes = passed.session_minutes
     except ValueError as exc:
         parser.error(str(exc))
     logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
@@ -5229,6 +5267,12 @@ def _run_id(value: str) -> str:
             # a wake must never crash on an unreadable/odd record — fall back to
             # the claude author (resume_author), same fail-safe as the sweep
             _wake_record = None
+        wake_limits = bound_limits(getattr(_wake_record, "author_limits", None))
+        if wake_limits is not None:
+            if args.job_minutes:
+                wake_limits = capped_limits(wake_limits, args.job_minutes)
+            args.max_turns = wake_limits.session_max_turns
+            args.session_minutes = wake_limits.session_minutes
         try:
             explicit_model = args.model  # what the operator typed, before any env fill-in
             if not getattr(_wake_record, "author_model", "") and not args.model:
@@ -5480,6 +5524,7 @@ def _run_id(value: str) -> str:
                 author_backend=args.author_backend,
                 author_model=args.model,
                 author_overridden=args.author_overridden,
+                author_limits=args.author_limits,
                 author_key_file=args.key_file,
                 task_hypothesis=(
                     __import__("base64").b64decode(args.hypothesis_b64).decode()

src/outerloop/author_overrides.py

@@ -10,6 +10,12 @@
 from functools import lru_cache
 from types import MappingProxyType
 
+from outerloop.limits import (
+    OVERRIDE_SESSION_MINUTES_CEILING,
+    OVERRIDE_SESSION_TURNS_CEILING,
+    SESSION_MINUTES_FLOOR,
+)
+
 SETTING = "OUTERLOOP_AUTHOR_OVERRIDES"
 
 
@@ -19,6 +25,9 @@ class AuthorOverride:
     model: str
     slots: tuple[str, ...] | None = None
 
+    session_minutes: int | None = None
+    session_max_turns: int | None = None
+
     def resolved_model(self) -> str:
         """Bind profile model defaults without inheriting the fleet endpoint."""
         from outerloop.endpoints import resolve_endpoint
@@ -58,8 +67,17 @@ def parse_overrides(raw: str) -> Mapping[str, tuple[AuthorOverride, ...]]:
 
 
 def _parse_one(target: str, value: object, listed: bool) -> AuthorOverride:
-    if not isinstance(value, dict) or set(value) - {"backend", "model", "slots"}:
-        raise ValueError(f"{target}: expected backend, model and optional slots")
+    if not isinstance(value, dict) or set(value) - {
+        "backend",
+        "model",
+        "slots",
+        "session_minutes",
+        "session_max_turns",
+    }:
+        raise ValueError(
+            f"{target}: expected backend, model and optional slots, "
+            "session_minutes, session_max_turns"
+        )
     backend, model, slots = value.get("backend"), value.get("model"), value.get("slots")
     if backend not in ("claude", "codex", "hermes"):
         raise ValueError(f"{target}: backend must be claude, codex or hermes")
@@ -88,7 +106,19 @@ def _parse_one(target: str, value: object, listed: bool) -> AuthorOverride:
         raise ValueError(f"{target}: slots must be distinct agent identities (agent-01, ...)")
     if listed and slots is None:
         raise ValueError(f"{target}: each override in a list must name its slots")
-    return AuthorOverride(backend, model, None if slots is None else tuple(slots))
+    for name, floor, ceiling in (
+        ("session_minutes", SESSION_MINUTES_FLOOR, OVERRIDE_SESSION_MINUTES_CEILING),
+        ("session_max_turns", 10, OVERRIDE_SESSION_TURNS_CEILING),
+    ):
+        if name in value and (type(value[name]) is not int or not floor <= value[name] <= ceiling):
+            raise ValueError(f"{target}: {name} must be an integer between {floor} and {ceiling}")
+    return AuthorOverride(
+        backend,
+        model,
+        None if slots is None else tuple(slots),
+        value.get("session_minutes"),
+        value.get("session_max_turns"),
+    )
 
 
 def overrides(

src/outerloop/limits.py

@@ -5,13 +5,13 @@
 — shorter sessions, tighter job walltimes — but must never be able to
 raise it: every contract value is clamped into [floor, ceiling], and the
 ceilings are code on the orchestrator side, not configuration a target
-can reach. Absent values fall back to the defaults the pilot has run with
-all along.
+can reach. Validated operator author overrides may raise the session ceilings.
+Absent values fall back to the defaults the pilot has run with all along.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass, replace
 from typing import Any
 
 # (default, floor, ceiling) per knob. Floors keep a hostile-or-typo'd
@@ -45,6 +45,10 @@
 # the tick's cap warning uses it as the no-runway threshold too.
 ATTEMPT_OVERHEAD_MINUTES = 20
 
+# Operator author overrides may raise sessions up to these, never further.
+OVERRIDE_SESSION_MINUTES_CEILING = 240
+OVERRIDE_SESSION_TURNS_CEILING = 300
+
 
 @dataclass(frozen=True)
 class EffectiveLimits:
@@ -61,20 +65,87 @@ def _clamp(name: str, value: int | None) -> int:
     return max(floor, min(int(value), ceiling))
 
 
-def effective_limits(budgets: Any = None) -> EffectiveLimits:
+def effective_limits(
+    budgets: Any = None, *, session_minutes: int | None = None, session_max_turns: int | None = None
+) -> EffectiveLimits:
     """Resolve a contract's optional budget knobs into enforceable limits.
 
     `budgets` is the contract's Budgets model (or None for pure defaults);
-    unknown/absent attributes read as None. The session is finally shrunk
+    unknown/absent attributes read as None. Optional session ceilings come only
+    from validated operator overrides (or their persisted binding), never the
+    contract. The session is finally shrunk
     to fit inside the climb job with room for the orchestrator's overhead —
     a session that outlives its job ends as a kill, not a report.
     """
     values = {
         name: _clamp(name, getattr(budgets, name, None) if budgets is not None else None)
         for name in _BOUNDS
     }
+    for name, override in (
+        ("session_minutes", session_minutes),
+        ("session_max_turns", session_max_turns),
+    ):
+        if override is not None:
+            requested = getattr(budgets, name, None)
+            values[name] = max(
+                _BOUNDS[name][1], min(requested, override) if requested is not None else override
+            )
+    if session_minutes is not None:
+        wanted = values["session_minutes"] + ATTEMPT_OVERHEAD_MINUTES
+        requested_job = getattr(budgets, "attempt_job_minutes", None)
+        values["attempt_job_minutes"] = (
+            min(wanted, max(ATTEMPT_JOB_MINUTES_FLOOR, requested_job))
+            if requested_job is not None
+            else wanted
+        )
     max_session = values["attempt_job_minutes"] - ATTEMPT_OVERHEAD_MINUTES
     if values["session_minutes"] > max_session:
         floor = _BOUNDS["session_minutes"][1]
         values["session_minutes"] = max(floor, max_session)
     return EffectiveLimits(**values)
+
+
+def capped_limits(limits: EffectiveLimits, job_minutes: int) -> EffectiveLimits:
+    """Shrink the author session to leave overhead inside the actual job."""
+    return replace(
+        limits,
+        session_minutes=max(
+            SESSION_MINUTES_FLOOR,
+            min(limits.session_minutes, job_minutes - ATTEMPT_OVERHEAD_MINUTES),
+        ),
+    )
+
+
+def clamp_bound_limits(limits: EffectiveLimits, budgets: Any) -> EffectiveLimits:
+    """Apply a newly loaded trusted contract without raising a bound budget."""
+    values = asdict(limits)
+    for name in values:
+        requested = getattr(budgets, name, None)
+        if requested is not None:
+            values[name] = min(values[name], max(_BOUNDS[name][1], requested))
+    if values["session_minutes"] < limits.session_minutes:
+        values["attempt_job_minutes"] = min(
+            values["attempt_job_minutes"], values["session_minutes"] + ATTEMPT_OVERHEAD_MINUTES
+        )
+    return capped_limits(EffectiveLimits(**values), values["attempt_job_minutes"])
+
+
+def bound_limits(value: Any) -> EffectiveLimits | None:
+    """A run's persisted limits, or None (today's defaults) when absent or unreadable."""
+    if not isinstance(value, dict):
+        return None
+    values: dict[str, int] = {}
+    for name in _BOUNDS:
+        stored = value.get(name)
+        if not isinstance(stored, int) or isinstance(stored, bool):
+            return None  # missing, bool, float or string: never coerced
+        values[name] = stored
+    ceilings = {
+        **{name: ceiling for name, (_, _, ceiling) in _BOUNDS.items()},
+        "session_minutes": OVERRIDE_SESSION_MINUTES_CEILING,
+        "session_max_turns": OVERRIDE_SESSION_TURNS_CEILING,
+        "attempt_job_minutes": OVERRIDE_SESSION_MINUTES_CEILING + ATTEMPT_OVERHEAD_MINUTES,
+    }
+    for name, (_, floor, _) in _BOUNDS.items():
+        values[name] = max(floor, min(values[name], ceilings[name]))
+    return EffectiveLimits(**values)

src/outerloop/runstate.py

@@ -137,6 +137,7 @@ class RunRecord:
     # the harness strips it only when constructing the backend session.
     author_backend: str = ""
     author_model: str = ""
+    author_limits: dict[str, int] | None = None  # bound operator session budgets
     author_overridden: bool = False  # judges inherit the fleet author for this run
     # The resolved author key FILE PATH (not the key) this run used, so a wake or
     # follow-up reproduces the exact key — an explicit --key-file survives, and an
@@ -265,6 +266,8 @@ def _save_record(root: Path, record: RunRecord, now: float) -> None:
     # same tmp file before the atomic replace
     tmp = directory / f".{RECORD_NAME}.{os.getpid()}.tmp"
     payload = asdict(stamped)
+    if stamped.author_limits is None:
+        payload.pop("author_limits")
     if not stamped.author_overridden:
         payload.pop("author_overridden")  # No-setting records retain their exact wire shape.
     path = directory / RECORD_NAME