baram2584/Agentic-AI-Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modify coding template

Browse Source
This commit is contained in:
NINI
2026-05-09 02:20:01 +09:00
parent 38291723f0
commit a1aea6a449
55 changed files with 437 additions and 2862 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Coding/Codex/scripts/__pycache__/execute.cpython-312.pyc
BIN
View File
Binary file not shown.
Coding/Codex/scripts/__pycache__/validate_workspace.cpython-312.pyc
BIN
View File
Binary file not shown.
Coding/Codex/scripts/execute.py
+11 -18
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Harness Step Executor - run phase steps sequentially with Codex and self-correction.
Harness Step Executor — phase 내 step을 순차 실행하고 자가 교정한다.
Usage:
python scripts/execute.py <phase-dir> [--push]
@@ -9,6 +9,7 @@ Usage:
import argparse
import contextlib
import json
import os
import subprocess
import sys
import threading
@@ -53,7 +54,7 @@ class StepExecutor:
"""Phase 디렉토리 안의 step들을 순차 실행하는 하네스."""
MAX_RETRIES = 3
FEAT_MSG = "feat({phase}): step {num} - {name}"
FEAT_MSG = "feat({phase}): step {num} — {name}"
CHORE_MSG = "chore({phase}): step {num} output"
TZ = timezone(timedelta(hours=9))
@@ -177,9 +178,7 @@ 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(f"## 프로젝트 규칙 (AGENTS.md)\n\n{agents_md.read_text(encoding='utf-8')}")
docs_dir = ROOT / "docs"
if docs_dir.is_dir():
for doc in sorted(docs_dir.glob("*.md")):
@@ -199,6 +198,9 @@ class StepExecutor:
def _build_preamble(self, guardrails: str, step_context: 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 = (
@@ -218,7 +220,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. 변경사항은 워킹 트리에 남겨라. step 완료 후 커밋은 execute.py가 정리한다.\n\n---\n\n"
f"6. 모든 변경사항을 커밋하라:\n"
f" {commit_example}\n\n---\n\n"
)
# --- Codex 호출 ---
@@ -232,14 +235,9 @@ class StepExecutor:
sys.exit(1)
prompt = preamble + step_file.read_text(encoding="utf-8")
last_message_path = self._phase_dir / f"step{step_num}-last-message.txt"
result = subprocess.run(
["codex", "exec", "--full-auto", "--json", "-C", self._root, "-o", str(last_message_path)],
cwd=self._root,
input=prompt,
capture_output=True,
text=True,
timeout=1800,
["codex", "exec", "--dangerously-bypass-approvals-and-sandbox", "--json", prompt],
cwd=self._root, capture_output=True, text=True, timeout=1800,
)
if result.returncode != 0:
@@ -247,14 +245,9 @@ class StepExecutor:
if result.stderr:
print(f" stderr: {result.stderr[:500]}")
final_message = None
if last_message_path.exists():
final_message = last_message_path.read_text(encoding="utf-8")
output = {
"step": step_num, "name": step_name,
"exitCode": result.returncode,
"finalMessage": final_message,
"stdout": result.stdout, "stderr": result.stderr,
}
out_path = self._phase_dir / f"step{step_num}-output.json"
Coding/Codex/scripts/test_execute.py
-562
View File
@@ -1,562 +0,0 @@
"""execute.py Codex migration safety-net tests."""
import json
import sys
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
sys.path.insert(0, str(Path(__file__).parent))
import execute as ex
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def tmp_project(tmp_path):
"""phases/, AGENTS.md, docs/ 를 갖춘 임시 프로젝트 구조."""
phases_dir = tmp_path / "phases"
phases_dir.mkdir()
agents_md = tmp_path / "AGENTS.md"
agents_md.write_text("# Rules\n- rule one\n- rule two")
docs_dir = tmp_path / "docs"
docs_dir.mkdir()
(docs_dir / "arch.md").write_text("# Architecture\nSome content")
(docs_dir / "guide.md").write_text("# Guide\nAnother doc")
return tmp_path
@pytest.fixture
def phase_dir(tmp_project):
"""step 3개를 가진 phase 디렉토리."""
d = tmp_project / "phases" / "0-mvp"
d.mkdir()
index = {
"project": "TestProject",
"phase": "mvp",
"steps": [
{"step": 0, "name": "setup", "status": "completed", "summary": "프로젝트 초기화 완료"},
{"step": 1, "name": "core", "status": "completed", "summary": "핵심 로직 구현"},
{"step": 2, "name": "ui", "status": "pending"},
],
}
(d / "index.json").write_text(json.dumps(index, indent=2, ensure_ascii=False))
(d / "step2.md").write_text("# Step 2: UI\n\nUI를 구현하세요.")
return d
@pytest.fixture
def top_index(tmp_project):
"""phases/index.json (top-level)."""
top = {
"phases": [
{"dir": "0-mvp", "status": "pending"},
{"dir": "1-polish", "status": "pending"},
]
}
p = tmp_project / "phases" / "index.json"
p.write_text(json.dumps(top, indent=2))
return p
@pytest.fixture
def executor(tmp_project, phase_dir):
"""테스트용 StepExecutor 인스턴스. git 호출은 별도 mock 필요."""
with patch.object(ex, "ROOT", tmp_project):
inst = ex.StepExecutor("0-mvp")
# 내부 경로를 tmp_project 기준으로 재설정
inst._root = str(tmp_project)
inst._phases_dir = tmp_project / "phases"
inst._phase_dir = phase_dir
inst._phase_dir_name = "0-mvp"
inst._index_file = phase_dir / "index.json"
inst._top_index_file = tmp_project / "phases" / "index.json"
return inst
# ---------------------------------------------------------------------------
# _stamp (= 이전 now_iso)
# ---------------------------------------------------------------------------
class TestStamp:
def test_returns_kst_timestamp(self, executor):
result = executor._stamp()
assert "+0900" in result
def test_format_is_iso(self, executor):
result = executor._stamp()
dt = datetime.strptime(result, "%Y-%m-%dT%H:%M:%S%z")
assert dt.tzinfo is not None
def test_is_current_time(self, executor):
before = datetime.now(ex.StepExecutor.TZ).replace(microsecond=0)
result = executor._stamp()
after = datetime.now(ex.StepExecutor.TZ).replace(microsecond=0) + timedelta(seconds=1)
parsed = datetime.strptime(result, "%Y-%m-%dT%H:%M:%S%z")
assert before <= parsed <= after
# ---------------------------------------------------------------------------
# _read_json / _write_json
# ---------------------------------------------------------------------------
class TestJsonHelpers:
def test_roundtrip(self, tmp_path):
data = {"key": "값", "nested": [1, 2, 3]}
p = tmp_path / "test.json"
ex.StepExecutor._write_json(p, data)
loaded = ex.StepExecutor._read_json(p)
assert loaded == data
def test_save_ensures_ascii_false(self, tmp_path):
p = tmp_path / "test.json"
ex.StepExecutor._write_json(p, {"한글": "테스트"})
raw = p.read_text()
assert "한글" in raw
assert "\\u" not in raw
def test_save_indented(self, tmp_path):
p = tmp_path / "test.json"
ex.StepExecutor._write_json(p, {"a": 1})
raw = p.read_text()
assert "\n" in raw
def test_load_nonexistent_raises(self, tmp_path):
with pytest.raises(FileNotFoundError):
ex.StepExecutor._read_json(tmp_path / "nope.json")
# ---------------------------------------------------------------------------
# _load_guardrails
# ---------------------------------------------------------------------------
class TestLoadGuardrails:
def test_loads_agents_md_and_docs(self, executor, tmp_project):
with patch.object(ex, "ROOT", tmp_project):
result = executor._load_guardrails()
assert "# Rules" in result
assert "rule one" in result
assert "# Architecture" in result
assert "# Guide" in result
def test_sections_separated_by_divider(self, executor, tmp_project):
with patch.object(ex, "ROOT", tmp_project):
result = executor._load_guardrails()
assert "---" in result
def test_docs_sorted_alphabetically(self, executor, tmp_project):
with patch.object(ex, "ROOT", tmp_project):
result = executor._load_guardrails()
arch_pos = result.index("arch")
guide_pos = result.index("guide")
assert arch_pos < guide_pos
def test_no_agents_md(self, executor, tmp_project):
(tmp_project / "AGENTS.md").unlink()
with patch.object(ex, "ROOT", tmp_project):
result = executor._load_guardrails()
assert "AGENTS.md" not in result
assert "Architecture" in result
def test_no_docs_dir(self, executor, tmp_project):
import shutil
shutil.rmtree(tmp_project / "docs")
with patch.object(ex, "ROOT", tmp_project):
result = executor._load_guardrails()
assert "Rules" in result
assert "Architecture" not in result
def test_empty_project(self, tmp_path):
with patch.object(ex, "ROOT", tmp_path):
# executor가 필요 없는 static-like 동작이므로 임시 인스턴스
phases_dir = tmp_path / "phases" / "dummy"
phases_dir.mkdir(parents=True)
idx = {"project": "T", "phase": "t", "steps": []}
(phases_dir / "index.json").write_text(json.dumps(idx))
inst = ex.StepExecutor.__new__(ex.StepExecutor)
result = inst._load_guardrails()
assert result == ""
# ---------------------------------------------------------------------------
# _build_step_context
# ---------------------------------------------------------------------------
class TestBuildStepContext:
def test_includes_completed_with_summary(self, phase_dir):
index = json.loads((phase_dir / "index.json").read_text())
result = ex.StepExecutor._build_step_context(index)
assert "Step 0 (setup): 프로젝트 초기화 완료" in result
assert "Step 1 (core): 핵심 로직 구현" in result
def test_excludes_pending(self, phase_dir):
index = json.loads((phase_dir / "index.json").read_text())
result = ex.StepExecutor._build_step_context(index)
assert "ui" not in result
def test_excludes_completed_without_summary(self, phase_dir):
index = json.loads((phase_dir / "index.json").read_text())
del index["steps"][0]["summary"]
result = ex.StepExecutor._build_step_context(index)
assert "setup" not in result
assert "core" in result
def test_empty_when_no_completed(self):
index = {"steps": [{"step": 0, "name": "a", "status": "pending"}]}
result = ex.StepExecutor._build_step_context(index)
assert result == ""
def test_has_header(self, phase_dir):
index = json.loads((phase_dir / "index.json").read_text())
result = ex.StepExecutor._build_step_context(index)
assert result.startswith("## 이전 Step 산출물")
# ---------------------------------------------------------------------------
# _build_preamble
# ---------------------------------------------------------------------------
class TestBuildPreamble:
def test_includes_project_name(self, executor):
result = executor._build_preamble("", "")
assert "TestProject" in result
def test_includes_guardrails(self, executor):
result = executor._build_preamble("GUARD_CONTENT", "")
assert "GUARD_CONTENT" in result
def test_includes_step_context(self, executor):
ctx = "## 이전 Step 산출물\n\n- Step 0: done"
result = executor._build_preamble("", ctx)
assert "이전 Step 산출물" in result
def test_mentions_executor_commits(self, executor):
result = executor._build_preamble("", "")
assert "커밋은 execute.py가 정리한다" in result
def test_includes_rules(self, executor):
result = executor._build_preamble("", "")
assert "작업 규칙" in result
assert "AC" in result
def test_no_retry_section_by_default(self, executor):
result = executor._build_preamble("", "")
assert "이전 시도 실패" not in result
def test_retry_section_with_prev_error(self, executor):
result = executor._build_preamble("", "", prev_error="타입 에러 발생")
assert "이전 시도 실패" in result
assert "타입 에러 발생" in result
def test_includes_max_retries(self, executor):
result = executor._build_preamble("", "")
assert str(ex.StepExecutor.MAX_RETRIES) in result
def test_includes_index_path(self, executor):
result = executor._build_preamble("", "")
assert "/phases/0-mvp/index.json" in result
# ---------------------------------------------------------------------------
# _update_top_index
# ---------------------------------------------------------------------------
class TestUpdateTopIndex:
def test_completed(self, executor, top_index):
executor._top_index_file = top_index
executor._update_top_index("completed")
data = json.loads(top_index.read_text())
mvp = next(p for p in data["phases"] if p["dir"] == "0-mvp")
assert mvp["status"] == "completed"
assert "completed_at" in mvp
def test_error(self, executor, top_index):
executor._top_index_file = top_index
executor._update_top_index("error")
data = json.loads(top_index.read_text())
mvp = next(p for p in data["phases"] if p["dir"] == "0-mvp")
assert mvp["status"] == "error"
assert "failed_at" in mvp
def test_blocked(self, executor, top_index):
executor._top_index_file = top_index
executor._update_top_index("blocked")
data = json.loads(top_index.read_text())
mvp = next(p for p in data["phases"] if p["dir"] == "0-mvp")
assert mvp["status"] == "blocked"
assert "blocked_at" in mvp
def test_other_phases_unchanged(self, executor, top_index):
executor._top_index_file = top_index
executor._update_top_index("completed")
data = json.loads(top_index.read_text())
polish = next(p for p in data["phases"] if p["dir"] == "1-polish")
assert polish["status"] == "pending"
def test_nonexistent_dir_is_noop(self, executor, top_index):
executor._top_index_file = top_index
executor._phase_dir_name = "no-such-dir"
original = json.loads(top_index.read_text())
executor._update_top_index("completed")
after = json.loads(top_index.read_text())
for p_before, p_after in zip(original["phases"], after["phases"]):
assert p_before["status"] == p_after["status"]
def test_no_top_index_file(self, executor, tmp_path):
executor._top_index_file = tmp_path / "nonexistent.json"
executor._update_top_index("completed") # should not raise
# ---------------------------------------------------------------------------
# _checkout_branch (mocked)
# ---------------------------------------------------------------------------
class TestCheckoutBranch:
def _mock_git(self, executor, responses):
call_idx = {"i": 0}
def fake_git(*args):
idx = call_idx["i"]
call_idx["i"] += 1
if idx < len(responses):
return responses[idx]
return MagicMock(returncode=0, stdout="", stderr="")
executor._run_git = fake_git
def test_already_on_branch(self, executor):
self._mock_git(executor, [
MagicMock(returncode=0, stdout="feat-mvp\n", stderr=""),
])
executor._checkout_branch() # should return without checkout
def test_branch_exists_checkout(self, executor):
self._mock_git(executor, [
MagicMock(returncode=0, stdout="main\n", stderr=""),
MagicMock(returncode=0, stdout="", stderr=""),
MagicMock(returncode=0, stdout="", stderr=""),
])
executor._checkout_branch()
def test_branch_not_exists_create(self, executor):
self._mock_git(executor, [
MagicMock(returncode=0, stdout="main\n", stderr=""),
MagicMock(returncode=1, stdout="", stderr="not found"),
MagicMock(returncode=0, stdout="", stderr=""),
])
executor._checkout_branch()
def test_checkout_fails_exits(self, executor):
self._mock_git(executor, [
MagicMock(returncode=0, stdout="main\n", stderr=""),
MagicMock(returncode=1, stdout="", stderr=""),
MagicMock(returncode=1, stdout="", stderr="dirty tree"),
])
with pytest.raises(SystemExit) as exc_info:
executor._checkout_branch()
assert exc_info.value.code == 1
def test_no_git_exits(self, executor):
self._mock_git(executor, [
MagicMock(returncode=1, stdout="", stderr="not a git repo"),
])
with pytest.raises(SystemExit) as exc_info:
executor._checkout_branch()
assert exc_info.value.code == 1
# ---------------------------------------------------------------------------
# _commit_step (mocked)
# ---------------------------------------------------------------------------
class TestCommitStep:
def test_two_phase_commit(self, executor):
calls = []
def fake_git(*args):
calls.append(args)
if args[:2] == ("diff", "--cached"):
return MagicMock(returncode=1)
return MagicMock(returncode=0, stdout="", stderr="")
executor._run_git = fake_git
executor._commit_step(2, "ui")
commit_calls = [c for c in calls if c[0] == "commit"]
assert len(commit_calls) == 2
assert "feat(mvp):" in commit_calls[0][2]
assert "chore(mvp):" in commit_calls[1][2]
def test_no_code_changes_skips_feat_commit(self, executor):
call_count = {"diff": 0}
calls = []
def fake_git(*args):
calls.append(args)
if args[:2] == ("diff", "--cached"):
call_count["diff"] += 1
if call_count["diff"] == 1:
return MagicMock(returncode=0)
return MagicMock(returncode=1)
return MagicMock(returncode=0, stdout="", stderr="")
executor._run_git = fake_git
executor._commit_step(2, "ui")
commit_msgs = [c[2] for c in calls if c[0] == "commit"]
assert len(commit_msgs) == 1
assert "chore" in commit_msgs[0]
# ---------------------------------------------------------------------------
# _invoke_codex (mocked)
# ---------------------------------------------------------------------------
class TestInvokeCodex:
def test_invokes_codex_with_correct_args(self, executor):
mock_result = MagicMock(returncode=0, stdout='{"result": "ok"}', stderr="")
step = {"step": 2, "name": "ui"}
preamble = "PREAMBLE\n"
with patch("subprocess.run", return_value=mock_result) as mock_run:
output = executor._invoke_codex(step, preamble)
cmd = mock_run.call_args[0][0]
kwargs = mock_run.call_args[1]
assert cmd[0] == "codex"
assert cmd[1] == "exec"
assert "--full-auto" in cmd
assert "--json" in cmd
assert "-o" in cmd
assert "PREAMBLE" in kwargs["input"]
assert "UI를 구현하세요" in kwargs["input"]
assert output["finalMessage"] is None
def test_saves_output_json(self, executor):
def fake_run(*args, **kwargs):
cmd = args[0]
last_message_path = Path(cmd[cmd.index("-o") + 1])
last_message_path.write_text("completed", encoding="utf-8")
return MagicMock(returncode=0, stdout='{"ok": true}', stderr="")
step = {"step": 2, "name": "ui"}
with patch("subprocess.run", side_effect=fake_run):
executor._invoke_codex(step, "preamble")
output_file = executor._phase_dir / "step2-output.json"
assert output_file.exists()
data = json.loads(output_file.read_text())
assert data["step"] == 2
assert data["name"] == "ui"
assert data["exitCode"] == 0
assert data["finalMessage"] == "completed"
def test_nonexistent_step_file_exits(self, executor):
step = {"step": 99, "name": "nonexistent"}
with pytest.raises(SystemExit) as exc_info:
executor._invoke_codex(step, "preamble")
assert exc_info.value.code == 1
def test_timeout_is_1800(self, executor):
mock_result = MagicMock(returncode=0, stdout="{}", stderr="")
step = {"step": 2, "name": "ui"}
with patch("subprocess.run", return_value=mock_result) as mock_run:
executor._invoke_codex(step, "preamble")
assert mock_run.call_args[1]["timeout"] == 1800
# ---------------------------------------------------------------------------
# progress_indicator (= 이전 Spinner)
# ---------------------------------------------------------------------------
class TestProgressIndicator:
def test_context_manager(self):
import time
with ex.progress_indicator("test") as pi:
time.sleep(0.15)
assert pi.elapsed >= 0.1
def test_elapsed_increases(self):
import time
with ex.progress_indicator("test") as pi:
time.sleep(0.2)
assert pi.elapsed > 0
# ---------------------------------------------------------------------------
# main() CLI 파싱 (mocked)
# ---------------------------------------------------------------------------
class TestMainCli:
def test_no_args_exits(self):
with patch("sys.argv", ["execute.py"]):
with pytest.raises(SystemExit) as exc_info:
ex.main()
assert exc_info.value.code == 2 # argparse exits with 2
def test_invalid_phase_dir_exits(self):
with patch("sys.argv", ["execute.py", "nonexistent"]):
with patch.object(ex, "ROOT", Path("/tmp/fake_nonexistent")):
with pytest.raises(SystemExit) as exc_info:
ex.main()
assert exc_info.value.code == 1
def test_missing_index_exits(self, tmp_project):
(tmp_project / "phases" / "empty").mkdir()
with patch("sys.argv", ["execute.py", "empty"]):
with patch.object(ex, "ROOT", tmp_project):
with pytest.raises(SystemExit) as exc_info:
ex.main()
assert exc_info.value.code == 1
# ---------------------------------------------------------------------------
# _check_blockers (= 이전 main() error/blocked 체크)
# ---------------------------------------------------------------------------
class TestCheckBlockers:
def _make_executor_with_steps(self, tmp_project, steps):
d = tmp_project / "phases" / "test-phase"
d.mkdir(exist_ok=True)
index = {"project": "T", "phase": "test", "steps": steps}
(d / "index.json").write_text(json.dumps(index))
with patch.object(ex, "ROOT", tmp_project):
inst = ex.StepExecutor.__new__(ex.StepExecutor)
inst._root = str(tmp_project)
inst._phases_dir = tmp_project / "phases"
inst._phase_dir = d
inst._phase_dir_name = "test-phase"
inst._index_file = d / "index.json"
inst._top_index_file = tmp_project / "phases" / "index.json"
inst._phase_name = "test"
inst._total = len(steps)
return inst
def test_error_step_exits_1(self, tmp_project):
steps = [
{"step": 0, "name": "ok", "status": "completed"},
{"step": 1, "name": "bad", "status": "error", "error_message": "fail"},
]
inst = self._make_executor_with_steps(tmp_project, steps)
with pytest.raises(SystemExit) as exc_info:
inst._check_blockers()
assert exc_info.value.code == 1
def test_blocked_step_exits_2(self, tmp_project):
steps = [
{"step": 0, "name": "ok", "status": "completed"},
{"step": 1, "name": "stuck", "status": "blocked", "blocked_reason": "API key"},
]
inst = self._make_executor_with_steps(tmp_project, steps)
with pytest.raises(SystemExit) as exc_info:
inst._check_blockers()
assert exc_info.value.code == 2
Coding/Codex/scripts/validate_workspace.py
-91
View File
@@ -1,91 +0,0 @@
#!/usr/bin/env python3
"""Run repository validation commands for the Harness template."""
from __future__ import annotations
import json
import os
import subprocess
import sys
from pathlib import Path
DEFAULT_NPM_ORDER = ("lint", "build", "test")
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 load_npm_commands(root: Path) -> list[str]:
package_json = root / "package.json"
if not package_json.exists():
return []
try:
payload = json.loads(package_json.read_text(encoding="utf-8"))
except json.JSONDecodeError:
return []
scripts = payload.get("scripts", {})
if not isinstance(scripts, dict):
return []
commands = []
for name in DEFAULT_NPM_ORDER:
value = scripts.get(name)
if isinstance(value, str) and value.strip():
commands.append(f"npm run {name}")
return commands
def discover_commands(root: Path) -> list[str]:
env_commands = load_env_commands()
if env_commands:
return env_commands
return load_npm_commands(root)
def run_command(command: str, root: Path) -> subprocess.CompletedProcess:
return subprocess.run(
command,
cwd=root,
shell=True,
capture_output=True,
text=True,
)
def emit_stream(prefix: str, content: str, *, stream) -> None:
text = content.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 validation commands configured.")
print("Set HARNESS_VALIDATION_COMMANDS or add npm scripts for lint/build/test.")
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())