fix(prax): quick_status self-sources mail counts from inbox.json (decouple from ai_mail section)

This commit is contained in:
AIOSAI
2026-06-12 23:57:08 -07:00
parent fc49928a4b
commit 37fb07ca16
5 changed files with 123 additions and 31 deletions
+5
View File
@@ -42,6 +42,11 @@ PyPI version — not the changelog header.
(not deleted). `DASHBOARD.local.json` overall 6.8 KB → 2.4 KB. Verified: seedgo
100%, 108 prax tests. (Follow-up: `@ai_mail`'s `dashboard_sync.py` section
writer to retire separately.)
- **quick_status now self-sources mail counts from `inbox.json`.** Decouples the
glance from the `ai_mail` section: prax's three quick_status calculators read
`.ai_mail.local/inbox.json` directly (`_read_mail_counts`) for `new_mail`/
`opened_mail`, so the `ai_mail` section is no longer a data dependency and can
be retired. 116 prax tests, seedgo 100%.
- **`.backupignore` is now a true `.gitignore` for the backup system — a single
source of truth (FPLAN-0269).** Replaced the hand-rolled `fnmatch`+part-loop
matcher (which broke leading-slash anchoring, `*`-crossing-`/`, dir-only `foo/`,
@@ -205,30 +205,47 @@ def _read_todo_count(branch_path: Path) -> int:
return 0
def _read_mail_counts(branch_path: Path) -> tuple:
"""Read new/opened mail counts from .ai_mail.local/inbox.json."""
inbox_path = branch_path / ".ai_mail.local" / "inbox.json"
if not inbox_path.exists():
return (0, 0)
try:
data = json.loads(inbox_path.read_text())
new_mail = 0
opened_mail = 0
for msg in data.get("messages", []):
status = msg.get("status", "")
if status == "new" or (not status and not msg.get("read", False)):
new_mail += 1
elif status == "opened":
opened_mail += 1
return (new_mail, opened_mail)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("Failed to read inbox from %s: %s", inbox_path, exc)
return (0, 0)
def _calculate_quick_status_standalone(sections: Dict, branch_path: Path) -> Dict:
"""
Calculate quick_status from live section data.
Calculate quick_status from branch data sources.
Self-contained version used by write_section() so it has no
external dependencies. Reads directly from section fields.
Self-contained version used by write_section(). Sources counts
directly from local files (inbox.json, local.json).
Args:
sections: All dashboard sections dict
branch_path: Path to branch root (for sourcing todo_count from local.json)
branch_path: Path to branch root (for sourcing counts from local files)
Returns:
Quick status dict with summary, action flags, and counts
"""
ai_mail = sections.get("ai_mail", {})
flow = sections.get("flow", {})
new_mail_raw = ai_mail.get("new", ai_mail.get("unread", 0))
opened_raw = ai_mail.get("opened", 0)
new_mail, opened_mail = _read_mail_counts(branch_path)
active_plans_raw = flow.get("active_plans", 0)
todo_count = _read_todo_count(branch_path)
new_mail = len(new_mail_raw) if isinstance(new_mail_raw, list) else int(new_mail_raw or 0)
opened_mail = len(opened_raw) if isinstance(opened_raw, list) else int(opened_raw or 0)
active_plans = len(active_plans_raw) if isinstance(active_plans_raw, list) else int(active_plans_raw or 0)
action_required = new_mail > 0 or active_plans > 0
@@ -166,22 +166,41 @@ def _read_todo_count(branch_path: Path) -> int:
return 0
def _read_mail_counts(branch_path: Path) -> tuple:
"""Read new/opened mail counts from .ai_mail.local/inbox.json."""
inbox_path = branch_path / ".ai_mail.local" / "inbox.json"
if not inbox_path.exists():
return (0, 0)
try:
data = json.loads(inbox_path.read_text())
new_mail = 0
opened_mail = 0
for msg in data.get("messages", []):
status = msg.get("status", "")
if status == "new" or (not status and not msg.get("read", False)):
new_mail += 1
elif status == "opened":
opened_mail += 1
return (new_mail, opened_mail)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("Failed to read inbox from %s: %s", inbox_path, exc)
return (0, 0)
def _calculate_quick_status(sections: Dict, branch_path: Path) -> Dict:
"""
Calculate quick_status from live section data (v3 schema).
Calculate quick_status from branch data sources.
Args:
sections: All dashboard sections dict
branch_path: Path to branch root (for sourcing todo_count from local.json)
branch_path: Path to branch root (for sourcing counts from local files)
Returns:
Quick status dict with counts, action flag, and summary
"""
ai_mail = sections.get("ai_mail", {})
flow = sections.get("flow", {})
new_mail = ai_mail.get("new", ai_mail.get("unread", 0))
opened_mail = ai_mail.get("opened", 0)
new_mail, opened_mail = _read_mail_counts(branch_path)
active_plans = flow.get("active_plans", 0)
todo_count = _read_todo_count(branch_path)
@@ -48,26 +48,48 @@ def _read_todo_count(branch_path: Path) -> int:
return 0
def _read_mail_counts(branch_path: Path) -> tuple:
"""Read new/opened mail counts from .ai_mail.local/inbox.json."""
inbox_path = branch_path / ".ai_mail.local" / "inbox.json"
if not inbox_path.exists():
return (0, 0)
try:
data = json.loads(inbox_path.read_text())
new_mail = 0
opened_mail = 0
for msg in data.get("messages", []):
status = msg.get("status", "")
if status == "new" or (not status and not msg.get("read", False)):
new_mail += 1
elif status == "opened":
opened_mail += 1
return (new_mail, opened_mail)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("Failed to read inbox from %s: %s", inbox_path, exc)
return (0, 0)
def calculate_quick_status(sections: Dict, branch_path: "Path | None" = None) -> Dict:
"""
Calculate quick status from live section data.
Calculate quick status from branch data sources.
Reads directly from section fields pushed by each service.
Sources counts directly from local files (inbox.json, local.json).
Args:
sections: All dashboard sections
branch_path: Optional path to branch root (for sourcing todo_count)
branch_path: Optional path to branch root (for sourcing counts)
Returns:
Quick status dict with summary data
"""
ai_mail = sections.get("ai_mail", {})
flow = sections.get("flow", {})
new_mail = ai_mail.get("new", ai_mail.get("unread", 0))
opened_mail = ai_mail.get("opened", 0)
if branch_path:
new_mail, opened_mail = _read_mail_counts(branch_path)
todo_count = _read_todo_count(branch_path)
else:
new_mail, opened_mail, todo_count = 0, 0, 0
active_plans = flow.get("active_plans", 0)
todo_count = _read_todo_count(branch_path) if branch_path else 0
action_required = new_mail > 0 or active_plans > 0
+39 -10
View File
@@ -321,10 +321,19 @@ class TestCalculateQuickStatusStandalone:
assert result["summary"] == "All clear"
def test_new_mail_triggers_action_required(self, tmp_path):
"""New mail count > 0 sets action_required to True."""
"""New mail count > 0 sets action_required to True (sourced from inbox.json)."""
ops = _load_ops()
sections = {"ai_mail": {"new": 3, "opened": 0}}
result = ops._calculate_quick_status_standalone(sections, tmp_path)
mail_dir = tmp_path / ".ai_mail.local"
mail_dir.mkdir()
inbox = {
"messages": [
{"id": "1", "status": "new"},
{"id": "2", "status": "new"},
{"id": "3", "status": "new"},
]
}
(mail_dir / "inbox.json").write_text(json.dumps(inbox))
result = ops._calculate_quick_status_standalone({}, tmp_path)
assert result["new_mail"] == 3
assert result["action_required"] is True
assert "3 new emails" in result["summary"]
@@ -341,21 +350,41 @@ class TestCalculateQuickStatusStandalone:
def test_combined_summary_includes_all_parts(self, tmp_path):
"""Summary string includes all active counts."""
ops = _load_ops()
sections = {
"ai_mail": {"new": 2, "opened": 1},
"flow": {"active_plans": 3},
mail_dir = tmp_path / ".ai_mail.local"
mail_dir.mkdir()
inbox = {
"messages": [
{"id": "1", "status": "new"},
{"id": "2", "status": "new"},
{"id": "3", "status": "opened"},
]
}
(mail_dir / "inbox.json").write_text(json.dumps(inbox))
sections = {"flow": {"active_plans": 3}}
result = ops._calculate_quick_status_standalone(sections, tmp_path)
assert result["action_required"] is True
assert "2 new emails" in result["summary"]
assert "1 opened" in result["summary"]
assert "3 active plans" in result["summary"]
def test_unread_field_falls_back_from_new(self, tmp_path):
"""ai_mail may use 'unread' instead of 'new' -- code checks both."""
def test_mail_counts_from_inbox_json(self, tmp_path):
"""Mail counts sourced from inbox.json status fields."""
ops = _load_ops()
sections = {"ai_mail": {"unread": 7}}
result = ops._calculate_quick_status_standalone(sections, tmp_path)
mail_dir = tmp_path / ".ai_mail.local"
mail_dir.mkdir()
inbox = {
"messages": [
{"id": "1", "status": "new"},
{"id": "2", "status": "new"},
{"id": "3", "status": "new"},
{"id": "4", "status": "new"},
{"id": "5", "status": "new"},
{"id": "6", "status": "new"},
{"id": "7", "status": "new"},
]
}
(mail_dir / "inbox.json").write_text(json.dumps(inbox))
result = ops._calculate_quick_status_standalone({}, tmp_path)
assert result["new_mail"] == 7
assert result["action_required"] is True