modify skill and plugin

This commit is contained in:
NINI
2026-05-13 22:27:48 +09:00
parent d414338780
commit 2d59191df2
24 changed files with 2051 additions and 1 deletions
@@ -0,0 +1,42 @@
---
name: harness-review
description: Use when reviewing this Harness repository: local changes, generated phase files, step outputs, implementation diffs, missing tests, build readiness, or compliance with AGENTS.md, docs/ARCHITECTURE.md, docs/ADR.md, and Harness acceptance criteria.
---
# Harness Review
## Overview
Use this skill to review Harness work against the repository's persistent rules, architecture docs, and executable verification requirements. Prioritize bugs, regressions, missing tests, and rule violations.
## Review Process
1. Read `/AGENTS.md`, `/docs/ARCHITECTURE.md`, and `/docs/ADR.md`.
2. Inspect the changed files with `git status --short` and `git diff`.
3. Check architecture, stack choices, tests, critical rules, and build readiness.
4. Run relevant verification commands when feasible. If a command cannot be run, report that as residual risk.
5. Lead with actionable findings. Keep summaries secondary.
## Checklist
| Item | Question |
| --- | --- |
| Architecture | Does the change follow `docs/ARCHITECTURE.md` directory and module boundaries? |
| Stack | Does the change stay within choices documented in `docs/ADR.md`? |
| Tests | Are new or changed behaviors covered by tests? |
| Critical Rules | Does the change violate any `AGENTS.md` CRITICAL rule? |
| Build | Do relevant build/test/lint commands pass? |
## Output Format
If there are findings, list them first in severity order with file and line references when possible. Then include this table:
| 항목 | 결과 | 비고 |
| --- | --- | --- |
| 아키텍처 준수 | PASS/FAIL | {상세} |
| 기술 스택 준수 | PASS/FAIL | {상세} |
| 테스트 존재 | PASS/FAIL | {상세} |
| CRITICAL 규칙 | PASS/FAIL | {상세} |
| 빌드 가능 | PASS/FAIL | {상세} |
When there are no findings, say that clearly, then mention any commands not run or remaining risk.
@@ -0,0 +1,4 @@
interface:
display_name: "Harness Review"
short_description: "Review Harness changes safely"
default_prompt: "Use $harness-review to review Harness repository changes."
@@ -0,0 +1,124 @@
---
name: harness-workflow
description: Use when planning or running this Harness framework: reading AGENTS.md and docs/*.md, discussing implementation scope, creating or updating phases/index.json, phases/{task}/index.json, phases/{task}/stepN.md, or invoking scripts/execute.py for staged Codex execution.
---
# Harness Workflow
## Overview
Use this skill to turn a user-approved task into small, self-contained Harness steps that another Codex session can execute reliably. Keep the workflow grounded in repository docs and executable acceptance criteria.
## Workflow
1. Read `AGENTS.md` and relevant files under `docs/`, especially `docs/PRD.md`, `docs/ARCHITECTURE.md`, and `docs/ADR.md`.
2. Discuss unresolved product or technical decisions with the user before writing phase files.
3. When the user asks for an implementation plan, draft steps and get approval before creating files.
4. Create or update `phases/index.json`, `phases/{task-name}/index.json`, and one `phases/{task-name}/stepN.md` per step.
5. Run the phase with `python scripts/execute.py {task-name}` when asked to execute it. Use `--push` only when the user asks to push.
## Step Design Rules
- Scope each step to one layer or module. Split steps when multiple modules would otherwise change together.
- Make every step self-contained. Do not rely on prior conversation; include all required context and file paths.
- Force context gathering. Each step must tell Codex which docs and previous outputs to read before editing.
- Specify interfaces and signatures, not full implementations, unless exact code is required for a constraint.
- Put core invariants directly in the step: idempotency, security, data integrity, API contracts, or other non-negotiables.
- Use executable acceptance criteria such as `npm run build && npm test`, not abstract statements.
- Write cautions concretely: "Do not do X. Reason: Y."
- Name steps with kebab-case slugs such as `project-setup`, `api-layer`, or `auth-flow`.
## Phase Files
Create or update `phases/index.json`:
```json
{
"phases": [
{
"dir": "0-mvp",
"status": "pending"
}
]
}
```
Create `phases/{task-name}/index.json`:
```json
{
"project": "<project-name>",
"phase": "<task-name>",
"steps": [
{ "step": 0, "name": "project-setup", "status": "pending" },
{ "step": 1, "name": "core-types", "status": "pending" },
{ "step": 2, "name": "api-layer", "status": "pending" }
]
}
```
Rules:
- `project` comes from `AGENTS.md`.
- `phase` matches the task directory name.
- `steps[].step` starts at `0`.
- Initial status is always `pending`.
- Do not add timestamps when creating files. `scripts/execute.py` records `created_at`, `started_at`, `completed_at`, `failed_at`, and `blocked_at`.
## Step Template
```markdown
# Step {N}: {name}
## 읽어야 할 파일
먼저 아래 파일들을 읽고 프로젝트의 아키텍처와 설계 의도를 파악하라:
- `/AGENTS.md`
- `/docs/ARCHITECTURE.md`
- `/docs/ADR.md`
- {previously created or modified files}
이전 step에서 만들어진 코드를 꼼꼼히 읽고, 설계 의도를 이해한 뒤 작업하라.
## 작업
{Concrete instructions with file paths, interfaces, signatures, and rules.}
## Acceptance Criteria
```bash
npm run build
npm test
```
## 검증 절차
1. 위 AC 커맨드를 실행한다.
2. 아키텍처 체크리스트를 확인한다:
- ARCHITECTURE.md 디렉토리 구조를 따르는가?
- ADR 기술 스택을 벗어나지 않았는가?
- AGENTS.md CRITICAL 규칙을 위반하지 않았는가?
3. 결과에 따라 `phases/{task-name}/index.json`의 해당 step을 업데이트한다:
- 성공: `"status": "completed"`, `"summary": "산출물 한 줄 요약"`
- 3회 수정 시도 후 실패: `"status": "error"`, `"error_message": "구체적 에러 내용"`
- 사용자 개입 필요: `"status": "blocked"`, `"blocked_reason": "구체적 사유"` 후 중단
## 금지사항
- {Do not do X. Reason: Y.}
- 기존 테스트를 깨뜨리지 마라.
```
## Execution And Recovery
Run:
```bash
python scripts/execute.py {task-name}
python scripts/execute.py {task-name} --push
```
`scripts/execute.py` creates or checks out `feat-{task-name}`, injects `AGENTS.md` and `docs/*.md` into each prompt, carries completed step summaries forward, retries failed steps up to three times, separates code and metadata commits, and records timestamps.
If a step is `error`, set it back to `pending` and remove `error_message` after fixing the cause. If a step is `blocked`, resolve `blocked_reason`, set it back to `pending`, remove `blocked_reason`, and rerun.
@@ -0,0 +1,4 @@
interface:
display_name: "Harness Workflow"
short_description: "Plan staged Harness workflow steps"
default_prompt: "Use $harness-workflow to plan Harness phases and step files."
+4
View File
@@ -0,0 +1,4 @@
#:schema https://developers.openai.com/codex/config-schema.json
[features]
hooks = true
+28
View File
@@ -0,0 +1,28 @@
{
"hooks": {
"PreToolUse": [
{
"matcher": "^Bash$",
"hooks": [
{
"type": "command",
"command": "python -c \"import pathlib, runpy, subprocess; root = pathlib.Path(subprocess.check_output(['git', 'rev-parse', '--show-toplevel'], text=True).strip()); runpy.run_path(str(root / '.codex' / 'hooks' / 'pre_commit_checks.py'), run_name='__main__')\"",
"timeout": 600,
"statusMessage": "Running pre-commit checks"
}
]
},
{
"matcher": "^(apply_patch|Edit|Write)$",
"hooks": [
{
"type": "command",
"command": "python -c \"import pathlib, runpy, subprocess; root = pathlib.Path(subprocess.check_output(['git', 'rev-parse', '--show-toplevel'], text=True).strip()); runpy.run_path(str(root / '.codex' / 'hooks' / 'tdd-guard.py'), run_name='__main__')\"",
"timeout": 30,
"statusMessage": "Checking TDD guard"
}
]
}
]
}
}
@@ -0,0 +1,111 @@
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path
CHECKS = ("lint", "build", "test")
def _repo_root(cwd: Path) -> Path:
try:
root = subprocess.check_output(
["git", "rev-parse", "--show-toplevel"],
cwd=cwd,
text=True,
stderr=subprocess.DEVNULL,
).strip()
except (subprocess.CalledProcessError, FileNotFoundError):
return cwd
return Path(root)
def _load_scripts(root: Path) -> dict[str, str]:
package_json = root / "package.json"
if not package_json.exists():
return {}
try:
package = json.loads(package_json.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
_deny(f"Invalid package.json: {exc}")
raise SystemExit(0) from exc
scripts = package.get("scripts", {})
if not isinstance(scripts, dict):
return {}
return {str(name): str(command) for name, command in scripts.items()}
def _is_git_commit(command: str) -> bool:
return re.search(r"\bgit(?:\s+(?:-[A-Za-z]\s+\S+|--[A-Za-z0-9-]+(?:=\S+)?))*\s+commit\b", command) is not None
def _deny(reason: str) -> None:
print(
json.dumps(
{
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": reason,
}
}
)
)
def _tail(text: str, limit: int = 1200) -> str:
text = text.strip()
if len(text) <= limit:
return text
return text[-limit:]
def _run_checks(root: Path, scripts: dict[str, str]) -> str | None:
npm = shutil.which("npm") or shutil.which("npm.cmd")
if npm is None:
return "npm was not found, so pre-commit checks could not run."
for check in CHECKS:
if check not in scripts:
continue
result = subprocess.run(
[npm, "run", check],
cwd=root,
capture_output=True,
text=True,
)
if result.returncode != 0:
details = _tail(result.stdout + "\n" + result.stderr)
if details:
return f"npm run {check} failed:\n{details}"
return f"npm run {check} failed with exit code {result.returncode}."
return None
def main() -> int:
try:
payload = json.load(sys.stdin)
except json.JSONDecodeError:
return 0
command = payload.get("tool_input", {}).get("command", "")
if not isinstance(command, str) or not _is_git_commit(command):
return 0
cwd = Path(payload.get("cwd") or Path.cwd())
root = _repo_root(cwd)
failure = _run_checks(root, _load_scripts(root))
if failure:
_deny(f"PRE-COMMIT CHECKS: {failure}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+189
View File
@@ -0,0 +1,189 @@
import json
import subprocess
import sys
from pathlib import Path
SOURCE_SUFFIXES = {".ts", ".tsx", ".js", ".jsx"}
TEST_SUFFIXES = ("ts", "tsx", "js", "jsx")
CONFIG_SUFFIXES = {".json", ".css", ".scss", ".md", ".yml", ".yaml"}
NEXT_SPECIAL_FILES = {
"layout.ts",
"layout.tsx",
"page.ts",
"page.tsx",
"loading.tsx",
"error.tsx",
"not-found.tsx",
"globals.css",
}
def _repo_root(cwd: Path) -> Path:
try:
root = subprocess.check_output(
["git", "rev-parse", "--show-toplevel"],
cwd=cwd,
text=True,
stderr=subprocess.DEVNULL,
).strip()
except (subprocess.CalledProcessError, FileNotFoundError):
return cwd
return Path(root)
def _extract_patch_paths(command: str) -> list[str]:
prefixes = (
"*** Add File: ",
"*** Update File: ",
"*** Delete File: ",
"*** Move to: ",
)
paths: list[str] = []
for raw_line in command.splitlines():
line = raw_line.strip()
for prefix in prefixes:
if line.startswith(prefix):
paths.append(line[len(prefix) :].strip())
break
return paths
def _touched_paths(payload: dict) -> list[str]:
tool_input = payload.get("tool_input", {})
if not isinstance(tool_input, dict):
return []
file_path = tool_input.get("file_path")
if isinstance(file_path, str) and file_path:
return [file_path]
command = tool_input.get("command")
if isinstance(command, str):
return _extract_patch_paths(command)
return []
def _normalize(path_text: str) -> str:
return path_text.replace("\\", "/").lower()
def _is_test_path(path_text: str) -> bool:
normalized = _normalize(path_text)
name = normalized.rsplit("/", 1)[-1]
return (
"__tests__/" in normalized
or ".test." in name
or ".spec." in name
or "test" in name
or "spec" in name
)
def _is_exempt(path_text: str) -> bool:
normalized = _normalize(path_text)
path = Path(path_text)
name = path.name.lower()
if _is_test_path(path_text):
return True
if name in NEXT_SPECIAL_FILES:
return True
if path.suffix.lower() in CONFIG_SUFFIXES:
return True
if ".env" in name or ".config." in name:
return True
if any(token in name for token in ("tailwind", "postcss", "next.config", "tsconfig")):
return True
if "/types/" in normalized or name in {"types.ts", "types.d.ts"}:
return True
return False
def _resolve_path(path_text: str, cwd: Path) -> Path:
path = Path(path_text)
if path.is_absolute():
return path
return (cwd / path).resolve()
def _base_name(path: Path) -> str:
for suffix in (".tsx", ".ts", ".jsx", ".js"):
if path.name.endswith(suffix):
return path.name[: -len(suffix)]
return path.stem
def _has_existing_test(path: Path, root: Path) -> bool:
directory = path.parent
parent = directory.parent
base = _base_name(path)
for ext in TEST_SUFFIXES:
if (directory / f"{base}.test.{ext}").exists():
return True
if (directory / f"{base}.spec.{ext}").exists():
return True
for ext in TEST_SUFFIXES:
if (parent / "__tests__" / f"{base}.test.{ext}").exists():
return True
if (directory / "__tests__" / f"{base}.test.{ext}").exists():
return True
for ext in TEST_SUFFIXES:
if (root / "src" / "__tests__" / f"{base}.test.{ext}").exists():
return True
return False
def _guarded_paths(paths: list[str], cwd: Path, root: Path) -> list[str]:
missing_tests: list[str] = []
for path_text in paths:
if _is_exempt(path_text):
continue
path = _resolve_path(path_text, cwd)
if path.suffix.lower() not in SOURCE_SUFFIXES:
continue
if not _has_existing_test(path, root):
missing_tests.append(_base_name(path))
return missing_tests
def main() -> int:
try:
payload = json.load(sys.stdin)
except json.JSONDecodeError:
return 0
cwd = Path(payload.get("cwd") or Path.cwd())
root = _repo_root(cwd)
missing_tests = _guarded_paths(_touched_paths(payload), cwd, root)
if not missing_tests:
return 0
names = ", ".join(sorted(set(missing_tests)))
print(
json.dumps(
{
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": (
"TDD GUARD: missing test file for "
f"{names}. Write or add the test first."
),
}
}
)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+9
View File
@@ -0,0 +1,9 @@
node_modules/
.next/
out/
next-env.d.ts
tsconfig.tsbuildinfo
# phase execution outputs
phases/**/phase*-output.json
phases/**/step*-output.json
+21
View File
@@ -0,0 +1,21 @@
# 프로젝트: {프로젝트명}
## 기술 스택
- {프레임워크 (예: Next.js 15)}
- {언어 (예: TypeScript strict mode)}
- {스타일링 (예: Tailwind CSS)}
## 아키텍처 규칙
- CRITICAL: {절대 지켜야 할 규칙 1 (예: 모든 API 로직은 app/api/ 라우트 핸들러에서만 처리)}
- CRITICAL: {절대 지켜야 할 규칙 2 (예: 클라이언트 컴포넌트에서 직접 외부 API를 호출하지 말 것)}
- {일반 규칙 (예: 컴포넌트는 components/ 폴더에, 타입은 types/ 폴더에 분리)}
## 개발 프로세스
- CRITICAL: 새 기능 구현 시 반드시 테스트를 먼저 작성하고, 테스트가 통과하는 구현을 작성할 것 (TDD)
- 커밋 메시지는 conventional commits 형식을 따를 것 (feat:, fix:, docs:, refactor:)
## 명령어
npm run dev # 개발 서버
npm run build # 프로덕션 빌드
npm run lint # ESLint
npm run test # 테스트
+21
View File
@@ -0,0 +1,21 @@
# Architecture Decision Records
## 철학
{프로젝트의 핵심 가치관 (예: MVP 속도 최우선. 외부 의존성 최소화. 작동하는 최소 구현을 선택.)}
---
### ADR-001: {결정 사항 (예: Next.js App Router 선택)}
**결정**: {뭘 선택했는지}
**이유**: {왜 선택했는지}
**트레이드오프**: {뭘 포기했는지}
### ADR-002: {결정 사항}
**결정**: {뭘 선택했는지}
**이유**: {왜 선택했는지}
**트레이드오프**: {뭘 포기했는지}
### ADR-003: {결정 사항}
**결정**: {뭘 선택했는지}
**이유**: {왜 선택했는지}
**트레이드오프**: {뭘 포기했는지}
+24
View File
@@ -0,0 +1,24 @@
# 아키텍처
## 디렉토리 구조
```
src/
├── app/ # 페이지 + API 라우트
├── components/ # UI 컴포넌트
├── types/ # TypeScript 타입 정의
├── lib/ # 유틸리티 + 헬퍼
└── services/ # 외부 API 래퍼
```
## 패턴
{사용하는 디자인 패턴 (예: Server Components 기본, 인터랙션이 필요한 곳만 Client Component)}
## 데이터 흐름
```
{데이터가 어떻게 흐르는지 (예:
사용자 입력 → Client Component → API Route → 외부 API → 응답 → UI 업데이트
)}
```
## 상태 관리
{상태 관리 방식 (예: 서버 상태는 Server Components, 클라이언트 상태는 useState/useReducer)}
+21
View File
@@ -0,0 +1,21 @@
# PRD: {프로젝트명}
## 목표
{이 프로젝트가 해결하려는 문제를 한 줄로 요약}
## 사용자
{누가 이 제품을 쓰는지}
## 핵심 기능
1. {기능 1}
2. {기능 2}
3. {기능 3}
## MVP 제외 사항
- {안 만들 것 1}
- {안 만들 것 2}
- {안 만들 것 3}
## 디자인
- {디자인 방향 (예: 다크모드 고정, 미니멀)}
- {색상 (예: 무채색 + 포인트 1가지)}
+76
View File
@@ -0,0 +1,76 @@
# UI 디자인 가이드
## 디자인 원칙
1. {원칙 1 — 예: "도구처럼 보여야 한다. 마케팅 페이지가 아니라 매일 쓰는 대시보드."}
2. {원칙 2}
3. {원칙 3}
## AI 슬롭 안티패턴 — 하지 마라
| 금지 사항 | 이유 |
|-----------|------|
| backdrop-filter: blur() | glass morphism은 AI 템플릿의 가장 흔한 징후 |
| gradient-text (배경 그라데이션 텍스트) | AI가 만든 SaaS 랜딩의 1번 특징 |
| "Powered by AI" 배지 | 기능이 아니라 장식. 사용자에게 가치 없음 |
| box-shadow 글로우 애니메이션 | 네온 글로우 = AI 슬롭 |
| 보라/인디고 브랜드 색상 | "AI = 보라색" 클리셰 |
| 모든 카드에 동일한 rounded-2xl | 균일한 둥근 모서리는 템플릿 느낌 |
| 배경 gradient orb (blur-3xl 원형) | 모든 AI 랜딩 페이지에 있는 장식 |
## 색상
### 배경
| 용도 | 값 |
|------|------|
| 페이지 | {예: #0a0a0a} |
| 카드 | {예: #141414} |
### 텍스트
| 용도 | 값 |
|------|------|
| 주 텍스트 | {예: text-white} |
| 본문 | {예: text-neutral-300} |
| 보조 | {예: text-neutral-400} |
| 비활성 | {예: text-neutral-500} |
### 데이터/시맨틱 색상
| 용도 | 값 |
|------|------|
| {긍정/성공} | {예: #22c55e} |
| {부정/에러} | {예: #ef4444} |
| {중립/기본} | {예: #525252} |
## 컴포넌트
### 카드
```
{예: rounded-lg bg-[#141414] border border-neutral-800 p-6}
```
### 버튼
```
Primary: {예: rounded-lg bg-white text-black hover:bg-neutral-200}
Text: {예: text-neutral-500 hover:text-neutral-300}
```
### 입력 필드
```
{예: rounded-lg bg-neutral-900 border border-neutral-800 px-4 py-3}
```
## 레이아웃
- 전체 너비: {예: max-w-5xl}
- 정렬: {예: 좌측 정렬 기본. 중앙 정렬 금지}
- 간격: {예: gap-3~4, 섹션 간 space-y-8}
## 타이포그래피
| 용도 | 스타일 |
|------|--------|
| 페이지 제목 | {예: text-4xl font-semibold text-white} |
| 카드 제목 | {예: text-sm font-medium text-neutral-400} |
| 본문 | {예: text-sm text-neutral-300 leading-relaxed} |
## 애니메이션
- {허용할 애니메이션만 나열. 예: fade-in (0.4s), slide-up (0.5s)}
- {그 외 모든 애니메이션 금지}
## 아이콘
- {예: SVG 인라인, strokeWidth 1.5}
- {예: 아이콘 컨테이너(둥근 배경 박스)로 감싸지 않는다}
+417
View File
@@ -0,0 +1,417 @@
#!/usr/bin/env python3
"""
Harness Step Executor — phase 내 step을 순차 실행하고 자가 교정한다.
Usage:
python scripts/execute.py <phase-dir> [--push]
"""
import argparse
import contextlib
import json
import os
import subprocess
import sys
import threading
import time
import types
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Optional
ROOT = Path(__file__).resolve().parent.parent
@contextlib.contextmanager
def progress_indicator(label: str):
"""터미널 진행 표시기. with 문으로 사용하며 .elapsed 로 경과 시간을 읽는다."""
frames = "◐◓◑◒"
stop = threading.Event()
t0 = time.monotonic()
def _animate():
idx = 0
while not stop.wait(0.12):
sec = int(time.monotonic() - t0)
sys.stderr.write(f"\r{frames[idx % len(frames)]} {label} [{sec}s]")
sys.stderr.flush()
idx += 1
sys.stderr.write("\r" + " " * (len(label) + 20) + "\r")
sys.stderr.flush()
th = threading.Thread(target=_animate, daemon=True)
th.start()
info = types.SimpleNamespace(elapsed=0.0)
try:
yield info
finally:
stop.set()
th.join()
info.elapsed = time.monotonic() - t0
class StepExecutor:
"""Phase 디렉토리 안의 step들을 순차 실행하는 하네스."""
MAX_RETRIES = 3
FEAT_MSG = "feat({phase}): step {num}{name}"
CHORE_MSG = "chore({phase}): step {num} output"
TZ = timezone(timedelta(hours=9))
def __init__(self, phase_dir_name: str, *, auto_push: bool = False):
self._root = str(ROOT)
self._phases_dir = ROOT / "phases"
self._phase_dir = self._phases_dir / phase_dir_name
self._phase_dir_name = phase_dir_name
self._top_index_file = self._phases_dir / "index.json"
self._auto_push = auto_push
if not self._phase_dir.is_dir():
print(f"ERROR: {self._phase_dir} not found")
sys.exit(1)
self._index_file = self._phase_dir / "index.json"
if not self._index_file.exists():
print(f"ERROR: {self._index_file} not found")
sys.exit(1)
idx = self._read_json(self._index_file)
self._project = idx.get("project", "project")
self._phase_name = idx.get("phase", phase_dir_name)
self._total = len(idx["steps"])
def run(self):
self._print_header()
self._check_blockers()
self._checkout_branch()
guardrails = self._load_guardrails()
self._ensure_created_at()
self._execute_all_steps(guardrails)
self._finalize()
# --- timestamps ---
def _stamp(self) -> str:
return datetime.now(self.TZ).strftime("%Y-%m-%dT%H:%M:%S%z")
# --- JSON I/O ---
@staticmethod
def _read_json(p: Path) -> dict:
return json.loads(p.read_text(encoding="utf-8"))
@staticmethod
def _write_json(p: Path, data: dict):
p.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
# --- git ---
def _run_git(self, *args) -> subprocess.CompletedProcess:
cmd = ["git"] + list(args)
return subprocess.run(cmd, cwd=self._root, capture_output=True, text=True)
def _checkout_branch(self):
branch = f"feat-{self._phase_name}"
r = self._run_git("rev-parse", "--abbrev-ref", "HEAD")
if r.returncode != 0:
print(f" ERROR: git을 사용할 수 없거나 git repo가 아닙니다.")
print(f" {r.stderr.strip()}")
sys.exit(1)
if r.stdout.strip() == branch:
return
r = self._run_git("rev-parse", "--verify", branch)
r = self._run_git("checkout", branch) if r.returncode == 0 else self._run_git("checkout", "-b", branch)
if r.returncode != 0:
print(f" ERROR: 브랜치 '{branch}' checkout 실패.")
print(f" {r.stderr.strip()}")
print(f" Hint: 변경사항을 stash하거나 commit한 후 다시 시도하세요.")
sys.exit(1)
print(f" Branch: {branch}")
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"
self._run_git("add", "-A")
self._run_git("reset", "HEAD", "--", output_rel)
self._run_git("reset", "HEAD", "--", index_rel)
if self._run_git("diff", "--cached", "--quiet").returncode != 0:
msg = self.FEAT_MSG.format(phase=self._phase_name, num=step_num, name=step_name)
r = self._run_git("commit", "-m", msg)
if r.returncode == 0:
print(f" Commit: {msg}")
else:
print(f" WARN: 코드 커밋 실패: {r.stderr.strip()}")
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)
r = self._run_git("commit", "-m", msg)
if r.returncode != 0:
print(f" WARN: housekeeping 커밋 실패: {r.stderr.strip()}")
# --- top-level index ---
def _update_top_index(self, status: str):
if not self._top_index_file.exists():
return
top = self._read_json(self._top_index_file)
ts = self._stamp()
for phase in top.get("phases", []):
if phase.get("dir") == self._phase_dir_name:
phase["status"] = status
ts_key = {"completed": "completed_at", "error": "failed_at", "blocked": "blocked_at"}.get(status)
if ts_key:
phase[ts_key] = ts
break
self._write_json(self._top_index_file, top)
# --- guardrails & context ---
def _load_guardrails(self) -> str:
sections = []
agents_md = ROOT / "AGENTS.md"
if agents_md.exists():
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")):
sections.append(f"## {doc.stem}\n\n{doc.read_text(encoding='utf-8')}")
return "\n\n---\n\n".join(sections) if sections else ""
@staticmethod
def _build_step_context(index: dict) -> str:
lines = [
f"- Step {s['step']} ({s['name']}): {s['summary']}"
for s in index["steps"]
if s["status"] == "completed" and s.get("summary")
]
if not lines:
return ""
return "## 이전 Step 산출물\n\n" + "\n".join(lines) + "\n\n"
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 = (
f"\n## ⚠ 이전 시도 실패 — 아래 에러를 반드시 참고하여 수정하라\n\n"
f"{prev_error}\n\n---\n\n"
)
return (
f"당신은 {self._project} 프로젝트의 개발자입니다. 아래 step을 수행하세요.\n\n"
f"{guardrails}\n\n---\n\n"
f"{step_context}{retry_section}"
f"## 작업 규칙\n\n"
f"1. 이전 step에서 작성된 코드를 확인하고 일관성을 유지하라.\n"
f"2. 이 step에 명시된 작업만 수행하라. 추가 기능이나 파일을 만들지 마라.\n"
f"3. 기존 테스트를 깨뜨리지 마라.\n"
f"4. AC(Acceptance Criteria) 검증을 직접 실행하라.\n"
f"5. /phases/{self._phase_dir_name}/index.json의 해당 step status를 업데이트하라:\n"
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"
)
# --- Codex 호출 ---
def _invoke_codex(self, step: dict, preamble: str) -> dict:
step_num, step_name = step["step"], step["name"]
step_file = self._phase_dir / f"step{step_num}.md"
if not step_file.exists():
print(f" ERROR: {step_file} not found")
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,
)
if result.returncode != 0:
print(f"\n WARN: Codex가 비정상 종료됨 (code {result.returncode})")
if result.stderr:
print(f" stderr: {result.stderr[:500]}")
output = {
"step": step_num, "name": step_name,
"exitCode": result.returncode,
"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)
return output
# --- 헤더 & 검증 ---
def _print_header(self):
print(f"\n{'='*60}")
print(f" Harness Step Executor")
print(f" Phase: {self._phase_name} | Steps: {self._total}")
if self._auto_push:
print(f" Auto-push: enabled")
print(f"{'='*60}")
def _check_blockers(self):
index = self._read_json(self._index_file)
for s in reversed(index["steps"]):
if s["status"] == "error":
print(f"\n ✗ Step {s['step']} ({s['name']}) failed.")
print(f" Error: {s.get('error_message', 'unknown')}")
print(f" Fix and reset status to 'pending' to retry.")
sys.exit(1)
if s["status"] == "blocked":
print(f"\n ⏸ Step {s['step']} ({s['name']}) blocked.")
print(f" Reason: {s.get('blocked_reason', 'unknown')}")
print(f" Resolve and reset status to 'pending' to retry.")
sys.exit(2)
if s["status"] != "pending":
break
def _ensure_created_at(self):
index = self._read_json(self._index_file)
if "created_at" not in index:
index["created_at"] = self._stamp()
self._write_json(self._index_file, index)
# --- 실행 루프 ---
def _execute_single_step(self, step: dict, guardrails: str) -> bool:
"""단일 step 실행 (재시도 포함). 완료되면 True, 실패/차단이면 False."""
step_num, step_name = step["step"], step["name"]
done = sum(1 for s in self._read_json(self._index_file)["steps"] if s["status"] == "completed")
prev_error = None
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, 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)
elapsed = int(pi.elapsed)
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()
if status == "completed":
for s in index["steps"]:
if s["step"] == step_num:
s["completed_at"] = ts
self._write_json(self._index_file, index)
self._commit_step(step_num, step_name)
print(f" ✓ Step {step_num}: {step_name} [{elapsed}s]")
return True
if status == "blocked":
for s in index["steps"]:
if s["step"] == step_num:
s["blocked_at"] = ts
self._write_json(self._index_file, index)
reason = next((s.get("blocked_reason", "") for s in index["steps"] if s["step"] == step_num), "")
print(f" ⏸ Step {step_num}: {step_name} blocked [{elapsed}s]")
print(f" Reason: {reason}")
self._update_top_index("blocked")
sys.exit(2)
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",
)
if attempt < self.MAX_RETRIES:
for s in index["steps"]:
if s["step"] == step_num:
s["status"] = "pending"
s.pop("error_message", None)
self._write_json(self._index_file, index)
prev_error = err_msg
print(f" ↻ Step {step_num}: retry {attempt}/{self.MAX_RETRIES}{err_msg}")
else:
for s in index["steps"]:
if s["step"] == step_num:
s["status"] = "error"
s["error_message"] = f"[{self.MAX_RETRIES}회 시도 후 실패] {err_msg}"
s["failed_at"] = ts
self._write_json(self._index_file, index)
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")
sys.exit(1)
return False # unreachable
def _execute_all_steps(self, guardrails: str):
while True:
index = self._read_json(self._index_file)
pending = next((s for s in index["steps"] if s["status"] == "pending"), None)
if pending is None:
print("\n All steps completed!")
return
step_num = pending["step"]
for s in index["steps"]:
if s["step"] == step_num and "started_at" not in s:
s["started_at"] = self._stamp()
self._write_json(self._index_file, index)
break
self._execute_single_step(pending, guardrails)
def _finalize(self):
index = self._read_json(self._index_file)
index["completed_at"] = self._stamp()
self._write_json(self._index_file, index)
self._update_top_index("completed")
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"{msg}")
if self._auto_push:
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()}")
sys.exit(1)
print(f" ✓ Pushed to origin/{branch}")
print(f"\n{'='*60}")
print(f" Phase '{self._phase_name}' completed!")
print(f"{'='*60}")
def main():
parser = argparse.ArgumentParser(description="Harness Step Executor")
parser.add_argument("phase_dir", help="Phase directory name (e.g. 0-mvp)")
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()
if __name__ == "__main__":
main()