feat(factory): persist guarded workflows and reconcile recovery

crates/core/prompts/rlm/repl_control.md

@@ -18,11 +18,13 @@ Continual harness state is available as `rlm.harness` and `rlm.get_harness_state
 
 Terminology: continual harness names the persisted prompt, memory, skill, and subagent layer; RLM names the runtime, Python REPL kernel, and native call interface exposed to the model.
 
-RLM-native call contract: installed Python skills are pre-imported modules. Read the matching SKILL.md and call its documented function, such as `await <skill_import>.<function>(...)`; when a CLI exists, use `<skill_import> ...` from shell. Continual harness skill entries are Python REPL skills with an explicit Python `reference` and `arguments` contract. Spawn a reusable delegation spec with `await rlm.spawn('sub-task', name='worker')` (host wire action `rlm.run`); admission returns a child handle immediately. Final assistant output is included in the child completion record; use an available messaging capability for timely updates and files for substantial artifacts. Results are never an `rlm.spawn()` return value. Do not invent non-native wrappers such as `call_skill(...)` or `run_subagent(...)`. Do not call `rlm.create_session` — daemon-backed top-level sessions are out of scope and denied by the host.
+RLM-native call contract: installed Python skills are pre-imported modules. Read the matching SKILL.md and call its documented function, such as `await <skill_import>.<function>(...)`; when a CLI exists, use `<skill_import> ...` from shell. Continual harness skill entries are Python REPL skills with an explicit Python `reference` and `arguments` contract. Spawn a reusable delegation spec with `await rlm.spawn('sub-task', name='worker')` (host wire action `rlm.run`); admission returns a child handle immediately. Final assistant output is included in the child completion record; use an available messaging capability for timely updates and files for substantial artifacts. Results are never an `rlm.spawn()` return value. Do not invent non-native wrappers such as `call_skill(...)` or `run_subagent(...)`. Durable Build roots can call `await rlm.create_session('task', name='independent')` for a separate visible root in the same working directory. It inherits permissions/model settings, has its own journal and no parent lifecycle, and is not addressable by collect/delete_subagent. Child, ephemeral, and Plan callers are rejected.
 
 Child model selection: `await rlm.find_models(query='', limit=8)` lists enabled, configured, routable models with `provider`, `id`, `name`, `selector`, and `thinking_levels`; it does not probe live health or credits. Use an exact returned selector with `await rlm.spawn('sub-task', name='worker', model=chosen.selector, thinking='high')`, choosing a supported thinking level. Model IDs may contain slashes. Omitted model/thinking (or `None`) inherit the immediate parent's values. Invalid selectors or unsupported explicit or inherited thinking fail admission without silent fallback; when changing models, check its supported levels. `limit` is an integer from 1 to 100. Catalog descriptors contain no credentials.
 
 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 `<proposed_plan>` instead.
 - `await rlm.request_user_input([{"id": "choice", "header": "Choice", "question": "Which option?", "options": [{"label": "A", "description": "..."}, {"label": "B", "description": "..."}]}])` uses the session's question UI.
 - `await rlm.web_search("query")` and `await rlm.web_fetch("https://...")` use Rust host operations; `rlm.web_search` requires local search configuration. When provider-hosted web tools are configured, use their provider-native capability directly; those are separate from the regular function-tool list.
+
+Python orchestration controls: `await rlm.collect(targets=None, timeout_ms=0)` snapshots direct children; a positive timeout (up to 300000 ms) waits only until completion, recovery is required, or the deadline. Timeout returns partial snapshots and never cancels work. Results retain status, settled, answer_text, answer_preview, answer_truncated, error, progress_note, and requires_resume. A saved unfinished turn with no live execution reports waiting_for_resume; explicitly recover it rather than replaying blindly. `await rlm.progress_note('running tests')` lets children publish non-steering progress, at most 512 UTF-16 units and one accepted note per 10 seconds. `await rlm.end_turn()` finishes normally after the current tool batch is committed, without an extra model call or canceling children. The rest of the cell and other calls in that batch still run.

crates/core/src/query/event.rs

@@ -141,6 +141,8 @@ pub struct QueryOptions {
     /// Acknowledged journal for durable sessions; absent for ephemeral callers.
     pub journal: Option<Arc<dyn crate::durable_execution::ToolIntentJournal>>,
     pub cancel_token: Option<CancellationToken>,
+    /// Graceful host request: finish after the current durable tool batch.
+    pub end_turn_requested: Option<Arc<std::sync::atomic::AtomicBool>>,
     /// Optional provider used only for compaction summaries. Servers use this
     /// seam to attach Compaction metering without misclassifying the main
     /// streaming query as compaction overhead.

crates/core/src/query/mod.rs

@@ -1323,6 +1323,14 @@ pub async fn query(
             session.end_turn();
             return Err(AgentError::Aborted);
         }
+        if options
+            .end_turn_requested
+            .as_ref()
+            .is_some_and(|requested| requested.load(std::sync::atomic::Ordering::Acquire))
+        {
+            session.end_turn();
+            return Ok(());
+        }
     }
 }
 

crates/core/src/rlm_prompts.rs

@@ -1,7 +1,7 @@
 //! Immutable RLM system-prompt doctrine for Devo sessions.
 //!
 //! Kernel-surface RLM prompt set (reference-agent design, Devo branding) with explicit
-//! denial of daemon `rlm.create_session`. Catalog skill progressive disclosure
+//! bounded subagent and independent-root controls. Catalog skill progressive disclosure
 //! uses XML (`format_catalog_skills_xml`); the markdown skill body from
 //! `devo_skills::render_available_skills_body` remains for non-RLM surfaces.
 
@@ -241,12 +241,14 @@ pub fn build_rlm_base_prompt(options: &RlmPromptOptions) -> String {
         );
     }
 
-    // Explicit deny: never teach daemon create_session (depth-0 path is out of scope).
-    parts.push(String::new());
-    parts.push(
-        "Do not call `rlm.create_session`. Daemon-backed top-level session creation is denied; use `await rlm.spawn(...)` only for child agents."
-            .to_string(),
-    );
+    if has_ipython {
+        parts.push(String::new());
+        parts.push(if options.depth == 0 {
+            "Durable Build roots may call `await rlm.create_session('task', name='independent')` to create a separately visible root in the same workspace. It inherits safety and model settings but has its own journal, no parent linkage, and no collect/child lifecycle. Plan and ephemeral callers are rejected."
+        } else {
+            "Child sessions cannot call `rlm.create_session`; use `await rlm.spawn(...)` for direct descendants."
+        }.to_string());
+    }
 
     if options.allow_recursion && has_ipython {
         parts.push(String::new());
@@ -259,7 +261,7 @@ pub fn build_rlm_base_prompt(options: &RlmPromptOptions) -> String {
                 .to_string(),
         );
         parts.push(
-            "Children inherit your model and thinking level. This host does not support `model` or `thinking` overrides on `rlm.spawn`; omit both options."
+            "Children inherit your model and thinking level by default. Discover exact model selectors and supported thinking levels with `await rlm.find_models()`, then optionally pass `model=chosen.selector` and `thinking` to `rlm.spawn`; unsupported selections fail without fallback."
                 .to_string(),
         );
         if has_agent_observe {
@@ -290,12 +292,12 @@ pub fn build_rlm_base_prompt(options: &RlmPromptOptions) -> String {
             );
         } else {
             parts.push(
-                "Inspect files a child wrote when you need to collect its work without an observation capability."
+                "Use `await rlm.collect(child, timeout_ms=0)` to inspect retained child results without an observation capability, or read the files the child wrote."
                     .to_string(),
             );
         }
         parts.push(
-            "Spawn independent children in separate calls and end your turn instead of awaiting completion. Multiple replies may arrive over multiple turns. Delete a direct child explicitly with `await rlm.delete_subagent(child)` when it is no longer needed."
+            "Use `await rlm.collect(children, timeout_ms=1000)` for a short bounded wait: timeout returns partial snapshots without cancellation. Check `settled`, `requires_resume`, and `answer_truncated`; saved unfinished work requires explicit recovery, not automatic replay. Children can report `await rlm.progress_note('running tests')` without steering the parent. `await rlm.end_turn()` ends normally after the current tool batch while children continue. Delete a direct child explicitly with `await rlm.delete_subagent(child)` when it is no longer needed."
                 .to_string(),
         );
     }
@@ -424,7 +426,7 @@ mod tests {
 
     use super::*;
 
-    /// Verifies: root RLM base prompt carries Devo doctrine strings and denies create_session.
+    /// Root prompt teaches the supported controls and their safety boundaries.
     #[test]
     fn build_rlm_base_prompt_contains_key_doctrine() {
         let mut options = RlmPromptOptions::root("/repo", "/tmp/messages.jsonl");
@@ -441,7 +443,11 @@ mod tests {
         assert!(prompt.contains("rlm.spawn"));
         assert!(prompt.contains("rlm.harness"));
         assert!(prompt.contains("rlm.create_session"));
-        assert!(prompt.contains("denied"));
+        assert!(prompt.contains("Durable Build roots"));
+        assert!(prompt.contains("rlm.collect"));
+        assert!(prompt.contains("rlm.progress_note"));
+        assert!(prompt.contains("rlm.end_turn"));
+        assert!(!prompt.contains("Daemon-backed top-level session creation is denied"));
         assert!(!prompt.contains("daemon-backed depth-0 session"));
         assert!(prompt.contains("Recursive agent depth: 0"));
         assert!(prompt.contains("await refine.run()"));
@@ -506,7 +512,7 @@ mod tests {
             assert!(prompt.contains("Kernel snapshots are separate"));
             assert!(prompt.contains("`rlm.shell()` defaults to the session workspace"));
             assert!(prompt.contains("Bash on Unix and PowerShell on Windows"));
-            assert!(prompt.contains("does not support `model` or `thinking` overrides"));
+            assert!(prompt.contains("unsupported selections fail without fallback"));
             assert!(prompt.contains("never relax relative to the parent") || options.depth == 0);
             assert!(
                 prompt.contains("Never use direct Python process or filesystem APIs to bypass")

crates/kernel/rlm-runtime/README.md

@@ -167,7 +167,8 @@ The main groups are:
 | `rlm.web_search`, `rlm.web_fetch` | Host-mediated network access. |
 | `rlm.update_plan`, `rlm.request_user_input` | Session UI integration. |
 | `rlm.read`, `rlm.write`, `rlm.read_once`, `rlm.write_once` | Direct file I/O, with host-mediated fallback on OS denial. |
-| `rlm.spawn`, `rlm.create_session`, `rlm.find_models`, `rlm.list_subagents`, `rlm.delete_subagent` | Recursive subagent control. |
+| `rlm.spawn`, `rlm.create_session`, `rlm.find_models`, `rlm.list_subagents`, `rlm.delete_subagent`, `rlm.collect`, `rlm.progress_note`, `rlm.end_turn` | Recursive subagent control. |
+| `rlm.factory` | Experimental, explicit-opt-in DAG workflows; active-cell advance/wait, durable checkpoints/recovery, and v2 guarded joins. See [Factory workflows](../../../docs/factory-workflows.md) and [durable workflows](../../../docs/factory-durable-workflows.md). |
 | `rlm.harness`, `rlm.get_harness_state` | Cached reads and async host-owned harness CRUD. |
 | `rlm.emit` | Ship a MIME-keyed display event. |
 
@@ -190,8 +191,12 @@ if chosen is not None:
 ```
 
 `rlm.spawn` returns when the child is admitted, not when it finishes. The child
