feat(runtime): manage member-authored child runs

examples/member-authored-workflow/README.md

@@ -0,0 +1,107 @@
+# Member-authored Workflows
+
+A Team member can write Python Workflow source with its normal editing tools,
+then execute that file through `manage_workflow`. The source is imported by a
+Python process inside that member's execution environment. The entry function
+receives the usual `ctx` and `args`, so it can use arbitrary Python control flow
+and the existing Workflow operations.
+
+Run the offline example from an installed checkout. It creates a local HTTP
+model, has the parent member write `generated.py`, starts that file, and returns
+the child result to the parent. Every model request stays on the local machine.
+
+```bash
+uv run python examples/member-authored-workflow/demo.py
+```
+
+The result includes `Local answer`, a computed total of `15`, and `90` tokens
+reported by the scripted model across the parent and child. The printed
+workspace retains the generated source, parent transcripts, child Workflow
+manifest, usage, and result.
+
+Declare the tool on the role that needs it, together with `bash` and the tools
+its Workflow agents will use.
+
+```yaml
+entry: author
+roles:
+  author:
+    prompt: Write and run Workflows to complete the task.
+    tools: [file_write, file_read, bash, manage_workflow]
+```
+
+`manage_workflow` shares the role's shell execution permission. Local host
+execution requires the Team caller to enable `allow_unisolated_shell`.
+The target environment must provide Python with OpenCollab installed. The
+`python` argument selects that environment's executable and defaults to
+`python3`. A child session's requested tool names must be available to its
+parent member.
+Its output caps, shell isolation requirement, and file-creation restriction
+also carry into child sessions.
+
+The five actions use the same tool. `launch` returns immediately with a child
+identifier belonging to the calling member. `status` observes progress, `wait`
+returns a result when execution finishes, `result` reads the retained outcome,
+and `stop` waits for child cleanup. An optional `timeout` on `wait` bounds that
+wait and leaves the child running.
+
+```json
+{"action": "launch", "source": "generated.py", "entry": "main", "args": {"count": 6}}
+{"action": "status", "id": "<child id>"}
+{"action": "wait", "id": "<child id>", "timeout": 30}
+{"action": "result", "id": "<child id>"}
+{"action": "stop", "id": "<child id>"}
+```
+
+Model credentials and supported generation settings inherit the member's
+effective configuration through the child process's standard input. Child usage is charged once to the parent's token
+total. Finite child budgets reserve the member's remaining budget across
+simultaneously active children and the parent's subsequent model requests.
+A child with an omitted budget receives half the unreserved remainder so the
+parent can coordinate its runs. Unbounded parents can launch unbounded
+children. An explicit child `budget` limits that child. The parent's step limit
+supplies the default for child sessions. A Workflow may select a per-call step
+limit with `ctx.agent_run(max_steps=...)`.
+
+A launch lasts until its Workflow finishes or its parent stops. Ordinary
+parent completion stops any remaining children before saving the final parent
+snapshot. Active child control files live in
+`.opencollab/child-runs/<id>` inside the environment. When the parent run has an
+artifacts directory, it receives each child's native Workflow manifest,
+Session transcripts, result, and command output. The finished child's
+`artifacts` field names that persistent host directory. Parent shutdown removes
+child control files before the Team captures the member's patch.
+Source errors, execution errors, and output serialization errors appear in the
+result. Workflow agent failures retain the existing `agent_failures` field,
+including cases where the Workflow chooses to return after a failed agent.
+
+Cancellation first asks the child Workflow runtime to stop its sessions and
+save its state. If it fails to settle, the owning Environment cancels its
+command and performs its existing process cleanup. A `wait` call uses the
+managed lifecycle, so an ordinary `bash` timeout does not determine how long
+the Workflow can run.
+
+An authored Workflow inherits its member's isolation strategy. Each
+`ctx.agent` or `ctx.agent_run` call can select `shared`, `worktree`,
+`docker-group`, `docker-session`, or a host-registered strategy through its
+`isolation` argument. See the [session isolation guide](../../docs/session-isolation.md)
+for provider configuration. For example, these calls create two independent
+Docker execution environments when the host configures the `docker-session`
+provider.
+
+```python
+from opencollab.tools import builtin_tools
+
+async def main(ctx, args):
+    tools = builtin_tools("bash")
+    first = await ctx.agent("inspect the source", tools=tools, isolation="docker-session")
+    second = await ctx.agent("verify the source", tools=tools, isolation="docker-session")
+    return first, second
+```
+
+The host owns the configured provider factories and each authored run's child
+environments. The member's Python process sends environment operations and
+model requests through its private run directory. The host calls the member's
+resolved model backend and returns its native response and usage. This also
+supports a member container with Docker's default `none` network. Parent
+cancellation waits for active child operations and releases their environments.

examples/member-authored-workflow/demo.py

