Setting the file. One moment.
File Cu Canary · Foundry Iq · microsoft/azure-skills · Skills Docs
Back to the top of the page File Cu Canary · references
class BoundedTransport
— line 287
This file
Number 10.46
Position 46 of 77
Type Python
Size 33 KB
Lines 540 helpers/ file_cu_canary.py
Python · 540 lines · 33 KB
13
from
email
import
policy
14 from email.parser import BytesParser
15 from pathlib import Path
16 from urllib.parse import urlencode
17
18 try :
19 from . import _bootstrap_io, file_source, file_ingest, file_cu_mi, search_reconcile
20 from ._common import (HelperFailure, MANAGEMENT_AUDIENCE , SEARCH_AUDIENCE , azure_cli_token,
21 blocked_result, digest, emit_result, http_request, load_approved_input,
22 require_allowed_fields, reject_secrets)
23 except ImportError :
24 import _bootstrap_io, file_source, file_ingest, file_cu_mi, search_reconcile
25 from _common import (HelperFailure, MANAGEMENT_AUDIENCE , SEARCH_AUDIENCE , azure_cli_token,
26 blocked_result, digest, emit_result, http_request, load_approved_input,
27 require_allowed_fields, reject_secrets)
28
29 PDF_NAME = "cu-mi-probe.pdf"
30 MARKER = re.compile( r "CUOCR [ 2-9 ] {8} " )
31 FIELD = re.compile( r " [ A-Za-z ][ A-Za-z0-9_ ] {0,127} " )
32 # Pinned implementation, not deployed proof; see references/file-cu-canary.md.
33 # Never populate this from caller attestations or an arbitrary readback field.
34 OCR_CONTENT_MAPPINGS = {
35 "2026-08-01-preview" : {
36 "field" : "snippet" , "parent_field" : "snippet_parent_id" , "path_field" : "metadata_storage_path" ,
37 "authority" : {
38 "repository" : "AzureSearch" , "commit" : "2e241af8939a0891836b78278df113dab31a5964" ,
39 "contract" : "file-standard-cu-index-v1" ,
40 },
41 },
42 }
43 DISCLOSURE = (
44 "One synthetic image-only PDF is uploaded to Search and processed by billable CU OCR; "
45 "CU may create an analyzer, content may cross regions, and Search retains generated data. "
46 "No embeddings, KB/chat or direct CU probes. Existing resources/roles/network/local-auth stay unchanged. "
47 "Client time/attempt caps are not a monetary cap or cancellation of remote processing. "
48 "Cleanup needs separate approval; retain the private source/file/index inventory."
49 )
50 GLYPHS = {
51 "C" : ( "01111" , "10000" , "10000" , "10000" , "10000" , "10000" , "01111" ),
52 "U" : ( "10001" , "10001" , "10001" , "10001" , "10001" , "10001" , "01110" ),
53 "O" : ( "01110" , "10001" , "10001" , "10001" , "10001" , "10001" , "01110" ),
54 "R" : ( "11110" , "10001" , "10001" , "11110" , "10100" , "10010" , "10001" ),
55 "2" : ( "01110" , "10001" , "00001" , "00010" , "00100" , "01000" , "11111" ),
56 "3" : ( "11110" , "00001" , "00001" , "01110" , "00001" , "00001" , "11110" ),
57 "4" : ( "00010" , "00110" , "01010" , "10010" , "11111" , "00010" , "00010" ),
58 "5" : ( "11111" , "10000" , "10000" , "11110" , "00001" , "00001" , "11110" ),
59 "6" : ( "01110" , "10000" , "10000" , "11110" , "10001" , "10001" , "01110" ),
60 "7" : ( "11111" , "00001" , "00010" , "00100" , "01000" , "01000" , "01000" ),
61 "8" : ( "01110" , "10001" , "10001" , "01110" , "10001" , "10001" , "01110" ),
62 "9" : ( "01110" , "10001" , "10001" , "01111" , "00001" , "00001" , "01110" ),
63 }
64
65
66 def fail (code, message, * , partial = False , request_id = None , status = None ):
67 return HelperFailure(code, message, blocked_at = "file-cu-canary" , partial = partial, request_id = request_id, status = status)
68
69
70 def pdf_bytes (marker):
71 if not isinstance (marker, str ) or MARKER .fullmatch(marker) is None :
72 raise fail( "canary-marker-invalid" , "Use CUOCR followed by eight digits 2–9." )
73 scale, margin = 12 , 24
74 width, height = len (marker) * 6 * scale + 2 * margin, 7 * scale + 2 * margin
75 image = bytearray ( b " \xff " * width * height)
76 for n, letter in enumerate (marker):
77 for y, row in enumerate ( GLYPHS [letter]):
78 for x, pixel in enumerate (row):
79 if pixel == "1" :
80 for yy in range (scale):
81 start = (margin + y * scale + yy) * width + margin + (n * 6 + x) * scale
82 image[start:start + scale] = b " \x00 " * scale
83 compressed = zlib.compress( bytes (image), 9 )
84 draw = b "q 600 0 0 84 20 20 cm /Image0 Do Q \n "
85 objects = [
86 b "<< /Type /Catalog /Pages 2 0 R >>" ,
87 b "<< /Type /Pages /Kids [3 0 R] /Count 1 >>" ,
88 b "<< /Type /Page /Parent 2 0 R /MediaBox [0 0 640 124] /Resources << /XObject << /Image0 4 0 R >> >> /Contents 5 0 R >>" ,
89 f "<< /Type /XObject /Subtype /Image /Width { width } /Height { height } /ColorSpace /DeviceGray /BitsPerComponent 8 /Filter /FlateDecode /Length { len (compressed) } >> \n stream \n " .encode()
90 + compressed + b " \n endstream" ,
91 f "<< /Length { len (draw) } >> \n stream \n " .encode() + draw + b "endstream" ,
92 ]
93 out = bytearray ( b "%PDF-1.4 \n % \xe2\xe3\xcf\xd3\n " )
94 offsets = [ 0 ]
95 for n, obj in enumerate (objects, 1 ):
96 offsets.append( len (out))
97 out.extend( f " { n } 0 obj \n " .encode() + obj + b " \n endobj \n " )
98 xref = len (out)
99 out.extend( b "xref \n 0 6 \n 0000000000 65535 f \n " )
100 for offset in offsets[ 1 :]:
101 out.extend( f " { offset :010d} 00000 n \n " .encode())
102 out.extend( f "trailer \n << /Size 6 /Root 1 0 R >> \n startxref \n{ xref }\n %%EOF \n " .encode())
103 if len (out) > 65536 or marker.encode() in out:
104 raise fail( "canary-pdf-invalid" , "Probe must be bounded image-only bytes without a text marker." )
105 return bytes (out)
106
107
108 def prepare (directory, marker = None ):
109 directory = _bootstrap_io.private_directory(directory)
110 marker = marker or "CUOCR" + "" .join(secrets.choice( "23456789" ) for _ in range ( 8 ))
111 data = pdf_bytes(marker)
112 _bootstrap_io.private_bytes(directory, PDF_NAME , data)
113 manifest = { "file" : PDF_NAME , "marker" : marker, "sha256" : hashlib.sha256(data).hexdigest(),
114 "bytes" : len (data), "pages" : 1 , "text_layer" : False }
115 path = _bootstrap_io.private_file(directory, "cu-mi-probe-manifest.json" , manifest)
116 return { "status" : "prepared" , "manifest" : str (path), "writes_performed" : [],
117 "azure_operations" : 0 , "note" : "Local synthetic artifacts only; no live approval or resource selection." }
118
119
120 def _bounds (value):
121 if not isinstance (value, dict ):
122 raise fail( "canary-bounds-missing" , "Select explicit timeout, HTTP request, polling attempt and interval caps." )
123 ranges = { "timeout_seconds" : ( 30 , 600 ), "max_requests" : ( 16 , 100 ),
124 "max_poll_attempts" : ( 1 , 10 ), "poll_interval_seconds" : ( 1 , 30 )}
125 require_allowed_fields(value, set (ranges), label = "canary bounds" )
126 if any ( type (value.get(k)) is not int or not low <= value[k] <= high for k, (low, high) in ranges.items()):
127 raise fail( "canary-bounds-invalid" , "Use timeout 30–600s, HTTP requests 16–100, polls 1–10 and interval 1–30s." )
128
129
130 def _ocr_mapping (api_version, field):
131 mapping = OCR_CONTENT_MAPPINGS .get(api_version)
132 if mapping is None :
133 raise fail( "canary-ocr-mapping-unverified" ,
134 "No pinned File Standard OCR-to-index field mapping is verified for this API. "
135 "Do not create/upload; verify the first-party mapping and update the supported contract before planning." )
136 if field != mapping[ "field" ]:
137 raise fail( "canary-field-unverified" , "The selected field does not match the pinned File Standard OCR-content mapping." )
138 return copy.deepcopy(mapping)
139
140
141 def _verified_file (source_result, ingestion):
142 files = source_result[ "verification" ][ "readback" ].get( "files" )
143 expected = ingestion[ "files" ][ 0 ]
144 if (
145 not isinstance (files, list ) or len (files) != 1 or not isinstance (files[ 0 ], dict )
146 or not isinstance (files[ 0 ].get( "fileId" ), str ) or not files[ 0 ][ "fileId" ].strip()
147 or files[ 0 ].get( "fileName" ) != expected[ "path" ]
148 or files[ 0 ].get( "sha256" ) != expected[ "sha256" ] or files[ 0 ].get( "size" ) != expected[ "size" ]
149 ):
150 raise fail( "canary-file-identity-unverified" , "Require the uploaded synthetic file's exact ID, path, hash and size readback." )
151 return files[ 0 ]
152
153
154 def _verify_index_fields (index, mapping):
155 fields = index.get( "fields" )
156 if not isinstance (fields, list ):
157 raise fail( "canary-field-unverified" , "Generated index fields must match the pinned File OCR mapping." )
158 for name in (mapping[ "field" ], mapping[ "parent_field" ], mapping[ "path_field" ]):
159 selected = [field for field in fields if isinstance (field, dict ) and field.get( "name" ) == name]
160 if len (selected) != 1 or selected[ 0 ].get( "type" ) != "Edm.String" or selected[ 0 ].get( "retrievable" ) is not True :
161 raise fail( "canary-field-unverified" , "Require unique retrievable string fields for pinned OCR content and file identity." )
162 if name == mapping[ "field" ] and (
163 selected[ 0 ].get( "searchable" ) is not True
164 or any (selected[ 0 ].get(flag) is not False for flag in ( "filterable" , "sortable" , "facetable" ))
165 ):
166 raise fail( "canary-field-unverified" , "The canonical snippet field does not match the pinned searchable OCR-content schema." )
167
168
169 def _contains_marker (value, marker):
170 if isinstance (value, str ):
171 return marker in re.sub( r " [ ^A-Z0-9 ] " , "" , value.upper())
172 if isinstance (value, dict ):
173 return any (_contains_marker(k, marker) or _contains_marker(v, marker) for k, v in value.items())
174 if isinstance (value, ( list , tuple )):
175 return any (_contains_marker(item, marker) for item in value)
176 return False
177
178
179 def _check_non_image_channels (plan, marker):
180 ingestion = plan[ "ingestion" ]
181 # Produce the actual multipart envelope with only the image bytes omitted.
182 try :
183 body, boundary = file_ingest._multipart(ingestion, ingestion[ "files" ][ 0 ], b "" , digest(plan))
184 message = BytesParser( policy = policy.default).parsebytes(
185 f "Content-Type: multipart/form-data; boundary= { boundary }\r\n\r\n " .encode( "ascii" ) + body
186 )
187 except ( ValueError , UnicodeError , TypeError ):
188 raise fail( "canary-upload-envelope-unverified" , "The non-image upload envelope could not be produced safely." ) from None
189 parts = list (message.iter_parts())
190 if (
191 message.defects or len (parts) != 2 or any (part.defects for part in parts)
192 or [part.get_param( "name" , header = "content-disposition" ) for part in parts] != [ "metadata" , "content" ]
193 or parts[ 0 ].get_content_type() != "application/json" or parts[ 1 ].get_payload( decode = True ) != b ""
194 ):
195 raise fail( "canary-upload-envelope-unverified" , "The produced non-image upload envelope is not the supported multipart contract." )
196 try :
197 metadata = json.loads(parts[ 0 ].get_payload( decode = True ))
198 channels = [plan[ "source" ][ "desired" ], file_ingest._list_url(ingestion),
199 body.decode( "utf-8" ), metadata, [ list (part.items()) for part in parts]]
200 except ( ValueError , UnicodeError , TypeError ):
201 raise fail( "canary-upload-envelope-unverified" , "The produced non-image upload envelope could not be inspected." ) from None
202 if _contains_marker(channels, marker):
203 raise fail( "canary-marker-in-metadata" , "The OCR marker must not occur in source/upload names, owner, metadata or other non-image upload channels." )
204
205
206 def _file_contract (plan, marker):
207 expected = pdf_bytes(marker)
208 if not isinstance (plan, dict ) or not isinstance (plan.get( "ingestion" ), dict ):
209 raise fail( "canary-file-contract-invalid" , "Retain the complete File plan." )
210 root = file_ingest.resolve_local_root(plan[ "ingestion" ].get( "local_root" ))
211 actual = file_ingest._resolve_inventory_path(root, PDF_NAME )
212 try :
213 if actual.stat().st_size != len (expected):
214 raise fail( "canary-content-drift" , "Probe size differs from the bounded synthetic image-only PDF." )
215 except OSError :
216 raise fail( "canary-content-drift" , "Selected synthetic PDF is unreadable." ) from None
217 file_source._validate_plan(plan)
218 source, ingestion = plan[ "source" ], plan[ "ingestion" ]
219 settings = source[ "desired" ][ "fileParameters" ][ "ingestionParameters" ]
220 if (
221 plan.get( "file_cu_plan_version" ) != "1.2" or source.get( "action" ) != "create"
222 or source[ "api_version" ] != "2026-08-01-preview"
223 or settings.get( "contentExtractionMode" ) != "standard" or "embedding" in plan
224 or settings.get( "embeddingModel" ) is not None or settings.get( "chatCompletionModel" ) is not None
225 or settings.get( "disableImageVerbalization" ) is not True
226 or len (ingestion[ "files" ]) != 1 or ingestion[ "files" ][ 0 ][ "path" ] != PDF_NAME
227 ):
228 raise fail( "canary-file-contract-invalid" , "Canary requires fresh conditional File Standard MI creation, one synthetic PDF, no embedding/chat and unchanged August API." )
229 _check_non_image_channels(plan, marker)
230 records = file_ingest.snapshot_inventory(root, [ PDF_NAME ], service_tier = ingestion[ "service_tier" ])
231 if records != ingestion[ "files" ] or records[ 0 ][ "sha256" ] != "sha256:" + hashlib.sha256(expected).hexdigest():
232 raise fail( "canary-content-drift" , "Synthetic inventory changed; regenerate the concrete plan before approval." )
233
234
235 def plan_canary (request, * , token_provider = azure_cli_token, transport = http_request, now = time.time):
236 if not isinstance (request, dict ):
237 raise fail( "canary-input-invalid" , "Canary intent must be an object." )
238 require_allowed_fields(request, { "schema_version" , "file_request" , "marker" , "content_field" , "bounds" , "receipt_directory" },
239 label = "File CU canary intent" )
240 reject_secrets(request)
241 _bounds(request.get( "bounds" ))
242 if request.get( "schema_version" ) != "1.0" or not isinstance (request.get( "content_field" ), str ) or not FIELD .fullmatch(request[ "content_field" ]):
243 raise fail( "canary-input-invalid" , "Use schema 1.0 and an exact simple generated content field name." )
244 pdf_bytes(request.get( "marker" ))
245 directory = _bootstrap_io.private_directory(request.get( "receipt_directory" ))
246 fr = request.get( "file_request" )
247 if not isinstance (fr, dict ):
248 raise fail( "canary-input-invalid" , "Supply an explicit existing Search/CU File request; no resource defaults or provisioning." )
249 # Reject a key mode before even read-only resource discovery.
250 cu = fr.get( "content_understanding" )
251 if not isinstance (cu, dict ) or cu.get( "auth" , "system-assigned" ) != "system-assigned" :
252 raise fail( "canary-auth-invalid" , "This canary only tests File managed identity; no keys or fallback." )
253 if fr.get( "paths" ) != [ PDF_NAME ] or fr.get( "extraction_mode" ) != "standard" or fr.get( "vectorization" ) != "none" :
254 raise fail( "canary-file-contract-invalid" , "Select exactly the generated probe PDF, Standard extraction and no vectors." )
255 version = fr.get( "api_version" , file_ingest. API_VERSION )
256 file_ingest.validate_api_version(version)
257 mapping = _ocr_mapping(version, request[ "content_field" ])
258 root = file_ingest.resolve_local_root(fr.get( "local_root" ))
259 selected = file_ingest._resolve_inventory_path(root, PDF_NAME )
260 try :
261 if selected.stat().st_size != len (pdf_bytes(request[ "marker" ])):
262 raise fail( "canary-content-drift" , "Selected PDF is not the bounded synthetic probe." )
263 except OSError :
264 raise fail( "canary-content-drift" , "Selected probe is unreadable." ) from None
265 result = file_source.plan_source(fr, token_provider = token_provider, transport = transport)
266 fp = result[ "execution_input" ][ "plan" ]
267 _file_contract(fp, request[ "marker" ])
268 created = int (now())
269 plan = {
270 "operation" : "validate-file-cu-mi" , "version" : "1.1" , "file_plan" : fp, "ocr_mapping" : mapping,
271 "marker" : request[ "marker" ], "content_field" : request[ "content_field" ],
272 "bounds" : copy.deepcopy(request[ "bounds" ]), "receipt_directory" : str (directory),
273 "created_at" : created, "expires_at" : created + 900 , "disclosure" : DISCLOSURE ,
274 }
275 fingerprint = digest(plan)
276 return {
277 "status" : "planned" , "plan_fingerprint" : fingerprint,
278 "execution_input" : { "schema_version" : "1.0" , "plan" : plan,
279 "approval" : { "confirmed" : False , "fingerprint" : fingerprint}},
280 "approval_summary" : { "source" : result[ "approval_summary" ], "bounds" : plan[ "bounds" ],
281 "expires_at" : plan[ "expires_at" ], "disclosure" : DISCLOSURE ,
282 "expected_outcome" : "Indexed OCR marker without client keys; principal attribution remains separate." },
283 "writes_performed" : [],
284 }
285
286
287 class BoundedTransport :
288 def __init__ (self, plan, raw, clock, record):
289 self .plan, self .raw, self .clock, self .record = plan, raw, clock, record
290 self .deadline = clock() + plan[ "bounds" ][ "timeout_seconds" ]
291 self .calls, self .puts, self .uploads = 0 , 0 , 0
292 self .upload_ack_failure = None
293 self .index_url = None
294 fp = plan[ "file_plan" ]
295 self .source_url = search_reconcile.resource_url(fp[ "source" ])
296 self .files_prefix = self .source_url.split( "?" )[ 0 ] + "/files"
297 cu = fp[ "content_understanding" ]
298 mi = cu[ "managed_identity" ]
299 self .arm_urls = {
300 f " { MANAGEMENT_AUDIENCE }{ cu[ 'resource_id' ] } ?api-version=2024-10-01" ,
301 f " { MANAGEMENT_AUDIENCE }{ mi[ 'search_resource_id' ] } ?api-version= { file_cu_mi. SEARCH_API } " ,
302 f " { MANAGEMENT_AUDIENCE }{ mi[ 'role_assignment_id' ] } ?api-version= { file_cu_mi. ROLE_API } " ,
303 }
304
305 def __call__ (self, method, url, token, ** kwargs):
306 if self .upload_ack_failure is not None :
307 raise self .upload_ack_failure
308 remaining = self .deadline - self .clock()
309 if remaining <= 0 or self .calls >= self .plan[ "bounds" ][ "max_requests" ]:
310 raise fail( "canary-bound-reached" , "Client request/deadline cap reached; remote processing may continue." , partial = bool ( self .puts))
311 file_url = url.startswith( self .files_prefix + "?" )
312 allowed = method == "GET" and (
313 url in self .arm_urls or url == self .source_url or file_url
314 or self .index_url is not None and (url == self .index_url or url.startswith( self .index_url.split( "?" )[ 0 ] + "/docs?" ))
315 )
316 if method == "PUT" and url == self .source_url and self .puts == 0 and kwargs.get( "headers" , {}).get( "If-None-Match" ) == "*" :
317 allowed = True
318 self .puts += 1
319 elif method == "POST" and file_url and self .uploads == 0 :
320 allowed = True
321 self .uploads += 1
322 if not allowed:
323 raise fail( "canary-operation-forbidden" , "Only exact resource GETs, one conditional source PUT and one selected upload are approved." , partial = bool ( self .puts))
324 self .calls += 1
325 kwargs.update( follow_redirects = False , max_response_bytes = min ( 524288 , kwargs.get( "max_response_bytes" , 524288 )),
326 response_deadline = min ( self .deadline, kwargs.get( "response_deadline" , self .deadline)),
327 timeout = max ( 0.01 , min ( float (kwargs.get( "timeout" , 60 )), remaining)))
328 try :
329 response = self .raw(method, url, token, ** kwargs)
330 except HelperFailure as failure:
331 failure.recovery_deadline = min ( self .deadline, failure.recovery_deadline or self .deadline)
332 raise
333 if method == "POST" and response.status in ( 200 , 201 , 202 ):
334 try :
335 self .record( "upload-http-ack" , { "status" : response.status, "request_id" : response.request_id,
336 "ingestion" : "not yet verified" })
337 except HelperFailure as error:
338 # Return the real ACK: persistence is not an ambiguous upload.
339 self .upload_ack_failure = error
340 return search_reconcile.HttpResult(
341 response.status, response.body, response.headers, response.etag_values,
342 min ( self .deadline, response.recovery_deadline or self .deadline),
343 self .upload_ack_failure,
344 )
345
346
347 def execute (document, * , token_provider = azure_cli_token, transport = http_request, now = time.time,
348 clock = time.monotonic, sleep = time.sleep):
349 if not isinstance (document, dict ):
350 raise fail( "canary-input-invalid" , "Canary execution requires an object envelope." )
351 require_allowed_fields(document, { "schema_version" , "plan" , "approval" , "_computed_fingerprint" }, label = "canary envelope" )
352 if document.get( "schema_version" ) != "1.0" :
353 raise fail( "canary-input-invalid" , "Use canary envelope schema_version 1.0." )
354 plan = document.get( "plan" )
355 if not isinstance (plan, dict ):
356 raise fail( "canary-input-invalid" , "Use the unchanged canary execution envelope." )
357 require_allowed_fields(plan, { "operation" , "version" , "file_plan" , "marker" , "content_field" , "ocr_mapping" , "bounds" ,
358 "receipt_directory" , "created_at" , "expires_at" , "disclosure" }, label = "canary plan" )
359 fingerprint = digest(plan)
360 if document.get( "approval" ) != { "confirmed" : True , "fingerprint" : fingerprint} or document.get( "_computed_fingerprint" ) != fingerprint:
361 raise fail( "approval-missing" , "Exact canary scope, synthetic content/cost and bounded verification need unchanged fingerprinted approval." )
362 if (
363 plan.get( "operation" ) != "validate-file-cu-mi" or plan.get( "version" ) != "1.1"
364 or plan.get( "disclosure" ) != DISCLOSURE or not isinstance (plan.get( "content_field" ), str )
365 or not FIELD .fullmatch(plan[ "content_field" ])
366 or type (plan.get( "created_at" )) is not int or type (plan.get( "expires_at" )) is not int
367 or plan[ "expires_at" ] - plan[ "created_at" ] != 900 or not plan[ "created_at" ] <= now() < plan[ "expires_at" ]
368 ):
369 raise fail( "canary-plan-stale" , "Canary plan is invalid or outside its 15-minute approval window; refresh discovery." )
370 _bounds(plan.get( "bounds" ))
371 _file_contract(plan[ "file_plan" ], plan[ "marker" ])
372 if plan.get( "ocr_mapping" ) != _ocr_mapping(plan[ "file_plan" ][ "source" ][ "api_version" ], plan[ "content_field" ]):
373 raise fail( "canary-ocr-mapping-stale" , "The approved OCR mapping differs from the pinned helper contract; a fresh plan and approval are required." )
374 directory = _bootstrap_io.private_directory(plan[ "receipt_directory" ])
375 prefix = "cu-mi-" + fingerprint.split( ":" )[ 1 ][: 16 ]
376 refs = []
377
378 def record (stage, value):
379 path = _bootstrap_io.private_file(directory, prefix + "-" + stage + ".json" , {
380 "fingerprint" : fingerprint, "stage" : stage, "value" : value,
381 })
382 refs.append( str (path))
383
384 record( "started" , { "approved_input" : {k: v for k, v in document.items() if k != "_computed_fingerprint" },
385 "state" : "No Azure mutation yet; later completion is not atomic." })
386 bounded = BoundedTransport(plan, transport, clock, record)
387 fp = plan[ "file_plan" ]
388 child_digest = digest(fp)
389 source_result = None
390 def source_ack ( ** evidence):
391 response = evidence[ "response" ]
392 record( "source-http-ack" , { "status" : response.status, "request_id" : response.request_id,
393 "etag_evidence" : search_reconcile.response_etags(response)})
394
395 try :
396 source_result = file_source.execute(
397 { "schema_version" : "1.0" , "plan" : fp, "_computed_fingerprint" : child_digest,
398 "approval" : { "confirmed" : True , "fingerprint" : child_digest}},
399 token_provider = token_provider, transport = bounded, mi_on_created = source_ack,
400 allow_upload_retry = False ,
401 )
402 if bounded.upload_ack_failure is not None :
403 raise bounded.upload_ack_failure
404 record( "file-completed" , source_result)
405 original_file = _verified_file(source_result, fp[ "ingestion" ])
406 mapping = plan[ "ocr_mapping" ]
407 token = token_provider( SEARCH_AUDIENCE )
408 current, _ = search_reconcile.read_resource(bounded.source_url, token, transport = bounded)
409 file_source.verify_content_understanding_readback(fp[ "content_understanding" ], current)
410 created = current.get( "fileParameters" , {}).get( "createdResources" )
411 if not isinstance (created, dict ) or set (created) != { "index" } or not isinstance (created[ "index" ], str ) or not re.fullmatch( r " [ a-z0-9 ][ a-z0-9_- ] {1,127} " , created[ "index" ]):
412 raise fail( "canary-index-unverified" , "Fresh File readback must identify exactly one generated index." )
413 retained = source_result[ "verification" ][ "readback" ][ "source" ]
414 if current.get( "@odata.etag" ) != retained[ "etag" ] or not search_reconcile.definitions_match(fp[ "source" ][ "desired" ], current):
415 raise fail( "canary-source-drift" , "Source version changed before indexed OCR verification." )
416 version = fp[ "source" ][ "api_version" ]
417 bounded.index_url = f " { fp[ 'source' ][ 'endpoint' ].rstrip( '/' ) } /indexes(' { created[ 'index' ] } ')?api-version= { version } "
418 index, _ = search_reconcile.read_resource(bounded.index_url, token, transport = bounded)
419 if not isinstance (index, dict ) or index.get( "name" ) != created[ "index" ] or not isinstance (index.get( "@odata.etag" ), str ) or not index[ "@odata.etag" ]:
420 raise fail( "canary-index-unverified" , "Require the generated index's matching name and fresh ETag." )
421 _verify_index_fields(index, mapping)
422 query = bounded.index_url.split( "?" )[ 0 ] + "/docs?" + urlencode({
423 "api-version" : version, "search" : "*" ,
424 "$select" : "," .join((mapping[ "field" ], mapping[ "parent_field" ], mapping[ "path_field" ])), "$top" : 3 ,
425 })
426 matched = False
427 for attempt in range (plan[ "bounds" ][ "max_poll_attempts" ]):
428 response = bounded( "GET" , query, token)
429 if response.status != 200 or not isinstance (response.body, dict ) or not isinstance (response.body.get( "value" ), list ):
430 raise fail( "canary-index-query-failed" , "Generated-index query did not return bounded document rows." ,
431 request_id = response.request_id, status = response.status)
432 rows = response.body[ "value" ]
433 if len (rows) > 3 :
434 raise fail( "canary-index-query-failed" , "Generated-index response exceeded the approved row cap." )
435 if any (
436 not isinstance (row, dict ) or not isinstance (row.get(mapping[ "field" ]), str )
437 or row.get(mapping[ "parent_field" ]) != original_file[ "fileId" ]
438 or row.get(mapping[ "path_field" ]) != original_file[ "fileName" ]
439 for row in rows
440 ):
441 raise fail( "canary-document-binding-unverified" , "Indexed rows must contain canonical OCR text and the uploaded file's exact parent ID/path; chunk IDs or metadata alone are not proof." )
442 matched = any (_contains_marker(row[mapping[ "field" ]], plan[ "marker" ]) for row in rows)
443 if matched:
444 break
445 if attempt + 1 < plan[ "bounds" ][ "max_poll_attempts" ]:
446 sleep( min (plan[ "bounds" ][ "poll_interval_seconds" ], max ( 0 , bounded.deadline - clock())))
447 if not matched:
448 raise fail( "canary-ocr-unverified" , "Upload/source creation is not extraction proof: indexed OCR marker was not observed within the caps." )
449 after, _ = search_reconcile.read_resource(bounded.source_url, token, transport = bounded)
450 if not isinstance (after, dict ) or after.get( "@odata.etag" ) != current.get( "@odata.etag" ) or after.get( "fileParameters" , {}).get( "createdResources" ) != created:
451 raise fail( "canary-source-drift" , "Generated source/index binding changed during OCR verification." )
452 final_index, _ = search_reconcile.read_resource(bounded.index_url, token, transport = bounded)
453 if final_index != index:
454 raise fail( "canary-index-drift" , "Generated index changed during OCR verification." )
455 binding, _ = file_cu_mi.read_binding(fp[ "content_understanding" ], fp[ "source" ][ "endpoint" ],
456 token_provider = token_provider, transport = bounded)
457 if binding != fp[ "cu_identity_state" ]:
458 raise fail( "canary-identity-drift" , "Search/CU identity-role binding changed during validation." )
459 result = { ** copy.deepcopy(source_result), "outcome" : "validate-file-cu-mi" ,
460 "approved_plan" : { "fingerprint" : fingerprint, "confirmed" : True },
461 "status" : "completed" , "verdict" : "keyless-functional-pass" , "indexed_ocr_marker" : True ,
462 "principal_attribution" : "unverified" , "backend_rollout" : "unverified" ,
463 "identity_evidence" : fp[ "cu_identity_state" ], "source_result" : source_result,
464 "ocr_evidence" : { "mapping" : mapping, "file" : original_file},
465 "index" : created[ "index" ], "request_count" : bounded.calls,
466 "cleanup" : { "status" : "separate-plan-and-approval-required" ,
467 "instructions" : "Retain inventory; separately approved guarded source cleanup only." },
468 "warnings" : [ "Functional evidence is for this run only, not an atomic deployment or release claim." ,
469 "No backend CU principal/telemetry was observed." ]}
470 except HelperFailure as error:
471 if bounded.upload_ack_failure is not None and error is not bounded.upload_ack_failure:
472 checkpoint_error = bounded.upload_ack_failure
473 checkpoint_error.writes = error.writes
474 checkpoint_error.resources_remaining = error.resources_remaining
475 checkpoint_error.resources_reused = error.resources_reused
476 checkpoint_error.resources_unverified = error.resources_unverified
477 checkpoint_error.warnings.extend(error.warnings)
478 checkpoint_error.partial = error.partial
479 error = checkpoint_error
480 if source_result is not None :
481 error.partial = True
482 error.writes = [
483 { ** resource, "action" : "created" , "type" : resource.get( "type" , "knowledge-source-file" )}
484 for resource in source_result[ "resources" ][ "created" ]
485 ] + error.writes
486 error.resources_remaining = source_result[ "ownership" ][ "run_owned" ] + error.resources_remaining
487 result = blocked_result(error, outcome = "validate-file-cu-mi" , fingerprint = fingerprint, owner = fp[ "owner" ])
488 result.update( verdict = "unverified" , indexed_ocr_marker = False , principal_attribution = "unverified" ,
489 backend_rollout = "unverified" , request_count = bounded.calls)
490 result.setdefault( "warnings" , []).append( "Remote processing/costs may continue; do not replay mutations or infer ownership from an ambiguous create." )
491 if bounded.upload_ack_failure is not None :
492 result[ "warnings" ].append( "Upload HTTP ACK could not be retained; validation stopped. Only listed receipt_refs are confirmed; do not replay the upload." )
493 try :
494 record( "result" , result)
495 except HelperFailure:
496 if result[ "status" ] == "completed" :
497 error = fail( "canary-receipt-failed" , "Validation finished but its final private receipt could not be persisted." , partial = True )
498 error.writes = [
499 { ** resource, "action" : "created" , "type" : resource.get( "type" , "knowledge-source-file" )}
500 for resource in source_result[ "resources" ][ "created" ]
501 ]
502 error.resources_remaining = source_result[ "ownership" ][ "run_owned" ]
503 result = { ** blocked_result(error, outcome = "validate-file-cu-mi" , fingerprint = fingerprint, owner = fp[ "owner" ]),
504 "indexed_ocr_marker" : True , "principal_attribution" : "unverified" , "backend_rollout" : "unverified" ,
505 "source_result" : source_result, "request_count" : bounded.calls}
506 result[ "verdict" ] = "unverified"
507 result.setdefault( "warnings" , []).append( "Final private receipt persistence failed; retain this ownership handoff and existing checkpoints. Do not replay." )
508 return { ** result, "receipt_refs" : refs}
509
510
511 def main (argv = None ):
512 parser = argparse.ArgumentParser()
513 modes = parser.add_mutually_exclusive_group( required = True )
514 modes.add_argument( "--prepare" , metavar = "EXISTING_PRIVATE_DIRECTORY" )
515 modes.add_argument( "--plan" , type = Path)
516 modes.add_argument( "--input" , type = Path)
517 args = parser.parse_args(argv)
518 try :
519 if args.prepare:
520 result = prepare(args.prepare)
521 elif args.plan:
522 result = plan_canary(_bootstrap_io.read_json(args.plan))
523 directory = result[ "execution_input" ][ "plan" ][ "receipt_directory" ]
524 path = _bootstrap_io.private_file(directory, "cu-mi-plan-" + result[ "plan_fingerprint" ].split( ":" )[ 1 ][: 16 ] + ".json" ,
525 result[ "execution_input" ])
526 result = {k: v for k, v in result.items() if k != "execution_input" }
527 result[ "execution_input_ref" ] = str (path)
528 else :
529 document, _, fingerprint = load_approved_input(args.input)
530 document[ "_computed_fingerprint" ] = fingerprint
531 result = execute(document)
532 emit_result(result)
533 return 0 if result[ "status" ] in ( "prepared" , "planned" , "completed" ) else 3 if result[ "status" ] == "partial" else 2
534 except HelperFailure as error:
535 emit_result(blocked_result(error, outcome = "validate-file-cu-mi" , fingerprint = None , owner = None ))
536 return 3 if error.partial else 2
537
538
539 if __name__ == "__main__" :
540 sys.exit(main())