Foundry Iq

helpers/file_upload.py

"""Synchronous File batches; one in-operation queue-429 retry, never unknown-outcome replay."""
from __future__ import annotations

import argparse
import copy
import hashlib
import re
import sys
import time
import uuid
from dataclasses import replace
from pathlib import Path

try:
    from . import (
        _bootstrap_io as private_io, cu_ingestion_auth, _file_inventory, _file_source_config, _search_read,
        _source_readback,
    )
    from ._common import (
        HelperFailure, ReadRecovery, RetryAfter, RetryAfterTiming, SEARCH_AUDIENCE, retry_after_timing,
        retry_after_not_before, valid_utc_timestamp, azure_cli_token, blocked_result, digest, emit_result,
        http_request, load_approved_input, reject_secrets, require_allowed_fields,
    )
    from ._progress import Progress, add_progress_argument, reporting
except ImportError:
    import _bootstrap_io as private_io
    import cu_ingestion_auth
    import _file_inventory
    import _file_source_config
    import _search_read
    import _source_readback
    from _common import (
        HelperFailure, ReadRecovery, RetryAfter, RetryAfterTiming, SEARCH_AUDIENCE, retry_after_timing,
        retry_after_not_before, valid_utc_timestamp, azure_cli_token, blocked_result, digest, emit_result,
        http_request, load_approved_input, reject_secrets, require_allowed_fields,
    )
    from _progress import Progress, add_progress_argument, reporting


RETAIN = " Retain the source and evidence; no creation replay, reset, deletion or new roles."
UPLOAD_TIMEOUT = 180
MAX_BACKOFF_RECORDS = 200


def _rejected(status):
    return type(status) is int and 400 <= status < 500 and status not in (408, 409)


def fail(code, message):
    return HelperFailure(code, message + RETAIN, blocked_at="file-upload-resume")


