"""Render, validate, and immutably publish initial worker prompts."""
from __future__ import annotations

import json
import os
import re
import tempfile
from dataclasses import dataclass, replace
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Literal, Mapping, Sequence

from .final_report_paths import final_report_data_path
from .agent.invocation import (
    AgentInstruction,
    AgentInstructionSource,
    AgentInvocationError,
    AgentInvocationRequest,
    EXECUTION_DELIVERY_CONTRACT,
    agent_model_assignment_from_payload,
    compose_agent_prompt,
    invocation_execution_identity_from_manifest,
    invocation_dispatch_ids,
    prepare_agent_invocation,
    verify_agent_invocation,
)
from .path_hints import hydrate_active_run_context
from .report_language import resolve_report_language
from .report_inputs import report_narrative_path, uses_report_contract_v3
from .report_synthesis_packet import (
    ReportSynthesisPacketError,
    materialize_report_synthesis_packet,
    report_synthesis_packet_paths,
)
from .json_boundary import JsonBoundaryError, load_owned_object
from .worker_prompt_body import (
    REPORT_WRITER_WORKER_ID,
    analysis_prompt_body,
    instruction_path,
    report_writer_prompt_body,
)
from .worker_prompt_contract import PromptRecord, validate_initial_prompt_records
from .worker_prompt_headers import source_worktree_headers, worker_prompt_headers
from .worker_prompt_policy import (
    APPROVED_PLAN_HEADER,
    IMPLEMENTATION_STAGE_HEADER,
    PromptPlan,
    resolve_prompt_plan_for_manifest,
)


class PromptDeliveryMode(str, Enum):
    EAGER_INCLUDE = "eager-include"
    LAZY_PATH_REFERENCE = "lazy-path-reference"


MaterializationReason = Literal[
    "existing_prompt_invalid",
    "existing_prompt_conflict",
    "required_input_missing",
    "render_failed",
    "validation_failed",
    "publication_failed",
]


@dataclass(frozen=True)
class InitialPromptWorkerRequest:
    worker_id: str
    model: str


@dataclass(frozen=True)
class InitialPromptMaterializationRequest:
    project_root: Path
    run_manifest_path: Path
    runtime_root: Path
    delivery_mode: PromptDeliveryMode
    workers: tuple[InitialPromptWorkerRequest, ...]

    def __post_init__(self) -> None:
        # A `str` in any of these surfaced as `'str' object has no attribute
        # 'resolve'`, wrapped into `render_failed`, which named neither the
        # field nor the caller. Coercing removes the failure rather than
        # improving its message.
        for field in ("project_root", "run_manifest_path", "runtime_root"):
            object.__setattr__(self, field, Path(getattr(self, field)))


class InitialPromptMaterializationError(RuntimeError):
    reason: MaterializationReason

    def __init__(self, reason: MaterializationReason, message: str) -> None:
        super().__init__(message)
        self.reason = reason


MaterializeInitialPrompts = Callable[
    [InitialPromptMaterializationRequest],
    dict[str, Path],
]

_MATERIALIZABLE_AUDIENCES = frozenset({
    "analysis",
    "implementation-executor",
    "implementation-verifier",
    "report-writer",
})
_CLARIFICATION_INPUT_PREFIX = "- Clarification response:"
# Both sides of a stage judge against the same answers. The verifier used to
# get only the path under `## Inputs`, so whether it read the override was left
# to the worker: one provider read it and judged on the replacement command,
# another did not and reported the superseded plan's command as a stage defect.
_CLARIFICATION_AUTHORITY_AUDIENCES = frozenset({
    "implementation-executor",
    "implementation-verifier",
})
_CLARIFICATION_AUTHORITY_HEADING = (
    "# Clarification answers carried in (authoritative)"
)
_CLARIFICATION_AUTHORITY_HEADING_PREFIX = (
    "# Clarification answers carried in"
)
_CLARIFICATION_AUTHORITY_INTRO = (
    "The user answered these before this run. Where an answer conflicts "
    "with the approved plan text, the answer wins — implement the answer "
    "and say so in your result."
)
_CLARIFICATION_SOURCE_PREFIX = "Source:"
_REQUIRED_PROMPT_RESOURCES_HEADING = "## Required prompt resources"
_HTML_COMMENTS = re.compile(r"<!--.*?-->\n?", re.DOTALL)
_AGENT_CONTRACT_KEYS = {
    "schemaVersion",
    "dutyRootPath",
    "catalogDigest",
    "invocationReservationRootPath",
    "allowedAudiences",
    "authorizedPaths",
}
_MODEL_ASSIGNMENT_KEYS = {
    "provider",
    "model",
    "modelExecutionValue",
    "runner",
    "hostRuntime",
    "hostModelValue",
}
_V2_EXECUTION_IDENTITY_KEYS = {
    "participantRef",
    "roleExecutionRef",
    "dutyId",
}
_RUN_INVOCATION_PATH_KEYS = {
    "runManifestPath",
    "workerPromptsDirectoryPath",
    "leadInstructionsPath",
    "leadExecutionPromptPath",
    "leadPromptMetadataPath",
    "invocationReservationRootPath",
}


@dataclass(frozen=True)
class _MaterializationContext:
    request: InitialPromptMaterializationRequest
    project_root: Path
    manifest: Mapping[str, Any]
    team_state: Mapping[str, Any]
    active_context: Mapping[str, Any]


@dataclass(frozen=True)
class _PromptItem:
    worker: InitialPromptWorkerRequest
    final_path: Path
    plan: PromptPlan
    existed: bool
    clarification_input: tuple[str, Path] | None = None
    resources: tuple["_PromptResource", ...] = ()
    temp_path: Path | None = None


@dataclass(frozen=True)
class _PromptResource:
    path: Path
    text: str


@dataclass(frozen=True)
class _PublishedPrompt:
    worker: InitialPromptWorkerRequest
    path: Path
    plan: PromptPlan
    existed: bool


