Foundry Iq

helpers/source_vector.py

from __future__ import annotations

import argparse
import copy
import json
import math
import re
import sys
import time
from pathlib import Path
from typing import Any

try:
    from ._common import (
        SEARCH_AUDIENCE, HelperFailure, TokenProvider, Transport, azure_cli_token, blocked_result, digest,
        emit_result, http_request, load_approved_input, odata_name, reject_secrets, require_allowed_fields,
    )
    from . import (
        _blob_source_read, _embedding, _file_inventory, _file_source_config, _search_read, _source_readback,
    )
except ImportError:
    from _common import (
        SEARCH_AUDIENCE, HelperFailure, TokenProvider, Transport, azure_cli_token, blocked_result, digest,
        emit_result, http_request, load_approved_input, odata_name, reject_secrets, require_allowed_fields,
    )
    import _blob_source_read
    import _embedding
    import _file_inventory
    import _file_source_config
    import _search_read
    import _source_readback


IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,127}$")


def _read_json(path: Any) -> dict[str, Any]:
    if not _embedding._text(path):
        raise _embedding.fail("input-schema-invalid", "Select an explicit JSON file.")
    try:
        value = json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, UnicodeError, ValueError) as exc:
        raise _embedding.fail("input-unreadable", "Selected evidence must be readable UTF-8 JSON.") from exc
    if not isinstance(value, dict):
        raise _embedding.fail("input-schema-invalid", "Selected JSON must contain an object.")
    _embedding._json_valid(value)
    return value


def _source_plan(path: str) -> dict[str, Any]:
    document = _read_json(path)
    reject_secrets(document)
    require_allowed_fields(document, {"schema_version", "plan", "approval"}, label="source artifact")
    plan, approval = document.get("plan"), document.get("approval")
    if (
        document.get("schema_version") != "1.0" or not isinstance(plan, dict)
        or not isinstance(approval, dict) or type(approval.get("confirmed")) is not bool
        or approval.get("fingerprint") != digest(plan)
    ):
        raise _embedding.fail("source-artifact-invalid", "Retain the unchanged source execution_input; prior consent is not query consent.")
    require_allowed_fields(approval, {"confirmed", "fingerprint"}, label="source approval")
    source = plan.get("source")
    desired = source.get("desired") if isinstance(source, dict) else None
    if (
        not isinstance(desired, dict) or desired.get("kind") not in ("file", "azureBlob")
        or not isinstance(source.get("api_version"), str)
        or source["api_version"] not in _search_read.SUPPORTED_API_VERSIONS
        or source.get("action") not in ("create", "reuse") or not _embedding._text(plan.get("owner"))
    ):
        raise _embedding.fail("source-artifact-invalid", "Source artifact needs a supported same-owner File/Blob definition and API.")
    parameters = desired.get("fileParameters" if desired["kind"] == "file" else "azureBlobParameters")
    if not isinstance(parameters, dict) or not isinstance(parameters.get("ingestionParameters"), dict):
        raise _embedding.fail("source-artifact-invalid", "Source ingestion parameters must be an object.")

    if plan.get("operation") == "reconcile-and-ingest":
        _file_source_config._validate_plan(plan)
    elif plan.get("operation") == "reconcile-and-monitor":
        _blob_source_read._validate_plan(plan)
    else:
        raise _embedding.fail("source-artifact-invalid", "Only File/Blob source execution artifacts are supported.")
    if "embedding" not in plan:
        if parameters["ingestionParameters"].get("embeddingModel") is None:
            raise _embedding.fail("embedding-not-configured", "Source vectorization is not configured; CU extraction does not enable vectors.")
        raise _embedding.fail("embedding-prerequisite-missing", "The source has vectors but lacks resolved embedding choice evidence; retain the original artifact and resolve a new plan, never rewrite historical consent.")
    _embedding.validate_plan_choice(plan)
    if parameters["ingestionParameters"].get("ingestionPermissionOptions") not in (None, []):
        raise _embedding.fail("permission-query-unsupported", "Permission-aware retrieval belongs to the KB workflow; direct-index verification is unsupported.")
    return plan


