feat(cli): record pull structured logs

data/state/logs/.gitignore

@@ -0,0 +1,2 @@
+*
+!.gitignore

library/src/iqb/cli/cache_pull.py

@@ -1,9 +1,15 @@
 """Cache pull command."""
 
+# TODO(bassosimone): this download logic overlaps with ghremote/cache.py;
+# consider unifying into a shared helper once both implementations stabilise.
+
 import hashlib
+import json
 import os
+import threading
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
@@ -29,13 +35,24 @@ def _short_name(file: str) -> str:
     return "/".join(parts[-2:]) if len(parts) >= 2 else file
 
 
+def _now() -> str:
+    """Return the current time formatted for metrics spans."""
+    return datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %z")
+
+
 def _download_one(
     entry: DiffEntry,
     data_dir: Path,
     session: requests.Session,
     progress: Progress,
-) -> str:
-    """Download a single file, verify SHA256, atomic-replace. Returns the file path."""
+) -> dict[str, object]:
+    """Download a single file, verify SHA256, atomic-replace. Returns a metrics span."""
+    t0 = _now()
+    worker_id = threading.get_ident()
+    total_bytes = 0
+    content_length: int | None = None
+    ok = True
+    error: str | None = None
     assert entry.url is not None
     assert entry.remote_sha256 is not None
     dest = data_dir / entry.file
