fix(agent): record piped verification runs and stop double nudges

The bash wrapper now reports PIPESTATUS, so a verifier at the head of a
pipeline is scored by its own exit status instead of the last filter's.

Command matching also collapses interpreter paths to their family and skips
the flags that leave ``-m <module>`` intact (-I, -B, -u, -O, -W/-X with their
value, py -3.x), so ``pytest | tail`` and ``.venv/bin/python -u -m pytest``
count as the runs they are. Flags that exit before the module runs stay
unmatched.

agent/verification_evidence.py

@@ -210,25 +210,48 @@ def _split_shell_segments(command: str, *, posix: bool = True) -> list[_ShellSeg
     return segments if all(s.tokens for s in segments) else []
 
 
-def _exit_status_is_attributable(segments: list[_ShellSegment], match_index: int, exit_code: int) -> bool:
-    """Whether the shell's status proves the matched segment's own status.
+def _segment_status(
+    segments: list[_ShellSegment], match_index: int, exit_code: int,
+    pipeline_status: Optional[list[int]] = None,
+) -> Optional[int]:
+    """The matched segment's own exit status when the shell proves it, else ``None``.
 
-    Only the last ``;``-separated sequence reports its status; backgrounding,
-    pipes and ``||`` hide it; an ``&&`` chain proves each member only on success.
+    Only the last ``;``-separated sequence reports its status; backgrounding and ``||``
+    hide it; an ``&&`` chain proves each member only on success. A pipeline hides every
+    status but the last segment's, unless ``pipeline_status`` (the shell's PIPESTATUS
+    array) is supplied and the match is the pipeline's leading command.
     """
     if not segments or not 0 <= match_index < len(segments) or any(s.following_operator == "&" for s in segments):
-        return False
+        return None
 
     sequence_start = max(
         (i + 1 for i, s in enumerate(segments[:-1]) if s.following_operator == ";"), default=0
     )
     if match_index < sequence_start:
-        return False
+        return None
 
     operators = {segment.following_operator for segment in segments[sequence_start:-1]}
-    if operators & {"|", "|&", "||"}:
-        return False
-    return not operators or (int(exit_code) == 0 and operators == {"&&"})
+    if "||" in operators:
+        return None
+    if operators & {"|", "|&"}:
+        pipeline_start = max(
+            (
+                sequence_start + index + 1
+                for index, operator in enumerate(
+                    segment.following_operator for segment in segments[sequence_start:-1]
+                )
+                if operator not in {"|", "|&"}
+            ),
+            default=sequence_start,
+        )
+        if pipeline_status and match_index == pipeline_start:
+            return pipeline_status[0]
+        return None
+    if not operators:
+        return int(exit_code)
+    if operators == {"&&"}:
+        return int(exit_code) if int(exit_code) == 0 else None
+    return None
 
 
 def _canonical_tokens(canonical: str) -> list[str]:
@@ -239,16 +262,62 @@ def _canonical_tokens(canonical: str) -> list[str]:
         return []
 
 
+_STATUS_PRESERVING_WRAPPERS = {"timeout", "nice", "nohup", "stdbuf"}
+
+
 def _strip_command_prefix(tokens: list[str]) -> list[str]:
-    """Remove harmless command prefixes (env, VAR=x, command/time/noglob)."""
+    """Remove harmless command prefixes: env assignments, command/time/noglob, and
+    status-preserving wrappers with their flags and duration/niceness arguments."""
     i = 1 if tokens and tokens[0] == "env" else 0
     while i < len(tokens) and "=" in tokens[i] and not tokens[i].startswith("-"):
         i += 1
-    while i < len(tokens) and tokens[i] in {"command", "time", "noglob"}:
-        i += 1
+    while i < len(tokens):
+        if tokens[i] in {"command", "time", "noglob"}:
+            i += 1
+        elif tokens[i] in _STATUS_PRESERVING_WRAPPERS:
+            i += 1
+            while i < len(tokens) and (tokens[i].startswith("-") or tokens[i].isdigit()):
+                i += 1
+        else:
+            break
     return list(tokens[i:])
 
 
+def _interpreter_family(token: str) -> Optional[str]:
+    """Family name for an interpreter token wherever it lives, else ``None``.
+
+    ``./venv/bin/python3.12``, ``python3`` and ``/usr/bin/python`` are all ``python``:
+    the detected command is ``pytest`` and the agent runs it through a venv interpreter.
+    """
+    name = token.replace("\\", "/").rsplit("/", 1)[-1]
+    name = _WINDOWS_EXE_SUFFIX_RE.sub("", re.sub(r"^(?:\./)+", "", name))
+    if not _INTERPRETER_NAME_RE.match(name):
+        return None
+    family = re.sub(r"[0-9.]+$", "", name)
+    return "python" if family == "py" else family
+
+
+_INTERPRETER_FLAG_RE = re.compile(r"^-(?:[IBuEsSOd]+|3(?:\.\d+)?|[WX].*)$")
+
+
+def _is_harmless_interpreter_flag(token: str) -> bool:
+    return bool(_INTERPRETER_FLAG_RE.match(token))
+
+
+def _normalize_tokens(tokens: list[str]) -> list[str]:
+    """Comparison form: no leading ``./``, interpreters collapsed, harmless flags dropped."""
+    normalized: list[str] = []
+    i = 0
+    while i < len(tokens):
+        family = _interpreter_family(tokens[i])
+        normalized.append(family or re.sub(r"^(?:\./)+", "", tokens[i]))
+        i += 1
+        if family:
+            while i < len(tokens) and _is_harmless_interpreter_flag(tokens[i]):
+                i += 2 if tokens[i] in {"-W", "-X"} else 1
+    return normalized
+
+
 def _equivalent_needles(needle: list[str]) -> list[list[str]]:
     """Return command spellings equivalent to the detected canonical command."""
     candidates = [needle]
@@ -261,18 +330,25 @@ def _equivalent_needles(needle: list[str]) -> list[list[str]]:
     return candidates
 
 
-def _find_canonical_match(command: str, canonical_commands: list[str], exit_code: int) -> Optional[tuple[str, list[str]]]:
-    """Return ``(canonical, trailing_args)`` for the first detected command."""
+def _find_canonical_match(
+    command: str, canonical_commands: list[str], exit_code: int,
+    pipeline_status: Optional[list[int]] = None,
+) -> Optional[tuple[str, list[str], int]]:
+    """Return ``(canonical, trailing_args, status)`` for the first detected command."""
     segments = _split_shell_segments(command)
     for canonical in canonical_commands:
         needle = _canonical_tokens(canonical)
         if not needle:
             continue
         for index, segment in enumerate(segments):
-            candidate_tokens = _strip_command_prefix(segment.tokens)
+            candidate_tokens = _normalize_tokens(_strip_command_prefix(segment.tokens))
+            status = _segment_status(segments, index, exit_code, pipeline_status)
+            if status is None:
+                continue
             for candidate in _equivalent_needles(needle):
-                if candidate_tokens[:len(candidate)] == candidate and _exit_status_is_attributable(segments, index, exit_code):
-                    return canonical, candidate_tokens[len(candidate):]
+                candidate = _normalize_tokens(candidate)
+                if candidate_tokens[:len(candidate)] == candidate:
+                    return canonical, candidate_tokens[len(candidate):], status
     return None
 
 
@@ -349,7 +425,7 @@ def _find_ad_hoc_match(command: str, root: str | Path | None, exit_code: int = 0
         segments = _split_shell_segments(command, posix=posix)
         for index, segment in enumerate(segments):
             trailing_args = _ad_hoc_script_args(segment.tokens, root)
-            if trailing_args is not None and _exit_status_is_attributable(segments, index, exit_code):
+            if trailing_args is not None and _segment_status(segments, index, exit_code) is not None:
                 return trailing_args
     return None
 
@@ -422,10 +498,15 @@ def _load_changed_paths(raw: Any) -> list[Any]:
 
 
 def classify_verification_command(
-    command: str, *, cwd: str | Path | None = None, session_id: str | None = None, exit_code: int = 0, output: str = ""
+    command: str, *, cwd: str | Path | None = None, session_id: str | None = None, exit_code: int = 0,
+    output: str = "", pipeline_status: Optional[list[int]] = None,
 ) -> Optional[VerificationEvidence]:
     """Classify a terminal command as verification evidence, if applicable.
 
