feat: progress thresholds on watch.agent; agent.completed carries name and answer

README.md

@@ -130,13 +130,14 @@ researcher = await rlm.agent.get("researcher")  # Recover by sibling name or ID
 Names are unique among siblings and remain reserved for the session, including after completion. Immutable IDs are shown alongside names. Metadata includes the parent ID, initial task, status, persistence flag, creation time, elapsed lifetime in seconds, session directory, and any failure. `list(recursive=True)` includes descendants, but only direct children can be retrieved as handles or controlled through the supervisor.
 
 ```python
-status = await researcher.wait(timeout=30)
-result = await researcher.result()
-if result is not None:
+result = await researcher.result(yield_after=60)
+if result.running:
+    print("still working, answer so far:", result.answer)
+else:
     print(result.answer)
 ```
 
-`result()` returns the latest successful `RLMResult`, including while a persistent agent is running again, or `None` before its first answer. Terminal failure or cancellation raises. Use `info()` or `wait()` for current activity; a retained answer does not mean a follow-up has finished. `wait()` returns current metadata after an outcome or its timeout (default 30 seconds, range 0–300). It waits inside the Python cell and uses the cell's normal execution timeout. Cancelling or timing out a wait does not cancel the agent.
+`result()` waits up to `yield_after` seconds (default 300) for the agent to finish and returns an `AgentResult` (`status`, `answer`, `usage`, `turns`, `session_dir`, `running`). `answer` is `None` while the agent is still working on its first answer; a persistent agent's latest answer stays available while it runs again. Terminal failure or cancellation raises. Use `info()` or `wait()` for current activity; a retained answer does not mean a follow-up has finished. `wait()` returns current metadata after an outcome or its timeout (default 30 seconds, range 0–300). It waits inside the Python cell and uses the cell's normal execution timeout. Cancelling or timing out a wait does not cancel the agent.
 
 `await researcher.cancel()` terminates the agent and its descendants and waits for cleanup. Ordinary agents release their kernels after answering. An agent spawned with `persistent=True` becomes idle after answering and retains its conversation and kernel; a parent instruction or new inbox event wakes it. Parent termination tears down all descendants, including persistent agents. Closing the ACP session tears down the tree. Cancelling an individual prompt or cell leaves its accepted children registered and recoverable.
 
@@ -156,7 +157,8 @@ await rlm.agent.send_to_parent("Found a missing permission check")
 # Inside its parent:
 for event in await rlm.inbox.list():
     report = await rlm.inbox.read(event["id"])