def _request(request: Any) -> dict[str, Any]:
    if not isinstance(request, dict):
        raise _embedding.fail("input-schema-invalid", "Verification input must be an object.")
    reject_secrets(request)
    require_allowed_fields(request, {
        "schema_version", "source_plan_file", "vector_field", "content_field", "citation_field",
        "expected_citation", "expected_text", "query", "mode", "k", "not_before",
        "reuse_input_file", "reuse_result_file",
    }, label="vector verification intent")
    try:
        json.dumps(request, ensure_ascii=False, allow_nan=False).encode("utf-8")
    except (UnicodeError, ValueError, TypeError) as exc:
        raise _embedding.fail("input-schema-invalid", "Verification choices must be valid UTF-8 JSON.") from exc
    if (
        request.get("schema_version") != "1.0" or request.get("mode") not in ("vector", "hybrid")
        or type(request.get("k")) is not int or not 1 <= request["k"] <= 10
        or any(not _embedding._text(request.get(key)) for key in ("query", "expected_citation", "expected_text"))
        or any(not isinstance(request.get(key), str) or not IDENTIFIER.fullmatch(request[key])
               for key in ("vector_field", "content_field", "citation_field"))
        or len({request["vector_field"], request["content_field"], request["citation_field"]}) != 3
    ):
        raise _embedding.fail("input-schema-invalid", "Resolve vector/hybrid, k 1–10, three distinct simple field names, query and expected citation/text.")
    plan = _source_plan(request.get("source_plan_file"))

    _blob_source_read._receipt_paths(request)
    if plan["source"]["desired"]["kind"] == "azureBlob":
        _blob_source_read._timestamp(request.get("not_before"))
    elif "not_before" not in request or request["not_before"] is not None:
        raise _embedding.fail("input-schema-invalid", "File verification requires explicit not_before null; uploads are synchronous.")
    return plan


def _named(items: Any, name: str) -> dict[str, Any]:
    if (
        not _embedding._text(name, maximum=128) or not isinstance(items, list)
        or not all(isinstance(item, dict) and _embedding._text(item.get("name"), maximum=128) for item in items)
        or len({item["name"] for item in items}) != len(items)
    ):
        raise _embedding.fail("vector-configuration-invalid", "Expected complete generated metadata arrays.")
    selected = [item for item in items if item.get("name") == name]
    if len(selected) != 1:
        raise _embedding.fail("vector-configuration-invalid", "Generated field/profile/vectorizer identity is absent or ambiguous.")
    return selected[0]


def verify_index(index: dict[str, Any], choice: dict[str, Any], request: dict[str, Any]) -> dict[str, Any]:
    fields = index.get("fields")
    vector = _named(fields, request["vector_field"])
    if index.get("permissionFilterOption") not in (None, "disabled") or any(field.get("permissionFilter") for field in fields):
        raise _embedding.fail("permission-query-unsupported", "Permission-enabled indexes require the owning permission-aware retrieval workflow.")
    low, high = _embedding.MODELS[choice["model"]]
    dimensions = vector.get("dimensions")
    if (
        vector.get("type") != "Collection(Edm.Single)" or vector.get("searchable") is not True
        or type(dimensions) is not int or not low <= dimensions <= high
    ):
        raise _embedding.fail("vector-configuration-invalid", "Generated vector field type/searchability/dimensions do not match the selected model.")
    configuration = index.get("vectorSearch")
    if not isinstance(configuration, dict):
        raise _embedding.fail("vector-configuration-invalid", "Generated index has no vectorSearch configuration.")
    profile = _named(configuration.get("profiles"), vector.get("vectorSearchProfile"))
    algorithm = _named(configuration.get("algorithms"), profile.get("algorithm"))
    if algorithm.get("kind") not in {"hnsw", "exhaustiveKnn"}:
        raise _embedding.fail("vector-configuration-invalid", "Unsupported generated vector algorithm.")
    vectorizer = _named(configuration.get("vectorizers"), profile.get("vectorizer"))
    _embedding.verify_model_readback(choice, {key: value for key, value in vectorizer.items() if key != "name"})
    for key in ("content_field", "citation_field"):
        field = _named(fields, request[key])
        if field.get("type") != "Edm.String" or field.get("retrievable") is not True:
            raise _embedding.fail("vector-configuration-invalid", "Content and citation fields must be retrievable strings.")
        if key == "content_field" and request["mode"] == "hybrid" and field.get("searchable") is not True:
            raise _embedding.fail("vector-configuration-invalid", "Hybrid requires the selected content field to be searchable.")
    keys = [field for field in fields if field.get("key") is True]
    if len(keys) != 1 or keys[0].get("type") != "Edm.String" or keys[0].get("retrievable") is not True:
        raise _embedding.fail("vector-configuration-invalid", "Require one retrievable string document key.")
    return {"dimensions": dimensions, "profile": profile["name"], "vectorizer": vectorizer["name"], "document_id_field": keys[0]["name"]}


