Subchapter 10.49

File Source

File Cu Canary

Intent Routing

Owner

65 more in the appendix

File Source

def main

This file

Number: 10.49
Position: 49 of 77
Type: Python
Size: 44 KB
Lines: 858

Nearby

Source

helpers/file_source.py

Python858 lines44 KB

1from __future__ import annotations
2
3import argparse
4import copy
5import json
6import re
7import sys
8from pathlib import Path
9from typing import Any
10
11try:
12    from

file_source.py on GitHub

._progress

import

Progress, add_progress_argument, reporting

from . import file_ingest, search_reconcile, source_vector, file_cu_mi

from . import cu_ingestion_auth as file_cu_auth

from ._common import (

SEARCH_AUDIENCE,

MANAGEMENT_AUDIENCE,

HelperFailure,

TokenProvider,

Transport,

azure_cli_token,

blocked_result,

digest,

emit_result,

http_request,

load_approved_input,

normalize_azure_location,

reject_secrets,

require_allowed_fields,

)

except ImportError:

from _progress import Progress, add_progress_argument, reporting

import file_ingest # type: ignore[no-redef]

import search_reconcile # type: ignore[no-redef]

import source_vector

import file_cu_mi

import cu_ingestion_auth as file_cu_auth

from _common import ( # type: ignore[no-redef]

SEARCH_AUDIENCE,

MANAGEMENT_AUDIENCE,

HelperFailure,

TokenProvider,

Transport,

azure_cli_token,

blocked_result,

digest,

emit_result,

http_request,

load_approved_input,

normalize_azure_location,

reject_secrets,

require_allowed_fields,

)

def _cu_failure(code: str, message: str, *, request_id: str | None = None) -> HelperFailure:

return HelperFailure(code, message, blocked_at="cu-prerequisites", request_id=request_id)

def validate_content_understanding(value: Any, *, enabled: bool) -> dict[str, Any] | None:

if not enabled:

if value is not None:

raise _cu_failure("cu-choice-conflict", "Minimal extraction must omit CU choices.")

return None

if not isinstance(value, dict):

raise _cu_failure("cu-prerequisite-missing", "Standard planning requires a resolved CU account, disclosed auth channel and owner-verified prerequisites.")

value = copy.deepcopy(value)

value.setdefault("auth", "system-assigned")

reject_secrets(value)

require_allowed_fields(value, {

"endpoint", "resource_id", "auth", "api_key_environment", "prerequisites", "managed_identity",

}, label="File CU choice")

if (

not isinstance(value.get("endpoint"), str)

or re.fullmatch(r"https://[a-z0-9][a-z0-9-]{0,62}\.services\.ai\.azure\.com/?", value["endpoint"]) is None

or not isinstance(value.get("resource_id"), str)

or re.fullmatch(

r"/subscriptions/[0-9a-fA-F-]{36}/resourceGroups/[A-Za-z0-9_.()-]{1,90}"

r"/providers/Microsoft\.CognitiveServices/accounts/[A-Za-z0-9][A-Za-z0-9_.-]{1,63}",

value["resource_id"], re.IGNORECASE,

) is None

or value.get("auth") not in ("api-key-environment", "api-key-arm", "system-assigned")

or (value.get("auth") == "api-key-environment" and (

not isinstance(value.get("api_key_environment"), str)

or search_reconcile.ENVIRONMENT_NAME.fullmatch(value["api_key_environment"]) is None

))

or (value.get("auth") != "api-key-environment" and "api_key_environment" in value)

or (value.get("auth") != "system-assigned" and "managed_identity" in value)

):

raise _cu_failure("cu-choice-invalid", "Select exact AIServices and system-assigned MI, or explicitly retain approved ARM/ENV key auth. No automatic auth fallback or setup changes.")

if value["auth"] == "system-assigned":

file_cu_mi.validate_choice(value.get("managed_identity"), value["resource_id"])

prerequisites = value.get("prerequisites")

if not isinstance(prerequisites, dict):

raise _cu_failure("cu-prerequisite-missing", "Supply CU region/capability, selected processing/required deployments, identity/local-auth and network evidence references.")

fields = {"resource", "configuration", "identity", "network"}

require_allowed_fields(prerequisites, fields, label="File CU prerequisites")

if any(not source_vector._text(prerequisites.get(key)) for key in fields):

raise _cu_failure("cu-prerequisite-missing", "Owner-verified CU capability/region, selected processing/required deployments, auth/access and reachability evidence is required.")

source_vector._json_valid(value)

return copy.deepcopy(value)

def _cu_account_state(choice: dict[str, Any], account: Any) -> dict[str, Any]:

properties = account.get("properties") if isinstance(account, dict) else None

