314 lines
11 KiB
Python
314 lines
11 KiB
Python
"""Upgrade, single-shot and redaction contracts for the Hermes goal gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"cli_lane_review_contract_test",
|
|
ROOT / "services/hermes/scripts/cli_lane_goal.py",
|
|
)
|
|
assert SPEC and SPEC.loader
|
|
goal = importlib.util.module_from_spec(SPEC)
|
|
sys.modules[SPEC.name] = goal
|
|
SPEC.loader.exec_module(goal)
|
|
|
|
# A synthetic 40-character lowercase hex string. It has the shape of a Gitea
|
|
# personal access token *and* of a git SHA-1; no real credential is read here.
|
|
SYNTHETIC_HEX = "0123456789abcdef" * 2 + "01234567"
|
|
REAL_SHA = "b080b5f622ae0998213f3287762aea30dc931a73"
|
|
|
|
REPORTS = {
|
|
"block_review": {
|
|
"status": "completed",
|
|
"summary": (
|
|
"Independent read-only review of PR #18. Verdict: BLOCK. Deployment "
|
|
"verification is pending on two of three nodes, which is exactly why "
|
|
"the reviewed change is unfit to ship."
|
|
),
|
|
"changed_files": [],
|
|
"tests_run": ["pytest testing/tests: 337 passed"],
|
|
"artifacts": [],
|
|
"findings": ["P0 - coordinator.py:248 compares int to str run ids."],
|
|
"blockers": [],
|
|
},
|
|
"bare_token_review": {
|
|
"status": "completed",
|
|
"summary": (
|
|
"Independent read-only review of PR #18 concludes BLOCK because the "
|
|
"coordinator drops a lease on its first tick and deployment "
|
|
"verification is pending on two of three nodes."
|
|
),
|
|
"changed_files": [],
|
|
"tests_run": [],
|
|
"artifacts": [],
|
|
"findings": ["P0 - the lease is dropped before the first heartbeat."],
|
|
"blockers": [],
|
|
},
|
|
"terse_ship_audit": {
|
|
"status": "completed",
|
|
"summary": "Decision: ship. Audit done.",
|
|
"changed_files": [],
|
|
"tests_run": ["ruff check: clean"],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": [],
|
|
},
|
|
"block_audit_without_findings": {
|
|
"status": "completed",
|
|
"summary": (
|
|
"The generated-strategy audit harness was inspected end to end and "
|
|
"the verdict is BLOCK on the release train for now."
|
|
),
|
|
"changed_files": [],
|
|
"tests_run": ["pytest -q: 61 passed"],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": [],
|
|
},
|
|
"finished_implementation": {
|
|
"status": "completed",
|
|
"summary": "Every acceptance criterion passed and the branch was pushed.",
|
|
"changed_files": ["services/hermes/scripts/cli_lane_goal.py"],
|
|
"tests_run": ["pytest -q: 12 passed"],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": [],
|
|
},
|
|
"unfinished_implementation": {
|
|
"status": "completed",
|
|
"summary": "The broad rerun remains active before the final push.",
|
|
"changed_files": ["services/hermes/scripts/cli_lane_goal.py"],
|
|
"tests_run": [],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": [],
|
|
},
|
|
"incomplete_turn": {
|
|
"status": "incomplete",
|
|
"summary": "Only two of six boundaries were read.",
|
|
"changed_files": [],
|
|
"tests_run": [],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": [],
|
|
},
|
|
"completed_with_blockers": {
|
|
"status": "completed",
|
|
"summary": "The review finished but the head could not be resolved.",
|
|
"changed_files": [],
|
|
"tests_run": [],
|
|
"artifacts": [],
|
|
"findings": [],
|
|
"blockers": ["the Gitea API was unreachable"],
|
|
},
|
|
}
|
|
|
|
|
|
def _base_unfinished_result_reason(result):
|
|
"""The completion gate exactly as main d8f2d818 shipped it.
|
|
|
|
Kept as a local oracle so the upgrade properties below are checked against
|
|
the previous semantics rather than against the implementation under test.
|
|
"""
|
|
status = str(result.get("status") or "")
|
|
summary = str(result.get("summary") or "").strip()
|
|
blockers = result.get("blockers")
|
|
if status == "incomplete":
|
|
return summary or "worker explicitly reported incomplete work"
|
|
if status != "completed":
|
|
return None
|
|
if isinstance(blockers, list) and any(str(item).strip() for item in blockers):
|
|
return "worker reported blockers while claiming completion"
|
|
tests = result.get("tests_run")
|
|
evidence = [summary]
|
|
if isinstance(tests, list):
|
|
evidence.extend(str(item) for item in tests)
|
|
match = goal.UNFINISHED_EVIDENCE.search("\n".join(evidence))
|
|
if match:
|
|
return f"completion evidence says work is unfinished: {match.group(0).strip()}"
|
|
return None
|
|
|
|
|
|
def _accepted(result, *, role):
|
|
return goal.unfinished_result_reason(result, role=role) is None
|
|
|
|
|
|
@pytest.mark.parametrize("name", sorted(REPORTS))
|
|
def test_a_role_blind_replay_accepts_whatever_the_lane_accepted(name):
|
|
"""PR15 re-runs this gate on journalled terminal records without a card.
|
|
|
|
A record accepted by any version of the lane must stay valid, or an upgrade
|
|
quarantines it and re-dispatches an already-accepted task.
|
|
"""
|
|
result = REPORTS[name]
|
|
accepted_somewhere = _base_unfinished_result_reason(result) is None or any(
|
|
_accepted(result, role=role)
|
|
for role in (goal.REVIEW_ROLE, goal.IMPLEMENTATION_ROLE)
|
|
)
|
|
|
|
if accepted_somewhere:
|
|
assert goal.unfinished_result_reason(result) is None
|
|
|
|
|
|
def test_role_blind_replay_keeps_the_unfinished_work_integrity_check():
|
|
"""PR15's ``_terminal_record_valid`` relies on this without a card."""
|
|
assert goal.unfinished_result_reason(REPORTS["completed_with_blockers"])
|
|
assert goal.unfinished_result_reason(REPORTS["incomplete_turn"])
|
|
assert goal.unfinished_result_reason(REPORTS["unfinished_implementation"])
|
|
assert goal.unfinished_result_reason(
|
|
{**REPORTS["terse_ship_audit"], "summary": "tests are still running"}
|
|
)
|
|
|
|
|
|
def test_role_blind_replay_exempts_only_a_complete_review_deliverable():
|
|
"""The exemption is exactly the shape the role-aware lane calls a review."""
|
|
complete = REPORTS["bare_token_review"]
|
|
mutated = {**complete, "changed_files": ["src/a.py"]}
|
|
verdictless = {**complete, "summary": complete["summary"].replace("BLOCK", "no")}
|
|
|
|
assert _base_unfinished_result_reason(complete)
|
|
assert _accepted(complete, role=goal.REVIEW_ROLE)
|
|
assert goal.unfinished_result_reason(complete) is None
|
|
assert goal.unfinished_result_reason(mutated)
|
|
assert goal.unfinished_result_reason(verdictless)
|
|
|
|
|
|
@pytest.mark.parametrize("name", sorted(REPORTS))
|
|
def test_implementation_completion_is_byte_for_byte_the_previous_gate(name):
|
|
result = REPORTS[name]
|
|
|
|
assert goal.unfinished_result_reason(
|
|
result, role=goal.IMPLEMENTATION_ROLE
|
|
) == (
|
|
goal.sanitize_reason(_base_unfinished_result_reason(result))
|
|
if _base_unfinished_result_reason(result)
|
|
else None
|
|
)
|
|
|
|
|
|
def test_the_two_single_shot_outcomes_that_change_are_the_documented_ones():
|
|
"""Single-shot runs this gate as its only completion check.
|
|
|
|
The verdict contract binds there too, so a review's own deliverable decides
|
|
the outcome rather than prose about the artifact it reviewed.
|
|
"""
|
|
now_completes = REPORTS["bare_token_review"]
|
|
now_blocks = (REPORTS["terse_ship_audit"], REPORTS["block_audit_without_findings"])
|
|
|
|
assert _base_unfinished_result_reason(now_completes)
|
|
assert _accepted(now_completes, role=goal.REVIEW_ROLE)
|
|
for result in now_blocks:
|
|
assert _base_unfinished_result_reason(result) is None
|
|
problem = goal.unfinished_result_reason(result, role=goal.REVIEW_ROLE)
|
|
assert problem and problem.startswith(goal.READ_ONLY_GUARD)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "secret"),
|
|
[
|
|
(
|
|
f"cloned https://hermes:{SYNTHETIC_HEX}@scm.bstein.dev/titan/x.git",
|
|
SYNTHETIC_HEX,
|
|
),
|
|
(f"exported GITEA_TOKEN={SYNTHETIC_HEX} into the lane", SYNTHETIC_HEX),
|
|
(f"the pat {SYNTHETIC_HEX} was reused by the askpass helper", SYNTHETIC_HEX),
|
|
(f"Authorization: token {SYNTHETIC_HEX}", SYNTHETIC_HEX),
|
|
("token ghp_" + "A" * 24, "ghp_" + "A" * 24),
|
|
("the secret is xoxb-1234567890-abcdefghij", "xoxb-1234567890-abcdefghij"),
|
|
(f"reviewed exact head {REAL_SHA} against base main", None),
|
|
],
|
|
)
|
|
def test_judge_reasons_redact_credentials_but_keep_commit_evidence(text, secret):
|
|
"""A 40-hex PAT and a 40-hex commit SHA are the same string in isolation.
|
|
|
|
Redaction is anchored on the surrounding syntax so credentials disappear
|
|
and the exact head a review pins its findings to survives as evidence.
|
|
"""
|
|
reason = goal.sanitize_reason(text)
|
|
|
|
if secret is None:
|
|
assert reason == text
|
|
else:
|
|
assert secret not in reason
|
|
assert "[redacted]" in reason or "***" in reason
|
|
|
|
|
|
def test_redaction_falls_back_when_the_agent_runtime_is_absent(monkeypatch):
|
|
monkeypatch.setattr(goal, "_canonical_redact", None)
|
|
|
|
reason = goal.sanitize_reason(
|
|
f"token ghp_{'A' * 24} and https://ci:{SYNTHETIC_HEX}@scm.bstein.dev/x.git"
|
|
)
|
|
|
|
assert "ghp_" not in reason
|
|
assert SYNTHETIC_HEX not in reason
|
|
assert reason.count("[redacted]") == 2
|
|
|
|
|
|
def test_redaction_delegates_to_the_canonical_helper_when_it_is_installed(monkeypatch):
|
|
seen = []
|
|
|
|
def spy(text, **kwargs):
|
|
seen.append(kwargs)
|
|
return text.replace("OPAQUE", "***")
|
|
|
|
monkeypatch.setattr(goal, "_canonical_redact", spy)
|
|
|
|
assert goal.sanitize_reason("value OPAQUE here") == "value *** here"
|
|
assert seen == [{"force": True}]
|
|
|
|
|
|
def test_the_canonical_redactor_is_the_one_the_lane_runtime_ships():
|
|
canonical = pytest.importorskip("agent.redact")
|
|
|
|
assert goal._canonical_redact is canonical.redact_sensitive_text
|
|
|
|
|
|
def test_the_upstream_worker_context_headings_are_still_the_ones_we_cut_on():
|
|
"""Pin ``hermes_cli.kanban_db.build_worker_context``'s H2 contract.
|
|
|
|
``card_scope`` allow-lists CARD_SECTIONS and drops every other H2. A rename
|
|
upstream fails closed rather than leaking history into role resolution, but
|
|
it would also truncate real cards, so the coupling is asserted explicitly.
|
|
"""
|
|
kanban_db = pytest.importorskip("hermes_cli.kanban_db")
|
|
source = Path(kanban_db.__file__).read_text(encoding="utf-8")
|
|
emitted = {
|
|
match.group(1).replace("@", "").strip()
|
|
for match in re.finditer(r'lines\.append\(f?"## ([^"{]+)', source)
|
|
}
|
|
|
|
assert emitted == {*goal.CARD_SECTIONS, *goal.HISTORY_SECTIONS}
|
|
|
|
|
|
def test_the_controller_evidence_heading_is_not_a_card_section():
|
|
assert goal.CONTROLLER_EVIDENCE_HEADING.startswith("## ")
|
|
assert goal.CONTROLLER_EVIDENCE_HEADING[3:] not in goal.CARD_SECTIONS
|
|
assert goal.card_scope(
|
|
f"## Body\nreview only\n\n{goal.CONTROLLER_EVIDENCE_HEADING}\npush it"
|
|
) == "## Body\nreview only\n\n"
|
|
|
|
|
|
def test_every_report_shape_round_trips_through_the_result_schema():
|
|
for name, result in REPORTS.items():
|
|
assert set(result) == {
|
|
"status",
|
|
"summary",
|
|
"changed_files",
|
|
"tests_run",
|
|
"artifacts",
|
|
"findings",
|
|
"blockers",
|
|
}, name
|
|
assert result["status"] in goal.RESULT_STATUSES
|
|
assert json.loads(json.dumps(result)) == result
|