+    ``pipeline_status`` is the shell's ``PIPESTATUS`` array: when the verifier is the
+    leading command of a pipeline, its own status comes from there rather than from the
+    shell's code (which belongs to the last filter).
+
     Ad-hoc temp scripts only count when the project has no canonical verify
     commands at all, so they never shadow a real suite.
     """
@@ -436,30 +517,35 @@ def classify_verification_command(
         return None
 
     verify_commands = list(facts.get("verifyCommands") or [])
-    match = _find_canonical_match(command, verify_commands, int(exit_code))
+    match = _find_canonical_match(command, verify_commands, int(exit_code), pipeline_status)
     if match is not None:
-        canonical, trailing_args = match
+        canonical, trailing_args, status = match
         kind = _kind_for_command(canonical)
         scope = "targeted" if any(map(_looks_like_target, trailing_args)) else "full"
     else:
         if verify_commands or _find_ad_hoc_match(command, facts.get("root"), int(exit_code)) is None:
             return None
         canonical, kind, scope = "ad-hoc verification script", "ad_hoc", "targeted"
+        status = int(exit_code)
     return VerificationEvidence(
         command=command, canonical_command=canonical, kind=kind, scope=scope,
-        status="passed" if int(exit_code) == 0 else "failed", exit_code=int(exit_code),
+        status="passed" if int(status) == 0 else "failed", exit_code=int(exit_code),
         cwd=str(Path(cwd or ".").resolve()), root=_root_for(facts, cwd),
         session_id=str(session_id or "default"), output_summary=_summarize_output(output),
     )
 
 
 def record_terminal_result(
-    *, command: str, cwd: str | Path | None, session_id: str | None, exit_code: int, output: str = ""
+    *, command: str, cwd: str | Path | None, session_id: str | None, exit_code: int, output: str = "",
+    pipeline_status: Optional[list[int]] = None,
 ) -> Optional[dict[str, Any]]:
     """Record a foreground terminal result when it is verification evidence."""
     if not _ledger_enabled():
         return None
-    evidence = classify_verification_command(command, cwd=cwd, session_id=session_id, exit_code=exit_code, output=output)
+    evidence = classify_verification_command(
+        command, cwd=cwd, session_id=session_id, exit_code=exit_code, output=output,
+        pipeline_status=pipeline_status,
+    )
     return None if evidence is None else _insert_evidence(evidence)
 
 

agent/verification_stop.py

@@ -15,6 +15,12 @@ from agent.compression_marker import elide
 _MAX_CHANGED_PATHS_IN_NUDGE = 8
 _MAX_STATUS_SUMMARY_CHARS = 1200
 
+_UNPIPED_RUN_WARNING = (
+    " Run the check bare: no `| tail`, `| head`, or `| grep` filter. A piped run's exit "
+    "status belongs to the filter, not the check, so it cannot count as evidence; the "
+    "terminal already truncates long output and spills the rest to a file."
+)
+
 # Prose/data extensions and extension-less prose filenames (case-insensitive) with
 # no verifiable runtime behavior: a turn touching ONLY these suppresses the nudge
 # (a SKILL.md/README edit must never demand a /tmp verification script).
@@ -161,8 +167,12 @@ def build_verify_on_stop_nudge(
     *, session_id: str | None, changed_paths: Iterable[str], attempts: int=0, max_attempts: int=2,
 ) -> str | None:
     """Return a synthetic follow-up when edited code lacks fresh verification."""
-    # Prose-only turns (markdown, skills, README, LICENSE, ...) have nothing to verify.
-    paths = sorted({str(p) for p in changed_paths if p and not _is_non_code_path(p)})
+    # Prose-only turns (markdown, skills, README, LICENSE, ...) have nothing to verify, and
+    # a path under a directory that no longer exists (a removed worktree) is not a workspace.
+    paths = sorted({
+        str(p) for p in changed_paths
+        if p and not _is_non_code_path(p) and Path(p).parent.is_dir()
+    })
     if not paths or attempts >= max_attempts:
         return None
 
@@ -221,8 +231,8 @@ def build_verify_on_stop_nudge(
         "fresh passing verification evidence yet.\n\n"
         f"Verification status: {_status_detail(status)}\n\n"
         f"Changed paths:\n{_format_changed_paths(paths)}\n\n"
-        f"{command_instruction} If verification is not possible, explain the "
-        "concrete blocker instead of claiming the work is fully verified."
+        f"{command_instruction}{_UNPIPED_RUN_WARNING} If verification is not possible, "
+        "explain the concrete blocker instead of claiming the work is fully verified."
         f"{addendum}]"
     )
 

tests/agent/test_verification_evidence.py

@@ -91,6 +91,104 @@ def test_shell_wrappers_match_but_echo_does_not(tmp_path, monkeypatch):
     assert echoed is None
 
 
+@pytest.mark.parametrize("pipeline_status, expected", [([0, 0], "passed"), ([1, 0], "failed")])
+def test_piped_verifier_uses_its_own_pipeline_status(tmp_path, monkeypatch, pipeline_status, expected):
+    """``pytest | tail`` exits with tail's status; PIPESTATUS carries pytest's own."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _python_project(tmp_path)
+
+    evidence = classify_verification_command(
+        "pytest -q 2>&1 | tail -20", cwd=tmp_path, session_id="s1",
+        exit_code=0, output="12 passed", pipeline_status=pipeline_status,
+    )
+
+    assert evidence is not None
+    assert evidence.canonical_command == "pytest"
+    assert evidence.status == expected
+
+
+def test_piped_verifier_without_pipeline_status_is_not_evidence(tmp_path, monkeypatch):
+    """Without PIPESTATUS the shell's code belongs to the filter, so nothing is proven."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _python_project(tmp_path)
+
+    assert classify_verification_command(
+        "pytest -q 2>&1 | tail -20", cwd=tmp_path, session_id="s1", exit_code=0,
+    ) is None
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        "./venv/bin/python -m pytest tests/test_widget.py -q",
+        "venv/bin/python3 -m pytest tests/test_widget.py -q",
+        "/opt/venv/bin/python3.12 -m pytest tests/test_widget.py -q",
+        ".venv/bin/python -I -B -m pytest -o pythonpath=.",
+        "/usr/bin/python3.12 -u -m pytest tests/test_widget.py -q",
+        "python -W ignore -X dev -m pytest tests/test_widget.py -q",
+        "py -3.11 -m pytest tests/test_widget.py -q",
+    ],
+)
+def test_venv_interpreter_path_matches_detected_pytest(tmp_path, monkeypatch, command):
+    """A venv interpreter path is the same ``pytest`` run: the agent's usual spelling."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _python_project(tmp_path)
+
+    evidence = classify_verification_command(
+        command, cwd=tmp_path, session_id="s1", exit_code=0, output="12 passed",
+    )
+
+    assert evidence is not None
+    assert evidence.canonical_command == "pytest"
+    assert evidence.status == "passed"
+
+
+@pytest.mark.parametrize(
+    "command",
+    [
+        "python --version -m pytest",
+        "python -c pass -m pytest",
+        "python -h -m pytest",
+    ],
+)
+def test_interpreter_flag_that_skips_the_module_is_not_evidence(tmp_path, monkeypatch, command):
+    """A flag that exits before ``-m`` runs cannot produce passing evidence."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _python_project(tmp_path)
+
+    assert classify_verification_command(
+        command, cwd=tmp_path, session_id="s1", exit_code=0,
+    ) is None
+
+
+def test_dot_slash_script_invocation_matches_detected_script(tmp_path, monkeypatch):
+    """``./scripts/run_tests.sh`` is the detected command run the obvious way."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _node_project(tmp_path)
+
+    evidence = classify_verification_command(
+        "./scripts/run_tests.sh", cwd=tmp_path, session_id="s1", exit_code=0,
+    )
+
+    assert evidence is not None
+    assert evidence.canonical_command == "scripts/run_tests.sh"
+
+
+@pytest.mark.parametrize("command", ["timeout 300 pytest -q", "nice -n 5 pytest -q"])
+def test_status_preserving_wrapper_still_attributes_the_run(tmp_path, monkeypatch, command):
+    """``timeout``/``nice`` return the child's status, so the run proves the verifier's."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _python_project(tmp_path)
+
+    evidence = classify_verification_command(
+        command, cwd=tmp_path, session_id="s1", exit_code=0,
+    )
+
+    assert evidence is not None
+    assert evidence.canonical_command == "pytest"
+    assert evidence.status == "passed"
+
+
 @pytest.mark.parametrize(
     "command",
     [

tests/agent/test_verification_stop.py

@@ -25,6 +25,13 @@ def _make_project(root: Path) -> None:
     root.mkdir()
     _node_project(root)
 
+def _edited(root: Path, rel: str) -> str:
+    """Create an edited file: the guard only asks about paths a real edit left on disk."""
+    path = root / rel
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text("x", encoding="utf-8")
+    return str(path)
+
 @pytest.fixture(autouse=True)
 def _ledger_on(monkeypatch):
     """The ledger is inert unless verify-on-stop is enabled; ``clear_verify_env`` (requested
@@ -74,8 +81,8 @@ def test_nudge_checks_all_edited_workspaces(tmp_path, monkeypatch):
     project_b = tmp_path / "b"
     _make_project(project_a)
     _make_project(project_b)
-    changed_a = str(project_a / "src" / "app.ts")
-    changed_b = str(project_b / "src" / "app.ts")
+    changed_a = _edited(project_a, "src/app.ts")
+    changed_b = _edited(project_b, "src/app.ts")
 
     record_terminal_result(
         command="pnpm test",
@@ -108,7 +115,7 @@ def test_no_suite_nudge_uses_canonical_temp_dir(tmp_path, monkeypatch):
 
     nudge = build_verify_on_stop_nudge(
         session_id="s1",
-        changed_paths=[str(project / "src" / "app.ts")],
+        changed_paths=[_edited(project, "src/app.ts")],
     )
 
     assert nudge is not None
@@ -118,7 +125,7 @@ def test_no_suite_nudge_uses_canonical_temp_dir(tmp_path, monkeypatch):
 def test_ad_hoc_pass_satisfies_no_suite_stop_loop(tmp_path, monkeypatch):
     monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
     (tmp_path / "package.json").write_text("{}", encoding="utf-8")
-    changed = str(tmp_path / "src" / "app.ts")
+    changed = _edited(tmp_path, "src/app.ts")
     script = Path(tempfile.gettempdir()) / f"hermes-ad-hoc-stop-{tmp_path.name}.py"
     script.write_text("print('ok')\n", encoding="utf-8")
     try:
@@ -137,7 +144,7 @@ def test_ad_hoc_pass_satisfies_no_suite_stop_loop(tmp_path, monkeypatch):
 def test_nudge_attempts_are_bounded(tmp_path, monkeypatch):
     monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
     _node_project(tmp_path)
-    changed = str(tmp_path / "src" / "app.ts")
+    changed = _edited(tmp_path, "src/app.ts")
     mark_workspace_edited(session_id="s1", cwd=tmp_path, paths=[changed])
 
     assert build_verify_on_stop_nudge(
@@ -147,6 +154,46 @@ def test_nudge_attempts_are_bounded(tmp_path, monkeypatch):
         max_attempts=2,
     ) is None
 
+
+def test_piped_verification_run_satisfies_the_guard(tmp_path, monkeypatch):
+    """A piped run counts once the shell's PIPESTATUS is recorded with it."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _node_project(tmp_path)
+    changed = _edited(tmp_path, "src/app.ts")
+    mark_workspace_edited(session_id="s1", cwd=tmp_path, paths=[changed])
+    record_terminal_result(
+        command="pnpm test 2>&1 | tail -20",
+        cwd=tmp_path,
+        session_id="s1",
+        exit_code=0,
+        output="ok",
+        pipeline_status=[0, 0],
+    )
+
+    assert build_verify_on_stop_nudge(session_id="s1", changed_paths=[changed]) is None
+
+
+def test_path_under_a_deleted_directory_does_not_nudge(tmp_path, monkeypatch):
+    """A removed worktree is not a workspace: its vanished paths must not ask for proof."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _node_project(tmp_path)
+    gone = str(tmp_path / "worktree-gone" / "agent" / "app.py")
+
+    assert build_verify_on_stop_nudge(session_id="s1", changed_paths=[gone]) is None
+
+
+def test_nudge_demands_an_unpiped_verification_run(tmp_path, monkeypatch):
+    """A piped run reports the filter's exit status, so the ledger can never record it."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    _node_project(tmp_path)
+    changed = _edited(tmp_path, "src/app.ts")
+    mark_workspace_edited(session_id="s1", cwd=tmp_path, paths=[changed])
+
+    nudge = build_verify_on_stop_nudge(session_id="s1", changed_paths=[changed])
+
+    assert nudge is not None
+    assert "no `| tail`" in nudge
+
 # ---------------------------------------------------------------------------
 # Fix C: documentation/prose edits carry no verifiable behavior and must never
 # trip the nudge, even on an unverified workspace.
@@ -155,8 +202,8 @@ def test_nudge_attempts_are_bounded(tmp_path, monkeypatch):
 def test_mixed_doc_and_code_edit_still_nudges(tmp_path, monkeypatch):
     monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
     _node_project(tmp_path)
-    doc = str(tmp_path / "README.md")
-    code = str(tmp_path / "src" / "app.ts")
+    doc = _edited(tmp_path, "README.md")
+    code = _edited(tmp_path, "src/app.ts")
     mark_workspace_edited(session_id="s1", cwd=tmp_path, paths=[code])
 
     nudge = build_verify_on_stop_nudge(

tests/tools/test_terminal_pipeline_status.py

@@ -0,0 +1,39 @@
+"""The bash wrapper reports each pipeline segment's own exit status."""
+
+import subprocess
+
+from tools.environments.base_session_env import _wrap_command_script
+
+CWD_MARKER = "__HERMES_CWD_test__"
+PIPE_MARKER = "__HERMES_PIPE_test__"
+
+
+def _script(command: str) -> str:
+    return _wrap_command_script(
+        command, quoted_cwd="/tmp", quoted_snap="/nonexistent-snapshot",
+        snap_tmp_template="/tmp/hermes-snap.XXXXXX", passthrough_names=(),
+        snapshot_ready=False, cwd_marker=CWD_MARKER, pipe_marker=PIPE_MARKER,
+    )
+
+
+def _run(command: str) -> str:
+    return subprocess.run(["bash", "-c", _script(command)], capture_output=True, text=True, check=False).stdout
+
+
+def test_pipeline_status_captures_the_leading_command():
+    output = _run("false | true")
+
+    assert f"{PIPE_MARKER}1 0{PIPE_MARKER}" in output
+
+
+def test_plain_command_reports_one_status():
+    output = _run("false")
+
+    assert f"{PIPE_MARKER}1{PIPE_MARKER}" in output
+
+
+def test_wrapper_captures_pipestatus_before_anything_else():
+    script = _script("echo hi")
+
+    assert script.index('__hermes_ps=("${PIPESTATUS[@]}")') < script.index("umask 077")
+    assert script.index("__hermes_ec=${__hermes_ps[${#__hermes_ps[@]}-1]}") < script.index(PIPE_MARKER)

tools/environments/base.py

@@ -25,7 +25,7 @@ from tools.environments.base_output import (
     ProcessHandle, _finalize_wait_result, _new_output_collector, _start_drain_thread,
 )
 from tools.environments.base_session_env import (
-    _SHELL_ENV_NAME_RE, _SNAP_TMP_SUFFIX, _cwd_marker, _snapshot_bootstrap_script, _split_cwd_marker,
+    _SHELL_ENV_NAME_RE, _SNAP_TMP_SUFFIX, _cwd_marker, _pipe_marker, _snapshot_bootstrap_script, _split_cwd_marker,
     _wrap_command_script,
 )
 from tools.environments.base_wait import _WaitTrace
@@ -319,6 +319,7 @@ class BaseEnvironment(ABC):
         self._snapshot_path = f"{temp_dir}/hermes-snap-{self._session_id}.sh"
         self._cwd_file = f"{temp_dir}/hermes-cwd-{self._session_id}.txt"
         self._cwd_marker = _cwd_marker(self._session_id)
+        self._pipe_marker = _pipe_marker(self._session_id)
         self._snapshot_ready = False
         self._snapshot_passthrough_names: set[str] = set()
         # True when login bash is unusable (e.g. broken Git-for-Windows startup)
@@ -473,6 +474,7 @@ class BaseEnvironment(ABC):
             passthrough_names=self._snapshot_excluded_passthrough_names(),
             snapshot_ready=self._snapshot_ready,
             persist_env=self._persist_env,
+            pipe_marker=self._pipe_marker,
             **self._snapshot_script_kwargs(cwd))
 
     @staticmethod
@@ -628,6 +630,14 @@ class BaseEnvironment(ABC):
         only when THIS command emitted a marker: a killed/timed-out command emits none and
         ``self.cwd`` keeps the previous value. The environment is shared across sessions, so
         concurrent callers must read ``result["cwd"]`` rather than ``self.cwd``."""
+        pipe_split = _split_cwd_marker(result.get("output", ""), self._pipe_marker)
+        if pipe_split is not None:
+            statuses, cleaned = pipe_split
+            result["output"] = cleaned
+            try:
+                result["pipeline_status"] = [int(s) for s in (statuses or "").split()]
+            except ValueError:
+                pass
         split = _split_cwd_marker(result.get("output", ""), self._cwd_marker)
         if split is None:
             return

tools/environments/base_session_env.py

@@ -41,6 +41,20 @@ def _cwd_marker(session_id: str) -> str:
     return f"__HERMES_CWD_{session_id}__"
 
 
+def _pipe_marker(session_id: str) -> str:
+    return f"__HERMES_PIPE_{session_id}__"
+
+
+def _pipe_marker_printf(marker: str) -> str:
+    """Emit the pipeline's per-segment exit statuses (``PIPESTATUS``) on their own line.
+
+    ``cmd | tail`` exits with tail's status, so the shell's code cannot prove ``cmd``
+    passed. The array holds each segment's own status and is captured before anything
+    else runs.
+    """
+    return f"printf '\\n{marker}%s{marker}\\n' \"${{__hermes_ps[*]-}}\""
+
+
 def _cwd_marker_printf(marker: str) -> str:
     """Emit the CWD marker on its own line (leading ``\\n`` guards against a
     command whose output lacks a trailing newline; ``_split_cwd_marker`` strips it)."""
@@ -131,7 +145,7 @@ def _passthrough_save_restore(names: Iterable[str]) -> tuple[list[str], list[str
 def _wrap_command_script(
     command: str, *, quoted_cwd: str, quoted_snap: str, snap_tmp_template: str,
     passthrough_names: Iterable[str], snapshot_ready: bool, cwd_marker: str,
-    persist_env: bool = True) -> str:
+    pipe_marker: str = "", persist_env: bool = True) -> str:
     """Per-command bash script: source snapshot, cd, run, re-dump env, emit CWD marker.
     ``persist_env=False`` skips the re-dump, so every command starts from the login-time
     snapshot and exports from one command never reach the next (the CWD marker still runs).
@@ -155,8 +169,8 @@ def _wrap_command_script(
         'export GIT_PAGER="${GIT_PAGER:-cat}" PAGER="${PAGER:-cat}"',
         # ``--`` keeps hyphen-prefixed directory names from being parsed as options.
         f"builtin cd -- {quoted_cwd} || exit 126",
-        f"eval '{escaped}'",
-        "__hermes_ec=$?",
+        f"eval '{escaped}\n__hermes_ps=(\"${{PIPESTATUS[@]}}\")'",
+        "__hermes_ec=${__hermes_ps[${#__hermes_ps[@]}-1]}",
         "umask 077"]
     if snapshot_ready and persist_env:
         parts.append(
@@ -164,6 +178,8 @@ def _wrap_command_script(
             f"{{ {_export_dump_excluding_session_vars(_SNAP_TMP, passthrough_names)} "
             f"&& mv -f {_SNAP_TMP} {quoted_snap}; }} "
             f"2>/dev/null || rm -f {_SNAP_TMP} 2>/dev/null || true")
+    if pipe_marker:
+        parts.append(_pipe_marker_printf(pipe_marker))
     parts += [_cwd_marker_printf(cwd_marker), "exit $__hermes_ec"]
     return "\n".join(parts)
 

tools/terminal_tool_result.py

@@ -193,11 +193,12 @@ def _redact_spill_file(path, total_chars, command) -> list[tuple[str, Any]]:
     return [("output_total_chars", total_chars), ("full_output_path", path), ("truncation_note", note)]
 
 
-def _verification_evidence(command, cwd, session_id, returncode, output) -> Optional[dict]:
+def _verification_evidence(command, cwd, session_id, returncode, output, pipeline_status=None) -> Optional[dict]:
     with _quiet("verification evidence recording"):
         from agent.verification_evidence import record_terminal_result
         evidence = record_terminal_result(command=command, cwd=cwd, session_id=session_id,
-                                          exit_code=returncode, output=output)
+                                          exit_code=returncode, output=output,
+                                          pipeline_status=pipeline_status)
         if evidence:
             return {k: evidence.get(k) for k in ("status", "kind", "scope", "canonical_command")}
     return None
@@ -264,7 +265,7 @@ def finalize_foreground_result(
         *_redact_spill_file(result.get("full_output_path"), result.get("output_total_chars"), command),
         ("verification_evidence", _verification_evidence(
             command, command_cwd, session_id or task_id or effective_task_id or "default",
-            returncode, output)),
+            returncode, output, result.get("pipeline_status"))),
         ("approval", approval_note or None),
         ("exit_code_meaning", exit_note or None),
         ("hint", failure_hint or None),