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
2 changes: 1 addition & 1 deletion docs/specs/2026-09-03-subagent-mode-tiers-design.md
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ Against `origin/refactor/raven_v0_2_0` at `9bc06091`.

| Step | Where |
|---|---|
| A sub-agent's modes are read from its session response | `raven/acp_client/capabilities.py:428` `_read_session_modes`, re-recorded by `:451` `relearn_session_modes` |
| A sub-agent's modes are read from its session response | `raven/acp_client/capabilities.py:428` `_read_session_modes`, re-recorded by `:451` `relearn_session_menus` (named `relearn_session_modes` at that revision) |
| They become the agent's advertised menu | `raven/agent/subagent/backends/__init__.py:135` -> `AgentMeta.modes`, read back by `raven/agent/subagent/manager.py:944` `agent_modes` |
| The menu is offered to the dispatching model | `raven/agent/subagent/spawn_tool.py:195` - one enum over every agent's ids, per-agent text in the description. Withdrawn by this change |
| One resolver serves every dispatch lane | `raven/agent/subagent/manager.py:958` `resolve_mode`: `requested` > the instance's standing override > `None`. `requested` is withdrawn too |
Expand Down
24 changes: 14 additions & 10 deletions raven/acp_client/acp_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@
from raven.acp_client.capabilities import (
CapabilitySnapshot,
read_session_current_model,
relearn_session_modes,
relearn_session_menus,
steer_offered,
)
from raven.acp_client.elicitor import Elicitor
Expand Down Expand Up @@ -1257,17 +1257,21 @@ def _note_session_model(self, session_id: str | None, result: Any) -> None:
if current:
self._model_baseline[session_id] = current

def _relearn_modes(self, result: Any) -> None:
"""Take the modes a session response advertises over the cached ones.
def _relearn_menus(self, result: Any) -> None:
"""Take the modes and models a session response advertises over the cached ones.

Best-effort at every step. The menu is a description the NEXT dispatch
reads, never something this turn depends on, so neither the write nor
the table rebuild may cost the turn that happened to carry the evidence.
Or, when a newer measurement has been stored since this backend was
built, take that record instead, so the next response builds on it.

Best-effort at every step. The menus are what the NEXT dispatch and the
row's picker read, never something this turn depends on, so neither the
write nor the table rebuild may cost the turn that happened to carry the
evidence.
"""
try:
updated = relearn_session_modes(self._snapshot, result)
updated = relearn_session_menus(self._snapshot, result)
except Exception as exc: # noqa: BLE001 - a cache refresh must not sink a turn
logger.warning("acp agent {!r}: refreshing modes from the session failed: {}", self.name, exc)
logger.warning("acp agent {!r}: refreshing menus from the session failed: {}", self.name, exc)
return
if updated is None:
return
Expand Down Expand Up @@ -2198,7 +2202,7 @@ async def request_or_busy(method: str, params: dict[str, Any]) -> Any:
raise
loaded = None
if loaded is not None:
self._relearn_modes(loaded)
self._relearn_menus(loaded)
self._note_session_model(known, loaded)
await self._set_mode(client, known, mode, budget=budget)
await self._set_model(client, known, session_model, budget=budget)
Expand All @@ -2223,7 +2227,7 @@ async def request_or_busy(method: str, params: dict[str, Any]) -> Any:
raise

result = await request_or_busy("session/new", {"cwd": cwd, "mcpServers": mcp_servers})
self._relearn_modes(result)
self._relearn_menus(result)
session_id = (result or {}).get("sessionId") if isinstance(result, dict) else None
if not isinstance(session_id, str) or not session_id:
raise AcpEmptyTurnError(f"acp agent {self.name!r}: session/new returned no sessionId")
Expand Down
114 changes: 93 additions & 21 deletions raven/acp_client/capabilities.py
Original file line number Diff line number Diff line change
Expand Up @@ -396,6 +396,49 @@ def merge(current: str | None) -> tuple[str, None]:
except OSError as exc: # a snapshot is a cache, never a dependency
logger.warning("acp capability snapshot write failed (not remembered): {}", exc)

def revise(self, snapshot: CapabilitySnapshot, **changes: Any) -> tuple[CapabilitySnapshot, bool] | None:
"""Apply ``changes`` to the stored record ``snapshot`` was read from.

For a write that re-measured some fields of a record, so only those are
written, onto the row as it stands inside the lock: another holder of
the same record may have revised a different field since ``snapshot``
was read, and writing the caller's whole copy back would undo it.

Nothing is written when the stored row carries another measurement
stamp than ``snapshot``. A Test or the boot backfill that recorded in
the meantime is newer evidence about everything, the same "newest
measurement wins" rule the boot backfill applies to its own late
verdicts. With no stored row, ``snapshot`` itself is the base.

