"""프로젝트 backfill / discover.""" from __future__ import annotations import json import os import re as _re from pathlib import Path from typing import List from okstra_project.dirs import tasks_root from .ids import build_run_id from .paths import runs_dir_of from .invocation import save_invocation from .jsonl import append_jsonl, read_jsonl, rotate_recent_if_needed from .json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic from .project_meta import _project_meta_path from .reconcile import _now_iso, normalize_central_status from .run_index_row import build_run_index_row _STAGE_DIR_RE = _re.compile(r"^stage-\d+$") def _iter_manifest_dirs(runs: Path): """`runs/` 아래에서 실제 run-manifest 가 사는 manifests 디렉터리를 모두 yield. 대부분 task-type 은 `runs//manifests/`. implementation 은 stage 격리로 산출물이 `runs/implementation/stage-/manifests/` 에 사므로, 직접 하위 `manifests` 뿐 아니라 `stage-/manifests` 까지 내려가 스캔해야 backfill/reindex 가 implementation run 을 누락하지 않는다. """ for type_dir in sorted(p for p in runs.iterdir() if p.is_dir()): direct = type_dir / "manifests" if direct.is_dir(): yield direct for stage_dir in sorted(p for p in type_dir.iterdir() if p.is_dir()): if not _STAGE_DIR_RE.match(stage_dir.name): continue stage_manifests = stage_dir / "manifests" if stage_manifests.is_dir(): yield stage_manifests def discover_project_roots(home: Path) -> List[tuple]: """`~/.okstra/projects//meta.json` 을 권위 소스로 (project_id, project_root) 목록을 반환한다. 신규 모델에서는 okstra.sh 가 첫 실행 시 PROJECT_ROOT 를 해석해 `/.okstra/project.json` 에 자기 등록하고, record_start 가 그 PROJECT_ROOT 를 meta.json 에 mirror 한다. ctl 입장 에서는 한 번이라도 record_start 를 거친 프로젝트는 meta.json 에 등재 되어 있으므로, examples/projects 같은 외부 등록 디렉토리가 필요하지 않다. meta.json 의 projectRoot 가 더 이상 디스크에 존재하지 않으면 스킵한다 (프로젝트가 이동/삭제된 경우). 기록 자체는 유지된다 — 사용자가 향후 이동된 위치에서 다시 record_start 를 수행하면 meta.json 이 갱신된다. """ out: List[tuple] = [] projects_dir = home / "projects" if not projects_dir.is_dir(): return out for project_dir in sorted(p for p in projects_dir.iterdir() if p.is_dir()): meta_file = project_dir / "meta.json" if not meta_file.is_file(): continue try: meta = load_owned_object(meta_file, artifact="project metadata") except JsonBoundaryError: continue pid = str(meta.get("projectId") or "") root = str(meta.get("projectRoot") or "") if not pid or not root: continue if not Path(root).is_dir(): continue out.append((pid, root)) return out def _apply_backfill_meta(home: Path, project_id: str, project_root: Path, *, started_times: List[str], finished_times: List[str], run_count_inc: int, active_count_inc: int) -> None: """Backfill 전용 monotonic meta 갱신. upsert_project_meta 는 lastRunAt 을 무조건 인자값으로 덮어쓰므로, backfill 처럼 lexicographic 순서로 N 행을 적용하면 마지막 방문 행의 시각이 lastRunAt 이 되어 시계열이 깨진다. 여기서는 기존 meta 와 새 행들의 min/max 로 firstRunAt/lastRunAt 을 단조 갱신한다. """ target = _project_meta_path(home, project_id) target.parent.mkdir(parents=True, exist_ok=True) if target.is_file(): meta = load_owned_object(target, artifact="project metadata") else: meta = {"projectId": project_id, "projectRoot": str(project_root), "firstRunAt": "", "lastRunAt": "", "runCount": 0, "activeCount": 0} meta["projectRoot"] = str(project_root) first_pool = [t for t in started_times if t] if meta.get("firstRunAt"): first_pool.append(meta["firstRunAt"]) if first_pool: meta["firstRunAt"] = min(first_pool) last_pool = [t for t in (started_times + finished_times) if t] if meta.get("lastRunAt"): last_pool.append(meta["lastRunAt"]) if last_pool: meta["lastRunAt"] = max(last_pool) meta["runCount"] = int(meta.get("runCount", 0)) + run_count_inc meta["activeCount"] = max( 0, int(meta.get("activeCount", 0)) + active_count_inc) write_owned_object_atomic(target, meta, artifact="project metadata") def backfill_project(home: Path, project_id: str, project_root: Path) -> int: """타깃 프로젝트의 okstra tasks 디렉토리를 스캔해 누락된 run 을 인덱스에 채움. 실제 okstra 레이아웃: runs//manifests/run-manifest--.json 이미 존재하는 runId 는 스킵. 새로 추가된 run 수를 반환. """ base = tasks_root(project_root) if not base.is_dir(): return 0 existing_index = home / "projects" / project_id / "index.jsonl" existing_ids = {r["runId"] for r in read_jsonl(existing_index)} added = 0 # meta 는 행 단위가 아니라 배치 단위로 한 번만 갱신해 시계열을 보존한다. started_times: List[str] = [] finished_times: List[str] = [] active_inc = 0 manifest_re = _re.compile(r"^run-manifest-(?P.+)-(?P\d+)\.json$") for group_dir in sorted(p for p in base.iterdir() if p.is_dir()): for task_dir in sorted(p for p in group_dir.iterdir() if p.is_dir()): runs = runs_dir_of(task_dir) if not runs.is_dir(): continue for manifests in _iter_manifest_dirs(runs): for mf in sorted(manifests.iterdir()): m = manifest_re.match(mf.name) if not m: continue task_type = m.group("tt") seq = int(m.group("seq")) try: manifest = load_owned_object(mf, artifact="run manifest") except JsonBoundaryError: # 부분 기록 / 손상된 매니페스트를 빈 dict 로 강등해 row 를 # 만들면, 빈 status 가 'completed' 로 normalize 되어 # recent 에 박히고, 같은 runId 가 existing_ids 로 인덱싱 # 되어 후속 reindex 가 정상화된 매니페스트를 다시 읽을 # 기회를 잃는다. 이번 backfill 에서는 건너뛰어 다음 # 스캔이 정상 데이터를 ingest 할 수 있도록 한다. continue # manifest 의 logical task key 가 디렉터리 슬러그와 다를 수 # 있으므로 manifest 값을 우선 사용한다(디렉터리는 fallback). actual_group = manifest.get("taskGroup") or group_dir.name actual_task = manifest.get("taskId") or task_dir.name actual_type = manifest.get("taskType") or task_type run_id = build_run_id(project_id, actual_group, actual_task, actual_type, seq) if run_id in existing_ids: continue # 실제 okstra.sh 가 쓰는 manifest 키: # - runDirectoryPath (디렉터리) # - expectedReportRecordPath (보고서, 호환: reportPath) # - validation: { status, passed, ... } 중첩 dict run_dir_rel = (manifest.get("runDirectoryPath") or manifest.get("runDirPath", "")) from .final_report_paths import report_record_rel_from_legacy_pointer final_rel = report_record_rel_from_legacy_pointer( manifest.get("expectedReportRecordPath") or manifest.get("expectedReportPath") or manifest.get("reportPath") or manifest.get("finalReportRecordPath") or manifest.get("finalReportPath") or "" ) # status 파일 경로는 RUN_STATUS_SEQ 기반이라 manifest seq # 와 어긋날 수 있다. tail 이 정확한 파일을 추적하도록 # 매니페스트가 기록한 expectedStatusPath / # finalStatusPath 를 우선 보존한다. final_status_rel = ( manifest.get("expectedStatusPath") or manifest.get("finalStatusPath", "")) # 실제 okstra.sh 매니페스트는 worker 목록을 # `recommendedWorkers`, lead model 을 `teamContract.leadModel` # 또는 `resultContract.leadModel` 에 저장한다. 평탄 # `workers`/`leadModel` 키는 호환을 위한 fallback 으로만 # 사용한다(없으면 backfill 시 메타데이터 손실). raw_workers = (manifest.get("recommendedWorkers") or manifest.get("workers") or []) if not isinstance(raw_workers, list): raw_workers = [] team_contract = manifest.get("teamContract") or {} result_contract = manifest.get("resultContract") or {} raw_lead_model = ( (team_contract.get("leadModel") if isinstance(team_contract, dict) else "") or (result_contract.get("leadModel") if isinstance(result_contract, dict) else "") or manifest.get("leadModel", "")) raw_validation = manifest.get("validation") if isinstance(raw_validation, dict): validation_status = str(raw_validation.get("status") or "not-run") elif isinstance(raw_validation, str): validation_status = raw_validation else: validation_status = "not-run" # okstra.sh manifest 는 createdAt/updatedAt 만 쓴다(P2 회귀). # 별도 startedAt/finishedAt 가 없으면 createdAt/updatedAt 으로 보강한다. started_at = (manifest.get("startedAt") or manifest.get("createdAt", "")) finished_at = (manifest.get("finishedAt") or manifest.get("updatedAt", "") or started_at) normalized_status = normalize_central_status( manifest.get("status", ""), validation_status) # 비-터미널(running/in-progress) 매니페스트만 active 로 보내 # lazy reconcile 의 정상 종결 경로에 합류시킨다. # prepared 는 record_start 가 terminal 로 취급하는 상태이고 # (--render-only 실행 산출물), final-report 가 없어 12h 후 # reconcile 이 aborted 로 오인 마킹할 위험이 있으므로 backfill # 단계에서 recent 로 직접 보낸다. is_non_terminal = normalized_status in { "running", "in-progress"} row = build_run_index_row( project_id=project_id, project_root=str(project_root), task_group=actual_group, task_id=actual_task, task_type=actual_type, run_seq=seq, status=normalized_status, started_at=started_at, finished_at=None if is_non_terminal else finished_at, workers=raw_workers, lead_model=raw_lead_model, validation=validation_status, run_dir_rel=run_dir_rel, final_report_record_rel=final_rel, final_status_rel=final_status_rel) if is_non_terminal: append_jsonl(home / "active.jsonl", row) else: append_jsonl(home / "recent.jsonl", row) append_jsonl(existing_index, row) save_invocation(home, project_id, actual_group, actual_task, actual_type, seq, { "runId": run_id, "okstraVersion": "", "invokedAt": row["startedAt"], "cwd": str(project_root), "argv": [], "envOverrides": {}, "briefSha256": "", "backfilled": True, }) if row["startedAt"]: started_times.append(row["startedAt"]) if row["finishedAt"]: finished_times.append(row["finishedAt"]) if is_non_terminal: active_inc += 1 added += 1 existing_ids.add(run_id) if added: # backfill 의 terminal 행도 reconcile 의 promote 경로를 거치지 않으므로 # 직접 rotation 을 트리거한다(과거 run 이 많은 프로젝트의 첫 backfill # 에서 recent.jsonl 이 임계를 넘는 회귀 차단). rotate_recent_if_needed(home) _apply_backfill_meta(home, project_id, project_root, started_times=started_times, finished_times=finished_times, run_count_inc=added, active_count_inc=active_inc) return added def mark_backfilled(home: Path) -> None: state_file = home / "state.json" state = ( load_owned_object(state_file, artifact="okstra home state") if state_file.is_file() else {} ) state["backfilledAt"] = _now_iso() write_owned_object_atomic(state_file, state, artifact="okstra home state")