sync: pull from agentic-framework v1.1.184

agent-ops/.version

@@ -1 +1 @@
-1.1.180
+1.1.184

agent-ops/skills/common/prepare-epic-work-items/SKILL.md

@@ -0,0 +1,121 @@
+---
+name: prepare-epic-work-items
+description: 현재 또는 지정 Milestone의 정확히 한 Epic을 작은 직접 작업과 큰 PLAN/CODE_REVIEW pair로 변환하고, fresh one-shot 자가검토, refine-plans 세분화, 최종 재검토와 push까지 수행할 때 사용한다. "현 마일스톤의 X Epic에서 작은 작업은 바로 처리하고 큰 작업은 plan으로 작성해", "X 마일스톤 Y Epic 작업 준비해" 요청에서 사용한다.
+---
+
+# Prepare Epic Work Items
+
+## 목적
+
+정확히 한 Epic을 한 사이클로 준비한다. 작은 작업은 구현·검증하고 큰 작업은 실행 가능한 PLAN/CODE_REVIEW pair로 만든 뒤 두 번의 fresh 검토와 한 번의 세분화를 거쳐 remote branch에 보존한다.
+
+## 입력
+
+- `workspace`: 준비된 feature worktree 절대 경로 (필수)
+- `target-milestone`: 활성 Milestone slug 또는 경로 (필수)
+- `target-epic`: 정확한 Epic id 또는 이름 (필수)
+- `planner-agent`: `codex`, `claude`, `gemini`, `pi` 중 하나 (생략 시 `codex`)
+- `review-agent`: 생략하면 `planner-agent`와 같다. (선택)
+- `planner-model`, `review-model`: provider별 model override. Codex 기본 사용 시 `planner-model` 생략 시 `gpt-5.6-sol`, 다른 provider는 해당 CLI 기본 모델을 사용한다. (선택)
+- `reasoning-effort`: 지원하는 provider의 reasoning/thinking override. 생략 시 `xhigh` (선택)
+- `pi-provider`: Pi provider override (선택)
+- `retry`: terminal failure의 원인을 사용자가 해소한 뒤 같은 Epic 상태를 재개할 때만 사용한다. (선택)
+
+## 범위 계약
+
+- 한 실행은 Epic 하나만 다룬다. Epic 범위 요청은 caller가 문서 순서대로 하나씩 실행한다.
+- 실행 identity는 `<milestone-slug>:<epic-id>`다.
+- 다음 Epic은 현재 Epic의 모든 Task가 workstate sync에서 완료된 `EPIC_COMPLETED` event 뒤에만 시작한다.
+- `EPIC_WORK_ITEMS_READY`는 큰 작업 plan이 준비됐다는 뜻이며 구현 완료가 아니다.
+- 같은 identity를 다시 실행하면 active pair, USER_REVIEW, runtime state를 먼저 대조하고 중복 plan을 만들지 않는다.
+
+작은 작업은 아래를 모두 만족해야 한다.
+
+- 하나의 응집된 변경이고 한 번의 bounded 실행과 명시 검증으로 완료할 수 있다.
+- 새 API, wire, schema, migration, 외부 side effect 또는 책임 경계 변경이 없다.
+- 사용자·SDD 결정이 필요하지 않고 큰 작업의 write set과 충돌하지 않는다.
+
+하나라도 거짓이거나 불명확하면 큰 작업으로 분류한다. 고정 LOC나 파일 수만으로 분류하지 않는다.
+
+## 실행 절차
+
+1. **Epic을 고정한다**
+   - target Milestone이 `[계획]` 또는 `[진행중]`, 구현 잠금 해제인지 확인한다.
+   - `### Epic: [<epic-id>] <title>`을 정확히 하나 찾고 그 아래 Task id를 고정한다.
+   - 다른 Epic의 active pair가 있거나 target이 모호하면 `FAILED`로 멈춘다.
+
+2. **foreground 사이클을 실행한다**
+   - 아래 스크립트를 한 번 실행하고 execution-layer event wait를 유지한다.
+
+```bash
+python3 agent-ops/skills/common/prepare-epic-work-items/scripts/run_epic_cycle.py \
+  --workspace "$WORKSPACE" \
+  --milestone "$MILESTONE" \
+  --epic "$EPIC"
+```
+
+   - 기본값은 `codex / gpt-5.6-sol / xhigh`다. 다른 provider를 지정하면 모델을 별도로 주지 않는 한 해당 provider의 CLI 기본 모델을 사용한다.
+   - 다른 agent, model, reasoning, Pi provider override와 `--retry`는 해당 입력이 있을 때만 전달한다.
+   - 스크립트는 각 agent를 새 one-shot session으로 실행한다. Codex, Claude, Gemini(`agy` adapter), Pi를 같은 normalized runner 계약으로 지원한다.
+   - model stdout/stderr는 git common dir의 locator log에만 저장한다. caller stdout에는 lifecycle/attention event만 출력한다.
+
+3. **상태 전이를 따른다**
+   - `MATERIALIZE`: 작은 작업을 먼저 구현·검증하고, 변경된 source를 기준으로 큰 작업에 `plan`을 적용한다.
+   - `INITIAL_REVIEW`: fresh reviewer가 전체 변경과 PLAN/CODE_REVIEW stub을 재검토하고 누락을 수정한다. 구현 전 stub에 공식 `code-review`를 실행하지 않는다.
+   - 첫 검토가 유효하면 변경을 commit/push한다.
+   - `REFINE`: target Epic Task id를 가진 모든 미착수 pair에 `refine-plans`를 한 번 적용한다. 분리 가치가 없으면 `no-change`를 허용한다.
+   - `FINAL_REVIEW`: fresh reviewer가 child scope 합집합, 중복, dependency, `milestone-task`, routing, 검증을 다시 확인하고 수정한다.
+   - 최종 validator가 통과하면 남은 변경을 commit/push하고 `EPIC_WORK_ITEMS_READY`를 낸다.
+
+4. **중단 상태를 처리한다**
+   - 사용자만 결정할 범위·설계 문제는 roadmap SDD `USER_REVIEW.md`로 남기고 유효한 stop artifact를 commit/push한 뒤 `USER_REVIEW`로 끝낸다.
+   - agent-task 구현 review gate가 아니므로 preparation agent가 `agent-task/**/USER_REVIEW.md`를 만들지 않는다.
+   - agent exit, invalid pair, plan validator, git commit/push 실패는 자동 삭제 없이 `FAILED`로 끝낸다.
+   - foreground wait가 끊겼지만 동일 PID/start-token의 one-shot이 살아 있으면 state를 `tracking`으로 유지하고 `AGENT_TRACKING`만 낸다. 재호출은 새 agent를 만들지 않는다.
+   - tracking handle이 종료되면 `AGENT_RECOVERY_REQUIRED`에서 멈춘다. locator 확인 뒤 `--retry`하면 현재 artifact를 먼저 채택·검증한다. 검증 실패 뒤의 명시적 `--retry`만 새 one-shot을 허용한다.
+   - `EPIC_WORK_ITEMS_READY` 뒤 dispatcher와 workstate sync가 Task와 active pair를 모두 닫으면 같은 identity 재호출이 clean HEAD를 재검증해 `EPIC_COMPLETED`로 승격한다.
+
+## 상태 이벤트
+
+- `EPIC_SCOPE_RESOLVED`
+- `MATERIALIZE_STARTED`, `MATERIALIZE_FINISHED`
+- `INITIAL_REVIEW_STARTED`, `INITIAL_REVIEW_FINISHED`
+- `INITIAL_CHECKPOINT_PUSHED`
+- `REFINE_STARTED`, `REFINE_FINISHED`
+- `FINAL_REVIEW_STARTED`, `FINAL_REVIEW_FINISHED`
+- `FINAL_ARTIFACTS_PUSHED`
+- `AGENT_TRACKING`, `AGENT_RECOVERY_REQUIRED`, `AGENT_RESULT_RECOVERED`
+- `EPIC_WORK_ITEMS_READY`, `EPIC_COMPLETED`, `USER_REVIEW`, `FAILED`
+
+routine event는 caller 판단을 요구하지 않는다. caller는 `USER_REVIEW`, `AGENT_RECOVERY_REQUIRED`, 복구 불가능한 `FAILED`, terminal completion에서만 깨어난다.
+
+## 실행 결과 검증
+
+- [ ] active PLAN/CODE_REVIEW가 항상 pair이고 첫 줄 metadata가 일치하는가
+- [ ] 모든 `milestone-task`가 target Epic Task id의 비어 있지 않은 부분집합인가
+- [ ] 모든 PLAN이 dispatcher `--validate-plan`을 통과하는가
+- [ ] refine 전후 Task id 합집합과 scope가 보존됐는가
+- [ ] repository에 unresolved template token이나 preparation runtime state가 추적되지 않는가
+- [ ] 완료 checkpoint가 현재 feature branch remote에 push됐는가
+- 검증 실패 시: partial artifact를 commit하지 않고 locator와 복구 조건을 남겨 `FAILED`로 끝낸다.
+
+## 출력 형식
+
+```text
+Epic work preparation
+- identity: <milestone-slug>:<epic-id>
+- direct work: <completed task ids 또는 없음>
+- plans: <active pair paths 또는 없음>
+- refinement: <split | no-change>
+- event: <EPIC_WORK_ITEMS_READY | EPIC_COMPLETED | USER_REVIEW | FAILED>
+- remote: <branch와 pushed commit>
+```
+
+## 금지 사항
+
+- 여러 Epic을 한 agent context에서 처리하지 않는다.
+- 같은 session을 self-review에 resume하지 않는다.
+- plan/refine agent가 nested agent나 task dispatcher를 실행하지 않는다.
+- 구현 전 CODE_REVIEW stub에 공식 code-review verdict를 쓰지 않는다.
+- timer polling, LLM keepalive, routine model stream 중계를 하지 않는다.
+- validation 실패 상태를 commit/push하거나 force push하지 않는다.

agent-ops/skills/common/prepare-epic-work-items/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "Prepare Epic Work Items"
+  short_description: "Turn one Epic into reviewed work and plans"
+  default_prompt: "Use $prepare-epic-work-items to complete small work and prepare reviewed plans for one Epic."

agent-ops/skills/common/prepare-epic-work-items/scripts/run_agent_once.py

