"""Artifact-derived implementation phase outcome reconciliation.""" from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path from typing import Any from .consumers import ( FAILED_CARRY_STATUSES, carry_head_commit, read_stage_consumer_state, ) from . import next_phase from .paths import RunRef, task_manifest_file from .stage_map import StageMapError, load_task_stage_map from .workflow import PHASE_SEQUENCE from .json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic @dataclass(frozen=True) class ImplementationOutcome: completed: bool stages: list[int] = field(default_factory=list) head_commit: str = "" report_path: str = "" reason: str = "" def derive_implementation_outcome(task_root: Path) -> ImplementationOutcome: task_root = Path(task_root) manifest = _load_json(task_manifest_file(task_root)) workflow = manifest.get("workflow") if isinstance(manifest.get("workflow"), dict) else {} if workflow.get("currentPhase") != "implementation": return ImplementationOutcome(completed=False, reason="current phase is not implementation") plan_run_root = RunRef.from_task_root(task_root, "implementation-planning").run_dir if not plan_run_root.is_dir(): return ImplementationOutcome(completed=False, reason="implementation-planning run root missing") try: stage_snapshot = load_task_stage_map(task_root, manifest) except StageMapError as exc: return ImplementationOutcome( completed=False, reason=f"{exc.code}: {exc.reason}" ) if stage_snapshot.state == "missing": return ImplementationOutcome(completed=False, reason="stage map missing") stage_map = stage_snapshot.stages consumers = read_stage_consumer_state(plan_run_root, recover_from_carry=True) required_stages = {stage["stage_number"] for stage in stage_map} if not required_stages or not required_stages.issubset(consumers.done_stages): return ImplementationOutcome(completed=False, reason="not all stages are done") carries = [_load_carry(task_root, stage) for stage in sorted(required_stages)] if any(carry is None for carry in carries): return ImplementationOutcome(completed=False, reason="carry sidecar missing") if any(not _carry_passed(carry) for carry in carries if carry is not None): return ImplementationOutcome(completed=False, reason="carry sidecar is not pass-grade") latest_stage = max(required_stages) latest_done = consumers.done_by_stage.get(latest_stage) or {} latest_carry = carries[-1] or {} return ImplementationOutcome( completed=True, stages=sorted(required_stages), head_commit=str(latest_done.get("head_commit") or "") or carry_head_commit(latest_carry), report_path=_latest_implementation_report(task_root), reason="all implementation stages have pass-grade carry and done rows", ) def reconcile_implementation_outcome(project_root: Path, task_root: Path) -> bool: project_root = Path(project_root) task_root = Path(task_root) outcome = derive_implementation_outcome(task_root) if not outcome.completed: return False manifest_path = task_manifest_file(task_root) manifest = _load_json(manifest_path) workflow = manifest.get("workflow") if not isinstance(workflow, dict): workflow = {} manifest["workflow"] = workflow changed = _promote_workflow(workflow, outcome) phase_outcome = manifest.get("phaseOutcome") if not isinstance(phase_outcome, dict): phase_outcome = {} manifest["phaseOutcome"] = phase_outcome contract = manifest.get("contractValidation") contract_status = contract.get("status", "") if isinstance(contract, dict) else "" implementation = { "state": "completed", "source": "implementation-carry", "stages": outcome.stages, "headCommit": outcome.head_commit, "reportPath": outcome.report_path, "contractValidationStatus": contract_status, "reason": outcome.reason, } if phase_outcome.get("implementation") != implementation: phase_outcome["implementation"] = implementation changed = True if changed: _write_json(manifest_path, manifest) return changed or _update_task_catalog(project_root, manifest) def _load_json(path: Path) -> dict[str, Any]: if not path.is_file(): return {} try: data = load_owned_object(path, artifact="implementation outcome") except JsonBoundaryError: return {} return data if isinstance(data, dict) else {} def _project_root_from_task_root(task_root: Path) -> Path: parts = task_root.resolve().parts if ".okstra" not in parts: return task_root.resolve() return Path(*parts[: parts.index(".okstra")]) def _load_carry(task_root: Path, stage: int) -> dict[str, Any] | None: data = _load_json(RunRef.from_task_root(task_root, "implementation").carry(stage)) return data or None def _carry_passed(carry: dict[str, Any]) -> bool: status = str(carry.get("status") or "").lower() if status in FAILED_CARRY_STATUSES: return False conformance = carry.get("conformance") if isinstance(conformance, dict): overall = str(conformance.get("overall") or "").upper() if overall and overall != "PASS": return False unverified = carry.get("unverified") return not (isinstance(unverified, list) and unverified) def _latest_implementation_report(task_root: Path) -> str: reports = sorted( RunRef.from_task_root(task_root, "implementation").run_dir.glob( "stage-*/reports/final-report-implementation-*.md" ) ) if not reports: return "" project_root = _project_root_from_task_root(task_root) try: return str(reports[-1].relative_to(project_root)) except ValueError: return str(reports[-1]) def _promote_workflow(workflow: dict[str, Any], outcome: ImplementationOutcome) -> bool: before = json.dumps(workflow, sort_keys=True, ensure_ascii=False) phase_states = workflow.get("phaseStates") if not isinstance(phase_states, dict): phase_states = {} workflow["phaseStates"] = phase_states for phase in PHASE_SEQUENCE: phase_states.setdefault(phase, "not-started") workflow["currentPhase"] = "implementation" workflow["currentPhaseState"] = "completed" phase_states["implementation"] = "completed" workflow["lastCompletedPhase"] = "implementation" # 재조정기의 사실은 "모든 stage 가 끝났다"는 상태다. 대상 phase 는 마지막 # stage 검증이 이미 정했으므로 건드리지 않는다 (ADR-0004). # # 대상 phase 가 비어 있으면 status 도 올리지 않는다. `ready` 는 "지금 이 # phase 를 시작할 수 있다"는 뜻인데 가리킬 phase 가 없으면 시작할 대상이 # 없다. 마지막 stage 리포트에 라우팅 대상이 없으면 이 자리에 빈 phase 의 # pending 포인터가 들어오므로, 그때는 그대로 둔다. pointer = next_phase.promote(workflow.get("nextRecommendedPhase")) if pointer["phase"]: workflow["nextRecommendedPhase"] = next_phase.make( phase=pointer["phase"], status=next_phase.STATUS_READY, rationale=pointer["rationale"], ) else: workflow["nextRecommendedPhase"] = pointer after = json.dumps(workflow, sort_keys=True, ensure_ascii=False) return before != after def _update_task_catalog(project_root: Path, manifest: dict[str, Any]) -> bool: catalog_path = project_root / ".okstra" / "discovery" / "task-catalog.json" catalog = _load_json(catalog_path) tasks = catalog.get("tasks") if not isinstance(tasks, list): return False task_key = manifest.get("taskKey") workflow = manifest.get("workflow") if isinstance(manifest.get("workflow"), dict) else {} outcome = manifest.get("phaseOutcome") if isinstance(manifest.get("phaseOutcome"), dict) else {} for entry in tasks: if not isinstance(entry, dict) or entry.get("taskKey") != task_key: continue before = json.dumps(entry, sort_keys=True, ensure_ascii=False) entry["currentPhase"] = workflow.get("currentPhase", "") entry["currentPhaseState"] = workflow.get("currentPhaseState", "") entry["lastCompletedPhase"] = workflow.get("lastCompletedPhase", "") entry["nextRecommendedPhase"] = next_phase.promote( workflow.get("nextRecommendedPhase") ) entry["latestReportRecordPath"] = manifest.get("latestReportRecordPath", "") entry["latestRunStatus"] = manifest.get("latestRunStatus", entry.get("latestRunStatus", "")) entry["currentStatus"] = manifest.get("currentStatus", entry.get("currentStatus", "")) entry["phaseOutcome"] = outcome if before == json.dumps(entry, sort_keys=True, ensure_ascii=False): return False _write_json(catalog_path, catalog) return True return False def _write_json(path: Path, payload: dict[str, Any]) -> None: write_owned_object_atomic(path, payload, artifact="implementation outcome")