fix(workflows): preserve authored execution and artifact delivery

opencollab/adapters/_docker_snapshot.py

@@ -5,6 +5,7 @@
 import posixpath
 import shlex
 import shutil
+import stat
 import tempfile
 from contextlib import asynccontextmanager
 from pathlib import Path
@@ -14,8 +15,10 @@
 from opencollab.adapters._env_docker import DockerEnvironment
 from opencollab.adapters._env_local import LocalEnvironment
 from opencollab.adapters._env_vm import VMEnvironment
+from opencollab.adapters._env_worktree import WorktreeEnvironment
 from opencollab.adapters._git_patch_bytes import _complete, _patch_temporary
 from opencollab.adapters._team_candidate_environment import _TeamCandidateEnvironment
+from opencollab.adapters.safe_anchored_files import open_regular_read_at
 from opencollab.application.async_timeout import await_owned_operation
 
 
@@ -40,6 +43,38 @@ async def _copy_to_host(environment: Any, source: str, target: Path, *, director
         raise OSError("Docker source workspace copy failed")
 
 
+async def _copy_artifact_to_host(environment: Any, source: str, target: Path) -> None:
+    relative = (
+        posixpath.relpath(source, environment.workspace) if posixpath.isabs(source) else posixpath.normpath(source)
+    )
+    if relative == ".." or relative.startswith("../"):
+        raise PermissionError("artifact path escapes workspace")
+    source = posixpath.join(environment.workspace, relative)
+    delegate = environment._delegate if isinstance(environment, _TeamCandidateEnvironment) else environment
+    if isinstance(delegate, WorktreeEnvironment):
+        delegate._ensure_active()
+        if delegate._local_env is None:
+            await delegate.setup()
+        delegate = delegate._local_env
+    if isinstance(delegate, LocalEnvironment):
+        def copy() -> None:
+            with open_regular_read_at(
+                delegate._root_fd(), delegate.workspace, delegate._relative_path(source),
+            ) as handle, target.open("wb") as output:
+                shutil.copyfileobj(handle, output)
+
+        await delegate._run_file_operation(copy)
+        return
+    if isinstance(delegate, DockerEnvironment):
+        source = await delegate._contained_file_path(source)
+    if isinstance(delegate, (DockerEnvironment, VMEnvironment)):
+        await _copy_to_host(delegate, source, target)
+        if not stat.S_ISREG(target.lstat().st_mode):
+            raise OSError("artifact transfer requires a regular file")
+        return
+    target.write_text(await environment.read_file(source), encoding="utf-8")
+
+
 def _remove_marker(path: Path) -> None:
     marker = path / ".git"
     if marker.is_dir() and not marker.is_symlink():

opencollab/adapters/_env_worktree.py

@@ -679,12 +679,12 @@ async def _ensure_directory_copy_changes_exported(self) -> None:
                 "call get_diff() and deliver its artifact first"
             )
 
-    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()
         if self._local_env is None:
             await self.setup()
         assert self._local_env is not None
-        return await self._local_env.exec_cmd(cmd, timeout)
+        return await self._local_env.exec_cmd(cmd, timeout, input_bytes=input_bytes)
 
     async def read_file(self, path: str) -> str:
         self._ensure_active()

opencollab/adapters/safe_anchored_files.py

@@ -10,7 +10,8 @@
 import os
 import stat
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from typing import BinaryIO
 
 from opencollab.adapters._env_base import TextFileRange