def _reader(transport: Transport, token_provider: TokenProvider):
    request_ids: list[str] = []

    def read(method: str, url: str, token: str, **kwargs: Any):
        if method != "GET":
            raise _embedding.fail("planning-write-forbidden", "Verification observation permits only GETs.")
        kwargs.update(
            follow_redirects=False,
            max_response_bytes=min(8 * 1024 * 1024, kwargs.get("max_response_bytes", 8 * 1024 * 1024)),
            response_deadline=min(
                time.monotonic() + min(60, kwargs.get("timeout", 60)),
                kwargs.get("response_deadline", float("inf")),
            ),
        )
        result = transport(method, url, token, **kwargs)
        _embedding._json_valid(result.body)
        if result.request_id:
            request_ids.append(result.request_id)
        return result

    def get(url: str):
        value, _ = _search_read.read_resource(url, token_provider(SEARCH_AUDIENCE), transport=read)
        if value is None:
            raise _embedding.fail("vector-resource-missing", "An exact selected source/generated index is absent.")
        if not _embedding._text(value.get("@odata.etag")):
            raise _embedding.fail("vector-evidence-missing", "Fresh source and index ETags are required.")
        return value

    return read, get, request_ids


def _prior_blob_ingestion(
    plan: dict[str, Any], request: dict[str, Any], current: dict[str, Any],
    generated: list[dict[str, Any]], readiness: dict[str, Any],
    receipts: tuple[dict[str, Any], dict[str, Any]] | None,
) -> dict[str, Any]:

    if receipts is None:
        raise _embedding.fail("ingestion-unverified", "A zero-work cycle requires retained approved input and successful nonempty creation result files.")
    prior, result = receipts
    _embedding._json_valid({"plan": prior, "result": result})
    _blob_source_read._verify_creation_binding(
        plan["source"], plan["boundary"], current, generated, receipts,
    )
    if (
        prior["owner"] != plan["owner"] or prior["inventory_digest"] != plan["inventory_digest"]
        or result["verification"].get("source_digest") != plan["inventory_digest"]
    ):
        raise _embedding.fail("ingestion-unverified", "Prior ingestion must bind the same owner and unchanged before/after Storage inventory.")
    previous = result.get("readiness")
    if not isinstance(previous, dict) or previous.get("status") != "verified":
        raise _embedding.fail("ingestion-unverified", "Retained creation must include verified ingestion readiness.")
    require_allowed_fields(previous, {
        "status", "synchronization", "not_before", "request_ids", "first_retry", "watch", "progress",
    }, label="retained ingestion readiness")
    cycle = previous.get("synchronization")
    fields = {"startTime", "endTime", "itemsUpdatesProcessed", "itemsUpdatesFailed", "itemsSkipped"}
    if not isinstance(cycle, dict) or set(cycle) != fields:
        raise _embedding.fail("ingestion-unverified", "Retain the complete creation synchronization interval and counters.")
    if "progress" in previous:
        try:
            from ._progress import validate_blob_progress
        except ImportError:
            from _progress import validate_blob_progress
        display = previous["progress"]
        try:
            validate_blob_progress(display)
        except ValueError as exc:
            raise _embedding.fail("verification-drift", "Retained Blob progress metadata is malformed.") from exc
        if (display["phase"] != "completed"
                or display["synchronization_status"] not in ("active", "not-reported")
                or display["next_check_seconds"] is not None
                or display["run_start"] != _blob_source_read._timestamp(cycle["startTime"]).isoformat()
                or [display[key] for key in ("processed", "failed", "skipped")] !=
                [cycle[key] for key in ("itemsUpdatesProcessed", "itemsUpdatesFailed", "itemsSkipped")]):
            raise _embedding.fail("verification-drift", "Retained Blob progress must describe its own completed cycle.")
    if "watch" in previous:
        watch = previous["watch"]
        expected_latest = {key: cycle[key] for key in fields - {"endTime"}}
        expected_latest["state"] = "completed-cycle-observed"
        latest = watch.get("latest") if isinstance(watch, dict) else None
        if isinstance(latest, dict) and "indexer_execution" in latest:
            execution = latest["indexer_execution"]
            if not _blob_source_read._execution_completed_by(
                execution, _blob_source_read._timestamp(cycle["endTime"]),
                max(_blob_source_read._timestamp(request["not_before"]), _blob_source_read._timestamp(previous.get("not_before"))),
            ):
                raise _embedding.fail("verification-drift", "Retained active execution must be resolved before its primary cycle ends.")
            expected_latest["indexer_execution"] = execution
        if (not isinstance(watch, dict)
                or set(watch) != {"schema_version", "state", "elapsed_seconds", "status_checks", "latest"}
                or watch["schema_version"] != "1.0" or watch["state"] != "completed"
                or type(watch["elapsed_seconds"]) not in (int, float)
                or not math.isfinite(watch["elapsed_seconds"]) or watch["elapsed_seconds"] < 0
                or type(watch["status_checks"]) is not int or not 1 <= watch["status_checks"] <= 1000
                or watch["latest"] != expected_latest):
            raise _embedding.fail("verification-drift", "Retained watch metadata must match its completed cycle; it cannot replace ingestion proof.")
    counts = [cycle[key] for key in ("itemsUpdatesProcessed", "itemsUpdatesFailed", "itemsSkipped")]
    if any(type(count) is not int or count < 0 for count in counts) or counts[0] == 0 or any(counts[1:]):
        raise _embedding.fail("ingestion-unverified", "Prior creation must have processed content without failures or skipped items.")
    start, end = (_blob_source_read._timestamp(cycle[key]) for key in ("startTime", "endTime"))
    if (
        start < max(_blob_source_read._timestamp(request["not_before"]), _blob_source_read._timestamp(previous.get("not_before")))
        or end < start or end > _blob_source_read._timestamp(readiness["synchronization"]["startTime"])
    ):
        raise _embedding.fail("ingestion-unverified", "Prior ingestion must follow both lower bounds and complete before the current zero-work cycle.")
    return {
        "synchronization": copy.deepcopy(cycle),
        "evidence_digest": digest({"creation_plan": prior, "creation_result": result}),
    }


