diff --git a/emrg/server/daemon.py b/emrg/server/daemon.py index 7da8a308..8e4f8867 100644 --- a/emrg/server/daemon.py +++ b/emrg/server/daemon.py @@ -160,6 +160,7 @@ def resolve_client_tier(session, req_sandbox: str | None = None) -> str: from emrg.tools.glob_tool import GlobTool from emrg.tools.grep_tool import GrepTool from emrg.tools.submit_rant_tool import SubmitRantTool +from emrg.tools.subagent_tool import SubagentTool from emrg.skills.loader import load_skills from emrg.skills.registry import ensure_catalog_file, load_catalog_skills, skill_is_managed from emrg.server.rants import append_rant, list_rants @@ -177,6 +178,13 @@ def resolve_client_tier(session, req_sandbox: str | None = None) -> str: #: roots — and two hand-written unions would be two chances to drift apart. _ROOT_CONSUMING_TOOLS = SHELL_TOOL_NAMES | {"write", "edit"} +#: How much of a child session's final answer a `subagent` call returns, in +#: characters (issue #2451). The answer is the one thing the parent gets back, so it +#: needs a ceiling — a child that dumps a whole file would spend the parent's context +#: on material the parent asked not to carry. Cut, and **said to be cut**, because a +#: silently truncated report reads as a complete one. +_SUBAGENT_ANSWER_CHARS = 6000 + # ── Jinja2 template environment for system prompt ── _jinja_env = None @@ -771,6 +779,20 @@ def __init__(self, llm_config: LlmConfig) -> None: self._session_pending: dict[str, list[tuple[TaskRequest, bool]]] = {} self._all_connections: set = set() # all authenticated connections + # Subagent delegation (issue #2451): nine slots per parent session, and the + # instant each was last *freed*. Built on first use, with every slot closed + # at daemon start, so the first choice is decided by the slot number alone + # rather than by dictionary order. A daemon restart clears this table — the + # same event that ends any child still running, so the two agree. + self._subagent_slots: dict[str, dict[int, float]] = {} + self._subagent_busy: dict[str, set[int]] = {} + self._subagent_started = time.time() + # Sessions whose memory reflection is still running. A `subagent` call waits + # on its child's entry before it stamps that slot closed: the reflection is + # fire-and-forget, so a slot cleared while the previous child is still + # writing into it has its directory recreated underneath the clear. + self._memory_tasks: dict[str, asyncio.Task] = {} + # Device-flow auth (rant 10:17 Stage 2b): background gh auth login --web task self._pending_web_auth: Optional[asyncio.Task] = None self._pending_web_auth_proc: Optional[asyncio.subprocess.Process] = None @@ -789,6 +811,9 @@ def __init__(self, llm_config: LlmConfig) -> None: self.tools.register(GlobTool()) self.tools.register(GrepTool()) self.tools.register(SubmitRantTool()) + # The runner is the daemon's own method, so the tool has no path to the + # session machinery of its own: everything that owns state stays here. + self.tools.register(SubagentTool(self._run_subagent_tool)) logger.info("tools registered: %s", self.tools.names) # Skills are **not** held here. A snapshot taken at startup cannot see a skill @@ -4212,6 +4237,15 @@ def _inject_tool_arguments(tc_name: str, args: dict, session, req: TaskRequest) # safe form is the unconditional one: whatever the session holds, # including nothing, is what the call runs with. args["writable_roots"] = session.sandbox_roots + if tc_name == "subagent": + # The parent's identity and cwd are facts the model must not choose + # (issue #2451): the slot is derived from both, so a caller that named + # its own would name a wrong one. Injected unconditionally, so a value + # the model put in the call is replaced rather than honoured — the same + # rule the writable roots follow, and for the same reason: the parent of + # a delegation cannot be the delegated. + args["_parent_session_id"] = session.session_id + args["_parent_cwd"] = str(session.cwd) async def _apply_escalation( self, tc_name: str, args: dict, session, req: TaskRequest, @@ -7355,6 +7389,272 @@ async def _generate_session_title(self, session: Session) -> str: # ── Memory reflection & consolidation ───────────────────── + # ── Subagent delegation (issue #2451) ───────────────────────────────── + + #: How many child sessions one parent may have in play. Fixed, so the session + #: directories under a parent cannot grow without bound: a tenth concurrent + #: call waits for one to be released instead. + SUBAGENT_SLOTS = 9 + #: The deepest session that may still delegate. A session at this depth refuses, + #: which bounds nesting without a timer, a budget or a guess. + SUBAGENT_MAX_DEPTH = 9 + + def _subagent_slot_closes(self, parent_session_id: str) -> dict[int, float]: + """This parent's slots: slot number → the instant it was last freed.""" + table = self._subagent_slots.get(parent_session_id) + if table is None: + table = {n: self._subagent_started for n in range(1, self.SUBAGENT_SLOTS + 1)} + self._subagent_slots[parent_session_id] = table + return table + + def _subagent_acquire(self, parent_session_id: str) -> int: + """Take the free slot closed longest ago; raise ``LookupError`` if none is free. + + No ``await`` sits between the choice and the mark, so two calls cannot pick + the same slot: the daemon runs on one event loop and this is the whole of + the critical section. + """ + busy = self._subagent_busy.setdefault(parent_session_id, set()) + closes = self._subagent_slot_closes(parent_session_id) + free = [n for n in closes if n not in busy] + if not free: + raise LookupError("no free subagent slot") + slot = min(free, key=lambda n: (closes[n], n)) + busy.add(slot) + return slot + + def _subagent_release(self, parent_session_id: str, slot: int) -> None: + self._subagent_busy.get(parent_session_id, set()).discard(slot) + self._subagent_slot_closes(parent_session_id)[slot] = time.time() + + def _memory_task_table(self) -> dict[str, "asyncio.Task"]: + """This daemon's session → pending memory write table, created on first use. + + Lazily, because `_maybe_reflect_memory` is also called on a server built + without `__init__`: the reflection tests make one with `__new__` and hand it + only the attributes it reads, so a table that exists only in the constructor + turns every one of them into an `AttributeError` (measured — twelve red rows + when this table was first added). + """ + table = getattr(self, "_memory_tasks", None) + if table is None: + table = self._memory_tasks = {} + return table + + async def _await_memory_reflection(self, session_id: str) -> None: + """Wait for a session's pending memory write, if it has one. + + The reflection is fire-and-forget, so nothing else in the daemon knows when + a session's memory has finished landing — and "the session is closed" has to + mean the whole teardown, this write included. + """ + task = self._memory_task_table().pop(session_id, None) + if task is None or task.done(): + return + try: + await asyncio.shield(task) + except asyncio.CancelledError: + # Cancellation belongs to the caller's turn, and the task keeps writing. + raise + except Exception: + logger.debug("memory reflection for %s failed", session_id, exc_info=True) + + async def _run_subagent_tool( + self, *, parent_session_id: str, parent_cwd: str, prompt: str, + timeout_s: int, intent: str = "", + ) -> dict: + """Run one child session for one `subagent` call. Answers text or an error. + + The bound is the call's own ``timeout_s`` and it covers **everything** — + waiting for a free slot included — so there is no second timeout to reason + about. When it fires, the child's turn is cancelled the way any cancelled + turn is (its own ``finally`` runs) and the slot is returned before this + returns, so a timed-out delegation leaves nothing occupied. + + The bound is measured by *whether the child finished*, not by catching a + timeout exception: the tool loop absorbs the cancellation it receives (it + broadcasts the turn's `cancelled` frame and returns — that is what a user's + cancel does), so a cancelled child ends **normally**, and `wait_for` would + hand back a finished-looking failure. What separates the two outcomes is + the clock, so the clock is what is read. + """ + state: dict = {"slot": None, "child": None} + child_task = asyncio.create_task(self._run_subagent_child( + parent_session_id=parent_session_id, + parent_cwd=parent_cwd, + prompt=prompt, + intent=intent, + state=state, + )) + try: + done, _ = await asyncio.wait({child_task}, timeout=timeout_s) + except asyncio.CancelledError: + # The parent's own turn was cancelled. The child is not left running + # behind it, and the slot is back before this returns. + child_task.cancel() + await asyncio.gather(child_task, return_exceptions=True) + raise + if done: + return child_task.result() + child_task.cancel() + # Awaited, never abandoned: this is where the child's own `finally` runs, + # so the slot, the busy flag and the memory write are all settled before + # the refusal is handed to the parent. + await asyncio.gather(child_task, return_exceptions=True) + label = f"[sub-{state['slot']}]" if state["slot"] else "[sub-?]" + logger.warning( + "%s delegate timed out after %ss: parent=%s child=%s", + label, timeout_s, parent_session_id, state["child"], + ) + return {"error": ( + f"subagent: the child did not finish within {timeout_s}s, so its turn " + "was cancelled and its slot returned before this call ended. Nothing " + f"from it was delivered. Child session: {state['child']}. Raise " + "`timeout_s`, or split the job into smaller ones." + )} + + async def _run_subagent_child( + self, *, parent_session_id: str, parent_cwd: str, prompt: str, + intent: str, state: dict, + ) -> dict: + """The body of one delegation: pick a slot, run the child, report its answer.""" + started = time.monotonic() + parent = self._get_or_create_session(parent_session_id, Path(parent_cwd)) + + depth = parent.depth + if depth >= self.SUBAGENT_MAX_DEPTH: + # A refusal, named: a silent no-op here reads as "the job ran and had + # nothing to say", which is the one thing it must not read as. + logger.info( + "[sub-0] delegate refused: delegation depth %d reached (parent=%s)", + depth, parent_session_id, + ) + return {"error": ( + f"subagent: refused — this session is already {depth} delegations " + f"deep and the limit is {self.SUBAGENT_MAX_DEPTH}. Nothing was " + "started. Do the work in this session, or have the session that " + "started this one do the delegating." + )} + + while True: + try: + slot = self._subagent_acquire(parent_session_id) + break + except LookupError: + # All nine are held. Waiting is the specified behaviour, and the + # call's own timeout is what ends it. + await asyncio.sleep(0.05) + + state["slot"] = slot + # The id carries the parent's, because the busy/cancel/turn tables are keyed + # by id across the whole daemon: a fixed `sub1` would make two parents' + # delegations indistinguishable there and each would block the other. + child_id = f"sub-{parent_session_id}-{slot}" + state["child"] = child_id + # The slot directory is the child's **cwd**, so the child's own session + # directory is built inside it like any other session's. That is what keeps + # a child's memory inside the parent's session tree without a second + # project root, and it is why nothing here fights `is_safe_session_id`, + # which constrains the id alone. + slot_cwd = ( + Path(parent_cwd) / ".emrg" / "sessions" / parent_session_id + / "sub-agent" / f"sub{slot}" + ) + + try: + # Empty the slot before it is used. `Session.delete` is the only way to + # do it that also drops the global index entry, which a hand-rolled + # rmtree leaves pointing at a directory that no longer exists. + Session.delete(child_id, slot_cwd) + child = Session.create_with_id(child_id, slot_cwd) + child.set_depth(depth + 1) + # The child's tier is the parent's, never the child's own: authorization + # a caller can choose for the thing it is authorizing is not a boundary. + if parent.sandbox: + child.set_sandbox(parent.sandbox) + if parent.sandbox_roots: + try: + child.set_sandbox_roots(parent.sandbox_roots) + except (ValueError, OSError): + logger.warning( + "[sub-%d] could not inherit the parent's writable roots", slot, + ) + + req = TaskRequest( + id=f"subagent-{slot}-{int(started * 1000)}", + session_id=child_id, + cwd=str(slot_cwd), + prompt=prompt, + timestamp="", + sandbox=parent.sandbox, + ) + assistants_before = sum( + 1 for r in child._read_history() if r.get("role") == "assistant" + ) + logger.info( + "[sub-%d] delegate start: parent=%s child=%s depth=%d intent=%s", + slot, parent_session_id, child_id, depth + 1, + redact_string(intent) if intent else "-", + ) + + self._session_busy[child_id] = True + cancel_event = asyncio.Event() + self._session_cancel[child_id] = cancel_event + try: + await self._run_tool_loop_locked(req, None, child, cancel_event) + finally: + # "Closed" is the whole teardown, the memory write included: this + # await is the one thing that makes the ordering observable, and + # without it the next child of this slot clears a directory the + # previous one is still writing into. + await self._await_memory_reflection(child_id) + self._session_busy[child_id] = False + self._session_cancel.pop(child_id, None) + + records = child._read_history() + rounds = max(0, sum( + 1 for r in records if r.get("role") == "assistant" + ) - assistants_before) + answer = "" + for record in reversed(records): + if record.get("role") != "assistant": + continue + content = record.get("content") + if isinstance(content, str) and content.strip(): + answer = content + break + elapsed = time.monotonic() - started + logger.info( + "[sub-%d] delegate done: child=%s rounds=%d elapsed=%.1fs", + slot, child_id, rounds, elapsed, + ) + return {"text": self._subagent_report(child_id, slot_cwd, answer, rounds, elapsed)} + finally: + self._subagent_release(parent_session_id, slot) + + @staticmethod + def _subagent_report( + child_id: str, slot_cwd: Path, answer: str, rounds: int, elapsed: float, + ) -> str: + """What the parent's round receives: the child's answer, and how it got there.""" + if not answer: + answer = "(the child produced no final answer)" + elif len(answer) > _SUBAGENT_ANSWER_CHARS: + answer = ( + answer[:_SUBAGENT_ANSWER_CHARS] + + f"\n\n… (truncated — the answer ran to {len(answer)} characters, " + f"the first {_SUBAGENT_ANSWER_CHARS} are shown)" + ) + return ( + "The child session finished.\n" + f"- session: {child_id}\n" + f"- cwd: {slot_cwd}\n" + f"- rounds: {rounds}\n" + f"- elapsed: {elapsed:.1f}s\n\n" + "Its final answer, and nothing else it said along the way:\n\n" + f"{answer}" + ) + def _maybe_reflect_memory( self, session: Session, user_prompt: str, assistant_content: str ) -> None: @@ -7579,7 +7879,22 @@ async def _reflect(): except Exception: logger.debug("memory reflection failed", exc_info=True) - asyncio.create_task(_reflect()) + task = asyncio.create_task(_reflect()) + # Kept, so the one caller that must know when this write has landed can ask + # (issue #2451): a `subagent` call does not close its child's slot until the + # child's memory is on disk. The entry is dropped by its own completion, and + # only by that task — a later reflection for the same session replaces it, + # and the reference check keeps the newcomer from being removed by the older + # task's callback. + tasks = self._memory_task_table() + tasks[session.session_id] = task + + def _forget(done: "asyncio.Task", sid: str = session.session_id) -> None: + table = self._memory_task_table() + if table.get(sid) is done: + table.pop(sid, None) + + task.add_done_callback(_forget) _EXIT_RECORD_PATH = logs_dir() / "emrgd-exit.log" diff --git a/emrg/session.py b/emrg/session.py index 6654532a..bdd22254 100644 --- a/emrg/session.py +++ b/emrg/session.py @@ -380,6 +380,13 @@ def __init__(self, session_id: str, cwd: Path) -> None: # Canonical spellings only — the writer canonicalizes before storing, so # a reader never has to wonder which spelling was meant. self._sandbox_roots: list[str] = [] + # How many delegations deep this session is (issue #2451): 0 for a session + # a client or the scheduler started, parent + 1 for one a `subagent` call + # started. It lives in `meta.json` rather than in a daemon table because + # the *child* is the one that has to know its own depth — a child's + # `subagent` call reads this, and a daemon restart must not reset the + # count of a session that already exists on disk. + self._depth: int = 0 # Lazy-initialized memory store self._memory_store = None @@ -472,6 +479,11 @@ def load(cls, session_id: str, cwd: Path) -> Session: stored_roots = meta.get("sandbox_roots") if isinstance(stored_roots, list): session._sandbox_roots = [r for r in stored_roots if isinstance(r, str)] + stored_depth = meta.get("depth") + # `isinstance(..., int)` is not enough on its own: `True` is an int, and + # a JSON `true` in this field would then read as depth 1. + if isinstance(stored_depth, int) and not isinstance(stored_depth, bool): + session._depth = max(0, stored_depth) logger.info("session loaded: %s (%d messages)", session_id, session._message_count) return session @@ -522,6 +534,24 @@ def rename(self, title: str) -> None: self._save_meta_with_title(title) logger.info("session renamed: %s -> %s", self.session_id, title) + # ── Delegation depth ────────────────────────────────────── + + @property + def depth(self) -> int: + """How many `subagent` delegations deep this session is; 0 at the top.""" + return self._depth + + def set_depth(self, depth: int) -> None: + """Persist this session's delegation depth into ``meta.json``. + + The only writer of the key. Written only when non-zero, so a session that + was never delegated to keeps the meta shape it has always had — the same + reason the sandbox tier is written only when one was declared. + """ + self._depth = int(depth) + self._save_meta() + logger.info("session delegation depth set: %s -> %d", self.session_id, self._depth) + # ── Sandbox tier ────────────────────────────────────────── @property @@ -968,6 +998,10 @@ def _save_meta_with_title(self, title: str | None) -> None: meta["sandbox"] = self._sandbox if self._sandbox_roots: meta["sandbox_roots"] = list(self._sandbox_roots) + # Same rule as `sandbox`: a session nobody delegated to must not gain a + # key saying it is zero deep. + if self._depth: + meta["depth"] = self._depth if title is not None: meta["title"] = title else: diff --git a/emrg/tools/subagent_tool.py b/emrg/tools/subagent_tool.py new file mode 100644 index 00000000..c4b84556 --- /dev/null +++ b/emrg/tools/subagent_tool.py @@ -0,0 +1,177 @@ +"""SubagentTool — delegate one job to a child session and wait for its answer. + +Issue #2451. Delegation existed only as a page of instructions in a user-level +skill: the model hand-assembled the ``auth``/``task``/``delta``/``done`` frames, +chose its own session id, asked the daemon whether that session was busy, and +decided for itself whether to delete and reuse a slot. A page of instructions is +not a capability — one wrong step costs a whole round, and the tool loop already +has everything needed to run a second session for one turn. This exposes it. + +**Blocking on purpose.** The parent's round stands still while the child runs and +resumes when it finishes, times out or is cancelled. One child per round, no +parallelism: a fan-out would report back a pile of half-answers, which the tool +layer cannot merge and the parent has no budget to reconcile. + +**What the parent gets back is the child's final answer**, plus its id, its cwd, +its round count and its elapsed time. None of the child's intermediate narration +travels: measured while a hand-rolled delegate was in use, the concatenated +running commentary was shrapnel, not a report. + +The pieces a caller must not choose are injected by the daemon, not declared in +this schema: ``cwd`` and the slot are computed (`_parent_session_id` / +`_parent_cwd` ride the argument dict from ``_inject_tool_arguments``), because a +model that picks its own slot picks a wrong one. So this tool refuses a call that +arrives without them, which is what happens outside a daemon turn. + +The slot path is the child's **project cwd**, not its own session directory — the +child's directory is built from its cwd like any other session's +(``/.emrg/sessions/``), which is how the child's memory lands +inside the parent's session directory without a second project root. + +The rest of the mechanism — the nine slots, the least-recently-closed choice, the +depth limit, the sandbox inherited from the parent, and the close stamp taken +after the child's memory write — belongs to the daemon that owns the sessions and +is documented where it lives (``EmrgServer._run_subagent_tool``). +""" + +from __future__ import annotations + +from typing import Awaitable, Callable + +from emrg.server.tool_types import ToolDefinition, ToolResult +from emrg.tools.base import ToolExecutor + +#: The runner the daemon supplies: ``await runner(parent_session_id=…, +#: parent_cwd=…, prompt=…, timeout_s=…, intent=…)`` answering either +#: ``{"text": …}`` or ``{"error": …}``. +Runner = Callable[..., Awaitable[dict]] + +#: A default that is generous enough for a real job and short enough that a +#: wedged child cannot hold the parent's round for ever. +DEFAULT_TIMEOUT_S = 600 + + +class SubagentTool(ToolExecutor): + """Run one child session to completion and return its final answer.""" + + def __init__(self, runner: Runner) -> None: + self._runner = runner + + def definition(self) -> ToolDefinition: + return ToolDefinition( + name="subagent", + description=( + "Delegate one self-contained job to a separate child session, " + "and wait for it. Use it when the job is big enough that doing " + "it here would flood this conversation with material you do not " + "need afterwards — a wide search, a long reading, a bulk edit — " + "and when the job can be stated completely in one prompt. " + "The child gets its own session, its own context and its own " + "tools; the parent's round stands still until the child " + "finishes, times out or is cancelled, and only the child's " + "FINAL answer comes back (its intermediate narration does not). " + "The result carries the child's session id, its cwd, how many " + "rounds it took and how long it ran. One child per call: this " + "does not run children in parallel. The child inherits this " + "session's sandbox tier, and a child may delegate again, up to a " + "fixed depth. Write the prompt as a complete brief — the child " + "does not see this conversation." + ), + parameters={ + "type": "object", + "properties": { + "prompt": { + "type": "string", + "description": ( + "The complete brief for the child: what to do, where " + "to work, and what to report back. The child cannot " + "see this conversation, so state the goal and the " + "constraints in full." + ), + }, + "timeout_s": { + "type": "integer", + "minimum": 1, + "description": ( + "Whole seconds the child may run — at least 1, and a " + "value outside that is refused. The bound covers the " + "whole call, so it is also how long this tool waits for " + "a free slot. When it is reached the child's turn is " + "cancelled and an error is returned. " + f"Default {DEFAULT_TIMEOUT_S}." + ), + }, + "intent": { + "type": "string", + "description": ( + "One sentence on why this delegation is made and what " + "it should reach; it is logged with the call." + ), + }, + }, + "required": ["prompt"], + }, + ) + + async def execute(self, arguments: dict) -> ToolResult: + prompt = arguments.get("prompt") + if not isinstance(prompt, str) or not prompt.strip(): + return ToolResult( + name="subagent", + content=( + "subagent: `prompt` is required and must not be empty — it is " + "the child's whole brief, since the child cannot see this " + "conversation." + ), + error=True, + ) + + timeout_s = arguments.get("timeout_s") + if timeout_s is None: + timeout_s = DEFAULT_TIMEOUT_S + if isinstance(timeout_s, bool) or not isinstance(timeout_s, int): + return ToolResult( + name="subagent", + content=( + f"subagent: `timeout_s` must be a whole number of seconds, so " + f"got {timeout_s!r}. It is the bound on the child's run, and a " + "value from outside that domain is a caller error, not a " + "default." + ), + error=True, + ) + if timeout_s < 1: + return ToolResult( + name="subagent", + content=( + f"subagent: `timeout_s` must be at least 1, so got {timeout_s}. " + "A non-positive bound would cancel the child before its first " + "round." + ), + error=True, + ) + + parent_session_id = arguments.get("_parent_session_id") or "" + parent_cwd = arguments.get("_parent_cwd") or "" + if not parent_session_id or not parent_cwd: + return ToolResult( + name="subagent", + content=( + "subagent: this call carries no parent session, so there is " + "nothing to delegate from. The daemon injects the parent's " + "session and cwd into every call it dispatches; a call without " + "them did not come from a daemon turn." + ), + error=True, + ) + + outcome = await self._runner( + parent_session_id=str(parent_session_id), + parent_cwd=str(parent_cwd), + prompt=prompt, + timeout_s=int(timeout_s), + intent=str(arguments.get("intent") or ""), + ) + if outcome.get("error"): + return ToolResult(name="subagent", content=outcome["error"], error=True) + return ToolResult(name="subagent", content=outcome.get("text", "")) diff --git a/tests/test_subagent_tool.py b/tests/test_subagent_tool.py new file mode 100644 index 00000000..4ae7509b --- /dev/null +++ b/tests/test_subagent_tool.py @@ -0,0 +1,526 @@ +"""Delegation to a child session is a tool, not a page of instructions. + +Issue #2451. Delegation already existed as a user-level skill that told the model +how to hand-assemble the websocket protocol itself: pick a session id, ask the +daemon whether it was busy, decide alone whether to delete and reuse a slot. A +page of instructions is not a capability — one wrong step costs a whole round — +and the tool loop already has everything needed to run a second session. + +What is measured here is the mechanism behind the `subagent` tool, item by item, +because each item is a way the delegation can be wrong without looking wrong: + +1. the parent receives the child's answer and its facts, and a cut answer says it + was cut; +2. a child that outlives ``timeout_s`` is **cancelled**, not merely abandoned; +3. cancelling the parent's turn cancels the child and returns the slot; +4. nine slots are a real bound, and the next one taken is the one closed longest + ago — not whichever dictionary order offers first; +5. a reused slot is emptied first, and emptied through `Session.delete` so the + global index does not keep pointing at a directory that is gone; +6. "the child is closed" includes its memory write: the close stamp is taken + after that write has landed, or the next child of the slot clears a directory + the previous one is still writing into; +7. a session at the depth limit refuses **and says so**, rather than silently + doing nothing; +8. every delegation is one greppable line carrying its slot. + +No daemon is started, stopped or restarted here: the runner is called directly, +the LLM is a stub, and `_broadcast` goes nowhere. The parent and the files live +under `tmp_path`. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import tempfile +import time +from pathlib import Path + +import pytest + +from emrg.config import LlmConfig +from emrg.server import daemon as daemon_mod +from emrg.server.daemon import EmrgServer +from emrg.session import Session +from emrg.tools.subagent_tool import DEFAULT_TIMEOUT_S, SubagentTool + + +def _run(coro): + return asyncio.run(coro) + + +def _server(tmp_path, monkeypatch) -> EmrgServer: + """A server whose planted-fire markers and projects log stay under `tmp_path`. + + The loop stamps host state at round finalization; a suite run must not write + `~/.emrg/logs`, and a test that leaves the real marker behind changes what the + next real cycle reads. + """ + monkeypatch.setattr(daemon_mod, "_PLANTED_FIRE_MARKER_PATH", + tmp_path / "planted-fire-heartbeat") + monkeypatch.setattr(daemon_mod, "_PLANTED_FIRE_ROUND_COMPLETE_PATH", + tmp_path / "planted-fire-round-complete") + server = EmrgServer(LlmConfig(base_url="http://localhost", api_key="test")) + server._projects_log = Path(tempfile.mkdtemp()) / "projects.yml" + # `_broadcast` is left real: with no connections it sends nowhere, and it is the + # one place that records a turn's terminal frame. Stubbing it made every child + # turn report "ended without a terminal frame", which is the stub talking. + return server + + +def _stub_reply(server: EmrgServer, text: str) -> None: + """The child's turn answers `text` and stops.""" + + async def fake_stream(messages, tools=None): + yield {"content": text, "tool_calls": None, "finish_reason": "stop", + "usage": {"prompt_tokens": 5, "completion_tokens": 2}} + + server.llm.chat_stream = fake_stream + + async def fake_chat(messages, tools=None): + return {"content": "", "tool_calls": None} + + server.llm.chat = fake_chat + + +def _delegate(server, parent_id: str, parent_cwd: Path, *, prompt="do the job", + timeout_s=10): + return _run(server._run_subagent_tool( + parent_session_id=parent_id, parent_cwd=str(parent_cwd), + prompt=prompt, timeout_s=timeout_s, intent="a test delegation", + )) + + +def _report_field(report: str, name: str) -> str: + for line in report.splitlines(): + if line.startswith(f"- {name}: "): + return line[len(f"- {name}: "):] + raise AssertionError(f"the report carries no `{name}` line:\n{report}") + + +def _force_slot_choice(server, parent_id: str, slot: int) -> None: + """Make `slot` the one closed longest ago, so the next acquire must take it. + + The choice is by close time, so a test that needs a *particular* slot says so + through the same fact the production code reads — never by relying on + dictionary order, which the LRU rule exists to ignore. + """ + table = server._subagent_slot_closes(parent_id) + for n in table: + table[n] = time.time() + n + table[slot] = 0.0 + + +# ── 1. the parent gets the answer and the child's facts ──────────────────── + + +def test_the_parent_receives_the_childs_answer_and_its_facts(tmp_path, monkeypatch): + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "the child did the job and says so") + + outcome = _delegate(server, "parent-1", tmp_path) + + assert "error" not in outcome, outcome + report = outcome["text"] + assert "the child did the job and says so" in report, ( + "the child's final answer is the one thing the parent asked for; a report " + f"without it is not a delegation:\n{report}" + ) + child_id = _report_field(report, "session") + assert child_id.startswith("sub-parent-1-"), child_id + cwd = Path(_report_field(report, "cwd")) + assert cwd.is_dir(), f"the report names a cwd that does not exist: {cwd}" + assert str(tmp_path) in str(cwd), ( + "a child's cwd must sit inside the parent's session tree, or its memory is " + f"planted in a project root nobody asked about: {cwd}" + ) + assert _report_field(report, "rounds").isdigit() + assert float(_report_field(report, "elapsed").rstrip("s")) >= 0.0 + + +def test_a_cut_answer_says_it_was_cut(tmp_path, monkeypatch): + """A silently truncated report reads as a complete one — the marker is the point.""" + server = _server(tmp_path, monkeypatch) + # The subject is a fixed answer, far longer than any ceiling this tool would + # plausibly cut at. Deriving its length from the constant would make the test + # move with the thing it measures: raise the ceiling and both sides rise, so no + # arm could show it red (measured — that is exactly how the first version of + # this test survived `_SUBAGENT_ANSWER_CHARS = 60000`). + whole = 50_000 + _stub_reply(server, "x" * whole) + + outcome = _delegate(server, "parent-cut", tmp_path) + + assert "error" not in outcome, outcome + report = outcome["text"] + answer = report.split( + "Its final answer, and nothing else it said along the way:\n\n", 1 + )[-1] + kept = len(answer) - len(answer.lstrip("x")) + assert kept < whole, ( + f"the whole {whole}-character answer came back, so nothing was cut and this " + "test has no subject to measure" + ) + assert kept == daemon_mod._SUBAGENT_ANSWER_CHARS, ( + f"the ceiling kept {kept} characters rather than " + f"{daemon_mod._SUBAGENT_ANSWER_CHARS}" + ) + assert "truncated" in answer, ( + "the answer was cut and nothing said so; a caller cannot tell a short " + f"answer from a lost one:\n{answer[-300:]}" + ) + assert str(whole) in answer, ( + "the marker does not say how much was dropped, so a caller cannot tell a " + f"slightly long answer from a huge one:\n{answer[-300:]}" + ) + + +# ── 2. the timeout cancels the child ─────────────────────────────────────── + + +def test_a_child_that_outlives_the_timeout_is_cancelled(tmp_path, monkeypatch): + server = _server(tmp_path, monkeypatch) + cancelled = {"stream": False} + + async def hanging_stream(messages, tools=None): + try: + await asyncio.Event().wait() + yield {"content": "", "tool_calls": None, "finish_reason": "stop", + "usage": {}} + except asyncio.CancelledError: + cancelled["stream"] = True + raise + + server.llm.chat_stream = hanging_stream + + async def fake_chat(messages, tools=None): + return {"content": "", "tool_calls": None} + + server.llm.chat = fake_chat + + started = time.monotonic() + outcome = _delegate(server, "parent-timeout", tmp_path, timeout_s=1) + elapsed = time.monotonic() - started + + assert outcome.get("error"), outcome + assert "1s" in outcome["error"], outcome["error"] + assert elapsed < 10, f"the timeout did not bound the call: {elapsed:.1f}s" + assert cancelled["stream"], ( + "the child's stream was left running after the parent gave up; the child " + "must be cancelled, not abandoned" + ) + assert not any( + sid.startswith("sub-parent-timeout") and busy + for sid, busy in server._session_busy.items() + ), "a timed-out child is still marked busy" + + +# ── 3. cancelling the parent returns the slot at once ───────────────────── + + +def test_cancelling_the_parent_cancels_the_child_and_returns_the_slot( + tmp_path, monkeypatch, +): + server = _server(tmp_path, monkeypatch) + + async def hanging_stream(messages, tools=None): + await asyncio.Event().wait() + yield {"content": "", "tool_calls": None, "finish_reason": "stop", + "usage": {}} + + server.llm.chat_stream = hanging_stream + + async def fake_chat(messages, tools=None): + return {"content": "", "tool_calls": None} + + server.llm.chat = fake_chat + + async def scenario(): + task = asyncio.create_task(server._run_subagent_tool( + parent_session_id="parent-cancel", parent_cwd=str(tmp_path), + prompt="go", timeout_s=60, intent="cancel me", + )) + # Let it take a slot and start the child's turn. + for _ in range(200): + if server._subagent_busy.get("parent-cancel"): + break + await asyncio.sleep(0.01) + assert server._subagent_busy.get("parent-cancel"), "no slot was ever taken" + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + _run(scenario()) + + assert not server._subagent_busy.get("parent-cancel"), ( + "the slot is still held after the parent's turn was cancelled, so the next " + "delegation waits for a child nobody is running" + ) + assert not any( + sid.startswith("sub-parent-cancel") and busy + for sid, busy in server._session_busy.items() + ), "the child is still marked busy after its parent was cancelled" + + +# ── 4. nine slots, and the least recently closed one goes next ──────────── + + +def test_the_tenth_delegation_waits_and_takes_nothing(tmp_path, monkeypatch): + """Nine are the bound; a tenth waits, and the call's own timeout ends the wait.""" + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "unused") + server._subagent_busy["parent-full"] = set(range(1, EmrgServer.SUBAGENT_SLOTS + 1)) + + outcome = _delegate(server, "parent-full", tmp_path, timeout_s=1) + + assert outcome.get("error"), outcome + assert "did not finish" in outcome["error"] or "slot" in outcome["error"], ( + outcome["error"] + ) + child_root = tmp_path / ".emrg" / "sessions" / "parent-full" / "sub-agent" + assert not child_root.exists() or not any(child_root.iterdir()), ( + "a delegation that never got a slot still created a child" + ) + + +def test_the_next_slot_is_the_one_closed_longest_ago(tmp_path, monkeypatch): + server = _server(tmp_path, monkeypatch) + parent = "parent-lru" + taken = [server._subagent_acquire(parent) for _ in range(EmrgServer.SUBAGENT_SLOTS)] + assert sorted(taken) == list(range(1, EmrgServer.SUBAGENT_SLOTS + 1)), taken + with pytest.raises(LookupError): + server._subagent_acquire(parent) + + # Two slots come free, and the *later-numbered* one was closed first. + server._subagent_release(parent, 5) + server._subagent_release(parent, 7) + table = server._subagent_slot_closes(parent) + table[5] = 200.0 + table[7] = 100.0 + + assert server._subagent_acquire(parent) == 7, ( + "the slot closed longest ago must go next; taking the first free slot " + "instead makes reuse depend on dictionary order" + ) + + +# ── 5. a reused slot is emptied first ───────────────────────────────────── + + +def test_a_reused_slot_is_emptied_before_it_is_used(tmp_path, monkeypatch): + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "first run") + parent = "parent-reuse" + + _force_slot_choice(server, parent, 1) + first = _delegate(server, parent, tmp_path) + assert "error" not in first, first + child_id = _report_field(first["text"], "session") + assert child_id == "sub-parent-reuse-1", child_id + child_dir = Path(_report_field(first["text"], "cwd")) / ".emrg" / "sessions" / child_id + assert child_dir.is_dir(), child_dir + (child_dir / "stale-marker.txt").write_text("from the first child", encoding="utf-8") + old_history = len(child_dir.joinpath("history.jsonl").read_text(encoding="utf-8").splitlines()) + assert old_history > 0 + + _stub_reply(server, "second run") + _force_slot_choice(server, parent, 1) + second = _delegate(server, parent, tmp_path) + assert "error" not in second, second + assert _report_field(second["text"], "session") == child_id + + assert not (child_dir / "stale-marker.txt").exists(), ( + "the slot was reused without being emptied first, so the previous child's " + "files are still in the directory the new child is told is its own" + ) + fresh = child_dir.joinpath("history.jsonl").read_text(encoding="utf-8").splitlines() + assert len(fresh) < old_history + 1 or "from the first child" not in "".join(fresh), ( + "the previous child's history survived into the reused slot" + ) + + +# ── 6. the close stamp is taken after the child's memory write ──────────── + + +def test_the_slot_is_closed_only_after_the_childs_memory_was_written( + tmp_path, monkeypatch, +): + """Measured, not asserted in prose: the reflection is slow, and the stamp waits. + + The reflection is fire-and-forget, so the only thing that can order the two is + an explicit wait. Without it the close precedes the write, and the next child of + that slot clears a directory the previous one is still writing into. + """ + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "a reply long enough to be worth reflecting on") + parent = "parent-memory" + reflection = {"finished_at": None, "seen": 0} + + async def slow_reflection(messages, tools=None): + text = json.dumps(messages, ensure_ascii=False).lower() + if "memory" in text: + await asyncio.sleep(0.2) + reflection["finished_at"] = time.time() + reflection["seen"] += 1 + return {"content": "", "tool_calls": None} + + server.llm.chat = slow_reflection + + _force_slot_choice(server, parent, 1) + outcome = _delegate(server, parent, tmp_path) + + assert "error" not in outcome, outcome + assert reflection["seen"] > 0, ( + "the child never reflected, so this test has no subject — the premise of " + "the ordering is that a memory write was actually scheduled" + ) + closed_at = server._subagent_slot_closes(parent)[1] + assert closed_at >= reflection["finished_at"], ( + f"the slot was stamped closed at {closed_at} but the child's memory write " + f"only finished at {reflection['finished_at']}: closing before the write " + "means the next child of this slot can clear a directory still in use" + ) + + +# ── 7. the depth limit refuses, and says so ─────────────────────────────── + + +def test_a_session_at_the_depth_limit_refuses_and_says_so(tmp_path, monkeypatch): + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "unused") + parent = Session.create_with_id("parent-deep", tmp_path) + parent.set_depth(EmrgServer.SUBAGENT_MAX_DEPTH) + + outcome = _delegate(server, "parent-deep", tmp_path) + + assert outcome.get("error"), outcome + assert "refused" in outcome["error"], outcome["error"] + assert str(EmrgServer.SUBAGENT_MAX_DEPTH) in outcome["error"], ( + "the refusal must name the limit it hit, or a caller cannot tell a boundary " + f"from a bug: {outcome['error']}" + ) + child_root = tmp_path / ".emrg" / "sessions" / "parent-deep" / "sub-agent" + assert not child_root.exists() or not any(child_root.iterdir()), ( + "a refused delegation still started a child" + ) + + +def test_a_childs_depth_is_its_parents_plus_one(tmp_path, monkeypatch): + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "child answer") + parent = Session.create_with_id("parent-depth", tmp_path) + parent.set_depth(2) + + outcome = _delegate(server, "parent-depth", tmp_path) + + assert "error" not in outcome, outcome + child_id = _report_field(outcome["text"], "session") + child_cwd = Path(_report_field(outcome["text"], "cwd")) + reloaded = Session.load(child_id, child_cwd) + assert reloaded.depth == 3, ( + "the child's depth must be written into its own meta: the child is the one " + "that has to know how deep it is when it delegates again" + ) + + +# ── 7. the child runs at its parent's tier, not one it can choose ───────── + + +def test_the_child_runs_at_its_parents_sandbox_tier(tmp_path, monkeypatch): + """Authorization a caller can pick for the thing it authorizes is not a boundary. + + The tool takes no tier argument on purpose, so the only tier a child can have is + the parent's. Asserted on the child's own `meta.json` — the file a later turn of + that child reads — rather than on the object this call happened to hold. + """ + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "child answer") + parent = Session.create_with_id("parent-tier", tmp_path) + parent.set_sandbox("read-only") + + outcome = _delegate(server, "parent-tier", tmp_path) + + assert "error" not in outcome, outcome + child_id = _report_field(outcome["text"], "session") + child_cwd = Path(_report_field(outcome["text"], "cwd")) + reloaded = Session.load(child_id, child_cwd) + assert reloaded.sandbox == "read-only", ( + "the child is stored at a different tier than its parent; a child that does " + "not inherit is a child nobody confined, and a child that chooses is a " + "delegation that escapes its parent's boundary" + ) + + +# ── 8. one greppable line per delegation ────────────────────────────────── + + +def test_every_delegation_is_one_greppable_log_line(tmp_path, monkeypatch, caplog): + server = _server(tmp_path, monkeypatch) + _stub_reply(server, "logged") + + with caplog.at_level(logging.INFO, logger="emrg.server.daemon"): + outcome = _delegate(server, "parent-log", tmp_path) + + assert "error" not in outcome, outcome + lines = [r.getMessage() for r in caplog.records] + delegations = [m for m in lines if "delegate" in m] + assert delegations, ( + "no line names the delegation, so one delegation cannot be picked out of " + f"emrgd.log:\n" + "\n".join(lines[-8:]) + ) + for message in delegations: + assert "[sub-" in message, ( + f"a delegation line carries no slot label: {message!r}" + ) + assert any("delegate start" in m for m in delegations) + assert any("delegate done" in m for m in delegations) + + +# ── the tool's own contract ─────────────────────────────────────────────── + + +def test_the_tool_refuses_a_call_with_no_parent(): + """A call outside a daemon turn has no parent, and inventing one is worse.""" + tool = SubagentTool(lambda **_: None) # never called + + result = _run(tool.execute({"prompt": "do it"})) + + assert result.error + assert "parent session" in result.content + + +def test_the_tool_refuses_an_impossible_timeout(): + tool = SubagentTool(lambda **_: None) + + for bad in (0, -5, "soon", 2.5): + result = _run(tool.execute({ + "prompt": "do it", + "_parent_session_id": "p", + "_parent_cwd": "/tmp", + "timeout_s": bad, + })) + assert result.error, f"timeout_s={bad!r} was accepted" + assert "timeout_s" in result.content + + +def test_the_tool_passes_the_default_timeout_through(): + seen = {} + + async def runner(**kwargs): + seen.update(kwargs) + return {"text": "ok"} + + tool = SubagentTool(runner) + result = _run(tool.execute({ + "prompt": "do it", + "_parent_session_id": "p", + "_parent_cwd": "/tmp", + "intent": "because", + })) + + assert not result.error, result.content + assert seen["timeout_s"] == DEFAULT_TIMEOUT_S + assert seen["parent_session_id"] == "p" + assert seen["intent"] == "because" diff --git a/tests/test_tool_argument_domains.py b/tests/test_tool_argument_domains.py index 0627760b..82932826 100644 --- a/tests/test_tool_argument_domains.py +++ b/tests/test_tool_argument_domains.py @@ -46,6 +46,7 @@ grep_tool, pwsh_tool_v2, read_tool, + subagent_tool, submit_rant_tool, write_tool, ) @@ -207,6 +208,7 @@ def test_a_positive_shell_timeout_still_runs(tmp_path): (grep_tool.GrepTool(), "ignore_case", ("true or false",)), (bash_tool_v2.BashToolV2(), "timeout", ("positive",)), (pwsh_tool_v2.PwshToolV2(), "timeout", ("positive",)), + (subagent_tool.SubagentTool(lambda **kwargs: None), "timeout_s", ("at least 1",)), ] @@ -403,6 +405,9 @@ def test_a_description_claim_nothing_reads_is_the_class_this_file_closes(): submit_rant_tool.SubmitRantTool(), bash_tool_v2.BashToolV2(), pwsh_tool_v2.PwshToolV2(), + # The runner is never called: this sweep reads `definition()` only, and the tool + # is here so its declared parameters are asked about like every other tool's. + subagent_tool.SubagentTool(lambda **kwargs: None), ]