if not isinstance(properties, dict):

raise _cu_failure("cu-prerequisite-invalid", "CU account readback is incomplete.")

endpoints = properties.get("endpoints", {})

candidates = [properties.get("endpoint")]

if isinstance(endpoints, dict):

candidates.extend(endpoints.values())

if (

str(account.get("id", "")).casefold() != choice["resource_id"].casefold()

or account.get("kind") != "AIServices"

or normalize_azure_location(account.get("location")) is None

or properties.get("provisioningState") != "Succeeded"

or (choice["auth"] != "system-assigned" and properties.get("disableLocalAuth") is not False)

or properties.get("publicNetworkAccess") not in ("Enabled", "Disabled")

or choice["endpoint"].rstrip("/") not in [v.rstrip("/") for v in candidates if isinstance(v, str)]

):

raise _cu_failure("cu-prerequisite-invalid", "Readback must bind the selected ready AIServices account/endpoint/location/network. Key modes also need enabled local auth; MI does not. Any required setup change needs separate approval.")

state = {

"id": choice["resource_id"], "kind": "AIServices", "location": account["location"],

"identity": copy.deepcopy(account.get("identity")),

"properties": {

"endpoint": choice["endpoint"].rstrip("/"), "provisioningState": "Succeeded",

"disableLocalAuth": properties.get("disableLocalAuth"), "publicNetworkAccess": properties["publicNetworkAccess"],

"networkAcls": copy.deepcopy(properties.get("networkAcls")),

},

}

if choice["auth"] == "system-assigned":

acl = properties.get("networkAcls")

if properties["publicNetworkAccess"] != "Enabled" or (

acl is not None and (not isinstance(acl, dict) or acl.get("defaultAction") != "Allow")

):

raise _cu_failure("cu-mi-network-unverified", "This MI path requires existing public CU reachability without default-deny ACLs. Restricted/private network compatibility needs separate verified setup and approval; no network changes or key fallback.")

reject_secrets(state)

source_vector._json_valid(state)

return state

def _cu_states_match(current: dict[str, Any], retained: dict[str, Any]) -> bool:

location = normalize_azure_location(retained.get("location"))

return location is not None and (

{**current, "location": normalize_azure_location(current.get("location"))}

== {**retained, "location": location}

)

def read_content_understanding(

choice: dict[str, Any], *, token_provider: TokenProvider, transport: Transport,

) -> tuple[dict[str, Any], list[str]]:

url = f"{MANAGEMENT_AUDIENCE}{choice['resource_id']}?api-version=2024-10-01"

response = transport("GET", url, token_provider(MANAGEMENT_AUDIENCE))

try:

if response.status != 200:

raise _cu_failure("cu-prerequisite-unavailable", "Selected CU account metadata could not be read; no provisioning or auth changes are allowed.")

state = _cu_account_state(choice, response.body)

except HelperFailure as failure:

failure.request_id = response.request_id

if response.status != 200:

failure.http_status = response.status

raise

return state, [response.request_id] if response.request_id else []

def verify_content_understanding_readback(choice: dict[str, Any], current: Any) -> None:

parameters = current.get("fileParameters") if isinstance(current, dict) else None

ingestion = parameters.get("ingestionParameters") if isinstance(parameters, dict) else None

ai = ingestion.get("aiServices") if isinstance(ingestion, dict) else None

if (

not isinstance(ai, dict) or ingestion.get("contentExtractionMode") != "standard"

or not isinstance(ai.get("uri"), str)

or ai["uri"].rstrip("/") != choice["endpoint"].rstrip("/")

or ingestion.get("identity") is not None

or (choice["auth"] == "system-assigned" and ai.get("apiKey") not in file_cu_mi.REDACTED)

):

raise _cu_failure("cu-readback-mismatch", "Observed File CU endpoint/extraction/auth conflicts with the selected configuration; credential details withheld.")

# File's approved key may be redacted in GET. It is not embedding auth,

# and source readback cannot prove its value or CU processing readiness.

def plan_source(

request: dict[str, Any],

*,

token_provider: TokenProvider = azure_cli_token,

transport: Transport = http_request,

context_provider=file_cu_auth.account_context,

) -> dict[str, Any]:

"""Build an unapproved File workflow using only local reads and GETs."""

if not isinstance(request, dict):

raise HelperFailure(

"input-schema-invalid", "Planning input must be an object.",

blocked_at="input-resolution",

)

reject_secrets(request)

require_allowed_fields(

request,

{"schema_version", "api_version", "endpoint", "name", "owner", "local_root", "paths",

"service_tier", "extraction_mode", "vectorization", "embedding", "rbac", "network",

"content_understanding", "reuse_input_file", "reuse_result_file"},

label="File planning input",

)