def observe(request: dict[str, Any], *, token_provider: TokenProvider, transport: Transport,
            storage_transport: Transport) -> tuple[dict[str, Any], dict[str, Any]]:
    plan = _request(request)
    source = plan["source"]
    try:
        from . import blob_inventory
    except ImportError:
        import blob_inventory
    read, get, ids = _reader(_source_readback.guard_readback_transport(plan, transport), token_provider)
    url = _search_read.resource_url(source)
    current = get(url)
    if not _search_read.definitions_match(source["desired"], current):
        raise _embedding.fail("definition-conflict", "Fresh source differs from the selected source artifact.")
    if source.get("expected_etag") and current["@odata.etag"] != source["expected_etag"]:
        raise _embedding.fail("definition-drift", "Selected reuse source ETag changed; rerun the source planner.")
    kind = source["desired"]["kind"]
    if kind == "file":
        created = current.get("fileParameters", {}).get("createdResources")
        if not isinstance(created, dict) or set(created) != {"index"}:
            raise _embedding.fail("vector-evidence-missing", "Require the File source's generated index identity.")
        files, _ = _file_inventory.read_inventory(plan["ingestion"], token_provider(SEARCH_AUDIENCE), transport=read)
        matches = _file_inventory.reconcile_inventory(plan["ingestion"], files)
        identities = [item.get("fileId") for item in matches.values()]
        if not identities or not all(_embedding._text(item) for item in identities) or len(set(identities)) != len(identities):
            raise _embedding.fail("ingestion-unverified", "Require complete unique ingested File identities and markers.")
        inventory = _file_inventory.inventory_digest(files)
        readiness = {"status": "verified", "files": len(matches), "basis": "synchronous File ingestion markers; no reported file errors"}
    else:
        created = {item["type"]: item["name"] for item in _blob_source_read.generated_resources(current, strict=True)}
        generated = _blob_source_read.generated_resources(current, strict=True)
        if "expected_generated" in plan and generated != plan["expected_generated"]:
            raise _embedding.fail("definition-drift", "Generated Blob identities changed.")
        receipts: tuple[dict[str, Any], dict[str, Any]] | None = None

        def load_receipts() -> tuple[dict[str, Any], dict[str, Any]] | None:
            nonlocal receipts
            if receipts is None:
                receipts = _blob_source_read._reuse_receipts(_blob_source_read._receipt_paths(request))
            return receipts

        _blob_source_read._verify_storage_binding(
            source, plan["boundary"], current, generated, load_receipts,
        )
        snapshot = blob_inventory.discover(plan["boundary"], plan["inventory_limits"],
                                           token_provider=token_provider, transport=storage_transport)
        ids.extend(snapshot["request_ids"])
        inventory = snapshot["inventory_digest"]
        if inventory != plan["inventory_digest"]:
            raise _embedding.fail("inventory-drift", "Storage content or ACL evidence changed; rerun source planning.")
        readiness = _blob_source_read.monitor(
            source, not_before=_blob_source_read._timestamp(request["not_before"]),
            limits={"max_requests": 1, "interval_seconds": 1, "deadline_seconds": 60},
            token_provider=token_provider, transport=read, sleep=lambda _: None,
        )
        if readiness.get("status") != "verified":
            raise _embedding.fail("ingestion-unverified", "No relevant completed zero-failure Blob synchronization; do not query.")
        cycle = readiness["synchronization"]
        if cycle["itemsSkipped"]:
            raise _embedding.fail("ingestion-unverified", "A completed cycle with no skipped items is required.")
        if cycle["itemsUpdatesProcessed"] == 0:
            readiness["prior_ingestion"] = _prior_blob_ingestion(
                plan, request, current, generated, readiness, load_receipts(),
            )
            readiness["basis"] = "Current zero-work success plus same-source, same-inventory nonempty creation proof."
    index_name = created.get("index")
    index_url = f"{source['endpoint'].rstrip('/')}/indexes('{odata_name(index_name)}')?api-version={source['api_version']}"
    index = get(index_url)
    if index.get("name") != index_name:
        raise _embedding.fail("vector-configuration-mismatch", "Generated index name does not match its source identity.")
    metadata = verify_index(index, plan["embedding"], request)
    refreshed, refreshed_index = get(url), get(index_url)
    source_changed = digest(current) != digest(refreshed)
    if "content_understanding" in plan:
        source_changed = (
            current["@odata.etag"] != refreshed["@odata.etag"]
            or not _search_read.definitions_match(current, refreshed)
            or (
                refreshed.get("fileParameters", {}).get("createdResources") != created
                if kind == "file" else _blob_source_read.generated_resources(refreshed, strict=True) != generated
            )
        )
    if source_changed or digest(index) != digest(refreshed_index):
        raise _embedding.fail("definition-drift", "Source or index changed during observation.")
    snapshot = {
        "source_plan_digest": digest(plan), "source_etag": current["@odata.etag"],
        "source_definition": digest(_search_read._definition(current)),
        "index": index_name, "index_etag": index["@odata.etag"], "index_digest": digest(index),
        "inventory_digest": inventory, "vector": metadata,
    }
    if "prior_ingestion" in readiness:
        snapshot["ingestion_evidence_digest"] = readiness["prior_ingestion"]["evidence_digest"]
    return snapshot, {"readiness": readiness, "request_ids": ids}


