# shellcheck shell=bash

write_validation_brief() {
  local brief_path="$1"
  local brief_title="$2"
  local task_group="$3"
  local task_id="$4"
  local validation_focus="$5"
  local task_key_value=""

  task_key_value="$(task_key "$task_group" "$task_id")"

  cat >"$brief_path" <<EOF
# Cross Verify Task Brief

## Brief Identity

- Brief Title: $brief_title
- Project ID: \`okstra-validation\`
- Task Group: \`$task_group\`
- Task ID: \`$task_id\`
- Task Key: \`$task_key_value\`
- Related Tasks:
- Task Type: \`$TASK_TYPE\`
- Requested Outcome: validate asset seeding, refresh behavior, and discovery catalog integrity
- Owner: \`Oz\`

## Problem or Request Summary

- Verify that okstra seeds all Claude Markdown assets into the target project on first run.
- Verify that refresh is optional and only forced with the refresh option.
- Verify that config and deployment expected states are discoverable by task bundle artifacts.
- Validation focus: $validation_focus

## Source Materials

- Primary problem statement: \`validation\`
- Existing analysis or notes:
- Related raw samples or logs:
- Related code paths: \`scripts/okstra.sh\`
- Related tickets or docs: \`OKSTRA_USAGE_MANUAL.md\`
- Previous reports in the same task history:

## Constraints and Assumptions

- Known constraints: validation must stay inside the temporary project root
- Known assumptions: task bundle artifacts are the canonical source for skill loading
- Things that are still uncertain: none

## Configuration References and Expected Values

- Config file: \`.claude/settings.json\`
  - Expected values:
    - installed okstra Claude assets must remain discoverable under \`~/.claude/skills/\` and \`~/.claude/agents/\` (managed by \`okstra install\`)
- Config file: \`.okstra/discovery/latest-task.json\`
  - Expected values:
    - latest prepared task pointer must include the current task key
    - task catalog path must be present
- Config file: \`.okstra/discovery/task-catalog.json\`
  - Expected values:
    - task catalog must preserve prepared task bundles by task key
    - task catalog must allow task-group and task-id level distinction

## Deployment Manifests and Expected Values

- Manifest file: \`deploy/values.yaml\`
  - Expected values:
    - image tag should match the validated release candidate for this task
    - rollout-specific values must be verified against the task brief before final approval
- Manifest file: \`k8s/deployment.yaml\`
  - Expected values:
    - env and image settings must match the task brief requirements
    - missing deployment expectations must be reported as missing information

## Questions for Workers

1. Are all required Claude assets seeded into the temporary target project?
2. Do the generated task artifacts preserve config and deployment expected states?
3. Does refresh remain optional rather than automatic?
4. Does the discovery catalog preserve distinct task entries?

## Expected Outputs

- Asset seed verification:
- Missing information:
- Risks:
- Recommended next actions:
EOF
}

write_worker_prompt_history_fixture() {
  local task_group="$1"
  local task_id="$2"
  local worker_id="$3"
  local expected_task_manifest_relative_path=""

  expected_task_manifest_relative_path="$(task_manifest_relative_path "$task_group" "$task_id")"

  python3 - "$PROJECT_ROOT" "$expected_task_manifest_relative_path" "$worker_id" <<'PY'
from pathlib import Path
import json
import sys

project_root = Path(sys.argv[1])
task_manifest_path = project_root / sys.argv[2]
target_worker_id = sys.argv[3]
task_manifest = json.loads(task_manifest_path.read_text())
team_state_path = project_root / task_manifest["teamStatePath"]
team_state = json.loads(team_state_path.read_text())

target_worker = None
for worker in team_state.get("workers", []):
    if isinstance(worker, dict) and worker.get("workerId") == target_worker_id:
        target_worker = worker
        break

if target_worker is None:
    raise SystemExit(f"worker not found in team-state: {target_worker_id}")

prompt_relative = str(target_worker.get("promptPath", "")).strip()
if not prompt_relative:
    raise SystemExit(f"worker promptPath is missing: {target_worker_id}")

prompt_path = project_root / prompt_relative
prompt_path.parent.mkdir(parents=True, exist_ok=True)
artifacts = task_manifest.get("artifacts", {})
analysis_packet_path = (
    str(artifacts.get("analysisPacketPath", "")).strip()
    if isinstance(artifacts, dict)
    else ""
)
prompt_path.write_text(
    "\n".join(
        [
            "**Worker Error Contract Path:** /okstra/templates/worker-error-contract.md",
            f"**Errors log path:** {project_root}/run/logs/errors.jsonl",
            f"**Errors sidecar path:** {project_root}/run/worker-results/"
            f"{target_worker_id}-errors.json",
            f"Assigned worker prompt history path: {prompt_relative}",
            "**Prompt Delivery Mode:** eager-include",
            f"**Model:** Fixture worker, {target_worker.get('modelExecutionValue', '')}",
            f"- Primary analysis packet: `{analysis_packet_path}`",
            "",
            "# Initial Analysis Prompt",
            f"Task Key: {task_manifest.get('taskKey', '')}",
            "Validation fixture prompt body.",
        ]
    )
    + "\n"
)
PY
}

prepare_run_validator_fixture() {
  local task_group="$1"
  local task_id="$2"
  local omitted_worker_id="$3"
  local expected_task_manifest_relative_path=""

  expected_task_manifest_relative_path="$(task_manifest_relative_path "$task_group" "$task_id")"

  python3 - "$PROJECT_ROOT" "$expected_task_manifest_relative_path" "$omitted_worker_id" <<'PY'
from datetime import datetime, timezone
from pathlib import Path
import json
import sys

project_root = Path(sys.argv[1])
task_manifest_path = project_root / sys.argv[2]
omitted_worker_id = sys.argv[3]


def load_json(path: Path) -> dict:
    return json.loads(path.read_text())


def write_json(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n")


task_manifest = load_json(task_manifest_path)
timeline_path = project_root / task_manifest["historyTimelinePath"]
timeline = load_json(timeline_path)
runs = timeline.get("runs", [])
latest_run = None
if isinstance(runs, list):
    for item in reversed(runs):
        if isinstance(item, dict):
            latest_run = item
            break

if latest_run is None:
    raise SystemExit("timeline does not contain a latest run entry")

run_manifest_path = project_root / latest_run["runManifestPath"]
run_manifest = load_json(run_manifest_path)
team_state_path = project_root / run_manifest["teamStatePath"]
team_state = load_json(team_state_path)
report_path = project_root / run_manifest["expectedReportRecordPath"]
final_status_path = project_root / run_manifest["expectedStatusPath"]

status_by_worker_id = {
    "claude": "completed",
    "codex": "timeout",
    "antigravity": "error",
    "report-writer": "completed",
}
reason_by_worker_id = {
    "claude": "",
    "codex": "Validation fixture timeout",
    "antigravity": "Validation fixture execution error",
    "report-writer": "",
}

for worker in team_state.get("workers", []):
    if not isinstance(worker, dict):
        continue
    worker_id = str(worker.get("workerId", "")).strip()
    if not worker_id:
        continue
    worker["status"] = status_by_worker_id.get(worker_id, "not-run")
    worker["reason"] = reason_by_worker_id.get(worker_id, "Validation fixture not used")

    prompt_relative = str(worker.get("promptPath", "")).strip()
    if prompt_relative:
        prompt_path = project_root / prompt_relative
        prompt_path.parent.mkdir(parents=True, exist_ok=True)
        prompt_lines = [
            "**Worker Error Contract Path:** /okstra/templates/worker-error-contract.md",
            f"**Errors log path:** {project_root}/run/logs/errors.jsonl",
            f"**Errors sidecar path:** {project_root}/run/worker-results/"
            f"{worker_id}-errors.json",
            f"Assigned worker prompt history path: {prompt_relative}",
            "**Prompt Delivery Mode:** eager-include",
            f"**Model:** Fixture worker, {worker.get('modelExecutionValue', '')}",
        ]
        if worker_id != "report-writer":
            prompt_lines.append(
                f"- Primary analysis packet: `{run_manifest.get('analysisPacketPath', '')}`"
            )
            prompt_lines.extend([
                "",
                "# Initial Analysis Prompt",
                f"Task Key: {task_manifest.get('taskKey', '')}",
                "Validation fixture prompt body.",
            ])
        else:
            prompt_lines.extend(["", "# Report Writer Prompt"])
        prompt_path.write_text(
            "\n".join(prompt_lines) + "\n"
        )

    result_relative = str(worker.get("resultPath", "")).strip()
    if worker["status"] == "completed" and result_relative:
        result_path = project_root / result_relative
        result_path.parent.mkdir(parents=True, exist_ok=True)
        result_path.write_text(
            "\n".join(
                [
                    "# Findings",
                    "- Validation fixture finding.",
                    "",
                    "# Missing Information or Assumptions",
                    "- None.",
                    "",
                    "# Safe or Reasonable Areas",
                    "- Fixture output format is valid.",
                    "",
                    "# Uncertain Points",
                    "- None.",
                    "",
                    "# Recommended Next Actions",
                    "- Continue validator coverage.",
                ]
            )
            + "\n"
        )
        # Mirror the audit sidecar contract — every completed worker-results
        # file ships alongside `<worker>-audit-<task-type>-<seq>.md` carrying
        # the Reading Confirmation block. Derive the sidecar path by
        # inserting `-audit` after the worker-role segment of the
        # result-file stem.
        result_stem = result_path.stem  # e.g. claude-worker-error-analysis-001
        audit_stem = result_stem.replace("-worker-", "-worker-audit-", 1)
        audit_path = result_path.with_name(f"{audit_stem}{result_path.suffix}")
        audit_lines = [
            f"# {worker.get('role', worker_id)} Audit",
            "",
            "- Read task-brief.md end-to-end (validation fixture).",
        ]
        if worker_id == "claude":
            # Heartbeat 계약 (agents/workers/claude-worker.md "Heartbeat") —
            # validate_session_conformance 가 claude-worker audit 사이드카의
            # `- PROGRESS:` cadence 를 검사하므로 fixture 도 계약을 준수한다.
            hb_ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
            audit_lines += [
                "",
                f"- PROGRESS: started {hb_ts}",
                f"- PROGRESS: read-task-brief.md {hb_ts}",
                f"- PROGRESS: analysis-start {hb_ts}",
                f"- PROGRESS: findings-draft-complete {hb_ts}",
                f"- PROGRESS: write-result-start {hb_ts}",
            ]
        audit_path.write_text("\n".join(audit_lines) + "\n")

lead = team_state.get("lead")
if isinstance(lead, dict):
    lead["status"] = "completed"
    # render-only fixture 에는 실 Claude 세션이 없어 sessionId 가 비어 있을 수
    # 있다 — session-conformance fixture jsonl 의 파일명과 맞춘 고정 id 를 부여.
    if not str(lead.get("sessionId") or "").strip():
        lead["sessionId"] = "fixture-lead-session-0001"
team_state["workflowState"] = "worker-results-collected"

# validate-run.py requires team-state.teamCreate.status == "implicit" once any
# worker has been dispatched (CC v2.1.178 removed TeamCreate; every session has
# an implicit team). Mirror that here so the fixture is a valid post-Phase-3 state.
team_state["teamCreate"] = {
    "attempted": False,
    "status": "implicit",
    "splitPane": False,
}

# Phase 7 token-usage collection is normally produced by okstra-token-usage.py.
# The validator (`team-state.usageSummary is empty`) treats absence as a contract
# violation, so the fixture must mirror that step with a synthetic-but-valid object.
team_state["usageSummary"] = {
    "leadTotalTokens": 0,
    "workerTotalTokens": 0,
    "grandTotalTokens": 0,
    "leadBillableEquivalentTokens": 0,
    "workerBillableEquivalentTokens": 0,
    "grandBillableEquivalentTokens": 0,
    "estimatedCostUsd": {
        "lead": 0.0,
        "claudeWorkers": 0.0,
        "cliWorkers": 0.0,
        "grandTotal": 0.0,
    },
    "collectedAt": "1970-01-01T00:00:00Z",
    "teamName": team_state.get("teamName", "validation-fixture"),
    "sessionsFound": 0,
    "definitions": {
        "totalTokens": "Validation fixture placeholder.",
        "billableEquivalentTokens": "Validation fixture placeholder.",
        "estimatedCostUsd": "Validation fixture placeholder.",
    },
    "intake": {
        "totalTokens": 0,
        "billableEquivalentTokens": 0,
        "estimatedCostUsd": 0,
    },
}

required_status_entries = run_manifest.get("teamContract", {}).get(
    "requiredAgentStatusEntries", []
)
if not isinstance(required_status_entries, list):
    required_status_entries = []

report_lines = [
    "# Validation Fixture Report",
    "",
    # Top-of-report Index — the renderer injects this on real runs; the
    # hand-crafted fixture mirrors it so validate_report's index/anchor
    # contract (introduced with the clickable-ID pass) is satisfied.
    '## Index <a id="report-index"></a>',
    "",
    "### Sections",
    "",
    "- [Verdict Card](#verdict-card)",
    "",
    "## Verdict Card",
    "",
    "| 항목 | 값 |",
    "|------|----|",
    "| Final Conclusion | validation fixture |",
    "| Verdict Token | `accepted` |",
    "| Direction | `continue-investigation` |",
    "| Approval Required? | `no` |",
    "| Next Step | fixture |",
    "",
    "## Agent Execution Status",
]
for label in required_status_entries:
    if isinstance(label, str) and label.strip():
        report_lines.append(f"- {label}: fixture status recorded")
report_lines.extend(
    [
        "",
        "## Token Usage Summary",
        "",
        "| 항목 | 처리 토큰 | 환산 토큰 | 비용 (USD) |",
        "|------|-----------|-----------|------------|",
        "| Lead | `1` | `1` | `$0.01` |",
        "| Worker 합계 | `1` | `1` | `$0.01` |",
        "| **전체 합계** | **`2`** | **`2`** | **`$0.02`** |",
        "| Codex/Antigravity CLI 추가 비용 |  |  | `$0.00` |",
        "",
        "## 7. Final Verdict",
        "",
        "| 항목 | 값 |",
        "|------|----|",
        "| Verdict Token | `accepted` |",
        "",
        "## 5.8 Final Verification Deliverables",
        "",
        "Source Implementation Report / Acceptance Blockers / Residual Risk / "
        "Validation Evidence / Read-only Command Log / Conditional Acceptance "
        "Conditions / Routing Recommendation: fixture stub.",
    ]
)
report_path.parent.mkdir(parents=True, exist_ok=True)
_report_name = report_path.name
if _report_name.endswith(".data.json"):
    data_path = report_path
    md_path = report_path.with_name(_report_name[: -len(".data.json")] + ".md")
elif _report_name.endswith(".md"):
    md_path = report_path
    data_path = report_path.with_name(_report_name[: -len(".md")] + ".data.json")
else:
    md_path = report_path
    data_path = report_path.with_suffix(".data.json")
md_path.write_text("\n".join(report_lines) + "\n")

# Phase 7 step 1.5 (BLOCKING) — render the html sibling artifact next
# to the final-report so validate-run.py's report-views hook passes.
# The workflow validator's fixture predates that step; we materialise
# the html file in-place using the same single-reference-point helper
# the CLI uses.
import os
WORKSPACE_ROOT = os.environ.get("OKSTRA_WORKSPACE_ROOT_FOR_FIXTURE", "")
if WORKSPACE_ROOT:
    # Write final-report .data.json SSOT next to the markdown. The validator's
    # validate_final_report_data() reads this via _data_path_for(report_path)
    # and the renderer treats it as the canonical source — the markdown alone
    # is no longer a valid run artifact (dual-format final-report rollout).
    # Sample bundled in tests/fixtures is patched with this run's task identity.
    task_type = str(task_manifest.get("taskType", ""))
    sample_path = (
        Path(WORKSPACE_ROOT)
        / "tests" / "fixtures" / "final-report-data-v2"
        / f"{task_type}-001.data.json"
    )
    if sample_path.is_file():
        sample = json.loads(sample_path.read_text(encoding="utf-8"))
        sample["frontmatter"]["taskGroup"] = str(task_manifest.get("taskGroup", ""))
        sample["frontmatter"]["taskId"] = str(task_manifest.get("taskId", ""))
        sample["frontmatter"]["taskType"] = task_type
        sample["frontmatter"]["projectId"] = str(task_manifest.get("projectId", ""))
        sample["header"]["taskKey"] = str(task_manifest.get("taskKey", ""))
        sample["header"]["taskType"] = task_type
        # The shipped fixture carries its own roster; this run's contract names
        # a different one, and the validator compares the report's agent rows
        # against that contract. Restate the rows under the contract's names so
        # the fixture exercises the check instead of tripping over it.
        _required = (
            (run_manifest.get("teamContract") or {}).get("requiredAgentStatusEntries")
            or (task_manifest.get("resultContract") or {}).get(
                "requiredAgentStatusEntries"
            )
            or []
        )
        _rows = sample.get("executionStatus") or []
        if _required and _rows:
            # `agent` is the runtime (a schema enum); `role` carries the label the
            # validator looks for in the rendered markdown.
            sample["executionStatus"] = [
                {**_rows[min(i, len(_rows) - 1)], "role": role}
                for i, role in enumerate(_required)
            ]
        data_path.write_text(
            json.dumps(sample, indent=2, ensure_ascii=False) + "\n",
            encoding="utf-8",
        )

    import sys as _sys
    _sys.path.insert(0, str(Path(WORKSPACE_ROOT) / "scripts"))
    # The markdown seeded above this block predates the data.json contract, so
    # re-render it from the data.json the fixture just wrote. Otherwise the pair
    # disagrees and the validator's AI-handoff heading scan fails on a fixture
    # that never claimed to be hand-authored.
    if sample_path.is_file():
        try:
            from okstra_ctl.render_final_report import render_to_file

            render_to_file(data_path, md_path)
        except Exception as exc:  # pragma: no cover — fixture path only
            raise SystemExit(f"failed to render final report in fixture: {exc}")
    # Go through the same CLI the run uses, so the fixture picks the schema's
    # own view (v2 task template) rather than a second copy of that choice.
    import subprocess as _subprocess
    _views = _subprocess.run(
        [
            _sys.executable,
            str(Path(WORKSPACE_ROOT) / "scripts" / "okstra-render-report-views.py"),
            str(data_path),
            "--task-key", str(task_manifest.get("taskKey", "validation/fixture")),
            "--task-type", str(task_manifest.get("taskType", "validation")),
            "--seq", "001",
        ],
        capture_output=True,
        text=True,
    )
    if _views.returncode != 0:
        raise SystemExit(
            f"failed to render report views in fixture: {_views.stderr or _views.stdout}"
        )

if final_status_path.exists():
    final_status_path.unlink()

# session-conformance fixture — validate_session_conformance 는 lead 세션
# jsonl 에서 run 윈도우 내 PROGRESS 체크포인트를 스캔한다. 계약을 준수한
# 합성 jsonl 을 주입 시드 디렉터리(.claude-projects-fixture)에 만들어 두고,
# 러너(run_validator_expectation)가 --claude-projects-dir 로 넘긴다.
lead_sid = str((team_state.get("lead") or {}).get("sessionId") or "").strip()
if lead_sid:
    now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")
    progress_lines = [
        "PROGRESS: phase-1-intake reading task bundle",
        "PROGRESS: phase-1-intake complete",
        f"PROGRESS: phase-2-prompts preparing {len(team_state.get('workers') or [])} worker prompts",
        "PROGRESS: phase-3-team-create using implicit team (split-pane=off)",
    ]
    for worker in team_state.get("workers", []):
        if not isinstance(worker, dict):
            continue
        worker_id = str(worker.get("workerId", "")).strip()
        status = str(worker.get("status", "")).strip()
        if not worker_id or worker_id == "report-writer":
            continue
        if status in ("completed", "timeout", "error"):
            progress_lines.append(
                f"PROGRESS: phase-4-dispatch worker={worker_id}-worker model=fixture"
            )
        if status == "completed":
            progress_lines.append(
                f"PROGRESS: phase-5-collect worker={worker_id}-worker status=completed"
            )
    progress_lines.append("PROGRESS: phase-batch-cleanup panes=1")
    progress_lines.append("PROGRESS: phase-6-synthesis dispatching report-writer-worker")
    progress_lines.append("PROGRESS: phase-7-persist updating manifests")
    records = [
        {
            "type": "assistant",
            "timestamp": now_iso,
            "message": {"content": [{"type": "text", "text": line}]},
        }
        for line in progress_lines
    ]
    # 인코딩 기준은 validator 가 쓰는 task-manifest.projectRoot — macOS 의
    # /tmp 심링크 때문에 셸의 $PROJECT_ROOT(/tmp/...)와 manifest 의
    # projectRoot(/private/tmp/...)가 다른 문자열일 수 있다.
    manifest_project_root = str(task_manifest.get("projectRoot") or project_root)
    encoded_cwd = "-" + manifest_project_root.strip("/").replace("/", "-")
    session_dir = project_root / ".claude-projects-fixture" / encoded_cwd
    session_dir.mkdir(parents=True, exist_ok=True)
    (session_dir / f"{lead_sid}.jsonl").write_text(
        "".join(json.dumps(r, ensure_ascii=False) + "\n" for r in records)
    )

write_json(team_state_path, team_state)
write_json(run_manifest_path, run_manifest)
write_json(task_manifest_path, task_manifest)

# The synthetic fixture accepts lead and completed-worker results, so it must
# create the same verified invocation records and result links as a real run.
# Keep the deliberately omitted worker prompt absent: the first validator pass
# still exercises the historical prompt-history failure before the helper
# restores that file.
if WORKSPACE_ROOT:
    from okstra_ctl.agent_invocation import (
        AgentInstruction,
        AgentInstructionSource,
        AgentInvocationRequest,
        agent_model_assignment_from_payload,
        invocation_execution_identity_from_manifest,
        prepare_agent_invocation,
    )
    from okstra_ctl.dispatch_state import (
        link_agent_dispatch_result,
        record_verified_agent_dispatch,
    )
    # The analysis duty depends on the task type, exactly as it does in a real
    # run — a fixture that assumed one audience for every phase would be
    # rejected by the manifest's own allowlist.
    from okstra_ctl.worker_prompt_policy import ANALYSIS_DUTY_BY_TASK_TYPE

    analysis_audience = ANALYSIS_DUTY_BY_TASK_TYPE.get(
        str(run_manifest.get("taskType") or ""), "analysis-worker"
    )

    lead_record = record_verified_agent_dispatch(
        project_root=project_root,
        run_manifest_path=run_manifest_path,
        metadata_path=project_root / run_manifest["leadPromptMetadataPath"],
        enforcement_mode="host-native-spec-link-gate",
    )
    link_agent_dispatch_result(
        project_root=project_root,
        run_manifest_path=run_manifest_path,
        dispatch_id=lead_record["dispatchId"],
        result_path=report_path,
    )

    assignments = run_manifest["invocationAssignments"]
    contract = run_manifest["agentContract"]
    reservation_root = project_root / contract["invocationReservationRootPath"]
    for worker in team_state.get("workers", []):
        if not isinstance(worker, dict):
            continue
        worker_id = str(worker.get("workerId") or "").strip()
        prompt_relative = str(worker.get("promptPath") or "").strip()
        result_relative = str(worker.get("resultPath") or "").strip()
        if not worker_id or not prompt_relative or not result_relative:
            continue
        source_prompt = project_root / prompt_relative
        if not source_prompt.is_file():
            continue
        assignment_ref = f"initial/{worker_id}"
        invocation_id = f"validation-fixture-{worker_id}"
        invocation_prompt = reservation_root / f"{invocation_id}.prompt.md"
        audience = (
            "report-writer" if worker_id == "report-writer"
            else analysis_audience
        )
        assignment = agent_model_assignment_from_payload(
            assignments[assignment_ref]
        )
        identity = invocation_execution_identity_from_manifest(
            run_manifest,
            assignment=assignment,
            assignment_ref=assignment_ref,
            duty_id=audience,
        )
        identity_args = {
            "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,
        }
        prepared = prepare_agent_invocation(AgentInvocationRequest(
            invocation_id=invocation_id,
            worker_id=worker_id if identity is None else None,
            audience=audience,
            assignment_ref=assignment_ref,
            purpose=None,
            assignment=assignment,
            instruction=AgentInstruction(
                anchor_lines=(),
                body=source_prompt.read_text(encoding="utf-8"),
                source_paths=(
                    AgentInstructionSource("project", prompt_relative),
                ),
            ),
            project_root=project_root,
            run_manifest_path=run_manifest_path,
            duty_root=project_root / contract["dutyRootPath"],
            prompt_path=invocation_prompt,
            metadata_path=invocation_prompt.with_name(
                invocation_prompt.name + ".meta.json"
            ),
            dispatch_kind="validation-fixture",
            **identity_args,
        ))
        dispatch = record_verified_agent_dispatch(
            project_root=project_root,
            run_manifest_path=run_manifest_path,
            metadata_path=prepared.metadata_path,
            enforcement_mode=(
                "host-native-spec-link-gate"
                if prepared.assignment.runner == "native-session"
                else "core-pre-dispatch"
            ),
        )
        if worker.get("status") == "completed":
            link_agent_dispatch_result(
                project_root=project_root,
                run_manifest_path=run_manifest_path,
                dispatch_id=dispatch["dispatchId"],
                result_path=project_root / result_relative,
            )

# Preserve the workflow's historical missing-prompt negative after the v2
# invocation record has captured the prompt bytes that were actually sent.
for worker in team_state.get("workers", []):
    if isinstance(worker, dict) and worker.get("workerId") == omitted_worker_id:
        omitted_prompt = str(worker.get("promptPath") or "").strip()
        if omitted_prompt:
            (project_root / omitted_prompt).unlink(missing_ok=True)
        break
PY
}