fetch details incrementally on listing changes or weekly refresh

docs/SSH_RUN.md

@@ -183,6 +183,8 @@ COLLECT_FAMILY=saab-900
 COLLECT_MAX_PAGES=1
 COLLECT_MAX_DETAILS=all
 COLLECT_DELAY=5
+COLLECT_DETAIL_POLICY=incremental
+COLLECT_DETAIL_REFRESH_DAYS=7
 ```
 
 The SSH wrapper reads resolved Compose settings and supplies a validated, fixed
@@ -202,6 +204,45 @@ increase runtime; the Actions timeout remains 45 minutes. The pipeline and local
 state still retain exit 1 for partial coverage; the SSH wrapper converts verified
 limit-only outcomes into an Actions warning/success. No run is launched by editing app.env.
 
+### Incremental details
+
+`COLLECT_DETAIL_POLICY=incremental` fetches detail pages only when:
+
+1. There is no successfully imported detail observation for the ad.
+2. The search-visible listing data changed since the search accompanying its last
+   successful detail fetch (price, source update timestamp, title/description,
+   vehicle attributes, location, status/category and publication/expiration dates).
+3. Its last imported detail is at least `COLLECT_DETAIL_REFRESH_DAYS` old (default 7).
+
+Counters, promotion/boost flags, attribute order and excluded personal attributes
+do not trigger a refresh. Detail-only changes invisible in search results are
+detected only on the age-based refresh. A weekly refresh is an eligibility rule
+for the next manually requested run, not a weekly schedule.
+
+History comes from accepted observations in committed PostgreSQL imports. The
+baseline search is the latest search at/before the last successful detail, not
+the most recent search overall: failed enrichment cannot acknowledge a change.
+Missing baselines conservatively trigger a request. Cache skips are recorded as
+`not_attempted` with reason `fresh_unchanged`; they do not create a detail observation,
+advance freshness, consume max-details, or claim a new detail was fetched. Such
+intentional skips may leave raw coverage partial but produce the existing Actions
+warning/success when there are no actual failures or rejected records.
+
+The pipeline loads a read-only history snapshot before collection. Failure to load
+it prevents collection rather than silently falling back to fetching everything.
+Standalone `car-tracker` collection rejects incremental mode; use the database-backed
+`car-tracker-pipeline`. No schema migration is required, and existing import/replay
+semantics are unchanged. For this homelab-sized dataset the snapshot contains the
+latest accepted detail and preceding search per listing.
+
+**Activate only after a new image supporting these flags is published.** Install
+the updated host_run.py and compose.runtime.yaml, then add the two policy variables
+to host app.env and select that new release. For older images use
+`COLLECT_DETAIL_POLICY=always`; defaults remain always for backward compatibility.
+COLLECT_MAX_DETAILS=all still means all eligible ads, not all discovered ads.
+Existing details older than seven days will be eligible on the first incremental
+run, so this does not guarantee fewer requests immediately or eliminate 403s.
+
 ### Execution outcomes
 
 Local outcomes: `~/containers/car-tracker/state/requests/<uuid>.json`.

ops/app/compose.runtime.yaml

@@ -34,6 +34,8 @@ services:
       COLLECT_MAX_PAGES: "${COLLECT_MAX_PAGES:-1}"
       COLLECT_MAX_DETAILS: "${COLLECT_MAX_DETAILS:-1}"
       COLLECT_DELAY: "${COLLECT_DELAY:-5}"
+      COLLECT_DETAIL_POLICY: "${COLLECT_DETAIL_POLICY:-always}"
+      COLLECT_DETAIL_REFRESH_DAYS: "${COLLECT_DETAIL_REFRESH_DAYS:-7}"
     extra_hosts:
       - "host.docker.internal:host-gateway"
     volumes:

ops/ssh/host_run.py

@@ -38,9 +38,9 @@ def limit_reasons(manifest, pipeline):
     for item in details.get("listings", {}).values():
         if item.get("status") == "complete":
             continue
-        if item.get("status") != "not_attempted" or item.get("reason") != "detail_limit":
+        if item.get("status") != "not_attempted" or item.get("reason") not in ("detail_limit", "fresh_unchanged"):
             return []
-        reasons.add("detail_limit")
+        reasons.add(item["reason"])
     return sorted(reasons)
 
 
@@ -59,12 +59,13 @@ def outcome_summary(manifest, pipeline):
     failed = sum(item.get("status") == "failed" for item in listings)
     skipped = sum(item.get("status") == "not_attempted" for item in listings)
     stopped = sum(item.get("reason") == "run_stopped" for item in listings)
+    cached = sum(item.get("reason") == "fresh_unchanged" for item in listings)
     lines = [
         f"Collection: {status(manifest.get('status'))}. Import: {status(pipeline.get('import_status'))}.",
         f"Import records: accepted={count(summary.get('accepted'))}, rejected={count(summary.get('rejected'))}, "
         f"field issues={count(summary.get('field_issues'))}.",
         f"Details: completed={completed}, failed={failed}, not attempted={skipped} "
-        f"(collection stopped early={stopped}).",
+        f"(fresh unchanged details reused={cached}, collection stopped early={stopped}).",
     ]
     items = [*manifest.get("searches", []), *listings]
     http = sorted({item["http_status"] for item in items
@@ -132,7 +133,8 @@ def actions_outcome(code, request, state, raw_root):
                 print("::error::" + " ".join(lines) + " See private host logs/manifest; no automatic retry.", flush=True)
             return code
         explanations = {"page_limit": "the configured page limit left additional search pages unfetched",
-                        "detail_limit": "the configured detail limit left some discovered ads without detail attempts"}
+                        "detail_limit": "the configured detail limit left some discovered ads without detail attempts",
+                        "fresh_unchanged": "unchanged ads with fresh previously imported details were intentionally not fetched again"}
         message = "; ".join(explanations[reason] for reason in reasons)
         print(f"::warning::Import succeeded with no rejected records, but coverage is intentionally partial: {message}. "
               "This is not full source coverage. Host state retains pipeline exit 1.", flush=True)
@@ -147,6 +149,12 @@ def collection_command(settings):
     pages = settings.get("COLLECT_MAX_PAGES", "1")
     details = settings.get("COLLECT_MAX_DETAILS", "1")
     delay = settings.get("COLLECT_DELAY", "5")
+    policy = settings.get("COLLECT_DETAIL_POLICY", "always")
+    refresh = settings.get("COLLECT_DETAIL_REFRESH_DAYS", "7")
+    if policy not in ("always", "incremental"):
+        raise ValueError("Detail policy must be always or incremental")
+    if not isinstance(refresh, str) or not refresh.isdecimal() or int(refresh) < 1:
+        raise ValueError("Detail refresh days must be positive")
     if not isinstance(family, str) or not family.strip():
         raise ValueError("A collection family or all is required")
     if not isinstance(pages, str) or not pages.isdecimal() or int(pages) < 1:
@@ -158,6 +166,8 @@ def collection_command(settings):
         raise ValueError("Delay must be finite and at least one second")
     args = ["car-tracker-pipeline", "--headed", "--config", "/app/searches.toml",
             "--output", "/data", "--max-pages", pages, "--delay", delay]
+    if policy == "incremental":
+        args.extend(["--detail-policy", policy, "--detail-refresh-days", refresh])
     if family != "all":
         args.extend(["--family", family])
     if details != "all":

src/car_tracker/collect.py

@@ -114,7 +114,7 @@ def write_records(path, records):
     temporary.replace(path)
 
 
-def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, max_details=None):
+def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, max_details=None, detail_policy=None):
     run_id = str(uuid.uuid4())
     directory = output / datetime.now(timezone.utc).strftime("%Y-%m-%d") / run_id
     directory.mkdir(mode=0o700, parents=True, exist_ok=False)
@@ -126,6 +126,8 @@ def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, m
         "planned_searches": searches,
         "details": {
             "enabled": not search_only, "max_details": max_details,
+            "policy": "incremental" if detail_policy is not None else "always",
+            "refresh_days": getattr(detail_policy, "refresh_days", None),
             "status": "disabled" if search_only else "running", "listings": {},
         },
     }
@@ -192,6 +194,12 @@ def collect(fetch, searches, output, max_pages=10, delay=5, search_only=False, m
                         item = details[identifier]
                         if item["status"] != "pending":
                             continue
+                        if detail_policy is not None:
+                            decision = detail_policy(ad)
+                            item["detail_decision"] = decision
+                            if decision == "fresh_unchanged":
+                                item.update(status="not_attempted", reason="fresh_unchanged")
+                                continue
                         if max_details is not None and detail_attempts >= max_details:
                             item.update(status="not_attempted", reason="detail_limit")
                             continue
@@ -280,11 +288,15 @@ def parse_collection_args(argv=None, description=None):
     parser.add_argument("--headed", action="store_true")
     parser.add_argument("--search-only", action="store_true", help="Skip individual listing pages")
     parser.add_argument("--max-details", type=int, help="Maximum unique detail requests per run")
+    parser.add_argument("--detail-policy", choices=("always", "incremental"), default="always")
+    parser.add_argument("--detail-refresh-days", type=int, default=7)
     args = parser.parse_args(argv)
     if args.max_pages < 1 or not math.isfinite(args.delay) or args.delay < 1:
         parser.error("max-pages must be positive and delay must be finite and >= 1 second")
     if args.max_details is not None and args.max_details < 1:
         parser.error("max-details must be positive")
+    if args.detail_refresh_days < 1:
+        parser.error("detail-refresh-days must be positive")
     with args.config.open("rb") as stream:
         searches = tomllib.load(stream).get("searches", [])
     if not isinstance(searches, list) or not searches or any(
@@ -330,6 +342,7 @@ def collect_browser(args):
                 result = collect(
                     fetch, args.searches, args.output, args.max_pages, args.delay,
                     args.search_only, args.max_details,
+                    detail_policy=getattr(args, "detail_decider", None),
                 )
             finally:
                 browser.close()
@@ -350,6 +363,10 @@ def collect_browser(args):
 
 
 def main():
-    path, status = collect_browser(parse_collection_args())
+    args = parse_collection_args()
+    if args.detail_policy == "incremental":
+        print("Incremental details require car-tracker-pipeline and imported database history.", file=sys.stderr)
+        return 3
+    path, status = collect_browser(args)
     print(f"{status}: {path}")
     return 0 if status == "complete" else 1

src/car_tracker/detail_policy.py

@@ -0,0 +1,67 @@
+"""Choose detail requests from successfully imported history, not failed attempts."""
+
+from datetime import datetime, timedelta, timezone
+import json
+
+from .collect import minimize
+
+
+CHANGE_FIELDS = {
+    "subject", "body", "price", "price_cents", "brand", "status", "ad_type",
+    "category_id", "category_name", "attributes", "location", "index_date",
+    "first_publication_date", "expiration_date",
+}
+
+
+def signature(ad):
+    value = {key: item for key, item in minimize(ad).items() if key in CHANGE_FIELDS}
+    if "attributes" in value:
+        value["attributes"] = sorted(value["attributes"], key=lambda item: json.dumps(item, sort_keys=True))
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+class DetailPolicy:
+    def __init__(self, history, refresh_days=7, clock=None):
+        self.history = history
+        self.refresh_days = refresh_days
+        self.clock = clock or (lambda: datetime.now(timezone.utc))
+
+    def __call__(self, ad):
+        previous = self.history.get(str(ad["list_id"]))
+        if previous is None:
+            return "never_detailed"
+        observed_at, search = previous
+        current = self.clock()
+        if observed_at > current:
+            return "invalid_history_time"
+        if current - observed_at >= timedelta(days=self.refresh_days):
+            return "weekly_refresh"
+        if search is None:
+            return "missing_search_baseline"
+        if signature(ad) != signature(search):
+            return "search_changed"
+        return "fresh_unchanged"
+
+
+def load_policy(conn, refresh_days=7):
+    # A detail must have an accepted observation from a committed import. Pair it
+    # with the search observed before that detail, NOT the latest search overall:
+    # otherwise a failed enrichment could incorrectly acknowledge a change.
+    rows = conn.execute("""
+        WITH details AS (
+            SELECT DISTINCT ON (source_listing_id)
+                source_listing_id, observed_at, observation_id
+            FROM silver.observations
+            WHERE source = 'leboncoin' AND record_type = 'detail'
+            ORDER BY source_listing_id, observed_at DESC, observation_id DESC
+        )
+        SELECT d.source_listing_id, d.observed_at, s.raw_payload
+        FROM details d
+        LEFT JOIN LATERAL (
+            SELECT raw_payload FROM silver.observations
+            WHERE source = 'leboncoin' AND source_listing_id = d.source_listing_id
+                AND record_type = 'search' AND observed_at <= d.observed_at
+            ORDER BY observed_at DESC, observation_id DESC LIMIT 1
+        ) s ON true
+    """).fetchall()
+    return DetailPolicy({identifier: (observed_at, search) for identifier, observed_at, search in rows}, refresh_days)

src/car_tracker/pipeline.py

@@ -12,6 +12,7 @@ import psycopg
 
 from .collect import collect_browser, parse_collection_args, write_json
 from .database import InputError, database_error_hint, import_run
+from .detail_policy import load_policy
 
 
 def run_pipeline(args):
@@ -35,6 +36,8 @@ def run_pipeline(args):
             try:
                 with psycopg.connect(conninfo, connect_timeout=10, autocommit=True) as conn:
                     conn.execute("SELECT 1 FROM silver.collection_runs LIMIT 0")
+                    if getattr(args, "detail_policy", "always") == "incremental":
+                        args.detail_decider = load_policy(conn, args.detail_refresh_days)
                 path, status = collect_browser(args)
                 print(f"Manifest: {path}", flush=True)
                 sidecar = path.parent / "pipeline.json"

tests/test_collect.py

@@ -150,6 +150,20 @@ class CollectorTests(unittest.TestCase):
         self.run_collection([SourceAccessError(429, capture=capture)])
         capture.assert_not_called()
 
+    def test_fresh_details_are_intentionally_skipped_without_requests(self):
+        path, manifest = self.run_collection([payload([123, 456], total=2)], search_only=False,
+                                            detail_policy=lambda ad: "fresh_unchanged")
+        self.assertEqual(manifest["details"]["attempted"], 0)
+        self.assertEqual(manifest["details"]["listings"]["123"]["reason"], "fresh_unchanged")
+        self.assertEqual(len(list(path.parent.glob("*.jsonl.gz"))), 1)
+
+    def test_fresh_skips_do_not_consume_detail_request_limit(self):
+        _, manifest = self.run_collection([payload([123, 456], total=2), detail_payload(456)],
+                                         search_only=False, max_details=1,
+                                         detail_policy=lambda ad: "fresh_unchanged" if ad["list_id"] == 123 else "never_detailed")
+        self.assertEqual(manifest["details"]["attempted"], 1)
+        self.assertEqual(manifest["details"]["listings"]["456"]["status"], "complete")
+
     def test_discovery_failure_does_not_claim_detail_completion(self):
         _, manifest = self.run_collection([SourceAccessError(403)], search_only=False)
         self.assertEqual(manifest["details"]["status"], "not_attempted")

tests/test_database_integration.py

@@ -21,6 +21,7 @@ from psycopg.conninfo import conninfo_to_dict
 
 from car_tracker import database
 from car_tracker.collect import search_url, write_json, write_records
+from car_tracker.detail_policy import load_policy
 
 
 @unittest.skipUnless(os.environ.get("TEST_DATABASE_URL"), "TEST_DATABASE_URL is not set")
@@ -128,6 +129,36 @@ class DatabaseIntegrationTests(unittest.TestCase):
         database.migrate(self.conn)
         self.assertEqual(self.state(), before)
 
+    def test_incremental_policy_pairs_search_with_last_successful_detail(self):
+        path, _ = self.fixture(detail={"list_id": 123, "price": ["12500"]}, day=1)
+        database.import_run(self.conn, path)
+        later, _ = self.fixture(pages=[[{"list_id": 123, "price": ["11000"]}]], day=2)
+        database.import_run(self.conn, later)
+        before = self.state()
+        policy = load_policy(self.conn)
+        policy.clock = lambda: datetime(2026, 1, 3, tzinfo=timezone.utc)
+        self.assertEqual(policy({"list_id": 123, "price": ["12000"]}), "fresh_unchanged")
+        self.assertEqual(policy({"list_id": 123, "price": ["11000"]}), "search_changed")
+        self.assertEqual(policy({"list_id": 456, "price": ["10000"]}), "never_detailed")
+        policy.clock = lambda: datetime(2026, 1, 9, tzinfo=timezone.utc)
+        self.assertEqual(policy({"list_id": 123, "price": ["12000"]}), "weekly_refresh")
+        self.assertEqual(self.state(), before)
+
+    def test_cached_skip_import_preserves_detail_age_and_reason(self):
+        first, _ = self.fixture(detail={"list_id": 123, "price": ["12000"]}, day=1)
+        database.import_run(self.conn, first)
+        path, manifest = self.fixture(day=2)
+        manifest["status"] = "partial"
+        manifest["details"] = {"enabled": True, "status": "partial", "listings": {
+            "123": {"status": "not_attempted", "reason": "fresh_unchanged"}}}
+        write_json(path, manifest)
+        database.import_run(self.conn, path)
+        self.assertEqual(self.conn.execute(
+            "SELECT reason FROM silver.detail_coverage WHERE run_id = %s", (manifest["run_id"],)
+        ).fetchone()[0], "fresh_unchanged")
+        policy = load_policy(self.conn)
+        self.assertEqual(policy.history["123"][0].day, 1)
+
     def test_applied_migration_checksum_mismatch_rolls_back(self):
         before = self.state()
         # The outer transaction also rolls back the deliberate checksum corruption.

tests/test_detail_policy.py

@@ -0,0 +1,53 @@
+from datetime import datetime, timedelta, timezone
+import unittest
+from unittest.mock import Mock
+
+from car_tracker.detail_policy import DetailPolicy, load_policy, signature
+
+
+class DetailPolicyTests(unittest.TestCase):
+    def setUp(self):
+        self.current = datetime(2026, 10, 12, 12, tzinfo=timezone.utc)
+        self.ad = {"list_id": 123, "price": [12000], "subject": "Synthetic car",
+                   "attributes": [{"key": "mileage", "value": "100000"}]}
+
+    def policy(self, age=1, baseline=True):
+        return DetailPolicy({"123": (self.current - timedelta(days=age), self.ad if baseline else None)},
+                            clock=lambda: self.current)
+
+    def test_no_successful_detail_is_eligible(self):
+        self.assertEqual(DetailPolicy({}, clock=lambda: self.current)(self.ad), "never_detailed")
+
+    def test_unchanged_fresh_ad_is_skipped(self):
+        self.assertEqual(self.policy()(self.ad), "fresh_unchanged")
+
+    def test_exact_week_boundary_is_eligible(self):
+        self.assertEqual(self.policy(age=7)(self.ad), "weekly_refresh")
+        self.assertEqual(self.policy(age=6)(self.ad), "fresh_unchanged")
+
+    def test_price_and_source_timestamp_changes_are_eligible(self):
+        self.assertEqual(self.policy()({**self.ad, "price": [11000]}), "search_changed")
+        self.assertEqual(self.policy()({**self.ad, "index_date": "2026-10-12"}), "search_changed")
+
+    def test_unknown_baseline_and_future_timestamp_are_conservative(self):
+        self.assertEqual(self.policy(baseline=False)(self.ad), "missing_search_baseline")
+        self.assertEqual(self.policy(age=-1)(self.ad), "invalid_history_time")
+
+    def test_counters_and_promotions_do_not_trigger_enrichment(self):
+        other = {**self.ad, "counters": {"views": 123}, "is_boosted": True,
+                 "options": {"highlight": True}, "owner": {"name": "private"}}
+        self.assertEqual(self.policy()(other), "fresh_unchanged")
+
+    def test_attribute_order_and_sensitive_fields_do_not_trigger_changes(self):
+        a = {**self.ad, "attributes": [*self.ad["attributes"], {"key": "fuel", "value": "petrol"}]}
+        b = {**a, "attributes": [*reversed(a["attributes"]), {"key": "phone", "value": "secret"}]}
+        self.assertEqual(signature(a), signature(b))
+
+    def test_loads_only_imported_history(self):
+        conn = Mock()
+        conn.execute.return_value.fetchall.return_value = [("123", self.current, self.ad)]
+        policy = load_policy(conn)
+        self.assertEqual(policy.history["123"], (self.current, self.ad))
+        query = conn.execute.call_args.args[0]
+        self.assertIn("observed_at <= d.observed_at", query)
+        self.assertIn("record_type = 'detail'", query)

tests/test_pipeline.py

@@ -103,6 +103,21 @@ class PipelineTests(unittest.TestCase):
         self.preflight.__exit__.assert_called_once()
         self.assertNotIn("TOP_SECRET", self.stderr.getvalue())
 
+    def test_incremental_history_is_loaded_before_collection(self):
+        self.args.detail_policy = "incremental"
+        self.args.detail_refresh_days = 7
+        with patch.object(pipeline, "load_policy", return_value="synthetic-policy") as load:
+            self.assertEqual(pipeline.run_pipeline(self.args), 0)
+        load.assert_called_once_with(self.preflight, 7)
+        self.assertEqual(self.args.detail_decider, "synthetic-policy")
+
+    def test_history_failure_does_not_start_collection(self):
+        self.args.detail_policy = "incremental"
+        self.args.detail_refresh_days = 7
+        with patch.object(pipeline, "load_policy", side_effect=pipeline.psycopg.OperationalError("secret")):
+            self.assertEqual(pipeline.run_pipeline(self.args), 3)
+        self.collect.assert_not_called()
+
     def test_finalized_interruption_is_not_imported(self):
         self.collect.side_effect = lambda args: self.fixture(args, "partial", "KeyboardInterrupt")
         self.assertEqual(pipeline.run_pipeline(self.args), 130)

tests/test_ssh_run.py

@@ -97,6 +97,22 @@ class SSHRunTests(unittest.TestCase):
         self.assertEqual(args[args.index("--max-details") + 1], "20")
         self.assertEqual(args[args.index("--max-pages") + 1], "3")
 
+    def test_incremental_policy_is_host_configurable_and_validated(self):
+        args = host.collection_command({"COLLECT_DETAIL_POLICY": "incremental", "COLLECT_DETAIL_REFRESH_DAYS": "7"})
+        self.assertEqual(args[args.index("--detail-policy") + 1], "incremental")
+        self.assertEqual(args[args.index("--detail-refresh-days") + 1], "7")
+        self.assertNotIn("--detail-policy", host.collection_command({}))
+        for settings in ({"COLLECT_DETAIL_POLICY": "unknown"}, {"COLLECT_DETAIL_REFRESH_DAYS": "0"}):
+            with self.assertRaises(ValueError):
+                host.collection_command(settings)
+
+    def test_fresh_unchanged_skips_are_not_source_failures(self):
+        manifest, pipeline = self.bounded_outcome()
+        manifest["searches"][0] = {"status": "complete"}
+        manifest["details"]["listings"]["1"] = {"status": "not_attempted", "reason": "fresh_unchanged"}
+        self.assertEqual(host.limit_reasons(manifest, pipeline), ["fresh_unchanged"])
+        self.assertIn("fresh unchanged details reused=1", " ".join(host.outcome_summary(manifest, pipeline)))
+
     def test_invalid_settings_are_refused(self):
         for key, value in [("COLLECT_MAX_DETAILS", "0"), ("COLLECT_MAX_DETAILS", "-1"),
                            ("COLLECT_MAX_DETAILS", "all; id"), ("COLLECT_MAX_PAGES", "all"),