def plan_verification(request: dict[str, Any], *, token_provider: TokenProvider = azure_cli_token,
                      transport: Transport = http_request, storage_transport: Transport = http_request) -> dict[str, Any]:
    source_plan = _request(request)
    snapshot, evidence = observe(request, token_provider=token_provider, transport=transport, storage_transport=storage_transport)
    if digest(source_plan) != snapshot["source_plan_digest"]:
        raise _embedding.fail("source-artifact-drift", "Source artifact changed while planning verification.")
    plan = {"operation": "verify-source-vector", "owner": source_plan["owner"], "cleanup_approved": False,
            "request": copy.deepcopy(request), "snapshot": snapshot}
    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": source_plan["source"]["name"], "index": snapshot["index"],
            "search_endpoint": source_plan["source"]["endpoint"], "api_version": source_plan["source"]["api_version"],
            "mode": request["mode"], "queries": 2 if request["mode"] == "hybrid" else 1,
            "k": request["k"], "query": request["query"],
            "embedding": _embedding.summary(source_plan["embedding"]), "dimensions_observed": snapshot["vector"]["dimensions"],
            "configuration": "verified", "ingestion": "verified", "retrieval": "unverified",
            "next_step": "Approve the exact query/content transfer and embedding cost, then apply this unchanged artifact. No index repair or cleanup.",
        },
        "evidence": evidence, "writes_performed": [],
    }


