feat(ledger): add download_report_bundle method

Mirror of the TypeScript SDK's getReportBundleDownloadUrl +
getReportBundleXbrlZip — one method, two flavors. Wraps the
backend's GET /reports/{id}/download endpoint and handles both
response shapes transparently:

* format='jsonld' (default): backend returns a JSON envelope with a
  short-lived presigned URL; SDK follows the URL and pulls the
  bytes. Two HTTP calls.
* format='xbrl-2.1': backend streams the zip body directly; SDK
  pulls one response. No presigned URL hop.

Signature:

  download_report_bundle(
    graph_id: str,
    report_id: str,
    *,
    format: str = 'jsonld',
    to: str | Path | None = None,
    expires_in: int = 300,
  ) -> ReportBundleDownload

Returns a ReportBundleDownload dataclass with:
  - content: bytes — raw artifact
  - filename: str — server-suggested filename (parsed from
    Content-Disposition; falls back to {report_id}.{ext})
  - format, content_type, generation_count
  - path: Path | None — set when `to` was provided (the SDK writes
    the bytes and creates parent dirs)

End-to-end verified against the live API: ~313KB JSON-LD bundle
(251 @graph nodes) + ~10KB XBRL zip (5 files: instance.xml,
report.xsd, report-pre/cal/def.xml).

Tests cover JSON-LD two-hop path, XBRL one-hop path, on-disk
write with parent dir creation, non-2xx error path, missing-token
gate, and the Content-Disposition filename parser (quoted /
unquoted / missing). 88 ledger-client tests pass.

Unlocks bundle download in demo scripts (next commit) and any
external Python consumer of the SDK.

Author: jfrench9

robosystems_client/clients/ledger_client.py

@@ -23,9 +23,14 @@
 from __future__ import annotations
 
 import datetime
+import re
+from dataclasses import dataclass
 from http import HTTPStatus
+from pathlib import Path
 from typing import Any
 
+import httpx
+
 from ..api.extensions_robo_ledger.op_auto_map_elements import (
   sync_detailed as op_auto_map_elements,
 )
@@ -308,6 +313,43 @@
 from ..types import UNSET
 
 
+# Captures the ``filename`` value from a Content-Disposition header, with
+# or without quotes. Used by ``download_report_bundle`` to recover the
+# server-suggested filename when writing the bundle to disk.
+_FILENAME_PATTERN = re.compile(r'filename="?([^";]+)"?', re.IGNORECASE)
+
+
+def _parse_filename(content_disposition: str) -> str | None:
+  """Extract the ``filename`` value from a Content-Disposition header.
+
+  Returns ``None`` when the header is empty or doesn't carry a
+  filename — the caller falls back to a synthesized name in that case.
+  """
+  if not content_disposition:
+    return None
+  match = _FILENAME_PATTERN.search(content_disposition)
+  return match.group(1) if match else None
+
+
+@dataclass
+class ReportBundleDownload:
+  """Result of downloading a Report's serialization bundle.
+
+  ``content`` is the raw artifact bytes. ``filename`` is the
+  server-suggested name (``{report_id}-g{generation}.{ext}``).
+  ``path`` is populated when the caller passed a ``to=`` argument to
+  :meth:`LedgerClient.download_report_bundle` — points at the file
+  the SDK wrote to disk.
+  """
+
+  content: bytes
+  filename: str
+  format: str
+  content_type: str
+  generation_count: int | None
+  path: Path | None = None
+
+
 class LedgerClient:
   """High-level facade for the RoboLedger domain.
 
@@ -1717,6 +1759,99 @@ def share_report(
     envelope = self._call_op("Share report", response)
     return self._typed_result("Share report", envelope, ShareReportResponse)
 
+  def download_report_bundle(
+    self,
+    graph_id: str,
+    report_id: str,
+    *,
+    format: str = "jsonld",
+    to: str | Path | None = None,
+    expires_in: int = 300,
+  ) -> ReportBundleDownload:
+    """Download a Report's serialization bundle (JSON-LD or XBRL 2.1).
+
+    JSON-LD path: backend returns a JSON envelope with a short-lived
+    presigned URL to the S3-stamped bundle; client follows the URL
+    and pulls the bytes. XBRL path: backend streams the zip directly
+    (per spec — XBRL is on-demand emit, not stored).
+
+    Args:
+        graph_id: Graph identifier owning the Report.
+        report_id: Report identifier (``rpt_``-prefixed ULID).
+        format: Serialization flavor — ``"jsonld"`` for the JSON-LD
+            bundle, ``"xbrl-2.1"`` for the XBRL 2.1 zip. Other RDF /
+            XBRL flavors slot in as their producers ship.
+        to: Optional file path to write the bytes to. When set, the
+            returned ``ReportBundleDownload.path`` points at the
+            written file.
+        expires_in: Presigned URL lifetime in seconds (JSON-LD only;
+            ignored for XBRL flavors).
+
+    Returns:
+        :class:`ReportBundleDownload` with the artifact bytes,
+        server-suggested filename, content type, generation count,
+        and (when ``to`` is set) the path written.
+
+    Raises:
+        RuntimeError: backend returned non-2xx or the presigned URL
+            could not be followed.
+    """
+    if not self.token:
+      raise RuntimeError("No API key provided. Set X-API-Key in headers.")
+    base = self.base_url.rstrip("/")
+    url = (
+      f"{base}/extensions/roboledger/{graph_id}/reports/{report_id}"
+      f"/download?format={format}&expires_in={expires_in}"
+    )
+    headers = {**self.headers, "X-API-Key": self.token}
+    with httpx.Client(timeout=self.timeout) as client:
+      response = client.get(url, headers=headers)
+      if response.status_code != 200:
+        raise RuntimeError(
+          f"Download bundle failed ({response.status_code}): {response.text}"
+        )
+      content_type = response.headers.get("content-type", "")
+      if "application/json" in content_type:
+        # JSON-LD path — envelope contains a presigned URL to follow
+        envelope = response.json()
+        download_url = envelope["download_url"]
+        artifact = client.get(download_url)
+        if artifact.status_code != 200:
+          raise RuntimeError(
+            f"Failed to follow presigned URL ({artifact.status_code}): {artifact.text}"
+          )
+        filename = (
+          _parse_filename(artifact.headers.get("content-disposition", ""))
+          or f"{report_id}-g{envelope.get('generation_count', 1)}.jsonld"
+        )
+        result = ReportBundleDownload(
+          content=artifact.content,
+          filename=filename,
+          format=str(envelope.get("format", format)),
+          content_type=str(envelope.get("content_type", "application/ld+json")),
+          generation_count=envelope.get("generation_count"),
+        )
+      else:
+        # XBRL path — body IS the zip
+        filename = (
+          _parse_filename(response.headers.get("content-disposition", ""))
+          or f"{report_id}.zip"
+        )
+        generation_header = response.headers.get("x-bundle-generation")
+        result = ReportBundleDownload(
+          content=response.content,
+          filename=filename,
+          format=response.headers.get("x-bundle-format", format),
+          content_type=content_type or "application/zip",
+          generation_count=int(generation_header) if generation_header else None,
+        )
+    if to is not None:
+      path = Path(to)
+      path.parent.mkdir(parents=True, exist_ok=True)
+      path.write_bytes(result.content)
+      result.path = path
+    return result
+
   def file_report(self, graph_id: str, report_id: str) -> ReportResponse:
     """Transition a Report's filing_status to 'filed' — locks the package.
 

