Foundry Iq

helpers/file_cu_canary.py

"""Approval-gated one-file OCR canary; never invoked automatically by ingestion."""
from __future__ import annotations

import argparse
import copy
import hashlib
import json
import re
import secrets
import sys
import time
import zlib
from email import policy
from email.parser import BytesParser
from pathlib import Path
from urllib.parse import urlencode

try:
    from . import (
        _bootstrap_io, file_source, file_cu_mi, search_reconcile, _file_inventory, _file_source_config,
        _search_read,
    )
    from ._common import (
        HelperFailure, MANAGEMENT_AUDIENCE, SEARCH_AUDIENCE, azure_cli_token, blocked_result, digest,
        emit_result, http_request, load_approved_input, require_allowed_fields, reject_secrets,
    )
except ImportError:
    import _bootstrap_io
    import file_source
    import file_cu_mi
    import search_reconcile
    import _file_inventory
    import _file_source_config
    import _search_read
    from _common import (
        HelperFailure, MANAGEMENT_AUDIENCE, SEARCH_AUDIENCE, azure_cli_token, blocked_result, digest,
        emit_result, http_request, load_approved_input, require_allowed_fields, reject_secrets,
    )


PDF_NAME = "cu-mi-probe.pdf"
MARKER = re.compile(r"CUOCR[2-9]{8}")
FIELD = re.compile(r"[A-Za-z][A-Za-z0-9_]{0,127}")
# Pinned implementation, not deployed proof; see references/file-cu-canary.md.
# Never populate this from caller attestations or an arbitrary readback field.
OCR_CONTENT_MAPPINGS = {
    "2026-08-01-preview": {
        "field": "snippet", "parent_field": "snippet_parent_id", "path_field": "metadata_storage_path",
        "authority": {
            "repository": "AzureSearch", "commit": "2e241af8939a0891836b78278df113dab31a5964",
            "contract": "file-standard-cu-index-v1",
        },
    },
}
DISCLOSURE = (
    "One synthetic image-only PDF is uploaded to Search and processed by billable CU OCR; "
    "CU may create an analyzer, content may cross regions, and Search retains generated data. "
    "No embeddings, KB/chat or direct CU probes. Existing resources/roles/network/local-auth stay unchanged. "
    "Client time/attempt caps are not a monetary cap or cancellation of remote processing. "
    "Cleanup needs separate approval; retain the private source/file/index inventory."
)
GLYPHS = {
    "C": ("01111", "10000", "10000", "10000", "10000", "10000", "01111"),
    "U": ("10001", "10001", "10001", "10001", "10001", "10001", "01110"),
    "O": ("01110", "10001", "10001", "10001", "10001", "10001", "01110"),
    "R": ("11110", "10001", "10001", "11110", "10100", "10010", "10001"),
    "2": ("01110", "10001", "00001", "00010", "00100", "01000", "11111"),
    "3": ("11110", "00001", "00001", "01110", "00001", "00001", "11110"),
    "4": ("00010", "00110", "01010", "10010", "11111", "00010", "00010"),
    "5": ("11111", "10000", "10000", "11110", "00001", "00001", "11110"),
    "6": ("01110", "10000", "10000", "11110", "10001", "10001", "01110"),
    "7": ("11111", "00001", "00010", "00100", "01000", "01000", "01000"),
    "8": ("01110", "10001", "10001", "01110", "10001", "10001", "01110"),
    "9": ("01110", "10001", "10001", "01111", "00001", "00001", "01110"),
}


def fail(code, message, *, partial=False, request_id=None, status=None):
    return HelperFailure(code, message, blocked_at="file-cu-canary", partial=partial, request_id=request_id, status=status)