def query_body(request: dict[str, Any], metadata: dict[str, Any], *, hybrid: bool) -> dict[str, Any]:
    body = {
        "vectorQueries": [{"kind": "text", "text": request["query"], "fields": request["vector_field"], "k": request["k"]}],
        "select": ",".join(dict.fromkeys((metadata["document_id_field"], request["content_field"], request["citation_field"]))),
        "top": request["k"], "minimumCoverage": 100,
    }
    if hybrid:
        body.update(search=request["query"], searchFields=request["content_field"], queryType="simple")
    return body


def verify_response(body: Any, request: dict[str, Any], metadata: dict[str, Any]) -> int:
    _embedding._json_valid(body)
    if (
        not isinstance(body, dict) or type(body.get("@search.coverage")) not in (int, float)
        or body["@search.coverage"] != 100
        or any(body.get(key) is not None for key in
               ("@odata.nextLink", "@search.nextPageParameters", "@search.semanticPartialResponseReason", "error"))
        or not isinstance(body.get("value"), list) or not 1 <= len(body["value"]) <= request["k"]
    ):
        raise _embedding.fail("vector-query-incomplete", "Require a nonempty, complete, full-coverage query response without continuation or partial errors.")
    identities, matched = set(), 0
    for hit in body["value"]:
        if not isinstance(hit, dict):
            raise _embedding.fail("vector-query-invalid", "Malformed query hit.")
        key, text, citation = (hit.get(field) for field in (metadata["document_id_field"], request["content_field"], request["citation_field"]))
        score = hit.get("@search.score")
        if (
            not all(_embedding._text(value, maximum=1024 * 1024) for value in (key, text, citation))
            or key in identities or type(score) not in (int, float) or not math.isfinite(score) or score < 0
        ):
            raise _embedding.fail("vector-query-invalid", "Require unique document IDs, content, provenance and finite scores.")
        identities.add(key)
        matched += citation == request["expected_citation"] and request["expected_text"].casefold() in text.casefold()
    if not matched:
        raise _embedding.fail("vector-query-mismatch", "No retrieved document matches both the expected source citation and expected content.")
    return matched


def _validate_snapshot(value: Any) -> None:
    fields = {"source_plan_digest", "source_etag", "source_definition", "index",
              "index_etag", "index_digest", "inventory_digest", "vector"}
    if not isinstance(value, dict) or set(value) not in (fields, fields | {"ingestion_evidence_digest"}):
        raise _embedding.fail("input-schema-invalid", "Retain the complete generated verification snapshot.")
    hashes = {"source_plan_digest", "source_definition", "index_digest", "inventory_digest"}
    for key in hashes | ({"ingestion_evidence_digest"} & value.keys()):
        if not isinstance(value[key], str) or not _search_read.SHA256.fullmatch(value[key]):
            raise _embedding.fail("input-schema-invalid", "Invalid generated snapshot integrity value; rerun planning.")
    if not all(_embedding._text(value[key]) for key in ("source_etag", "index_etag", "index")):
        raise _embedding.fail("input-schema-invalid", "Snapshot identities and ETags are required.")
    odata_name(value["index"])
    vector = value["vector"]
    if (
        not isinstance(vector, dict)
        or set(vector) != {"dimensions", "profile", "vectorizer", "document_id_field"}
        or type(vector.get("dimensions")) is not int or not 1 <= vector["dimensions"] <= 3072
        or not all(_embedding._text(vector[key], maximum=128) for key in ("profile", "vectorizer", "document_id_field"))
    ):
        raise _embedding.fail("input-schema-invalid", "Invalid generated vector snapshot; rerun planning.")