tests/test_ledger_client.py

@@ -1821,3 +1821,165 @@ def test_financial_statement_analysis_with_ticker(
     body = mock_op.call_args.kwargs["body"]
     assert body.ticker == "NVDA"
     assert body.fiscal_year == 2025
+
+
+# ── Bundle download (REST GET to /reports/{id}/download) ──────────────
+
+
+@pytest.mark.unit
+class TestDownloadReportBundle:
+  """``download_report_bundle`` bypasses the typed op_* generators and
+  hits the endpoint via httpx directly — these tests stub the httpx
+  Client at the call site."""
+
+  def _mock_json_envelope(self, gen: int = 1) -> Mock:
+    """Mock the initial JSON envelope response for the JSON-LD path."""
+    resp = Mock()
+    resp.status_code = 200
+    resp.headers = {"content-type": "application/json"}
+    resp.json.return_value = {
+      "download_url": "https://s3.example.com/bundles/rpt_01/g1.jsonld",
+      "expires_at": "2026-05-28T12:30:00Z",
+      "content_type": "application/ld+json",
+      "format": "jsonld",
+      "generation_count": gen,
+    }
+    return resp
+
+  def _mock_artifact(self, content: bytes, filename: str = "rpt_01-g1.jsonld") -> Mock:
+    resp = Mock()
+    resp.status_code = 200
+    resp.headers = {"content-disposition": f'attachment; filename="{filename}"'}
+    resp.content = content
+    return resp
+
+  def _mock_xbrl_response(self, content: bytes, gen: int = 1) -> Mock:
+    """Mock the direct binary-stream response for the XBRL path."""
+    resp = Mock()
+    resp.status_code = 200
+    resp.headers = {
+      "content-type": "application/zip",
+      "content-disposition": f'attachment; filename="rpt_01-g{gen}.zip"',
+      "x-bundle-format": "xbrl-2.1",
+      "x-bundle-generation": str(gen),
+    }
+    resp.content = content
+    resp.text = ""
+    return resp
+
+  @patch("robosystems_client.clients.ledger_client.httpx.Client")
+  def test_jsonld_download_follows_presigned_url(
+    self, mock_client_cls, mock_config, graph_id
+  ):
+    """JSON-LD path: envelope GET → presigned URL GET → bytes."""
+    mock_client = Mock()
+    mock_client.__enter__ = Mock(return_value=mock_client)
+    mock_client.__exit__ = Mock(return_value=None)
+    mock_client.get.side_effect = [
+      self._mock_json_envelope(gen=2),
+      self._mock_artifact(b'{"@graph": []}', "rpt_01-g2.jsonld"),
+    ]
+    mock_client_cls.return_value = mock_client
+
+    result = LedgerClient(mock_config).download_report_bundle(
+      graph_id, "rpt_01", format="jsonld"
+    )
+
+    assert result.content == b'{"@graph": []}'
+    assert result.filename == "rpt_01-g2.jsonld"
+    assert result.format == "jsonld"
+    assert result.content_type == "application/ld+json"
+    assert result.generation_count == 2
+    # First request hits our endpoint with X-API-Key auth.
+    first_call = mock_client.get.call_args_list[0]
+    assert "/reports/rpt_01/download" in first_call.args[0]
+    assert "format=jsonld" in first_call.args[0]
+    headers = first_call.kwargs["headers"]
+    assert headers["X-API-Key"] == "test-api-key"
+
+  @patch("robosystems_client.clients.ledger_client.httpx.Client")
+  def test_xbrl_download_streams_zip_directly(
+    self, mock_client_cls, mock_config, graph_id
+  ):
+    """XBRL path: single GET, binary response, no presigned URL hop."""
+    zip_bytes = b"PK\x03\x04dummy-zip-payload"
+    mock_client = Mock()
+    mock_client.__enter__ = Mock(return_value=mock_client)
+    mock_client.__exit__ = Mock(return_value=None)
+    mock_client.get.return_value = self._mock_xbrl_response(zip_bytes, gen=3)
+    mock_client_cls.return_value = mock_client
+
+    result = LedgerClient(mock_config).download_report_bundle(
+      graph_id, "rpt_01", format="xbrl-2.1"
+    )
+
+    assert result.content == zip_bytes
+    assert result.filename == "rpt_01-g3.zip"
+    assert result.format == "xbrl-2.1"
+    assert result.content_type == "application/zip"
+    assert result.generation_count == 3
+    # XBRL path only makes one HTTP call.
+    assert mock_client.get.call_count == 1
+
+  @patch("robosystems_client.clients.ledger_client.httpx.Client")
+  def test_to_arg_writes_bytes_to_disk(
+    self, mock_client_cls, mock_config, graph_id, tmp_path
+  ):
+    """``to=path`` writes the artifact and exposes ``result.path``."""
+    zip_bytes = b"PK\x03\x04zip"
+    mock_client = Mock()
+    mock_client.__enter__ = Mock(return_value=mock_client)
+    mock_client.__exit__ = Mock(return_value=None)
+    mock_client.get.return_value = self._mock_xbrl_response(zip_bytes)
+    mock_client_cls.return_value = mock_client
+
+    target = tmp_path / "subdir" / "out.zip"
+    result = LedgerClient(mock_config).download_report_bundle(
+      graph_id, "rpt_01", format="xbrl-2.1", to=str(target)
+    )
+
+    assert result.path == target
+    assert target.read_bytes() == zip_bytes
+    # Parent directories get created automatically.
+    assert target.parent.is_dir()
+
+  @patch("robosystems_client.clients.ledger_client.httpx.Client")
+  def test_non_200_endpoint_response_raises(
+    self, mock_client_cls, mock_config, graph_id
+  ):
+    err_resp = Mock(status_code=404, text="not found")
+    err_resp.headers = {}
+    mock_client = Mock()
+    mock_client.__enter__ = Mock(return_value=mock_client)
+    mock_client.__exit__ = Mock(return_value=None)
+    mock_client.get.return_value = err_resp
+    mock_client_cls.return_value = mock_client
+
+    with pytest.raises(RuntimeError, match="Download bundle failed"):
+      LedgerClient(mock_config).download_report_bundle(graph_id, "rpt_missing")
+
+  def test_no_token_raises(self, mock_config, graph_id):
+    mock_config["token"] = None
+    with pytest.raises(RuntimeError, match="No API key"):
+      LedgerClient(mock_config).download_report_bundle(graph_id, "rpt_01")
+
+
+def test_parse_filename_quoted():
+  from robosystems_client.clients.ledger_client import _parse_filename
+
+  assert _parse_filename('attachment; filename="rpt_01-g1.jsonld"') == (
+    "rpt_01-g1.jsonld"
+  )
+
+
+def test_parse_filename_unquoted():
+  from robosystems_client.clients.ledger_client import _parse_filename
+
+  assert _parse_filename("attachment; filename=rpt_01-g1.zip") == "rpt_01-g1.zip"
+
+
+def test_parse_filename_missing_returns_none():
+  from robosystems_client.clients.ledger_client import _parse_filename
+
+  assert _parse_filename("") is None
+  assert _parse_filename("attachment") is None