def materialize_initial_prompts(
    request: InitialPromptMaterializationRequest,
) -> dict[str, Path]:
    """Materialize the selected initial prompts, keyed by worker id.

    A positional sequence made every caller re-pair prompts with workers by
    index; a pairing that slips still type-checks and dispatches the wrong
    prompt to the wrong worker.
    """
    try:
        return _materialize_initial_prompts(request)
    except InitialPromptMaterializationError:
        raise
    except Exception as exc:
        raise InitialPromptMaterializationError(
            "render_failed",
            f"unexpected initial prompt materialization failure: {exc}",
        ) from exc


def _materialize_initial_prompts(
    request: InitialPromptMaterializationRequest,
) -> dict[str, Path]:
    context = _load_materialization_context(request)
    resolved_items = [
        _resolve_prompt_item(context, worker)
        for worker in request.workers
    ]
    items: list[_PromptItem] = []
    completed = False
    try:
        for item in resolved_items:
            items.append(_prepare_prompt_item(context, item))
        _validate_prepublication_set(context, items)
        published = tuple(
            _publish_or_reuse(context, item)
            for item in items
        )
        _validate_published_set(context, published)
        completed = True
        return {prompt.worker.worker_id: prompt.path for prompt in published}
    finally:
        cleanup_error = _remove_temp_files(items)
        if completed and cleanup_error is not None:
            raise InitialPromptMaterializationError(
                "publication_failed",
                f"cannot remove prompt staging file: {cleanup_error}",
            ) from cleanup_error


def _load_materialization_context(
    request: InitialPromptMaterializationRequest,
) -> _MaterializationContext:
    project_root = request.project_root.resolve()
    if not isinstance(request.delivery_mode, PromptDeliveryMode):
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "delivery_mode must be a PromptDeliveryMode",
        )
    manifest_path = _resolve_input_path(project_root, request.run_manifest_path)
    manifest = _load_json_object(manifest_path, "run manifest")
    _validate_invocation_manifest_authority(manifest)
    team_state = _load_referenced_json(project_root, manifest, "teamStatePath")
    active_context = hydrate_active_run_context(
        _load_referenced_json(
            project_root,
            manifest,
            "activeRunContextPath",
        )
    )
    return _MaterializationContext(
        request=request,
        project_root=project_root,
        manifest=manifest,
        team_state=team_state,
        active_context=active_context,
    )


def _validate_invocation_manifest_authority(
    manifest: Mapping[str, Any],
) -> None:
    has_contract = "agentContract" in manifest
    has_assignments = "invocationAssignments" in manifest
    if not has_contract and not has_assignments:
        return
    contract = manifest.get("agentContract")
    assignments = manifest.get("invocationAssignments")
    schema = manifest.get("schemaVersion")
    identity_version = manifest.get("executionIdentityVersion")
    is_v1 = schema in (None, 1, "1", "1.0") and identity_version is None
    is_v2 = schema == "2.0" and identity_version == 2
    if not is_v1 and not is_v2:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "invocation manifest mixes v1 and v2 execution identity",
        )
    allowed_assignment_shapes = {frozenset(_MODEL_ASSIGNMENT_KEYS)}
    if is_v2 and isinstance(assignments, Mapping):
        if any(
            isinstance(item, Mapping)
            and bool(set(item) & _V2_EXECUTION_IDENTITY_KEYS)
            for item in assignments.values()
        ):
            raise InitialPromptMaterializationError(
                "required_input_missing",
                "invocation assignment duplicates v2 execution identity",
            )
    complete = (
        isinstance(contract, Mapping)
        and set(contract) == _AGENT_CONTRACT_KEYS
        and contract.get("schemaVersion") == 1
        and isinstance(assignments, Mapping)
        and bool(assignments)
        and all(
            isinstance(item, Mapping)
            and frozenset(item) in allowed_assignment_shapes
            for item in assignments.values()
        )
        and all(
            isinstance(manifest.get(key), str)
            and bool(manifest.get(key))
            for key in _RUN_INVOCATION_PATH_KEYS
        )
    )
    if not complete:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "new invocation manifest is incomplete",
        )


def _resolve_prompt_item(
    context: _MaterializationContext,
    worker: InitialPromptWorkerRequest,
) -> _PromptItem:
    _validate_worker_request(worker)
    plan = _prompt_plan(context.manifest, worker.worker_id)
    if plan.audience not in _MATERIALIZABLE_AUDIENCES:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            f"prompt audience is not materializable: {plan.audience}",
        )
    clarification_input = _resolve_clarification_input(context, plan)
    final_path = _worker_prompt_path(context, worker.worker_id)
    existed = final_path.exists()
    return _PromptItem(
        worker=worker,
        final_path=final_path,
        plan=plan,
        existed=existed,
        clarification_input=clarification_input,
    )


def _prepare_prompt_item(
    context: _MaterializationContext,
    item: _PromptItem,
) -> _PromptItem:
    resources = _resolve_prompt_resources(
        context,
        item.plan,
        existed=item.existed,
    )
    prepared_item = _with_prompt_resources(item, resources)
    if item.existed and _anchor_predates_this_contract(context, prepared_item):
        item = replace(item, existed=False)
        prepared_item = _with_prompt_resources(
            item,
            _resolve_prompt_resources(context, item.plan, existed=False),
        )
    if item.existed:
        _validate_existing_prompt(context, prepared_item)
        return prepared_item
    temp_path: Path | None = None
    try:
        item.final_path.parent.mkdir(parents=True, exist_ok=True)
        text = _render_prompt(context, prepared_item)
        with tempfile.NamedTemporaryFile(
            mode="w",
            encoding="utf-8",
            dir=item.final_path.parent,
            prefix=f".{item.final_path.name}.",
            suffix=".tmp",
            delete=False,
        ) as handle:
            temp_path = Path(handle.name)
            handle.write(text)
    except InitialPromptMaterializationError:
        raise
    except Exception as exc:
        leaked = ""
        if temp_path is not None:
            try:
                temp_path.unlink(missing_ok=True)
            except OSError as cleanup_exc:
                # 정리 실패를 버리면 남은 .tmp 파일이 원인 없이 쌓인다. 원 오류를
                # 가리지 않도록 던지지 않고 메시지에 실어 보낸다.
                leaked = (f"; the staging file {temp_path} could not be removed "
                          f"({cleanup_exc})")
        raise InitialPromptMaterializationError(
            "render_failed",
            f"cannot render initial prompt for {item.worker.worker_id}: {exc}{leaked}",
        ) from exc
    if temp_path is None:
        raise InitialPromptMaterializationError(
            "render_failed",
            f"no staging file was created for {item.worker.worker_id}",
        )
    return _PromptItem(
        worker=item.worker,
        final_path=item.final_path,
        plan=item.plan,
        existed=False,
        clarification_input=item.clarification_input,
        resources=resources,
        temp_path=temp_path,
    )