@@ -0,0 +1,103 @@
+"""Run a generated Workflow against an offline HTTP model."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+import tempfile
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+from opencollab import OpenCollab
+
+SOURCE = '''from opencollab.workflows import workflow
+
+@workflow
+async def main(ctx, args):
+    answer = await ctx.agent(args["question"], tools=[])
+    return {"answer": answer, "total": sum(range(args["count"]))}
+'''
+
+
+async def run(workspace: Path) -> dict:
+    observed = {}
+    parent_step = 0
+
+    class Model(BaseHTTPRequestHandler):
+        def log_message(self, *args):
+            pass
+
+        def do_POST(self):
+            nonlocal parent_step
+            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+            messages = body["messages"]
+            is_parent = messages[0].get("content") == "Offline workflow author"
+            tool_call = None
+            if is_parent:
+                parent_step += 1
+                if parent_step == 1:
+                    name, args = "file_write", {"path": "generated.py", "mode": "create", "content": SOURCE}
+                elif parent_step == 2:
+                    name, args = "manage_workflow", {"action": "launch", "source": "generated.py",
+                                                    "python": sys.executable,
+                                                    "args": {"question": "Give a local answer", "count": 6}}
+                elif parent_step == 3:
+                    launched = json.loads([m for m in messages if m["role"] == "tool"][-1]["content"])
+                    name, args = "manage_workflow", {"action": "wait", "id": launched["id"]}
+                else:
+                    observed.update(json.loads([m for m in messages if m["role"] == "tool"][-1]["content"]))
+                if parent_step <= 3:
+                    tool_call = {"id": f"call-{parent_step}", "type": "function",
+                                 "function": {"name": name, "arguments": json.dumps(args)}}
+            message = {"role": "assistant", "content": "Collected child result" if is_parent else "Local answer"}
+            if tool_call is not None:
+                message.update(content=None, tool_calls=[tool_call])
+            payload = json.dumps({
+                "id": "offline", "object": "chat.completion", "created": 0, "model": "gpt-4o",
+                "choices": [{"index": 0, "message": message,
+                             "finish_reason": "tool_calls" if tool_call else "stop"}],
+                "usage": {"prompt_tokens": 11, "completion_tokens": 7, "total_tokens": 18},
+            }).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            self.wfile.write(payload)
+
+    workspace.mkdir(parents=True, exist_ok=True)
+    team = workspace / "team.yaml"
+    team.write_text("entry: author\nroles:\n  author:\n    prompt: Offline workflow author\n"
+                    "    tools: [file_write, bash, manage_workflow]\n", encoding="utf-8")
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Model)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        result = await OpenCollab(
+            workspace, model="gpt-4o", provider="openai",
+            api_key="offline-model",  # pragma: allowlist secret
+            base_url=f"http://127.0.0.1:{server.server_port}/v1",
+            config={"thinking": False, "budget": 10_000},
+        ).team(
+            "Write and execute your own Workflow", config=team, trace=False,
+            artifacts=workspace / "parent-run", use_worktrees=False, allow_unisolated_shell=True,
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join()
+    result.raise_for_status()
+    if observed.get("status") != "completed":
+        raise RuntimeError(f"child failed: {observed}")
+    return {"parent_output": result.output, "total_tokens": result.tokens, "child": observed,
+            "workspace": str(workspace)}
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--workspace", type=Path, default=None)
+    options = parser.parse_args()
+    directory = options.workspace or Path(tempfile.mkdtemp(prefix="opencollab-authored-"))
+    print(json.dumps(asyncio.run(run(directory.resolve())), indent=2))

opencollab/README.md

@@ -282,6 +282,11 @@ another async workflow with the same context or use `ctx.candidate_workflow`
 for an isolated candidate. Team spawning continues to create members within the
 same Team. Workflow sessions receive the tools selected by their Python workflow.
 
+Members that write new Workflow source during a run can use `manage_workflow`
+to launch the file and later inspect, wait for, or stop its child run.
+The [member-authored Workflow example](../examples/member-authored-workflow/README.md)
+documents its permissions, budget sharing, lifecycle, and offline demonstration.
+
 ### Workflow authoring
 
 A workflow is a plain async Python function tagged with `@workflow`. Create

opencollab/adapters/_env_base.py

@@ -58,7 +58,7 @@ async def setup(self, mount_dir: str | None = None) -> str:
             raise ValueError("mount_dir is supported only by container environments")
         return self.workspace
 
-    async def exec_cmd(self, cmd: str, timeout: float = 120.0) -> ExecResult:
+    async def exec_cmd(self, cmd: str, timeout: float = 120.0, *, input_bytes: bytes | None = None) -> ExecResult:
         raise NotImplementedError
 
     async def read_file(self, path: str) -> str:

opencollab/adapters/_env_docker.py

@@ -548,8 +548,8 @@ async def _exec(
                     )
         return decoded
 
-    async def exec_cmd(self, cmd: str, timeout: float = 120.0) -> ExecResult:
-        return await self._exec(cmd, timeout=timeout)
+    async def exec_cmd(self, cmd: str, timeout: float = 120.0, *, input_bytes: bytes | None = None) -> ExecResult:
+        return await self._exec(cmd, timeout=timeout, input_bytes=input_bytes)
 
     async def read_file(self, path: str) -> str:
         result = await self._exec(

opencollab/adapters/_env_local.py

@@ -94,7 +94,7 @@ def _close_workspace(self) -> None:
             os.close(self._workspace_fd)
             self._workspace_fd = None
 
-    async def exec_cmd(self, cmd: str, timeout: float = 120.0) -> ExecResult:
+    async def exec_cmd(self, cmd: str, timeout: float = 120.0, *, input_bytes: bytes | None = None) -> ExecResult:
         self._ensure_active()
         try:
             result = await run_process(
@@ -104,6 +104,7 @@ async def exec_cmd(self, cmd: str, timeout: float = 120.0) -> ExecResult:
                 timeout=timeout,
                 registry=self._processes,
                 output_limit=PROCESS_OUTPUT_CAPTURE_BYTES,
+                input_bytes=input_bytes,
             )
         except asyncio.TimeoutError as exc:
             return timed_out_result(exc, -1, timeout)

opencollab/adapters/session_environments.py

@@ -48,6 +48,20 @@ def __init__(
         self._source = base_environment or LocalEnvironment(workspace)
         self._owned_source = self._source if base_environment is None else None
 
+    @property
+    def default_isolation(self) -> str:
+        return self._default
+
+    @property
+    def strategy_names(self) -> tuple[str, ...]:
+        return tuple(dict.fromkeys(("shared", "worktree", *self._providers)))
+
+    def child_pool(self, environment: EnvironmentPort, *, role: str) -> SessionEnvironmentPool:
+        return SessionEnvironmentPool(
+            self._workspace, isolation=self._roles.get(role) or self._default,
+            base_environment=environment, providers=self._providers,
+        )
+
     def _validate(self, name: str) -> None:
         isolation_name(name)
         if name not in {"shared", "worktree"} and name not in self._providers:

opencollab/adapters/tools/_paths.py

@@ -9,6 +9,9 @@
 
 from __future__ import annotations
 
+import os
+import posixpath
+
 from opencollab.application.tool_execution import ToolRuntime
 
 
@@ -20,4 +23,12 @@ def checked_path(runtime: ToolRuntime, path: str) -> str:
     workspace — callers that prefer a soft error catch it and return a string.
     """
     policy = runtime.safety_policy
