diff --git a/opencollab/adapters/_docker_snapshot.py b/opencollab/adapters/_docker_snapshot.py
index f233dc15..aaa10778 100644
--- a/opencollab/adapters/_docker_snapshot.py
+++ b/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():
diff --git a/opencollab/adapters/tools/evidence/bash_evidence.py b/opencollab/adapters/tools/evidence/bash_evidence.py
index 8c62d2de..84e35711 100644
--- a/opencollab/adapters/tools/evidence/bash_evidence.py
+++ b/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,7 +108,8 @@ 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
@@ -116,6 +117,8 @@ def with_pass_reporting(self, command: str) -> str:
             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:]
diff --git a/opencollab/adapters/tools/evidence/command_evidence.py b/opencollab/adapters/tools/evidence/command_evidence.py
index 46c24c79..704a14a1 100644
--- a/opencollab/adapters/tools/evidence/command_evidence.py
+++ b/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
diff --git a/opencollab/builtin_workflows/_weave_execution.py b/opencollab/builtin_workflows/_weave_execution.py
index 941e0cf5..8ab6e3ab 100644
--- a/opencollab/builtin_workflows/_weave_execution.py
+++ b/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),
diff --git a/tests/agents/test_session_construction.py b/tests/agents/test_session_construction.py
index 1da86eb9..87732561 100644
--- a/tests/agents/test_session_construction.py
+++ b/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"))
diff --git a/tests/runtime/test_authored_artifact_transfer.py b/tests/runtime/test_authored_artifact_transfer.py
index 2fa7d6c6..35846d06 100644
--- a/tests/runtime/test_authored_artifact_transfer.py
+++ b/tests/runtime/test_authored_artifact_transfer.py
@@ -1,19 +1,23 @@
 from __future__ import annotations
 
 import asyncio
+import json
 import os
 import shutil
 from pathlib import Path
+from types import SimpleNamespace
 from unittest.mock import AsyncMock
 
 import pytest
 
 from opencollab.adapters import _docker_snapshot as snapshot_module
 from opencollab.adapters._docker_snapshot import _copy_artifact_to_host
-from opencollab.adapters._env_vm import VMEnvironment
+from opencollab.adapters._env_process import ProcessResult
+from opencollab.adapters._env_vm import VMEnvironment, VMTransportError
 from opencollab.adapters.env import DockerEnvironment, ExecResult, LocalEnvironment, WorktreeEnvironment
 from opencollab.adapters.session_environments import SessionEnvironmentPool
 from opencollab.application.tool_execution import ToolRuntime
+from opencollab.bootstrap.authored_workflow import AuthoredWorkflowTool, _Child
 from opencollab.bootstrap.authored_workflow_bridge import AuthoredWorkflowBridge, BridgeClient, BridgeEnvironment
 
 
@@ -118,6 +122,117 @@ async def download(source, destination, *, directory=False):
         await _copy_artifact_to_host(environment, "/outside/result.json", target)
 
 