def execute(document: dict[str, Any], *, token_provider: TokenProvider = azure_cli_token,
            transport: Transport = http_request, storage_transport: Transport = http_request) -> dict[str, Any]:
    if not isinstance(document, dict):
        raise _embedding.fail("input-schema-invalid", "Query execution needs an approved artifact.")
    _embedding._json_valid(document)
    require_allowed_fields(document, {"schema_version", "plan", "approval"}, label="query artifact")
    plan = document.get("plan")
    if not isinstance(plan, dict):
        raise _embedding.fail("input-schema-invalid", "Query execution needs a planned artifact.")
    reject_secrets(document)
    require_allowed_fields(plan, {"operation", "owner", "cleanup_approved", "request", "snapshot"}, label="vector query plan")
    fingerprint = digest(plan)
    approval = document.get("approval")
    if (
        document.get("schema_version") != "1.0" or not isinstance(approval, dict)
        or approval.get("confirmed") is not True
        or approval != {"confirmed": True, "fingerprint": fingerprint}
    ):
        raise _embedding.fail("approval-required", "Explicit approval of this unchanged billable query plan is required.")
    if plan.get("operation") != "verify-source-vector" or plan.get("cleanup_approved") is not False:
        raise _embedding.fail("input-schema-invalid", "Only separately approved vector verification is supported.")
    _validate_snapshot(plan.get("snapshot"))
    request = plan.get("request")
    source_plan = _request(request)
    if plan.get("owner") != source_plan["owner"]:
        raise _embedding.fail("source-artifact-invalid", "Query owner must match the source artifact.")
    snapshot, evidence = observe(request, token_provider=token_provider, transport=transport, storage_transport=storage_transport)
    if snapshot != plan.get("snapshot") or digest(source_plan) != snapshot["source_plan_digest"]:
        raise _embedding.fail("verification-drift", "Evidence changed; rerun --plan, replace the artifact and discard consent.")
    source = source_plan["source"]
    url = f"{source['endpoint'].rstrip('/')}/indexes('{odata_name(snapshot['index'])}')/docs/search.post.search?api-version={source['api_version']}"
    queries = []
    attempted = 0
    try:
        for hybrid in ([False, True] if request["mode"] == "hybrid" else [False]):
            token = token_provider(SEARCH_AUDIENCE)
            attempted += 1
            response = transport(
                "POST", url, token,
                headers={"Content-Type": "application/json"},
                body=json.dumps(query_body(request, snapshot["vector"], hybrid=hybrid)).encode("utf-8"),
                follow_redirects=False, max_response_bytes=8 * 1024 * 1024,
                timeout=60, response_deadline=time.monotonic() + 60,
            )
            if response.status != 200 or not _embedding._text(response.request_id):
                raise _embedding.fail("vector-query-incomplete", "Require HTTP 200 and observed Search request provenance.")
            count = verify_response(response.body, request, snapshot["vector"])
            queries.append({"mode": "hybrid" if hybrid else "vector", "request_id": response.request_id,
                            "matched_documents": count, "response_digest": digest(response.body)})
        after, _ = observe(request, token_provider=token_provider, transport=transport, storage_transport=storage_transport)
        if after != snapshot:
            raise _embedding.fail("verification-drift", "Source, index or inventory changed across query execution.")
    except HelperFailure as failure:
        result = blocked_result(
            HelperFailure(failure.code, failure.message, blocked_at=failure.blocked_at,
                          status=failure.http_status, request_id=failure.request_id),
            outcome="source-vector-verification", fingerprint=fingerprint, owner=plan["owner"],
        )
        result.update(queries_attempted=attempted, billing_possible=attempted > 0, queries=queries, writes_performed=[])
        return result
    return {
        "status": "retrieval-verified", "outcome": "source-vector-verification",
        "approved_plan": {"confirmed": True, "fingerprint": fingerprint},
        "verification": {"configuration": "verified", "ingestion": "verified",
                         "retrieval": request["mode"], "queries": queries, "evidence": evidence},
        "writes_performed": [], "cleanup": "not-applicable",
        "warnings": ["Observed index retrieval only: not KB synthesis, agent tool use, or a corpus-wide relevance guarantee. Reobserve before later use."],
    }


def main(argv: list[str] | None = None) -> int:
    try:
        from .private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
    except ImportError:
        from private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
    parser = argparse.ArgumentParser(description="Plan or apply separately approved File/Blob vector verification.")
    modes = parser.add_mutually_exclusive_group(required=True)
    modes.add_argument("--plan", type=Path)
    modes.add_argument("--input", type=Path)
    add_execution_output_argument(parser)
    args = parser.parse_args(argv)
    try:
        validate_execution_output_mode(args)
        if args.plan:
            result = plan_verification(_read_json(str(args.plan)))
            emit_plan_result(result, args.execution_output)
            return 0 if result["status"] == "planned" else 2
        else:
            document, _, _ = load_approved_input(args.input)
            result = execute(document)
    except HelperFailure as failure:
        result = blocked_result(failure, outcome="source-vector-verification", fingerprint=None)
    emit_result(result)
    return 0 if result["status"] in {"planned", "retrieval-verified"} else 2


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