"""active/projects 인덱스 행 치환·기록.""" from __future__ import annotations import os from pathlib import Path from .ids import build_run_id from .invocation import save_invocation from .jsonl import ( append_jsonl, read_jsonl, remove_jsonl_row, rewrite_jsonl, rotate_recent_if_needed, ) from .project_meta import upsert_project_meta from .json_boundary import load_owned_object, write_owned_object_atomic from .run_index_row import build_run_index_row, slim_run_row def _replace_or_append_row(path: Path, run_id: str, row: dict) -> None: """run_id 매칭 row 를 교체하고 없으면 append. 디스크에는 항상 slim 으로 기록한다 — 파생 4필드가 새지 않도록 모든 인덱스 쓰기가 거치는 단일 길목.""" slim = slim_run_row(row) rows = read_jsonl(path) for i, r in enumerate(rows): if r.get("runId") == run_id: rows[i] = slim path.parent.mkdir(parents=True, exist_ok=True) rewrite_jsonl(path, rows) return append_jsonl(path, slim) def record_start(home: Path, *, project_id: str, project_root: str, task_group: str, task_id: str, task_type: str, run_seq: int, when: str, workers: list, lead_model: str, run_dir_rel: str, final_report_record_rel: str, argv: list, cwd: str, env_overrides: dict, okstra_version: str = "", brief_sha256: str = "", initial_status: str = "running", final_status_rel: str = "", execution_manifest_path: str = "", role_execution_refs: list | None = None) -> str: """run 시작 시점의 인덱스/invocation/meta 를 1 트랜잭션으로 기록. 반환값: 생성된 runId. """ run_id = build_run_id(project_id, task_group, task_id, task_type, run_seq) # 종결 상태(prepared/aborted/etc) 로 시작하는 경우 finishedAt 도 같이 채워 # 인덱스 위치를 active 가 아닌 recent 로 라우팅한다. is_terminal = initial_status != "running" row = build_run_index_row( project_id=project_id, project_root=project_root, task_group=task_group, task_id=task_id, task_type=task_type, run_seq=run_seq, status=initial_status, started_at=when, finished_at=when if is_terminal else None, workers=workers, lead_model=lead_model, validation="not-run", run_dir_rel=run_dir_rel, final_report_record_rel=final_report_record_rel, final_status_rel=final_status_rel, execution_manifest_path=execution_manifest_path, role_execution_refs=role_execution_refs) # okstra-ctl 가 미리 'reserving' 예약 row 를 작성했을 수 있다(P1-2). 같은 # runId 의 예약 row 가 있으면 update; 없으면 append. existing_rows = read_jsonl(home / "active.jsonl") had_reservation = any(r.get("runId") == run_id for r in existing_rows) if is_terminal: # active 에 들어가지 않도록 — 이미 있던 reservation 이 있으면 제거하고 recent 로. if had_reservation: remove_jsonl_row(home / "active.jsonl", lambda r: r.get("runId") == run_id) # 동일 runId 가 이미 recent.jsonl 에 있으면(예: 매니페스트 직후 사용자가 # 처음 okstra-ctl 을 실행해 자동 backfill 이 같은 terminal run 을 미리 # ingest 한 경우) 중복 append 를 방지하기 위해 기존 row 를 제거하고 새로 # 쓴다(replace). 그렇지 않으면 동일 runId 가 두 줄로 남아 prefix 해석이 # 모호해지고 runCount 도 재증가한다. had_recent = bool(remove_jsonl_row( home / "recent.jsonl", lambda r: r.get("runId") == run_id)) append_jsonl(home / "recent.jsonl", row) # --render-only 나 launch-failure 같은 terminal 시작은 reconcile 의 # promote 경로를 거치지 않으므로 별도 rotation hook 이 필요하다. # 그렇지 않으면 그런 run 만 누적된 환경에서 recent.jsonl 이 임계를 # 넘어도 archive 로 이동되지 않는다. rotate_recent_if_needed(home) else: _replace_or_append_row(home / "active.jsonl", run_id, row) had_recent = False _replace_or_append_row( home / "projects" / project_id / "index.jsonl", run_id, row) save_invocation(home, project_id, task_group, task_id, task_type, run_seq, { "runId": run_id, "okstraVersion": okstra_version or os.environ.get("OKSTRA_SCRIPT_VERSION", ""), "invokedAt": when, "cwd": cwd, "argv": argv, "envOverrides": env_overrides, "briefSha256": brief_sha256, "backfilled": False, }) # 예약이 이미 있었다면 runCount/activeCount 는 reserve 단계에서 이미 증가시켰다. # 종결 상태로 시작했다면(terminal) activeCount 를 즉시 되돌린다. # 또한 backfill 이 같은 runId 를 recent 에 미리 ingest 했었다면(had_recent) # runCount 는 _apply_backfill_meta 에서 이미 가산되었고 activeCount 는 # 증가된 적이 없으므로, 여기서는 started/finished 모두 건너뛴다. if had_recent: upsert_project_meta(home, project_id, project_root=project_root, when=when, started=False, finished=False) else: upsert_project_meta(home, project_id, project_root=project_root, when=when, started=not had_reservation, finished=is_terminal) return run_id def reserve_run_in_active(home: Path, *, project_id: str, project_root: str, task_group: str, task_id: str, task_type: str, run_seq: int, when: str, run_dir_rel: str, final_report_record_rel: str, final_status_rel: str = "") -> None: """예약 row 를 active.jsonl 에 즉시 기록한다. 중앙 락 보호 필수. record_start 가 나중에 같은 (project, group, task, task_type, seq) 를 만나면 이 row 를 update 한다. """ row = build_run_index_row( project_id=project_id, project_root=project_root, task_group=task_group, task_id=task_id, task_type=task_type, run_seq=run_seq, status="reserving", started_at=when, finished_at=None, workers=[], lead_model="", validation="not-run", run_dir_rel=run_dir_rel, final_report_record_rel=final_report_record_rel, final_status_rel=final_status_rel) append_jsonl(home / "active.jsonl", row) # project index 도 미리 한 줄 — record_start 가 나중에 update 하므로 동일 row 형태로. append_jsonl(home / "projects" / project_id / "index.jsonl", row) # runCount/activeCount 는 예약 시점에 1회만 증가. record_start 는 had_reservation # 일 때 started=False 로 호출되므로 중복 카운트 없음. upsert_project_meta(home, project_id, project_root=project_root, when=when, started=True) def remove_reservation(home: Path, *, project_id: str, task_group: str, task_id: str, task_type: str, run_seq: int) -> None: """예약 row 를 제거한다(spawn 실패 시 cleanup). 중앙 락 보호 필요. 같은 (project, group, task, task_type, seq) 의 'reserving' 행만 지운다. 예약 시점에 증가시켰던 runCount/activeCount 도 되돌린다. """ run_id = build_run_id(project_id, task_group, task_id, task_type, run_seq) def _match(r): return (r.get("runId") == run_id and r.get("status") == "reserving") removed = remove_jsonl_row(home / "active.jsonl", _match) proj_index = home / "projects" / project_id / "index.jsonl" if proj_index.is_file(): remove_jsonl_row(proj_index, _match) if removed is None: return # reserve 가 카운터를 1 증가시켰으므로 활성/총 카운트를 되돌린다. target = home / "projects" / project_id / "meta.json" if not target.is_file(): return meta = load_owned_object(target, artifact="project metadata") meta["runCount"] = max(0, int(meta.get("runCount", 0)) - 1) meta["activeCount"] = max(0, int(meta.get("activeCount", 0)) - 1) write_owned_object_atomic(target, meta, artifact="project metadata")