def _with_prompt_resources(
    item: _PromptItem,
    resources: tuple[_PromptResource, ...],
) -> _PromptItem:
    return _PromptItem(
        worker=item.worker,
        final_path=item.final_path,
        plan=item.plan,
        existed=item.existed,
        clarification_input=item.clarification_input,
        resources=resources,
        temp_path=item.temp_path,
    )


def _render_prompt(
    context: _MaterializationContext,
    item: _PromptItem,
) -> str:
    request = _agent_invocation_request(context, item)
    if request is not None:
        return compose_agent_prompt(request)
    anchors, body = _render_prompt_parts(context, item)
    return "\n".join([*anchors, *body]).rstrip() + "\n"


def _render_prompt_parts(
    context: _MaterializationContext,
    item: _PromptItem,
) -> tuple[list[str], list[str]]:
    worker = item.worker
    state = _worker_state(context.team_state, worker.worker_id)
    prompt_rel = _project_relative(context.project_root, item.final_path)
    result_rel, audit_source_rel = _result_paths(context, state, worker.worker_id)
    try:
        anchors = worker_prompt_headers(
            project_root=context.project_root,
            prompt_rel=prompt_rel,
            result_rel=result_rel,
            audit_source_rel=audit_source_rel,
            worker_id=worker.worker_id,
            dispatch_kind="initial",
            manifest=context.manifest,
            active_context=context.active_context,
        )
        anchors.append(
            f"**Prompt Delivery Mode:** {context.request.delivery_mode.value}"
        )
        anchors.extend(_implementation_anchor_lines(context, item.plan))
        synthesis_packet_path = _materialize_report_writer_packet(
            context,
            item.plan,
        )
        body = _prompt_body(
            context,
            state,
            worker,
            item.plan,
            synthesis_packet_path,
        )
        body.extend(_resource_lines(context, item))
    except InitialPromptMaterializationError:
        raise
    except Exception as exc:
        raise InitialPromptMaterializationError(
            "render_failed",
            f"cannot compose initial prompt for {worker.worker_id}: {exc}",
        ) from exc
    return anchors, body


def _worker_invocation_id(manifest: Mapping[str, Any], worker_id: str) -> str:
    """Scope a worker's invocation id to this run, the way the lead's already is.

    The lead reserves `<task-type>-<seq>-lead` while workers reserved a bare
    `initial-<worker>`, so a second run of the same task-type asked for a
    reservation the first run already holds and every re-run died on
    `invocation reservation conflicts with existing invocation`. The asymmetry
    was the whole bug: nothing about a worker makes its call less run-specific
    than the lead's.
    """
    task_type = str(manifest.get("taskType") or "").strip()
    sequences = manifest.get("runSequencesByCategory")
    seq = ""
    if isinstance(sequences, Mapping):
        seq = str(sequences.get("prompts") or "").strip()
    scope = "-".join(part for part in (task_type, seq) if part)
    return f"{scope}-initial-{worker_id}" if scope else f"initial-{worker_id}"


def _agent_invocation_request(
    context: _MaterializationContext,
    item: _PromptItem,
) -> AgentInvocationRequest | None:
    contract = context.manifest.get("agentContract")
    if not isinstance(contract, Mapping):
        return None
    assignment_ref = f"initial/{item.worker.worker_id}"
    assignments = context.manifest.get("invocationAssignments")
    payload = assignments.get(assignment_ref) if isinstance(assignments, Mapping) else None
    try:
        assignment = agent_model_assignment_from_payload(payload)
        identity = invocation_execution_identity_from_manifest(
            context.manifest,
            assignment=assignment,
            assignment_ref=assignment_ref,
            duty_id=item.plan.duty_audience,
        )
    except AgentInvocationError as exc:
        raise InitialPromptMaterializationError(
            "required_input_missing", str(exc)
        ) from exc
    if assignment.model_execution_value != item.worker.model:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            f"requested model does not match {assignment_ref}",
        )
    anchors, body = _render_prompt_parts(context, item)
    source_paths = [
        AgentInstructionSource(
            "project",
            _project_relative(context.project_root, context.request.run_manifest_path),
        ),
        AgentInstructionSource(
            "project", _required_string(context.manifest, "activeRunContextPath")
        ),
        AgentInstructionSource(
            "project", _required_string(context.manifest, "teamStatePath")
        ),
    ]
    packet_source = _report_writer_packet_source(context, item.plan)
    if packet_source is not None:
        source_paths.append(packet_source)
    invocation_id = _worker_invocation_id(context.manifest, item.worker.worker_id)
    return AgentInvocationRequest(
        invocation_id=invocation_id,
        worker_id=item.worker.worker_id if identity is None else None,
        audience=item.plan.duty_audience,
        assignment_ref=assignment_ref,
        purpose=None,
        assignment=assignment,
        instruction=AgentInstruction(
            anchor_lines=tuple(anchors),
            body="\n".join(body).rstrip() + "\n",
            source_paths=tuple(source_paths),
        ),
        project_root=context.project_root,
        run_manifest_path=context.request.run_manifest_path,
        duty_root=context.project_root / _required_string(contract, "dutyRootPath"),
        prompt_path=item.final_path,
        metadata_path=item.final_path.with_name(item.final_path.name + ".meta.json"),
        dispatch_kind="initial",
        # 옛 계약으로 발행된 프롬프트를 같은 invocation id 위에 다시 쓰기 위한
        # 열쇠. `_replace_undispatched` 가 team-state 의 디스패치 행을 확인해
        # 이미 워커가 돌린 프롬프트는 그대로 둔다.
        replace_undispatched=True,
        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=invocation_id if identity is not None else None,
        attempt=1,
    )