@@ -46,24 +63,39 @@ def _download_one(
             tmp_file = Path(tmp_dir) / dest.name
             resp = session.get(entry.url, stream=True)
             resp.raise_for_status()
-            content_length = resp.headers.get("Content-Length")
-            if content_length is not None:
-                progress.update(task_id, total=int(content_length))
+            cl = resp.headers.get("Content-Length")
+            if cl is not None:
+                content_length = int(cl)
+                progress.update(task_id, total=content_length)
             sha256 = hashlib.sha256()
             with open(tmp_file, "wb") as fp:
                 for chunk in resp.iter_content(chunk_size=8192):
                     fp.write(chunk)
                     sha256.update(chunk)
+                    total_bytes += len(chunk)
                     progress.update(task_id, advance=len(chunk))
             got = sha256.hexdigest()
             if got != entry.remote_sha256:
                 raise ValueError(
                     f"SHA256 mismatch for {entry.file}: expected {entry.remote_sha256}, got {got}"
                 )
             os.replace(tmp_file, dest)
+    except Exception as exc:
+        ok = False
+        error = str(exc)
     finally:
         progress.remove_task(task_id)
-    return entry.file
+    return {
+        "t0": t0,
+        "t": _now(),
+        "worker_id": worker_id,
+        "file": entry.file,
+        "url": entry.url,
+        "content_length": content_length,
+        "bytes": total_bytes,
+        "ok": ok,
+        "error": error,
+    }
 
 
 @cache.command()
@@ -90,6 +122,7 @@ def pull(data_dir: str | None, force: bool, jobs: int) -> None:
 
     session = requests.Session()
     failed: list[tuple[str, str]] = []
+    spans: list[dict[str, object]] = []
     t0 = time.monotonic()
     with (
         Progress(
@@ -107,13 +140,27 @@ def pull(data_dir: str | None, force: bool, jobs: int) -> None:
         for future in as_completed(futures):
             entry = futures[future]
             try:
-                future.result()
+                span = future.result()
+                spans.append(span)
+                if not span["ok"]:
+                    failed.append((str(span["file"]), str(span["error"] or "unknown")))
             except Exception as exc:
+                # Defensive: bug in _download_one itself
                 failed.append((entry.file, str(exc)))
     elapsed = time.monotonic() - t0
 
-    ok = len(targets) - len(failed)
-    click.echo(f"Downloaded {ok}/{len(targets)} file(s) in {elapsed:.1f}s.")
+    # Write metrics
+    log_dir = resolved / "state" / "logs"
+    log_dir.mkdir(parents=True, exist_ok=True)
+    now = datetime.now().astimezone().strftime("%Y%m%dT%H%M%S")
+    log_file = log_dir / f"{now}_{time.time_ns()}_pull.jsonl"
+    with open(log_file, "w") as fp:
+        for span in spans:
+            fp.write(json.dumps(span) + "\n")
+
+    ok_count = len(targets) - len(failed)
+    click.echo(f"Downloaded {ok_count}/{len(targets)} file(s) in {elapsed:.1f}s.")
+    click.echo(f"Detailed logs: {log_file.relative_to(resolved)}")
 
     if failed:
         click.echo(f"{len(failed)} download(s) failed:", err=True)

library/tests/iqb/cli/cache_pull_test.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import json
+from datetime import datetime
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -36,6 +37,24 @@ def _make_cache_file(data_dir: Path, rel_path: str, content: bytes) -> Path:
 _FILE_B = f"cache/v1/{_TS1}/{_TS2}/uploads/data.parquet"
 
 
+def _find_log_file(data_dir: Path) -> Path | None:
+    """Find the single JSONL log file under state/logs/."""
+    log_dir = data_dir / "state" / "logs"
+    if not log_dir.exists():
+        return None
+    files = list(log_dir.glob("*_pull.jsonl"))
+    return files[0] if len(files) == 1 else None
+
+
+def _read_spans(log_file: Path) -> list[dict]:
+    """Read all spans from a JSONL log file."""
+    spans = []
+    for line in log_file.read_text().splitlines():
+        if line.strip():
+            spans.append(json.loads(line))
+    return spans
+
+
 def _fake_response(content: bytes) -> MagicMock:
     """Create a mock response that yields content in chunks."""
     resp = MagicMock()
@@ -82,6 +101,15 @@ def test_downloads_only_remote(self, mock_session_cls: MagicMock, tmp_path: Path
         # Session.get was called with the right URL
         mock_session.get.assert_called_once_with(url, stream=True)
 
+        # Metrics log should exist with one span
+        log_file = _find_log_file(tmp_path)
+        assert log_file is not None
+        spans = _read_spans(log_file)
+        assert len(spans) == 1
+        assert spans[0]["ok"] is True
+        assert spans[0]["file"] == _FILE_A
+        assert spans[0]["bytes"] == len(content)
+
 
 class TestCachePullMatchingSkipped:
     """MATCHING entries are not downloaded."""
@@ -178,6 +206,15 @@ def test_sha256_failure(self, mock_session_cls: MagicMock, tmp_path: Path):
         # File should NOT exist on disk (atomic write prevented partial file)
         assert not (tmp_path / _FILE_A).exists()
 
+        # Metrics log should record the failure
+        log_file = _find_log_file(tmp_path)
+        assert log_file is not None
+        spans = _read_spans(log_file)
+        assert len(spans) == 1
+        assert spans[0]["ok"] is False
+        assert spans[0]["error"] is not None
+        assert "SHA256 mismatch" in spans[0]["error"]
+
 
 class TestCachePullMultipleFiles:
     """Multiple ONLY_REMOTE entries are all downloaded."""
@@ -243,3 +280,87 @@ def test_no_partial_file_on_failure(self, mock_session_cls: MagicMock, tmp_path:
 
         # No partial file should exist
         assert not (tmp_path / _FILE_A).exists()
+
+        # Metrics log should record the connection failure
+        log_file = _find_log_file(tmp_path)
+        assert log_file is not None
+        spans = _read_spans(log_file)
+        assert len(spans) == 1
+        assert spans[0]["ok"] is False
+        assert "connection refused" in spans[0]["error"]
+
+
+class TestCachePullMetrics:
+    """Download metrics JSONL file is created with correct schema."""
+
+    _EXPECTED_KEYS = {
+        "t0",
+        "t",
+        "worker_id",
+        "file",
+        "url",
+        "content_length",
+        "bytes",
+        "ok",
+        "error",
+    }
+
+    @patch("iqb.cli.cache_pull.requests.Session")
+    def test_two_files_produce_two_spans(self, mock_session_cls: MagicMock, tmp_path: Path):
+        content_a = b"content a"
+        content_b = b"content b"
+        url_a = "https://example.com/a"
+        url_b = "https://example.com/b"
+        _write_manifest(
+            tmp_path,
+            {
+                _FILE_A: {"sha256": _sha256(content_a), "url": url_a},
+                _FILE_B: {"sha256": _sha256(content_b), "url": url_b},
+            },
+        )
+
+        mock_session = MagicMock()
+
+        def side_effect(url: str, **kwargs):
+            if url == url_a:
+                return _fake_response(content_a)
+            return _fake_response(content_b)
+
+        mock_session.get.side_effect = side_effect
+        mock_session_cls.return_value = mock_session
+
+        runner = CliRunner()
+        result = runner.invoke(cli, ["cache", "pull", "-d", str(tmp_path)])
+        assert result.exit_code == 0
+
+        log_file = _find_log_file(tmp_path)
+        assert log_file is not None
+        assert log_file.name.endswith("_pull.jsonl")
+        spans = _read_spans(log_file)
+        assert len(spans) == 2
+
+        for span in spans:
+            # All expected keys are present
+            assert set(span.keys()) == self._EXPECTED_KEYS
+
+            # ok is True
+            assert span["ok"] is True
+            assert span["error"] is None
+
+            # worker_id is an integer
+            assert isinstance(span["worker_id"], int)
+
+            # Timestamps are parseable and t0 <= t
+            t0 = datetime.strptime(span["t0"], "%Y-%m-%d %H:%M:%S %z")
+            t = datetime.strptime(span["t"], "%Y-%m-%d %H:%M:%S %z")
+            assert t0 <= t
+
+            # URL is present
+            assert span["url"] in (url_a, url_b)
+
+        # Check per-file details
+        spans_by_file = {s["file"]: s for s in spans}
+        assert spans_by_file[_FILE_A]["bytes"] == len(content_a)
+        assert spans_by_file[_FILE_A]["content_length"] == len(content_a)
+        assert spans_by_file[_FILE_B]["bytes"] == len(content_b)
+        assert spans_by_file[_FILE_B]["content_length"] == len(content_b)