chore: prefer DB runtime reads and prune local cache

.env

@@ -71,7 +71,7 @@ CONTENT_SUPPLY_DB_PORT=3306
 CONTENT_SUPPLY_DB_NAME=content-deconstruction-supply
 CONTENT_SUPPLY_DB_USER=content_rw
 CONTENT_SUPPLY_DB_PASSWORD=bC1aH4bA1lB0
-CONTENT_AGENT_DB_RUNTIME_ENABLED=0
+CONTENT_AGENT_DB_RUNTIME_ENABLED=1
 
 # -----------------------------------------------------------------------------
 # Platform APIs: Crawapi / Douyin(V3 双渠道 抖音+视频号 均走此 host)

content_agent/integrations/composite_runtime.py

@@ -31,13 +31,22 @@ class CompositeRuntimeStore:
         return self.export.append_jsonl(run_id, filename, rows)
 
     def read_json(self, run_id: str, filename: str) -> dict[str, Any]:
-        return self.export.read_json(run_id, filename)
+        try:
+            return self.primary.read_json(run_id, filename)
+        except Exception:
+            return self.export.read_json(run_id, filename)
 
     def read_jsonl(self, run_id: str, filename: str) -> list[dict[str, Any]]:
-        return self.export.read_jsonl(run_id, filename)
+        try:
+            return self.primary.read_jsonl(run_id, filename)
+        except Exception:
+            return self.export.read_jsonl(run_id, filename)
 
     def file_status(self, run_id: str) -> dict[str, bool]:
-        return self.export.file_status(run_id)
+        try:
+            return self.primary.file_status(run_id)
+        except Exception:
+            return self.export.file_status(run_id)
 
     def create_run_record(self, record: dict[str, Any]) -> None:
         self.primary.create_run_record(record)

scripts/prune_runtime_cache.py

@@ -0,0 +1,102 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import shutil
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+
+DEFAULT_RETENTION_DAYS = 30
+
+
+@dataclass(frozen=True)
+class PruneCandidate:
+    path: Path
+    age_days: float
+    size_bytes: int
+
+
+def collect_candidates(
+    runtime_root: Path,
+    *,
+    retention_days: int = DEFAULT_RETENTION_DAYS,
+    now: float | None = None,
+) -> list[PruneCandidate]:
+    reference_time = time.time() if now is None else now
+    cutoff_seconds = retention_days * 24 * 60 * 60
+    if not runtime_root.exists():
+        return []
+    candidates: list[PruneCandidate] = []
+    for path in sorted(runtime_root.iterdir()):
+        if not path.is_dir() or not path.name.startswith("v1_run_"):
+            continue
+        age_seconds = max(0.0, reference_time - path.stat().st_mtime)
+        if age_seconds <= cutoff_seconds:
+            continue
+        candidates.append(
+            PruneCandidate(
+                path=path,
+                age_days=age_seconds / (24 * 60 * 60),
+                size_bytes=_dir_size(path),
+            )
+        )
+    return candidates
+
+
+def prune_candidates(candidates: list[PruneCandidate], *, execute: bool) -> int:
+    for candidate in candidates:
+        if execute:
+            shutil.rmtree(candidate.path)
+        print(
+            f"{'DELETE' if execute else 'DRY-RUN'}\t"
+            f"{candidate.age_days:.1f}d\t"
+            f"{_format_size(candidate.size_bytes)}\t"
+            f"{candidate.path}"
+        )
+    return len(candidates)
+
+
+def main() -> int:
+    args = _parse_args()
+    candidates = collect_candidates(
+        args.runtime_root,
+        retention_days=args.retention_days,
+    )
+    count = prune_candidates(candidates, execute=args.execute)
+    action = "deleted" if args.execute else "would_delete"
+    print(f"{action}={count} retention_days={args.retention_days} runtime_root={args.runtime_root}")
+    return 0
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Prune local runtime/v1 run directories. DB and OSS records are not touched."
+    )
+    parser.add_argument("--runtime-root", default=Path("runtime/v1"), type=Path)
+    parser.add_argument("--retention-days", default=DEFAULT_RETENTION_DAYS, type=int)
+    parser.add_argument(
+        "--execute",
+        action="store_true",
+        help="Actually delete matched run directories. Default is dry-run.",
+    )
+    return parser.parse_args()
+
+
+def _dir_size(path: Path) -> int:
+    return sum(item.stat().st_size for item in path.rglob("*") if item.is_file())
+
+
+def _format_size(size_bytes: int) -> str:
+    value = float(size_bytes)
+    for unit in ["B", "KB", "MB", "GB"]:
+        if value < 1024 or unit == "GB":
+            return f"{value:.1f}{unit}"
+        value /= 1024
+    return f"{value:.1f}GB"
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_p0d_p0g.py

@@ -57,6 +57,35 @@ def test_composite_runtime_db_failure_blocks_local_export(tmp_path):
     assert export.calls == []
 
 
