Setting the file. One moment.
Source Vector · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary
def _read_json
— line 176
This file
Number 10.68
Position 68 of 77
Type Python
Size 41 KB
Lines 706 helpers/ source_vector.py
Python · 706 lines · 41 KB
12
from
urllib.parse
import
urlsplit
13
14 try :
15 from . import search_reconcile
16 from ._common import (
17 SEARCH_AUDIENCE , HelperFailure, TokenProvider, Transport, azure_cli_token,
18 blocked_result, digest, emit_result, http_request, load_approved_input,
19 odata_name, reject_secrets, require_allowed_fields,
20 )
21 except ImportError :
22 import search_reconcile
23 from _common import (
24 SEARCH_AUDIENCE , HelperFailure, TokenProvider, Transport, azure_cli_token,
25 blocked_result, digest, emit_result, http_request, load_approved_input,
26 odata_name, reject_secrets, require_allowed_fields,
27 )
28
29
30 MODELS = { "text-embedding-ada-002" : ( 1536 , 1536 ),
31 "text-embedding-3-small" : ( 1 , 1536 ), "text-embedding-3-large" : ( 1 , 3072 )}
32 IDENTIFIER = re.compile( r " ^[ A-Za-z ][ A-Za-z0-9_ ] {0,127} $ " )
33
34
35 def fail (code: str , message: str ) -> HelperFailure:
36 return HelperFailure(code, message, blocked_at = "vector-verification" )
37
38
39 def _text (value: Any, * , maximum: int = 4096 ) -> bool :
40 return isinstance (value, str ) and bool (value.strip()) and len (value) <= maximum
41
42
43 def _json_valid (value: Any) -> None :
44 try :
45 json.dumps(value, ensure_ascii = False , allow_nan = False ).encode( "utf-8" )
46 except ( UnicodeError , TypeError , ValueError ) as exc:
47 raise fail( "input-schema-invalid" , "Evidence must be finite, valid UTF-8 JSON." ) from exc
48
49
50 def validate_choice (value: Any, * , enabled: bool , api_version: str ) -> dict[ str , Any] | None :
51 if not enabled:
52 if value is not None :
53 raise fail( "embedding-choice-conflict" , "Lexical planning must omit embedding configuration." )
54 return None
55 if not isinstance (value, dict ):
56 raise fail( "embedding-prerequisite-missing" , "Vector planning requires resolved embedding choices." )
57 reject_secrets(value)
58 require_allowed_fields(value, {
59 "endpoint" , "deployment" , "model" , "dimensions" , "model_version" , "auth" , "prerequisites" ,
60 }, label = "embedding choices" )
61 if api_version not in { "2026-04-01" , "2026-08-01-preview" }:
62 raise fail( "embedding-version-unsupported" , "Unsupported Search embedding API version." )
63 try :
64 json.dumps(value, ensure_ascii = False , allow_nan = False ).encode( "utf-8" )
65 uri = urlsplit(value.get( "endpoint" ) if isinstance (value.get( "endpoint" ), str ) else "" )
66 valid_endpoint = (
67 uri.scheme == "https" and uri.port is None and not uri.username and not uri.password
68 and not uri.query and not uri.fragment and uri.path in { "" , "/" }
69 and re.fullmatch(
70 r " [ a-z0-9 ][ a-z0-9- ] {0,62} \. ( openai \. azure \. com | services \. ai \. azure \. com | cognitiveservices \. azure \. com ) " ,
71 uri.netloc,
72 )
73 )
74 except ( UnicodeError , ValueError , TypeError ) as exc:
75 raise fail( "embedding-choice-invalid" , "Embedding choices must be valid UTF-8 and an exact HTTPS endpoint." ) from exc
76 if (
77 not valid_endpoint or not _text(value.get( "deployment" ), maximum = 64 )
78 or not re.fullmatch( r " [ A-Za-z0-9 ][ A-Za-z0-9_.- ] * " , value[ "deployment" ])
79 or not isinstance (value.get( "model" ), str ) or value[ "model" ] not in MODELS
80 ):
81 raise fail( "embedding-choice-invalid" , "Select a supported Azure OpenAI endpoint, deployment and embedding model." )
82 if value.get( "dimensions" ) != "service-managed" or value.get( "model_version" ) != "deployment-managed" :
83 raise fail( "embedding-parameter-unsupported" , "Knowledge-source APIs expose neither dimensions nor a model-version pin; custom values are unsupported." )
84 if value.get( "auth" ) != "system-assigned" :
85 raise fail( "embedding-auth-unsupported" , "This planner supports only the existing Search system-assigned identity; no keys or identity changes." )
86 prerequisites = value.get( "prerequisites" )
87 if not isinstance (prerequisites, dict ):
88 raise fail( "embedding-prerequisite-missing" , "Provide current deployment, identity and network evidence references." )
89 require_allowed_fields(prerequisites, { "deployment" , "identity" , "network" }, label = "embedding prerequisites" )
90 if any ( not _text(prerequisites.get(key)) for key in ( "deployment" , "identity" , "network" )):
91 raise fail( "embedding-prerequisite-missing" , "Owner-verified deployment/model, Search identity/RBAC and network evidence is required." )
92 return copy.deepcopy(value)
93
94
95 def model_definition (choice: dict[ str , Any]) -> dict[ str , Any]:
96 return { "kind" : "azureOpenAI" , "azureOpenAIParameters" : {
97 "resourceUri" : choice[ "endpoint" ].rstrip( "/" ), "deploymentId" : choice[ "deployment" ],
98 "modelName" : choice[ "model" ], "authIdentity" : None ,
99 }}
100
101
102 def verify_model_readback (choice: dict[ str , Any], model: Any) -> None :
103 parameters = model.get( "azureOpenAIParameters" ) if isinstance (model, dict ) else None
104 if not isinstance (parameters, dict ):
105 raise fail( "embedding-readback-invalid" , "Observed embedding parameters must be an object." )
106 # Generic definition normalization omits apiKey; check auth evidence before matching.
107 api_key = parameters.get( "apiKey" )
108 if (
109 api_key is not None and not ( isinstance (api_key, str ) and api_key == "" )
110 or parameters.get( "authIdentity" ) is not None
111 ):
112 raise fail( "embedding-auth-conflict" , "Observed embedding authentication does not prove the selected system-assigned mode; key/identity details withheld." )
113 if not search_reconcile.definitions_match(model_definition(choice), model):
114 raise fail( "embedding-readback-mismatch" , "Observed embedding endpoint/deployment/model differs from the selected configuration." )
115
116
117 def verify_source_readback (choice: dict[ str , Any] | None , current: Any) -> None :
118 if choice is None or current is None :
119 return
120 kind = current.get( "kind" ) if isinstance (current, dict ) else None
121 if kind not in ( "file" , "azureBlob" ):
122 raise fail( "embedding-readback-invalid" , "Expected a File/Blob embedding source readback." )
123 parameters = current.get( "fileParameters" if kind == "file" else "azureBlobParameters" )
124 ingestion = parameters.get( "ingestionParameters" ) if isinstance (parameters, dict ) else None
125 verify_model_readback(choice, ingestion.get( "embeddingModel" ) if isinstance (ingestion, dict ) else None )
126
127
128 def guard_readback_transport (plan: dict[ str , Any], transport: Transport) -> Transport:
129 if "embedding" not in plan and "content_understanding" not in plan:
130 return transport
131 url = search_reconcile.resource_url(plan[ "source" ])
132
133 def guarded (method: str , target: str , token: str , ** kwargs: Any):
134 response = transport(method, target, token, ** kwargs)
135 if method == "GET" and target == url and response.status == 200 :
136 try :
137 verify_source_readback(plan.get( "embedding" ), response.body)
138 if "content_understanding" in plan:
139 try :
140 from . import blob_source, file_source
141 except ImportError :
142 import blob_source, file_source
143 owner = file_source if plan[ "source" ][ "desired" ][ "kind" ] == "file" else blob_source
144 owner.verify_content_understanding_readback(plan[ "content_understanding" ], response.body)
145 except HelperFailure as failure:
146 failure.request_id = response.request_id
147 raise
148 return response
149
150 return guarded
151
152
153 def validate_plan_choice (plan: dict[ str , Any]) -> None :
154 if "embedding" not in plan:
155 return
156 source = plan.get( "source" , {})
157 choice = validate_choice(plan[ "embedding" ], enabled = True , api_version = source.get( "api_version" ))
158 desired = source.get( "desired" , {})
159 parameters = desired.get( "fileParameters" if desired.get( "kind" ) == "file" else "azureBlobParameters" , {})
160 ingestion = parameters.get( "ingestionParameters" , {})
161 if ingestion.get( "contentExtractionMode" ) not in ( "minimal" , "standard" ) or ingestion.get( "embeddingModel" ) != model_definition(choice):
162 raise fail( "embedding-plan-mismatch" , "Embedding choices and the selected extraction definition must match." )
163
164
165 def summary (choice: dict[ str , Any]) -> dict[ str , Any]:
166 return {
167 key: choice[key] for key in ( "endpoint" , "deployment" , "model" , "dimensions" , "model_version" , "auth" )
168 } | {
169 "purpose" : "Source vectors for hybrid/vector search; not CU extraction or KB chat." ,
170 "cost" : "Creating/ingesting this source invokes billable embeddings and moves content to the selected model. Query vectorization needs separate approval." ,
171 "prerequisites" : "Supplied evidence references are owner-verified, not proof of effective access or deployed model version." ,
172 "answer_synthesis" : "Not configured; ingestion dependencies do not configure KB reasoning, reranking or answer synthesis." ,
173 }
174
175
176 def _read_json (path: Any) -> dict[ str , Any]:
177 if not _text(path):
178 raise fail( "input-schema-invalid" , "Select an explicit JSON file." )
179 try :
180 value = json.loads(Path(path).read_text( encoding = "utf-8" ))
181 except ( OSError , UnicodeError , ValueError ) as exc:
182 raise fail( "input-unreadable" , "Selected evidence must be readable UTF-8 JSON." ) from exc
183 if not isinstance (value, dict ):
184 raise fail( "input-schema-invalid" , "Selected JSON must contain an object." )
185 _json_valid(value)
186 return value
187
188
189 def _source_plan (path: str ) -> dict[ str , Any]:
190 document = _read_json(path)
191 reject_secrets(document)
192 require_allowed_fields(document, { "schema_version" , "plan" , "approval" }, label = "source artifact" )
193 plan, approval = document.get( "plan" ), document.get( "approval" )
194 if (
195 document.get( "schema_version" ) != "1.0" or not isinstance (plan, dict )
196 or not isinstance (approval, dict ) or type (approval.get( "confirmed" )) is not bool
197 or approval.get( "fingerprint" ) != digest(plan)
198 ):
199 raise fail( "source-artifact-invalid" , "Retain the unchanged source execution_input; prior consent is not query consent." )
200 require_allowed_fields(approval, { "confirmed" , "fingerprint" }, label = "source approval" )
201 source = plan.get( "source" )
202 desired = source.get( "desired" ) if isinstance (source, dict ) else None
203 if (
204 not isinstance (desired, dict ) or desired.get( "kind" ) not in ( "file" , "azureBlob" )
205 or not isinstance (source.get( "api_version" ), str )
206 or source[ "api_version" ] not in search_reconcile. SUPPORTED_API_VERSIONS
207 or source.get( "action" ) not in ( "create" , "reuse" ) or not _text(plan.get( "owner" ))
208 ):
209 raise fail( "source-artifact-invalid" , "Source artifact needs a supported same-owner File/Blob definition and API." )
210 parameters = desired.get( "fileParameters" if desired[ "kind" ] == "file" else "azureBlobParameters" )
211 if not isinstance (parameters, dict ) or not isinstance (parameters.get( "ingestionParameters" ), dict ):
212 raise fail( "source-artifact-invalid" , "Source ingestion parameters must be an object." )
213 try :
214 from . import file_source, blob_source
215 except ImportError :
216 import file_source, blob_source
217 if plan.get( "operation" ) == "reconcile-and-ingest" :
218 file_source._validate_plan(plan)
219 elif plan.get( "operation" ) == "reconcile-and-monitor" :
220 blob_source._validate_plan(plan)
221 else :
222 raise fail( "source-artifact-invalid" , "Only File/Blob source execution artifacts are supported." )
223 if "embedding" not in plan:
224 if parameters[ "ingestionParameters" ].get( "embeddingModel" ) is None :
225 raise fail( "embedding-not-configured" , "Source vectorization is not configured; CU extraction does not enable vectors." )
226 raise 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." )
227 validate_plan_choice(plan)
228 if parameters[ "ingestionParameters" ].get( "ingestionPermissionOptions" ) not in ( None , []):
229 raise fail( "permission-query-unsupported" , "Permission-aware retrieval belongs to the KB workflow; direct-index verification is unsupported." )
230 return plan
231
232
233 def _request (request: Any) -> dict[ str , Any]:
234 if not isinstance (request, dict ):
235 raise fail( "input-schema-invalid" , "Verification input must be an object." )
236 reject_secrets(request)
237 require_allowed_fields(request, {
238 "schema_version" , "source_plan_file" , "vector_field" , "content_field" , "citation_field" ,
239 "expected_citation" , "expected_text" , "query" , "mode" , "k" , "not_before" ,
240 "reuse_input_file" , "reuse_result_file" ,
241 }, label = "vector verification intent" )
242 try :
243 json.dumps(request, ensure_ascii = False , allow_nan = False ).encode( "utf-8" )
244 except ( UnicodeError , ValueError , TypeError ) as exc:
245 raise fail( "input-schema-invalid" , "Verification choices must be valid UTF-8 JSON." ) from exc
246 if (
247 request.get( "schema_version" ) != "1.0" or request.get( "mode" ) not in ( "vector" , "hybrid" )
248 or type (request.get( "k" )) is not int or not 1 <= request[ "k" ] <= 10
249 or any ( not _text(request.get(key)) for key in ( "query" , "expected_citation" , "expected_text" ))
250 or any ( not isinstance (request.get(key), str ) or not IDENTIFIER .fullmatch(request[key])
251 for key in ( "vector_field" , "content_field" , "citation_field" ))
252 or len ({request[ "vector_field" ], request[ "content_field" ], request[ "citation_field" ]}) != 3
253 ):
254 raise fail( "input-schema-invalid" , "Resolve vector/hybrid, k 1–10, three distinct simple field names, query and expected citation/text." )
255 plan = _source_plan(request.get( "source_plan_file" ))
256 try :
257 from . import blob_source
258 except ImportError :
259 import blob_source
260 blob_source._receipt_paths(request)
261 if plan[ "source" ][ "desired" ][ "kind" ] == "azureBlob" :
262 blob_source._timestamp(request.get( "not_before" ))
263 elif "not_before" not in request or request[ "not_before" ] is not None :
264 raise fail( "input-schema-invalid" , "File verification requires explicit not_before null; uploads are synchronous." )
265 return plan
266
267
268 def _named (items: Any, name: str ) -> dict[ str , Any]:
269 if (
270 not _text(name, maximum = 128 ) or not isinstance (items, list )
271 or not all ( isinstance (item, dict ) and _text(item.get( "name" ), maximum = 128 ) for item in items)
272 or len ({item[ "name" ] for item in items}) != len (items)
273 ):
274 raise fail( "vector-configuration-invalid" , "Expected complete generated metadata arrays." )
275 selected = [item for item in items if item.get( "name" ) == name]
276 if len (selected) != 1 :
277 raise fail( "vector-configuration-invalid" , "Generated field/profile/vectorizer identity is absent or ambiguous." )
278 return selected[ 0 ]
279
280
281 def verify_index (index: dict[ str , Any], choice: dict[ str , Any], request: dict[ str , Any]) -> dict[ str , Any]:
282 fields = index.get( "fields" )
283 vector = _named(fields, request[ "vector_field" ])
284 if index.get( "permissionFilterOption" ) not in ( None , "disabled" ) or any (field.get( "permissionFilter" ) for field in fields):
285 raise fail( "permission-query-unsupported" , "Permission-enabled indexes require the owning permission-aware retrieval workflow." )
286 low, high = MODELS [choice[ "model" ]]
287 dimensions = vector.get( "dimensions" )
288 if (
289 vector.get( "type" ) != "Collection(Edm.Single)" or vector.get( "searchable" ) is not True
290 or type (dimensions) is not int or not low <= dimensions <= high
291 ):
292 raise fail( "vector-configuration-invalid" , "Generated vector field type/searchability/dimensions do not match the selected model." )
293 configuration = index.get( "vectorSearch" )
294 if not isinstance (configuration, dict ):
295 raise fail( "vector-configuration-invalid" , "Generated index has no vectorSearch configuration." )
296 profile = _named(configuration.get( "profiles" ), vector.get( "vectorSearchProfile" ))
297 algorithm = _named(configuration.get( "algorithms" ), profile.get( "algorithm" ))
298 if algorithm.get( "kind" ) not in { "hnsw" , "exhaustiveKnn" }:
299 raise fail( "vector-configuration-invalid" , "Unsupported generated vector algorithm." )
300 vectorizer = _named(configuration.get( "vectorizers" ), profile.get( "vectorizer" ))
301 verify_model_readback(choice, {key: value for key, value in vectorizer.items() if key != "name" })
302 for key in ( "content_field" , "citation_field" ):
303 field = _named(fields, request[key])
304 if field.get( "type" ) != "Edm.String" or field.get( "retrievable" ) is not True :
305 raise fail( "vector-configuration-invalid" , "Content and citation fields must be retrievable strings." )
306 if key == "content_field" and request[ "mode" ] == "hybrid" and field.get( "searchable" ) is not True :
307 raise fail( "vector-configuration-invalid" , "Hybrid requires the selected content field to be searchable." )
308 keys = [field for field in fields if field.get( "key" ) is True ]
309 if len (keys) != 1 or keys[ 0 ].get( "type" ) != "Edm.String" or keys[ 0 ].get( "retrievable" ) is not True :
310 raise fail( "vector-configuration-invalid" , "Require one retrievable string document key." )
311 return { "dimensions" : dimensions, "profile" : profile[ "name" ], "vectorizer" : vectorizer[ "name" ], "document_id_field" : keys[ 0 ][ "name" ]}
312
313
314 def _reader (transport: Transport, token_provider: TokenProvider):
315 request_ids: list[ str ] = []
316
317 def read (method: str , url: str , token: str , ** kwargs: Any):
318 if method != "GET" :
319 raise fail( "planning-write-forbidden" , "Verification observation permits only GETs." )
320 kwargs.update(
321 follow_redirects = False ,
322 max_response_bytes = min ( 8 * 1024 * 1024 , kwargs.get( "max_response_bytes" , 8 * 1024 * 1024 )),
323 response_deadline = min (
324 time.monotonic() + min ( 60 , kwargs.get( "timeout" , 60 )),
325 kwargs.get( "response_deadline" , float ( "inf" )),
326 ),
327 )
328 result = transport(method, url, token, ** kwargs)
329 _json_valid(result.body)
330 if result.request_id:
331 request_ids.append(result.request_id)
332 return result
333
334 def get (url: str ):
335 value, _ = search_reconcile.read_resource(url, token_provider( SEARCH_AUDIENCE ), transport = read)
336 if value is None :
337 raise fail( "vector-resource-missing" , "An exact selected source/generated index is absent." )
338 if not _text(value.get( "@odata.etag" )):
339 raise fail( "vector-evidence-missing" , "Fresh source and index ETags are required." )
340 return value
341
342 return read, get, request_ids
343
344
345 def _prior_blob_ingestion (
346 plan: dict[ str , Any], request: dict[ str , Any], current: dict[ str , Any],
347 generated: list[dict[ str , Any]], readiness: dict[ str , Any],
348 receipts: tuple[dict[ str , Any], dict[ str , Any]] | None ,
349 ) -> dict[ str , Any]:
350 try :
351 from . import blob_source
352 except ImportError :
353 import blob_source
354 if receipts is None :
355 raise fail( "ingestion-unverified" , "A zero-work cycle requires retained approved input and successful nonempty creation result files." )
356 prior, result = receipts
357 _json_valid({ "plan" : prior, "result" : result})
358 blob_source._verify_creation_binding(
359 plan[ "source" ], plan[ "boundary" ], current, generated, receipts,
360 )
361 if (
362 prior[ "owner" ] != plan[ "owner" ] or prior[ "inventory_digest" ] != plan[ "inventory_digest" ]
363 or result[ "verification" ].get( "source_digest" ) != plan[ "inventory_digest" ]
364 ):
365 raise fail( "ingestion-unverified" , "Prior ingestion must bind the same owner and unchanged before/after Storage inventory." )
366 previous = result.get( "readiness" )
367 if not isinstance (previous, dict ) or previous.get( "status" ) != "verified" :
368 raise fail( "ingestion-unverified" , "Retained creation must include verified ingestion readiness." )
369 require_allowed_fields(previous, {
370 "status" , "synchronization" , "not_before" , "request_ids" , "first_retry" , "watch" , "progress" ,
371 }, label = "retained ingestion readiness" )
372 cycle = previous.get( "synchronization" )
373 fields = { "startTime" , "endTime" , "itemsUpdatesProcessed" , "itemsUpdatesFailed" , "itemsSkipped" }
374 if not isinstance (cycle, dict ) or set (cycle) != fields:
375 raise fail( "ingestion-unverified" , "Retain the complete creation synchronization interval and counters." )
376 if "progress" in previous:
377 try :
378 from ._progress import validate_blob_progress
379 except ImportError :
380 from _progress import validate_blob_progress
381 display = previous[ "progress" ]
382 try :
383 validate_blob_progress(display)
384 except ValueError as exc:
385 raise fail( "verification-drift" , "Retained Blob progress metadata is malformed." ) from exc
386 if (display[ "phase" ] != "completed"
387 or display[ "synchronization_status" ] not in ( "active" , "not-reported" )
388 or display[ "next_check_seconds" ] is not None
389 or display[ "run_start" ] != blob_source._timestamp(cycle[ "startTime" ]).isoformat()
390 or [display[key] for key in ( "processed" , "failed" , "skipped" )] !=
391 [cycle[key] for key in ( "itemsUpdatesProcessed" , "itemsUpdatesFailed" , "itemsSkipped" )]):
392 raise fail( "verification-drift" , "Retained Blob progress must describe its own completed cycle." )
393 if "watch" in previous:
394 watch = previous[ "watch" ]
395 expected_latest = {key: cycle[key] for key in fields - { "endTime" }}
396 expected_latest[ "state" ] = "completed-cycle-observed"
397 latest = watch.get( "latest" ) if isinstance (watch, dict ) else None
398 if isinstance (latest, dict ) and "indexer_execution" in latest:
399 execution = latest[ "indexer_execution" ]
400 if not blob_source._execution_completed_by(
401 execution, blob_source._timestamp(cycle[ "endTime" ]),
402 max (blob_source._timestamp(request[ "not_before" ]), blob_source._timestamp(previous.get( "not_before" ))),
403 ):
404 raise fail( "verification-drift" , "Retained active execution must be resolved before its primary cycle ends." )
405 expected_latest[ "indexer_execution" ] = execution
406 if ( not isinstance (watch, dict )
407 or set (watch) != { "schema_version" , "state" , "elapsed_seconds" , "status_checks" , "latest" }
408 or watch[ "schema_version" ] != "1.0" or watch[ "state" ] != "completed"
409 or type (watch[ "elapsed_seconds" ]) not in ( int , float )
410 or not math.isfinite(watch[ "elapsed_seconds" ]) or watch[ "elapsed_seconds" ] < 0
411 or type (watch[ "status_checks" ]) is not int or not 1 <= watch[ "status_checks" ] <= 1000
412 or watch[ "latest" ] != expected_latest):
413 raise fail( "verification-drift" , "Retained watch metadata must match its completed cycle; it cannot replace ingestion proof." )
414 counts = [cycle[key] for key in ( "itemsUpdatesProcessed" , "itemsUpdatesFailed" , "itemsSkipped" )]
415 if any ( type (count) is not int or count < 0 for count in counts) or counts[ 0 ] == 0 or any (counts[ 1 :]):
416 raise fail( "ingestion-unverified" , "Prior creation must have processed content without failures or skipped items." )
417 start, end = (blob_source._timestamp(cycle[key]) for key in ( "startTime" , "endTime" ))
418 if (
419 start < max (blob_source._timestamp(request[ "not_before" ]), blob_source._timestamp(previous.get( "not_before" )))
420 or end < start or end > blob_source._timestamp(readiness[ "synchronization" ][ "startTime" ])
421 ):
422 raise fail( "ingestion-unverified" , "Prior ingestion must follow both lower bounds and complete before the current zero-work cycle." )
423 return {
424 "synchronization" : copy.deepcopy(cycle),
425 "evidence_digest" : digest({ "creation_plan" : prior, "creation_result" : result}),
426 }
427
428
429 def observe (request: dict[ str , Any], * , token_provider: TokenProvider, transport: Transport,
430 storage_transport: Transport) -> tuple[dict[ str , Any], dict[ str , Any]]:
431 plan = _request(request)
432 source = plan[ "source" ]
433 try :
434 from . import file_ingest, blob_inventory, blob_source
435 except ImportError :
436 import file_ingest, blob_inventory, blob_source
437 read, get, ids = _reader(guard_readback_transport(plan, transport), token_provider)
438 url = search_reconcile.resource_url(source)
439 current = get(url)
440 if not search_reconcile.definitions_match(source[ "desired" ], current):
441 raise fail( "definition-conflict" , "Fresh source differs from the selected source artifact." )
442 if source.get( "expected_etag" ) and current[ "@odata.etag" ] != source[ "expected_etag" ]:
443 raise fail( "definition-drift" , "Selected reuse source ETag changed; rerun the source planner." )
444 kind = source[ "desired" ][ "kind" ]
445 if kind == "file" :
446 created = current.get( "fileParameters" , {}).get( "createdResources" )
447 if not isinstance (created, dict ) or set (created) != { "index" }:
448 raise fail( "vector-evidence-missing" , "Require the File source's generated index identity." )
449 files, _ = file_ingest.read_inventory(plan[ "ingestion" ], token_provider( SEARCH_AUDIENCE ), transport = read)
450 matches = file_ingest.reconcile_inventory(plan[ "ingestion" ], files)
451 identities = [item.get( "fileId" ) for item in matches.values()]
452 if not identities or not all (_text(item) for item in identities) or len ( set (identities)) != len (identities):
453 raise fail( "ingestion-unverified" , "Require complete unique ingested File identities and markers." )
454 inventory = file_ingest.inventory_digest(files)
455 readiness = { "status" : "verified" , "files" : len (matches), "basis" : "synchronous File ingestion markers; no reported file errors" }
456 else :
457 created = {item[ "type" ]: item[ "name" ] for item in blob_source.generated_resources(current, strict = True )}
458 generated = blob_source.generated_resources(current, strict = True )
459 if "expected_generated" in plan and generated != plan[ "expected_generated" ]:
460 raise fail( "definition-drift" , "Generated Blob identities changed." )
461 receipts: tuple[dict[ str , Any], dict[ str , Any]] | None = None
462
463 def load_receipts () -> tuple[dict[ str , Any], dict[ str , Any]] | None :
464 nonlocal receipts
465 if receipts is None :
466 receipts = blob_source._reuse_receipts(blob_source._receipt_paths(request))
467 return receipts
468
469 blob_source._verify_storage_binding(
470 source, plan[ "boundary" ], current, generated, load_receipts,
471 )
472 snapshot = blob_inventory.discover(plan[ "boundary" ], plan[ "inventory_limits" ],
473 token_provider = token_provider, transport = storage_transport)
474 ids.extend(snapshot[ "request_ids" ])
475 inventory = snapshot[ "inventory_digest" ]
476 if inventory != plan[ "inventory_digest" ]:
477 raise fail( "inventory-drift" , "Storage content or ACL evidence changed; rerun source planning." )
478 readiness = blob_source.monitor(
479 source, not_before = blob_source._timestamp(request[ "not_before" ]),
480 limits = { "max_requests" : 1 , "interval_seconds" : 1 , "deadline_seconds" : 60 },
481 token_provider = token_provider, transport = read, sleep =lambda _: None ,
482 )
483 if readiness.get( "status" ) != "verified" :
484 raise fail( "ingestion-unverified" , "No relevant completed zero-failure Blob synchronization; do not query." )
485 cycle = readiness[ "synchronization" ]
486 if cycle[ "itemsSkipped" ]:
487 raise fail( "ingestion-unverified" , "A completed cycle with no skipped items is required." )
488 if cycle[ "itemsUpdatesProcessed" ] == 0 :
489 readiness[ "prior_ingestion" ] = _prior_blob_ingestion(
490 plan, request, current, generated, readiness, load_receipts(),
491 )
492 readiness[ "basis" ] = "Current zero-work success plus same-source, same-inventory nonempty creation proof."
493 index_name = created.get( "index" )
494 index_url = f " { source[ 'endpoint' ].rstrip( '/' ) } /indexes(' { odata_name(index_name) } ')?api-version= { source[ 'api_version' ] } "
495 index = get(index_url)
496 if index.get( "name" ) != index_name:
497 raise fail( "vector-configuration-mismatch" , "Generated index name does not match its source identity." )
498 metadata = verify_index(index, plan[ "embedding" ], request)
499 refreshed, refreshed_index = get(url), get(index_url)
500 source_changed = digest(current) != digest(refreshed)
501 if "content_understanding" in plan:
502 source_changed = (
503 current[ "@odata.etag" ] != refreshed[ "@odata.etag" ]
504 or not search_reconcile.definitions_match(current, refreshed)
505 or (
506 refreshed.get( "fileParameters" , {}).get( "createdResources" ) != created
507 if kind == "file" else blob_source.generated_resources(refreshed, strict = True ) != generated
508 )
509 )
510 if source_changed or digest(index) != digest(refreshed_index):
511 raise fail( "definition-drift" , "Source or index changed during observation." )
512 snapshot = {
513 "source_plan_digest" : digest(plan), "source_etag" : current[ "@odata.etag" ],
514 "source_definition" : digest(search_reconcile._definition(current)),
515 "index" : index_name, "index_etag" : index[ "@odata.etag" ], "index_digest" : digest(index),
516 "inventory_digest" : inventory, "vector" : metadata,
517 }
518 if "prior_ingestion" in readiness:
519 snapshot[ "ingestion_evidence_digest" ] = readiness[ "prior_ingestion" ][ "evidence_digest" ]
520 return snapshot, { "readiness" : readiness, "request_ids" : ids}
521
522
523 def plan_verification (request: dict[ str , Any], * , token_provider: TokenProvider = azure_cli_token,
524 transport: Transport = http_request, storage_transport: Transport = http_request) -> dict[ str , Any]:
525 source_plan = _request(request)
526 snapshot, evidence = observe(request, token_provider = token_provider, transport = transport, storage_transport = storage_transport)
527 if digest(source_plan) != snapshot[ "source_plan_digest" ]:
528 raise fail( "source-artifact-drift" , "Source artifact changed while planning verification." )
529 plan = { "operation" : "verify-source-vector" , "owner" : source_plan[ "owner" ], "cleanup_approved" : False ,
530 "request" : copy.deepcopy(request), "snapshot" : snapshot}
531 fingerprint = digest(plan)
532 return {
533 "status" : "planned" , "plan_fingerprint" : fingerprint,
534 "execution_input" : { "schema_version" : "1.0" , "plan" : plan, "approval" : { "confirmed" : False , "fingerprint" : fingerprint}},
535 "approval_summary" : {
536 "source" : source_plan[ "source" ][ "name" ], "index" : snapshot[ "index" ],
537 "search_endpoint" : source_plan[ "source" ][ "endpoint" ], "api_version" : source_plan[ "source" ][ "api_version" ],
538 "mode" : request[ "mode" ], "queries" : 2 if request[ "mode" ] == "hybrid" else 1 ,
539 "k" : request[ "k" ], "query" : request[ "query" ],
540 "embedding" : summary(source_plan[ "embedding" ]), "dimensions_observed" : snapshot[ "vector" ][ "dimensions" ],
541 "configuration" : "verified" , "ingestion" : "verified" , "retrieval" : "unverified" ,
542 "next_step" : "Approve the exact query/content transfer and embedding cost, then apply this unchanged artifact. No index repair or cleanup." ,
543 },
544 "evidence" : evidence, "writes_performed" : [],
545 }
546
547
548 def query_body (request: dict[ str , Any], metadata: dict[ str , Any], * , hybrid: bool ) -> dict[ str , Any]:
549 body = {
550 "vectorQueries" : [{ "kind" : "text" , "text" : request[ "query" ], "fields" : request[ "vector_field" ], "k" : request[ "k" ]}],
551 "select" : "," .join( dict .fromkeys((metadata[ "document_id_field" ], request[ "content_field" ], request[ "citation_field" ]))),
552 "top" : request[ "k" ], "minimumCoverage" : 100 ,
553 }
554 if hybrid:
555 body.update( search = request[ "query" ], searchFields = request[ "content_field" ], queryType = "simple" )
556 return body
557
558
559 def verify_response (body: Any, request: dict[ str , Any], metadata: dict[ str , Any]) -> int :
560 _json_valid(body)
561 if (
562 not isinstance (body, dict ) or type (body.get( "@search.coverage" )) not in ( int , float )
563 or body[ "@search.coverage" ] != 100
564 or any (body.get(key) is not None for key in
565 ( "@odata.nextLink" , "@search.nextPageParameters" , "@search.semanticPartialResponseReason" , "error" ))
566 or not isinstance (body.get( "value" ), list ) or not 1 <= len (body[ "value" ]) <= request[ "k" ]
567 ):
568 raise fail( "vector-query-incomplete" , "Require a nonempty, complete, full-coverage query response without continuation or partial errors." )
569 identities, matched = set (), 0
570 for hit in body[ "value" ]:
571 if not isinstance (hit, dict ):
572 raise fail( "vector-query-invalid" , "Malformed query hit." )
573 key, text, citation = (hit.get(field) for field in (metadata[ "document_id_field" ], request[ "content_field" ], request[ "citation_field" ]))
574 score = hit.get( "@search.score" )
575 if (
576 not all (_text(value, maximum = 1024 * 1024 ) for value in (key, text, citation))
577 or key in identities or type (score) not in ( int , float ) or not math.isfinite(score) or score < 0
578 ):
579 raise fail( "vector-query-invalid" , "Require unique document IDs, content, provenance and finite scores." )
580 identities.add(key)
581 matched += citation == request[ "expected_citation" ] and request[ "expected_text" ].casefold() in text.casefold()
582 if not matched:
583 raise fail( "vector-query-mismatch" , "No retrieved document matches both the expected source citation and expected content." )
584 return matched
585
586
587 def _validate_snapshot (value: Any) -> None :
588 fields = { "source_plan_digest" , "source_etag" , "source_definition" , "index" ,
589 "index_etag" , "index_digest" , "inventory_digest" , "vector" }
590 if not isinstance (value, dict ) or set (value) not in (fields, fields | { "ingestion_evidence_digest" }):
591 raise fail( "input-schema-invalid" , "Retain the complete generated verification snapshot." )
592 hashes = { "source_plan_digest" , "source_definition" , "index_digest" , "inventory_digest" }
593 for key in hashes | ({ "ingestion_evidence_digest" } & value.keys()):
594 if not isinstance (value[key], str ) or not search_reconcile. SHA256 .fullmatch(value[key]):
595 raise fail( "input-schema-invalid" , "Invalid generated snapshot integrity value; rerun planning." )
596 if not all (_text(value[key]) for key in ( "source_etag" , "index_etag" , "index" )):
597 raise fail( "input-schema-invalid" , "Snapshot identities and ETags are required." )
598 odata_name(value[ "index" ])
599 vector = value[ "vector" ]
600 if (
601 not isinstance (vector, dict )
602 or set (vector) != { "dimensions" , "profile" , "vectorizer" , "document_id_field" }
603 or type (vector.get( "dimensions" )) is not int or not 1 <= vector[ "dimensions" ] <= 3072
604 or not all (_text(vector[key], maximum = 128 ) for key in ( "profile" , "vectorizer" , "document_id_field" ))
605 ):
606 raise fail( "input-schema-invalid" , "Invalid generated vector snapshot; rerun planning." )
607
608
609 def execute (document: dict[ str , Any], * , token_provider: TokenProvider = azure_cli_token,
610 transport: Transport = http_request, storage_transport: Transport = http_request) -> dict[ str , Any]:
611 if not isinstance (document, dict ):
612 raise fail( "input-schema-invalid" , "Query execution needs an approved artifact." )
613 _json_valid(document)
614 require_allowed_fields(document, { "schema_version" , "plan" , "approval" }, label = "query artifact" )
615 plan = document.get( "plan" )
616 if not isinstance (plan, dict ):
617 raise fail( "input-schema-invalid" , "Query execution needs a planned artifact." )
618 reject_secrets(document)
619 require_allowed_fields(plan, { "operation" , "owner" , "cleanup_approved" , "request" , "snapshot" }, label = "vector query plan" )
620 fingerprint = digest(plan)
621 approval = document.get( "approval" )
622 if (
623 document.get( "schema_version" ) != "1.0" or not isinstance (approval, dict )
624 or approval.get( "confirmed" ) is not True
625 or approval != { "confirmed" : True , "fingerprint" : fingerprint}
626 ):
627 raise fail( "approval-required" , "Explicit approval of this unchanged billable query plan is required." )
628 if plan.get( "operation" ) != "verify-source-vector" or plan.get( "cleanup_approved" ) is not False :
629 raise fail( "input-schema-invalid" , "Only separately approved vector verification is supported." )
630 _validate_snapshot(plan.get( "snapshot" ))
631 request = plan.get( "request" )
632 source_plan = _request(request)
633 if plan.get( "owner" ) != source_plan[ "owner" ]:
634 raise fail( "source-artifact-invalid" , "Query owner must match the source artifact." )
635 snapshot, evidence = observe(request, token_provider = token_provider, transport = transport, storage_transport = storage_transport)
636 if snapshot != plan.get( "snapshot" ) or digest(source_plan) != snapshot[ "source_plan_digest" ]:
637 raise fail( "verification-drift" , "Evidence changed; rerun --plan, replace the artifact and discard consent." )
638 source = source_plan[ "source" ]
639 url = f " { source[ 'endpoint' ].rstrip( '/' ) } /indexes(' { odata_name(snapshot[ 'index' ]) } ')/docs/search.post.search?api-version= { source[ 'api_version' ] } "
640 queries = []
641 attempted = 0
642 try :
643 for hybrid in ([ False , True ] if request[ "mode" ] == "hybrid" else [ False ]):
644 token = token_provider( SEARCH_AUDIENCE )
645 attempted += 1
646 response = transport(
647 "POST" , url, token,
648 headers = { "Content-Type" : "application/json" },
649 body = json.dumps(query_body(request, snapshot[ "vector" ], hybrid = hybrid)).encode( "utf-8" ),
650 follow_redirects = False , max_response_bytes = 8 * 1024 * 1024 ,
651 timeout = 60 , response_deadline = time.monotonic() + 60 ,
652 )
653 if response.status != 200 or not _text(response.request_id):
654 raise fail( "vector-query-incomplete" , "Require HTTP 200 and observed Search request provenance." )
655 count = verify_response(response.body, request, snapshot[ "vector" ])
656 queries.append({ "mode" : "hybrid" if hybrid else "vector" , "request_id" : response.request_id,
657 "matched_documents" : count, "response_digest" : digest(response.body)})
658 after, _ = observe(request, token_provider = token_provider, transport = transport, storage_transport = storage_transport)
659 if after != snapshot:
660 raise fail( "verification-drift" , "Source, index or inventory changed across query execution." )
661 except HelperFailure as failure:
662 result = blocked_result(
663 HelperFailure(failure.code, failure.message, blocked_at = failure.blocked_at,
664 status = failure.http_status, request_id = failure.request_id),
665 outcome = "source-vector-verification" , fingerprint = fingerprint, owner = plan[ "owner" ],
666 )
667 result.update( queries_attempted = attempted, billing_possible = attempted > 0 , queries = queries, writes_performed = [])
668 return result
669 return {
670 "status" : "retrieval-verified" , "outcome" : "source-vector-verification" ,
671 "approved_plan" : { "confirmed" : True , "fingerprint" : fingerprint},
672 "verification" : { "configuration" : "verified" , "ingestion" : "verified" ,
673 "retrieval" : request[ "mode" ], "queries" : queries, "evidence" : evidence},
674 "writes_performed" : [], "cleanup" : "not-applicable" ,
675 "warnings" : [ "Observed index retrieval only: not KB synthesis, agent tool use, or a corpus-wide relevance guarantee. Reobserve before later use." ],
676 }
677
678
679 def main (argv: list[ str ] | None = None ) -> int :
680 try :
681 from .private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
682 except ImportError :
683 from private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
684 parser = argparse.ArgumentParser( description = "Plan or apply separately approved File/Blob vector verification." )
685 modes = parser.add_mutually_exclusive_group( required = True )
686 modes.add_argument( "--plan" , type = Path)
687 modes.add_argument( "--input" , type = Path)
688 add_execution_output_argument(parser)
689 args = parser.parse_args(argv)
690 try :
691 validate_execution_output_mode(args)
692 if args.plan:
693 result = plan_verification(_read_json( str (args.plan)))
694 emit_plan_result(result, args.execution_output)
695 return 0 if result[ "status" ] == "planned" else 2
696 else :
697 document, _, _ = load_approved_input(args.input)
698 result = execute(document)
699 except HelperFailure as failure:
700 result = blocked_result(failure, outcome = "source-vector-verification" , fingerprint = None )
701 emit_result(result)
702 return 0 if result[ "status" ] in { "planned" , "retrieval-verified" } else 2
703
704
705 if __name__ == "__main__" :
706 sys.exit(main())