def pdf_bytes(marker):
    if not isinstance(marker, str) or MARKER.fullmatch(marker) is None:
        raise fail("canary-marker-invalid", "Use CUOCR followed by eight digits 2–9.")
    scale, margin = 12, 24
    width, height = len(marker) * 6 * scale + 2 * margin, 7 * scale + 2 * margin
    image = bytearray(b"\xff" * width * height)
    for n, letter in enumerate(marker):
        for y, row in enumerate(GLYPHS[letter]):
            for x, pixel in enumerate(row):
                if pixel == "1":
                    for yy in range(scale):
                        start = (margin + y * scale + yy) * width + margin + (n * 6 + x) * scale
                        image[start:start + scale] = b"\x00" * scale
    compressed = zlib.compress(bytes(image), 9)
    draw = b"q 600 0 0 84 20 20 cm /Image0 Do Q\n"
    objects = [
        b"<< /Type /Catalog /Pages 2 0 R >>",
        b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
        b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 640 124] /Resources << /XObject << /Image0 4 0 R >> >> /Contents 5 0 R >>",
        f"<< /Type /XObject /Subtype /Image /Width {width} /Height {height} /ColorSpace /DeviceGray /BitsPerComponent 8 /Filter /FlateDecode /Length {len(compressed)} >>\nstream\n".encode()
        + compressed + b"\nendstream",
        f"<< /Length {len(draw)} >>\nstream\n".encode() + draw + b"endstream",
    ]
    out = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
    offsets = [0]
    for n, obj in enumerate(objects, 1):
        offsets.append(len(out))
        out.extend(f"{n} 0 obj\n".encode() + obj + b"\nendobj\n")
    xref = len(out)
    out.extend(b"xref\n0 6\n0000000000 65535 f \n")
    for offset in offsets[1:]:
        out.extend(f"{offset:010d} 00000 n \n".encode())
    out.extend(f"trailer\n<< /Size 6 /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n".encode())
    if len(out) > 65536 or marker.encode() in out:
        raise fail("canary-pdf-invalid", "Probe must be bounded image-only bytes without a text marker.")
    return bytes(out)


def prepare(directory, marker=None):
    directory = _bootstrap_io.private_directory(directory)
    marker = marker or "CUOCR" + "".join(secrets.choice("23456789") for _ in range(8))
    data = pdf_bytes(marker)
    _bootstrap_io.private_bytes(directory, PDF_NAME, data)
    manifest = {"file": PDF_NAME, "marker": marker, "sha256": hashlib.sha256(data).hexdigest(),
                "bytes": len(data), "pages": 1, "text_layer": False}
    path = _bootstrap_io.private_file(directory, "cu-mi-probe-manifest.json", manifest)
    return {"status": "prepared", "manifest": str(path), "writes_performed": [],
            "azure_operations": 0, "note": "Local synthetic artifacts only; no live approval or resource selection."}


def _bounds(value):
    if not isinstance(value, dict):
        raise fail("canary-bounds-missing", "Select explicit timeout, HTTP request, polling attempt and interval caps.")
    ranges = {"timeout_seconds": (30, 600), "max_requests": (16, 100),
              "max_poll_attempts": (1, 10), "poll_interval_seconds": (1, 30)}
    require_allowed_fields(value, set(ranges), label="canary bounds")
    if any(type(value.get(k)) is not int or not low <= value[k] <= high for k, (low, high) in ranges.items()):
        raise fail("canary-bounds-invalid", "Use timeout 30–600s, HTTP requests 16–100, polls 1–10 and interval 1–30s.")


def _ocr_mapping(api_version, field):
    mapping = OCR_CONTENT_MAPPINGS.get(api_version)
    if mapping is None:
        raise fail("canary-ocr-mapping-unverified",
                   "No pinned File Standard OCR-to-index field mapping is verified for this API. "
                   "Do not create/upload; verify the first-party mapping and update the supported contract before planning.")
    if field != mapping["field"]:
        raise fail("canary-field-unverified", "The selected field does not match the pinned File Standard OCR-content mapping.")
    return copy.deepcopy(mapping)