def _prompt_body(
    context: _MaterializationContext,
    state: Mapping[str, Any],
    worker: InitialPromptWorkerRequest,
    plan: PromptPlan,
    synthesis_packet_path: str = "",
) -> list[str]:
    if plan.audience == "report-writer":
        return report_writer_prompt_body(
            context.manifest,
            context.active_context,
            context.team_state,
            worker.model,
            resolve_report_language(
                context.project_root,
                context.manifest,
                context.active_context,
            ),
            synthesis_packet_path,
        )
    return analysis_prompt_body(
        context.manifest,
        context.active_context,
        worker.worker_id,
        worker.model,
        plan,
        _optional_approved_plan_path(context) if plan.include_approved_plan else "",
    )


def _report_writer_narrative_path(context: _MaterializationContext) -> Path:
    return report_narrative_path(context.project_root, context.manifest)


def _report_writer_packet_source(
    context: _MaterializationContext,
    plan: PromptPlan,
) -> AgentInstructionSource | None:
    if not _report_writer_reads_a_packet(context, plan):
        return None
    _, markdown_path = report_synthesis_packet_paths(
        _report_writer_narrative_path(context)
    )
    return AgentInstructionSource(
        "project",
        _project_relative(context.project_root, markdown_path),
    )


def _report_writer_reads_a_packet(
    context: "_MaterializationContext",
    plan: PromptPlan,
) -> bool:
    """report-writer 의 `## Inputs` 가 합성 묶음 한 줄로 접히는가.

    ADR-0023 이후 report-writer 는 합성 묶음 하나만 받는다 — clarification
    응답을 포함한 개별 경로는 열거되지 않는다. contract v3 가 아닌 매니페스트는
    아직 옛 목록(`report_writer_input_lines`)을 쓰고 그 목록에는 clarification
    줄이 있다. 즉 이 답은 정적 plan 플래그가 아니라 이 run 의 계약 버전에
    달려 있고, 본문을 조립한 쪽과 그 본문을 검사하는 쪽이 같은 술어를 써야
    한다 — 갈리면 clarification 을 받은 planning 재실행이 Phase 6 에서
    `existing_prompt_invalid` 로 막힌다.
    """
    return plan.audience == "report-writer" and uses_report_contract_v3(
        context.manifest
    )


def _materialize_report_writer_packet(
    context: _MaterializationContext,
    plan: PromptPlan,
) -> str:
    packet_source = _report_writer_packet_source(context, plan)
    if packet_source is None:
        return ""
    try:
        _, markdown_path = materialize_report_synthesis_packet(
            project_root=context.project_root,
            manifest=context.manifest,
            active_context=context.active_context,
            team_state=context.team_state,
            narrative_path=_report_writer_narrative_path(context),
        )
    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 InitialPromptMaterializationError(
            "required_input_missing",
            f"report synthesis packet contract defects: {defects}",
        ) from exc
    return _project_relative(context.project_root, markdown_path)


def _resource_lines(
    context: _MaterializationContext,
    item: _PromptItem,
) -> list[str]:
    if not item.resources:
        return []
    if context.request.delivery_mode is PromptDeliveryMode.LAZY_PATH_REFERENCE:
        return [
            "",
            _REQUIRED_PROMPT_RESOURCES_HEADING,
            "",
            *(f"- `{resource.path}`" for resource in item.resources),
        ]
    bodies = [resource.text.rstrip() for resource in item.resources]
    clarification = (
        _eager_clarification_authority(item.clarification_input)
        if item.plan.audience in _CLARIFICATION_AUTHORITY_AUDIENCES
        else ""
    )
    if clarification:
        bodies.append(clarification)
    carry = _stage_fix_carry_body(context, item.plan)
    if carry:
        bodies.append(carry)
    return ["", "\n\n".join(bodies)]


def _stage_fix_carry_body(
    context: _MaterializationContext,
    plan: PromptPlan,
) -> str:
    """The fix-run carry block, for the two audiences that act on it.

    The executor's scope is the carried blocking findings, and the verifier MUST
    cite each of them as resolved or still-failing — a fix-run verifier result
    citing none is recorded as a contract violation. Both facts live in the
    rendered analysis profile, which no CLI worker can read, so the block used to
    reach them only if the lead transcribed it by hand. A missed transcription
    made the worker answerable for a list it was never given.
    """
    if plan.audience not in ("implementation-executor", "implementation-verifier"):
        return ""
    run = context.active_context.get("run")
    if not isinstance(run, Mapping):
        return ""
    return _string_value(run.get("fixRunCarry")).strip()


def _resolve_clarification_input(
    context: _MaterializationContext,
    plan: PromptPlan,
) -> tuple[str, Path] | None:
    relative = instruction_path(
        context.manifest,
        context.active_context,
        "clarificationResponsePath",
    )
    if not relative:
        return None
    path = _resolve_owned_artifact_path(
        context.project_root,
        Path(relative),
        "clarification response path",
    )
    return relative, path


def _eager_clarification_authority(
    clarification_input: tuple[str, Path] | None,
) -> str:
    if clarification_input is None:
        return ""
    relative, _ = clarification_input
    body = _clarification_body(clarification_input)
    if not body:
        return ""
    return (
        f"{_CLARIFICATION_AUTHORITY_HEADING}\n\n"
        f"{_CLARIFICATION_AUTHORITY_INTRO}\n\n"
        f"Source: `{relative}`\n\n{body}"
    )


def _clarification_body(
    clarification_input: tuple[str, Path] | None,
) -> str:
    if clarification_input is None:
        return ""
    _, path = clarification_input
    if not path.is_file():
        return ""
    return _read_required_text(path, "required clarification response").strip()


def _read_required_nonempty_text(path: Path, label: str) -> str:
    text = _read_required_text(path, label)
    if text.strip():
        return text
    cause = ValueError(f"{label} is empty: {path}")
    raise InitialPromptMaterializationError(
        "required_input_missing",
        str(cause),
    ) from cause


def _read_required_text(path: Path, label: str) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            f"{label} cannot be read: {path}",
        ) from exc