if request.get("schema_version") != "1.0":

raise HelperFailure(

"input-schema-invalid", "Planning requires schema_version 1.0.",

blocked_at="input-resolution",

)

file_ingest.validate_api_version(request.get("api_version", file_ingest.API_VERSION))

for field in ("name", "owner"):

if not isinstance(request.get(field), str) or not request[field].strip():

raise HelperFailure(

"input-schema-invalid", f"An explicit non-empty {field} is required.",

blocked_at="input-resolution",

)

if request.get("extraction_mode") not in ("minimal", "standard") or request.get("vectorization") not in ("none", "azureOpenAI"):

raise HelperFailure(

"planning-processing-unsupported",

"Select minimal or standard extraction and independent vectorization none or azureOpenAI.",

blocked_at="input-resolution",

)

embedding = source_vector.validate_choice(

request.get("embedding"), enabled=request["vectorization"] == "azureOpenAI",

api_version=file_ingest.API_VERSION,

)

cu = validate_content_understanding(

request.get("content_understanding"), enabled=request["extraction_mode"] == "standard",

)

if (request.get("reuse_input_file") is not None or request.get("reuse_result_file") is not None) and (

cu is None or cu["auth"] != "system-assigned"

):

raise _cu_failure("cu-choice-conflict", "File MI provenance inputs are only for managed-identity exact reuse.")

rbac, network = request.get("rbac"), request.get("network")

if (

not isinstance(rbac, dict)

or not isinstance(rbac.get("assignments"), list)

or not rbac["assignments"]

or not all(isinstance(item, dict) and item for item in rbac["assignments"])

or not isinstance(network, dict)

or not isinstance(network.get("posture"), str)

or not network["posture"].strip()

or not isinstance(network.get("evidence"), str)

or not network["evidence"].strip()

):

raise HelperFailure(

"planning-evidence-missing",

"Supply observed RBAC assignments and network posture/evidence; the policy owner must refresh and verify them before approval.",

blocked_at="input-resolution",

)

root = file_ingest.resolve_local_root(request.get("local_root"))

records = file_ingest.snapshot_inventory(

root, request.get("paths"), service_tier=request.get("service_tier")

)

common = {

"endpoint": request.get("endpoint"), "name": request.get("name"),

"api_version": file_ingest.API_VERSION, "owner": request.get("owner"),

"cleanup_approved": False, "rbac": copy.deepcopy(rbac),

"network": copy.deepcopy(network),

}

source = {

**common, "operation": "reconcile", "resource_type": "knowledge-source",

"outcome": "create-file-knowledge-source", "action": "create",

"desired": {

"name": common["name"], "kind": "file",

"fileParameters": {"ingestionParameters": {"contentExtractionMode": request["extraction_mode"]}},

},

}

ingestion = {

**copy.deepcopy(common), "operation": "ingest", "local_root": str(root),

"files": records, "inventory_digest": digest(records),

"expected_server_inventory_digest": file_ingest.inventory_digest([]),

"service_tier": request["service_tier"], "extraction_mode": request["extraction_mode"],

}

plan = {

"operation": "reconcile-and-ingest", "outcome": "create-file-knowledge-source",

"owner": common["owner"], "cleanup_approved": False,

"source": source, "ingestion": ingestion,

}

if embedding is not None:

plan["embedding"] = embedding

source["desired"]["fileParameters"]["ingestionParameters"]["embeddingModel"] = source_vector.model_definition(embedding)

if cu is not None:

automatic = cu["auth"] == "api-key-arm"

mi = cu["auth"] == "system-assigned"

plan.update(file_cu_plan_version="1.2" if mi else "1.1" if automatic else "1.0", content_understanding=cu)

if automatic:

source["ai_services_key_acquisition"] = file_cu_auth.acquisition(cu, context_provider())

elif mi:

source["ai_services_managed_identity"] = True

else:

source["ai_services_api_key_environment"] = cu["api_key_environment"]

source["desired"]["fileParameters"]["ingestionParameters"].update(

aiServices={"uri": cu["endpoint"].rstrip("/")}, disableImageVerbalization=True,

)

# Validate the local inventory and all choices before any authentication.

_validate_plan(plan, require_cu_readback=False)

cu_request_ids = []

if cu is not None:

state, cu_request_ids = read_content_understanding(cu, token_provider=token_provider, transport=transport)

plan["cu_resource_state"] = state

if cu["auth"] == "system-assigned":

plan["cu_identity_state"], ids = file_cu_mi.read_binding(

cu, common["endpoint"], token_provider=token_provider, transport=transport,

)

cu_request_ids.extend(ids)

_validate_plan(plan)