+@pytest.mark.parametrize("kind", ["docker", "vm"])
+async def test_missing_remote_artifact_stays_distinguishable_from_transport_failure(
+    tmp_path, monkeypatch, kind,
+):
+    environment = object.__new__(DockerEnvironment if kind == "docker" else VMEnvironment)
+    environment.workspace = "/workspace"
+    target = tmp_path / "copied.json"
+
+    async def absent(command, timeout=120.0, **kwargs):
+        assert command == "test ! -e /workspace/result.json"
+        return ExecResult(0, "", "")
+
+    monkeypatch.setattr(environment, "exec_cmd", absent)
+    if kind == "docker":
+        environment._container_id = "owned-test-container"
+        monkeypatch.setattr(environment, "setup", AsyncMock())
+        monkeypatch.setattr(environment, "_contained_file_path", AsyncMock(return_value="/workspace/result.json"))
+
+        async def docker(*args):
+            return ExecResult(1, "", "No such file")
+
+        monkeypatch.setattr(environment, "_docker", docker)
+    else:
+        async def download(source, destination, *, directory=False):
+            raise VMTransportError("missing", operation_id="download")
+
+        monkeypatch.setattr(environment, "copy_to_host", download)
+    with pytest.raises(FileNotFoundError):
+        await _copy_artifact_to_host(environment, "/workspace/result.json", target)
+    assert not target.exists()
+
+    async def present(command, timeout=120.0, **kwargs):
+        return ExecResult(1, "", "")
+
+    monkeypatch.setattr(environment, "exec_cmd", present)
+    with pytest.raises((OSError, VMTransportError), match="copy failed|missing"):
+        await _copy_artifact_to_host(environment, "/workspace/result.json", target)
+
+
+@pytest.mark.parametrize("probe", [
+    ExecResult(1, "", "Cannot connect to the Docker daemon"),
+    ExecResult(0, "", "", timed_out=True),
+    ExecResult(0, "", "", stdout_truncated=True),
+    ExecResult(0, "", "", stderr_truncated=True),
+    ExecResult(0, "", "", stdout_dropped_bytes=1),
+    ExecResult(0, "", "", stderr_dropped_bytes=1),
+])
+async def test_failed_remote_absence_probe_preserves_transfer_error(tmp_path, monkeypatch, probe):
+    environment = object.__new__(VMEnvironment)
+    environment.workspace = "/workspace"
+    monkeypatch.setattr(environment, "exec_cmd", AsyncMock(return_value=probe))
+    error = VMTransportError("download failed", operation_id="download")
+    monkeypatch.setattr(environment, "copy_to_host", AsyncMock(side_effect=error))
+
+    with pytest.raises(VMTransportError) as caught:
+        await _copy_artifact_to_host(environment, "/workspace/result.json", tmp_path / "copied.json")
+
+    assert caught.value is error
+
+
+async def test_authored_docker_transport_failure_retains_native_artifacts(tmp_path, monkeypatch):
+    native = tmp_path / "workspace" / "child"
+    native.mkdir(parents=True)
+    saved = tmp_path / "saved"
+    saved.mkdir()
+    (native / "workflow.json").write_text('{"evidence": "retain this workflow"}')
+    (native / "result.json").write_text(json.dumps({
+        "status": "completed", "output": 42, "tokens": 0, "artifact_files": ["workflow.json"],
+    }))
+    environment = DockerEnvironment(workspace=str(native.parent))
+    environment._container_id = "owned-test-container"
+    monkeypatch.setattr(environment, "setup", AsyncMock(return_value=environment._container_id))
+    monkeypatch.setattr(environment, "_contained_file_path", AsyncMock(side_effect=lambda path: path))
+    monkeypatch.setattr(environment, "read_file", AsyncMock(side_effect=FileNotFoundError))
+    commands = []
+
+    async def docker(*args, **kwargs):
+        if args[0] == "cp":
+            source = args[2].split(":", 1)[1]
+            if source.endswith("/workflow.json"):
+                return ProcessResult(1, b"", b"Cannot connect to the Docker daemon")
+            shutil.copyfile(source, args[3])
+            return ProcessResult(0, b"", b"")
+        assert args[0] == "exec"
+        command = args[-1]
+        commands.append(command)
+        if command.startswith("test "):
+            return ProcessResult(1, b"", b"Cannot connect to the Docker daemon")
+        if command.startswith("rm -rf -- "):
+            shutil.rmtree(native)
+        return ProcessResult(0, b"", b"")
+
+    monkeypatch.setattr(environment, "_docker", docker)
+    parent = SimpleNamespace(runner=SimpleNamespace(_run_cancellation_requested=lambda: False))
+    tool = AuthoredWorkflowTool(lambda: parent, None, str(saved))
+    bridge = SimpleNamespace(tokens=0, poll=AsyncMock(), aclose=AsyncMock())
+    child = _Child("child", environment, str(native), None, bridge=bridge)
+    tool._children[child.id] = child
+
+    await tool._execute(child, "fixture-runner", b"{}")
+
+    assert child.status == "failed"
+    assert isinstance(child.cleanup_error, OSError)
+    assert "child result persistence failed" in child.result["reason"]
+    with pytest.raises(RuntimeError, match="Docker source workspace copy failed"):
+        await tool.aclose()
+    assert (native / "workflow.json").is_file()
+    assert not child.artifacts_removed
+    assert all(not command.startswith("rm -rf -- ") for command in commands)
+
+
 async def test_artifact_copy_failure_closes_source_descriptor(tmp_path, monkeypatch):
     (tmp_path / "result.json").write_text('{"ready": true}')
     environment = LocalEnvironment(str(tmp_path))
