From 05495fc477b5ed226976a659517e447fe2232ff1 Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Mon, 9 Nov 2026 00:23:41 +0800 Subject: [PATCH] feat(rlm): add bounded subagent controls and independent root sessions --- crates/core/prompts/rlm/repl_control.md | 4 +- crates/core/src/query/event.rs | 2 + crates/core/src/query/mod.rs | 8 + crates/core/src/rlm_prompts.rs | 32 +- crates/kernel/rlm-runtime/README.md | 10 +- crates/kernel/rlm-runtime/src/rlm/__init__.py | 31 +- crates/kernel/rlm-runtime/src/rlm/controls.py | 113 ++++ crates/kernel/rlm-runtime/src/rlm/harness.py | 7 +- .../kernel/rlm-runtime/tests/test_controls.py | 67 +++ crates/server/src/persistence.rs | 55 +- crates/server/src/runtime.rs | 8 + crates/server/src/runtime/agents.rs | 18 +- crates/server/src/runtime/goal_background.rs | 15 + crates/server/src/runtime/handlers/session.rs | 110 +++- crates/server/src/runtime/kernel_host.rs | 14 +- .../server/src/runtime/kernel_host_bridge.rs | 2 + .../src/runtime/kernel_host_controls.rs | 421 ++++++++++++++ .../server/src/runtime/kernel_host_session.rs | 329 +++++++++++ crates/server/src/runtime/session_cache.rs | 3 + crates/server/src/runtime/session_deletion.rs | 3 + .../src/runtime/turn_exec/approval_resume.rs | 1 + crates/server/src/runtime/turn_exec/query.rs | 3 + crates/server/src/runtime/turn_reservation.rs | 5 +- crates/server/tests/child_model_selection.rs | 130 +++++ .../server/tests/subagent_python_controls.rs | 513 ++++++++++++++++++ docs/rlm-orchestration.md | 93 ++++ 26 files changed, 1949 insertions(+), 48 deletions(-) create mode 100644 crates/kernel/rlm-runtime/src/rlm/controls.py create mode 100644 crates/kernel/rlm-runtime/tests/test_controls.py create mode 100644 crates/server/src/runtime/kernel_host_controls.rs create mode 100644 crates/server/src/runtime/kernel_host_session.rs create mode 100644 crates/server/tests/subagent_python_controls.rs create mode 100644 docs/rlm-orchestration.md diff --git a/crates/core/prompts/rlm/repl_control.md b/crates/core/prompts/rlm/repl_control.md index 7ce588fc..ca1bc3fb 100644 --- a/crates/core/prompts/rlm/repl_control.md +++ b/crates/core/prompts/rlm/repl_control.md @@ -18,7 +18,7 @@ Continual harness state is available as `rlm.harness` and `rlm.get_harness_state Terminology: continual harness names the persisted prompt, memory, skill, and subagent layer; RLM names the runtime, Python REPL kernel, and native call interface exposed to the model. -RLM-native call contract: installed Python skills are pre-imported modules. Read the matching SKILL.md and call its documented function, such as `await .(...)`; when a CLI exists, use ` ...` from shell. Continual harness skill entries are Python REPL skills with an explicit Python `reference` and `arguments` contract. Spawn a reusable delegation spec with `await rlm.spawn('sub-task', name='worker')` (host wire action `rlm.run`); admission returns a child handle immediately. Final assistant output is included in the child completion record; use an available messaging capability for timely updates and files for substantial artifacts. Results are never an `rlm.spawn()` return value. Do not invent non-native wrappers such as `call_skill(...)` or `run_subagent(...)`. Do not call `rlm.create_session` — daemon-backed top-level sessions are out of scope and denied by the host. +RLM-native call contract: installed Python skills are pre-imported modules. Read the matching SKILL.md and call its documented function, such as `await .(...)`; when a CLI exists, use ` ...` from shell. Continual harness skill entries are Python REPL skills with an explicit Python `reference` and `arguments` contract. Spawn a reusable delegation spec with `await rlm.spawn('sub-task', name='worker')` (host wire action `rlm.run`); admission returns a child handle immediately. Final assistant output is included in the child completion record; use an available messaging capability for timely updates and files for substantial artifacts. Results are never an `rlm.spawn()` return value. Do not invent non-native wrappers such as `call_skill(...)` or `run_subagent(...)`. Durable Build roots can call `await rlm.create_session('task', name='independent')` for a separate visible root in the same working directory. It inherits permissions/model settings, has its own journal and no parent lifecycle, and is not addressable by collect/delete_subagent. Child, ephemeral, and Plan callers are rejected. Child model selection: `await rlm.find_models(query='', limit=8)` lists enabled, configured, routable models with `provider`, `id`, `name`, `selector`, and `thinking_levels`; it does not probe live health or credits. Use an exact returned selector with `await rlm.spawn('sub-task', name='worker', model=chosen.selector, thinking='high')`, choosing a supported thinking level. Model IDs may contain slashes. Omitted model/thinking (or `None`) inherit the immediate parent's values. Invalid selectors or unsupported explicit or inherited thinking fail admission without silent fallback; when changing models, check its supported levels. `limit` is an integer from 1 to 100. Catalog descriptors contain no credentials. @@ -26,3 +26,5 @@ RLM host APIs are called from Python through `ipython`, not as separate model to - `await rlm.update_plan([{"step": "Inspect", "status": "in_progress"}])` publishes a session plan during execution. Plan Mode should keep its draft in Python and present the final `` instead. - `await rlm.request_user_input([{"id": "choice", "header": "Choice", "question": "Which option?", "options": [{"label": "A", "description": "..."}, {"label": "B", "description": "..."}]}])` uses the session's question UI. - `await rlm.web_search("query")` and `await rlm.web_fetch("https://...")` use Rust host operations; `rlm.web_search` requires local search configuration. When provider-hosted web tools are configured, use their provider-native capability directly; those are separate from the regular function-tool list. + +Python orchestration controls: `await rlm.collect(targets=None, timeout_ms=0)` snapshots direct children; a positive timeout (up to 300000 ms) waits only until completion, recovery is required, or the deadline. Timeout returns partial snapshots and never cancels work. Results retain status, settled, answer_text, answer_preview, answer_truncated, error, progress_note, and requires_resume. A saved unfinished turn with no live execution reports waiting_for_resume; explicitly recover it rather than replaying blindly. `await rlm.progress_note('running tests')` lets children publish non-steering progress, at most 512 UTF-16 units and one accepted note per 10 seconds. `await rlm.end_turn()` finishes normally after the current tool batch is committed, without an extra model call or canceling children. The rest of the cell and other calls in that batch still run. diff --git a/crates/core/src/query/event.rs b/crates/core/src/query/event.rs index 9b0a8600..826cdb9b 100644 --- a/crates/core/src/query/event.rs +++ b/crates/core/src/query/event.rs @@ -141,6 +141,8 @@ pub struct QueryOptions { /// Acknowledged journal for durable sessions; absent for ephemeral callers. pub journal: Option>, pub cancel_token: Option, + /// Graceful host request: finish after the current durable tool batch. + pub end_turn_requested: Option>, /// Optional provider used only for compaction summaries. Servers use this /// seam to attach Compaction metering without misclassifying the main /// streaming query as compaction overhead. diff --git a/crates/core/src/query/mod.rs b/crates/core/src/query/mod.rs index 16430fcc..9d177d44 100644 --- a/crates/core/src/query/mod.rs +++ b/crates/core/src/query/mod.rs @@ -1323,6 +1323,14 @@ pub async fn query( session.end_turn(); return Err(AgentError::Aborted); } + if options + .end_turn_requested + .as_ref() + .is_some_and(|requested| requested.load(std::sync::atomic::Ordering::Acquire)) + { + session.end_turn(); + return Ok(()); + } } } diff --git a/crates/core/src/rlm_prompts.rs b/crates/core/src/rlm_prompts.rs index 25681931..05a46674 100644 --- a/crates/core/src/rlm_prompts.rs +++ b/crates/core/src/rlm_prompts.rs @@ -1,7 +1,7 @@ //! Immutable RLM system-prompt doctrine for Devo sessions. //! //! Kernel-surface RLM prompt set (reference-agent design, Devo branding) with explicit -//! denial of daemon `rlm.create_session`. Catalog skill progressive disclosure +//! bounded subagent and independent-root controls. Catalog skill progressive disclosure //! uses XML (`format_catalog_skills_xml`); the markdown skill body from //! `devo_skills::render_available_skills_body` remains for non-RLM surfaces. @@ -241,12 +241,14 @@ pub fn build_rlm_base_prompt(options: &RlmPromptOptions) -> String { ); } - // Explicit deny: never teach daemon create_session (depth-0 path is out of scope). - parts.push(String::new()); - parts.push( - "Do not call `rlm.create_session`. Daemon-backed top-level session creation is denied; use `await rlm.spawn(...)` only for child agents." - .to_string(), - ); + if has_ipython { + parts.push(String::new()); + parts.push(if options.depth == 0 { + "Durable Build roots may call `await rlm.create_session('task', name='independent')` to create a separately visible root in the same workspace. It inherits safety and model settings but has its own journal, no parent linkage, and no collect/child lifecycle. Plan and ephemeral callers are rejected." + } else { + "Child sessions cannot call `rlm.create_session`; use `await rlm.spawn(...)` for direct descendants." + }.to_string()); + } if options.allow_recursion && has_ipython { parts.push(String::new()); @@ -259,7 +261,7 @@ pub fn build_rlm_base_prompt(options: &RlmPromptOptions) -> String { .to_string(), ); parts.push( - "Children inherit your model and thinking level. This host does not support `model` or `thinking` overrides on `rlm.spawn`; omit both options." + "Children inherit your model and thinking level by default. Discover exact model selectors and supported thinking levels with `await rlm.find_models()`, then optionally pass `model=chosen.selector` and `thinking` to `rlm.spawn`; unsupported selections fail without fallback." .to_string(), ); if has_agent_observe { @@ -290,12 +292,12 @@ pub fn build_rlm_base_prompt(options: &RlmPromptOptions) -> String { ); } else { parts.push( - "Inspect files a child wrote when you need to collect its work without an observation capability." + "Use `await rlm.collect(child, timeout_ms=0)` to inspect retained child results without an observation capability, or read the files the child wrote." .to_string(), ); } parts.push( - "Spawn independent children in separate calls and end your turn instead of awaiting completion. Multiple replies may arrive over multiple turns. Delete a direct child explicitly with `await rlm.delete_subagent(child)` when it is no longer needed." + "Use `await rlm.collect(children, timeout_ms=1000)` for a short bounded wait: timeout returns partial snapshots without cancellation. Check `settled`, `requires_resume`, and `answer_truncated`; saved unfinished work requires explicit recovery, not automatic replay. Children can report `await rlm.progress_note('running tests')` without steering the parent. `await rlm.end_turn()` ends normally after the current tool batch while children continue. Delete a direct child explicitly with `await rlm.delete_subagent(child)` when it is no longer needed." .to_string(), ); } @@ -424,7 +426,7 @@ mod tests { use super::*; - /// Verifies: root RLM base prompt carries Devo doctrine strings and denies create_session. + /// Root prompt teaches the supported controls and their safety boundaries. #[test] fn build_rlm_base_prompt_contains_key_doctrine() { let mut options = RlmPromptOptions::root("/repo", "/tmp/messages.jsonl"); @@ -441,7 +443,11 @@ mod tests { assert!(prompt.contains("rlm.spawn")); assert!(prompt.contains("rlm.harness")); assert!(prompt.contains("rlm.create_session")); - assert!(prompt.contains("denied")); + assert!(prompt.contains("Durable Build roots")); + assert!(prompt.contains("rlm.collect")); + assert!(prompt.contains("rlm.progress_note")); + assert!(prompt.contains("rlm.end_turn")); + assert!(!prompt.contains("Daemon-backed top-level session creation is denied")); assert!(!prompt.contains("daemon-backed depth-0 session")); assert!(prompt.contains("Recursive agent depth: 0")); assert!(prompt.contains("await refine.run()")); @@ -506,7 +512,7 @@ mod tests { assert!(prompt.contains("Kernel snapshots are separate")); assert!(prompt.contains("`rlm.shell()` defaults to the session workspace")); assert!(prompt.contains("Bash on Unix and PowerShell on Windows")); - assert!(prompt.contains("does not support `model` or `thinking` overrides")); + assert!(prompt.contains("unsupported selections fail without fallback")); assert!(prompt.contains("never relax relative to the parent") || options.depth == 0); assert!( prompt.contains("Never use direct Python process or filesystem APIs to bypass") diff --git a/crates/kernel/rlm-runtime/README.md b/crates/kernel/rlm-runtime/README.md index 220919f8..01ea8232 100644 --- a/crates/kernel/rlm-runtime/README.md +++ b/crates/kernel/rlm-runtime/README.md @@ -167,7 +167,7 @@ The main groups are: | `rlm.web_search`, `rlm.web_fetch` | Host-mediated network access. | | `rlm.update_plan`, `rlm.request_user_input` | Session UI integration. | | `rlm.read`, `rlm.write`, `rlm.read_once`, `rlm.write_once` | Direct file I/O, with host-mediated fallback on OS denial. | -| `rlm.spawn`, `rlm.create_session`, `rlm.find_models`, `rlm.list_subagents`, `rlm.delete_subagent` | Recursive subagent control. | +| `rlm.spawn`, `rlm.create_session`, `rlm.find_models`, `rlm.list_subagents`, `rlm.delete_subagent`, `rlm.collect`, `rlm.progress_note`, `rlm.end_turn` | Recursive subagent control. | | `rlm.harness`, `rlm.get_harness_state` | Cached reads and async host-owned harness CRUD. | | `rlm.emit` | Ship a MIME-keyed display event. | @@ -190,8 +190,12 @@ if chosen is not None: ``` `rlm.spawn` returns when the child is admitted, not when it finishes. The child -runs as a separate session; results arrive through `agent_message` replies or -files, usually on a later turn. +runs as a separate session. Use `await rlm.collect(reviewer, timeout_ms=10000)` +for retained typed results; timeout returns partial snapshots without cancellation. +Children can report `await rlm.progress_note("running tests")` without steering the +parent. `await rlm.end_turn()` finishes normally after the current tool batch. +`create_session` creates a separate durable root in the same workspace. See +[orchestration semantics](../../../docs/rlm-orchestration.md) for limits and recovery. Omitted `model` and `thinking` (or `None`) inherit the immediate parent's values. An unsupported inherited thinking level fails just like an explicit unsupported diff --git a/crates/kernel/rlm-runtime/src/rlm/__init__.py b/crates/kernel/rlm-runtime/src/rlm/__init__.py index 9fff1294..806e48ff 100644 --- a/crates/kernel/rlm-runtime/src/rlm/__init__.py +++ b/crates/kernel/rlm-runtime/src/rlm/__init__.py @@ -50,6 +50,7 @@ class RLMSubagent: session_name: str session_dir: Path status: str + progress_note: str | None = None def _spawn_handle_from_payload(payload: Any) -> RLMSpawnHandle: @@ -327,13 +328,23 @@ async def create_session( thinking: str | None = None, cwd: str | None = None, ) -> RLMCreateSessionHandle: - """Create and prompt a resident depth-0 daemon session. + """Create and prompt an independent, durable root in the same workspace. - Only daemon-backed depth-0 sessions support this operation. The optional - arguments set the session name, model, thinking level, and working directory. + Requires a durable root in Build mode. Model and safety settings inherit; + an explicit model uses an exact selector from find_models. The result is + visible in session/list and has its own journal, with no parent lifecycle. + An explicit cwd must resolve to the current working directory. """ if not isinstance(prompt, str): raise TypeError(f"prompt must be str, got {type(prompt).__name__}") + if not prompt.strip(): + raise ValueError("prompt must be a non-empty string") + for label, value in (("name", name), ("model", model), ("thinking", thinking), ("cwd", cwd)): + if value is not None: + if not isinstance(value, str): + raise TypeError(f"{label} must be str") + if not value.strip() or value.strip() != value: + raise ValueError(f"{label} must be a nonempty string without surrounding whitespace") kwargs: dict[str, Any] = {} if name is not None: kwargs["name"] = name @@ -397,6 +408,7 @@ def _subagent_from_payload(payload: Any, operation: str = "rlm.list_subagents") session_name=session_name, session_dir=Path(session_dir), status=status, + progress_note=payload.get("progress_note"), ) @@ -484,6 +496,15 @@ async def update_plan( async def request_user_input(self, questions: list[dict[str, Any]]) -> dict[str, Any]: return await request_user_input(questions) + async def collect(self, targets=None, *, timeout_ms: int = 0) -> list[RLMChildResult]: + return await collect(targets, timeout_ms=timeout_ms) + + async def progress_note(self, message: str) -> RLMProgressNoteResult: + return await progress_note(message) + + async def end_turn(self) -> dict[str, Any]: + return await end_turn() + async def list_subagents(self) -> list[RLMSubagent]: return await list_subagents() @@ -556,3 +577,7 @@ def __getattr__(name: str) -> Any: # noqa: D401 - module-level lazy attr hook if name == "run": raise AttributeError(_RENAMED_RUN_MESSAGE) raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + +# Imported after handle definitions to keep the orchestration parser independent. +from .controls import RLMChildResult, RLMProgressNoteResult, collect, progress_note, end_turn +__all__ += ["RLMChildResult", "RLMProgressNoteResult", "collect", "progress_note", "end_turn"] diff --git a/crates/kernel/rlm-runtime/src/rlm/controls.py b/crates/kernel/rlm-runtime/src/rlm/controls.py new file mode 100644 index 00000000..53ee5ce2 --- /dev/null +++ b/crates/kernel/rlm-runtime/src/rlm/controls.py @@ -0,0 +1,113 @@ +"""Typed, bounded Python orchestration over the host's session lifecycle.""" +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +@dataclass(frozen=True) +class RLMChildResult: + rlm_child_id: str + session_name: str + session_dir: Path | None + status: str + settled: bool + answer_text: str | None = None + answer_preview: str | None = None + error: str | None = None + progress_note: str | None = None + answer_truncated: bool = False + requires_resume: bool = False + + +@dataclass(frozen=True) +class RLMProgressNoteResult: + accepted: bool + retry_after_ms: int | None = None + + +async def collect(targets=None, *, timeout_ms: int = 0) -> list[RLMChildResult]: + """Snapshot direct children, waiting at most 300000 ms for all to settle. + + Accepts a name, spawn handle, subagent row, or a list/tuple of these. + None/[] snapshots all direct children at admission. Timeout returns partial + snapshots, never cancels children or consumes their retained results. + """ + from . import RLMSpawnHandle, RLMSubagent, host_request + if isinstance(timeout_ms, bool) or not isinstance(timeout_ms, int): + raise TypeError("timeout_ms must be an int") + if not 0 <= timeout_ms <= 300000: + raise ValueError("timeout_ms must be between 0 and 300000") + if targets is None: + targets = [] + elif isinstance(targets, (str, RLMSpawnHandle, RLMSubagent)): + targets = [targets] + elif not isinstance(targets, (list, tuple)): + raise TypeError("targets must be a target or a list/tuple of targets") + selectors = [] + for target in targets: + if isinstance(target, (RLMSpawnHandle, RLMSubagent)): + target = target.rlm_child_id + if not isinstance(target, str): + raise TypeError("each target must be a name, spawn handle, or subagent") + if not target.strip(): + raise ValueError("target must not be empty") + selectors.append(target.strip()) + payload = await host_request("rlm.collect", {"targets": selectors, "timeout_ms": timeout_ms}) + if not isinstance(payload, dict): + raise RuntimeError("rlm.collect returned an invalid payload") + entries = payload.get("results") + if not isinstance(entries, list): + raise RuntimeError("rlm.collect returned an invalid results list") + results = [] + for entry in entries: + if not isinstance(entry, dict): + raise RuntimeError("rlm.collect returned an invalid entry") + if any(not isinstance(entry.get(k), str) or not entry[k] for k in ("rlm_child_id", "session_name", "status")): + raise RuntimeError("rlm.collect entry is missing identity/status") + if not isinstance(entry.get("settled"), bool): + raise RuntimeError("rlm.collect entry has invalid settled flag") + for key in ("answer_truncated", "requires_resume"): + if not isinstance(entry.get(key, False), bool): + raise RuntimeError(f"rlm.collect entry has invalid {key} flag") + for key in ("session_dir", "answer_text", "answer_preview", "error", "progress_note"): + if entry.get(key) is not None and not isinstance(entry[key], str): + raise RuntimeError(f"rlm.collect entry has invalid {key}") + results.append(RLMChildResult( + rlm_child_id=entry["rlm_child_id"], session_name=entry["session_name"], + status=entry["status"], settled=entry["settled"], + session_dir=Path(entry["session_dir"]) if entry.get("session_dir") else None, + answer_text=entry.get("answer_text"), answer_preview=entry.get("answer_preview"), + error=entry.get("error"), progress_note=entry.get("progress_note"), + answer_truncated=entry.get("answer_truncated", False), + requires_resume=entry.get("requires_resume", False), + )) + return results + + +async def progress_note(message: str) -> RLMProgressNoteResult: + """Update a child's progress without steering its parent; throttled to 10s.""" + from . import host_request + if not isinstance(message, str): + raise TypeError("message must be str") + message = message.strip() + if not message or len(message.encode("utf-16-le", "surrogatepass")) // 2 > 512: + raise ValueError("message must contain 1 to 512 UTF-16 code units") + payload = await host_request("rlm.progress.note", {"message": message}) + if not isinstance(payload, dict): + raise RuntimeError("rlm.progress.note returned an invalid payload") + accepted, retry = payload.get("accepted"), payload.get("retry_after_ms") + if not isinstance(accepted, bool) or (retry is not None and (isinstance(retry, bool) or not isinstance(retry, int) or retry < 0)): + raise RuntimeError("rlm.progress.note returned an invalid result") + return RLMProgressNoteResult(accepted, retry) + + +async def end_turn() -> dict[str, Any]: + """Request normal completion after this tool batch, without canceling children. + + The remainder of this cell and other calls in the batch still execute. No + additional model response is requested. Idle/obsolete host bridges are rejected. + """ + from . import host_request + return await host_request("rlm.end_turn", {}) diff --git a/crates/kernel/rlm-runtime/src/rlm/harness.py b/crates/kernel/rlm-runtime/src/rlm/harness.py index 5f5d629a..e53ac0df 100644 --- a/crates/kernel/rlm-runtime/src/rlm/harness.py +++ b/crates/kernel/rlm-runtime/src/rlm/harness.py @@ -633,7 +633,12 @@ def overview(self, *, max_entries_per_kind: int = 20, global_: bool = False, **k "and thinking_levels, not credentials or live health guarantees. Pass model=chosen.selector and a supported " "thinking level to spawn; omitted options inherit the immediate parent. Invalid selectors or unsupported " "explicit/inherited thinking fail without fallback. The handle contains admission metadata " - "and never the child's answer. Results arrive through child completion, explicit agent_message replies or " + "and never the child's answer. Use await rlm.collect(handle, timeout_ms=10000) for retained typed results " + "(0 is nonblocking, timeout returns partial snapshots, inspect settled). Children can call " + "await rlm.progress_note('running tests') without steering the parent. await rlm.end_turn() completes " + "after the current tool batch without canceling children. Durable Build roots can use " + "await rlm.create_session('task', name='independent') for a separate visible root in the same workspace. " + "Results also arrive through child completion, explicit agent_message replies or " "files; children reply with await agent_message.send(message, receiver_role='parent'). Use " "await rlm.list_subagents() to recover direct children. A recovered child uses child.session_name " "(the spawn handle uses handle.name); send await agent_message.send(..., " diff --git a/crates/kernel/rlm-runtime/tests/test_controls.py b/crates/kernel/rlm-runtime/tests/test_controls.py new file mode 100644 index 00000000..2f3ac0e7 --- /dev/null +++ b/crates/kernel/rlm-runtime/tests/test_controls.py @@ -0,0 +1,67 @@ +import unittest +from pathlib import Path +from unittest.mock import AsyncMock, patch +import rlm + + +class ControlsTests(unittest.IsolatedAsyncioTestCase): + async def test_collect_mixed_targets_and_retained_partial_result(self): + entry = dict(rlm_child_id="id", session_name="worker", session_dir="artifacts", status="running", settled=False, progress_note="testing") + host = AsyncMock(return_value={"results": [entry]}) + handle = rlm.RLMSpawnHandle("id", "worker", Path("artifacts"), "model") + with patch.object(rlm, "host_request", host): + result = await rlm.rlm.collect([handle, "worker"], timeout_ms=25) + self.assertEqual(result, [rlm.RLMChildResult("id", "worker", Path("artifacts"), "running", False, progress_note="testing")]) + host.assert_awaited_once_with("rlm.collect", {"targets": ["id", "worker"], "timeout_ms": 25}) + + async def test_invalid_wait_and_target_do_not_call_host(self): + host = AsyncMock() + with patch.object(rlm, "host_request", host): + for timeout in (True, None, "1", 1.0): + with self.assertRaises(TypeError): + await rlm.collect(timeout_ms=timeout) + for timeout in (-1, 300001): + with self.assertRaises(ValueError): + await rlm.collect(timeout_ms=timeout) + for targets in (False, [False], {}): + with self.assertRaises(TypeError): + await rlm.collect(targets) + with self.assertRaises(ValueError): + await rlm.collect([" "]) + host.assert_not_awaited() + + async def test_progress_utf16_boundary_and_throttle(self): + host = AsyncMock(return_value={"accepted": False, "retry_after_ms": 9000}) + with patch.object(rlm, "host_request", host): + self.assertEqual(await rlm.rlm.progress_note(" " + "🎉" * 256 + " "), rlm.RLMProgressNoteResult(False, 9000)) + with self.assertRaises(ValueError): + await rlm.progress_note("🎉" * 257) + with self.assertRaises(ValueError): + await rlm.progress_note(" ") + host.assert_awaited_once_with("rlm.progress.note", {"message": "🎉" * 256}) + + async def test_end_turn_and_host_failure(self): + host = AsyncMock(return_value={"ended": True}) + with patch.object(rlm, "host_request", host): + self.assertEqual(await rlm.rlm.end_turn(), {"ended": True}) + host.assert_awaited_once_with("rlm.end_turn", {}) + with patch.object(rlm, "host_request", AsyncMock(side_effect=RuntimeError("inactive"))): + with self.assertRaisesRegex(RuntimeError, "inactive"): + await rlm.end_turn() + + async def test_invalid_host_payload(self): + for payload in ({}, {"results": [None]}, {"results": [{}]}): + with patch.object(rlm, "host_request", AsyncMock(return_value=payload)): + with self.assertRaises(RuntimeError): + await rlm.collect() + + async def test_create_session_validates_options_before_host(self): + host = AsyncMock() + with patch.object(rlm, "host_request", host): + for kwargs in ({"prompt": " "}, {"prompt": "task", "cwd": " "}, {"prompt": "task", "thinking": " high "}): + with self.assertRaises(ValueError): + await rlm.create_session(**kwargs) + for field in ("name", "model", "thinking", "cwd"): + with self.assertRaises(TypeError): + await rlm.create_session("task", **{field: True}) + host.assert_not_awaited() diff --git a/crates/server/src/persistence.rs b/crates/server/src/persistence.rs index 45ba781e..446056bb 100644 --- a/crates/server/src/persistence.rs +++ b/crates/server/src/persistence.rs @@ -382,6 +382,47 @@ impl RolloutStore { ) } + /// Persist inherited settings before the first metadata header in one locked + /// create batch. A crash leaving only a prefix cannot expose a root with + /// relaxed defaults: either its settings precede its header or no header exists. + pub(crate) fn append_seeded_session_meta_create_at( + &self, + rollout_path: &Path, + session: &devo_protocol::native::session::Session, + extras: devo_core::SessionPersistenceExtras, + settings: &[(SessionSettingsField, serde_json::Value)], + ) -> Result<()> { + let session_id = session.id; + self.append_rollout_lines_with( + rollout_path, + session_id, + JournalWriteMode::CreateIfMissing, + |state| { + let mut lines = Vec::with_capacity(settings.len() + 1); + let session_state = state.session_state_mut(session_id); + for (field, value) in settings { + let epoch = session_state.allocate_settings_epoch(); + lines.push(devo_core::settings_line( + Utc::now(), + session_id, + *field, + value.clone(), + epoch, + )); + } + let merged = + merge_session_extras(state.session_extras.get(&session_id), Some(&extras)); + state.session_extras.insert(session_id, merged.clone()); + lines.push(devo_core::session_line( + session.clone(), + Some(merged), + Utc::now(), + )); + Ok(lines) + }, + ) + } + fn append_session_meta_with_mode( &self, rollout_path: &Path, @@ -2964,10 +3005,22 @@ impl ReplayState { native.active_turn_id = None; native.queued_count = 0; native.last_activity_at = last_activity_at; + // The actor's next turn resolves its selection from these strings. + // Keeping only the base binding silently drops a named variant after + // restart even when the display-only `variant` field still says fast. + let persisted_selection = turn_config.variant.as_ref().map(|variant| { + let base = turn_config + .model_binding_id + .as_ref() + .unwrap_or(&turn_config.model.slug); + format!("{base}/{variant}") + }); native.model.provider = turn_config .model_binding_id + .map(|binding| persisted_selection.clone().unwrap_or(binding)) .unwrap_or_else(|| "unknown".into()); - native.model.model = turn_config.model.slug.clone(); + native.model.model = persisted_selection.unwrap_or_else(|| turn_config.model.slug.clone()); + native.model.variant = turn_config.variant.clone(); native.model.reasoning_effort = summary_reasoning_effort; native.settings.reasoning_effort = summary_reasoning_effort_selection.clone(); native.settings.mode = Some(match core_session.collaboration_mode { diff --git a/crates/server/src/runtime.rs b/crates/server/src/runtime.rs index 5a7e58fd..289b0877 100644 --- a/crates/server/src/runtime.rs +++ b/crates/server/src/runtime.rs @@ -120,8 +120,10 @@ mod items; mod kernel_host; mod kernel_host_agent_message; mod kernel_host_bridge; +mod kernel_host_controls; mod kernel_host_heartbeat; mod kernel_host_models; +mod kernel_host_session; mod kernel_idle_bash; mod lifecycle; mod mcp; @@ -204,6 +206,9 @@ pub struct ServerRuntime { background_work: std::sync::Mutex, /// Per-root-session agent registries for subagent coordination. agent_registries: Mutex>, + rlm_lazy_children: Mutex>, + rlm_progress_notes: Mutex>, + rlm_child_changed: tokio::sync::Notify, /// Per-session inboxes used by agent tools to exchange ordered messages. agent_mailboxes: Mutex>, /// Per-parent child-output buffers used by wait_agent polling. @@ -446,6 +451,9 @@ impl ServerRuntime { goal_stores: Mutex::new(HashMap::new()), background_work: std::sync::Mutex::new(Default::default()), agent_registries: Mutex::new(HashMap::new()), + rlm_lazy_children: Mutex::new(HashSet::new()), + rlm_progress_notes: Mutex::new(HashMap::new()), + rlm_child_changed: tokio::sync::Notify::new(), agent_mailboxes: Mutex::new(HashMap::new()), agent_output_buffers: Mutex::new(HashMap::new()), agent_wait_cursors: Mutex::new(HashMap::new()), diff --git a/crates/server/src/runtime/agents.rs b/crates/server/src/runtime/agents.rs index 966eceb8..fbc25e6b 100644 --- a/crates/server/src/runtime/agents.rs +++ b/crates/server/src/runtime/agents.rs @@ -647,6 +647,7 @@ impl ServerRuntime { .get_mut(&parent_session_id) { registry.update_status(child_session_id, status); + self.rlm_child_changed.notify_waiters(); } } @@ -1870,8 +1871,15 @@ impl ServerRuntime { /// Kernel `host_request("rlm.list_subagents")` — direct children retained /// by this session, in the payload shape `rlm.list_subagents()` parses. - pub(crate) async fn host_rlm_list_subagents(&self, session_id: &str) -> serde_json::Value { + pub(crate) async fn host_rlm_list_subagents( + self: &Arc, + session_id: &str, + ) -> serde_json::Value { let sid = SessionId::from_string(session_id.to_owned()); + if let Err(error) = self.restore_direct_children(sid).await { + return serde_json::json!({"status": "error", "error": error}); + } + let notes = self.rlm_progress_notes.lock().await.clone(); let entries: Vec = { let registries = self.agent_registries.lock().await; registries @@ -1881,7 +1889,13 @@ impl ServerRuntime { .children_of(&sid) .into_iter() .filter_map(|child_id| registry.get(&child_id)) - .map(Self::rlm_subagent_entry) + .map(|meta| { + let mut entry = Self::rlm_subagent_entry(meta); + if let Some((_, note)) = notes.get(&meta.session_id) { + entry["progress_note"] = serde_json::json!(note); + } + entry + }) .collect() }) .unwrap_or_default() diff --git a/crates/server/src/runtime/goal_background.rs b/crates/server/src/runtime/goal_background.rs index a2a3c57b..acf0fa10 100644 --- a/crates/server/src/runtime/goal_background.rs +++ b/crates/server/src/runtime/goal_background.rs @@ -92,6 +92,9 @@ impl ServerRuntime { None } }; + if work_id.starts_with("agent:") { + self.rlm_child_changed.notify_waiters(); + } let Some(deferred) = deferred else { return; }; @@ -201,6 +204,18 @@ impl ServerRuntime { .is_some_and(|state| !state.pending.is_empty()) } + /// Admission tickets bridge mailbox drain and queued-turn activation, so + /// collection cannot report a child settled while accepted work is in transit. + pub(super) fn has_pending_subagent_work(&self, child_id: SessionId) -> bool { + let prefix = format!("agent:{child_id}:"); + self.background_work + .lock() + .expect("background work mutex poisoned") + .sessions + .values() + .any(|state| state.pending.iter().any(|id| id.starts_with(&prefix))) + } + pub(super) fn clear_background_work(&self, session_id: SessionId) { self.background_work .lock() diff --git a/crates/server/src/runtime/handlers/session.rs b/crates/server/src/runtime/handlers/session.rs index 6d58db3a..4de4db35 100644 --- a/crates/server/src/runtime/handlers/session.rs +++ b/crates/server/src/runtime/handlers/session.rs @@ -147,6 +147,24 @@ impl ServerRuntime { request_id: serde_json::Value, params: SessionStartParams, tool_registry: Option>, + ) -> serde_json::Value { + self.start_session_with_seed( + connection_id, + request_id, + params, + tool_registry, + /*seed*/ None, + ) + .await + } + + pub(crate) async fn start_session_with_seed( + &self, + connection_id: u64, + request_id: serde_json::Value, + params: SessionStartParams, + tool_registry: Option>, + seed: Option, ) -> serde_json::Value { let now = Utc::now(); let session_id = devo_protocol::native::ids::SessionId::new(); @@ -164,13 +182,46 @@ impl ServerRuntime { .model_binding_id .as_deref() .or(params.model.as_deref()); - let initial_turn_config = runtime_context.resolve_turn_config(requested_model, None); + let initial_turn_config = runtime_context.resolve_turn_config( + requested_model, + seed.as_ref() + .and_then(|seed| seed.settings.reasoning_effort.clone()), + ); let model = initial_turn_config.model.slug.clone(); - let model_binding_id = initial_turn_config.model_binding_id.clone(); + let mut model_binding_id = initial_turn_config.model_binding_id.clone(); + let persisted_model = if seed.is_some() + && let Some(variant) = initial_turn_config.variant.as_ref() + { + let base = model_binding_id.as_ref().unwrap_or(&model); + let selected = format!("{base}/{variant}"); + model_binding_id = model_binding_id.as_ref().map(|_| selected.clone()); + selected + } else { + model.clone() + }; + let seeded_settings = seed.as_ref().map(|seed| { + let mut settings = vec![ + ( + SessionSettingsField::Model, + serde_json::json!(persisted_model), + ), + ( + SessionSettingsField::ModelBindingId, + serde_json::json!(model_binding_id), + ), + ]; + if let Some(sandbox) = seed.config.sandbox_profile.as_ref() { + settings.push(( + SessionSettingsField::SandboxProfile, + serde_json::json!(sandbox), + )); + } + settings + }); let initial_effort = initial_turn_config.reasoning_effort_selection.clone(); let mut invented_persistence = (!params.ephemeral).then(|| self.rollout_store.invent_session_persistence(&session_id)); - let native_session = devo_protocol::native::session::Session { + let mut native_session = devo_protocol::native::session::Session { id: session_id, version: 1, cwd: params.cwd.clone(), @@ -194,8 +245,10 @@ impl ServerRuntime { .unwrap_or(SessionTitleState::Unset), model: devo_protocol::native::model::ModelBinding { provider: model_binding_id.unwrap_or_else(|| "unknown".to_string()), - model: model.clone(), - variant: None, + model: persisted_model, + variant: seed + .as_ref() + .and_then(|_| initial_turn_config.variant.clone()), reasoning_effort: None, }, settings: devo_protocol::native::session::SessionSettings { @@ -224,6 +277,12 @@ impl ServerRuntime { updated_at: now, }, }; + if let Some(seed) = &seed { + native_session.settings = seed.settings.clone(); + if let Some(invented) = invented_persistence.as_mut() { + invented.extras.collaboration_mode = Some(seed.collaboration_mode); + } + } let applied_compaction_limit = crate::runtime::context_occupancy::resolved_compaction_limit( &initial_turn_config.model, ); @@ -250,6 +309,13 @@ impl ServerRuntime { params.cwd.clone(), params.additional_directories.clone(), ); + if let Some(seed) = seed { + core_session.config.permission_mode = seed.config.permission_mode; + core_session.config.permission_profile = seed.config.permission_profile; + core_session.config.sandbox_profile = seed.config.sandbox_profile; + core_session.collaboration_mode = seed.collaboration_mode; + summary.collaboration_mode = seed.collaboration_mode; + } let permission_preset = protocol_preset_from_safety(core_session.config.permission_profile.preset); summary.set_permission_preset(permission_preset); @@ -298,18 +364,28 @@ impl ServerRuntime { } None => None, }; - if let Some(invented) = &invented_persistence - && let Err(error) = self.rollout_store.append_session_meta_create_at( - &invented.rollout_path, - &summary.native, - Some(invented.extras.clone()), - ) - { - return self.error_response( - request_id, - ProtocolErrorCode::InternalError, - format!("failed to persist session metadata: {error}"), - ); + if let Some(invented) = &invented_persistence { + let persisted = if let Some(settings) = seeded_settings { + self.rollout_store.append_seeded_session_meta_create_at( + &invented.rollout_path, + &summary.native, + invented.extras.clone(), + &settings, + ) + } else { + self.rollout_store.append_session_meta_create_at( + &invented.rollout_path, + &summary.native, + Some(invented.extras.clone()), + ) + }; + if let Err(error) = persisted { + return self.error_response( + request_id, + ProtocolErrorCode::InternalError, + format!("failed to persist session metadata: {error}"), + ); + } } crate::runtime::context_occupancy::apply_resolved_compaction_limit( &mut core_session.config, diff --git a/crates/server/src/runtime/kernel_host.rs b/crates/server/src/runtime/kernel_host.rs index 593727af..311d20d4 100644 --- a/crates/server/src/runtime/kernel_host.rs +++ b/crates/server/src/runtime/kernel_host.rs @@ -46,9 +46,7 @@ pub async fn dispatch_host_request_with_bridge(bridge: &HostBridge, data: Value) action if action.starts_with("harness.") => { crate::runtime::kernel_host_bridge::handle_harness(bridge, action, ¶ms).await } - "rlm.create_session" => { - error_reply("denied: rlm.create_session (daemon out of scope this release)") - } + "rlm.create_session" => super::kernel_host_session::create_session(bridge, ¶ms).await, "rlm_heartbeat.list" => { let Some(runtime) = bridge.runtime() else { return error_reply("runtime not available for rlm_heartbeat.list"); @@ -77,6 +75,9 @@ pub async fn dispatch_host_request_with_bridge(bridge: &HostBridge, data: Value) error_reply(format!("unknown rlm_heartbeat action: {a}")) } "rlm.run" => super::kernel_host_models::spawn(bridge, ¶ms).await, + "rlm.collect" => super::kernel_host_controls::collect(bridge, ¶ms).await, + "rlm.progress.note" => super::kernel_host_controls::progress_note(bridge, ¶ms).await, + "rlm.end_turn" => super::kernel_host_controls::end_turn(bridge).await, "rlm.find_models" => super::kernel_host_models::find_models(bridge, ¶ms).await, "rlm.list_subagents" => { let Some(runtime) = bridge.runtime() else { @@ -379,9 +380,9 @@ mod tests { use devo_core::tools::PermissionChecker; use devo_protocol::CollaborationMode; - /// Verifies: rlm.create_session is denied. + /// Invalid root creation fails before side effects. #[tokio::test] - async fn denies_create_session() { + async fn create_session_requires_a_prompt() { let reply = dispatch_host_request( "ses_00000000-0000-0000-0000-000000000001", json!({ "type": "rlm.create_session" }), @@ -389,7 +390,7 @@ mod tests { .await; assert_eq!(reply["status"], "error"); assert!( - reply["error"].as_str().unwrap_or("").contains("denied"), + reply["error"].as_str().unwrap_or("").contains("prompt"), "{reply}" ); } @@ -520,6 +521,7 @@ mod tests { file_read_ledger: Arc::new(FileReadLedger::new()), sandbox_profile: None, cancel_token: tokio_util::sync::CancellationToken::new(), + end_turn_requested: None, kernel_lifetime: None, mcp_manager: None, agent_scope: ToolAgentScope::Parent, diff --git a/crates/server/src/runtime/kernel_host_bridge.rs b/crates/server/src/runtime/kernel_host_bridge.rs index a221f464..2a6f7ae6 100644 --- a/crates/server/src/runtime/kernel_host_bridge.rs +++ b/crates/server/src/runtime/kernel_host_bridge.rs @@ -53,6 +53,7 @@ pub(crate) struct HostBridge { pub file_read_ledger: Arc, pub sandbox_profile: Option, pub cancel_token: CancellationToken, + pub end_turn_requested: Option>, pub kernel_lifetime: Option, pub mcp_manager: Option>, pub agent_scope: ToolAgentScope, @@ -82,6 +83,7 @@ impl HostBridge { file_read_ledger: Arc::new(FileReadLedger::new()), sandbox_profile: None, cancel_token: CancellationToken::new(), + end_turn_requested: None, kernel_lifetime: None, mcp_manager: None, agent_scope: ToolAgentScope::Parent, diff --git a/crates/server/src/runtime/kernel_host_controls.rs b/crates/server/src/runtime/kernel_host_controls.rs new file mode 100644 index 00000000..a12f252c --- /dev/null +++ b/crates/server/src/runtime/kernel_host_controls.rs @@ -0,0 +1,421 @@ +//! Non-steering orchestration. Waits run outside actors and never own admission locks. + +use std::collections::HashSet; +use std::sync::Arc; +use std::time::Duration; + +use devo_protocol::native::ids::SessionId; +use devo_protocol::native::item::Item; +use serde_json::{Value, json}; +use tokio::time::Instant; + +use super::ServerRuntime; +use super::kernel_host_bridge::HostBridge; +use crate::subagent::SubagentStatus; + +pub(super) async fn collect(bridge: &HostBridge, params: &Value) -> Value { + let result = async { + let timeout_ms = match params.get("timeout_ms") { + None => 0, + Some(value) => value.as_u64().filter(|ms| *ms <= 300_000) + .ok_or("timeout_ms must be an integer between 0 and 300000")?, + }; + let targets = match params.get("targets") { + None => Vec::new(), + Some(Value::Array(values)) => values.iter().map(|value| { + value.as_str().filter(|target| !target.trim().is_empty()) + .ok_or("targets must contain nonempty strings") + }).collect::, _>>()?, + Some(_) => return Err("targets must be an array".to_string()), + }; + let runtime = bridge.runtime().ok_or("runtime not available for rlm.collect")?; + // Rehydrate only durable direct children, not arbitrary user sessions. The + // registry is process-local; the index and journal survive host restarts. + runtime.restore_direct_children(bridge.session_id).await?; + let ids = if targets.is_empty() { + runtime.agent_registries.lock().await.get(&bridge.session_id) + .map(|registry| registry.children_of(&bridge.session_id)).unwrap_or_default() + } else { + let mut seen = HashSet::new(); + let mut ids = Vec::new(); + for target in targets { + let child = runtime.resolve_child_agent(bridge.session_id, target).await + .map_err(|error| error.to_string())?; + if child.parent_session_id != bridge.session_id { + return Err("collect can only read direct children".to_string()); + } + if seen.insert(child.session_id) { ids.push(child.session_id); } + } + ids + }; + let deadline = Instant::now() + Duration::from_millis(timeout_ms); + loop { + // Register before observing state so a concurrent merge cannot be lost. + let changed = runtime.rlm_child_changed.notified(); + tokio::pin!(changed); + changed.as_mut().enable(); + if bridge.cancel_token.is_cancelled() { + return Err("rlm.collect interrupted".to_string()); + } + let mut results = Vec::with_capacity(ids.len()); + for id in &ids { results.push(runtime.collect_child(bridge.session_id, *id).await?); } + if results.iter().all(|result| result["settled"] == true || result["requires_resume"] == true) || Instant::now() >= deadline { + return Ok(json!({"results": results})); + } + tokio::select! { + _ = changed => {}, + _ = tokio::time::sleep_until(deadline) => {}, + _ = bridge.cancel_token.cancelled() => return Err("rlm.collect interrupted".to_string()), + } + } + }.await; + match result { + Ok(result) => json!({"status": "ok", "result": result}), + Err(error) => json!({"status": "error", "error": error}), + } +} + +impl ServerRuntime { + pub(super) async fn restore_direct_children( + self: &Arc, + parent: SessionId, + ) -> Result<(), String> { + let rows = self + .deps + .db + .list_sessions() + .map_err(|error| error.to_string())?; + for row in rows + .into_iter() + .filter(|row| row.parent_session_id == Some(parent) && row.agent_path.is_some()) + { + if self.session(row.session_id).await.is_none() { + self.get_or_load_parent_session(row.session_id) + .await + .map_err(|error| format!("restore child {}: {error}", row.session_id))?; + } + } + Ok(()) + } + + async fn collect_child(&self, parent: SessionId, child: SessionId) -> Result { + let meta = self + .resolve_child_agent(parent, child.as_str()) + .await + .map_err(|error| error.to_string())?; + let handle = self + .session(child) + .await + .ok_or("child session is unavailable")?; + let (snapshot, queued, active) = { + // Queue drain pops an input before registering its next turn. A + // short admission-consistent snapshot must not mistake that gap + // for completion. Release the gate before transcript/recovery I/O + // or any wait for child progress. + let _admission = handle.lock_state_change().await; + let snapshot = handle + .turn_reservation_snapshot() + .await + .ok_or("child snapshot is unavailable")?; + let queued = !snapshot + .pending_turn_queue + .lock() + .map_err(|_| "child queue is unavailable")? + .is_empty() + || self.has_pending_subagent_work(child); + let active = self.active_turns.active_turn_id(child).await.is_some(); + (snapshot, queued, active) + }; + // Terminal registry status is published before MergeTurn. Keep it running + // until the working set is installed and the execution reservation clears. + // Recovery is a journal-owned state, not just an unfinished status. + // Provider failures and shutdown interruptions can also be resumable. + // Queued follow-ups must not mask an abandoned current turn. + let requires_resume = if !meta.close_requested && !active && snapshot.active_turn.is_none() + { + self.turn_recovery(child) + .await + .map_err(|error| format!("read child recovery: {error}"))? + .is_some() + || (queued && self.rlm_lazy_children.lock().await.contains(&child)) + || snapshot.latest_turn.as_ref().is_some_and(|turn| { + turn.native.status == devo_protocol::native::turn::TurnStatus::WaitingApproval + }) + } else { + false + }; + let settled = !matches!( + meta.status, + SubagentStatus::Spawning | SubagentStatus::Running + ) && snapshot.active_turn.is_none() + && !queued + && !active + && !requires_resume; + let answer = if settled { + let latest_id = snapshot.latest_turn.as_ref().map(|turn| turn.turn_id()); + handle + .transcript_items() + .await + .unwrap_or_default() + .into_iter() + .filter(|item| Some(item.turn_id) == latest_id) + .filter_map(|item| match item.item { + Item::AssistantMessage { text, .. } => Some(text), + _ => None, + }) + .collect::>() + .join("\n") + } else { + String::new() + }; + // Explicit host bound; callers can read the retained journal for larger output. + let truncated = answer.chars().count() > 128 * 1024; + let answer: String = answer.chars().take(128 * 1024).collect(); + let error = if (settled || requires_resume) && meta.status == SubagentStatus::Failed { + if let Some(turn) = snapshot.latest_turn.as_ref() { + // Native turn/read includes the canonical persisted terminal + // error; the actor's latest-turn mirror can omit its details. + let read = self + .handle_native_turn_read( + json!("rlm.collect"), + json!({ + "sessionId": child, "turnId": turn.turn_id(), + }), + ) + .await; + Some( + read["result"]["turn"]["error"]["message"] + .as_str() + .unwrap_or("child turn failed; inspect its transcript") + .to_string(), + ) + } else { + Some("child failed before a turn was admitted".to_string()) + } + } else { + None + }; + let note = self + .rlm_progress_notes + .lock() + .await + .get(&child) + .map(|(_, note)| note.clone()); + Ok(json!({ + "rlm_child_id": child.to_string(), "session_name": meta.nickname, + "session_dir": meta.session_dir, "status": if requires_resume { "waiting_for_resume" } else if settled { meta.status.as_str() } else { "running" }, + "settled": settled, "answer_preview": if answer.is_empty() { None } else { Some(answer.chars().take(512).collect::()) }, + "answer_text": if answer.is_empty() { None } else { Some(answer) }, + "answer_truncated": truncated, "error": error, "progress_note": note, + "requires_resume": requires_resume, + })) + } +} + +pub(super) async fn progress_note(bridge: &HostBridge, params: &Value) -> Value { + let result = async { + let note = params + .get("message") + .and_then(Value::as_str) + .map(str::trim) + .filter(|note| !note.is_empty() && note.encode_utf16().count() <= 512) + .ok_or("message must contain 1 to 512 UTF-16 code units")?; + let runtime = bridge + .runtime() + .ok_or("runtime not available for rlm.progress.note")?; + if runtime + .child_parent_and_path(bridge.session_id) + .await + .is_none() + { + return Err("progress_note requires a child session"); + } + let mut notes = runtime.rlm_progress_notes.lock().await; + if let Some((at, _)) = notes.get(&bridge.session_id) { + let remaining = Duration::from_secs(/*secs*/ 10).saturating_sub(at.elapsed()); + if !remaining.is_zero() { + return Ok( + json!({"accepted": false, "retry_after_ms": remaining.as_millis().max(1)}), + ); + } + } + notes.insert( + bridge.session_id, + (std::time::Instant::now(), note.to_string()), + ); + Ok(json!({"accepted": true, "retry_after_ms": null})) + } + .await; + match result { + Ok(result) => json!({"status": "ok", "result": result}), + Err(error) => json!({"status": "error", "error": error}), + } +} + +/// The bridge may outlive a turn when a Python task is detached. Never let an +/// obsolete bridge end a later turn or claim success after its turn merged. +pub(super) async fn end_turn(bridge: &HostBridge) -> Value { + if let (Some(runtime), Some(turn_id), Some(requested)) = + (bridge.runtime(), bridge.turn_id, &bridge.end_turn_requested) + && !bridge.cancel_token.is_cancelled() + && runtime.active_turns.active_turn_id(bridge.session_id).await == Some(turn_id) + { + requested.store(true, std::sync::atomic::Ordering::Release); + json!({"status": "ok", "result": {"ended": true}}) + } else { + json!({"status": "error", "error": "rlm.end_turn requires an active model turn"}) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + + #[tokio::test] + async fn malformed_bounds_and_targets_are_rejected_before_runtime_access() { + let bridge = HostBridge::for_tests(SessionId::new()); + for timeout in [ + json!(-1), + json!(true), + json!(1.5), + json!(300001), + json!(null), + ] { + assert_eq!( + collect(&bridge, &json!({"timeout_ms":timeout})).await, + json!({"status":"error","error":"timeout_ms must be an integer between 0 and 300000"}) + ); + } + for target in [json!([null]), json!([false]), json!([""]), json!([" "])] { + assert_eq!( + collect(&bridge, &json!({"targets":target})).await, + json!({"status":"error","error":"targets must contain nonempty strings"}) + ); + } + assert_eq!( + end_turn(&bridge).await, + json!({"status":"error","error":"rlm.end_turn requires an active model turn"}) + ); + } + + #[tokio::test] + async fn progress_limit_counts_utf16_not_unicode_scalars() { + let bridge = HostBridge::for_tests(SessionId::new()); + for message in [String::new(), " ".to_string(), "🎉".repeat(257)] { + assert_eq!( + progress_note(&bridge, &json!({"message":message})).await, + json!({"status":"error","error":"message must contain 1 to 512 UTF-16 code units"}) + ); + } + assert_eq!( + progress_note(&bridge, &json!({"message":"🎉".repeat(256)})).await, + json!({"status":"error","error":"runtime not available for rlm.progress.note"}) + ); + } + #[tokio::test] + async fn saved_unfinished_child_requires_explicit_recovery_without_waiting() { + use devo_core::tools::AgentToolCoordinator; + let home = tempfile::TempDir::new().unwrap(); + let runtime = crate::test_support::TestRuntime::noop().runtime(home.path()); + let started = runtime + .start_session_with_registry( + /*connection_id*/ 0, + json!(1), + devo_protocol::SessionStartParams { + cwd: home.path().to_path_buf(), + additional_directories: Vec::new(), + ephemeral: false, + title: None, + model: None, + model_binding_id: None, + }, + /*tool_registry*/ None, + ) + .await; + let parent: SessionId = + serde_json::from_value(started["result"]["session"]["id"].clone()).unwrap(); + let child = Arc::clone(&runtime) + .spawn_agent( + serde_json::from_value(json!({ + "session_id":parent,"message":"task","nickname":"worker","fork_turns":"none" + })) + .unwrap(), + ) + .await + .unwrap(); + let handle = runtime.session(child.child_session_id).await.unwrap(); + let mut saved = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { + loop { + let snapshot = handle.turn_reservation_snapshot().await.unwrap(); + if runtime + .active_turns + .active_turn_id(child.child_session_id) + .await + .is_none() + && let Some(turn) = snapshot.latest_turn + { + break turn; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + let mut completed = saved.clone(); + completed.native.status = devo_protocol::native::turn::TurnStatus::Completed; + handle + .enqueue_pending_turn_input(devo_protocol::PendingInputItem::new( + devo_protocol::PendingInputKind::UserText { + text: "queued follow-up".to_string(), + }, + None, + chrono::Utc::now(), + )) + .await; + // Simulate the durable state exposed after a crash, without starting or + // silently replaying the saved turn's tool intents. Pending follow-ups + // must not mask the abandoned current turn. + saved.native.status = devo_protocol::native::turn::TurnStatus::InProgress; + handle.set_runtime_session_idle(Some(saved)).await; + runtime + .set_agent_status(parent, child.child_session_id, SubagentStatus::Running) + .await; + let mut bridge = HostBridge::for_tests(parent); + bridge.runtime = Arc::downgrade(&runtime); + let result = tokio::time::timeout( + Duration::from_secs(/*secs*/ 1), + collect(&bridge, &json!({"targets":["worker"],"timeout_ms":300000})), + ) + .await + .unwrap(); + let item = &result["result"]["results"][0]; + assert_eq!( + json!({"status":item["status"],"settled":item["settled"], + "requires_resume":item["requires_resume"],"answer_text":item["answer_text"]}), + json!({"status":"waiting_for_resume","settled":false,"requires_resume":true,"answer_text":null}) + ); + // A normal live post-merge scheduling gap is not orphaned work. + handle.set_runtime_session_idle(Some(completed)).await; + runtime + .set_agent_status(parent, child.child_session_id, SubagentStatus::Completed) + .await; + let live_gap = collect(&bridge, &json!({"targets":["worker"],"timeout_ms":0})).await; + assert_eq!( + live_gap["result"]["results"][0]["requires_resume"], + json!(false) + ); + // Cold hydration marks the same queued-only state as needing explicit + // session resume. Starting any real turn clears this runtime marker. + runtime + .rlm_lazy_children + .lock() + .await + .insert(child.child_session_id); + let dormant = collect(&bridge, &json!({"targets":["worker"],"timeout_ms":300000})).await; + assert_eq!( + dormant["result"]["results"][0]["requires_resume"], + json!(true) + ); + } +} diff --git a/crates/server/src/runtime/kernel_host_session.rs b/crates/server/src/runtime/kernel_host_session.rs new file mode 100644 index 00000000..64a9c99f --- /dev/null +++ b/crates/server/src/runtime/kernel_host_session.rs @@ -0,0 +1,329 @@ +//! Independent root creation uses the canonical session constructor and Native turn admission. + +use devo_core::SessionConfig; +use devo_protocol::native::session::SessionSettings; +use devo_protocol::{CollaborationMode, SessionStartParams}; +use serde_json::{Value, json}; + +use super::kernel_host_bridge::HostBridge; + +/// Initial safety/settings seed, installed before persistence and SessionStart hooks. +/// This is creation state, not an alternate settings-update protocol. +pub(crate) struct RootSessionSeed { + pub config: SessionConfig, + pub settings: SessionSettings, + pub collaboration_mode: CollaborationMode, +} + +pub(super) async fn create_session(bridge: &HostBridge, params: &Value) -> Value { + let result = async { + let prompt = params + .get("prompt") + .and_then(Value::as_str) + .filter(|prompt| !prompt.trim().is_empty()) + .ok_or("create_session requires a nonempty prompt")?; + let kwargs = params + .get("kwargs") + .and_then(Value::as_object) + .ok_or("create_session requires kwargs")?; + for key in kwargs.keys() { + if !["name", "model", "thinking", "cwd"].contains(&key.as_str()) { + return Err(format!("unsupported create_session option: {key}")); + } + } + let option = |key: &str| -> Result, String> { + match kwargs.get(key) { + None => Ok(None), + Some(Value::String(value)) if !value.trim().is_empty() && value.trim() == value => { + Ok(Some(value)) + } + _ => Err(format!( + "{key} must be a nonempty string without surrounding whitespace" + )), + } + }; + let name = option("name")?; + let selected_model = option("model")?; + let thinking = option("thinking")?; + let cwd = option("cwd")?; + let runtime = bridge + .runtime() + .ok_or("runtime not available for rlm.create_session")?; + if matches!(bridge.collaboration_mode, CollaborationMode::Plan) { + return Err("create_session is blocked in Plan mode".to_string()); + } + let caller = runtime + .session(bridge.session_id) + .await + .ok_or("caller session is unavailable")?; + let snapshot = caller + .spawn_snapshot() + .await + .ok_or("caller snapshot is unavailable")?; + if matches!( + snapshot.parent_summary.collaboration_mode, + CollaborationMode::Plan + ) { + return Err("create_session is blocked in Plan mode".to_string()); + } + if snapshot.parent_summary.agent_path.is_some() + || snapshot.parent_summary.ephemeral + || matches!( + snapshot.parent_summary.native.parent, + Some(devo_protocol::native::session::SessionParent::Agent { .. }) + ) + { + return Err("create_session requires a durable root session".to_string()); + } + // Do not use independent sessions to escape the caller's filesystem + // capability. A cross-workspace creation needs a separate approval flow. + if let Some(cwd) = cwd { + let requested = std::path::Path::new(cwd); + let requested = if requested.is_absolute() { + requested.to_path_buf() + } else { + bridge.cwd.join(requested) + }; + if std::fs::canonicalize(requested).map_err(|error| error.to_string())? + != std::fs::canonicalize(&bridge.cwd).map_err(|error| error.to_string())? + { + return Err( + "create_session currently requires the caller's working directory".to_string(), + ); + } + } + // Like child spawn, inherit the current decision-point configuration, + // including named variants and mid-turn model/thinking changes. + let live_config = if let Some(stream) = runtime.active_stream_state(bridge.session_id).await + { + let stream = stream.lock().await; + stream.turn_inline.as_ref().and_then(|inline| { + inline + .live_turn_settings + .lock() + .ok() + .and_then(|live| live.turn_config.clone()) + }) + } else { + None + }; + let parent_model = live_config + .as_ref() + .map(|config| { + let base = config + .model_binding_id + .as_ref() + .unwrap_or(&config.model.slug); + config + .variant + .as_ref() + .map(|variant| format!("{base}/{variant}")) + .unwrap_or_else(|| base.clone()) + }) + .or_else(|| { + super::session_model_selection(&snapshot.parent_summary).map(str::to_string) + }) + .ok_or("caller has no model binding")?; + let parent_thinking = match live_config.as_ref() { + Some(config) => config.reasoning_effort_selection.as_deref(), + None => snapshot.parent_summary.settings.reasoning_effort.as_deref(), + }; + let (_, resolved) = snapshot + .runtime_context + .resolve_child_model(selected_model, &parent_model, thinking, parent_thinking) + .map_err(|error| error.to_string())?; + let model = resolved + .model_binding_id + .clone() + .unwrap_or_else(|| resolved.model.slug.clone()); + let model_selection = resolved + .variant + .as_ref() + .map(|variant| format!("{model}/{variant}")) + .unwrap_or_else(|| model.clone()); + let mut settings = snapshot.parent_summary.native.settings.clone(); + settings.reasoning_effort = resolved.reasoning_effort_selection; + // A root is independently durable and visible; it has no parent usage + // ledger, ancestry, shared journal, or implicit parent completion notice. + let connection = runtime + .active_turns + .active_connection_id(bridge.session_id) + .await; + let response = runtime + .start_session_with_seed( + connection.unwrap_or(0), + json!("rlm.create_session"), + SessionStartParams { + cwd: bridge.cwd.clone(), + additional_directories: snapshot.parent_summary.additional_directories.clone(), + ephemeral: false, + title: name.map(str::to_string), + model: Some(model_selection), + model_binding_id: None, + }, + snapshot.parent_tool_registry, + Some(RootSessionSeed { + config: snapshot.parent_config, + settings, + collaboration_mode: bridge.collaboration_mode, + }), + ) + .await; + if let Some(error) = response.get("error") { + return Err(format!("create_session failed: {error}")); + } + let session: devo_protocol::native::session::Session = + serde_json::from_value(response["result"]["session"].clone()) + .map_err(|error| format!("invalid created session: {error}"))?; + let session_id = session.id; + let handle = runtime + .session(session_id) + .await + .ok_or_else(|| format!("created session {session_id} is unavailable"))?; + let session_file = handle + .rollout_path() + .await + .flatten() + .ok_or_else(|| format!("created session {session_id} has no journal"))?; + let turn = runtime + .handle_turn_start_for_connection( + connection, + json!("rlm.create_session.turn"), + json!({ + "sessionId": session_id, "input": [{"type":"text", "text":prompt}], + "idempotencyKey": format!("rlm-create-{session_id}"), + }), + ) + .await; + if let Some(error) = turn.get("error") { + // Creation is durable. Do not hide the id or retry with another root. + return Err(format!( + "created session {session_id}, but its initial turn was not admitted: {error}" + )); + } + Ok::<_, String>(json!({ + "active_session_id": session_id.to_string(), "session_id": session_id.to_string(), + "name": name.map(str::to_string).unwrap_or_else(|| session_id.to_string()), + "session_file": session_file, "model": model, + })) + } + .await; + match result { + Ok(result) => json!({"status":"ok", "result":result}), + Err(error) => json!({"status":"error", "error":error}), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + use std::sync::Arc; + + #[tokio::test] + async fn root_inherits_restrictive_safety_and_invalid_requests_create_nothing() { + let home = tempfile::TempDir::new().unwrap(); + let runtime = crate::test_support::TestRuntime::noop().runtime(home.path()); + let started = runtime + .start_session_with_registry( + 0, + json!(1), + SessionStartParams { + cwd: home.path().to_path_buf(), + additional_directories: Vec::new(), + ephemeral: false, + title: Some("caller".into()), + model: None, + model_binding_id: None, + }, + None, + ) + .await; + let caller_id: devo_protocol::SessionId = + serde_json::from_value(started["result"]["session"]["id"].clone()).unwrap(); + let updated = runtime + .handle_native_session_metadata_update( + json!(2), + json!({ + "sessionId": caller_id, "expectedVersion":0, + "settings":{"permissionProfile":"default", "sandboxProfile":"strict"} + }), + ) + .await; + assert!(updated.get("error").is_none(), "{updated}"); + let mut bridge = HostBridge::for_tests(caller_id); + bridge.cwd = home.path().to_path_buf(); + bridge.runtime = Arc::downgrade(&runtime); + let parent = runtime + .session(caller_id) + .await + .unwrap() + .spawn_snapshot() + .await + .unwrap(); + assert_eq!( + parent.parent_summary.native.settings.permission_profile, + devo_protocol::native::model::PermissionProfile::Default + ); + for params in [ + json!({"prompt":"task","kwargs":{"cwd":".."}}), + json!({"prompt":"task","kwargs":{"model":"unavailable/model"}}), + json!({"prompt":"task","kwargs":{"unexpected":true}}), + ] { + assert_eq!( + create_session(&bridge, ¶ms).await["status"], + json!("error") + ); + assert_eq!(runtime.deps.db.list_root_sessions().unwrap().len(), 1); + } + bridge.collaboration_mode = CollaborationMode::Plan; + assert_eq!( + create_session(&bridge, &json!({"prompt":"task","kwargs":{}})).await, + json!({"status":"error","error":"create_session is blocked in Plan mode"}) + ); + assert_eq!(runtime.deps.db.list_root_sessions().unwrap().len(), 1); + bridge.collaboration_mode = CollaborationMode::Build; + let created = create_session( + &bridge, + &json!({"prompt":"task","kwargs":{"name":"independent"}}), + ) + .await; + assert_eq!(created["status"], json!("ok"), "{created}"); + let id = serde_json::from_value(created["result"]["session_id"].clone()).unwrap(); + let child = runtime + .session(id) + .await + .unwrap() + .spawn_snapshot() + .await + .unwrap(); + assert_eq!( + child.parent_config.permission_mode, + parent.parent_config.permission_mode + ); + assert_eq!( + child.parent_config.permission_profile, + parent.parent_config.permission_profile + ); + assert_eq!( + child.parent_config.sandbox_profile, + parent.parent_config.sandbox_profile + ); + assert_eq!(child.parent_summary.native.parent, None); + assert_eq!(runtime.deps.db.list_root_sessions().unwrap().len(), 2); + runtime.shutdown().await; + drop(runtime); + let restored = crate::test_support::TestRuntime::noop().runtime(home.path()); + restored.load_persisted_sessions().await.unwrap(); + let resumed = restored.get_or_load_parent_session(id).await.unwrap(); + let resumed = resumed.spawn_snapshot().await.unwrap(); + assert_eq!( + resumed.parent_config.permission_profile, + parent.parent_config.permission_profile + ); + assert_eq!( + resumed.parent_config.sandbox_profile.as_deref(), + Some("strict") + ); + } +} diff --git a/crates/server/src/runtime/session_cache.rs b/crates/server/src/runtime/session_cache.rs index 42b5decd..4b1229eb 100644 --- a/crates/server/src/runtime/session_cache.rs +++ b/crates/server/src/runtime/session_cache.rs @@ -209,6 +209,9 @@ impl ServerRuntime { .await; let mut state = SessionActorState::from_runtime_session(runtime_session); state.session_file_lock = Some(session_lock); + // This cold child has no queue scheduler yet. Set under the load + // gate before exposing the actor; a real turn admission clears it. + self.rlm_lazy_children.lock().await.insert(session_id); self.insert_session_actor(state).await } else { self.insert_root_session_actor(runtime_session, Some(session_lock)) diff --git a/crates/server/src/runtime/session_deletion.rs b/crates/server/src/runtime/session_deletion.rs index ce3f0809..e4e31b4a 100644 --- a/crates/server/src/runtime/session_deletion.rs +++ b/crates/server/src/runtime/session_deletion.rs @@ -721,6 +721,9 @@ impl ServerRuntime { self.clear_background_work(session_id); self.agent_mailboxes.lock().await.remove(&session_id); self.agent_output_buffers.lock().await.remove(&session_id); + self.rlm_lazy_children.lock().await.remove(&session_id); + self.rlm_progress_notes.lock().await.remove(&session_id); + self.rlm_child_changed.notify_waiters(); self.agent_wait_cursors.lock().await.remove(&session_id); { let mut registries = self.agent_registries.lock().await; diff --git a/crates/server/src/runtime/turn_exec/approval_resume.rs b/crates/server/src/runtime/turn_exec/approval_resume.rs index 228b69f7..f75bf24a 100644 --- a/crates/server/src/runtime/turn_exec/approval_resume.rs +++ b/crates/server/src/runtime/turn_exec/approval_resume.rs @@ -756,6 +756,7 @@ impl ServerRuntime { file_read_ledger: Arc::clone(&working.state.file_read_ledger), sandbox_profile: working.state.core.config.sandbox_profile.clone(), cancel_token: host_cancel_token.clone(), + end_turn_requested: None, kernel_lifetime: Some(kernel.lifetime_token()), mcp_manager: Some(Arc::clone(&working.state.runtime_context.mcp_manager)), agent_scope, diff --git a/crates/server/src/runtime/turn_exec/query.rs b/crates/server/src/runtime/turn_exec/query.rs index 7cd07f17..d986bbb1 100644 --- a/crates/server/src/runtime/turn_exec/query.rs +++ b/crates/server/src/runtime/turn_exec/query.rs @@ -166,6 +166,7 @@ impl ServerRuntime { .await .unwrap_or_else(CancellationToken::new); let query_cancel_token = turn_cancel_token.clone(); + let end_turn_requested = Arc::new(std::sync::atomic::AtomicBool::new(/*v*/ false)); let provider_http = runtime_context .config_store .lock() @@ -324,6 +325,7 @@ impl ServerRuntime { file_read_ledger: Arc::clone(&state.file_read_ledger), sandbox_profile: state.core.config.sandbox_profile.clone(), cancel_token: turn_cancel_token.clone(), + end_turn_requested: Some(Arc::clone(&end_turn_requested)), kernel_lifetime: Some(kernel.lifetime_token()), mcp_manager: Some(Arc::clone(&runtime_context.mcp_manager)), agent_scope, @@ -473,6 +475,7 @@ impl ServerRuntime { as Arc }), cancel_token: Some(query_cancel_token.clone()), + end_turn_requested: Some(end_turn_requested), compaction_provider: Some(compaction_provider), live_settings: live_turn_settings.clone(), live_goal, diff --git a/crates/server/src/runtime/turn_reservation.rs b/crates/server/src/runtime/turn_reservation.rs index 7702aacd..68875cb2 100644 --- a/crates/server/src/runtime/turn_reservation.rs +++ b/crates/server/src/runtime/turn_reservation.rs @@ -77,13 +77,16 @@ impl ServerRuntime { self.active_turns .register_turn(session_id, turn.native) .await; + self.rlm_lazy_children.lock().await.remove(&session_id); } pub(super) async fn clear_active_turn_runtime_handles( &self, session_id: devo_protocol::native::ids::SessionId, ) -> bool { - self.active_turns.clear_runtime_handles(session_id).await + let activation = self.active_turns.clear_runtime_handles(session_id).await; + self.rlm_child_changed.notify_waiters(); + activation } } diff --git a/crates/server/tests/child_model_selection.rs b/crates/server/tests/child_model_selection.rs index ef50bfa3..e34bb97b 100644 --- a/crates/server/tests/child_model_selection.rs +++ b/crates/server/tests/child_model_selection.rs @@ -567,3 +567,133 @@ async fn invalid_selection_does_not_poison_later_spawn_during_active_parent_turn wait_for_parent_turn_completed(&mut notifications, root).await?; Ok(()) } + +#[tokio::test] +async fn independent_root_inherits_named_non_effort_variant() -> Result<()> { + let home = TempDir::new()?; + write_config(home.path())?; + let path = home.path().join("providers.json"); + let mut config: serde_json::Value = serde_json::from_slice(&std::fs::read(&path)?)?; + config["provider"]["parent"]["models"]["vendor/main"]["variants"] = json!({ + "fast": {"request": {"speed":"fast"}} + }); + std::fs::write(&path, serde_json::to_vec(&config)?)?; + let parent = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +root = await rlm.create_session('Use the inherited variant', name='variant') +Path('variant-root').write_text(root.session_id) +await rlm.end_turn() +"#, + ), + ScriptedProvider::completed("variant answer"), + ])); + let runtime = runtime( + home.path(), + parent.clone(), + Arc::new(ScriptedProvider::pending()), + ); + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let session = start_parent_session(&runtime, connection, home.path()).await?; + let response = runtime + .handle_incoming( + connection, + json!({ + "id": 91, "method": "session/metadata/update", "params": { + "sessionId": session, "expectedVersion":0, + "model":{"provider":"parent", "model":"vendor/main", "variant":"fast"} + } + }), + ) + .await + .context("select named variant")?; + assert!(response.get("error").is_none(), "{response}"); + start_turn_with_approval_policy( + &runtime, + connection, + session, + "Create an independent root", + Some("never"), + ) + .await?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", session).await?; + assert_eq!( + completed["params"]["turn"]["status"], + json!("completed"), + "{completed}" + ); + let child: SessionId = std::fs::read_to_string(home.path().join("variant-root"))?.parse()?; + wait_for_stream_calls(&parent, 2).await?; + assert_eq!( + parent + .requests() + .iter() + .map(|request| request.extra_body.clone()) + .collect::>(), + vec![Some(json!({"speed":"fast"})); 2] + ); + let read = runtime + .handle_incoming( + connection, + json!({ + "id":92, "method":"session/read", "params":{"sessionId": child} + }), + ) + .await + .context("read variant root")?; + assert_eq!(read["result"]["session"]["model"]["variant"], json!("fast")); + runtime.shutdown().await; + drop(runtime); + let resumed_provider = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "after restart", + )])); + let restored = self::runtime( + home.path(), + resumed_provider.clone(), + Arc::new(ScriptedProvider::pending()), + ); + restored.load_persisted_sessions().await?; + let (connection, mut notifications) = initialize_connection(&restored).await?; + let resumed = restored + .handle_incoming( + connection, + json!({ + "id":93, "method":"session/resume", "params":{"sessionId": child} + }), + ) + .await + .context("resume variant root")?; + assert!(resumed.get("error").is_none(), "{resumed}"); + assert_eq!( + resumed["result"]["session"]["model"]["variant"], + json!("fast") + ); + while notifications.try_recv().is_ok() {} + let started = start_turn( + &restored, + connection, + child, + "Keep the variant after restart", + ) + .await?; + let completed = loop { + let notification = + wait_for_session_notification(&mut notifications, "turn/completed", child).await?; + if notification["params"]["turn"]["id"] == json!(started.turn.id) { + break notification; + } + }; + assert_eq!( + completed["params"]["turn"]["status"], + json!("completed"), + "{completed}" + ); + wait_for_stream_calls(&resumed_provider, 1).await?; + assert_eq!( + resumed_provider.requests()[0].extra_body, + Some(json!({"speed":"fast"})) + ); + Ok(()) +} diff --git a/crates/server/tests/subagent_python_controls.rs b/crates/server/tests/subagent_python_controls.rs new file mode 100644 index 00000000..b3a99850 --- /dev/null +++ b/crates/server/tests/subagent_python_controls.rs @@ -0,0 +1,513 @@ +//! Real CPython cells through the production host bridge and session actors. +#[path = "support/subagent_lifecycle.rs"] +mod support; + +use anyhow::{Context, Result}; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::sync::Arc; +use std::time::Duration; +use support::*; +use tempfile::TempDir; + +#[tokio::test] +async fn python_collect_progress_and_graceful_end_turn() -> Result<()> { + let home = TempDir::new()?; + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +worker = await rlm.spawn('Report progress, then finish', name='worker') +import asyncio +results, concurrent = await asyncio.gather( + rlm.collect([worker, 'worker'], timeout_ms=10000), + rlm.collect(worker, timeout_ms=10000), +) +assert results == concurrent +assert len(results) == 1, results +result = results[0] +assert result.settled and result.status == 'completed', result +assert result.answer_text == 'child final answer', result +assert result.progress_note == 'testing', result +assert (await rlm.collect(worker))[0] == result +assert (await rlm.list_subagents())[0].progress_note == 'testing' +assert (await rlm.end_turn())['ended'] +Path('controls-ok').write_text('complete') +"#, + ), + ScriptedProvider::ipython_code_tool_call( + r#" +assert (await rlm.progress_note('testing')).accepted +again = await rlm.progress_note('too soon') +assert not again.accepted and 0 < again.retry_after_ms <= 10000 +"#, + ), + ScriptedProvider::completed("child final answer"), + ])); + let runtime = build_runtime(home.path(), provider.clone())?; + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let parent = start_parent_session(&runtime, connection, home.path()).await?; + start_turn_with_approval_policy( + &runtime, + connection, + parent, + "Coordinate children", + Some("never"), + ) + .await?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!( + completed["params"]["turn"]["status"], + json!("completed"), + "{completed}" + ); + assert_eq!( + std::fs::read_to_string(home.path().join("controls-ok"))?, + "complete" + ); + // No synthetic parent message and no fourth model request after end_turn. + assert_eq!(provider.stream_calls(), 3); + Ok(()) +} + +#[tokio::test] +async fn collect_timeout_does_not_cancel_child_or_block_read_rpcs() -> Result<()> { + let home = TempDir::new()?; + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +worker = await rlm.spawn('Keep working', name='worker') +partial = await rlm.collect(worker, timeout_ms=50) +assert len(partial) == 1 and not partial[0].settled, partial +Path('timeout-ok').write_text('still running') +await rlm.collect(worker, timeout_ms=300000) +"#, + ), + StreamScript::Pending, + ])); + let runtime = build_runtime(home.path(), provider.clone())?; + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let parent = start_parent_session(&runtime, connection, home.path()).await?; + start_turn_with_approval_policy( + &runtime, + connection, + parent, + "Wait for child", + Some("never"), + ) + .await?; + wait_for_stream_calls(&provider, 2).await?; + // Observable file proves the real kernel returned a partial snapshot. + tokio::time::timeout(Duration::from_secs(10), async { + while !home.path().join("timeout-ok").exists() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + for (index, method, params) in [ + (41, "runtime/ping", json!({})), + (42, "session/read", json!({"sessionId":parent})), + (43, "session/list", json!({})), + ] { + let result = tokio::time::timeout( + Duration::from_secs(1), + runtime.handle_incoming( + connection, + json!({"id":index, "method":method, "params":params}), + ), + ) + .await? + .context("read response")?; + assert!(result.get("error").is_none(), "{result}"); + } + let agents = request_agent_list(&runtime, connection, parent).await?; + assert_eq!(agents.agents[0].status, "running"); + let interrupt = runtime + .handle_incoming( + connection, + json!({ + "id":44, "method":"session/interrupt", "params":{"scope":{"scope":"session", "sessionId":parent}} + }), + ) + .await + .context("interrupt")?; + assert!(interrupt.get("error").is_none(), "{interrupt}"); + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!(completed["params"]["turn"]["status"], json!("interrupted")); + // Explicit session interruption retains the existing cascading child policy. + request_agent_close(&runtime, connection, parent, "worker").await?; + Ok(()) +} + +#[tokio::test] +async fn collect_restores_direct_children_and_answers_after_restart() -> Result<()> { + let home = TempDir::new()?; + let first_provider = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "durable answer", + )])); + let first = build_runtime(home.path(), first_provider)?; + let (connection, mut notifications) = initialize_connection(&first).await?; + let parent = start_parent_session(&first, connection, home.path()).await?; + let child = spawn_child_named( + &first, + parent, + "Complete the task", + Some("none"), + Some("retained"), + ) + .await?; + wait_for_session_notification(&mut notifications, "turn/completed", child.child_session_id) + .await?; + drop(first); + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +result = (await rlm.collect('retained', timeout_ms=1000))[0] +assert result.settled and result.answer_text == 'durable answer', result +Path('restored-ok').write_text(result.answer_text) +await rlm.end_turn() +"#, + ), + ])); + let restored = build_runtime(home.path(), provider.clone())?; + restored.load_persisted_sessions().await?; + let (connection, mut notifications) = initialize_connection(&restored).await?; + let response = restored + .handle_incoming( + connection, + json!({ + "id":45,"method":"session/resume","params":{"sessionId":parent} + }), + ) + .await + .context("resume parent")?; + assert!(response.get("error").is_none(), "{response}"); + start_turn_with_approval_policy( + &restored, + connection, + parent, + "Collect retained answer", + Some("never"), + ) + .await?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!( + completed["params"]["turn"]["status"], + json!("completed"), + "{completed}" + ); + assert_eq!( + std::fs::read_to_string(home.path().join("restored-ok"))?, + "durable answer" + ); + assert_eq!(provider.stream_calls(), 1); + Ok(()) +} + +#[tokio::test] +async fn python_create_session_is_an_independent_visible_durable_root() -> Result<()> { + let home = TempDir::new()?; + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +import json +root = await rlm.create_session('Work independently', name='independent', cwd='.') +assert root.active_session_id == root.session_id +assert root.session_file.is_file() +assert not await rlm.list_subagents() +Path('root-handle.json').write_text(json.dumps({'id': root.session_id, 'journal': str(root.session_file)})) +await rlm.end_turn() +"#, + ), + ScriptedProvider::completed_after(Duration::from_millis(100), "independent answer"), + ])); + let runtime = build_runtime(home.path(), provider.clone())?; + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let parent = start_parent_session(&runtime, connection, home.path()).await?; + start_turn_with_approval_policy( + &runtime, + connection, + parent, + "Create an independent root", + Some("never"), + ) + .await?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!( + completed["params"]["turn"]["status"], + json!("completed"), + "{completed}" + ); + let handle: serde_json::Value = + serde_json::from_slice(&std::fs::read(home.path().join("root-handle.json"))?)?; + let created_id: devo_protocol::SessionId = handle["id"].as_str().context("root id")?.parse()?; + assert_ne!(created_id, parent); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let read = runtime + .handle_incoming( + connection, + json!({"id":46, "method":"session/read", "params":{"sessionId":created_id}}), + ) + .await + .context("read independent root")?; + assert!(read.get("error").is_none(), "{read}"); + let session = &read["result"]["session"]; + assert!(session["parent"].is_null(), "{session}"); + assert_eq!(session["title"], json!("independent")); + assert_eq!( + session["settings"]["permissionProfile"], + json!("fullAccess") + ); + if session["status"] == "idle" { + return Ok::<_, anyhow::Error>(()); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await??; + let list = runtime + .handle_incoming( + connection, + json!({"id":47,"method":"session/list","params":{}}), + ) + .await + .context("list roots")?; + assert!(list.to_string().contains(created_id.as_str()), "{list}"); + assert!( + request_agent_list(&runtime, connection, parent) + .await? + .agents + .is_empty() + ); + assert_eq!(provider.stream_calls(), 2); + let journal = std::fs::read_to_string(handle["journal"].as_str().context("journal")?)?; + assert!(journal.contains("independent answer")); + Ok(()) +} + +#[tokio::test] +async fn end_turn_keeps_a_running_child_alive() -> Result<()> { + let home = TempDir::new()?; + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +worker = await rlm.spawn('Continue after parent returns', name='worker') +partial = await rlm.collect(worker, timeout_ms=10) +assert not partial[0].settled +await rlm.end_turn() +Path('ended-ok').write_text('ended') +"#, + ), + StreamScript::Pending, + ])); + let runtime = build_runtime(home.path(), provider.clone())?; + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let parent = start_parent_session(&runtime, connection, home.path()).await?; + start_turn_with_approval_policy( + &runtime, + connection, + parent, + "Yield normally", + Some("never"), + ) + .await?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!(completed["params"]["turn"]["status"], json!("completed")); + assert_eq!( + std::fs::read_to_string(home.path().join("ended-ok"))?, + "ended" + ); + let agents = request_agent_list(&runtime, connection, parent).await?; + assert_eq!(agents.agents[0].status, "running"); + wait_for_stream_calls(&provider, 2).await?; + assert_eq!(provider.stream_calls(), 2); + request_agent_close(&runtime, connection, parent, "worker").await?; + Ok(()) +} + +#[tokio::test] +async fn collect_reports_recoverable_failure_with_its_canonical_error() -> Result<()> { + let home = TempDir::new()?; + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +worker = await rlm.spawn('Fail the model request', name='failure') +result = (await rlm.collect(worker, timeout_ms=10000))[0] +assert not result.settled and result.status == 'waiting_for_resume', result +assert result.requires_resume and result.answer_text is None, result +assert 'scripted provider stream script exhausted' in result.error, result +await rlm.delete_subagent(worker.rlm_child_id) +closed = (await rlm.collect(worker))[0] +assert closed.settled and not closed.requires_resume and closed.status == 'failed', closed +assert closed.error == result.error, closed +Path('failure-ok').write_text('reported') +await rlm.end_turn() +"#, + ), + ])); + let runtime = build_runtime(home.path(), provider.clone())?; + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let parent = start_parent_session(&runtime, connection, home.path()).await?; + start_turn_with_approval_policy( + &runtime, + connection, + parent, + "Collect failure", + Some("never"), + ) + .await?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!(completed["params"]["turn"]["status"], json!("completed")); + assert_eq!( + std::fs::read_to_string(home.path().join("failure-ok"))?, + "reported" + ); + Ok(()) +} + +#[tokio::test] +async fn collect_recognizes_shutdown_interrupted_child_recovery() -> Result<()> { + let home = TempDir::new()?; + let first_provider = Arc::new(ScriptedProvider::pending()); + let first = build_runtime(home.path(), first_provider.clone())?; + let (connection, _) = initialize_connection(&first).await?; + let parent = start_parent_session(&first, connection, home.path()).await?; + spawn_child_named( + &first, + parent, + "Keep working", + Some("none"), + Some("recoverable"), + ) + .await?; + wait_for_stream_calls(&first_provider, 1).await?; + first.shutdown().await; + drop(first); + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +result = (await rlm.collect('recoverable', timeout_ms=300000))[0] +assert result.status == 'waiting_for_resume' and result.requires_resume, result +assert not result.settled and result.answer_text is None, result +Path('shutdown-recovery-ok').write_text('no replay') +await rlm.end_turn() +"#, + ), + ])); + let restored = build_runtime(home.path(), provider.clone())?; + restored.load_persisted_sessions().await?; + let (connection, mut notifications) = initialize_connection(&restored).await?; + let response = restored + .handle_incoming( + connection, + json!({ + "id": 81, "method": "session/resume", "params": {"sessionId": parent} + }), + ) + .await + .context("resume parent")?; + assert!(response.get("error").is_none(), "{response}"); + start_turn_with_approval_policy( + &restored, + connection, + parent, + "Inspect recovery", + Some("never"), + ) + .await?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!( + completed["params"]["turn"]["status"], + json!("completed"), + "{completed}" + ); + assert_eq!( + std::fs::read_to_string(home.path().join("shutdown-recovery-ok"))?, + "no replay" + ); + assert_eq!(provider.stream_calls(), 1); + Ok(()) +} + +#[tokio::test] +async fn collect_waits_through_a_live_queued_followup() -> Result<()> { + let home = TempDir::new()?; + let provider = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +from pathlib import Path +worker = await rlm.spawn('Do the first task', name='queued') +result = (await rlm.collect(worker, timeout_ms=10000))[0] +print(result) +assert result.settled and not result.requires_resume, result +assert result.answer_text == 'queued answer', result +Path('queued-ok').write_text('joined both turns') +await rlm.end_turn() +"#, + ), + ScriptedProvider::ipython_code_tool_call( + r#" +import asyncio +from pathlib import Path +while not Path('queue-installed').exists(): + await asyncio.sleep(0.01) +"#, + ), + ScriptedProvider::completed("first answer"), + ScriptedProvider::completed("queued answer"), + ])); + let runtime = build_runtime(home.path(), provider.clone())?; + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let parent = start_parent_session(&runtime, connection, home.path()).await?; + start_turn_with_approval_policy( + &runtime, + connection, + parent, + "Join queued work", + Some("never"), + ) + .await?; + wait_for_stream_calls(&provider, 2).await?; + request_agent_send_message(&runtime, connection, parent, "queued", "Do the follow-up").await?; + std::fs::write(home.path().join("queue-installed"), "ready")?; + let completed = + wait_for_session_notification(&mut notifications, "turn/completed", parent).await?; + assert_eq!( + completed["params"]["turn"]["status"], + json!("completed"), + "{completed}" + ); + let outputs = provider + .requests() + .into_iter() + .flat_map(|r| r.messages) + .flat_map(|m| m.content) + .filter_map(|content| match content { + devo_protocol::RequestContent::ToolResult { content, .. } => Some(content), + _ => None, + }) + .collect::>(); + assert!( + home.path().join("queued-ok").exists(), + "Python outputs: {outputs:?}" + ); + assert_eq!( + std::fs::read_to_string(home.path().join("queued-ok"))?, + "joined both turns" + ); + Ok(()) +} diff --git a/docs/rlm-orchestration.md b/docs/rlm-orchestration.md new file mode 100644 index 00000000..75347270 --- /dev/null +++ b/docs/rlm-orchestration.md @@ -0,0 +1,93 @@ +# Python subagent orchestration + +These operations run through the real kernel host bridge. They use existing +session actors, durable journals and Native turn admission; they do not create a +second execution engine or require model-provider network access to coordinate. + +```python +worker = await rlm.spawn("Run the focused tests", name="tests") +results = await rlm.collect([worker], timeout_ms=10_000) +for result in results: + print(result.status, result.progress_note, result.answer_text) +await rlm.end_turn() +``` + +## Collect and progress + +`collect(targets=None, *, timeout_ms=0)` accepts a child name/id, spawn handle, +subagent row, or a list/tuple mixing those selectors. Empty selection snapshots +all direct children at admission. Duplicate selectors are returned once. Foreign +sessions and unknown targets fail rather than being silently skipped. + +Timeout is an integer from 0 to 300000 milliseconds. Zero is a snapshot. Positive +values wait until all selected children settle or the deadline expires, then +return current snapshots. Timeout does not cancel work. Interrupted waits fail +promptly. Waiting does not hold the session mailbox or an admission lock, does +not send steering messages, and does not consume results or other wait cursors. +The notification subscription precedes the snapshot to avoid lost wakeups. + +Each `RLMChildResult` contains `rlm_child_id`, `session_name`, `session_dir`, +`status`, `settled`, `answer_text`, `answer_preview`, `answer_truncated`, `error`, +`progress_note`, and `requires_resume`. Status preserves native terminal distinctions (`completed`, +`failed`, `interrupted`, `canceled`, `closed`, `waiting_for_input`); unfinished +snapshots report `running`. Settlement waits for the completed working set to +merge and execution admission to clear, and excludes queued follow-ups and accepted mailbox messages still in transit. Results +read the latest completed turn's assistant text, bounded to 128 Ki characters +with a truncation flag and a 512-character preview. The journal retains larger +output. A resumed child never presents its previous answer as the new answer. + +The host rehydrates durable direct children from the index/journal after restart. +Corrupt or missing child journals are explicit restore errors. Saved unfinished +turns, recoverable provider failures, and shutdown interruptions with no live execution report `waiting_for_resume`, `requires_resume=True`, +and `settled=False`. Collect stops waiting once every target is settled or needs +explicit Native recovery; it never silently starts a new run or replays tools. +An idle restored child with queued follow-ups also requires resume. Use Native +turn/recovery/read to identify resumable turns; approval recovery and queued-only +work use their normal approval/session resume controls. In-process +progress is intentionally transient, not recovered as if it were fresh. + +Children call `await rlm.progress_note("running tests")`. Notes contain 1–512 +UTF-16 code units after trimming and are accepted at most once per 10 seconds. +The result has `accepted` and an optional `retry_after_ms`. Notes appear in +`collect` and `list_subagents`, without messaging or waking the parent model. +Root calls fail. Use `agent_message` for a message that needs a response. + +## End the current turn + +`await rlm.end_turn()` requests normal completion at the next tool-batch boundary. +The rest of the Python cell and sibling calls in that batch still run. Tool +results are committed before completion; no additional model response is +requested. This is not cancellation and does not stop children or background +jobs. Calling it from an idle or obsolete host bridge fails. Approval-replayed +tools do not support this control until normal model execution resumes. + +## Independent root sessions + +```python +root = await rlm.create_session("Investigate the next issue", name="Next issue") +print(root.session_id, root.session_file) +``` + +`create_session(prompt, name=None, model=None, thinking=None, cwd=None)` requires +a durable root in Build mode. It creates a separately visible, durable root with +its own journal, no parent link, no inherited transcript, and no parent usage or +completion lifecycle. Normal session/list, session/resume and turn controls apply. +It is not a child and cannot be collected or deleted through subagent controls. + +Model/thinking use the same exact validated resolver as spawn, with inherited +defaults and no silent fallback. Permissions and sandbox settings are installed +before the first durable metadata write and before SessionStart hooks. Initial +turn admission uses the Native handler. The handle returns only after admission, +not after the initial prompt completes. If creation succeeds but admission +fails, the error includes the existing session id; callers must not blindly retry. + +For this slice, an explicit cwd must canonicalize to the caller's current working +directory. Cross-workspace roots need a separate authorization flow. Child, +ephemeral and Plan-mode callers are rejected before any root is created. + +## Verification + +`crates/server/tests/subagent_python_controls.rs` exercises real CPython cells, +production host dispatch, durable actor merge, nonblocking read RPCs, bounded +waits and interruption, root isolation, and child result recovery after restart. +Python wire and malformed-input cases live in `test_controls.py`.