-runs as a separate session; results arrive through `agent_message` replies or
-files, usually on a later turn.
+runs as a separate session. Use `await rlm.collect(reviewer, timeout_ms=10000)`
+for retained typed results; timeout returns partial snapshots without cancellation.
+Children can report `await rlm.progress_note("running tests")` without steering the
+parent. `await rlm.end_turn()` finishes normally after the current tool batch.
+`create_session` creates a separate durable root in the same workspace. See
+[orchestration semantics](../../../docs/rlm-orchestration.md) for limits and recovery.
 
 Omitted `model` and `thinking` (or `None`) inherit the immediate parent's values.
 An unsupported inherited thinking level fails just like an explicit unsupported

crates/kernel/rlm-runtime/src/rlm/__init__.py

@@ -11,6 +11,7 @@
 from .bash import BashHandle, BashResult, bash
 from .shell import shell
 from .harness import HarnessEntry, HarnessScope, HarnessState, RefinementEvent, get_harness_state
+from .factory import FactoryNamespace
 
 _NOT_CALLABLE_MESSAGE = "'rlm' is not callable; spawn a child with: handle = await rlm.spawn('sub-task', name='worker')"
 _RENAMED_RUN_MESSAGE = "rlm.run was renamed; spawn a child with: handle = await rlm.spawn('sub-task', name='worker')"
@@ -50,6 +51,7 @@ class RLMSubagent:
     session_name: str
     session_dir: Path
     status: str
+    progress_note: str | None = None
 
 
 def _spawn_handle_from_payload(payload: Any) -> RLMSpawnHandle:
@@ -327,13 +329,23 @@ async def create_session(
     thinking: str | None = None,
     cwd: str | None = None,
 ) -> RLMCreateSessionHandle:
-    """Create and prompt a resident depth-0 daemon session.
+    """Create and prompt an independent, durable root in the same workspace.
 
-    Only daemon-backed depth-0 sessions support this operation. The optional
-    arguments set the session name, model, thinking level, and working directory.
+    Requires a durable root in Build mode. Model and safety settings inherit;
+    an explicit model uses an exact selector from find_models. The result is
+    visible in session/list and has its own journal, with no parent lifecycle.
+    An explicit cwd must resolve to the current working directory.
     """
     if not isinstance(prompt, str):
         raise TypeError(f"prompt must be str, got {type(prompt).__name__}")
+    if not prompt.strip():
+        raise ValueError("prompt must be a non-empty string")
+    for label, value in (("name", name), ("model", model), ("thinking", thinking), ("cwd", cwd)):
+        if value is not None:
+            if not isinstance(value, str):
+                raise TypeError(f"{label} must be str")
+            if not value.strip() or value.strip() != value:
+                raise ValueError(f"{label} must be a nonempty string without surrounding whitespace")
     kwargs: dict[str, Any] = {}
     if name is not None:
         kwargs["name"] = name
@@ -397,6 +409,7 @@ def _subagent_from_payload(payload: Any, operation: str = "rlm.list_subagents")
         session_name=session_name,
         session_dir=Path(session_dir),
         status=status,
+        progress_note=payload.get("progress_note"),
     )
 
 
@@ -438,11 +451,13 @@ def __repr__(self) -> str:
 
 
 _harness_state = _HarnessProxy()
+factory = FactoryNamespace()
 
 
 class _RLMNamespace:
     shell = staticmethod(shell)
     harness = _harness_state
+    factory = factory
     get_harness_state = staticmethod(get_harness_state)
 
     async def spawn(
@@ -484,6 +499,15 @@ async def update_plan(
     async def request_user_input(self, questions: list[dict[str, Any]]) -> dict[str, Any]:
         return await request_user_input(questions)
 
+    async def collect(self, targets=None, *, timeout_ms: int = 0) -> list[RLMChildResult]:
+        return await collect(targets, timeout_ms=timeout_ms)
+
+    async def progress_note(self, message: str) -> RLMProgressNoteResult:
+        return await progress_note(message)
+
+    async def end_turn(self) -> dict[str, Any]:
+        return await end_turn()
+
     async def list_subagents(self) -> list[RLMSubagent]:
         return await list_subagents()
 
@@ -515,6 +539,7 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any:
     "BashHandle",
     "BashResult",
     "HarnessEntry",
+    "FactoryNamespace",
     "HarnessScope",
     "HarnessState",
     "McpIntegration",
@@ -531,6 +556,7 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any:
     "delete_subagent",
     "emit",
     "find_models",
+    "factory",
     "get_harness_state",
     "harness",
     "host_request",
@@ -556,3 +582,7 @@ def __getattr__(name: str) -> Any:  # noqa: D401 - module-level lazy attr hook
     if name == "run":
         raise AttributeError(_RENAMED_RUN_MESSAGE)
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+# Imported after handle definitions to keep the orchestration parser independent.
+from .controls import RLMChildResult, RLMProgressNoteResult, collect, progress_note, end_turn
+__all__ += ["RLMChildResult", "RLMProgressNoteResult", "collect", "progress_note", "end_turn"]