def _resolve_prompt_resources(
    context: _MaterializationContext,
    plan: PromptPlan,
    *,
    existed: bool,
) -> tuple[_PromptResource, ...]:
    paths = _required_resource_path_candidates(context, plan)
    if existed and context.request.delivery_mode is PromptDeliveryMode.EAGER_INCLUDE:
        return ()
    return tuple(
        _PromptResource(
            path=path.resolve(),
            text=_worker_visible_text(
                _read_required_nonempty_text(path, "required prompt resource")
            ),
        )
        for path in paths
    )


def _worker_visible_text(text: str) -> str:
    """Strip lead-and-maintainer-only passages before a body reaches a worker.

    These sidecars have two readers: the lead lazy-reads the file to run the
    phase, and the worker receives the same body inlined in its prompt. An
    HTML comment is the seam between them — rendered markdown hides it, so it
    holds the delivery plumbing (which path feeds this file, which heading the
    CLI wrapper greps for) that the worker would otherwise pay for and read as
    an instruction addressed to itself.
    """
    return _HTML_COMMENTS.sub("", text).lstrip("\n")


def _required_resource_path_candidates(
    context: _MaterializationContext,
    plan: PromptPlan,
) -> tuple[Path, ...]:
    profiles = context.request.runtime_root.resolve() / "prompts" / "profiles"
    if plan.audience == "implementation-executor":
        paths = (
            _executor_profile_path(context),
            profiles / "_stage-discipline.md",
            profiles / "_coding-conventions-preflight.md",
            profiles / "_implementation-diff-review.md",
            profiles / "_implementation-self-check.md",
        )
    elif plan.audience == "implementation-verifier":
        # The self-check body is executor-only: it is written to the worker that
        # owns the diff ("fix it or surface the violation", break-and-restore
        # mutation checks), and a verifier is barred from every edit it asks
        # for. Its own blocking taxonomy is inlined in the verifier sidecar.
        paths = (profiles / "_implementation-verifier.md",)
    else:
        return ()
    return paths


def _executor_profile_path(context: _MaterializationContext) -> Path:
    instruction_set = context.active_context.get("instructionSet")
    value = ""
    if isinstance(instruction_set, Mapping):
        value = _string_value(instruction_set.get("analysisProfilePath"))
    value = value or _string_value(context.manifest.get("analysisProfilePath"))
    if not value:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "implementation executor profile path is missing",
        )
    return _resolve_owned_artifact_path(
        context.project_root,
        _resolve_input_path(context.project_root, Path(value)).parent
        / "implementation-executor.md",
        "implementation executor profile path",
    )


def _validate_prepublication_set(
    context: _MaterializationContext,
    items: Sequence[_PromptItem],
) -> None:
    records = [_item_record(context, item) for item in items]
    _validate_existing_record_subset(
        context.manifest,
        [
            record
            for item, record in zip(items, records, strict=True)
            if item.existed
        ],
        "prepublication",
    )
    for item, record in zip(items, records, strict=True):
        if item.existed:
            continue
        errors = validate_initial_prompt_records(
            manifest=context.manifest,
            records=[record],
            require_evidence_ledger=True,
        )
        if errors:
            raise InitialPromptMaterializationError(
                "validation_failed",
                "rendered initial prompt contract: " + "; ".join(errors),
            )
    errors = validate_initial_prompt_records(
        manifest=context.manifest,
        records=records,
        require_evidence_ledger=True,
    )
    if errors:
        raise InitialPromptMaterializationError(
            "validation_failed",
            "prepublication initial prompt contract: " + "; ".join(errors),
        )


def _publish_or_reuse(
    context: _MaterializationContext,
    item: _PromptItem,
) -> _PublishedPrompt:
    if item.existed:
        return _published_prompt(item, existed=True)
    invocation = _agent_invocation_request(context, item)
    if invocation is None:
        temp_path = _required_temp_path(item)
        try:
            os.link(temp_path, item.final_path)
        except FileExistsError:
            _validate_existing_prompt(context, item)
            return _published_prompt(item, existed=True)
        except OSError as exc:
            raise InitialPromptMaterializationError(
                "publication_failed",
                f"cannot publish initial prompt {item.final_path}: {exc}",
            ) from exc
        return _published_prompt(item, existed=False)
    try:
        prepare_agent_invocation(invocation)
    except AgentInvocationError as exc:
        reason: MaterializationReason = (
            "existing_prompt_conflict"
            if exc.reason == "existing_invocation_conflict"
            else "publication_failed"
        )
        raise InitialPromptMaterializationError(
            reason,
            f"cannot publish initial prompt {item.final_path}: {exc}",
        ) from exc
    return _published_prompt(item, existed=False)


def _validate_existing_prompt(
    context: _MaterializationContext,
    item: _PromptItem,
) -> None:
    expects_invocation = isinstance(context.manifest.get("agentContract"), Mapping)
    text = _require_readable_prompt(
        item.final_path,
        item.worker.worker_id,
        "existing_prompt_invalid",
    )
    _validate_existing_clarification_metadata(context, item, text)
    _validate_existing_resource_metadata(context, item, text)
    basic_record = PromptRecord(
        item.worker.worker_id,
        "initial",
        item.final_path,
        metadata_path=_metadata_path(item.final_path) if expects_invocation else None,
        expected_duty_audience=(
            item.plan.duty_audience if expects_invocation else None
        ),
    )
    basic_errors = validate_initial_prompt_records(
        manifest=context.manifest,
        records=[basic_record],
        require_evidence_ledger=True,
    )
    if basic_errors:
        raise InitialPromptMaterializationError(
            "existing_prompt_invalid",
            "existing initial prompt contract: " + "; ".join(basic_errors),
        )
    expected_record = PromptRecord(
        item.worker.worker_id,
        "initial",
        item.final_path,
        expected_model=item.worker.model,
        expected_delivery_mode=context.request.delivery_mode.value,
        metadata_path=_metadata_path(item.final_path) if expects_invocation else None,
        expected_duty_audience=(
            item.plan.duty_audience if expects_invocation else None
        ),
    )
    expected_errors = validate_initial_prompt_records(
        manifest=context.manifest,
        records=[expected_record],
        require_evidence_ledger=True,
    )
    if expected_errors:
        reason = _existing_metadata_reason(expected_errors)
        raise InitialPromptMaterializationError(
            reason,
            "existing initial prompt metadata: " + "; ".join(expected_errors),
        )
    invocation = _agent_invocation_request(context, item)
    if invocation is not None:
        errors = verify_agent_invocation(
            invocation.metadata_path,
            project_root=context.project_root,
            expected_run_manifest_path=context.request.run_manifest_path,
            expected_assignment=invocation.assignment,
            expected_invocation_id=invocation.invocation_id,
            expected_worker_id=invocation.worker_id,
            expected_assignment_ref=invocation.assignment_ref,
            expected_audience=invocation.audience,
        )
        if errors:
            raise InitialPromptMaterializationError(
                "existing_prompt_invalid",
                "existing initial prompt invocation: " + "; ".join(errors),
            )


