From e7f373a851b5df188a0f1e9b4b55115bb460e60a Mon Sep 17 00:00:00 2001 From: Saagar Date: Sat, 4 Jul 2026 05:25:36 -0700 Subject: [PATCH] fix: guard weekly command center freshness --- src/cli.py | 14 +++- src/weekly_command_center.py | 110 +++++++++++++++++++++++----- tests/test_cli_hardening.py | 33 +++++++++ tests/test_weekly_command_center.py | 40 ++++++++++ 4 files changed, 178 insertions(+), 19 deletions(-) diff --git a/src/cli.py b/src/cli.py index 1506074f..4ba88116 100644 --- a/src/cli.py +++ b/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}") diff --git a/src/weekly_command_center.py b/src/weekly_command_center.py index 189a3604..55fd17ed 100644 --- a/src/weekly_command_center.py +++ b/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", diff --git a/tests/test_cli_hardening.py b/tests/test_cli_hardening.py index 9ebcfada..3b0427dc 100644 --- a/tests/test_cli_hardening.py +++ b/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) diff --git a/tests/test_weekly_command_center.py b/tests/test_weekly_command_center.py index 040f1191..6cbfcc33 100644 --- a/tests/test_weekly_command_center.py +++ b/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,