Returns the record now stored and whether this call wrote it, so a
caller that stood down holds the newer record and builds its next
revision on that stamp rather than being refused again for as long as
it lives. Flagged stale when it was measured under another launch
config than ``snapshot``, or when ``snapshot`` already was: for the
caller's launch it proves no more than that. ``None`` when the store
could not be written, or holds only a row that cannot be read back.
"""

def merge(current: str | None) -> tuple[str | None, tuple[CapabilitySnapshot, bool] | None]:
rows = self._parse(current)
stored = next((r for r in rows if r.get("agent") == snapshot.agent), None)
if stored is not None and int(stored.get("measuredAtMs") or 0) != snapshot.measured_at_ms:
newer = CapabilitySnapshot.from_row(stored)
if newer is None:
return None, None
return None, (replace(newer, stale=snapshot.stale or newer.fingerprint != snapshot.fingerprint), False)
base = (CapabilitySnapshot.from_row(stored) if stored is not None else None) or snapshot
updated = replace(base, stale=snapshot.stale, **changes)
rows = [r for r in rows if r.get("agent") != snapshot.agent]
rows.append(updated.to_row())
return self._dump(rows), (updated, True)

try:
return atomic_update(self._path, merge)
except OSError as exc: # a snapshot is a cache, never a dependency
logger.warning("acp capability snapshot write failed (not remembered): {}", exc)
return None

def load(self, configs: Sequence[Any], *, allow_stale: bool = False) -> dict[str, CapabilitySnapshot]:
"""Snapshots for these configs, keyed by agent name.

Expand Down Expand Up @@ -670,43 +713,72 @@ def _read_session_modes(result: Any) -> tuple[AcpMode, ...]:
)