def _validate_existing_clarification_metadata(
    context: _MaterializationContext,
    item: _PromptItem,
    text: str,
) -> None:
    expected = (
        item.clarification_input[0]
        if item.clarification_input is not None
        else None
    )
    enumerated = (
        item.plan.audience in _MATERIALIZABLE_AUDIENCES
        and not item.plan.packet_only
        and not _report_writer_reads_a_packet(context, item.plan)
    )
    input_values = _backticked_line_values(
        _markdown_section(text, "## Inputs"),
        _CLARIFICATION_INPUT_PREFIX,
    )
    expected_inputs = [expected] if expected is not None and enumerated else []
    if input_values != expected_inputs:
        _raise_existing_clarification_invalid(
            item,
            "Clarification response",
        )
    _validate_eager_clarification_authority(
        context,
        item,
        text,
        expected,
    )


def _validate_eager_clarification_authority(
    context: _MaterializationContext,
    item: _PromptItem,
    text: str,
    expected: str | None,
) -> None:
    authority_expected = (
        item.plan.audience in _CLARIFICATION_AUTHORITY_AUDIENCES
        and context.request.delivery_mode is PromptDeliveryMode.EAGER_INCLUDE
        and bool(_clarification_body(item.clarification_input))
    )
    lines = text.splitlines()
    heading_indexes = [
        index
        for index, line in enumerate(lines)
        if line.startswith(_CLARIFICATION_AUTHORITY_HEADING_PREFIX)
    ]
    if not authority_expected and not heading_indexes:
        return
    if not authority_expected or expected is None:
        _raise_existing_clarification_invalid(item, "clarification authority")
    expected_lines = [
        _CLARIFICATION_AUTHORITY_HEADING,
        "",
        _CLARIFICATION_AUTHORITY_INTRO,
        "",
        f"{_CLARIFICATION_SOURCE_PREFIX} `{expected}`",
        "",
    ]
    # The block okstra injects is what this check owns; the carry-in body that
    # follows it is the user's own document and may legitimately open with the
    # same heading. Requiring the phrase to be unique made such a document
    # undispatchable — the run failed `existing_prompt_invalid` with nothing in
    # any contract telling the author to avoid that title.
    matches = [
        index
        for index in heading_indexes
        if lines[index:index + len(expected_lines)] == expected_lines
    ]
    if not matches or any(index < matches[0] for index in heading_indexes):
        _raise_existing_clarification_invalid(item, "clarification authority")


def _validate_existing_resource_metadata(
    context: _MaterializationContext,
    item: _PromptItem,
    text: str,
) -> None:
    if (
        context.request.delivery_mode is not PromptDeliveryMode.LAZY_PATH_REFERENCE
        or not item.resources
    ):
        return
    expected_section = "\n".join([
        "",
        *(f"- `{resource.path}`" for resource in item.resources),
    ])
    actual_section = _markdown_section(
        text,
        _REQUIRED_PROMPT_RESOURCES_HEADING,
    )
    if (
        text.count(_REQUIRED_PROMPT_RESOURCES_HEADING) != 1
        or actual_section != expected_section
    ):
        raise InitialPromptMaterializationError(
            "existing_prompt_invalid",
            "existing required prompt resources do not match canonical "
            f"paths for {item.worker.worker_id}",
        )


def _backticked_line_values(text: str, prefix: str) -> list[str]:
    values = []
    for line in text.splitlines():
        if not line.startswith(prefix):
            continue
        suffix = line[len(prefix):].strip()
        if len(suffix) >= 2 and suffix.startswith("`") and suffix.endswith("`"):
            values.append(suffix[1:-1])
        else:
            values.append("")
    return values


def _markdown_section(text: str, heading: str) -> str:
    lines = text.splitlines()
    try:
        start = lines.index(heading) + 1
    except ValueError:
        return ""
    section = []
    for line in lines[start:]:
        if line.startswith("## "):
            break
        section.append(line)
    return "\n".join(section)


def _raise_existing_clarification_invalid(
    item: _PromptItem,
    field: str,
) -> None:
    raise InitialPromptMaterializationError(
        "existing_prompt_invalid",
        f"existing {field} metadata does not match active context "
        f"for {item.worker.worker_id}",
    )


def _validate_published_set(
    context: _MaterializationContext,
    prompts: Sequence[_PublishedPrompt],
) -> None:
    records = [_published_record(context, prompt) for prompt in prompts]
    for prompt, record in zip(prompts, records, strict=True):
        reason: MaterializationReason = (
            "existing_prompt_invalid"
            if prompt.existed
            else "validation_failed"
        )
        _require_readable_prompt(prompt.path, prompt.worker.worker_id, reason)
    _validate_existing_record_subset(
        context.manifest,
        [
            record
            for prompt, record in zip(prompts, records, strict=True)
            if prompt.existed
        ],
        "published",
    )
    for prompt, record in zip(prompts, records, strict=True):
        if prompt.existed:
            continue
        errors = validate_initial_prompt_records(
            manifest=context.manifest,
            records=[record],
            require_evidence_ledger=True,
        )
        if errors:
            raise InitialPromptMaterializationError(
                "validation_failed",
                "published initial prompt contract: " + "; ".join(errors),
            )
    errors = validate_initial_prompt_records(
        manifest=context.manifest,
        records=records,
        require_evidence_ledger=True,
    )
    if errors:
        raise InitialPromptMaterializationError(
            "validation_failed",
            "published initial prompt contract: " + "; ".join(errors),
        )


