fix: guard weekly command center freshness

src/cli.py

@@ -2388,7 +2388,7 @@ def _run_control_center_mode(args, parser) -> None:
         report_path,
         _parse_iso_dt(normalized.get("generated_at")) or datetime.now(timezone.utc),
     )
-    json_artifact, md_artifact, weekly_json, weekly_md, _payload = _write_control_center_artifacts(
+    json_artifact, md_artifact, weekly_json, weekly_md, payload = _write_control_center_artifacts(
         normalized,
         snapshot,
         output_dir,
@@ -2397,7 +2397,17 @@ def _run_control_center_mode(args, parser) -> None:
         report_reference=str(report_path),
         diff_dict=diff_dict,
     )
-    _print_control_center_summary(snapshot)
+    weekly_digest = payload.get("weekly_command_center_digest_v1", {})
+    source_freshness = weekly_digest.get("source_freshness", {})
+    if source_freshness.get("status") and source_freshness.get("status") != "current":
+        print_info(weekly_digest.get("headline") or "Refresh the audit report before acting.")
+        print_info(
+            source_freshness.get("summary")
+            or "Control-center source freshness could not be proven."
+        )
+        print_info(weekly_digest.get("decision") or "Refresh the audit report, then rerun.")
+    else:
+        _print_control_center_summary(snapshot)
     print_info(f"Control center JSON: {json_artifact}")
     print_info(f"Control center Markdown: {md_artifact}")
     print_info(f"Weekly command center JSON: {weekly_json}")

src/weekly_command_center.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
@@ -30,6 +30,44 @@ def _mapping(value: Any) -> dict[str, Any]:
     return {}
 
 
+def _parse_datetime(value: Any) -> datetime | None:
+    text = _safe_text(value)
+    if not text:
+        return None
+    try:
+        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _source_freshness(report_data: dict[str, Any], portfolio_truth: dict[str, Any]) -> dict[str, Any]:
+    report_generated_at = _parse_datetime(report_data.get("generated_at"))
+    truth_generated_at = _parse_datetime(portfolio_truth.get("generated_at"))
+    status = "current"
+    summary = "Control-center source report and portfolio truth are aligned enough to read together."
+    if truth_generated_at and report_generated_at and truth_generated_at > report_generated_at:
+        status = "portfolio-truth-newer"
+        summary = (
+            "Portfolio truth is newer than the audit report feeding the control-center queue; "
+            "refresh the audit report before acting on queue pressure."
+        )
+    elif truth_generated_at and not report_generated_at:
+        status = "unknown-report-age"
+        summary = (
+            "Portfolio truth is available, but the audit report timestamp is missing; "
+            "refresh the audit report before treating queue pressure as current."
+        )
+    return {
+        "status": status,
+        "summary": summary,
+        "report_generated_at": report_generated_at.isoformat() if report_generated_at else "",
+        "portfolio_truth_generated_at": truth_generated_at.isoformat() if truth_generated_at else "",
+    }
+
+
 def _weekly_pack_source(report_data: dict[str, Any], snapshot: dict[str, Any]) -> dict[str, Any]:
     source = dict(report_data)
     snapshot_summary = _mapping(snapshot.get("operator_summary"))
@@ -95,6 +133,8 @@ def build_weekly_command_center_digest(
         _safe_text(decision_quality.get("decision_quality_status")) or "insufficient-data"
     )
     truth = portfolio_truth or {}
+    freshness = _source_freshness(report_data, truth)
+    source_is_stale = freshness["status"] != "current"
     truth_summary = _build_truth_summary(truth)
     decision_queue = build_decision_queue(truth)
     decision_queue_summary = summarize_decision_queue(decision_queue)
@@ -104,6 +144,49 @@ def build_weekly_command_center_digest(
     )
     queue_pressure_summary = operator_why or _safe_text(weekly_pack.get("queue_pressure_summary"))
 
+    headline = (
+        "Refresh the audit report before acting on control-center queue pressure."
+        if source_is_stale
+        else _safe_text(operator_summary.get("headline"))
+        or _safe_text(weekly_story.get("headline"))
+        or "No weekly headline is recorded yet."
+    )
+    decision = (
+        "Refresh the audit report, then rerun the read-only control center before choosing a repo action."
+        if source_is_stale
+        else operator_decision
+        or _safe_text(weekly_story.get("decision"))
+        or "Continue the normal operator review loop."
+    )
+    why_this_week = (
+        freshness["summary"]
+        if source_is_stale
+        else operator_why
+        or _safe_text(weekly_story.get("why_this_week"))
+        or "No weekly rationale is recorded yet."
+    )
+    section_digest = (
+        [
+            {
+                "id": "source-freshness",
+                "label": "Source Freshness",
+                "state": "refresh-needed",
+                "headline": freshness["summary"],
+                "next_step": (
+                    "Refresh the audit report, then rerun the read-only control center."
+                ),
+                "reason_codes": [freshness["status"]],
+            }
+        ]
+        if source_is_stale
+        else _build_section_digest(
+            weekly_story,
+            operator_decision=operator_decision,
+            operator_why=operator_why,
+        )
+    )
+    top_repo_briefings = [] if source_is_stale else repo_briefings[:MAX_REPO_BRIEFINGS]
+
     return {
         "contract_version": CONTRACT_VERSION,
         "authority_cap": AUTHORITY_CAP,
@@ -113,18 +196,13 @@ def build_weekly_command_center_digest(
         "report_reference": report_reference or _safe_text(report_data.get("latest_report_path")),
         "control_center_reference": control_center_reference,
         "portfolio_truth_reference": portfolio_truth_reference,
-        "headline": _safe_text(operator_summary.get("headline"))
-        or _safe_text(weekly_story.get("headline"))
-        or "No weekly headline is recorded yet.",
-        "decision": operator_decision
-        or _safe_text(weekly_story.get("decision"))
-        or "Continue the normal operator review loop.",
-        "why_this_week": operator_why
-        or _safe_text(weekly_story.get("why_this_week"))
-        or "No weekly rationale is recorded yet.",
+        "source_freshness": freshness,
+        "headline": headline,
+        "decision": decision,
+        "why_this_week": why_this_week,
         "next_step": _safe_text(weekly_story.get("next_step"))
         or "Open the workbook first, then use the read-only control center.",
-        "queue_pressure_summary": queue_pressure_summary,
+        "queue_pressure_summary": freshness["summary"] if source_is_stale else queue_pressure_summary,
         "operating_paths_summary": _safe_text(weekly_pack.get("operating_paths_summary")),
         "decision_quality": {
             "status": decision_quality_status,
@@ -155,11 +233,7 @@ def build_weekly_command_center_digest(
             **_build_security_summary(truth),
             "top_alerts": _build_security_attention_items(truth),
         },
-        "section_digest": _build_section_digest(
-            weekly_story,
-            operator_decision=operator_decision,
-            operator_why=operator_why,
-        ),
+        "section_digest": section_digest,
         "top_repo_briefings": [
             {
                 "repo": _safe_text(item.get("repo")) or "Repo",
@@ -170,7 +244,7 @@ def build_weekly_command_center_digest(
                 "operating_path_line": _safe_text(item.get("operating_path_line")),
                 "operator_focus_line": _safe_text(item.get("operator_focus_line")),
             }
-            for item in repo_briefings[:MAX_REPO_BRIEFINGS]
+            for item in top_repo_briefings
         ],
         "report_only_guardrail": (
             "This digest is descriptive only. It may highlight path, trust, and pressure, "
@@ -181,6 +255,7 @@ def build_weekly_command_center_digest(
 
 def render_weekly_command_center_markdown(digest: dict[str, Any]) -> str:
     decision_quality = _mapping(digest.get("decision_quality"))
+    source_freshness = _mapping(digest.get("source_freshness"))
     portfolio_truth = _mapping(digest.get("portfolio_truth"))
     risk_posture = _mapping(digest.get("risk_posture"))
     tier_counts = _mapping(risk_posture.get("risk_tier_counts"))
@@ -195,6 +270,7 @@ def render_weekly_command_center_markdown(digest: dict[str, Any]) -> str:
         f"- Decision: {_safe_text(digest.get('decision'))}",
         f"- Why This Week: {_safe_text(digest.get('why_this_week'))}",
         f"- Next Step: {_safe_text(digest.get('next_step'))}",
+        f"- Source Freshness: `{_safe_text(source_freshness.get('status')) or 'unknown'}` — {_safe_text(source_freshness.get('summary')) or 'No source freshness summary is recorded yet.'}",
         f"- Decision Quality: `{_safe_text(decision_quality.get('status'))}` — {_safe_text(decision_quality.get('summary'))}",
         f"- Operating Paths: {_safe_text(digest.get('operating_paths_summary')) or 'No operating-path summary is recorded yet.'}",
         f"- Portfolio Truth: {portfolio_truth.get('project_count', 0)} projects, {portfolio_truth.get('active_project_count', 0)} active registry entries, {portfolio_truth.get('default_attention_count', 0)} default attention, {portfolio_truth.get('decision_queue_count', 0)} decision queue",

tests/test_cli_hardening.py

@@ -537,6 +537,39 @@ def test_main_control_center_writes_artifacts_without_audit(monkeypatch, tmp_pat
     assert "Move into Action Sync only when the local weekly story is already" in combined
 
 
+def test_main_control_center_suppresses_queue_when_portfolio_truth_is_newer(
+    monkeypatch,
+    tmp_path,
+    sample_metadata,
+    capsys,
+):
+    args = _make_args(control_center=True, output_dir=str(tmp_path))
+    report_path = tmp_path / "audit-report-testuser-2026-03-29.json"
+    report_data = _make_report_dict(sample_metadata)
+    report_data["generated_at"] = "2026-03-29T12:00:00+00:00"
+    report_data["operator_summary"] = {
+        "headline": "Urgent stale pressure is active.",
+        "what_to_do_next": "Act now on StaleRepo.",
+    }
+    report_data["operator_queue"] = [{"repo": "StaleRepo"}]
+    report_path.write_text("{}")
+    (tmp_path / "portfolio-truth-latest.json").write_text(
+        json.dumps({"generated_at": "2026-03-30T12:00:00+00:00", "projects": []})
+    )
+
+    monkeypatch.setattr(cli, "build_parser", lambda: FakeParser(args))
+    monkeypatch.setattr(cli, "_load_latest_report", lambda _output_dir: (report_path, report_data))
+    monkeypatch.setattr("src.history.find_previous", lambda *_args, **_kwargs: None)
+
+    cli.main()
+
+    captured = capsys.readouterr()
+    combined = captured.out + captured.err
+    assert "Refresh the audit report before acting on control-center queue pressure." in combined
+    assert "Portfolio truth is newer than the audit report" in combined
+    assert "Act now on StaleRepo" not in combined
+
+
 def test_main_control_center_requires_latest_report(monkeypatch):
     args = _make_args(control_center=True)
 

tests/test_weekly_command_center.py

@@ -8,6 +8,7 @@
 
 def _make_portfolio_truth() -> dict:
     return {
+        "generated_at": "2026-04-14T12:00:00+00:00",
         "projects": [
             {
                 "identity": {"display_name": "GithubRepoAuditor"},
@@ -277,6 +278,45 @@ def test_build_weekly_command_center_digest_prefers_control_center_snapshot_focu
     assert digest["top_repo_briefings"][0]["repo"] == "codexkit"
 
 
+def test_build_weekly_command_center_digest_blocks_stale_queue_when_truth_is_newer() -> None:
+    portfolio_truth = _make_portfolio_truth()
+    portfolio_truth["generated_at"] = "2026-04-15T12:00:00+00:00"
+    report_data = {
+        "username": "testuser",
+        "generated_at": "2026-04-14T12:00:00+00:00",
+        "operator_summary": {
+            "headline": "Urgent queue pressure is active.",
+            "what_to_do_next": "Act now on StaleRepo.",
+            "trend_summary": "StaleRepo is the top pressure item.",
+            "decision_quality_v1": {},
+        },
+        "operator_queue": [{"repo": "StaleRepo"}],
+        "audits": [],
+    }
+    snapshot = {
+        "operator_summary": report_data["operator_summary"],
+        "operator_queue": report_data["operator_queue"],
+    }
+
+    digest = build_weekly_command_center_digest(
+        report_data,
+        snapshot,
+        portfolio_truth=portfolio_truth,
+        generated_at="2026-04-14T12:00:00+00:00",
+    )
+
+    assert digest["source_freshness"]["status"] == "portfolio-truth-newer"
+    assert "Refresh the audit report" in digest["headline"]
+    assert "Refresh the audit report" in digest["decision"]
+    assert "StaleRepo" not in digest["decision"]
+    assert "StaleRepo" not in digest["queue_pressure_summary"]
+    assert digest["top_repo_briefings"] == []
+
+    rendered_md = render_weekly_command_center_markdown(digest)
+    assert "Source Freshness: `portfolio-truth-newer`" in rendered_md
+    assert "StaleRepo" not in rendered_md
+
+
 def _sec(available: bool, critical: int = 0, high: int = 0) -> dict:
     return {
         "alerts_available": available,