"""프롬프트를 실제로 찍어 낸다 — run 에 매인 것과 독립 실행 두 갈래. `--run-manifest` 가 있으면 run 갈래다: 매니페스트에서 신원을 확인하고, 필요하면 동적 검증자를 예약하고, 그 run 의 산출물 경로로 프롬프트를 쓴다. 없으면 독립 갈래다: 인자로 받은 audience·duty 를 스냅샷해 감사 가능한 명세를 만든다. 두 갈래가 같은 파일에 있는 이유는 `_materialize` 가 둘 중 하나를 고르는 유일한 분기점이기 때문이다. """ from __future__ import annotations import argparse import os from pathlib import Path import shlex import shutil import tempfile from typing import Any, Mapping, get_args from ..invocation import ( AgentInstruction, AgentInstructionSource, AgentAudience, AgentInvocationRequest, AgentModelAssignment, PreparedAgentInvocation, agent_model_assignment_from_payload, digest_duty_catalog, invocation_execution_identity_from_manifest, prepare_agent_invocation, compose_agent_prompt, compose_unbound_run_prompt, ) from ...assignment_environment import load_assignment_context from ...assignment_resolver import AssignmentContext, resolve_dispatch_assignment from ...path_hints import hydrate_active_run_context from ...worker_prompt_headers import worker_prompt_headers from ...worker_prompt_contract import complete_reverify_instruction, validate_reverify_prompt from ...worker_prompt_policy import ( critic_assignment_ref, is_plan_critic_verification, is_verification_dispatch_kind, ) from ...paths import okstra_home from ...final_report_paths import final_report_data_path from ...final_report_schema import load_schema_version from ...report_inputs import report_narrative_path, uses_report_contract_v3 from ...report_narrative import NarrativeContractError, parse_narrative_structure from ...report_synthesis_packet import ( ReportSynthesisPacketError, materialize_report_synthesis_packet, ) from ...worker_prompt_body import report_writer_input_lines from ...dispatch_state import detect_terminal_backend from ...report_corrections import ( CorrectionsCheck, body_owned_section_conflicts, render_corrections_section, render_output_section, ) from .corrections import ( corrections_payload, run_corrections_apply, run_corrections_check, ) from .dynamic_verifier import ( _dynamic_verifier_source, _reserve_dynamic_verifier_request, ) from .inputs import ( AgentPromptCliError, _authorized_path, _exact_path, _is_relative_to, _is_slug, _mapping, _project_input, _project_manifest_path, _project_root, _read_json_object, _relative, ) from .run_identity import _validate_run_identity _AUDIENCES = frozenset(get_args(AgentAudience)) # 결과 파일과 별개의 워커 결과 포인터를 가진 audience. 이들만 `--audit-source` 로 # 두 번째 경로를 받고, 프롬프트에 `**Worker Result Path:**` 앵커가 생긴다. # 나머지 audience 의 결과는 하나뿐이고 감사 사이드카는 그 이름에서 파생된다. _POINTER_AUDIENCES = frozenset({"report-writer", "translator"}) def _materialize(args: argparse.Namespace) -> PreparedAgentInvocation: project_root = _project_root(args.project_root) if args.run_manifest: return _materialize_run(args, project_root) return _materialize_standalone(args, project_root) def _materialize_run( args: argparse.Namespace, project_root: Path, ) -> PreparedAgentInvocation: required = { "worker-id": args.worker_id, "dispatch-kind": args.dispatch_kind, "assignment-ref": args.assignment_ref, "result": args.result, } missing = [name for name, value in required.items() if not value] if missing: raise AgentPromptCliError( f"run materialization requires: {', '.join('--' + name for name in missing)}" ) forbidden = { "host-runtime": args.host_runtime, "provider": args.provider, "model-role": args.model_role, "model": args.model, "purpose": args.purpose, } present = [name for name, value in forbidden.items() if value] if present: raise AgentPromptCliError( f"run materialization forbids: {', '.join('--' + name for name in present)}" ) manifest_path = _project_input(project_root, args.run_manifest, "run manifest") manifest = _read_json_object(manifest_path, "run manifest") contract = _mapping(manifest.get("agentContract"), "run agent contract") authorized = _mapping(contract.get("authorizedPaths"), "authorized paths") instruction_path = _authorized_path( project_root, args.instruction, authorized.get("instructionRoots"), "instruction", must_exist=True, ) prompt_path = _authorized_path( project_root, args.prompt, authorized.get("promptRoots"), "prompt", must_exist=False, ) result_path = _authorized_path( project_root, args.result, authorized.get("resultRoots"), "result", must_exist=False, ) audit_source_path = ( _authorized_path( project_root, args.audit_source, authorized.get("resultRoots"), "audit source", must_exist=False, ) if args.audit_source else None ) if args.audience not in _POINTER_AUDIENCES: # 결과가 하나뿐인 audience 다. 두 번째 경로는 워커(`**Result Path:**` 에 # 쓴다)와 수집기(jobs-file 의 workerResultPath 를 기다린다)에게 서로 다른 # 파일을 말하는 것밖에 못 한다. 실측(2026-09-02, fontsninja-v3-site # dev-10626-1 error-analysis r04): 리드가 reverify 에 `-worker-` 이름을 # `--audit-source` 로 따로 넘겨 검증 디스패치 9건 중 8건이 결과를 다 # 쓰고도 `required worker artifact was not produced` 로 끝났다. if ( audit_source_path is not None and _normalized(audit_source_path) != _normalized(result_path) ): raise AgentPromptCliError( f"{args.audience} materialization forbids --audit-source: this " "audience writes one result and its audit sidecar derives from " "that result's name. Name --result with the canonical `-worker-` " "token instead (reverify: " "-worker-reverify-r--.md) and put " "the same path in the jobs file's workerResultPath" ) audit_source_path = result_path elif args.audience == "report-writer": if audit_source_path is None: # The report writer is the one audience whose result path is not its # own worker result: it writes the report body, while the audit # sidecar is derived from its `.md`. With both collapsed into one # value the prompt loses its `**Worker Result Path:**` anchor and the # writer puts the report where the audit file belongs — silently, # because every header is still present and well-formed. The roster # path derives both from the manifest; a dynamic call has to name # them. # # Which artifact `--result` names depends on the report contract, and # `dispatch_state.dispatch_result_path` is what decides it. Naming # only the v2 answer here sent v3 runs to the data.json, so the # narrative `report-finalize` assembles from was never written and # the phase failed later, at a place that does not point back here. if uses_report_contract_v3(manifest): expected = report_narrative_path(project_root, manifest) raise AgentPromptCliError( "report-writer materialization requires --audit-source: " f"under report contract 3.0 --result is the narrative " f"({expected}), which `report-finalize` assembles the report " "data.json from, and --audit-source the worker-result .md " "the audit sidecar is derived from" ) raise AgentPromptCliError( "report-writer materialization requires --audit-source: under " "report contract 2.0 --result is the report data.json, and " "--audit-source the worker-result .md the audit sidecar is " "derived from" ) _validate_report_writer_paths( project_root, manifest, worker_id=args.worker_id, result_path=result_path, audit_source_path=audit_source_path, ) elif ( audit_source_path is None or _normalized(audit_source_path) == _normalized(result_path) ): # translator 다. report-writer 와 같은 두-경로 audience 인데 종전엔 빠진 # `--audit-source` 를 `--result` 로 조용히 접었다. 그러면 프롬프트에 # `**Worker Result Path:**` 앵커가 안 생기는데, `worker-dispatch` 는 그 # 앵커가 `**Result Path:**` 와 다를 것을 요구하므로 발행은 되지만 어떤 # 디스패치도 받지 못하는 예약이 남는다. 예약은 회수 명령이 없어 리드가 # 두 번째 id 로 다시 만드는 순간 "exactly one reservation" 이 영영 # 깨졌다(2026-09-08 실측, fontsninja-v3-site dev-10627-2 # requirements-discovery). 발행 시점에 거절한다. raise AgentPromptCliError( "translator materialization requires --audit-source distinct from " "--result: --result is the translator's own report " "(e.g. worker-results/translator-translations--.md) " "and --audit-source the worker-result .md the audit sidecar is " "derived from (worker-results/translator-worker--.md). " "`worker-dispatch --workers translator` reads both from the prompt's " "**Result Path:** and **Worker Result Path:** anchors and refuses a " "prompt without the second. To fix a prompt already published " "without it, re-run this command with the same --invocation-id and " "--prompt plus --replace-undispatched; a new invocation id would " "leave two reservations and block the translator for this run" ) _validate_run_identity( manifest, worker_id=args.worker_id, dispatch_kind=args.dispatch_kind, assignment_ref=args.assignment_ref, audience=args.audience, ) assignments = _mapping( manifest.get("invocationAssignments"), "run invocation assignments", ) if args.assignment_ref not in assignments: # 유효값은 이 매니페스트 안에만 있고 리드는 그것을 추측할 수 없다. # 이름만 거절하면 읽는 쪽이 매니페스트를 직접 열어 키를 세게 된다. critic_ref = critic_assignment_ref(str(manifest.get("taskType", ""))) recovery = "" if ( args.assignment_ref == "reverify/critic-worker" and critic_ref in assignments and is_plan_critic_verification( task_type=str(manifest.get("taskType", "")), assignment_ref=critic_ref, dispatch_kind=args.dispatch_kind, ) ): recovery = ( f"; for the rostered plan critic, use --assignment-ref {critic_ref} " "with --worker-id critic-worker and --audience reverification-worker. " "Keep the plan-verify-r dispatch kind and, on v2 runs, the selected " "critic's --source-role-execution-ref. This is a reference correction; " "the existing critic assignment does not need to be added or changed" ) raise AgentPromptCliError( f"assignment reference is missing from run manifest: " f"{args.assignment_ref}; this run declares " + (", ".join(sorted(assignments)) or "no invocation assignment") + recovery ) assignment_payload = assignments[args.assignment_ref] assignment = agent_model_assignment_from_payload(assignment_payload) dynamic_source = _dynamic_verifier_source(args, manifest) identity = ( None if dynamic_source is not None else invocation_execution_identity_from_manifest( manifest, assignment=assignment, assignment_ref=args.assignment_ref, duty_id=args.audience, ) ) duty_root = _project_manifest_path( project_root, contract.get("dutyRootPath"), "duty root", must_exist=True, ) if contract.get("catalogDigest") != digest_duty_catalog(duty_root): raise AgentPromptCliError("run duty snapshot catalog digest does not match") active_context_path = _project_manifest_path( project_root, manifest.get("activeRunContextPath"), "active run context", must_exist=True, ) active_context = hydrate_active_run_context( _read_json_object(active_context_path, "active run context") ) prompt_rel = _relative(project_root, prompt_path) result_rel = _relative(project_root, result_path) audit_rel = _relative(project_root, audit_source_path) anchor_lines = tuple(worker_prompt_headers( project_root=project_root, prompt_rel=prompt_rel, result_rel=result_rel, audit_source_rel=audit_rel, worker_id=args.worker_id, dispatch_kind=args.dispatch_kind, manifest=manifest, active_context=active_context, )) body = instruction_path.read_text(encoding="utf-8") is_reverify = is_verification_dispatch_kind(args.dispatch_kind) if is_reverify: body = _complete_run_reverify_body(args, manifest, active_context, assignment, body, instruction_path) if args.audience == "report-writer" and not getattr(args, "corrections", None): body = _with_inputs_section( body, _report_writer_input_lines( project_root, manifest, active_context, narrative_path=result_path, ), ) if args.audience == "report-writer" and uses_report_contract_v3(manifest): body = _with_report_writer_sections( args, project_root, manifest, authorized, active_context, body, narrative_path=result_path, ) elif getattr(args, "corrections", None): if args.audience != "report-writer": raise AgentPromptCliError( "--corrections is a report-writer materialization option; " f"audience {args.audience} has no corrections ledger" ) raise AgentPromptCliError( "--corrections requires report contract 3.0 (a Markdown narrative); " "this run uses an older contract" ) request = AgentInvocationRequest( invocation_id=args.invocation_id, worker_id=args.worker_id if identity is None else None, audience=args.audience, assignment_ref=args.assignment_ref, purpose=None, assignment=assignment, instruction=AgentInstruction( anchor_lines=anchor_lines, body=body, source_paths=(AgentInstructionSource( kind="project", path=_relative(project_root, instruction_path), ),) + ((AgentInstructionSource( kind="runtime", path="templates/reverify-output-contract.md", ),) if is_reverify else ()), ), project_root=project_root, run_manifest_path=manifest_path, duty_root=duty_root, prompt_path=prompt_path, metadata_path=prompt_path.with_name(prompt_path.name + ".meta.json"), dispatch_kind=args.dispatch_kind, participant_ref=identity.participant_ref if identity is not None else None, role_execution_ref=( identity.role_execution_ref if identity is not None else None ), duty_id=identity.duty_id if identity is not None else None, invocation_ref=args.invocation_id if identity is not None else None, attempt=1, replace_undispatched=args.replace_undispatched, ) if is_reverify: candidate = compose_unbound_run_prompt(request).decode("utf-8") if dynamic_source else compose_agent_prompt(request) errors = validate_reverify_prompt( candidate, task_type=str(manifest["taskType"]), forbidden_actions=str(active_context["workflow"]["forbiddenActions"]), expected_model=assignment.model_execution_value, dispatch_kind=args.dispatch_kind, ) if errors: raise AgentPromptCliError("; ".join(errors)) if dynamic_source is not None: request = _reserve_dynamic_verifier_request( request, args=args, manifest=manifest, manifest_path=manifest_path, source_role_execution_ref=dynamic_source, ) return prepare_agent_invocation(request) def _complete_run_reverify_body( args: argparse.Namespace, manifest: Mapping[str, Any], active_context: Mapping[str, Any], assignment: AgentModelAssignment, body: str, instruction_path: Path, ) -> str: workflow = _mapping(active_context.get("workflow"), "active workflow") forbidden = workflow.get("forbiddenActions") if not isinstance(forbidden, str) or not forbidden.strip(): raise AgentPromptCliError("active workflow.forbiddenActions is missing or invalid") resources = active_context.get("runtimeResources") or {} error_contract = resources.get("workerErrorContractPath") template = Path( error_contract or okstra_home() / "templates/worker-error-contract.md" ).with_name("reverify-output-contract.md") try: return complete_reverify_instruction( body, task_type=str(manifest["taskType"]), forbidden_actions=forbidden, model=assignment.model_execution_value, output_contract=template.read_text(encoding="utf-8"), ) except ValueError as exc: raise AgentPromptCliError( f"{exc}; correct {instruction_path} and rerun agent-prompt materialize " f"with --invocation-id {args.invocation_id} --prompt {args.prompt} " "(use a fresh ID and path if already dispatched or dynamically reserved)" ) from exc def _normalized(path: Path) -> Path: return Path(os.path.normpath(path)) def _with_report_writer_sections( args: argparse.Namespace, project_root: Path, manifest: Mapping[str, Any], authorized: Mapping[str, Any], active_context: Mapping[str, Any], body: str, *, narrative_path: Path, ) -> str: """run 갈래 report-writer 프롬프트의 okstra 소유 절을 본문 끝에 렌더한다. `## Output` 은 매 디스패치에 렌더한다 — 세 산출물을 리드가 손으로 열거하다 하나를 빠뜨리면 `required worker artifact was not produced` 로 끝났다. `--corrections` 가 있으면 원장을 대조해 `## Corrections` 를 그 앞에 둔다. 리드가 자유 서술로 적던 교정 지시는 기계가 대조할 수 없어 값 오류가 조립에서야 드러났다(2026-09-03 실측: 재실행 6회 중 4회). 그래서 원장 없는 교정 디스패치(서사가 이미 있고 파싱되는 경우)는 거절한다. 두 절은 okstra 가 렌더하므로 본문에 같은 제목이 있으면 거절한다. """ conflicts = body_owned_section_conflicts(body) if conflicts: raise AgentPromptCliError( f"instruction body must not contain {', '.join(conflicts)}: okstra " "renders those sections itself (## Output from the run manifest, " "## Corrections from the --corrections ledger, ## Previous Attempt " "from the narrative a re-authoring dispatch replaces)" ) sections: list[str] = [] if getattr(args, "corrections", None): corrections_path = _authorized_path( project_root, args.corrections, authorized.get("instructionRoots"), "corrections", must_exist=True, ) check = run_corrections_check( project_root=project_root, manifest=manifest, active_context=active_context, team_state=_load_team_state(project_root, manifest), corrections_path=corrections_path, narrative_path=narrative_path, ) if check.defects: raise AgentPromptCliError( "report-writer corrections defects: " + "; ".join(check.defects) ) sections.extend(render_corrections_section( check, base_narrative_rel=str(check.ledger.get("baseNarrativePath") or ""), corrections_rel=_relative(project_root, corrections_path), )) body = "## Inputs\n\nCorrection-only task: use the checked field values and evidence below." sections.extend(_render_correction_application(args, project_root, narrative_path, corrections_path, check)) sections.append("") else: _refuse_free_form_correction(project_root, narrative_path) preserved = _preserve_reauthored_narrative( project_root, narrative_path, invocation_id=str(args.invocation_id), ) if preserved is not None: sections.extend(_render_previous_attempt_section(preserved)) sections.append("") sections.extend(render_output_section()) return body.rstrip("\n") + "\n\n" + "\n".join(sections) + "\n" def _render_correction_application( args: argparse.Namespace, project_root: Path, narrative_path: Path, corrections_path: Path, check: CorrectionsCheck, ) -> list[str]: result_path = narrative_path.with_name(f"{narrative_path.stem}.{args.invocation_id}.rewrites.json") command = [ "okstra", "agent-prompt", "apply-corrections", "--project-root", str(project_root), "--run-manifest", str(args.run_manifest), "--corrections", str(corrections_path), ] sections: list[str] = [] if not check.mechanical: command.extend(["--rewrite-results", str(result_path)]) sections.extend([ "", f"Write only rewrite replacement values to `{_relative(project_root, result_path)}`:", '{"baseNarrativeSha256":"' + check.ledger["baseNarrativeSha256"] + '","replacements":[{"id":"RC-...","replacement":"the complete value for that field"}]}', "Include every rewrite id exactly once. Do not repeat unchanged narrative fields. " "Use the JSON value type required by each target's schema constraint.", ]) sections.extend([ "", "Run this command to validate all replacements and generate the complete narrative:", "```sh", shlex.join(command), "```", "The command reports every remaining schema and semantic defect together. " "Repair the replacement values and retry only when it reports a defect. " "Do not write the complete narrative yourself. Then write the pointer record and reading audit.", ]) return sections def _preserve_reauthored_narrative( project_root: Path, narrative_path: Path, *, invocation_id: str, ) -> str | None: """재저작 디스패치 전에 기존 서사를 사본으로 남기고 그 상대 경로를 돌려준다. 교정(ledger) 경로는 리드가 `baseNarrativePath` 사본을 만들지만, 구조가 안 읽히는 서사는 원장 없이 재저작으로 통과해 사본 요구가 없었다. 실측 (2026-09-09 dev-10642 requirements-discovery 001): 들여쓰기만 고치는 재저작에서 작성자의 변환 명령이 실패해 579줄 서사가 0 바이트로 덮였고, `.okstra` 는 gitignore 라 복구본이 없었다. 재저작은 live 파일을 제자리에서 덮어쓰므로 okstra 가 사본을 남긴다. 빈 파일은 남길 내용이 없어 건너뛴다. """ if not narrative_path.is_file() or narrative_path.stat().st_size == 0: return None copy_path = narrative_path.with_name( f"{narrative_path.stem}.pre-{invocation_id}{narrative_path.suffix}" ) if os.path.normpath(copy_path) == os.path.normpath(narrative_path): raise AgentPromptCliError("preserved narrative copy resolves to the live narrative") shutil.copyfile(narrative_path, copy_path) return _relative(project_root, copy_path) def _render_previous_attempt_section(preserved_rel: str) -> list[str]: return [ "## Previous Attempt", "", f"Your previous attempt is preserved at `{preserved_rel}`. Read that copy for " "its content and write the re-authored narrative to `**Result Path:**` as a " "fresh file. Do not transform the file at `**Result Path:**` in place with a " "shell or script command: a failed command leaves an empty file and the run " "loses the attempt. The preserved copy is read-only for you.", ] def _refuse_free_form_correction(project_root: Path, narrative_path: Path) -> None: """서사가 이미 있고 구조가 읽히면 이 디스패치는 교정이다 — 원장 없이는 거절한다. 구조가 읽히지 않는 서사(줄 문법·소유권 결함)는 교정 대상이 아니라 재저작 대상이다(원장의 경로가 해소될 자료가 없다). 그 디스패치는 원장 없이 통과하고, `_preserve_reauthored_narrative` 가 기존 파일의 사본을 남긴다. 값 결함(패턴 밖 id, enum 밖 값)은 구조가 읽히는 서사이고, 그것이 원장이 고치는 자리다 — 실측(dev-10626 a3)의 `SC-` id 20곳이 이 경우다. """ if not narrative_path.is_file(): return try: parse_narrative_structure( narrative_path.read_text(encoding="utf-8"), load_schema_version("3.0"), ) except NarrativeContractError: return narrative_rel = _relative(project_root, narrative_path) raise AgentPromptCliError( f"report-writer narrative already exists at {narrative_rel} and parses, " "so this is a corrective dispatch, and a free-form correction is refused: " "nothing can check it before the writer runs. Write a corrections ledger " "(schemas/report-writer-corrections-v1.0.schema.json; baseNarrativePath " "names a preserved copy of that attempt), run `okstra agent-prompt " "check-corrections --project-root --run-manifest " "--corrections ` until it reports no defect, then pass the same " "--corrections here. A ledger of only replace/remove/add/move entries needs no " "writer round: `okstra agent-prompt apply-corrections` writes the " "narrative and records the activity row. Only a narrative whose " "structure does not parse (line grammar, unknown top-level field) is " "re-authored without a ledger; value defects are what the ledger fixes" ) def _corrections_context( args: argparse.Namespace, ) -> tuple[Path, Path, dict[str, Any], Mapping[str, Any], Path]: """`check-corrections`·`apply-corrections` 가 공유하는 입력 해소.""" project_root = _project_root(args.project_root) manifest_path = _project_input(project_root, args.run_manifest, "run manifest") manifest = _read_json_object(manifest_path, "run manifest") contract = _mapping(manifest.get("agentContract"), "run agent contract") authorized = _mapping(contract.get("authorizedPaths"), "authorized paths") corrections_path = _authorized_path( project_root, args.corrections, authorized.get("instructionRoots"), "corrections", must_exist=True, ) active_context_path = _project_manifest_path( project_root, manifest.get("activeRunContextPath"), "active run context", must_exist=True, ) active_context = hydrate_active_run_context( _read_json_object(active_context_path, "active run context") ) return project_root, manifest_path, manifest, active_context, corrections_path def _check_corrections(args: argparse.Namespace) -> dict[str, Any]: """`okstra agent-prompt check-corrections` — 실체화 없이 원장만 대조한다.""" project_root, _manifest_path, manifest, active_context, corrections_path = ( _corrections_context(args) ) narrative_path = ( report_narrative_path(project_root, manifest) if uses_report_contract_v3(manifest) else project_root ) check = run_corrections_check( project_root=project_root, manifest=manifest, active_context=active_context, team_state=_load_team_state(project_root, manifest), corrections_path=corrections_path, narrative_path=narrative_path, ) return corrections_payload( check, project_root=project_root, corrections_path=corrections_path, ) def _apply_corrections(args: argparse.Namespace) -> dict[str, Any]: """`okstra agent-prompt apply-corrections` — 기계적 원장을 서사에 쓴다.""" project_root, manifest_path, manifest, active_context, corrections_path = ( _corrections_context(args) ) results_path = None if getattr(args, "rewrite_results", None): contract = _mapping(manifest.get("agentContract"), "agent contract") authorized = _mapping(contract.get("authorizedPaths"), "authorized paths") results_path = _authorized_path( project_root, args.rewrite_results, authorized.get("resultRoots"), "rewrite results", must_exist=True, ) return run_corrections_apply( project_root=project_root, manifest=manifest, manifest_path=manifest_path, active_context=active_context, team_state=_load_team_state(project_root, manifest), corrections_path=corrections_path, rewrite_results_path=results_path, ) def _validate_report_writer_paths( project_root: Path, manifest: Mapping[str, Any], *, worker_id: str, result_path: Path, audit_source_path: Path, ) -> None: """report-writer 의 두 경로를 run 이 이미 정한 값에 못 박는다. `--result` 는 조립이 읽는 서술문 경로다: 계약 3.0 이면 매니페스트의 `reportNarrativePath`, 2.0 이면 `expectedReportRecordPath` 의 data.json. 다른 경로에 쓴 서술문은 `report-finalize` 가 읽지 않는다. `--audit-source` 는 명부(team-state `workers[].resultPath`)의 워커 결과다: `okstra team await` 는 그 파일이 있어야 명부 행을 completed 로 적는다. 실측(2026-09-02, fontsninja-v3-site dev-10626-1 r04): 리드가 재시도마다 `-a2`/`-a3` 접미를 붙인 경로를 넘겨 서술문 세 벌이 조립 밖에 쌓였고, 결국 손으로 `cp` 했다. 교정 디스패치는 프롬프트 경로와 invocation ID 만 새로 하고 이 두 경로는 그대로 쓴다. """ if uses_report_contract_v3(manifest): expected_result = report_narrative_path(project_root, manifest) what = "the narrative report-finalize assembles from (run manifest reportNarrativePath)" else: expected_result = final_report_data_path(_project_manifest_path( project_root, manifest.get("expectedReportRecordPath"), "expected report record", must_exist=False, )) what = "the report data.json (run manifest expectedReportRecordPath)" if _normalized(result_path) != _normalized(expected_result): raise AgentPromptCliError( f"report-writer --result must be {what}: expected {expected_result}, " f"got {result_path}. A corrective dispatch reuses this path with a " "fresh prompt path and invocation ID" ) roster_result = _roster_result_path(project_root, manifest, worker_id) if ( roster_result is not None and _normalized(audit_source_path) != _normalized(roster_result) ): raise AgentPromptCliError( "report-writer --audit-source must be the roster's worker result " f"(team-state workers[].resultPath): expected {roster_result}, got " f"{audit_source_path}. `okstra team await` records the roster row " "completed only when that file exists" ) def _load_team_state(project_root: Path, manifest: Mapping[str, Any]) -> Mapping[str, Any]: """매니페스트가 가리키는 team-state. 없으면 빈 매핑.""" team_state_value = manifest.get("teamStatePath") if not isinstance(team_state_value, str) or not team_state_value: return {} team_state_path = Path(team_state_value) if not team_state_path.is_absolute(): team_state_path = project_root / team_state_path if not team_state_path.is_file(): return {} return _read_json_object(team_state_path, "team state") def _report_writer_input_lines( project_root: Path, manifest: Mapping[str, Any], active_context: Mapping[str, Any], *, narrative_path: Path, ) -> list[str]: """report-writer 의 `## Inputs` 줄. 로스터 경로(`worker_prompt_body`)와 같은 규칙. 계약 3.0 이면 합성 묶음을 여기서 만들고 그 한 줄을 싣는다; 그 전 계약은 옛 목록이다. 실측(2026-09-02, fontsninja-v3-site dev-10626-1 r04): run 갈래의 프롬프트에는 이 절이 없어, 작성자가 "읽을 게 없다"며 서술문을 거부했고 리드가 경로 22개를 손으로 열거하고서야 통과했다. 두 갈래가 같은 입력 표면을 내야 하는 이유다. """ team_state = _load_team_state(project_root, manifest) if not uses_report_contract_v3(manifest): return report_writer_input_lines(manifest, active_context, team_state) try: _, markdown_path = materialize_report_synthesis_packet( project_root=project_root, manifest=manifest, active_context=active_context, team_state=team_state, narrative_path=narrative_path, ) except ReportSynthesisPacketError as exc: defects = "; ".join( f"owner={issue.owner} source={issue.label} path={issue.path} " f"reason={issue.reason}" for issue in exc.issues ) raise AgentPromptCliError( f"report synthesis packet contract defects: {defects}" ) from exc return [f"- Report synthesis packet: `{_relative(project_root, markdown_path)}`"] def _with_inputs_section(body: str, input_lines: list[str]) -> str: """런타임 소유 입력 줄을 본문의 `## Inputs` 첫머리에 넣는다. 리드가 이미 그 절을 썼으면 그 제목 바로 아래에 끼워 한 절로 두고, 없으면 머리의 `**…:**` 헤더 블록 뒤에 절을 새로 연다. """ lines = body.splitlines() for index, line in enumerate(lines): if line.strip() == "## Inputs": merged = lines[: index + 1] + input_lines + lines[index + 1 :] return "\n".join(merged) + "\n" insert_at = 0 while insert_at < len(lines) and ( not lines[insert_at].strip() or lines[insert_at].startswith("**") ): insert_at += 1 section = ["## Inputs", *input_lines, ""] if insert_at and lines[insert_at - 1].strip(): section = ["", *section] return "\n".join(lines[:insert_at] + section + lines[insert_at:]) + "\n" def _roster_result_path( project_root: Path, manifest: Mapping[str, Any], worker_id: str, ) -> Path | None: """명부가 이 워커에 적어 둔 `resultPath`. 행이 없으면 None.""" workers = _load_team_state(project_root, manifest).get("workers") if not isinstance(workers, list): return None for row in workers: if not isinstance(row, Mapping) or row.get("workerId") != worker_id: continue value = row.get("resultPath") if not isinstance(value, str) or not value.strip(): return None path = Path(value.strip()) return path if path.is_absolute() else project_root / path return None def _materialize_standalone( args: argparse.Namespace, project_root: Path, ) -> PreparedAgentInvocation: required = { "host-runtime": args.host_runtime, "provider": args.provider, "model-role": args.model_role, "purpose": args.purpose, } missing = [name for name, value in required.items() if not value] if missing: raise AgentPromptCliError( "standalone materialization requires: " + ", ".join("--" + name for name in missing) ) forbidden = { "worker-id": args.worker_id, "dispatch-kind": args.dispatch_kind, "assignment-ref": args.assignment_ref, "result": args.result, "audit-source": args.audit_source, "source-role-execution-ref": args.source_role_execution_ref, } present = [name for name, value in forbidden.items() if value] if present: raise AgentPromptCliError( "standalone materialization forbids: " + ", ".join("--" + name for name in present) ) if not _is_slug(args.purpose) or not _is_slug(args.invocation_id): raise AgentPromptCliError("standalone purpose and invocation ID must be slugs") if args.audience not in _AUDIENCES: raise AgentPromptCliError(f"unknown duty audience: {args.audience}") root = project_root / ".okstra" / "agent-invocations" / args.purpose if not root.is_dir(): raise AgentPromptCliError(f"standalone invocation root not found: {root}") resolved_root = root.resolve(strict=True) if not _is_relative_to(resolved_root, project_root): raise AgentPromptCliError("standalone invocation root escapes project root") root = resolved_root expected_instruction = root / f"{args.invocation_id}.instructions.md" expected_prompt = root / f"{args.invocation_id}.prompt.md" instruction_path = _exact_path(args.instruction, expected_instruction, True, "instruction") prompt_path = _exact_path(args.prompt, expected_prompt, False, "prompt") assignment_context = load_assignment_context( host_runtime=args.host_runtime, terminal_backend=args.terminal_backend or detect_terminal_backend(), execution_provider_ids=(args.provider,), include_native_provider=False, ) assignment = _standalone_assignment( context=assignment_context, provider=args.provider, model_role=args.model_role, model=args.model, ) duty_root = root / f"{args.invocation_id}.duty-contracts" _snapshot_standalone_duties(duty_root) return prepare_agent_invocation(AgentInvocationRequest( invocation_id=args.invocation_id, worker_id=None, audience=args.audience, assignment_ref=None, purpose=args.purpose, assignment=assignment, instruction=AgentInstruction( anchor_lines=(), body=instruction_path.read_text(encoding="utf-8"), source_paths=(AgentInstructionSource( kind="project", path=_relative(project_root, instruction_path), ),), ), project_root=project_root, run_manifest_path=None, duty_root=duty_root, prompt_path=prompt_path, metadata_path=prompt_path.with_name(prompt_path.name + ".meta.json"), dispatch_kind="standalone", participant_ref=None, role_execution_ref=None, duty_id=None, invocation_ref=None, attempt=1, )) def _standalone_assignment( *, context: AssignmentContext, provider: str, model_role: str, model: str, ) -> AgentModelAssignment: host_runtime = context.environment.host_descriptor.id resolved = resolve_dispatch_assignment( context=context, host_runtime=host_runtime, role=model_role, duty_id=model_role, provider=provider, model=model, ) binding = resolved.binding if binding is None: raise AgentPromptCliError("standalone assignment requires a model binding") return AgentModelAssignment( provider=resolved.provider_id, model=resolved.display_name, model_execution_value=binding.resolved_execution_value, runner=binding.runner, host_runtime=host_runtime, host_model_value=binding.host_model_value, ) def _snapshot_standalone_duties(destination: Path) -> None: # 부모 개수를 세면 체크아웃(`/scripts/okstra_ctl/agent/prompt_cli/`) # 에서만 맞는다 — 설치본은 패키지가 `~/.okstra/lib/python/` 이라 prompts 가 # 한 단계 위(`~/.okstra/prompts/`)에 있다. duties 를 실제로 가진 루트를 # 탐색해서 해소한다. from ...paths import find_asset_root duties_relative = ("prompts", "duties") root = find_asset_root(duties_relative, is_present=Path.is_dir) if root is None: raise AgentPromptCliError( "duty contract source not found: no prompts/duties under " "OKSTRA_HOME or this checkout" ) source = root.joinpath(*duties_relative) if destination.exists(): if not destination.is_dir() or digest_duty_catalog(destination) != digest_duty_catalog(source): raise AgentPromptCliError("standalone duty snapshot conflicts with existing files") return temporary = Path(tempfile.mkdtemp(prefix=f".{destination.name}.", dir=destination.parent)) try: shutil.copytree(source, temporary, dirs_exist_ok=True) try: os.rename(temporary, destination) except FileExistsError: if digest_duty_catalog(destination) != digest_duty_catalog(source): raise AgentPromptCliError("standalone duty snapshot publication conflict") finally: if temporary.exists(): shutil.rmtree(temporary)