Foundry Iq

helpers/_progress.py

"""Bounded, content-free observations; never a service execution controller."""
from __future__ import annotations

from functools import wraps
from contextlib import contextmanager
import copy
from datetime import datetime
import json
import math
import os
import sys
import time
import threading

try:
    from ._common import HelperFailure
except ImportError:
    from _common import HelperFailure


STAGES = {
    "file-source": ("validation", "source-reconciliation", "file-inventory", "file-upload", "file-readback"),
    "file-upload": ("file-inventory", "file-upload", "file-readback"),
    "blob-source": ("validation", "blob-inventory", "source-reconciliation", "ingestion-cycle",
                    "blob-readback", "source-readback"),
    "blob-monitor": ("ingestion-cycle",),
    "blob-capture": ("evidence-validation", "context-check", "source-binding", "blob-inventory", "checkpoint"),
    "blob-recheck": ("evidence-validation", "context-check", "source-binding", "blob-inventory",
                     "ingestion-cycle", "blob-readback", "source-readback", "context-readback"),
    "search-bootstrap": ("validation", "context-check", "region-check", "absence-check", "search-submit",
                         "arm-wait", "arm-readback"),
}
COUNTS = frozenset(("uploads_acknowledged", "files_reused", "files_verified",
                    "status_checks", "cycle_updates_processed", "cycle_items_skipped",
                    "files_failed", "files_unverified", "files_not_attempted", "files_pending", "files_ingested"))
FILE_HEARTBEAT_SECONDS = 5
IO_WARNING = "progress-output-failed: stderr progress could not be written; execution result is authoritative."
SHUTDOWN_WARNING = "progress-shutdown-flush-unresolved: failed stderr could not be redirected; shutdown may override the exit code."


def validate_blob_progress(value):
    fields = {"schema_version", "phase", "run_start", "processed", "failed", "skipped",
              "unit", "total", "remaining", "denominator", "synchronization_status",
              "elapsed_seconds", "next_check_seconds"}
    if not isinstance(value, dict) or set(value) != fields:
        raise ValueError("Invalid Blob progress shape.")
    if (value["schema_version"] != "1.0" or value["unit"] != "item-updates"
            or value["denominator"] != "not-comparable-to-files"
            or value["total"] is not None or value["remaining"] is not None
            or value["phase"] not in ("ingesting", "waiting", "throttled", "paused", "completed", "failed")
            or value["synchronization_status"] not in ("not-reported", "active", "creating", "deleting")):
        raise ValueError("Invalid Blob progress labels.")
    for field in ("processed", "failed", "skipped"):
        count = value[field]
        if count is not None and (type(count) is not int or count < 0):
            raise ValueError("Invalid Blob progress count.")
    for field in ("elapsed_seconds", "next_check_seconds"):
        number = value[field]
        if field == "next_check_seconds" and number is None:
            continue
        if type(number) not in (int, float) or not math.isfinite(number) or number < 0:
            raise ValueError("Invalid Blob progress timing.")
    if value["next_check_seconds"] is not None and value["next_check_seconds"] > 60:
        raise ValueError("Invalid Blob progress wait.")
    start = value["run_start"]
    if start is not None:
        if not isinstance(start, str) or len(start) > 40:
            raise ValueError("Invalid Blob progress run.")
        parsed = datetime.fromisoformat(start.replace("Z", "+00:00"))
        if parsed.tzinfo is None or parsed.isoformat() != start:
            raise ValueError("Blob progress run must be a canonical timestamp.")