-    print(report["type"], report["content"])
+    if report["type"] == "agent.message":
+        print(report["content"]["name"], report["content"]["text"])
 ```
 
 Parent instructions are pushed into the child's conversation. Steering does not interrupt a running model request or tool. Queued messages wait until the child answers or calls the native `wait` tool. Both operations wake an idle persistent child; sending to a terminated child raises.

src/rlm/agent.py

@@ -9,7 +9,7 @@
 
 from rlm import broker
 from rlm.history import History, history
-from rlm.types import RLMResult
+from rlm.types import AgentResult
 
 
 @dataclass(frozen=True)
@@ -26,6 +26,7 @@ class AgentInfo:
     elapsed_seconds: float
     session_dir: Path
     error: str | None
+    turns: int = 0
     cleanup_error: str | None = None
 
     @classmethod
@@ -37,6 +38,7 @@ def from_payload(cls, payload: dict) -> AgentInfo:
 class AgentHandle:
     id: str
     session_dir: Path
+    name: str | None = None
 
     async def info(self) -> AgentInfo:
         """Read current metadata from the supervisor."""
@@ -48,12 +50,16 @@ async def history(self) -> History:
         """Read a fresh snapshot of this agent's local conversation history."""
         return await history(session_dir=self.session_dir)
 
-    async def result(self) -> RLMResult | None:
-        """Return the latest answer, even while running again; None before the first answer.
+    async def result(self, *, yield_after: float = 300) -> AgentResult:
+        """Wait up to yield_after seconds for the child to finish, then return its latest state.
 
-        Terminal failure/cancellation raises. Use info/wait to inspect current activity."""
-        payload = await broker.agent_request("agent.result", agent_id=self.id)
-        return broker.result_from_payload(payload) if payload is not None else None
+        .answer is None while the child is still working on its first answer (.running
+        is True); a persistent child's latest answer stays available while it runs
+        again. Terminal failure/cancellation raises."""
+        payload = await broker.agent_request(
+            "agent.result", agent_id=self.id, yield_after=yield_after
+        )
+        return broker.result_from_payload(payload)
 
     async def wait(self, timeout: float = 30) -> AgentInfo:
         """Wait up to timeout seconds for an outcome, then return current metadata.
@@ -103,15 +109,15 @@ async def spawn(
             "agent.spawn", task=task, name=name, persistent=persistent
         )
     )
-    return AgentHandle(info.id, info.session_dir)
+    return AgentHandle(info.id, info.session_dir, info.name)
 
 
 async def get(name_or_id: str) -> AgentHandle:
     """Recover a direct child's handle by sibling name or immutable ID."""
     info = AgentInfo.from_payload(
         await broker.agent_request("agent.get", name_or_id=name_or_id)
     )
-    return AgentHandle(info.id, info.session_dir)
+    return AgentHandle(info.id, info.session_dir, info.name)
 
 
 async def send_to_parent(message: str) -> str:

src/rlm/broker.py

@@ -12,9 +12,9 @@
 from typing import Annotated, Any, Literal
 
 from pydantic import ConfigDict, Field, TypeAdapter, ValidationError
-from typing_extensions import TypedDict
+from typing_extensions import NotRequired, TypedDict
 
-from rlm.types import RLMResult
+from rlm.types import AgentResult, RLMResult
 
 
 MAX_REQUEST_BYTES = 1024 * 1024
@@ -69,12 +69,21 @@ class BrokerAgentListRequest(TypedDict):
 
 class BrokerAgentHandleRequest(TypedDict):
     __pydantic_config__ = ConfigDict(extra="forbid", strict=True)
-    op: Literal["agent.info", "agent.result", "agent.cancel"]
+    op: Literal["agent.info", "agent.cancel"]
     capability: Annotated[str, Field(min_length=1)]
     scope_id: Annotated[str, Field(min_length=1)]
     agent_id: Annotated[str, Field(min_length=1)]
 
 
+class BrokerAgentResultRequest(TypedDict):
+    __pydantic_config__ = ConfigDict(extra="forbid", strict=True)
+    op: Literal["agent.result"]
+    capability: Annotated[str, Field(min_length=1)]
+    scope_id: Annotated[str, Field(min_length=1)]
+    agent_id: Annotated[str, Field(min_length=1)]
+    yield_after: Annotated[float, Field(ge=0)]
+
+
 class BrokerAgentWaitRequest(TypedDict):
     __pydantic_config__ = ConfigDict(extra="forbid", strict=True)
     op: Literal["agent.wait"]
@@ -185,6 +194,8 @@ class BrokerWatchAgentRequest(TypedDict):
     capability: Annotated[str, Field(min_length=1)]
     scope_id: Annotated[str, Field(min_length=1)]
     agent_id: Annotated[str, Field(min_length=1)]
+    every_turns: NotRequired[Annotated[int, Field(gt=0)] | None]
+    every_tokens: NotRequired[Annotated[int, Field(gt=0)] | None]
 
 
 class BrokerWatchJobRequest(TypedDict):
@@ -233,6 +244,7 @@ class BrokerSkillRequest(TypedDict):
     | BrokerAgentGetRequest
     | BrokerAgentListRequest
     | BrokerAgentHandleRequest
+    | BrokerAgentResultRequest
     | BrokerAgentWaitRequest
     | BrokerSkillRequest
     | BrokerAgentMessageRequest
@@ -271,7 +283,8 @@ class _BrokerFailure(TypedDict):
 
 
 _RESULT_ADAPTER = TypeAdapter(RLMResult)
-_RESULT_FIELDS = {"answer", "session_dir", "usage", "turns"}
+_AGENT_RESULT_ADAPTER = TypeAdapter(AgentResult)
+_RESULT_FIELDS = {"status", "answer", "session_dir", "usage", "turns"}
 _USAGE_FIELDS = {"prompt_tokens", "completion_tokens"}
 
 
@@ -286,14 +299,18 @@ def result_to_payload(result: RLMResult) -> dict[str, Any]:
     return _RESULT_ADAPTER.dump_python(result, mode="json")
 
 
-def result_from_payload(value: dict[str, Any]) -> RLMResult:
+def agent_result_to_payload(result: AgentResult) -> dict[str, Any]:
+    return _AGENT_RESULT_ADAPTER.dump_python(result, mode="json")
+
+
+def result_from_payload(value: dict[str, Any]) -> AgentResult:
     usage = value.get("usage")
     if set(value) != _RESULT_FIELDS or not isinstance(usage, dict):
         raise RuntimeError("invalid response from RLM supervisor")
     if set(usage) != _USAGE_FIELDS:
         raise RuntimeError("invalid response from RLM supervisor")
     try:
-        return _RESULT_ADAPTER.validate_python(value)
+        return _AGENT_RESULT_ADAPTER.validate_python(value)
     except ValidationError:
         raise RuntimeError("invalid response from RLM supervisor") from None
 

src/rlm/engine.py

@@ -1108,7 +1108,7 @@ async def _call_model(
             if checkpoint:
                 self._supervisor.record_usage(new_tokens)
             else:
-                self._supervisor.record_call(new_tokens)
+                self._supervisor.record_call(new_tokens, self._invocation_id)
         if not checkpoint:
             self._last_prompt_tokens = usage.prompt_tokens
             self._last_call_id = request_id

src/rlm/prompt.py

@@ -208,9 +208,9 @@
 `await rlm.inbox.list()` returns unread event dictionaries: ["id"], ["type"],
 ["sender_id"], ["created_at"], ["read"]; listed items carry no ["content"] and listing
 does not mark events read. `event = await rlm.inbox.read(event_id)` returns a dictionary
-with ["content"] and marks it read. For supervisor events (shell.completed, agent.completed,
-watch.*) content is a dictionary: index its keys, do not slice it; for agent.message it is
-the string a child sent. `list(unread_only=False)` includes read events; reads are repeatable.
+with ["content"] and marks it read. Content is always a dictionary: index its keys, do not
+slice it. An `agent.message` a child sent has ["agent_id"], ["name"] and ["text"].
+`list(unread_only=False)` includes read events; reads are repeatable.
 A read flag means retrieved, not completed or acted upon.
 
 Supervisor notifications show the unread count when it changes, plus occasional one-line hints about
@@ -261,19 +261,23 @@
 
 AGENT_PROMPT = """## Delegation
 `child = await rlm.agent.spawn(task, name="researcher", persistent=False)` returns
-an AgentHandle immediately. Give the child a self-contained task, relevant constraints,
+an AgentHandle (.id, .name, .session_dir) immediately. Give the child a self-contained task, relevant constraints,
 and an expected result. Names are unique among siblings and reserved for the session.
 `await rlm.agent.list()` returns AgentInfo objects with .id, .parent_id, .name, .task,
-.status, .persistent, .session_dir, and timing. `recursive=True` also lists descendants;
+.status, .persistent, .turns, .session_dir, and timing. `recursive=True` also lists descendants;
 only direct children can be controlled. Finished children remain discoverable. Recover a direct child with
 `await rlm.agent.get(name_or_id)`. Reassigning/deleting a Python handle does not stop it.
 
-`await child.info()` reads metadata. `await child.result()` returns an RLMResult
-(.answer, .usage, .turns, .session_dir), or None before its first answer. The latest answer
-remains available while a persistent child runs again; use info/wait for current activity.
-Terminal failure/cancellation raises.
-Child completion/failure posts `agent.completed` automatically;
-`event["content"]["agent_id"]` identifies the child and ["status"] gives its state. Inspect the event and recover the handle rather than assuming success.
+`await child.info()` reads metadata. `await child.result()` waits up to 300 s (or its
+yield_after=) for the child to finish and returns an AgentResult (.status, .answer, .usage,
+.turns, .session_dir, .running). If the child is still working on its first answer, .answer is
+None and .running is True: keep going and collect later, or from the `agent.completed` event.
+A persistent child's latest answer stays available while it runs again. Terminal
+failure/cancellation raises.
+Child completion/failure posts `agent.completed` automatically; its content has
+["agent_id"], ["name"], ["status"], ["turns"], ["error"] and ["answer"] (the last 4 KiB of
+the child's answer), so the event alone tells you what came back; `await child.result()`
+has the full answer. Inspect the event rather than assuming success.
 `await child.history()` returns a fresh history snapshot. `await child.cancel()` terminates
 that child and its descendants. Terminating a parent ends its whole subtree.
 
@@ -290,8 +294,12 @@
 `await rlm.watch.agent(child)` watches a direct child's conversation after complete
 assistant/tool steps, including final answers. Its `watch.agent` event content identifies
 the child via target and gives start:end indices for
-`(await child.history()).messages[start:end]`. It observes progress without waiting for an explicit
-report. Read history, then steer if needed; the subscription itself does not direct the child.
+`(await child.history()).messages[start:end]`. `await rlm.watch.agent(child, every_turns=10)` (and/or
+`every_tokens=50000`) instead posts a `watch.progress` event each time the child's own model
+calls or new tokens cross the next multiple, with ["turns"], ["tokens"], ["name"], ["status"]
+and the same start:end slice — the way to keep a long-running child in view: read its recent
+history, then `await child.steer("report what you have and stop")` if it should wrap up. The
+subscription itself does not direct the child.
 """
 
 HISTORY_PROMPT = """## Conversation history
@@ -392,7 +400,7 @@ def build_system_prompt(
             )
         if depth > 0:
             parts.append(
-                "Use `await rlm.agent.send_to_parent(message)` to put a report in your immediate parent's inbox; it returns an event ID. Your parent chooses when to read it. Parent instructions are pushed automatically: queued input at an answer/wait boundary, steering at the next model/tool boundary. You cannot steer your parent or message siblings. If you have children, their reports enter your own pull-based inbox in the same way."
+                "Your final answer is your deliverable: it reaches your parent as an `agent.completed` event and through `result()`. Use `await rlm.agent.send_to_parent(message)` for interim findings, blockers or questions, not to repeat the final report; it returns an event ID and your parent chooses when to read it. Parent instructions are pushed automatically: queued input at an answer/wait boundary, steering at the next model/tool boundary. You cannot steer your parent or message siblings. If you have children, their reports enter your own pull-based inbox in the same way."
             )
         else:
             parts.append(

src/rlm/subscriptions.py

@@ -28,6 +28,7 @@ class Subscription:
     timer: asyncio.TimerHandle | None = None
     task: asyncio.Task | None = None
     ready: asyncio.Event = field(default_factory=asyncio.Event)
+    thresholds: dict = field(default_factory=dict)
 
 
 class Subscriptions:
@@ -49,6 +50,7 @@ def register(
         cursor: int = 0,
         recursive: bool = False,
         completed: bool = False,
+        thresholds: dict | None = None,
     ) -> Subscription:
         if len(self.items) >= MAX_SUBSCRIPTIONS:
             raise RuntimeError("subscription limit reached")
@@ -70,10 +72,48 @@ def register(
             ),
             cursor,
         )
+        if thresholds:
+            sub.thresholds = dict(thresholds)
         self.record(sub)
         self.items[sub.info.id] = sub
         return sub
 
+    def progress(
+        self, target: str, turns: int, tokens: int, end: int, extra: dict
+    ) -> None:
+        """Fire `watch.progress` for every active progress subscription on `target` whose
+        turn or token threshold was crossed since it last fired; the content carries the
+        child's counters and the history slice start:end since the previous event."""
+        for sub in self.items.values():
+            if sub.info.status != "active" or sub.info.kind != "progress":
+                continue
+            if sub.info.target != target:
+                continue
+            th = sub.thresholds
+            fire = False
+            every = th.get("every_turns")
+            if every and turns >= th.get("next_turns", every):
+                th["next_turns"] = (turns // every + 1) * every
+                fire = True
+            every = th.get("every_tokens")
+            if every and tokens >= th.get("next_tokens", every):
+                th["next_tokens"] = (tokens // every + 1) * every
+                fire = True
+            if not fire:
+                continue
+            payload = {
+                "turns": turns,
+                "tokens": tokens,
+                "start": sub.cursor,
+                "end": end,
+                **extra,
+            }
+            sub.cursor = end
+            try:
+                self.publish(sub, "watch.progress", payload)
+            except Exception as exc:
+                self._fail(sub, str(exc))
+
     async def path(self, owner_id: str, target: Path, recursive: bool) -> Subscription:
         if not target.exists():
             raise FileNotFoundError(target)