Nested key layout for archive manifests and reports

code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_layout.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+from forecasting_tools.agents_and_tools.source_archive import layout
+from forecasting_tools.agents_and_tools.source_archive import manifest as manifest_io
+from forecasting_tools.agents_and_tools.source_archive.catalog import _load_all_records
+from forecasting_tools.agents_and_tools.source_archive.config import ArchiveConfig
+from forecasting_tools.agents_and_tools.source_archive.models import CitationRecord
+from forecasting_tools.agents_and_tools.source_archive.pipeline import (
+    CaptureOutcome,
+    PipelineSummary,
+)
+from forecasting_tools.agents_and_tools.source_archive.reports import (
+    read_outcomes,
+    write_run_report,
+)
+from forecasting_tools.agents_and_tools.source_archive.storage import LocalBlobStore
+
+
+# --- key helpers --------------------------------------------------------------
+def test_manifest_key_nests_by_group_daily_or_adhoc():
+    assert (
+        layout.manifest_key("r1", group="sprints/myrun")
+        == "manifests/sprints/myrun/r1.jsonl"
+    )
+    assert (
+        layout.manifest_key("daily-2026-07-01")
+        == "manifests/daily/2026-07/daily-2026-07-01.jsonl"
+    )
+    assert layout.manifest_key("r1") == "manifests/adhoc/r1.jsonl"
+
+
+def test_report_key_nests_and_keeps_suffix():
+    assert (
+        layout.report_key("daily-2026-07-01", ".json")
+        == "reports/daily/2026-07/daily-2026-07-01.json"
+    )
+    assert layout.report_key("r1", "_cost.json") == "reports/adhoc/r1_cost.json"
+    assert (
+        layout.report_key("r1", ".json", group="sprints/myrun")
+        == "reports/sprints/myrun/r1.json"
+    )
+
+
+def test_group_slashes_are_normalized():
+    assert layout.manifest_key("r1", group="/sprints/myrun/") == (
+        "manifests/sprints/myrun/r1.jsonl"
+    )
+
+
+def test_candidates_prefer_nested_then_legacy_flat():
+    assert layout.manifest_key_candidates("r1") == [
+        "manifests/adhoc/r1.jsonl",
+        "manifests/r1.jsonl",
+    ]
+    assert layout.report_key_candidates("r1", ".json") == [
+        "reports/adhoc/r1.json",
+        "reports/r1.json",
+    ]
+
+
+# --- readers ------------------------------------------------------------------
+def test_read_blob_falls_back_to_legacy_flat_key(tmp_path):
+    store = LocalBlobStore(tmp_path)
+    cfg = ArchiveConfig(s3_prefix="t")
+    legacy = manifest_io.dumps([CitationRecord(url="https://old.test", run_id="r1")])
+    store.put("t/manifests/r1.jsonl", legacy.encode("utf-8"))
+
+    assert manifest_io.read_blob(store, "r1", cfg)[0].url == "https://old.test"
+
+
+def test_read_blob_prefers_nested_over_legacy(tmp_path):
+    store = LocalBlobStore(tmp_path)
+    cfg = ArchiveConfig(s3_prefix="t")
+    legacy = manifest_io.dumps([CitationRecord(url="https://old.test", run_id="r1")])
+    store.put("t/manifests/r1.jsonl", legacy.encode("utf-8"))
+    manifest_io.write_blob(
+        store, "r1", [CitationRecord(url="https://new.test", run_id="r1")], cfg
+    )
+
+    assert manifest_io.read_blob(store, "r1", cfg)[0].url == "https://new.test"
+
+
+def test_catalog_loads_nested_and_flat_manifests(tmp_path):
+    store = LocalBlobStore(tmp_path)
+    cfg = ArchiveConfig(s3_prefix="t")
+    store.put(
+        "t/manifests/old.jsonl",
+        manifest_io.dumps([CitationRecord(url="https://old.test")]).encode("utf-8"),
+    )
+    manifest_io.write_blob(
+        store, "daily-2026-07-01", [CitationRecord(url="https://daily.test")], cfg
+    )
+    manifest_io.write_blob(
+        store,
+        "sprint-run",
+        [CitationRecord(url="https://sprint.test")],
+        cfg,
+        group="sprints/myrun",
+    )
+
+    urls = {r.url for r in _load_all_records(store, "t")}
+    assert urls == {"https://old.test", "https://daily.test", "https://sprint.test"}
+
+
+def test_read_outcomes_sees_nested_and_flat_reports(tmp_path):
+    store = LocalBlobStore(tmp_path)
+    cfg = ArchiveConfig(s3_prefix="t")
+    store.put(
+        "t/reports/old.json",
+        b'[{"url": "https://old.test", "status": "stored", "reason": ""}]',
+    )
+    write_run_report(
+        store,
+        "daily-2026-07-01",
+        PipelineSummary(
+            outcomes=[CaptureOutcome(url="https://daily.test", status="error")]
+        ),
+        cfg,
+    )
+
+    out = read_outcomes(store, cfg)
+    assert out["https://old.test"] == "stored"
+    assert out["https://daily.test"] == "error"

