async def parse_files(
    input_bucket: str,
    engine: str = "mistral",
    input_prefix: str | None = None,
    output_bucket: str | None = None,
    output_prefix_base: str | None = "parsed",
) -> dict:
    """
    Parse files from an input GCS bucket and upload parsed outputs.

    Files are discovered by listing blobs in ``input_bucket`` and optionally
    filtering by ``input_prefix``. This is the single async entry point for
    batch parsing.

    For ``engine='docai'``, a single DocumentAIProcessor is created once and
    reused across all files — this avoids per-file SSL handshakes and reuses
    the persistent retry session.
    """
    logger = logging.getLogger(__name__)

    if not str(output_prefix_base or "").strip() and not output_bucket:
        raise ValueError(
            "output_bucket is required when output_prefix_base is not provided"
        )

    cfg        = get_settings()
    concurrency = int(cfg.get("parsing", {}).get("batch", {}).get("max_concurrency", 5))
    semaphore  = asyncio.Semaphore(max(1, int(concurrency)))

    destination_bucket = output_bucket or input_bucket
    gcs                = GCSBucketConnector()
    object_names       = await gcs.list_files(input_bucket, prefix=input_prefix)
    object_names       = [name for name in object_names if not name.endswith("/")]

    # ── FIX: Create DocumentAIProcessor ONCE, reuse across all files ──────────
    # Previously created inside _run_one per file — caused new SSL session
    # and new credentials object for every single file processed.
    docai_processor: DocumentAIProcessor | None = None
    if engine == "docai":
        logger.info("🔌 [PARSER] Initializing DocumentAIProcessor (shared across all files)")
        docai_processor = DocumentAIProcessor()
        logger.info("✅ [PARSER] DocumentAIProcessor ready")

    logger.info(
        f"🚀 [PARSER] Starting batch parsing | "
        f"input_bucket={input_bucket} | "
        f"input_prefix={input_prefix} | "
        f"output_bucket={destination_bucket} | "
        f"engine={engine} | "
        f"total_files={len(object_names)} | "
        f"concurrency={concurrency}"
    )

    async def _run_one(object_name: str) -> dict:
        result = {
            "gcs_uri":        f"gs://{input_bucket}/{object_name}",
            "output_gcs_uri": None,
            "status":         "failed",
            "error":          None,
        }

        async with semaphore:
            try:
                ext                = Path(object_name).suffix.lower()
                destination_object = _build_output_object_name(object_name, output_prefix_base)

                logger.info(
                    f"📄 [PARSER] Processing file | "
                    f"gcs_uri=gs://{input_bucket}/{object_name} | "
                    f"ext={ext} | "
                    f"engine={engine}"
                )

                # ── Download file from GCS ────────────────────────────────────
                payload = await gcs.download(input_bucket, object_name)
                if not isinstance(payload, bytes):
                    raise TypeError(
                        f"Downloaded object is not bytes for "
                        f"'gs://{input_bucket}/{object_name}'"
                    )
                logger.info(
                    f"⬇️  [PARSER] Downloaded | "
                    f"file={object_name} | "
                    f"size={len(payload) / 1024:.1f}KB"
                )

                # ── Parse file content ────────────────────────────────────────
                parsed_output = await asyncio.to_thread(
                    _parse_content_bytes,
                    payload,
                    ext,
                    engine,
                    docai_processor,  # ← FIX: pass shared processor
                )

                # ── Serialize output ──────────────────────────────────────────
                if isinstance(parsed_output, (dict, list)):
                    destination_object = str(Path(destination_object).with_suffix(".json"))
                    raw_json           = cast(dict | list[dict], parsed_output)
                    output_bytes       = json.dumps(raw_json, ensure_ascii=True, indent=2).encode("utf-8")
                    content_type       = "application/json"
                else:
                    destination_object = str(Path(destination_object).with_suffix(".md"))
                    markdown_text      = cast(str, parsed_output)
                    output_bytes       = markdown_text.encode("utf-8")
                    content_type       = "text/markdown"

                # ── Upload parsed output to GCS ───────────────────────────────
                await gcs.save(
                    destination_bucket,
                    destination_object,
                    data=output_bytes,
                    content_type=content_type,
                )

                logger.info(
                    f"⬆️  [PARSER] Uploaded | "
                    f"output=gs://{destination_bucket}/{destination_object} | "
                    f"size={len(output_bytes) / 1024:.1f}KB"
                )

                result["status"]         = "success"
                result["output_gcs_uri"] = f"gs://{destination_bucket}/{destination_object}"

                logger.info(
                    f"✅ [PARSER] File complete | "
                    f"gcs_uri=gs://{input_bucket}/{object_name}"
                )
                return result

            except Exception as exc:
                err            = _error_payload(exc)
                result["error"] = err
                logger.error(
                    f"❌ [PARSER] Failed | "
                    f"gcs_uri=gs://{input_bucket}/{object_name} | "
                    f"error_type={err['type']} | "
                    f"message={err['message']}\n"
                    f"{err['traceback']}"
                )
                return result

    # ── Run all files concurrently ────────────────────────────────────────────
    results = (
        await asyncio.gather(*[_run_one(name) for name in object_names])
        if object_names
        else []
    )

    succeeded = sum(1 for item in results if item.get("status") == "success")
    failed    = len(results) - succeeded

    logger.info(
        f"\n{'─' * 50}\n"
        f"  📊 PARSING SUMMARY\n"
        f"{'─' * 50}\n"
        f"  📁 Total files   : {len(results)}\n"
        f"  ✅ Succeeded     : {succeeded}\n"
        f"  ❌ Failed        : {failed}\n"
        f"  🪣 Input bucket  : {input_bucket}\n"
        f"  🪣 Output bucket : {destination_bucket}\n"
        f"{'─' * 50}"
    )

    return {
        "input_bucket":  input_bucket,
        "input_prefix":  input_prefix,
        "output_bucket": destination_bucket,
        "processed":     len(results),
        "succeeded":     succeeded,
        "failed":        failed,
        "results":       list(results),
    }