transport = source_vector.guard_readback_transport(plan, transport)

url = search_reconcile.resource_url(source)

token = token_provider(SEARCH_AUDIENCE)

current, request_id = search_reconcile.read_resource(url, token, transport=transport)

request_ids = cu_request_ids + ([request_id] if request_id else [])

matched = {}

if current is not None:

if cu is not None and cu["auth"] == "system-assigned":

file_cu_mi.verify_reuse(request, plan, current)

if not search_reconcile.definitions_match(source["desired"], current):

raise HelperFailure(

"definition-conflict",

"The exact source has a different definition; planning never overwrites or chooses another name.",

blocked_at="reconciliation",

)

etag = current.get("@odata.etag")

if not isinstance(etag, str) or not etag:

raise HelperFailure(

"definition-evidence-missing", "Exact reuse requires the current source ETag.",

blocked_at="reconciliation",

)

before, ids = file_ingest.read_inventory(ingestion, token, transport=transport)

request_ids.extend(ids)

matched = file_ingest.reconcile_inventory(ingestion, before)

file_ids = [item.get("fileId") for item in matched.values()]

if (

not all(isinstance(value, str) and value for value in file_ids)

or len(set(file_ids)) != len(file_ids)

):

raise HelperFailure(

"file-identity-ambiguous", "Exact reuse requires unique non-empty server file IDs.",

blocked_at="reconciliation",

)

source.update(action="reuse", expected_etag=etag)

ingestion["expected_server_inventory_digest"] = file_ingest.inventory_digest(before)

refreshed, refresh_id = search_reconcile.read_resource(url, token, transport=transport)

if refresh_id:

request_ids.append(refresh_id)

if (

refreshed is None

or refreshed.get("@odata.etag") != etag

or not search_reconcile.definitions_match(source["desired"], refreshed)

):

raise HelperFailure(

"definition-drift",

"Source definition or ETag changed during file inventory readback.",

blocked_at="reconciliation",

)

# Do not return a snapshot that changed while Search discovery was running.

_validate_plan(plan)

if cu is not None:

state, ids = read_content_understanding(cu, token_provider=token_provider, transport=transport)

request_ids.extend(ids)

if not _cu_states_match(state, plan["cu_resource_state"]):

raise _cu_failure(

"cu-prerequisite-drift", "CU account access or configuration changed during planning; refresh the plan.",

request_id=ids[-1] if ids else None,

)

if cu["auth"] == "api-key-arm":

file_cu_auth.check_context(source["ai_services_key_acquisition"]["context"], context_provider)

elif cu["auth"] == "system-assigned":

binding, ids = file_cu_mi.read_binding(cu, common["endpoint"], token_provider=token_provider, transport=transport)

request_ids.extend(ids)

if binding != plan["cu_identity_state"]:

raise _cu_failure("cu-mi-identity-drift", "Search identity, CU scoped role or network changed during planning.")

fingerprint = digest(plan)

mutation_required = source["action"] == "create"

return {

"status": "planned", "outcome": plan["outcome"],

"plan_fingerprint": fingerprint,

"execution_input": {

"schema_version": "1.0", "plan": plan,

"approval": {"confirmed": False, "fingerprint": fingerprint},

},

"approval_summary": {

"target": {"endpoint": common["endpoint"], "name": common["name"],

"api_version": common["api_version"], "preview": True},

"source_action": source["action"], "owner": common["owner"],

"execution_required": mutation_required,

"mutation_approval_required": mutation_required,

"processing": (

"standard CU extraction" + (" with source embeddings" if embedding else "; source vectors off")

if cu else "minimal extraction with embeddings" if embedding else "minimal lexical; no models"

),

**({"content_understanding": {

"purpose": "Standard document extraction only; not source vectorization or KB answer synthesis.",

"endpoint": cu["endpoint"],

"authentication": file_cu_auth.approval_summary("file", cu["auth"], creating=mutation_required),

"auth": (

"Search system-assigned MI for CU; no API key or credential reads, no local-auth requirement. Service implementation inspected; live compatibility unverified."

if cu["auth"] == "system-assigned" else

"Existing File CU key-auth configuration reused; no credential acquisition. Search remains keyless."

if not mutation_required else

"Approved private ARM listKeys acquisition of key1 for this source PUT only; Search remains keyless. Local authentication means key auth, not manual local setup."

if cu["auth"] == "api-key-arm" else "Explicit existing CU API-key ENV channel; Search remains keyless."

),

**({"credential_acquisition": copy.deepcopy(source["ai_services_key_acquisition"])} if cu["auth"] == "api-key-arm" and mutation_required else {}),

**({"managed_identity": copy.deepcopy(plan["cu_identity_state"])} if cu["auth"] == "system-assigned" else {}),

"cost_and_data": "Billable CU processing, no daily free document allowance; uploaded content moves from Search to CU, possibly across regions. Search retains outputs.",

"verification": (

"ARM verifies Search identity, exact CU-scoped role and account/network metadata, not backend MI rollout, effective access or extraction. A separately approved bounded OCR canary can validate functionality."

if cu["auth"] == "system-assigned" else

"ARM verifies account binding/local-auth/network metadata, not effective access, key validity or processing success. Verify selected processing/required deployment evidence; no blanket account-defaults confirmation."

),

"kb_reasoning": "Unchanged; source CU does not enable KB chat or source vectors.",

}} if cu else {}),

**({"embedding": source_vector.summary(embedding)} if embedding else {}),

"data_boundary": {"paths": [r["path"] for r in records],

"file_count": len(records), "total_bytes": sum(r["size"] for r in records)},

"uploads": 0 if matched else len(records),

"reused_files": [{"path": path, "fileId": item["fileId"]} for path, item in matched.items()],

"service_tier": request["service_tier"],

"rbac": rbac, "network": network,

"cost_and_retention": (

"Existing Search charges remain; review File ingestion/storage charges and retention before approval."

if mutation_required else "Existing charges and retention are unchanged; no new uploads or resources."

),

"ownership": "New source and uploaded files only; reused Search/source/files are not run-owned.",

"format_verification": "Filename/MIME hints are not detected types; Search checks actual content support during ingestion.",

"verification": (

"Execution rechecks definition/ETag and local/server inventories, then verifies uploaded file markers."

if mutation_required else

"Fresh source definition/ETag and complete file markers/IDs match the local inventory; this does not verify ingestion readiness or retrieval."

),

"cleanup": "Excluded; separate run-owned source cleanup plan and approval required.",

"next_step": (

"Creation owner refreshes identity, RBAC, network, source state and cost/data consent, then obtains explicit approval of these changes before applying the unchanged execution input."

if mutation_required else

"Reuse the verified identity and file markers without mutation approval or invoking the mutation helper. Refresh discovery before later use; this observation is not future consent."

),

},

"read_only_evidence": {"request_ids": request_ids, "source_state": source["action"]},

"writes_performed": [],

"warnings": [

"Planning is not approval or completed ingestion. Uploader RBAC/network remain supplied evidence; MI identity/role metadata readbacks do not prove backend attribution or effective access."

if cu is not None and cu["auth"] == "system-assigned" else

"Planning is not approval, policy evaluation, or proof of completed ingestion. RBAC/network are caller-supplied evidence, not verified by this helper."

],

}

def _validate_plan(

plan: dict[str, Any],

*,

require_cu_readback: bool = True,

) -> tuple[dict[str, Any], dict[str, Any]]:

reject_secrets(plan)

require_allowed_fields(

plan,

{

"operation",

"outcome",

"cleanup_approved",

"owner",

"source",

"ingestion",

"embedding",

"content_understanding",

"file_cu_plan_version",

"cu_resource_state",

"cu_identity_state",

},

label="File source plan",

)

if (

plan.get("operation") != "reconcile-and-ingest"

or plan.get("cleanup_approved") is not False

):

raise HelperFailure(

"operation-invalid",

"File source application requires reconcile-and-ingest with cleanup excluded.",

blocked_at="input-resolution",

)

source = plan.get("source")

ingestion = plan.get("ingestion")

if not isinstance(source, dict) or not isinstance(ingestion, dict):

raise HelperFailure(

"input-schema-invalid",

"File source application requires source and ingestion objects.",

blocked_at="input-resolution",

)

desired = source.get("desired")

if (

source.get("operation") != "reconcile"

or source.get("resource_type") != "knowledge-source"

or source.get("action") not in {"create", "reuse"}

or not isinstance(desired, dict)

or desired.get("kind") != "file"

or ingestion.get("operation") != "ingest"

or plan.get("owner") != source.get("owner")

or any(

source.get(field) != ingestion.get(field)

for field in ("endpoint", "name", "api_version", "owner")

)

):

raise HelperFailure(

"step-contract-mismatch",

"Source reconciliation and ingestion must target the same approved File source.",

blocked_at="input-resolution",

)

source_mode = desired.get("fileParameters", {}).get(

"ingestionParameters", {}

).get("contentExtractionMode")

if source_mode != ingestion.get("extraction_mode"):

raise HelperFailure(

"step-contract-mismatch",

"Source and ingestion extraction modes must match exactly.",

blocked_at="input-resolution",

)

search_reconcile._validate_plan(source)

file_ingest._validate_plan(ingestion)