@@ -0,0 +1,341 @@
+#!/usr/bin/env python3
+"""Run one fresh Codex, Claude, Gemini/agy, or Pi agent without polling."""
+
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+from typing import Any, Iterable
+import uuid
+
+
+AGENT_COMMAND = {"codex": "codex", "claude": "claude", "gemini": "agy", "pi": "pi"}
+DEFAULT_AGENT = "codex"
+DEFAULT_MODEL = "gpt-5.6-sol"
+DEFAULT_REASONING_EFFORT = "xhigh"
+LABEL_PATTERN = re.compile(r"^[A-Za-z0-9._-]+$")
+PROBE_EXPECTED = "MILESTONE_AGENT_READY"
+
+
+class AgentRunError(RuntimeError):
+    """One-shot runner contract error."""
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def emit(event: str, **payload: Any) -> None:
+    print(json.dumps({"event": event, **payload}, ensure_ascii=False, sort_keys=True), flush=True)
+
+
+def atomic_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + f".tmp.{os.getpid()}")
+    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    os.replace(temporary, path)
+
+
+def process_start_token(pid: int) -> str | None:
+    """Return a best-effort token that distinguishes PID reuse."""
+    stat = Path(f"/proc/{pid}/stat")
+    try:
+        remainder = stat.read_text(encoding="utf-8").rsplit(")", 1)[1].split()
+        return f"proc:{remainder[19]}"
+    except (OSError, IndexError):
+        return None
+
+
+def workspace_root(raw: str) -> Path:
+    workspace = Path(raw).expanduser().resolve()
+    if not workspace.is_dir():
+        raise AgentRunError(f"workspace directory not found: {workspace}")
+    result = subprocess.run(
+        ["git", "rev-parse", "--show-toplevel"],
+        cwd=workspace,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+    if result.returncode != 0 or Path(result.stdout.strip()).resolve() != workspace:
+        raise AgentRunError(f"workspace must be a git repository root: {workspace}")
+    return workspace
+
+
+def state_root(workspace: Path) -> Path:
+    result = subprocess.run(
+        ["git", "rev-parse", "--git-common-dir"],
+        cwd=workspace,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+    if result.returncode == 0:
+        raw = Path(result.stdout.strip())
+        common = (workspace / raw).resolve() if not raw.is_absolute() else raw.resolve()
+        if os.access(common, os.W_OK):
+            return common / "epic-work-preparation"
+    fallback = Path(os.environ.get("XDG_STATE_HOME", str(Path.home() / ".local" / "state")))
+    identity = hashlib.sha256(str(workspace).encode()).hexdigest()[:16]
+    return fallback / "epic-work-preparation" / identity
+
+
+def result_file(workspace: Path, raw: str | None) -> Path | None:
+    if raw is None:
+        return None
+    path = Path(raw).expanduser().resolve()
+    root = state_root(workspace).resolve()
+    try:
+        path.relative_to(root)
+    except ValueError as exc:
+        raise AgentRunError(f"--result-file must be inside runner state root: {root}") from exc
+    return path
+
+
+def persist(locator: Path, result: Path | None, record: dict[str, Any]) -> None:
+    atomic_json(locator, record)
+    if result is not None:
+        atomic_json(result, record)
+
+
+def prompt_text(args: argparse.Namespace) -> str:
+    if args.probe:
+        return (
+            "Reply only with the result of joining MILESTONE, _AGENT, and _READY. "
+            "Do not inspect files, call tools, or modify the workspace."
+        )
+    if args.prompt is not None:
+        return args.prompt
+    if args.prompt_file is None:
+        raise AgentRunError("--prompt or --prompt-file is required")
+    path = Path(args.prompt_file).expanduser().resolve()
+    if not path.is_file():
+        raise AgentRunError(f"prompt file not found: {path}")
+    return path.read_text(encoding="utf-8")
+
+
+def build_command(
+    *,
+    agent: str,
+    prompt: str,
+    workspace: Path,
+    model: str | None,
+    reasoning_effort: str | None,
+    pi_provider: str | None,
+    session_id: str,
+    attempt_dir: Path,
+    probe: bool = False,
+) -> list[str]:
+    if agent == "codex":
+        command = ["codex", "exec", "--json", "-C", str(workspace)]
+        if model:
+            command.extend(["-m", model])
+        if reasoning_effort:
+            command.extend(["-c", f'model_reasoning_effort="{reasoning_effort}"'])
+        if not probe:
+            command.append("--dangerously-bypass-approvals-and-sandbox")
+        command.append(prompt)
+        return command
+    if agent == "claude":
+        command = [
+            "claude",
+            "-p",
+            "--output-format",
+            "stream-json",
+            "--verbose",
+            "--session-id",
+            session_id,
+        ]
+        if model:
+            command.extend(["--model", model])
+        if reasoning_effort:
+            command.extend(["--effort", reasoning_effort])
+        if not probe:
+            command.append("--dangerously-skip-permissions")
+        command.append(prompt)
+        return command
+    if agent == "gemini":
+        command = ["agy", "--print", prompt, "--print-timeout", "8h"]
+        if model:
+            command.extend(["--model", model])
+        if not probe:
+            command.append("--dangerously-skip-permissions")
+        command.extend(["--log-file", str(attempt_dir / "agy-cli.log")])
+        return command
+    if agent == "pi":
+        command = [
+            "pi",
+            "-p",
+            "--mode",
+            "json",
+            "--session-id",
+            session_id,
+            "--session-dir",
+            str(attempt_dir / "pi-sessions"),
+        ]
+        if not probe:
+            command.append("--approve")
+        if pi_provider:
+            command.extend(["--provider", pi_provider])
+        if model:
+            command.extend(["--model", model])
+        if reasoning_effort:
+            command.extend(["--thinking", reasoning_effort])
+        command.append(prompt)
+        return command
+    raise AgentRunError(f"unsupported agent: {agent}")
+
+
+def sanitized_command(command: list[str], prompt: str) -> list[str]:
+    return ["<prompt>" if value == prompt else value for value in command]
+
+
+def parser() -> argparse.ArgumentParser:
+    value = argparse.ArgumentParser(description=__doc__)
+    value.add_argument("--agent", choices=sorted(AGENT_COMMAND), default=DEFAULT_AGENT)
+    value.add_argument("--workspace", required=True)
+    prompt_group = value.add_mutually_exclusive_group()
+    prompt_group.add_argument("--prompt")
+    prompt_group.add_argument("--prompt-file")
+    value.add_argument("--model")
+    value.add_argument("--reasoning-effort", default=DEFAULT_REASONING_EFFORT)
+    value.add_argument("--pi-provider")
+    value.add_argument("--label", default="one-shot")
+    value.add_argument("--probe", action="store_true")
+    value.add_argument("--result-file")
+    return value
+
+
+def execute(args: argparse.Namespace) -> int:
+    workspace = workspace_root(args.workspace)
+    if args.model is None and args.agent == DEFAULT_AGENT:
+        args.model = DEFAULT_MODEL
+    if not LABEL_PATTERN.fullmatch(args.label):
+        raise AgentRunError("--label may contain only letters, digits, dot, underscore, and hyphen")
+    prompt = prompt_text(args)
+    result = result_file(workspace, args.result_file)
+    executable = AGENT_COMMAND[args.agent]
+    resolved = shutil.which(executable)
+    if resolved is None:
+        raise AgentRunError(f"agent command not found: agent={args.agent} command={executable}")
+
+    execution_id = f"{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}-{uuid.uuid4().hex[:12]}"
+    root = state_root(workspace)
+    attempt_dir = root / "runs" / f"{args.label}-{execution_id}"
+    attempt_dir.mkdir(parents=True, exist_ok=False)
+    stream = attempt_dir / "stream.log"
+    locator = attempt_dir / "locator.json"
+    session_id = str(uuid.uuid4())
+    command = build_command(
+        agent=args.agent,
+        prompt=prompt,
+        workspace=workspace,
+        model=args.model,
+        reasoning_effort=args.reasoning_effort,
+        pi_provider=args.pi_provider,
+        session_id=session_id,
+        attempt_dir=attempt_dir,
+        probe=args.probe,
+    )
+    record: dict[str, Any] = {
+        "execution_id": execution_id,
+        "label": args.label,
+        "workspace": str(workspace),
+        "agent": args.agent,
+        "command": sanitized_command(command, prompt),
+        "model": args.model,
+        "reasoning_effort": args.reasoning_effort,
+        "prompt_sha256": hashlib.sha256(prompt.encode()).hexdigest(),
+        "session_id": session_id,
+        "stream_log": str(stream),
+        "locator": str(locator),
+        "started_at": now(),
+        "status": "starting",
+    }
+    persist(locator, result, record)
+    emit(
+        "AGENT_STARTED",
+        agent=args.agent,
+        execution_id=execution_id,
+        label=args.label,
+        locator=str(locator),
+        model=args.model or "default",
+    )
+    with stream.open("wb") as output:
+        try:
+            process = subprocess.Popen(
+                command,
+                cwd=workspace,
+                env={
+                    **os.environ,
+                    "MILESTONE_PREPARATION_EXECUTION_ID": execution_id,
+                },
+                stdout=output,
+                stderr=subprocess.STDOUT,
+                start_new_session=True,
+            )
+        except OSError as exc:
+            record.update(status="failed", finished_at=now(), exit_code=127, error=str(exc))
+            persist(locator, result, record)
+            emit("AGENT_FINISHED", execution_id=execution_id, label=args.label, result="failed", exit_code=127)
+            return 127
+        record.update(
+            status="running",
+            agent_pid=process.pid,
+            agent_process_start_token=process_start_token(process.pid),
+        )
+        persist(locator, result, record)
+        try:
+            exit_code = process.wait()
+        except KeyboardInterrupt:
+            record.update(status="tracking", interrupted_at=now(), agent_pid=process.pid)
+            persist(locator, result, record)
+            emit(
+                "AGENT_TRACKING",
+                execution_id=execution_id,
+                label=args.label,
+                locator=str(locator),
+                pid=process.pid,
+            )
+            return 3
+
+    size = stream.stat().st_size
+    status = "succeeded" if exit_code == 0 and size > 0 else "failed"
+    if args.probe and status == "succeeded":
+        content = stream.read_text(encoding="utf-8", errors="replace")
+        if PROBE_EXPECTED not in content.upper():
+            status = "failed"
+            exit_code = 2
+    record.update(status=status, finished_at=now(), exit_code=exit_code, output_bytes=size)
+    persist(locator, result, record)
+    emit(
+        "AGENT_FINISHED",
+        execution_id=execution_id,
+        exit_code=exit_code,
+        label=args.label,
+        locator=str(locator),
+        result=status,
+    )
+    return exit_code if status == "succeeded" else (exit_code or 2)
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    args = parser().parse_args(argv)
+    try:
+        return execute(args)
+    except (AgentRunError, OSError) as exc:
+        emit("AGENT_FINISHED", label=getattr(args, "label", "one-shot"), result="failed", reason=str(exc))
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

agent-ops/skills/common/prepare-epic-work-items/scripts/run_epic_cycle.py

@@ -0,0 +1,848 @@
+#!/usr/bin/env python3
+"""Run the event-driven preparation cycle for exactly one Milestone Epic."""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+import fcntl
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import sys
+from typing import Any, Iterable
+
+
+STAGES = ("materialize", "initial-review", "refine", "final-review")
+DEFAULT_PLANNER_AGENT = "codex"
+DEFAULT_PLANNER_MODEL = "gpt-5.6-sol"
+DEFAULT_REASONING_EFFORT = "xhigh"
+PLAN_PATTERN = "PLAN-*-G??.md"
+REVIEW_PATTERN = "CODE_REVIEW-*-G??.md"
+HEADER = re.compile(r"^<!--\s+(?P<body>.*?)\s+-->$")
+MILESTONE_PATTERN = re.compile(
+    r"^agent-roadmap/phase/(?P<phase>[a-z0-9-]+)/milestones/(?P<slug>[a-z0-9-]+)\.md$"
+)
+EPIC_HEADING = re.compile(r"^### Epic:\s*\[(?P<id>[a-z0-9-]+)\]\s*(?P<title>.+?)\s*$")
+TASK_LINE = re.compile(r"^- \[(?P<done>[ xX])\] \[(?P<id>[a-z0-9-]+)\]\s+(?P<body>.+)$")
+SENSITIVE_PARTS = {".env", "secret", "secrets", "credential", "credentials", "password", "passwords"}
+SENSITIVE_SUFFIXES = {".pem", ".key", ".p12"}
+
+
+class CycleError(RuntimeError):
+    """A fail-closed Epic cycle error."""
+
+
+class TrackingRequired(CycleError):
+    """A one-shot agent is still owned by an existing execution handle."""
+
+    def __init__(self, stage: str, result: dict[str, Any]) -> None:
+        self.stage = stage
+        self.result = result
+        super().__init__(f"agent execution is still running: stage={stage}")
+
+
+class TrackingRecoveryRequired(TrackingRequired):
+    """A detached execution ended and its artifacts need explicit adoption."""
+
+
+@dataclass(frozen=True)
+class Epic:
+    epic_id: str
+    title: str
+    task_ids: tuple[str, ...]
+    incomplete_ids: tuple[str, ...]
+    body: str
+
+
+def emit(event: str, **payload: Any) -> None:
+    print(json.dumps({"event": event, **payload}, ensure_ascii=False, sort_keys=True), flush=True)
+
+
+def run(
+    command: list[str],
+    *,
+    cwd: Path,
+    check: bool = True,
+    capture: bool = True,
+) -> subprocess.CompletedProcess[str]:
+    result = subprocess.run(
+        command,
+        cwd=cwd,
+        text=True,
+        stdout=subprocess.PIPE if capture else None,
+        stderr=subprocess.PIPE if capture else None,
+        check=False,
+    )
+    if check and result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        raise CycleError(f"command failed ({result.returncode}): {' '.join(command)}: {detail}")
+    return result
+
+
+def git(workspace: Path, *arguments: str, check: bool = True) -> str:
+    result = run(["git", *arguments], cwd=workspace, check=check)
+    return (result.stdout or "").strip()
+
+
+def resolve_workspace(raw: str) -> Path:
+    workspace = Path(raw).expanduser().resolve()
+    if not workspace.is_dir():
+        raise CycleError(f"workspace not found: {workspace}")
+    top = Path(git(workspace, "rev-parse", "--show-toplevel")).resolve()
+    if top != workspace:
+        raise CycleError(f"workspace must be git root: expected={top} actual={workspace}")
+    branch = git(workspace, "branch", "--show-current")
+    if not branch:
+        raise CycleError("detached HEAD is not supported")
+    develop = git(workspace, "config", "--get", "gitflow.branch.develop")
+    feature_prefix = git(workspace, "config", "--get", "gitflow.prefix.feature")
+    if not develop or not feature_prefix or not branch.startswith(feature_prefix):
+        raise CycleError(f"Epic preparation requires a Git Flow feature branch: actual={branch}")
+    return workspace
+
+
+def resolve_milestone(workspace: Path, raw: str) -> tuple[Path, re.Match[str]]:
+    candidate = Path(raw).expanduser()
+    path = (workspace / candidate).resolve() if not candidate.is_absolute() else candidate.resolve()
+    try:
+        relative = path.relative_to(workspace).as_posix()
+    except ValueError as exc:
+        raise CycleError(f"milestone outside workspace: {path}") from exc
+    match = MILESTONE_PATTERN.fullmatch(relative)
+    if match is None or not path.is_file():
+        raise CycleError(f"active milestone path required: {relative}")
+    return path, match
+
+
+def section(text: str, heading: str) -> str:
+    match = re.search(
+        rf"^## {re.escape(heading)}\s*$\n(?P<body>.*?)(?=^##\s|\Z)",
+        text,
+        re.MULTILINE | re.DOTALL,
+    )
+    return match.group("body").strip() if match else ""
+
+
+def verify_milestone_gate(text: str) -> None:
+    status_body = section(text, "상태")
+    status_match = re.search(r"^\[(.+?)\]\s*$", status_body, re.MULTILINE)
+    status = status_match.group(1).strip() if status_match else ""
+    if status not in {"계획", "진행중"}:
+        raise CycleError(f"milestone must be [계획] or [진행중]: actual={status or 'missing'}")
+    lock = section(text, "구현 잠금")
+    if not re.search(r"^- 상태:\s*해제\s*$", lock, re.MULTILINE):
+        raise CycleError("milestone implementation lock is not 해제")
+    if not re.search(r"^- 결정 필요:\s*없음\s*$", lock, re.MULTILINE):
+        raise CycleError("milestone has unresolved 결정 필요")
+
+
+def parse_epics(text: str) -> list[Epic]:
+    lines = text.splitlines()
+    starts: list[tuple[int, re.Match[str]]] = []
+    for index, line in enumerate(lines):
+        match = EPIC_HEADING.fullmatch(line)
+        if match:
+            starts.append((index, match))
+    epics: list[Epic] = []
+    for position, (start, match) in enumerate(starts):
+        end = starts[position + 1][0] if position + 1 < len(starts) else len(lines)
+        body_lines = lines[start + 1 : end]
+        tasks = [TASK_LINE.fullmatch(line) for line in body_lines]
+        task_matches = [value for value in tasks if value is not None]
+        epics.append(
+            Epic(
+                epic_id=match.group("id"),
+                title=match.group("title"),
+                task_ids=tuple(value.group("id") for value in task_matches),
+                incomplete_ids=tuple(
+                    value.group("id") for value in task_matches if value.group("done") == " "
+                ),
+                body="\n".join(body_lines).strip(),
+            )
+        )
+    return epics
+
+
+def select_epic(epics: list[Epic], selector: str) -> Epic:
+    exact_id = [epic for epic in epics if epic.epic_id == selector]
+    if len(exact_id) == 1:
+        return exact_id[0]
+    normalized = selector.casefold().strip()
+    exact_title = [epic for epic in epics if epic.title.casefold().strip() == normalized]
+    if len(exact_title) == 1:
+        return exact_title[0]
+    raise CycleError(f"target Epic must resolve exactly once: selector={selector}")
+
+
+def git_common_dir(workspace: Path) -> Path:
+    raw = Path(git(workspace, "rev-parse", "--git-common-dir"))
+    return (workspace / raw).resolve() if not raw.is_absolute() else raw.resolve()
+
+
+def atomic_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + f".tmp.{os.getpid()}")
+    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    os.replace(temporary, path)
+
+
+def read_state(path: Path) -> dict[str, Any] | None:
+    if not path.exists():
+        return None
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise CycleError(f"invalid cycle state: {path}: {exc}") from exc
+    if not isinstance(value, dict):
+        raise CycleError(f"cycle state must be an object: {path}")
+    return value
+
+
+def process_start_token(pid: int) -> str | None:
+    stat = Path(f"/proc/{pid}/stat")
+    try:
+        remainder = stat.read_text(encoding="utf-8").rsplit(")", 1)[1].split()
+        return f"proc:{remainder[19]}"
+    except (OSError, IndexError):
+        return None
+
+
+def process_is_same(pid: object, expected_token: object) -> bool:
+    if not isinstance(pid, int) or pid <= 0:
+        return False
+    try:
+        os.kill(pid, 0)
+    except (OSError, ValueError):
+        return False
+    actual_token = process_start_token(pid)
+    if expected_token is None or actual_token is None:
+        return True
+    return actual_token == expected_token
+
+
+def stage_result(path: Path, *, workspace: Path, label: str) -> dict[str, Any] | None:
+    value = read_state(path)
+    if value is None:
+        return None
+    if value.get("workspace") != str(workspace) or value.get("label") != label:
+        raise CycleError(f"agent result identity mismatch: {path}")
+    locator = value.get("locator")
+    if not isinstance(locator, str) or not Path(locator).is_file():
+        raise CycleError(f"agent result locator is missing: {path}")
+    return value
+
+
+def changed_paths(workspace: Path) -> list[str]:
+    unmerged = git(workspace, "diff", "--name-only", "--diff-filter=U")
+    if unmerged:
+        raise CycleError(f"workspace has unmerged paths: {','.join(unmerged.splitlines())}")
+    values: set[str] = set()
+    for arguments in (
+        ("diff", "--name-only", "--no-renames", "-z", "HEAD"),
+        ("ls-files", "--others", "--exclude-standard", "-z"),
+    ):
+        result = subprocess.run(
+            ["git", *arguments],
+            cwd=workspace,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            check=False,
+        )
+        if result.returncode != 0:
+            raise CycleError(result.stderr.decode("utf-8", "replace").strip())
+        values.update(
+            part.decode("utf-8", "surrogateescape")
+            for part in result.stdout.split(b"\0")
+            if part
+        )
+    return sorted(values)
+
+
+def sensitive(path: str) -> bool:
+    candidate = Path(path)
+    lowered = {part.casefold() for part in candidate.parts}
+    if lowered & SENSITIVE_PARTS:
+        return True
+    name = candidate.name.casefold()
+    return any(token in name for token in ("secret", "credential", "password")) or candidate.suffix.casefold() in SENSITIVE_SUFFIXES
+
+
+def parse_header(path: Path) -> dict[str, str]:
+    try:
+        first = path.read_text(encoding="utf-8").splitlines()[0]
+    except (OSError, IndexError) as exc:
+        raise CycleError(f"missing first-line metadata: {path}") from exc
+    match = HEADER.fullmatch(first)
+    if not match:
+        raise CycleError(f"invalid first-line metadata: {path}")
+    values: dict[str, str] = {}
+    for token in match.group("body").split():
+        key, separator, value = token.partition("=")
+        if separator:
+            values[key] = value
+    required = {"task", "plan", "tag", "milestone-task"}
+    missing = sorted(required - values.keys())
+    if missing:
+        raise CycleError(f"metadata fields missing in {path}: {','.join(missing)}")
+    return values
+
+
+def active_pairs(workspace: Path, task_group: str) -> list[tuple[Path, Path, dict[str, str]]]:
+    root = workspace / "agent-task" / task_group
+    if not root.exists():
+        return []
+    directories = [root, *sorted(path for path in root.iterdir() if path.is_dir())]
+    pairs: list[tuple[Path, Path, dict[str, str]]] = []
+    for directory in directories:
+        plans = sorted(directory.glob(PLAN_PATTERN))
+        reviews = sorted(directory.glob(REVIEW_PATTERN))
+        if not plans and not reviews:
+            continue
+        if len(plans) != 1 or len(reviews) != 1:
+            raise CycleError(f"active PLAN/CODE_REVIEW pair required: {directory}")
+        plan_header = parse_header(plans[0])
+        review_header = parse_header(reviews[0])
+        if plan_header != review_header:
+            raise CycleError(f"PLAN/CODE_REVIEW metadata mismatch: {directory}")
+        if plan_header["task"].split("/", 1)[0] != task_group:
+            raise CycleError(f"task group mismatch: {plans[0]}")
+        pairs.append((plans[0], reviews[0], plan_header))
+    return pairs
+
+
+def validate_pairs(
+    workspace: Path,
+    task_group: str,
+    epic_task_ids: set[str],
+) -> tuple[list[tuple[Path, Path, dict[str, str]]], set[str]]:
+    pairs = active_pairs(workspace, task_group)
+    union: set[str] = set()
+    dispatcher = (
+        workspace
+        / "agent-ops"
+        / "skills"
+        / "common"
+        / "orchestrate-agent-task-loop"
+        / "scripts"
+        / "dispatch.py"
+    )
+    for plan, review, header in pairs:
+        ids = header["milestone-task"].split(",")
+        if not ids or any(not value for value in ids) or len(ids) != len(set(ids)):
+            raise CycleError(f"invalid milestone-task list: {plan}")
+        outside = sorted(set(ids) - epic_task_ids)
+        if outside:
+            raise CycleError(f"plan includes Task ids outside target Epic: {plan}: {','.join(outside)}")
+        union.update(ids)
+        contents = plan.read_text(encoding="utf-8") + "\n" + review.read_text(encoding="utf-8")
+        if "[TODO" in contents or "<task_group>" in contents or "<milestone-slug>" in contents:
+            raise CycleError(f"unresolved template token: {plan.parent}")
+        if dispatcher.is_file():
+            run(
+                [sys.executable, str(dispatcher), "--workspace", str(workspace), "--validate-plan", str(plan)],
+                cwd=workspace,
+            )
+    git(workspace, "diff", "--check")
+    return pairs, union
+
+
+def active_task_user_reviews(workspace: Path, task_group: str) -> list[Path]:
+    root = workspace / "agent-task" / task_group
+    if not root.exists():
+        return []
+    return sorted(root.glob("USER_REVIEW.md")) + sorted(root.glob("*/USER_REVIEW.md"))
+
+
+def sdd_user_review(workspace: Path, phase_slug: str, milestone_slug: str) -> Path:
+    return workspace / "agent-roadmap" / "sdd" / phase_slug / milestone_slug / "USER_REVIEW.md"
+
+
+def stage_prompt(
+    *,
+    stage: str,
+    workspace: Path,
+    milestone: Path,
+    epic: Epic,
+    task_group: str,
+    base_head: str,
+    checkpoint_head: str | None,
+) -> str:
+    common = f"""You are a fresh child agent launched for one bounded Epic preparation stage, not the caller or monitor.
+Work only in {workspace}.
+Read the repository AGENTS.md completely, then read agent-ops/skills/common/router.md and only the skills required for this stage.
+Do not start subagents, orchestration dispatchers, prepare-milestone-workspace, prepare-epic-work-items, or any monitoring loop.
+Do not commit or push; the parent runtime owns Git checkpoints.
+Target Milestone: {milestone}
+Target Epic: [{epic.epic_id}] {epic.title}
+Allowed Milestone Task ids: {','.join(epic.task_ids)}
+Active task group: agent-task/{task_group}
+Keep every change inside this Epic and preserve user changes. Final in Korean.
+"""
+    if stage == "materialize":
+        return common + f"""
+Materialize this Epic once. Read current source, tests, SDD, matching spec and contracts required by AGENTS.md.
+Classify cohesive slices as direct-small only when each is one bounded change, has explicit verification, changes no API/wire/schema/migration/external side effect/responsibility boundary, needs no user decision, and does not collide with planned work. Treat every uncertain slice as large.
+Implement and verify all direct-small slices first. Then, against that updated source, use agent-ops/skills/common/plan/SKILL.md in write mode to create valid PLAN/CODE_REVIEW pairs for every remaining large slice. Preserve exact milestone-task ids and let plan perform final routing. Do not use official code-review on unimplemented stubs.
+If a genuine product/scope decision is required, use roadmap-sdd review-ready for this Milestone and stop without inventing a decision.
+Starting HEAD: {base_head}
+"""
+    if stage == "initial-review":
+        return common + f"""
+Review everything produced for this Epic since {base_head}, including direct code/test/document changes and every active PLAN/CODE_REVIEW stub. This is the explicit self-review request: review the work and fix every material omission you find.
+Re-run appropriate verification for direct-small work. For semantic plan defects, use the plan skill's explicit write/replan path so routing and paired files remain valid. Do not append an official code-review verdict.
+Run sync-milestone-workstate mode=sync only for exact Task ids whose direct work is fully implemented and evidenced; never complete an id that still has pending plan scope.
+If no material defect exists, leave correct artifacts unchanged.
+"""
+    if stage == "refine":
+        return common + """
+Read agent-ops/skills/common/refine-plans/SKILL.md and apply it once to every eligible unstarted active pair in the target task group whose milestone-task ids belong to this Epic. Preserve original scope and do not re-read source/tests or run verification. A justified no-change decision is valid. Do not recursively split a child created in this pass.
+"""
+    if stage == "final-review":
+        return common + f"""
+Review the refined active pair set for this Epic from a fresh context. Compare the child scope union, milestone-task union, write sets, verification, dependencies, indices, PLAN/CODE_REVIEW metadata, and routing against the Milestone, SDD, current source, and the pre-refine intent at checkpoint {checkpoint_head or base_head}. Fix every material defect using the owning plan/refine procedure; do not append an official code-review verdict. If a semantic replan replaces a pair, apply refine-plans once to that replacement when it remains eligible. Finish only with valid unstarted pairs or no pairs when all Epic work was direct-small.
+"""
+    raise CycleError(f"unsupported stage: {stage}")
+
+
+def run_agent_stage(
+    *,
+    workspace: Path,
+    state_root: Path,
+    identity: str,
+    stage: str,
+    prompt: str,
+    agent: str,
+    model: str | None,
+    reasoning_effort: str | None,
+    pi_provider: str | None,
+    prior_cycle_status: str,
+    retry: bool,
+) -> Path:
+    runner = Path(__file__).resolve().with_name("run_agent_once.py")
+    prompt_path = state_root / "prompts" / f"{stage}.txt"
+    result_path = state_root / "attempts" / f"{stage}.json"
+    prompt_path.parent.mkdir(parents=True, exist_ok=True)
+    prompt_path.write_text(prompt, encoding="utf-8")
+    label = f"{identity}-{stage}"
+    previous = stage_result(result_path, workspace=workspace, label=label)
+    if previous is not None:
+        previous_status = previous.get("status")
+        is_live = previous_status in {"running", "tracking"} and process_is_same(
+            previous.get("agent_pid"), previous.get("agent_process_start_token")
+        )
+        if is_live:
+            raise TrackingRequired(stage, previous)
+        if previous_status == "succeeded" and not (
+            retry and prior_cycle_status == "failed"
+        ):
+            emit(
+                "AGENT_RESULT_RECOVERED",
+                stage=stage,
+                locator=previous["locator"],
+                result="succeeded",
+            )
+            return result_path
+        if previous_status in {"running", "tracking"} and prior_cycle_status in {
+            "running",
+            "tracking",
+        }:
+            if not retry:
+                raise TrackingRecoveryRequired(stage, previous)
+            emit(
+                "AGENT_RESULT_RECOVERED",
+                stage=stage,
+                locator=previous["locator"],
+                result="detached-artifacts",
+            )
+            return result_path
+        if not retry:
+            raise CycleError(
+                f"prior agent result requires --retry: stage={stage} status={previous_status}"
+            )
+    command = [
+        sys.executable,
+        str(runner),
+        "--agent",
+        agent,
+        "--workspace",
+        str(workspace),
+        "--prompt-file",
+        str(prompt_path),
+        "--label",
+        label,
+        "--result-file",
+        str(result_path),
+    ]
+    if model:
+        command.extend(["--model", model])
+    if reasoning_effort:
+        command.extend(["--reasoning-effort", reasoning_effort])
+    if pi_provider:
+        command.extend(["--pi-provider", pi_provider])
+    result = run(command, cwd=workspace, check=False, capture=False)
+    if result.returncode != 0:
+        if result.returncode == 3:
+            tracked = stage_result(result_path, workspace=workspace, label=label)
+            if tracked is None:
+                raise CycleError(f"agent tracking result missing: stage={stage}")
+            raise TrackingRequired(stage, tracked)
+        raise CycleError(f"agent stage failed: stage={stage} exit={result.returncode}")
+    final = stage_result(result_path, workspace=workspace, label=label)
+    if final is None or final.get("status") != "succeeded":
+        raise CycleError(f"agent stage returned without succeeded result: stage={stage}")
+    return result_path
+
+
+def publish(workspace: Path, epic: Epic, phase: str) -> str:
+    paths = changed_paths(workspace)
+    if paths:
+        unsafe = [path for path in paths if sensitive(path)]
+        if unsafe:
+            raise CycleError(f"sensitive path refused: {','.join(unsafe)}")
+        for path in paths:
+            git(workspace, "add", "--", path)
+        git(workspace, "diff", "--cached", "--check")
+        message = (
+            f"feat(epic): {epic.epic_id} 작업을 준비한다"
+            if phase == "initial"
+            else f"chore(epic): {epic.epic_id} 준비 결과를 검증한다"
+        )
+        git(workspace, "commit", "-m", message)
+    branch = git(workspace, "branch", "--show-current")
+    upstream = git(workspace, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
+    if not upstream:
+        raise CycleError(f"feature branch has no upstream: {branch}")
+    if not upstream.endswith(f"/{branch}"):
+        raise CycleError(f"feature branch upstream mismatch: branch={branch} upstream={upstream}")
+    git(workspace, "push")
+    if changed_paths(workspace):
+        raise CycleError("workspace is dirty after publish")
+    return git(workspace, "rev-parse", "HEAD")
+
+
+def parser() -> argparse.ArgumentParser:
+    value = argparse.ArgumentParser(description=__doc__)
+    value.add_argument("--workspace", required=True)
+    value.add_argument("--milestone", required=True)
+    value.add_argument("--epic", required=True)
+    value.add_argument(
+        "--planner-agent",
+        choices=("codex", "claude", "gemini", "pi"),
+        default=DEFAULT_PLANNER_AGENT,
+    )
+    value.add_argument("--review-agent", choices=("codex", "claude", "gemini", "pi"))
+    value.add_argument("--planner-model")
+    value.add_argument("--review-model")
+    value.add_argument("--reasoning-effort", default=DEFAULT_REASONING_EFFORT)
+    value.add_argument("--pi-provider")
+    value.add_argument("--retry", action="store_true")
+    return value
+
+
+def apply_defaults(args: argparse.Namespace) -> argparse.Namespace:
+    if args.planner_model is None and args.planner_agent == DEFAULT_PLANNER_AGENT:
+        args.planner_model = DEFAULT_PLANNER_MODEL
+    if args.reasoning_effort is None:
+        args.reasoning_effort = DEFAULT_REASONING_EFFORT
+    return args
+
+
+def cycle(args: argparse.Namespace) -> int:
+    apply_defaults(args)
+    workspace = resolve_workspace(args.workspace)
+    milestone_path, milestone_match = resolve_milestone(workspace, args.milestone)
+    milestone_text = milestone_path.read_text(encoding="utf-8")
+    verify_milestone_gate(milestone_text)
+    epic = select_epic(parse_epics(milestone_text), args.epic)
+    milestone_slug = milestone_match.group("slug")
+    phase_slug = milestone_match.group("phase")
+    task_group = f"m-{milestone_slug}"
+    identity = f"{milestone_slug}:{epic.epic_id}"
+    if not epic.task_ids:
+        raise CycleError(f"target Epic has no Task ids: {epic.epic_id}")
+    feature_prefix = git(workspace, "config", "--get", "gitflow.prefix.feature")
+    branch = git(workspace, "branch", "--show-current")
+    expected_branch = f"{feature_prefix}{milestone_slug}"
+    if branch != expected_branch:
+        raise CycleError(f"workspace branch does not match Milestone slug: expected={expected_branch} actual={branch}")
+    current_path = workspace / "agent-roadmap" / "current.md"
+    expected_current_target = f"phase/{phase_slug}/milestones/{milestone_slug}.md"
+    if not current_path.is_file() or expected_current_target not in current_path.read_text(encoding="utf-8"):
+        raise CycleError(f"workspace-local current does not select target Milestone: {current_path}")
+    state_root = git_common_dir(workspace) / "epic-work-preparation" / milestone_slug / epic.epic_id
+    state_path = state_root / "state.json"
+    state_root.mkdir(parents=True, exist_ok=True)
+    with (state_root / "cycle.lock").open("a+", encoding="utf-8") as lock:
+        try:
+            fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+            raise CycleError(f"Epic cycle already running: {identity}") from exc
+
+        state = read_state(state_path)
+        current_head = git(workspace, "rev-parse", "HEAD")
+        if state and state.get("identity") != identity:
+            raise CycleError(f"cycle state identity mismatch: {state_path}")
+        if (
+            state
+            and state.get("task_ids") is not None
+            and state.get("task_ids") != list(epic.task_ids)
+        ):
+            raise CycleError("target Epic Task ids changed after cycle scope was fixed")
+        if state and state.get("status") == "completed":
+            if changed_paths(workspace):
+                raise CycleError("completed cycle requires a clean feature workspace")
+            refreshed = select_epic(
+                parse_epics(milestone_path.read_text(encoding="utf-8")), epic.epic_id
+            )
+            pairs, task_union = validate_pairs(workspace, task_group, set(refreshed.task_ids))
+            completed_with_plan = sorted(task_union - set(refreshed.incomplete_ids))
+            if completed_with_plan:
+                raise CycleError(
+                    "completed Task ids still have active plans: " + ",".join(completed_with_plan)
+                )
+            remaining_without_plan = sorted(set(refreshed.incomplete_ids) - task_union)
+            if remaining_without_plan:
+                raise CycleError(
+                    "completed cycle no longer has evidence for incomplete Task ids; run workstate sync or recover plans: "
+                    + ",".join(remaining_without_plan)
+                )
+            terminal = "EPIC_COMPLETED" if not refreshed.incomplete_ids else "EPIC_WORK_ITEMS_READY"
+            state.update(event=terminal, head=current_head)
+            atomic_json(state_path, state)
+            emit(terminal, identity=identity, resumed=True, head=current_head, plans=len(pairs))
+            return 0
+        review_path = sdd_user_review(workspace, phase_slug, milestone_slug)
+        if state and state.get("status") == "user-review" and review_path.exists():
+            emit("USER_REVIEW", identity=identity, path=str(review_path), resumed=True)
+            return 2
+        prior_cycle_status = str(state.get("status")) if state else "new"
+        if state and state.get("status") == "failed" and not args.retry:
+            raise CycleError(f"prior terminal failure requires --retry: {state.get('reason', 'unknown')}")
+        if state is None:
+            if changed_paths(workspace):
+                raise CycleError("clean feature workspace required before a new Epic cycle")
+            if review_path.exists():
+                emit("USER_REVIEW", identity=identity, path=str(review_path))
+                return 2
+            if active_task_user_reviews(workspace, task_group):
+                raise CycleError("preparation cannot resume from agent-task USER_REVIEW")
+            if not epic.incomplete_ids:
+                existing, _ = validate_pairs(workspace, task_group, set(epic.task_ids))
+                if existing:
+                    raise CycleError(
+                        "completed Epic still has active PLAN/CODE_REVIEW pairs; reconcile them before completion"
+                    )
+                state = {
+                    "identity": identity,
+                    "task_ids": list(epic.task_ids),
+                    "status": "completed",
+                    "event": "EPIC_COMPLETED",
+                    "head": current_head,
+                }
+                atomic_json(state_path, state)
+                emit("EPIC_COMPLETED", identity=identity)
+                return 0
+            existing, _ = validate_pairs(workspace, task_group, set(epic.task_ids))
+            if existing:
+                raise CycleError("active pair already exists before new Epic cycle; select recovery explicitly")
+            state = {
+                "identity": identity,
+                "task_ids": list(epic.task_ids),
+                "status": "active",
+                "next_stage": STAGES[0],
+                "base_head": current_head,
+                "checkpoint_head": None,
+                "pre_refine_ids": [],
+            }
+            atomic_json(state_path, state)
+            emit(
+                "EPIC_SCOPE_RESOLVED",
+                identity=identity,
+                task_ids=list(epic.task_ids),
+                incomplete_ids=list(epic.incomplete_ids),
+            )
+        elif changed_paths(workspace) and not args.retry:
+            raise CycleError("dirty recovery state requires explicit --retry")
+
+        reviewer_agent = args.review_agent or args.planner_agent
+        reviewer_model = args.review_model or (
+            args.planner_model if reviewer_agent == args.planner_agent else None
+        )
+        start_index = STAGES.index(str(state.get("next_stage", STAGES[0])))
+        for stage in STAGES[start_index:]:
+            stage_head = git(workspace, "rev-parse", "HEAD")
+            event_prefix = stage.upper().replace("-", "_")
+            emit(f"{event_prefix}_STARTED", identity=identity)
+            agent = args.planner_agent if stage in {"materialize", "refine"} else reviewer_agent
+            model = args.planner_model if stage in {"materialize", "refine"} else reviewer_model
+            prompt = stage_prompt(
+                stage=stage,
+                workspace=workspace,
+                milestone=milestone_path,
+                epic=epic,
+                task_group=task_group,
+                base_head=str(state["base_head"]),
+                checkpoint_head=state.get("checkpoint_head"),
+            )
+            result_path = state_root / "attempts" / f"{stage}.json"
+            state.update(
+                status="running",
+                current_stage=stage,
+                active_result=str(result_path),
+            )
+            atomic_json(state_path, state)
+            try:
+                run_agent_stage(
+                    workspace=workspace,
+                    state_root=state_root,
+                    identity=identity.replace(":", "-"),
+                    stage=stage,
+                    prompt=prompt,
+                    agent=agent,
+                    model=model,
+                    reasoning_effort=args.reasoning_effort,
+                    pi_provider=args.pi_provider,
+                    prior_cycle_status=prior_cycle_status,
+                    retry=args.retry,
+                )
+            except TrackingRequired as exc:
+                state.update(
+                    status="tracking",
+                    current_stage=stage,
+                    active_result=str(result_path),
+                    locator=exc.result.get("locator"),
+                )
+                atomic_json(state_path, state)
+                if isinstance(exc, TrackingRecoveryRequired):
+                    state["recovery_required"] = True
+                    atomic_json(state_path, state)
+                    emit(
+                        "AGENT_RECOVERY_REQUIRED",
+                        identity=identity,
+                        stage=stage,
+                        locator=exc.result.get("locator"),
+                        action="inspect locator, then rerun with --retry to adopt artifacts",
+                    )
+                else:
+                    state.pop("recovery_required", None)
+                    atomic_json(state_path, state)
+                    emit(
+                        "AGENT_TRACKING",
+                        identity=identity,
+                        stage=stage,
+                        locator=exc.result.get("locator"),
+                        pid=exc.result.get("agent_pid"),
+                    )
+                return 3
+            if git(workspace, "rev-parse", "HEAD") != stage_head:
+                raise CycleError(f"child agent committed unexpectedly: stage={stage}")
+            refreshed_scope = select_epic(
+                parse_epics(milestone_path.read_text(encoding="utf-8")), epic.epic_id
+            )
+            if refreshed_scope.task_ids != epic.task_ids:
+                raise CycleError(f"target Epic Task ids changed unexpectedly: stage={stage}")
+            pairs, task_union = validate_pairs(workspace, task_group, set(epic.task_ids))
+            if active_task_user_reviews(workspace, task_group):
+                raise CycleError("preparation agent created forbidden agent-task USER_REVIEW")
+            if review_path.exists():
+                head = publish(workspace, epic, "user-review")
+                state.update(status="user-review", event="USER_REVIEW", head=head, next_stage=stage)
+                atomic_json(state_path, state)
+                emit("USER_REVIEW", identity=identity, path=str(review_path), head=head)
+                return 2
+            if stage == "materialize" and not changed_paths(workspace) and not pairs:
+                raise CycleError("materialize produced neither direct work nor PLAN pairs")
+            if stage == "refine":
+                expected_union = set(state.get("pre_refine_ids", []))
+                if task_union != expected_union:
+                    raise CycleError(
+                        f"refine changed milestone-task union: before={sorted(expected_union)} after={sorted(task_union)}"
+                    )
+            if stage == "final-review":
+                expected_union = set(state.get("pre_refine_ids", []))
+                if task_union != expected_union:
+                    raise CycleError(
+                        f"final review changed milestone-task union: before={sorted(expected_union)} after={sorted(task_union)}"
+                    )
+            emit(f"{event_prefix}_FINISHED", identity=identity, plans=len(pairs))
+            state.pop("current_stage", None)
+            state.pop("active_result", None)
+            state.pop("locator", None)
+            state.pop("recovery_required", None)
+
+            if stage == "initial-review":
+                state["pre_refine_ids"] = sorted(task_union)
+                checkpoint = publish(workspace, epic, "initial")
+                state["checkpoint_head"] = checkpoint
+                emit("INITIAL_CHECKPOINT_PUSHED", identity=identity, head=checkpoint)
+            elif stage == "final-review":
+                final_head = publish(workspace, epic, "final")
+                emit("FINAL_ARTIFACTS_PUSHED", identity=identity, head=final_head)
+                refreshed = select_epic(
+                    parse_epics(milestone_path.read_text(encoding="utf-8")), epic.epic_id
+                )
+                completed_with_plan = sorted(task_union - set(refreshed.incomplete_ids))
+                if completed_with_plan:
+                    raise CycleError(
+                        "completed Task ids still have active plans: "
+                        + ",".join(completed_with_plan)
+                    )
+                remaining_without_plan = sorted(set(refreshed.incomplete_ids) - task_union)
+                if remaining_without_plan:
+                    raise CycleError(
+                        "incomplete Epic Task ids have neither completion sync nor active plans: "
+                        + ",".join(remaining_without_plan)
+                    )
+                terminal = "EPIC_COMPLETED" if not refreshed.incomplete_ids else "EPIC_WORK_ITEMS_READY"
+                state.update(status="completed", event=terminal, head=final_head, next_stage=None)
+                atomic_json(state_path, state)
+                emit(terminal, identity=identity, head=final_head, plans=len(pairs))
+                return 0
+
+            next_index = STAGES.index(stage) + 1
+            state["next_stage"] = STAGES[next_index]
+            state["status"] = "active"
+            atomic_json(state_path, state)
+            prior_cycle_status = "active"
+        raise CycleError("cycle ended without terminal state")
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    args = apply_defaults(parser().parse_args(argv))
+    state_path: Path | None = None
+    identity = "unknown"
+    try:
+        return cycle(args)
+    except (CycleError, OSError, ValueError) as exc:
+        try:
+            workspace = Path(args.workspace).expanduser().resolve()
+            milestone = Path(args.milestone)
+            if not milestone.is_absolute():
+                milestone = (workspace / milestone).resolve()
+            match = MILESTONE_PATTERN.fullmatch(milestone.relative_to(workspace).as_posix())
+            if match:
+                selected = select_epic(
+                    parse_epics(milestone.read_text(encoding="utf-8")), args.epic
+                )
+                identity = f"{match.group('slug')}:{selected.epic_id}"
+                state_path = (
+                    git_common_dir(workspace)
+                    / "epic-work-preparation"
+                    / match.group("slug")
+                    / selected.epic_id
+                    / "state.json"
+                )
+                prior = read_state(state_path) or {"identity": identity}
+                prior.update(status="failed", event="FAILED", reason=str(exc))
+                atomic_json(state_path, prior)
+        except Exception:
+            pass
+        emit("FAILED", identity=identity, reason=str(exc), state=str(state_path) if state_path else None)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

agent-ops/skills/common/prepare-epic-work-items/tests/test_run_agent_once.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "run_agent_once.py"
+SPEC = importlib.util.spec_from_file_location("run_agent_once", SCRIPT)
+assert SPEC and SPEC.loader
+MODULE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(MODULE)
+REPOSITORY = Path(__file__).resolve().parents[5]
+
+
+class AgentCommandTest(unittest.TestCase):
+    def build(self, agent: str) -> list[str]:
+        with tempfile.TemporaryDirectory() as raw:
+            path = Path(raw)
+            return MODULE.build_command(
+                agent=agent,
+                prompt="prompt",
+                workspace=path,
+                model="model-name",
+                reasoning_effort="high",
+                pi_provider="provider-name",
+                session_id="session-id",
+                attempt_dir=path,
+            )
+
+    def test_codex_contract(self) -> None:
+        command = self.build("codex")
+        self.assertEqual(command[:3], ["codex", "exec", "--json"])
+        self.assertIn("--dangerously-bypass-approvals-and-sandbox", command)
+
+    def test_claude_contract(self) -> None:
+        command = self.build("claude")
+        self.assertEqual(command[0], "claude")
+        self.assertIn("--output-format", command)
+        self.assertIn("--session-id", command)
+
+    def test_gemini_maps_to_agy(self) -> None:
+        command = self.build("gemini")
+        self.assertEqual(command[0], "agy")
+        self.assertEqual(command[1:3], ["--print", "prompt"])
+
+    def test_pi_contract(self) -> None:
+        command = self.build("pi")
+        self.assertEqual(command[0], "pi")
+        self.assertIn("--mode", command)
+        self.assertIn("--session-id", command)
+
+    def test_probe_commands_do_not_enable_mutating_permission_bypass(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            path = Path(raw)
+            for agent in MODULE.AGENT_COMMAND:
+                command = MODULE.build_command(
+                    agent=agent,
+                    prompt="READY",
+                    workspace=path,
+                    model=None,
+                    reasoning_effort=None,
+                    pi_provider=None,
+                    session_id="session-id",
+                    attempt_dir=path,
+                    probe=True,
+                )
+                self.assertNotIn("--dangerously-bypass-approvals-and-sandbox", command)
+                self.assertNotIn("--dangerously-skip-permissions", command)
+                self.assertNotIn("--approve", command)
+
+    def test_probe_executes_selected_command_once(self) -> None:
+        with tempfile.TemporaryDirectory(dir=REPOSITORY) as raw:
+            root = Path(raw)
+            workspace = root / "workspace"
+            binary = root / "bin"
+            workspace.mkdir()
+            binary.mkdir()
+            subprocess.run(
+                ["git", "init", "-b", "main", str(workspace)],
+                check=True,
+                stdout=subprocess.DEVNULL,
+            )
+            fake = binary / "codex"
+            fake.write_text(
+                "#!/bin/sh\nprintf '%s\\n' MILESTONE_AGENT_READY\n",
+                encoding="utf-8",
+            )
+            fake.chmod(0o755)
+            result_file = (
+                workspace
+                / ".git"
+                / "epic-work-preparation"
+                / "probe-cycle"
+                / "result.json"
+            )
+            with mock.patch.dict(os.environ, {"PATH": f"{binary}:{os.environ['PATH']}"}):
+                result = MODULE.main(
+                    [
+                        "--agent",
+                        "codex",
+                        "--workspace",
+                        str(workspace),
+                        "--probe",
+                        "--result-file",
+                        str(result_file),
+                    ]
+                )
+            self.assertEqual(result, 0)
+            recorded = json.loads(result_file.read_text(encoding="utf-8"))
+            self.assertEqual(recorded["status"], "succeeded")
+            self.assertEqual(recorded["model"], "gpt-5.6-sol")
+            self.assertEqual(recorded["reasoning_effort"], "xhigh")
+            self.assertIn("agent_process_start_token", recorded)
+            locators = list((workspace / ".git" / "epic-work-preparation" / "runs").glob("*/locator.json"))
+            self.assertEqual(len(locators), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

agent-ops/skills/common/prepare-epic-work-items/tests/test_run_epic_cycle.py

@@ -0,0 +1,281 @@
+from __future__ import annotations
+
+import importlib.util
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "run_epic_cycle.py"
+SPEC = importlib.util.spec_from_file_location("run_epic_cycle", SCRIPT)
+assert SPEC and SPEC.loader
+MODULE = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = MODULE
+SPEC.loader.exec_module(MODULE)
+
+
+def command(cwd: Path, *args: str) -> str:
+    result = subprocess.run(
+        list(args),
+        cwd=cwd,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=True,
+    )
+    return result.stdout.strip()
+
+
+class EpicCycleContractTest(unittest.TestCase):
+    def test_cycle_defaults_to_codex_top_model_and_reasoning(self) -> None:
+        args = MODULE.parser().parse_args(
+            ["--workspace", "/workspace", "--milestone", "milestone.md", "--epic", "epic"]
+        )
+        MODULE.apply_defaults(args)
+        self.assertEqual(args.planner_agent, "codex")
+        self.assertEqual(args.planner_model, "gpt-5.6-sol")
+        self.assertEqual(args.reasoning_effort, "xhigh")
+
+        other = MODULE.parser().parse_args(
+            [
+                "--workspace",
+                "/workspace",
+                "--milestone",
+                "milestone.md",
+                "--epic",
+                "epic",
+                "--planner-agent",
+                "claude",
+            ]
+        )
+        MODULE.apply_defaults(other)
+        self.assertIsNone(other.planner_model)
+        self.assertEqual(other.reasoning_effort, "xhigh")
+
+    def test_live_stage_result_requires_tracking_without_relaunch(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            root = Path(raw)
+            workspace = root / "workspace"
+            state_root = root / "state"
+            workspace.mkdir()
+            locator = state_root / "runs" / "locator.json"
+            locator.parent.mkdir(parents=True)
+            locator.write_text("{}\n", encoding="utf-8")
+            result_path = state_root / "attempts" / "materialize.json"
+            MODULE.atomic_json(
+                result_path,
+                {
+                    "workspace": str(workspace),
+                    "label": "sample-epic-materialize",
+                    "status": "tracking",
+                    "agent_pid": os.getpid(),
+                    "agent_process_start_token": MODULE.process_start_token(os.getpid()),
+                    "locator": str(locator),
+                },
+            )
+            with self.assertRaises(MODULE.TrackingRequired):
+                MODULE.run_agent_stage(
+                    workspace=workspace,
+                    state_root=state_root,
+                    identity="sample-epic",
+                    stage="materialize",
+                    prompt="prompt",
+                    agent="codex",
+                    model=None,
+                    reasoning_effort=None,
+                    pi_provider=None,
+                    prior_cycle_status="tracking",
+                    retry=False,
+                )
+
+    def test_dead_tracking_result_requires_explicit_artifact_adoption(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            root = Path(raw)
+            workspace = root / "workspace"
+            state_root = root / "state"
+            workspace.mkdir()
+            locator = state_root / "runs" / "locator.json"
+            locator.parent.mkdir(parents=True)
+            locator.write_text("{}\n", encoding="utf-8")
+            result_path = state_root / "attempts" / "materialize.json"
+            MODULE.atomic_json(
+                result_path,
+                {
+                    "workspace": str(workspace),
+                    "label": "sample-epic-materialize",
+                    "status": "tracking",
+                    "agent_pid": 999_999_999,
+                    "agent_process_start_token": "proc:missing",
+                    "locator": str(locator),
+                },
+            )
+            values = {
+                "workspace": workspace,
+                "state_root": state_root,
+                "identity": "sample-epic",
+                "stage": "materialize",
+                "prompt": "prompt",
+                "agent": "codex",
+                "model": None,
+                "reasoning_effort": None,
+                "pi_provider": None,
+                "prior_cycle_status": "tracking",
+            }
+            with self.assertRaises(MODULE.TrackingRecoveryRequired):
+                MODULE.run_agent_stage(**values, retry=False)
+            self.assertEqual(
+                MODULE.run_agent_stage(**values, retry=True),
+                result_path,
+            )
+
+    def test_parse_and_select_exact_epic(self) -> None:
+        epics = MODULE.parse_epics(
+            """## 기능
+
+### Epic: [first] First Epic
+
+- [x] [done-task] completed
+- [ ] [open-task] pending
+
+### Epic: [second] Second Epic
+
+- [ ] [other-task] pending
+"""
+        )
+        selected = MODULE.select_epic(epics, "first")
+        self.assertEqual(selected.task_ids, ("done-task", "open-task"))
+        self.assertEqual(selected.incomplete_ids, ("open-task",))
+
+    def test_validate_pair_rejects_task_outside_epic(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            workspace = Path(raw)
+            task = workspace / "agent-task" / "m-sample" / "01_work"
+            task.mkdir(parents=True)
+            header = "<!-- task=m-sample/01_work plan=0 tag=TEST milestone-task=outside -->\n"
+            (task / "PLAN-local-G01.md").write_text(header + "# Plan\n", encoding="utf-8")
+            (task / "CODE_REVIEW-local-G01.md").write_text(header + "# Review\n", encoding="utf-8")
+            with self.assertRaises(MODULE.CycleError):
+                MODULE.validate_pairs(workspace, "m-sample", {"inside"})
+
+    def test_full_cycle_with_fresh_fake_codex_passes_and_pushes(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            root = Path(raw)
+            remote = root / "remote.git"
+            workspace = root / "workspace"
+            command(root, "git", "init", "--bare", str(remote))
+            command(root, "git", "init", "-b", "dev", str(workspace))
+            command(workspace, "git", "config", "user.name", "Test Agent")
+            command(workspace, "git", "config", "user.email", "agent@example.test")
+            command(workspace, "git", "config", "gitflow.branch.develop", "dev")
+            command(workspace, "git", "config", "gitflow.prefix.feature", "feature/")
+            command(workspace, "git", "remote", "add", "origin", str(remote))
+            milestone = (
+                workspace
+                / "agent-roadmap"
+                / "phase"
+                / "phase-one"
+                / "milestones"
+                / "sample-milestone.md"
+            )
+            milestone.parent.mkdir(parents=True)
+            milestone.write_text(
+                "# Milestone: [sample-01] Sample\n\n"
+                "## 상태\n\n[계획]\n\n"
+                "## 구현 잠금\n\n- 상태: 해제\n- 결정 필요: 없음\n\n"
+                "## 기능\n\n"
+                "### Epic: [sample-epic] Sample Epic\n\n"
+                "- [ ] [large-task] large work\n",
+                encoding="utf-8",
+            )
+            current = workspace / "agent-roadmap" / "current.md"
+            current.write_text(
+                "# 현재 로드맵 컨텍스트\n\n"
+                "## 활성 Milestone\n\n"
+                "- [계획] Sample\n"
+                "  - 경로: [Sample](phase/phase-one/milestones/sample-milestone.md)\n",
+                encoding="utf-8",
+            )
+            (workspace / ".gitignore").write_text("agent-roadmap/current.md\n", encoding="utf-8")
+            command(workspace, "git", "add", ".gitignore", "agent-roadmap")
+            command(workspace, "git", "commit", "-m", "init")
+            command(workspace, "git", "push", "-u", "origin", "dev")
+            command(workspace, "git", "switch", "-c", "feature/sample-milestone")
+            command(workspace, "git", "push", "-u", "origin", "feature/sample-milestone")
+
+            def fake_agent_stage(**values: object) -> None:
+                if "Materialize this Epic once" not in str(values["prompt"]):
+                    return
+                task = workspace / "agent-task" / "m-sample-milestone"
+                task.mkdir(parents=True, exist_ok=True)
+                header = "<!-- task=m-sample-milestone plan=0 tag=TEST milestone-task=large-task -->\n"
+                (task / "PLAN-local-G01.md").write_text(header + "# Plan\n", encoding="utf-8")
+                (task / "CODE_REVIEW-local-G01.md").write_text(
+                    header + "# Review\n", encoding="utf-8"
+                )
+
+            with mock.patch.object(MODULE, "run_agent_stage", side_effect=fake_agent_stage):
+                result = MODULE.main(
+                    [
+                        "--workspace",
+                        str(workspace),
+                        "--milestone",
+                        str(milestone.relative_to(workspace)),
+                        "--epic",
+                        "sample-epic",
+                        "--planner-agent",
+                        "codex",
+                    ]
+                )
+            self.assertEqual(result, 0)
+            self.assertFalse(command(workspace, "git", "status", "--porcelain=v1"))
+            self.assertEqual(
+                command(workspace, "git", "rev-parse", "HEAD"),
+                command(workspace, "git", "rev-parse", "origin/feature/sample-milestone"),
+            )
+            state_path = (
+                MODULE.git_common_dir(workspace)
+                / "epic-work-preparation"
+                / "sample-milestone"
+                / "sample-epic"
+                / "state.json"
+            )
+            state = MODULE.read_state(state_path)
+            self.assertEqual(state["event"], "EPIC_WORK_ITEMS_READY")
+
+            task_root = workspace / "agent-task" / "m-sample-milestone"
+            for path in task_root.iterdir():
+                path.unlink()
+            task_root.rmdir()
+            milestone.write_text(
+                milestone.read_text(encoding="utf-8").replace(
+                    "- [ ] [large-task] large work",
+                    "- [x] [large-task] large work",
+                ),
+                encoding="utf-8",
+            )
+            command(workspace, "git", "add", "-A", "agent-roadmap", "agent-task")
+            command(workspace, "git", "commit", "-m", "complete epic work")
+            command(workspace, "git", "push")
+
+            resumed = MODULE.main(
+                [
+                    "--workspace",
+                    str(workspace),
+                    "--milestone",
+                    str(milestone.relative_to(workspace)),
+                    "--epic",
+                    "sample-epic",
+                    "--planner-agent",
+                    "codex",
+                ]
+            )
+            self.assertEqual(resumed, 0)
+            self.assertEqual(MODULE.read_state(state_path)["event"], "EPIC_COMPLETED")
+
+
+if __name__ == "__main__":
+    unittest.main()

agent-ops/skills/common/prepare-milestone-workspace/SKILL.md

@@ -0,0 +1,100 @@
+---
+name: prepare-milestone-workspace
+description: 명시된 workspace에 계획 상태의 Milestone을 현재 작업현황과 동기화한 뒤 Git Flow feature branch를 Milestone slug로 생성·push하고 worktree로 준비해 첫 Epic 작업 사이클로 넘길 때 사용한다. "workspace에 milestone 작업 준비해" 요청에서 사용한다.
+---
+
+# Prepare Milestone Workspace
+
+`<workspace>에 <milestone> 작업 준비해` 요청은 workspace만 추정하거나 만드는 요청이 아니라, 지정된 workspace에 대상 Milestone의 정합성 확인부터 첫 Epic 준비 사이클 연결까지 수행하는 전체 흐름으로 해석한다.
+
+## 목적
+
+계획 가능한 Milestone 하나를 검증된 `feature/<milestone-slug>` workspace로 전환한다. 의미 정합성은 기존 roadmap 스킬이, branch/worktree/current와 lifecycle event는 번들 스크립트가 소유한다.
+
+## 입력
+
+- `target-milestone`: 활성 Milestone 이름, id, slug 또는 경로 (필수)
+- `workspace`: feature worktree로 사용할 절대 경로 (필수)
+- `planner-agent`: `codex`, `claude`, `gemini`, `pi` 중 하나 (생략 시 `codex`)
+- `review-agent`: 생략하면 `planner-agent`와 같다. (선택)
+- `planner-model`, `review-model`: provider별 model override. Codex 기본 사용 시 `planner-model` 생략 시 `gpt-5.6-sol`, 다른 provider는 해당 CLI 기본 모델을 사용한다. (선택)
+- `reasoning-effort`: 지원하는 provider의 reasoning/thinking override. 생략 시 `xhigh` (선택)
+- `pi-provider`: Pi provider override (선택)
+- `target-epic`: 준비 직후 실행할 정확한 Epic id 또는 이름. 생략하면 첫 미완료 Epic을 선택한다. (선택)
+
+트리거의 첫 번째 위치 표현(`<workspace>`)은 workspace 절대 경로로, 두 번째 표현(`<milestone>`)은 대상 Milestone으로 각각 확정한다. workspace가 생략되거나 둘 이상의 경로로 해석되면 임의로 `current`나 기본 경로를 사용하지 않고 사용자에게 확인한다.
+
+## 사전 조건
+
+- 대상 checkout은 Git Flow develop branch이고 tracked/untracked 변경이 없어야 한다.
+- 대상 Milestone은 정확히 `[계획]`, `구현 잠금: 해제`, `결정 필요: 없음`이어야 한다.
+- `sync-milestone-workstate mode=consistency-check`가 `ready`여야 한다.
+- remote와 `gitflow.branch.develop`, `gitflow.prefix.feature`를 확인할 수 있어야 한다.
+- 선택 agent의 비대화식 one-shot capability probe가 branch 생성 전에 성공해야 한다.
+
+## 절차
+
+1. **대상과 정합성을 고정한다**
+   - `agent-roadmap/current.md`를 선택 근거로만 사용하고 target이 명시됐으면 target을 우선한다.
+   - `sync-milestone-workstate mode=consistency-check`를 먼저 실행한다.
+   - 기계적으로 고칠 수 있는 drift는 보고된 owner 스킬로 갱신하고 commit/push한 뒤 consistency check를 다시 실행한다.
+   - 사용자만 결정할 수 있는 drift는 `roadmap-sdd mode=review-ready`의 `agent-roadmap/sdd/<phase-slug>/<milestone-slug>/USER_REVIEW.md`로 남기고 commit/push한 뒤 `USER_REVIEW`로 멈춘다. pre-plan 문제에 `agent-task/**/USER_REVIEW.md`를 만들지 않는다.
+
+2. **workspace를 준비한다**
+   - 아래 스크립트를 foreground로 한 번 실행한다. 실행 중 caller LLM이 timer polling, `ps`, state 파일 검사 또는 중복 실행을 하지 않는다.
+
+```bash
+python3 agent-ops/skills/common/prepare-milestone-workspace/scripts/prepare_workspace.py \
+  --repo "$REPO" \
+  --milestone "$MILESTONE" \
+  --workspace "$WORKSPACE"
+```
+
+   - 기본값은 `codex / gpt-5.6-sol / xhigh`다. 다른 provider를 지정하면 모델을 별도로 주지 않는 한 해당 provider의 CLI 기본 모델을 사용한다.
+   - 다른 agent, model, reasoning, Pi provider override가 있으면 해당 인자를 전달한다.
+   - 스크립트는 develop HEAD와 remote develop의 일치, agent probe, branch 충돌, worktree 소유권을 mutation 전에 검사한다.
+   - branch는 Milestone id가 아니라 파일 basename을 사용한 `feature/<milestone-slug>`다.
+   - 기존 branch/worktree는 정확히 같은 branch·경로이고 clean할 때만 재개한다.
+   - remote branch 생성 뒤 후속 단계가 실패해도 branch/worktree를 자동 삭제하지 않는다.
+
+3. **Epic 사이클로 handoff한다**
+   - `WORKSPACE_READY` event에서만 대상 workspace로 이동한다.
+   - target Epic을 명시했으면 그대로, 없으면 Milestone 문서 순서상 첫 미완료 Epic 하나를 선택한다.
+   - 같은 caller turn에서 `prepare-epic-work-items`를 정확히 한 번 시작하고 그 foreground 실행을 기다린다.
+   - `USER_REVIEW`, 복구 불가능한 `FAILED`, `EPIC_WORK_ITEMS_READY`, `EPIC_COMPLETED` 외 routine event에는 caller가 개입하지 않는다.
+
+## lifecycle event
+
+- `PREFLIGHT_READY`: mutation 전 검증 완료
+- `FEATURE_BRANCH_PUSHED`: remote feature branch 준비 완료
+- `WORKSPACE_READY`: worktree와 local current 준비 완료
+- `USER_REVIEW`: 사전 정합화 또는 연결된 Epic 준비에서 사용자 결정이 필요해 정상 정지
+- `FAILED`: 안전하게 계속할 수 없음
+
+## 실행 결과 검증
+
+- [ ] branch가 `gitflow.prefix.feature + milestone file slug`인가
+- [ ] branch가 remote에 존재하고 upstream이 연결됐는가
+- [ ] 요청 workspace가 정확히 해당 branch의 clean worktree인가
+- [ ] workspace-local `agent-roadmap/current.md`가 대상 Milestone만 가리키는가
+- [ ] state와 locator가 repository 문서가 아니라 git common dir 아래에 있는가
+- 검증 실패 시: branch/worktree를 삭제하지 않고 `FAILED` event와 복구 조건을 보고한다.
+
+## 출력 형식
+
+```text
+Milestone workspace preparation
+- milestone: <path>
+- branch: <feature/milestone-slug>
+- workspace: <absolute path>
+- event: <WORKSPACE_READY | USER_REVIEW | FAILED>
+- next: <prepare-epic-work-items target | user action | recovery condition>
+```
+
+## 금지 사항
+
+- `[계획]`이 아니거나 잠긴 Milestone의 branch를 만들지 않는다.
+- consistency check의 `refresh-required`를 `ready`로 간주하지 않는다.
+- 사용자 소유 변경이 있는 develop checkout이나 기존 workspace를 덮어쓰지 않는다.
+- `git push --force`, destructive rollback, branch/worktree 자동 삭제를 하지 않는다.
+- caller LLM이 주기적으로 process나 상태 파일을 모니터링하지 않는다.

agent-ops/skills/common/prepare-milestone-workspace/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "Prepare Milestone Workspace"
+  short_description: "Prepare a Git Flow workspace for one Milestone"
+  default_prompt: "Use $prepare-milestone-workspace to prepare the named Milestone in the explicitly provided workspace and start its first Epic."

agent-ops/skills/common/prepare-milestone-workspace/scripts/prepare_workspace.py

@@ -0,0 +1,451 @@
+#!/usr/bin/env python3
+"""Prepare one Git Flow feature worktree for an active Milestone."""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+from typing import Any, Iterable
+
+
+VALID_AGENTS = {"codex", "claude", "gemini", "pi"}
+AGENT_COMMAND = {"codex": "codex", "claude": "claude", "gemini": "agy", "pi": "pi"}
+DEFAULT_PLANNER_AGENT = "codex"
+DEFAULT_PLANNER_MODEL = "gpt-5.6-sol"
+DEFAULT_REASONING_EFFORT = "xhigh"
+MILESTONE_PATTERN = re.compile(
+    r"^agent-roadmap/phase/(?P<phase>[a-z0-9-]+)/milestones/(?P<slug>[a-z0-9-]+)\.md$"
+)
+
+
+class PreparationError(RuntimeError):
+    """A fail-closed preparation error."""
+
+
+def emit(event: str, **payload: Any) -> None:
+    print(json.dumps({"event": event, **payload}, ensure_ascii=False, sort_keys=True), flush=True)
+
+
+def run(
+    command: list[str],
+    *,
+    cwd: Path,
+    check: bool = True,
+    capture: bool = True,
+) -> subprocess.CompletedProcess[str]:
+    result = subprocess.run(
+        command,
+        cwd=cwd,
+        text=True,
+        stdout=subprocess.PIPE if capture else None,
+        stderr=subprocess.PIPE if capture else None,
+        check=False,
+    )
+    if check and result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        raise PreparationError(f"command failed ({result.returncode}): {' '.join(command)}: {detail}")
+    return result
+
+
+def git(repo: Path, *arguments: str, check: bool = True) -> str:
+    result = run(["git", *arguments], cwd=repo, check=check)
+    return (result.stdout or "").strip()
+
+
+def resolve_repo(raw: str) -> Path:
+    repo = Path(raw).expanduser().resolve()
+    if not repo.is_dir():
+        raise PreparationError(f"repository directory not found: {repo}")
+    top = Path(git(repo, "rev-parse", "--show-toplevel")).resolve()
+    if top != repo:
+        raise PreparationError(f"--repo must be the repository root: expected={top} actual={repo}")
+    return repo
+
+
+def resolve_milestone(repo: Path, raw: str) -> tuple[Path, re.Match[str]]:
+    candidate = Path(raw).expanduser()
+    path = (repo / candidate).resolve() if not candidate.is_absolute() else candidate.resolve()
+    try:
+        relative = path.relative_to(repo).as_posix()
+    except ValueError as exc:
+        raise PreparationError(f"milestone is outside repository: {path}") from exc
+    match = MILESTONE_PATTERN.fullmatch(relative)
+    if match is None or not path.is_file():
+        raise PreparationError(f"active milestone path required: {relative}")
+    return path, match
+
+
+def section(text: str, heading: str) -> str:
+    pattern = re.compile(
+        rf"^## {re.escape(heading)}\s*$\n(?P<body>.*?)(?=^##\s|\Z)",
+        re.MULTILINE | re.DOTALL,
+    )
+    match = pattern.search(text)
+    return match.group("body").strip() if match else ""
+
+
+def first_heading(text: str, prefix: str) -> str:
+    match = re.search(rf"^{re.escape(prefix)}\s*(.+?)\s*$", text, re.MULTILINE)
+    if not match:
+        raise PreparationError(f"missing heading: {prefix}")
+    return match.group(1).strip()
+
+
+def milestone_contract(path: Path) -> dict[str, str]:
+    text = path.read_text(encoding="utf-8")
+    status_body = section(text, "상태")
+    status_match = re.search(r"^\[(.+?)\]\s*$", status_body, re.MULTILINE)
+    status = status_match.group(1).strip() if status_match else ""
+    lock = section(text, "구현 잠금")
+    lock_state = re.search(r"^- 상태:\s*(.+?)\s*$", lock, re.MULTILINE)
+    decision = re.search(r"^- 결정 필요:\s*(.+?)\s*$", lock, re.MULTILINE)
+    if status != "계획":
+        raise PreparationError(f"milestone must be [계획]: actual={status or 'missing'}")
+    if not lock_state or lock_state.group(1).strip() != "해제":
+        raise PreparationError("milestone implementation lock is not 해제")
+    if not decision or decision.group(1).strip() != "없음":
+        raise PreparationError("milestone has unresolved 결정 필요")
+    return {
+        "title": first_heading(text, "# Milestone:"),
+        "status": status,
+    }
+
+
+def phase_contract(repo: Path, phase_slug: str) -> dict[str, str]:
+    path = repo / "agent-roadmap" / "phase" / phase_slug / "PHASE.md"
+    if not path.is_file():
+        raise PreparationError(f"phase document not found: {path}")
+    text = path.read_text(encoding="utf-8")
+    status_body = section(text, "상태")
+    status_match = re.search(r"^\[(.+?)\]\s*$", status_body, re.MULTILINE)
+    return {
+        "title": first_heading(text, "# Phase:"),
+        "status": status_match.group(1).strip() if status_match else "계획",
+    }
+
+
+def git_common_dir(repo: Path) -> Path:
+    raw = Path(git(repo, "rev-parse", "--git-common-dir"))
+    return (repo / raw).resolve() if not raw.is_absolute() else raw.resolve()
+
+
+def atomic_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + f".tmp.{os.getpid()}")
+    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    os.replace(temporary, path)
+
+
+def ref_exists(repo: Path, ref: str) -> bool:
+    return run(["git", "show-ref", "--verify", "--quiet", ref], cwd=repo, check=False).returncode == 0
+
+
+def ensure_clean(repo: Path, label: str) -> None:
+    status = git(repo, "status", "--porcelain=v1", "--untracked-files=all")
+    if status:
+        raise PreparationError(f"{label} has repository changes; clean checkout required")
+
+
+def worktrees(repo: Path) -> list[dict[str, str]]:
+    records: list[dict[str, str]] = []
+    current: dict[str, str] = {}
+    for line in git(repo, "worktree", "list", "--porcelain").splitlines() + [""]:
+        if not line:
+            if current:
+                records.append(current)
+                current = {}
+            continue
+        key, _, value = line.partition(" ")
+        current[key] = value
+    return records
+
+
+def probe_agents(
+    repo: Path,
+    planner_agent: str,
+    reviewer_agent: str,
+    planner_model: str | None,
+    reviewer_model: str | None,
+    reasoning_effort: str | None,
+    pi_provider: str | None,
+) -> None:
+    runner = (
+        Path(__file__).resolve().parents[2]
+        / "prepare-epic-work-items"
+        / "scripts"
+        / "run_agent_once.py"
+    )
+    if not runner.is_file():
+        raise PreparationError(f"agent runner not found: {runner}")
+    seen: set[tuple[str, str | None]] = set()
+    for agent, model in (
+        (planner_agent, planner_model),
+        (reviewer_agent, reviewer_model),
+    ):
+        identity = (agent, model)
+        if identity in seen:
+            continue
+        seen.add(identity)
+        command = [
+            sys.executable,
+            str(runner),
+            "--agent",
+            agent,
+            "--workspace",
+            str(repo),
+            "--probe",
+        ]
+        if model:
+            command.extend(["--model", model])
+        if reasoning_effort:
+            command.extend(["--reasoning-effort", reasoning_effort])
+        if agent == "pi" and pi_provider:
+            command.extend(["--pi-provider", pi_provider])
+        result = run(command, cwd=repo, check=False, capture=False)
+        if result.returncode != 0:
+            raise PreparationError(f"agent capability probe failed: agent={agent} model={model or 'default'}")
+
+
+def render_current(
+    *,
+    phase_slug: str,
+    phase: dict[str, str],
+    milestone_slug: str,
+    milestone: dict[str, str],
+) -> str:
+    return f"""# 현재 로드맵 컨텍스트
+
+## 고정 참조
+
+- Phase를 가로지르는 다음 작업 후보는 [전역 마일스톤 실행 순서](priority-queue.md)를 먼저 확인한다.
+- Phase는 도메인/책임 영역이며 순차 실행 게이트가 아니다.
+
+## 활성 Phase
+
+- [{phase['status']}] {phase['title']}
+  - 경로: [PHASE.md](phase/{phase_slug}/PHASE.md)
+
+## 활성 Milestone
+
+- [{milestone['status']}] {milestone['title']}
+  - Phase: [PHASE.md](phase/{phase_slug}/PHASE.md)
+  - 경로: [{milestone['title']}](phase/{phase_slug}/milestones/{milestone_slug}.md)
+
+## 선택 규칙
+
+- 이 문서는 현재 feature workspace의 활성 Phase와 Milestone 후보를 가리킨다.
+- 실제 진행·완료 상태는 Milestone 문서와 active task evidence로 판정한다.
+- `[완료]` 또는 `[폐기]` 항목은 활성 항목에 남기지 않는다.
+"""
+
+
+def parser() -> argparse.ArgumentParser:
+    value = argparse.ArgumentParser(description=__doc__)
+    value.add_argument("--repo", required=True)
+    value.add_argument("--milestone", required=True)
+    value.add_argument("--workspace", required=True)
+    value.add_argument(
+        "--planner-agent",
+        choices=sorted(VALID_AGENTS),
+        default=DEFAULT_PLANNER_AGENT,
+    )
+    value.add_argument("--review-agent", choices=sorted(VALID_AGENTS))
+    value.add_argument("--planner-model")
+    value.add_argument("--review-model")
+    value.add_argument("--reasoning-effort", default=DEFAULT_REASONING_EFFORT)
+    value.add_argument("--pi-provider")
+    value.add_argument("--remote", default="origin")
+    value.add_argument(
+        "--skip-agent-probe",
+        action="store_true",
+        help="tests only; requires --dry-run or AGENT_OPS_TESTING=1",
+    )
+    value.add_argument("--dry-run", action="store_true")
+    return value
+
+
+def apply_defaults(args: argparse.Namespace) -> argparse.Namespace:
+    if args.planner_model is None and args.planner_agent == DEFAULT_PLANNER_AGENT:
+        args.planner_model = DEFAULT_PLANNER_MODEL
+    if args.reasoning_effort is None:
+        args.reasoning_effort = DEFAULT_REASONING_EFFORT
+    return args
+
+
+def prepare(args: argparse.Namespace) -> int:
+    apply_defaults(args)
+    if args.skip_agent_probe and not (
+        args.dry_run or os.environ.get("AGENT_OPS_TESTING") == "1"
+    ):
+        raise PreparationError(
+            "--skip-agent-probe is test-only; use --dry-run or AGENT_OPS_TESTING=1"
+        )
+    repo = resolve_repo(args.repo)
+    milestone_path, milestone_match = resolve_milestone(repo, args.milestone)
+    milestone = milestone_contract(milestone_path)
+    phase_slug = milestone_match.group("phase")
+    milestone_slug = milestone_match.group("slug")
+    phase = phase_contract(repo, phase_slug)
+    workspace = Path(args.workspace).expanduser().resolve()
+    if workspace == repo:
+        raise PreparationError("feature workspace must differ from the develop checkout")
+    try:
+        workspace.relative_to(repo)
+    except ValueError:
+        pass
+    else:
+        raise PreparationError("feature workspace must not be nested inside the develop checkout")
+    reviewer_agent = args.review_agent or args.planner_agent
+    reviewer_model = args.review_model or (
+        args.planner_model if reviewer_agent == args.planner_agent else None
+    )
+    for agent in {args.planner_agent, reviewer_agent}:
+        command = AGENT_COMMAND[agent]
+        if shutil.which(command) is None and not args.skip_agent_probe:
+            raise PreparationError(f"agent command not found: agent={agent} command={command}")
+
+    common = git_common_dir(repo)
+    state_root = common / "milestone-work-preparation" / milestone_slug
+    state_path = state_root / "workspace-state.json"
+    state_root.mkdir(parents=True, exist_ok=True)
+    lock_path = state_root / "workspace.lock"
+    with lock_path.open("a+", encoding="utf-8") as lock:
+        try:
+            fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+            raise PreparationError(f"workspace preparation already running: {lock_path}") from exc
+
+        develop = git(repo, "config", "--get", "gitflow.branch.develop")
+        feature_prefix = git(repo, "config", "--get", "gitflow.prefix.feature")
+        if not develop or not feature_prefix:
+            raise PreparationError("gitflow.branch.develop and gitflow.prefix.feature are required")
+        current_branch = git(repo, "branch", "--show-current")
+        if current_branch != develop:
+            raise PreparationError(f"prepare from Git Flow develop branch: expected={develop} actual={current_branch}")
+        ensure_clean(repo, "develop checkout")
+        if not ref_exists(repo, f"refs/remotes/{args.remote}/{develop}") and args.dry_run:
+            raise PreparationError(f"remote develop ref missing: {args.remote}/{develop}")
+        if not args.dry_run:
+            git(repo, "fetch", args.remote, develop)
+        remote_develop = f"refs/remotes/{args.remote}/{develop}"
+        if git(repo, "rev-parse", "HEAD") != git(repo, "rev-parse", remote_develop):
+            raise PreparationError(f"develop checkout is not exactly synchronized with {args.remote}/{develop}")
+        if run(
+            ["git", "check-ignore", "--quiet", "agent-roadmap/current.md"],
+            cwd=repo,
+            check=False,
+        ).returncode != 0:
+            raise PreparationError(
+                "agent-roadmap/current.md must be ignored for workspace-local Milestone selection"
+            )
+
+        if not args.skip_agent_probe and not args.dry_run:
+            probe_agents(
+                repo,
+                args.planner_agent,
+                reviewer_agent,
+                args.planner_model,
+                reviewer_model,
+                args.reasoning_effort,
+                args.pi_provider,
+            )
+            ensure_clean(repo, "develop checkout after agent probe")
+
+        branch = f"{feature_prefix}{milestone_slug}"
+        local_ref = f"refs/heads/{branch}"
+        remote_ref = f"refs/remotes/{args.remote}/{branch}"
+        branch_worktrees = [
+            item
+            for item in worktrees(repo)
+            if item.get("branch") == local_ref
+        ]
+        if branch_worktrees and Path(branch_worktrees[0]["worktree"]).resolve() != workspace:
+            raise PreparationError(
+                f"feature branch already belongs to another worktree: {branch_worktrees[0]['worktree']}"
+            )
+
+        emit(
+            "PREFLIGHT_READY",
+            branch=branch,
+            milestone=str(milestone_path),
+            workspace=str(workspace),
+        )
+        if args.dry_run:
+            return 0
+
+        local_exists = ref_exists(repo, local_ref)
+        remote_exists = ref_exists(repo, remote_ref)
+        if not local_exists:
+            if remote_exists:
+                git(repo, "branch", "--track", branch, f"{args.remote}/{branch}")
+            else:
+                git(repo, "branch", branch, remote_develop)
+        if run(
+            ["git", "merge-base", "--is-ancestor", remote_develop, local_ref],
+            cwd=repo,
+            check=False,
+        ).returncode != 0:
+            raise PreparationError(f"feature branch does not contain current {args.remote}/{develop}")
+        if remote_exists and git(repo, "rev-parse", local_ref) != git(repo, "rev-parse", remote_ref):
+            raise PreparationError("existing local and remote feature branches differ")
+        if not remote_exists:
+            git(repo, "push", "--set-upstream", args.remote, branch)
+        else:
+            git(repo, "branch", "--set-upstream-to", f"{args.remote}/{branch}", branch)
+        emit("FEATURE_BRANCH_PUSHED", branch=branch, remote=args.remote)
+
+        if branch_worktrees:
+            ensure_clean(workspace, "feature workspace")
+        else:
+            if workspace.exists() and any(workspace.iterdir()):
+                raise PreparationError(f"workspace exists and is not empty: {workspace}")
+            workspace.parent.mkdir(parents=True, exist_ok=True)
+            git(repo, "worktree", "add", str(workspace), branch)
+            ensure_clean(workspace, "feature workspace")
+
+        current_path = workspace / "agent-roadmap" / "current.md"
+        current_path.parent.mkdir(parents=True, exist_ok=True)
+        current_path.write_text(
+            render_current(
+                phase_slug=phase_slug,
+                phase=phase,
+                milestone_slug=milestone_slug,
+                milestone=milestone,
+            ),
+            encoding="utf-8",
+        )
+        ensure_clean(workspace, "feature workspace")
+        target_milestone_path = workspace / milestone_path.relative_to(repo)
+        state = {
+            "status": "workspace-ready",
+            "milestone": str(target_milestone_path),
+            "milestone_slug": milestone_slug,
+            "branch": branch,
+            "workspace": str(workspace),
+            "planner_agent": args.planner_agent,
+            "review_agent": reviewer_agent,
+            "reasoning_effort": args.reasoning_effort,
+        }
+        atomic_json(state_path, state)
+        emit("WORKSPACE_READY", **state)
+        return 0
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    args = apply_defaults(parser().parse_args(argv))
+    try:
+        return prepare(args)
+    except (OSError, PreparationError) as exc:
+        emit("FAILED", reason=str(exc))
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

agent-ops/skills/common/prepare-milestone-workspace/tests/test_prepare_workspace.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import contextlib
+import importlib.util
+import io
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "prepare_workspace.py"
+SPEC = importlib.util.spec_from_file_location("prepare_workspace", SCRIPT)
+assert SPEC and SPEC.loader
+MODULE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(MODULE)
+
+
+def command(cwd: Path, *args: str) -> str:
+    result = subprocess.run(
+        list(args),
+        cwd=cwd,
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=True,
+    )
+    return result.stdout.strip()
+
+
+class PrepareWorkspaceTest(unittest.TestCase):
+    def test_workspace_defaults_to_codex_top_model_and_reasoning(self) -> None:
+        args = MODULE.parser().parse_args(
+            ["--repo", "/repo", "--milestone", "milestone.md", "--workspace", "/workspace"]
+        )
+        MODULE.apply_defaults(args)
+        self.assertEqual(args.planner_agent, "codex")
+        self.assertEqual(args.planner_model, "gpt-5.6-sol")
+        self.assertEqual(args.reasoning_effort, "xhigh")
+
+        other = MODULE.parser().parse_args(
+            [
+                "--repo",
+                "/repo",
+                "--milestone",
+                "milestone.md",
+                "--workspace",
+                "/workspace",
+                "--planner-agent",
+                "claude",
+            ]
+        )
+        MODULE.apply_defaults(other)
+        self.assertIsNone(other.planner_model)
+        self.assertEqual(other.reasoning_effort, "xhigh")
+
+    def test_agent_probe_bypass_is_test_only(self) -> None:
+        output = io.StringIO()
+        with contextlib.redirect_stdout(output), mock.patch.dict(
+            os.environ, {"AGENT_OPS_TESTING": ""}
+        ):
+            result = MODULE.main(
+                [
+                    "--repo",
+                    "/missing-repository",
+                    "--milestone",
+                    "missing.md",
+                    "--workspace",
+                    "/missing-workspace",
+                    "--planner-agent",
+                    "codex",
+                    "--skip-agent-probe",
+                ]
+            )
+        self.assertEqual(result, 2)
+        self.assertIn("test-only", output.getvalue())
+
+    def test_prepares_git_flow_branch_worktree_and_current(self) -> None:
+        with tempfile.TemporaryDirectory() as raw:
+            root = Path(raw)
+            remote = root / "remote.git"
+            repo = root / "source"
+            worktree = root / "feature-worktree"
+            command(root, "git", "init", "--bare", str(remote))
+            command(root, "git", "init", "-b", "dev", str(repo))
+            command(repo, "git", "config", "user.name", "Test Agent")
+            command(repo, "git", "config", "user.email", "agent@example.test")
+            command(repo, "git", "config", "gitflow.branch.develop", "dev")
+            command(repo, "git", "config", "gitflow.prefix.feature", "feature/")
+            command(repo, "git", "remote", "add", "origin", str(remote))
+            phase = repo / "agent-roadmap" / "phase" / "phase-one"
+            milestone = phase / "milestones" / "sample-milestone.md"
+            milestone.parent.mkdir(parents=True)
+            (repo / ".gitignore").write_text("agent-roadmap/current.md\n", encoding="utf-8")
+            (phase / "PHASE.md").write_text(
+                "# Phase: Sample Phase\n\n## 상태\n\n[계획]\n",
+                encoding="utf-8",
+            )
+            milestone.write_text(
+                "# Milestone: [sample-01] Sample Milestone\n\n"
+                "## 상태\n\n[계획]\n\n"
+                "## 구현 잠금\n\n- 상태: 해제\n- 결정 필요: 없음\n",
+                encoding="utf-8",
+            )
+            command(repo, "git", "add", ".gitignore", "agent-roadmap")
+            command(repo, "git", "commit", "-m", "init")
+            command(repo, "git", "push", "-u", "origin", "dev")
+
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output), mock.patch.dict(
+                os.environ, {"AGENT_OPS_TESTING": "1"}
+            ):
+                result = MODULE.main(
+                    [
+                        "--repo",
+                        str(repo),
+                        "--milestone",
+                        str(milestone.relative_to(repo)),
+                        "--workspace",
+                        str(worktree),
+                        "--planner-agent",
+                        "codex",
+                        "--skip-agent-probe",
+                    ]
+                )
+            self.assertEqual(result, 0, output.getvalue())
+            self.assertIn('"event": "WORKSPACE_READY"', output.getvalue())
+            self.assertEqual(command(worktree, "git", "branch", "--show-current"), "feature/sample-milestone")
+            self.assertEqual(
+                command(repo, "git", "rev-parse", "feature/sample-milestone"),
+                command(repo, "git", "rev-parse", "origin/feature/sample-milestone"),
+            )
+            current = (worktree / "agent-roadmap" / "current.md").read_text(encoding="utf-8")
+            self.assertIn("sample-milestone.md", current)
+            self.assertFalse(command(worktree, "git", "status", "--porcelain=v1"))
+
+
+if __name__ == "__main__":
+    unittest.main()

agent-ops/skills/common/router.md

@@ -47,6 +47,8 @@
 | 이 마일스톤은 X가 끝나야 가능해, A 전까지 B 잠가둬, 잠금 해제 조건은 X야, X 프로젝트 작업 뒤에 현재 마일스톤 진행, 의존성 설정해, 외부 의존 잠금 | `agent-ops/skills/common/update-roadmap/SKILL.md` |
 | roadmap dependency 확인, locks.yaml 판별, 외부 의존 잠금 확인, unlock-ready 판별, 잠금 해제 조건 충족 여부 확인, roadmap-dependency-checker.sh | `agent-ops/skills/common/check-roadmap-dependency/SKILL.md` |
 | 지금 작업이 뭐지?, 현재 작업 분석, 어디까지 했지?, 로드맵상 현 위치, 현재 마일스톤 위치, current 기준 breadcrumb | `agent-ops/skills/common/analyze-roadmap-position/SKILL.md` |
+| X에 Y 작업 준비해, X에 Y 작업준비해, X에 Y 마일스톤 작업 준비해 | `agent-ops/skills/common/prepare-milestone-workspace/SKILL.md` |
+| 현 마일스톤 Epic 작업 준비해, 마일스톤 Epic 작업 준비해, 이 Epic의 작은 작업은 바로 처리하고 큰 작업은 plan으로 작성해, Epic 작업을 작은 작업과 plan으로 나눠 | `agent-ops/skills/common/prepare-epic-work-items/SKILL.md` |
 | 계획 세워줘, 계획 작성해, 계획 만들어줘, 구현 계획, PLAN.md, plan, plan 작성해, plan 만들어줘 | `agent-ops/skills/common/plan/SKILL.md` |
 | 현재 plan들 세분화해, 현재 plan 세분화, 기존 plan 더 나눠, task 세분화해, plan 분리해 | `agent-ops/skills/common/refine-plans/SKILL.md` |
 | 최종 라우팅, task routing, cloud/local 재평가, lane/G 판단, G 등급 재평가, routed filename 결정 | `agent-ops/skills/common/finalize-task-routing/SKILL.md` |
@@ -59,6 +61,8 @@
 
 라우팅 우선순위:
 
+- `X에 Y 작업 준비해`처럼 workspace 위치와 대상 Milestone이 함께 명시되면 `prepare-milestone-workspace`를 선택하고, `WORKSPACE_READY` 뒤 정확히 한 Epic에 `prepare-epic-work-items`를 연결한다. workspace 위치가 없으면 이 흐름으로 라우팅하지 않고 확인을 요청한다.
+- 한 Epic 안에서 작은 작업 직접 처리와 큰 작업 plan 작성을 함께 요청하면 `prepare-epic-work-items`를 선택한다. 이미 존재하는 plan만 세분화하는 요청과 새로운 plan만 작성하는 요청에는 이 스킬을 선택하지 않는다.
 - 이미 생성된 미착수 pair의 분할만 요청하면 lane과 관계없이 `refine-plans`를 선택한다. 새 plan 작성이나 구현 범위 재분석이 포함되면 `plan`을 선택한다.
 - `refine-plans` 대상이 아닌 PLAN/CODE_REVIEW 작성 또는 재작성이 요청 범위에 포함되면 `plan`을 선택한다. `plan`이 최종 단계에서 `finalize-task-routing`을 필수 호출한다.
 - lane/G/canonical filename 판단만 요청되고 plan 문서 작성은 요청되지 않았을 때만 `finalize-task-routing`을 직접 선택한다.