+def test_composite_runtime_reads_primary_before_local_export(tmp_path):
+    primary = LocalRuntimeFileStore(tmp_path / "primary")
+    export = LocalRuntimeFileStore(tmp_path / "export")
+    primary.prepare_run("run_001")
+    export.prepare_run("run_001")
+    primary.write_json("run_001", "final_output.json", {"run_id": "run_001", "source": "db"})
+    export.write_json("run_001", "final_output.json", {"run_id": "run_001", "source": "local"})
+    primary.append_jsonl("run_001", "run_events.jsonl", [{"run_id": "run_001", "source": "db"}])
+    export.append_jsonl("run_001", "run_events.jsonl", [{"run_id": "run_001", "source": "local"}])
+    store = CompositeRuntimeStore(primary, export)
+
+    assert store.read_json("run_001", "final_output.json")["source"] == "db"
+    assert store.read_jsonl("run_001", "run_events.jsonl")[0]["source"] == "db"
+    assert store.file_status("run_001")["final_output.json"] is True
+
+
+def test_composite_runtime_falls_back_to_local_export_on_primary_read_failure(tmp_path):
+    primary = _ReadFailingRuntimeStore()
+    export = LocalRuntimeFileStore(tmp_path / "export")
+    export.prepare_run("run_001")
+    export.write_json("run_001", "final_output.json", {"run_id": "run_001", "source": "local"})
+    export.append_jsonl("run_001", "run_events.jsonl", [{"run_id": "run_001", "source": "local"}])
+    store = CompositeRuntimeStore(primary, export)
+
+    assert store.read_json("run_001", "final_output.json")["source"] == "local"
+    assert store.read_jsonl("run_001", "run_events.jsonl")[0]["source"] == "local"
+    assert store.file_status("run_001")["final_output.json"] is True
+
+
 def test_run_service_success_records_run_policy_and_lifecycle_events(tmp_path):
     runtime = _SpyRuntimeStore(tmp_path / "runtime")
     demand_source = FakeDemandSource(real_source_payload(demand_content_id=123))
@@ -477,6 +506,17 @@ class _FakeRuntimeStore(_SpyRuntimeStore):
         return self.run_dir(run_id) / filename
 
 
+class _ReadFailingRuntimeStore(_FakeRuntimeStore):
+    def read_json(self, run_id: str, filename: str) -> dict[str, Any]:
+        raise FileNotFoundError(filename)
+
+    def read_jsonl(self, run_id: str, filename: str) -> list[dict[str, Any]]:
+        raise FileNotFoundError(filename)
+
+    def file_status(self, run_id: str) -> dict[str, bool]:
+        raise FileNotFoundError(run_id)
+
+
 class _MalformedQueryVariantClient:
     def generate_variant(self, *, seed_term: str, evidence_context: dict[str, Any]):
         return None

tests/test_prune_runtime_cache.py

@@ -0,0 +1,50 @@
+import os
+import time
+from pathlib import Path
+
+from scripts.prune_runtime_cache import collect_candidates, prune_candidates
+
+
+def _touch_dir(path: Path, mtime: float) -> None:
+    path.mkdir(parents=True)
+    (path / "final_output.json").write_text("{}\n", encoding="utf-8")
+    os.utime(path, (mtime, mtime))
+
+
+def test_collect_candidates_only_matches_old_v1_run_dirs(tmp_path):
+    now = time.time()
+    old_run = tmp_path / "v1_run_old"
+    recent_run = tmp_path / "v1_run_recent"
+    other_dir = tmp_path / "notes"
+    _touch_dir(old_run, now - 31 * 24 * 60 * 60)
+    _touch_dir(recent_run, now - 2 * 24 * 60 * 60)
+    _touch_dir(other_dir, now - 60 * 24 * 60 * 60)
+
+    candidates = collect_candidates(tmp_path, retention_days=30, now=now)
+
+    assert [candidate.path for candidate in candidates] == [old_run]
+    assert candidates[0].size_bytes > 0
+
+
+def test_prune_candidates_dry_run_keeps_directories(tmp_path):
+    now = time.time()
+    old_run = tmp_path / "v1_run_old"
+    _touch_dir(old_run, now - 31 * 24 * 60 * 60)
+    candidates = collect_candidates(tmp_path, retention_days=30, now=now)
+
+    count = prune_candidates(candidates, execute=False)
+
+    assert count == 1
+    assert old_run.exists()
+
+
+def test_prune_candidates_execute_deletes_directories(tmp_path):
+    now = time.time()
+    old_run = tmp_path / "v1_run_old"
+    _touch_dir(old_run, now - 31 * 24 * 60 * 60)
+    candidates = collect_candidates(tmp_path, retention_days=30, now=now)
+
+    count = prune_candidates(candidates, execute=True)
+
+    assert count == 1
+    assert not old_run.exists()