class Session:
    def __init__(self, directory, document=None, *, context_provider=cu_ingestion_auth.account_context):
        self.directory = private_io.validate_private_artifact_directory(str(directory))
        self.context_provider = context_provider
        self.records = {}
        self.network_failure = None
        self.active_attempt = None
        if document is not None:
            self._validate_document(document)
            if any(self.directory.iterdir()):
                raise fail("file-upload-journal-exists", "Use a new empty private receipt directory for original creation.")
            context = context_provider()
            cu_ingestion_auth.validate_context(context)
            self.seed = {"document": {key: copy.deepcopy(document[key]) for key in ("schema_version", "plan", "approval")},
                         "context": context, "backoff_version": "1.0"}
            self._write("run.json", self.seed)
        else:
            self._load()
            self.seed = self.records["run.json"]
            self._validate_document(self.seed["document"])
            cu_ingestion_auth.validate_context(self.seed["context"])
        self.document = self.seed["document"]
        self.plan = self.document["plan"]
        self.ingestion = self.plan["ingestion"]

    @staticmethod
    def _validate_document(document):

        if (not isinstance(document, dict) or document.get("schema_version") != "1.0"
                or not isinstance(document.get("plan"), dict)
                or "_computed_fingerprint" in document and document["_computed_fingerprint"] != digest(document.get("plan"))
                or not isinstance(document.get("approval"), dict) or document["approval"].get("confirmed") is not True
                or document.get("approval") != {"confirmed": True, "fingerprint": digest(document.get("plan"))}):
            raise fail("file-upload-approval-missing", "Original unchanged File creation approval is required.")
        require_allowed_fields(document, {"schema_version", "plan", "approval", "_computed_fingerprint"}, label="original File envelope")
        _file_source_config._validate_plan(document["plan"])
        if document["plan"]["source"]["action"] != "create":
            raise fail("file-upload-provenance-missing", "This continuation is only for original acknowledged creation, not generic reuse.")

    def _write(self, name, payload):
        value = {"schema_version": "1.0", "kind": "file-upload-journal", "payload": payload, "integrity": digest(payload)}
        reject_secrets(value)
        try:
            private_io.atomic_private_file(self.directory, name, value, max_bytes=private_io.MAX_BYTES)
        except HelperFailure as error:
            raise HelperFailure("file-upload-receipt-failed", "Private upload evidence persistence failed." + RETAIN,
                                blocked_at="local-persistence", warnings=error.warnings) from error
        self.records[name] = copy.deepcopy(payload)

    def _load(self):
        # Reuse the existing bounded, private, no-link evidence reader.
        try:
            from . import _private_json
        except ImportError:
            import _private_json
        try:
            paths = list(self.directory.iterdir())
        except OSError as error:
            raise fail("file-upload-evidence-unreadable", "Original upload evidence is inaccessible.") from error
        if len(paths) > 803 + MAX_BACKOFF_RECORDS or not {"run.json", "source-ack.json"}.issubset({p.name for p in paths}):
            raise fail("file-upload-provenance-missing", "Necessary original ACK and journal were not retained.")
        for path in paths:
            if (path.name not in {"run.json", "source-ack.json", "pending-request.json"}
                    and not re.fullmatch(r"(?:[0-9]{4}-(retry-)?(attempt|result)|backoff-[0-9]{4})\.json", path.name)):
                raise fail("file-upload-evidence-invalid", "Unexpected or incomplete journal entry; do not discard it to resume.")
            value = _private_json.read_private(path)
            if (set(value) != {"schema_version", "kind", "payload", "integrity"}
                    or value["schema_version"] != "1.0" or value["kind"] != "file-upload-journal"
                    or value["integrity"] != digest(value["payload"])):
                raise fail("file-upload-evidence-invalid", "Original private upload evidence changed.")
            self.records[path.name] = value["payload"]
        seed = self.records["run.json"]
        if (not isinstance(seed, dict) or set(seed) not in ({"document", "context"}, {"document", "context", "backoff_version"})
                or "backoff_version" in seed and seed["backoff_version"] != "1.0"):
            raise fail("file-upload-evidence-invalid", "Original approval/context evidence is incomplete.")

    def acknowledge(self, metadata):
        self._write("source-ack.json", {"original_plan_digest": digest(self.plan),
                                       "source_url": _search_read.resource_url(self.plan["source"]),
                                       "acknowledgement": metadata})

    def require_ack(self):
        value = self.records.get("source-ack.json")
        if (not isinstance(value, dict) or set(value) != {"original_plan_digest", "source_url", "acknowledgement"}
                or value["original_plan_digest"] != digest(self.plan)
                or value["source_url"] != _search_read.resource_url(self.plan["source"])):
            raise fail("file-upload-provenance-missing", "An original acknowledged conditional source creation must be retained.")
        ack = value["acknowledgement"]
        if (not isinstance(ack, dict) or set(ack) != {"status", "request_id", "etag_evidence"}
                or ack["status"] not in (200, 201) or not ack["request_id"]
                or not isinstance(ack["request_id"], str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", ack["request_id"])
                or not isinstance(ack["etag_evidence"], dict) or set(ack["etag_evidence"]) != {"body", "headers"}
                or not isinstance(ack["etag_evidence"]["headers"], list)):
            raise fail("file-upload-provenance-missing", "Original successful creation ACK/request ID is required, not GET ownership.")
        etag = _search_read.resolve_etag(ack["etag_evidence"], ack["request_id"])
        if not etag:
            raise fail("file-upload-provenance-missing", "The original creation ACK lacks version proof; GET cannot supply it.")
        return etag

    def states(self):
        states = []
        events = self.backoff_events()
        allowed = {"run.json", "source-ack.json", "pending-request.json"} | set(events)
        pending = self.records.get("pending-request.json")
        if "pending-request.json" in self.records and (
            not isinstance(pending, dict) or set(pending) != {"version", "plan_digest", "nonce", "method", "url_digest"}
            or pending["version"] != "1.0" or pending["plan_digest"] != digest(self.plan)
            or not isinstance(pending["nonce"], str) or not re.fullmatch(r"[0-9a-f]{32}", pending["nonce"])
            or pending["method"] not in ("GET", "POST", "PUT", "HEAD")
            or not isinstance(pending["url_digest"], str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", pending["url_digest"])
        ):
            raise fail("file-upload-evidence-invalid", "Pending request evidence is invalid.")
        for index, record in enumerate(self.ingestion["files"]):
            state = {"upload": "not_attempted", "request_ids": [], "ingested": False}
            for retry in (False, True):
                prefix = f"{index:04}-" + ("retry-" if retry else "")
                attempt_name, result_name = prefix + "attempt.json", prefix + "result.json"
                allowed.update((attempt_name, result_name))
                attempt, result = self.records.get(attempt_name), self.records.get(result_name)
                expected = self._attempt_record(index, retry=retry) if attempt is not None else None
                if attempt is not None and attempt != expected or result is not None and attempt is None:
                    raise fail("file-upload-evidence-invalid", "Upload attempt evidence does not bind the original source/corpus.")
                if attempt is not None:
                    state.update(upload="unverified", ingested=False)
                    state.pop("file_proof", None)
                    state.pop("ack_file_id", None)
                if result is not None:
                    if (not isinstance(result, dict) or not {"upload", "status", "request_id"} <= set(result)
                            or set(result) - {"upload", "status", "request_id", "file_proof", "file_id", "throttle_event"}
                            or not isinstance(result["upload"], str)
                            or result["upload"] not in {"accepted", "rejected", "unverified"}
                            or result["upload"] == "accepted" and result["status"] not in (200, 201)
                            or result["upload"] == "rejected" and not _rejected(result["status"])
                            or result["status"] is not None and (type(result["status"]) is not int or not 100 <= result["status"] <= 599)
                            or result["request_id"] is not None and ReadRecovery.safe_id(result["request_id"]) != result["request_id"]):
                        raise fail("file-upload-evidence-invalid", "Upload result is not an original supported ACK or failure.")
                    linked = [event for event in events.values() if event["attempt"] == attempt_name]
                    if "throttle_event" in result:
                        if (result["status"] != 429 or len(linked) != 1
                                or result["throttle_event"] != digest(linked[0])
                                or result["request_id"] != linked[0]["request_id"]):
                            raise fail("file-upload-evidence-invalid", "Upload 429 timing does not bind its original attempt/result.")
                    elif result["status"] == 429 and self.seed.get("backoff_version") is not None:
                        raise fail("file-upload-evidence-invalid", "Required original upload 429 timing evidence is missing.")
                    if "file_id" in result:
                        if result["upload"] != "accepted" or not _file_id(result["file_id"]):
                            raise fail("file-upload-evidence-invalid", "Original upload ACK file identity is invalid.")
                        state["ack_file_id"] = result["file_id"]
                    proof = result.get("file_proof")
                    if proof is not None:
                        if (result["upload"] != "accepted" or not isinstance(proof, dict)
                                or set(proof) != {"file_id", "record_digest"} or not _file_id(proof["file_id"])
                                or proof["record_digest"] != digest(record)
                                or result.get("file_id", proof["file_id"]) != proof["file_id"]):
                            raise fail("file-upload-evidence-invalid", "Persisted File completion proof does not match the approved record.")
                        state.update(file_proof=proof, ingested=True)
                    state.update(upload=result["upload"], status=result["status"])
                    if result["request_id"]:
                        state["request_ids"].append(result["request_id"])
            states.append(state)
        if set(self.records) - allowed:
            raise fail("file-upload-evidence-invalid", "Upload journal contains records outside the approved corpus.")
        return states

    def _attempt_record(self, index, *, retry=False):
        record = {
            "original_plan_digest": digest(self.plan), "record_digest": digest(self.ingestion["files"][index]),
            "creation_ack_digest": digest(self.records["source-ack.json"]),
        }
        if retry:
            rejection = self.records.get(f"{index:04}-result.json")
            if not isinstance(rejection, dict) or rejection.get("status") != 429 or rejection.get("upload") != "rejected":
                raise fail("file-upload-replay-forbidden", "A retry attempt requires the same operation's retained File upload 429.")
            record["rejected_result_digest"] = digest(rejection)
        return record

    def attempt(self, index, *, retry=False):
        prefix = f"{index:04}-" + ("retry-" if retry else "")
        self._write(prefix + "attempt.json", self._attempt_record(index, retry=retry))
        self.active_attempt = prefix + "attempt.json"

    def result(self, index, upload, status, request_id, *, retry=False, proof=None, file_id=None):
        prefix = f"{index:04}-" + ("retry-" if retry else "")
        event = {}
        if status == 429:
            linked = [value for value in self.backoff_events().values() if value["attempt"] == prefix + "attempt.json"]
            if len(linked) != 1:
                raise fail("file-upload-evidence-invalid", "Original upload 429 timing was not retained.")
            event["throttle_event"] = digest(linked[0])
        self._write(prefix + "result.json", {"upload": upload, "status": status,
                    "request_id": ReadRecovery.safe_id(request_id) if request_id else None,
                    **({"file_id": file_id} if file_id is not None else {}),
                    **({"file_proof": proof} if proof is not None else {}), **event})
        self.active_attempt = None

    def backoff_events(self):
        events = {key: self.records[key] for key in sorted(self.records) if key.startswith("backoff-")}
        if len(events) > MAX_BACKOFF_RECORDS:
            raise fail("file-upload-evidence-invalid", "Too many retained backoff records.")
        plan_digest = digest(self.plan)
        previous = None
        for index, (name, event) in enumerate(events.items()):
            if (name != f"backoff-{index:04}.json" or not isinstance(event, dict)
                    or set(event) != {"version", "plan_digest", "previous", "method", "url_digest",
                                      "attempt", "attempt_digest", "request_id", "metadata", "timing", "origin"}
                    or event["version"] != "1.0" or event["plan_digest"] != plan_digest
                    or event["previous"] != previous or event["method"] not in ("GET", "POST", "PUT", "HEAD")
                    or not isinstance(event["url_digest"], str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", event["url_digest"])
                    or event["request_id"] is not None and ReadRecovery.safe_id(event["request_id"]) != event["request_id"]
                    or event["origin"] not in ("response-headers", "native-http-error", "typed-metadata", "unavailable")):
                raise fail("file-upload-evidence-invalid", "Original backoff provenance is invalid or incomplete.")
            if event["attempt"] is not None:
                attempt = self.records.get(event["attempt"]) if isinstance(event["attempt"], str) else None
                if (attempt is None or not re.fullmatch(r"[0-9]{4}-(retry-)?attempt\.json", event["attempt"])
                        or event["attempt_digest"] != digest(attempt) or event["method"] != "POST"
                        or event["url_digest"] != digest(_file_inventory._list_url(self.ingestion))):
                    raise fail("file-upload-evidence-invalid", "Backoff does not bind the original upload attempt.")
            elif event["attempt_digest"] is not None:
                raise fail("file-upload-evidence-invalid", "Backoff attempt provenance is inconsistent.")
            metadata, timing = event["metadata"], event["timing"]
            if (not isinstance(metadata, dict) or set(metadata) != {"kind", "value"}
                    or metadata["kind"] not in ("seconds", "date", "date-rfc850", "missing", "invalid", "overlong")
                    or not valid_utc_timestamp(metadata["value"])
                    or metadata["kind"] == "seconds" and (type(metadata["value"]) is not int or not 0 <= metadata["value"] <= 30)
                    or metadata["kind"] in ("missing", "invalid", "overlong") and metadata["value"] != 0
                    or not isinstance(timing, dict) or set(timing) != {"received_at_utc", "not_before_utc", "server_delay_seconds"}):
                raise fail("file-upload-evidence-invalid", "Retained Retry-After metadata is invalid.")
            received, deadline = timing["received_at_utc"], timing["not_before_utc"]
            seconds = timing["server_delay_seconds"]
            if (received is not None and not valid_utc_timestamp(received)
                    or deadline is not None and (received is None or not valid_utc_timestamp(deadline))
                    or seconds is not None and (type(seconds) is not int or not 0 <= seconds < 315537897600
                                               or metadata["kind"] not in ("seconds", "overlong"))
                    or metadata["kind"] == "seconds" and seconds not in (None, metadata["value"])):
                raise fail("file-upload-evidence-invalid", "Retained backoff UTC timing is invalid.")
            if deadline is not None:
                if (event["origin"] == "unavailable"
                        or event["origin"] == "typed-metadata" and metadata["kind"] not in ("seconds", "date", "date-rfc850")
                        or metadata["kind"] == "overlong" and (seconds is None or seconds <= 30 or deadline != received + seconds)
                        or metadata["kind"] != "overlong"
                        and deadline != retry_after_not_before(RetryAfter(**metadata), received)):
                    raise fail("file-upload-evidence-invalid", "Retained backoff deadline contradicts original metadata.")
            previous = digest(event)
        return events

    def backoff_status(self):
        events = self.backoff_events()
        attempts = {name: value for name, value in self.records.items()
                    if re.fullmatch(r"[0-9]{4}-(retry-)?(attempt|result)\.json", name)}
        if any(not isinstance(value, dict) for value in attempts.values()):
            raise fail("file-upload-evidence-invalid", "Attempt/result evidence is malformed.")
        # Legacy 429 or an interrupted receipt cannot prove what restriction was received.
        unknown = "pending-request.json" in self.records or any(
            name.endswith("result.json") and value.get("status") == 429 and "throttle_event" not in value
            or name.endswith("attempt.json") and name.replace("attempt.json", "result.json") not in self.records
            and name != self.active_attempt
            for name, value in attempts.items()
        )
        now = time.time()
        if unknown or any(event["timing"]["not_before_utc"] is None for event in events.values()):
            return {"status": "unresolved", "reason": "Original response timing is unavailable; no deadline is inferred."}
        if not events:
            return {"status": "clear"}
        if (not valid_utc_timestamp(now)
                or any(now < event["timing"]["received_at_utc"] for event in events.values())):
            return {"status": "unresolved", "reason": "UTC clock is invalid or precedes retained response receipt."}
        deadline = max(event["timing"]["not_before_utc"] for event in events.values())
        return {"status": "waiting" if now < deadline else "elapsed", "not_before_utc": deadline,
                "seconds_remaining": max(0, deadline - now)}

    def require_backoff(self):
        status = self.backoff_status()
        if status["status"] in ("waiting", "unresolved"):
            error = fail("file-upload-backoff-" + status["status"],
                         "Server backoff still applies; fresh approval does not waive it. "
                         + ("Wait until the retained UTC not-before time." if status["status"] == "waiting"
                            else "Original timing evidence or scoped operator recovery is required."))
            error.partial = "source-ack.json" in self.records
            error.file_batch = _summary(self.ingestion, self.states(), self)
            raise error

    def transport(self, raw):
        if getattr(raw, "_file_journal_owner", None) is self:
            return raw

        def invoke(method, url, token, **kwargs):
            if self.network_failure is not None:
                raise self.network_failure
            self.require_backoff()
            if len(self.backoff_events()) >= MAX_BACKOFF_RECORDS:
                raise fail("file-upload-backoff-limit", "Backoff journal limit reached; no more requests.")
            pending = {"version": "1.0", "plan_digest": digest(self.plan), "nonce": uuid.uuid4().hex,
                       "method": method, "url_digest": digest(url)}
            self._write("pending-request.json", pending)
            try:
                response = raw(method, url, token, **kwargs)
            except HelperFailure as error:
                if error.http_status == 429:
                    self.record_backoff(method, url, error, None)
                self.finish_request(error)
                raise
            if response.status == 429:
                error = HelperFailure("azure-http-error", "Azure request failed with HTTP 429.",
                                      blocked_at="execution", status=429, request_id=response.request_id,
                                      retry_after=response.retry_after, recovery_deadline=response.recovery_deadline)
                self.record_backoff(method, url, error, response.headers)
            try:
                self.finish_request()
            except HelperFailure as persistence:
                if method in ("PUT", "POST") and response.status in (200, 201):
                    return replace(response, ack_failure=persistence)
                raise
            return response

        invoke._file_journal_owner = self
        return invoke

    def finish_request(self, original=None):
        try:
            path = self.directory / "pending-request.json"
            expected = self.records["pending-request.json"]
            retained = private_io.read_json(path)
            if not isinstance(retained, dict) or retained.get("payload") != expected or retained.get("integrity") != digest(expected):
                raise fail("file-upload-evidence-invalid", "Pending request receipt changed.")
            path.unlink()
            self.records.pop("pending-request.json")
        except (OSError, HelperFailure) as cleanup:
            error = original or HelperFailure("file-upload-receipt-failed", "Pending request evidence could not be finalized.",
                                              blocked_at="local-persistence")
            error.blocked_at = "local-persistence"
            error.warnings.append("Pending request evidence remains unresolved; no further requests.")
            self.network_failure = error
            raise error from cleanup

    def record_backoff(self, method, url, error, headers):
        metadata = error.retry_after
        origin = "response-headers" if headers is not None else "native-http-error"
        timing = retry_after_timing(headers) if headers is not None else error.retry_after_timing
        if timing is None:
            received = time.time()
            received = received if valid_utc_timestamp(received) else None
            known = isinstance(metadata, RetryAfter) and metadata.kind in ("seconds", "date", "date-rfc850")
            timing = RetryAfterTiming(received, retry_after_not_before(metadata, received) if known else None,
                                      metadata.value if known and metadata.kind == "seconds" else None)
            origin = "typed-metadata" if known else "unavailable"
        if not isinstance(metadata, RetryAfter):
            metadata, origin = RetryAfter("missing"), "unavailable"
            timing = RetryAfterTiming(timing.received_at_utc, None)
        events = self.backoff_events()
        attempt = self.active_attempt if method == "POST" and url == _file_inventory._list_url(self.ingestion) else None
        event = {"version": "1.0", "plan_digest": digest(self.plan),
                 "previous": digest(next(reversed(events.values()))) if events else None,
                 "method": method, "url_digest": digest(url), "attempt": attempt,
                 "attempt_digest": digest(self.records[attempt]) if attempt else None,
                 "request_id": ReadRecovery.safe_id(error.request_id) if error.request_id else None,
                 "metadata": metadata._asdict(), "timing": timing._asdict(), "origin": origin}
        try:
            self._write(f"backoff-{len(events):04}.json", event)
        except HelperFailure as persistence:
            error.blocked_at = "local-persistence"
            error.warnings.extend(persistence.warnings)
            error.warnings.append("Original HTTP 429 retained; backoff receipt persistence failed. No further requests.")
            self.network_failure = error
            raise error from persistence


def _file_id(value):
    return isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", value) is not None


def _proof(plan, record, item):
    if isinstance(item, dict) and _file_id(item.get("fileId")) and _file_inventory._matches(item, plan, record):
        return {"file_id": item["fileId"], "record_digest": digest(record)}
    return None


def _observe(plan, states, files):
    names = {record["path"] for record in plan["files"]}
    if any(item.get("fileName") not in names for item in files):
        raise fail("server-inventory-conflict", "File inventory contains an unapproved identity.")
    observed = {}
    for index, record in enumerate(plan["files"]):
        matches = [item for item in files if item.get("fileName") == record["path"]]
        if len(matches) > 1 or matches and not _file_inventory._matches({**matches[0], "errorMessage": None}, plan, record):
            raise fail("file-record-conflict", "File inventory has duplicate or conflicting approved markers.")
        state = states[index]
        state["verification"] = {"accepted": "pending", "rejected": "failed"}.get(state["upload"], state["upload"])
        if matches:
            item = matches[0]
            proof = _proof(plan, record, item)
            if (proof and state.get("file_proof") and proof != state["file_proof"]
                    or state.get("ack_file_id") and item.get("fileId") is not None
                    and item["fileId"] != state["ack_file_id"]):
                raise fail("file-record-conflict", "Current File identity differs from the persisted upload ACK.")
            state["verification"] = ("failed" if state["upload"] == "rejected" and state.get("status") != 429
                                     or item.get("errorMessage") is not None else
                                     "confirmed" if proof else "unverified")
            state["ingested"] = state["verification"] == "confirmed"
            observed[index] = item
    return observed


def _summary(plan, states, session):
    counts = {name: 0 for name in ("accepted", "confirmed", "failed", "pending", "unverified", "not_attempted")}
    files = []
    for record, state in zip(plan["files"], states):
        counts["accepted"] += state["upload"] == "accepted"
        verification = state.get("verification", {"accepted": "pending", "rejected": "failed"}.get(state["upload"], state["upload"]))
        counts[verification] += 1
        files.append({"fileName": record["path"], "sha256": record["sha256"], "upload": state["upload"],
                      "verification": verification, "http_status": state.get("status"),
                      "request_ids": state["request_ids"][:3]})
    ingested = sum(bool(state.get("ingested")) for state in states)
    return {"counts": counts, "files": files, "ingested": ingested,
            "readiness": "ingested" if ingested == len(states) and counts["confirmed"] == len(states) else "partial",
            "retrieval": "unverified",
            **({"backoff": session.backoff_status()} if session else {}),
            **({"upload_retry": {
                "status": "blocked", "code": "file-upload-retry-safety-unproven",
                "reason": "Unknown upload outcomes cannot be replayed; only a received File upload 429 permits one bounded in-operation retry.",
            }} if counts["unverified"] else {}),
            "resume": ("Plan newly approved upload-only continuation for never-attempted files; uncertain attempts are never replayed."
                       if session else "Original durable ACK/attempt evidence was not retained; uploads cannot safely resume from GET alone."),
            "subset_handoff": "Report the proved ingested subset and remaining outcomes; subset isolation and KB retrieval are separate. KB changes need separate approval."}


def _progress(progress, plan, states):
    summary = _summary(plan, states, None)
    counts = summary["counts"]
    progress.update("file-upload", uploads_acknowledged=counts["accepted"], files_verified=counts["confirmed"],
                    files_ingested=summary["ingested"],
                    files_reused=sum(state["upload"] == "not_attempted" and state.get("verification") == "confirmed" for state in states),
                    files_failed=counts["failed"], files_unverified=counts["unverified"],
                    files_not_attempted=counts["not_attempted"], files_pending=counts["pending"])


def run_batch(document, *, token_provider, transport, progress, allow_new_uploads=False, session=None,
              eligible=None, source_check=None, allow_upload_retry=True):
    plan, fingerprint = document["plan"], document["_computed_fingerprint"]
    root, records = _file_inventory._validate_plan(plan)
    states = session.states() if session else [{"upload": "not_attempted", "request_ids": []} for _ in records]
    if session:
        session.require_backoff()
        transport = session.transport(transport)
    created, reused, request_ids, warnings = [], [], [], []
    primary = None
    stopped = False
    retry_used = False
    readback_failed = False
    token = token_provider(SEARCH_AUDIENCE)
    url = _file_inventory._list_url(plan)
    recovery = ReadRecovery(on_wait=progress.waiting)
    progress.update("file-inventory")
    try:
        before, ids = _file_inventory._list_files(url, token, transport=transport, recovery=recovery)
        request_ids.extend(ids)
        if session is None and _file_inventory.inventory_digest(before) != plan["expected_server_inventory_digest"]:
            raise fail("server-inventory-drift", "Approved initial file inventory changed.")
        observed = _observe(plan, states, before)
        if not allow_new_uploads and len(observed) != len(records):
            raise fail("reused-source-upload-forbidden", "Generic reuse cannot authorize new uploads.")
    except HelperFailure as failure:
        failure.file_batch = _summary(plan, states, session)
        failure.file_batch["request_ids"] = recovery.request_ids
        raise
    warnings.extend(recovery.warnings)
    eligible = set(range(len(records))) if eligible is None else set(eligible)
    after = None
    for index, record in enumerate(records):
        _progress(progress, plan, states)
        state = states[index]
        if index in observed:
            reused.append({"fileId": observed[index].get("fileId"), "fileName": record["path"], "sha256": record["sha256"]})
            continue
        if index not in eligible or state["upload"] != "not_attempted":
            continue
        try:
            path = _file_inventory._resolve_file(root, record)
            content = path.read_bytes()
            if (len(content) != record["size"] or "sha256:" + hashlib.sha256(content).hexdigest() != record["sha256"]
                    or path.stat().st_mtime_ns != record["mtime_ns"]):
                raise fail("inventory-drift", "Approved corpus changed before upload.")
        except OSError:
            primary = primary or fail("inventory-unreadable", "Approved corpus became inaccessible.")
            warnings.append("Batch stopped: approved corpus became inaccessible.")
            stopped = True
            break
        except HelperFailure as failure:
            primary = primary or failure
            warnings.append(f"Batch stopped ({failure.code}); no further uploads were attempted.")
            stopped = True
            break
        body, boundary = _file_inventory._multipart(plan, record, content, fingerprint)
        for attempt in range(2):
            try:
                if attempt:
                    _file_inventory._resolve_file(root, record)
                if session:
                    session.attempt(index, retry=bool(attempt))
            except HelperFailure as failure:
                primary = primary or failure
                warnings.append(f"Upload attempt evidence/drift check failed ({failure.code}); no further requests.")
                stopped = True
                break
            state.update(upload="unverified", verification="unverified", status=None)
            _progress(progress, plan, states)
            response = None
            transport_failed = False
            try:
                with progress.processing_file(index + 1, len(records), attempt + 1):
                    response = transport("POST", url, token, body=body,
                                         headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
                                         follow_redirects=False, timeout=UPLOAD_TIMEOUT,
                                         response_deadline=time.monotonic() + UPLOAD_TIMEOUT,
                                         max_response_bytes=1024 * 1024)
            except HelperFailure as caught:
                transport_failed = True
                failure = caught
                failure_status, failure_id = failure.http_status, failure.request_id
            else:
                failure_status, failure_id = response.status, response.request_id
                failure = HelperFailure("upload-failed", f"Upload returned HTTP {response.status}.",
                                        blocked_at="execution", status=response.status, request_id=response.request_id,
                                        retry_after=response.retry_after, recovery_deadline=response.recovery_deadline)
            if failure_id:
                safe_id = ReadRecovery.safe_id(failure_id)
                failure.request_id = safe_id
                state["request_ids"].append(safe_id)
                request_ids.append(safe_id)
            state["status"] = failure_status
            proof = None
            if failure_status in (200, 201):
                state.update(upload="accepted", verification="pending")
                created.append({"fileName": record["path"], "sha256": record["sha256"]})
                if response is not None:
                    proof = _proof(plan, record, response.body)
                    if isinstance(response.body, dict) and _file_id(response.body.get("fileId")):
                        state["ack_file_id"] = response.body["fileId"]
            elif _rejected(failure_status):
                state.update(upload="rejected", verification="failed")
            try:
                if response is not None and response.ack_failure is not None:
                    raise response.ack_failure
                if session:
                    session.result(index, state["upload"], failure_status, failure_id, retry=bool(attempt),
                                   proof=proof, file_id=state.get("ack_file_id"))
            except HelperFailure as persistence:
                primary = primary or (failure if failure_status not in (200, 201) else persistence)
                primary.warnings.extend(persistence.warnings)
                warnings.append("Upload journal persistence failed; no further requests were issued.")
                stopped = True
                break
            if failure_status in (200, 201) and not transport_failed:
                if proof:
                    state.update(file_proof=proof, ingested=True, verification="confirmed")
                elif isinstance(response.body, dict) and (
                    response.body.get("errorMessage") is not None
                    or "fileName" in response.body and not _file_inventory._matches(response.body, plan, record)
                ):
                    primary = primary or fail("upload-metadata-unverified", "Upload ACK metadata conflicts with the approved file.")
                    stopped = True
                _progress(progress, plan, states)
                break
            primary = primary or failure
            if failure_status == 415:
                break
            _progress(progress, plan, states)
            if failure_status == 429 and not retry_used and allow_upload_retry and failure.blocked_at != "local-persistence":
                retry_used = True
                recovery = ReadRecovery(on_wait=progress.waiting)
                try:
                    recovery.delay(failure)
                    if source_check is not None:
                        source_check(recovery, token)
                    inventory, _ = _file_inventory._list_files(url, token, transport=transport, recovery=recovery)
                    seen = _observe(plan, states, inventory)
                    if index in seen:
                        if state["verification"] != "confirmed":
                            raise fail("file-upload-retry-conflict", "Existing file does not prove the approved completed upload.")
                        created.append({"fileName": record["path"], "sha256": record["sha256"]})
                        break
                    _file_inventory._resolve_file(root, record)
                    if time.monotonic() >= recovery.deadline:
                        raise fail("read-recovery-budget-exhausted", "Retry preflight exceeded its complete read budget.")
                    warnings.append("File upload HTTP 429: one same-operation retry after bounded backoff; no source recreation.")
                except HelperFailure as read_failure:
                    warnings.append(f"File retry stopped ({read_failure.code}; HTTP {read_failure.http_status}).")
                    warnings.extend(read_failure.warnings)
                    stopped = True
                finally:
                    request_ids.extend(recovery.request_ids)
                    warnings.extend(recovery.diagnostics())
                if not stopped:
                    continue
            else:
                stopped = True
                warnings.append(f"Batch stopped ({failure.code}; HTTP {failure_status}); no further uploads.")
                if failure.blocked_at != "local-persistence" and (
                    failure_status in (408, 409) or failure_status is None or failure_status >= 500
                ):
                    recovery = ReadRecovery(on_wait=progress.waiting)
                    try:
                        after, _ = _file_inventory._list_files(url, token, transport=transport, recovery=recovery)
                        _observe(plan, states, after)
                        if state["verification"] == "confirmed":
                            created.append({"fileName": record["path"], "sha256": record["sha256"]})
                    except HelperFailure as read_failure:
                        warnings.append(f"File readback stopped ({read_failure.code}; HTTP {read_failure.http_status}).")
                        warnings.extend(read_failure.warnings)
                    request_ids.extend(recovery.request_ids)
                    warnings.extend(recovery.diagnostics())
            break
        if stopped:
            break
    if session:
        session.active_attempt = None
    _progress(progress, plan, states)
    progress.update("file-readback")
    if all(state["upload"] == "rejected" and state["verification"] != "confirmed" for state in states):
        stopped = True
    if after is None and not stopped:
        recovery = ReadRecovery(on_wait=progress.waiting)
        try:
            after, ids = _file_inventory._list_files(url, token, transport=transport, recovery=recovery)
            _observe(plan, states, after)
        except HelperFailure as failure:
            primary = primary or failure
            readback_failed = True
            warnings.append(f"File readback stopped ({failure.code}; HTTP {failure.http_status}).")
        request_ids.extend(recovery.request_ids)
        warnings.extend(recovery.diagnostics())
    batch = _summary(plan, states, session)
    batch["request_ids"] = request_ids[:804]
    batch["upload_retry_used"] = retry_used
    progress.update("file-readback", files_verified=batch["counts"]["confirmed"], files_failed=batch["counts"]["failed"],
                    files_pending=batch["counts"]["pending"], files_unverified=batch["counts"]["unverified"],
                    files_not_attempted=batch["counts"]["not_attempted"], files_ingested=batch["ingested"])
    if batch["counts"]["confirmed"] != len(records) or readback_failed:
        failure = primary or fail("readback-mismatch", "Some approved files remain pending, failed or unverified.")
        uncertain = [{"action": "upload-unverified", "type": "knowledge-source-file",
                      "fileName": record["path"], "sha256": record["sha256"]}
                     for record, state in zip(records, states)
                     if state["upload"] == "unverified" and state["verification"] != "confirmed"]
        failure.writes = created + uncertain + failure.writes
        failure.resources_remaining = _file_inventory._remaining_files(created) + uncertain + failure.resources_remaining
        failure.partial = bool(created) or any(s["upload"] == "unverified" for s in states) or (
            failure.partial and not _rejected(failure.http_status))
        failure.warnings.extend(warnings)
        failure.file_batch = batch
        raise failure
    verified = [{"fileId": item.get("fileId"), "fileName": record["path"], "sha256": record["sha256"], "size": record["size"]}
                for record in records for item in after or before if item.get("fileName") == record["path"]]
    return {"status": "completed", "outcome": plan.get("outcome", "file-knowledge-source-ingestion"),
            "approved_plan": {"fingerprint": fingerprint, "confirmed": True},
            "resources": {"created": created, "reused": reused, "updated": [], "skipped": []},
            "api_contracts": [{"operation": "upload-file", "version": _file_inventory.API_VERSION, "preview": True}],
            "data_movement": {"boundary": {"local_root_digest": digest(str(root))}, "result": "Approved direct File uploads"},
            "auth": {"mode": "entra-user", "principals": []}, "rbac": plan.get("rbac", {"assignments": []}),
            "network": plan.get("network", {"posture": "preserved", "evidence": None}),
            "verification": {"readback": verified, "request_ids": request_ids, "server_inventory_digest": _file_inventory.inventory_digest(after or before),
                             "idempotency": "Only a received File upload 429 permits one bounded same-operation retry; unknown outcomes are never replayed."},
            "warnings": warnings, "file_batch": batch,
            "ownership": {"run_owned": created, "reused_not_owned": reused, "owner": plan["owner"]},
            "cleanup": {"status": "not-requested", "separate_confirmation_required": True}}


def _source_check(session, token, transport, recovery):

    plan = session.plan
    _file_inventory._validate_plan(session.ingestion)
    etag = session.require_ack()
    current, _ = _search_read._get(_search_read.resource_url(plan["source"]), token,
                                       transport=_source_readback.guard_readback_transport(plan, transport), recovery=recovery)
    if (current is None or current.get("@odata.etag") != etag
            or not _search_read.definitions_match(plan["source"]["desired"], current)):
        raise fail("file-upload-source-drift", "Current source does not match original acknowledged identity/version/definition.")
    return current


def _verify(session, token_provider, transport, context_provider):
    try:
        from . import file_cu_mi
    except ImportError:
        import file_cu_mi
    session.require_ack()
    session.states()
    session.require_backoff()
    transport = session.transport(transport)
    context = context_provider()
    cu_ingestion_auth.validate_context(context)
    if context != session.seed["context"]:
        raise fail("file-upload-context-drift", "Original tenant/subscription/principal changed.")
    plan = session.plan
    if "content_understanding" in plan:
        state, _ = _file_source_config.read_content_understanding(plan["content_understanding"], token_provider=token_provider, transport=transport)
        if not _file_source_config._cu_states_match(state, plan["cu_resource_state"]):
            raise fail("file-upload-auth-drift", "Original CU account/auth/network state changed.")
        if plan["source"].get("ai_services_managed_identity"):
            binding, _ = file_cu_mi.read_binding(plan["content_understanding"], plan["source"]["endpoint"],
                                               token_provider=token_provider, transport=transport)
            if binding != plan["cu_identity_state"]:
                raise fail("file-upload-auth-drift", "Original Search MI/CU role/network binding changed.")
    return _source_check(session, token_provider(SEARCH_AUDIENCE), transport, ReadRecovery())


def plan_resume(request, *, token_provider=azure_cli_token, transport=http_request,
                context_provider=cu_ingestion_auth.account_context):
    if not isinstance(request, dict) or not isinstance(request.get("receipt_directory"), str):
        raise fail("file-upload-input-invalid", "Use an object with the original private receipt directory.")
    require_allowed_fields(request, {"schema_version", "receipt_directory"}, label="upload resume request")
    if request.get("schema_version") != "1.0":
        raise fail("file-upload-input-invalid", "Use the supported resume request schema.")
    session = Session(request.get("receipt_directory"))
    transport = session.transport(transport)
    _verify(session, token_provider, transport, context_provider)
    states = session.states()
    inventory, _ = _file_inventory._list_files(_file_inventory._list_url(session.ingestion), token_provider(SEARCH_AUDIENCE),
                                         transport=transport, recovery=ReadRecovery())
    observed = _observe(session.ingestion, states, inventory)
    eligible = [i for i, state in enumerate(states) if state["upload"] == "not_attempted" and i not in observed]
    plan = {"operation": "resume-file-uploads", "version": "1.0", "receipt_directory": str(session.directory),
            "original_plan_digest": digest(session.plan), "creation_ack_digest": digest(session.records["source-ack.json"]),
            "journal_digest": digest(session.records), "eligible": eligible, "owner": session.plan["owner"], "cleanup_approved": False}
    return {"status": "planned", "execution_input": {"schema_version": "1.0", "plan": plan,
                                                   "approval": {"confirmed": False, "fingerprint": digest(plan)}},
            "approval_summary": {"execution_required": bool(eligible), "mutation_approval_required": bool(eligible), "uploads": len(eligible)},
            "file_batch": _summary(session.ingestion, states, session), "writes_performed": [],
            "next_step": "Approve only never-attempted uploads. No eligible files: retain this observation; do not replay uncertain files."}


@reporting("file-source")
def execute(document, *, token_provider=azure_cli_token, transport=http_request,
            context_provider=cu_ingestion_auth.account_context, progress=None):
    if not isinstance(document, dict) or not isinstance(document.get("plan"), dict):
        raise fail("file-upload-input-invalid", "Use the newly approved upload-only envelope.")
    require_allowed_fields(document, {"schema_version", "plan", "approval", "_computed_fingerprint"}, label="upload resume envelope")
    plan = document.get("plan")
    reject_secrets(document)
    require_allowed_fields(plan, {"operation", "version", "receipt_directory", "original_plan_digest", "creation_ack_digest",
                                 "journal_digest", "eligible", "owner", "cleanup_approved"}, label="upload resume plan")
    if (document.get("schema_version") != "1.0" or plan.get("operation") != "resume-file-uploads" or plan.get("version") != "1.0"
            or set(plan) != {"operation", "version", "receipt_directory", "original_plan_digest", "creation_ack_digest",
                             "journal_digest", "eligible", "owner", "cleanup_approved"}
            or not isinstance(document.get("approval"), dict) or document["approval"].get("confirmed") is not True
            or plan.get("cleanup_approved") is not False or document.get("approval") != {"confirmed": True, "fingerprint": digest(plan)}
            or document.get("_computed_fingerprint") != digest(plan)):
        raise fail("file-upload-approval-missing", "A newly approved unchanged upload-only plan is required.")
    session = Session(plan.get("receipt_directory"))
    transport = session.transport(transport)
    if (plan["original_plan_digest"] != digest(session.plan) or plan["journal_digest"] != digest(session.records)
            or plan["creation_ack_digest"] != digest(session.records["source-ack.json"]) or plan["owner"] != session.plan["owner"]
            or not isinstance(plan["eligible"], list) or any(type(i) is not int or not 0 <= i < len(session.ingestion["files"]) for i in plan["eligible"])
            or len(set(plan["eligible"])) != len(plan["eligible"])):
        raise fail("file-upload-plan-drift", "Original approval/ACK/journal changed; refresh the upload-only plan.")
    states = session.states()
    if any(states[i]["upload"] != "not_attempted" for i in plan["eligible"]):
        raise fail("file-upload-replay-forbidden", "Attempted files cannot be submitted again, even after an empty inventory.")
    progress.update("source-reconciliation")
    try:
        _verify(session, token_provider, transport, context_provider)
        result = run_batch({"plan": session.ingestion, "_computed_fingerprint": digest(session.plan)}, token_provider=token_provider,
                           transport=transport, progress=progress, allow_new_uploads=True, session=session, eligible=plan["eligible"],
                           source_check=lambda recovery, token: _source_check(session, token, transport, recovery))
    except HelperFailure as failure:
        if failure.file_batch is None:
            failure.file_batch = _summary(session.ingestion, states, session)
        failure.partial = True
        failure.resources_remaining.insert(0, {"type": "knowledge-source", "name": session.plan["source"]["name"]})
        raise
    result["approved_plan"] = document["approval"]
    result["original_run"] = {"plan_digest": digest(session.plan), "creation_ack_digest": plan["creation_ack_digest"],
                              "source_retained": session.plan["source"]["name"]}
    return result


def main(argv=None):
    try:
        from .private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
    except ImportError:
        from private_artifacts import add_execution_output_argument, emit_plan_result, validate_execution_output_mode
    parser = argparse.ArgumentParser()
    modes = parser.add_mutually_exclusive_group(required=True)
    modes.add_argument("--plan", type=Path)
    modes.add_argument("--input", type=Path)
    add_execution_output_argument(parser)
    add_progress_argument(parser)
    args = parser.parse_args(argv)
    fingerprint = None
    try:
        validate_execution_output_mode(args)
        if args.plan:
            emit_plan_result(plan_resume(private_io.read_json(args.plan)), args.execution_output)
            return 0
        document, _, fingerprint = load_approved_input(args.input)
        document["_computed_fingerprint"] = fingerprint
        result = execute(document, progress=Progress("file-source", enabled=args.progress))
    except HelperFailure as failure:
        result = blocked_result(failure, outcome="resume-file-uploads", fingerprint=fingerprint)
        result["safe_next_decision"] = "Retain the source. Missing/changed original evidence blocks upload continuation, not retention; no creation replay or cleanup workaround."
        emit_result(result)
        return 3 if result["status"] == "partial" else 2
    emit_result(result)
    return 0


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