"""Materialize and verify one auditable LLM invocation specification.""" from __future__ import annotations import argparse from dataclasses import replace from datetime import datetime, timezone import hashlib import json import os from pathlib import Path from pathlib import PurePosixPath import shutil import sys import tempfile from typing import Any, Mapping, get_args from .agent_invocation import ( AgentInstruction, AgentInstructionSource, AgentAudience, AgentInvocationError, AgentInvocationRequest, AgentModelAssignment, PreparedAgentInvocation, agent_model_assignment_from_payload, compose_agent_prompt, compose_unbound_run_prompt, digest_duty_catalog, invocation_execution_identity_from_manifest, materialize_standalone_result, prepare_agent_invocation, publish_standalone_completion, verify_standalone_completion, verify_agent_invocation, v2_role_assignment_authority_errors, ) from .fixed_text import line from .json_boundary import JsonBoundaryError, load_owned_object from .convergence_store import ( DYNAMIC_VERIFIER_SOURCE_ROLES, reserve_dynamic_verifier, ) 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 WorkerPromptHeaderError, worker_prompt_headers from .worker_artifact_paths import audit_sidecar_rel from .wrapper_status import ( log_path_for_prompt, prompt_derived_paths, status_path_for_prompt, ) from .report_inputs import report_narrative_path, uses_report_contract_v3 from .worker_prompt_policy import ( CRITIC_DUTY_BY_ASSIGNMENT_SEGMENT, resolve_prompt_plan_for_manifest, ) from .dispatch_state import ( BACKEND_CLI_WRAPPER, BACKEND_CMUX_PANE, DispatchError, detect_terminal_backend, link_agent_dispatch_result, record_verified_agent_dispatch, reject_agent_dispatch_result, ) _AUDIENCES = frozenset(get_args(AgentAudience)) class AgentPromptCliError(RuntimeError): """Raised when CLI input cannot become a valid invocation request.""" def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="okstra agent-prompt") commands = parser.add_subparsers(dest="command", required=True) _add_materialize_parser(commands) _add_completion_parsers(commands) _add_dispatch_parsers(commands) return parser def _add_materialize_parser(commands: argparse._SubParsersAction) -> None: materialize = commands.add_parser("materialize") _common_paths(materialize) materialize.add_argument("--invocation-id", required=True) materialize.add_argument("--audience", required=True) materialize.add_argument("--instruction", required=True) materialize.add_argument("--prompt", required=True) materialize.add_argument("--run-manifest") materialize.add_argument("--worker-id") materialize.add_argument("--dispatch-kind") materialize.add_argument("--assignment-ref") materialize.add_argument("--source-role-execution-ref") materialize.add_argument("--result") materialize.add_argument("--audit-source") materialize.add_argument("--host-runtime") materialize.add_argument( "--terminal-backend", choices=(BACKEND_CLI_WRAPPER, BACKEND_CMUX_PANE), ) materialize.add_argument("--provider") materialize.add_argument("--model-role") materialize.add_argument("--model", default="") materialize.add_argument("--purpose") materialize.add_argument( "--replace-undispatched", action="store_true", help="rewrite a prompt this invocation id already wrote, allowed only " "while no dispatch has referenced it — the exit for a prompt that " "failed a pre-dispatch gate and never ran. Not available for v2 " "reverify prompts (--assignment-ref reverify/…): their reservation " "records the prompt's input digest in the run manifest and " "reservations are append-only, so a rewrite needs a fresh " "invocation id and prompt path", ) materialize.add_argument("--json", action="store_true") verify = commands.add_parser("verify") _common_paths(verify) verify.add_argument("--run-manifest") verify.add_argument("--metadata", required=True) verify.add_argument("--json", action="store_true") verify.add_argument("--text", action="store_true") def _add_completion_parsers(commands: argparse._SubParsersAction) -> None: materialize_result = commands.add_parser("materialize-result") _standalone_completion_args(materialize_result) materialize_result.add_argument("--returned-body-file", required=True) complete = commands.add_parser("complete") _standalone_completion_args(complete) verify_completion = commands.add_parser("verify-completion") _common_paths(verify_completion) verify_completion.add_argument("--purpose", required=True) verify_completion.add_argument("--completion", required=True) verify_completion.add_argument("--json", action="store_true") def _add_dispatch_parsers(commands: argparse._SubParsersAction) -> None: record_dispatch = commands.add_parser("record-dispatch") _common_paths(record_dispatch) record_dispatch.add_argument("--run-manifest", required=True) record_dispatch.add_argument("--metadata", required=True) record_dispatch.add_argument( "--enforcement-mode", required=True, choices=("core-pre-dispatch", "host-native-spec-link-gate"), ) record_dispatch.add_argument("--json", action="store_true") reject_result = commands.add_parser( "reject-result", help="mark a linked result rejected so a corrective re-dispatch can " "claim its path", ) _common_paths(reject_result) reject_result.add_argument("--run-manifest", required=True) reject_result.add_argument("--dispatch-id", required=True) reject_result.add_argument("--superseded-by", required=True) reject_result.add_argument("--reason", required=True) reject_result.add_argument("--json", action="store_true") link_result = commands.add_parser("link-result") _common_paths(link_result) link_result.add_argument("--run-manifest", required=True) link_result.add_argument("--dispatch-id", required=True) link_result.add_argument("--result", required=True) link_result.add_argument("--json", action="store_true") def _common_paths(parser: argparse.ArgumentParser) -> None: parser.add_argument("--project-root", required=True) def _standalone_completion_args(parser: argparse.ArgumentParser) -> None: _common_paths(parser) parser.add_argument("--purpose", required=True) parser.add_argument("--metadata", required=True) parser.add_argument("--json", action="store_true") def main(argv: list[str] | None = None) -> int: try: args = _parser().parse_args(argv) if args.command == "materialize": prepared = _materialize(args) _emit(_prepared_payload(prepared, args.audience), args.json) return 0 if args.command == "verify": _verify(args) _emit( {"ok": True, "metadataPath": str(Path(args.metadata).resolve())}, args.json, args.text, ) return 0 if args.command == "materialize-result": _materialize_result(args) return 0 if args.command == "complete": _complete(args) return 0 if args.command == "verify-completion": _verify_completion(args) return 0 if args.command == "record-dispatch": _record_dispatch(args) return 0 if args.command == "reject-result": _reject_result(args) return 0 _link_result(args) return 0 except ( AgentPromptCliError, AgentInvocationError, DispatchError, WorkerPromptHeaderError, OSError, ValueError, ) as exc: print(f"error: {exc}", file=sys.stderr) return 2 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 _record_dispatch(args: argparse.Namespace) -> None: project_root = _project_root(args.project_root) record = record_verified_agent_dispatch( project_root=project_root, run_manifest_path=_project_input( project_root, args.run_manifest, "run manifest" ), metadata_path=_project_input( project_root, args.metadata, "agent invocation metadata" ), enforcement_mode=args.enforcement_mode, ) _emit(record, args.json) def _reject_result(args: argparse.Namespace) -> None: project_root = _project_root(args.project_root) row = reject_agent_dispatch_result( project_root=project_root, run_manifest_path=_project_input( project_root, args.run_manifest, "run manifest" ), dispatch_id=args.dispatch_id, superseded_by=args.superseded_by, reason=args.reason, ) _emit(row, args.json) def _link_result(args: argparse.Namespace) -> None: project_root = _project_root(args.project_root) link = link_agent_dispatch_result( project_root=project_root, run_manifest_path=_project_input( project_root, args.run_manifest, "run manifest" ), dispatch_id=args.dispatch_id, result_path=_project_input(project_root, args.result, "agent result"), ) _emit(link, args.json) 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, ) if args.audience == "report-writer" and not args.audit_source: # 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: under " f"report contract 3.0 --result is the narrative ({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" ) audit_source_path = ( _authorized_path( project_root, args.audit_source, authorized.get("resultRoots"), "audit source", must_exist=False, ) if args.audit_source else result_path ) _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: raise AgentPromptCliError( f"assignment reference is missing from run manifest: {args.assignment_ref}" ) 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, )) 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=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=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 dynamic_source is not None: request = _reserve_dynamic_verifier_request( request, args=args, manifest=manifest, manifest_path=manifest_path, active_context=active_context, source_role_execution_ref=dynamic_source, artifact_paths=_dynamic_verifier_artifact_paths( project_root, prompt_path, result_path, audit_source_path, audit_rel, active_context, args.worker_id, ), ) return prepare_agent_invocation(request) def _run_worktree( manifest: Mapping[str, Any], active_context: Mapping[str, Any] ) -> Path | None: """이 런의 워커 루트. 디스패치가 job 에 싣는 값과 같은 seam 에서 읽는다.""" from .dispatch_state import worktree_path value = worktree_path(manifest, active_context) return Path(value) if value else None def _reserve_dynamic_verifier_request( request: AgentInvocationRequest, *, args: argparse.Namespace, manifest: Mapping[str, Any], manifest_path: Path, active_context: Mapping[str, Any], source_role_execution_ref: str, artifact_paths: tuple[Path, ...], ) -> AgentInvocationRequest: _validate_dynamic_source_assignment( manifest, source_role_execution_ref, request.assignment, ) prompt_bytes = compose_unbound_run_prompt(request) verifier, invocation = reserve_dynamic_verifier( manifest_path, source_role_execution_ref=source_role_execution_ref, duty_id=args.audience, round_number=_reverify_round(args.dispatch_kind), task_key=_required_manifest_string(manifest, "taskKey"), input_digest="sha256:" + hashlib.sha256(prompt_bytes).hexdigest(), invocation_ref=args.invocation_id, artifact_paths=artifact_paths, worktree=_run_worktree(manifest, active_context), ) bound = replace( request, worker_id=None, participant_ref=verifier.participant_ref, role_execution_ref=verifier.role_execution_ref, duty_id=args.audience, invocation_ref=invocation.invocation_ref, ) if compose_agent_prompt(bound).encode("utf-8") != prompt_bytes: raise AgentPromptCliError( "dynamic verifier identity changed immutable prompt bytes" ) return bound def _dynamic_verifier_artifact_paths( project_root: Path, prompt_path: Path, result_path: Path, audit_source_path: Path, audit_rel: str, active_context: Mapping[str, Any], worker_id: str, ) -> tuple[Path, ...]: paths = { result_path, audit_source_path, project_root / audit_sidecar_rel(audit_rel), *prompt_derived_paths(prompt_path), } error_logs = active_context.get("errorLogs") sidecars = error_logs.get("sidecarsByWorkerId") if isinstance(error_logs, Mapping) else None value = sidecars.get(worker_id) if isinstance(sidecars, Mapping) else None if isinstance(value, str) and value: path = Path(value) paths.add(path if path.is_absolute() else project_root / path) return tuple(sorted(paths, key=str)) def _dynamic_verifier_source( args: argparse.Namespace, manifest: Mapping[str, Any], ) -> str | None: is_v2 = ( manifest.get("schemaVersion") == "2.0" and manifest.get("executionIdentityVersion") == 2 ) is_reverify = ( args.assignment_ref.startswith("reverify/") and args.audience == "reverification-worker" ) source = args.source_role_execution_ref if is_v2 and is_reverify: if args.replace_undispatched: raise AgentPromptCliError( "v2 dynamic verifier prompts are append-only: this " "invocation's reservation already recorded this prompt's input " "digest in the run manifest, and a reservation cannot be " "rewritten. Use a fresh invocation ID and prompt path." ) if not source: raise AgentPromptCliError( "v2 reverify materialization requires " "--source-role-execution-ref" ) return source if source: raise AgentPromptCliError( "--source-role-execution-ref is allowed only for v2 reverify" ) return None def _reverify_round(dispatch_kind: str) -> int: prefix = "reverify-r" value = dispatch_kind.removeprefix(prefix) if not dispatch_kind.startswith(prefix) or not value.isdigit() or int(value) < 1: raise AgentPromptCliError( "dynamic verifier dispatch kind must be reverify-r" ) return int(value) def _validate_dynamic_source_assignment( manifest: Mapping[str, Any], source_role_execution_ref: str, assignment: AgentModelAssignment, ) -> None: rows = manifest.get("roleExecutions") source = next( ( row for row in rows if isinstance(row, Mapping) and row.get("roleExecutionRef") == source_role_execution_ref ), None, ) if isinstance(rows, list) else None if source is None: raise AgentPromptCliError( f"source role execution is unknown: {source_role_execution_ref}" ) if source.get("role") not in DYNAMIC_VERIFIER_SOURCE_ROLES: raise AgentPromptCliError( "dynamic verifier source role is not eligible for re-verification" ) errors = v2_role_assignment_authority_errors( manifest, role_execution_ref=source_role_execution_ref, participant_ref=str(source.get("participantRef") or ""), assignment=assignment, ) if errors: raise AgentPromptCliError( "dynamic verifier source does not match the selected assignment: " + "; ".join(errors) ) def _required_manifest_string( manifest: Mapping[str, Any], key: str, ) -> str: value = manifest.get(key) if not isinstance(value, str) or not value.strip(): raise AgentPromptCliError(f"run manifest has no {key}") return value def _validate_run_identity( manifest: Mapping[str, Any], *, worker_id: str, dispatch_kind: str, assignment_ref: str, audience: str, ) -> None: expected: str if assignment_ref == "lead": expected = "lead" if worker_id != "lead": raise AgentPromptCliError("lead assignment requires worker ID 'lead'") elif assignment_ref == "translator": expected = "translator" if worker_id != "translator": raise AgentPromptCliError("translator assignment requires worker ID 'translator'") elif assignment_ref.startswith("critic/"): scope = assignment_ref.split("/", 1)[1] expected = CRITIC_DUTY_BY_ASSIGNMENT_SEGMENT.get(scope, "") if not expected or dispatch_kind != "critic": raise AgentPromptCliError("critic assignment identity is invalid") # provider 대조는 여기 두지 않는다. critic 의 `worker_id` 는 배정 참조의 # 마지막 마디(`scope` / `acceptance`)이고 provider 이름이 아니다 # (`render.py` 의 로스터 행이 `assignment_ref.rsplit("/", 1)[-1]` 로 만든다). # 종전에 있던 `provider != worker_id` 검사는 두 어휘를 비교해서 늘 참이 되는 # 형태였고, 설정을 run manifest 에서 찾다 못 찾아 조용히 건너뛰었기에 # 드러나지 않았다. 정본에서 읽게 고치면 모든 critic 디스패치가 거부된다. else: try: expected = resolve_prompt_plan_for_manifest( manifest=manifest, worker_id=worker_id, dispatch_kind=dispatch_kind, ).duty_audience except ValueError as exc: raise AgentPromptCliError(str(exc)) from exc if audience != expected and _manifest_issued_role_execution( manifest, audience=audience, assignment_ref=assignment_ref ): # The prompt plan names a worker's role by worker id, so on an # implementation run the executor's provider resolves to the # executor role no matter which role execution is being targeted. # But the manifest issues one role execution per role, and a # provider serving as both executor and verifier gets two — a # normal roster, since the verifier contract accepts reusing the # executor's model behind its own session. Refusing the second # audience made a role execution okstra had itself issued # undispatchable, which costs the run an independent verifier. expected = audience if audience != expected: raise AgentPromptCliError( f"audience {audience!r} does not match required audience {expected!r}" ) def _manifest_issued_role_execution( manifest: Mapping[str, Any], *, audience: str, assignment_ref: str, ) -> bool: """Whether this run issued a role execution for that audience's role. Keyed on the manifest's own rows, not on what the prompt plan infers from a worker id: the question is whether okstra created the execution being targeted, and only the manifest answers that. """ from .domain.role import RoleCatalogError, role_for_duty try: role = role_for_duty(audience) except RoleCatalogError: return False assignments = manifest.get("invocationAssignments") assignment = ( assignments.get(assignment_ref) if isinstance(assignments, Mapping) else None ) provider = ( assignment.get("provider") if isinstance(assignment, Mapping) else None ) if not provider: return False executions = manifest.get("roleExecutions") return any( isinstance(row, Mapping) and row.get("role") == role and row.get("provider") == provider for row in (executions if isinstance(executions, list) else []) ) 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: source = Path(__file__).resolve().parents[2] / "prompts" / "duties" if not source.is_dir(): raise AgentPromptCliError(f"duty contract source not found: {source}") 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) def _verify(args: argparse.Namespace) -> None: project_root = _project_root(args.project_root) metadata_path = _project_input(project_root, args.metadata, "metadata") metadata = _read_json_object(metadata_path, "metadata") contract_source = _mapping(metadata.get("contractSource"), "contract source") mode = contract_source.get("mode") expected_manifest: Path | None = None expected_assignment: AgentModelAssignment | None = None if mode == "run": if not args.run_manifest: raise AgentPromptCliError("run invocation verification requires --run-manifest") expected_manifest = _project_input(project_root, args.run_manifest, "run manifest") manifest = _read_json_object(expected_manifest, "run manifest") assignments = _mapping(manifest.get("invocationAssignments"), "run assignments") reference = metadata.get("assignmentRef") if not isinstance(reference, str) or reference not in assignments: raise AgentPromptCliError("assignment reference is missing from run manifest") expected_assignment = agent_model_assignment_from_payload(assignments[reference]) elif mode == "standalone": if args.run_manifest: raise AgentPromptCliError("standalone invocation verification forbids --run-manifest") else: raise AgentPromptCliError("contract source mode is invalid") errors = verify_agent_invocation( metadata_path, project_root=project_root, expected_run_manifest_path=expected_manifest, expected_assignment=expected_assignment, ) if errors: raise AgentPromptCliError("; ".join(errors)) def _materialize_result(args: argparse.Namespace) -> None: project_root = _project_root(args.project_root) metadata_path = _project_input(project_root, args.metadata, "metadata") root = project_root / ".okstra" / "agent-invocations" / args.purpose temporary_root = root / ".tmp" returned_body_path = _project_input( project_root, args.returned_body_file, "returned body", ) if not temporary_root.is_dir() or not _is_relative_to( returned_body_path, temporary_root.resolve(strict=True), ): raise AgentPromptCliError( "returned body file must be inside the standalone purpose .tmp directory" ) result = materialize_standalone_result( project_root=project_root, purpose=args.purpose, metadata_path=metadata_path, returned_body=returned_body_path.read_bytes(), ) _emit({ "ok": True, "purpose": result.purpose, "invocationId": result.invocation_id, "metadataPath": str(result.prompt_metadata_path), "resultPath": str(result.result_path), "resultEnvelopeDigest": result.result_envelope_digest, }, args.json) def _complete(args: argparse.Namespace) -> None: project_root = _project_root(args.project_root) metadata_path = _project_input(project_root, args.metadata, "metadata") completion = publish_standalone_completion( project_root=project_root, purpose=args.purpose, metadata_path=metadata_path, completed_at=datetime.now(timezone.utc), ) _emit({ "ok": True, "purpose": args.purpose, "metadataPath": str(metadata_path), "completionPath": str(completion), }, args.json) def _verify_completion(args: argparse.Namespace) -> None: project_root = _project_root(args.project_root) completion_path = _project_input( project_root, args.completion, "completion", ) verified = verify_standalone_completion( completion_path, project_root=project_root, expected_purpose=args.purpose, ) _emit({ "ok": True, "purpose": verified.purpose, "invocationId": verified.invocation_id, "metadataPath": str(verified.prompt_metadata_path), "resultPath": str(verified.result_path), "completionPath": str(verified.completion_path), "resultEnvelopeDigest": verified.result_envelope_digest, "returnedBody": verified.returned_body, }, args.json) def _authorized_path( project_root: Path, raw_path: str, raw_roots: object, label: str, *, must_exist: bool, ) -> Path: if not isinstance(raw_roots, list) or not raw_roots: raise AgentPromptCliError(f"run manifest has no authorized {label} roots") path = _candidate(project_root, raw_path, must_exist=must_exist, label=label) roots = [ _project_manifest_path(project_root, value, f"{label} root", must_exist=True) for value in raw_roots ] if not any(_is_relative_to(path, root) for root in roots): raise AgentPromptCliError(f"{label} path is outside authorized roots: {path}") return path def _candidate(project_root: Path, raw: str, *, must_exist: bool, label: str) -> Path: candidate = Path(raw) if not candidate.is_absolute(): candidate = project_root / candidate try: if must_exist or candidate.exists() or candidate.is_symlink(): return candidate.resolve(strict=True) return candidate.parent.resolve(strict=True) / candidate.name except OSError as exc: raise AgentPromptCliError(f"{label} path is invalid: {candidate}") from exc def _project_input(project_root: Path, raw: str, label: str) -> Path: path = _candidate(project_root, raw, must_exist=True, label=label) if not _is_relative_to(path, project_root): raise AgentPromptCliError(f"{label} path escapes project root: {path}") return path def _project_manifest_path( project_root: Path, value: object, label: str, *, must_exist: bool, ) -> Path: if not isinstance(value, str) or not value: raise AgentPromptCliError(f"{label} path is invalid") pure = PurePosixPath(value) if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts): raise AgentPromptCliError(f"{label} path is invalid") path = _candidate(project_root, value, must_exist=must_exist, label=label) if not _is_relative_to(path, project_root): raise AgentPromptCliError(f"{label} path escapes project root: {path}") return path def _exact_path(raw: str, expected: Path, must_exist: bool, label: str) -> Path: actual = ( Path(raw).resolve(strict=True) if must_exist else Path(raw).parent.resolve(strict=True) / Path(raw).name ) expected_actual = ( expected.resolve(strict=True) if must_exist else expected.parent.resolve(strict=True) / expected.name ) if actual != expected_actual: raise AgentPromptCliError(f"{label} does not use canonical standalone path") return actual def _project_root(value: str) -> Path: try: return Path(value).resolve(strict=True) except OSError as exc: raise AgentPromptCliError(f"project root not found: {value}") from exc def _relative(project_root: Path, path: Path) -> str: try: return path.relative_to(project_root).as_posix() except ValueError as exc: raise AgentPromptCliError(f"path escapes project root: {path}") from exc def _read_json_object(path: Path, label: str) -> dict[str, Any]: try: value = load_owned_object(path, artifact="agent prompt metadata") except JsonBoundaryError as exc: raise AgentPromptCliError(f"{label} is invalid: {path}") from exc if not isinstance(value, dict): raise AgentPromptCliError(f"{label} must be a JSON object: {path}") return value def _mapping(value: object, label: str) -> Mapping[str, Any]: if not isinstance(value, Mapping): raise AgentPromptCliError(f"{label} is missing or invalid") return value def _is_relative_to(path: Path, root: Path) -> bool: try: path.relative_to(root) except ValueError: return False return True def _is_slug(value: str) -> bool: if not value or value.startswith("-") or value.endswith("-"): return False return all(part.isalnum() and part.lower() == part for part in value.split("-")) def _prepared_payload( prepared: PreparedAgentInvocation, audience: str, ) -> dict[str, object]: assignment = prepared.assignment return { "ok": True, "invocationId": prepared.invocation_id, "workerId": prepared.worker_id, "assignmentRef": prepared.assignment_ref, "audience": audience, "promptPath": str(prepared.prompt_path), "metadataPath": str(prepared.metadata_path), "modelAssignment": { "provider": assignment.provider, "model": assignment.model, "modelExecutionValue": assignment.model_execution_value, "runner": assignment.runner, "hostRuntime": assignment.host_runtime, "hostModelValue": assignment.host_model_value, }, "digests": { "catalogDigest": prepared.catalog_digest, "assignmentDigest": prepared.assignment_digest, "dutyDigest": prepared.duty_digest, "instructionDigest": prepared.instruction_digest, "promptDigest": prepared.prompt_digest, }, } def _emit( payload: Mapping[str, object], as_json: bool, as_text: bool = False ) -> None: if as_json: print(json.dumps(payload, ensure_ascii=False, indent=2)) elif as_text: print(render_agent_prompt_text(payload), end="") else: print(payload.get("promptPath") or payload.get("metadataPath") or "ok") def render_agent_prompt_text(payload: Mapping[str, object]) -> str: """agent prompt 검증 결과의 승인된 경로만 투영한다.""" rows = ["Okstra agent prompt\n"] rows.append(line("Status", "ready" if payload.get("ok", True) else "error")) for label, key in (("Prompt path", "promptPath"), ("Metadata path", "metadataPath"), ("Assignment ID", "assignmentId"), ("Worker role", "workerRole")): if key in payload: rows.append(line(label, payload.get(key))) return "".join(rows) if __name__ == "__main__": raise SystemExit(main())