def _verified_file(source_result, ingestion):
    files = source_result["verification"]["readback"].get("files")
    expected = ingestion["files"][0]
    if (
        not isinstance(files, list) or len(files) != 1 or not isinstance(files[0], dict)
        or not isinstance(files[0].get("fileId"), str) or not files[0]["fileId"].strip()
        or files[0].get("fileName") != expected["path"]
        or files[0].get("sha256") != expected["sha256"] or files[0].get("size") != expected["size"]
    ):
        raise fail("canary-file-identity-unverified", "Require the uploaded synthetic file's exact ID, path, hash and size readback.")
    return files[0]


def _verify_index_fields(index, mapping):
    fields = index.get("fields")
    if not isinstance(fields, list):
        raise fail("canary-field-unverified", "Generated index fields must match the pinned File OCR mapping.")
    for name in (mapping["field"], mapping["parent_field"], mapping["path_field"]):
        selected = [field for field in fields if isinstance(field, dict) and field.get("name") == name]
        if len(selected) != 1 or selected[0].get("type") != "Edm.String" or selected[0].get("retrievable") is not True:
            raise fail("canary-field-unverified", "Require unique retrievable string fields for pinned OCR content and file identity.")
        if name == mapping["field"] and (
            selected[0].get("searchable") is not True
            or any(selected[0].get(flag) is not False for flag in ("filterable", "sortable", "facetable"))
        ):
            raise fail("canary-field-unverified", "The canonical snippet field does not match the pinned searchable OCR-content schema.")


def _contains_marker(value, marker):
    if isinstance(value, str):
        return marker in re.sub(r"[^A-Z0-9]", "", value.upper())
    if isinstance(value, dict):
        return any(_contains_marker(k, marker) or _contains_marker(v, marker) for k, v in value.items())
    if isinstance(value, (list, tuple)):
        return any(_contains_marker(item, marker) for item in value)
    return False


def _check_non_image_channels(plan, marker):
    ingestion = plan["ingestion"]
    # Produce the actual multipart envelope with only the image bytes omitted.
    try:
        body, boundary = _file_inventory._multipart(ingestion, ingestion["files"][0], b"", digest(plan))
        message = BytesParser(policy=policy.default).parsebytes(
            f"Content-Type: multipart/form-data; boundary={boundary}\r\n\r\n".encode("ascii") + body
        )
    except (ValueError, UnicodeError, TypeError):
        raise fail("canary-upload-envelope-unverified", "The non-image upload envelope could not be produced safely.") from None
    parts = list(message.iter_parts())
    if (
        message.defects or len(parts) != 2 or any(part.defects for part in parts)
        or [part.get_param("name", header="content-disposition") for part in parts] != ["metadata", "content"]
        or parts[0].get_content_type() != "application/json" or parts[1].get_payload(decode=True) != b""
    ):
        raise fail("canary-upload-envelope-unverified", "The produced non-image upload envelope is not the supported multipart contract.")
    try:
        metadata = json.loads(parts[0].get_payload(decode=True))
        channels = [plan["source"]["desired"], _file_inventory._list_url(ingestion),
                    body.decode("utf-8"), metadata, [list(part.items()) for part in parts]]
    except (ValueError, UnicodeError, TypeError):
        raise fail("canary-upload-envelope-unverified", "The produced non-image upload envelope could not be inspected.") from None
    if _contains_marker(channels, marker):
        raise fail("canary-marker-in-metadata", "The OCR marker must not occur in source/upload names, owner, metadata or other non-image upload channels.")