class Progress:
    def __init__(self, workflow, *, enabled=True, stream=None, clock=time.monotonic):
        self.stages = STAGES[workflow]
        self.workflow = workflow
        self.enabled = enabled
        self.stream = stream
        self.clock = clock
        self.depth = 0
        self.stage = self.stages[0]
        self.counts = {}
        self.started = None
        self.elapsed = 0.0
        self.last_sent = None
        self.last_stage = None
        self.output_failed = False
        self.shutdown_flush_unresolved = False
        self.blob_progress = None
        self.last_blob_phase = None

    def blob_update(self, value):
        if not self.workflow.startswith("blob-"):
            raise ValueError("Blob progress cannot describe a File or ARM workflow.")
        validate_blob_progress(value)
        self.blob_progress = copy.deepcopy(value)
        for field, target in (("processed", "cycle_updates_processed"), ("skipped", "cycle_items_skipped")):
            self.counts.pop(target, None)
            if value[field] is not None:
                self.counts[target] = value[field]
        self.update("ingestion-cycle")

    def update(self, stage, **counts):
        if stage not in self.stages or self.stages.index(stage) < self.stages.index(self.stage):
            raise ValueError("Invalid progress stage transition.")
        if any(key not in COUNTS or type(value) is not int or value < 0 for key, value in counts.items()):
            raise ValueError("Invalid progress count.")
        self.stage = stage
        self.counts.update(counts)
        self._emit("running")

    def waiting(self, seconds):
        if not isinstance(seconds, (int, float)) or not math.isfinite(seconds) or not 0 <= seconds <= 30:
            raise ValueError("Invalid recovery wait.")
        self._emit("waiting", wait={"reason": "http-429", "seconds": seconds})

    @contextmanager
    def processing_file(self, ordinal, total, attempt):
        if (self.workflow not in {"file-source", "file-upload"} or type(ordinal) is not int
                or type(total) is not int or not 1 <= ordinal <= total <= 200 or attempt not in (1, 2)):
            raise ValueError("Invalid active file.")
        active = {"ordinal": ordinal, "total": total, "attempt": attempt}
        if not self.enabled or self.output_failed:
            yield
            return
        started = self.clock()
        stop = threading.Event()

        def pulse():
            while not stop.wait(FILE_HEARTBEAT_SECONDS):
                if self.output_failed:
                    return
                self._emit("processing", active=active, active_started=started)

        self._emit("processing", active=active, active_started=started)
        worker = threading.Thread(target=pulse, name="foundry-file-progress", daemon=True)
        worker.start()
        try:
            yield
        finally:
            stop.set()
            worker.join()

    def _emit(self, state, *, wait=None, active=None, active_started=None):
        if not self.enabled or self.output_failed:
            return
        observed = self.clock()
        # Clock regressions/nonfinite observations must not create negative time
        # or bypass throttling. Progress never shares the service deadline clock.
        if math.isfinite(observed):
            if self.started is None:
                self.started = observed
            delta = observed - self.started
            if math.isfinite(delta):
                self.elapsed = max(self.elapsed, delta)
        phase = self.blob_progress["phase"] if self.blob_progress is not None else None
        if (state == "running" and self.stage == self.last_stage and phase == self.last_blob_phase
                and self.last_sent is not None and self.elapsed - self.last_sent < 1.0):
            return
        event = {
            "event": "progress", "workflow": self.workflow, "activity": self.stage,
            "state": state, "elapsed_seconds": round(self.elapsed, 3),
            "remaining_checks": [] if state == "completed" else list(self.stages[self.stages.index(self.stage) + 1:]),
        }
        if self.counts:
            event["completed_counts"] = dict(self.counts)
        if wait is not None:
            event["wait"] = wait
        if active is not None:
            duration = observed - active_started
            event["processing_file"] = {**active, "elapsed_seconds": round(max(0, duration), 3) if math.isfinite(duration) else 0}
        if self.blob_progress is not None:
            value = self.blob_progress
            event["blob_progress"] = copy.deepcopy(value)
            counts = "; ".join(f"{field}={value[field] if value[field] is not None else 'unknown'}"
                               for field in ("processed", "failed", "skipped"))
            next_check = ("paused/resumable" if value["phase"] == "paused" else "none") if value["next_check_seconds"] is None else f"{value['next_check_seconds']:g}s"
            event["message"] = (
                f"Blob ingestion observation {value['phase']}: item updates {counts}; file total/remaining unknown "
                f"(not comparable); elapsed={value['elapsed_seconds']:g}s; next check={next_check}."
            )
        stream = self.stream if self.stream is not None else sys.stderr
        try:
            text = json.dumps(event, sort_keys=True, separators=(",", ":")) + "\n"
            if stream.write(text) != len(text):
                raise OSError("Short progress write.")
            stream.flush()
        except (OSError, UnicodeError, ValueError):
            # Continue the approved operation, retaining a fixed secondary warning
            # even when stderr itself is broken. Never replace a primary failure.
            self.output_failed = True
            self._disable_failed_default_stderr(stream)
        self.last_sent, self.last_stage = self.elapsed, self.stage
        self.last_blob_phase = phase

    def _disable_failed_default_stderr(self, stream):
        if self.stream is not None or stream is not sys.__stderr__ or stream.closed:
            return
        try:
            if stream.fileno() == 2:
                # A failed TextIOWrapper flush can retain pending bytes. Redirect
                # only the failed native stderr so shutdown can drain that buffer
                # without replacing the authoritative process exit status.
                sink = os.open(os.devnull, os.O_WRONLY)
                try:
                    os.dup2(sink, 2)
                finally:
                    if sink != 2:
                        os.close(sink)
        except (OSError, ValueError):
            self.shutdown_flush_unresolved = True

    def finish(self, result=None, failure=None):
        if self.blob_progress is not None and failure is not None:
            self.blob_progress.update(phase="failed", next_check_seconds=None)
        if failure is not None:
            state = "partial" if failure.partial or failure.writes else "blocked"
        else:
            state = {"verified": "completed", "unverified": "blocked"}.get(result["status"], result["status"])
        if state not in {"completed", "blocked", "partial"}:
            raise ValueError("Invalid progress terminal state.")
        self._emit(state)
        if self.output_failed:
            warnings = failure.warnings if failure is not None else result.setdefault("warnings", [])
            if IO_WARNING not in warnings:
                warnings.append(IO_WARNING)
            if self.shutdown_flush_unresolved and SHUTDOWN_WARNING not in warnings:
                warnings.append(SHUTDOWN_WARNING)


def reporting(workflow):
    """One reporter and terminal event across nested public entrypoints."""
    def decorate(function):
        @wraps(function)
        def wrapped(*args, progress=None, **kwargs):
            if progress is None:
                progress = Progress(workflow, enabled=False)
            outer = progress.depth == 0
            progress.depth += 1
            try:
                result = function(*args, progress=progress, **kwargs)
            except HelperFailure as failure:
                if outer:
                    progress.finish(failure=failure)
                raise
            else:
                if outer:
                    progress.finish(result=result)
                return result
            finally:
                progress.depth -= 1
        return wrapped
    return decorate


def add_progress_argument(parser):
    parser.add_argument("--no-progress", dest="progress", action="store_false",
                        help="Suppress content-free execution progress on stderr.")