def _item_record(
    context: _MaterializationContext,
    item: _PromptItem,
) -> PromptRecord:
    path = item.final_path if item.existed else _required_temp_path(item)
    expects_invocation = (
        item.existed
        and isinstance(context.manifest.get("agentContract"), Mapping)
    )
    return PromptRecord(
        item.worker.worker_id,
        "initial",
        path,
        expected_model=item.worker.model,
        expected_delivery_mode=context.request.delivery_mode.value,
        metadata_path=_metadata_path(item.final_path) if expects_invocation else None,
        expected_duty_audience=(
            item.plan.duty_audience if expects_invocation else None
        ),
    )


def _published_record(
    context: _MaterializationContext,
    prompt: _PublishedPrompt,
) -> PromptRecord:
    expects_invocation = isinstance(context.manifest.get("agentContract"), Mapping)
    return PromptRecord(
        prompt.worker.worker_id,
        "initial",
        prompt.path,
        expected_model=prompt.worker.model,
        expected_delivery_mode=context.request.delivery_mode.value,
        metadata_path=_metadata_path(prompt.path) if expects_invocation else None,
        expected_duty_audience=(
            prompt.plan.duty_audience if expects_invocation else None
        ),
    )


def _metadata_path(prompt_path: Path) -> Path:
    return prompt_path.with_name(prompt_path.name + ".meta.json")


def _worker_prompt_path(
    context: _MaterializationContext,
    worker_id: str,
) -> Path:
    paths = context.manifest.get("workerPromptPathByWorkerId")
    if not isinstance(paths, Mapping):
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "run manifest has no workerPromptPathByWorkerId object",
        )
    value = _required_string(paths, worker_id)
    return _resolve_owned_artifact_path(
        context.project_root,
        Path(value),
        "initial prompt path",
    )


def _resolve_owned_artifact_path(
    project_root: Path,
    path: Path,
    label: str,
) -> Path:
    resolved = _resolve_input_path(project_root, path).resolve()
    artifact_root = (project_root / ".okstra").resolve()
    if resolved == artifact_root or artifact_root not in resolved.parents:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            f"{label} is outside .okstra: {resolved}",
        )
    return resolved


def _result_paths(
    context: _MaterializationContext,
    state: Mapping[str, Any],
    worker_id: str,
) -> tuple[str, str | None]:
    worker_result = _required_string(state, "resultPath")
    if worker_id != REPORT_WRITER_WORKER_ID:
        return worker_result, None
    if uses_report_contract_v3(context.manifest):
        narrative = report_narrative_path(context.project_root, context.manifest)
        return _project_relative(context.project_root, narrative), worker_result
    expected_report = _required_string(context.manifest, "expectedReportRecordPath")
    return str(final_report_data_path(Path(expected_report))), worker_result


def _anchor_predates_this_contract(
    context: _MaterializationContext,
    item: _PromptItem,
) -> bool:
    """이 프롬프트가 작업 트리 앵커 없이 발행된 옛 계약본인가.

    프롬프트는 불변 발행이지만 계약은 계속 조여진다. 그 사이에 발행된
    프롬프트를 그대로 디스패치하면 워커는 지금 계약이 요구하는 값을 못 받고,
    그 어긋남을 알아채는 자리(리드, 검증기)에는 고칠 수단이 없어 run 이 선다.
    그래서 여기서 조용히 다시 만든다 — 발행 경로가 `replace_undispatched` 로
    같은 invocation id 위에 덮어쓰고, 이미 디스패치된 프롬프트는 그 경로가
    감사 기록으로 지켜 준다.

    판정은 지금 렌더가 실을 앵커 한 줄이 기존 본문에 있는지로 한다. 작업
    트리가 없는 run 은 실을 줄이 없으므로 언제나 최신이다.
    """
    lines = source_worktree_headers(context.active_context) if context.manifest.get("taskType") != "final-verification" else []
    try:
        text = item.final_path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        # 읽을 수 없는 기존 프롬프트는 `_validate_existing_prompt` 가 원인을
        # 보존해 보고한다(디렉터리, 깨진 UTF-8). 여기서 재발행으로 덮어 그
        # 진단을 지우지 않는다.
        return False
    if context.manifest.get("executionIdentityVersion") == 2:
        request = _agent_invocation_request(context, item)
        if request is not None and not invocation_dispatch_ids(request):
            marker = f"**Execution delivery contract:** {EXECUTION_DELIVERY_CONTRACT}"
            if marker not in text:
                return True
    return bool(lines) and lines[0] not in text


def _implementation_anchor_lines(
    context: _MaterializationContext,
    plan: PromptPlan,
) -> list[str]:
    """소스 작업 트리와 산출물 루트를 구분하고 구현 입력을 덧붙인다."""
    if context.manifest.get("taskType") == "final-verification":
        return []
    lines = []
    if plan.audience not in {
        "implementation-executor",
        "implementation-verifier",
    }:
        return lines
    worktree = _required_worktree_path(context)
    if plan.audience == "implementation-executor":
        lines.append(f"cwd for every mutating command: {worktree}")
    lines.append(
        f"{APPROVED_PLAN_HEADER} {_required_approved_plan_path(context)}"
    )
    lines.append(
        f"{IMPLEMENTATION_STAGE_HEADER} {_required_stage_number(context)}"
    )
    return lines


def _required_worktree_path(context: _MaterializationContext) -> str:
    executor_worktree = context.active_context.get("executorWorktree")
    value = (
        _string_value(executor_worktree.get("path"))
        if isinstance(executor_worktree, Mapping)
        else ""
    )
    if not value:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "implementation prompt generation requires a worktree path",
        )
    return value