-    return policy.check_path(path) if policy else path
+    checked = policy.check_path(path) if policy else path
+    environment = runtime.environment
+    workspace = getattr(environment, "workspace", None)
+    if policy and workspace is not None and not getattr(environment, "local_filesystem", False):
+        # Keep a remote workspace in its own namespace after host policy validation.
+        host_root = os.path.realpath(workspace)
+        if os.path.isabs(checked) and os.path.commonpath((host_root, checked)) == host_root:
+            checked = posixpath.join(workspace, os.path.relpath(checked, host_root))
+    return checked

opencollab/application/_session_run_completion.py

@@ -693,13 +693,8 @@ def _request_output_limit(
                 self._run_control is not None and self.hard_budget_tokens is not None
             ),
         )
-        ceilings = [self.max_budget_tokens]
-        if self._run_control is not None:
-            ceilings.append(self.hard_budget_tokens)
-        for ceiling in ceilings:
-            if ceiling is None:
-                continue
-            remaining_budget = int(ceiling) - int(self.state.used_tokens)
+        remaining_budget = self.remaining_budget_tokens
+        if remaining_budget is not None:
             output_budget = remaining_budget - reserved_input_tokens
             if output_budget < minimum_output_tokens:
                 raise _TokenBudgetStop(
@@ -709,6 +704,16 @@ def _request_output_limit(
             max_output_tokens = min(max_output_tokens, output_budget)
         return max_output_tokens
 
+    @property
+    def remaining_budget_tokens(self) -> int | None:
+        ceilings = [self.max_budget_tokens]
+        if self._run_control is not None:
+            ceilings.append(self.hard_budget_tokens)
+        finite = [int(ceiling) for ceiling in ceilings if ceiling is not None]
+        if not finite:
+            return None
+        return min(finite) - int(self.state.used_tokens) - sum(reserve() for reserve in self.budget_reservations)
+
     def _completion_accepts_output_limit(self) -> bool:
         """Include the default limit when the injected completion accepts it."""
         try:

opencollab/application/ports.py

@@ -27,7 +27,7 @@ async def setup(self, mount_dir: str | None = None) -> str:
         """Prepare the environment and return its ready workspace identity."""
         ...
 
-    async def exec_cmd(self, cmd: str, timeout: float = 120.0) -> Any:
+    async def exec_cmd(self, cmd: str, timeout: float = 120.0, *, input_bytes: bytes | None = None) -> Any:
         ...
 
     async def read_file(self, path: str) -> str: