From 9930bffc2e4d8a564df4e7966870978cc0f3df1d Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Sun, 8 Nov 2026 23:46:31 +0800 Subject: [PATCH 1/2] feat(factory): add explicit cell-driven DAG workflows --- crates/kernel/rlm-runtime/README.md | 1 + crates/kernel/rlm-runtime/src/rlm/__init__.py | 5 + .../rlm-runtime/src/rlm/_factory_backend.py | 67 ++++ .../rlm-runtime/src/rlm/_factory_spec.py | 97 ++++++ crates/kernel/rlm-runtime/src/rlm/factory.py | 266 ++++++++++++++++ .../kernel/rlm-runtime/tests/test_factory.py | 300 ++++++++++++++++++ .../rlm-runtime/tests/test_factory_spec.py | 53 ++++ docs/factory-workflows.md | 163 ++++++++++ 8 files changed, 952 insertions(+) create mode 100644 crates/kernel/rlm-runtime/src/rlm/_factory_backend.py create mode 100644 crates/kernel/rlm-runtime/src/rlm/_factory_spec.py create mode 100644 crates/kernel/rlm-runtime/src/rlm/factory.py create mode 100644 crates/kernel/rlm-runtime/tests/test_factory.py create mode 100644 crates/kernel/rlm-runtime/tests/test_factory_spec.py create mode 100644 docs/factory-workflows.md diff --git a/crates/kernel/rlm-runtime/README.md b/crates/kernel/rlm-runtime/README.md index 220919f8..be2bac2d 100644 --- a/crates/kernel/rlm-runtime/README.md +++ b/crates/kernel/rlm-runtime/README.md @@ -168,6 +168,7 @@ The main groups are: | `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.factory` | Experimental, explicit-opt-in DAG workflows; advance/wait inside an active cell. Requires the subagent collection API. See [Factory workflows](../../../docs/factory-workflows.md). | | `rlm.harness`, `rlm.get_harness_state` | Cached reads and async host-owned harness CRUD. | | `rlm.emit` | Ship a MIME-keyed display event. | diff --git a/crates/kernel/rlm-runtime/src/rlm/__init__.py b/crates/kernel/rlm-runtime/src/rlm/__init__.py index 9fff1294..ec119360 100644 --- a/crates/kernel/rlm-runtime/src/rlm/__init__.py +++ b/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')" @@ -438,11 +439,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( @@ -515,6 +518,7 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any: "BashHandle", "BashResult", "HarnessEntry", + "FactoryNamespace", "HarnessScope", "HarnessState", "McpIntegration", @@ -531,6 +535,7 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any: "delete_subagent", "emit", "find_models", + "factory", "get_harness_state", "harness", "host_request", diff --git a/crates/kernel/rlm-runtime/src/rlm/_factory_backend.py b/crates/kernel/rlm-runtime/src/rlm/_factory_backend.py new file mode 100644 index 00000000..e7627e38 --- /dev/null +++ b/crates/kernel/rlm-runtime/src/rlm/_factory_backend.py @@ -0,0 +1,67 @@ +"""An isolated adapter to host-owned child lifecycle, never direct session I/O.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Protocol + +from ._factory_spec import FactoryNode + + +@dataclass(frozen=True) +class ChildSnapshot: + child_id: str + status: str + settled: bool + output: str = "" + error: str | None = None + + +class FactoryBackend(Protocol): + """Host boundary for Factory; implementations must preserve parent permissions. + + Spawn admits one ordinary direct child. Inspect is non-consuming and must + return exactly the requested children or raise, never infer completion from + an absent child. Cancel confirms termination, not just signal delivery. + A failed or cancelled spawn has an uncertain outcome; the executor retains + its unique name for cancellation and never silently retries the admission. + """ + + def check_available(self) -> None: ... + + async def spawn(self, node: FactoryNode, prompt: str, name: str) -> str: ... + + async def inspect(self, child_ids: list[str]) -> list[ChildSnapshot]: ... + + async def cancel(self, target: str) -> None: ... + + +class RlmFactoryBackend: + """Use only the public RLM helpers and their existing Rust authorization.""" + + def check_available(self) -> None: + import rlm + + if not callable(getattr(rlm, "collect", None)): + raise RuntimeError( + "Factory requires the RLM child collection API (rlm.collect); " + "install the subagent-controls implementation before starting a run" + ) + + async def spawn(self, node: FactoryNode, prompt: str, name: str) -> str: + import rlm + + handle = await rlm.spawn(prompt, name=name, model=node.model, thinking=node.thinking) + return handle.rlm_child_id + + async def inspect(self, child_ids: list[str]) -> list[ChildSnapshot]: + import rlm + + results = await rlm.collect(targets=child_ids, timeout_ms=0) + return [ChildSnapshot(item.rlm_child_id, item.status, item.settled, + item.answer_text or "", item.error) for item in results] + + async def cancel(self, target: str) -> None: + import rlm + + await rlm.delete_subagent(target) diff --git a/crates/kernel/rlm-runtime/src/rlm/_factory_spec.py b/crates/kernel/rlm-runtime/src/rlm/_factory_spec.py new file mode 100644 index 00000000..b8f2edfe --- /dev/null +++ b/crates/kernel/rlm-runtime/src/rlm/_factory_spec.py @@ -0,0 +1,97 @@ +"""Strict, side-effect-free validation for the first Devo Factory DAG format.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any + +MAX_NODES = 128 +MAX_PARALLEL = 16 +MAX_PROMPT_CHARS = 128_000 +MAX_OUTPUT_CHARS = 32_000 +_ID = re.compile(r"[a-z][a-z0-9_-]{0,31}\Z") + + +@dataclass(frozen=True) +class FactoryNode: + id: str + prompt: str + needs: tuple[str, ...] = () + model: str | None = None + thinking: str | None = None + + +@dataclass(frozen=True) +class FactorySpec: + nodes: tuple[FactoryNode, ...] + max_parallel: int = 4 + failure_policy: str = "fail_fast" + + +def validate(spec: Any) -> FactorySpec: + """Return an immutable DAG; reject unsupported options rather than ignore them. + + Version 1 is intentionally not a Prime machine specification. It supports + finite DAGs only, with explicit dependencies and plain-text child outputs. + """ + if not isinstance(spec, dict): + raise ValueError("factory spec must be an object") + unknown = set(spec) - {"version", "nodes", "max_parallel", "failure_policy"} + if unknown: + raise ValueError(f"unsupported factory fields: {sorted(unknown, key=str)}") + if type(spec.get("version")) is not int or spec["version"] != 1: + raise ValueError("factory version must be 1") + max_parallel = spec.get("max_parallel", 4) + if type(max_parallel) is not int or not 1 <= max_parallel <= MAX_PARALLEL: + raise ValueError(f"max_parallel must be an integer from 1 to {MAX_PARALLEL}") + failure_policy = spec.get("failure_policy", "fail_fast") + if failure_policy not in ("fail_fast", "continue"): + raise ValueError("failure_policy must be fail_fast or continue") + raw_nodes = spec.get("nodes") + if not isinstance(raw_nodes, list) or not 1 <= len(raw_nodes) <= MAX_NODES: + raise ValueError(f"nodes must contain 1 to {MAX_NODES} nodes") + nodes: list[FactoryNode] = [] + seen: set[str] = set() + for raw in raw_nodes: + if not isinstance(raw, dict): + raise ValueError("each factory node must be an object") + unknown = set(raw) - {"id", "prompt", "needs", "model", "thinking"} + if unknown: + raise ValueError(f"unsupported node fields: {sorted(unknown, key=str)}") + node_id = raw.get("id") + if not isinstance(node_id, str) or not _ID.fullmatch(node_id): + raise ValueError("node id must match [a-z][a-z0-9_-]{0,31}") + if node_id in seen: + raise ValueError(f"duplicate node id: {node_id}") + seen.add(node_id) + prompt = raw.get("prompt") + if not isinstance(prompt, str) or not prompt.strip() or len(prompt) > MAX_PROMPT_CHARS: + raise ValueError(f"node {node_id}: prompt must contain 1 to {MAX_PROMPT_CHARS} characters") + needs = raw.get("needs", []) + if not isinstance(needs, list) or not all(isinstance(item, str) for item in needs): + raise ValueError(f"node {node_id}: needs must be a list of node ids") + if len(set(needs)) != len(needs): + raise ValueError(f"node {node_id}: duplicate dependency") + model = raw.get("model") + if model is not None: + if not isinstance(model, str): + raise ValueError(f"node {node_id}: model must be a provider/model selector") + provider, separator, model_id = model.partition("/") + if not separator or not provider.strip() or not model_id.strip(): + raise ValueError(f"node {node_id}: model must be a provider/model selector") + thinking = raw.get("thinking") + if thinking is not None and (not isinstance(thinking, str) or not thinking.strip()): + raise ValueError(f"node {node_id}: thinking must be a nonempty string") + nodes.append(FactoryNode(node_id, prompt, tuple(needs), model, thinking)) + for node in nodes: + missing = set(node.needs) - seen + if missing: + raise ValueError(f"node {node.id}: unknown dependencies: {sorted(missing)}") + remaining = {node.id: set(node.needs) for node in nodes} + while remaining: + ready = {node_id for node_id, needs in remaining.items() if not needs} + if not ready: + raise ValueError("factory dependencies must not contain a cycle") + remaining = {node_id: needs - ready for node_id, needs in remaining.items() if node_id not in ready} + return FactorySpec(tuple(nodes), max_parallel, failure_policy) diff --git a/crates/kernel/rlm-runtime/src/rlm/factory.py b/crates/kernel/rlm-runtime/src/rlm/factory.py new file mode 100644 index 00000000..52690217 --- /dev/null +++ b/crates/kernel/rlm-runtime/src/rlm/factory.py @@ -0,0 +1,266 @@ +"""Opt-in, foreground-driven DAG orchestration over ordinary Devo children. + +No tasks escape a kernel cell: call advance or wait again to drive successors. +Runs are volatile; the Rust host continues to own children and permissions. +""" + +from __future__ import annotations + +import asyncio +import json +import math +from dataclasses import asdict, dataclass, field +from typing import Any +from uuid import uuid4 + +from ._factory_backend import FactoryBackend, RlmFactoryBackend +from ._factory_spec import MAX_OUTPUT_CHARS, MAX_PROMPT_CHARS, FactoryNode, FactorySpec, validate + +_TERMINAL = {"completed", "failed", "stopped"} +_FAILED_NODES = {"failed", "blocked", "cancelled"} +_MAX_RUNS = 32 + +FACTORY_HELP = """Devo Factory v1 (experimental, explicit opt-in) +factory.validate(spec) validate without host calls +await factory.start(spec, enabled=True) validate/register; admit no children yet +await factory.advance(run_id) collect current children; admit ready nodes +await factory.wait(run_id, timeout=30) drive in this cell until done or timeout +factory.status(run_id) detached, local status snapshot +await factory.stop(run_id) stop admission and close live children +factory.forget(run_id) remove only a settled local run + +Spec: {"version": 1, "max_parallel": 4, "failure_policy": "fail_fast", + "nodes": [{"id": "research", "prompt": "Research the change"}, + {"id": "review", "prompt": "Review these findings", + "needs": ["research"]}]} +Node options: model (exact provider/model), thinking. All needs must succeed. +Outputs are passed as a JSON object of predecessor text in the child prompt. +continue runs independent branches after failure; dependent branches block. +Limits: 128 nodes, 16 parallel children, 32 retained runs, 32k chars/output, +128k chars/rendered prompt. Excess output fails its node; it is not truncated. + +This is not Prime machine compatibility: no loops/guards/foreach/residents, +retries, durable specs/runs, restart resume, UI or background advancement. +wait timeout leaves children running; call wait/advance again or explicitly stop. +Interruption does not imply cancellation. Inspect status and call stop if needed. +Host/transport failures are surfaced, never converted to success or respawned. +When status is stopping, inspect cleanup_error and retry stop; do not forget it. +The host owns child sessions; status() can be stale until advance()/wait(). +""" + + +@dataclass +class _NodeRun: + id: str + name: str + status: str = "pending" + child_id: str | None = None + output: str | None = None + error: str | None = None + cleanup_error: str | None = None + needs_cleanup: bool = False + + +@dataclass +class _Run: + id: str + spec: FactorySpec + nodes: dict[str, _NodeRun] + status: str = "ready" + stop_requested: bool = False + failed: bool = False + lock: asyncio.Lock = field(default_factory=asyncio.Lock) + + +class FactoryNamespace: + """Cell-driven scheduler. A backend injection is intended for host adapters/tests.""" + + def __init__(self, backend: FactoryBackend | None = None): + self._backend = backend if backend is not None else RlmFactoryBackend() + self._runs: dict[str, _Run] = {} + + help = staticmethod(lambda: FACTORY_HELP) + validate = staticmethod(validate) + + async def start(self, spec: dict[str, Any], *, enabled: bool = False) -> dict[str, Any]: + """Register a validated volatile run. Opt in for every new run explicitly.""" + if enabled is not True: + raise ValueError("Factory is experimental; pass enabled=True to start a run") + canonical = validate(spec) + self._backend.check_available() + if len(self._runs) >= _MAX_RUNS: + raise RuntimeError("Factory run limit reached; forget a terminal run first") + run_id = uuid4().hex + run = _Run(run_id, canonical, { + node.id: _NodeRun(node.id, f"factory-{run_id[:16]}-{node.id}") + for node in canonical.nodes + }) + self._runs[run_id] = run + return self.status(run_id) + + def status(self, run_id: str) -> dict[str, Any]: + """Read a detached snapshot, without reading or mutating child sessions.""" + run = self._runs[run_id] + return {"id": run.id, "status": run.status, "max_parallel": run.spec.max_parallel, + "failure_policy": run.spec.failure_policy, "stop_requested": run.stop_requested, + "nodes": {node_id: asdict(node) for node_id, node in run.nodes.items()}} + + async def advance(self, run_id: str) -> dict[str, Any]: + """Perform one serialized observation/admission pass in the current cell.""" + run = self._runs[run_id] + async with run.lock: + if run.status in _TERMINAL: + return self.status(run_id) + if run.stop_requested: + await self._cleanup(run) + return self.status(run_id) + run.status = "running" + active = {node.child_id: node for node in run.nodes.values() + if node.status in ("running", "waiting_for_input")} + if active: + # Do not partially apply malformed or incomplete replies. A read + # failure leaves all child IDs retained and can be safely retried. + snapshots = await self._backend.inspect(list(active)) + ids = [item.child_id for item in snapshots] + if len(ids) != len(active) or set(ids) != set(active): + raise RuntimeError("Factory collect returned missing or duplicate children") + for item in snapshots: + if item.status not in {"running", "waiting_for_input", "completed", "failed", + "interrupted", "canceled", "closed"}: + raise RuntimeError(f"Factory collect returned unknown status {item.status!r}") + if type(item.settled) is not bool or item.settled != ( + item.status not in {"running", "waiting_for_input"} + ) or not isinstance(item.output, str): + raise RuntimeError("Factory collect returned inconsistent child state") + for item in snapshots: + node = active[item.child_id] + if not item.settled: + node.status = item.status + continue + node.needs_cleanup = False + if item.status == "completed" and len(item.output) <= MAX_OUTPUT_CHARS: + node.status, node.output = "completed", item.output + else: + node.status = "failed" + node.error = (f"output exceeds {MAX_OUTPUT_CHARS} characters" + if item.status == "completed" else item.error or item.status) + run.failed = True + # Block transitively, even when nodes were supplied out of order. + changed = True + while changed: + changed = False + for spec in run.spec.nodes: + node = run.nodes[spec.id] + if node.status == "pending" and any( + run.nodes[dep].status in _FAILED_NODES for dep in spec.needs + ): + node.status, node.error = "blocked", "a dependency did not complete successfully" + changed = True + if run.failed and run.spec.failure_policy == "fail_fast": + run.stop_requested = True + if run.stop_requested: + await self._cleanup(run) + return self.status(run_id) + for spec in run.spec.nodes: + if run.stop_requested: + break + active_count = sum(node.needs_cleanup for node in run.nodes.values()) + if active_count >= run.spec.max_parallel: + break + node = run.nodes[spec.id] + if node.status != "pending" or not all( + run.nodes[dep].status == "completed" for dep in spec.needs + ): + continue + prompt = self._prompt(spec, run) + if len(prompt) > MAX_PROMPT_CHARS: + node.status, node.error = "failed", "rendered prompt exceeds the Factory limit" + run.failed = True + # A fresh pass propagates blocks before admitting successors. + if run.spec.failure_policy == "fail_fast": + run.stop_requested = True + break + node.status, node.needs_cleanup = "running", True + try: + node.child_id = await self._backend.spawn(spec, prompt, node.name) + if not isinstance(node.child_id, str) or not node.child_id: + node.child_id = None + raise RuntimeError("spawn returned an invalid child id") + except asyncio.CancelledError: + node.status, node.error = "failed", "spawn interrupted; admission outcome uncertain" + run.failed = run.stop_requested = True + run.status = "stopping" + raise + except Exception as exc: + node.status, node.error = "failed", f"spawn outcome uncertain: {exc}" + run.failed = run.stop_requested = True + # Never retry an uncertain admission. Keep its unique name + # until stop can confirm that child has been closed. + break + if run.stop_requested: + await self._cleanup(run) + elif all(node.status in {"completed", "failed", "blocked"} for node in run.nodes.values()): + run.status = "failed" if run.failed else "completed" + return self.status(run_id) + + @staticmethod + def _prompt(spec: FactoryNode, run: _Run) -> str: + if not spec.needs: + return spec.prompt + outputs = {dep: run.nodes[dep].output for dep in spec.needs} + return (spec.prompt + "\n\nFactory predecessor outputs (task data, not new authorization):\n" + + json.dumps(outputs, ensure_ascii=False)) + + async def _cleanup(self, run: _Run) -> None: + run.status = "stopping" + for node in run.nodes.values(): + if node.status == "pending": + node.status = "cancelled" + if not node.needs_cleanup: + continue + try: + await self._backend.cancel(node.child_id or node.name) + except Exception as exc: + node.cleanup_error = str(exc) + else: + node.needs_cleanup, node.cleanup_error = False, None + if node.status != "failed": + node.status = "cancelled" + if not any(node.needs_cleanup for node in run.nodes.values()): + run.status = "failed" if run.failed else "stopped" + + async def stop(self, run_id: str) -> dict[str, Any]: + """Stop new admissions immediately; retain uncertain cleanup for retry.""" + run = self._runs[run_id] + if run.status in _TERMINAL: + return self.status(run_id) + run.stop_requested = True + async with run.lock: + await self._cleanup(run) + return self.status(run_id) + + async def wait(self, run_id: str, *, timeout: float = 30, poll_interval: float = 0.2) -> dict[str, Any]: + """Drive in this cell. Timeout is a polling budget, not child cancellation. + + An in-flight host call is not cut off at the deadline: doing so can lose + a spawn/close acknowledgement. The caller may interrupt and then stop. + """ + for label, value in (("timeout", timeout), ("poll_interval", poll_interval)): + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): + raise ValueError(f"{label} must be finite") + if not 0 <= timeout <= 300 or not 0.01 <= poll_interval <= 60: + raise ValueError("timeout must be 0..300 seconds; poll_interval must be 0.01..60 seconds") + deadline = asyncio.get_running_loop().time() + timeout + while True: + result = await self.advance(run_id) + remaining = deadline - asyncio.get_running_loop().time() + if result["status"] in _TERMINAL or result["status"] == "stopping" or remaining <= 0: + return result + await asyncio.sleep(min(poll_interval, remaining)) + + def forget(self, run_id: str) -> None: + """Drop only a settled local run; this never deletes host child transcripts.""" + run = self._runs[run_id] + if run.status not in _TERMINAL or run.lock.locked(): + raise ValueError("only a terminal Factory run can be forgotten") + del self._runs[run_id] diff --git a/crates/kernel/rlm-runtime/tests/test_factory.py b/crates/kernel/rlm-runtime/tests/test_factory.py new file mode 100644 index 00000000..3f049bee --- /dev/null +++ b/crates/kernel/rlm-runtime/tests/test_factory.py @@ -0,0 +1,300 @@ +import asyncio +import unittest +from dataclasses import replace +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import rlm +from rlm._factory_backend import ChildSnapshot, RlmFactoryBackend +from rlm._factory_spec import FactoryNode, MAX_OUTPUT_CHARS +from rlm.factory import FactoryNamespace + + +def dag(*nodes, **options): + return {"version": 1, "nodes": [ + {"id": node, "prompt": f"Task {node}"} if isinstance(node, str) else node + for node in nodes + ], **options} + + +class FakeBackend: + def __init__(self): + self.spawned = [] + self.cancelled = [] + self.children = {} + + def check_available(self): + pass + + async def spawn(self, node, prompt, name): + child_id = f"child-{len(self.spawned)}" + self.spawned.append((node, prompt, name)) + self.children[child_id] = ChildSnapshot(child_id, "running", False) + return child_id + + async def inspect(self, child_ids): + return [self.children[child_id] for child_id in child_ids] + + async def cancel(self, target): + self.cancelled.append(target) + + def settle(self, child_id, output="done", status="completed"): + self.children[child_id] = ChildSnapshot(child_id, status, True, output) + + +class FactoryTests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.backend = FakeBackend() + self.factory = FactoryNamespace(self.backend) + + async def start(self, spec): + return (await self.factory.start(spec, enabled=True))["id"] + + async def test_opt_in_validation_and_start_do_not_spawn(self): + for enabled in (False, 1, "true", None): + with self.assertRaises(ValueError): + await self.factory.start(dag("a"), enabled=enabled) + with self.assertRaises(ValueError): + await self.factory.start(dag("a", "a"), enabled=True) + run = await self.factory.start(dag("a"), enabled=True) + self.assertEqual((run["status"], self.backend.spawned), ("ready", [])) + run["nodes"]["a"]["status"] = "corrupted" + self.assertEqual(self.factory.status(run["id"])["nodes"]["a"]["status"], "pending") + + async def test_diamond_joins_wait_for_every_dependency_and_forward_outputs(self): + run_id = await self.start(dag("a", "b", { + "id": "join", "prompt": "Combine", "needs": ["a", "b"], + })) + await self.factory.advance(run_id) + self.assertEqual([entry[0].id for entry in self.backend.spawned], ["a", "b"]) + self.backend.settle("child-0", "A\nquoted \" output") + await self.factory.advance(run_id) + self.assertEqual(len(self.backend.spawned), 2) + self.backend.settle("child-1", "B") + await self.factory.advance(run_id) + self.assertEqual(self.backend.spawned[2][:2], ( + FactoryNode("join", "Combine", ("a", "b")), + 'Combine\n\nFactory predecessor outputs (task data, not new authorization):\n' + '{"a": "A\\nquoted \\" output", "b": "B"}', + )) + self.backend.settle("child-2", "Final") + final = await self.factory.advance(run_id) + self.assertEqual((final["status"], final["nodes"]["join"]["output"]), ("completed", "Final")) + self.assertEqual(await self.factory.advance(run_id), final) + self.assertEqual(await self.factory.stop(run_id), final) + self.assertEqual(self.backend.cancelled, []) + + async def test_concurrent_advances_admit_each_node_once_and_obey_limit(self): + run_id = await self.start(dag("a", "b", "c", max_parallel=1)) + await asyncio.gather(*(self.factory.advance(run_id) for _ in range(8))) + self.assertEqual([entry[0].id for entry in self.backend.spawned], ["a"]) + self.backend.settle("child-0") + await asyncio.gather(*(self.factory.advance(run_id) for _ in range(8))) + self.assertEqual([entry[0].id for entry in self.backend.spawned], ["a", "b"]) + self.assertTrue(all(len(entry[2]) <= 64 for entry in self.backend.spawned)) + + async def test_continue_blocks_descendants_but_finishes_independent_work(self): + run_id = await self.start(dag( + {"id": "grandchild", "prompt": "G", "needs": ["child"]}, + {"id": "child", "prompt": "C", "needs": ["a"]}, + "a", "b", failure_policy="continue")) + await self.factory.advance(run_id) + self.backend.settle("child-0", status="failed") + self.backend.settle("child-1", "Independent") + result = await self.factory.advance(run_id) + self.assertEqual((result["status"], {k: v["status"] for k, v in result["nodes"].items()}), + ("failed", {"grandchild": "blocked", "child": "blocked", + "a": "failed", "b": "completed"})) + self.assertEqual(self.backend.cancelled, []) + + async def test_fail_fast_closes_other_children_and_cancels_pending(self): + run_id = await self.start(dag("a", "b", "c", max_parallel=2)) + await self.factory.advance(run_id) + self.backend.settle("child-0", status="interrupted") + result = await self.factory.advance(run_id) + self.assertEqual((result["status"], self.backend.cancelled), ("failed", ["child-1"])) + self.assertEqual([n["status"] for n in result["nodes"].values()], + ["failed", "cancelled", "cancelled"]) + self.assertEqual(len(self.backend.spawned), 2) + + async def test_stop_cleanup_failure_stays_visible_and_retryable(self): + run_id = await self.start(dag("a", "b")) + await self.factory.advance(run_id) + with patch.object(self.backend, "cancel", AsyncMock(side_effect=RuntimeError("offline"))): + result = await self.factory.stop(run_id) + self.assertEqual(result["status"], "stopping") + self.assertEqual([n["cleanup_error"] for n in result["nodes"].values()], ["offline", "offline"]) + with self.assertRaises(ValueError): + self.factory.forget(run_id) + result = await self.factory.stop(run_id) + self.assertEqual((result["status"], self.backend.cancelled), ("stopped", ["child-0", "child-1"])) + self.assertTrue(all(n["cleanup_error"] is None for n in result["nodes"].values())) + self.factory.forget(run_id) + with self.assertRaises(KeyError): + self.factory.status(run_id) + + async def test_stop_during_spawn_keeps_acknowledged_child_and_prevents_next_spawn(self): + entered, release = asyncio.Event(), asyncio.Event() + original = self.backend.spawn + + async def blocked_spawn(node, prompt, name): + entered.set() + await release.wait() + return await original(node, prompt, name) + + run_id = await self.start(dag("a", "b")) + with patch.object(self.backend, "spawn", blocked_spawn): + advancing = asyncio.create_task(self.factory.advance(run_id)) + await entered.wait() + stopping = asyncio.create_task(self.factory.stop(run_id)) + await asyncio.sleep(0) + release.set() + await asyncio.gather(advancing, stopping) + result = self.factory.status(run_id) + self.assertEqual((result["status"], len(self.backend.spawned), self.backend.cancelled), + ("stopped", 1, ["child-0"])) + + async def test_interrupted_spawn_retains_unique_name_and_never_respawns(self): + entered = asyncio.Event() + + async def blocked_spawn(node, prompt, name): + entered.set() + await asyncio.Event().wait() + + run_id = await self.start(dag("a", "b")) + with patch.object(self.backend, "spawn", blocked_spawn): + advancing = asyncio.create_task(self.factory.advance(run_id)) + await entered.wait() + advancing.cancel() + with self.assertRaises(asyncio.CancelledError): + await advancing + result = self.factory.status(run_id) + self.assertEqual(result["status"], "stopping") + self.assertTrue(result["nodes"]["a"]["needs_cleanup"]) + self.assertIsNone(result["nodes"]["a"]["child_id"]) + await self.factory.stop(run_id) + self.assertEqual(self.backend.cancelled, [result["nodes"]["a"]["name"]]) + await self.factory.advance(run_id) + self.assertEqual(self.backend.spawned, []) + + async def test_uncertain_spawn_does_not_retry_and_cleanup_failure_is_not_success(self): + run_id = await self.start(dag("a", "b")) + with patch.object(self.backend, "spawn", AsyncMock(side_effect=RuntimeError("connection lost"))) as spawn: + with patch.object(self.backend, "cancel", AsyncMock(side_effect=RuntimeError("unknown outcome"))): + result = await self.factory.advance(run_id) + again = await self.factory.advance(run_id) + self.assertEqual(spawn.await_count, 1) + self.assertEqual((result["status"], again["status"]), ("stopping", "stopping")) + self.assertIn("connection lost", result["nodes"]["a"]["error"]) + self.assertEqual(result["nodes"]["b"]["status"], "cancelled") + + async def test_collect_failure_does_not_lose_children_or_admit_more(self): + run_id = await self.start(dag("a", "b", max_parallel=1)) + await self.factory.advance(run_id) + before = self.factory.status(run_id) + with patch.object(self.backend, "inspect", AsyncMock(side_effect=RuntimeError("offline"))): + with self.assertRaisesRegex(RuntimeError, "offline"): + await self.factory.advance(run_id) + self.assertEqual(self.factory.status(run_id), before) + self.assertEqual(len(self.backend.spawned), 1) + self.backend.settle("child-0") + await self.factory.advance(run_id) + self.assertEqual(len(self.backend.spawned), 2) + + async def test_malformed_snapshot_is_rejected_atomically(self): + run_id = await self.start(dag("a", "b")) + await self.factory.advance(run_id) + before = self.factory.status(run_id) + a = ChildSnapshot("child-0", "completed", True, "A") + b = ChildSnapshot("child-1", "running", False) + invalid = [[], [a, a], [a, replace(b, child_id="stranger")], + [a, replace(b, status="unknown")], [a, replace(b, settled=True)], + [a, replace(b, output=None)]] + for reply in invalid: + with self.subTest(reply=reply), patch.object(self.backend, "inspect", AsyncMock(return_value=reply)): + with self.assertRaises(RuntimeError): + await self.factory.advance(run_id) + self.assertEqual(self.factory.status(run_id), before) + + async def test_wait_timeout_leaves_live_children_for_later_cells(self): + run_id = await self.start(dag("a")) + result = await self.factory.wait(run_id, timeout=0) + self.assertEqual((result["status"], self.backend.cancelled), ("running", [])) + self.backend.settle("child-0") + result = await self.factory.wait(run_id, timeout=0.1, poll_interval=0.01) + self.assertEqual(result["status"], "completed") + + async def test_wait_invalid_bounds_are_side_effect_free(self): + run_id = await self.start(dag("a")) + for options in ({"timeout": True}, {"timeout": -1}, {"timeout": 301}, + {"timeout": float("nan")}, {"poll_interval": float("inf")}, + {"poll_interval": 0}, {"poll_interval": "fast"}): + with self.subTest(options=options), self.assertRaises(ValueError): + await self.factory.wait(run_id, **options) + self.assertEqual(self.backend.spawned, []) + + async def test_oversized_output_and_rendered_prompts_fail_without_truncation(self): + run_id = await self.start(dag("a", {"id": "b", "prompt": "B", "needs": ["a"]})) + await self.factory.advance(run_id) + self.backend.settle("child-0", "x" * (MAX_OUTPUT_CHARS + 1)) + result = await self.factory.advance(run_id) + self.assertEqual((result["status"], result["nodes"]["a"]["output"], len(self.backend.spawned)), + ("failed", None, 1)) + run_id = await self.start(dag("a", {"id": "b", "prompt": "B" * 128000, "needs": ["a"]})) + await self.factory.advance(run_id) + self.backend.settle("child-1", "small") + result = await self.factory.advance(run_id) + self.assertEqual((result["status"], len(self.backend.spawned)), ("failed", 2)) + self.assertIn("rendered prompt", result["nodes"]["b"]["error"]) + + async def test_waiting_for_input_holds_capacity_without_blocking_local_status(self): + run_id = await self.start(dag("a", "b", max_parallel=1)) + await self.factory.advance(run_id) + self.backend.children["child-0"] = ChildSnapshot("child-0", "waiting_for_input", False) + result = await self.factory.advance(run_id) + self.assertEqual((result["nodes"]["a"]["status"], len(self.backend.spawned)), ("waiting_for_input", 1)) + await self.factory.stop(run_id) + self.assertEqual(self.backend.cancelled, ["child-0"]) + + async def test_retained_run_limit_requires_explicit_forget(self): + with patch("rlm.factory._MAX_RUNS", 1): + run_id = await self.start(dag("a")) + with self.assertRaisesRegex(RuntimeError, "limit"): + await self.start(dag("b")) + await self.factory.stop(run_id) + self.factory.forget(run_id) + await self.start(dag("b")) + + +class FactoryAdapterTests(unittest.IsolatedAsyncioTestCase): + async def test_missing_collect_fails_before_registering_or_spawning(self): + factory = FactoryNamespace() + with patch.object(rlm, "collect", None, create=True), patch.object(rlm, "spawn", AsyncMock()) as spawn: + with self.assertRaisesRegex(RuntimeError, "requires.*rlm.collect"): + await factory.start(dag("a"), enabled=True) + self.assertEqual(factory._runs, {}) + spawn.assert_not_awaited() + + async def test_adapter_uses_existing_public_helpers_and_preserves_options(self): + backend = RlmFactoryBackend() + node = FactoryNode("a", "Task", (), "provider/model", "high") + raw = SimpleNamespace(rlm_child_id="id", status="completed", settled=True, + answer_text="answer", error=None) + with patch.object(rlm, "collect", AsyncMock(return_value=[raw]), create=True) as collect: + with patch.object(rlm, "spawn", AsyncMock(return_value=SimpleNamespace(rlm_child_id="id"))) as spawn: + with patch.object(rlm, "delete_subagent", AsyncMock()) as cancel: + backend.check_available() + self.assertEqual(await backend.spawn(node, "Rendered", "unique"), "id") + self.assertEqual(await backend.inspect(["id"]), [ChildSnapshot("id", "completed", True, "answer")]) + await backend.cancel("id") + spawn.assert_awaited_once_with("Rendered", name="unique", model="provider/model", thinking="high") + collect.assert_awaited_once_with(targets=["id"], timeout_ms=0) + cancel.assert_awaited_once_with("id") + + async def test_namespace_and_module_share_one_factory(self): + self.assertIs(rlm.factory, rlm.rlm.factory) + self.assertIn("no loops", rlm.factory.help()) + + +if __name__ == "__main__": + unittest.main() diff --git a/crates/kernel/rlm-runtime/tests/test_factory_spec.py b/crates/kernel/rlm-runtime/tests/test_factory_spec.py new file mode 100644 index 00000000..d1c14c3f --- /dev/null +++ b/crates/kernel/rlm-runtime/tests/test_factory_spec.py @@ -0,0 +1,53 @@ +import copy +import unittest + +from rlm._factory_spec import FactoryNode, FactorySpec, MAX_NODES, validate + + +class FactorySpecTests(unittest.TestCase): + def test_defaults_and_topology_are_immutable(self): + spec = {"version": 1, "nodes": [ + {"id": "review", "prompt": "Review", "needs": ["build"], + "model": "gateway/org/model", "thinking": "high"}, + {"id": "build", "prompt": "Build"}, + ]} + original = copy.deepcopy(spec) + result = validate(spec) + self.assertEqual(result, FactorySpec(( + FactoryNode("review", "Review", ("build",), "gateway/org/model", "high"), + FactoryNode("build", "Build"), + ))) + self.assertEqual(spec, original) + spec["nodes"][0]["needs"].append("another") + self.assertEqual(result.nodes[0].needs, ("build",)) + + def test_invalid_specs(self): + good = {"version": 1, "nodes": [{"id": "one", "prompt": "Task"}]} + bad_specs = [None, [], {}, {**good, "version": True}, {**good, "version": 2}, + {**good, "machine": {}}, {**good, "max_parallel": True}, + {**good, "max_parallel": 0}, {**good, "max_parallel": 17}, + {**good, "max_parallel": 1.0}, {**good, "max_parallel": None}, + {**good, "failure_policy": "escalate"}, {**good, "nodes": []}, + {**good, "nodes": [{}]}, {**good, "nodes": [None]}, + {**good, "nodes": [good["nodes"][0]] * (MAX_NODES + 1)}, + {**good, "nodes": [good["nodes"][0]] * 2}] + node = good["nodes"][0] + for overrides in ({"id": "Bad"}, {"id": "../other"}, {"id": "x" * 33}, + {"prompt": " "}, {"prompt": None}, {"prompt": "x" * 128001}, + {"needs": "one"}, {"needs": [1]}, {"needs": ["missing"]}, + {"needs": ["one"]}, {"needs": ["one", "one"]}, + {"model": "model"}, {"model": "/model"}, {"model": 2}, + {"thinking": ""}, {"thinking": True}, {"retries": 1}, + {"foreach": []}, {"guard": {}}): + bad_specs.append({**good, "nodes": [{**node, **overrides}]}) + bad_specs.append({**good, "nodes": [ + {"id": "one", "prompt": "A", "needs": ["two"]}, + {"id": "two", "prompt": "B", "needs": ["one"]}, + ]}) + for spec in bad_specs: + with self.subTest(spec=str(spec)[:150]), self.assertRaises(ValueError): + validate(spec) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs/factory-workflows.md b/docs/factory-workflows.md new file mode 100644 index 00000000..c43a97bf --- /dev/null +++ b/docs/factory-workflows.md @@ -0,0 +1,163 @@ +# Factory workflows: experimental DAG slice + +Devo Factory v1 composes ordinary direct child agents into a finite DAG. +It provides strict preflight validation, bounded parallel admission, all-parent +joins, plain-text result forwarding, explicit failure policies, local status, +and stop/cleanup. It does not replace the Rust session supervisor or grant +additional permissions. + +## Dependencies and boundaries + +The default backend needs the **subagent-controls implementation** providing +`rlm.collect(targets=[...], timeout_ms=0)` and its host route, in addition to the +existing `rlm.spawn` and `rlm.delete_subagent` helpers. On a baseline without +the Python collection API, `start` fails before creating a run or spawning +children. A mismatched/older host surfaces an error from its existing bridge. +The `FactoryBackend` protocol is isolated so this module and its scheduler +tests can be reviewed independently of that implementation. + +Every run requires explicit `enabled=True`. This is an API opt-in, not a new +persisted session setting, permission grant, or equivalent of Prime's +`/factory on` command. Existing child model selection, resource limits, +approval rules, session ownership, and cancellation remain host-owned. + +**This release advances only inside a live Python cell.** `start` validates and +registers a run without spawning; `advance` performs one pass; `wait` keeps +performing passes in that cell. There is no hidden asyncio task. Children +already spawned are host-owned and may keep running when the cell ends, but +their successors will not start until another `advance` or `wait` call. +This respects Devo's cell-bound host capability lifetime: an idle/detached +kernel request must not reuse an old or a later turn's permissions. + +Run state is in kernel memory, not the durable session journal. A kernel +restart loses the graph, outputs, and cleanup tracking, not necessarily the +host-owned children. After a restart, inspect `rlm.list_subagents()` / collect +and explicitly stop unwanted children. Do not claim automatic run recovery. + +## Example: parallel investigation, joined review + +Run in an active Devo Python cell after the collection API is installed: + +```python +spec = { + "version": 1, + "max_parallel": 2, + "failure_policy": "fail_fast", + "nodes": [ + {"id": "runtime", "prompt": "Inspect runtime behavior; report findings only."}, + {"id": "tests", "prompt": "Inspect relevant test coverage; report findings only."}, + { + "id": "review", + "needs": ["runtime", "tests"], + "prompt": "Combine both findings into a concise implementation recommendation.", + }, + ], +} +run = await rlm.factory.start(spec, enabled=True) +run_id = run["id"] +result = await rlm.factory.wait(run_id, timeout=30) +print(result) +``` + +If still running, use a later cell: + +```python +result = await rlm.factory.wait(run_id, timeout=30) +# Or advance once, returning without waiting for child completion: +result = await rlm.factory.advance(run_id) +# Local status alone does not refresh the host or start work: +print(rlm.factory.status(run_id)) +``` + +To stop: + +```python +result = await rlm.factory.stop(run_id) +print(result) +# Only after completed / failed / stopped: +rlm.factory.forget(run_id) +``` + +`stop` prevents further admissions, then closes known active children using the +existing host API. Completed child transcripts are retained. Cleanup errors +remain visible under each node's `cleanup_error`, and the run remains +`stopping` rather than claiming it stopped. Resolve the host issue and retry +`stop`. A running/stopping run cannot be forgotten. + +## Specification and execution semantics + +- Top-level fields: `version` (exact integer 1), `nodes`, `max_parallel` + (1–16, default 4), and `failure_policy` (`fail_fast` or `continue`). +- Nodes: unique `id` matching `[a-z][a-z0-9_-]{0,31}`, nonempty `prompt`, + optional `needs` list, optional exact `provider/model` selector, optional + `thinking`. Omitted model/thinking inherit the immediate parent through + `rlm.spawn`. The host still validates the resolved model and thinking level. +- Up to 128 nodes. Duplicate IDs/dependencies, missing predecessors, cycles, + unknown fields, and unsupported Prime machine options are rejected, never + silently ignored. Nodes may be declared in any order. +- A node starts only after **all** dependencies have completed successfully. + Predecessor text is appended as a JSON object keyed by predecessor ID and + labeled as task data, not additional authorization. No eval or template + execution occurs. A graph does not authorize its children to exceed the + user's original task. +- Each node is admitted at most once. There are no automatic retries. Admission + is serialized per run, including concurrent `advance`/`wait` callers. + `max_parallel` counts admitted children, including those waiting for input. + This does not increase the host's own global/session child limit. +- `fail_fast` stops admission and closes remaining live children after a node + fails. `continue` allows independent branches to finish; descendants of any + failed node become blocked. A run with any failed node ends `failed`. +- Child `failed`, `interrupted`, `canceled`, or `closed` never becomes success. + Missing, duplicate, or inconsistent collection replies raise without partly + applying observations or spawning more work. +- Outputs are limited to 32,000 characters per node and the rendered successor + prompt to 128,000 characters. Oversized content fails explicitly; it is not + silently truncated. There are at most 32 retained runs; explicitly forget a + terminal run to release its local state. +- `wait` timeout (0–300 seconds) is a polling budget, not a transaction deadline + or child cancellation. An in-flight spawn/collect/close may exceed that budget + so its acknowledgement is not lost. Interruption during spawn is treated as + uncertain admission; the unique child name is retained for cleanup, and the + node is never automatically respawned. If the host cannot resolve that name, + cleanup remains unresolved and the run stays `stopping`. +- Status is a detached local snapshot: `ready`, `running`, `stopping`, + `completed`, `failed`, or `stopped`. Host state may have changed since the last + advance; status is not a live provider-health or process-liveness promise. + +## Prime comparison and next safe milestones + +Reference: [Prime Factory skill at 74591cd](https://github.com/PrimeIntellect-ai/prime-agent/blob/74591cd/skills/factory/SKILL.md) +and [runtime implementation](https://github.com/PrimeIntellect-ai/prime-agent/blob/74591cd/prime-agent-runtime/src/rlm/factory.py). +Devo baseline inspected: [4228c9d](https://github.com/7df-lab/devo/commit/4228c9d). + +| Capability | Prime reference | This Devo slice | +| --- | --- | --- | +| Validated definitions | Harness `factory` entries; DAG compiles to state machine | Inline versioned DAG; no persistent harness kind | +| Fan-out / fan-in | State transitions and joins | Bounded fan-out; all-dependency success joins | +| Child ownership | Ordinary supervisor-owned children | Existing Rust-owned child sessions | +| Failure control | Retries, per-state policies, escalation | Fail-fast/continue; no admission retries | +| Complex transitions | Guards, bounded cycles, foreach, residents | Rejected as unsupported | +| Run lifetime | Background kernel driver; registry lost on kernel restart | Explicit active-cell advance/wait; volatile registry | +| User controls | Persistent gate, graph/watch, stop/resume, TUI factory page | Explicit API opt-in, local status, stop/cleanup | +| Machine library | MACHINE.md seeds and import/export | Not implemented | + +Next milestones should be independently reviewed: (1) integrate and exercise +the real collection API end to end; (2) design a host-owned durable run record +and bounded background scheduler with explicit permission/session lifetime; +(3) add Native-only run status/control events and UI; (4) add guarded bounded +loops, validated retry budgets, and machine-library storage. A background +Python task with a stale host bridge is not a substitute for milestone 2. + +## Verification + +```sh +PYTHONPATH=crates/kernel/rlm-runtime/src python -m unittest discover \ + -s crates/kernel/rlm-runtime/tests -p 'test_factory*.py' -v +``` + +The scheduler tests use an injected backend and cover validation, joins, +concurrency, failure propagation, malformed observations, uncertain admissions, +interruption, stop races/retries, timeouts, capacity, and output bounds. Adapter +tests verify the public helper calls. These are not a substitute for a real +provider-backed Devo session test, nor evidence of background continuation, +restart recovery, UI support, or complete Prime parity. From d2a2ca8d988c5c981ed691c8a7d56c3ac16a825c Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Sun, 8 Nov 2026 23:51:01 +0800 Subject: [PATCH 2/2] fix(factory): preserve waiting children and surface explicit resume --- .../rlm-runtime/src/rlm/_factory_backend.py | 4 +++- crates/kernel/rlm-runtime/src/rlm/factory.py | 22 ++++++++++++------ .../kernel/rlm-runtime/tests/test_factory.py | 23 ++++++++++++++++--- docs/factory-workflows.md | 7 +++++- 4 files changed, 44 insertions(+), 12 deletions(-) diff --git a/crates/kernel/rlm-runtime/src/rlm/_factory_backend.py b/crates/kernel/rlm-runtime/src/rlm/_factory_backend.py index e7627e38..d9b5b70a 100644 --- a/crates/kernel/rlm-runtime/src/rlm/_factory_backend.py +++ b/crates/kernel/rlm-runtime/src/rlm/_factory_backend.py @@ -15,6 +15,7 @@ class ChildSnapshot: settled: bool output: str = "" error: str | None = None + requires_resume: bool = False class FactoryBackend(Protocol): @@ -59,7 +60,8 @@ async def inspect(self, child_ids: list[str]) -> list[ChildSnapshot]: results = await rlm.collect(targets=child_ids, timeout_ms=0) return [ChildSnapshot(item.rlm_child_id, item.status, item.settled, - item.answer_text or "", item.error) for item in results] + item.answer_text or "", item.error, + getattr(item, "requires_resume", False)) for item in results] async def cancel(self, target: str) -> None: import rlm diff --git a/crates/kernel/rlm-runtime/src/rlm/factory.py b/crates/kernel/rlm-runtime/src/rlm/factory.py index 52690217..715609ab 100644 --- a/crates/kernel/rlm-runtime/src/rlm/factory.py +++ b/crates/kernel/rlm-runtime/src/rlm/factory.py @@ -42,6 +42,7 @@ This is not Prime machine compatibility: no loops/guards/foreach/residents, retries, durable specs/runs, restart resume, UI or background advancement. wait timeout leaves children running; call wait/advance again or explicitly stop. +waiting_for_resume returns from wait immediately; explicitly resume the child or stop. Interruption does not imply cancellation. Inspect status and call stop if needed. Host/transport failures are surfaced, never converted to success or respawned. When status is stopping, inspect cleanup_error and retry stop; do not forget it. @@ -116,7 +117,7 @@ async def advance(self, run_id: str) -> dict[str, Any]: return self.status(run_id) run.status = "running" active = {node.child_id: node for node in run.nodes.values() - if node.status in ("running", "waiting_for_input")} + if node.status in ("running", "waiting_for_input", "waiting_for_resume")} if active: # Do not partially apply malformed or incomplete replies. A read # failure leaves all child IDs retained and can be safely retried. @@ -125,16 +126,21 @@ async def advance(self, run_id: str) -> dict[str, Any]: if len(ids) != len(active) or set(ids) != set(active): raise RuntimeError("Factory collect returned missing or duplicate children") for item in snapshots: - if item.status not in {"running", "waiting_for_input", "completed", "failed", + if item.status not in {"running", "waiting_for_input", "waiting_for_resume", "completed", "failed", "interrupted", "canceled", "closed"}: raise RuntimeError(f"Factory collect returned unknown status {item.status!r}") - if type(item.settled) is not bool or item.settled != ( - item.status not in {"running", "waiting_for_input"} - ) or not isinstance(item.output, str): + if (type(item.settled) is not bool + or type(item.requires_resume) is not bool + or (item.requires_resume and item.status != "waiting_for_resume") + or (item.status in {"running", "waiting_for_resume"} and item.settled) + or (item.status not in {"running", "waiting_for_input", "waiting_for_resume"} and not item.settled) + or not isinstance(item.output, str)): raise RuntimeError("Factory collect returned inconsistent child state") for item in snapshots: node = active[item.child_id] - if not item.settled: + # The host may settle a turn while its child still waits + # for input. That is not workflow success or a free slot. + if not item.settled or item.status == "waiting_for_input": node.status = item.status continue node.needs_cleanup = False @@ -201,6 +207,8 @@ async def advance(self, run_id: str) -> dict[str, Any]: await self._cleanup(run) elif all(node.status in {"completed", "failed", "blocked"} for node in run.nodes.values()): run.status = "failed" if run.failed else "completed" + elif any(node.status == "waiting_for_resume" for node in run.nodes.values()): + run.status = "waiting_for_resume" return self.status(run_id) @staticmethod @@ -254,7 +262,7 @@ async def wait(self, run_id: str, *, timeout: float = 30, poll_interval: float = while True: result = await self.advance(run_id) remaining = deadline - asyncio.get_running_loop().time() - if result["status"] in _TERMINAL or result["status"] == "stopping" or remaining <= 0: + if result["status"] in _TERMINAL | {"stopping", "waiting_for_resume"} or remaining <= 0: return result await asyncio.sleep(min(poll_interval, remaining)) diff --git a/crates/kernel/rlm-runtime/tests/test_factory.py b/crates/kernel/rlm-runtime/tests/test_factory.py index 3f049bee..9beecf36 100644 --- a/crates/kernel/rlm-runtime/tests/test_factory.py +++ b/crates/kernel/rlm-runtime/tests/test_factory.py @@ -250,9 +250,10 @@ async def test_oversized_output_and_rendered_prompts_fail_without_truncation(sel async def test_waiting_for_input_holds_capacity_without_blocking_local_status(self): run_id = await self.start(dag("a", "b", max_parallel=1)) await self.factory.advance(run_id) - self.backend.children["child-0"] = ChildSnapshot("child-0", "waiting_for_input", False) - result = await self.factory.advance(run_id) - self.assertEqual((result["nodes"]["a"]["status"], len(self.backend.spawned)), ("waiting_for_input", 1)) + for settled in (False, True): + self.backend.children["child-0"] = ChildSnapshot("child-0", "waiting_for_input", settled) + result = await self.factory.advance(run_id) + self.assertEqual((result["nodes"]["a"]["status"], len(self.backend.spawned)), ("waiting_for_input", 1)) await self.factory.stop(run_id) self.assertEqual(self.backend.cancelled, ["child-0"]) @@ -265,6 +266,22 @@ async def test_retained_run_limit_requires_explicit_forget(self): self.factory.forget(run_id) await self.start(dag("b")) + async def test_restored_child_needing_resume_returns_from_wait_without_replaying(self): + run_id = await self.start(dag("a", "b", max_parallel=1)) + await self.factory.advance(run_id) + self.backend.children["child-0"] = ChildSnapshot( + "child-0", "waiting_for_resume", False, requires_resume=True) + # A very long polling budget must return on this actionable state. + result = await asyncio.wait_for(self.factory.wait(run_id, timeout=300), timeout=1) + self.assertEqual((result["status"], len(self.backend.spawned)), ("waiting_for_resume", 1)) + with self.assertRaises(ValueError): + self.factory.forget(run_id) + self.backend.settle("child-0") + result = await self.factory.advance(run_id) + self.assertEqual((result["status"], len(self.backend.spawned)), ("running", 2)) + await self.factory.stop(run_id) + self.assertEqual(self.backend.cancelled, ["child-1"]) + class FactoryAdapterTests(unittest.IsolatedAsyncioTestCase): async def test_missing_collect_fails_before_registering_or_spawning(self): diff --git a/docs/factory-workflows.md b/docs/factory-workflows.md index c43a97bf..a8ee503e 100644 --- a/docs/factory-workflows.md +++ b/docs/factory-workflows.md @@ -104,6 +104,11 @@ remain visible under each node's `cleanup_error`, and the run remains is serialized per run, including concurrent `advance`/`wait` callers. `max_parallel` counts admitted children, including those waiting for input. This does not increase the host's own global/session child limit. +- A host child requiring explicit resume is reported as `waiting_for_resume`. + `wait` returns that actionable state immediately instead of using its entire + polling budget. The child keeps its slot; no automatic replay occurs. Resume + the child through the existing session controls, then advance again, or stop + the workflow. A settled turn waiting for input also keeps its slot. - `fail_fast` stops admission and closes remaining live children after a node fails. `continue` allows independent branches to finish; descendants of any failed node become blocked. A run with any failed node ends `failed`. @@ -120,7 +125,7 @@ remain visible under each node's `cleanup_error`, and the run remains uncertain admission; the unique child name is retained for cleanup, and the node is never automatically respawned. If the host cannot resolve that name, cleanup remains unresolved and the run stays `stopping`. -- Status is a detached local snapshot: `ready`, `running`, `stopping`, +- Status is a detached local snapshot: `ready`, `running`, `waiting_for_resume`, `stopping`, `completed`, `failed`, or `stopped`. Host state may have changed since the last advance; status is not a live provider-health or process-liveness promise.