def _file_contract(plan, marker):
    expected = pdf_bytes(marker)
    if not isinstance(plan, dict) or not isinstance(plan.get("ingestion"), dict):
        raise fail("canary-file-contract-invalid", "Retain the complete File plan.")
    root = _file_inventory.resolve_local_root(plan["ingestion"].get("local_root"))
    actual = _file_inventory._resolve_inventory_path(root, PDF_NAME)
    try:
        if actual.stat().st_size != len(expected):
            raise fail("canary-content-drift", "Probe size differs from the bounded synthetic image-only PDF.")
    except OSError:
        raise fail("canary-content-drift", "Selected synthetic PDF is unreadable.") from None
    _file_source_config._validate_plan(plan)
    source, ingestion = plan["source"], plan["ingestion"]
    settings = source["desired"]["fileParameters"]["ingestionParameters"]
    if (
        plan.get("file_cu_plan_version") != "1.2" or source.get("action") != "create"
        or source["api_version"] != "2026-08-01-preview"
        or settings.get("contentExtractionMode") != "standard" or "embedding" in plan
        or settings.get("embeddingModel") is not None or settings.get("chatCompletionModel") is not None
        or settings.get("disableImageVerbalization") is not True
        or len(ingestion["files"]) != 1 or ingestion["files"][0]["path"] != PDF_NAME
    ):
        raise fail("canary-file-contract-invalid", "Canary requires fresh conditional File Standard MI creation, one synthetic PDF, no embedding/chat and unchanged August API.")
    _check_non_image_channels(plan, marker)
    records = _file_inventory.snapshot_inventory(root, [PDF_NAME], service_tier=ingestion["service_tier"])
    if records != ingestion["files"] or records[0]["sha256"] != "sha256:" + hashlib.sha256(expected).hexdigest():
        raise fail("canary-content-drift", "Synthetic inventory changed; regenerate the concrete plan before approval.")


def plan_canary(request, *, token_provider=azure_cli_token, transport=http_request, now=time.time):
    if not isinstance(request, dict):
        raise fail("canary-input-invalid", "Canary intent must be an object.")
    require_allowed_fields(request, {"schema_version", "file_request", "marker", "content_field", "bounds", "receipt_directory"},
                           label="File CU canary intent")
    reject_secrets(request)
    _bounds(request.get("bounds"))
    if request.get("schema_version") != "1.0" or not isinstance(request.get("content_field"), str) or not FIELD.fullmatch(request["content_field"]):
        raise fail("canary-input-invalid", "Use schema 1.0 and an exact simple generated content field name.")
    pdf_bytes(request.get("marker"))
    directory = _bootstrap_io.private_directory(request.get("receipt_directory"))
    fr = request.get("file_request")
    if not isinstance(fr, dict):
        raise fail("canary-input-invalid", "Supply an explicit existing Search/CU File request; no resource defaults or provisioning.")
    # Reject a key mode before even read-only resource discovery.
    cu = fr.get("content_understanding")
    if not isinstance(cu, dict) or cu.get("auth", "system-assigned") != "system-assigned":
        raise fail("canary-auth-invalid", "This canary only tests File managed identity; no keys or fallback.")
    if fr.get("paths") != [PDF_NAME] or fr.get("extraction_mode") != "standard" or fr.get("vectorization") != "none":
        raise fail("canary-file-contract-invalid", "Select exactly the generated probe PDF, Standard extraction and no vectors.")
    version = fr.get("api_version", _file_inventory.API_VERSION)
    _file_inventory.validate_api_version(version)
    mapping = _ocr_mapping(version, request["content_field"])
    root = _file_inventory.resolve_local_root(fr.get("local_root"))
    selected = _file_inventory._resolve_inventory_path(root, PDF_NAME)
    try:
        if selected.stat().st_size != len(pdf_bytes(request["marker"])):
            raise fail("canary-content-drift", "Selected PDF is not the bounded synthetic probe.")
    except OSError:
        raise fail("canary-content-drift", "Selected probe is unreadable.") from None
    result = file_source.plan_source(fr, token_provider=token_provider, transport=transport)
    fp = result["execution_input"]["plan"]
    _file_contract(fp, request["marker"])
    created = int(now())
    plan = {
        "operation": "validate-file-cu-mi", "version": "1.1", "file_plan": fp, "ocr_mapping": mapping,
        "marker": request["marker"], "content_field": request["content_field"],
        "bounds": copy.deepcopy(request["bounds"]), "receipt_directory": str(directory),
        "created_at": created, "expires_at": created + 900, "disclosure": DISCLOSURE,
    }
    fingerprint = digest(plan)
    return {
        "status": "planned", "plan_fingerprint": fingerprint,
        "execution_input": {"schema_version": "1.0", "plan": plan,
                            "approval": {"confirmed": False, "fingerprint": fingerprint}},
        "approval_summary": {"source": result["approval_summary"], "bounds": plan["bounds"],
                             "expires_at": plan["expires_at"], "disclosure": DISCLOSURE,
                             "expected_outcome": "Indexed OCR marker without client keys; principal attribution remains separate."},
        "writes_performed": [],
    }