def relearn_session_modes(
def relearn_session_menus(
snapshot: CapabilitySnapshot | None,
result: Any,
*,
store: SnapshotStore | None = None,
) -> CapabilitySnapshot | None:
"""Re-record an agent's modes from a live session response, if they moved.
"""Re-record an agent's modes and model menu from a live session response, if they moved.

The menu is measured once at probe time and cached, but the agent
re-advertises it on every route into a session (``_with_modes`` in the
fork's ACP methods), so the response a dispatch already holds is newer
Both menus are measured once at probe time and cached, but the agent
re-advertises them on its way into a session -- the modes on every route
(``_with_modes`` in the fork's ACP methods), the models at least on
``session/new`` -- so the response a dispatch already holds is newer
evidence than the snapshot. Without this the cache only ever moves when
someone presses Test: ``snapshot_fingerprint`` covers how the agent is
LAUNCHED, and an agent that reworded its own modes launches identically, so
nothing marks the stored text stale and the dispatching model goes on
reading a menu the agent no longer serves.

The stored fingerprint is kept rather than recomputed. Only the modes were
re-measured here, and claiming a fresh handshake would clear a staleness
flag the rest of the snapshot has not earned.
LAUNCHED, and an agent that reworded its own modes or renamed its models
launches identically, so nothing marks the stored menus stale -- the
dispatching model goes on reading modes the agent no longer serves, and the
row's picker goes on offering model ids it no longer takes.

Read with the probe's own parsers, from the response this route was sending
for anyway: nothing here opens a session or starts a process.

The stored fingerprint and measurement stamp are kept rather than renewed.
Only the menus were re-measured here, and claiming a fresh handshake would
clear a staleness flag the rest of the snapshot has not earned. Only the
menus this response reported are written, and none when a newer
measurement has replaced the record ``snapshot`` was read from
(``SnapshotStore.revise``). That newer record is returned instead, unrevised:
this response may predate it, so its menus are not taken over it, but the
caller's next response is then judged against the stamp actually stored.

``None`` when there was nothing to learn, which includes a response carrying
no modes at all. That is read as "this route did not report them", never as
"the agent dropped them": erasing the menu takes the ``mode`` argument out
of the spawn schema entirely, and a stale description is the smaller harm.
neither menu. Each menu is judged on its own, and an absent or empty one is
read as "this route did not report it", never as "the agent dropped it":
erasing the modes takes the ``mode`` argument out of the spawn schema, and
erasing the models takes the picker off the row, both worse than a stale
entry. A model menu the agent flagged as unread is not taken either, for
the reason ``_read_session_menu_unread`` gives.
"""
if snapshot is None:
return None
changes: dict[str, Any] = {}
modes = _read_session_modes(result)
if not modes or modes == snapshot.available_modes:
if modes and modes != snapshot.available_modes:
changes["available_modes"] = modes
choices = () if _read_session_menu_unread(result) else _read_session_model_choices(result)
if choices and choices != snapshot.model_choices:
changes["model_choices"] = choices
changes["model_menu_measured"] = True
if not changes:
return None
updated = replace(snapshot, available_modes=modes)
(store or SnapshotStore()).record(updated)
revised = (store or SnapshotStore()).revise(snapshot, **changes)
if revised is None:
return None
updated, written = revised
if not written:
logger.info(
"acp agent {!r}: session menus not re-recorded over a newer measurement; building on that one next",
snapshot.agent,
)
return updated
logger.info(
"acp agent {!r}: re-recorded {} mode(s) from a session response",
"acp agent {!r}: re-recorded {} mode(s) and {} model(s) from a session response",
snapshot.agent,
len(modes),
len(updated.available_modes),
len(updated.model_choices),
)
return updated

Expand Down Expand Up @@ -1079,7 +1151,7 @@ def done(
"looks_like_auth",
"npx_fetch_failure",
"npx_fetch_lead",
"relearn_session_modes",
"relearn_session_menus",
"snapshot_fingerprint",
"steer_offered",
"verify_agent",
Expand Down
6 changes: 3 additions & 3 deletions raven/agent/subagent/probe.py
Original file line number Diff line number Diff line change
Expand Up @@ -1349,7 +1349,7 @@ def _measured_at(snapshot: Any) -> int | None:
"""When the record's verdict was measured; ``None`` for no record.

Every measurement stamps its own time (``_test_record`` carries it over), and
a modes relearn keeps the stamp it found, so a changed stamp means a newer
a menu relearn keeps the stamp it found, so a changed stamp means a newer
verdict and nothing else does.
"""
return None if snapshot is None else getattr(snapshot, "measured_at_ms", 0)
Expand Down Expand Up @@ -1545,9 +1545,9 @@ async def _verify_missing_snapshots(manager: Any, rows: list[Any], *, configured
# Read again with nothing awaited between here and the write: the
# handshake held the row for up to the ready timeout. A measurement
# recorded meanwhile -- a manual Test, a connect -- is newer evidence
# than this one and stands. A dispatch relearning the agent's modes
# than this one and stands. A dispatch relearning the agent's menus
# measures nothing and keeps the stamp, so it does not stop this
# write, and a failure below is kept over the re-read record, modes
# write, and a failure below is kept over the re-read record, menus
# included.
current = acp_snapshot_for(cfg)
if _measured_at(current) != _measured_at(snapshot):
Expand Down
62 changes: 62 additions & 0 deletions tests/test_rpc_subagents.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
subagents_test,
subagents_test_cancel,
)
from raven.utils.commands import command_quote


@pytest.fixture
Expand Down Expand Up @@ -3198,6 +3199,67 @@ async def test_list_marks_a_config_row_whose_handshake_named_raven_as_ravens_own
assert rows["Other-Quiet"]["own"] is False and rows["Other-Quiet"]["model_source"] == "agent"


async def test_list_shows_the_model_menu_a_dispatch_relearned(
config_path: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The row's picker follows the agent without anyone pressing Test.

The listing reads the stored snapshot, and the snapshot never expires by
age, so before a dispatch could refresh it the card went on offering the
ids the agent held at its last Test. One dispatch now carries the agent's
current menu to the card, under the same ``agent`` rule as before.
A spaced stub path exercises launch quoting on every test host.
"""
from raven.acp_client.capabilities import AcpModelChoice, CapabilitySnapshot, SnapshotStore, snapshot_fingerprint
from raven.acp_client.pool import close_pool
from raven.agent.subagent.backends import build_third_party_backend
from raven.config.schema import SubagentsConfig

stub = tmp_path / "acp stub server.py"
stub.write_bytes(Path(__file__).with_name("acp_stub_server.py").read_bytes())
raw = json.loads(config_path.read_text())
raw["subagents"]["agents"].append(
{
"name": "Stubby",
"kind": "acp",
"command": f"{command_quote(sys.executable)} {command_quote(str(stub))}",
"env": {"ACP_STUB_MODE": "ok"},
"description": "d",
"enabled": True,
}
)
config_path.write_text(json.dumps(raw), encoding="utf-8")
store_path = tmp_path / "caps.json"
monkeypatch.setattr("raven.acp_client.capabilities.default_snapshot_path", lambda: store_path)
cfg = next(c for c in SubagentsConfig(agents=raw["subagents"]["agents"]).agents if c.name == "Stubby")
SnapshotStore(path=store_path).record(
CapabilitySnapshot(
agent="Stubby",
fingerprint=snapshot_fingerprint(cfg),
status="ready",
detail="",
measured_at_ms=1,
agent_name="stub-agent",
model_choices=(AcpModelChoice(value="stub:retired", name="retired", group="Stub"),),
)
)
before = {r["name"]: r for r in (await subagents_list({"probe": False}))["rows"]}["Stubby"]
assert before["model_choices"] == [{"value": "stub:retired", "name": "retired", "group": "Stub"}]

try:
reply = await build_third_party_backend(cfg).run("ping", task_id="t1", workspace=tmp_path, executor=None)
finally:
await close_pool()
assert reply == "pong"

after = {r["name"]: r for r in (await subagents_list({"probe": False}))["rows"]}["Stubby"]
assert after["model_choices"] == [
{"value": "stub:model-a", "name": "model-a", "group": "Stub"},
{"value": "stub:model-b", "name": "model-b", "group": "Stub"},
]
assert after["model_source"] == before["model_source"] == "agent"


async def test_test_can_target_a_discovered_product(
config_path: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
Expand Down
Loading
Loading