@@ -3,8 +3,10 @@ name: Pipeline liveness
 # Weekly check that community reports are being consumed and that upstream
 # sources with a timetable are still being imported. A failure opens one issue
 # labelled pipeline-stalled, later failures comment on it, and the first passing
-# run closes it. The queue is expected to go stale between drains, so this lives
-# apart from the push workflow instead of turning it permanently red.
+# run closes it. A source within a week of going stale (exit code 3) opens or
+# updates the same issue as a warning without failing the job. The queue is
+# expected to go stale between drains, so this lives apart from the push
+# workflow instead of turning it permanently red.
 
 on:
   schedule:
@@ -38,7 +40,7 @@ jobs:
           cat liveness.txt
 
       - name: Open or update the tracking issue
-        if: steps.check.outputs.status != '0'
+        if: steps.check.outputs.status != '0' && steps.check.outputs.status != '3'
         env:
           GH_TOKEN: ${{ github.token }}
         run: |
@@ -52,6 +54,20 @@ jobs:
           fi
           exit 1
 
+      - name: Warn about a source close to stale
+        if: steps.check.outputs.status == '3'
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          body="$(printf 'The weekly pipeline check passed on %s, with a warning: a source goes stale within a week.\n\n```\n%s\n```\n' "$(date -u +%Y-%m-%d)" "$(cat liveness.txt)")"
+          number="$(gh issue list --state open --label pipeline-stalled --json number --jq '.[0].number // empty')"
+          if [ -n "$number" ]; then
+            gh issue comment "$number" --body "$body"
+          else
+            gh label create pipeline-stalled --color B60205 --description "The weekly pipeline check is failing" 2>/dev/null || true
+            gh issue create --title "Report pipeline warning: a source goes stale within a week" --label pipeline-stalled --body "$body"
+          fi
+
       - name: Close the tracking issue once healthy
         if: steps.check.outputs.status == '0'
         env:
 
@@ -137,6 +137,10 @@ All notable changes to CallShield will be documented in this file.
   Singapore and Thailand. You can turn on "Allow apps from unverified
   developers" in Developer options and wait a day, or install over USB with
   `adb install`.
+- The weekly pipeline check warns a week before an upstream source goes
+  stale, without failing, so an overdue FCC or FTC import shows up while
+  there is still time to run it. FTC went stale on 2026-10-07 and nothing
+  said so until the next weekly run would have failed.
 
 ## v1.11.0 (2026-09-30)
 
 
@@ -29,7 +29,8 @@
 With `--scheduled` (the weekly workflow) the age check measures against the
 current time instead of the database date, upstream sources with a regular
 cadence are checked against `stale_after_days`, and published rows whose
-evidence expires within a month are reported. Both are left out of
+evidence expires within a month are reported. A source within a week of its
+limit is a warning, exit code 3, which the workflow reports without failing. Both are left out of
 `verifyPipelineTests`: that task runs inside `check`, where a clock would fail
 every later build of an old tag. Measured against the database date, the age
 check cannot fire in the weeks after a drain, because every report queued since
@@ -76,6 +77,12 @@
 REGULAR_CADENCES = frozenset({"daily", "weekly"})
 # How far ahead the weekly run looks for downloaded rows whose evidence runs out.
 EVIDENCE_EXPIRY_WARNING_DAYS = 30
+# A source this many days from its stale_after_days limit gets a warning. The
+# run is weekly, so a source that only failed once it lapsed could sit stale
+# for up to a week first (FTC lapsed 2026-10-07 and the next run was 10-12).
+SOURCE_STALE_WARNING_DAYS = 7
+# main()'s exit code when nothing failed but a source is close to its limit.
+EXIT_WARNING = 3
 
 
 def _parse_updated(value: object) -> datetime | None:
