toki
0e17a9d9ec
Edge 내부 오류가 provider HTTP 거부로 기록되지 않도록 실제 tunnel status에서만 관측하고, dev release가 stale tracking ref와 존재하지 않는 package root에 막히지 않게 한다.
1061 lines
51 KiB
Python
1061 lines
51 KiB
Python
"""
|
|
Credential-free contract tests binding the benchmark skill to the CLI surface.
|
|
|
|
Covers: template/frontmatter invariants, project rule routing, documented
|
|
command forms against --help and required options, supported/unsupported capability matrix,
|
|
absence of the internal workspace API from user routing, provider/dispatcher/secret/prepare prohibitions,
|
|
durable state vs isolated cache boundaries, and mutation resistance.
|
|
|
|
Invokes only help and tracked text; never runs a stateful benchmark command.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
# Ensure repo root is importable
|
|
_REPO_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
if str(_REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(_REPO_ROOT))
|
|
|
|
_SKILL_DIR = _REPO_ROOT / "agent-ops" / "skills" / "project" / "iop-agent-comparison-benchmark"
|
|
_SKILL_FILE = _SKILL_DIR / "SKILL.md"
|
|
_DEPLOY_SKILL_FILE = (
|
|
_REPO_ROOT / "agent-ops" / "skills" / "project" / "dev-runtime-deploy" / "SKILL.md"
|
|
)
|
|
_RULES_FILE = _REPO_ROOT / "agent-ops" / "rules" / "project" / "rules.md"
|
|
_CLI_SCRIPT = _REPO_ROOT / "scripts" / "agent_comparison_benchmark.py"
|
|
|
|
# Commands documented by the CLI --help
|
|
_CLI_HELP_COMMANDS = {"validate", "preflight", "run", "resume", "status", "score", "report"}
|
|
|
|
# Cached exact option sets per subcommand, derived from each subcommand --help.
|
|
_CLI_OPTION_CACHE: dict[str, set[str]] = {}
|
|
|
|
|
|
class BenchmarkSkillContractTest(unittest.TestCase):
|
|
"""Contract tests for the iop-agent-comparison-benchmark skill."""
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helper methods for section parsing & semantic assertions
|
|
# ------------------------------------------------------------------
|
|
|
|
def _get_section(self, skill_text: str, section_name: str) -> str:
|
|
in_section = False
|
|
lines = []
|
|
target = f"## {section_name.strip()}"
|
|
for line in skill_text.splitlines():
|
|
if line.startswith("## "):
|
|
if line.strip().lower() == target.lower():
|
|
in_section = True
|
|
continue
|
|
elif in_section:
|
|
break
|
|
if in_section:
|
|
lines.append(line)
|
|
return "\n".join(lines)
|
|
|
|
def _get_procedure_step(self, skill_text: str, command: str) -> str:
|
|
"""Extract the numbered `Delegate <command>` Procedure step body."""
|
|
procedure = self._get_section(skill_text, "Procedure")
|
|
pattern = rf"\d+\.\s+\*\*Delegate {command}.*?(?=\n\d+\.|\Z)"
|
|
match = re.search(pattern, procedure, re.DOTALL)
|
|
self.assertTrue(
|
|
match,
|
|
f"Procedure step delegating '{command}' must be present",
|
|
)
|
|
return match.group(0)
|
|
|
|
def _get_cli_help_commands(self) -> set[str]:
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(
|
|
result.returncode,
|
|
0,
|
|
f"CLI --help exited {result.returncode}: {result.stderr}",
|
|
)
|
|
commands: set[str] = set()
|
|
in_subparsers = False
|
|
for line in result.stdout.splitlines():
|
|
stripped = line.strip()
|
|
if "positional arguments:" in line:
|
|
in_subparsers = True
|
|
continue
|
|
if in_subparsers and stripped and not stripped.startswith("-"):
|
|
if stripped == "options:":
|
|
continue
|
|
if stripped.startswith("{") and stripped.endswith("}"):
|
|
for cmd in stripped[1:-1].split(","):
|
|
commands.add(cmd)
|
|
else:
|
|
commands.add(stripped.split()[0])
|
|
return commands
|
|
|
|
def _assert_no_public_prepare(self, skill_text: str) -> None:
|
|
"""Parse all sections; fail if public prepare is exposed as an operation or trigger."""
|
|
allowed_prepare_fragments = (
|
|
"no public `prepare` operation was exposed or referenced.",
|
|
"do not expose a public `prepare` operation.",
|
|
)
|
|
for line in skill_text.splitlines():
|
|
stripped = line.strip()
|
|
if "prepare" in stripped.lower():
|
|
lower = stripped.lower()
|
|
if not any(fragment in lower for fragment in allowed_prepare_fragments):
|
|
self.fail(
|
|
"Section or step exposes or mentions prepare operation "
|
|
f"outside the approved prohibition forms: '{stripped}'"
|
|
)
|
|
|
|
def _assert_provider_prohibition(self, skill_text: str) -> None:
|
|
"""Assert provider APIs/services are prohibited and never invoked in any section."""
|
|
prohibitions = self._get_section(skill_text, "Prohibitions")
|
|
self.assertRegex(
|
|
prohibitions,
|
|
r"(?i)do not invoke.*provider",
|
|
"Prohibitions must explicitly forbid provider invocations",
|
|
)
|
|
allowed_provider_fragments = (
|
|
"no caller or provider was invoked outside the deterministic cli.",
|
|
"do not invoke a caller or provider outside the deterministic benchmark cli.",
|
|
"stop without fallback, fabricated evidence, ad-hoc provider calls, subagents, or orchestration dispatchers.",
|
|
)
|
|
for line in skill_text.splitlines():
|
|
stripped = line.strip()
|
|
if "provider" in stripped.lower():
|
|
lower = stripped.lower()
|
|
if not any(fragment in lower for fragment in allowed_provider_fragments):
|
|
self.fail(
|
|
"Line contains an unapproved provider operation or "
|
|
f"prohibition form: '{stripped}'"
|
|
)
|
|
|
|
def _get_cli_subcommand_options(self, cmd: str) -> set[str]:
|
|
"""Derive the exact long-option set for a subcommand from its live --help output."""
|
|
if cmd in _CLI_OPTION_CACHE:
|
|
return set(_CLI_OPTION_CACHE[cmd])
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), cmd, "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(
|
|
result.returncode,
|
|
0,
|
|
f"CLI {cmd} --help exited {result.returncode}: {result.stderr}",
|
|
)
|
|
options = set(re.findall(r"--[a-z][a-z0-9-]*", result.stdout))
|
|
options.discard("--help")
|
|
_CLI_OPTION_CACHE[cmd] = set(options)
|
|
return options
|
|
|
|
def _documented_command_options(self, cmd_line: str) -> set[str]:
|
|
"""Extract the documented long-option set, excluding --help and value placeholders."""
|
|
options: set[str] = set()
|
|
for match in re.finditer(r"--[a-z][a-z0-9-]*", cmd_line):
|
|
token = match.group(0)
|
|
if token != "--help":
|
|
options.add(token)
|
|
return options
|
|
|
|
def _assert_command_options(self, skill_text: str) -> None:
|
|
"""Parse each documented CLI invocation and require exact option parity with subcommand --help."""
|
|
procedure = self._get_section(skill_text, "Procedure")
|
|
for cmd in ("validate", "preflight", "run", "resume", "status", "score", "report"):
|
|
pattern = rf"python3 scripts/agent_comparison_benchmark\.py {cmd}\b[^\n]*"
|
|
matches = re.findall(pattern, procedure)
|
|
self.assertTrue(matches, f"Documented command line for '{cmd}' missing from Procedure")
|
|
cli_options = self._get_cli_subcommand_options(cmd)
|
|
for index, cmd_line in enumerate(matches, start=1):
|
|
documented = self._documented_command_options(cmd_line)
|
|
self.assertEqual(
|
|
documented,
|
|
cli_options,
|
|
f"Documented options {sorted(documented)} for '{cmd}' invocation "
|
|
f"#{index} must exactly match CLI --help options "
|
|
f"{sorted(cli_options)}: '{cmd_line}'",
|
|
)
|
|
|
|
def _assert_boundary_wording(self, skill_text: str) -> None:
|
|
"""Assert consistent durable-state, isolated cache, read-only testbed, and output boundaries."""
|
|
self.assertIn("Durable run/attempt state", skill_text)
|
|
self.assertIn("agent-test/runs/<output-id>/<run-id>/", skill_text)
|
|
self.assertRegex(
|
|
skill_text,
|
|
r"fresh and isolated for every cell, repetition, and attempt",
|
|
"Skill text must specify per-cell/per-repetition/per-attempt freshness and isolation",
|
|
)
|
|
self.assertIn("../iop-s2", skill_text)
|
|
self.assertIn("read-only", skill_text)
|
|
forbidden_phrases = [
|
|
"do not cache or persist state between invocations",
|
|
"do not read or write files outside the benchmark workspace",
|
|
"The benchmark workspace root is fixed at `../iop-s2`.",
|
|
"Output is contained within the benchmark workspace.",
|
|
"fresh, isolated per run, and never shared across runs",
|
|
]
|
|
for phrase in forbidden_phrases:
|
|
self.assertNotIn(phrase, skill_text, f"Forbidden contradictory boundary phrase found: {phrase}")
|
|
|
|
prohibitions = self._get_section(skill_text, "Prohibitions")
|
|
self.assertRegex(
|
|
prohibitions,
|
|
r"(?i)do not share session or cache state within a run",
|
|
"Prohibitions must explicitly forbid session or cache state sharing within a run",
|
|
)
|
|
|
|
for line in skill_text.splitlines():
|
|
stripped = line.strip()
|
|
lower = stripped.lower()
|
|
if (
|
|
"session" in lower
|
|
and "cache" in lower
|
|
and any(token in lower for token in ("share", "shared", "sharing"))
|
|
):
|
|
allowed_boundary_fragments = (
|
|
"caller sessions, output workspaces, and caches are fresh and isolated for every cell, repetition, and attempt; session or cache state is never shared within a run or across runs.",
|
|
"do not share session or cache state within a run across cells, repetitions, or attempts, or across run invocations.",
|
|
)
|
|
if not any(fragment in lower for fragment in allowed_boundary_fragments):
|
|
self.fail(
|
|
"Line contains an unapproved affirmative or contradictory "
|
|
f"session/cache sharing statement: '{stripped}'"
|
|
)
|
|
elif "cache" in lower and "sharing" in lower:
|
|
self.fail(f"Line contains an unapproved cache-sharing statement: '{stripped}'")
|
|
|
|
def _assert_error_ordering(self, skill_text: str) -> None:
|
|
"""Assert invalid state is handled before execution preflight blockers."""
|
|
procedure = self._get_section(skill_text, "Procedure")
|
|
for cmd in ("run", "resume", "status", "score", "report"):
|
|
pattern = rf"\d+\.\s+\*\*Delegate {cmd}.*?(?=\n\d+\.|\Z)"
|
|
match = re.search(pattern, procedure, re.DOTALL)
|
|
self.assertTrue(match, f"Procedure step for '{cmd}' missing")
|
|
step_text = match.group(0)
|
|
if cmd in ("run", "resume"):
|
|
pos_invalid = step_text.find("benchmark state is unavailable")
|
|
pos_blocked = step_text.find("preflight blocked")
|
|
self.assertTrue(
|
|
pos_invalid != -1
|
|
and pos_blocked != -1
|
|
and pos_invalid < pos_blocked,
|
|
f"In step '{cmd}', invalid state must precede preflight blockers",
|
|
)
|
|
if cmd == "report":
|
|
self.assertIn(
|
|
"benchmark report is unavailable",
|
|
step_text,
|
|
f"In step '{cmd}', invalid state must be reported",
|
|
)
|
|
|
|
def _assert_capabilities(self, skill_text: str) -> None:
|
|
"""Assert report is a supported operation alongside run/resume."""
|
|
procedure = self._get_section(skill_text, "Procedure")
|
|
self.assertNotIn("capability-unavailable: caller-adapter", procedure)
|
|
self.assertIn("append a fresh all-cell preflight before attempt allocation", procedure)
|
|
self.assertIn("invoke each eligible cell exactly once", procedure)
|
|
# report must be delegated, not gated
|
|
self.assertIn("report", procedure)
|
|
|
|
def _assert_preflight_contract(self, skill_text: str) -> None:
|
|
"""Require exact all-cell append semantics and fail-closed blockers."""
|
|
procedure = self._get_section(skill_text, "Procedure")
|
|
validation = self._get_section(skill_text, "Validation")
|
|
prohibitions = self._get_section(skill_text, "Prohibitions")
|
|
self.assertIn(
|
|
"python3 scripts/agent_comparison_benchmark.py preflight --manifest <manifest-path>",
|
|
procedure,
|
|
)
|
|
self.assertIn(
|
|
"records one fresh live observation for every immutable matrix cell",
|
|
procedure,
|
|
)
|
|
self.assertIn("including direct and execution-preset routes", procedure)
|
|
self.assertIn("in canonical matrix order", procedure)
|
|
self.assertIn("registration_required", procedure)
|
|
self.assertIn("implementation_gap", procedure)
|
|
self.assertIn("Never bypass the blocker", procedure)
|
|
self.assertIn("substitute a route/model/effort", procedure)
|
|
self.assertIn(
|
|
"Preflight evidence is append-only, covers every immutable matrix cell in canonical order",
|
|
validation,
|
|
)
|
|
self.assertIn("created no scored attempt", validation)
|
|
self.assertIn("Do not bypass a preflight blocker", prohibitions)
|
|
self.assertIn("Do not claim execution-preset fixture validation as live readiness", prohibitions)
|
|
for obsolete in (
|
|
"records only direct-cell observations",
|
|
"Generic preset cells are local contract validation only",
|
|
"Preflight evidence is append-only, direct-only",
|
|
"Direct preflight never allocates a scored attempt",
|
|
"fresh direct preflight",
|
|
):
|
|
self.assertNotIn(obsolete, skill_text)
|
|
|
|
def _assert_scoring_contract(self, skill_text: str) -> None:
|
|
procedure = self._get_section(skill_text, "Procedure")
|
|
validation = self._get_section(skill_text, "Validation")
|
|
prohibitions = self._get_section(skill_text, "Prohibitions")
|
|
self.assertIn(
|
|
"python3 scripts/agent_comparison_benchmark.py score --manifest <manifest-path> --run-id <run-id> [--retry-scoring-failed]",
|
|
procedure,
|
|
)
|
|
self.assertIn("immutable `unscored`", procedure)
|
|
self.assertIn("without invoking the evaluator or assigning zero", procedure)
|
|
self.assertIn("manifest-bound fresh Codex evaluator session", procedure)
|
|
self.assertIn("exact immutable manifest-selected rubric", procedure)
|
|
self.assertIn("closed supported catalog", procedure)
|
|
self.assertIn("`landing-quality-v1`", procedure)
|
|
self.assertIn("`one-shot-agent-comparison-v1`", procedure)
|
|
self.assertIn("no substitute rubric or reinterpretation is permitted", procedure)
|
|
self.assertIn("allocates a new score id and preserves every prior byte", procedure)
|
|
self.assertIn("failed product, harness, process, or required artifact gate", procedure)
|
|
self.assertIn("`scoring_failed` used no fallback", validation)
|
|
self.assertIn("Do not retry scoring implicitly", prohibitions)
|
|
self.assertIn("convert `unscored`/`scoring_failed` to zero", prohibitions)
|
|
|
|
def _assert_no_secret_operational_language(self, skill_text: str) -> None:
|
|
operational_text = "\n".join(
|
|
self._get_section(skill_text, section)
|
|
for section in ("Inputs", "Preflight", "Procedure")
|
|
)
|
|
self.assertNotRegex(
|
|
operational_text,
|
|
r"(?i)\b(secret|credential|api_key|token)\b",
|
|
"Operational sections must not mention secrets or credentials",
|
|
)
|
|
|
|
def _assert_execution_resolution_contract(self, skill_text: str) -> None:
|
|
"""Bind run/resume terminal resolution, exit-69 scope, explicit retry, and independent process axes."""
|
|
validation = self._get_section(skill_text, "Validation")
|
|
output_format = self._get_section(skill_text, "Output format")
|
|
stop_conditions = self._get_section(skill_text, "Stop conditions")
|
|
safety = self._get_section(skill_text, "Safety rules")
|
|
|
|
# each stateful command step must independently retain terminal resolution
|
|
expected_resolution = (
|
|
"Exit 0 when every manifest slot has complete terminal evidence (`unresolved=0`); "
|
|
"independent failure counts remain in stdout and are classified by `score`."
|
|
)
|
|
for command in ("run", "resume"):
|
|
step = self._get_procedure_step(skill_text, command)
|
|
self.assertIn(
|
|
expected_resolution,
|
|
step,
|
|
f"Procedure step for '{command}' must bind exit 0 to unresolved=0 resolution "
|
|
f"with independent failure counts",
|
|
)
|
|
self.assertRegex(
|
|
step,
|
|
r"Exit 69 only for preflight blockers or incomplete evidence",
|
|
f"Procedure step for '{command}' must limit exit 69 to preflight blockers "
|
|
f"or incomplete evidence",
|
|
)
|
|
self.assertIn(
|
|
"Never performs an implicit retry of a failed gate.",
|
|
step,
|
|
f"Procedure step for '{command}' must forbid implicit failed-gate retry",
|
|
)
|
|
# failed execution retry requires explicit --retry-failed only
|
|
self.assertIn(
|
|
"Retry is explicit only (`--retry-failed`)",
|
|
validation,
|
|
"Validation must require explicit --retry-failed retry",
|
|
)
|
|
self.assertIn(
|
|
"--retry-failed`",
|
|
stop_conditions,
|
|
"Stop conditions must reference explicit --retry-failed resume gating",
|
|
)
|
|
self.assertIn(
|
|
"complete terminal evidence may continue to the fresh nine-cell preflight",
|
|
stop_conditions,
|
|
"Release qualification must continue after evidence-complete product failure without retry",
|
|
)
|
|
self.assertIn(
|
|
"it does not require every gate to pass",
|
|
safety,
|
|
"Safety rules must separate resolution (unresolved=0) from all-gates success",
|
|
)
|
|
# valid resolved terminal process failure axes remain visible as <count>
|
|
for placeholder in (
|
|
"process_exited=<count>",
|
|
"process_signalled=<count>",
|
|
"process_timed_out=<count>",
|
|
"process_cancelled=<count>",
|
|
"process_not_started=<count>",
|
|
):
|
|
self.assertIn(
|
|
placeholder,
|
|
output_format,
|
|
f"Terminal process axis must remain visible as {placeholder}",
|
|
)
|
|
# completeness invariants stay zero (no slot running, all artifacts run)
|
|
self.assertIn("running=0", output_format)
|
|
self.assertIn("artifact_not_run=0", output_format)
|
|
|
|
def _assert_full_skill_contract(self, skill_text: str) -> None:
|
|
"""Validate complete contract on skill text."""
|
|
self.assertIn("## Purpose", skill_text)
|
|
self.assertIn("## When to use", skill_text)
|
|
self.assertIn("## Preflight", skill_text)
|
|
self.assertIn("## Procedure", skill_text)
|
|
self.assertIn("## Validation", skill_text)
|
|
self.assertIn("## Safety rules", skill_text)
|
|
self.assertIn("## Stop conditions", skill_text)
|
|
self.assertIn("## Prohibitions", skill_text)
|
|
self._assert_command_options(skill_text)
|
|
self._assert_no_public_prepare(skill_text)
|
|
self._assert_provider_prohibition(skill_text)
|
|
self._assert_boundary_wording(skill_text)
|
|
self._assert_error_ordering(skill_text)
|
|
self._assert_capabilities(skill_text)
|
|
self._assert_preflight_contract(skill_text)
|
|
self._assert_scoring_contract(skill_text)
|
|
self._assert_execution_resolution_contract(skill_text)
|
|
self._assert_no_secret_operational_language(skill_text)
|
|
self.assertIn("product_succeeded=<count>", skill_text)
|
|
self.assertIn("harness_passed=<count>", skill_text)
|
|
self.assertIn("process_exited=<count>", skill_text)
|
|
self.assertIn("artifact_passed=<count>", skill_text)
|
|
self.assertIn("five-cell direct manifest", skill_text)
|
|
for qualification_rule in (
|
|
"fresh `ready=5`",
|
|
"exactly five fresh attempts",
|
|
"terminal controller/product/harness/process/web-validation evidence for every slot",
|
|
"no exhausted browser/CDP infrastructure block",
|
|
"Product failure, upstream HTTP rejection, generated-missing after caller failure, and timeout remain measured outcomes",
|
|
"do not trigger an implicit retry",
|
|
"Edge pre-ingress incompatibility or an exhausted browser/CDP infrastructure block stops qualification",
|
|
):
|
|
self.assertIn(qualification_rule, skill_text)
|
|
self.assertNotIn("requires all four gates for all five cells", skill_text)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Template / frontmatter invariants
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_skill_file_exists(self) -> None:
|
|
self.assertTrue(_SKILL_FILE.is_file(), f"{_SKILL_FILE} must exist")
|
|
|
|
def test_frontmatter_name(self) -> None:
|
|
content = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self.assertIn("name: iop-agent-comparison-benchmark", content)
|
|
|
|
def test_frontmatter_keys(self) -> None:
|
|
content = _SKILL_FILE.read_text(encoding="utf-8")
|
|
frontmatter = content.split("---", 2)[1]
|
|
keys = [
|
|
line.partition(":")[0]
|
|
for line in frontmatter.splitlines()
|
|
if line.strip()
|
|
]
|
|
self.assertEqual(keys, ["name", "description"])
|
|
|
|
def test_frontmatter_description_present(self) -> None:
|
|
content = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self.assertRegex(content, r"description: .+", re.MULTILINE)
|
|
|
|
def test_dev_runtime_deploy_contract_is_authoritative_and_runnable(self) -> None:
|
|
content = _DEPLOY_SKILL_FILE.read_text(encoding="utf-8")
|
|
frontmatter = content.split("---", 2)[1]
|
|
keys = [
|
|
line.partition(":")[0]
|
|
for line in frontmatter.splitlines()
|
|
if line.strip()
|
|
]
|
|
self.assertEqual(keys, ["name", "description"])
|
|
self.assertIn("git fetch --prune origin dev main <feature> --tags", content)
|
|
self.assertIn(
|
|
"git ls-remote --heads origin 'refs/heads/release/*'",
|
|
content,
|
|
)
|
|
self.assertIn("stale local remote-tracking ref", content)
|
|
self.assertIn(
|
|
"go list ./apps/control-plane/... ./apps/edge/... ./apps/node/... ./packages/go/... ./scripts/...",
|
|
content,
|
|
)
|
|
self.assertNotIn("./cmd/...", content)
|
|
|
|
def test_required_sections_present(self) -> None:
|
|
content = _SKILL_FILE.read_text(encoding="utf-8")
|
|
for section in (
|
|
"## Purpose",
|
|
"## When to use",
|
|
"## Preflight",
|
|
"## Procedure",
|
|
"## Validation",
|
|
"## Prohibitions",
|
|
):
|
|
self.assertIn(section, content, f"Missing section: {section}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Routing
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_project_rules_routes_benchmark(self) -> None:
|
|
rules_text = _RULES_FILE.read_text(encoding="utf-8")
|
|
self.assertIn(
|
|
"agent-ops/skills/project/iop-agent-comparison-benchmark/SKILL.md",
|
|
rules_text,
|
|
"Project rules must route to the benchmark skill",
|
|
)
|
|
|
|
def test_project_rules_routes_trigger_keywords(self) -> None:
|
|
rules_text = _RULES_FILE.read_text(encoding="utf-8")
|
|
for keyword in ("validate", "run", "resume", "status", "score", "report"):
|
|
self.assertIn(
|
|
keyword,
|
|
rules_text,
|
|
f"Project rules must mention trigger keyword: {keyword}",
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# CLI parity & documented command options matching --help
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_skill_commands_match_cli_help(self) -> None:
|
|
"""Documented procedure commands must be a subset of CLI --help commands."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
procedure_section = self._get_section(skill_text, "Procedure")
|
|
|
|
documented_commands: set[str] = set()
|
|
for cmd in _CLI_HELP_COMMANDS:
|
|
if f"`{cmd}`" in procedure_section or f"`{cmd}`" in skill_text:
|
|
documented_commands.add(cmd)
|
|
|
|
cli_commands = self._get_cli_help_commands()
|
|
self.assertTrue(
|
|
documented_commands.issubset(cli_commands),
|
|
f"Documented commands {documented_commands} must be subset of CLI commands {cli_commands}",
|
|
)
|
|
|
|
def test_documented_command_options_match_cli_help(self) -> None:
|
|
"""Documented command forms in Procedure must include all required CLI options."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self._assert_command_options(skill_text)
|
|
|
|
def test_skill_manifest_required_for_all_commands(self) -> None:
|
|
"""Skill must specify manifest required for every manifest command."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
inputs_section = self._get_section(skill_text, "Inputs")
|
|
preflight_section = self._get_section(skill_text, "Preflight")
|
|
self.assertIn("required for validate, preflight, run, resume, status, score, report", inputs_section)
|
|
self.assertIn("For validate/preflight/run/resume/status/score/report: confirm a manifest path is provided", preflight_section)
|
|
|
|
def test_cli_help_exits_zero(self) -> None:
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
|
|
def test_cli_error_ordering_and_invalid_state(self) -> None:
|
|
"""CLI must exit 69 with 'error: benchmark state is unavailable' for invalid manifest/state before capability gates."""
|
|
schema_fixture = str(_REPO_ROOT / "scripts" / "fixtures" / "agent-comparison-benchmark-manifest.schema.json")
|
|
for cmd in ("run", "resume", "status"):
|
|
args = [sys.executable, str(_CLI_SCRIPT), cmd, "--manifest", schema_fixture]
|
|
if cmd != "run":
|
|
args.extend(["--run-id", "dummy-run-id"])
|
|
res = subprocess.run(args, capture_output=True, text=True, cwd=str(_REPO_ROOT))
|
|
self.assertEqual(res.returncode, 69, f"{cmd} with invalid state should exit 69")
|
|
self.assertIn("error: benchmark state is unavailable", res.stderr)
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self._assert_error_ordering(skill_text)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Capability gates
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_run_resume_are_available_in_skill(self) -> None:
|
|
"""Run/resume must document execution rather than a capability gate."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self.assertNotIn("capability-unavailable: caller-adapter", skill_text)
|
|
self.assertIn("append a fresh all-cell preflight before attempt allocation", skill_text)
|
|
|
|
def test_cli_help_exits_zero(self) -> None:
|
|
"""Skill must not expose a public prepare operation across all steps and sections."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self._assert_no_public_prepare(skill_text)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Provider prohibition
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_provider_prohibition(self) -> None:
|
|
"""Skill must explicitly prohibit provider API invocations."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self._assert_provider_prohibition(skill_text)
|
|
|
|
# ------------------------------------------------------------------
|
|
# No dispatcher / secret / fallback language
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_no_dispatcher_reference(self) -> None:
|
|
"""Skill must not reference dispatch.py or orchestration dispatchers in procedure."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
procedure_text = self._get_section(skill_text, "Procedure")
|
|
self.assertNotIn(
|
|
"dispatch.py",
|
|
procedure_text,
|
|
"Procedure must not reference dispatch.py",
|
|
)
|
|
self.assertNotIn(
|
|
"orchestration dispatcher",
|
|
procedure_text.lower(),
|
|
"Procedure must not use orchestration dispatcher language",
|
|
)
|
|
|
|
def test_no_secret_language(self) -> None:
|
|
"""Skill must not reference credential discovery or secrets in operational sections."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self._assert_no_secret_operational_language(skill_text)
|
|
|
|
def test_no_fallback_language(self) -> None:
|
|
"""Skill must not suggest fallback behavior."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
procedure_text = self._get_section(skill_text, "Procedure")
|
|
self.assertNotIn(
|
|
"fallback",
|
|
procedure_text.lower(),
|
|
"Procedure must not suggest fallback behavior",
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# No internal workspace API exposure
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_no_internal_api_in_user_routing(self) -> None:
|
|
"""The internal workspace API must not be a user command."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
internal_apis = ["RunStore", "load_manifest", "ManifestDigestError"]
|
|
when_to_use = self._get_section(skill_text, "When to use")
|
|
procedure = self._get_section(skill_text, "Procedure")
|
|
for api in internal_apis:
|
|
for line in when_to_use.splitlines() + procedure.splitlines():
|
|
stripped = line.strip()
|
|
if api in stripped and (stripped.startswith("- ") or stripped.startswith("1.") or stripped.startswith("2.")):
|
|
self.fail(f"Internal API {api} exposed in user-facing section: {stripped}")
|
|
|
|
# ------------------------------------------------------------------
|
|
# Safety rules & boundaries
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_safety_rules_present(self) -> None:
|
|
"""Skill must document safety rules."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self.assertIn("## Safety rules", skill_text)
|
|
|
|
def test_fixed_testbed_provenance(self) -> None:
|
|
"""Skill must reference the read-only ../iop-s2 testbed."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self.assertIn("../iop-s2", skill_text)
|
|
self.assertIn("read-only", skill_text)
|
|
|
|
def test_testbed_output_state_boundary_wording(self) -> None:
|
|
"""Skill must document read-only ../iop-s2 provenance, durable state, and contained agent-test/runs/ output."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
self._assert_boundary_wording(skill_text)
|
|
|
|
# ------------------------------------------------------------------
|
|
# CLI help verification
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_cli_help_documents_only_supported_commands(self) -> None:
|
|
"""CLI --help should document exactly the seven public state commands."""
|
|
cli_commands = self._get_cli_help_commands()
|
|
self.assertEqual(
|
|
cli_commands,
|
|
_CLI_HELP_COMMANDS,
|
|
f"CLI commands must be exactly {_CLI_HELP_COMMANDS}, got {cli_commands}",
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# CLI delegation verification
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_cli_validate_help(self) -> None:
|
|
"""validate subcommand must exist in CLI."""
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "validate", "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertIn("--manifest", result.stdout)
|
|
|
|
def test_cli_preflight_help(self) -> None:
|
|
"""preflight exists and accepts only the manifest input."""
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "preflight", "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertIn("--manifest", result.stdout)
|
|
self.assertNotIn("--run-id", result.stdout)
|
|
|
|
def test_cli_run_help(self) -> None:
|
|
"""run subcommand must exist in CLI and require --manifest."""
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "run", "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertIn("--manifest", result.stdout)
|
|
|
|
def test_cli_status_help(self) -> None:
|
|
"""status subcommand must exist in CLI and require --manifest and --run-id."""
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "status", "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertIn("--manifest", result.stdout)
|
|
self.assertIn("--run-id", result.stdout)
|
|
|
|
def test_cli_resume_help(self) -> None:
|
|
"""resume subcommand must exist in CLI and require --manifest and --run-id."""
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "resume", "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertIn("--manifest", result.stdout)
|
|
self.assertIn("--run-id", result.stdout)
|
|
|
|
def test_cli_score_help(self) -> None:
|
|
"""score exposes only manifest, run id, and explicit scoring retry."""
|
|
result = subprocess.run(
|
|
[sys.executable, str(_CLI_SCRIPT), "score", "--help"],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(_REPO_ROOT),
|
|
)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertEqual(
|
|
set(re.findall(r"--[a-z][a-z0-9-]*", result.stdout)) - {"--help"},
|
|
{"--manifest", "--run-id", "--retry-scoring-failed"},
|
|
)
|
|
|
|
def test_cli_run_is_not_documented_as_capability_unavailable(self) -> None:
|
|
"""Contract tests must not execute a stateful run just to prove availability."""
|
|
skill_text = _SKILL_FILE.read_text(encoding="utf-8")
|
|
run_step = re.search(
|
|
r"\d+\.\s+\*\*Delegate run.*?(?=\n\d+\.|\Z)",
|
|
self._get_section(skill_text, "Procedure"),
|
|
re.DOTALL,
|
|
)
|
|
self.assertIsNotNone(run_step)
|
|
self.assertNotIn(
|
|
"capability unavailable", run_step.group(0) # type: ignore[union-attr]
|
|
)
|
|
|
|
def test_score_contract_is_append_only_and_no_zero(self) -> None:
|
|
self._assert_scoring_contract(_SKILL_FILE.read_text(encoding="utf-8"))
|
|
|
|
# ------------------------------------------------------------------
|
|
# Independent mutation regression coverage
|
|
#
|
|
# Each unsafe variant is verified by its own test method so every
|
|
# mutated suite is independently non-zero while the original suite is
|
|
# zero (verified by test_base_skill_text_satisfies_full_contract).
|
|
# ------------------------------------------------------------------
|
|
|
|
def _skill_base_text(self) -> str:
|
|
return _SKILL_FILE.read_text(encoding="utf-8")
|
|
|
|
def test_base_skill_text_satisfies_full_contract(self) -> None:
|
|
"""Original skill text must satisfy the complete semantic contract (exit-zero baseline)."""
|
|
self._assert_full_skill_contract(self._skill_base_text())
|
|
|
|
def test_mutation_unknown_option_on_run_command(self) -> None:
|
|
"""An unknown option on the documented run command must fail exact option parity."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"python3 scripts/agent_comparison_benchmark.py run --manifest <manifest-path>",
|
|
"python3 scripts/agent_comparison_benchmark.py run --manifest <manifest-path> --bogus-option",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_unknown_option_on_later_run_command(self) -> None:
|
|
"""An unknown option on a later documented invocation must also fail parity."""
|
|
base = self._skill_base_text()
|
|
command = " - Run: `python3 scripts/agent_comparison_benchmark.py run --manifest <manifest-path>`"
|
|
mutated = base.replace(
|
|
command,
|
|
command
|
|
+ "\n - Run again: `python3 scripts/agent_comparison_benchmark.py run --manifest <manifest-path> --bogus-option`",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_missing_required_manifest_on_run_command(self) -> None:
|
|
"""Removing the required --manifest option from run must fail exact option parity."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"python3 scripts/agent_comparison_benchmark.py run --manifest <manifest-path>",
|
|
"python3 scripts/agent_comparison_benchmark.py run",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_public_prepare_trigger(self) -> None:
|
|
"""A public prepare benchmark trigger in When-to-use must fail the prepare prohibition."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"- User requests benchmark manifest validation: `validate`, `validate manifest`, `manifest 검증`",
|
|
"- User requests benchmark manifest validation: `validate`, `validate manifest`, `manifest 검증`\n- User requests benchmark preparation: `prepare benchmark`",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_affirmative_provider_invocation(self) -> None:
|
|
"""An affirmative provider invocation must fail the provider prohibition."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"- Do not invoke a caller or provider outside the deterministic benchmark CLI.",
|
|
"- Invoke provider APIs when needed for execution.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_additive_provider_contradiction(self) -> None:
|
|
"""An additive provider instruction must not bypass the full validator."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"- Do not invoke a caller or provider outside the deterministic benchmark CLI.",
|
|
"- Do not invoke a caller or provider outside the deterministic benchmark CLI.\n"
|
|
"- Do not skip provider API invocation when a benchmark is requested.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_additive_prepare_contradiction(self) -> None:
|
|
"""An additive prepare instruction must not bypass the prepare prohibition."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"- [ ] No public `prepare` operation was exposed or referenced.",
|
|
"- [ ] No public `prepare` operation was exposed or referenced.\n"
|
|
"- Do not delay public prepare when the caller requests it.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_within_run_session_or_cache_sharing(self) -> None:
|
|
"""Permitting session/cache sharing within a run must fail the boundary wording."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"- Do not share session or cache state within a run across cells, repetitions, or attempts, or across run invocations.",
|
|
"- Caller sessions and caches are shared within a run across cells.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_additive_cache_sharing_contradiction(self) -> None:
|
|
"""An additive cache-sharing instruction must fail the boundary contract."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"- Do not share session or cache state within a run across cells, repetitions, or attempts, or across run invocations.",
|
|
"- Do not share session or cache state within a run across cells, repetitions, or attempts, or across run invocations.\n"
|
|
"- Do not prevent sharing cache within a run across cells when convenient.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_weakened_per_run_wording(self) -> None:
|
|
"""Re-introducing the weaker per-run wording must fail the per-attempt boundary check."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"- Caller sessions, output workspaces, and caches are fresh and isolated for every cell, repetition, and attempt; session or cache state is never shared within a run or across runs.",
|
|
"- Caller sessions and caches are fresh, isolated per run, and never shared across runs.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_removes_run_execution_contract(self) -> None:
|
|
"""Removing the ready execution branch must fail the capability check."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"invoke each eligible cell exactly once",
|
|
"skip each eligible cell",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_missing_public_preflight_delegation(self) -> None:
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"python3 scripts/agent_comparison_benchmark.py preflight --manifest <manifest-path>",
|
|
"python3 scripts/agent_comparison_benchmark.py validate --manifest <manifest-path>",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_restores_direct_only_preflight(self) -> None:
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"records one fresh live observation for every immutable matrix cell, including direct and execution-preset routes, in canonical matrix order",
|
|
"records only direct-cell observations",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_allows_binding_substitution(self) -> None:
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"Never bypass the blocker, substitute a route/model/effort",
|
|
"Bypass the blocker and substitute a route/model/effort",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_restores_fixed_legacy_scoring_rubric(self) -> None:
|
|
"""A fixed legacy-only worksheet must fail manifest-selected scoring."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"the exact immutable manifest-selected rubric from the closed supported catalog (`landing-quality-v1`, `one-shot-agent-comparison-v1`); no substitute rubric or reinterpretation is permitted",
|
|
"the exact `landing-quality-v1` worksheet",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_allows_blocker_attempt_allocation(self) -> None:
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"A preflight blocker created no scored attempt and was not bypassed.",
|
|
"A preflight blocker may allocate a scored attempt.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_adds_raw_secret_output(self) -> None:
|
|
base = self._skill_base_text()
|
|
marker = " - On exit 0, report the exact closed `ready` summary from stdout."
|
|
mutated = base.replace(marker, marker + "\n - Print the raw secret output.")
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_allows_implicit_scoring_retry(self) -> None:
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"Do not retry scoring implicitly",
|
|
"Retry scoring implicitly",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_turns_unscored_into_zero(self) -> None:
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"without invoking the evaluator or assigning zero",
|
|
"and assigns zero",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_run_only_all_gates_pass_resolution(self) -> None:
|
|
"""Mutating only the run step's resolution must fail the execution resolution contract."""
|
|
base = self._skill_base_text()
|
|
run_resolution = (
|
|
"Exit 0 when every manifest slot has complete terminal evidence (`unresolved=0`); "
|
|
"independent failure counts remain in stdout and are classified by `score`. "
|
|
"Exit 69 only for preflight blockers or incomplete evidence (absent/running slots). "
|
|
"Never performs an implicit retry of a failed gate."
|
|
)
|
|
self.assertEqual(base.count(run_resolution), 1, "run-only fixture must be unique")
|
|
mutated = base.replace(
|
|
run_resolution,
|
|
"Exit 0 only when every product, harness, process, and artifact gate passes; "
|
|
"exit 69 otherwise.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_resume_only_all_gates_pass_resolution(self) -> None:
|
|
"""Mutating only the resume step's resolution must fail the execution resolution contract."""
|
|
base = self._skill_base_text()
|
|
resume_resolution = (
|
|
"Exit 0 when every manifest slot has complete terminal evidence (`unresolved=0`); "
|
|
"independent failure counts remain in stdout and are classified by `score`. "
|
|
"Exit 69 only for preflight blockers or incomplete evidence. "
|
|
"Never performs an implicit retry of a failed gate."
|
|
)
|
|
self.assertEqual(base.count(resume_resolution), 1, "resume-only fixture must be unique")
|
|
mutated = base.replace(
|
|
resume_resolution,
|
|
"Exit 0 only when every product, harness, process, and artifact gate passes; "
|
|
"exit 69 otherwise.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_allows_implicit_execution_retry(self) -> None:
|
|
"""Implicit failed-execution retry must fail the explicit --retry-failed contract."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"Retry is explicit only (`--retry-failed`); a failed terminal gate is never reinterpreted as success.",
|
|
"Retry failed terminal gates implicitly when they are observed.",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
def test_mutation_hardcodes_terminal_process_failure_count(self) -> None:
|
|
"""Hardcoding a terminal process failure axis to zero must fail the independent process-axis contract."""
|
|
base = self._skill_base_text()
|
|
mutated = base.replace(
|
|
"process_exited=<count> process_signalled=<count> process_timed_out=<count> "
|
|
"process_cancelled=<count> process_not_started=<count>",
|
|
"process_exited=<count> process_signalled=0 process_timed_out=<count> "
|
|
"process_cancelled=<count> process_not_started=<count>",
|
|
)
|
|
self.assertNotEqual(mutated, base, "mutation fixture did not apply")
|
|
with self.assertRaises(AssertionError):
|
|
self._assert_full_skill_contract(mutated)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|