Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,15 @@ All notable changes to Raven are documented here.

### Fixed

- An on-call campaign bills only its own runs. Campaigns of one task usually
keep their rounds in one `remote_dir`, and the spend measure walked that
directory's `jobs/`, so a later campaign was charged for every earlier one's
runs the moment it started: a second-round campaign with a 130-minute budget
read 125.6 spent on its first look, 34.6 of it its own, and stopped with 142
real minutes unspent. The backend now counts the jobs in the campaign's own
ledger (plus what it submitted itself), and `ops_declare` refuses a
`remote_dir` a live sibling campaign is still writing rounds into.

- The web file viewer opens a sub-agent's report again. `/file` anchored the
state-directory fence on the session's working directory whenever the page
named a session, so the fence exempted `~/.raven/tmp/<channel>` and refused
Expand Down
16 changes: 16 additions & 0 deletions agents/raven-oncall/plugins/oncall-flow/oncall_flow/backends.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,22 @@ def backend_from_meta(meta: dict[str, Any]) -> JobBackend:
return factory(meta)


def billing_only(backend: JobBackend, ledger: Any) -> JobBackend:
"""``backend``, told to bill only the jobs in this campaign's ``ledger``.

Campaigns share a ``remote_dir`` more often than not, and a backend that
measures spend by walking that directory then sees every sibling's jobs as
its own. The ledger is the one record of which jobs are this campaign's, so
it is handed over here, at the seam every tool builds its backend through. A
backend that does not measure spend by directory has nothing to restrict and
comes back as it was; so does one built by a test double.
"""
restrict = getattr(backend, "restrict_spend_to", None)
if ledger is not None and callable(restrict):
restrict(ledger)
return backend


def prepare_from_meta(meta: dict[str, Any], *, app_dir: str) -> None:
"""Round-zero setup for backends that need it; a no-op for those that don't."""
if backend_name(meta) != DEFAULT_BACKEND:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,13 @@ async def spent_minutes(self) -> float:
self._unmeasured.clear()
self._cores.clear()
for key, procs, alive, started, clock, last_touch, _end in rows:
# Filtered here rather than after the sum, because _cores and the
# timeline are read by the caller and a sibling's width in either
# one is the same cross-campaign bill in a different place. This
# override does not go through the base class's loop, so it has to
# ask the same question itself.
if not self._counts_toward_spend(key):
continue
np_ = int(_num(procs) or 0) or self._default_cores
self._cores[key] = np_
st, ct, lt = _num(started), _num(clock), _num(last_touch)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,8 @@
import base64
import json
import shlex
from collections.abc import Callable
from typing import Any
from collections.abc import Callable, Iterable
from typing import TYPE_CHECKING, Any

from oncall_flow.backend import (
JobBackend,
Expand All @@ -43,6 +43,9 @@
JobStatus,
)
from oncall_flow.budget import ADDITIVE, SHARED, Budget, accumulate

if TYPE_CHECKING:
from oncall_flow.ledger import Ledger
from oncall_flow.budget import from_meta as budget_from_meta