@@ -116,15 +117,13 @@ def open_parent_beneath(
         os.close(current_fd)
         raise
 
-def read_regular_bytes_at(
+@contextmanager
+def open_regular_read_at(
     root_fd: int,
     root_path: str | os.PathLike[str],
     path: str | os.PathLike[str],
-    *,
-    max_bytes: int,
-) -> bytes:
-    """Read a bounded regular file beneath a pinned directory descriptor."""
-    limit = _require_limit(max_bytes)
+) -> Iterator[BinaryIO]:
+    """Open a regular file beneath a pinned directory without following symlinks."""
     parent_fd, name = open_parent_beneath(
         root_fd,
         root_path,
@@ -141,6 +140,28 @@ def read_regular_bytes_at(
         opened = os.fstat(fd)
         if not stat.S_ISREG(opened.st_mode):
             raise OSError(f"read target is not a regular file: {path}")
+        handle = os.fdopen(fd, "rb")
+        fd = -1
+        with handle:
+            yield handle
+    finally:
+        if fd >= 0:
+            os.close(fd)
+        os.close(parent_fd)
+
+
+def read_regular_bytes_at(
+    root_fd: int,
+    root_path: str | os.PathLike[str],
+    path: str | os.PathLike[str],
+    *,
+    max_bytes: int,
+) -> bytes:
+    """Read a bounded regular file beneath a pinned directory descriptor."""
+    limit = _require_limit(max_bytes)
+    with open_regular_read_at(root_fd, root_path, path) as handle:
+        fd = handle.fileno()
+        opened = os.fstat(fd)
         if opened.st_size > limit:
             raise ValueError(f"read target exceeds {limit}-byte limit: {path}")
         chunks: list[bytes] = []
@@ -155,10 +176,6 @@ def read_regular_bytes_at(
         if len(payload) > limit:
             raise ValueError(f"read target exceeds {limit}-byte limit: {path}")
         return payload
-    finally:
-        if fd >= 0:
-            os.close(fd)
-        os.close(parent_fd)
 
 def read_regular_text_range_at(
     root_fd: int,

opencollab/bootstrap/authored_workflow.py

@@ -16,6 +16,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, get_args
 
+from opencollab.adapters._docker_snapshot import _copy_artifact_to_host
+from opencollab.adapters._env_process import ProcessCleanupError
 from opencollab.adapters.session_environments import SessionEnvironmentPool
 from opencollab.adapters.tools._paths import checked_path
 from opencollab.adapters.tools.base import Tool
@@ -320,15 +322,16 @@ def _account_usage(self, child: _Child, tokens: Any) -> None:
             child.tokens = tokens
 
     async def _execute(self, child: _Child, command: str, request_bytes: bytes) -> None:
-        execution = asyncio.create_task(child.environment.exec_cmd(
-            command, timeout=math.inf, input_bytes=request_bytes,
-        ))
-        child.execution = execution
-        if child.stop_requested:
-            execution.cancel()
+        execution = None
         logs: dict[str, Any] = {}
         status = "running"
         try:
+            execution = asyncio.create_task(child.environment.exec_cmd(
+                command, timeout=math.inf, input_bytes=request_bytes,
+            ))
+            child.execution = execution
+            if child.stop_requested:
+                execution.cancel()
             while not execution.done():
                 await asyncio.wait({execution}, timeout=0.1)
                 assert child.bridge is not None
@@ -340,7 +343,12 @@ async def _execute(self, child: _Child, command: str, request_bytes: bytes) -> N
             logs = {"returncode": outcome.returncode, "stdout": outcome.stdout, "stderr": outcome.stderr}
             if outcome.returncode:
                 raise RuntimeError(f"Workflow runner exited {outcome.returncode}: {outcome.stderr[-2000:]}")
-            child.result = json.loads(await child.environment.read_file(posixpath.join(child.artifacts, "result.json")))
+            with tempfile.TemporaryDirectory(prefix="opencollab-child-result-") as temporary:
+                result_path = Path(temporary) / "result.json"
+                await _copy_artifact_to_host(
+                    child.environment, posixpath.join(child.artifacts, "result.json"), result_path,
+                )
+                child.result = json.loads(result_path.read_text(encoding="utf-8"))
             self._account_usage(child, child.result["tokens"])
             status = child.result["status"]
         except asyncio.CancelledError:
@@ -350,14 +358,16 @@ async def _execute(self, child: _Child, command: str, request_bytes: bytes) -> N
             status = "failed"
             child.result = {"status": "failed", "output": None, "reason": f"{type(exc).__name__}: {exc}"}
         finally:
-            if not execution.done():
-                execution.cancel()
-            settled = await asyncio.gather(execution, return_exceptions=True)
-            if isinstance(settled[0], Exception):
-                status = "failed"
-                child.cleanup_error = settled[0]
-                child.result = {"status": "failed", "output": None,
-                                "reason": f"child process cleanup failed: {settled[0]}"}
+            if execution is not None:
+                cancelling = not execution.done()
+                if cancelling:
+                    execution.cancel()
+                settled = await asyncio.gather(execution, return_exceptions=True)
+                if isinstance(settled[0], ProcessCleanupError) or (cancelling and isinstance(settled[0], Exception)):
+                    status = "failed"
+                    child.cleanup_error = settled[0]
+                    child.result = {"status": "failed", "output": None,
+                                    "reason": f"child process cleanup failed: {settled[0]}"}
             try:
                 assert child.bridge is not None
                 await await_owned_operation(child.bridge.aclose(), propagate_cancellation=False)
@@ -388,10 +398,11 @@ async def _execute(self, child: _Child, command: str, request_bytes: bytes) -> N
                         ):
                             raise ValueError("invalid child artifact name")
                         try:
-                            content = await child.environment.read_file(posixpath.join(child.artifacts, name))
+                            await _copy_artifact_to_host(
+                                child.environment, posixpath.join(child.artifacts, name), directory / name,
+                            )
                         except FileNotFoundError:
                             continue
-                        (directory / name).write_text(content, encoding="utf-8")
                     child.saved_artifacts = str(directory)
                     (directory / "result.json").write_text(json.dumps(self._snapshot(child, include_result=True)),
                                                            encoding="utf-8")

tests/interfaces/test_authored_workflow_execution.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from opencollab.adapters._env_process import ProcessCleanupError
+from opencollab.adapters._team_candidate_environment import _TeamCandidateEnvironment
+from opencollab.adapters.env import LocalEnvironment, WorktreeEnvironment
+from tests.interfaces.test_member_authored_workflow import _launch, _manager
+from tests.support.paths import REPO_ROOT
+
+
+@pytest.fixture(autouse=True)
+def _child_import_path(monkeypatch):
+    monkeypatch.setenv("PYTHONPATH", str(REPO_ROOT))
+    monkeypatch.setenv("OPENCOLLAB_API_USAGE_LOG", os.devnull)
+
+
+@pytest.mark.parametrize("kind", ["local", "copy-worktree", "git-worktree", "team-candidate"])
+@pytest.mark.parametrize("size", [1024, 4 * 1024 * 1024 + 1024])
+async def test_authored_result_and_native_artifacts_cross_file_read_limit(tmp_path, kind, size):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    (workspace / "authored.py").write_text(
+        "async def main(ctx, args):\n    return {'payload': args['payload']}\n"
+    )
+    (workspace / ".gitignore").write_text(".opencollab/\n")
+    if kind == "git-worktree":
+        for args in (("init", "-q"), ("add", "."), (
+            "-c", "user.name=Test", "-c", "user.email=test@example.invalid",
+            "commit", "-qm", "test: prepare workflow fixture",
+        )):
+            subprocess.run(["git", "-C", str(workspace), *args], check=True, capture_output=True)
+    environment = (LocalEnvironment if kind in {"local", "team-candidate"} else WorktreeEnvironment)(str(workspace))
+    if kind == "team-candidate":
+        environment = _TeamCandidateEnvironment(
+            SimpleNamespace(environment=environment, cleanup=environment.cleanup), str(workspace),
+        )
+    await environment.setup()
+    parent, tool, runtime = _manager(environment)
+    saved = tmp_path / "saved"
+    saved.mkdir()
+    tool._save_dir = str(saved)
+    payload = "x" * size
+    try:
+        child = await _launch(tool, runtime, args={"payload": payload})
+        result = json.loads(await tool.execute_with_runtime({"action": "wait", "id": child["id"]}, runtime))
+        native_path = Path(child["artifacts"]) / "result.json"
+        assert json.loads(native_path.read_text())["status"] == "completed"
+        assert result["status"] == "completed", result
+        assert result["result"]["output"] == {"payload": payload}
+        assert tool._reserved_tokens() == 0
+        persisted = Path(result["artifacts"])
+        assert json.loads((persisted / "result.json").read_text())["result"]["output"] == {"payload": payload}
+        assert json.loads((persisted / "workflow.json").read_text())["args"] == {"payload": payload}
+        if size > 4 * 1024 * 1024:
+            assert native_path.stat().st_size > 4 * 1024 * 1024
+            assert (persisted / "workflow.json").stat().st_size > 4 * 1024 * 1024
+            with pytest.raises(ValueError, match="read target exceeds"):
+                await environment.read_file(str(native_path))
+    finally:
+        try:
+            await parent.aclose()
+        finally:
+            if isinstance(environment, WorktreeEnvironment):
+                await environment.get_diff()
+            await environment.cleanup()
+
+
+@pytest.mark.parametrize("failure", ["signature", "async-start"])
+async def test_executor_start_failure_finishes_child_and_releases_budget(tmp_path, monkeypatch, failure):
+    (tmp_path / "authored.py").write_text("async def main(ctx, args):\n    return 1\n")
+    environment = LocalEnvironment(str(tmp_path))
+    parent, tool, runtime = _manager(environment)
+    native_exec = environment.exec_cmd
+
+    if failure == "signature":
+        def fail_to_start(command, timeout=120):
+            raise AssertionError("the input_bytes argument must fail during binding")
+    else:
+        async def fail_to_start(command, timeout=120, *, input_bytes=None):
+            raise OSError("executor startup failed")
+
+    monkeypatch.setattr(environment, "exec_cmd", fail_to_start)
+    try:
+        child = await _launch(tool, runtime)
+        assert tool._reserved_tokens() == 500
+        result = json.loads(await tool.execute_with_runtime({"action": "wait", "id": child["id"]}, runtime))
+        assert result["status"] == "failed"
+        assert ("TypeError" if failure == "signature" else "executor startup failed") in result["result"]["reason"]
+        assert tool._reserved_tokens() == 0
+        owned = tool._children[child["id"]]
+        assert owned.task.done() and owned.task.exception() is None
+        assert owned.bridge._pool._closed
+        assert owned.cleanup_error is None
+        assert not environment._processes._processes
+    finally:
+        monkeypatch.setattr(environment, "exec_cmd", native_exec)
+        try:
+            await parent.aclose()
+        finally:
+            await environment.cleanup()
+
+
+async def test_executor_cleanup_failure_retains_recovery_error(tmp_path, monkeypatch):
+    (tmp_path / "authored.py").write_text("async def main(ctx, args):\n    return 1\n")
+    environment = LocalEnvironment(str(tmp_path))
+    parent, tool, runtime = _manager(environment)
+    native_exec = environment.exec_cmd
+
+    async def cleanup_failed(command, timeout=120, *, input_bytes=None):
+        raise ProcessCleanupError("process group still alive")
+
+    monkeypatch.setattr(environment, "exec_cmd", cleanup_failed)
+    try:
+        child = await _launch(tool, runtime)
+        result = json.loads(await tool.execute_with_runtime({"action": "wait", "id": child["id"]}, runtime))
+        assert result["status"] == "failed"
+        assert "process group still alive" in result["result"]["reason"]
+        owned = tool._children[child["id"]]
+        assert isinstance(owned.cleanup_error, ProcessCleanupError)
+        assert tool._reserved_tokens() == 0
+        with pytest.raises(RuntimeError, match="process group still alive"):
+            await parent.aclose()
+        assert not owned.artifacts_removed
+    finally:
+        monkeypatch.setattr(environment, "exec_cmd", native_exec)
+        await environment.cleanup()

tests/interfaces/test_member_authored_workflow.py

@@ -266,22 +266,21 @@ async def test_stop_grace_leaves_finished_runner_result_persistence_owned(
     environment = local_environment_factory(str(tmp_path))
     parent, tool, runtime = _manager(environment)
     entered, finish = asyncio.Event(), asyncio.Event()
-    native_read = environment.read_file
+    native_copy = authored_workflow._copy_artifact_to_host
     read_cancelled = False
 
-    async def delayed_result(path):
+    async def delayed_result(environment, path, target):
         nonlocal read_cancelled
-        value = await native_read(path)
+        await native_copy(environment, path, target)
         if path.endswith("/result.json"):
             entered.set()
             try:
                 await finish.wait()
             except asyncio.CancelledError:
                 read_cancelled = True
                 raise
-        return value
 
-    monkeypatch.setattr(environment, "read_file", delayed_result)
+    monkeypatch.setattr(authored_workflow, "_copy_artifact_to_host", delayed_result)
     child = await _launch(tool, runtime)
     stopping = None
     try: