fix: preserve Weave checks and remote artifact recovery

opencollab/adapters/_docker_snapshot.py

@@ -14,7 +14,7 @@
 from opencollab.adapters._candidate_physical import _initialized_submodule_paths
 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_vm import VMEnvironment, VMTransportError
 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
@@ -68,13 +68,32 @@ def copy() -> None:
     if isinstance(delegate, DockerEnvironment):
         source = await delegate._contained_file_path(source)
     if isinstance(delegate, (DockerEnvironment, VMEnvironment)):
-        await _copy_to_host(delegate, source, target)
+        try:
+            await _copy_to_host(delegate, source, target)
+        except (OSError, VMTransportError) as exc:
+            if await _remote_path_absent(delegate, source):
+                target.unlink(missing_ok=True)
+                raise FileNotFoundError(source) from exc
+            raise
         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")
 
 
+async def _remote_path_absent(environment: Any, source: str) -> bool:
+    """A successful absence probe keeps transport failures visible."""
+    try:
+        result = await environment.exec_cmd(f"test ! -e {shlex.quote(source)}", timeout=30)
+    except Exception:
+        return False
+    return (
+        result.returncode == 0 and not result.timed_out
+        and not result.stdout_truncated and not result.stderr_truncated
+        and not result.stdout_dropped_bytes and not result.stderr_dropped_bytes
+    )
+
+
 def _remove_marker(path: Path) -> None:
     marker = path / ".git"
     if marker.is_dir() and not marker.is_symlink():

opencollab/adapters/tools/evidence/bash_evidence.py

@@ -17,7 +17,7 @@
     _verification_targets_overlap,
     has_pass_evidence,
 )
-from .command_evidence import _strip_command_padding, parse_test_command, relative_target
+from .command_evidence import _shell_comment_start, _strip_command_padding, parse_test_command, relative_target
 
 
 def _pytest_target(target: str, workspace: str | None, local_filesystem: bool) -> str:
@@ -108,14 +108,17 @@ def with_pass_reporting(self, command: str) -> str:
         spec = parse_test_command(command, None)
         if spec is None or not spec.provable or not _is_pytest_runner(spec.runner):
             return command
-        lexer = shlex.shlex(command, posix=True)
+        comment_start = _shell_comment_start(command)
+        lexer = shlex.shlex(command[:comment_start], posix=True)
         lexer.whitespace_split = True
         lexer.commenters = ""
         runner_seen = False
         while True:
             start = lexer.instream.tell()
             token = lexer.get_token()
             if token is None:
+                if comment_start < len(command):
+                    return command[:comment_start] + "-rA " + command[comment_start:]
                 return command + " -rA"
             if runner_seen and token == "--":
                 return command[:start] + "-rA " + command[start:]

opencollab/adapters/tools/evidence/command_evidence.py

@@ -51,14 +51,29 @@ def _strip_command_padding(command: str) -> str:
         end += 1
     return command[:end].lstrip()
 
+def _shell_comment_start(command: str) -> int:
+    lexer = shlex.shlex(command, posix=True)
+    lexer.whitespace_split = True
+    lexer.commenters = ""
+    while True:
+        start = lexer.instream.tell()
+        remaining = command[start:].lstrip()
+        if remaining.startswith("#"):
+            return len(command) - len(remaining)
+        if lexer.get_token() is None:
+            return len(command)
+
 def parse_test_command(command: str, workspace: str | None) -> TestCommand | None:
     """Recognize direct pytest, Go and Django invocations, never shell output replay."""
     if not isinstance(command, str):
         return None
     command = _strip_command_padding(command)
-    if any(char in command for char in "\n\r;|<>`"):
+    if any(char in command for char in "\n\r"):
         return None
     try:
+        command = command[:_shell_comment_start(command)]
+        if any(char in command for char in ";|<>`"):
+            return None
         parts = shlex.split(command)
     except ValueError:
         return None

opencollab/builtin_workflows/_weave_execution.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import math
+import re
 import shlex
 import time
 from collections.abc import Callable, Mapping, Sequence
@@ -117,6 +118,8 @@ def _command_segments(command: str) -> list[str]:
             if command[start:index].strip():
                 segments.append(command[start:index])
             start = index + 2
+        elif quote is None and command[index:index + 3] == "<<<":
+            return []
         elif quote is None and char in ";|&":
             return []
         elif quote is None and char in "\n\r" and command[:index].strip() and command[index:].strip():
@@ -125,22 +128,165 @@ def _command_segments(command: str) -> list[str]:
         segments.append(command[start:])
     return segments
 
-def _command_words(tokens: list[str]) -> list[str]:
-    for index, token in enumerate(tokens):
+_WRAPPERS = frozenset({
+    "exec", "eval", "nice", "nohup", "time", "timeout", "setsid", "sudo",
+    "ionice", "chrt", "taskset", "flock", "xargs", "stdbuf", "busybox",
+    "unbuffer", "script",
+})
+_SHELLS = frozenset({"bash", "sh", "dash", "zsh", "ksh", "ash"})
+_STATIC_COMMANDS = frozenset({"true", "false", ":", "echo", "printf", "test", "["})
+_TEXT_TOOLS = frozenset({"awk", "gawk", "nawk", "mawk", "sed", "jq"})
+_PY_INLINE = re.compile(r"-[bBdEihIOPqRsSuvVx?]*c")
+_PERL_INLINE = re.compile(r"-[nplad]*[eE]")
+_RUBY_INLINE = re.compile(r"-[npl]*e")
+
+
+def _command_basename(token: str) -> str:
+    return token.rsplit("/", 1)[-1]
+
+
+def _effective_tokens(tokens: list[str]) -> list[str] | None:
+    """Return the program argv, or None when a wrapper can hide a printer.
+
+    Assignments and ``env`` / ``command`` / ``builtin`` are skipped, as before.
+    Their options stay in place, so ``env -i python`` remains unverified.
+    Shells accept a literal script path with execution options, while inline
+    commands and wrappers remain unverified.
+    """
+    index = 0
+    while index < len(tokens):
+        token = tokens[index]
         variable, separator, _ = token.partition("=")
         if separator and variable.isidentifier():
+            index += 1
             continue
-        name = token.rsplit("/", 1)[-1]
+        name = _command_basename(token)
         if name in {"env", "command", "builtin", "--"}:
+            index += 1
             continue
+        if name in _WRAPPERS:
+            return None
+        if name in _SHELLS and not _shell_file_script(tokens[index + 1:]):
+            return None
         return tokens[index:]
     return []
 
+
+def _shell_file_script(arguments: list[str]) -> bool:
+    index = 0
+    while index < len(arguments):
+        option = arguments[index]
+        if option == "--":
+            index += 1
+            break
+        if not option.startswith("-"):
+            break
+        if re.fullmatch(r"-[eux]*o", option):
+            index += 1
+            if index == len(arguments) or arguments[index] not in {"errexit", "nounset", "xtrace", "pipefail"}:
+                return False
+        elif not re.fullmatch(r"-[eux]+", option):
+            return False
+        index += 1
+    return index < len(arguments) and bool(arguments[index]) and not arguments[index].startswith("-") and not any(
+        char in arguments[index] for char in "$<>"
+    )
+
+
+def _command_words(tokens: list[str]) -> list[str]:
+    words = _effective_tokens(tokens)
+    return [] if words is None else words
+
 def _has_behavior_command(tokens: list[str]) -> bool:
-    words = _command_words(tokens)
-    return bool(words) and not words[0].startswith("-") and words[0].rsplit("/", 1)[-1] not in {
-        "true", ":", "echo", "printf",
-    }
+    words = _effective_tokens(tokens)
+    if not words or words[0].startswith("-"):
+        return False
+    name = _command_basename(words[0])
+    return name not in _STATIC_COMMANDS and name not in _TEXT_TOOLS
+
+
+def _static_printer_supplies(command: str, expected: str) -> bool:
+    """True when a printer or one-liner already contains the expected text.
+
+    ``echo CHECK_OK && python -c 'exit(0)'`` would otherwise match CHECK_OK
+    from the printer while the second command only has to exit zero.
+    """
+    if not expected:
+        return False
+    for segment in _command_segments(command):
+        try:
+            tokens = shlex.split(segment)
+        except ValueError:
+            return True
+        words = _effective_tokens(tokens)
+        printer = not words or _command_basename(words[0]) in _STATIC_COMMANDS | _TEXT_TOOLS
+        if printer and any(expected in token for token in tokens):
+            return True
+        if _contains_inline_script(tokens):
+            # Clustered flags (``python -ic``), ``node -p``, ``perl -E``, and a
+            # runner such as ``uv run python -c`` can still print arbitrary text.
+            # ``python -m pytest -c`` stays a real test command.
+            return True
+    return False
+
+
+def _contains_inline_script(tokens: list[str]) -> bool:
+    words = _effective_tokens(tokens)
+    if words is None:
+        return True
+    if words and _command_basename(words[0]) == "uv" and words[1:2] == ["run"]:
+        words = _effective_tokens(words[2:])
+        if words is None or words and words[0].startswith("-"):
+            return True
+    if not words:
+        return False
+    name, rest = _command_basename(words[0]), words[1:]
+    if name.startswith("python"):
+        return _python_inline(rest)
+    if name in {"node", "nodejs"}:
+        return _node_inline(rest)
+    if name == "perl":
+        return any(_PERL_INLINE.match(item) for item in _interpreter_options(rest, {"-I", "-M", "-m"}))
+    if name == "ruby":
+        return any(_RUBY_INLINE.match(item) for item in _interpreter_options(rest, {"-I", "-r", "-C", "-E", "-F"}))
+    if name.startswith("php"):
+        return any(item.startswith("-r") and not item.startswith("--")
+                   for item in _interpreter_options(rest, {"-c", "-d"}))
+    if name == "lua":
+        return any(item.startswith("-e") and not item.startswith("--")
+                   for item in _interpreter_options(rest, {"-l"}))
+    return False
+
+
+def _interpreter_options(rest: list[str], value_options: set[str]) -> list[str]:
+    options = []
+    arguments = iter(rest)
+    for token in arguments:
+        if token == "--" or not token.startswith("-"):
+            break
+        options.append(token)
+        if token in value_options:
+            next(arguments, None)
+    return options
+
+
+def _node_inline(rest: list[str]) -> bool:
+    options = _interpreter_options(rest, {
+        "-r", "--require", "--import", "--loader", "--experimental-loader",
+        "-C", "--conditions", "--input-type",
+    })
+    return any(token.startswith(("--eval", "--print")) or (
+        not token.startswith("--") and token.startswith(("-e", "-p"))
+    ) for token in options)
+
+
+def _python_inline(rest: list[str]) -> bool:
+    for token in _interpreter_options(rest, {"-W", "-X", "--check-hash-based-pycs"}):
+        if token == "-m" or (token.startswith("-m") and not token.startswith("--")):
+            return False
+        if _PY_INLINE.match(token):
+            return True
+    return False
 
 def _verification_mode(command: str) -> bool:
     try:
@@ -193,6 +339,9 @@ async def _verify_commands(ctx: Any, commands: Sequence[Any], seconds: float | N
                 row["expected_output"] = expected
                 row["matched"] = expected in output
                 row["ok"] = (row["ok"] if records else not requires_test_evidence) and completed and row["matched"]
+                if row["ok"] and not records and _static_printer_supplies(command, expected):
+                    row["ok"] = False
+                    row["reason"] = "expected output is supplied by a static printer"
         checks.append(row)
     return {
         "ok": bool(checks) and all(row["ok"] for row in checks),

tests/agents/test_session_construction.py

@@ -427,6 +427,19 @@ def test_session_snapshot_returns_independent_session_without_autosave(local_env
     snap.messages.append({"role": "user", "content": "isolated"})
     assert {"role": "user", "content": "isolated"} not in session.messages
 
+def test_session_snapshot_preserves_enforcement_after_the_soft_cap_shrinks(local_environment_factory):
+    session = _new_session(local_environment_factory, env=_ForkableEnvironment(), max_budget_tokens=100_000)
+    session.runner.configure_enforcement(enforcement_strength="needs-enforcement", commit_reserve=25_000)
+    session.max_budget_tokens = 1_000
+
+    snap = snapshot_session(session)
+
+    assert snap.max_budget_tokens == 1_000
+    assert snap.runner.hard_budget_tokens == 100_000
+    assert snap.runner._enforcement_strength == "needs-enforcement"
+    assert snap.runner._commit_reserve == 25_000
+
+
 def test_session_snapshot_preserves_queued_external_turn(local_environment_factory):
     session = _new_session(local_environment_factory, env=_ForkableEnvironment())
     run(session.add_user_message("resume this request"))