From 63fa71d047f02869e71d0255d0e3ff0ed8395bc9 Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Sun, 8 Nov 2026 22:49:04 +0800 Subject: [PATCH] feat(rlm): support independently routed child models and discovery --- README.md | 16 + apps/tui/src/generated/protocol/native.ts | 12 +- apps/tui/src/generated/protocol/schema.json | 16 +- crates/core/prompts/rlm/repl_control.md | 2 + crates/core/src/tools/handlers/agent.rs | 4 + crates/kernel/rlm-runtime/README.md | 23 +- crates/kernel/rlm-runtime/src/rlm/__init__.py | 45 +- crates/kernel/rlm-runtime/src/rlm/harness.py | 6 +- .../rlm-runtime/tests/test_child_models.py | 103 ++++ crates/protocol/src/agent.rs | 10 +- crates/provider/src/provider.rs | 6 + crates/provider/src/router.rs | 39 ++ crates/server/src/provider_config.rs | 8 + crates/server/src/runtime.rs | 2 + crates/server/src/runtime/agents.rs | 85 ++- crates/server/src/runtime/agents/lifecycle.rs | 9 +- .../server/src/runtime/approval_checkpoint.rs | 1 + crates/server/src/runtime/command_exec.rs | 2 + .../server/src/runtime/goal_continuation.rs | 21 +- .../server/src/runtime/handlers/compaction.rs | 64 +- .../src/runtime/handlers/message_edit.rs | 17 +- crates/server/src/runtime/handlers/session.rs | 103 +++- crates/server/src/runtime/handlers/turn.rs | 18 +- crates/server/src/runtime/kernel_host.rs | 97 +-- .../server/src/runtime/kernel_host_models.rs | 147 +++++ crates/server/src/runtime/refine_planner.rs | 12 +- .../src/runtime/session_actor/actor_loop.rs | 15 +- .../src/runtime/session_model_config.rs | 32 + .../src/runtime/turn_exec/approval_resume.rs | 14 +- .../server/src/runtime/turn_exec/followup.rs | 108 ++-- .../src/runtime/turn_exec/followup/tests.rs | 196 ++++++ .../server/src/runtime/turn_exec/recovery.rs | 22 +- crates/server/src/session_context.rs | 110 ++-- .../src/session_context/child_models.rs | 439 ++++++++++++++ .../src/session_context/child_models/tests.rs | 530 ++++++++++++++++ .../server/tests/child_metadata_selection.rs | 107 ++++ crates/server/tests/child_model_selection.rs | 569 ++++++++++++++++++ crates/server/tests/goal_background_agents.rs | 2 + crates/server/tests/subagent_lifecycle.rs | 2 + .../tests/support/subagent_lifecycle.rs | 2 + docs/prompt-assembly.md | 5 +- docs/rlm-child-models.md | 68 +++ 42 files changed, 2835 insertions(+), 254 deletions(-) create mode 100644 crates/kernel/rlm-runtime/tests/test_child_models.py create mode 100644 crates/server/src/runtime/kernel_host_models.rs create mode 100644 crates/server/src/runtime/session_model_config.rs create mode 100644 crates/server/src/runtime/turn_exec/followup/tests.rs create mode 100644 crates/server/src/session_context/child_models.rs create mode 100644 crates/server/src/session_context/child_models/tests.rs create mode 100644 crates/server/tests/child_metadata_selection.rs create mode 100644 crates/server/tests/child_model_selection.rs create mode 100644 docs/rlm-child-models.md diff --git a/README.md b/README.md index eca60f25..64c51e8a 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,22 @@ The agent can filter large outputs in Python, reuse earlier work, and delegate independent subtasks to child agents with their own sessions and Python state. MCP, skills, permissions, and resumable sessions round out the workflow. +Children can independently select a configured model and reasoning level: + +```python +models = await rlm.find_models(limit=100) +chosen = next((m for m in models if "high" in m.thinking_levels), None) +if chosen is not None: + reviewer = await rlm.spawn( + "Review the current diff.", name="reviewer", + model=chosen.selector, thinking="high", + ) +``` + +Omitted options inherit the immediate parent. Invalid selections fail without +fallback. Discovery reflects configuration, not live health or credits. See the +[child model API](docs/rlm-child-models.md). + ## Quick start **Desktop:** Download the package for your OS and architecture from diff --git a/apps/tui/src/generated/protocol/native.ts b/apps/tui/src/generated/protocol/native.ts index 6f5933d8..be27533a 100644 --- a/apps/tui/src/generated/protocol/native.ts +++ b/apps/tui/src/generated/protocol/native.ts @@ -1002,7 +1002,15 @@ export type SearchFailedPayload = { searchId: ReferenceSearchId, query: string, export type AgentToolPolicy = "inherit" | "deny_all"; -export type SpawnAgentParams = { session_id: SessionId, message: string, fork_turns?: string | null, max_turns?: number | null, tool_policy: AgentToolPolicy, ephemeral: boolean, +export type SpawnAgentParams = { session_id: SessionId, message: string, +/** + * Exact connection/model selector. Omission inherits the immediate parent. + */ +model?: string | null, +/** + * Reasoning selection validated against the child model, never clamped. + */ +thinking?: string | null, fork_turns?: string | null, max_turns?: number | null, tool_policy: AgentToolPolicy, ephemeral: boolean, /** * Caller-requested child name. RLM `spawn(name=…)` addressing (messaging, * observation, delete) keys off this name, so it must be honored when @@ -1017,7 +1025,7 @@ export type SpawnAgentResult = { task_id: TaskId, child_session_id: SessionId, a */ session_dir?: string | null, /** - * Model binding the child runs with (inherited from the parent). + * Canonical model selector bound to the child session. */ model?: string | null, }; diff --git a/apps/tui/src/generated/protocol/schema.json b/apps/tui/src/generated/protocol/schema.json index 0727403e..bf6357bd 100644 --- a/apps/tui/src/generated/protocol/schema.json +++ b/apps/tui/src/generated/protocol/schema.json @@ -173689,6 +173689,13 @@ "message": { "type": "string" }, + "model": { + "description": "Exact connection/model selector. Omission inherits the immediate parent.", + "type": [ + "string", + "null" + ] + }, "nickname": { "description": "Caller-requested child name. RLM `spawn(name=…)` addressing (messaging, observation, delete) keys off this name, so it must be honored when present; `None` keeps the generated adjective-noun nickname.", "type": [ @@ -173699,6 +173706,13 @@ "session_id": { "$ref": "#/definitions/SessionId" }, + "thinking": { + "description": "Reasoning selection validated against the child model, never clamped.", + "type": [ + "string", + "null" + ] + }, "tool_policy": { "default": "inherit", "allOf": [ @@ -173743,7 +173757,7 @@ "$ref": "#/definitions/SessionId" }, "model": { - "description": "Model binding the child runs with (inherited from the parent).", + "description": "Canonical model selector bound to the child session.", "type": [ "string", "null" diff --git a/crates/core/prompts/rlm/repl_control.md b/crates/core/prompts/rlm/repl_control.md index 15927da8..7ce588fc 100644 --- a/crates/core/prompts/rlm/repl_control.md +++ b/crates/core/prompts/rlm/repl_control.md @@ -20,6 +20,8 @@ Terminology: continual harness names the persisted prompt, memory, skill, and su 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. +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. + RLM host APIs are called from Python through `ipython`, not as separate model tools: - `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. diff --git a/crates/core/src/tools/handlers/agent.rs b/crates/core/src/tools/handlers/agent.rs index 9fb79b78..5dea9be2 100644 --- a/crates/core/src/tools/handlers/agent.rs +++ b/crates/core/src/tools/handlers/agent.rs @@ -166,6 +166,8 @@ impl ToolHandler for AgentToolHandler { let input: SpawnAgentInput = parse_input(input)?; let result = coordinator .spawn_agent(SpawnAgentParams { + model: None, + thinking: None, session_id, message: input.message, fork_turns: input.fork_turns, @@ -707,6 +709,8 @@ mod tests { assert_eq!( coordinator.spawned.lock().await.as_slice(), &[SpawnAgentParams { + model: None, + thinking: None, session_id, message: "review this".to_string(), fork_turns: Some("all".to_string()), diff --git a/crates/kernel/rlm-runtime/README.md b/crates/kernel/rlm-runtime/README.md index 766e0366..220919f8 100644 --- a/crates/kernel/rlm-runtime/README.md +++ b/crates/kernel/rlm-runtime/README.md @@ -179,18 +179,29 @@ import rlm result = await rlm.bash("cargo test -p devo-kernel") print(result.exit_code, result.output) -reviewer = await rlm.spawn( - "Review the current diff for concurrency bugs.", - name="reviewer", - model="openai-codex/gpt-5.4-mini", -) -print(reviewer.model) +models = await rlm.find_models(limit=100) +chosen = next((m for m in models if "high" in m.thinking_levels), None) +if chosen is not None: + reviewer = await rlm.spawn( + "Review the current diff for concurrency bugs.", + name="reviewer", model=chosen.selector, thinking="high", + ) + print(reviewer.model) ``` `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. +Omitted `model` and `thinking` (or `None`) inherit the immediate parent's values. +An unsupported inherited thinking level fails just like an explicit unsupported +level; there is no silent fallback. `find_models` returns enabled, configured, +routable models and supported thinking levels without credentials or network +probes. It does not guarantee provider health or credits. Selectors are exact +`provider/model` strings, and model IDs can contain slashes. `limit` must be an +integer from 1 to 100, not a boolean. See the +[child model API](../../../docs/rlm-child-models.md) for details. + `rlm.read` and `rlm.write` try the direct filesystem call first. On `OSError` they ask the host to perform the operation through its approval pipeline. A host denial surfaces as `PermissionError`. These helpers are an entry point, not diff --git a/crates/kernel/rlm-runtime/src/rlm/__init__.py b/crates/kernel/rlm-runtime/src/rlm/__init__.py index a675e079..9fff1294 100644 --- a/crates/kernel/rlm-runtime/src/rlm/__init__.py +++ b/crates/kernel/rlm-runtime/src/rlm/__init__.py @@ -4,7 +4,7 @@ import sys import types -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -39,6 +39,7 @@ class RLMModel: id: str name: str selector: str + thinking_levels: list[str] = field(default_factory=list) @dataclass(frozen=True) @@ -268,12 +269,27 @@ async def spawn( """Spawn a recursive Devo child and return once its task is admitted. ``name`` is required and must be unique among siblings. - ``model`` selects a child with an exact ``provider/model`` selector. + ``model`` selects a child with an exact ``provider/model`` selector from + ``find_models`` (model IDs may contain slashes); there is no fallback. ``thinking`` sets the child reasoning level (e.g. 'off', 'low', 'medium', 'high'); - defaults to the parent level; levels invalid for the resolved model fail the spawn. + Omitted or None options inherit the immediate parent's model and thinking. + Levels invalid for the resolved model, including inherited levels, fail the spawn. """ 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)): + if value is None and label != "name": + continue + if not isinstance(value, str): + raise TypeError(f"{label} must be str, got {type(value).__name__}") + if not value.strip(): + raise ValueError(f"{label} must be a non-empty string") + if model is not None: + provider, separator, model_id = model.partition("/") + if not separator or not provider.strip() or not model_id.strip(): + raise ValueError("model must be an exact provider/model selector") kwargs: dict[str, Any] = {"name": name} if model is not None: kwargs["model"] = model @@ -293,7 +309,15 @@ def _model_from_payload(payload: Any) -> RLMModel: selector = payload.get("selector") if not all(isinstance(value, str) and value for value in (provider, model_id, name, selector)): raise RuntimeError("rlm.find_models returned an invalid model entry") - return RLMModel(provider=provider, id=model_id, name=name, selector=selector) + thinking_levels = payload.get("thinking_levels", []) + if not isinstance(thinking_levels, list) or not all( + isinstance(level, str) and level.strip() for level in thinking_levels + ): + raise RuntimeError("rlm.find_models returned invalid thinking_levels") + return RLMModel( + provider=provider, id=model_id, name=name, selector=selector, + thinking_levels=thinking_levels, + ) async def create_session( @@ -324,12 +348,21 @@ async def create_session( async def find_models(query: str = "", limit: int = 8) -> list[RLMModel]: - """Search a bounded list of models backed by active user credentials.""" + """Search enabled, configured, routable models without network probes. + + Availability describes local configuration, not live health or credit. Each + descriptor includes supported thinking levels; older hosts may omit them. + ``limit`` must be an integer from 1 to 100 (not bool). + """ if not isinstance(query, str): raise TypeError(f"query must be str, got {type(query).__name__}") - if not isinstance(limit, int): + if isinstance(limit, bool) or not isinstance(limit, int): raise TypeError(f"limit must be int, got {type(limit).__name__}") + if not 1 <= limit <= 100: + raise ValueError("limit must be between 1 and 100") payload = await host_request("rlm.find_models", {"query": query, "limit": limit}) + if not isinstance(payload, dict): + raise RuntimeError("rlm.find_models returned an invalid models list") models = payload.get("models") if not isinstance(models, list): raise RuntimeError("rlm.find_models returned an invalid models list") diff --git a/crates/kernel/rlm-runtime/src/rlm/harness.py b/crates/kernel/rlm-runtime/src/rlm/harness.py index 11f9c6ba..5f5d629a 100644 --- a/crates/kernel/rlm-runtime/src/rlm/harness.py +++ b/crates/kernel/rlm-runtime/src/rlm/harness.py @@ -629,7 +629,11 @@ def overview(self, *, max_entries_per_kind: int = 20, global_: bool = False, **k "harness skill entries are Python REPL skills and must include a Python reference plus arguments. " "Spawn a subagent spec by composing a concise task prompt and calling " "handle = await rlm.spawn('sub-task', name='worker'); admission returns immediately with rlm_child_id, name, session_dir, " - "and model, never the child's answer. Results arrive only through explicit agent_message replies or " + "and model. Discover configured models with await rlm.find_models(query='', limit=8); descriptors include selector " + "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 " "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_child_models.py b/crates/kernel/rlm-runtime/tests/test_child_models.py new file mode 100644 index 00000000..6e62dae5 --- /dev/null +++ b/crates/kernel/rlm-runtime/tests/test_child_models.py @@ -0,0 +1,103 @@ +import unittest +from importlib import import_module +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import rlm + +repl = import_module("rlm.repl") + + +class ChildModelTests(unittest.IsolatedAsyncioTestCase): + async def test_harness_documents_discovery_and_no_fallback(self): + harness = import_module("rlm.harness") + overview = harness.HarnessState(in_memory=True).overview() + for guidance in ("rlm.find_models", "thinking_levels", "model=chosen.selector", + "immediate parent", "without fallback"): + self.assertIn(guidance, overview) + + async def test_spawn_wire_options_and_inheritance(self): + response = { + "rlm_child_id": "child", "name": "worker", + "session_dir": "child-session", "model": "gateway/org/model", + } + bridge = AsyncMock(return_value={"status": "ok", "result": response}) + with patch.object(repl, "host_request", bridge): + for options in ({}, {"model": None, "thinking": None}, + {"model": "gateway/org/model", "thinking": "high"}): + handle = await rlm.rlm.spawn("Review code", name="worker", **options) + self.assertEqual(handle, rlm.RLMSpawnHandle( + **{**response, "session_dir": Path("child-session")})) + bridge.assert_awaited_with({ + "type": "rlm.run", "prompt": "Review code", + "kwargs": {"name": "worker", **{ + key: value for key, value in options.items() if value is not None}}, + }) + + async def test_invalid_spawn_arguments_do_not_reach_host(self): + cases = [({"prompt": value}, TypeError) for value in (None, 1, True)] + cases += [({"prompt": value}, ValueError) for value in ("", " ")] + for field in ("name", "model", "thinking"): + cases += [({field: value}, TypeError) for value in (1, True, [])] + cases += [({field: value}, ValueError) for value in ("", " ")] + cases += [({"name": None}, TypeError)] + cases += [({"model": value}, ValueError) for value in ("model", "/id", "provider/", " /id")] + bridge = AsyncMock() + with patch.object(repl, "host_request", bridge): + for overrides, error in cases: + with self.subTest(overrides=overrides), self.assertRaises(error): + await rlm.spawn(**{"prompt": "Task", "name": "worker", **overrides}) + bridge.assert_not_awaited() + + async def test_find_models_wire_and_legacy_descriptors(self): + legacy = {"provider": "gateway", "id": "org/model", "name": "Model", + "selector": "gateway/org/model"} + current = {**legacy, "thinking_levels": ["off", "high"]} + bridge = AsyncMock(return_value={"status": "ok", "result": {"models": [legacy, current]}}) + with patch.object(repl, "host_request", bridge): + self.assertEqual(await rlm.rlm.find_models("org", 100), [ + rlm.RLMModel(**legacy), rlm.RLMModel(**current)]) + bridge.assert_awaited_with({"type": "rlm.find_models", "query": "org", "limit": 100}) + await rlm.find_models() + bridge.assert_awaited_with({"type": "rlm.find_models", "query": "", "limit": 8}) + await rlm.find_models(limit=1) + + async def test_find_models_invalid_arguments_do_not_reach_host(self): + bridge = AsyncMock() + with patch.object(repl, "host_request", bridge): + for query in (None, 1, [], True): + with self.subTest(query=query), self.assertRaises(TypeError): + await rlm.find_models(query) + for limit in (True, False, "8", 1.0, None): + with self.subTest(limit=limit), self.assertRaises(TypeError): + await rlm.find_models(limit=limit) + for limit in (0, -1, 101): + with self.subTest(limit=limit), self.assertRaises(ValueError): + await rlm.find_models(limit=limit) + bridge.assert_not_awaited() + + async def test_invalid_catalog_responses(self): + base = {"provider": "p", "id": "m", "name": "M", "selector": "p/m"} + payloads = [None, [], {}, {"models": None}, {"models": [None]}, {"models": [{}]}] + payloads += [{"models": [{**base, "thinking_levels": levels}]} + for levels in (None, "high", [True], [""], [" "])] + for payload in payloads: + with self.subTest(payload=payload), patch.object(repl, "host_request", AsyncMock( + return_value={"status": "ok", "result": payload} + )), self.assertRaises(RuntimeError): + await rlm.find_models() + + async def test_host_errors_propagate_without_retry_or_fallback(self): + for operation in ("spawn", "find_models"): + bridge = AsyncMock(return_value={"status": "error", "error": "unsupported inherited thinking"}) + with self.subTest(operation=operation), patch.object(repl, "host_request", bridge): + with self.assertRaisesRegex(RuntimeError, "unsupported inherited thinking"): + if operation == "spawn": + await rlm.spawn("Task", name="worker", model="gateway/org/model") + else: + await rlm.find_models() + self.assertEqual(bridge.await_count, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/crates/protocol/src/agent.rs b/crates/protocol/src/agent.rs index c7d5209d..4dc3301b 100644 --- a/crates/protocol/src/agent.rs +++ b/crates/protocol/src/agent.rs @@ -37,6 +37,12 @@ impl AgentOutputEventKind { pub struct SpawnAgentParams { pub session_id: SessionId, pub message: String, + /// Exact connection/model selector. Omission inherits the immediate parent. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub model: Option, + /// Reasoning selection validated against the child model, never clamped. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thinking: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub fork_turns: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -63,7 +69,7 @@ pub struct SpawnAgentResult { /// RLM `rlm.spawn` handles require it to address child artifacts. #[serde(default, skip_serializing_if = "Option::is_none")] pub session_dir: Option, - /// Model binding the child runs with (inherited from the parent). + /// Canonical model selector bound to the child session. #[serde(default, skip_serializing_if = "Option::is_none")] pub model: Option, } @@ -285,6 +291,8 @@ mod tests { let child_session_id = SessionId::new(); let payloads = serde_json::json!({ "spawn": SpawnAgentParams { + model: None, + thinking: None, session_id, message: "review this".to_string(), fork_turns: Some("all".to_string()), diff --git a/crates/provider/src/provider.rs b/crates/provider/src/provider.rs index b97da3bd..31ff5b8d 100644 --- a/crates/provider/src/provider.rs +++ b/crates/provider/src/provider.rs @@ -95,6 +95,12 @@ pub trait ProviderAdapter: ModelProviderSDK { #[allow(clippy::double_must_use)] #[async_trait] pub trait ModelProviderSDK: Send + Sync { + /// Whether this adapter is locally configured for requests. This never probes + /// the service; unavailable placeholders must override it to return false. + fn is_available(&self) -> bool { + true + } + /// Send a request and get a complete response. async fn completion(&self, request: ModelRequest) -> anyhow::Result; diff --git a/crates/provider/src/router.rs b/crates/provider/src/router.rs index c7d1e707..75eb76bb 100644 --- a/crates/provider/src/router.rs +++ b/crates/provider/src/router.rs @@ -74,6 +74,12 @@ pub trait ProviderRouter: Send + Sync { request: ModelRequest, ) -> Result; + /// Checks whether a route has a usable adapter without making a request. + /// Custom routers must opt in; unknown routes are never assumed routable. + fn has_available_route(&self, _route: &ProviderRoute) -> bool { + false + } + /// Human-readable name of the router (e.g. "multi-provider", "openai-only"). fn name(&self) -> &str; } @@ -94,6 +100,10 @@ impl SingleProviderRouter { #[async_trait] impl ProviderRouter for SingleProviderRouter { + fn has_available_route(&self, route: &ProviderRoute) -> bool { + matches!(route, ProviderRoute::Default) && self.provider.is_available() + } + async fn stream( &self, _route: ProviderRoute, @@ -170,6 +180,11 @@ impl MultiProviderRouter { #[async_trait] impl ProviderRouter for MultiProviderRouter { + fn has_available_route(&self, route: &ProviderRoute) -> bool { + self.provider_for_route(route) + .is_ok_and(|provider| provider.is_available()) + } + async fn stream( &self, route: ProviderRoute, @@ -222,6 +237,7 @@ mod tests { #[derive(Default)] struct CapturingProvider { requests: Mutex>, + unavailable: bool, } impl CapturingProvider { @@ -235,6 +251,10 @@ mod tests { #[async_trait] impl ModelProviderSDK for CapturingProvider { + fn is_available(&self) -> bool { + !self.unavailable + } + async fn completion(&self, request: ModelRequest) -> anyhow::Result { self.requests .lock() @@ -266,6 +286,25 @@ mod tests { } } + #[test] + fn route_availability_checks_adapters_without_sending_requests() { + let provider = Arc::new(CapturingProvider::default()); + let mut router = MultiProviderRouter::new(provider.clone()); + let route = ProviderRoute::connection("local", ProviderWireApi::OpenAIResponses); + assert!(!router.has_available_route(&route)); + router.insert_route(route.clone(), provider.clone()); + assert!(router.has_available_route(&route)); + router.insert_route( + route.clone(), + Arc::new(CapturingProvider { + unavailable: true, + ..Default::default() + }), + ); + assert!(!router.has_available_route(&route)); + assert_eq!(provider.requests(), Vec::::new()); + } + fn request(model: &str) -> ModelRequest { ModelRequest { model_slug: devo_protocol::ModelProfileKey::Generic, diff --git a/crates/server/src/provider_config.rs b/crates/server/src/provider_config.rs index 75549822..77d9a189 100644 --- a/crates/server/src/provider_config.rs +++ b/crates/server/src/provider_config.rs @@ -147,6 +147,10 @@ struct MissingProvider; #[async_trait::async_trait] impl ModelProviderSDK for MissingProvider { + fn is_available(&self) -> bool { + false + } + async fn completion(&self, _request: ModelRequest) -> Result { anyhow::bail!(NO_PROVIDER_CONFIGURED_MESSAGE) } @@ -175,6 +179,10 @@ impl UnavailableProvider { #[async_trait::async_trait] impl ModelProviderSDK for UnavailableProvider { + fn is_available(&self) -> bool { + false + } + async fn completion(&self, _request: ModelRequest) -> Result { anyhow::bail!("{}", self.message) } diff --git a/crates/server/src/runtime.rs b/crates/server/src/runtime.rs index a70035a2..5a7e58fd 100644 --- a/crates/server/src/runtime.rs +++ b/crates/server/src/runtime.rs @@ -121,6 +121,7 @@ mod kernel_host; mod kernel_host_agent_message; mod kernel_host_bridge; mod kernel_host_heartbeat; +mod kernel_host_models; mod kernel_idle_bash; mod lifecycle; mod mcp; @@ -141,6 +142,7 @@ mod session_cache; mod session_deletion; pub(crate) mod session_file_lock; mod session_interactive; +mod session_model_config; mod session_title; mod skills; mod subagent_usage; diff --git a/crates/server/src/runtime/agents.rs b/crates/server/src/runtime/agents.rs index 5acbc24c..966eceb8 100644 --- a/crates/server/src/runtime/agents.rs +++ b/crates/server/src/runtime/agents.rs @@ -110,9 +110,68 @@ impl ServerRuntime { .clone() .unwrap_or_else(|| "root".to_string()); let agent_path = AgentPath::new(parent_path).join(&nickname).0; - let model = parent_summary.model_name().map(str::to_string); - let model_binding_id = parent_summary.model_binding_id().map(str::to_string); - let reasoning_effort_selection = parent_summary.settings.reasoning_effort.clone(); + // Resolve before persistence/admission. The child owns this binding; + // neither the shared context nor its global defaults are changed. + // A frozen spawn snapshot supplies history, but a live settings change + // may have selected a new parent model since this turn began. + let live_parent_config = + if let Some(stream) = self.active_stream_state(parent_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_parent_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(|| session_model_selection(&parent_summary).map(str::to_string)) + .ok_or_else(|| { + ToolCallError::InvalidInput("parent session has no model binding".to_string()) + })?; + let parent_thinking = match live_parent_config.as_ref() { + Some(config) => config.reasoning_effort_selection.as_deref(), + None => parent_summary.settings.reasoning_effort.as_deref(), + }; + let (_, child_turn_config) = runtime_context + .resolve_child_model( + params.model.as_deref(), + &parent_model, + params.thinking.as_deref(), + parent_thinking, + ) + .map_err(|error| ToolCallError::InvalidInput(error.to_string()))?; + let model = Some( + child_turn_config + .model_binding_id + .clone() + .unwrap_or_else(|| child_turn_config.model.slug.clone()), + ); + let persisted_model = child_turn_config + .variant + .as_ref() + .map(|variant| format!("{}/{variant}", model.as_deref().unwrap_or_default())) + .unwrap_or_else(|| model.clone().unwrap_or_default()); + let model_binding_id = child_turn_config + .model_binding_id + .as_ref() + .map(|_| persisted_model.clone()); + let reasoning_effort_selection = child_turn_config.reasoning_effort_selection.clone(); let mut native_session = parent_summary.native.clone(); native_session.id = child_session_id; @@ -146,7 +205,7 @@ impl ServerRuntime { native_session.title_state = SessionTitleState::Final(SessionTitleFinalSource::ExplicitCreate); native_session.model.provider = model_binding_id.unwrap_or_else(|| "unknown".to_string()); - native_session.model.model = model.clone().unwrap_or_default(); + native_session.model.model = persisted_model; native_session.model.reasoning_effort = None; native_session.settings.reasoning_effort = reasoning_effort_selection; native_session.settings.mode = Some("build".to_string()); @@ -575,7 +634,7 @@ impl ServerRuntime { .await; } - async fn set_agent_status( + pub(in crate::runtime) async fn set_agent_status( &self, parent_session_id: SessionId, child_session_id: SessionId, @@ -776,10 +835,14 @@ impl ServerRuntime { return Ok(active_turn); } - let turn_config = reservation.runtime_context.resolve_turn_config( - session_model_selection(&reservation.summary), - reservation.summary.settings.reasoning_effort.clone(), - ); + let turn_config = reservation + .runtime_context + .resolve_session_turn_config( + &reservation.summary, + session_model_selection(&reservation.summary), + reservation.summary.settings.reasoning_effort.clone(), + ) + .map_err(|error| ToolCallError::InvalidInput(error.to_string()))?; let resolved_request = turn_config .model .resolve_reasoning_effort_selection(turn_config.reasoning_effort_selection.as_deref()); @@ -806,7 +869,7 @@ impl ServerRuntime { .clone() .unwrap_or_else(|| "unknown".to_string()), model: request_model, - variant: None, + variant: turn_config.variant.clone(), reasoning_effort: resolved_request.effective_reasoning_effort, }, collaboration_mode: Some(devo_protocol::CollaborationMode::Build), @@ -1265,7 +1328,7 @@ impl ServerRuntime { .await; } - async fn record_subagent_status_event_with_text( + pub(in crate::runtime) async fn record_subagent_status_event_with_text( &self, parent_session_id: SessionId, child_session_id: SessionId, diff --git a/crates/server/src/runtime/agents/lifecycle.rs b/crates/server/src/runtime/agents/lifecycle.rs index a2ec4d1a..cd879b93 100644 --- a/crates/server/src/runtime/agents/lifecycle.rs +++ b/crates/server/src/runtime/agents/lifecycle.rs @@ -102,7 +102,7 @@ impl ServerRuntime { .await; } - pub(super) async fn fail_child_agent_startup( + pub(in crate::runtime) async fn fail_child_agent_startup( &self, parent_session_id: SessionId, child_session_id: SessionId, @@ -110,7 +110,8 @@ impl ServerRuntime { ) { self.set_agent_status(parent_session_id, child_session_id, SubagentStatus::Failed) .await; - if let Some(session_handle) = self.sessions.lock().await.get(&child_session_id).cloned() { + let session_handle = self.sessions.lock().await.get(&child_session_id).cloned(); + if let Some(session_handle) = session_handle { session_handle.set_runtime_session_idle(None).await; } self.broadcast_notification( @@ -473,6 +474,8 @@ mod tests { let child = Arc::clone(&runtime) .spawn_agent(SpawnAgentParams { + model: None, + thinking: None, session_id: parent_session_id, message: "spawn a nested worker".into(), fork_turns: Some("none".into()), @@ -484,6 +487,8 @@ mod tests { .await?; let grandchild = Arc::clone(&runtime) .spawn_agent(SpawnAgentParams { + model: None, + thinking: None, session_id: child.child_session_id, message: "verify a nested path".into(), fork_turns: Some("none".into()), diff --git a/crates/server/src/runtime/approval_checkpoint.rs b/crates/server/src/runtime/approval_checkpoint.rs index b6092484..b64274c2 100644 --- a/crates/server/src/runtime/approval_checkpoint.rs +++ b/crates/server/src/runtime/approval_checkpoint.rs @@ -269,6 +269,7 @@ impl ServerRuntime { "modelSlug": snapshot.turn_config.model.slug, "requestModel": snapshot.turn_config.request_model, "modelBindingId": snapshot.turn_config.model_binding_id, + "variant": snapshot.turn_config.variant, "reasoningEffortSelection": snapshot.turn_config.reasoning_effort_selection, "collaborationMode": format!("{:?}", snapshot.collaboration_mode), }); diff --git a/crates/server/src/runtime/command_exec.rs b/crates/server/src/runtime/command_exec.rs index 461c3fdd..5603ec79 100644 --- a/crates/server/src/runtime/command_exec.rs +++ b/crates/server/src/runtime/command_exec.rs @@ -738,6 +738,8 @@ impl ServerRuntime { agent_policies.expect("agent branch always carries policies"); let result = match Arc::clone(self) .spawn_agent(devo_protocol::SpawnAgentParams { + model: None, + thinking: None, session_id: legacy_session_id, message: input, fork_turns, diff --git a/crates/server/src/runtime/goal_continuation.rs b/crates/server/src/runtime/goal_continuation.rs index bbe7d1f7..0bc90ddc 100644 --- a/crates/server/src/runtime/goal_continuation.rs +++ b/crates/server/src/runtime/goal_continuation.rs @@ -57,9 +57,22 @@ impl ServerRuntime { let requested_model = session_model_selection(&reservation.summary); let requested_reasoning_effort_selection = reservation.summary.settings.reasoning_effort.clone(); - let turn_config = reservation - .runtime_context - .resolve_turn_config(requested_model, requested_reasoning_effort_selection); + let turn_config = match reservation.runtime_context.resolve_session_turn_config( + &reservation.summary, + requested_model, + requested_reasoning_effort_selection, + ) { + Ok(config) => config, + Err(error) => { + tracing::warn!(%session_id, %error, "child goal model is unavailable"); + drop(_state_change_guard); + if let Some(parent) = reservation.summary.parent_session_id() { + self.fail_child_agent_startup(parent, session_id, error.to_string()) + .await; + } + return; + } + }; let resolved_request = turn_config.model.resolve_reasoning_effort_selection( turn_config.reasoning_effort_selection.as_deref(), ); @@ -85,7 +98,7 @@ impl ServerRuntime { .clone() .unwrap_or_else(|| "unknown".to_string()), model: request_model, - variant: None, + variant: turn_config.variant.clone(), reasoning_effort: resolved_request.effective_reasoning_effort, }, collaboration_mode: None, diff --git a/crates/server/src/runtime/handlers/compaction.rs b/crates/server/src/runtime/handlers/compaction.rs index d7a3057b..3e6a8a05 100644 --- a/crates/server/src/runtime/handlers/compaction.rs +++ b/crates/server/src/runtime/handlers/compaction.rs @@ -142,9 +142,20 @@ impl ServerRuntime { let requested_model = session_model_selection(&reservation.summary); let requested_reasoning_effort_selection = reservation.summary.settings.reasoning_effort.clone(); - let turn_config = reservation - .runtime_context - .resolve_turn_config(requested_model, requested_reasoning_effort_selection); + let turn_config = match reservation.runtime_context.resolve_session_turn_config( + &reservation.summary, + requested_model, + requested_reasoning_effort_selection, + ) { + Ok(config) => config, + Err(error) => { + return self.error_response( + request_id, + ProtocolErrorCode::InvalidParams, + error.to_string(), + ); + } + }; let resolved_request = turn_config .model .resolve_reasoning_effort_selection(turn_config.reasoning_effort_selection.as_deref()); @@ -169,7 +180,7 @@ impl ServerRuntime { .clone() .unwrap_or_else(|| "unknown".to_string()), model: request_model, - variant: None, + variant: turn_config.variant.clone(), reasoning_effort: resolved_request.effective_reasoning_effort, }, collaboration_mode: None, @@ -388,9 +399,22 @@ impl ServerRuntime { let requested_model = session_model_selection(&reservation.summary); let requested_reasoning_effort_selection = reservation.summary.settings.reasoning_effort.clone(); - let turn_config = reservation - .runtime_context - .resolve_turn_config(requested_model, requested_reasoning_effort_selection); + let turn_config = match reservation.runtime_context.resolve_session_turn_config( + &reservation.summary, + requested_model, + requested_reasoning_effort_selection, + ) { + Ok(config) => config, + Err(error) => { + tracing::warn!(%session_id, %error, "child compaction model is unavailable"); + drop(_state_change_guard); + if let Some(parent) = reservation.summary.parent_session_id() { + self.fail_child_agent_startup(parent, session_id, error.to_string()) + .await; + } + return AgentCompactionOutcome::Consumed; + } + }; let resolved_request = turn_config.model.resolve_reasoning_effort_selection( turn_config.reasoning_effort_selection.as_deref(), ); @@ -415,7 +439,7 @@ impl ServerRuntime { .clone() .unwrap_or_else(|| "unknown".to_string()), model: request_model, - variant: None, + variant: turn_config.variant.clone(), reasoning_effort: resolved_request.effective_reasoning_effort, }, collaboration_mode: None, @@ -616,10 +640,28 @@ impl ServerRuntime { let model_selection = session_model_selection(&runtime_session.summary) .unwrap_or(&runtime_session.runtime_context.default_model); - let turn_config = runtime_session.runtime_context.resolve_turn_config( + let turn_config = match runtime_session.runtime_context.resolve_session_turn_config( + &runtime_session.summary, Some(model_selection), - /*reasoning_effort_selection*/ None, - ); + /*thinking*/ None, + ) { + Ok(config) => config, + Err(error) => { + drop(core_session); + drop(runtime_session); + self.finalize_manual_compaction_turn( + &session_handle, + session_id, + turn, + CompactionTurnOutcome::Failed { + message: error.to_string(), + }, + Some(compaction_item_id), + ) + .await; + return; + } + }; let resolved_request = turn_config.model.resolve_reasoning_effort_selection(None); let model_slug = resolved_request.request_model; let request_model = turn_config.provider_request_model(&model_slug); diff --git a/crates/server/src/runtime/handlers/message_edit.rs b/crates/server/src/runtime/handlers/message_edit.rs index df053cc7..9a4ddfc7 100644 --- a/crates/server/src/runtime/handlers/message_edit.rs +++ b/crates/server/src/runtime/handlers/message_edit.rs @@ -492,16 +492,27 @@ impl ServerRuntime { let requested_reasoning_effort_selection = session.summary.settings.reasoning_effort.clone(); let runtime_context = Arc::clone(&session.runtime_context); + let model_summary = session.summary.clone(); let collaboration_mode = { let core_session = session.core_session.lock().await; core_session.collaboration_mode }; drop(session); - let turn_config = runtime_context.resolve_turn_config( + let turn_config = match runtime_context.resolve_session_turn_config( + &model_summary, requested_model.as_deref(), requested_reasoning_effort_selection.clone(), - ); + ) { + Ok(config) => config, + Err(error) => { + return self.error_response( + request_id, + ProtocolErrorCode::InvalidParams, + error.to_string(), + ); + } + }; let resolved_request = turn_config .model .resolve_reasoning_effort_selection(turn_config.reasoning_effort_selection.as_deref()); @@ -580,7 +591,7 @@ impl ServerRuntime { .clone() .unwrap_or_else(|| "unknown".to_string()), model: request_model, - variant: None, + variant: turn_config.variant.clone(), reasoning_effort: resolved_request.effective_reasoning_effort, }, collaboration_mode: None, diff --git a/crates/server/src/runtime/handlers/session.rs b/crates/server/src/runtime/handlers/session.rs index 7d491f23..6d58db3a 100644 --- a/crates/server/src/runtime/handlers/session.rs +++ b/crates/server/src/runtime/handlers/session.rs @@ -755,6 +755,62 @@ impl ServerRuntime { index_session.settings.clone(), ) }; + // Indexed children are rejected above. Ephemeral children can reach + // this path through the actor-backed metadata fallback; validate their + // exact selection before normalizing effort or persisting any settings. + let child_session = durable_rollout_session + .as_deref() + .or(index_session.as_ref()) + .filter(|session| session.parent.is_some()); + let validated_child_config = if let Some(session) = child_session + && (params.model.is_some() + || params.model_binding_id.is_some() + || params + .settings + .as_ref() + .is_some_and(|settings| settings.reasoning_effort.is_some())) + { + let summary = crate::runtime_session_summary::RuntimeSessionSummary::new( + session.clone(), + Utc::now(), + devo_protocol::CollaborationMode::default(), + ); + let model = params.model_binding_id.clone().or_else(|| { + params.model.as_ref().map(|binding| { + let mut selection = native_model_slug(binding); + if let Some(variant) = binding.variant.as_deref() + && !selection.ends_with(&format!("/{variant}")) + { + selection = format!("{selection}/{variant}"); + } + selection + }) + }); + let thinking = params + .settings + .as_ref() + .and_then(|settings| settings.reasoning_effort.clone()) + .or_else(|| session.settings.reasoning_effort.clone()); + let result = match self.deps.context_for_workspace(&session_cwd).await { + Ok(context) => { + context.resolve_session_turn_config(&summary, model.as_deref(), thinking) + } + Err(error) => Err(error), + }; + match result { + Ok(config) => Some(config), + Err(error) => { + return self.error_response( + request_id, + ProtocolErrorCode::InvalidParams, + error.to_string(), + ); + } + } + } else { + None + }; + let is_child_session = child_session.is_some(); let mut overlay_profile: Option = None; let mut overlay_sandbox: Option = None; let mut overlay_effort: Option = None; @@ -838,21 +894,30 @@ impl ServerRuntime { .as_ref() .map(native_model_slug) .unwrap_or_else(|| session_model_slug.clone()); - let normalized = match self.deps.context_for_workspace(&session_cwd).await { - Ok(runtime_context) => runtime_context - .resolve_turn_config(Some(&requested_model), Some(effort.clone())) + let normalized = if let Some(config) = &validated_child_config { + config .reasoning_effort_selection - .unwrap_or_else(|| { - devo_protocol::normalize_reasoning_effort_literal(effort) - }), - Err(_) => self - .deps - .model_catalog - .get(&requested_model) - .and_then(|model| model.normalize_reasoning_effort_selection(Some(effort))) - .unwrap_or_else(|| { - devo_protocol::normalize_reasoning_effort_literal(effort) - }), + .clone() + .unwrap_or_else(|| effort.clone()) + } else { + match self.deps.context_for_workspace(&session_cwd).await { + Ok(runtime_context) => runtime_context + .resolve_turn_config(Some(&requested_model), Some(effort.clone())) + .reasoning_effort_selection + .unwrap_or_else(|| { + devo_protocol::normalize_reasoning_effort_literal(effort) + }), + Err(_) => self + .deps + .model_catalog + .get(&requested_model) + .and_then(|model| { + model.normalize_reasoning_effort_selection(Some(effort)) + }) + .unwrap_or_else(|| { + devo_protocol::normalize_reasoning_effort_literal(effort) + }), + } }; if current_effort.as_ref() != Some(&normalized) { if rollout_path.is_some() { @@ -1017,7 +1082,7 @@ impl ServerRuntime { // Mirror session model / effort into user defaults (config.toml). // Provider is not a separate preference — it is the `provider/` prefix // of the model slug. - if overlay_model.is_some() || overlay_effort.is_some() { + if !is_child_session && (overlay_model.is_some() || overlay_effort.is_some()) { { let mut store = self .deps @@ -1094,7 +1159,8 @@ impl ServerRuntime { || overlay_effort.is_some() || overlay_model.is_some() || overlay_compact_limit.is_some() - || overlay_python_cell_first_wait_ms.is_some()) + || overlay_python_cell_first_wait_ms.is_some() + || validated_child_config.is_some()) && let Some(stream) = self.active_stream_state(legacy_session_id).await { let mut stream = stream.lock().await; @@ -1126,12 +1192,15 @@ impl ServerRuntime { if overlay_effort.is_some() || overlay_model.is_some() || overlay_compact_limit.is_some() + || validated_child_config.is_some() { let mut live = inline .live_turn_settings .lock() .expect("live settings mutex poisoned"); - if let Some(model_slug) = &overlay_model { + if let Some(config) = &validated_child_config { + live.turn_config = Some(config.clone()); + } else if let Some(model_slug) = &overlay_model { let effort = overlay_effort.clone().or_else(|| { live.turn_config .as_ref() diff --git a/crates/server/src/runtime/handlers/turn.rs b/crates/server/src/runtime/handlers/turn.rs index b9ea6a1c..2d5ba453 100644 --- a/crates/server/src/runtime/handlers/turn.rs +++ b/crates/server/src/runtime/handlers/turn.rs @@ -549,8 +549,20 @@ impl ServerRuntime { let requested_reasoning_effort_selection = params .reasoning_effort_selection .or_else(|| reservation.summary.settings.reasoning_effort.clone()); - let turn_config = runtime_context - .resolve_turn_config(requested_model, requested_reasoning_effort_selection); + let turn_config = match runtime_context.resolve_session_turn_config( + &reservation.summary, + requested_model, + requested_reasoning_effort_selection, + ) { + Ok(config) => config, + Err(error) => { + return self.error_response( + request_id, + ProtocolErrorCode::InvalidParams, + error.to_string(), + ); + } + }; let model_binding_id = turn_config .model_binding_id .clone() @@ -575,7 +587,7 @@ impl ServerRuntime { model: devo_protocol::native::model::ModelBinding { provider: model_binding_id, model: request_model, - variant: None, + variant: turn_config.variant.clone(), reasoning_effort: resolved_request.effective_reasoning_effort, }, collaboration_mode: Some(params.collaboration_mode), diff --git a/crates/server/src/runtime/kernel_host.rs b/crates/server/src/runtime/kernel_host.rs index 7785aae2..593727af 100644 --- a/crates/server/src/runtime/kernel_host.rs +++ b/crates/server/src/runtime/kernel_host.rs @@ -76,94 +76,8 @@ pub async fn dispatch_host_request_with_bridge(bridge: &HostBridge, data: Value) a if a.starts_with("rlm_heartbeat") => { error_reply(format!("unknown rlm_heartbeat action: {a}")) } - "rlm.run" => { - // Wire action behind kernel `await rlm.spawn(...)`: spawn a real - // child session through the agent coordinator. Fresh namespace per - // the child doctrine (no parent history fork), caller-named. - let prompt = params - .get("prompt") - .and_then(|v| v.as_str()) - .map(str::trim) - .unwrap_or(""); - if prompt.is_empty() { - return error_reply("rlm.run requires prompt"); - } - let kwargs = params.get("kwargs").cloned().unwrap_or_else(|| json!({})); - let name = kwargs - .get("name") - .and_then(|v| v.as_str()) - .map(str::trim) - .filter(|name| !name.is_empty()) - .map(str::to_string); - if let Some(model) = kwargs - .get("model") - .and_then(|v| v.as_str()) - .map(str::trim) - .filter(|model| !model.is_empty()) - { - return error_reply(format!( - "model override '{model}' is not supported by this host; children inherit \ - the parent model (omit 'model')" - )); - } - if let Some(thinking) = kwargs - .get("thinking") - .and_then(|v| v.as_str()) - .map(str::trim) - .filter(|thinking| !thinking.is_empty()) - { - return error_reply(format!( - "thinking override '{thinking}' is not supported by this host; children \ - inherit the parent level (omit 'thinking')" - )); - } - let Some(runtime) = bridge.runtime() else { - return error_reply("runtime not available for rlm.run"); - }; - let Some(sid) = parse_session_id(session_id) else { - return error_reply("invalid session id"); - }; - use devo_core::tools::AgentToolCoordinator; - use devo_protocol::AgentToolPolicy; - use devo_protocol::SpawnAgentParams; - match runtime - .spawn_agent(SpawnAgentParams { - session_id: sid, - message: prompt.to_string(), - // RLM children start a fresh namespace ("[task from parent]" - // labeling); forking parent history would leak the parent - // transcript into the child instead. - fork_turns: Some("none".to_string()), - max_turns: None, - tool_policy: AgentToolPolicy::Inherit, - ephemeral: false, - nickname: name, - }) - .await - { - Ok(result) => { - let session_dir = result.session_dir.clone().unwrap_or_default(); - if session_dir.is_empty() { - return error_reply( - "spawned child has no persisted session dir (ephemeral?)", - ); - } - ok_result(json!({ - "rlm_child_id": result.child_session_id.to_string(), - "name": result.agent_nickname, - "session_dir": session_dir, - "model": result.model.unwrap_or_default(), - })) - } - Err(error) => error_reply(error.to_string()), - } - } - "rlm.find_models" => { - // Valid payload shape for the kernel parser (`models` list). This - // host runs children on the inherited parent model, so there is no - // alternate-model catalog to offer here. - ok_result(json!({ "models": [] })) - } + "rlm.run" => super::kernel_host_models::spawn(bridge, ¶ms).await, + "rlm.find_models" => super::kernel_host_models::find_models(bridge, ¶ms).await, "rlm.list_subagents" => { let Some(runtime) = bridge.runtime() else { return error_reply("runtime not available for rlm.list_subagents"); @@ -705,16 +619,15 @@ mod tests { assert!( model_override["error"] .as_str() - .is_some_and(|error| error.contains("not supported")) + .is_some_and(|error| error.contains("runtime not available")) ); let listing = dispatch_host_request(&sid, json!({ "type": "rlm.list_subagents" })).await; assert_eq!(listing["status"], "error"); - // find_models must return the `models` list shape the kernel parses. + // Discovery cannot claim an empty catalog without a live session. let models = dispatch_host_request(&sid, json!({ "type": "rlm.find_models" })).await; - assert_eq!(models["status"], "ok"); - assert!(models["result"]["models"].as_array().is_some()); + assert_eq!(models["status"], "error"); } /// Verifies: bash.completed enqueues an in-memory notice. diff --git a/crates/server/src/runtime/kernel_host_models.rs b/crates/server/src/runtime/kernel_host_models.rs new file mode 100644 index 00000000..4e1a162b --- /dev/null +++ b/crates/server/src/runtime/kernel_host_models.rs @@ -0,0 +1,147 @@ +//! Python child model operations share the session's configured model resolver. + +use devo_core::tools::AgentToolCoordinator; +use devo_protocol::{AgentToolPolicy, SpawnAgentParams}; +use serde_json::{Value, json}; + +use super::kernel_host_bridge::HostBridge; + +fn optional_string<'a>(params: &'a Value, key: &str) -> Result, String> { + match params.get(key) { + None => Ok(None), + Some(Value::String(value)) if !value.trim().is_empty() && value.trim() == value => { + Ok(Some(value)) + } + Some(_) => Err(format!( + "{key} must be a nonempty string without surrounding whitespace" + )), + } +} + +pub(super) async fn spawn(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_else(|| "rlm.run requires prompt".to_string())?; + let kwargs = params + .get("kwargs") + .filter(|value| value.is_object()) + .ok_or_else(|| "rlm.run requires a kwargs object".to_string())?; + let name = + optional_string(kwargs, "name")?.ok_or_else(|| "rlm.run requires name".to_string())?; + let model = optional_string(kwargs, "model")?.map(str::to_string); + let thinking = optional_string(kwargs, "thinking")?.map(str::to_string); + let runtime = bridge + .runtime() + .ok_or_else(|| "runtime not available for rlm.run".to_string())?; + let spawned = runtime + .spawn_agent(SpawnAgentParams { + session_id: bridge.session_id, + message: prompt.to_string(), + model, + thinking, + fork_turns: Some("none".to_string()), + max_turns: None, + tool_policy: AgentToolPolicy::Inherit, + ephemeral: false, + nickname: Some(name.to_string()), + }) + .await + .map_err(|error| error.to_string())?; + let session_dir = spawned + .session_dir + .filter(|dir| !dir.is_empty()) + .ok_or_else(|| "spawned child has no persisted session dir (ephemeral?)".to_string())?; + Ok::<_, String>(json!({ + "rlm_child_id": spawned.child_session_id.to_string(), + "name": spawned.agent_nickname, + "session_dir": session_dir, + "model": spawned.model.unwrap_or_default(), + })) + } + .await; + match result { + Ok(result) => json!({"status": "ok", "result": result}), + Err(error) => json!({"status": "error", "error": error}), + } +} + +pub(super) async fn find_models(bridge: &HostBridge, params: &Value) -> Value { + let result = async { + let query = match params.get("query") { + None => "", + Some(Value::String(query)) => query, + Some(_) => return Err("query must be a string".to_string()), + }; + let limit = match params.get("limit") { + None => 8, + Some(value) => value + .as_u64() + .filter(|limit| (1..=100).contains(limit)) + .ok_or_else(|| "limit must be an integer between 1 and 100".to_string())?, + }; + let runtime = bridge + .runtime() + .ok_or_else(|| "runtime not available for rlm.find_models".to_string())?; + let session = runtime + .session(bridge.session_id) + .await + .ok_or_else(|| "session not available for rlm.find_models".to_string())?; + let snapshot = session + .turn_reservation_snapshot() + .await + .ok_or_else(|| "session not available for rlm.find_models".to_string())?; + let models = snapshot + .runtime_context + .find_child_models(Some(query), limit as usize) + .map_err(|error| error.to_string())?; + Ok::<_, String>(json!({"models": models})) + } + .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; + + #[tokio::test] + async fn rejects_malformed_spawn_arguments_without_runtime() { + let bridge = HostBridge::for_tests("ses_00000000-0000-0000-0000-000000000001"); + for value in [ + json!(null), + json!(true), + json!(8), + json!(""), + json!(" x/y "), + ] { + let reply = spawn( + &bridge, + &json!({"prompt":"task", "kwargs":{"name":"child", "model":value}}), + ) + .await; + assert_eq!(reply["status"], "error"); + assert!(reply["error"].as_str().unwrap().starts_with("model must")); + } + for value in [ + json!(null), + json!(true), + json!(0), + json!(-1), + json!(101), + json!(1.5), + ] { + let reply = find_models(&bridge, &json!({"limit":value})).await; + assert_eq!( + reply, + json!({"status":"error", "error":"limit must be an integer between 1 and 100"}) + ); + } + } +} diff --git a/crates/server/src/runtime/refine_planner.rs b/crates/server/src/runtime/refine_planner.rs index ce0e39c8..9bac06a0 100644 --- a/crates/server/src/runtime/refine_planner.rs +++ b/crates/server/src/runtime/refine_planner.rs @@ -287,10 +287,18 @@ impl ServerRuntime { .model_selection .unwrap_or_else(|| title_context.runtime_context.default_model.clone()); let runtime_context = title_context.runtime_context; - let turn_config = runtime_context.resolve_turn_config( + let snapshot = session_handle.turn_reservation_snapshot().await?; + let turn_config = match runtime_context.resolve_session_turn_config( + &snapshot.summary, Some(primary_selection.as_str()), title_context.reasoning_effort_selection.clone(), - ); + ) { + Ok(config) => config, + Err(error) => { + tracing::warn!(%session_id, %error, "refine child model is unavailable"); + return None; + } + }; let resolved_request = turn_config .model .resolve_reasoning_effort_selection(turn_config.reasoning_effort_selection.as_deref()); diff --git a/crates/server/src/runtime/session_actor/actor_loop.rs b/crates/server/src/runtime/session_actor/actor_loop.rs index 1879d382..fad5ea11 100644 --- a/crates/server/src/runtime/session_actor/actor_loop.rs +++ b/crates/server/src/runtime/session_actor/actor_loop.rs @@ -187,15 +187,14 @@ pub(super) async fn run_session_actor( .and_then(|live| live.turn_config.clone()) }) .or_else(|| { - Some( - state.runtime_context.resolve_turn_config( - state - .summary - .model_binding_id() - .or(state.summary.model_name()), + state + .runtime_context + .resolve_session_turn_config( + &state.summary, + state.summary.model_selection(), state.summary.settings.reasoning_effort.clone(), - ), - ) + ) + .ok() }); turn_config.map(|turn_config| ApprovalCheckpointSnapshot { messages: state.core.messages.clone(), diff --git a/crates/server/src/runtime/session_model_config.rs b/crates/server/src/runtime/session_model_config.rs new file mode 100644 index 00000000..15b3e78f --- /dev/null +++ b/crates/server/src/runtime/session_model_config.rs @@ -0,0 +1,32 @@ +//! Child turns revalidate their persisted binding instead of falling back to +//! process defaults after a configuration change or session restore. + +use crate::runtime_session_summary::RuntimeSessionSummary; +use crate::session_context::SessionRuntimeContext; + +impl SessionRuntimeContext { + pub(crate) fn resolve_session_turn_config( + &self, + summary: &RuntimeSessionSummary, + requested_model: Option<&str>, + thinking: Option, + ) -> anyhow::Result { + if summary.parent_session_id().is_none() { + return Ok(self.resolve_turn_config(requested_model, thinking)); + } + let bound = summary + .model_selection() + .ok_or_else(|| anyhow::anyhow!("child session has no persisted model binding"))?; + let selected = requested_model.unwrap_or(bound); + let changed_thinking = thinking + .as_deref() + .filter(|thinking| Some(*thinking) != summary.settings.reasoning_effort.as_deref()); + self.resolve_child_model( + None, + selected, + changed_thinking, + summary.settings.reasoning_effort.as_deref(), + ) + .map(|(_, config)| config) + } +} diff --git a/crates/server/src/runtime/turn_exec/approval_resume.rs b/crates/server/src/runtime/turn_exec/approval_resume.rs index 250d6f6f..228b69f7 100644 --- a/crates/server/src/runtime/turn_exec/approval_resume.rs +++ b/crates/server/src/runtime/turn_exec/approval_resume.rs @@ -602,15 +602,25 @@ impl ServerRuntime { }) .or(snapshot.summary.model_binding_id()) .or(snapshot.summary.model_name()); + let selected_with_variant = requested_model + .zip( + checkpoint + .turn_config + .get("variant") + .and_then(|value| value.as_str()), + ) + .map(|(model, variant)| format!("{model}/{variant}")); + let requested_model = selected_with_variant.as_deref().or(requested_model); let reasoning_effort = checkpoint .turn_config .get("reasoningEffortSelection") .and_then(|value| value.as_str()) .map(str::to_string) .or(snapshot.summary.settings.reasoning_effort.clone()); - Ok(snapshot + snapshot .runtime_context - .resolve_turn_config(requested_model, reasoning_effort)) + .resolve_session_turn_config(&snapshot.summary, requested_model, reasoning_effort) + .map_err(|error| error.to_string()) } async fn turn_has_tool_result( diff --git a/crates/server/src/runtime/turn_exec/followup.rs b/crates/server/src/runtime/turn_exec/followup.rs index 0d64bb8f..9d2d7a6f 100644 --- a/crates/server/src/runtime/turn_exec/followup.rs +++ b/crates/server/src/runtime/turn_exec/followup.rs @@ -68,18 +68,22 @@ impl ServerRuntime { return false; }; let state_change_guard = session_handle.lock_state_change().await; - let Some(queued) = self - .pop_next_queued_turn_input( - session_id, - /*require_idle_session*/ matches!(policy, QueueDrainPolicy::OnlyWhenIdle), - ) - .await - else { - return false; - }; - let Some((turn, turn_config)) = self.prepare_queued_turn_start(session_id, &queued).await - else { - return false; + let (queued, turn, turn_config) = loop { + let Some(queued) = self + .pop_next_queued_turn_input( + session_id, + /*require_idle_session*/ + matches!(policy, QueueDrainPolicy::OnlyWhenIdle), + ) + .await + else { + return false; + }; + if let Some((turn, config)) = self.prepare_queued_turn_start(session_id, &queued).await + { + break (queued, turn, config); + } + // A rejected binding settles this input, not the remaining queue. }; if let Some((parent_session_id, parent_turn_id)) = queued.subagent_usage_owner { self.register_subagent_usage_owner(parent_session_id, session_id, parent_turn_id) @@ -150,28 +154,22 @@ impl ServerRuntime { .await .map(|snapshot| snapshot.pending_count) .unwrap_or_default(); - let Some(queued) = self - .pop_next_queued_turn_input(session_id, /*require_idle_session*/ false) - .await - else { - if queue_len_before > 0 { - tracing::warn!( - session_id = %session_id, - queue_len_before, - "post-turn chain found a non-empty queue but popped nothing (entries un-drainable?)" - ); + let (queued, turn, turn_config) = loop { + let Some(queued) = self + .pop_next_queued_turn_input(session_id, /*require_idle_session*/ false) + .await + else { + if queue_len_before > 0 { + tracing::debug!(%session_id, queue_len_before, "post-turn queue drain finished without an admissible input"); + } + return false; + }; + tracing::info!(%session_id, queued_input_id = %queued.queued_item_id, queue_len_before, + "post-turn chain draining queued follow-up"); + if let Some((turn, config)) = self.prepare_queued_turn_start(session_id, &queued).await + { + break (queued, turn, config); } - return false; - }; - tracing::info!( - session_id = %session_id, - queued_input_id = %queued.queued_item_id, - queue_len_before, - "post-turn chain draining queued follow-up" - ); - let Some((turn, turn_config)) = self.prepare_queued_turn_start(session_id, &queued).await - else { - return false; }; if let Some((parent_session_id, parent_turn_id)) = queued.subagent_usage_owner { self.register_subagent_usage_owner(parent_session_id, session_id, parent_turn_id) @@ -272,10 +270,41 @@ impl ServerRuntime { .model_selection .as_deref() .or_else(|| session_model_selection(&reservation.summary)); - let turn_config = reservation.runtime_context.resolve_turn_config( + let turn_config = match reservation.runtime_context.resolve_session_turn_config( + &reservation.summary, model_override, reservation.summary.settings.reasoning_effort.clone(), - ); + ) { + Ok(config) => config, + Err(error) => { + let message = format!("Queued child turn could not start: {error}"); + if let Some(parent) = reservation.summary.parent_session_id() { + self.complete_background_work( + parent, + format!("agent:{session_id}:queued:{}", queued.queued_item_id), + message.clone(), + ); + self.set_agent_status(parent, session_id, SubagentStatus::Failed) + .await; + self.record_subagent_status_event_with_text( + parent, + session_id, + SubagentStatus::Failed, + TurnId::new(), + Some(message), + ) + .await; + } + self.broadcast_queue_updated( + session_id, + devo_protocol::native::queue::QueueChange::Removed, + queued.queued_item_id, + None, + ) + .await; + return None; + } + }; let resolved_request = turn_config .model .resolve_reasoning_effort_selection(turn_config.reasoning_effort_selection.as_deref()); @@ -299,7 +328,7 @@ impl ServerRuntime { .clone() .unwrap_or_else(|| "unknown".to_string()), model: request_model, - variant: None, + variant: turn_config.variant.clone(), reasoning_effort: turn_config .reasoning_effort_selection .as_deref() @@ -346,6 +375,13 @@ impl ServerRuntime { .await; self.add_active_turn_user_text(session_id, turn.turn_id(), display_input.to_string()) .await; + if let Some((parent, _)) = self.child_parent_and_path(session_id).await { + self.set_agent_status(parent, session_id, SubagentStatus::Running) + .await; + } } } } + +#[cfg(test)] +mod tests; diff --git a/crates/server/src/runtime/turn_exec/followup/tests.rs b/crates/server/src/runtime/turn_exec/followup/tests.rs new file mode 100644 index 00000000..5b76011f --- /dev/null +++ b/crates/server/src/runtime/turn_exec/followup/tests.rs @@ -0,0 +1,196 @@ +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use async_trait::async_trait; +use chrono::Utc; +use devo_core::tools::AgentToolCoordinator; +use devo_protocol::{ + ModelRequest, ModelResponse, PendingInputItem, PendingInputKind, ResponseContent, + ResponseMetadata, SpawnAgentParams, StopReason, StreamEvent, Usage, +}; +use devo_provider::ModelProviderSDK; +use pretty_assertions::assert_eq; +use serde_json::json; +use tokio::sync::Semaphore; + +use crate::{ClientTransportKind, test_support::TestRuntime}; + +struct GatedChildProvider { + requests: Mutex>, + started: Semaphore, + release: Semaphore, +} + +#[async_trait] +impl ModelProviderSDK for GatedChildProvider { + async fn completion(&self, _request: ModelRequest) -> Result { + Ok(ModelResponse { + id: "title".into(), + content: vec![ResponseContent::Text("Child task".into())], + stop_reason: Some(StopReason::EndTurn), + usage: Usage::default(), + metadata: ResponseMetadata::default(), + }) + } + + async fn completion_stream( + &self, + request: ModelRequest, + ) -> Result> + Send>>> { + let first = { + let mut requests = self.requests.lock().expect("requests"); + requests.push(request); + requests.len() == 1 + }; + if first { + self.started.add_permits(1); + self.release.acquire().await?.forget(); + } + Ok(Box::pin(futures::stream::iter([Ok( + StreamEvent::MessageDone { + response: ModelResponse { + id: "child-reply".into(), + content: vec![ResponseContent::Text("Child complete".into())], + stop_reason: Some(StopReason::EndTurn), + usage: Usage::default(), + metadata: ResponseMetadata::default(), + }, + }, + )]))) + } + + fn name(&self) -> &str { + "gated-child" + } +} + +#[tokio::test] +async fn invalid_queued_child_binding_does_not_strand_next_valid_input() -> Result<()> { + let home = tempfile::TempDir::new()?; + let provider = Arc::new(GatedChildProvider { + requests: Mutex::new(Vec::new()), + started: Semaphore::new(/*permits*/ 0), + release: Semaphore::new(/*permits*/ 0), + }); + let runtime = TestRuntime::new(provider.clone()) + .with_test_model() + .runtime(home.path()); + let (tx, mut notifications) = crate::test_outbound_channel(128); + let connection = runtime + .register_connection(ClientTransportKind::Stdio, tx) + .await; + let initialized = runtime.handle_incoming(connection, json!({ + "id": 1, "method": "initialize", "params": { + "protocolVersion": 1, "clientCapabilities": {}, "_meta": {"devo": {"protocol": "native"}}, + "clientInfo": {"name": "queue-child-test", "title": "test", "version": "1"} + } + })).await.context("initialize")?; + assert!(initialized.get("error").is_none(), "{initialized}"); + let created = runtime.handle_incoming(connection, json!({ + "id": 2, "method": "session/new", "params": {"cwd": home.path(), "idempotencyKey": "queue-parent"} + })).await.context("create parent")?; + let root = devo_protocol::SessionId::from( + created["result"]["session"]["id"] + .as_str() + .context("parent id")?, + ); + let child = runtime + .clone() + .spawn_agent(serde_json::from_value::(json!({ + "session_id": root, "message": "Keep initial child turn active", "fork_turns": "none" + }))?) + .await?; + tokio::time::timeout(Duration::from_secs(/*secs*/ 15), provider.started.acquire()) + .await?? + .forget(); + + let subscribed = runtime + .handle_incoming( + connection, + json!({ + "id": 3, "method": "subscription/create", "params": { + "selectors": [{"kind": "session", "sessionId": child.child_session_id}], + "includeSnapshot": false + } + }), + ) + .await + .context("subscribe to child queue changes")?; + assert!(subscribed.get("error").is_none(), "{subscribed}"); + + // Native queue payloads no longer expose model overrides. Seed the same + // durable actor queue metadata used by restored/legacy queued inputs. + let handle = runtime + .session(child.child_session_id) + .await + .context("child actor")?; + let invalid = PendingInputItem::new( + PendingInputKind::UserText { + text: "invalid queued input".into(), + }, + Some(json!({"model": "missing/model"})), + Utc::now(), + ); + let invalid_id = invalid.id; + let valid = PendingInputItem::new( + PendingInputKind::UserText { + text: "valid queued input".into(), + }, + /*metadata*/ None, + Utc::now(), + ); + let valid_id = valid.id; + handle.enqueue_pending_turn_input(invalid).await; + handle.enqueue_pending_turn_input(valid).await; + // Snapshot is a mailbox barrier ensuring both enqueues precede completion. + let snapshot = handle + .turn_reservation_snapshot() + .await + .context("queued snapshot")?; + assert_eq!(snapshot.pending_turn_queue.lock().expect("queue").len(), 2); + provider.release.add_permits(1); + + let (removed, drained) = tokio::time::timeout(Duration::from_secs(/*secs*/ 15), async { + let mut completed = 0; + let mut removed = false; + let mut drained = false; + while let Some(event) = notifications.recv().await { + if event["method"] == "queue/updated" { + removed |= event["params"]["change"] == "removed" + && event["params"]["queueItemId"] == json!(invalid_id); + drained |= event["params"]["change"] == "drained" + && event["params"]["queueItemId"] == json!(valid_id); + } + if event["method"] == "turn/completed" + && event["params"]["turn"]["sessionId"] == json!(child.child_session_id) + { + completed += 1; + if completed == 2 { + return Ok::<_, anyhow::Error>((removed, drained)); + } + } + } + anyhow::bail!("notification stream closed before valid queued turn completed") + }) + .await + .context("queue did not advance past rejected binding")??; + assert_eq!((removed, drained), (true, true)); + let snapshot = handle + .turn_reservation_snapshot() + .await + .context("drained snapshot")?; + assert!( + snapshot + .pending_turn_queue + .lock() + .expect("queue") + .is_empty() + ); + let requests = provider.requests.lock().expect("requests"); + assert_eq!(requests.len(), 2); + assert!(serde_json::to_string(&requests[1])?.contains("valid queued input")); + assert!(!serde_json::to_string(&requests[1])?.contains("invalid queued input")); + Ok(()) +} diff --git a/crates/server/src/runtime/turn_exec/recovery.rs b/crates/server/src/runtime/turn_exec/recovery.rs index f533aee7..e3b3708d 100644 --- a/crates/server/src/runtime/turn_exec/recovery.rs +++ b/crates/server/src/runtime/turn_exec/recovery.rs @@ -299,13 +299,23 @@ impl ServerRuntime { .turn_reservation_snapshot() .await .context("session unavailable")?; - let turn_config = snapshot.runtime_context.resolve_turn_config( - saved - .turn - .model_binding_id() - .or(Some(saved.turn.logical_model())), + let saved_model = saved + .turn + .model_binding_id() + .unwrap_or(saved.turn.logical_model()); + let saved_selection = saved + .turn + .native + .model + .variant + .as_ref() + .map(|variant| format!("{saved_model}/{variant}")) + .unwrap_or_else(|| saved_model.to_string()); + let turn_config = snapshot.runtime_context.resolve_session_turn_config( + &snapshot.summary, + Some(&saved_selection), saved.turn.reasoning_effort_selection(), - ); + )?; let mut turn = saved.turn; turn.native.status = devo_protocol::native::turn::TurnStatus::InProgress; turn.native.completed_at = None; diff --git a/crates/server/src/session_context.rs b/crates/server/src/session_context.rs index d1272dd3..e98a26a3 100644 --- a/crates/server/src/session_context.rs +++ b/crates/server/src/session_context.rs @@ -1,3 +1,5 @@ +mod child_models; + use std::collections::HashSet; use std::path::Path; use std::path::PathBuf; @@ -66,6 +68,7 @@ pub(crate) struct SessionRuntimeContext { /// creation, and without the slot a connection added after session start /// would route through the boot-time table forever. pub(crate) provider_router: Arc>>, + child_route_inputs: StdMutex, /// Live tool registry. Wrapped so MCP enable/disable can swap the Arc for /// the next turn without rebuilding the whole process context graph. pub(crate) registry: Arc>>, @@ -137,7 +140,7 @@ impl SessionRuntimeContext { agents_md: AgentsMdConfig, config_store: Arc>, ) -> Self { - let (provider_catalog_snapshot, provider_http_snapshot) = { + let (provider_catalog_snapshot, provider_http_snapshot, child_route_inputs) = { let config_store = config_store .lock() .expect("app config store mutex should not be poisoned"); @@ -145,11 +148,13 @@ impl SessionRuntimeContext { ( config.provider_catalog_config(), config.provider_http.clone(), + child_models::ChildRouteInputs::capture(config, config_store.user_config_dir()), ) }; Self { provider, provider_router: Arc::new(StdMutex::new(provider_router)), + child_route_inputs: StdMutex::new(child_route_inputs), registry: Arc::new(StdMutex::new(registry)), mcp_manager, default_model, @@ -187,13 +192,6 @@ impl SessionRuntimeContext { ) } - pub(crate) fn replace_provider_router(&self, router: Arc) { - *self - .provider_router - .lock() - .expect("provider router mutex should not be poisoned") = router; - } - /// Rebuilds this context's Connection routes from its current on-disk /// provider config and installs them in place. /// @@ -204,7 +202,7 @@ impl SessionRuntimeContext { /// default provider is kept so sessions on the Default route keep their /// model selection. pub(crate) async fn reload_provider_router(&self) -> anyhow::Result> { - let (provider_config, provider_http, user_config_dir) = { + let (provider_config, provider_http, user_config_dir, route_config) = { let mut config_store = self .config_store .lock() @@ -221,6 +219,7 @@ impl SessionRuntimeContext { config.provider_catalog_config(), config.provider_http.clone(), config_store.user_config_dir().to_path_buf(), + config, ) }; let effective = devo_core::effective_provider_catalog_with_home( @@ -228,21 +227,42 @@ impl SessionRuntimeContext { Some(user_config_dir.as_path()), ) .context("failed to merge live provider catalog with user configuration")?; - let auth = read_user_auth_config(&user_config_dir.join(AUTH_CONFIG_FILE_NAME))?; - let routes = crate::provider_config::build_connection_route_table( - &provider_config, - &effective, - &provider_http, - &auth, - &user_config_dir, - ) - .await?; + let mut auth = read_user_auth_config(&user_config_dir.join(AUTH_CONFIG_FILE_NAME))?; + let mut stable_routes = None; + // OAuth refresh may update auth during adapter creation. Rebuild from + // that result, rather than claiming an adapter used a later credential. + // Continual external edits fail closed instead of publishing stale routes. + for _ in 0..3 { + let routes = crate::provider_config::build_connection_route_table( + &provider_config, + &effective, + &provider_http, + &auth, + &user_config_dir, + ) + .await?; + let latest_auth = read_user_auth_config(&user_config_dir.join(AUTH_CONFIG_FILE_NAME))?; + if latest_auth == auth { + stable_routes = Some(routes); + break; + } + auth = latest_auth; + } + let routes = stable_routes + .context("provider authentication changed during reload; retry reconnecting")?; let mut router = MultiProviderRouter::new(Arc::clone(&self.provider)); for (route, provider) in routes { router.insert_route(route, provider); } let router = Arc::new(router); - self.replace_provider_router(Arc::clone(&router) as Arc); + let mut inputs = child_models::ChildRouteInputs::capture(&route_config, &user_config_dir); + inputs.auth = Some(auth); + let mut installed = self.provider_router.lock().expect("provider router mutex"); + *self + .child_route_inputs + .lock() + .expect("child route inputs mutex") = inputs; + *installed = Arc::clone(&router) as Arc; Ok(router) } @@ -321,26 +341,37 @@ impl SessionRuntimeContext { .resolve_for_turn(None) .ok() .map(|model| model.slug.clone()); - let (provider, provider_router, provider_default_model) = if provider_runtime_config_changed - { - let provider = - load_server_provider(&config, default_model.as_deref(), &user_config_dir) - .await - .context("load server provider for session workspace")?; - ( - provider.provider, - provider.provider_router, - provider.default_model, - ) - } else { - ( - Arc::clone(&inherited_context.provider), - inherited_context.provider_router(), - // An empty discovery catalog must not erase the model of the - // provider we are retaining (including injected providers). - default_model.unwrap_or_else(|| inherited_context.default_model.clone()), - ) - }; + let (provider, provider_router, provider_default_model, child_route_inputs) = + if provider_runtime_config_changed { + let provider = + load_server_provider(&config, default_model.as_deref(), &user_config_dir) + .await + .context("load server provider for session workspace")?; + ( + provider.provider, + provider.provider_router, + provider.default_model, + child_models::ChildRouteInputs::capture(&config, &user_config_dir), + ) + } else { + let router = inherited_context + .provider_router + .lock() + .expect("provider router mutex"); + let inputs = inherited_context + .child_route_inputs + .lock() + .expect("child route inputs mutex") + .clone(); + ( + Arc::clone(&inherited_context.provider), + Arc::clone(&router), + // An empty discovery catalog must not erase the model of the + // provider we are retaining (including injected providers). + default_model.unwrap_or_else(|| inherited_context.default_model.clone()), + inputs, + ) + }; let skill_workspace_root = workspace_root .map(Path::to_path_buf) .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))); @@ -356,6 +387,7 @@ impl SessionRuntimeContext { Ok(Arc::new(Self { provider, provider_router: Arc::new(StdMutex::new(provider_router)), + child_route_inputs: StdMutex::new(child_route_inputs), registry, mcp_manager, default_model: provider_default_model, diff --git a/crates/server/src/session_context/child_models.rs b/crates/server/src/session_context/child_models.rs new file mode 100644 index 00000000..4b2e2d29 --- /dev/null +++ b/crates/server/src/session_context/child_models.rs @@ -0,0 +1,439 @@ +//! Strict, local-only model selection for kernel children. No completion probes. +use super::*; +use devo_core::{AuthCredentialKind, ProviderConfigEntry, ProviderModelConfig, UserAuthConfigFile}; +use devo_protocol::ProviderWireApi; +use serde::Serialize; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ChildModelEntry { + pub(crate) provider: String, + pub(crate) id: String, + pub(crate) name: String, + pub(crate) selector: String, + pub(crate) thinking_levels: Vec, +} + +/// Inputs associated with the installed adapters, never serialized or logged. +#[derive(Clone)] +pub(super) struct ChildRouteInputs { + providers: ProviderConfigFile, + http: ProviderHttpConfig, + pub(super) auth: Option, +} + +impl ChildRouteInputs { + pub(super) fn capture(config: &AppConfig, home: &Path) -> Self { + Self { + providers: config.provider_catalog_config(), + http: config.provider_http.clone(), + auth: read_user_auth_config(&home.join(AUTH_CONFIG_FILE_NAME)).ok(), + } + } + + fn matches(&self, current: &ChildModelCatalog, provider_id: &str, model_id: &str) -> bool { + let Some(previous) = self.providers.providers.get(provider_id) else { + return false; + }; + let Some(provider) = current.configured.providers.get(provider_id) else { + return false; + }; + let Some(auth) = self.auth.as_ref() else { + return false; + }; + let previous_model = previous.models.get(model_id); + let model = provider.models.get(model_id); + self.http == current.config.provider_http + && previous.base_url == provider.base_url + && previous.wire_api == provider.wire_api + && previous.credential == provider.credential + && previous.headers == provider.headers + && previous.model_overrides == provider.model_overrides + && previous_model.and_then(|model| model.base_url.as_ref()) + == model.and_then(|model| model.base_url.as_ref()) + && previous_model.and_then(|model| model.wire_api) + == model.and_then(|model| model.wire_api) + && bound_credential(provider_id, previous, auth) + == bound_credential(provider_id, provider, ¤t.auth) + } +} + +struct ChildModelCatalog { + config: AppConfig, + home: PathBuf, + configured: ProviderConfigFile, + effective: ProviderConfigFile, + catalog: PresetModelCatalog, + auth: UserAuthConfigFile, + router: Arc, + route_inputs: ChildRouteInputs, +} + +impl SessionRuntimeContext { + fn child_model_catalog(&self) -> anyhow::Result { + let (config, home) = { + let mut store = self.config_store.lock().expect("config store mutex"); + store + .reload() + .map_err(|_| anyhow::anyhow!("cannot reload child model configuration"))?; + ( + store.effective_config().clone(), + store.user_config_dir().to_path_buf(), + ) + }; + let configured = config.provider_catalog_config(); + let mut effective = + devo_core::effective_provider_catalog_with_home(&configured, Some(&home)) + .map_err(|_| anyhow::anyhow!("cannot load configured child models"))?; + // A user-pinned primary/small model is itself an explicit declaration, + // even when the directory has no cached metadata for that model. Match + // PresetModelCatalog's exact-model-before-variant interpretation. + for selector in [ + configured.model.as_deref(), + configured.small_model.as_deref(), + ] + .into_iter() + .flatten() + { + let Ok((provider_id, requested_id)) = parse_selector(selector) else { + continue; + }; + if !configured.providers.contains_key(provider_id) { + continue; + } + let Some(provider) = effective.providers.get_mut(provider_id) else { + continue; + }; + if provider.models.contains_key(requested_id) + || requested_id.rsplit_once('/').is_some_and(|(id, variant)| { + provider + .models + .get(id) + .is_some_and(|model| model.variants.contains_key(variant)) + }) + { + continue; + } + provider.models.entry(requested_id.to_string()).or_default(); + } + effective.apply_model_overrides(&config.provider.model_overrides); + let catalog = PresetModelCatalog::load_from_provider_config_with_overrides( + &effective, + &config.provider.model_overrides, + ) + .map_err(|_| anyhow::anyhow!("cannot load child model capabilities"))?; + let auth = read_user_auth_config(&home.join(AUTH_CONFIG_FILE_NAME)) + .map_err(|_| anyhow::anyhow!("cannot read provider authentication configuration"))?; + // Take both snapshots under the router lock, matching the reload writer. + let router = self.provider_router.lock().expect("provider router mutex"); + let route_inputs = self + .child_route_inputs + .lock() + .expect("child route inputs mutex") + .clone(); + Ok(ChildModelCatalog { + config, + home, + configured, + effective, + catalog, + auth, + router: Arc::clone(&router), + route_inputs, + }) + } + + /// Returns only configured, enabled, currently routable models, never secrets. + pub(crate) fn find_child_models( + &self, + query: Option<&str>, + limit: usize, + ) -> anyhow::Result> { + let snapshot = self.child_model_catalog()?; + let query = query.unwrap_or_default().trim().to_lowercase(); + let mut entries = Vec::new(); + for model in snapshot.catalog.list_visible() { + if snapshot.resolve(&model.slug).is_err() { + continue; + } + let Some((provider, id)) = model.slug.split_once('/') else { + continue; + }; + if !query.is_empty() + && !format!("{} {}", model.slug, model.display_name) + .to_lowercase() + .contains(&query) + { + continue; + } + entries.push(ChildModelEntry { + provider: provider.to_string(), + id: id.to_string(), + name: model.display_name.clone(), + selector: model.slug.clone(), + thinking_levels: thinking_levels(model), + }); + } + entries.sort_by(|a, b| a.selector.cmp(&b.selector)); + entries.truncate(limit.min(100)); + Ok(entries) + } + + /// Resolves exact selectors, preserving transport/request metadata. Omitted + /// values inherit the immediate parent, never the global session default. + pub(crate) fn resolve_child_model( + &self, + requested: Option<&str>, + parent_model: &str, + thinking: Option<&str>, + parent_thinking: Option<&str>, + ) -> anyhow::Result<(Model, TurnConfig)> { + let snapshot = self.child_model_catalog()?; + let selector = requested.unwrap_or(parent_model); + // Preserve injected single-provider runtimes only when inheriting their + // exact current parent model. Explicit model choices never use this path. + if requested.is_none() + && snapshot.configured.providers.is_empty() + && selector == self.default_model + && !selector.contains('/') + { + anyhow::ensure!( + self.provider.is_available(), + "parent provider is unavailable" + ); + let model = self.resolve_turn_model(Some(selector)); + anyhow::ensure!(model.slug == selector, "parent model is unavailable"); + let effort = strict_thinking(&model, thinking.or(parent_thinking))?; + return Ok((model.clone(), TurnConfig::new(model, effort))); + } + let (selector, inherited_variant) = + if requested.is_none() && snapshot.resolve(selector).is_err() { + if let Some((base, variant)) = selector.rsplit_once('/') + && let Ok((provider, id, _)) = snapshot.resolve(base) + && snapshot.effective.providers[provider].models[id] + .variants + .get(variant) + .is_some_and(|variant| !variant.disabled) + { + (base, Some(variant)) + } else { + (selector, None) + } + } else { + (selector, None) + }; + let (provider_id, model_id, route) = snapshot.resolve(selector)?; + let model = snapshot + .catalog + .get(selector) + .context("configured child model has no capability metadata")? + .clone(); + let effort = strict_thinking(&model, thinking.or(parent_thinking))?; + let provider = &snapshot.effective.providers[provider_id]; + let mut model_config = provider.models[model_id].clone(); + model_config.migrate_reasoning_implementation_into_variants(); + let variant = model_config.resolve_turn_variant_id(inherited_variant, effort.as_deref()); + let (defaults, headers) = provider_request_config( + &snapshot.effective, + provider_id, + model_id, + variant.as_deref(), + ); + let request_models = provider + .models + .iter() + .map(|(id, config)| { + ( + format!("{provider_id}/{id}"), + config.request_model.clone().unwrap_or_else(|| id.clone()), + ) + }) + .collect(); + let mut turn = TurnConfig::with_provider_route_and_web_tools( + model.clone(), + model_config + .request_model + .clone() + .unwrap_or_else(|| model_id.to_string()), + ProviderRequestModelMap::new(request_models).with_request_config(defaults, headers), + route, + self.resolve_turn_web_search( + &snapshot.config, + &snapshot.home, + provider.web_search.as_ref(), + model_config.web_search.as_ref(), + ), + self.resolve_turn_web_fetch( + &snapshot.config, + provider.web_fetch.as_ref(), + model_config.web_fetch.as_ref(), + ), + effort, + ); + turn.model_binding_id = Some(selector.to_string()); + turn.variant = variant; + Ok((model, turn)) + } +} + +impl ChildModelCatalog { + fn resolve<'a>(&self, selector: &'a str) -> anyhow::Result<(&'a str, &'a str, ProviderRoute)> { + let (provider_id, model_id) = parse_selector(selector)?; + let configured = self + .configured + .providers + .get(provider_id) + .context("child provider is not configured")?; + anyhow::ensure!( + configured.enabled != Some(false), + "child provider is disabled" + ); + let provider = self + .effective + .providers + .get(provider_id) + .context("child provider is unavailable")?; + anyhow::ensure!( + provider.enabled != Some(false), + "child provider is disabled" + ); + let model = provider + .models + .get(model_id) + .context("child model is not configured; use rlm.find_models()")?; + anyhow::ensure!(model.enabled != Some(false), "child model is disabled"); + let wire_api = model + .wire_api + .or(provider.wire_api) + .context("child provider has no configured protocol")?; + anyhow::ensure!( + model + .base_url + .as_ref() + .or(provider.base_url.as_ref()) + .is_some_and(|url| !url.trim().is_empty()), + "child provider has no configured endpoint" + ); + anyhow::ensure!( + credentials_available(provider_id, provider, wire_api, &self.auth), + "child provider credentials are missing, empty, or expired" + ); + anyhow::ensure!( + self.route_inputs.matches(self, provider_id, model_id), + "child provider configuration changed; reconnect the provider before selecting this model" + ); + let route = model_route(provider_id, model_id, model, wire_api); + anyhow::ensure!( + self.router.has_available_route(&route), + "child model route is unavailable; reconnect the provider" + ); + Ok((provider_id, model_id, route)) + } +} + +fn parse_selector(selector: &str) -> anyhow::Result<(&str, &str)> { + let (provider, id) = selector + .split_once('/') + .context("child model must use an exact provider/model selector")?; + anyhow::ensure!( + !provider.is_empty() + && !id.is_empty() + && !selector + .chars() + .any(|character| character.is_whitespace() || character.is_control()) + && !id.split('/').any(str::is_empty), + "invalid child model selector" + ); + Ok((provider, id)) +} + +fn model_route( + provider: &str, + id: &str, + model: &ProviderModelConfig, + wire_api: ProviderWireApi, +) -> ProviderRoute { + if model.base_url.is_some() { + ProviderRoute::connection_model(provider, wire_api, id) + } else { + ProviderRoute::connection(provider, wire_api) + } +} + +fn bound_credential<'a>( + provider_id: &str, + provider: &ProviderConfigEntry, + auth: &'a UserAuthConfigFile, +) -> Option<&'a devo_core::AuthCredentialConfig> { + if let Some(id) = provider + .credential + .as_deref() + .map(str::trim) + .filter(|id| !id.is_empty()) + { + return auth.credentials.get(id).or_else(|| { + // Only the documented legacy-id migration is allowed for explicit bindings. + auth.credentials + .get(&devo_core::provider_id_from_credential_id(id)) + }); + } + crate::provider_config::resolve_provider_credential_id(provider_id, provider, auth) + .and_then(|id| auth.credentials.get(&id)) +} + +fn credentials_available( + provider_id: &str, + provider: &ProviderConfigEntry, + wire_api: ProviderWireApi, + auth: &UserAuthConfigFile, +) -> bool { + if let Some(credential) = bound_credential(provider_id, provider, auth) { + return match credential.kind { + AuthCredentialKind::ApiKey => !credential.value.trim().is_empty(), + AuthCredentialKind::Oauth => { + credential + .access + .as_deref() + .is_some_and(|access| !access.trim().is_empty()) + && credential + .expires_at + .is_none_or(|expires| expires > chrono::Utc::now().timestamp()) + } + }; + } + // Match the adapters: OpenAI-compatible endpoints can be credentialless + // (e.g. local servers). An explicit binding must actually resolve. + provider + .credential + .as_deref() + .is_none_or(|id| id.trim().is_empty()) + && matches!( + wire_api, + ProviderWireApi::OpenAIChatCompletions | ProviderWireApi::OpenAIResponses + ) +} + +fn thinking_levels(model: &Model) -> Vec { + devo_protocol::resolve_thinking_fields_for_model_info( + &model.reasoning_capability, + model.reasoning, + model.thinking_level_map.as_ref(), + ) + .2 +} + +fn strict_thinking(model: &Model, requested: Option<&str>) -> anyhow::Result> { + let Some(requested) = requested else { + return Ok(model.normalize_reasoning_effort_selection(None)); + }; + let available = thinking_levels(model); + anyhow::ensure!( + available.iter().any(|value| value == requested), + "thinking level `{requested}` is unsupported for child model `{}`; supported levels: {}", + model.slug, + available.join(", ") + ); + Ok(Some(requested.to_string())) +} + +#[cfg(test)] +mod tests; diff --git a/crates/server/src/session_context/child_models/tests.rs b/crates/server/src/session_context/child_models/tests.rs new file mode 100644 index 00000000..fb4b292e --- /dev/null +++ b/crates/server/src/session_context/child_models/tests.rs @@ -0,0 +1,530 @@ +use super::*; +use devo_core::AuthCredentialConfig; +use pretty_assertions::assert_eq; +use std::collections::BTreeMap; + +struct NoRequests; +#[async_trait::async_trait] +impl ModelProviderSDK for NoRequests { + async fn completion(&self, _: ModelRequest) -> anyhow::Result { + panic!("discovery must not send requests") + } + async fn completion_stream( + &self, + _: ModelRequest, + ) -> anyhow::Result> + Send>>> { + panic!("discovery must not send requests") + } + fn name(&self) -> &str { + "test" + } +} + +fn snapshot() -> ChildModelCatalog { + let configured: ProviderConfigFile = serde_json::from_value(serde_json::json!({ + "providers": {"local": {"base_url":"http://localhost:8000", "wire_api":"openai_chat_completions", "models": { + "vendor/model": {"name":"Local model", "request_model":"actual-wire-name"} + }}} + })).unwrap(); + let catalog = PresetModelCatalog::load_from_provider_config(&configured).unwrap(); + let mut router = MultiProviderRouter::new(Arc::new(NoRequests)); + router.insert_route( + ProviderRoute::connection("local", ProviderWireApi::OpenAIChatCompletions), + Arc::new(NoRequests), + ); + ChildModelCatalog { + config: AppConfig::default(), + home: PathBuf::new(), + effective: configured.clone(), + configured: configured.clone(), + catalog, + auth: UserAuthConfigFile::default(), + router: Arc::new(router), + route_inputs: ChildRouteInputs { + providers: configured.clone(), + http: ProviderHttpConfig::default(), + auth: Some(UserAuthConfigFile::default()), + }, + } +} + +fn context(home: &Path, providers: &ProviderConfigFile) -> SessionRuntimeContext { + std::fs::write( + home.join("providers.json"), + serde_json::to_vec(providers).unwrap(), + ) + .unwrap(); + let store = AppConfigStore::load(home.to_path_buf(), None).unwrap(); + let catalog = PresetModelCatalog::load_from_provider_config(providers).unwrap(); + let mut router = MultiProviderRouter::new(Arc::new(NoRequests)); + router.insert_route( + ProviderRoute::connection("local", ProviderWireApi::OpenAIChatCompletions), + Arc::new(NoRequests), + ); + SessionRuntimeContext::from_parts( + Arc::new(NoRequests), + Arc::new(router), + Arc::new(ToolRegistry::new()), + crate::empty_mcp_manager(), + "local/vendor/model".to_string(), + Arc::new(catalog), + Arc::new(StdMutex::new(Box::new(FileSystemSkillCatalog::new( + Default::default(), + )))), + AgentsMdConfig::default(), + Arc::new(StdMutex::new(store)), + ) +} + +#[test] +fn explicitly_pinned_models_work_without_directory_metadata() { + let home = tempfile::tempdir().unwrap(); + let mut providers = snapshot().configured; + providers.model = Some("local/vendor/model".to_string()); + providers.small_model = Some("local/vendor/small".to_string()); + providers.providers.get_mut("local").unwrap().models.clear(); + let context = context(home.path(), &providers); + let entries = context.find_child_models(None, 100).unwrap(); + assert_eq!( + entries + .iter() + .map(|entry| (entry.selector.as_str(), entry.thinking_levels.clone())) + .collect::>(), + vec![ + ("local/vendor/model", vec!["off".to_string()]), + ("local/vendor/small", vec!["off".to_string()]) + ] + ); + let (_, turn) = context + .resolve_child_model(None, "local/vendor/model", None, None) + .unwrap(); + assert_eq!( + (turn.model_binding_id, turn.request_model), + ( + Some("local/vendor/model".to_string()), + "vendor/model".to_string() + ) + ); + assert!( + context + .resolve_child_model( + Some("local/vendor/unknown"), + "local/vendor/model", + None, + None + ) + .is_err() + ); + assert!( + context + .resolve_child_model(None, "local/vendor/model", Some("high"), None) + .is_err() + ); +} + +#[test] +fn discovery_reloads_external_provider_disable_and_removal() { + let home = tempfile::tempdir().unwrap(); + let mut providers = snapshot().configured; + let context = context(home.path(), &providers); + assert_eq!(context.find_child_models(None, 100).unwrap().len(), 1); + providers.providers.get_mut("local").unwrap().enabled = Some(false); + std::fs::write( + home.path().join("providers.json"), + serde_json::to_vec(&providers).unwrap(), + ) + .unwrap(); + assert_eq!(context.find_child_models(None, 100).unwrap(), Vec::new()); + assert!( + context + .resolve_child_model(Some("local/vendor/model"), "other/model", None, None) + .is_err() + ); + providers.providers.clear(); + std::fs::write( + home.path().join("providers.json"), + serde_json::to_vec(&providers).unwrap(), + ) + .unwrap(); + assert_eq!(context.find_child_models(None, 100).unwrap(), Vec::new()); +} + +#[test] +fn changed_endpoint_or_account_cannot_use_installed_adapter() { + let mut current = snapshot(); + assert!( + current + .route_inputs + .matches(¤t, "local", "vendor/model") + ); + current + .configured + .providers + .get_mut("local") + .unwrap() + .base_url = Some("http://localhost:9000".to_string()); + assert!( + !current + .route_inputs + .matches(¤t, "local", "vendor/model") + ); + current.configured = current.route_inputs.providers.clone(); + current.auth.credentials.insert( + "local".to_string(), + AuthCredentialConfig { + kind: AuthCredentialKind::ApiKey, + value: "new-account".to_string(), + access: None, + refresh: None, + expires_at: None, + account_id: None, + enterprise_url: None, + }, + ); + assert!( + !current + .route_inputs + .matches(¤t, "local", "vendor/model") + ); +} + +#[test] +fn inherited_non_effort_variant_preserves_request_defaults() { + let home = tempfile::tempdir().unwrap(); + let mut providers = snapshot().configured; + providers + .providers + .get_mut("local") + .unwrap() + .models + .get_mut("vendor/model") + .unwrap() + .variants + .insert( + "special".to_string(), + devo_core::ProviderModelVariantConfig { + request: Some(serde_json::json!({"special_mode": true})), + ..Default::default() + }, + ); + let model = providers + .providers + .get_mut("local") + .unwrap() + .models + .get_mut("vendor/model") + .unwrap(); + model.reasoning = Some(true); + model.thinking_level_map = Some(BTreeMap::from([ + ("low".to_string(), Some("low".to_string())), + ("high".to_string(), Some("high".to_string())), + ])); + let context = context(home.path(), &providers); + let (_, turn) = context + .resolve_child_model(None, "local/vendor/model/special", None, Some("off")) + .unwrap(); + assert_eq!( + ( + turn.model_binding_id, + turn.variant, + turn.provider_request_models.request_defaults().cloned() + ), + ( + Some("local/vendor/model".to_string()), + Some("special".to_string()), + Some(serde_json::json!({"special_mode": true})) + ) + ); + let (_, changed_thinking) = context + .resolve_child_model( + None, + "local/vendor/model/special", + Some("high"), + Some("low"), + ) + .unwrap(); + assert_eq!( + ( + changed_thinking.variant, + changed_thinking.reasoning_effort_selection, + changed_thinking + .provider_request_models + .request_defaults() + .cloned() + ), + ( + Some("special".to_string()), + Some("high".to_string()), + Some(serde_json::json!({"special_mode": true})) + ) + ); +} + +#[test] +fn missing_explicit_binding_never_falls_back_to_another_account() { + let provider = ProviderConfigEntry { + credential: Some("missing".to_string()), + ..Default::default() + }; + let auth = UserAuthConfigFile { + credentials: BTreeMap::from([( + "local".to_string(), + AuthCredentialConfig { + kind: AuthCredentialKind::ApiKey, + value: "wrong-account".to_string(), + access: None, + refresh: None, + expires_at: None, + account_id: None, + enterprise_url: None, + }, + )]), + ..Default::default() + }; + assert!(!credentials_available( + "local", + &provider, + ProviderWireApi::OpenAIChatCompletions, + &auth + )); +} + +#[test] +fn selectors_preserve_slashes_and_reject_malformed_values() { + assert_eq!( + parse_selector("provider/vendor/model").unwrap(), + ("provider", "vendor/model") + ); + for selector in [ + "model", + "", + "/model", + "provider/", + "provider//model", + "provider/model/", + "provider /model", + "provider/mo del", + "provider/model\0", + ] { + assert!(parse_selector(selector).is_err(), "{selector}"); + } +} + +#[test] +fn routes_must_be_configured_enabled_and_live() { + let mut state = snapshot(); + assert_eq!( + state.resolve("local/vendor/model").unwrap(), + ( + "local", + "vendor/model", + ProviderRoute::connection("local", ProviderWireApi::OpenAIChatCompletions) + ) + ); + assert!(state.resolve("LOCAL/vendor/model").is_err()); + assert!(state.resolve("local/unknown").is_err()); + state.configured.providers.get_mut("local").unwrap().enabled = Some(false); + assert!(state.resolve("local/vendor/model").is_err()); + state.configured.providers.get_mut("local").unwrap().enabled = Some(true); + state.router = Arc::new(MultiProviderRouter::new(Arc::new(NoRequests))); + assert!( + state.resolve("local/vendor/model").is_err(), + "stale catalog does not imply live route" + ); +} + +#[test] +fn directory_only_providers_and_disabled_models_are_excluded() { + let mut state = snapshot(); + state.configured.providers.clear(); + assert!(state.resolve("local/vendor/model").is_err()); + state.configured = state.effective.clone(); + state + .effective + .providers + .get_mut("local") + .unwrap() + .models + .get_mut("vendor/model") + .unwrap() + .enabled = Some(false); + assert!(state.resolve("local/vendor/model").is_err()); +} + +#[test] +fn credential_references_must_resolve_but_local_routes_can_be_credentialless() { + let mut state = snapshot(); + let provider = state.effective.providers.get_mut("local").unwrap(); + assert!(credentials_available( + "local", + provider, + ProviderWireApi::OpenAIChatCompletions, + &state.auth + )); + assert!(!credentials_available( + "local", + provider, + ProviderWireApi::AnthropicMessages, + &state.auth + )); + provider.credential = Some("explicit".to_string()); + assert!(!credentials_available( + "local", + provider, + ProviderWireApi::OpenAIChatCompletions, + &state.auth + )); + state.auth.credentials.insert( + "explicit".to_string(), + AuthCredentialConfig { + kind: AuthCredentialKind::ApiKey, + value: "secret-never-serialized".to_string(), + access: None, + refresh: None, + expires_at: None, + account_id: None, + enterprise_url: None, + }, + ); + assert!(credentials_available( + "local", + provider, + ProviderWireApi::OpenAIChatCompletions, + &state.auth + )); + state.auth.credentials.get_mut("explicit").unwrap().value = " ".to_string(); + assert!(!credentials_available( + "local", + provider, + ProviderWireApi::OpenAIChatCompletions, + &state.auth + )); +} + +#[test] +fn expired_and_empty_oauth_tokens_are_not_available() { + let provider = ProviderConfigEntry { + credential: Some("oauth".to_string()), + ..Default::default() + }; + let mut auth = UserAuthConfigFile { + credentials: BTreeMap::from([( + "oauth".to_string(), + AuthCredentialConfig { + kind: AuthCredentialKind::Oauth, + value: String::new(), + access: Some("token".to_string()), + refresh: Some("refresh".to_string()), + expires_at: Some(0), + account_id: None, + enterprise_url: None, + }, + )]), + ..Default::default() + }; + assert!(!credentials_available( + "local", + &provider, + ProviderWireApi::OpenAIResponses, + &auth + )); + auth.credentials.get_mut("oauth").unwrap().expires_at = None; + assert!(credentials_available( + "local", + &provider, + ProviderWireApi::OpenAIResponses, + &auth + )); + auth.credentials.get_mut("oauth").unwrap().access = None; + assert!(!credentials_available( + "local", + &provider, + ProviderWireApi::OpenAIResponses, + &auth + )); +} + +#[test] +fn thinking_requires_exact_supported_levels_and_is_never_clamped() { + let model = Model { + thinking_level_map: Some(BTreeMap::from([ + ("minimal".to_string(), None), + ("medium".to_string(), None), + ("low".to_string(), Some("low".to_string())), + ("high".to_string(), Some("high".to_string())), + ])), + reasoning: Some(true), + ..Default::default() + }; + assert_eq!( + thinking_levels(&model), + vec!["off".to_string(), "low".to_string(), "high".to_string()] + ); + for level in thinking_levels(&model) { + assert_eq!(strict_thinking(&model, Some(&level)).unwrap(), Some(level)); + } + assert!(strict_thinking(&model, Some("xhigh")).is_err()); + assert!(strict_thinking(&model, Some(" HIGH ")).is_err()); + assert!(strict_thinking(&model, Some("HIGH")).is_err()); + assert_eq!( + strict_thinking(&model, Some("high")).unwrap(), + Some("high".to_string()) + ); + assert!(strict_thinking(&model, Some("medium")).is_err()); + assert!(strict_thinking(&model, Some("")).is_err()); + assert!(strict_thinking(&Model::default(), Some("high")).is_err()); + assert_eq!( + strict_thinking(&Model::default(), Some("off")).unwrap(), + Some("off".to_string()) + ); +} + +#[test] +fn derived_capability_levels_reject_unlisted_effort() { + let model = Model { + reasoning_capability: devo_protocol::ReasoningCapability::Levels(vec![ + devo_protocol::ReasoningLevelChoice::Effort(devo_protocol::ReasoningEffort::Low), + devo_protocol::ReasoningLevelChoice::Effort(devo_protocol::ReasoningEffort::High), + ]), + ..Default::default() + }; + assert!(strict_thinking(&model, Some("medium")).is_err()); + assert_eq!( + strict_thinking(&model, Some("high")).unwrap(), + Some("high".to_string()) + ); +} + +#[test] +fn model_specific_route_keeps_entire_model_identifier() { + let model = ProviderModelConfig { + base_url: Some("http://localhost:9000".to_string()), + ..Default::default() + }; + assert_eq!( + model_route( + "local", + "vendor/model", + &model, + ProviderWireApi::OpenAIResponses + ), + ProviderRoute::connection_model("local", ProviderWireApi::OpenAIResponses, "vendor/model") + ); +} + +#[test] +fn discovery_entry_serializes_only_safe_metadata() { + let entry = ChildModelEntry { + provider: "local".to_string(), + id: "vendor/model".to_string(), + name: "Local model".to_string(), + selector: "local/vendor/model".to_string(), + thinking_levels: vec!["off".to_string()], + }; + assert_eq!( + serde_json::to_value(entry).unwrap(), + serde_json::json!({ + "provider":"local", "id":"vendor/model", "name":"Local model", "selector":"local/vendor/model", "thinking_levels":["off"] + }) + ); +} diff --git a/crates/server/tests/child_metadata_selection.rs b/crates/server/tests/child_metadata_selection.rs new file mode 100644 index 00000000..9976ab4f --- /dev/null +++ b/crates/server/tests/child_metadata_selection.rs @@ -0,0 +1,107 @@ +//! Strict settings updates for ephemeral children without SQLite metadata. +#[path = "support/subagent_lifecycle.rs"] +mod support; + +use anyhow::{Context, Result}; +use devo_core::tools::AgentToolCoordinator; +use devo_protocol::{ProviderWireApi, SpawnAgentParams}; +use devo_provider::{MultiProviderRouter, ProviderRoute}; +use devo_server::test_support::TestRuntime; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::sync::Arc; +use support::*; +use tempfile::TempDir; + +#[tokio::test] +async fn ephemeral_child_metadata_rejects_fallback_without_changing_defaults() -> Result<()> { + let home = TempDir::new()?; + let providers = json!({"provider": {"local": { + "wire_api": "openai_chat_completions", "base_url": "https://local.invalid/v1", + "models": { + "main": {"reasoning_capability": {"levels": ["off", "high"]}}, + "other": {"reasoning_capability": {"levels": ["off", "high"]}} + } + }}}); + std::fs::write( + home.path().join("providers.json"), + serde_json::to_vec(&providers)?, + )?; + let provider = Arc::new(ScriptedProvider::pending()); + let mut router = MultiProviderRouter::new(provider.clone()); + router.insert_route( + ProviderRoute::connection("local", ProviderWireApi::OpenAIChatCompletions), + provider.clone(), + ); + let runtime = TestRuntime::new(provider.clone()) + .router(Arc::new(router)) + .default_model("local/main") + .runtime(home.path()); + let (connection, _notifications) = initialize_connection(&runtime).await?; + let parent = start_parent_session(&runtime, connection, home.path()).await?; + let params: SpawnAgentParams = serde_json::from_value(json!({ + "session_id": parent, "message": "Keep working", "ephemeral": true, + "model": "local/main", "thinking": "high" + }))?; + let child = Arc::clone(&runtime).spawn_agent(params).await?; + wait_for_stream_calls(&provider, 1).await?; + let config_before = std::fs::read(home.path().join("config.toml")).ok(); + for (patch, expected_error) in [ + ( + json!({"model": {"provider": "local", "model": "missing"}}), + "child model is not configured", + ), + ( + json!({"settings": {"reasoningEffort": "impossible"}}), + "unsupported for child model", + ), + ] { + let mut params = patch; + params["sessionId"] = json!(child.child_session_id); + params["expectedVersion"] = json!(0); + let response = runtime + .handle_incoming( + connection, + json!({ + "id": 50, "method": "session/metadata/update", "params": params + }), + ) + .await + .context("invalid metadata response")?; + assert!(response.get("error").is_some(), "{response}"); + assert!(response.to_string().contains(expected_error), "{response}"); + assert!( + !response.to_string().contains("cannot be updated directly"), + "test must exercise ephemeral validation, not indexed-child rejection" + ); + } + let response = runtime + .handle_incoming( + connection, + json!({ + "id": 51, "method": "session/metadata/update", "params": { + "sessionId": child.child_session_id, "expectedVersion": 0, + "model": {"provider": "local", "model": "other"}, + "settings": {"reasoningEffort": "off"} + } + }), + ) + .await + .context("valid metadata response")?; + assert!(response.get("error").is_none(), "{response}"); + assert_eq!(response["result"]["appliedToActiveTurn"], json!(true)); + assert_eq!( + response["result"]["session"]["model"]["model"], + json!("local/other") + ); + assert_eq!( + response["result"]["session"]["settings"]["reasoningEffort"], + json!("off") + ); + assert_eq!( + std::fs::read(home.path().join("config.toml")).ok(), + config_before + ); + assert_eq!(provider.requests().len(), 1); + Ok(()) +} diff --git a/crates/server/tests/child_model_selection.rs b/crates/server/tests/child_model_selection.rs new file mode 100644 index 00000000..ef50bfa3 --- /dev/null +++ b/crates/server/tests/child_model_selection.rs @@ -0,0 +1,569 @@ +//! Real Python kernel -> host -> independently routed child model regressions. +#[path = "support/subagent_lifecycle.rs"] +mod support; + +use std::path::Path; +use std::sync::Arc; + +use anyhow::{Context, Result}; +use devo_core::PresetModelCatalog; +use devo_core::tools::{AgentToolCoordinator, create_default_tool_registry}; +use devo_protocol::{ + ModelProfileKey, ModelRequest, ProviderWireApi, ReasoningEffort, SessionId, SpawnAgentParams, +}; +use devo_provider::{MultiProviderRouter, ProviderRoute}; +use devo_server::{ServerRuntime, test_support::TestRuntime}; +use pretty_assertions::assert_eq; +use serde_json::json; +use support::*; +use tempfile::TempDir; + +const PARENT: &str = "parent/vendor/main"; +const CHILD: &str = "alternate/vendor/child"; + +fn write_config(home: &Path) -> Result<()> { + std::fs::write( + home.join("providers.json"), + serde_json::to_vec_pretty(&json!({ + "model": PARENT, + "provider": { + "parent": { + "name": "Parent", "wire_api": "openai_chat_completions", + "base_url": "https://parent.invalid/v1", "enabled": true, + "models": {"vendor/main": { + "name": "Parent model", "request_model": "wire-parent", + "default_reasoning_selection": "off", + "reasoning_capability": {"levels": ["off", "high"]} + }} + }, + "alternate": { + "name": "Alternate", "wire_api": "openai_chat_completions", + "base_url": "https://alternate.invalid/v1", "enabled": true, + "models": {"vendor/child": { + "name": "Child model", "request_model": "wire-child", + "reasoning_capability": {"levels": ["off", "high"]} + }, "disabled": {"name": "Disabled model", "enabled": false}} + }, + "unrouted": { + "name": "Missing adapter", "wire_api": "openai_chat_completions", + "base_url": "https://unrouted.invalid/v1", "enabled": true, + "models": {"missing": {"name": "Unrouted model"}} + } + } + }))?, + )?; + Ok(()) +} + +fn runtime( + home: &Path, + parent: Arc, + child: Arc, +) -> Arc { + let mut router = MultiProviderRouter::new(parent.clone()); + router.insert_route( + ProviderRoute::connection("parent", ProviderWireApi::OpenAIChatCompletions), + parent.clone(), + ); + router.insert_route( + ProviderRoute::connection("alternate", ProviderWireApi::OpenAIChatCompletions), + child, + ); + TestRuntime::new(parent) + .router(Arc::new(router)) + .default_model(PARENT) + .catalog(Arc::new(PresetModelCatalog::new( + [PARENT, CHILD] + .into_iter() + .map(|slug| devo_protocol::Model { + slug: slug.to_string(), + display_name: slug.to_string(), + reasoning_capability: devo_protocol::ReasoningCapability::Levels(vec![ + devo_protocol::ReasoningLevelChoice::Off, + devo_protocol::ReasoningLevelChoice::Effort(ReasoningEffort::High), + ]), + default_reasoning_selection: Some("off".to_string()), + ..devo_protocol::Model::default() + }) + .collect(), + ))) + .registry(Arc::new(create_default_tool_registry())) + .db_file("child-model-selection.db") + .runtime(home) +} + +async fn start_parent_session( + runtime: &Arc, + connection: u64, + cwd: &Path, +) -> Result { + let session = support::start_parent_session(runtime, connection, cwd).await?; + let read = runtime + .handle_incoming( + connection, + json!({ + "id": 50, "method": "session/read", "params": {"sessionId": session} + }), + ) + .await + .context("read parent model selection")?; + assert_eq!( + read["result"]["session"]["model"]["model"], + json!(PARENT), + "fixture must select its configured parent model: {read}" + ); + Ok(session) +} + +fn bindings(requests: &[ModelRequest]) -> Vec<(ModelProfileKey, String, Option)> { + requests + .iter() + .map(|request| { + ( + request.model_slug.clone(), + request.model.clone(), + request.reasoning_effort, + ) + }) + .collect() +} + +fn assert_python_stdout(requests: &[ModelRequest], expected: &str) -> Result<()> { + let outputs = requests + .iter() + .flat_map(|request| &request.messages) + .flat_map(|message| &message.content) + .filter_map(|content| match content { + devo_protocol::RequestContent::ToolResult { + content, is_error, .. + } => Some((content, is_error)), + _ => None, + }) + .map(|(content, is_error)| { + let output: serde_json::Value = serde_json::from_str(content)?; + Ok(( + is_error.unwrap_or(false), + output["details"]["stdout"].clone(), + )) + }) + .collect::>>()?; + assert!( + outputs.contains(&(false, json!(expected))), + "missing Python success output: {outputs:?}" + ); + assert!( + outputs.iter().all(|(error, _)| !error), + "Python tool returned an error: {outputs:?}" + ); + Ok(()) +} + +async fn wait_completed( + runtime: &Arc, + connection: u64, + parent: SessionId, + target: &str, +) -> Result<()> { + tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 15), async { + loop { + let result = request_agent_wait_with( + runtime, + connection, + parent, + Some(target), + /*after_sequence*/ None, + /*timeout_secs*/ 10, + ) + .await?; + for event in result.events { + match event.status.as_deref() { + Some("completed") => return Ok(()), + Some("failed" | "interrupted" | "closed") => { + anyhow::bail!("child {target} terminated: {event:?}") + } + _ => {} + } + } + } + }) + .await + .context("waiting for completed child status")? +} + +async fn spawn_selected( + runtime: &Arc, + parent: SessionId, + model: &str, + thinking: &str, +) -> Result { + let params: SpawnAgentParams = serde_json::from_value(json!({ + "session_id": parent, "message": "Run the isolated child task", "fork_turns": "none", + "model": model, "thinking": thinking + }))?; + Ok(Arc::clone(runtime).spawn_agent(params).await?) +} + +#[tokio::test] +async fn python_spawn_routes_child_sibling_grandchild_and_followup_independently() -> Result<()> { + let home = TempDir::new()?; + write_config(home.path())?; + let parent = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +available = await rlm.find_models() +assert all(not model.selector.startswith('unrouted/') for model in available) +models = await rlm.find_models('alternate') +assert len(models) == 1 +assert models[0].selector == 'alternate/vendor/child' +assert 'high' in models[0].thinking_levels +child = await rlm.spawn('Use your chosen model and spawn a grandchild', name='selected', model=models[0].selector, thinking='high') +assert child.model == models[0].selector +sibling = await rlm.spawn('Inherit the parent model', name='inherited') +assert sibling.model == 'parent/vendor/main' +print('SELECTION_AND_DISCOVERY_OK') +"#, + ), + ScriptedProvider::completed("parent or sibling complete"), + ScriptedProvider::completed("parent or sibling complete"), + ])); + let child = Arc::new(ScriptedProvider::new([ + ScriptedProvider::ipython_code_tool_call( + r#" +grandchild = await rlm.spawn('Inherit the immediate parent model and thinking', name='grandchild') +assert grandchild.model == 'alternate/vendor/child' +print('GRANDCHILD_INHERITANCE_OK') +"#, + ), + ScriptedProvider::completed("child or grandchild complete"), + ScriptedProvider::completed("child or grandchild complete"), + ScriptedProvider::completed("followup complete"), + ])); + let runtime = runtime(home.path(), parent.clone(), child.clone()); + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let root = start_parent_session(&runtime, connection, home.path()).await?; + start_turn_with_approval_policy( + &runtime, + connection, + root, + "Discover and select child models", + Some("never"), + ) + .await?; + let parent_completed = + wait_for_session_notification(&mut notifications, "turn/completed", root).await?; + assert_eq!( + parent_completed["params"]["turn"]["status"], + json!("completed"), + "parent turn failed: {parent_completed}; requests: {:?}", + bindings(&parent.requests()) + ); + wait_for_stream_calls(&parent, 3).await.with_context(|| { + format!( + "parent requests: {:?}; child requests: {:?}", + bindings(&parent.requests()), + bindings(&child.requests()) + ) + })?; + wait_for_stream_calls(&child, 3).await?; + let agents = request_agent_list(&runtime, connection, root).await?; + let selected = agents + .agents + .iter() + .find(|agent| agent.agent_nickname == "selected") + .context("selected child")?; + let selected_id = selected.session_id; + wait_completed(&runtime, connection, root, "selected").await?; + request_agent_send_message( + &runtime, + connection, + root, + "selected", + "Follow up using the same model", + ) + .await?; + wait_for_stream_calls(&child, 4).await?; + wait_completed(&runtime, connection, root, "selected").await?; + assert_eq!( + bindings(&parent.requests()), + vec![ + ( + ModelProfileKey::CatalogSlug(PARENT.to_string()), + "wire-parent".to_string(), + None + ); + 3 + ] + ); + assert_eq!( + bindings(&child.requests()), + vec![ + ( + ModelProfileKey::CatalogSlug(CHILD.to_string()), + "wire-child".to_string(), + Some(ReasoningEffort::High) + ); + 4 + ] + ); + let descendants = request_agent_list(&runtime, connection, selected_id).await?; + assert!( + descendants + .agents + .iter() + .any(|agent| agent.agent_nickname == "grandchild") + ); + assert_python_stdout(&parent.requests(), "SELECTION_AND_DISCOVERY_OK\n")?; + assert_python_stdout(&child.requests(), "GRANDCHILD_INHERITANCE_OK\n")?; + Ok(()) +} + +#[tokio::test] +async fn child_binding_and_thinking_survive_runtime_restart() -> Result<()> { + let home = TempDir::new()?; + write_config(home.path())?; + let parent = Arc::new(ScriptedProvider::new([])); + let child = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "initial child", + )])); + let first = runtime(home.path(), parent.clone(), child.clone()); + let (connection, _notifications) = initialize_connection(&first).await?; + let root = start_parent_session(&first, connection, home.path()).await?; + let spawned = spawn_selected(&first, root, CHILD, "high").await?; + wait_completed(&first, connection, root, &spawned.agent_path).await?; + assert_eq!( + bindings(&child.requests()), + vec![( + ModelProfileKey::CatalogSlug(CHILD.to_string()), + "wire-child".to_string(), + Some(ReasoningEffort::High) + )] + ); + drop(first); + + let restored_child = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "restored child", + )])); + let restored = runtime(home.path(), parent.clone(), restored_child.clone()); + restored.load_persisted_sessions().await?; + let (connection, mut notifications) = initialize_connection(&restored).await?; + let response = restored.handle_incoming(connection, json!({ + "id": 20, "method": "session/resume", "params": {"sessionId": spawned.child_session_id} + })).await.context("resume child")?; + assert!(response.get("error").is_none(), "{response}"); + assert_eq!( + response["result"]["session"]["model"]["model"], + json!(CHILD) + ); + start_turn( + &restored, + connection, + spawned.child_session_id, + "Continue after restart", + ) + .await?; + wait_for_parent_turn_completed(&mut notifications, spawned.child_session_id).await?; + assert_eq!( + bindings(&restored_child.requests()), + vec![( + ModelProfileKey::CatalogSlug(CHILD.to_string()), + "wire-child".to_string(), + Some(ReasoningEffort::High) + )] + ); + assert_eq!(parent.requests().len(), 0); + Ok(()) +} + +#[tokio::test] +async fn invalid_child_selection_is_rejected_before_agent_creation_or_provider_call() -> Result<()> +{ + let home = TempDir::new()?; + write_config(home.path())?; + let parent = Arc::new(ScriptedProvider::new([])); + let child = Arc::new(ScriptedProvider::new([])); + let runtime = runtime(home.path(), parent.clone(), child.clone()); + let (connection, _notifications) = initialize_connection(&runtime).await?; + let root = start_parent_session(&runtime, connection, home.path()).await?; + for (model, thinking) in [ + ("child", "high"), + ("alternate/no-such-model", "high"), + ("alternate/disabled", "off"), + (CHILD, "impossible"), + ("unrouted/missing", "off"), + ] { + let result = spawn_selected(&runtime, root, model, thinking).await; + assert!( + result.is_err(), + "unexpected successful selection: {model} / {thinking}" + ); + assert_eq!( + request_agent_list(&runtime, connection, root) + .await? + .agents + .len(), + 0 + ); + } + assert_eq!((parent.requests().len(), child.requests().len()), (0, 0)); + Ok(()) +} + +#[tokio::test] +async fn restored_child_with_disconnected_route_never_falls_back_to_parent() -> Result<()> { + let home = TempDir::new()?; + write_config(home.path())?; + let parent = Arc::new(ScriptedProvider::new([])); + let child = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "initial child", + )])); + let first = runtime(home.path(), parent.clone(), child.clone()); + let (connection, _notifications) = initialize_connection(&first).await?; + let root = start_parent_session(&first, connection, home.path()).await?; + let spawned = spawn_selected(&first, root, CHILD, "high").await?; + wait_completed(&first, connection, root, &spawned.agent_path).await?; + drop(first); + + let mut router = MultiProviderRouter::new(parent.clone()); + router.insert_route( + ProviderRoute::connection("parent", ProviderWireApi::OpenAIChatCompletions), + parent.clone(), + ); + let restored = TestRuntime::new(parent.clone()) + .router(Arc::new(router)) + .default_model(PARENT) + .db_file("child-model-selection.db") + .runtime(home.path()); + restored.load_persisted_sessions().await?; + let (connection, _notifications) = initialize_connection(&restored).await?; + let response = restored.handle_incoming(connection, json!({ + "id": 30, "method": "session/resume", "params": {"sessionId": spawned.child_session_id} + })).await.context("read disconnected child")?; + assert!( + response.get("error").is_none(), + "read-only resume must remain available: {response}" + ); + let response = restored + .handle_incoming( + connection, + json!({ + "id": 31, "method": "turn/start", "params": { + "sessionId": spawned.child_session_id, + "input": [{"type": "text", "text": "Continue with the persisted model"}], + "idempotencyKey": "disconnected-child" + } + }), + ) + .await + .context("start disconnected child")?; + assert!( + response["error"] + .to_string() + .contains("route is unavailable"), + "missing child route must fail explicitly: {response}" + ); + assert_eq!((parent.requests().len(), child.requests().len()), (0, 1)); + Ok(()) +} + +#[tokio::test] +async fn concurrent_sibling_admissions_keep_distinct_model_bindings() -> Result<()> { + let home = TempDir::new()?; + write_config(home.path())?; + let parent = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "inherited sibling", + )])); + let child = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "selected sibling", + )])); + let runtime = runtime(home.path(), parent.clone(), child.clone()); + let (connection, _notifications) = initialize_connection(&runtime).await?; + let root = start_parent_session(&runtime, connection, home.path()).await?; + let (first, second) = tokio::join!( + spawn_selected(&runtime, root, PARENT, "off"), + spawn_selected(&runtime, root, CHILD, "high"), + ); + let (first, second) = (first?, second?); + wait_completed(&runtime, connection, root, &first.agent_path).await?; + wait_completed(&runtime, connection, root, &second.agent_path).await?; + assert_ne!(first.child_session_id, second.child_session_id); + assert_eq!( + request_agent_list(&runtime, connection, root) + .await? + .agents + .len(), + 2 + ); + assert_eq!( + bindings(&parent.requests()), + vec![( + ModelProfileKey::CatalogSlug(PARENT.to_string()), + "wire-parent".to_string(), + None + )] + ); + assert_eq!( + bindings(&child.requests()), + vec![( + ModelProfileKey::CatalogSlug(CHILD.to_string()), + "wire-child".to_string(), + Some(ReasoningEffort::High) + )] + ); + Ok(()) +} + +#[tokio::test] +async fn invalid_selection_does_not_poison_later_spawn_during_active_parent_turn() -> Result<()> { + let home = TempDir::new()?; + write_config(home.path())?; + let parent = Arc::new(ScriptedProvider::pending()); + let child = Arc::new(ScriptedProvider::new([ScriptedProvider::completed( + "valid child", + )])); + let runtime = runtime(home.path(), parent.clone(), child.clone()); + let (connection, mut notifications) = initialize_connection(&runtime).await?; + let root = start_parent_session(&runtime, connection, home.path()).await?; + start_turn(&runtime, connection, root, "Keep the parent turn active").await?; + wait_for_stream_calls(&parent, 1).await.with_context(|| { + format!( + "parent requests: {:?}; child requests: {:?}", + bindings(&parent.requests()), + bindings(&child.requests()) + ) + })?; + assert!( + spawn_selected(&runtime, root, "alternate/no-such-model", "high") + .await + .is_err() + ); + let selected = spawn_selected(&runtime, root, CHILD, "high").await?; + wait_completed(&runtime, connection, root, &selected.agent_path).await?; + assert_eq!( + request_agent_list(&runtime, connection, root) + .await? + .agents + .len(), + 1 + ); + assert_eq!( + bindings(&child.requests()), + vec![( + ModelProfileKey::CatalogSlug(CHILD.to_string()), + "wire-child".to_string(), + Some(ReasoningEffort::High) + )] + ); + assert_eq!(parent.requests().len(), 1); + let interrupted = runtime + .handle_incoming( + connection, + json!({ + "id": 40, "method": "session/interrupt", "params": {"scope": {"scope": "session", "sessionId": root}} + }), + ) + .await + .context("interrupt pending parent")?; + assert!(interrupted.get("error").is_none(), "{interrupted}"); + wait_for_parent_turn_completed(&mut notifications, root).await?; + Ok(()) +} diff --git a/crates/server/tests/goal_background_agents.rs b/crates/server/tests/goal_background_agents.rs index 8e7966af..368cfcb2 100644 --- a/crates/server/tests/goal_background_agents.rs +++ b/crates/server/tests/goal_background_agents.rs @@ -70,6 +70,8 @@ async fn busy_child_mailbox_keeps_goal_waiting_through_followup() -> Result<()> support::wait_for_captured_request_count(&provider.requests, /*expected*/ 1).await?; let child = Arc::clone(&runtime) .spawn_agent(devo_protocol::SpawnAgentParams { + model: None, + thinking: None, session_id: parent, message: "first child task".to_string(), fork_turns: Some("none".to_string()), diff --git a/crates/server/tests/subagent_lifecycle.rs b/crates/server/tests/subagent_lifecycle.rs index 05a3b926..a1a6a0bf 100644 --- a/crates/server/tests/subagent_lifecycle.rs +++ b/crates/server/tests/subagent_lifecycle.rs @@ -778,6 +778,8 @@ async fn ephemeral_deny_all_child_agent_has_no_tools_and_one_turn() -> Result<() let child = Arc::clone(&runtime) .spawn_agent(devo_protocol::SpawnAgentParams { + model: None, + thinking: None, session_id: parent_session_id, message: "answer this side question".to_string(), fork_turns: Some("all".to_string()), diff --git a/crates/server/tests/support/subagent_lifecycle.rs b/crates/server/tests/support/subagent_lifecycle.rs index a074286e..d6b317ec 100644 --- a/crates/server/tests/support/subagent_lifecycle.rs +++ b/crates/server/tests/support/subagent_lifecycle.rs @@ -287,6 +287,8 @@ pub async fn spawn_child_named( ) -> Result { Ok(Arc::clone(runtime) .spawn_agent(SpawnAgentParams { + model: None, + thinking: None, session_id: parent_session_id, message: message.to_string(), fork_turns: fork_turns.map(str::to_string), diff --git a/docs/prompt-assembly.md b/docs/prompt-assembly.md index e09ef262..9b876d75 100644 --- a/docs/prompt-assembly.md +++ b/docs/prompt-assembly.md @@ -33,7 +33,10 @@ available for callers that do not have agent ancestry. ## Verified runtime details - `rlm.spawn` returns admission metadata, not the child's answer. Native - `kernel_host.rs` rejects model/thinking overrides; children inherit both. + `kernel_host.rs` accepts exact model selectors and supported thinking overrides. + Omitted options inherit the immediate parent; unsupported inherited thinking + also fails admission. `rlm.find_models` discovers configured routable models, + not live health or credit availability. See [the API](rlm-child-models.md). Final assistant text is retained as completion detail in `runtime/agents.rs`. Explicit `agent_message` updates remain useful for timely delivery; substantial artifacts can be shared through files. Child `refine.run` is rejected. diff --git a/docs/rlm-child-models.md b/docs/rlm-child-models.md new file mode 100644 index 00000000..456f77a5 --- /dev/null +++ b/docs/rlm-child-models.md @@ -0,0 +1,68 @@ +# Independent child models + +Native recursive agents can select their own model and reasoning level without +changing the parent session. Selection also applies to children of children. + +## Discover configured models + +```python +models = await rlm.find_models(query="", limit=8) +for model in models: + print(model.selector, model.name, model.thinking_levels) +``` + +`query` is a string; an empty query browses the catalog. `limit` is an integer +from 1 to 100, defaults to 8, and cannot be a boolean. Each `RLMModel` has: + +- `provider`: the configured provider/connection identifier. +- `id`: the model identifier, which may itself contain slashes. +- `name`: the display name. +- `selector`: the exact `provider/model` selector accepted by `spawn`. +- `thinking_levels`: a list of supported reasoning levels. + +Discovery reads enabled, configured, routable connection models. It returns no +credentials and performs no network calls, credential probes, or billing checks. +A catalog result is not a guarantee of live provider health or available credit. +Older hosts returning only the original four fields remain readable; their +`thinking_levels` defaults to an empty list (capabilities are unknown). + +## Spawn with independent settings + +```python +models = await rlm.find_models(query="", limit=100) +chosen = next((model for model in models if "high" in model.thinking_levels), None) +if chosen is not None: + reviewer = await rlm.spawn( + "Review the current diff for concurrency bugs.", + name="reviewer", + model=chosen.selector, + thinking="high", + ) + print(reviewer.model) +``` + +The signature remains `spawn(prompt, *, name, model=None, thinking=None)`. +`prompt` and `name` must be nonempty strings; sibling names must be unique. +`model`, when supplied, must be an exact selector, not a display name or an +unqualified model ID. The first slash separates provider from model; remaining +slashes are part of the model ID. `thinking`, when supplied, must be a nonempty +string supported by that model. Invalid types and empty strings are rejected. + +Omitting either option, or passing `None`, inherits that option from the +**immediate parent**. Selecting a different model does not reset the inherited +thinking level. If that level is unsupported, admission fails; specify a supported +level from discovery instead. Unknown/disabled/unconfigured model selectors and +unsupported explicit thinking levels also fail, without silently substituting +another model or level. Host failures surface as Python `RuntimeError`. + +The model binding is saved on the child session and reused for follow-up turns, +restores, and inherited grandchild settings. Changes to the parent's model or +global defaults do not change an existing child. If a bound connection disappears +or becomes unusable, the child reports an error instead of using a default model. +Direct edits to a connection's endpoint or credentials require reconnecting that +provider so the installed adapter matches the new configuration. + +The handle still contains `rlm_child_id`, `name`, `session_dir`, and the effective +`model`. It represents admission, not completion or the child's answer. Child +completion records, explicit agent messages, and shared files deliver results. +The compatibility wire action remains `rlm.run`; no new RPC or tool is needed.