From 84d46ec68e98cad29d6c59df9d33df5a063dfd80 Mon Sep 17 00:00:00 2001 From: "xiaotian.luo" Date: Sun, 20 Sep 2026 16:18:30 +0800 Subject: [PATCH] feat(raven-oncall): let the on-call instance add the machine the owner describes An on-call instance that finds its connection registry empty asks the owner for the machine. Until 2026-09-06 the host consumed that answer: a check before every spawn read the registry and, when it held nothing, ran `raven ops connection add --non-interactive` from what the owner had typed in conversation. That check went with the pre-dispatch registry gate, and the plugin's own reply kept only half of the exchange -- it asked the owner to run the command themselves in a terminal. Measured 2026-09-20: the owner answered the question in the TUI and nothing wrote the row. The write now belongs to the instance that needs the machine. A new tool, `ops_connection_add`, takes the owner's answer and files the row through functions in the plugin's own registry module (`probe`, `write`, `write_ssh_alias`, mirroring the CLI as the reader already does): the machine is reached before anything is written, and its cores, memory and devices are read off it rather than typed. The ask is cut to what only the owner knows. Required: what they call the machine, whether it is this computer or one reached over ssh, and an address for the latter. Port, username and key path are taken when given; left out, `ssh -G` resolves them the way the owner's own terminal would and the keys it names are probed in turn, keeping the one that connects. A username ssh cannot resolve is refused, not guessed. Installed software, work directories and the budget unit are optional; concurrency is kept only on a row that reports no cores or devices, where admission has nothing else to go on. A key this tool picks has to be shown to be the one that opened the session, and that takes reading no ssh config at all. `-i` is a preference rather than a restriction, and `IdentitiesOnly=yes` is defined as excluding what an agent or a provider adds, not what the config names. Measured with OpenSSH 9.9p2 against a config carrying two `IdentityFile` lines: `-i cand` with IdentitiesOnly and no agent still resolves to three identities. A candidate is therefore probed with `-F /dev/null` alongside both options, through an opt-in parameter on the ssh runner; credited otherwise, the registry would hold a path that merely looked verified and would stop working the day that agent or config changed. The cost is that an alias or a jump host is not in play while candidates are tried, so a machine reached that way is refused with a line asking for the key path, and a path the owner named is probed the way a job will reach the machine, agent included. The empty-registry reply, the tool prompt and the heat2d demo walk-through say the new shape. Both on-call parity contracts are updated on the product side: the visible-tool set names the new tool, and the prompt guard keeps byte equality with the frozen fork through an explicit list of this side's additions. The frozen A-side fixture is a record and is left untouched. Co-authored-by: Claude (claude-opus-5) --- .../oncall-flow/oncall_flow/connections.py | 233 ++++++++++++++++-- .../oncall-flow/oncall_flow/docker_backend.py | 25 ++ .../oncall-flow/oncall_flow/tools/__init__.py | 8 + .../oncall_flow/tools/ops_connection_add.py | 203 +++++++++++++++ .../oncall-flow/prompts/TOOLS_ONCALL.md | 6 +- .../plugins/oncall-flow/raven-plugin.toml | 4 + demos/oncall_heat2d/README.md | 10 +- .../test_agents_oncall_flow_connection_add.py | 194 +++++++++++++++ tests/test_agents_oncall_flow_watcher.py | 1 + tests/test_agents_oncall_launcher.py | 44 +++- 10 files changed, 693 insertions(+), 35 deletions(-) create mode 100644 agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/ops_connection_add.py create mode 100644 tests/test_agents_oncall_flow_connection_add.py diff --git a/agents/raven-oncall/plugins/oncall-flow/oncall_flow/connections.py b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/connections.py index 7f1caa325..0aa41fca0 100644 --- a/agents/raven-oncall/plugins/oncall-flow/oncall_flow/connections.py +++ b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/connections.py @@ -26,12 +26,17 @@ one means a keychain or a master password, and neither is worth building before something asks for it. -Read-only here: the file is written by hand or by the host's -``raven ops connection add``. This module is the plugin's own reader over the -same store the trunk module reads (RAVEN_CONNECTIONS, else the owner's -``~/.raven/connections.json``): the trunk module is deliberately not a -contracts paper, so the plugin cannot import it at runtime -- behaviour is -aligned to the trunk version and any drift is a parity-ledger entry (C2). +The file is written by hand, by the host's ``raven ops connection add``, or -- +since 2026-09-20 -- by this instance's own ``ops_connection_add`` once the owner +has answered what that command asks (the host stopped gating a spawn on the +registry on 2026-09-06, and with it went the step that ran the +command on the owner's behalf; the ask survived, the write did not). This module +is the plugin's own reader and writer over the same store the trunk module +reads (RAVEN_CONNECTIONS, else the owner's ``~/.raven/connections.json``): the +trunk module is deliberately not a contracts paper, so the plugin cannot import +it at runtime -- behaviour is aligned to the trunk version and any drift is a +parity-ledger entry (C2). ``probe``, ``write`` and ``write_ssh_alias`` mirror +``raven/cli/ops_connection_commands.py`` for the same reason. """ from __future__ import annotations @@ -425,23 +430,27 @@ def resolve_into(meta: dict[str, Any]) -> dict[str, Any]: return merged -# What the owner has to be asked for, when there is nobody to ask it of here. The -# loop cannot fill any of it in: the address and the key are deliberately outside -# what it can see, and the rest -- what is installed, how the budget is metered, -# which directories are the owner's -- is not on the machine to be discovered -# before there is a way onto the machine. +# What the owner has to be asked for, when there is nobody to ask it of here. +# Kept to what only they know: the address and the key are deliberately outside +# what the loop can see, and a name is theirs to give. Everything a machine can +# say about itself -- cores, memory, devices -- is read off it once reached, and +# ssh's own config resolves a port, a user or a key the owner left out. Once +# they have answered, the write is the loop's own: ops_connection_add. _REQUEST = """\ -Hand this to the owner as it stands. Ask them to run `raven ops connection add`, -which asks for, and checks: - - what they call this machine - - whether it is reached over ssh or is this very computer - - for ssh: address, port, username, and the path to the private key - - what is installed on it, with paths -- "CalculiX 2.17 (/opt/calculix)" - - whether its budget is counted in minutes, core-minutes or gpu-minutes - - how many jobs it will run at once - - which directories on it hold the owner's work - -Until one exists there is no machine to run on. Do not submit anything, +Ask the owner, in one message, and stop there until they answer (there is no +campaign yet, so end the turn with the question; whoever called you relays the +answer): + - is the machine this very computer, or another one reached over ssh? + - what do they call it? + - if another one: its address -- and port, username and private-key path + only if they know them; left out, ssh's own config is tried and whatever + connects is kept +Optional, only if they care to say: what is installed on it (with paths), +which directories on it hold their work, and how its budget is counted. + +With the answers, call ops_connection_add: it reaches the machine before +writing anything, and what the machine says about itself is read off it. +Until a machine is listed there is nothing to run on. Do not submit anything, and do not look for a way in with exec or ssh.""" @@ -472,9 +481,8 @@ def describe() -> str: f"The machine registry for this instance cannot be read.\n{found.detail}\n\n" "This is NOT the same as having no machines: the file may well list " "several, and this instance cannot see any of them. Say exactly this " - "to the owner -- the file needs fixing, or rewriting with " - "`raven ops connection add`. Do not submit anything and do not look " - "for a way in yourself." + "to the owner -- the file needs fixing by hand. Do not submit anything " + "and do not look for a way in yourself." ) rows = load() if not rows: @@ -534,6 +542,181 @@ def describe() -> str: ) +# What a machine is asked about itself, the moment it is reached. Read off the +# box rather than typed by the owner, so the row cannot say 64 cores about a +# machine with 32. Mirrors the CLI's probe line. +_PROBE = ( + "echo CORES=$(nproc 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null); " + "echo MEM=$(free -g 2>/dev/null | awk '/^Mem:/{print $2}'); " + "echo GPU=$(nvidia-smi --query-gpu=name,memory.total --format=csv,noheader 2>/dev/null | paste -sd'|' -); " + "echo LIBC=$(ldd --version 2>/dev/null | head -1)" +) + + +def ssh_defaults(host: str, *, port: int = 0, user: str = "") -> dict[str, Any]: + """What ssh itself would use for this host: ``port``, ``user``, ``keys``. + + ``ssh -G`` prints the resolved configuration -- ``~/.ssh/config`` blocks + included -- so an owner who left the port, the user or the key path out + gets the values their own ssh would pick, not a guess. ``keys`` lists only + identity files that exist here, in ssh's own order; the caller probes each + and keeps the one that connects. Empty when ssh is not on this computer. + """ + import subprocess + + argv = ["ssh", "-G"] + (["-p", str(port)] if port else []) + [f"{user}@{host}" if user else host] + try: + done = subprocess.run(argv, capture_output=True, text=True, timeout=10, check=False) + except (FileNotFoundError, subprocess.TimeoutExpired): + return {} + if done.returncode != 0: + return {} + out: dict[str, Any] = {"keys": []} + for line in done.stdout.splitlines(): + key, _, value = line.partition(" ") + if key == "port" and value.strip().isdigit(): + out["port"] = int(value) + elif key == "user" and value.strip(): + out["user"] = value.strip() + elif key == "identityfile" and value.strip(): + if Path(os.path.expanduser(value.strip())).exists(): + out["keys"].append(value.strip()) + return out + + +def probe(row: dict[str, Any], *, timeout: float = 30.0, isolate_key: bool = False) -> tuple[bool, str, dict[str, Any]]: + """Reach the machine and read what it is. ``(reached, message, detected)``. + + The reaching is the point and the reading is the bonus. A row that cannot be + reached is not written, because an unreachable row in the registry is worse + than an absent one: absent is a question the loop knows to ask, unreachable + is a fact it acts on. Rides the plugin's own transport seam, so the way a + machine is reached here is the way its jobs will be reached. + + ``isolate_key`` is for the one caller that has to say WHICH key opened the + session rather than merely that one opened: trying the keys ssh named when + the owner gave no path. Without it, ``-i`` is a preference and not a + restriction -- the other configured identities and the agent are offered too + -- so the first candidate could be credited with a session a different key + authenticated, and the registry would then hold a path that stops working + the day that agent or config changes. + """ + from oncall_flow.backend import JobBackendError + from oncall_flow.transport import runner_from + + where = "this computer" if transport_of(row) == LOCAL else f"{row.get('user')}@{row.get('host')}:{row.get('port')}" + try: + if isolate_key and transport_of(row) == SSH: + from oncall_flow.docker_backend import make_ssh_runner + + runner = make_ssh_runner( + str(row.get("host") or ""), + int(row.get("port") or 22), + os.path.expanduser(str(row.get("key") or "")), + user=str(row.get("user") or "root"), + identities_only=True, + ) + else: + runner = runner_from(row, what="connection", cap_seconds=timeout) + code, out = runner(_PROBE) + except JobBackendError as exc: + return False, f"could not reach {where}: {exc}", {} + if code != 0: + return False, f"could not reach {where}: {out.strip() or f'exit {code}'}", {} + return True, f"reached {where}", _parse_probe(out) + + +def _parse_probe(out: str) -> dict[str, Any]: + seen: dict[str, str] = {} + for line in out.splitlines(): + key, _, value = line.partition("=") + if value.strip(): + seen[key.strip()] = value.strip() + found: dict[str, Any] = {} + if seen.get("CORES", "").isdigit(): + found["cores"] = int(seen["CORES"]) + if seen.get("MEM", "").isdigit() and int(seen["MEM"]) > 0: + found["memory"] = f"{seen['MEM']} GB" + gpu = seen.get("GPU", "") + if gpu: + cards = [c.strip() for c in gpu.split("|") if c.strip()] + found["gpus"] = len(cards) + found["device"] = f"{len(cards)} x {cards[0]}" if len(set(cards)) == 1 else " + ".join(cards) + found["kind"] = "gpu" + elif "cores" in found: + found["kind"] = "cpu" + if seen.get("LIBC"): + found["note"] = seen["LIBC"] + return found + + +def write(row: dict[str, Any]) -> Path: + """Append one machine, keeping whatever shape the file already had. + + Refuses rather than rewrites when the file holds entries this reader skips: + serialising the survivors would erase rows the owner wrote by hand, and a + recoverable typo is theirs to fix, not ours to delete on the way past. + """ + path = store_path() + found = read() + if found.state == UNREADABLE: + raise ValueError(f"{found.detail}\nFix or move that file before adding to it.") + if found.detail: + raise ValueError( + f"{found.detail}\nAdding a machine here would rewrite the file without them. " + "Give those entries an id (or remove them) first, then add this machine." + ) + rows = list(found.rows) + if any(str(r.get("id")).strip() == row["id"] for r in rows): + raise ValueError(f"a machine with id {row['id']!r} is already listed in {path}") + rows.append(row) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"connections": rows}, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + return path + + +_ALIAS_MARK = "# raven connection {conn_id} (managed; rewritten on every add)" + + +def write_ssh_alias(row: dict[str, Any]) -> str | None: + """Keep a ``Host `` alias in ``~/.ssh/config`` for an ssh row. + + Transfers cannot ride the exec machine channel -- rsync and scp run their + client HERE and only address the machine -- so without an alias every + transfer carries the raw address into the model's context. One managed + block per id, replaced in full on re-add; everything outside the markers + is the owner's and is never touched. Returns the alias, or None for a + local row. A failure is the caller's to report as a warning, never as a + refusal: the alias is a convenience beside the registry, not part of it. + """ + if transport_of(row) == LOCAL: + return None + conn_id = str(row["id"]) + mark = _ALIAS_MARK.format(conn_id=conn_id) + block = "\n".join( + [ + mark, + f"Host {conn_id}", + f" HostName {row.get('host')}", + f" Port {int(row.get('port') or 22)}", + f" User {row.get('user') or 'root'}", + f" IdentityFile {row.get('key') or '~/.ssh/id_rsa'}", + mark, + ] + ) + path = Path(os.path.expanduser("~/.ssh/config")) + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + text = path.read_text(encoding="utf-8") if path.exists() else "" + if mark in text: + head, _, rest = text.partition(mark) + _, _, tail = rest.partition(mark) + text = head.rstrip("\n") + ("\n" if head.strip() else "") + tail.lstrip("\n") + text = (text.rstrip("\n") + "\n\n" if text.strip() else "") + block + "\n" + path.write_text(text, encoding="utf-8") + path.chmod(0o600) + return conn_id + + # A claim shallower than this is a whole filesystem, not a case: "/", "/opt", # "/Users/admin". Honouring one would put a line about machines on every ordinary # look, and a note that fires everywhere is read as noise and then not read at diff --git a/agents/raven-oncall/plugins/oncall-flow/oncall_flow/docker_backend.py b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/docker_backend.py index 5ecfc4072..0f7351768 100644 --- a/agents/raven-oncall/plugins/oncall-flow/oncall_flow/docker_backend.py +++ b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/docker_backend.py @@ -227,6 +227,13 @@ def _parse_docker_time(value: str) -> float | None: return None +#: What it takes for a session to be attributable to the key passed with ``-i``: +#: no config read (so the owner's own IdentityFile lines are out), no agent, and +#: no provider identities. Read by the guard that measures this through ssh's own +#: resolver rather than by eye -- see the connection-add suite. +ISOLATE_IDENTITY = ("-F", "/dev/null", "-o", "IdentitiesOnly=yes", "-o", "IdentityAgent=none") + + def make_ssh_runner( host: str, port: int, @@ -234,7 +241,21 @@ def make_ssh_runner( *, user: str = "root", connect_timeout: int = 15, + identities_only: bool = False, ) -> CommandRunner: + """Run a command over ssh. ``identities_only`` narrows auth to ``key`` alone. + + Off by default, which is how a job reaches its machine: whatever the owner's + ssh would use gets to work, an agent included. Turned on only where the + session has to prove WHICH key opened it, and it takes all three options in + ISOLATE_IDENTITY to get there. ``-i`` is a preference, not a restriction; + ``IdentitiesOnly=yes`` is defined as excluding what an agent or provider adds, + not what the config names, so the owner's own ``IdentityFile`` lines are still + offered beside the candidate. Measured with OpenSSH 9.9p2 against a config + carrying two ``IdentityFile`` lines: ``-i cand`` with IdentitiesOnly and no + agent resolves to three identities, and only reading no config at all leaves + the one. An existing ``-i`` file suppresses the built-in defaults by itself. + """ import subprocess def run(cmd: str) -> tuple[int, str]: @@ -250,6 +271,10 @@ def run(cmd: str) -> tuple[int, str]: f"ConnectTimeout={connect_timeout}", "-o", "StrictHostKeyChecking=accept-new", + ] + if identities_only: + argv += list(ISOLATE_IDENTITY) + argv += [ f"{user}@{host}", cmd, ] diff --git a/agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/__init__.py b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/__init__.py index 52ba72c54..e8c305a94 100644 --- a/agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/__init__.py +++ b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/__init__.py @@ -102,6 +102,14 @@ def make_ops_connections(ctx: "PluginContext"): return base.adopt(OpsConnectionsTool()) +def make_ops_connection_add(ctx: "PluginContext"): + if not _wired(ctx): + return None + from oncall_flow.tools.ops_connection_add import OpsConnectionAddTool + + return base.adopt(OpsConnectionAddTool()) + + def make_ops_declare(ctx: "PluginContext"): if not _wired(ctx): return None diff --git a/agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/ops_connection_add.py b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/ops_connection_add.py new file mode 100644 index 000000000..14e562a08 --- /dev/null +++ b/agents/raven-oncall/plugins/oncall-flow/oncall_flow/tools/ops_connection_add.py @@ -0,0 +1,203 @@ +"""Add one machine to the registry, from what the owner answered. + +The counterpart of ``ops_connections``' empty-registry reply: that reply asks +the owner what only they know, and this face takes the answer. Until 2026-09-06 +the host ran ``raven ops connection add --non-interactive`` on the owner's +behalf before dispatching a spawn; that step went with the pre-dispatch registry +gate, so an owner who answered in conversation had nothing consuming the answer. Here the write belongs to the instance that needs the machine. + +Two things keep the ask short. What a machine can say about itself -- cores, +memory, devices -- is read off it once reached, never typed. And what the owner +leaves out of the way in -- port, username, key path -- is resolved the way +their own terminal would resolve it (``ssh -G``, config blocks included), then +tried: the probe decides, and only what connected is written. Nothing is written +until the machine itself answers; an unreachable row in the registry is a fact +every later turn acts on, an absent one is a question the loop knows to ask. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +from raven.contracts.tool import Tool + +_BUDGET_UNITS = ("minute", "core-minute", "gpu-minute") + + +class OpsConnectionAddTool(Tool): + """Write one machine the owner described, after reaching it.""" + + @property + def name(self) -> str: + return "ops_connection_add" + + @property + def description(self) -> str: + return ( + "Add a machine to the registry from the owner's own answers, when ops_connections " + "lists none that fits. Needed: what they call it, and whether it is this very computer " + "or another one reached over ssh; for ssh, its address. Port, username and private-key " + "path only when the owner gave them -- left out, ssh's own config is consulted and " + "whatever connects is kept, so never invent one: a guess that lands in the registry " + "stops being a guess. What is installed, which directories hold their work and how the " + "budget is counted are optional; cores, memory and devices are read off the machine. " + "Nothing is written until the machine answers. Then hand the new id to ops_declare." + ) + + @property + def parameters(self) -> dict[str, Any]: + return { + "type": "object", + "properties": { + "name": {"type": "string", "description": "What the owner calls this machine."}, + "transport": { + "type": "string", + "enum": ["ssh", "local"], + "description": "'ssh' for another computer, 'local' for this very one.", + }, + "host": {"type": "string", "description": "ssh only: the address."}, + "port": {"type": "integer", "description": "ssh only, if the owner said; else ssh's config."}, + "user": {"type": "string", "description": "ssh only, if the owner said; else ssh's config."}, + "key": { + "type": "string", + "description": "ssh only, if the owner said: path to the private key (the path is stored, " + "not the key). Left out, the keys ssh itself would try are probed in turn.", + }, + "software": {"type": "string", "description": "Optional: what is installed, with paths."}, + "budget_unit": { + "type": "string", + "enum": list(_BUDGET_UNITS), + "description": "Optional: gpu-minute for a GPU machine, core-minute otherwise.", + }, + "concurrency": { + "type": "integer", + "description": "Optional: jobs at once, read only when the machine reports no cores or devices.", + }, + "paths": { + "type": "array", + "items": {"type": "string"}, + "description": "Optional: absolute directories on it that hold the owner's work.", + }, + "id": { + "type": "string", + "description": "Short id; minted from the name when omitted. Never renamed, campaigns store it.", + }, + "note": {"type": "string"}, + }, + "required": ["name", "transport"], + } + + async def execute( + self, + name: str, + transport: str, + host: str = "", + port: int = 0, + user: str = "", + key: str = "", + software: str = "", + budget_unit: str = "", + concurrency: int = 0, + paths: list | None = None, + id: str = "", # noqa: A002 -- the registry's own field name, kept so the model sees one spelling + note: str = "", + **kwargs: Any, + ) -> str: + from oncall_flow import connections + from raven.utils.paths import mint_slug + + label = str(name).strip() + if not label: + return "REFUSED: the machine needs the owner's name for it. Nothing was written." + kind = connections.LOCAL if str(transport).strip().lower() == connections.LOCAL else connections.SSH + conn_id = str(id).strip() or mint_slug(label) + row: dict[str, Any] = {"id": conn_id, "display_name": label, "transport": kind} + if kind == connections.LOCAL: + reached, message, found = await asyncio.to_thread(connections.probe, row) + else: + host = str(host).strip() + if not host: + return "REFUSED: an ssh machine needs its address; ask the owner rather than guessing. Nothing was written." + # What the owner left out, ssh's own config supplies -- the same + # answer their terminal would give -- and the probe decides. Asked + # only when something is actually missing: with all three given there + # is nothing for it to answer. + given = str(key).strip() + resolved: dict[str, Any] = {} + if not (port and str(user).strip() and given): + resolved = await asyncio.to_thread(connections.ssh_defaults, host, port=int(port or 0), user=str(user)) + row.update( + { + "host": host, + "port": int(port or resolved.get("port") or 22), + "user": str(user).strip() or str(resolved.get("user") or "root"), + } + ) + keys = [given] if given else list(resolved.get("keys") or []) + if not keys: + return ( + f"REFUSED: no private key to reach {row['user']}@{host}:{row['port']} with -- the owner gave " + "no key path and none of ssh's default keys exists here. Ask them for the path. Nothing was written." + ) + reached, message, found = False, "", {} + for candidate in keys: + row["key"] = candidate + # A path the owner named is their assertion, and is probed the way a + # job will reach the machine -- an agent doing the authenticating is + # their setup working. A path this tool picked is a claim of its own, + # so it is probed with that key alone: credited otherwise, the row + # would hold a key that only looked like the one that opened the + # session, and stop working the day the agent or the config changes. + reached, message, found = await asyncio.to_thread(connections.probe, row, isolate_key=not given) + if reached: + break + if not reached and not given: + # Attribution costs the owner's ssh config: a candidate is tried + # with no config read, so an alias or a jump host that makes this + # machine reachable is not in play here. Naming the key puts the + # probe back on the path a job will take. + message = ( + f"{message} (tried, each on its own, the key(s) ssh named: {', '.join(keys)}; " + "a candidate is probed with no ssh config read, so a machine reached through " + "an alias or a jump host needs its key path named)" + ) + if not reached: + return f"REFUSED: {message}. Nothing was written. Check the address, port, user and key with the owner." + row.update(found) + if str(software).strip(): + row["software"] = str(software).strip() + row["budget_unit"] = str(budget_unit).strip() or ("gpu-minute" if row.get("kind") == "gpu" else "core-minute") + # Not read on a row that reports cores or devices (capacity decides); + # kept only where it is the one thing admission has to go on. + if int(concurrency or 0) > 0 and not connections.resource_unit(row): + row["concurrency"] = int(concurrency) + if paths: + row["paths"] = [str(p) for p in paths] + if note: + row["note"] = str(note) + + faults = connections.row_problems(row) + blocking = [f for f in faults if f.blocking] + if blocking: + return "REFUSED: " + "; ".join(str(f) for f in blocking) + ". Nothing was written." + try: + written = connections.write(row) + except ValueError as exc: + return f"REFUSED: {exc} Nothing was written." + + how = f" with key {row['key']}" if kind == connections.SSH else "" + lines = [f"{message}{how}; wrote {conn_id} to {written}."] + if found: + lines.append("Read off the machine: " + " ".join(f"{k} {v}" for k, v in found.items())) + for fault in faults: + lines.append(f"Worth telling the owner: {fault}") + try: + alias = connections.write_ssh_alias(row) + except OSError as exc: + lines.append(f"ssh alias not written ({exc}); transfers will need the address by hand.") + else: + if alias: + lines.append(f"ssh alias '{alias}' written to ~/.ssh/config, so rsync/scp reach it by id.") + lines.append(f"Now hand '{conn_id}' to ops_declare as 'connection'.") + return "\n".join(lines) diff --git a/agents/raven-oncall/plugins/oncall-flow/prompts/TOOLS_ONCALL.md b/agents/raven-oncall/plugins/oncall-flow/prompts/TOOLS_ONCALL.md index dfc4b3acf..72e9b50f4 100644 --- a/agents/raven-oncall/plugins/oncall-flow/prompts/TOOLS_ONCALL.md +++ b/agents/raven-oncall/plugins/oncall-flow/prompts/TOOLS_ONCALL.md @@ -119,7 +119,11 @@ is 1440 of them. **`ops_connections` lists the machines this instance can run on** — the owner's own name for each, what it is, and what it has installed. You never need a host, a port, a user or a key: those belong to the connection and the tools use them -for you. A task statement that names no machine is normal. +for you. A task statement that names no machine is normal. When it lists none +that fits, ask the owner what it says to ask -- is it this computer or another, +what they call it, and an address if another -- and hand the answer to +`ops_connection_add`; ssh's own config fills a port, user or key they left out, +and the machine is reached before anything is written. **`exec` takes a `machine`** — an id from `ops_connections`, or a campaign name — and runs the command there instead of on this computer. Use it to look at anything diff --git a/agents/raven-oncall/plugins/oncall-flow/raven-plugin.toml b/agents/raven-oncall/plugins/oncall-flow/raven-plugin.toml index 1398a7ade..31f05731e 100644 --- a/agents/raven-oncall/plugins/oncall-flow/raven-plugin.toml +++ b/agents/raven-oncall/plugins/oncall-flow/raven-plugin.toml @@ -75,6 +75,10 @@ factory = "oncall_flow.tools:make_ops_campaigns" name = "ops_connections" factory = "oncall_flow.tools:make_ops_connections" +[[plugin.contributes.tools]] +name = "ops_connection_add" +factory = "oncall_flow.tools:make_ops_connection_add" + [[plugin.contributes.tools]] name = "ops_declare" factory = "oncall_flow.tools:make_ops_declare" diff --git a/demos/oncall_heat2d/README.md b/demos/oncall_heat2d/README.md index b8561221e..96c9072e1 100644 --- a/demos/oncall_heat2d/README.md +++ b/demos/oncall_heat2d/README.md @@ -71,11 +71,11 @@ Tool call: read_file(arena/run_heat.sh) <- looks at the case Tool call: spawn(subagent="Raven-Oncall", ...) -> "Subagent [...] started" ... the on-call agent's first call is ops_connections, which answers - "No machine is set up in this instance ... Hand this to the owner as - it stands. Ask them to run `raven ops connection add`" and lists the - seven things that command asks for -$ raven ops connection add --transport local <- you, once - -> probed the machine, wrote the registry row + "No machine is set up in this instance ... Ask the owner for these" + and lists seven things; the question reaches you through the host + (you answer in the conversation: this computer, what is installed, ...) +Tool call: ops_connection_add(name=..., transport="local", ...) + -> reached this computer, read its cores and memory, wrote the registry row ... the on-call agent declares a campaign, submits trials, takes readings, and concludes with the best stable configuration and its L2 error ``` diff --git a/tests/test_agents_oncall_flow_connection_add.py b/tests/test_agents_oncall_flow_connection_add.py new file mode 100644 index 000000000..400a41ad1 --- /dev/null +++ b/tests/test_agents_oncall_flow_connection_add.py @@ -0,0 +1,194 @@ +"""The on-call instance writes the machine the owner described, after reaching it. + +Until 2026-09-06 the host ran `raven ops connection add` on the owner's behalf +before a spawn; that step went with the pre-dispatch registry gate, and an owner who answered the seven questions in conversation had nothing +consuming the answer. These pin the guarantee the write carries over hand-editing +-- nothing lands until the machine itself has answered -- and that what lands is +what the listing then shows. +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parent.parent +PLUGIN_DIR = REPO / "agents" / "raven-oncall" / "plugins" / "oncall-flow" +sys.path.insert(0, str(PLUGIN_DIR)) + +from oncall_flow import connections # noqa: E402 +from oncall_flow.tools import base as tools_base # noqa: E402 +from oncall_flow.tools.ops_connection_add import OpsConnectionAddTool # noqa: E402 + + +@pytest.fixture(autouse=True) +def _home(tmp_path, monkeypatch): + tools_base.set_home(tmp_path / "ops") + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setattr(connections, "store_path", lambda: tmp_path / "connections.json") + return tmp_path + + +@pytest.fixture +def reached(monkeypatch): + monkeypatch.setattr(connections, "probe", lambda row, **kw: (True, "reached it", {"cores": 32, "kind": "cpu"})) + + +OWNER_SAID = { + "name": "my CPU box", + "transport": "ssh", + "host": "14.103.100.27", + "port": 58717, + "user": "root", + "key": "~/.ssh/id_rsa", + "software": "OpenFOAM v2312 (/opt/openfoam)", + "budget_unit": "core-minute", + "concurrency": 1, + "paths": ["/home/cfd/work"], +} + + +def add(**overrides): + return asyncio.run(OpsConnectionAddTool().execute(**{**OWNER_SAID, **overrides})) + + +def test_reached_machine_is_written_and_listed(_home, reached): + out = add() + assert "wrote my-cpu-box" in out + rows = json.loads((_home / "connections.json").read_text())["connections"] + assert rows[0]["id"] == "my-cpu-box" + assert rows[0]["cores"] == 32, "what the machine said about itself is on the row" + assert "my CPU box" in connections.describe() + assert "Host my-cpu-box" in (_home / ".ssh" / "config").read_text() + + +def test_unreachable_machine_writes_nothing(_home, monkeypatch): + monkeypatch.setattr(connections, "probe", lambda row, **kw: (False, "could not reach it: refused", {})) + out = add() + assert out.startswith("REFUSED") + assert not (_home / "connections.json").exists() + + +def test_ssh_without_an_address_is_refused_before_probing(_home, monkeypatch): + monkeypatch.setattr(connections, "probe", lambda row, **kw: pytest.fail("probed with no address")) + out = add(host="") + assert "REFUSED" in out and "address" in out + assert not (_home / "connections.json").exists() + + +def test_what_the_owner_left_out_comes_from_ssh_config_and_the_key_that_connects(_home, monkeypatch): + """Ssh's own resolution names the candidates, and each is tried with that key ALONE. + + `-i` is a preference, not a restriction: OpenSSH still offers the other + configured identities and whatever the agent holds. A candidate credited with + a session some other key authenticated would put a path in the registry that + stops working the day that agent or that config goes. + """ + monkeypatch.setattr( + connections, + "ssh_defaults", + lambda host, **kw: {"port": 58717, "user": "cfd", "keys": ["~/.ssh/id_ed25519", "~/.ssh/id_rsa"]}, + ) + tried = [] + + def probe(row, **kw): + tried.append((row["key"], kw.get("isolate_key"))) + hit = row["key"] == "~/.ssh/id_rsa" + return hit, "reached it" if hit else "refused", {} + + monkeypatch.setattr(connections, "probe", probe) + out = add(port=0, user="", key="") + assert tried == [("~/.ssh/id_ed25519", True), ("~/.ssh/id_rsa", True)] + row = json.loads((_home / "connections.json").read_text())["connections"][0] + assert (row["port"], row["user"], row["key"]) == (58717, "cfd", "~/.ssh/id_rsa") + assert "with key ~/.ssh/id_rsa" in out + + +def test_a_key_the_owner_named_is_probed_the_way_a_job_will_reach_the_machine(_home, monkeypatch): + """No isolation for a path the owner asserted: an agent doing the + authenticating is their setup working, and refusing it here would refuse a + machine every later job can reach.""" + seen = [] + + def probe(row, **kw): + seen.append(kw.get("isolate_key")) + return True, "reached it", {} + + monkeypatch.setattr(connections, "probe", probe) + monkeypatch.setattr(connections, "ssh_defaults", lambda host, **kw: pytest.fail("resolved a key the owner gave")) + add(key="~/.ssh/id_rsa") + assert seen == [False] + + +def test_an_isolated_candidate_leaves_ssh_exactly_one_identity_to_offer(tmp_path, monkeypatch): + """Measured through ssh's own resolver, because the argv does not settle it. + + `-i` is a preference, not a restriction, and `IdentitiesOnly=yes` is defined + as excluding what an agent or a provider adds -- not what the config names. + With the owner's own `IdentityFile` lines still in play, any one of them can + authenticate the first candidate's session and be credited to it, and the + registry then holds a path that merely looked verified. + + The owner's real config cannot be pointed somewhere else for a test (ssh + finds it through the password database, not $HOME), so one is put in front + of the runner's own arguments instead. The last `-F` wins, which is what + makes this sensitive: drop the runner's and this config is read. + """ + import shutil + import subprocess + + from oncall_flow.docker_backend import make_ssh_runner + + if not shutil.which("ssh"): + pytest.skip("no ssh on this computer") + + cand, other_a, other_b = (tmp_path / n for n in ("cand", "other_a", "other_b")) + for f in (cand, other_a, other_b): + f.write_text("") + config = tmp_path / "config" + config.write_text(f"Host target\n HostName 127.0.0.1\n IdentityFile {other_a}\n IdentityFile {other_b}\n") + + def resolved(argv: list[str]) -> list[str]: + out = subprocess.run(argv, capture_output=True, text=True, check=False) + return [ln.split(None, 1)[1] for ln in out.stdout.splitlines() if ln.startswith("identityfile ")] + + def as_ssh_reads_it(*, isolated: bool) -> list[str]: + seen: list[list[str]] = [] + monkeypatch.setattr( + subprocess, "run", lambda argv, **kw: seen.append(argv) or subprocess.CompletedProcess(argv, 0, "", "") + ) + make_ssh_runner("target", 22, str(cand), user="u", identities_only=isolated)("true") + monkeypatch.undo() + # The runner's own argv, handed back to ssh as a question rather than a + # connection, with this owner's config in front of it. + return resolved(["ssh", "-G", "-F", str(config)] + seen[0][1:-2] + [seen[0][-2]]) + + assert as_ssh_reads_it(isolated=True) == [str(cand)], "only the candidate may authenticate its own probe" + + open_to_all = as_ssh_reads_it(isolated=False) + assert str(other_a) in open_to_all and str(other_b) in open_to_all, ( + "the guard means nothing unless this config would otherwise put other keys in play" + ) + + # The reviewer's finding, kept as a measurement: shutting the agent out is + # not enough on its own, which is why the option set also reads no config. + agent_only = resolved( + [ + "ssh", + "-G", + "-F", + str(config), + "-o", + "IdentitiesOnly=yes", + "-o", + "IdentityAgent=none", + "-i", + str(cand), + "u@target", + ] + ) + assert str(other_a) in agent_only and str(other_b) in agent_only diff --git a/tests/test_agents_oncall_flow_watcher.py b/tests/test_agents_oncall_flow_watcher.py index 99d22a6b7..973276b33 100644 --- a/tests/test_agents_oncall_flow_watcher.py +++ b/tests/test_agents_oncall_flow_watcher.py @@ -111,6 +111,7 @@ def test_the_manifest_contributes_the_watcher_service_and_the_ops_tools() -> Non "ops_note", "ops_campaigns", "ops_connections", + "ops_connection_add", "ops_declare", "ops_kill", "ops_outputs", diff --git a/tests/test_agents_oncall_launcher.py b/tests/test_agents_oncall_launcher.py index 9ff806099..e639c70aa 100644 --- a/tests/test_agents_oncall_launcher.py +++ b/tests/test_agents_oncall_launcher.py @@ -46,14 +46,43 @@ def grounded(launcher, tmp_path, monkeypatch): # --- byte parity: the one prompt asset and the roster identity -------------- +#: Where this product's guide deliberately says more than the frozen twin's, as +#: ``(the fork's text, what stands here instead)``. Applied to the fork's copy +#: before the comparison, so the guard still reads byte for byte everywhere else: +#: an addition quietly dropped, reworded, or landing on a fork line that has since +#: moved fails exactly as a silent drift would. The A side is a record and is never +#: edited to make this green (tests/fixtures/vendored_fork/README.md). +#: +#: ops_connection_add is this side's own tool. The fork never had it: its host +#: filled the connection registry from the owner's answer before dispatching, and +#: that step was removed on 2026-09-06, so the ask now has to name the tool that +#: consumes the answer. +B_SIDE_ADDITIONS = ( + ( + "for you. A task statement that names no machine is normal.\n", + "for you. A task statement that names no machine is normal. When it lists none\n" + "that fits, ask the owner what it says to ask -- is it this computer or another,\n" + "what they call it, and an address if another -- and hand the answer to\n" + "`ops_connection_add`; ssh's own config fills a port, user or key they left out,\n" + "and the machine is reached before anything is written.\n", + ), +) + + def test_the_oncall_section_is_the_vendored_twins(): - """Byte parity, no exception list. + """Byte parity, with every divergence named in B_SIDE_ADDITIONS. The respelling table this test used to carry existed because the fork's exec(machine=...) face was contributed under its own name (ops_exec). - Trunk exec now carries the machine parameter itself, so the fork's - wording is simply true again and the guide is the twin's, byte for byte.""" - assert PRODUCT_SECTION.read_text(encoding="utf-8") == FORK_SECTION.read_text(encoding="utf-8") + Trunk exec now carries the machine parameter itself, so that entry is gone + and the guide is the twin's, byte for byte -- except where this side has + deliberately grown a tool the fork never had, which is listed above rather + than written into the frozen record.""" + expected = FORK_SECTION.read_text(encoding="utf-8") + for fork_text, product_text in B_SIDE_ADDITIONS: + assert fork_text in expected, f"the fork text this addition extends has moved: {fork_text!r}" + expected = expected.replace(fork_text, product_text, 1) + assert PRODUCT_SECTION.read_text(encoding="utf-8") == expected def test_the_seeded_guide_is_the_trunk_template_plus_the_section(grounded, tmp_path): @@ -243,6 +272,12 @@ def test_a_preexisting_own_registry_stays(grounded, tmp_path): #: be disabled by the product config, not by luck. The two playbook tools and #: the everos understand_media only register outside this hermetic fixture; #: their disable rows are pinned below instead. +#: +#: ops_connection_add is the one row here the fork never had. Its host filled the +#: connection registry from the owner's answer before dispatching a spawn; that +#: step was removed on 2026-09-06, so the instance that needs the machine now +#: writes the row itself. Listed explicitly, not folded in: a tool reaching this +#: face without a line saying why is the drift this guard exists to catch. VENDORED_TOOL_FACE = { "ask_user", "edit_file", @@ -255,6 +290,7 @@ def test_a_preexisting_own_registry_stays(grounded, tmp_path): "ops_campaigns", "ops_case_changes", "ops_check_later", + "ops_connection_add", "ops_connections", "ops_declare", "ops_edit_case_dict",