"""whole-task stage 통합(머지) + worktree teardown 코어. Phase A: stage done commit 을 task 브랜치에 위상순 --no-ff 머지(이미 머지면 skip, 충돌이면 해당 머지만 abort 후 IntegrateError). Phase B: 머지/skip 된 stage 의 worktree 디렉터리와 registry 키를 정리(dirty 면 skip). stage 브랜치는 스택 보존을 위해 삭제하지 않는다. 설계: docs/superpowers/specs/2026-06-20-stage-auto-integrate-teardown-design.md """ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path from typing import Any from .worktree import _git class IntegrateError(Exception): """stage 머지 충돌 등으로 통합을 중단해야 함.""" @dataclass class IntegrateResult: merged: list[int] = field(default_factory=list) already_merged: list[int] = field(default_factory=list) torn_down: list[int] = field(default_factory=list) teardown_skipped: list[tuple[int, str]] = field(default_factory=list) warnings: list[str] = field(default_factory=list) def _merge_stage(task_wt: str, stage_n: int, branch: str, done_commit: str) -> str: """한 stage 의 done commit 을 task 브랜치에 머지. 'merged'|'already_merged'. 충돌 시 abort 후 IntegrateError. already_merged 판정과 실제 머지 대상을 모두 done_commit 으로 통일한다 — consumers 의 done.head_commit 은 reconcile 로 `-sN` 브랜치 tip 과 다른 commit 일 수 있어, ancestor 는 done_commit 으로 보고 머지는 브랜치 ref 로 하면 tip 이 아직 안 머지됐는데 skip 하거나 이미 머지된 변경을 중복 재도입할 수 있다. branch 는 정리(teardown) 식별용으로만 쓰고 머지 대상은 아니다.""" from .worktree import MergeError, is_ancestor, merge_branch head = _git(task_wt, "rev-parse", "HEAD").stdout.strip() if is_ancestor(task_wt, done_commit, head): return "already_merged" try: conflicts = merge_branch(task_wt, done_commit, no_ff=True) except MergeError as exc: raise IntegrateError( f"stage {stage_n} ({branch}) 머지 실패(내용 충돌 아님) — abort 함. " f"{exc}. 원인 해소 후 재시도." ) from exc if conflicts is not None: raise IntegrateError( f"stage {stage_n} ({branch}) 머지 충돌 — abort 함. " f"충돌 파일: {', '.join(conflicts) or '(unknown)'}. 수동 머지로 해소 후 재시도." ) return "merged" def integrate_stages( *, project_id: str, task_group: str, task_id: str, task_worktree_path: str, stage_map: list[dict[str, Any]], done_rows: list[dict[str, Any]], teardown: bool = True, ) -> IntegrateResult: """whole-task 통합: Phase A 머지 → Phase B teardown. teardown=False 면 Phase B(stage worktree 디렉터리·registry 키 정리)를 건너뛴다 — container 는 머지만 하고 stage 트리를 보존한다(스펙 §10). final-verification 호출은 기본값(True)으로 기존 행위를 유지한다.""" from .consumers import latest_done_by_stage res = IntegrateResult() if not task_worktree_path or not Path(task_worktree_path).exists(): return res # non-git / degraded — no-op done = latest_done_by_stage(done_rows) ordered = sorted((s["stage_number"] for s in stage_map)) for n in ordered: row = done.get(n) if not row or not row.get("head_commit"): continue # 미완 stage 는 merged 게이트가 별도로 잡음 branch = _stage_branch_name(project_id, task_group, task_id, n) if branch is None: res.warnings.append(f"stage {n}: registry 에 stage-key 없음 — 머지 skip") continue try: outcome = _merge_stage(task_worktree_path, n, branch, row["head_commit"]) except IntegrateError as exc: # 앞선 stage 머지는 이미 task 브랜치에 남는다(롤백 없음, 재시도 시 # already_merged 로 skip). 부분 통합 상태를 명시해 사용자가 인지하게 한다. if res.merged: raise IntegrateError( f"{exc} 이미 task 브랜치에 머지된 stage: {res.merged} " "— 재시도 시 자동 skip 됩니다.") from exc raise (res.merged if outcome == "merged" else res.already_merged).append(n) if teardown: _teardown_all(project_id, task_group, task_id, task_worktree_path, res) return res def _stage_branch_name(project_id, task_group, task_id, stage_n) -> str | None: from . import worktree_registry row = worktree_registry.get_stage_row(project_id, task_group, task_id, stage_n) if not row: return None return row.get("branch") or None def _teardown_all(project_id, task_group, task_id, task_wt: str, res: IntegrateResult) -> None: from . import worktree_registry for n in [*res.merged, *res.already_merged]: row = worktree_registry.get_stage_row(project_id, task_group, task_id, n) outcome, detail = _teardown_stage(project_id, task_group, task_id, task_wt, n, row) if outcome == "torn_down": res.torn_down.append(n) elif outcome == "skipped": res.teardown_skipped.append((n, detail)) if outcome == "warn": res.warnings.append(detail) def _remove_clean_worktree(task_wt: str, wt: str, stage_n: int) -> tuple[str, str] | None: """clean stage worktree 를 물리 제거. dirty/실패면 (outcome, detail), 성공이면 None.""" from .worktree import is_dirty_excluding_okstra, remove_worktree_force if is_dirty_excluding_okstra(wt): return ("skipped", "dirty — 미커밋 변경 보존") rm = remove_worktree_force(task_wt, wt) if rm.returncode != 0: return ("warn", f"stage {stage_n} worktree remove 실패: {rm.stderr.strip()}") return None def _unmerged_branch_tip_warning(task_wt: str, stage_n: int, branch: str) -> str: """Warn text when the stage branch tip is not an ancestor of the task worktree HEAD, or "" when it is (or cannot be resolved). Phase A merges `done.head_commit`, never the branch ref, so a stage branch can carry commits that whole-task verification never saw. `local-checkout --stage ` hands that branch to the user as a verified branch, so the divergence has to be reported.""" from .worktree import is_ancestor tip = _git(task_wt, "rev-parse", "--verify", "--quiet", branch) if tip.returncode != 0: return "" sha = tip.stdout.strip() head = _git(task_wt, "rev-parse", "HEAD").stdout.strip() if is_ancestor(task_wt, sha, head): return "" return (f"stage {stage_n} 브랜치 {branch} 의 tip {sha[:12]} 이 검증된 task " f"HEAD 에 포함되지 않습니다 — 검증에 포함되지 않은 커밋이 있습니다") def _teardown_stage(project_id, task_group, task_id, task_wt, stage_n, row): """Remove the clean stage worktree directory and mark the registry row released. The stage branch is deliberately kept: the per-stage stack branches must survive whole-task verification so that `okstra handoff local-checkout --stage ` still has a target. The branch's registry slot is not touched here — `consumers`' `_release_stage_reservation` already freed it at stage-done time. Warns when the surviving branch tip carries commits the verified merge never included. Teardown itself is already complete at that point, so the warning reports the divergence rather than undoing the cleanup. Returns ('torn_down'|'skipped'|'warn', detail).""" from . import worktree_registry wt = (row or {}).get("worktree_path") or "" branch = (row or {}).get("branch") or "" if wt and Path(wt).exists(): outcome = _remove_clean_worktree(task_wt, wt, stage_n) if outcome is not None: return outcome worktree_registry.release_status(project_id, task_group, task_id, stage_number=stage_n) if branch: warning = _unmerged_branch_tip_warning(task_wt, stage_n, branch) if warning: return ("warn", warning) return ("torn_down", "") def main(argv: list[str] | None = None) -> int: import argparse import json import sys from .consumers import read_consumers p = argparse.ArgumentParser(prog="okstra integrate-stages") p.add_argument("--project-id", required=True) p.add_argument("--task-group", required=True) p.add_argument("--task-id", required=True) p.add_argument("--plan-run-root", required=True) p.add_argument("--task-worktree-path", required=True) p.add_argument("--stage-map-json", default="[]", help="[{stage_number, depends_on}] JSON") a = p.parse_args(argv) # done 필터는 integrate_stages → latest_done_by_stage 가 단독 책임(SSOT). try: res = integrate_stages( project_id=a.project_id, task_group=a.task_group, task_id=a.task_id, task_worktree_path=a.task_worktree_path, stage_map=json.loads(a.stage_map_json), done_rows=read_consumers(Path(a.plan_run_root)), ) except IntegrateError as exc: print(f"integrate-stages: {exc}", file=sys.stderr) return 2 print(f"merged={res.merged} already={res.already_merged} " f"torn_down={res.torn_down} skipped={res.teardown_skipped}") for w in res.warnings: print(f"warning: {w}") return 0 if __name__ == "__main__": raise SystemExit(main())