ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(agent): reject masked verification results

Browse Source
This commit is contained in:
fangliquanflq
2026-08-10 17:44:10 +08:00
committed by Teknium
parent 1f8f79b320
commit c6dfdcbf8d
2 changed files with 263 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/verification_evidence.py
+123 -21
View File
@@ -8,7 +8,6 @@ blocks completion, and never upgrades targeted checks into "repo green".
from __future__ import annotations
import json
import re
import shlex
import sqlite3
import tempfile
@@ -29,7 +28,12 @@ _MAX_EVENTS_PER_SESSION_ROOT = 100
_MAX_TOTAL_UNREFERENCED_EVENTS = 10_000
_AD_HOC_SCRIPT_NAME_PREFIXES = ("hermes-verify-", "hermes-ad-hoc-")
_VERIFY_SCHEMA_VERSION = 1
_SHELL_SPLIT_RE = re.compile(r"\s*(?:&&|\|\||;)\s*")
@dataclass(frozen=True)
class _ShellSegment:
tokens: list[str]
following_operator: str | None = None
@dataclass(frozen=True)
@@ -150,20 +154,104 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
conn.commit()
def _split_segment_tokens(command: str, *, posix: bool = True) -> list[list[str]]:
segments: list[list[str]] = []
for segment in _SHELL_SPLIT_RE.split(command.strip()):
if not segment:
def _split_shell_segments(command: str, *, posix: bool = True) -> list[_ShellSegment]:
"""Tokenize top-level shell commands while preserving their control operators."""
raw_segments: list[tuple[str, str | None]] = []
start = 0
quote: str | None = None
escaped = False
index = 0
while index < len(command):
char = command[index]
if escaped:
escaped = False
index += 1
continue
if char == "\\" and quote != "'":
escaped = True
index += 1
continue
if quote:
if char == quote:
quote = None
index += 1
continue
if char in {"'", '"'}:
quote = char
index += 1
continue
operator = None
if command.startswith(("&&", "||", "|&"), index):
operator = command[index:index + 2]
elif char == "\n":
operator = ";"
elif char in ";|":
operator = char
elif (
char == "&"
and (index == 0 or command[index - 1] not in "<>")
and not command.startswith(("&>", "&>>"), index)
):
operator = char
if operator is None:
index += 1
continue
raw = command[start:index].strip()
if not raw:
return []
raw_segments.append((raw, operator))
index += 1 if char == "\n" else len(operator)
start = index
if quote or escaped:
return []
trailing = command[start:].strip()
if trailing:
raw_segments.append((trailing, None))
elif raw_segments and raw_segments[-1][1] not in {";"}:
return []
segments: list[_ShellSegment] = []
for raw, operator in raw_segments:
try:
tokens = shlex.split(segment, posix=posix)
tokens = shlex.split(raw, posix=posix)
except ValueError:
continue
if tokens:
segments.append(tokens)
return []
if not tokens:
return []
segments.append(_ShellSegment(tokens=tokens, following_operator=operator))
return segments
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."""
if not segments or not 0 <= match_index < len(segments):
return False
if any(segment.following_operator == "&" for segment in segments):
return False
sequence_start = 0
for index, segment in enumerate(segments[:-1]):
if segment.following_operator == ";":
sequence_start = index + 1
if match_index < sequence_start:
return False
sequence = segments[sequence_start:]
operators = [segment.following_operator for segment in sequence[:-1]]
if any(operator in {"|", "|&", "||"} for operator in operators):
return False
if len(sequence) == 1:
return True
return int(exit_code) == 0 and all(operator == "&&" for operator in operators)
def _clean_token(token: str) -> str:
token = token.strip()
while token.startswith("./"):
@@ -223,18 +311,25 @@ def _equivalent_needles(needle: list[str]) -> list[list[str]]:
return candidates
def _find_canonical_match(command: str, canonical_commands: list[str]) -> Optional[tuple[str, list[str]]]:
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."""
segments = _split_segment_tokens(command)
segments = _split_shell_segments(command)
for canonical in canonical_commands:
needle = _canonical_tokens(canonical)
if not needle:
continue
for tokens in segments:
candidate_tokens = _strip_command_prefix(tokens)
for index, segment in enumerate(segments):
candidate_tokens = _strip_command_prefix(segment.tokens)
for candidate in _equivalent_needles(needle):
if candidate_tokens[:len(candidate)] == candidate:
if (
candidate_tokens[:len(candidate)] == candidate
and _exit_status_is_attributable(segments, index, exit_code)
):
return canonical, candidate_tokens[len(candidate):]
return None
@@ -325,13 +420,20 @@ def _ad_hoc_script_args(tokens: list[str], root: str | Path | None) -> Optional[
return None
def _find_ad_hoc_match(command: str, root: str | Path | None) -> Optional[list[str]]:
def _find_ad_hoc_match(
command: str,
root: str | Path | None,
exit_code: int = 0,
) -> Optional[list[str]]:
# Try both posix=True (default) and posix=False (Windows backslash paths)
# so ad-hoc verification scripts with backslash paths are matched on Windows.
for posix in (True, False):
for tokens in _split_segment_tokens(command, posix=posix):
trailing_args = _ad_hoc_script_args(tokens, root)
if trailing_args is not None:
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
):
return trailing_args
return None
@@ -433,10 +535,10 @@ def classify_verification_command(
return None
verify_commands = list(facts.get("verifyCommands") or [])
match = _find_canonical_match(command, verify_commands)
match = _find_canonical_match(command, verify_commands, int(exit_code))
is_ad_hoc = False
if match is None and not verify_commands:
ad_hoc_args = _find_ad_hoc_match(command, facts.get("root"))
ad_hoc_args = _find_ad_hoc_match(command, facts.get("root"), int(exit_code))
if ad_hoc_args is not None:
match = ("ad-hoc verification script", ad_hoc_args)
is_ad_hoc = True
tests/agent/test_verification_evidence.py
+140
View File
@@ -4,6 +4,8 @@ import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
from agent.verification_evidence import (
classify_verification_command,
mark_workspace_edited,
@@ -82,6 +84,144 @@ def test_shell_wrappers_match_but_echo_does_not(tmp_path, monkeypatch):
assert echoed is None
@pytest.mark.parametrize(
"command",
[
"pytest || true",
"pytest ; true",
"pytest | tee test.log",
"pytest &",
],
)
def test_masking_shell_control_is_not_verification_evidence(
tmp_path, monkeypatch, command
):
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 None
@pytest.mark.parametrize("command", ["prepare && pytest", "pytest && report"])
def test_successful_and_chain_preserves_passing_evidence(
tmp_path, monkeypatch, command
):
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.status == "passed"
@pytest.mark.parametrize("exit_code, expected", [(0, "passed"), (1, "failed")])
def test_final_verifier_after_sequence_owns_shell_exit_status(
tmp_path, monkeypatch, exit_code, expected
):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
_python_project(tmp_path)
evidence = classify_verification_command(
"prepare; pytest",
cwd=tmp_path,
session_id="s1",
exit_code=exit_code,
)
assert evidence is not None
assert evidence.status == expected
def test_quoted_shell_operator_remains_a_verifier_argument(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
_python_project(tmp_path)
evidence = classify_verification_command(
"pytest -k 'passes || fails'",
cwd=tmp_path,
session_id="s1",
exit_code=0,
)
assert evidence is not None
assert evidence.status == "passed"
@pytest.mark.parametrize("redirect", ["2>&1", "&> test.log"])
def test_shell_redirection_does_not_hide_simple_verifier(tmp_path, monkeypatch, redirect):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
_python_project(tmp_path)
evidence = classify_verification_command(
f"pytest {redirect}",
cwd=tmp_path,
session_id="s1",
exit_code=0,
)
assert evidence is not None
assert evidence.status == "passed"
def test_masked_ad_hoc_script_is_not_verification_evidence(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
(tmp_path / "package.json").write_text("{}", encoding="utf-8")
script = Path(tempfile.gettempdir()) / f"hermes-ad-hoc-{tmp_path.name}.py"
script.write_text("raise SystemExit(1)\n", encoding="utf-8")
try:
evidence = classify_verification_command(
f"python {script} || true",
cwd=tmp_path,
session_id="s1",
exit_code=0,
)
finally:
script.unlink(missing_ok=True)
assert evidence is None
def test_masked_verifier_does_not_clear_edited_ledger_state(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
_python_project(tmp_path)
record_terminal_result(
command="pytest",
cwd=tmp_path,
session_id="s1",
exit_code=0,
output="passed",
)
mark_workspace_edited(
session_id="s1",
cwd=tmp_path,
paths=[str(tmp_path / "changed.py")],
)
result = record_terminal_result(
command="pytest || true",
cwd=tmp_path,
session_id="s1",
exit_code=0,
output="1 failed",
)
assert result is None
assert verification_status(session_id="s1", cwd=tmp_path)["status"] == "stale"
def test_temp_script_records_ad_hoc_evidence_without_canonical_suite(tmp_path, monkeypatch):