diff --git a/tests/tools/test_native_bash_evidence.py b/tests/tools/test_native_bash_evidence.py
index 29bb7cf5..9ac7a912 100644
--- a/tests/tools/test_native_bash_evidence.py
+++ b/tests/tools/test_native_bash_evidence.py
@@ -61,6 +61,12 @@ def test_native_bash_schema_output_command_and_timeout_are_unchanged():
     ('python -m pytest test.py && echo finished', 'python -m pytest test.py && echo finished'),
     ('python -m pytest --collect-only', 'python -m pytest --collect-only'),
     ('python -m pytest --help', 'python -m pytest --help'),
+    ('python -m pytest tests.py -q # keep this note',
+     'python -m pytest tests.py -q -rA # keep this note'),
+    ("python -m pytest tests.py --junit-prefix '#suite'",
+     "python -m pytest tests.py --junit-prefix '#suite' -rA"),
+    (r"python -m pytest tests.py --junit-prefix \#suite",
+     r"python -m pytest tests.py --junit-prefix \#suite -rA"),
 ])
 def test_opt_in_pass_reporting_preserves_shell_text_and_target_separator(command, expected):
     tool = evidence_tools("bash")[0]
diff --git a/tests/tools/test_native_test_option_values.py b/tests/tools/test_native_test_option_values.py
index 6b72e4e8..e877cb8b 100644
--- a/tests/tools/test_native_test_option_values.py
+++ b/tests/tools/test_native_test_option_values.py
@@ -22,6 +22,13 @@
     ("python -m pytest --report-chars A tests/test_example.py", ("tests/test_example.py",)),
     ("python -m pytest --max-warnings 5 tests/test_example.py", ("tests/test_example.py",)),
     ("python -m pytest -r A -k example tests", ("tests",)),
+    ("python -m pytest test_example.py # keep this note", ("test_example.py",)),
+    ("python -m pytest test_example.py #don't parse ' | ; < > `", ("test_example.py",)),
+    ("python -m pytest --junit-prefix '#suite' test_example.py", ("test_example.py",)),
+    (r"python -m pytest --junit-prefix \#suite test_example.py", ("test_example.py",)),
+    ("python -m pytest '#test_example.py'", ("#test_example.py",)),
+    (r"python -m pytest \#test_example.py", ("#test_example.py",)),
+    ("python -m pytest test#example.py", ("test#example.py",)),
     ("go test -json -coverprofile=coverage.out .", (".",)),
     ("go test -json -coverprofile coverage.out .", (".",)),
     ("go test -json -covermode count -coverpkg ./... -coverprofile coverage.out ./pkg", ("./pkg",)),
@@ -36,6 +43,13 @@ def test_missing_native_option_value_cannot_prove_execution(command):
     parsed = parse_test_command(command, "/workspace")
     assert parsed is None or not parsed.provable
 
+@pytest.mark.parametrize("command", [
+    "python -m pytest test_example.py # note\npython other.py",
+    "python -m pytest\n test_example.py",
+])
+def test_native_test_commands_keep_multiline_execution_unverified(command):
+    assert parse_test_command(command, "/workspace") is None
+
 def test_pytest_core_required_values_follow_installed_option_metadata():
     for group in get_config()._parser._groups:
         for option in group.options:
