Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 41 additions & 0 deletions src/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -5310,6 +5310,46 @@ def _load_security_alerts_by_name(*, output_dir: Path, username: str) -> dict[st
return {name: entry for name, entry in data.items() if isinstance(entry, dict)}


WAREHOUSE_REPORT_STALE_DAYS = 7


def _warn_if_warehouse_report_stale(output_dir: Path, username: str) -> None:
"""Warn when the legacy warehouse report is missing or stale (F2).

Notion OS's external-signal-sync reads ``audit-report-<username>-*.json`` (the
3.7 warehouse report), but ``--portfolio-truth`` mode does NOT regenerate it —
the truth pipeline scans the workspace and never runs the GitHub audit that
produces warehouse data. Per the F2 "keep both artifacts live" decision, surface
the gap at generation time so the operator runs ``audit report <username>`` to
keep Notion's Repo Auditor signal fresh. Complements the cross-system-smoke C2
check, which catches the same drift at smoke time.
"""
from datetime import date

reports = sorted(output_dir.glob(f"audit-report-{username}-*.json"))
if not reports:
print_warning(
f"No audit-report-{username}-*.json in {output_dir}: Notion's Repo Auditor "
f"signal reads that warehouse report and this --portfolio-truth run did not "
f"create one. Run `audit report {username}` to generate it (F2)."
)
return
match = re.search(r"(\d{4}-\d{2}-\d{2})", reports[-1].name)
if not match:
return
try:
report_date = date.fromisoformat(match.group(1))
except ValueError:
return
age = (date.today() - report_date).days
if age > WAREHOUSE_REPORT_STALE_DAYS:
print_warning(
f"Newest warehouse report {reports[-1].name} is {age}d old: Notion's Repo "
f"Auditor signal reads it and is now stale. Run `audit report {username}` to "
f"refresh the warehouse report (F2 — both artifacts kept live by decision)."
)


def _run_portfolio_truth_mode(args) -> None:
from src.portfolio_truth_publish import publish_portfolio_truth

Expand Down Expand Up @@ -5361,6 +5401,7 @@ def _run_portfolio_truth_mode(args) -> None:
f"(registry {'updated' if result.registry_changed else 'unchanged'}, "
f"report {'updated' if result.report_changed else 'unchanged'})"
)
_warn_if_warehouse_report_stale(output_dir, args.username)


def _run_portfolio_context_recovery_mode(args) -> None:
Expand Down
39 changes: 39 additions & 0 deletions tests/test_portfolio_truth.py
Original file line number Diff line number Diff line change
Expand Up @@ -1387,3 +1387,42 @@ def test_git_default_branch_empty_when_origin_head_unset(tmp_path: Path) -> None

# A freshly init'd repo has no origin/HEAD → "" so callers fall back.
assert _git_default_branch(repo) == ""


# ── F2: warehouse-report staleness reminder ────────────────────────────────
from src.cli import _warn_if_warehouse_report_stale # noqa: E402


def _write_warehouse_report(d: Path, username: str, date_str: str) -> None:
(d / f"audit-report-{username}-{date_str}.json").write_text("{}", encoding="utf-8")


class TestWarehouseStalenessReminder:
"""F2 (keep-dual): --portfolio-truth mode warns when the warehouse report Notion
reads is missing or stale, so the operator refreshes it."""

def test_missing_report_warns(self, tmp_path: Path, capsys) -> None:
import re

_warn_if_warehouse_report_stale(tmp_path, "saagpatel")
captured = capsys.readouterr()
# print_warning word-wraps, so normalize whitespace before substring checks
combined = re.sub(r"\s+", " ", captured.out + captured.err)
assert "No audit-report-saagpatel" in combined
assert "audit report saagpatel" in combined

def test_stale_report_warns(self, tmp_path: Path, capsys) -> None:
_write_warehouse_report(tmp_path, "saagpatel", "2020-01-01")
_warn_if_warehouse_report_stale(tmp_path, "saagpatel")
captured = capsys.readouterr()
assert "stale" in (captured.out + captured.err).lower()

def test_fresh_report_no_warning(self, tmp_path: Path, capsys) -> None:
from datetime import date

_write_warehouse_report(tmp_path, "saagpatel", date.today().isoformat())
_warn_if_warehouse_report_stale(tmp_path, "saagpatel")
captured = capsys.readouterr()
combined = captured.out + captured.err
assert "stale" not in combined.lower()
assert "No audit-report" not in combined