source_vector.validate_plan_choice(plan)

if any(field in plan for field in ("file_cu_plan_version", "content_understanding", "cu_resource_state")):

if plan.get("file_cu_plan_version") not in ("1.0", "1.1", "1.2") or source_mode != "standard":

raise _cu_failure("cu-plan-mismatch", "New File CU plans require their supported CU-specific version and standard extraction.")

cu = validate_content_understanding(plan.get("content_understanding"), enabled=True)

automatic = cu["auth"] == "api-key-arm"

mi = cu["auth"] == "system-assigned"

acquisition = source.get("ai_services_key_acquisition")

if automatic:

file_cu_auth.validate_acquisition(acquisition, cu["endpoint"])

if (

plan["file_cu_plan_version"] != ("1.2" if mi else "1.1" if automatic else "1.0")

or (automatic and acquisition["resource_id"] != cu["resource_id"])

or (not automatic and acquisition is not None)

or source.get("ai_services_managed_identity") is not (True if mi else None)

or (not mi and "cu_identity_state" in plan)

):

raise _cu_failure("cu-plan-mismatch", "CU auth mode, version and exact acquisition scope must match the approved plan.")

settings = desired["fileParameters"]["ingestionParameters"]

if (

settings.get("aiServices") != {"uri": cu["endpoint"].rstrip("/")}

or settings.get("identity") is not None

or settings.get("disableImageVerbalization") is not True

or settings.get("chatCompletionModel") is not None

or source.get("ai_services_api_key_environment") != cu.get("api_key_environment")

or ("embedding" in plan) != (settings.get("embeddingModel") is not None)

):

raise _cu_failure("cu-plan-mismatch", "CU/embedding choices, credential channel and source processing must match the approved definition.")

if require_cu_readback:

state = plan.get("cu_resource_state")

if not isinstance(state, dict) or state != _cu_account_state(cu, state):

raise _cu_failure("cu-prerequisite-missing", "Retain the planner's selected CU account readback.")

if mi:

file_cu_mi.validate_state(plan.get("cu_identity_state"), cu, source["endpoint"])

elif source.get("ai_services_key_acquisition") is not None or source.get("ai_services_managed_identity") is not None or "cu_identity_state" in plan:

raise _cu_failure("cu-plan-mismatch", "Automatic acquisition requires the complete versioned File CU workflow.")

return source, ingestion

def _writes(result: dict[str, Any]) -> list[dict[str, Any]]:

writes: list[dict[str, Any]] = []

for action in ("created", "updated"):

for resource in result["resources"].get(action, []):

writes.append(

{

"action": action,

"type": resource["type"],

"name": resource["name"],

}

)

return writes

@reporting("file-source")

def execute(

document: dict[str, Any],

*,

token_provider: TokenProvider = azure_cli_token,

transport: Transport = http_request,

progress: Progress | None = None,

context_provider=file_cu_auth.account_context,

mi_on_created=None,

cleanup_capture=None,

upload_receipt_dir=None,

allow_upload_retry=True,

) -> dict[str, Any]:

progress.update("validation")

plan = document["plan"]

fingerprint = document["_computed_fingerprint"]

source, ingestion = _validate_plan(plan)

upload_session = None

if upload_receipt_dir is not None:

try:

from .file_upload import Session

except ImportError:

from file_upload import Session

upload_session = Session(upload_receipt_dir, document, context_provider=context_provider)

private_key = None

acquisition = source.get("ai_services_key_acquisition")

mi = source.get("ai_services_managed_identity") is True

mi_callback = None

if mi_on_created is not None and (not mi or not callable(mi_on_created)):

raise _cu_failure("creation-callback-unsupported", "Private MI checkpoints cannot receive File key-auth wire.")

if acquisition is not None or mi:

approval = document.get("approval")

if (

not isinstance(approval, dict) or approval.get("confirmed") is not True

or approval.get("fingerprint") != digest(plan) or fingerprint != digest(plan)

):

raise _cu_failure("approval-missing", "Private credential acquisition requires the unchanged fingerprinted source approval.")

if acquisition is not None:

file_cu_auth.check_context(acquisition["context"], context_provider)

child = {"_computed_fingerprint": fingerprint}

cu_ids = []

if "content_understanding" in plan:

state, cu_ids = read_content_understanding(

plan["content_understanding"], token_provider=token_provider, transport=transport,

)

if not _cu_states_match(state, plan["cu_resource_state"]):

raise _cu_failure(

"cu-prerequisite-drift", "CU account access or configuration changed since approval; refresh the plan.",

request_id=cu_ids[-1] if cu_ids else None,

)

if mi:

binding, ids = file_cu_mi.read_binding(

plan["content_understanding"], source["endpoint"], token_provider=token_provider, transport=transport,

)

cu_ids.extend(ids)

if binding != plan["cu_identity_state"]:

raise _cu_failure("cu-mi-identity-drift", "Search identity, CU role assignment or network changed since approval; refresh the concrete plan.")

raw_transport = transport

def recheck_mi():

state, _ = read_content_understanding(

plan["content_understanding"], token_provider=token_provider, transport=raw_transport,

)

binding, _ = file_cu_mi.read_binding(

plan["content_understanding"], source["endpoint"], token_provider=token_provider, transport=raw_transport,

)

if not _cu_states_match(state, plan["cu_resource_state"]) or binding != plan["cu_identity_state"]:

raise _cu_failure("cu-mi-identity-drift", "CU account or Search identity/role/network changed immediately before source PUT.")

transport, mi_callback = file_cu_mi.guard_create(plan, transport, recheck_mi, mi_on_created)

if acquisition is not None:

def recheck():

current, _ = read_content_understanding(

plan["content_understanding"], token_provider=token_provider, transport=transport,

)

if not _cu_states_match(current, plan["cu_resource_state"]):

raise _cu_failure("cu-prerequisite-drift", "CU account changed before credential acquisition; refresh the plan and approval.")

private_key = file_cu_auth.PrivateKey(

acquisition, token_provider=token_provider, transport=transport,

context_provider=context_provider, recheck=recheck,

)

transport = private_key.transport

if upload_session is not None:

transport = upload_session.transport(transport)

progress.update("source-reconciliation")

source_result = search_reconcile.execute(

{**child, "plan": source},

token_provider=token_provider,

transport=source_vector.guard_readback_transport(plan, transport),

credential_provider=private_key.acquire if private_key else None,

**({"managed_identity_verified": True, "on_created": mi_callback} if mi else {}),

**({"cleanup_capture": cleanup_capture} if cleanup_capture is not None else {}),

**({"on_file_acknowledged": upload_session.acknowledge} if upload_session is not None else {}),

)

source_writes = _writes(source_result)

def check_retry_source(recovery, token):

_validate_plan(plan)

current, _ = search_reconcile._get(

search_reconcile.resource_url(source), token,

transport=source_vector.guard_readback_transport(plan, transport), recovery=recovery,

)

etag = source_result["verification"]["readback"]["etag"]

if (not etag or current is None or current.get("@odata.etag") != etag

or not search_reconcile.definitions_match(source["desired"], current)):

raise HelperFailure("file-upload-source-drift", "Retry requires the unchanged acknowledged source version/definition.",

blocked_at="verification")

try:

if upload_session is not None:

etag = upload_session.require_ack()

if source_result["verification"]["readback"]["etag"] != etag:

raise HelperFailure("file-upload-source-drift", "Source readback differs from the retained creation ACK; retain the source.",

blocked_at="verification")

ingestion_result = file_ingest.execute(

{**child, "plan": ingestion},

token_provider=token_provider,

transport=transport,

allow_new_uploads=bool(source_result["resources"]["created"]),

progress=progress,

source_check=check_retry_source,

allow_upload_retry=allow_upload_retry,

**({"upload_session": upload_session} if upload_session is not None else {}),

)

except HelperFailure as failure:

combined = HelperFailure(

failure.code,

failure.message,

blocked_at=failure.blocked_at,

writes=source_writes + failure.writes,

resources_remaining=(

[

{

"type": str(write["type"]),

"name": str(write["name"]),

}

for write in source_writes

]

+ failure.resources_remaining

),

resources_reused=failure.resources_reused,

resources_unverified=failure.resources_unverified,

warnings=failure.warnings,

request_id=failure.request_id,

status=failure.http_status,

partial=bool(source_writes or failure.writes or failure.partial),

)

combined.file_batch = failure.file_batch

raise combined from failure

resources = {"created": [], "reused": [], "updated": [], "skipped": []}

for action in resources:

resources[action].extend(source_result["resources"].get(action, []))

resources[action].extend(ingestion_result["resources"].get(action, []))