code_tests/unit_tests/test_agents_and_tools/test_source_archive/test_pipeline_and_manifest.py

@@ -142,7 +142,7 @@ def test_manifest_blob_roundtrip(tmp_path):
     cfg = ArchiveConfig(s3_prefix="t")
     records = [CitationRecord(url="https://a.test", run_id="r1")]
     manifest.write_blob(store, "r1", records, cfg)
-    assert store.exists("t/manifests/r1.jsonl")
+    assert store.exists("t/manifests/adhoc/r1.jsonl")
     assert manifest.read_blob(store, "r1", cfg)[0].url == "https://a.test"
 
 

forecasting_tools/agents_and_tools/source_archive/README.md

@@ -271,8 +271,15 @@ are written up in [ROADMAP.md](ROADMAP.md).
 <prefix>/content/<url_hash>/<content_hash>.html
 <prefix>/content/<url_hash>/<content_hash>.webp     (screenshot)
 <prefix>/content/<url_hash>/<content_hash>.md
-<prefix>/manifests/<run_id>.jsonl                  the run's citation manifest
-<prefix>/reports/<run_id>.json                     per-URL capture outcomes (for coverage)
+<prefix>/manifests/<nest>/<run_id>.jsonl           the run's citation manifest
+<prefix>/reports/<nest>/<run_id>.json              per-URL capture outcomes (for coverage)
+<prefix>/reports/<nest>/<run_id>_cost.json         the run's estimated cost breakdown
 <prefix>/catalog/index.html                        browsable catalog (by question/bot/site)
 <prefix>/catalog/by-question/<id>.{html,csv}
 ```
+
+`<nest>` keeps manifests/reports from piling into one flat folder: it is
+`daily/<YYYY-MM>` for `daily-YYYY-MM-DD` run ids, `adhoc` otherwise, or
+whatever you pin with `--group` (e.g. `--group sprints/myrun`). Readers list by
+prefix (and fall back to the old flat keys), so archives written before this
+layout keep working unchanged.

forecasting_tools/agents_and_tools/source_archive/catalog.py

@@ -16,10 +16,9 @@
     catalog/by-bot/<bot>.html         one bot's sources across questions
     catalog/by-domain/<domain>.html   sources grouped by site
 
-The question view is the default because that's how post-mortems and
-non-technical coworkers think ("what did we know about question X?"); ``by-bot``
-covers profiling/"what is the top bot-maker doing", always next to how other
-bots handled the same question.
+The question view is the default because that's how post-mortems think
+("what sources informed question X?"); ``by-bot`` groups a bot's sources
+across questions, and ``by-domain`` groups them by site.
 """
 
 from __future__ import annotations

forecasting_tools/agents_and_tools/source_archive/cli.py

@@ -119,10 +119,14 @@ def _cmd_capture(args, config: ArchiveConfig) -> int:
     if run_id:
         from forecasting_tools.agents_and_tools.source_archive import reports
 
-        reports.write_run_report(store.blobs, run_id, summary, config)
-        print(f"Wrote run outcomes -> {config.s3_prefix}/reports/{run_id}.json")
-        cost_mod.write_cost_report(store.blobs, run_id, run_cost, config)
-        print(f"Wrote cost report -> {config.s3_prefix}/reports/{run_id}_cost.json")
+        key = reports.write_run_report(
+            store.blobs, run_id, summary, config, group=args.group
+        )
+        print(f"Wrote run outcomes -> {key}")
+        key = cost_mod.write_cost_report(
+            store.blobs, run_id, run_cost, config, group=args.group
+        )
+        print(f"Wrote cost report -> {key}")
 
     # Failures leave no cache entry, so re-running retries exactly them. Write a
     # retry manifest (with provenance) so coming back — e.g. with hyperbrowser
@@ -147,8 +151,10 @@ def _cmd_capture(args, config: ArchiveConfig) -> int:
     if args.upload_manifest:
         if not run_id:
             sys.exit("--upload-manifest needs --run-id (no run_id found in records)")
-        manifest_io.write_blob(store.blobs, run_id, records, config)
-        print(f"Uploaded manifest -> {config.s3_prefix}/manifests/{run_id}.jsonl")
+        manifest_io.write_blob(store.blobs, run_id, records, config, group=args.group)
+        print(
+            f"Uploaded manifest -> {manifest_io.manifest_key(run_id, config, args.group)}"
+        )
     return 0
 
 
@@ -173,8 +179,10 @@ def _cmd_ingest_traces(args, config: ArchiveConfig) -> int:
         if not run_id:
             sys.exit("--upload needs a run id (pass --run-id; none found in records)")
         store = _make_blob_store(config, None, args.bucket)
-        manifest_io.write_blob(store, run_id, records, config)
-        print(f"Uploaded manifest -> {config.s3_prefix}/manifests/{run_id}.jsonl")
+        manifest_io.write_blob(store, run_id, records, config, group=args.group)
+        print(
+            f"Uploaded manifest -> {manifest_io.manifest_key(run_id, config, args.group)}"
+        )
     return 0
 
 
@@ -244,6 +252,12 @@ def main(argv: list[str] | None = None) -> int:
         help="also upload the manifest itself to manifests/<run_id>.jsonl",
     )
     cap.add_argument("--run-id", help="run id for the uploaded manifest")
+    cap.add_argument(
+        "--group",
+        help="nest the uploaded manifest/reports under this folder, e.g. "
+        "'sprints/myrun' (default: daily/<YYYY-MM>/ for daily-YYYY-MM-DD "
+        "run ids, else adhoc/)",
+    )
     cap.add_argument(
         "--no-hyperbrowser",
         action="store_true",
@@ -287,6 +301,11 @@ def main(argv: list[str] | None = None) -> int:
     ing.add_argument(
         "--upload", action="store_true", help="upload the manifest to S3 manifests/"
     )
+    ing.add_argument(
+        "--group",
+        help="nest the uploaded manifest under this folder, e.g. 'sprints/myrun' "
+        "(default: daily/<YYYY-MM>/ for daily-YYYY-MM-DD run ids, else adhoc/)",
+    )
     ing.add_argument("--bucket", help="override the S3 bucket")
 
     cat = sub.add_parser(

forecasting_tools/agents_and_tools/source_archive/cost.py

@@ -19,6 +19,7 @@
 
 from pydantic import BaseModel
 
+from forecasting_tools.agents_and_tools.source_archive import layout
 from forecasting_tools.agents_and_tools.source_archive.config import ArchiveConfig
 
 # $ per vendor credit (public list pricing, 2026-06).
@@ -116,13 +117,18 @@ def estimate_run_cost(
     )
 
 
-def cost_report_key(run_id: str, config: ArchiveConfig) -> str:
-    return f"{config.s3_prefix.rstrip('/')}/reports/{run_id}_cost.json"
+def cost_report_key(
+    run_id: str, config: ArchiveConfig, group: str | None = None
+) -> str:
+    prefix = config.s3_prefix.rstrip("/")
+    return f"{prefix}/{layout.report_key(run_id, '_cost.json', group)}"
 
 
-def write_cost_report(store, run_id: str, cost: RunCost, config: ArchiveConfig) -> str:
-    """Persist the cost breakdown next to the run report (``reports/<id>_cost.json``)."""
-    key = cost_report_key(run_id, config)
+def write_cost_report(
+    store, run_id: str, cost: RunCost, config: ArchiveConfig, group: str | None = None
+) -> str:
+    """Persist the cost breakdown next to the run report (``<id>_cost.json``)."""
+    key = cost_report_key(run_id, config, group)
     store.put(
         key,
         json.dumps(cost.model_dump(), indent=2).encode("utf-8"),

forecasting_tools/agents_and_tools/source_archive/layout.py

@@ -0,0 +1,62 @@
+"""Key layout for manifests and run reports in the blob store.
+
+Manifests and reports used to be written flat (``manifests/<run_id>.jsonl``,
+``reports/<run_id>.json``), which turns into one giant folder as runs
+accumulate. New writes nest by run family instead:
+
+- an explicit ``group`` pins the folder (e.g. ``sprints/myrun``);
+- ``daily-YYYY-MM-DD`` run ids nest under ``daily/<YYYY-MM>/``;
+- everything else lands under ``adhoc/``.
+
+Readers that list by prefix (catalog, coverage) pick up both layouts for free;
+exact-key readers should try :func:`manifest_key_candidates` /
+:func:`report_key_candidates` — nested first, then the legacy flat key — so
+archives written before the nesting keep working.
+
+Keys here are store-relative (no ``s3_prefix``); callers prepend the prefix.
+"""
+
+from __future__ import annotations
+
+import re
+
+_DAILY_RUN_ID = re.compile(r"^daily-(\d{4}-\d{2})-\d{2}$")
+
+
+def _folder(run_id: str, group: str | None) -> str:
+    if group:
+        return group.strip("/")
+    match = _DAILY_RUN_ID.match(run_id)
+    if match:
+        return f"daily/{match.group(1)}"
+    return "adhoc"
+
+
+def manifest_key(run_id: str, group: str | None = None) -> str:
+    """Nested key for a run's citation manifest."""
+    return f"manifests/{_folder(run_id, group)}/{run_id}.jsonl"
+
+
+def report_key(run_id: str, suffix: str, group: str | None = None) -> str:
+    """Nested key for a run report artifact (``suffix`` e.g. ``.json``)."""
+    return f"reports/{_folder(run_id, group)}/{run_id}{suffix}"
+
+
+def legacy_manifest_key(run_id: str) -> str:
+    return f"manifests/{run_id}.jsonl"
+
+
+def legacy_report_key(run_id: str, suffix: str) -> str:
+    return f"reports/{run_id}{suffix}"
+
+
+def manifest_key_candidates(run_id: str, group: str | None = None) -> list[str]:
+    """Where a run's manifest may live, preferred (nested) first."""
+    return [manifest_key(run_id, group), legacy_manifest_key(run_id)]
+
+
+def report_key_candidates(
+    run_id: str, suffix: str, group: str | None = None
+) -> list[str]:
+    """Where a run's report may live, preferred (nested) first."""
+    return [report_key(run_id, suffix, group), legacy_report_key(run_id, suffix)]

forecasting_tools/agents_and_tools/source_archive/manifest.py

@@ -1,7 +1,8 @@
 """Per-run citation manifest: one JSONL record per (URL, citation).
 
 This is the provenance layer a bot emits and the input to the capture pipeline.
-One manifest per run, stored as ``manifests/<run_id>.jsonl`` in the blob store.
+One manifest per run, stored in the blob store under the nested ``manifests/``
+layout (see :mod:`layout`); reads fall back to the legacy flat key.
 """
 
 from __future__ import annotations
@@ -10,6 +11,7 @@
 from collections.abc import Iterable, Iterator
 from pathlib import Path
 
+from forecasting_tools.agents_and_tools.source_archive import layout
 from forecasting_tools.agents_and_tools.source_archive.canonicalize import (
     canonicalize_url,
 )
@@ -60,25 +62,39 @@ def write_file(path: str | Path, records: Iterable[CitationRecord]) -> None:
 
 
 # --- blob store io ---------------------------------------------------------
-def manifest_key(run_id: str, config: ArchiveConfig | None = None) -> str:
+def manifest_key(
+    run_id: str, config: ArchiveConfig | None = None, group: str | None = None
+) -> str:
     prefix = (config or ArchiveConfig()).s3_prefix.rstrip("/")
-    return f"{prefix}/manifests/{run_id}.jsonl"
+    return f"{prefix}/{layout.manifest_key(run_id, group)}"
 
 
 def read_blob(
-    store: BlobStore, run_id: str, config: ArchiveConfig | None = None
+    store: BlobStore,
+    run_id: str,
+    config: ArchiveConfig | None = None,
+    group: str | None = None,
 ) -> list[CitationRecord]:
-    return loads(store.get(manifest_key(run_id, config)).decode("utf-8"))
+    prefix = (config or ArchiveConfig()).s3_prefix.rstrip("/")
+    candidates = [
+        f"{prefix}/{k}" for k in layout.manifest_key_candidates(run_id, group)
+    ]
+    # Prefer the nested key; fall back to the legacy flat key for old runs.
+    for key in candidates[:-1]:
+        if store.exists(key):
+            return loads(store.get(key).decode("utf-8"))
+    return loads(store.get(candidates[-1]).decode("utf-8"))
 
 
 def write_blob(
     store: BlobStore,
     run_id: str,
     records: Iterable[CitationRecord],
     config: ArchiveConfig | None = None,
+    group: str | None = None,
 ) -> None:
     store.put(
-        manifest_key(run_id, config),
+        manifest_key(run_id, config, group),
         dumps(records).encode("utf-8"),
         content_type="application/x-ndjson",
     )