class BoundedTransport:
    def __init__(self, plan, raw, clock, record):
        self.plan, self.raw, self.clock, self.record = plan, raw, clock, record
        self.deadline = clock() + plan["bounds"]["timeout_seconds"]
        self.calls, self.puts, self.uploads = 0, 0, 0
        self.upload_ack_failure = None
        self.index_url = None
        fp = plan["file_plan"]
        self.source_url = _search_read.resource_url(fp["source"])
        self.files_prefix = self.source_url.split("?")[0] + "/files"
        cu = fp["content_understanding"]
        mi = cu["managed_identity"]
        self.arm_urls = {
            f"{MANAGEMENT_AUDIENCE}{cu['resource_id']}?api-version=2024-10-01",
            f"{MANAGEMENT_AUDIENCE}{mi['search_resource_id']}?api-version={file_cu_mi.SEARCH_API}",
            f"{MANAGEMENT_AUDIENCE}{mi['role_assignment_id']}?api-version={file_cu_mi.ROLE_API}",
        }

    def __call__(self, method, url, token, **kwargs):
        if self.upload_ack_failure is not None:
            raise self.upload_ack_failure
        remaining = self.deadline - self.clock()
        if remaining <= 0 or self.calls >= self.plan["bounds"]["max_requests"]:
            raise fail("canary-bound-reached", "Client request/deadline cap reached; remote processing may continue.", partial=bool(self.puts))
        file_url = url.startswith(self.files_prefix + "?")
        allowed = method == "GET" and (
            url in self.arm_urls or url == self.source_url or file_url
            or self.index_url is not None and (url == self.index_url or url.startswith(self.index_url.split("?")[0] + "/docs?"))
        )
        if method == "PUT" and url == self.source_url and self.puts == 0 and kwargs.get("headers", {}).get("If-None-Match") == "*":
            allowed = True
            self.puts += 1
        elif method == "POST" and file_url and self.uploads == 0:
            allowed = True
            self.uploads += 1
        if not allowed:
            raise fail("canary-operation-forbidden", "Only exact resource GETs, one conditional source PUT and one selected upload are approved.", partial=bool(self.puts))
        self.calls += 1
        kwargs.update(follow_redirects=False, max_response_bytes=min(524288, kwargs.get("max_response_bytes", 524288)),
                      response_deadline=min(self.deadline, kwargs.get("response_deadline", self.deadline)),
                      timeout=max(0.01, min(float(kwargs.get("timeout", 60)), remaining)))
        try:
            response = self.raw(method, url, token, **kwargs)
        except HelperFailure as failure:
            failure.recovery_deadline = min(self.deadline, failure.recovery_deadline or self.deadline)
            raise
        if method == "POST" and response.status in (200, 201, 202):
            try:
                self.record("upload-http-ack", {"status": response.status, "request_id": response.request_id,
                                               "ingestion": "not yet verified"})
            except HelperFailure as error:
                # Return the real ACK: persistence is not an ambiguous upload.
                self.upload_ack_failure = error
        return search_reconcile.HttpResult(
            response.status, response.body, response.headers, response.etag_values,
            min(self.deadline, response.recovery_deadline or self.deadline),
            self.upload_ack_failure,
        )