def _required_approved_plan_path(context: _MaterializationContext) -> str:
    """Read the approved plan path from this run's user-input snapshot.

    run-inputs owns it because it is a user input, not a rendered path — and
    reading it there also keeps runs prepared before this anchor existed
    materializable, since every run has always recorded it.
    """
    source_artifacts = context.active_context.get("sourceArtifacts")
    run_inputs_rel = (
        _string_value(source_artifacts.get("runInputsPath"))
        if isinstance(source_artifacts, Mapping)
        else ""
    )
    if not run_inputs_rel:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "implementation prompt generation requires a run inputs path",
        )
    payload = _load_json_object(
        _resolve_input_path(context.project_root, Path(run_inputs_rel)),
        "run inputs",
    )
    inputs = payload.get("inputs")
    value = (
        _string_value(inputs.get("approvedPlanPath"))
        if isinstance(inputs, Mapping)
        else ""
    )
    if not value:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "implementation prompt generation requires an approved plan path",
        )
    return str(_resolve_input_path(context.project_root, Path(value)))


def _optional_approved_plan_path(context: _MaterializationContext) -> str:
    """The approved plan path when this run recorded one, otherwise empty.

    `_required_approved_plan_path` speaks for implementation, where a missing
    plan is a contract violation. An acceptance verifier enumerates the same
    file as an input, and a run prepared without one still has to produce a
    prompt — so the absence is reported by the empty line being dropped, not by
    refusing to materialize.
    """
    try:
        return _required_approved_plan_path(context)
    except InitialPromptMaterializationError:
        return ""


def _required_stage_number(context: _MaterializationContext) -> str:
    value = _string_value(_active_run_field(context, "stage"))
    if not value:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            "implementation prompt generation requires a stage number",
        )
    return value


def _active_run_field(context: _MaterializationContext, key: str) -> Any:
    run = context.active_context.get("run")
    return run.get(key) if isinstance(run, Mapping) else ""


def _prompt_plan(
    manifest: Mapping[str, Any],
    worker_id: str,
) -> PromptPlan:
    try:
        return resolve_prompt_plan_for_manifest(
            manifest=manifest,
            worker_id=worker_id,
            dispatch_kind="initial",
        )
    except ValueError as exc:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            str(exc),
        ) from exc


def _worker_state(
    team_state: Mapping[str, Any],
    worker_id: str,
) -> Mapping[str, Any]:
    workers = team_state.get("workers")
    if isinstance(workers, list):
        for worker in workers:
            if isinstance(worker, Mapping) and worker.get("workerId") == worker_id:
                return worker
    raise InitialPromptMaterializationError(
        "required_input_missing",
        f"team-state has no workerId={worker_id}",
    )


def _load_referenced_json(
    project_root: Path,
    payload: Mapping[str, Any],
    key: str,
) -> Mapping[str, Any]:
    value = _required_string(payload, key)
    return _load_json_object(
        _resolve_input_path(project_root, Path(value)),
        key,
    )


def _load_json_object(path: Path, label: str) -> Mapping[str, Any]:
    try:
        payload = load_owned_object(path, artifact=label)
    except JsonBoundaryError as exc:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            f"cannot load {label}: {path}: {exc}",
        ) from (exc.__cause__ or exc)
    if not isinstance(payload, dict):
        raise InitialPromptMaterializationError(
            "required_input_missing",
            f"{label} must be a JSON object: {path}",
        )
    return payload


def _required_string(payload: Mapping[str, Any], key: str) -> str:
    value = _string_value(payload.get(key))
    if not value:
        raise InitialPromptMaterializationError(
            "required_input_missing",
            f"missing required string field: {key}",
        )
    return value


def _validate_worker_request(worker: InitialPromptWorkerRequest) -> None:
    if worker.worker_id.strip() and worker.model.strip():
        return
    raise InitialPromptMaterializationError(
        "required_input_missing",
        "worker_id and model are required",
    )


def _existing_metadata_reason(
    errors: Sequence[str],
) -> MaterializationReason:
    if any("does not match requested" in error for error in errors):
        return "existing_prompt_conflict"
    return "existing_prompt_invalid"


def _validate_existing_record_subset(
    manifest: Mapping[str, Any],
    records: Sequence[PromptRecord],
    stage: str,
) -> None:
    errors = validate_initial_prompt_records(
        manifest=manifest,
        records=records,
        require_evidence_ledger=True,
    )
    if errors:
        raise InitialPromptMaterializationError(
            "existing_prompt_invalid",
            f"{stage} existing initial prompt contract: " + "; ".join(errors),
        )


def _published_prompt(
    item: _PromptItem,
    *,
    existed: bool,
) -> _PublishedPrompt:
    return _PublishedPrompt(
        worker=item.worker,
        path=item.final_path,
        plan=item.plan,
        existed=existed,
    )


def _require_readable_prompt(
    path: Path,
    worker_id: str,
    reason: MaterializationReason,
) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        raise InitialPromptMaterializationError(
            reason,
            f"cannot read prompt for {worker_id}: {path}: {exc}",
        ) from exc


def _required_temp_path(item: _PromptItem) -> Path:
    if item.temp_path is not None:
        return item.temp_path
    raise InitialPromptMaterializationError(
        "render_failed",
        f"no staging file exists for {item.worker.worker_id}",
    )


def _project_relative(project_root: Path, path: Path) -> str:
    # macOS commonly exposes the same temporary directory as both `/var/...`
    # and `/private/var/...`; project-root resolution already canonicalizes the
    # former. Resolve the candidate too so provenance does not reject an alias
    # for the same directory. `strict=False` is required for a prompt that has
    # not been published yet, while still resolving every existing parent and
    # therefore preserving the symlink-escape check in `relative_to`.
    return path.resolve(strict=False).relative_to(
        project_root.resolve(strict=True)
    ).as_posix()


def _resolve_input_path(project_root: Path, path: Path) -> Path:
    return path if path.is_absolute() else project_root / path


def _remove_temp_files(
    items: Sequence[_PromptItem],
) -> OSError | None:
    first_error = None
    for item in items:
        if item.temp_path is None:
            continue
        try:
            item.temp_path.unlink(missing_ok=True)
        except OSError as exc:
            first_error = first_error or exc
    return first_error


def _string_value(value: Any) -> str:
    return value.strip() if isinstance(value, str) else ""