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

import json
import os
import re
import tempfile
from dataclasses import dataclass
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,
    agent_model_assignment_from_payload,
    compose_agent_prompt,
    invocation_execution_identity_from_manifest,
    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 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:
        _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:
        if temp_path is not None:
            try:
                temp_path.unlink(missing_ok=True)
            except OSError:
                pass
        raise InitialPromptMaterializationError(
            "render_failed",
            f"cannot render initial prompt for {item.worker.worker_id}: {exc}",
        ) 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",
        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,
    )


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 (
        plan.audience != "report-writer"
        or not uses_report_contract_v3(context.manifest)
    ):
        return None
    _, markdown_path = report_synthesis_packet_paths(
        _report_writer_narrative_path(context)
    )
    return AgentInstructionSource(
        "project",
        _project_relative(context.project_root, markdown_path),
    )


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
    )
    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
    expected_report = _required_string(context.manifest, "expectedReportRecordPath")
    return str(final_report_data_path(Path(expected_report))), worker_result


def _implementation_anchor_lines(
    context: _MaterializationContext,
    plan: PromptPlan,
) -> list[str]:
    """Render the anchors only an implementation audience carries.

    The approved plan and the stage number are absolute here because a CLI
    worker's cwd is the stage worktree, where a `.okstra/...` relative path
    resolves against the wrong root.
    """
    if plan.audience not in {
        "implementation-executor",
        "implementation-verifier",
    }:
        return []
    worktree = _required_worktree_path(context)
    lines = [f"**Worktree:** {worktree}"]
    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 _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 ""