def execute(document, *, token_provider=azure_cli_token, transport=http_request, now=time.time,
            clock=time.monotonic, sleep=time.sleep):
    if not isinstance(document, dict):
        raise fail("canary-input-invalid", "Canary execution requires an object envelope.")
    require_allowed_fields(document, {"schema_version", "plan", "approval", "_computed_fingerprint"}, label="canary envelope")
    if document.get("schema_version") != "1.0":
        raise fail("canary-input-invalid", "Use canary envelope schema_version 1.0.")
    plan = document.get("plan")
    if not isinstance(plan, dict):
        raise fail("canary-input-invalid", "Use the unchanged canary execution envelope.")
    require_allowed_fields(plan, {"operation", "version", "file_plan", "marker", "content_field", "ocr_mapping", "bounds",
                                  "receipt_directory", "created_at", "expires_at", "disclosure"}, label="canary plan")
    fingerprint = digest(plan)
    if document.get("approval") != {"confirmed": True, "fingerprint": fingerprint} or document.get("_computed_fingerprint") != fingerprint:
        raise fail("approval-missing", "Exact canary scope, synthetic content/cost and bounded verification need unchanged fingerprinted approval.")
    if (
        plan.get("operation") != "validate-file-cu-mi" or plan.get("version") != "1.1"
        or plan.get("disclosure") != DISCLOSURE or not isinstance(plan.get("content_field"), str)
        or not FIELD.fullmatch(plan["content_field"])
        or type(plan.get("created_at")) is not int or type(plan.get("expires_at")) is not int
        or plan["expires_at"] - plan["created_at"] != 900 or not plan["created_at"] <= now() < plan["expires_at"]
    ):
        raise fail("canary-plan-stale", "Canary plan is invalid or outside its 15-minute approval window; refresh discovery.")
    _bounds(plan.get("bounds"))
    _file_contract(plan["file_plan"], plan["marker"])
    if plan.get("ocr_mapping") != _ocr_mapping(plan["file_plan"]["source"]["api_version"], plan["content_field"]):
        raise fail("canary-ocr-mapping-stale", "The approved OCR mapping differs from the pinned helper contract; a fresh plan and approval are required.")
    directory = _bootstrap_io.private_directory(plan["receipt_directory"])
    prefix = "cu-mi-" + fingerprint.split(":")[1][:16]
    refs = []

    def record(stage, value):
        path = _bootstrap_io.private_file(directory, prefix + "-" + stage + ".json", {
            "fingerprint": fingerprint, "stage": stage, "value": value,
        })
        refs.append(str(path))

    record("started", {"approved_input": {k: v for k, v in document.items() if k != "_computed_fingerprint"},
                       "state": "No Azure mutation yet; later completion is not atomic."})
    bounded = BoundedTransport(plan, transport, clock, record)
    fp = plan["file_plan"]
    child_digest = digest(fp)
    source_result = None
    def source_ack(**evidence):
        response = evidence["response"]
        record("source-http-ack", {"status": response.status, "request_id": response.request_id,
                                   "etag_evidence": _search_read.response_etags(response)})

    try:
        source_result = file_source.execute(
            {"schema_version": "1.0", "plan": fp, "_computed_fingerprint": child_digest,
             "approval": {"confirmed": True, "fingerprint": child_digest}},
            token_provider=token_provider, transport=bounded, mi_on_created=source_ack,
            allow_upload_retry=False,
        )
        if bounded.upload_ack_failure is not None:
            raise bounded.upload_ack_failure
        record("file-completed", source_result)
        original_file = _verified_file(source_result, fp["ingestion"])
        mapping = plan["ocr_mapping"]
        token = token_provider(SEARCH_AUDIENCE)
        current, _ = _search_read.read_resource(bounded.source_url, token, transport=bounded)
        _file_source_config.verify_content_understanding_readback(fp["content_understanding"], current)
        created = current.get("fileParameters", {}).get("createdResources")
        if not isinstance(created, dict) or set(created) != {"index"} or not isinstance(created["index"], str) or not re.fullmatch(r"[a-z0-9][a-z0-9_-]{1,127}", created["index"]):
            raise fail("canary-index-unverified", "Fresh File readback must identify exactly one generated index.")
        retained = source_result["verification"]["readback"]["source"]
        if current.get("@odata.etag") != retained["etag"] or not _search_read.definitions_match(fp["source"]["desired"], current):
            raise fail("canary-source-drift", "Source version changed before indexed OCR verification.")
        version = fp["source"]["api_version"]
        bounded.index_url = f"{fp['source']['endpoint'].rstrip('/')}/indexes('{created['index']}')?api-version={version}"
        index, _ = _search_read.read_resource(bounded.index_url, token, transport=bounded)
        if not isinstance(index, dict) or index.get("name") != created["index"] or not isinstance(index.get("@odata.etag"), str) or not index["@odata.etag"]:
            raise fail("canary-index-unverified", "Require the generated index's matching name and fresh ETag.")
        _verify_index_fields(index, mapping)
        query = bounded.index_url.split("?")[0] + "/docs?" + urlencode({
            "api-version": version, "search": "*",
            "$select": ",".join((mapping["field"], mapping["parent_field"], mapping["path_field"])), "$top": 3,
        })
        matched = False
        for attempt in range(plan["bounds"]["max_poll_attempts"]):
            response = bounded("GET", query, token)
            if response.status != 200 or not isinstance(response.body, dict) or not isinstance(response.body.get("value"), list):
                raise fail("canary-index-query-failed", "Generated-index query did not return bounded document rows.",
                           request_id=response.request_id, status=response.status)
            rows = response.body["value"]
            if len(rows) > 3:
                raise fail("canary-index-query-failed", "Generated-index response exceeded the approved row cap.")
            if any(
                not isinstance(row, dict) or not isinstance(row.get(mapping["field"]), str)
                or row.get(mapping["parent_field"]) != original_file["fileId"]
                or row.get(mapping["path_field"]) != original_file["fileName"]
                for row in rows
            ):
                raise fail("canary-document-binding-unverified", "Indexed rows must contain canonical OCR text and the uploaded file's exact parent ID/path; chunk IDs or metadata alone are not proof.")
            matched = any(_contains_marker(row[mapping["field"]], plan["marker"]) for row in rows)
            if matched:
                break
            if attempt + 1 < plan["bounds"]["max_poll_attempts"]:
                sleep(min(plan["bounds"]["poll_interval_seconds"], max(0, bounded.deadline - clock())))
        if not matched:
            raise fail("canary-ocr-unverified", "Upload/source creation is not extraction proof: indexed OCR marker was not observed within the caps.")
        after, _ = _search_read.read_resource(bounded.source_url, token, transport=bounded)
        if not isinstance(after, dict) or after.get("@odata.etag") != current.get("@odata.etag") or after.get("fileParameters", {}).get("createdResources") != created:
            raise fail("canary-source-drift", "Generated source/index binding changed during OCR verification.")
        final_index, _ = _search_read.read_resource(bounded.index_url, token, transport=bounded)
        if final_index != index:
            raise fail("canary-index-drift", "Generated index changed during OCR verification.")
        binding, _ = file_cu_mi.read_binding(fp["content_understanding"], fp["source"]["endpoint"],
                                             token_provider=token_provider, transport=bounded)
        if binding != fp["cu_identity_state"]:
            raise fail("canary-identity-drift", "Search/CU identity-role binding changed during validation.")
        result = {**copy.deepcopy(source_result), "outcome": "validate-file-cu-mi",
                  "approved_plan": {"fingerprint": fingerprint, "confirmed": True},
                  "status": "completed", "verdict": "keyless-functional-pass", "indexed_ocr_marker": True,
                  "principal_attribution": "unverified", "backend_rollout": "unverified",
                  "identity_evidence": fp["cu_identity_state"], "source_result": source_result,
                  "ocr_evidence": {"mapping": mapping, "file": original_file},
                  "index": created["index"], "request_count": bounded.calls,
                  "cleanup": {"status": "separate-plan-and-approval-required",
                              "instructions": "Retain inventory; separately approved guarded source cleanup only."},
                  "warnings": ["Functional evidence is for this run only, not an atomic deployment or release claim.",
                               "No backend CU principal/telemetry was observed."]}
    except HelperFailure as error:
        if bounded.upload_ack_failure is not None and error is not bounded.upload_ack_failure:
            checkpoint_error = bounded.upload_ack_failure
            checkpoint_error.writes = error.writes
            checkpoint_error.resources_remaining = error.resources_remaining
            checkpoint_error.resources_reused = error.resources_reused
            checkpoint_error.resources_unverified = error.resources_unverified
            checkpoint_error.warnings.extend(error.warnings)
            checkpoint_error.partial = error.partial
            error = checkpoint_error
        if source_result is not None:
            error.partial = True
            error.writes = [
                {**resource, "action": "created", "type": resource.get("type", "knowledge-source-file")}
                for resource in source_result["resources"]["created"]
            ] + error.writes
            error.resources_remaining = source_result["ownership"]["run_owned"] + error.resources_remaining
        result = blocked_result(error, outcome="validate-file-cu-mi", fingerprint=fingerprint, owner=fp["owner"])
        result.update(verdict="unverified", indexed_ocr_marker=False, principal_attribution="unverified",
                      backend_rollout="unverified", request_count=bounded.calls)
        result.setdefault("warnings", []).append("Remote processing/costs may continue; do not replay mutations or infer ownership from an ambiguous create.")
        if bounded.upload_ack_failure is not None:
            result["warnings"].append("Upload HTTP ACK could not be retained; validation stopped. Only listed receipt_refs are confirmed; do not replay the upload.")
    try:
        record("result", result)
    except HelperFailure:
        if result["status"] == "completed":
            error = fail("canary-receipt-failed", "Validation finished but its final private receipt could not be persisted.", partial=True)
            error.writes = [
                {**resource, "action": "created", "type": resource.get("type", "knowledge-source-file")}
                for resource in source_result["resources"]["created"]
            ]
            error.resources_remaining = source_result["ownership"]["run_owned"]
            result = {**blocked_result(error, outcome="validate-file-cu-mi", fingerprint=fingerprint, owner=fp["owner"]),
                      "indexed_ocr_marker": True, "principal_attribution": "unverified", "backend_rollout": "unverified",
                      "source_result": source_result, "request_count": bounded.calls}
        result["verdict"] = "unverified"
        result.setdefault("warnings", []).append("Final private receipt persistence failed; retain this ownership handoff and existing checkpoints. Do not replay.")
    return {**result, "receipt_refs": refs}