return {

"status": "completed",

"outcome": str(plan.get("outcome") or "create-file-knowledge-source"),

"approved_plan": {"fingerprint": fingerprint, "confirmed": True},

"resources": resources,

"api_contracts": source_result["api_contracts"]

+ ingestion_result["api_contracts"],

"data_movement": ingestion_result["data_movement"],

"auth": ingestion_result["auth"],

"rbac": ingestion_result["rbac"],

"network": ingestion_result["network"],

"verification": {

**({"cu_managed_identity": {

"mode": "system-assigned", "binding_digest": digest(plan["cu_identity_state"]),

"processing": "unverified until indexed OCR marker validation",

"principal_attribution": "unverified; no backend identity telemetry collected",

}} if mi else {}),

"readback": {

"source": source_result["verification"]["readback"],

"files": ingestion_result["verification"]["readback"],

},

"request_ids": cu_ids + source_result["verification"]["request_ids"]

+ ingestion_result["verification"]["request_ids"],

"idempotency": (

"exact source and file marker readback is zero-write"

),

},

"warnings": source_result["warnings"] + ingestion_result["warnings"],

**({"file_batch": ingestion_result["file_batch"]} if "file_batch" in ingestion_result else {}),

"ownership": {

"run_owned": source_result["ownership"]["run_owned"]

+ ingestion_result["ownership"]["run_owned"],

"reused_not_owned": source_result["ownership"]["reused_not_owned"]

+ ingestion_result["ownership"]["reused_not_owned"],

"owner": plan.get("owner"),

},

"cleanup": {

"status": "not-requested",

"separate_confirmation_required": True,

},

}

def main(argv: list[str] | None = None) -> int:

try:

from . import _cleanup_receipts as cleanup_receipts

from .private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode

except ImportError:

import _cleanup_receipts as cleanup_receipts

from private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode

parser = argparse.ArgumentParser()

modes = parser.add_mutually_exclusive_group(required=True)

modes.add_argument("--input", type=Path)

modes.add_argument("--plan", type=Path)

add_execution_output_argument(parser)

add_progress_argument(parser)

cleanup_receipts.add_argument(parser)

parser.add_argument("--upload-receipt-dir", type=Path,

help="Required for creation: existing empty private directory for original File ACK and pre-upload attempts.")

args = parser.parse_args(argv)

fingerprint: str | None = None

owner: Any = None

outcome = "create-file-knowledge-source"

capture = None

try:

validate_execution_output_mode(args)

if args.plan and (args.cleanup_receipt_dir or args.upload_receipt_dir):

raise HelperFailure("input-schema-invalid", "Creation capture requires approved --input.", blocked_at="confirmation")

if args.plan:

try:

request = json.loads(args.plan.read_text(encoding="utf-8"))

except (OSError, UnicodeError, json.JSONDecodeError) as exc:

raise HelperFailure(

"input-unreadable", "Planning input must be readable UTF-8 JSON.",

blocked_at="input-resolution",

) from exc

result = plan_source(request)

emit_plan_result(result, args.execution_output)

return 0

document, plan, fingerprint = load_approved_input(args.input)

document["_computed_fingerprint"] = fingerprint

owner = plan.get("owner")

outcome = str(plan.get("outcome") or outcome)

source, _ = _validate_plan(plan)

if source["action"] == "create" and args.cleanup_receipt_dir is None:

raise HelperFailure(

"file-creation-receipt-required",

"File creation requires explicit --cleanup-receipt-dir pointing to an existing protected private directory; no default is inferred.",

blocked_at="confirmation",

)

if args.cleanup_receipt_dir is not None:

directory = cleanup_receipts.private_io.validate_private_artifact_directory(str(args.cleanup_receipt_dir))

if args.upload_receipt_dir is not None:

upload_directory = cleanup_receipts.private_io.validate_private_artifact_directory(str(args.upload_receipt_dir))

if directory == upload_directory:

raise HelperFailure("file-receipt-directory-conflict",

"Use separate private cleanup and upload journal directories.",

blocked_at="confirmation")

elif source["action"] == "create":

raise HelperFailure(

"file-upload-receipt-required",

"File creation requires explicit --upload-receipt-dir for durable original ACK and pre-upload attempt records; use a separate existing empty private directory.",

blocked_at="confirmation",

)

capture = cleanup_receipts.Capture(directory, document)

result = execute(document, progress=Progress("file-source", enabled=args.progress),

**({"cleanup_capture": capture} if capture else {}),

**({"upload_receipt_dir": args.upload_receipt_dir} if args.upload_receipt_dir else {}))

except HelperFailure as failure:

result = blocked_result(

failure,

outcome=outcome,

fingerprint=fingerprint,

owner=owner,

)

if capture is not None:

result["cleanup_receipts"] = capture.summaries

result["safe_next_decision"] = (

"Retain any acknowledged source and confirmed uploads. Do not rerun the creation envelope, "

"replay uncertain uploads, reset or delete resources. Use file_upload.py --plan with the "

"original upload journal for newly approved never-attempted files; missing evidence blocks continuation, not retention."

)

emit_result(result)

return 3 if result["status"] == "partial" else 2

if capture is not None:

result["cleanup_receipts"] = capture.summaries

emit_result(result)

return 0

if __name__ == "__main__":

sys.exit(main())