"""목적별 Okstra 소유 JSON에서 고정 Markdown 입력을 렌더한다.""" from __future__ import annotations import argparse import os import sys from pathlib import Path from typing import Any, Mapping from okstra_project import ( StateError, parse_task_key, slugify, ) from .convergence import ( ConvergenceContractError, canonical_run_state_artifact, validated_run_authority, ) from .json_boundary import JsonBoundaryError, load_owned_object from .error_zip import last_output_path from .fixed_text import line as _line, scalar as _value from .paths import okstra_home from .recap import assemble_recap def _mapping(payload: Mapping[str, Any], key: str) -> Mapping[str, Any]: value = payload.get(key) return value if isinstance(value, Mapping) else {} def _project_policy_lines(project: Mapping[str, Any]) -> str: packs = project.get("reviewRulePacks") pack_lines = ( [ _line("Review rule pack", path) for path in packs if isinstance(path, str) ] if isinstance(packs, list) else [] ) qa_lines: list[str] = [] qa_commands = project.get("qaCommands") if isinstance(qa_commands, Mapping): for category, entries in sorted(qa_commands.items()): if not isinstance(entries, list): continue for entry in entries: if not isinstance(entry, Mapping): continue qa_lines.append( f"- QA `{_value(category)}` / `{_value(entry.get('label'))}`: " f"`{_value(entry.get('cmd'))}`\n" ) return ( "\n## Project Review Rule Packs\n\n" + ("".join(pack_lines) or "- None\n") + "\n## Project QA Commands\n\n" + ("".join(qa_lines) or "- None\n") ) def _assignment_line(prefix: str, assignment: Mapping[str, Any], identifier: object = None) -> str: label = prefix if identifier is None else f"{prefix} `{_value(identifier)}`" return ( f"- {label}: role `{_value(assignment.get('role'))}`; " f"provider `{_value(assignment.get('provider'))}`; " f"model `{_value(assignment.get('model'))}`\n" ) def _worker_assignment_lines(run: Mapping[str, Any]) -> str: assignments = run.get("workerAssignments") if not isinstance(assignments, list): return "- None\n" lines = [ _assignment_line("Worker", assignment, assignment.get("workerId")) for assignment in assignments if isinstance(assignment, Mapping) ] return "".join(lines) or "- None\n" def _worker_prompt_lines(run: Mapping[str, Any]) -> str: paths = run.get("workerPromptPathByWorkerId") if not isinstance(paths, Mapping): return "- None\n" lines = [ f"- Worker prompt `{_value(worker_id)}`: `{_value(path)}`\n" for worker_id, path in sorted(paths.items(), key=lambda item: _value(item[0])) ] return "".join(lines) or "- None\n" def _fixed_owned_artifact_path( project_root: Path, relative_path: Path, artifact: str ) -> Path: resolved_project_root = project_root.resolve() okstra_root = resolved_project_root / ".okstra" if okstra_root.is_symlink(): raise JsonBoundaryError(okstra_root, artifact, "Okstra root is not a fixed directory") resolved_okstra_root = okstra_root.resolve() try: resolved_okstra_root.relative_to(resolved_project_root) except ValueError as exc: raise JsonBoundaryError(okstra_root, artifact, "outside this project") from exc path = okstra_root / relative_path expected_path = resolved_okstra_root / relative_path if path.resolve() != expected_path: raise JsonBoundaryError(path, artifact, "not at its fixed Okstra path") return path def _owned_tasks_root(project_root: Path, artifact: str) -> Path: path = _fixed_owned_artifact_path(project_root, Path("tasks"), artifact) if not path.is_dir(): raise JsonBoundaryError(path, artifact, "tasks root is not a directory") return path def _load_project(project_root: Path) -> dict[str, Any]: path = _fixed_owned_artifact_path( project_root, Path("project.json"), "project metadata" ) return load_owned_object(path, artifact="project metadata") def _latest_task_pointer(project_root: Path) -> Mapping[str, Any]: path = _fixed_owned_artifact_path( project_root, Path("discovery") / "latest-task.json", "latest task pointer" ) if not path.is_file(): return {} pointer = load_owned_object(path, artifact="latest task pointer") manifest_path = _owned_task_manifest_path( project_root, pointer.get("taskManifestPath"), "latest task pointer" ) return _task_pointer_from_manifest( project_root, manifest_path, expected_task_key=pointer.get("taskKey"), pointer_run_manifest=pointer.get("latestRunManifestPath"), artifact="latest task pointer", ) def _project_relative(project_root: Path, path: Path) -> str: try: return str(path.resolve().relative_to(project_root.resolve())) except ValueError: return str(path) def _owned_project_path(project_root: Path, value: str, artifact: str) -> Path: path = Path(value) path = path if path.is_absolute() else project_root / path try: path.resolve().relative_to(project_root.resolve()) except ValueError as exc: raise JsonBoundaryError(path, artifact, "outside this project") from exc return path def _task_reference_error(project_root: Path, reason: str) -> JsonBoundaryError: return JsonBoundaryError(project_root, "task reference", reason) def _catalog_tasks(project_root: Path) -> list[Mapping[str, Any]]: path = _fixed_owned_artifact_path( project_root, Path("discovery") / "task-catalog.json", "task catalog" ) if not path.is_file(): return [] catalog = load_owned_object(path, artifact="task catalog") tasks = catalog.get("tasks") return [task for task in tasks if isinstance(task, Mapping)] if isinstance(tasks, list) else [] def _catalog_task_root(project_root: Path, task: Mapping[str, Any]) -> Path | None: value = task.get("taskRootPath") or task.get("taskRoot") if not isinstance(value, str) or not value: return None path = Path(value) path = path if path.is_absolute() else project_root / path return path if path.is_dir() else None def _normal_task_root(project_root: Path, task_key: str) -> Path | None: _, task_group, task_id = parse_task_key(task_key) path = _owned_tasks_root(project_root, "task reference") / slugify(task_group) / slugify(task_id) return path if path.is_dir() else None def _catalog_task_reference( project_root: Path, reference: str ) -> tuple[Path | None, str | None]: tasks = _catalog_tasks(project_root) if ":" in reference: parse_task_key(reference) matches = [ task for task in tasks if isinstance(task.get("taskKey"), str) and task["taskKey"].lower() == reference.lower() ] task_key = matches[0]["taskKey"] if matches else reference task_root = _catalog_task_root(project_root, matches[0]) if matches else None return task_root or _normal_task_root(project_root, task_key), task_key matches = [ task for task in tasks if isinstance(task.get("taskId"), str) and task["taskId"].lower() == reference.lower() ] if len(matches) != 1: raise _task_reference_error( project_root, "task reference must resolve to exactly one task" ) task_key = matches[0].get("taskKey") if not isinstance(task_key, str): return None, None return _catalog_task_root(project_root, matches[0]) or _normal_task_root( project_root, task_key ), task_key def _owned_task_manifest_path( project_root: Path, value: object, artifact: str ) -> Path: if not isinstance(value, str) or not value: raise JsonBoundaryError(project_root, artifact, "task manifest path is missing") path = Path(value) path = path if path.is_absolute() else project_root / path if path.name != "task-manifest.json": raise JsonBoundaryError(path, artifact, "task manifest path is invalid") tasks_root = _owned_tasks_root(project_root, artifact) try: path.resolve().relative_to(tasks_root.resolve()) except ValueError as exc: raise JsonBoundaryError(path, artifact, "task manifest is outside this project") from exc return path def _selected_run_manifest_path( project_root: Path, task_root: Path, value: str ) -> Path: path = Path(value) path = path if path.is_absolute() else project_root / path tasks_root = _owned_tasks_root(project_root, "run manifest") resolved_task_root = task_root.resolve() resolved_runs_root = (task_root / "runs").resolve() try: resolved_task_root.relative_to(tasks_root.resolve()) resolved_runs_root.relative_to(resolved_task_root) path.resolve().relative_to(resolved_runs_root) path.resolve().relative_to(resolved_task_root) except ValueError as exc: raise JsonBoundaryError(path, "run manifest", "outside selected task") from exc if not path.name.startswith("run-manifest-") or path.suffix != ".json": raise JsonBoundaryError(path, "run manifest", "not a normal run manifest path") return path def _owned_run_manifest_path(run_manifest: Path) -> Path: for runs_root in run_manifest.parents: if runs_root.name != "runs": continue task_root = runs_root.parent tasks_root = task_root.parent.parent if tasks_root.name != "tasks" or tasks_root.parent.name != ".okstra": continue project_root = tasks_root.parent.parent return _selected_run_manifest_path(project_root, task_root, str(run_manifest)) raise JsonBoundaryError( run_manifest, "run manifest", "outside owned task runs" ) def _task_manifest_from_reference(project_root: Path, task_ref: str) -> tuple[Path, str | None]: reference = task_ref.strip() candidate = Path(reference).expanduser() candidate = candidate if candidate.is_absolute() else project_root / candidate if candidate.is_file(): try: return ( _owned_task_manifest_path(project_root, str(candidate), "task reference"), None, ) except JsonBoundaryError as exc: raise _task_reference_error(project_root, exc.reason) from exc try: task_root, task_key = _catalog_task_reference(project_root, reference) except StateError as exc: raise _task_reference_error(project_root, str(exc)) from exc if task_root is None: raise _task_reference_error(project_root, "task not found") try: manifest_path = _owned_task_manifest_path( project_root, str(task_root / "task-manifest.json"), "task reference" ) except JsonBoundaryError as exc: raise _task_reference_error(project_root, exc.reason) from exc return manifest_path, task_key if isinstance(task_key, str) else None def _task_pointer_from_manifest( project_root: Path, manifest_path: Path, *, expected_task_key: object = None, pointer_run_manifest: object = None, artifact: str, ) -> Mapping[str, Any]: manifest = load_owned_object(manifest_path, artifact="task manifest") task_key = manifest.get("taskKey") if not isinstance(task_key, str): raise JsonBoundaryError(manifest_path, artifact, "task key is invalid") try: _, task_group, task_id = parse_task_key(task_key) except StateError as exc: raise JsonBoundaryError(manifest_path, artifact, "task key is invalid") from exc normal_manifest_path = ( _owned_tasks_root(project_root, artifact) / slugify(task_group) / slugify(task_id) / "task-manifest.json" ) if manifest_path.resolve() != normal_manifest_path.resolve(): raise JsonBoundaryError( manifest_path, artifact, "task key does not match task manifest path" ) if expected_task_key is not None and task_key != expected_task_key: raise JsonBoundaryError(manifest_path, artifact, "task key does not match task manifest") task_root = manifest_path.parent expected_timeline_path = task_root / "history" / "timeline.json" try: expected_timeline_path.resolve().relative_to(task_root.resolve()) except ValueError as exc: raise JsonBoundaryError( expected_timeline_path, "task timeline", "outside selected task" ) from exc timeline_value = manifest.get("historyTimelinePath") if isinstance(timeline_value, str) and timeline_value: timeline_path = _owned_project_path( project_root, timeline_value, "task timeline" ) if timeline_path.resolve() != expected_timeline_path.resolve(): raise JsonBoundaryError( timeline_path, "task timeline", "outside selected task" ) else: timeline_path = expected_timeline_path latest_run_manifest: object = None latest_run_path: Path | None = None if timeline_path.is_file(): timeline = load_owned_object(timeline_path, artifact="task timeline") runs = timeline.get("runs") if isinstance(runs, list): for run in reversed(runs): if isinstance(run, Mapping) and isinstance(run.get("runManifestPath"), str): latest_run_path = _selected_run_manifest_path( project_root, task_root, run["runManifestPath"] ) latest_run_manifest = _project_relative(project_root, latest_run_path) break if pointer_run_manifest is not None: if not isinstance(pointer_run_manifest, str): raise JsonBoundaryError(manifest_path, artifact, "run manifest path is invalid") pointer_run_path = _selected_run_manifest_path( project_root, task_root, pointer_run_manifest ) if latest_run_path is None or pointer_run_path.resolve() != latest_run_path.resolve(): raise JsonBoundaryError( manifest_path, artifact, "run manifest does not match task timeline" ) return { "taskKey": task_key, "taskManifestPath": _project_relative(project_root, manifest_path), "latestRunManifestPath": latest_run_manifest, } def _selected_task_pointer(project_root: Path, task_ref: str) -> Mapping[str, Any]: manifest_path, expected_task_key = _task_manifest_from_reference(project_root, task_ref) return _task_pointer_from_manifest( project_root, manifest_path, expected_task_key=expected_task_key, artifact="task reference", ) def render_project_context(project_root: Path, task_ref: str = "") -> str: """모델의 프로젝트 문맥에 필요한 안정 필드를 렌더한다.""" project = _load_project(project_root) architecture = _mapping(project, "architecture") current_task = ( _selected_task_pointer(project_root, task_ref) if task_ref else _latest_task_pointer(project_root) ) return ( "# Okstra Project Context\n\n" + _line("Project ID", project.get("projectId")) + _line("Project root", str(project_root)) + _line("Report language", project.get("reportLanguage")) + _line("Architecture style", architecture.get("style")) + _project_policy_lines(project) + "\n## Current Task Pointer\n\n" + _line("Task key", current_task.get("taskKey")) + _line("Task manifest", current_task.get("taskManifestPath")) + _line("Latest run manifest", current_task.get("latestRunManifestPath")) ) def render_run_input(run_manifest: Path) -> str: """런 매니페스트의 간결한 비밀 없는 실행 식별자를 렌더한다.""" run_manifest = _owned_run_manifest_path(run_manifest) run = load_owned_object(run_manifest, artifact="run manifest") scope = _mapping(run, "analysisScopeConfirmation") workflow = _mapping(run, "workflowSnapshot") lead = _mapping(run, "leadAssignment") adapter = _mapping(run, "leadAdapter") return ( "# Okstra Run Input\n\n" "## Run Context\n\n" + _line("Task key", run.get("taskKey")) + _line("Task type", run.get("taskType")) + _line("Run ID", run.get("runDateTimeSegment")) + _line("Work category", run.get("workCategory")) + _line("Lead runtime", run.get("leadRuntime")) + _line("Lead adapter name", adapter.get("name")) + _line("Lead dispatch mode", adapter.get("dispatchMode")) + _line("Lead session accounting", adapter.get("sessionAccounting")) + _line("Current phase", workflow.get("currentPhase")) + _line("Current phase state", workflow.get("currentPhaseState")) + _line("Scope confirmation", scope.get("status")) + "\n## Worker Roster\n\n" + _assignment_line("Lead", lead) + _worker_assignment_lines(run) + "\n## Artifact Paths\n\n" + _line("Worker prompts directory", run.get("workerPromptsDirectoryPath")) + _worker_prompt_lines(run) + _line("Worker results directory", run.get("workerResultsDirectoryPath")) + _line("Expected report", run.get("expectedReportRecordPath")) + _line("Expected status", run.get("expectedStatusPath")) + _line("Validator", run.get("validatorScriptPath")) + _line("Resume command", run.get("resumeCommandPath")) + "\n## Configuration References\n\n" + _line("Task manifest", run.get("taskManifestPath")) + _line("Instruction set", run.get("instructionSetPath")) + _line("Reference expectations", run.get("referenceExpectationsPath")) + _line("Report template", run.get("reportTemplatePath")) ) def render_active_context_input(project_root: Path, run_manifest: Path) -> str: """이전 run의 executor base ref만 목적별 텍스트로 투영한다.""" root = Path(project_root) if not root.is_absolute() or root.absolute() != root.resolve(): raise JsonBoundaryError(root, "active run context", "project root is not canonical") try: authority = validated_run_authority(run_manifest) if authority.project_root != root: raise ConvergenceContractError( "active run context project root does not match run authority" ) path = canonical_run_state_artifact( authority, manifest_field="activeRunContextPath", prefix="active-run-context", label="active run context path", ) except ConvergenceContractError as exc: raise JsonBoundaryError( run_manifest, "active run context", str(exc) ) from exc payload = load_owned_object(path, artifact="active run context") executor = _mapping(payload, "executorWorktree") return "# Okstra Active Context\n\n" + _line("Executor base ref", executor.get("baseRef")) def _group_segment(value: str) -> str: return "".join(character for character in value.lower() if character.isalnum()) def _catalog_task_path(project_root: Path, task: Mapping[str, Any]) -> tuple[Path, str]: task_key = task.get("taskKey") if not isinstance(task_key, str): raise JsonBoundaryError(project_root, "task catalog", "task key is invalid") try: _, task_group, task_id = parse_task_key(task_key) except StateError as exc: raise JsonBoundaryError(project_root, "task catalog", "task key is invalid") from exc expected_group_segment = slugify(task_group) expected_task_segment = slugify(task_id) group_segment = task.get("taskGroupPathSegment", expected_group_segment) task_segment = task.get("taskIdPathSegment", expected_task_segment) if group_segment != expected_group_segment or task_segment != expected_task_segment: raise JsonBoundaryError( project_root, "task catalog", "task path segment does not match task key" ) path = ( _owned_tasks_root(project_root, "task catalog") / expected_group_segment / expected_task_segment / "task-manifest.json" ) return _owned_task_manifest_path(project_root, str(path), "task catalog"), task_key def render_schedule_input(project_root: Path, task_group: str) -> str: """한 그룹의 카탈로그 작업에 대한 일정용 메타데이터를 렌더한다.""" project = _load_project(project_root) catalog_path = _fixed_owned_artifact_path( project_root, Path("discovery") / "task-catalog.json", "task catalog" ) catalog = load_owned_object(catalog_path, artifact="task catalog") target_group = _group_segment(task_group) task_blocks: list[str] = [] catalog_tasks = catalog.get("tasks") if isinstance(catalog_tasks, list): for entry in catalog_tasks: if not isinstance(entry, Mapping): continue task_key = entry.get("taskKey") if not isinstance(task_key, str): continue try: _, entry_group, _ = parse_task_key(task_key) except StateError: continue if _group_segment(entry_group) != target_group: continue manifest_path, expected_task_key = _catalog_task_path(project_root, entry) _task_pointer_from_manifest( project_root, manifest_path, expected_task_key=expected_task_key, artifact="task catalog", ) manifest = load_owned_object(manifest_path, artifact="task manifest") workflow = _mapping(manifest, "workflow") task_blocks.append( "### " + _value(manifest.get("taskId")) + "\n\n" + _line("Task key", manifest.get("taskKey")) + _line("Work status", manifest.get("workStatus")) + _line("Work category", manifest.get("workCategory")) + _line("Task type", manifest.get("taskType")) + _line("Current phase", workflow.get("currentPhase")) ) tasks = "".join(task_blocks) or "- None\n" return ( "# Okstra Schedule Input\n\n" + _line("Project ID", project.get("projectId")) + _line("Task group", task_group) + "\n## Tasks\n\n" + tasks ) def render_code_review_input(project_root: Path, base: str, head: str) -> str: """리뷰에 필요한 고정 프로젝트 식별자와 git 범위를 렌더한다.""" project = _load_project(project_root) return ( "# Okstra Code Review Input\n\n" + _line("Project ID", project.get("projectId")) + _line("Project root", str(project_root)) + _line("Base ref", base) + _line("Head ref", head) ) def _selected_task_data( project_root: Path, task_ref: str ) -> tuple[Mapping[str, Any], list[Mapping[str, Any]]]: manifest_path, expected_key = _task_manifest_from_reference(project_root, task_ref) _task_pointer_from_manifest( project_root, manifest_path, expected_task_key=expected_key, artifact="task reference" ) manifest = load_owned_object(manifest_path, artifact="task manifest") timeline_path = manifest_path.parent / "history" / "timeline.json" timeline = ( load_owned_object(timeline_path, artifact="task timeline") if timeline_path.is_file() else {} ) runs = timeline.get("runs") return manifest, [run for run in runs if isinstance(run, Mapping)] if isinstance(runs, list) else [] def _next_phase_lines(workflow: Mapping[str, Any]) -> str: pointer = _mapping(workflow, "nextRecommendedPhase") return ( _line("Next phase", pointer.get("phase")) + _line("Next phase status", pointer.get("status")) + _line("Next phase rationale", pointer.get("rationale")) ) def _latest_timeline_run(manifest_path: Path) -> Mapping[str, Any]: timeline_path = manifest_path.parent / "history" / "timeline.json" if not timeline_path.is_file(): return {} timeline = load_owned_object(timeline_path, artifact="task timeline") runs = timeline.get("runs") if not isinstance(runs, list): return {} return next( (run for run in reversed(runs) if isinstance(run, Mapping)), {}, ) def _overview_rows( project_root: Path, *, task_type: str = "", latest_run_status: str = "", task_group: str = "", ) -> list[dict[str, object]]: rows: list[dict[str, object]] = [] for task in _catalog_tasks(project_root): manifest_path, task_key = _catalog_task_path(project_root, task) pointer = _task_pointer_from_manifest( project_root, manifest_path, expected_task_key=task_key, artifact="task catalog" ) manifest = load_owned_object(manifest_path, artifact="task manifest") latest_run = _latest_timeline_run(manifest_path) _, group, _ = parse_task_key(task_key) workflow = _mapping(manifest, "workflow") next_phase = _mapping(workflow, "nextRecommendedPhase") row = { "taskKey": task_key, "taskGroup": manifest.get("taskGroup") or task.get("taskGroup") or group, "taskType": manifest.get("taskType") or latest_run.get("taskType"), "latestRunStatus": latest_run.get("status") or manifest.get("latestRunStatus"), "updatedAt": manifest.get("updatedAt") or task.get("updatedAt") or latest_run.get("runTimestamp"), "latestRunManifestPath": pointer.get("latestRunManifestPath"), "workCategory": manifest.get("workCategory"), "currentStatus": manifest.get("currentStatus"), "currentPhase": workflow.get("currentPhase"), "currentPhaseState": workflow.get("currentPhaseState"), "nextPhase": next_phase.get("phase"), "nextPhaseStatus": next_phase.get("status"), "nextPhaseRationale": next_phase.get("rationale"), "awaitingApproval": workflow.get("awaitingApproval"), "latestReportRecordPath": latest_run.get("reportRecordPath") or manifest.get("latestReportRecordPath"), "latestResumeCommandPath": latest_run.get("resumeCommandPath") or manifest.get("latestResumeCommandPath"), "workStatus": manifest.get("workStatus"), "lastRun": latest_run.get("runTimestamp"), } if task_type and row["taskType"] != task_type: continue if latest_run_status and row["latestRunStatus"] != latest_run_status: continue if task_group and row["taskGroup"] != task_group: continue rows.append(row) rows.sort(key=lambda row: _value(row.get("taskKey"))) rows.sort(key=lambda row: _value(row.get("updatedAt")), reverse=True) return rows def _status_overview_text(rows: list[dict[str, object]]) -> str: blocks = [_line("Task count", len(rows))] for row in rows: blocks.extend(( "\n## Task\n\n", _line("Task key", row.get("taskKey")), _line("Task group", row.get("taskGroup")), _line("Task type", row.get("taskType")), _line("Latest run status", row.get("latestRunStatus")), _line("Updated at", row.get("updatedAt")), _line("Latest run manifest", row.get("latestRunManifestPath")), _line("Work category", row.get("workCategory")), _line("Current status", row.get("currentStatus")), _line("Current phase", row.get("currentPhase")), _line("Current phase state", row.get("currentPhaseState")), _line("Next phase", row.get("nextPhase")), _line("Next phase status", row.get("nextPhaseStatus")), _line("Next phase rationale", row.get("nextPhaseRationale")), _line("Awaiting approval", row.get("awaitingApproval")), _line("Latest report", row.get("latestReportRecordPath")), _line("Latest resume command", row.get("latestResumeCommandPath")), _line("Work status", row.get("workStatus")), )) return "# Okstra Status Input\n\n" + "".join(blocks) def _history_overview_text(rows: list[dict[str, object]], limit: int) -> str: shown = rows[:limit] blocks = [ _line("Task count", len(rows)), _line("Shown task count", len(shown)), _line("Remaining task count", max(0, len(rows) - len(shown))), ] for row in shown: blocks.extend(( "\n## Task\n\n", _line("Task key", row.get("taskKey")), _line("Task group", row.get("taskGroup")), _line("Task type", row.get("taskType")), _line("Current status", row.get("currentStatus")), _line("Latest run status", row.get("latestRunStatus")), _line("Last run", row.get("lastRun")), _line("Latest report", row.get("latestReportRecordPath")), _line("Updated at", row.get("updatedAt")), _line("Latest run manifest", row.get("latestRunManifestPath")), )) return "# Okstra History Input\n\n" + "".join(blocks) def render_status_input( project_root: Path, task_ref: str, *, task_type: str = "", latest_run_status: str = "", task_group: str = "", ) -> str: if not task_ref: return _status_overview_text( _overview_rows( project_root, task_type=task_type, latest_run_status=latest_run_status, task_group=task_group, ), ) manifest, _ = _selected_task_data(project_root, task_ref) workflow = _mapping(manifest, "workflow") checkpoint = _mapping(workflow, "lastSafeCheckpoint") phases = _mapping(workflow, "phaseStates") phase_text = ", ".join(f"{key}={_value(value)}" for key, value in sorted(phases.items())) return ( "# Okstra Status Input\n\n" + _line("Task key", manifest.get("taskKey")) + _line("Task type", manifest.get("taskType")) + _line("Work category", manifest.get("workCategory")) + _line("Current status", manifest.get("currentStatus")) + _line("Latest run status", manifest.get("latestRunStatus")) + _line("Current phase", workflow.get("currentPhase")) + _line("Current phase state", workflow.get("currentPhaseState")) + _line("Phase states", phase_text) + _line("Last completed phase", workflow.get("lastCompletedPhase")) + _next_phase_lines(workflow) + _line("Awaiting approval", workflow.get("awaitingApproval")) + _line("Work status", manifest.get("workStatus")) + _line("Work status updated at", manifest.get("workStatusUpdatedAt")) + _line("Work status note", manifest.get("workStatusNote")) + _line("Latest report", manifest.get("latestReportRecordPath")) + _line("Latest resume command", manifest.get("latestResumeCommandPath")) + _line("History timeline", manifest.get("historyTimelinePath")) + _line("Safe checkpoint label", checkpoint.get("label")) + _line("Safe checkpoint run manifest", checkpoint.get("runManifestPath")) + _line("Safe checkpoint team state", checkpoint.get("teamStatePath")) + _line("Safe checkpoint report", checkpoint.get("reportRecordPath")) + _line("Safe checkpoint resume command", checkpoint.get("resumeCommandPath")) ) def render_history_input( project_root: Path, task_ref: str, *, task_type: str = "", latest_run_status: str = "", task_group: str = "", limit: int = 20, ) -> str: if not task_ref: return _history_overview_text( _overview_rows( project_root, task_type=task_type, latest_run_status=latest_run_status, task_group=task_group, ), limit, ) manifest, runs = _selected_task_data(project_root, task_ref) blocks: list[str] = [] for index, run in enumerate(runs, 1): blocks.append( f"\n## Run {index}\n\n" + _line("Run timestamp", run.get("runTimestamp")) + _line("Run date-time segment", run.get("runDateTimeSegment")) + _line("Task type", run.get("taskType")) + _line("Status", run.get("status")) + _line("Run manifest", run.get("runManifestPath")) + _line("Report", run.get("reportRecordPath")) + _line("Resume command", run.get("resumeCommandPath")) + _line("Related tasks", ",".join(run.get("relatedTasks", [])) if isinstance(run.get("relatedTasks"), list) else None) ) return "# Okstra History Input\n\n" + _line("Task key", manifest.get("taskKey")) + "".join(blocks) def render_recap_input(project_root: Path, task_ref: str) -> str: manifest_path, _ = _task_manifest_from_reference(project_root, task_ref) recap = assemble_recap(manifest_path.parent, project_root) blocks = [ "# Okstra Recap Input\n\n", _line("Task key", recap.get("taskKey")), _line("Run count", recap.get("runCount")), ] transitions = recap.get("transitions") for transition in transitions if isinstance(transitions, list) else []: if not isinstance(transition, Mapping): continue pointer = _mapping(transition, "nextRecommendedPhase") blocks.extend(( "\n## Transition\n\n", _line("Run timestamp", transition.get("runTimestamp")), _line("Task type", transition.get("taskType")), _line("From phase", transition.get("fromPhase")), _line("To phase", transition.get("toPhase")), _line("Status", transition.get("status")), _line("Last completed phase", transition.get("lastCompletedPhase")), _line("Next phase", pointer.get("phase")), _line("Next phase status", pointer.get("status")), _line("Next phase rationale", pointer.get("rationale")), _line("Report", transition.get("reportRecordPath")), )) return "".join(blocks) def render_report_input(project_root: Path, task_ref: str) -> str: manifest, _ = _selected_task_data(project_root, task_ref) workflow = _mapping(manifest, "workflow") return ( "# Okstra Report Input\n\n" + _line("Task key", manifest.get("taskKey")) + _line("Task type", manifest.get("taskType")) + _line("Current status", manifest.get("currentStatus")) + _line("Work status", manifest.get("workStatus")) + _line("Latest report", manifest.get("latestReportRecordPath")) + _next_phase_lines(workflow) ) def _csv(value: object) -> str: return ",".join(str(item) for item in value) if isinstance(value, list) else "" def _provider_models(run: Mapping[str, Any]) -> dict[str, str]: values: dict[str, set[str]] = {} assignments = run.get("workerAssignments") if isinstance(assignments, list): for assignment in assignments: if not isinstance(assignment, Mapping): continue provider = assignment.get("provider") model = assignment.get("model") if isinstance(provider, str) and isinstance(model, str) and model: values.setdefault(provider, set()).add(model) for provider, models in values.items(): if len(models) > 1: raise JsonBoundaryError( Path("workerAssignments"), "run manifest", f"conflicting models for provider {provider}" ) return {provider: next(iter(models)) for provider, models in values.items()} def render_rerun_input(run_manifest: Path) -> str: run_manifest = _owned_run_manifest_path(run_manifest) run = load_owned_object(run_manifest, artifact="run manifest") models = _provider_models(run) team = _mapping(run, "teamContract") executor = _mapping(team, "executor") executor_provider = executor.get("provider") executor_model = executor.get("model") if ( isinstance(executor_provider, str) and isinstance(executor_model, str) and executor_model and models.get(executor_provider) and models[executor_provider] != executor_model ): raise JsonBoundaryError( run_manifest, "run manifest", f"executor model conflicts with provider {executor_provider}", ) return ( "# Okstra Rerun Input\n\n" + _line("Project ID", run.get("projectId")) + _line("Task group", run.get("taskGroup")) + _line("Task ID", run.get("taskId")) + _line("Task type", run.get("taskType")) + _line("Task brief", run.get("taskBriefPath")) + _line("Recommended workers", _csv(run.get("recommendedWorkers"))) + _line("Related tasks", _csv(run.get("relatedTasks"))) + _line("Claude model", models.get("claude")) + _line("Codex model", models.get("codex")) + _line("Antigravity model", models.get("antigravity")) + _line("Executor provider", executor.get("provider")) + _line("Executor model", executor.get("model")) ) def _symlink_ancestor(path: Path) -> Path | None: lexical = Path(os.path.abspath(path)) current = Path(lexical.anchor) for part in lexical.parts[1:-1]: current /= part if current.is_symlink(): return current return None def render_error_zip_input() -> str: home = okstra_home() linked_ancestor = _symlink_ancestor(home) if linked_ancestor is not None: raise JsonBoundaryError( home, "error zip configuration", f"Okstra home contains a symbolic link ({linked_ancestor})", ) if home.is_symlink(): raise JsonBoundaryError(home, "error zip configuration", "Okstra home is not a fixed directory") configuration = home / "error-zip.json" if configuration.is_symlink(): raise JsonBoundaryError( configuration, "error zip configuration", "configuration is not at its fixed path", ) return "# Okstra Error Zip Input\n\n" + _line( "Previous output path", last_output_path(home.resolve()) ) def render_task_selection_input(project_root: Path, task_ref: str = "") -> str: tasks = _catalog_tasks(project_root) token = task_ref.casefold() matches: list[tuple[Mapping[str, Any], str]] = [] for task in tasks: task_id = str(task.get("taskId") or "") task_group = str(task.get("taskGroup") or "") via = "taskId" if token and task_id.casefold() == token else "" via = "taskGroup" if token and task_group.casefold() == token else via if not token or via: matches.append((task, via or "catalog")) matches.sort(key=lambda item: (_value(item[0].get("updatedAt")), _value(item[0].get("taskKey"))), reverse=True) lines = ["# Okstra Task Selection Input\n\n", _line("Match count", len(matches))] for task, via in matches[:10]: lines.extend( ( _line("Task", task.get("taskKey")), _line("Updated at", task.get("updatedAt")), _line("Matched via", via), ) ) return "".join(lines) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="okstra model-io", description="Render fixed Markdown views of Okstra-owned JSON.", ) commands = parser.add_subparsers(dest="command", required=True) project = commands.add_parser("project-context") project.add_argument("--project-root", required=True) project.add_argument("--task-ref", default="") run = commands.add_parser("run-input") run.add_argument("--run-manifest", required=True) active = commands.add_parser("active-context-input") active.add_argument("--project-root", required=True) active.add_argument("--run-manifest", required=True) schedule = commands.add_parser("schedule-input") schedule.add_argument("--project-root", required=True) schedule.add_argument("--task-group", required=True) review = commands.add_parser("code-review-input") review.add_argument("--project-root", required=True) review.add_argument("--base", required=True) review.add_argument("--head", required=True) for name in ("status-input", "history-input", "report-input"): inspect = commands.add_parser(name) inspect.add_argument("--project-root", required=True) inspect.add_argument("--task-ref", required=name == "report-input", default="") if name != "report-input": inspect.add_argument("--task-type", default="") inspect.add_argument("--latest-run-status", default="") inspect.add_argument("--task-group", default="") if name == "history-input": inspect.add_argument("--limit", type=int, default=20) recap = commands.add_parser("recap-input") recap.add_argument("--project-root", required=True) recap.add_argument("--task-ref", required=True) rerun = commands.add_parser("rerun-input") rerun.add_argument("--run-manifest", required=True) commands.add_parser("error-zip-input") selection = commands.add_parser("task-selection-input") selection.add_argument("--project-root", required=True) selection.add_argument("--task-ref", default="") return parser def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) try: if args.command == "project-context": text = render_project_context(Path(args.project_root), args.task_ref) elif args.command == "run-input": text = render_run_input(Path(args.run_manifest)) elif args.command == "active-context-input": text = render_active_context_input( Path(args.project_root), Path(args.run_manifest) ) elif args.command == "schedule-input": text = render_schedule_input(Path(args.project_root), args.task_group) elif args.command == "code-review-input": text = render_code_review_input(Path(args.project_root), args.base, args.head) elif args.command == "status-input": text = render_status_input( Path(args.project_root), args.task_ref, task_type=args.task_type, latest_run_status=args.latest_run_status, task_group=args.task_group, ) elif args.command == "history-input": text = render_history_input( Path(args.project_root), args.task_ref, task_type=args.task_type, latest_run_status=args.latest_run_status, task_group=args.task_group, limit=max(0, args.limit), ) elif args.command == "recap-input": text = render_recap_input(Path(args.project_root), args.task_ref) elif args.command == "report-input": text = render_report_input(Path(args.project_root), args.task_ref) elif args.command == "rerun-input": text = render_rerun_input(Path(args.run_manifest)) elif args.command == "error-zip-input": text = render_error_zip_input() elif args.command == "task-selection-input": text = render_task_selection_input(Path(args.project_root), args.task_ref) else: # argparse가 도달 불가로 만들지만 명시적 종료 경계를 유지한다. raise ValueError(f"unsupported command: {args.command}") except JsonBoundaryError as exc: print(f"model-io: {exc}", file=sys.stderr) return 2 print(text, end="") return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))