# How long a job script may take to write its result after the pid it advertised
Expand Down Expand Up @@ -201,6 +204,7 @@ def __init__(
checkpoint as confidently as the best.
"""
self._declared_width: dict[str, float] = {}
self._own_jobs: Callable[[], set[str]] | None = None
self._run = run
self._remote_dir = remote_dir.rstrip("/")
self._command = command
Expand Down Expand Up @@ -282,6 +286,49 @@ async def _arun(self, cmd: str) -> tuple[int, str]:

# ---- budget ----

def restrict_spend_to(self, own: Ledger | Iterable[str]) -> None:
Comment thread
gloryfromca marked this conversation as resolved.
"""Count only this campaign's own jobs when measuring spend.

Campaigns share a ``remote_dir`` more often than not -- an agent that runs
one task in rounds points every round at the same ``runs/`` -- and the spend
scan below walks ``{remote_dir}/jobs/*``, which then holds every round's
jobs. Measured 2026-09-11: a second-round campaign with a 130-minute budget
read 125.6 spent on its first look, of which 34.6 was its own; the rest was
the two rounds before it in the same directory. It stopped with 142 real
minutes unspent. The job directory names carry no campaign identity (the
suffix is a digest of the config), so the only record of ownership is the
campaign's ledger, which is what is read here -- live, on every measure, so
a job submitted after this call is counted too. Jobs this executor itself
submitted are always counted, ledger or not.
"""
# Imported here, not at module scope: the module-level import is under
# TYPE_CHECKING, and narrowing the union by isinstance needs the class at
# run time. ledger.py imports nothing from this module, so the local
# import costs a dict lookup and risks no cycle. isinstance rather than
# hasattr because the type checker cannot narrow a union on an attribute
# probe, and every caller -- ops, ops_observe, the tests -- hands over a
# real Ledger.
from oncall_flow.ledger import Ledger as _Ledger

if isinstance(own, _Ledger):
ledger = own

def _keys() -> set[str]:
return {r.idem_key for r in ledger.all()}

else:
fixed = set(own)

def _keys() -> set[str]:
return set(fixed)

self._own_jobs = _keys

def _counts_toward_spend(self, key: str) -> bool:
if self._own_jobs is None:
return True
return key in self._own_jobs() or key in self._declared_width

async def spent_minutes(self) -> float:
"""Minutes this campaign has held the device, whatever ended the runs.

Expand Down Expand Up @@ -343,6 +390,8 @@ async def spent_minutes(self) -> float:
exclusive_spans: list[tuple[float, float, float]] = []
self._unmeasured.clear()
for key, result_min, alive, started, elapsed, prog_mtime, width, exclusive in rows:
if not self._counts_toward_spend(key):
continue
rm, st, el, pm = _num(result_min), _num(started), _num(elapsed), _num(prog_mtime)
if rm is not None:
minutes = rm
Expand Down
17 changes: 9 additions & 8 deletions agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -1070,7 +1070,7 @@ async def execute(self, ledger: str = "", metric: str = "", **kwargs: Any) -> st
import json as _json

from oncall_flow.backend import JobStatus
from oncall_flow.backends import backend_from_meta
from oncall_flow.backends import backend_from_meta, billing_only
from oncall_flow.ledger import Ledger

# No ledger given: fall back to the one campaign under the ops home, so the
Expand Down Expand Up @@ -1155,7 +1155,7 @@ async def execute(self, ledger: str = "", metric: str = "", **kwargs: Any) -> st
try:
from oncall_flow.instrument import log_event

backend = backend_from_meta(meta)
backend = billing_only(backend_from_meta(meta), led)
try:
budget_spent = await backend.spent_minutes()
budget_remaining = await backend.remaining_minutes()
Expand Down Expand Up @@ -1818,7 +1818,7 @@ async def execute( # noqa: C901 (cc 69: pre-existing, above the ceiling)
**kwargs: Any,
) -> str:
from oncall_flow.backend import JobBackendError, JobResult, JobSpec, JobStatus
from oncall_flow.backends import backend_from_meta, prepare_from_meta
from oncall_flow.backends import backend_from_meta, billing_only, prepare_from_meta
from oncall_flow.ledger import Ledger
from oncall_flow.proposer import config_key

Expand Down Expand Up @@ -2159,7 +2159,7 @@ def _split_hostport(h: str, p: int) -> tuple[str, int]:
if round == 0:
prepare_from_meta(meta, app_dir=app_dir)

backend = backend_from_meta(meta)
backend = billing_only(backend_from_meta(meta), Ledger(ledger))

# The apparatus, checked against what this campaign was set up with. The
# first submit is where the baseline is taken -- it is the last moment the
Expand Down Expand Up @@ -2782,9 +2782,10 @@ async def execute(self, **kwargs: Any) -> str:
# stay cheap enough to ask casually.
if not concluded.exists():
try:
from oncall_flow.backends import backend_from_meta
from oncall_flow.backends import backend_from_meta, billing_only
from oncall_flow.ledger import Ledger

spent = await backend_from_meta(meta).spent_minutes()
spent = await billing_only(backend_from_meta(meta), Ledger(d / "ledger.json")).spent_minutes()
total = (meta.get("budget") or {}).get("total")
unit = (meta.get("budget") or {}).get("unit") or ""
bits.append(f"spent {spent:.2f}" + (f" of {total} {unit}" if total else ""))
Expand Down Expand Up @@ -2906,7 +2907,7 @@ async def execute(
import json as _json

from oncall_flow.backend import JobResult, JobStatus
from oncall_flow.backends import backend_from_meta
from oncall_flow.backends import backend_from_meta, billing_only
from oncall_flow.instrument import log_event
from oncall_flow.ledger import Ledger

Expand All @@ -2919,7 +2920,7 @@ async def execute(
if refusal:
return refusal
meta = _json.loads(meta_path.read_text(encoding="utf-8"))
backend = backend_from_meta(meta)
backend = billing_only(backend_from_meta(meta), Ledger(ledger_path))
led = Ledger(ledger_path)
killed, skipped, notes = [], [], []
for key in trials:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ def _dict_cmd(meta: dict, args: str) -> str:


def _campaign(campaign: str, ledger: str | None):
from oncall_flow.backends import backend_from_meta
from oncall_flow.backends import backend_from_meta, billing_only
from oncall_flow.ledger import Ledger

cdir = _resolve_campaign_dir(campaign, ledger)
Expand All @@ -82,7 +82,8 @@ def _campaign(campaign: str, ledger: str | None):
if not meta_path.exists():
return f"No campaign meta under {cdir} (need meta.json)."
meta = _json.loads(meta_path.read_text(encoding="utf-8"))
return backend_from_meta(meta), Ledger(ledger_path), cdir, meta
ledger = Ledger(ledger_path)
return billing_only(backend_from_meta(meta), ledger), ledger, cdir, meta


def _case_root(backend: Any, meta: dict, led: Any, trial: str | None) -> tuple[str, str] | str:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,64 @@ def _round_dir_inside_case(remote_dir: str, staged_case: str) -> tuple[str, str]
return None


def _remote_dir_taken(remote_dir: str, cdir: Path, ops_home: Path) -> str | None:
"""The sibling campaign already keeping its rounds in ``remote_dir``, or None.

Two campaigns writing rounds into one directory bill each other: the spend
measure walks ``{remote_dir}/jobs/*``, and a directory name carries no campaign
identity. The backend now reads only the campaign's own ledger, but a shared
directory still mixes the two campaigns' job trees, so it is refused at the
declaration, where the fix is one word. Re-declaring this same campaign is
not a collision.

A concluded sibling counts too, as long as it left jobs behind. A job
directory is named from a digest of its config and carries no campaign, so
two campaigns that run the same trial name the same directory: the newcomer's
ledger then records that key as its own, the ownership filter reads the
predecessor's minutes as this campaign's, and submit finds the old
``result.json`` and calls the trial already done instead of running it. A
concluded sibling that never submitted anything left nothing to adopt and
does not stand in the way.
"""
if not remote_dir:
return None
from oncall_flow.tools.ops import _campaign_meta

mine = Path(remote_dir).expanduser()
try:
siblings = sorted(d for d in ops_home.iterdir() if d.is_dir())
except OSError:
return None
for sib in siblings:
if sib == cdir:
continue
theirs = str(_campaign_meta(sib).get("remote_dir") or "")
if not theirs or Path(theirs).expanduser() != mine:
continue
if (sib / "concluded.json").exists() and not _left_jobs_behind(sib):
continue
return sib.name
return None


def _left_jobs_behind(sib: Path) -> bool:
"""Whether a campaign ever recorded a job, read straight off its ledger file.

Read as plain JSON rather than through ``Ledger``: this is a sibling's file
and the question is only whether it holds a record, so a ledger this campaign
cannot parse must not raise here. An unreadable one is treated as occupied,
which is the safe direction -- the alternative hands the newcomer a directory
whose contents nobody could account for.
"""
path = sib / "ledger.json"
if not path.exists():
return False
try:
return bool(json.loads(path.read_text(encoding="utf-8")).get("records") or {})
except Exception: # noqa: BLE001 -- see the docstring: unreadable means occupied
return True


# Commands whose whole job is to put a copy of something somewhere else. Only the
# first word of a step is checked against this, so a "cp" inside a filename or a
# message is not one of them.
Expand Down Expand Up @@ -854,6 +912,16 @@ async def execute(
f"would stop being something you only read. Put rounds beside the case or "
f"anywhere else writable. Nothing was written."
)
taken = _remote_dir_taken(remote_dir, cdir, _ops_home())
if taken:
return (
f"REFUSED: remote_dir {remote_dir!r} is where campaign {taken!r} keeps its rounds, "
f"and that campaign has not concluded.\n"
f"Two campaigns in one rounds directory cannot tell their jobs apart, so each "
f"would bill the other's runs against its own budget. Give this campaign a "
f"rounds directory of its own (for example a sibling named after it). "
f"Nothing was written."
)
# Only what this shape has. An absent metric is what tells every later
# reader there is no ranking to do -- writing an empty one would leave
# them looking for a number the campaign never had.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,15 +27,16 @@


def _campaign_backend(campaign: str, ledger: str | None) -> tuple[Any, Any, Path] | str:
from oncall_flow.backends import backend_from_meta
from oncall_flow.backends import backend_from_meta, billing_only
from oncall_flow.ledger import Ledger

cdir = _resolve_campaign_dir(campaign, ledger)
ledger_path, meta_path = cdir / "ledger.json", cdir / "meta.json"
if not ledger_path.exists() or not meta_path.exists():
return f"No campaign state under {cdir} (need ledger.json + meta.json)."
meta = _json.loads(meta_path.read_text(encoding="utf-8"))
return backend_from_meta(meta), Ledger(ledger_path), cdir
ledger = Ledger(ledger_path)
return billing_only(backend_from_meta(meta), ledger), ledger, cdir


class OpsOutputsTool(Tool):
Expand Down
5 changes: 4 additions & 1 deletion docs/specs/2026-09-03-resource-admission-design.md
Original file line number Diff line number Diff line change
Expand Up @@ -214,7 +214,10 @@ frees its cards while still running.
- Spend: `spent_minutes` reads width and whether ids were assigned from the
staged `.raven-resources` file (written at submit next to `config.json`, so a
restart loses nothing), builds spans `width` wide, and bills assigned-device
spans additively and the rest under the declared overlap (decision 5). The
spans additively and the rest under the declared overlap (decision 5). Only
jobs in the campaign's own ledger (and those this executor submitted) are
placed on the timeline: campaigns of one task share a `remote_dir`, and a
directory walk billed each campaign for its siblings' runs. The
openfoam backend keeps its own measured
width and, when the measured `processor*` count differs from the declared
`cores_per_job`, records `{"declared": 8, "measured": 16}` on the record and
Expand Down
Loading
Loading