@@ -190,14 +197,29 @@ def evaluate_source_freshness(manifest: object, freshness: object, now: datetime
     A manifest this can't read is a failure rather than a pass: a broken
     manifest would otherwise switch the whole check off.
     """
+    return _source_freshness(manifest, freshness, now)[0]
+
+
+def warn_source_staleness(manifest: object, freshness: object, now: datetime) -> list[str]:
+    """One warning per regular-cadence source within SOURCE_STALE_WARNING_DAYS of its limit.
+
+    The source is still fresh, so the weekly run reports it without failing
+    (EXIT_WARNING), and a source past its limit is evaluate_source_freshness's.
+    """
+    return _source_freshness(manifest, freshness, now)[1]
+
+
+def _source_freshness(manifest: object, freshness: object, now: datetime) -> tuple[list[str], list[str]]:
+    """The failures and the warnings for every regular-cadence source."""
     sources = manifest.get("sources") if isinstance(manifest, dict) else None
     if not isinstance(sources, list):
-        return [f"{MANIFEST_FILE.name} is missing or unreadable, so upstream freshness can't be checked"]
+        return [f"{MANIFEST_FILE.name} is missing or unreadable, so upstream freshness can't be checked"], []
     recorded = freshness.get("last_success") if isinstance(freshness, dict) else None
     last_success = recorded if isinstance(recorded, dict) else {}
     recorded_dates = freshness.get("newest_record_date") if isinstance(freshness, dict) else None
     newest_record_date = recorded_dates if isinstance(recorded_dates, dict) else {}
     problems: list[str] = []
+    warnings: list[str] = []
     for source in sources:
         if not isinstance(source, dict):
             problems.append(f"{MANIFEST_FILE.name} has a source entry that isn't an object")
@@ -222,13 +244,15 @@ def evaluate_source_freshness(manifest: object, freshness: object, now: datetime
                     f"{source_id} ({cadence} source) has no {'record date' if record_dated else 'successful import'} recorded in "
                     f"{FRESHNESS_FILE.name} - {refresh}"
                 )
-        elif now - stamp > timedelta(days=limit):
-            age_description = "newest record is from" if record_dated else "was last imported"
-            problems.append(
-                f"{source_id} {age_description} {stamp.date().isoformat()}, {(now - stamp).days} days ago, "
-                f"past its {limit:g}-day limit - {refresh}"
-            )
-    return problems
+            continue
+        age_description = "newest record is from" if record_dated else "was last imported"
+        age = f"{source_id} {age_description} {stamp.date().isoformat()}, {(now - stamp).days} days ago"
+        if now - stamp > timedelta(days=limit):
+            problems.append(f"{age}, past its {limit:g}-day limit - {refresh}")
+        elif now - stamp >= timedelta(days=limit - SOURCE_STALE_WARNING_DAYS):
+            stale_on = (stamp + timedelta(days=limit)).date().isoformat()
+            warnings.append(f"{age}, and passes its {limit:g}-day limit after {stale_on} - {refresh}")
+    return problems, warnings
 
 
 def row_expiry_epoch_ms(row: dict) -> int | None:
@@ -359,14 +383,22 @@ def main(argv: list[str] | None = None) -> int:
     reports, unreadable = load_queue(REPORTS_DIR)
     total = len(reports) + unreadable
     problems = evaluate_queue_health(reports, load_database_updated(DB_FILE), unreadable, now=now)
+    warnings: list[str] = []
     if now is not None:
-        problems += evaluate_source_freshness(_load_json(MANIFEST_FILE), _load_json(FRESHNESS_FILE), now)
+        manifest, freshness = _load_json(MANIFEST_FILE), _load_json(FRESHNESS_FILE)
+        problems += evaluate_source_freshness(manifest, freshness, now)
         problems += evaluate_evidence_expiry(_load_json(DB_FILE), now)
+        warnings = warn_source_staleness(manifest, freshness, now)
     if problems:
         print(f"Report pipeline is not healthy ({total} queued file(s) in {REPORTS_DIR}):", file=sys.stderr)
-        for problem in problems:
+        for problem in problems + [f"warning: {warning}" for warning in warnings]:
             print(f"  - {problem}", file=sys.stderr)
         return 1
+    if warnings:
+        print(f"Report pipeline warning, a source is close to stale ({total} queued file(s) in {REPORTS_DIR}):")
+        for warning in warnings:
+            print(f"  - {warning}")
+        return EXIT_WARNING
     suffix = f", {unreadable} unreadable" if unreadable else ""
     print(f"Report queue liveness OK ({total} file(s) pending{suffix}).")
     return 0
 
@@ -13,10 +13,13 @@
     MIN_BUCKET_SAMPLE,
     evaluate_evidence_expiry,
     evaluate_queue_health,
+    EXIT_WARNING,
+    SOURCE_STALE_WARNING_DAYS,
     evaluate_source_freshness,
     load_database_updated,
     load_queue,
     row_expiry_epoch_ms,
+    warn_source_staleness,
 )
 
 BUCKET = "0123456789abcdef"
@@ -234,8 +237,84 @@ def scheduled_checks() -> None:
     problems = evaluate_source_freshness(real_manifest, default_import, BASE)
     assert len(problems) == 1 and problems[0].startswith("fcc_complaints newest record is from 2026-08-01"), problems
 
+    source_warning_checks()
     evidence_expiry_checks()
     check_scheduled_switch()
+    check_scheduled_warning()
+
+
+def source_warning_checks() -> None:
+    """A source a week from its limit warns without failing; past the limit it fails instead."""
+    assert SOURCE_STALE_WARNING_DAYS == 7
+    manifest = {
+        "sources": [
+            {"id": "ftc_complaints", "cadence": "daily", "stale_after_days": 14},
+            {"id": "toastedspam", "cadence": "weekly", "stale_after_days": 30, "import_flag": "--allow-insecure-sources"},
+        ]
+    }
+
+    def ftc_aged(days: float) -> dict:
+        return {"newest_record_date": {"ftc_complaints": (BASE - timedelta(days=days)).isoformat()}}
+
+    assert warn_source_staleness(manifest, ftc_aged(6), BASE) == []
+    for days in (7, 13, 14):
+        warnings = warn_source_staleness(manifest, ftc_aged(days), BASE)
+        assert len(warnings) == 1, (days, warnings)
+        assert warnings[0].startswith(f"ftc_complaints newest record is from {(BASE - timedelta(days=days)).date().isoformat()}"), warnings
+        if days == 14:
+            assert warnings[0].endswith("limit after 2026-09-04 - run scripts/import_all_sources.py"), warnings
+        assert evaluate_source_freshness(manifest, ftc_aged(days), BASE) == [], days
+    stale = ftc_aged(15)
+    assert warn_source_staleness(manifest, stale, BASE) == []
+    assert len(evaluate_source_freshness(manifest, stale, BASE)) == 1
+
+    # A 30-day source warns from day 23, by its last import, naming its flag.
+    imported = ftc_aged(0)
+    imported["last_success"] = {"toastedspam": (BASE - timedelta(days=22)).isoformat()}
+    assert warn_source_staleness(manifest, imported, BASE) == []
+    imported["last_success"]["toastedspam"] = (BASE - timedelta(days=23)).isoformat()
+    warnings = warn_source_staleness(manifest, imported, BASE)
+    assert len(warnings) == 1 and warnings[0].startswith("toastedspam was last imported"), warnings
+    assert warnings[0].endswith("--allow-insecure-sources"), warnings
+
+    # A broken manifest stays a failure and adds no warnings.
+    assert warn_source_staleness(None, ftc_aged(10), BASE) == []
+
+
+def check_scheduled_warning() -> None:
+    """A source close to its limit exits with EXIT_WARNING and says so; nothing else fails."""
+    import contextlib
+    import io
+
+    import pipeline_liveness as liveness
+
+    saved = (liveness.REPORTS_DIR, liveness.DB_FILE, liveness.MANIFEST_FILE, liveness.FRESHNESS_FILE)
+    with tempfile.TemporaryDirectory() as directory:
+        data = Path(directory)
+        (data / "reports").mkdir()
+        today = datetime.now(timezone.utc)
+        (data / "spam_numbers.json").write_text(json.dumps({"updated": today.date().isoformat(), "numbers": []}), encoding="utf-8")
+        (data / "source-manifest.json").write_text(
+            json.dumps({"sources": [{"id": "ftc_complaints", "cadence": "daily", "stale_after_days": 14}]}),
+            encoding="utf-8",
+        )
+        aged = (today - timedelta(days=9)).date().isoformat()
+        (data / "source-freshness.json").write_text(json.dumps({"newest_record_date": {"ftc_complaints": aged}}), encoding="utf-8")
+        liveness.REPORTS_DIR = data / "reports"
+        liveness.DB_FILE = data / "spam_numbers.json"
+        liveness.MANIFEST_FILE = data / "source-manifest.json"
+        liveness.FRESHNESS_FILE = data / "source-freshness.json"
+        try:
+            output = io.StringIO()
+            with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
+                assert liveness.main(["--scheduled"]) == EXIT_WARNING, output.getvalue()
+            assert "close to stale" in output.getvalue() and f"newest record is from {aged}" in output.getvalue(), output.getvalue()
+
+            # Without --scheduled there's no clock, so nothing to warn about.
+            with contextlib.redirect_stdout(io.StringIO()):
+                assert liveness.main([]) == 0
+        finally:
+            liveness.REPORTS_DIR, liveness.DB_FILE, liveness.MANIFEST_FILE, liveness.FRESHNESS_FILE = saved
 
 
 def evidence_row(expires_in_days: float | None, *, number: str = "+15125550100") -> dict: