modify harness framework

This commit is contained in:
KOKO\Mimi
2026-08-05 01:42:21 +09:00
parent 6646344113
commit 41020d78d8
74 changed files with 2663 additions and 3145 deletions
+104 -185
View File
@@ -8,10 +8,8 @@ Usage:
import argparse
import contextlib
import fnmatch
import json
import os
import re
import subprocess
import sys
import threading
@@ -24,6 +22,10 @@ from typing import Optional
ROOT = Path(__file__).resolve().parent.parent
class CodexEnvironmentError(RuntimeError):
"""재시도로 해결할 수 없는 Codex CLI 환경 오류."""
@contextlib.contextmanager
def progress_indicator(label: str):
"""터미널 진행 표시기. with 문으로 사용하며 .elapsed 로 경과 시간을 읽는다."""
@@ -56,10 +58,6 @@ class StepExecutor:
"""Phase 디렉토리 안의 step들을 순차 실행하는 하네스."""
MAX_RETRIES = 3
VALIDATION_COMMANDS = (
[sys.executable, "-m", "unittest", "discover", "-s", "scripts", "-p", "test_*.py"],
[sys.executable, "scripts/validate_workspace.py"],
)
FEAT_MSG = "feat({phase}): step {num}{name}"
CHORE_MSG = "chore({phase}): step {num} output"
TZ = timezone(timedelta(hours=9))
@@ -89,7 +87,6 @@ class StepExecutor:
def run(self):
self._print_header()
self._check_blockers()
self._assert_clean_worktree("before branch checkout")
self._checkout_branch()
guardrails = self._load_guardrails()
self._ensure_created_at()
@@ -117,117 +114,8 @@ class StepExecutor:
cmd = ["git"] + list(args)
return subprocess.run(cmd, cwd=self._root, capture_output=True, text=True)
def _validate_before_commit(self, commit_message: str):
print(f" Validation before commit: {commit_message}")
for cmd in self.VALIDATION_COMMANDS:
r = subprocess.run(cmd, cwd=self._root, capture_output=True, text=True)
if r.returncode != 0:
print(f" ERROR: validation failed before commit: {' '.join(cmd)}")
if r.stdout:
print(r.stdout[-2000:])
if r.stderr:
print(r.stderr[-2000:])
sys.exit(1)
def _branch_name(self) -> str:
slug = re.sub(r"[^A-Za-z0-9._-]+", "-", self._phase_name.strip())
slug = slug.strip("/.-")
if not slug:
slug = self._phase_dir_name
return f"codex/{slug}"
def _assert_clean_worktree(self, context: str):
r = self._run_git("status", "--porcelain")
if r.returncode != 0:
print(" ERROR: git status failed.")
print(f" {r.stderr.strip()}")
sys.exit(1)
dirty = r.stdout.strip()
if dirty:
print(f" ERROR: dirty worktree detected {context}.")
print(" Commit, stash, or remove these changes before running scripts/execute.py:")
for line in dirty.splitlines():
print(f" {line}")
sys.exit(1)
@staticmethod
def _normalize_rel_path(path: str) -> str:
return path.replace("\\", "/").lstrip("./")
def _path_allowed(self, path: str, patterns: list[str]) -> bool:
rel = self._normalize_rel_path(path)
for raw in patterns:
pattern = self._normalize_rel_path(str(raw))
if not pattern:
continue
if pattern.endswith("/") and rel.startswith(pattern):
return True
if any(ch in pattern for ch in "*?[") and fnmatch.fnmatchcase(rel, pattern):
return True
if rel == pattern:
return True
return False
def _validate_step_allowlist(self, step: dict):
allowed = step.get("allowed_paths")
if (
not isinstance(allowed, list)
or not allowed
or not all(isinstance(p, str) and p.strip() for p in allowed)
):
print(f" ERROR: Step {step.get('step')} must define non-empty allowed_paths.")
sys.exit(1)
def _changed_paths(self) -> list[str]:
paths: list[str] = []
tracked = self._run_git("diff", "--name-only")
if tracked.returncode != 0:
print(" ERROR: git diff --name-only failed.")
print(f" {tracked.stderr.strip()}")
sys.exit(1)
paths.extend(tracked.stdout.splitlines())
staged = self._run_git("diff", "--cached", "--name-only")
if staged.returncode != 0:
print(" ERROR: git diff --cached --name-only failed.")
print(f" {staged.stderr.strip()}")
sys.exit(1)
paths.extend(staged.stdout.splitlines())
untracked = self._run_git("ls-files", "--others", "--exclude-standard")
if untracked.returncode != 0:
print(" ERROR: git ls-files --others failed.")
print(f" {untracked.stderr.strip()}")
sys.exit(1)
paths.extend(untracked.stdout.splitlines())
return sorted({self._normalize_rel_path(p) for p in paths if p.strip()})
def _housekeeping_paths(self, step_num: int) -> set[str]:
return {
f"phases/{self._phase_dir_name}/index.json",
f"phases/{self._phase_dir_name}/step{step_num}-output.json",
"phases/index.json",
}
def _classify_step_changes(self, step_num: int, step: dict, changed_paths: list[str]) -> tuple[list[str], list[str], list[str]]:
allowed_patterns = step.get("allowed_paths", [])
housekeeping_set = self._housekeeping_paths(step_num)
allowed: list[str] = []
housekeeping: list[str] = []
disallowed: list[str] = []
for path in changed_paths:
rel = self._normalize_rel_path(path)
if rel in housekeeping_set:
housekeeping.append(rel)
elif self._path_allowed(rel, allowed_patterns):
allowed.append(rel)
else:
disallowed.append(rel)
return allowed, housekeeping, disallowed
def _checkout_branch(self):
branch = self._branch_name()
branch = f"feat-{self._phase_name}"
r = self._run_git("rev-parse", "--abbrev-ref", "HEAD")
if r.returncode != 0:
@@ -249,45 +137,28 @@ class StepExecutor:
print(f" Branch: {branch}")
def _stage_paths(self, paths: list[str]):
if not paths:
return
r = self._run_git("add", "--", *paths)
if r.returncode != 0:
print(" ERROR: git add failed.")
print(f" {r.stderr.strip()}")
sys.exit(1)
def _commit_step(self, step_num: int, step_name: str):
output_rel = f"phases/{self._phase_dir_name}/step{step_num}-output.json"
index_rel = f"phases/{self._phase_dir_name}/index.json"
def _commit_step(self, step: dict, step_name: str):
step_num = step["step"]
changed = self._changed_paths()
allowed, housekeeping, disallowed = self._classify_step_changes(step_num, step, changed)
if disallowed:
print(f" ERROR: Step {step_num} modified files outside allowed_paths:")
for path in disallowed:
print(f" {path}")
sys.exit(1)
self._run_git("add", "-A")
self._run_git("reset", "HEAD", "--", output_rel)
self._run_git("reset", "HEAD", "--", index_rel)
if allowed:
if self._run_git("diff", "--cached", "--quiet").returncode != 0:
msg = self.FEAT_MSG.format(phase=self._phase_name, num=step_num, name=step_name)
self._validate_before_commit(msg)
self._stage_paths(allowed)
if self._run_git("diff", "--cached", "--quiet").returncode != 0:
r = self._run_git("commit", "-m", msg)
if r.returncode != 0:
print(f" ERROR: code commit failed: {r.stderr.strip()}")
sys.exit(1)
r = self._run_git("commit", "-m", msg)
if r.returncode == 0:
print(f" Commit: {msg}")
else:
print(f" WARN: 코드 커밋 실패: {r.stderr.strip()}")
if housekeeping:
self._run_git("add", "-A")
if self._run_git("diff", "--cached", "--quiet").returncode != 0:
msg = self.CHORE_MSG.format(phase=self._phase_name, num=step_num)
self._validate_before_commit(msg)
self._stage_paths(housekeeping)
if self._run_git("diff", "--cached", "--quiet").returncode != 0:
r = self._run_git("commit", "-m", msg)
if r.returncode != 0:
print(f" ERROR: housekeeping commit failed: {r.stderr.strip()}")
sys.exit(1)
r = self._run_git("commit", "-m", msg)
if r.returncode != 0:
print(f" WARN: housekeeping 커밋 실패: {r.stderr.strip()}")
# --- top-level index ---
@@ -311,11 +182,16 @@ class StepExecutor:
sections = []
agents_md = ROOT / "AGENTS.md"
if agents_md.exists():
sections.append(f"## 프로젝트 규칙 (AGENTS.md)\n\n{agents_md.read_text(encoding='utf-8')}")
sections.append(
"## 프로젝트 규칙 (AGENTS.md)\n\n"
f"{agents_md.read_text(encoding='utf-8')}"
)
docs_dir = ROOT / "docs"
if docs_dir.is_dir():
for doc in sorted(docs_dir.glob("*.md")):
sections.append(f"## {doc.stem}\n\n{doc.read_text(encoding='utf-8')}")
sections.append(
f"## {doc.stem}\n\n{doc.read_text(encoding='utf-8')}"
)
return "\n\n---\n\n".join(sections) if sections else ""
@staticmethod
@@ -330,11 +206,7 @@ class StepExecutor:
return "## 이전 Step 산출물\n\n" + "\n".join(lines) + "\n\n"
def _build_preamble(self, guardrails: str, step_context: str,
allowed_paths: list[str],
prev_error: Optional[str] = None) -> str:
commit_example = self.FEAT_MSG.format(
phase=self._phase_name, num="N", name="<step-name>"
)
retry_section = ""
if prev_error:
retry_section = (
@@ -345,9 +217,6 @@ class StepExecutor:
f"당신은 {self._project} 프로젝트의 개발자입니다. 아래 step을 수행하세요.\n\n"
f"{guardrails}\n\n---\n\n"
f"{step_context}{retry_section}"
f"## Step file allowlist\n\n"
f"This step may modify only these repository-relative paths:\n"
f"{chr(10).join(f'- {p}' for p in allowed_paths)}\n\n"
f"## 작업 규칙\n\n"
f"1. 이전 step에서 작성된 코드를 확인하고 일관성을 유지하라.\n"
f"2. 이 step에 명시된 작업만 수행하라. 추가 기능이나 파일을 만들지 마라.\n"
@@ -357,8 +226,8 @@ class StepExecutor:
f" - AC 통과 → \"completed\" + \"summary\" 필드에 이 step의 산출물을 한 줄로 요약\n"
f" - {self.MAX_RETRIES}회 수정 시도 후에도 실패 → \"error\" + \"error_message\" 기록\n"
f" - 사용자 개입이 필요한 경우 (API 키, 인증, 수동 설정 등) → \"blocked\" + \"blocked_reason\" 기록 후 즉시 중단\n"
f"6. 모든 변경사항을 커밋하라:\n"
f" {commit_example}\n\n---\n\n"
f"6. 변경사항을 직접 커밋하지 마라. Git 커밋과 timestamp는 실행기가 처리한다.\n\n"
f"---\n\n"
)
# --- Codex 호출 ---
@@ -372,10 +241,30 @@ class StepExecutor:
sys.exit(1)
prompt = preamble + step_file.read_text(encoding="utf-8")
result = subprocess.run(
["codex", "exec", "--dangerously-bypass-approvals-and-sandbox", "--json", prompt],
cwd=self._root, capture_output=True, text=True, timeout=1800,
)
command = [
"codex",
"exec",
"--json",
"--sandbox",
"workspace-write",
"--dangerously-bypass-hook-trust",
"--cd",
self._root,
"-",
]
try:
result = subprocess.run(
command,
cwd=self._root,
input=prompt,
capture_output=True,
text=True,
timeout=1800,
)
except FileNotFoundError as exc:
raise CodexEnvironmentError(
"Codex CLI를 찾을 수 없습니다. codex를 설치하고 PATH를 확인하세요."
) from exc
if result.returncode != 0:
print(f"\n WARN: Codex가 비정상 종료됨 (code {result.returncode})")
@@ -388,11 +277,28 @@ class StepExecutor:
"stdout": result.stdout, "stderr": result.stderr,
}
out_path = self._phase_dir / f"step{step_num}-output.json"
with open(out_path, "w", encoding="utf-8") as f:
json.dump(output, f, indent=2, ensure_ascii=False)
self._write_json(out_path, output)
return output
@staticmethod
def _codex_environment_failure(output: dict) -> Optional[str]:
if output.get("exitCode", 0) == 0:
return None
diagnostic = (
f"{output.get('stderr', '')}\n{output.get('stdout', '')}"
).lower()
markers = {
"not logged in": "Codex 인증이 필요합니다.",
"authentication": "Codex 인증에 실패했습니다.",
"unexpected argument": "현재 Codex CLI가 필요한 옵션을 지원하지 않습니다.",
"unrecognized option": "현재 Codex CLI가 필요한 옵션을 지원하지 않습니다.",
}
for marker, message in markers.items():
if marker in diagnostic:
return message
return None
# --- 헤더 & 검증 ---
def _print_header(self):
@@ -436,16 +342,20 @@ class StepExecutor:
for attempt in range(1, self.MAX_RETRIES + 1):
index = self._read_json(self._index_file)
step_context = self._build_step_context(index)
preamble = self._build_preamble(guardrails, step_context, step.get("allowed_paths", []), prev_error)
preamble = self._build_preamble(guardrails, step_context, prev_error)
tag = f"Step {step_num}/{self._total - 1} ({done} done): {step_name}"
if attempt > 1:
tag += f" [retry {attempt}/{self.MAX_RETRIES}]"
with progress_indicator(tag) as pi:
self._invoke_codex(step, preamble)
output = self._invoke_codex(step, preamble)
elapsed = int(pi.elapsed)
environment_failure = self._codex_environment_failure(output)
if environment_failure:
raise CodexEnvironmentError(environment_failure)
index = self._read_json(self._index_file)
status = next((s.get("status", "pending") for s in index["steps"] if s["step"] == step_num), "pending")
ts = self._stamp()
@@ -455,7 +365,7 @@ class StepExecutor:
if s["step"] == step_num:
s["completed_at"] = ts
self._write_json(self._index_file, index)
self._commit_step(step, step_name)
self._commit_step(step_num, step_name)
print(f" ✓ Step {step_num}: {step_name} [{elapsed}s]")
return True
@@ -470,9 +380,22 @@ class StepExecutor:
self._update_top_index("blocked")
sys.exit(2)
runtime_error = ""
if output["exitCode"] != 0:
runtime_error = (
output.get("stderr", "").strip()
or output.get("stdout", "").strip()
)
err_msg = next(
(s.get("error_message", "Step did not update status") for s in index["steps"] if s["step"] == step_num),
"Step did not update status",
(
s.get(
"error_message",
runtime_error or "Step did not update status",
)
for s in index["steps"]
if s["step"] == step_num
),
runtime_error or "Step did not update status",
)
if attempt < self.MAX_RETRIES:
@@ -490,7 +413,7 @@ class StepExecutor:
s["error_message"] = f"[{self.MAX_RETRIES}회 시도 후 실패] {err_msg}"
s["failed_at"] = ts
self._write_json(self._index_file, index)
self._commit_step(step, step_name)
self._commit_step(step_num, step_name)
print(f" ✗ Step {step_num}: {step_name} failed after {self.MAX_RETRIES} attempts [{elapsed}s]")
print(f" Error: {err_msg}")
self._update_top_index("error")
@@ -506,7 +429,6 @@ class StepExecutor:
print("\n All steps completed!")
return
self._validate_step_allowlist(pending)
step_num = pending["step"]
for s in index["steps"]:
if s["step"] == step_num and "started_at" not in s:
@@ -522,22 +444,15 @@ class StepExecutor:
self._write_json(self._index_file, index)
self._update_top_index("completed")
final_paths = [f"phases/{self._phase_dir_name}/index.json"]
if self._top_index_file.exists():
final_paths.append("phases/index.json")
self._validate_before_commit(f"chore({self._phase_name}): mark phase completed")
self._stage_paths(final_paths)
self._run_git("add", "-A")
if self._run_git("diff", "--cached", "--quiet").returncode != 0:
msg = f"chore({self._phase_name}): mark phase completed"
r = self._run_git("commit", "-m", msg)
if r.returncode != 0:
print(f" ERROR: phase completion commit failed: {r.stderr.strip()}")
sys.exit(1)
else:
if r.returncode == 0:
print(f"{msg}")
if self._auto_push:
branch = self._branch_name()
branch = f"feat-{self._phase_name}"
r = self._run_git("push", "-u", "origin", branch)
if r.returncode != 0:
print(f"\n ERROR: git push 실패: {r.stderr.strip()}")
@@ -555,7 +470,11 @@ def main():
parser.add_argument("--push", action="store_true", help="Push branch after completion")
args = parser.parse_args()
StepExecutor(args.phase_dir, auto_push=args.push).run()
try:
StepExecutor(args.phase_dir, auto_push=args.push).run()
except CodexEnvironmentError as exc:
print(f"ERROR: {exc}")
sys.exit(1)
if __name__ == "__main__":
+118
View File
@@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""Codex PreToolUse 정책: 위험 명령과 테스트 없는 구현 파일 수정을 차단한다."""
import json
import re
import sys
from pathlib import Path
from typing import Any
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
from msvc_harness.config import ConfigError, load_config
from msvc_harness.tdd_policy import CPP_SUFFIXES, evaluate_paths
DANGEROUS_PATTERNS = (
re.compile(r"\bgit\s+reset\s+--hard\b", re.IGNORECASE),
re.compile(r"\bgit\s+push\b[^\n]*--force(?:-with-lease)?\b", re.IGNORECASE),
re.compile(r"\brm\s+-rf\b", re.IGNORECASE),
re.compile(
r"\bRemove-Item\b(?=[^\n]*-Recurse\b)(?=[^\n]*-Force\b)",
re.IGNORECASE,
),
re.compile(
r"\b(?:rmdir|rd)\b(?=[^\n]*/s\b)(?=[^\n]*/q\b)",
re.IGNORECASE,
),
re.compile(r"\bDROP\s+TABLE\b", re.IGNORECASE),
)
PATCH_PATH = re.compile(
r"^\*\*\* (?:(?:Add|Update|Delete) File:|Move to:) (?P<path>.+?)\s*$",
re.MULTILINE,
)
def _tool_input(payload: dict[str, Any]) -> dict[str, Any]:
value = payload.get("tool_input", {})
return value if isinstance(value, dict) else {}
def _candidate_paths(payload: dict[str, Any], root: Path) -> list[Path]:
tool_input = _tool_input(payload)
raw_paths: list[str] = []
for key in ("path", "file_path"):
value = tool_input.get(key)
if isinstance(value, str) and value.strip():
raw_paths.append(value.strip())
edits = tool_input.get("edits")
if isinstance(edits, list):
for edit in edits:
if not isinstance(edit, dict):
continue
value = edit.get("path")
if isinstance(value, str) and value.strip():
raw_paths.append(value.strip())
for key in ("patch", "input"):
value = tool_input.get(key)
if isinstance(value, str):
raw_paths.extend(match.group("path") for match in PATCH_PATH.finditer(value))
paths: list[Path] = []
for raw_path in raw_paths:
path = Path(raw_path)
paths.append(path.resolve() if path.is_absolute() else (root / path).resolve())
return list(dict.fromkeys(paths))
def evaluate(payload: dict[str, Any], root: Path) -> str | None:
"""Return a blocking reason, or None when the tool call is allowed."""
tool_name = str(payload.get("tool_name", ""))
tool_input = _tool_input(payload)
if tool_name in {"Bash", "shell_command", "PowerShell"}:
command = tool_input.get("command", "")
if isinstance(command, str) and any(
pattern.search(command) for pattern in DANGEROUS_PATTERNS
):
return "위험한 명령어가 감지되어 실행을 차단했습니다."
if tool_name not in {"apply_patch", "Edit", "MultiEdit", "Write"}:
return None
paths = _candidate_paths(payload, root)
try:
config = load_config(root)
except ConfigError as exc:
if any(path.suffix.lower() in CPP_SUFFIXES for path in paths):
return f"TDD GUARD: .harness/config.json must be repaired: {exc}"
return None
return evaluate_paths(paths, root, config.tdd)
def main() -> int:
try:
payload = json.load(sys.stdin)
if not isinstance(payload, dict):
raise TypeError("hook input must be a JSON object")
cwd = payload.get("cwd")
if cwd is not None and not isinstance(cwd, str):
raise TypeError("hook cwd must be a string")
except (json.JSONDecodeError, TypeError) as exc:
print(f"TDD GUARD: hook 입력을 해석하지 못해 검사를 건너뜁니다: {exc}", file=sys.stderr)
return 0
root = Path(payload.get("cwd") or Path.cwd()).resolve()
reason = evaluate(payload, root)
if reason:
print(reason, file=sys.stderr)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())
+148
View File
@@ -0,0 +1,148 @@
#!/usr/bin/env python3
"""Codex Stop hook: validate discovered C/C++ projects with MSVC."""
import json
import os
import subprocess
import sys
import time
from pathlib import Path
SCRIPTS = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(SCRIPTS))
from msvc_harness.adapters.base import AdapterError
from msvc_harness.adapters.cmake import CMakeAdapter
from msvc_harness.adapters.msbuild import MSBuildAdapter
from msvc_harness.config import ConfigError, load_config
from msvc_harness.discovery import DiscoveryError, discover_project
from msvc_harness.models import ProjectKind
from msvc_harness.process import ValidationFailure, execute_plan
from msvc_harness.toolchain import ToolchainError, discover_toolchain
REENTRY_ENV = "CODEX_STOP_VALIDATION_ACTIVE"
TOTAL_TIMEOUT_SECONDS = 1800
def _remaining(deadline, clock, stage):
remaining = deadline - clock()
if remaining <= 0:
raise ValidationFailure(f"{stage} timed out before it could start")
return remaining
def _project_root(
cwd: Path,
*,
deadline: float,
run=subprocess.run,
clock=time.monotonic,
) -> Path:
argv = ["git", "rev-parse", "--show-toplevel"]
resolved_cwd = cwd.resolve()
try:
result = run(
argv,
cwd=resolved_cwd,
capture_output=True,
text=True,
shell=False,
timeout=_remaining(deadline, clock, "repository discovery"),
)
except subprocess.TimeoutExpired as exc:
raise ValidationFailure(
f"repository discovery timed out; argv={argv!r}; "
f"cwd={str(resolved_cwd)!r}"
) from exc
if result.returncode == 0 and result.stdout.strip():
return Path(result.stdout.strip()).resolve()
return resolved_cwd
def run_validations(
root: Path,
*,
deadline: float | None = None,
clock=time.monotonic,
) -> tuple[bool, str]:
"""Build and test a discovered C/C++ project, when one exists."""
if deadline is None:
deadline = clock() + TOTAL_TIMEOUT_SECONDS
try:
config = load_config(root)
discovery = discover_project(root, config)
if discovery.selection is None:
return True, ""
selection = discovery.selection
tools = discover_toolchain(
selection.kind,
deadline=deadline,
clock=clock,
)
adapter = (
CMakeAdapter()
if selection.kind is ProjectKind.CMAKE
else MSBuildAdapter()
)
plan = adapter.create_plan(root, selection, config, tools)
child_env = os.environ.copy()
child_env[REENTRY_ENV] = "1"
execute_plan(
plan,
root,
env=child_env,
deadline=deadline,
clock=clock,
)
return True, ""
except (
ConfigError,
DiscoveryError,
ToolchainError,
AdapterError,
ValidationFailure,
OSError,
) as exc:
return False, str(exc)
def _emit_stop_response(message: str) -> None:
print(
json.dumps(
{
"continue": False,
"stopReason": message,
"systemMessage": message,
},
ensure_ascii=False,
)
)
def main(*, run=subprocess.run, clock=time.monotonic) -> int:
if os.environ.get(REENTRY_ENV) == "1":
return 0
deadline = clock() + TOTAL_TIMEOUT_SECONDS
try:
root = _project_root(
Path.cwd(),
deadline=deadline,
run=run,
clock=clock,
)
ok, message = run_validations(root, deadline=deadline, clock=clock)
except (OSError, ValidationFailure) as exc:
_emit_stop_response(f"validation hook failed: {exc}")
return 0
if ok:
return 0
_emit_stop_response(message)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+3
View File
@@ -0,0 +1,3 @@
from .config import ConfigError, load_config
__all__ = ["ConfigError", "load_config"]
@@ -0,0 +1,5 @@
from .base import AdapterError, ValidationAdapter
from .cmake import CMakeAdapter
from .msbuild import MSBuildAdapter
__all__ = ["AdapterError", "CMakeAdapter", "MSBuildAdapter", "ValidationAdapter"]
+19
View File
@@ -0,0 +1,19 @@
from pathlib import Path
from typing import Protocol
from ..models import HarnessConfig, ProjectSelection, Toolchain, ValidationPlan
class AdapterError(RuntimeError):
pass
class ValidationAdapter(Protocol):
def create_plan(
self,
root: Path,
selection: ProjectSelection,
config: HarnessConfig,
tools: Toolchain,
) -> ValidationPlan:
raise NotImplementedError
+99
View File
@@ -0,0 +1,99 @@
from pathlib import Path
from ..models import (
CommandSpec,
HarnessConfig,
ProjectKind,
ProjectSelection,
ResultCheck,
ResultCheckKind,
Toolchain,
ValidationPlan,
ValidationStep,
)
from .base import AdapterError
class CMakeAdapter:
def create_plan(
self,
root: Path,
selection: ProjectSelection,
config: HarnessConfig,
tools: Toolchain,
) -> ValidationPlan:
if selection.kind is not ProjectKind.CMAKE:
raise AdapterError("CMakeAdapter requires a CMake project selection")
if tools.cmake is None:
raise AdapterError("CMake tool is required")
if tools.ctest is None:
raise AdapterError("CTest tool is required")
if config.cmake.configure_preset is None:
command_cwd = root
binary = root / ".harness/build"
configure = (
str(tools.cmake),
"-S",
str(config.cmake.source_dir),
"-B",
str(binary),
"-A",
"x64",
)
build = (str(tools.cmake), "--build", str(binary), "--config", "Debug")
discover = (
str(tools.ctest),
"--test-dir",
str(binary),
"-C",
"Debug",
"--show-only=json-v1",
)
test = (
str(tools.ctest),
"--test-dir",
str(binary),
"-C",
"Debug",
"--output-on-failure",
)
else:
command_cwd = config.cmake.source_dir
binary = config.cmake.binary_dir
if (
binary is None
or config.cmake.build_preset is None
or config.cmake.test_preset is None
):
raise AdapterError("CMake presets require binary, build, and test settings")
configure = (str(tools.cmake), "--preset", config.cmake.configure_preset)
build = (str(tools.cmake), "--build", "--preset", config.cmake.build_preset)
discover = (
str(tools.ctest),
"--preset",
config.cmake.test_preset,
"--show-only=json-v1",
)
test = (
str(tools.ctest),
"--preset",
config.cmake.test_preset,
"--output-on-failure",
)
return ValidationPlan(
ProjectKind.CMAKE,
(
ValidationStep(
CommandSpec(configure, command_cwd, "configure"),
(ResultCheck(ResultCheckKind.CMAKE_COMPILER_IS_MSVC, binary),),
),
ValidationStep(CommandSpec(build, command_cwd, "build")),
ValidationStep(
CommandSpec(discover, command_cwd, "test-discovery"),
(ResultCheck(ResultCheckKind.CTEST_HAS_TESTS),),
),
ValidationStep(CommandSpec(test, command_cwd, "test")),
),
)
+60
View File
@@ -0,0 +1,60 @@
from pathlib import Path
from ..models import (
CommandSpec,
HarnessConfig,
ProjectKind,
ProjectSelection,
Toolchain,
ValidationPlan,
ValidationStep,
)
from .base import AdapterError
def _test_argv(root: Path, command: tuple[str, ...]) -> tuple[str, ...]:
if "/" not in command[0] and "\\" not in command[0]:
return command
first = Path(command[0].replace("\\", "/"))
executable = first if first.is_absolute() else (root / first).resolve()
try:
executable.relative_to(root)
except ValueError as exc:
raise AdapterError(
"msbuild.testCommand resolves outside the repository"
) from exc
return (str(executable), *command[1:])
class MSBuildAdapter:
def create_plan(
self,
root: Path,
selection: ProjectSelection,
config: HarnessConfig,
tools: Toolchain,
) -> ValidationPlan:
if selection.kind is not ProjectKind.MSBUILD:
raise AdapterError("MSBuildAdapter received a non-MSBuild project")
if not config.msbuild.test_command:
raise AdapterError(
"msbuild.testCommand is required for .sln/.vcxproj validation"
)
build = ValidationStep(
CommandSpec(
(
str(tools.msbuild),
str(selection.project_file),
"/m",
"/nologo",
f"/p:Configuration={config.msbuild.configuration}",
f"/p:Platform={config.msbuild.platform}",
),
root,
"build",
)
)
test = ValidationStep(
CommandSpec(_test_argv(root, config.msbuild.test_command), root, "test")
)
return ValidationPlan(ProjectKind.MSBUILD, (build, test))
+146
View File
@@ -0,0 +1,146 @@
import json
from pathlib import Path
from typing import Any
from .models import CMakeConfig, HarnessConfig, MSBuildConfig, TDDConfig
DEFAULT_TEST_PATTERNS = (
"{stem}_test.cpp",
"{stem}_tests.cpp",
"test_{stem}.cpp",
"{stem}.test.cpp",
)
class ConfigError(ValueError):
pass
def _reject_unknown(data: dict[str, Any], allowed: set[str], label: str) -> None:
unknown = sorted(set(data) - allowed)
if unknown:
raise ConfigError(f"{label} contains unexpected field: {unknown[0]}")
def _repo_path(root: Path, raw: str, label: str) -> Path:
candidate = Path(raw)
if candidate.is_absolute():
raise ConfigError(f"{label} must be repository-relative")
resolved = (root / candidate).resolve()
try:
resolved.relative_to(root)
except ValueError as exc:
raise ConfigError(f"{label} resolves outside the repository") from exc
return resolved
def _mapping(value: Any, label: str) -> dict[str, Any]:
if not isinstance(value, dict):
raise ConfigError(f"{label} must be an object")
return value
def _nonempty_string(value: Any, label: str) -> str:
if not isinstance(value, str) or not value:
raise ConfigError(f"{label} must be a non-empty string")
return value
def _string_array(value: Any, label: str) -> tuple[str, ...]:
if not isinstance(value, list):
raise ConfigError(f"{label} must be an array")
if not value or any(not isinstance(item, str) or not item for item in value):
raise ConfigError(f"{label} must be a non-empty string array")
return tuple(value)
def load_config(root: Path) -> HarnessConfig:
root = root.resolve()
config_path = root / ".harness" / "config.json"
if config_path.is_file():
try:
data = json.loads(config_path.read_text(encoding="utf-8"))
except UnicodeDecodeError as exc:
raise ConfigError(
f"{config_path}: configuration must be valid UTF-8: {exc}"
) from exc
except json.JSONDecodeError as exc:
raise ConfigError(f"{config_path}: {exc}") from exc
else:
data = {"version": 1}
data = _mapping(data, "config")
_reject_unknown(data, {"version", "projectType", "cmake", "msbuild", "tdd"}, "config")
if data.get("version") != 1 or isinstance(data.get("version"), bool):
raise ConfigError("version must be 1")
project_type = data.get("projectType", "auto")
if project_type not in {"auto", "cmake", "msbuild"}:
raise ConfigError("projectType must be auto, cmake, or msbuild")
cmake_data = _mapping(data.get("cmake", {}), "cmake")
_reject_unknown(
cmake_data,
{"sourceDir", "binaryDir", "configurePreset", "buildPreset", "testPreset"},
"cmake",
)
source_dir = _repo_path(
root, _nonempty_string(cmake_data.get("sourceDir", "."), "cmake.sourceDir"), "cmake.sourceDir"
)
binary_dir = None
if "binaryDir" in cmake_data:
binary_dir = _repo_path(
root, _nonempty_string(cmake_data["binaryDir"], "cmake.binaryDir"), "cmake.binaryDir"
)
preset_keys = ("configurePreset", "buildPreset", "testPreset", "binaryDir")
if any(key in cmake_data for key in preset_keys) and not all(key in cmake_data for key in preset_keys):
raise ConfigError("cmake configurePreset, buildPreset, testPreset, and binaryDir must be specified together")
configure_preset = (
_nonempty_string(cmake_data["configurePreset"], "cmake.configurePreset")
if "configurePreset" in cmake_data
else None
)
build_preset = (
_nonempty_string(cmake_data["buildPreset"], "cmake.buildPreset")
if "buildPreset" in cmake_data
else None
)
test_preset = (
_nonempty_string(cmake_data["testPreset"], "cmake.testPreset")
if "testPreset" in cmake_data
else None
)
msbuild_data = _mapping(data.get("msbuild", {}), "msbuild")
_reject_unknown(msbuild_data, {"solution", "configuration", "platform", "testCommand"}, "msbuild")
solution = None
if "solution" in msbuild_data:
solution = _repo_path(
root, _nonempty_string(msbuild_data["solution"], "msbuild.solution"), "msbuild.solution"
)
configuration = _nonempty_string(msbuild_data.get("configuration", "Debug"), "msbuild.configuration")
platform = _nonempty_string(msbuild_data.get("platform", "x64"), "msbuild.platform")
test_command = (
_string_array(msbuild_data["testCommand"], "msbuild.testCommand")
if "testCommand" in msbuild_data
else None
)
tdd_data = _mapping(data.get("tdd", {}), "tdd")
_reject_unknown(tdd_data, {"testRoots", "testPatterns", "exclude"}, "tdd")
raw_test_roots = _string_array(tdd_data["testRoots"], "tdd.testRoots") if "testRoots" in tdd_data else ("tests", "test")
test_roots = tuple(_repo_path(root, item, "tdd.testRoots") for item in raw_test_roots)
test_patterns = _string_array(tdd_data["testPatterns"], "tdd.testPatterns") if "testPatterns" in tdd_data else DEFAULT_TEST_PATTERNS
if any("{stem}" not in pattern for pattern in test_patterns):
raise ConfigError("every tdd.testPatterns entry must contain {stem}")
exclude = _string_array(tdd_data["exclude"], "tdd.exclude") if "exclude" in tdd_data else ()
for item in exclude:
_repo_path(root, item, "tdd.exclude")
return HarnessConfig(
version=1,
project_type=project_type,
cmake=CMakeConfig(source_dir, binary_dir, configure_preset, build_preset, test_preset),
msbuild=MSBuildConfig(solution, configuration, platform, test_command),
tdd=TDDConfig(test_roots, test_patterns, exclude),
)
+77
View File
@@ -0,0 +1,77 @@
from pathlib import Path
from .models import DiscoveryResult, HarnessConfig, ProjectKind, ProjectSelection
from .tdd_policy import find_cpp_files
class DiscoveryError(RuntimeError):
pass
def _cmake_entry(source: Path) -> Path | None:
for name in ("CMakePresets.json", "CMakeUserPresets.json", "CMakeLists.txt"):
candidate = source / name
if candidate.is_file():
return candidate
return None
def _single_msbuild(root: Path, explicit: Path | None) -> Path:
if explicit is not None:
if explicit.is_file() and explicit.suffix.lower() in {".sln", ".vcxproj"}:
return explicit
raise DiscoveryError(f"configured MSBuild project is invalid: {explicit}")
solutions = sorted(root.glob("*.sln"))
if len(solutions) > 1:
raise DiscoveryError("multiple .sln files found; set msbuild.solution")
if solutions:
return solutions[0]
projects = sorted(root.glob("*.vcxproj"))
if len(projects) > 1:
raise DiscoveryError("multiple .vcxproj files found; set msbuild.solution")
if projects:
return projects[0]
raise DiscoveryError("no root .sln or .vcxproj was found")
def discover_project(root: Path, config: HarnessConfig) -> DiscoveryResult:
root = root.resolve()
if config.project_type == "cmake":
entry = _cmake_entry(config.cmake.source_dir)
if entry is None:
raise DiscoveryError(
f"no CMake project found in {config.cmake.source_dir}"
)
return DiscoveryResult(ProjectSelection(ProjectKind.CMAKE, entry))
if config.project_type == "msbuild":
project = _single_msbuild(root, config.msbuild.solution)
return DiscoveryResult(ProjectSelection(ProjectKind.MSBUILD, project))
entry = _cmake_entry(root)
if entry is not None:
return DiscoveryResult(ProjectSelection(ProjectKind.CMAKE, entry))
solutions = sorted(root.glob("*.sln"))
if len(solutions) > 1:
raise DiscoveryError("multiple .sln files found; set msbuild.solution")
if solutions:
return DiscoveryResult(
ProjectSelection(ProjectKind.MSBUILD, solutions[0])
)
projects = sorted(root.glob("*.vcxproj"))
if len(projects) > 1:
raise DiscoveryError("multiple .vcxproj files found; set msbuild.solution")
if projects:
return DiscoveryResult(
ProjectSelection(ProjectKind.MSBUILD, projects[0])
)
cpp_files = find_cpp_files(root)
if cpp_files:
raise DiscoveryError(
"C/C++ files exist but no CMakeLists.txt, preset, .sln, or .vcxproj "
"was found; configure projectType and project path"
)
return DiscoveryResult(None, ())
+100
View File
@@ -0,0 +1,100 @@
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
class ProjectKind(Enum):
CMAKE = "cmake"
MSBUILD = "msbuild"
class ResultCheckKind(Enum):
CMAKE_COMPILER_IS_MSVC = "cmake_compiler_is_msvc"
CTEST_HAS_TESTS = "ctest_has_tests"
@dataclass(frozen=True)
class CMakeConfig:
source_dir: Path
binary_dir: Path | None = None
configure_preset: str | None = None
build_preset: str | None = None
test_preset: str | None = None
@dataclass(frozen=True)
class MSBuildConfig:
solution: Path | None = None
configuration: str = "Debug"
platform: str = "x64"
test_command: tuple[str, ...] | None = None
@dataclass(frozen=True)
class TDDConfig:
test_roots: tuple[Path, ...]
test_patterns: tuple[str, ...]
exclude: tuple[str, ...] = ()
@dataclass(frozen=True)
class HarnessConfig:
version: int
project_type: str
cmake: CMakeConfig
msbuild: MSBuildConfig
tdd: TDDConfig
@dataclass(frozen=True)
class ProjectSelection:
kind: ProjectKind
project_file: Path
@dataclass(frozen=True)
class DiscoveryResult:
selection: ProjectSelection | None
cpp_files: tuple[Path, ...] = ()
@dataclass(frozen=True)
class Toolchain:
installation: Path
msbuild: Path
cmake: Path | None = None
ctest: Path | None = None
@dataclass(frozen=True)
class CommandSpec:
argv: tuple[str, ...]
cwd: Path
stage: str
timeout_seconds: int = 1800
@dataclass(frozen=True)
class ResultCheck:
kind: ResultCheckKind
path: Path | None = None
@dataclass(frozen=True)
class ValidationStep:
command: CommandSpec
checks: tuple[ResultCheck, ...] = ()
@dataclass(frozen=True)
class ValidationPlan:
project_kind: ProjectKind
steps: tuple[ValidationStep, ...]
@dataclass(frozen=True)
class CommandResult:
command: CommandSpec
returncode: int
stdout: str
stderr: str
+140
View File
@@ -0,0 +1,140 @@
import json
import locale
import re
import subprocess
import time
from .models import CommandResult, ResultCheckKind
DIAGNOSTIC_LIMIT = 8000
class ValidationFailure(RuntimeError):
pass
def _decode(value):
if value is None:
return ""
if isinstance(value, str):
return value
for encoding in ("utf-8", locale.getpreferredencoding(False)):
try:
return value.decode(encoding)
except (UnicodeDecodeError, LookupError):
continue
return value.decode("utf-8", errors="replace")
def _check_result(check, result):
if check.kind is ResultCheckKind.CMAKE_COMPILER_IS_MSVC:
if check.path is None:
raise ValidationFailure("MSVC check is missing binaryDir")
files = sorted(check.path.glob("CMakeFiles/*/CMakeCXXCompiler.cmake"))
if not files:
raise ValidationFailure("CMake did not generate compiler metadata")
text = files[-1].read_text(encoding="utf-8", errors="replace")
if not re.search(
r'set\s*\(\s*CMAKE_CXX_COMPILER_ID\s+"MSVC"\s*\)', text
):
raise ValidationFailure("CMake selected a compiler other than MSVC")
elif check.kind is ResultCheckKind.CTEST_HAS_TESTS:
try:
payload = json.loads(result.stdout)
except json.JSONDecodeError as exc:
raise ValidationFailure("CTest discovery did not return JSON") from exc
if not isinstance(payload, dict):
raise ValidationFailure("CTest discovery JSON must be an object")
if not isinstance(payload.get("tests"), list) or not payload["tests"]:
raise ValidationFailure("CTest discovered no tests")
def _diagnostic(stdout, stderr):
diagnostic = (stdout + "\n" + stderr).rstrip()[-DIAGNOSTIC_LIMIT:]
return diagnostic.replace("\r\n", " | ").replace("\n", " | ")
def _command_context(argv, cwd):
return f"argv={list(argv)!r}; cwd={str(cwd)!r}"
def execute_plan(
plan,
root,
*,
env=None,
total_timeout_seconds=1800,
deadline=None,
run=subprocess.run,
clock=time.monotonic,
):
root = root.resolve()
if deadline is None:
deadline = clock() + total_timeout_seconds
results = []
for step in plan.steps:
cwd = step.command.cwd.resolve()
context = _command_context(step.command.argv, cwd)
try:
cwd.relative_to(root)
except ValueError as exc:
raise ValidationFailure(
f"{step.command.stage} working directory is outside the repository; "
f"{context}"
) from exc
remaining = deadline - clock()
if remaining <= 0:
raise ValidationFailure(
f"{step.command.stage} timed out before it could start; {context}"
)
timeout = min(step.command.timeout_seconds, remaining)
try:
completed = run(
list(step.command.argv),
cwd=cwd,
env=env,
capture_output=True,
shell=False,
timeout=timeout,
)
except subprocess.TimeoutExpired as exc:
stdout = _decode(exc.output)
stderr = _decode(exc.stderr)
diagnostic = _diagnostic(stdout, stderr)
message = (
f"{step.command.stage} timed out after {timeout} seconds; {context}"
)
if diagnostic:
message = f"{message}; output tail: {diagnostic}"
raise ValidationFailure(message) from exc
stdout = _decode(completed.stdout)
stderr = _decode(completed.stderr)
result = CommandResult(step.command, completed.returncode, stdout, stderr)
if completed.returncode != 0:
diagnostic = _diagnostic(stdout, stderr)
message = (
f"{step.command.stage} failed with exit code {completed.returncode}; "
f"{context}"
)
if diagnostic:
message = f"{message}; output tail: {diagnostic}"
raise ValidationFailure(message)
try:
for check in step.checks:
_check_result(check, result)
except ValidationFailure as exc:
diagnostic = _diagnostic(stdout, stderr)
message = (
f"{step.command.stage} result check failed: {exc}; {context}"
)
if diagnostic:
message = f"{message}; output tail: {diagnostic}"
raise ValidationFailure(message) from exc
results.append(result)
return tuple(results)
+88
View File
@@ -0,0 +1,88 @@
from fnmatch import fnmatch
from pathlib import Path
CPP_SUFFIXES = frozenset({".c", ".cc", ".cpp", ".cxx", ".h", ".hpp", ".hxx"})
DEFAULT_EXCLUDES = (
".harness/build/**",
"build/**",
"out/**",
"cmake-build-*/**",
"third_party/**",
"external/**",
"vendor/**",
"generated/**",
)
def is_excluded_path(path: Path, root: Path, extra: tuple[str, ...] = ()) -> bool:
try:
relative = path.resolve().relative_to(root.resolve()).as_posix()
except ValueError:
return False
return any(fnmatch(relative, pattern) for pattern in DEFAULT_EXCLUDES + extra)
def find_cpp_files(
root: Path,
extra_excludes: tuple[str, ...] = (),
) -> tuple[Path, ...]:
found = (
path
for path in root.resolve().rglob("*")
if path.is_file()
and path.suffix.lower() in CPP_SUFFIXES
and not is_excluded_path(path, root, extra_excludes)
)
return tuple(sorted(found))
def is_test_file(path: Path, root: Path, config) -> bool:
resolved = path.resolve()
if any(root == resolved or root in resolved.parents for root in config.test_roots):
return True
relative = resolved.relative_to(root.resolve())
if {"test", "tests"} & {part.lower() for part in relative.parts}:
return True
stem = resolved.stem.lower()
return stem.startswith("test_") or stem.endswith(("_test", "_tests", ".test"))
def matching_test_exists(path: Path, config) -> bool:
names = tuple(pattern.format(stem=path.stem) for pattern in config.test_patterns)
roots = (
*config.test_roots,
path.parent / "tests",
path.parent / "test",
)
for root in roots:
if not root.is_dir():
continue
for name in names:
if any(candidate.is_file() for candidate in root.rglob(name)):
return True
return False
def evaluate_paths(paths, root: Path, config) -> str | None:
root = root.resolve()
for raw in paths:
path = raw.resolve()
try:
path.relative_to(root)
except ValueError:
return f"TDD GUARD: '{path}' is outside the repository"
if path.suffix.lower() not in CPP_SUFFIXES:
continue
if (
path.name.lower() == "main.cpp"
or is_excluded_path(path, root, config.exclude)
or is_test_file(path, root, config)
):
continue
if not matching_test_exists(path, config):
expected = config.test_patterns[0].format(stem=path.stem)
return (
f"TDD GUARD: '{path.name}' requires an existing test such as "
f"'{expected}'. Add the test in a configured test root first."
)
return None
+169
View File
@@ -0,0 +1,169 @@
import os
import shutil
import subprocess
import time
from pathlib import Path
from .models import ProjectKind, Toolchain
VC_WORKLOAD = "Microsoft.VisualStudio.Component.VC.Tools.x86.x64"
VS_CMAKE_RELATIVE = Path(
"Common7/IDE/CommonExtensions/Microsoft/CMake/CMake/bin"
)
class ToolchainError(RuntimeError):
pass
def _decode(value):
if isinstance(value, str):
return value
return (value or b"").decode("utf-8", errors="replace")
def _first_path(result, label, argv, cwd):
text = _decode(result.stdout).strip()
if result.returncode != 0:
diagnostic = (
_decode(result.stdout) + "\n" + _decode(result.stderr)
).rstrip()[-8000:]
diagnostic = diagnostic.replace("\r\n", " | ").replace("\n", " | ")
message = (
f"{label} probe failed with exit code {result.returncode}; "
f"argv={argv!r}; cwd={str(cwd)!r}"
)
if diagnostic:
message = f"{message}; output tail: {diagnostic}"
raise ToolchainError(
f"{message}; install Visual Studio Desktop development with C++"
)
if not text:
raise ToolchainError(
f"{label} probe returned no path; argv={argv!r}; cwd={str(cwd)!r}; "
"install Visual Studio Desktop development with C++"
)
path = Path(text.splitlines()[0]).resolve()
if not path.exists():
diagnostic = (
_decode(result.stdout) + "\n" + _decode(result.stderr)
).rstrip()[-8000:]
diagnostic = diagnostic.replace("\r\n", " | ").replace("\n", " | ")
message = (
f"{label} probe returned nonexistent path with exit code 0; "
f"argv={argv!r}; cwd={str(cwd)!r}"
)
if diagnostic:
message = f"{message}; output tail: {diagnostic}"
raise ToolchainError(
f"{message}; install or repair Visual Studio Desktop development with C++"
)
return path
def _require_path(path, label):
if path is None or not path.exists():
raise ToolchainError(f"{label} not found")
return path.resolve()
def _remaining(deadline, clock, label):
if deadline is None:
return None
remaining = deadline - clock()
if remaining <= 0:
raise ToolchainError(f"{label} probe timed out before it could start")
return remaining
def _probe(run, argv, label, *, deadline, clock):
cwd = Path.cwd().resolve()
kwargs = {
"cwd": cwd,
"capture_output": True,
"check": False,
"shell": False,
}
remaining = _remaining(deadline, clock, label)
if remaining is not None:
kwargs["timeout"] = remaining
try:
return run(argv, **kwargs)
except subprocess.TimeoutExpired as exc:
raise ToolchainError(
f"{label} probe timed out; argv={argv!r}; cwd={str(cwd)!r}; "
"install or repair Visual Studio Desktop development with C++"
) from exc
def discover_toolchain(
kind: ProjectKind,
*,
env=None,
which=shutil.which,
run=subprocess.run,
deadline=None,
clock=time.monotonic,
) -> Toolchain:
environment = os.environ if env is None else env
vswhere_path = which("vswhere.exe")
if vswhere_path is None:
program_files = environment.get("ProgramFiles(x86)")
if program_files:
candidate = Path(program_files) / "Microsoft Visual Studio/Installer/vswhere.exe"
vswhere_path = str(candidate) if candidate.exists() else None
vswhere = _require_path(
Path(vswhere_path) if vswhere_path else None,
"vswhere.exe; install Visual Studio Desktop development with C++",
)
install_argv = [
str(vswhere), "-latest", "-products", "*", "-requires", VC_WORKLOAD,
"-property", "installationPath",
]
install_result = _probe(
run,
install_argv,
"Visual Studio installation",
deadline=deadline,
clock=clock,
)
probe_cwd = Path.cwd().resolve()
installation = _require_path(
_first_path(
install_result,
"Visual Studio installation",
install_argv,
probe_cwd,
),
"Visual Studio installation",
)
msbuild_argv = [
str(vswhere), "-latest", "-products", "*", "-requires", VC_WORKLOAD,
"-find", r"MSBuild\**\Bin\MSBuild.exe",
]
msbuild_result = _probe(
run,
msbuild_argv,
"MSBuild",
deadline=deadline,
clock=clock,
)
msbuild = _require_path(
_first_path(msbuild_result, "MSBuild", msbuild_argv, probe_cwd),
"MSBuild.exe",
)
if kind is not ProjectKind.CMAKE:
return Toolchain(installation=installation, msbuild=msbuild)
bundled_cmake_dir = installation / VS_CMAKE_RELATIVE
cmake = _require_path(
Path(which("cmake.exe")) if which("cmake.exe") else bundled_cmake_dir / "cmake.exe",
"cmake.exe",
)
ctest = _require_path(
Path(which("ctest.exe")) if which("ctest.exe") else bundled_cmake_dir / "ctest.exe",
"ctest.exe",
)
return Toolchain(installation=installation, msbuild=msbuild, cmake=cmake, ctest=ctest)
-75
View File
@@ -1,75 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENTS_ROOT = ROOT / ".codex" / "agents"
AGENT_SKILL_REFERENCES = {
"coordinator-agent.toml": (
"fesa-requirements-baseline",
"fesa-reference-models",
"fesa-release-readiness",
),
"requirement-agent.toml": ("fesa-requirements-baseline",),
"research-agent.toml": (
"fesa-research-evidence",
"fem-theory-query",
),
"formulation-agent.toml": (
"fesa-formulation-spec",
"fem-theory-query",
),
"numerical-review-agent.toml": (
"fesa-numerical-review",
"fem-theory-query",
),
"io-definition-agent.toml": (
"fesa-io-contract",
"fem-theory-query",
),
"reference-model-agent.toml": (
"fesa-reference-models",
"fem-theory-query",
),
"implementation-planning-agent.toml": (
"fesa-formulation-spec",
"fesa-reference-models",
"fesa-cpp-msvc-tdd",
"fem-theory-query",
),
"implementation-agent.toml": ("fesa-cpp-msvc-tdd",),
"build-test-executor-agent.toml": ("fesa-cpp-msvc-tdd",),
"correction-agent.toml": ("fesa-cpp-msvc-tdd",),
"reference-verification-agent.toml": (
"fesa-reference-comparison",
"fesa-io-contract",
),
"physics-evaluation-agent.toml": (
"fesa-physics-sanity",
"fem-theory-query",
),
"release-agent.toml": ("fesa-release-readiness",),
}
class AgentSkillReferenceTests(unittest.TestCase):
def test_agents_reference_their_solver_skills(self):
for agent_file, skill_names in AGENT_SKILL_REFERENCES.items():
with self.subTest(agent=agent_file):
text = (AGENTS_ROOT / agent_file).read_text(encoding="utf-8")
data = tomllib.loads(text)
instructions = data["developer_instructions"]
self.assertIn("Skill references:", instructions)
for skill_name in skill_names:
self.assertIn(f"${skill_name}", instructions)
if __name__ == "__main__":
unittest.main()
@@ -1,86 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "build-test-executor-agent.toml"
BUILD_TEST_REPORTS_README = ROOT / "docs" / "build-test-reports" / "README.md"
class BuildTestExecutorAgentConfigTests(unittest.TestCase):
def test_build_test_executor_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "build-test-executor-agent")
self.assertIn("C++/MSVC/CMake/CTest validation", data["description"])
self.assertEqual(data["sandbox_mode"], "workspace-write")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_build_test_executor_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not edit source code.",
"Do not edit tests.",
"Do not edit CMake.",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not approve release readiness.",
):
self.assertIn(required_text, instructions)
def test_build_test_executor_agent_instructions_define_validation_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"python scripts/validate_workspace.py",
"HARNESS_VALIDATION_COMMANDS",
"msvc-debug",
"CMake/MSVC x64 Debug",
"ctest --test-dir",
"--output-on-failure",
"-C Debug",
):
self.assertIn(required_text, instructions)
def test_build_test_executor_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Execution Environment",
"Command Log Summary",
"Validation Results",
"Failure Classification",
"Failed Test Inventory",
"Handoff Recommendation",
"No-Change Assertion",
):
self.assertIn(required_text, instructions)
def test_build_test_report_guide_defines_template_and_status_values(self):
guide = BUILD_TEST_REPORTS_README.read_text(encoding="utf-8")
for required_text in (
"docs/build-test-reports/<feature-id>-build-test.md",
"Execution Environment",
"Command Log Summary",
"Validation Results",
"Failure Classification",
"Failed Test Inventory",
"Handoff Recommendation",
"No-Change Assertion",
"pass-for-reference-verification",
"needs-correction",
"needs-environment-fix",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-102
View File
@@ -1,102 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "coordinator-agent.toml"
COORDINATION_README = ROOT / "docs" / "coordination" / "README.md"
class CoordinatorAgentConfigTests(unittest.TestCase):
def test_coordinator_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "coordinator-agent")
self.assertIn("workflow state", data["description"])
self.assertEqual(data["sandbox_mode"], "workspace-write")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_coordinator_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not edit source code.",
"Do not edit tests.",
"Do not edit CMake.",
"Do not run build/test validation.",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not automatically spawn subagents.",
"Do not approve release readiness independently.",
):
self.assertIn(required_text, instructions)
def test_coordinator_agent_instructions_define_workflow_and_status_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"INTAKE -> STATE AUDIT -> GATE DECISION -> HANDOFF PACKAGE -> STATUS REPORT",
"Requirement Agent",
"Research Agent",
"Formulation Agent",
"Numerical Review Agent",
"I/O Definition Agent",
"Reference Model Agent",
"Implementation Planning Agent",
"Implementation Agent",
"Build/Test Executor Agent",
"Correction Agent",
"Reference Verification Agent",
"Physics Evaluation Agent",
"Release Agent",
"ready-for-implementation",
"pass-for-reference-verification",
"pass-for-physics-evaluation",
"pass-for-release-agent",
"ready-for-release",
"completed",
):
self.assertIn(required_text, instructions)
def test_coordinator_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Gate Evidence Inventory",
"Decision Log",
"Next Agent Handoff",
"Traceability Snapshot",
"Risk and Blocker Register",
"Rework Loop Control",
"No-Change Assertion",
):
self.assertIn(required_text, instructions)
def test_coordination_guide_defines_template_and_status_values(self):
guide = COORDINATION_README.read_text(encoding="utf-8")
for required_text in (
"docs/coordination/<feature-id>-coordination.md",
"Gate Evidence Inventory",
"Decision Log",
"Next Agent Handoff",
"Traceability Snapshot",
"Risk and Blocker Register",
"Rework Loop Control",
"No-Change Assertion",
"needs-user-decision",
"blocked",
"python scripts/validate_workspace.py",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-90
View File
@@ -1,90 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "correction-agent.toml"
CORRECTIONS_README = ROOT / "docs" / "corrections" / "README.md"
class CorrectionAgentConfigTests(unittest.TestCase):
def test_correction_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "correction-agent")
self.assertIn("C++/MSVC/CMake/CTest fixes", data["description"])
self.assertEqual(data["sandbox_mode"], "workspace-write")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_correction_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not change requirements",
"Do not change formulations",
"Do not change I/O contracts",
"Do not change reference artifacts",
"Do not change tolerance policies",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not approve release readiness.",
):
self.assertIn(required_text, instructions)
def test_correction_agent_instructions_define_triage_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"TRIAGE -> MINIMAL FIX -> VERIFY -> REPORT",
"configure",
"compile",
"link",
"test",
"reference-comparison",
"harness",
"environment",
"upstream-contract",
):
self.assertIn(required_text, instructions)
def test_correction_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Failure Triage",
"Root Cause Summary",
"Correction Scope",
"Verification Evidence",
"Traceability",
"Handoff Recommendation",
"Stop Condition",
):
self.assertIn(required_text, instructions)
def test_correction_report_guide_defines_template_and_status_values(self):
guide = CORRECTIONS_README.read_text(encoding="utf-8")
for required_text in (
"docs/corrections/<feature-id>-correction.md",
"Failure Triage",
"Root Cause Summary",
"Correction Scope",
"Verification Evidence",
"Traceability",
"Handoff Recommendation",
"Stop Condition",
"corrected-for-build-test",
"needs-upstream-decision",
"python scripts/validate_workspace.py",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-286
View File
@@ -1,286 +0,0 @@
import importlib.util
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
def load_execute():
module_path = Path(__file__).resolve().parent / "execute.py"
spec = importlib.util.spec_from_file_location("execute", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def write_phase(root: Path, phase_dir: str = "0-mvp", phase_name: str = "0-mvp", steps=None):
phase_path = root / "phases" / phase_dir
phase_path.mkdir(parents=True)
if steps is None:
steps = [
{
"step": 1,
"name": "Docs",
"status": "pending",
"summary": "",
"allowed_paths": ["docs/*.md"],
}
]
(phase_path / "index.json").write_text(
json.dumps({"project": "FESA", "phase": phase_name, "steps": steps}, indent=2),
encoding="utf-8",
)
(phase_path / "step1.md").write_text("# Step 1\n", encoding="utf-8")
return phase_path
def make_executor(execute, root: Path, phase_dir: str = "0-mvp"):
with patch.object(execute, "ROOT", root):
return execute.StepExecutor(phase_dir)
class ExecuteRunnerSafetyTests(unittest.TestCase):
def test_scaffold_loads_execute_module(self):
execute = load_execute()
self.assertTrue(hasattr(execute, "StepExecutor"))
def test_branch_name_uses_codex_prefix_and_sanitized_phase(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root, phase_name="linear truss/1d")
executor = make_executor(execute, root)
self.assertEqual(executor._branch_name(), "codex/linear-truss-1d")
def test_finalize_push_uses_codex_branch_name(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root, phase_name="0-mvp")
executor = make_executor(execute, root)
executor._auto_push = True
calls = []
def fake_git(*args):
calls.append(args)
if args == ("diff", "--cached", "--quiet"):
return subprocess.CompletedProcess(args, 0, "", "")
return subprocess.CompletedProcess(args, 0, "", "")
with patch.object(executor, "_run_git", side_effect=fake_git):
with patch.object(executor, "_validate_before_commit", create=True):
with patch("builtins.print"):
executor._finalize()
self.assertIn(("push", "-u", "origin", "codex/0-mvp"), calls)
def test_finalize_stages_only_phase_indexes(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
(root / "phases" / "index.json").write_text('{"phases":[]}', encoding="utf-8")
executor = make_executor(execute, root)
calls = []
def fake_git(*args):
calls.append(args)
if args == ("diff", "--cached", "--quiet"):
return subprocess.CompletedProcess(args, 1, "", "")
return subprocess.CompletedProcess(args, 0, "", "")
with patch.object(executor, "_run_git", side_effect=fake_git):
with patch.object(executor, "_validate_before_commit"):
with patch("builtins.print"):
executor._finalize()
self.assertNotIn(("add", "-A"), calls)
self.assertIn(("add", "--", "phases/0-mvp/index.json", "phases/index.json"), calls)
def test_assert_clean_worktree_exits_when_git_status_has_changes(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
executor = make_executor(execute, root)
with patch.object(
executor,
"_run_git",
return_value=subprocess.CompletedProcess([], 0, " M AGENTS.md\n?? scratch.txt\n", ""),
):
with patch("builtins.print"):
with self.assertRaises(SystemExit) as cm:
executor._assert_clean_worktree("before checkout")
self.assertEqual(cm.exception.code, 1)
def test_run_checks_clean_worktree_before_checkout(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
executor = make_executor(execute, root)
calls = []
def record(name):
def inner(*args, **kwargs):
calls.append(name)
return inner
with patch.object(executor, "_assert_clean_worktree", side_effect=record("clean")):
with patch.object(executor, "_checkout_branch", side_effect=record("checkout")):
with patch.object(executor, "_print_header"):
with patch.object(executor, "_check_blockers"):
with patch.object(executor, "_load_guardrails", return_value=""):
with patch.object(executor, "_ensure_created_at"):
with patch.object(executor, "_execute_all_steps"):
with patch.object(executor, "_finalize"):
executor.run()
self.assertLess(calls.index("clean"), calls.index("checkout"))
def test_step_allowlist_accepts_exact_prefix_and_glob_paths(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
executor = make_executor(execute, root)
patterns = ["AGENTS.md", "docs/", "scripts/*.py"]
self.assertTrue(executor._path_allowed("AGENTS.md", patterns))
self.assertTrue(executor._path_allowed("docs/PRD.md", patterns))
self.assertTrue(executor._path_allowed("scripts/execute.py", patterns))
self.assertFalse(executor._path_allowed(".codex/hooks.json", patterns))
def test_step_without_allowed_paths_is_rejected_before_codex_invocation(self):
execute = load_execute()
steps = [{"step": 1, "name": "Unsafe", "status": "pending", "summary": ""}]
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root, steps=steps)
executor = make_executor(execute, root)
with patch("builtins.print"):
with self.assertRaises(SystemExit) as cm:
executor._validate_step_allowlist(steps[0])
self.assertEqual(cm.exception.code, 1)
def test_classify_step_changes_splits_allowed_housekeeping_and_disallowed_paths(self):
execute = load_execute()
step = {
"step": 1,
"name": "Docs",
"status": "completed",
"summary": "",
"allowed_paths": ["docs/*.md"],
}
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
executor = make_executor(execute, root)
changed = [
"docs/PRD.md",
"phases/0-mvp/index.json",
"phases/0-mvp/step1-output.json",
"scripts/execute.py",
]
allowed, housekeeping, disallowed = executor._classify_step_changes(1, step, changed)
self.assertEqual(allowed, ["docs/PRD.md"])
self.assertEqual(housekeeping, ["phases/0-mvp/index.json", "phases/0-mvp/step1-output.json"])
self.assertEqual(disallowed, ["scripts/execute.py"])
def test_commit_step_stages_only_explicit_allowed_and_housekeeping_paths(self):
execute = load_execute()
step = {
"step": 1,
"name": "Docs",
"status": "completed",
"summary": "",
"allowed_paths": ["docs/*.md"],
}
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
executor = make_executor(execute, root)
calls = []
def fake_git(*args):
calls.append(args)
if args in {
("diff", "--quiet", "--cached", "--"),
("diff", "--cached", "--quiet"),
}:
return subprocess.CompletedProcess(args, 1, "", "")
return subprocess.CompletedProcess(args, 0, "", "")
with patch.object(
executor,
"_changed_paths",
return_value=[
"docs/PRD.md",
"phases/0-mvp/index.json",
"phases/0-mvp/step1-output.json",
],
):
with patch.object(executor, "_run_git", side_effect=fake_git):
with patch.object(executor, "_validate_before_commit", create=True):
with patch("builtins.print"):
executor._commit_step(step, "Docs")
self.assertNotIn(("add", "-A"), calls)
self.assertIn(("add", "--", "docs/PRD.md"), calls)
self.assertIn(("add", "--", "phases/0-mvp/index.json", "phases/0-mvp/step1-output.json"), calls)
def test_validate_before_commit_runs_python_selftest_then_workspace_validation(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
executor = make_executor(execute, root)
commands = []
def fake_run(cmd, **kwargs):
commands.append(cmd)
return subprocess.CompletedProcess(cmd, 0, "ok", "")
with patch.object(execute.subprocess, "run", side_effect=fake_run):
with patch("builtins.print"):
executor._validate_before_commit("feat(0-mvp): step 1")
self.assertEqual(
commands,
[
[sys.executable, "-m", "unittest", "discover", "-s", "scripts", "-p", "test_*.py"],
[sys.executable, "scripts/validate_workspace.py"],
],
)
def test_validate_before_commit_exits_before_commit_when_validation_fails(self):
execute = load_execute()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
write_phase(root)
executor = make_executor(execute, root)
def fake_run(cmd, **kwargs):
return subprocess.CompletedProcess(cmd, 1, "bad", "failed")
with patch.object(execute.subprocess, "run", side_effect=fake_run):
with patch("builtins.print"):
with self.assertRaises(SystemExit) as cm:
executor._validate_before_commit("feat(0-mvp): step 1")
self.assertEqual(cm.exception.code, 1)
if __name__ == "__main__":
unittest.main()
-344
View File
@@ -1,344 +0,0 @@
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SKILLS_ROOT = ROOT / ".codex" / "skills"
COMMON_SECTIONS = (
"## Inputs",
"## Workflow",
"## Output Contract",
"## Boundaries",
"## Quality Gate",
"## Handoff",
)
ACTIVE_CONTRACT_FILES = (
ROOT / "AGENTS.md",
ROOT / "docs" / "ProjectInitialPlanNote.md",
ROOT / "docs" / "PRD.md",
ROOT / "docs" / "ARCHITECTURE.md",
ROOT / "docs" / "ADR.md",
ROOT / "docs" / "SOLVER_AGENT_DESIGN.md",
ROOT / "docs" / "SOLVER_SKILL_DESIGN.md",
ROOT / "docs" / "reference-models" / "README.md",
ROOT / "docs" / "reference-verifications" / "README.md",
ROOT / "docs" / "io-definitions" / "README.md",
ROOT / "docs" / "implementation-plans" / "README.md",
ROOT / "docs" / "physics-evaluations" / "README.md",
ROOT / "docs" / "requirements" / "README.md",
ROOT / ".codex" / "agents" / "reference-model-agent.toml",
ROOT / ".codex" / "agents" / "reference-verification-agent.toml",
ROOT / ".codex" / "agents" / "io-definition-agent.toml",
ROOT / ".codex" / "agents" / "implementation-planning-agent.toml",
ROOT / ".codex" / "agents" / "implementation-agent.toml",
ROOT / ".codex" / "agents" / "physics-evaluation-agent.toml",
ROOT / ".codex" / "agents" / "release-agent.toml",
ROOT / ".codex" / "agents" / "requirement-agent.toml",
ROOT / ".codex" / "agents" / "coordinator-agent.toml",
ROOT / ".codex" / "skills" / "fesa-reference-models" / "SKILL.md",
ROOT / ".codex" / "skills" / "fesa-reference-comparison" / "SKILL.md",
ROOT / ".codex" / "skills" / "fesa-io-contract" / "SKILL.md",
ROOT / ".codex" / "skills" / "fesa-physics-sanity" / "SKILL.md",
ROOT / ".codex" / "skills" / "fesa-cpp-msvc-tdd" / "SKILL.md",
ROOT / ".codex" / "skills" / "fesa-release-readiness" / "SKILL.md",
ROOT / ".codex" / "skills" / "fesa-requirements-baseline" / "SKILL.md",
)
STALE_REFERENCE_CONTRACT_PHRASES = (
"reference" ".h5",
"stored reference " "HDF5",
"reference " "HDF5 artifact",
"results.h5 and " "reference" ".h5",
"results.h5` and `reference" ".h5",
"derived from reference" ".h5",
"references/" "<feature-id>/<model-id>/",
)
SKILLS = {
"fesa-requirements-baseline": {
"description_terms": (
"Use when",
"FESA solver",
"requirements",
"acceptance criteria",
"verification matrix",
),
"body_terms": (
"docs/requirements/<feature-id>.md",
"Requirement Verification Matrix",
"shall",
"FESA-REQ-<FEATURE>-###",
"Verification Quantities",
"Tolerance Policy",
"Reference Artifact Requirements",
"Do not implement C++ code.",
),
},
"fesa-research-evidence": {
"description_terms": (
"Use when",
"FESA solver",
"research",
"FEM theory",
"benchmarks",
),
"body_terms": (
"docs/research/<feature-id>-research.md",
"Source Inventory",
"Source Reliability Tier",
"Candidate Benchmarks",
"Verification Relevance",
"Applicability Limits",
"Separate verified facts from inference.",
),
},
"fesa-formulation-spec": {
"description_terms": (
"Use when",
"FESA FEM",
"formulation",
"element equations",
"output recovery",
),
"body_terms": (
"docs/formulations/<feature-id>-formulation.md",
"Strong Form",
"Weak or Variational Form",
"Discretization",
"Kinematics",
"Element Equations",
"Jacobian",
"Output Recovery",
"Do not design C++ APIs.",
),
},
"fesa-numerical-review": {
"description_terms": (
"Use when",
"FESA FEM",
"numerical review",
"stability",
"implementation planning",
),
"body_terms": (
"docs/numerical-reviews/<feature-id>-review.md",
"pass-for-implementation-planning",
"rigid body modes",
"patch test",
"hourglass",
"locking",
"Jacobian",
"Do not edit formulations directly.",
),
},
"fesa-io-contract": {
"description_terms": (
"Use when",
"FESA solver",
"Abaqus .inp",
"HDF5",
"CSV",
"I/O",
),
"body_terms": (
"docs/io-definitions/<feature-id>-io.md",
"Abaqus Input Scope",
"Internal Model Contract",
"Output HDF5 Schema",
"FESA HDF5 to Reference CSV Comparison Schema",
"results.h5",
"reference/<model-id>/",
"*NODE",
"*ELEMENT",
"*MATERIAL",
"*BOUNDARY",
"*STEP",
"Do not implement parsers.",
),
},
"fesa-reference-models": {
"description_terms": (
"Use when",
"FESA",
"reference model",
"Abaqus input",
"CSV",
),
"body_terms": (
"docs/reference-models/<feature-id>-reference-models.md",
"reference/<model-id>/",
"model.inp",
"metadata.json",
"<model-id>_displacements.csv",
"<model-id>_reactions.csv",
"<model-id>_internalforces.csv",
"<model-id>_stresses.csv",
"Coverage Matrix",
"Do not generate or modify Abaqus reference CSV files.",
),
},
"fesa-cpp-msvc-tdd": {
"description_terms": (
"Use when",
"FESA solver",
"C++",
"MSVC",
"TDD",
),
"body_terms": (
"docs/implementation-plans/<feature-id>-implementation-plan.md",
"RED -> GREEN -> VERIFY",
"python -m unittest discover -s scripts -p \"test_*.py\"",
"python scripts/validate_workspace.py",
"ctest",
"configure | compile | link | test | reference-comparison",
"Do not change requirements.",
),
},
"fesa-reference-comparison": {
"description_terms": (
"Use when",
"FESA solver",
"HDF5",
"reference CSV",
"tolerance",
"comparison",
),
"body_terms": (
"docs/reference-verifications/<feature-id>-reference-verification.md",
"ARTIFACT CHECK -> COMPARE -> CLASSIFY -> REPORT",
"results.h5",
"Abaqus reference CSV",
"reference/<model-id>/",
"<model-id>_displacements.csv",
"<model-id>_reactions.csv",
"<model-id>_internalforces.csv",
"<model-id>_stresses.csv",
"max absolute error",
"max relative error",
"RMS error",
"missing rows",
"extra rows",
"pass-for-physics-evaluation",
"Do not change tolerance policies.",
),
},
"fesa-physics-sanity": {
"description_terms": (
"Use when",
"FESA solver",
"physical plausibility",
"equilibrium",
"physics",
),
"body_terms": (
"docs/physics-evaluations/<feature-id>-physics-evaluation.md",
"global equilibrium",
"reaction consistency",
"displacement direction",
"symmetry",
"element force balance",
"model coverage",
"pass-for-release-agent",
"Do not approve release readiness.",
),
},
"fesa-release-readiness": {
"description_terms": (
"Use when",
"FESA solver",
"release readiness",
"release notes",
"known limitations",
),
"body_terms": (
"docs/releases/<feature-id>-release.md",
"GATE AUDIT -> TRACEABILITY CHECK -> RELEASE DOCUMENTATION -> RELEASE VERDICT",
"ready-for-release",
"Known Limitations",
"Release Notes Draft",
"pass-for-reference-verification",
"pass-for-physics-evaluation",
"pass-for-release-agent",
"Do not publish, deploy, package, tag, commit, or externally release anything unless the user explicitly asks.",
),
},
}
def read_skill(skill_name):
return (SKILLS_ROOT / skill_name / "SKILL.md").read_text(encoding="utf-8")
def parse_frontmatter(text):
lines = text.splitlines()
if not lines or lines[0] != "---":
raise AssertionError("SKILL.md must start with YAML frontmatter")
fields = {}
for line in lines[1:]:
if line == "---":
return fields
key, sep, value = line.partition(":")
if not sep:
raise AssertionError(f"Invalid frontmatter line: {line}")
fields[key.strip()] = value.strip()
raise AssertionError("SKILL.md frontmatter must be closed")
class FesaSolverSkillTests(unittest.TestCase):
def test_all_solver_skill_files_exist_with_required_frontmatter(self):
for skill_name, spec in SKILLS.items():
with self.subTest(skill=skill_name):
skill_path = SKILLS_ROOT / skill_name / "SKILL.md"
self.assertTrue(skill_path.exists(), f"{skill_name} SKILL.md is missing")
fields = parse_frontmatter(read_skill(skill_name))
self.assertEqual(set(fields), {"name", "description"})
self.assertEqual(fields["name"], skill_name)
for term in spec["description_terms"]:
self.assertIn(term, fields["description"])
def test_all_solver_skills_define_common_contract_sections(self):
for skill_name in SKILLS:
with self.subTest(skill=skill_name):
body = read_skill(skill_name)
for section in COMMON_SECTIONS:
self.assertIn(section, body)
self.assertIn("AGENTS.md", body)
self.assertIn("docs/SOLVER_AGENT_DESIGN.md", body)
self.assertNotIn("docs/SOLVER_SKILL_DESIGN.md", body)
def test_solver_skills_define_skill_specific_contracts(self):
for skill_name, spec in SKILLS.items():
with self.subTest(skill=skill_name):
body = read_skill(skill_name)
for term in spec["body_terms"]:
self.assertIn(term, body)
def test_solver_skills_have_openai_ui_metadata(self):
for skill_name in SKILLS:
with self.subTest(skill=skill_name):
metadata = SKILLS_ROOT / skill_name / "agents" / "openai.yaml"
self.assertTrue(metadata.exists(), f"{skill_name} openai.yaml is missing")
text = metadata.read_text(encoding="utf-8")
self.assertIn("interface:", text)
self.assertIn("display_name:", text)
self.assertIn("short_description:", text)
self.assertIn("default_prompt:", text)
self.assertIn(f"${skill_name}", text)
def test_active_contracts_do_not_require_reference_hdf5(self):
for path in ACTIVE_CONTRACT_FILES:
with self.subTest(path=str(path.relative_to(ROOT))):
text = path.read_text(encoding="utf-8")
for stale_phrase in STALE_REFERENCE_CONTRACT_PHRASES:
self.assertNotIn(stale_phrase, text)
if __name__ == "__main__":
unittest.main()
-88
View File
@@ -1,88 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "formulation-agent.toml"
FORMULATIONS_README = ROOT / "docs" / "formulations" / "README.md"
class FormulationAgentConfigTests(unittest.TestCase):
def test_formulation_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "formulation-agent")
self.assertIn("FEM formulation", data["description"])
self.assertEqual(data["sandbox_mode"], "read-only")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_formulation_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not design C++ APIs",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not approve release readiness.",
"docs/SOLVER_AGENT_DESIGN.md",
"docs/requirements/<feature-id>.md",
"docs/research/<feature-id>-research.md",
):
self.assertIn(required_text, instructions)
def test_formulation_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Strong Form and Boundary Conditions",
"Weak or Variational Form",
"Discretization",
"Kinematics",
"Constitutive Contract",
"Element Equations",
"Mapping and Numerical Integration",
"Output Recovery",
"Numerical Risks",
"Downstream Handoff",
):
self.assertIn(required_text, instructions)
def test_formulation_agent_instructions_define_numerical_risk_policy(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"rigid body modes",
"patch test",
"hourglass",
"locking",
"Jacobian",
):
self.assertIn(required_text, instructions)
def test_formulation_document_guide_defines_output_contract(self):
guide = FORMULATIONS_README.read_text(encoding="utf-8")
for required_text in (
"Strong Form and Boundary Conditions",
"Weak or Variational Form",
"Discretization",
"Kinematics",
"Constitutive Contract",
"Element Equations",
"Mapping and Numerical Integration",
"Output Recovery",
"Numerical Risks",
"Downstream Handoff",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
@@ -1,81 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "implementation-agent.toml"
class ImplementationAgentConfigTests(unittest.TestCase):
def test_implementation_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "implementation-agent")
self.assertIn("C++17/MSVC", data["description"])
self.assertEqual(data["sandbox_mode"], "workspace-write")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_implementation_agent_instructions_define_tdd_execution_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Write tests first",
"verify failure",
"minimum code",
"C++17",
"MSVC",
"CMake",
"CTest",
"RED -> GREEN -> VERIFY",
):
self.assertIn(required_text, instructions)
def test_implementation_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not approve release readiness.",
"Do not change requirements",
"Do not change formulations",
"Do not change I/O contracts",
"Do not change reference artifacts",
"Do not produce the final reference verification report.",
):
self.assertIn(required_text, instructions)
def test_implementation_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Test Evidence",
"Code Changes",
"Validation Evidence",
"Traceability",
"Downstream Handoff",
"Build/Test Executor Agent",
"Correction Agent",
"Reference Verification Agent",
):
self.assertIn(required_text, instructions)
def test_implementation_agent_instructions_define_validation_commands(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"python -m unittest discover -s scripts -p \"test_*.py\"",
"python scripts/validate_workspace.py",
"ctest -C Debug",
):
self.assertIn(required_text, instructions)
if __name__ == "__main__":
unittest.main()
@@ -1,87 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "implementation-planning-agent.toml"
IMPLEMENTATION_PLANS_README = ROOT / "docs" / "implementation-plans" / "README.md"
class ImplementationPlanningAgentConfigTests(unittest.TestCase):
def test_implementation_planning_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "implementation-planning-agent")
self.assertIn("TDD-first C++/MSVC implementation plans", data["description"])
self.assertEqual(data["sandbox_mode"], "read-only")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_implementation_planning_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not write tests.",
"Do not edit CMake.",
"Do not run CMake/CTest.",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not compare solver results.",
"Do not approve release readiness.",
):
self.assertIn(required_text, instructions)
def test_implementation_planning_agent_instructions_define_tdd_msvc_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"C++17",
"MSVC",
"CMake",
"CTest",
"TDD",
"failing unit tests first",
"reference comparison tests",
):
self.assertIn(required_text, instructions)
def test_implementation_planning_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Readiness Check",
"Work Breakdown",
"TDD Test Plan",
"CMake/CTest Plan",
"Acceptance Traceability Matrix",
"Validation Commands",
"Downstream Handoff",
):
self.assertIn(required_text, instructions)
def test_implementation_planning_document_guide_defines_output_contract(self):
guide = IMPLEMENTATION_PLANS_README.read_text(encoding="utf-8")
for required_text in (
"docs/implementation-plans/<feature-id>-implementation-plan.md",
"Readiness Check",
"Work Breakdown",
"TDD Test Plan",
"CMake/CTest Plan",
"Acceptance Traceability Matrix",
"Validation Commands",
"Downstream Handoff",
"python scripts/validate_workspace.py",
"ctest -C Debug",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-101
View File
@@ -1,101 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "io-definition-agent.toml"
IO_DEFINITIONS_README = ROOT / "docs" / "io-definitions" / "README.md"
class IoDefinitionAgentConfigTests(unittest.TestCase):
def test_io_definition_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "io-definition-agent")
self.assertIn("Abaqus input-file subsets", data["description"])
self.assertEqual(data["sandbox_mode"], "read-only")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_io_definition_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement parsers.",
"Do not design C++ APIs",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not approve release readiness.",
"Do not claim full Abaqus compatibility",
):
self.assertIn(required_text, instructions)
def test_io_definition_agent_instructions_define_abaqus_input_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"FESA solver input files are Abaqus input files.",
"Abaqus input files use keyword lines, data lines, and comment lines.",
"Model data and history data",
"supported Abaqus keyword subset",
"HDF5 result schema",
"reference CSV comparison row schema",
):
self.assertIn(required_text, instructions)
def test_io_definition_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Abaqus Input Scope",
"Syntax Policy",
"Model Data Mapping",
"History Data Mapping",
"Internal Model Contract",
"Output HDF5 Schema",
"FESA HDF5 to Reference CSV Comparison Schema",
"Validation Rules",
"Downstream Handoff",
):
self.assertIn(required_text, instructions)
def test_io_definition_agent_instructions_define_keyword_policy(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"*NODE",
"*ELEMENT",
"*MATERIAL",
"*BOUNDARY",
"*STEP",
"*OUTPUT",
"*NODE OUTPUT",
"*ELEMENT OUTPUT",
):
self.assertIn(required_text, instructions)
def test_io_definition_document_guide_defines_output_contract(self):
guide = IO_DEFINITIONS_README.read_text(encoding="utf-8")
for required_text in (
"Abaqus Input Scope",
"Syntax Policy",
"Model Data Mapping",
"History Data Mapping",
"Internal Model Contract",
"Output HDF5 Schema",
"FESA HDF5 to Reference CSV Comparison Schema",
"Validation Rules",
"Downstream Handoff",
"FESA 솔버의 입력 파일은 Abaqus input file이다.",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
@@ -1,86 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "numerical-review-agent.toml"
NUMERICAL_REVIEWS_README = ROOT / "docs" / "numerical-reviews" / "README.md"
class NumericalReviewAgentConfigTests(unittest.TestCase):
def test_numerical_review_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "numerical-review-agent")
self.assertIn("numerical correctness", data["description"])
self.assertEqual(data["sandbox_mode"], "read-only")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_numerical_review_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not edit formulations directly.",
"Do not design C++ APIs",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not approve release readiness.",
"docs/SOLVER_AGENT_DESIGN.md",
"docs/formulations/<feature-id>-formulation.md",
):
self.assertIn(required_text, instructions)
def test_numerical_review_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Review Verdict",
"Critical Findings",
"Numerical Risk Assessment",
"Consistency Checks",
"Verification Readiness",
"Required Revisions",
"Downstream Handoff",
):
self.assertIn(required_text, instructions)
def test_numerical_review_agent_instructions_define_risk_policy(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"rigid body modes",
"patch test",
"symmetry",
"positive definiteness",
"hourglass",
"locking",
"singular Jacobian",
"conditioning",
):
self.assertIn(required_text, instructions)
def test_numerical_review_document_guide_defines_output_contract(self):
guide = NUMERICAL_REVIEWS_README.read_text(encoding="utf-8")
for required_text in (
"Review Verdict",
"Critical Findings",
"Numerical Risk Assessment",
"Consistency Checks",
"Verification Readiness",
"Required Revisions",
"Downstream Handoff",
"pass-for-implementation-planning",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
@@ -1,89 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "physics-evaluation-agent.toml"
PHYSICS_EVALUATIONS_README = ROOT / "docs" / "physics-evaluations" / "README.md"
class PhysicsEvaluationAgentConfigTests(unittest.TestCase):
def test_physics_evaluation_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "physics-evaluation-agent")
self.assertIn("physical plausibility", data["description"])
self.assertEqual(data["sandbox_mode"], "workspace-write")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_physics_evaluation_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not edit source code.",
"Do not edit tests.",
"Do not edit CMake.",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not change tolerances.",
"Do not approve release readiness.",
):
self.assertIn(required_text, instructions)
def test_physics_evaluation_agent_instructions_define_physics_checks(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"global equilibrium",
"reaction consistency",
"displacement direction",
"symmetry",
"element force balance",
"stress/strain",
"rigid body mode",
"energy/residual",
):
self.assertIn(required_text, instructions)
def test_physics_evaluation_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"HDF5",
"Abaqus reference CSV files",
"Input Evidence",
"Physics Checks",
"Failure Classification",
"Evaluation Verdict",
"Handoff Recommendation",
"No-Change Assertion",
):
self.assertIn(required_text, instructions)
def test_physics_evaluation_report_guide_defines_template_and_status_values(self):
guide = PHYSICS_EVALUATIONS_README.read_text(encoding="utf-8")
for required_text in (
"docs/physics-evaluations/<feature-id>-physics-evaluation.md",
"Input Evidence",
"Physics Checks",
"Failure Classification",
"Evaluation Verdict",
"Handoff Recommendation",
"No-Change Assertion",
"pass-for-release-agent",
"needs-correction",
"needs-reference-model",
"needs-upstream-decision",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-41
View File
@@ -1,41 +0,0 @@
import importlib.util
import sys
import tempfile
import unittest
from pathlib import Path
def load_pre_commit_checks():
module_path = Path(__file__).resolve().parent.parent / ".codex" / "hooks" / "pre_commit_checks.py"
spec = importlib.util.spec_from_file_location("pre_commit_checks", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class PreCommitChecksTests(unittest.TestCase):
def test_git_commit_runs_python_self_tests_and_workspace_validation(self):
pre_commit_checks = load_pre_commit_checks()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
commands = pre_commit_checks._build_pre_commit_commands(root)
self.assertEqual(
commands,
[
[sys.executable, "-m", "unittest", "discover", "-s", "scripts", "-p", "test_*.py"],
[sys.executable, "scripts/validate_workspace.py"],
],
)
self.assertFalse(any("npm" in part.lower() for command in commands for part in command))
def test_only_git_commit_commands_trigger_checks(self):
pre_commit_checks = load_pre_commit_checks()
self.assertTrue(pre_commit_checks._is_git_commit('git commit -m "change"'))
self.assertTrue(pre_commit_checks._is_git_commit('git -c core.editor=true commit -m "change"'))
self.assertFalse(pre_commit_checks._is_git_commit("git status --short"))
self.assertFalse(pre_commit_checks._is_git_commit("echo git commit"))
if __name__ == "__main__":
unittest.main()
@@ -1,90 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "reference-model-agent.toml"
REFERENCE_MODELS_README = ROOT / "docs" / "reference-models" / "README.md"
class ReferenceModelAgentConfigTests(unittest.TestCase):
def test_reference_model_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "reference-model-agent")
self.assertIn("reference model packages", data["description"])
self.assertEqual(data["sandbox_mode"], "read-only")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_reference_model_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not implement parsers.",
"Do not design C++ APIs",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not compare solver results.",
"Do not approve release readiness.",
):
self.assertIn(required_text, instructions)
def test_reference_model_agent_instructions_define_reference_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"FESA reference models use Abaqus input files.",
"reference/<model-id>/",
"model.inp",
"metadata.json",
"<model-id>_displacements.csv",
"<model-id>_reactions.csv",
"<model-id>_internalforces.csv",
"<model-id>_stresses.csv",
):
self.assertIn(required_text, instructions)
self.assertNotIn("reference" ".h5", instructions)
self.assertNotIn("references/" "<feature-id>/<model-id>/", instructions)
def test_reference_model_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Reference Strategy",
"Model Inventory",
"Abaqus Input Requirements",
"Artifact Bundle Contract",
"Metadata JSON Contract",
"Abaqus Reference CSV Requirements",
"Coverage Matrix",
"Downstream Handoff",
):
self.assertIn(required_text, instructions)
def test_reference_model_document_guide_defines_output_contract(self):
guide = REFERENCE_MODELS_README.read_text(encoding="utf-8")
for required_text in (
"Reference Strategy",
"Model Inventory",
"Abaqus Input Requirements",
"Artifact Bundle Contract",
"Metadata JSON Contract",
"Abaqus Reference CSV Requirements",
"Coverage Matrix",
"Downstream Handoff",
"reference/<model-id>/",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
@@ -1,91 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "reference-verification-agent.toml"
REFERENCE_VERIFICATIONS_README = ROOT / "docs" / "reference-verifications" / "README.md"
class ReferenceVerificationAgentConfigTests(unittest.TestCase):
def test_reference_verification_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "reference-verification-agent")
self.assertIn("HDF5", data["description"])
self.assertIn("Abaqus reference CSV", data["description"])
self.assertEqual(data["sandbox_mode"], "workspace-write")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_reference_verification_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not edit source code.",
"Do not edit tests.",
"Do not edit CMake.",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not approve release readiness.",
"Do not change tolerance policies.",
):
self.assertIn(required_text, instructions)
def test_reference_verification_agent_instructions_define_artifact_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"results.h5",
"Abaqus reference CSV",
"reference/<model-id>/",
"<model-id>_displacements.csv",
"<model-id>_reactions.csv",
"<model-id>_internalforces.csv",
"<model-id>_stresses.csv",
"metadata.json",
):
self.assertIn(required_text, instructions)
self.assertNotIn("reference" ".h5", instructions)
self.assertNotIn("stored reference " "HDF5", instructions)
def test_reference_verification_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Artifact Inventory",
"Comparison Contract",
"Quantity Results",
"Failure Classification",
"Handoff Recommendation",
"No-Change Assertion",
):
self.assertIn(required_text, instructions)
def test_reference_verification_report_guide_defines_template_and_status_values(self):
guide = REFERENCE_VERIFICATIONS_README.read_text(encoding="utf-8")
for required_text in (
"docs/reference-verifications/<feature-id>-reference-verification.md",
"Artifact Inventory",
"Comparison Contract",
"Quantity Results",
"Failure Classification",
"Handoff Recommendation",
"No-Change Assertion",
"pass-for-physics-evaluation",
"needs-correction",
"needs-reference-artifacts",
"needs-upstream-decision",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-92
View File
@@ -1,92 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "release-agent.toml"
RELEASES_README = ROOT / "docs" / "releases" / "README.md"
class ReleaseAgentConfigTests(unittest.TestCase):
def test_release_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "release-agent")
self.assertIn("release readiness", data["description"])
self.assertEqual(data["sandbox_mode"], "workspace-write")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_release_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not edit source code.",
"Do not edit tests.",
"Do not edit CMake.",
"Do not change requirements",
"Do not change formulations",
"Do not change I/O contracts",
"Do not change reference artifacts",
"Do not change tolerance policies",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Do not override failed or missing upstream gates.",
):
self.assertIn(required_text, instructions)
def test_release_agent_instructions_define_gate_and_status_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"GATE AUDIT -> TRACEABILITY CHECK -> RELEASE DOCUMENTATION -> RELEASE VERDICT",
"pass-for-release-agent",
"pass-for-physics-evaluation",
"pass-for-reference-verification",
"ready-for-release",
"known limitations",
):
self.assertIn(required_text, instructions)
def test_release_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Gate Evidence Inventory",
"Acceptance Traceability",
"Validation Evidence",
"Known Limitations",
"Release Notes Draft",
"Release Verdict",
"No-Change Assertion",
):
self.assertIn(required_text, instructions)
def test_release_report_guide_defines_template_and_status_values(self):
guide = RELEASES_README.read_text(encoding="utf-8")
for required_text in (
"docs/releases/<feature-id>-release.md",
"Gate Evidence Inventory",
"Acceptance Traceability",
"Validation Evidence",
"Known Limitations",
"Release Notes Draft",
"Release Verdict",
"No-Change Assertion",
"ready-for-release",
"needs-documentation",
"needs-upstream-decision",
"python scripts/validate_workspace.py",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-55
View File
@@ -1,55 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "requirement-agent.toml"
REQUIREMENTS_README = ROOT / "docs" / "requirements" / "README.md"
class RequirementAgentConfigTests(unittest.TestCase):
def test_requirement_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "requirement-agent")
self.assertIn("verifiable requirements", data["description"])
self.assertEqual(data["sandbox_mode"], "read-only")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_requirement_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not write finite element formulations.",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"Requirement Verification Matrix",
"docs/SOLVER_AGENT_DESIGN.md",
"reference/<model-id>/",
):
self.assertIn(required_text, instructions)
def test_requirement_document_guide_defines_output_contract(self):
guide = REQUIREMENTS_README.read_text(encoding="utf-8")
for required_text in (
"feature_id",
"Verification Quantities",
"Tolerance Policy",
"Reference Artifact Requirements",
"Requirement Verification Matrix",
"Downstream Handoff",
"FESA-REQ-<FEATURE>-001",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-79
View File
@@ -1,79 +0,0 @@
import unittest
from pathlib import Path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
import tomli as tomllib
ROOT = Path(__file__).resolve().parents[1]
AGENT_PATH = ROOT / ".codex" / "agents" / "research-agent.toml"
RESEARCH_README = ROOT / "docs" / "research" / "README.md"
class ResearchAgentConfigTests(unittest.TestCase):
def test_research_agent_toml_has_required_codex_fields(self):
data = tomllib.loads(AGENT_PATH.read_text(encoding="utf-8"))
self.assertEqual(data["name"], "research-agent")
self.assertIn("FEM theory", data["description"])
self.assertEqual(data["sandbox_mode"], "read-only")
self.assertEqual(data["model_reasoning_effort"], "extra high")
self.assertIn("developer_instructions", data)
def test_research_agent_instructions_enforce_boundaries(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Do not implement code.",
"Do not finalize FEM formulations.",
"Do not run Abaqus, Nastran, or any reference solver.",
"Do not generate or modify Abaqus reference CSV files.",
"docs/SOLVER_AGENT_DESIGN.md",
"docs/requirements/<feature-id>.md",
"Separate verified facts from inference.",
):
self.assertIn(required_text, instructions)
def test_research_agent_instructions_define_output_contract(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"Source Inventory",
"Candidate Benchmarks",
"Verification Relevance",
"Applicability Limits",
"Downstream Handoff",
):
self.assertIn(required_text, instructions)
def test_research_agent_instructions_define_source_policy(self):
instructions = AGENT_PATH.read_text(encoding="utf-8")
for required_text in (
"ASME V&V 10",
"Abaqus Verification Guide",
"Abaqus Benchmarks Guide",
"NAFEMS benchmarks",
"NASA FEMCI",
"MMS and MES papers",
):
self.assertIn(required_text, instructions)
def test_research_document_guide_defines_output_contract(self):
guide = RESEARCH_README.read_text(encoding="utf-8")
for required_text in (
"Source Reliability Tier",
"Source Inventory",
"Candidate Benchmarks",
"Verification Relevance",
"Applicability Limits",
"Downstream Handoff",
):
self.assertIn(required_text, guide)
if __name__ == "__main__":
unittest.main()
-61
View File
@@ -1,61 +0,0 @@
import importlib.util
import tempfile
import unittest
from pathlib import Path
def load_tdd_guard():
module_path = Path(__file__).resolve().parent.parent / ".codex" / "hooks" / "tdd-guard.py"
spec = importlib.util.spec_from_file_location("tdd_guard", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class CppTddGuardTests(unittest.TestCase):
def test_cpp_production_file_without_related_test_is_blocked(self):
tdd_guard = load_tdd_guard()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "include" / "fesa" / "Core" / "DofManager.hpp"
source.parent.mkdir(parents=True)
source.write_text("#pragma once\n", encoding="utf-8")
self.assertEqual(tdd_guard._guarded_paths([str(source)], root, root), ["DofManager"])
def test_cpp_production_file_with_module_test_is_allowed(self):
tdd_guard = load_tdd_guard()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "include" / "fesa" / "Core" / "DofManager.hpp"
source.parent.mkdir(parents=True)
source.write_text("#pragma once\n", encoding="utf-8")
tests_dir = root / "tests"
tests_dir.mkdir()
(tests_dir / "test_core_module_includes.cpp").write_text("int main() { return 0; }\n", encoding="utf-8")
self.assertEqual(tdd_guard._guarded_paths([str(source)], root, root), [])
def test_cpp_production_file_with_basename_test_in_same_patch_is_allowed(self):
tdd_guard = load_tdd_guard()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source = root / "src" / "Math" / "DenseMatrix.cpp"
source.parent.mkdir(parents=True)
source.write_text("void f() {}\n", encoding="utf-8")
test_path = root / "tests" / "test_dense_matrix.cpp"
self.assertEqual(tdd_guard._guarded_paths([str(source), str(test_path)], root, root), [])
def test_cmake_and_docs_are_exempt(self):
tdd_guard = load_tdd_guard()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
self.assertEqual(
tdd_guard._guarded_paths(["CMakeLists.txt", "docs/ARCHITECTURE.md", "cmake/toolchain.cmake"], root, root),
[],
)
if __name__ == "__main__":
unittest.main()
-94
View File
@@ -1,94 +0,0 @@
import importlib.util
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
def load_validate_workspace():
module_path = Path(__file__).resolve().parent / "validate_workspace.py"
spec = importlib.util.spec_from_file_location("validate_workspace", module_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class ValidateWorkspaceTests(unittest.TestCase):
def test_env_commands_override_cmake_detection(self):
validate_workspace = load_validate_workspace()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / "CMakeLists.txt").write_text("cmake_minimum_required(VERSION 3.20)\n", encoding="utf-8")
with patch.dict(os.environ, {"HARNESS_VALIDATION_COMMANDS": "echo first\n echo second \n"}, clear=True):
self.assertEqual(validate_workspace.discover_commands(root), ["echo first", "echo second"])
def test_msvc_debug_cmake_commands_are_default_for_cmake_project(self):
validate_workspace = load_validate_workspace()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / "CMakeLists.txt").write_text("cmake_minimum_required(VERSION 3.20)\n", encoding="utf-8")
build_dir = root / "build" / "msvc-debug"
with patch.dict(os.environ, {}, clear=True):
self.assertEqual(
validate_workspace.discover_commands(root),
[
f'cmake -S "{root}" -B "{build_dir}" -G "Visual Studio 17 2022" -A x64',
f'cmake --build "{build_dir}" --config Debug',
f'ctest --test-dir "{build_dir}" --output-on-failure -C Debug',
],
)
def test_msvc_debug_configure_preset_is_preferred_when_present(self):
validate_workspace = load_validate_workspace()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / "CMakeLists.txt").write_text("cmake_minimum_required(VERSION 3.20)\n", encoding="utf-8")
(root / "CMakePresets.json").write_text(
"""
{
"version": 3,
"configurePresets": [
{
"name": "msvc-debug",
"generator": "Visual Studio 17 2022",
"binaryDir": "${sourceDir}/out/msvc-debug"
}
]
}
""",
encoding="utf-8",
)
with patch.dict(os.environ, {}, clear=True):
self.assertEqual(
validate_workspace.discover_commands(root),
[
"cmake --preset msvc-debug",
f'cmake --build "{root / "out" / "msvc-debug"}" --config Debug',
f'ctest --test-dir "{root / "out" / "msvc-debug"}" --output-on-failure -C Debug',
],
)
def test_no_cmake_project_has_no_validation_commands(self):
validate_workspace = load_validate_workspace()
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
with patch.dict(os.environ, {}, clear=True):
self.assertEqual(validate_workspace.discover_commands(root), [])
def test_common_cmake_install_path_is_prepended_when_cmake_is_not_on_path(self):
validate_workspace = load_validate_workspace()
with tempfile.TemporaryDirectory() as tmp:
common_bin = Path(tmp) / "CMake" / "bin"
common_bin.mkdir(parents=True)
(common_bin / "cmake.exe").write_text("", encoding="utf-8")
(common_bin / "ctest.exe").write_text("", encoding="utf-8")
with patch.object(validate_workspace, "COMMON_CMAKE_BIN", common_bin):
with patch.object(validate_workspace.shutil, "which", return_value=None):
env = validate_workspace.validation_environment({"PATH": "C:\\Windows\\System32"})
self.assertTrue(env["PATH"].startswith(str(common_bin)))
if __name__ == "__main__":
unittest.main()
-151
View File
@@ -1,151 +0,0 @@
#!/usr/bin/env python3
"""Run C++/MSVC Harness validation commands."""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
DEFAULT_GENERATOR = "Visual Studio 17 2022"
DEFAULT_PLATFORM = "x64"
DEFAULT_CONFIG = "Debug"
DEFAULT_BUILD_DIR = "build/msvc-debug"
PRESET_NAME = "msvc-debug"
COMMON_CMAKE_BIN = Path(r"C:\Program Files\CMake\bin")
def load_env_commands() -> list[str]:
raw = os.environ.get("HARNESS_VALIDATION_COMMANDS", "")
return [line.strip() for line in raw.splitlines() if line.strip()]
def _cmake_config() -> tuple[str, str, str, Path]:
generator = os.environ.get("HARNESS_CMAKE_GENERATOR", DEFAULT_GENERATOR)
platform = os.environ.get("HARNESS_CMAKE_PLATFORM", DEFAULT_PLATFORM)
config = os.environ.get("HARNESS_CMAKE_CONFIG", DEFAULT_CONFIG)
build_dir = Path(os.environ.get("HARNESS_BUILD_DIR", DEFAULT_BUILD_DIR))
return generator, platform, config, build_dir
def _read_presets(root: Path) -> dict:
presets_file = root / "CMakePresets.json"
if not presets_file.exists():
return {}
try:
return json.loads(presets_file.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return {}
def _preset_binary_dir(root: Path, preset: dict) -> Path:
binary_dir = str(preset.get("binaryDir") or DEFAULT_BUILD_DIR)
binary_dir = binary_dir.replace("${sourceDir}", str(root))
binary_dir = binary_dir.replace("$sourceDir", str(root))
path = Path(binary_dir)
return path if path.is_absolute() else root / path
def load_preset_commands(root: Path) -> list[str]:
payload = _read_presets(root)
config = os.environ.get("HARNESS_CMAKE_CONFIG", DEFAULT_CONFIG)
for preset in payload.get("configurePresets", []):
if isinstance(preset, dict) and preset.get("name") == PRESET_NAME:
build_dir = _preset_binary_dir(root, preset)
return [
f"cmake --preset {PRESET_NAME}",
f'cmake --build "{build_dir}" --config {config}',
f'ctest --test-dir "{build_dir}" --output-on-failure -C {config}',
]
return []
def load_cmake_commands(root: Path) -> list[str]:
if not (root / "CMakeLists.txt").exists():
return []
generator, platform, config, build_dir = _cmake_config()
if not build_dir.is_absolute():
build_dir = root / build_dir
return [
f'cmake -S "{root}" -B "{build_dir}" -G "{generator}" -A {platform}',
f'cmake --build "{build_dir}" --config {config}',
f'ctest --test-dir "{build_dir}" --output-on-failure -C {config}',
]
def discover_commands(root: Path) -> list[str]:
env_commands = load_env_commands()
if env_commands:
return env_commands
preset_commands = load_preset_commands(root)
if preset_commands:
return preset_commands
return load_cmake_commands(root)
def run_command(command: str, root: Path) -> subprocess.CompletedProcess:
return subprocess.run(
command,
cwd=root,
shell=True,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
env=validation_environment(os.environ),
)
def validation_environment(base_env: os._Environ | dict[str, str]) -> dict[str, str]:
env = dict(base_env)
if shutil.which("cmake") is not None:
return env
cmake_exe = COMMON_CMAKE_BIN / "cmake.exe"
if not cmake_exe.exists():
return env
current_path = env.get("PATH", "")
paths = [part for part in current_path.split(os.pathsep) if part]
common_bin_text = str(COMMON_CMAKE_BIN)
if not any(part.lower() == common_bin_text.lower() for part in paths):
env["PATH"] = common_bin_text + (os.pathsep + current_path if current_path else "")
return env
def emit_stream(prefix: str, content: str, *, stream) -> None:
text = (content or "").strip()
if not text:
return
print(prefix, file=stream)
print(text, file=stream)
def main() -> int:
root = Path(__file__).resolve().parent.parent
commands = discover_commands(root)
if not commands:
print("No C++ validation commands configured.")
print("Add CMakeLists.txt or set HARNESS_VALIDATION_COMMANDS.")
return 0
for command in commands:
print(f"$ {command}")
result = run_command(command, root)
emit_stream("[stdout]", result.stdout, stream=sys.stdout)
emit_stream("[stderr]", result.stderr, stream=sys.stderr)
if result.returncode != 0:
print(f"Validation failed: {command}", file=sys.stderr)
return result.returncode
print("Validation succeeded.")
return 0
if __name__ == "__main__":
raise SystemExit(main())