"""목적별 고정 Markdown 입력을 조립한다. 각 `render_*` 는 CLI 하위명령 하나에 대응한다. 참조 해석은 `.references`, 줄 서식은 `.lines` 에 있고 여기서는 둘을 엮기만 한다. """ from __future__ import annotations 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 ..error_zip import last_output_path from ..fixed_text import line as _line, scalar as _value from ..json_boundary import JsonBoundaryError, load_owned_object from ..paths import okstra_home from ..recap import assemble_group_recap, assemble_recap from ..timeline_runs import current_run_facts from ..worker_artifacts import worker_provider_id from .lines import ( _assignment_line, _csv, _history_overview_text, _mapping, _next_phase_lines, _project_policy_lines, _provider_models, _status_overview_text, _worker_assignment_lines, _worker_prompt_lines, ) from .references import ( _catalog_task_path, _catalog_tasks, _fixed_owned_artifact_path, _group_segment, _latest_task_pointer, _load_project, _owned_run_manifest_path, _owned_task_manifest_path, _owned_tasks_root, _selected_task_data, _selected_task_pointer, _symlink_ancestor, _task_manifest_from_reference, _task_pointer_from_manifest, ) def _latest_timeline_run(project_root: Path, 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 {} latest = next( (run for run in reversed(runs) if isinstance(run, Mapping)), None, ) return current_run_facts(project_root, latest) if latest is not None else {} 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(project_root, manifest_path) _, group, _ = parse_task_key(task_key) workflow = _mapping(manifest, "workflow") next_phase = _mapping(workflow, "nextRecommendedPhase") # task 단위 현재 사실은 task-manifest 가 권위다(validate-run 이 run 종료 # 시 거기에 쓴다). timeline 의 마지막 run 은 task-manifest 에 그 필드가 # 없을 때의 폴백이고, 그 값도 run-manifest 로 덮어쓴 현재 사실이다. row = { "taskKey": task_key, "taskGroup": manifest.get("taskGroup") or task.get("taskGroup") or group, "taskType": manifest.get("taskType") or latest_run.get("taskType"), "latestRunStatus": manifest.get("latestRunStatus") or latest_run.get("status"), "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": manifest.get("latestReportRecordPath") or latest_run.get("reportRecordPath"), "latestResumeCommandPath": latest_run.get("resumeCommandPath") or manifest.get("latestResumeCommandPath"), "workStatus": manifest.get("workStatus"), "latestWorkRecordPath": manifest.get("latestWorkRecordPath"), "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 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 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 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("Direct work record", manifest.get("latestWorkRecordPath")) + _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((current_run_facts(project_root, run) for run in 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")), _line("Work status", recap.get("workStatus")), _line("Direct work record", recap.get("latestWorkRecordPath")), ] 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_group_recap_input(project_root: Path, task_group: str) -> str: recap = assemble_group_recap(project_root, task_group) following = _mapping(recap, "nextInGroup") blocks = [ "# Okstra Group Recap Input\n\n", _line("Task group", recap.get("taskGroup")), _line("Group context", recap.get("groupContextPath")), _line("Human sections", "present" if recap.get("hasHumanSections") else "absent"), _line("Brief count", recap.get("briefCount")), _line("Task count", recap.get("taskCount")), _line("Next in group", following.get("taskId")), _line("Next in group brief", following.get("brief")), ] queue = recap.get("queue") for position, row in enumerate(queue if isinstance(queue, list) else [], 1): if not isinstance(row, Mapping): continue waits = row.get("waitsFor") blocks.extend(( "\n## Queue entry\n\n", _line("Position", position), _line("Task id", row.get("taskId")), _line("Brief", row.get("brief")), _line("Status", row.get("status")), _line("Progress", row.get("progress")), _line("Waits for", ", ".join(waits) if isinstance(waits, list) and waits else ""), )) tasks = recap.get("tasks") for task in tasks if isinstance(tasks, list) else []: if not isinstance(task, Mapping): continue pointer = _mapping(task, "nextRecommendedPhase") memory = _mapping(task, "memory") blocks.extend(( "\n## Task\n\n", _line("Task id", task.get("taskId")), _line("Task key", task.get("taskKey")), _line("Current phase", task.get("currentPhase")), _line("Phase state", task.get("currentPhaseState")), _line("Latest run status", task.get("latestRunStatus")), _line("Work status", task.get("workStatus")), _line("Direct work record", task.get("latestWorkRecordPath")), _line("Memory source", memory.get("source")), _line("Run count", task.get("runCount")), _line("Next phase", pointer.get("phase")), _line("Next phase status", pointer.get("status")), _line("Next phase rationale", pointer.get("rationale")), _line("Report", task.get("reportPath")), _line("Memory date", memory.get("date")), _line("Memory run", ( f"{memory.get('taskType')} #{memory.get('seq')}" if memory.get("taskType") else "" )), _line("Headline", memory.get("headline")), )) for label, key in (("Decision", "decisions"), ("Watch out", "watchOut"), ("Follow-up", "followUps")): items = memory.get(key) for index, item in enumerate(items if isinstance(items, list) else [], 1): blocks.append(_line(f"{label} {index}", item)) 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 _worker_providers_csv(recommended: Any) -> str: """로스터 슬롯 id 를 `--workers` 가 받는 provider id 로 접는다. `recommendedWorkers` 는 슬롯 id(`codex-verifier`)를 싣고 `--workers` 는 프로파일이 선언한 provider 허용목록만 받는다([`workers.py`](../workers.py)). 슬롯 id 를 그대로 넘기면 재실행이 `unknown workers` 로 죽으므로, 재실행 입력을 만드는 이 자리에서 어휘를 맞춘다. 슬롯 수는 `--role-count` 가 나른다. """ if not isinstance(recommended, list): return "" seen: list[str] = [] for item in recommended: if not isinstance(item, str) or not item.strip(): continue provider = worker_provider_id(item.strip()) if provider not in seen: seen.append(provider) return ",".join(seen) 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("Workers", _worker_providers_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 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)