diff --git a/tests/workflows/test_weave_command_checks.py b/tests/workflows/test_weave_command_checks.py
index ef85f762..eb283076 100644
--- a/tests/workflows/test_weave_command_checks.py
+++ b/tests/workflows/test_weave_command_checks.py
@@ -3,12 +3,14 @@
 from __future__ import annotations
 
 import shlex
+import shutil
 import sys
 
 import pytest
 
 from opencollab import OpenCollab
 from opencollab.builtin_workflows import WeaveAdapter
+from opencollab.builtin_workflows._weave_execution import _static_printer_supplies, _verification_mode
 from tests.workflows.test_weave_execution import FileModel
 
 
@@ -22,6 +24,249 @@ async def flow(ctx, _args):
     return result.output
 
 
+@pytest.mark.parametrize("command,executable,supplied", [
+    ("exec echo CHECK_OK", False, True),
+    ("busybox echo CHECK_OK", False, True),
+    ("bash -c 'echo CHECK_OK'", False, True),
+    ("nice -n 0 echo CHECK_OK", False, True),
+    ("echo CHECK_OK && python -c 'import sys; sys.exit(0)'", True, True),
+    ("python -c 'print(\"CHECK_OK\")'", True, True),
+    ("node -e 'console.log(\"CHECK_OK\")'", True, True),
+    ("eval 'echo CHECK_OK'", False, True),
+    ("env -i python check.py", False, False),
+    ("echo 'Starting check' && python check.py", True, False),
+    ("awk 'BEGIN{print \"CHECK_OK\"}'", False, True),
+    ("awk 'BEGIN{print \"CHECK\" \"_OK\"}'", False, False),
+    ("sed -n 's/.*/CHECK_OK/p'", False, True),
+    ("python -c 'print(\"CHE\"+\"CK_OK\")'", True, True),
+    ("python -c 'x=\"open(\"; print(\"CHE\"+\"CK_OK\")'", True, True),
+    ("python -c 'print(open(\"result.txt\").read())'", True, True),
+    ('python -c"print(\'CHE\'+\'CK_OK\')"', True, True),
+    ('SCRIPT=\'print("CHE"+"CK_OK")\' python -c"$SCRIPT"', True, True),
+    ("python -ic 'print(\"CHE\"+\"CK_OK\")'", True, True),
+    ("python -bc 'print(\"CHE\"+\"CK_OK\")'", True, True),
+    ("node -p '\"CHE\"+\"CK_OK\"'", True, True),
+    ("perl -E 'say \"CHE\".\"CK_OK\"'", True, True),
+    ("uv run python -c 'print(\"CHE\"+\"CK_OK\")'", True, True),
+    ("python -m pytest -c pytest.ini tests.py", True, False),
+    ("python -W ignore -c 'print(1)'", True, True),
+    ("python -X dev -c 'print(1)'", True, True),
+    ("python check.py -c settings.json", True, False),
+    ("python check.py node -p artifact.txt", True, False),
+    ("./check.sh python -c config.json", True, False),
+    ("node check.js -p artifact.txt", True, False),
+    ("node --require setup.js -e 'console.log(1)'", True, True),
+    ("perl check.pl -e settings.json", True, False),
+    ("ruby check.rb -e settings.json", True, False),
+    ("php check.php -r settings.json", True, False),
+    ("lua check.lua -e settings.json", True, False),
+    ("perl -I lib -E 'say 1'", True, True),
+    ("ruby -I lib -e 'puts 1'", True, True),
+    ("php -c config.ini -r 'echo 1;'", True, True),
+    ("lua -l module -e 'print(1)'", True, True),
+    ("bash check.sh", True, False),
+    ("bash -- check.sh -c settings.json", True, False),
+    ("bash -e check.sh", True, False),
+    ("bash -eux -- check.sh -c settings.json", True, False),
+    ("bash -euo pipefail check.sh", True, False),
+    ("bash -o errexit -o nounset check.sh", True, False),
+    ("bash -e", False, True),
+    ("bash -euo", False, True),
+    ("bash -s check.sh", False, True),
+    ("bash -n check.sh", False, True),
+    ("bash -o noexec check.sh", False, True),
+    ("bash -ec 'echo CHECK_OK'", False, True),
+    ("python -c 'import sys; print(sys.stdin.read())' <<< CHECK_OK", False, False),
+    ("python -c 'import sys; print(sys.stdin.read())' <<< \"$X\"", False, False),
+    ("python check.py < /dev/null", True, False),
+])
+def test_static_output_cannot_satisfy_an_expected_probe(command, executable, supplied):
+    assert _verification_mode(command) is executable
+    assert _static_printer_supplies(command, "CHECK_OK") is supplied
+
+
+@pytest.mark.parametrize("arguments", ["-c settings.json", "node -p artifact.txt"])
+@pytest.mark.parametrize("passed", [True, False])
+async def test_python_script_arguments_preserve_behavior_probe(tmp_path, arguments, passed):
+    (tmp_path / "artifact.txt").write_text("42\n", encoding="utf-8")
+    (tmp_path / "check.py").write_text(
+        "from pathlib import Path\nimport sys\n"
+        f"assert sys.argv[1:] == {shlex.split(arguments)!r}\n"
+        "Path('executed').write_text('yes')\n"
+        f"assert Path('artifact.txt').read_text() == {('42' if passed else '43')!r} + '\\n'\n"
+        "print('CHECK_OK')\n",
+        encoding="utf-8",
+    )
+    command = f"{shlex.quote(sys.executable)} check.py {arguments}"
+    output = await _check(tmp_path, {"command": command, "expected_output": "CHECK_OK"})
+    assert (tmp_path / "executed").read_text() == "yes"
+    assert output["executed"] is True and output["ok"] is passed
+
+
+@pytest.mark.parametrize("passed", [True, False])
+async def test_node_script_arguments_preserve_behavior_probe(tmp_path, passed):
+    node = shutil.which("node")
+    if node is None:
+        pytest.skip("Node is required for the JavaScript assertion probe")
+    (tmp_path / "artifact.txt").write_text("42\n", encoding="utf-8")
+    (tmp_path / "check.js").write_text(
+        "const fs = require('node:fs');\nconst assert = require('node:assert/strict');\n"
+        "assert.deepEqual(process.argv.slice(2), ['-p', 'artifact.txt']);\n"
+        "fs.writeFileSync('executed', 'yes');\n"
+        f"assert.equal(fs.readFileSync(process.argv[3], 'utf8'), {('42' if passed else '43')!r} + '\\n');\n"
+        "console.log('CHECK_OK');\n",
+        encoding="utf-8",
+    )
+    command = f"{shlex.quote(node)} check.js -p artifact.txt"
+    output = await _check(tmp_path, {"command": command, "expected_output": "CHECK_OK"})
+    assert (tmp_path / "executed").read_text() == "yes"
+    assert output["executed"] is True and output["ok"] is passed
+
+
+@pytest.mark.parametrize("interpreter,suffix,flag,source", [
+    ("perl", "pl", "-e",
+     'die "args" unless "@ARGV" eq "-e settings.json";\n'
+     'open(my $input, "<", "artifact.txt") or die $!;\n'
+     'my $actual = <$input>; chomp $actual;\n'
+     'open(my $marker, ">", "executed") or die $!; print $marker "yes";\n'
+     'die "bad artifact" unless $actual eq "VALUE";\nprint "CHECK_OK\\n";\n'),
+    ("ruby", "rb", "-e",
+     "raise 'args' unless ARGV == ['-e', 'settings.json']\n"
+     "File.write('executed', 'yes')\n"
+     "raise 'bad artifact' unless File.read('artifact.txt').strip == 'VALUE'\nputs 'CHECK_OK'\n"),
+    ("php", "php", "-r",
+     "<?php\nif (array_slice($argv, 1) !== ['-r', 'settings.json']) { exit(2); }\n"
+     "file_put_contents('executed', 'yes');\n"
+     "if (trim(file_get_contents('artifact.txt')) !== 'VALUE') { exit(1); }\n"
+     'echo "CHECK_OK\\n";\n'),
+    ("lua", "lua", "-e",
+     'assert(arg[1] == "-e" and arg[2] == "settings.json")\n'
+     'local input = assert(io.open("artifact.txt")); local actual = input:read("*a"); input:close()\n'
+     'local marker = assert(io.open("executed", "w")); marker:write("yes"); marker:close()\n'
+     'assert(actual == "VALUE\\n")\nprint("CHECK_OK")\n'),
+])
+@pytest.mark.parametrize("passed", [True, False])
+async def test_other_interpreter_script_arguments_preserve_behavior_probe(
+    tmp_path, interpreter, suffix, flag, source, passed,
+):
+    executable = shutil.which(interpreter)
+    if executable is None:
+        pytest.skip(f"{interpreter} is required for its assertion probe")
+    (tmp_path / "artifact.txt").write_text("42\n", encoding="utf-8")
+    script = f"check.{suffix}"
+    (tmp_path / script).write_text(source.replace("VALUE", "42" if passed else "43"), encoding="utf-8")
+    command = f"{shlex.quote(executable)} {script} {flag} settings.json"
+    output = await _check(tmp_path, {"command": command, "expected_output": "CHECK_OK"})
+    assert (tmp_path / "executed").read_text() == "yes"
+    assert output["executed"] is True and output["ok"] is passed
+
+
+@pytest.mark.parametrize("options,arguments", [
+    ("", ""),
+    ("--", ""),
+    ("", "python -c settings.json"),
+    ("--", "python -c settings.json"),
+    ("-e", ""),
+    ("-u", ""),
+    ("-x", ""),
+    ("-eu", "python -c settings.json"),
+    ("-eux --", "python -c settings.json"),
+    ("-o pipefail", ""),
+    ("-euo pipefail", ""),
+])
+@pytest.mark.parametrize("passed", [True, False])
+async def test_shell_file_script_preserves_assertion_probe(tmp_path, options, arguments, passed):
+    shell = shutil.which("bash")
+    if shell is None:
+        pytest.skip("Bash is required for the shell assertion probe")
+    (tmp_path / "artifact.txt").write_text("42\n", encoding="utf-8")
+    (tmp_path / "check.sh").write_text(
+        "test -f artifact.txt || exit 1\n"
+        f"test \"$*\" = {shlex.quote(arguments)} || exit 1\n"
+        "printf yes > executed\n"
+        "read -r actual < artifact.txt\n"
+        f"test \"$actual\" = {('42' if passed else '43')} || exit 1\n"
+        "printf 'CHECK_OK\\n'\n",
+        encoding="utf-8",
+    )
+    command = f"{shlex.quote(shell)} {options} check.sh {arguments}"
+    output = await _check(tmp_path, {"command": command, "expected_output": "CHECK_OK"})
+    assert (tmp_path / "executed").read_text() == "yes"
+    assert output["executed"] is True and output["ok"] is passed
+
+
+async def test_expected_output_can_overlap_a_real_script_argument(tmp_path):
+    (tmp_path / "check.py").write_text(
+        "import sys\nfrom pathlib import Path\n"
+        "assert Path('artifact.txt').read_text() == sys.argv[1]\n"
+        "print('correct')\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "artifact.txt").write_text("correct", encoding="utf-8")
+    command = f"{shlex.quote(sys.executable)} check.py correct"
+    output = await _check(tmp_path, {"command": command, "expected_output": "correct"})
+    assert output["executed"] is True and output["ok"] is True
+
+
+@pytest.mark.parametrize("target,options", [
+    ("test_result.py", ""),
+    ("test_result.py", "--junit-prefix '#suite'"),
+    ("test_result.py", r"--junit-prefix \#suite"),
+    ("test_result.py", "# keep this note"),
+    ("test_result.py", "#don't parse this ' comment; | < > `"),
+    ("#test_result.py", ""),
+    ("test#result.py", ""),
+])
+async def test_pytest_target_output_and_quoted_hash_option_preserve_test_execution(tmp_path, target, options):
+    (tmp_path / target).write_text(
+        "from pathlib import Path\ndef test_result():\n"
+        "    Path('executed').write_text('yes')\n"
+        "    assert 6 * 7 == 42\n",
+        encoding="utf-8",
+    )
+    command = f"{shlex.quote(sys.executable)} -m pytest -q -p no:cacheprovider {shlex.quote(target)} {options}"
+    output = await _check(tmp_path, {"command": command, "expected_output": target})
+    assert (tmp_path / "executed").read_text() == "yes"
+    assert output["executed"] is True and output["ok"] is True
+    assert all(row["verified"] for row in output["report"]["checks"][0]["test_records"])
+
+
+async def test_clustered_python_flag_cannot_authenticate_a_one_liner(tmp_path):
+    command = f"{shlex.quote(sys.executable)} -bc 'print(\"CHE\"+\"CK_OK\")'"
+    output = await _check(tmp_path, {"command": command, "expected_output": "CHECK_OK"})
+    assert output["executed"] is True
+    assert output["ok"] is False
+    assert output["report"]["checks"][0]["reason"] == "expected output is supplied by a static printer"
+
+
+async def test_decoy_file_read_cannot_authenticate_a_one_liner(tmp_path):
+    command = f"{shlex.quote(sys.executable)} -c 'x=\"open(\"; print(\"CHE\"+\"CK_OK\")'"
+    output = await _check(tmp_path, {"command": command, "expected_output": "CHECK_OK"})
+    assert output["executed"] is True
+    assert output["ok"] is False
+    assert output["report"]["checks"][0]["reason"] == "expected output is supplied by a static printer"
+
+
+async def test_here_string_cannot_authenticate_a_noop_program(tmp_path):
+    output = await _check(
+        tmp_path,
+        {"command": "python -c 'import sys; print(sys.stdin.read())' <<< CHECK_OK", "expected_output": "CHECK_OK"},
+    )
+    assert output["executed"] is False
+    assert output["ok"] is False
+
+
+async def test_printer_prefix_cannot_authenticate_a_noop_program(tmp_path):
+    command = f"echo CHECK_OK && {shlex.quote(sys.executable)} -c 'import sys; sys.exit(0)'"
+    output = await _check(
+        tmp_path,
+        {"command": command, "expected_output": "CHECK_OK"},
+    )
+    assert output["executed"] is True
+    assert output["ok"] is False
+    assert output["report"]["checks"][0]["reason"] == "expected output is supplied by a static printer"
+
+
 @pytest.mark.parametrize("passed", [False, True])
 @pytest.mark.parametrize("wrapper", ["", "env CHECK_ENV=yes "])
 async def test_echo_prefix_executes_actual_behavior_probe(tmp_path, passed, wrapper):
@@ -53,6 +298,15 @@ async def test_echo_prefix_executes_actual_behavior_probe(tmp_path, passed, wrap
     "echo Starting && true\nprintf CHECK_OK", "echo Starting && :\nprintf CHECK_OK",
     "echo Starting && false\nprintf CHECK_OK",
     "echo Starting && false # a comment\nprintf CHECK_OK",
+    "exec echo CHECK_OK",
+    "exec printf CHECK_OK",
+    "eval echo CHECK_OK",
+    "busybox echo CHECK_OK",
+    "nice echo CHECK_OK",
+    "xargs echo CHECK_OK",
+    "/usr/bin/time echo CHECK_OK",
+    "bash -c 'echo CHECK_OK'",
+    "sh -c 'printf CHECK_OK'",
 ])
 async def test_static_printing_compounds_remain_unverified(tmp_path, command):
     output = await _check(tmp_path, {"command": command, "expected_output": "CHECK_OK"})