def main(argv=None):
    parser = argparse.ArgumentParser()
    modes = parser.add_mutually_exclusive_group(required=True)
    modes.add_argument("--prepare", metavar="EXISTING_PRIVATE_DIRECTORY")
    modes.add_argument("--plan", type=Path)
    modes.add_argument("--input", type=Path)
    args = parser.parse_args(argv)
    try:
        if args.prepare:
            result = prepare(args.prepare)
        elif args.plan:
            result = plan_canary(_bootstrap_io.read_json(args.plan))
            directory = result["execution_input"]["plan"]["receipt_directory"]
            path = _bootstrap_io.private_file(directory, "cu-mi-plan-" + result["plan_fingerprint"].split(":")[1][:16] + ".json",
                                               result["execution_input"])
            result = {k: v for k, v in result.items() if k != "execution_input"}
            result["execution_input_ref"] = str(path)
        else:
            document, _, fingerprint = load_approved_input(args.input)
            document["_computed_fingerprint"] = fingerprint
            result = execute(document)
        emit_result(result)
        return 0 if result["status"] in ("prepared", "planned", "completed") else 3 if result["status"] == "partial" else 2
    except HelperFailure as error:
        emit_result(blocked_result(error, outcome="validate-file-cu-mi", fingerprint=None, owner=None))
        return 3 if error.partial else 2


if __name__ == "__main__":
    sys.exit(main())