"""Stage readiness and verification target rules. This module owns the policy that decides which Stage Map stage can run, which commit it branches from, and what final-verification should inspect. It keeps the stage lifecycle rules behind one interface instead of leaking raw ``consumers.jsonl`` rows and git ancestry checks into prepare callers. """ from __future__ import annotations import heapq import subprocess import sys from dataclasses import dataclass, replace from pathlib import Path from typing import Any from .stage_integrate import IntegrateResult RUN_STEP_BUDGET = 8 class StageTargetError(Exception): """Stage target selection or verification precondition failed.""" class PrepareError(Exception): """surface to caller — task bundle prepare failed. Defined here(저수준 모듈)so both run.py(final-verification)와 container.py 가 whole-task 통합 헬퍼의 실패를 동일 타입으로 받을 수 있다. run.py 는 이 클래스를 그대로 재노출한다(단일 참조점).""" @dataclass class FinalVerificationTarget: scope: str # "whole-task" | "single-stage" base: str head: str worktree_path: str stages: list[int] reports: list[str] @dataclass(frozen=True) class FinalVerificationTargetRequest: """Semantic inputs needed to acquire one final-verification target.""" project_root: Path project_id: str task_group: str task_id: str work_category: str approved_plan_path: Path stage: int | None stage_map: tuple[dict[str, Any], ...] @dataclass(frozen=True) class FinalVerificationTargetAcquisition: """Resolved target facts returned to the prepare adapter.""" target: FinalVerificationTarget worktree_branch: str integration_result: IntegrateResult | None @dataclass(frozen=True) class StageLifecycle: stage: int depends_on: list[int] done: bool started: bool reserved: bool verified_accepted: bool pr_covered: bool head_commit: str report_path: str step_count: int deps_satisfied: bool blocked_by: list[int] @property def status(self) -> str: """Lifecycle state, in precedence order: done > active > ready > blocked. ``started`` and ``reserved`` collapse into ``active`` because every caller that distinguishes them already words the two the same way (see the numeric-request error in ``resolve_effective_stages``). """ if self.done: return "done" if self.started or self.reserved: return "active" return "ready" if self.deps_satisfied else "blocked" def handoff_eligibility_record(self) -> dict[str, Any]: reasons: list[str] = [] if self.pr_covered: reasons.append("already-in-pr") else: if not self.done: reasons.append("not-done") if not self.verified_accepted: reasons.append("not-verified-accepted") return { "stage": self.stage, "depends_on": list(self.depends_on), "eligible": not reasons, "reasons": reasons, "head_commit": self.head_commit, } @dataclass(frozen=True) class StageLifecycleSnapshot: stage_map: list[dict[str, Any]] rows: list[dict[str, Any]] done_rows: list[dict[str, Any]] done_by_stage: dict[int, dict[str, Any]] done_stages: set[int] started_stages: set[int] reserved_stages: set[int] verified_accepted_stages: set[int] pr_covered_stages: set[int] lifecycles: list[StageLifecycle] def ledger_records(self) -> list[dict[str, Any]]: """계획 저작 쪽에 넘길 stage 사실 기록. 상태 어휘는 `StageLifecycle.status` 를 그대로 쓴다 — 원장이 자기 어휘를 따로 가지면 같은 stage 가 소비처마다 다른 상태로 읽힌다. """ titles = { int(row["stage_number"]): str(row.get("title") or "") for row in self.stage_map } return [ { "stage": lifecycle.stage, "title": titles.get(lifecycle.stage, ""), "status": lifecycle.status, "dependsOn": list(lifecycle.depends_on), "headCommit": lifecycle.head_commit, } for lifecycle in self.lifecycles ] def lifecycle_for(self, stage_number: int) -> StageLifecycle: for lifecycle in self.lifecycles: if lifecycle.stage == stage_number: return lifecycle raise StageTargetError(f"stage {stage_number} not in Stage Map") def resolve_effective_stages( self, requested: str, budget: int = RUN_STEP_BUDGET, ) -> list[int]: """Return ordered stage numbers this run should execute. ``requested`` is ``"auto"`` or a decimal string. Auto selection returns every stage reporting ``ready`` within the step budget; numeric selection returns one forced stage. Readiness itself is not decided here — ``StageLifecycle.status`` is the single judgment point. """ if requested != "auto": return [self._forced_stage(requested)] ready = [lc for lc in self.lifecycles if lc.status == "ready"] if not ready: raise StageTargetError(self._no_ready_stage_reason()) batch: list[int] = [] total = 0 for lifecycle in ready: if batch and total + lifecycle.step_count > budget: break batch.append(lifecycle.stage) total += lifecycle.step_count return batch def _forced_stage(self, requested: str) -> int: try: n = int(requested) except ValueError as exc: raise StageTargetError( f"--stage must be 'auto' or an integer, got {requested!r}" ) from exc by_number = {lc.stage: lc for lc in self.lifecycles} lifecycle = by_number.get(n) if lifecycle is None: raise StageTargetError( f"--stage {n} not in Stage Map (have {list(by_number)})" ) if lifecycle.status == "done": raise StageTargetError( f"--stage {n} already completed (consumers.jsonl status:done exists)" ) if lifecycle.status == "active": # Wording is load-bearing: okstra-run's unattended chain matches it # to terminate normally instead of raising an exception gate. raise StageTargetError( f"--stage {n} already in progress or reserved by another run" ) return n def _no_ready_stage_reason(self) -> str: blocked = [lc for lc in self.lifecycles if lc.status == "blocked"] if not blocked: return ( "no stage is ready: every stage is done or occupied by " "another run" ) detail = "; ".join( f"stage {lc.stage} blocked by " + ", ".join(str(d) for d in lc.blocked_by) for lc in blocked ) return f"no stage is ready: {detail}" def resolve_implementation_stage(self, requested: str) -> int: return self.resolve_effective_stages(requested)[0] def concurrent_stage_numbers(self, *, selected_stage: int) -> list[int]: return sorted(self.reserved_stages - self.done_stages - {selected_stage}) def handoff_eligibility(self) -> list[dict[str, Any]]: return [ lifecycle.handoff_eligibility_record() for lifecycle in self.lifecycles ] def _stage_consumer_state_from_rows(rows: list[dict[str, Any]]) -> Any: from .consumers import stage_consumer_state_from_rows return stage_consumer_state_from_rows(rows) def stage_lifecycle_snapshot_from_state( stage_map: list[dict[str, Any]], consumer_state: Any, *, reserved_stages: set[int] | None = None, ) -> StageLifecycleSnapshot: reserved = set(reserved_stages or set()) lifecycles: list[StageLifecycle] = [] for stage in stage_map: stage_number = stage["stage_number"] done_row = consumer_state.done_by_stage.get(stage_number) or {} depends_on = list(stage.get("depends_on") or []) blocked_by = [ d for d in depends_on if d not in consumer_state.done_stages ] lifecycles.append(StageLifecycle( stage=stage_number, depends_on=depends_on, step_count=int(stage.get("step_count") or 0), deps_satisfied=not blocked_by, blocked_by=blocked_by, done=stage_number in consumer_state.done_stages, started=stage_number in consumer_state.started_stages, reserved=stage_number in reserved, verified_accepted=stage_number in ( consumer_state.verified_accepted_stages ), pr_covered=stage_number in consumer_state.pr_covered_stages, head_commit=str(done_row.get("head_commit") or ""), report_path=str(done_row.get("report_path") or ""), )) return StageLifecycleSnapshot( stage_map=stage_map, rows=list(consumer_state.rows), done_rows=list(consumer_state.done_rows), done_by_stage=dict(consumer_state.done_by_stage), done_stages=set(consumer_state.done_stages), started_stages=set(consumer_state.started_stages), reserved_stages=reserved, verified_accepted_stages=set(consumer_state.verified_accepted_stages), pr_covered_stages=set(consumer_state.pr_covered_stages), lifecycles=lifecycles, ) def stage_lifecycle_snapshot_from_rows( stage_map: list[dict[str, Any]], rows: list[dict[str, Any]], *, reserved_stages: set[int] | None = None, ) -> StageLifecycleSnapshot: return stage_lifecycle_snapshot_from_state( stage_map, _stage_consumer_state_from_rows(rows), reserved_stages=reserved_stages, ) def read_stage_lifecycle_snapshot( stage_map: list[dict[str, Any]], plan_run_root: Path, *, recover_from_carry: bool = False, reserved_stages: set[int] | None = None, ) -> StageLifecycleSnapshot: from .consumers import read_stage_consumer_state return stage_lifecycle_snapshot_from_state( stage_map, read_stage_consumer_state( plan_run_root, recover_from_carry=recover_from_carry, ), reserved_stages=reserved_stages, ) def resolve_effective_stages( stages: list[dict[str, Any]], done_stages: set[int], requested: str, budget: int = RUN_STEP_BUDGET, started_stages: set[int] | None = None, reserved_stages: set[int] | None = None, ) -> list[int]: """Adapter over ``StageLifecycleSnapshot.resolve_effective_stages``. Entry point for callers holding raw stage-map and consumer sets rather than a snapshot. It only assembles the snapshot; the readiness judgment lives on ``StageLifecycle.status``. """ rows: list[dict[str, Any]] = [ {"stage": n, "status": "done"} for n in sorted(done_stages) ] rows += [ {"stage": n, "status": "started"} for n in sorted(started_stages or set()) ] snapshot = stage_lifecycle_snapshot_from_rows( stages, rows, reserved_stages=reserved_stages, ) return snapshot.resolve_effective_stages(requested, budget=budget) def resolve_implementation_stage( stages: list[dict[str, Any]], done_stages: set[int], requested: str, *, started_stages: set[int] | None = None, reserved_stages: set[int] | None = None, ) -> int: """Return the single Stage Map stage selected for one implementation run.""" return resolve_effective_stages( stages, done_stages, requested, started_stages=started_stages, reserved_stages=reserved_stages, )[0] def order_stage_closure( stages, selected: set[int], done: set[int], ) -> list[int]: """selected ∪ (미완료 depends-on closure) 를 의존성 위상순서로 정렬해 반환. `want` 부분그래프에 Kahn 위상정렬을 적용한다(타이브레이커: stage 번호 오름차순). done 인 의존성은 want 밖이므로 이미 충족된 것으로 보고 간선에서 제외한다. 어떤 비순환 Stage Map(전방 참조 포함)에서도 정확하다 — 정렬이 stage 번호 순서와 일치한다는 가정에 의존하지 않는다.""" deps = {n: list(d) for n, d in stages} want = set(selected) frontier = list(selected) while frontier: n = frontier.pop() for d in deps.get(n, []): if d not in done and d not in want: want.add(d) frontier.append(d) indeg = {n: 0 for n in want} adj: dict[int, list[int]] = {n: [] for n in want} for n in want: for d in deps.get(n, []): if d in want: adj[d].append(n) indeg[n] += 1 ready = [n for n in want if indeg[n] == 0] heapq.heapify(ready) out: list[int] = [] while ready: n = heapq.heappop(ready) out.append(n) for m in adj[n]: indeg[m] -= 1 if indeg[m] == 0: heapq.heappush(ready, m) return out def downstream_stage_closure( stages: list[tuple[int, list[int]]], seed: set[int], ) -> set[int]: """Stages that (transitively) depend on any stage in ``seed``, plus ``seed``. ``stages`` mirrors ``order_stage_closure``: ``(stage_number, [depends_on_numbers])``. ``order_stage_closure`` walks dependencies (upstream); this walks dependents (downstream) — the reverse edge — so an answered clarification that changes a seed stage re-verifies everything built on top of it. """ dependents: dict[int, list[int]] = {} for num, deps in stages: for dep in deps: dependents.setdefault(dep, []).append(num) result = set(seed) queue = list(seed) while queue: cur = queue.pop() for child in dependents.get(cur, []): if child not in result: result.add(child) queue.append(child) return result def commit_is_ancestor(project_root: Path, ancestor: str, descendant: str) -> bool: """True iff ``ancestor`` is an ancestor of ``descendant`` in git history.""" from .worktree import is_ancestor return is_ancestor(project_root, ancestor, descendant) def check_multi_dep_merged( project_root: Path, plan_run_root: Path | None, latest: dict[int, dict[str, Any]], pred_commits: dict[int, str], candidate_base: str, stage_n: int, ) -> None: """Ensure all predecessor commits are merged into the candidate base.""" from .git_reconcile import content_merged, _record_reconciled for dep_stage, head in pred_commits.items(): if commit_is_ancestor(project_root, head, candidate_base): continue match = content_merged(project_root, head, candidate_base) if match.status in ("ancestor", "patch-equivalent"): if plan_run_root is not None: _record_reconciled( plan_run_root, impl_task_key=(latest.get(dep_stage) or {}).get("impl_task_key", ""), stage=dep_stage, new_commit=match.matched_commit, replaced=head, reason="auto-patch-id", ) continue raise StageTargetError( f"multi-dependency stage {stage_n}: predecessor stage {dep_stage} " f"({head[:8]}) is not merged into the task worktree " f"({candidate_base[:8]}). Merge stage branches " f"(e.g. the `-s{dep_stage}` branches) into the task worktree " "(or into main, then refresh the worktree) and retry." ) def resolve_stage_base_commit( stage: dict[str, Any], consumer_done_rows: list[dict[str, Any]], anchor_base_commit: str, candidate_base: str = "", project_root: Path | None = None, plan_run_root: Path | None = None, ) -> str: """Pick the git base commit a stage's isolated worktree branches from.""" from .consumers import latest_done_by_stage latest = latest_done_by_stage(consumer_done_rows) deps = stage.get("depends_on") or [] if len(deps) >= 2: stage_number = stage["stage_number"] pred_commits: dict[int, str] = {} for dep_stage in deps: head = (latest.get(dep_stage) or {}).get("head_commit") if not head: raise StageTargetError( f"predecessor stage {dep_stage} has no done row with head_commit " f"in consumers.jsonl; multi-dependency stage {stage_number} cannot start" ) pred_commits[dep_stage] = head if not candidate_base or project_root is None: raise StageTargetError( f"candidate base missing for multi-dependency stage {stage_number}; " "task-key worktree HEAD could not be resolved" ) check_multi_dep_merged( project_root, plan_run_root, latest, pred_commits, candidate_base, stage_number, ) return candidate_base if not deps: if not anchor_base_commit: raise StageTargetError( f"anchor base commit missing for independent stage " f"{stage['stage_number']}; first-stage prepare should have " "fixed it via worktree_registry.set_implementation_base" ) return anchor_base_commit pred = deps[0] head = (latest.get(pred) or {}).get("head_commit") or "" if head: _warn_if_branch_moved_past(project_root, pred, head, candidate_base) return head raise StageTargetError( f"predecessor stage {pred} has no done row with head_commit in " "consumers.jsonl; cannot derive base for stage " f"{stage['stage_number']}" ) def _warn_if_branch_moved_past( project_root: Path | None, predecessor: int, recorded_head: str, branch_tip: str, ) -> None: """Say so when the branch has advanced past the commit this stage branches from. A stage branches from its predecessor's recorded `head_commit`, not from the branch tip — that is what makes the stage's base reproducible. It also means a commit landing on the branch after the predecessor was settled is invisible to this stage, and the divergence only surfaces later as a rebase. Two such rebases were reported from one six-stage migration; both would have been caught by this line. The base rule is unchanged: this only names the gap. """ if project_root is None or not branch_tip or branch_tip == recorded_head: return try: contains = subprocess.run( ["git", "-C", str(project_root), "merge-base", "--is-ancestor", recorded_head, branch_tip], capture_output=True, ) except OSError: return if contains.returncode != 0: return print( f"okstra: stage branches from stage {predecessor}'s recorded head " f"{recorded_head[:12]}, but the branch tip is {branch_tip[:12]} — " "commits landed after that stage was settled and this stage will not " "carry them. Integrate them first if they belong to this work.", file=sys.stderr, ) def _resolve_whole_task_target( *, stage_map: list[dict[str, Any]], done_rows: list[dict[str, Any]], anchor_base: str, task_worktree_path: str, task_head: str, task_dirty: bool, merged: dict[int, bool], ) -> FinalVerificationTarget: """Resolve whole-task final-verification target, enforcing all gates.""" from .consumers import latest_done_by_stage done_by_stage = latest_done_by_stage(done_rows) for stage in stage_map: n = stage["stage_number"] if n not in done_by_stage: raise StageTargetError( f"final-verification(whole-task): stage {n} not done — " f"run implementation --stage {n} first" ) if not merged.get(n, False): sha = done_by_stage[n].get("head_commit", "") raise StageTargetError( f"final-verification(whole-task): stage {n} done commit " f"{sha} not merged into task worktree HEAD — merge stage " "branches then retry" ) if task_dirty: raise StageTargetError( "final-verification: worktree has uncommitted source changes " "(outside .okstra/) — commit or stash before verifying" ) stages = [s["stage_number"] for s in stage_map] reports = [done_by_stage[n].get("report_path", "") for n in stages] return FinalVerificationTarget( scope="whole-task", base=anchor_base, head=task_head, worktree_path=task_worktree_path, stages=stages, reports=reports, ) def _merged_stage_map( task_worktree_path: str, done_rows: list[dict[str, Any]], head: str, ) -> dict[int, bool]: """각 stage 의 done commit 이 실제 task HEAD 의 ancestor 인지 ground-truth 로 판정한다 — stage-key 가 teardown 됐어도 커밋이 HEAD 에 남아있으면 merged. run.py 의 기존 패턴을 그대로 옮긴 것.""" from .consumers import latest_done_by_stage from .worktree import is_ancestor return { s: is_ancestor(task_worktree_path, r.get("head_commit", ""), head) for s, r in latest_done_by_stage(done_rows).items() } def _resolve_and_integrate_whole_task_unlocked( *, 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]], anchor_base: str, teardown: bool, ) -> dict[str, Any]: """Integrate and resolve a whole-task target while the caller owns the lock.""" from .stage_integrate import IntegrateError, integrate_stages from .worktree import _git, is_dirty_excluding_okstra try: integration = integrate_stages( project_id=project_id, task_group=task_group, task_id=task_id, task_worktree_path=task_worktree_path, stage_map=stage_map, done_rows=done_rows, teardown=teardown, ) except IntegrateError as exc: raise PrepareError(str(exc)) from exc head = _git(task_worktree_path, "rev-parse", "HEAD").stdout.strip() merged = _merged_stage_map(task_worktree_path, done_rows, head) try: target = _resolve_whole_task_target( stage_map=stage_map, done_rows=done_rows, anchor_base=anchor_base, task_worktree_path=task_worktree_path, task_head=head, task_dirty=is_dirty_excluding_okstra(task_worktree_path), merged=merged, ) except StageTargetError as exc: raise PrepareError(str(exc)) from exc return { "head_commit": head, "merged_stage_map": merged, "target_ref": target.head, "target": target, "integrate_result": integration, } def resolve_and_integrate_whole_task( *, 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]], anchor_base: str, teardown: bool, ) -> dict[str, Any]: """Locked whole-task integration interface shared with the container.""" from okstra_project.dirs import okstra_home from okstra_project.state import slugify from .locks import worktree_provision_mutex with worktree_provision_mutex( okstra_home(), project_id, slugify(task_group), slugify(task_id), ): return _resolve_and_integrate_whole_task_unlocked( project_id=project_id, task_group=task_group, task_id=task_id, task_worktree_path=task_worktree_path, stage_map=stage_map, done_rows=done_rows, anchor_base=anchor_base, teardown=teardown, ) def _resolve_single_stage_target( *, requested_stage: int, done_rows: list[dict[str, Any]], stage_base: str, stage_worktree_path: str, stage_head: str, stage_dirty: bool, ) -> FinalVerificationTarget: """Resolve single-stage final-verification target, enforcing all gates.""" from .consumers import latest_done_by_stage n = requested_stage done_by_stage = latest_done_by_stage(done_rows) if n not in done_by_stage: raise StageTargetError( f"final-verification(single-stage): stage {n} not done — " f"run implementation --stage {n} first" ) if not stage_worktree_path: raise StageTargetError( f"final-verification(single-stage): stage worktree not found for " f"stage {n} (torn down?) — use whole-task mode (--stage auto)" ) if stage_dirty: raise StageTargetError( "final-verification: worktree has uncommitted source changes " "(outside .okstra/) — commit or stash before verifying" ) return FinalVerificationTarget( scope="single-stage", base=stage_base, head=stage_head, worktree_path=stage_worktree_path, stages=[n], reports=[done_by_stage[n].get("report_path", "")], ) def _read_final_verification_done_rows( request: FinalVerificationTargetRequest, registry_coordinates: tuple[str, str, str], ) -> list[dict[str, Any]]: from .consumers import backfill_done_from_carry, read_stage_consumer_state from .plan_run_root import plan_run_root_from_approved_plan from .stage_reconcile import auto_reconcile_best_effort plan_run_root = plan_run_root_from_approved_plan(request.approved_plan_path) backfill_done_from_carry(plan_run_root) project_id, task_group, task_id = registry_coordinates auto_reconcile_best_effort( replace( request, project_id=project_id, task_group=task_group, task_id=task_id, ), plan_run_root, ) return read_stage_consumer_state(plan_run_root).done_rows def _final_verification_registry_coordinates( request: FinalVerificationTargetRequest, ) -> tuple[str, str, str]: from okstra_project.state import slugify def segment(value: str) -> str: return slugify(value) or "_" return ( segment(request.project_id), segment(request.task_group), segment(request.task_id), ) def _acquire_single_stage_target( request: FinalVerificationTargetRequest, done_rows: list[dict[str, Any]], registry_coordinates: tuple[str, str, str], ) -> FinalVerificationTargetAcquisition: from . import worktree_registry from .worktree import _git, is_dirty_excluding_okstra stage = request.stage assert stage is not None row = worktree_registry.get_stage_row(*registry_coordinates, stage) worktree_path = (row or {}).get("worktree_path", "") head = "" if worktree_path and Path(worktree_path).is_dir(): head_result = _git(worktree_path, "rev-parse", "HEAD") if head_result.returncode == 0: head = head_result.stdout.strip() if not head: worktree_path = "" target = _resolve_single_stage_target( requested_stage=stage, done_rows=done_rows, stage_base=(row or {}).get("base_ref", ""), stage_worktree_path=worktree_path, stage_head=head, stage_dirty=( is_dirty_excluding_okstra(worktree_path) if worktree_path else False ), ) return FinalVerificationTargetAcquisition( target=target, worktree_branch=(row or {}).get("branch", ""), integration_result=None, ) def _acquire_whole_task_target( request: FinalVerificationTargetRequest, done_rows: list[dict[str, Any]], registry_coordinates: tuple[str, str, str], ) -> FinalVerificationTargetAcquisition: from . import worktree_registry project_id, task_group, task_id = registry_coordinates entry = worktree_registry.lookup(project_id, task_group, task_id) worktree_path = ( entry.worktree_path if entry is not None else str(request.project_root) ) whole = _resolve_and_integrate_whole_task_unlocked( project_id=project_id, task_group=task_group, task_id=task_id, task_worktree_path=worktree_path, stage_map=list(request.stage_map), done_rows=done_rows, anchor_base=( worktree_registry.get_implementation_base( project_id, task_group, task_id ) or "" ), # 정리는 판정 뒤로 미룬다(Phase 7 `teardown-stages`). 되돌릴 수 없는 정리를 # 판정 앞에 두면, 재작업이 가장 필요한 blocked 판정에서 stage 작업물이 이미 # 사라져 있다. 여기서는 통합만 하고 worktree/registry 키는 남긴다. teardown=False, ) return FinalVerificationTargetAcquisition( target=whole["target"], worktree_branch=entry.branch if entry is not None else "", integration_result=whole["integrate_result"], ) def integrate_and_teardown_whole_task( *, project_root: Path, task_group: str, task_id: str, ) -> dict[str, Any]: """판정이 끝난 whole-task 검증의 stage worktree 와 registry 키를 회수한다. 진입은 통합만 하고 정리를 남겨 두므로(`_resolve_whole_task_acquisition`), 정리는 판정 뒤인 Phase 7 에서 여기로 들어온다. 통합은 이미 끝나 있어 Phase A 는 전부 `already_merged` 로 지나가고 Phase B 만 실제 일을 한다. 두 번 불려도 결과는 같다 — 사라진 worktree 는 건너뛰고, 미커밋 변경이 남은 stage 트리는 보존한다. stage_map 은 done 행에서 만든다. 정리 대상은 완료된 stage 뿐이고, 계획에만 있고 완료되지 않은 stage 는 Phase A 가 어차피 건너뛰기 때문이다. """ from okstra_project.dirs import okstra_home, project_json_path from . import consumers, worktree_registry from .json_boundary import load_owned_object from .locks import worktree_provision_mutex from .paths import task_runs_dir from .stage_integrate import integrate_stages try: project_id = load_owned_object( project_json_path(project_root), artifact="project config" ).get("projectId", "") except (OSError, ValueError): project_id = "" if not project_id: return {"skipped": "project.json declares no projectId"} plan_run_root = task_runs_dir( project_root, task_group, task_id ) / "implementation-planning" done_rows = [ row for row in consumers.read_consumers(plan_run_root) if row.get("status") == "done" ] if not done_rows: return {"skipped": "no done stage rows to reclaim"} stage_map = [ {"stage_number": stage} for stage in sorted(consumers.latest_done_by_stage(done_rows)) ] entry = worktree_registry.lookup(project_id, task_group, task_id) if entry is None: return {"skipped": "task worktree is no longer registered"} with worktree_provision_mutex(okstra_home(), project_id, task_group, task_id): result = integrate_stages( project_id=project_id, task_group=task_group, task_id=task_id, task_worktree_path=entry.worktree_path, stage_map=stage_map, done_rows=done_rows, teardown=True, ) return { "tornDown": result.torn_down, "teardownSkipped": [ {"stage": stage, "reason": reason} for stage, reason in result.teardown_skipped ], "warnings": result.warnings, } def acquire_final_verification_target( request: FinalVerificationTargetRequest, ) -> FinalVerificationTargetAcquisition: """Acquire a stable final-verification target behind one task-key lock.""" from okstra_project.dirs import okstra_home from okstra_project.state import slugify from .locks import worktree_provision_mutex try: with worktree_provision_mutex( okstra_home(), request.project_id, slugify(request.task_group), slugify(request.task_id), ): registry_coordinates = _final_verification_registry_coordinates(request) done_rows = _read_final_verification_done_rows( request, registry_coordinates, ) if request.stage is not None: return _acquire_single_stage_target( request, done_rows, registry_coordinates, ) return _acquire_whole_task_target( request, done_rows, registry_coordinates, ) except PrepareError: raise except (OSError, RuntimeError, StageTargetError, ValueError) as exc: raise PrepareError(str(exc)) from exc