diff --git a/src/portfolio_decision_queue.py b/src/portfolio_decision_queue.py new file mode 100644 index 0000000..3691ffe --- /dev/null +++ b/src/portfolio_decision_queue.py @@ -0,0 +1,147 @@ +"""Decision-queue compression for portfolio truth. + +This layer is intentionally narrower than default attention. ``active-product`` +and ``active-infra`` form the watch set; the decision queue is only for current +truth entries that already carry a concrete decision signal. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +CONTRACT_VERSION = "decision_queue_v1" +MAX_DECISION_QUEUE_ITEMS = 5 + +NON_DEFAULT_STATES = frozenset( + {"parked", "archived", "experiment", "evidence-history", "manual-only"} +) + + +def _mapping(value: Any) -> dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def _text(value: Any) -> str: + return value.strip() if isinstance(value, str) else "" + + +@dataclass(frozen=True) +class DecisionQueueItem: + project: str + path: str + attention_state: str + decision_type: str + why_now: str + evidence: tuple[str, ...] + source_freshness: str + recommended_action: str + do_not_refresh_docs_unless: str + + def to_dict(self) -> dict[str, Any]: + return { + "project": self.project, + "path": self.path, + "attention_state": self.attention_state, + "decision_type": self.decision_type, + "why_now": self.why_now, + "evidence": list(self.evidence), + "source_freshness": self.source_freshness, + "recommended_action": self.recommended_action, + "do_not_refresh_docs_unless": self.do_not_refresh_docs_unless, + } + + +def _decision_for_project( + project: dict[str, Any], *, generated_at: str +) -> DecisionQueueItem | None: + identity = _mapping(project.get("identity")) + derived = _mapping(project.get("derived")) + risk = _mapping(project.get("risk")) + security = _mapping(project.get("security")) + + attention_state = _text(derived.get("attention_state")) or "manual-only" + project_name = _text(identity.get("display_name")) or "Repo" + path = _text(identity.get("path")) or project_name + + if attention_state in {"archived", "evidence-history"}: + return None + + evidence: list[str] = [] + decision_type = "" + why_now = "" + recommended_action = "" + + if bool(risk.get("security_risk")): + critical = int(security.get("dependabot_critical") or 0) + high = int(security.get("dependabot_high") or 0) + decision_type = "security follow-up" + why_now = "Current portfolio truth marks this project with security risk." + evidence.append(f"security_risk=true; dependabot critical={critical}, high={high}") + recommended_action = "Decide whether to run the repo's security follow-up lane." + elif attention_state in NON_DEFAULT_STATES: + return None + elif attention_state == "decision-needed": + decision_type = "owner or human decision" + why_now = "Current portfolio truth marks this project as decision-needed." + evidence.append("attention_state=decision-needed") + risk_summary = _text(risk.get("risk_summary")) + if risk_summary: + evidence.append(risk_summary) + recommended_action = "Resolve the explicit portfolio decision before expanding scope." + else: + return None + + return DecisionQueueItem( + project=project_name, + path=path, + attention_state=attention_state, + decision_type=decision_type, + why_now=why_now, + evidence=tuple(evidence), + source_freshness=generated_at or "unknown", + recommended_action=recommended_action, + do_not_refresh_docs_unless=( + "Do not refresh context, roadmap, handoff, AGENTS, or docs unless " + "that work directly resolves this decision." + ), + ) + + +def build_decision_queue(portfolio_truth: dict[str, Any]) -> list[dict[str, Any]]: + """Return the small decision queue from current portfolio truth. + + This is deliberately stricter than the watch set: active product or active + infrastructure projects are ignored unless current truth also contains a + concrete decision signal. + """ + projects = portfolio_truth.get("projects") or [] + generated_at = _text(portfolio_truth.get("generated_at")) + queue: list[DecisionQueueItem] = [] + for project in projects: + if not isinstance(project, dict): + continue + item = _decision_for_project(project, generated_at=generated_at) + if item is not None: + queue.append(item) + + decision_rank = {"security follow-up": 0, "owner or human decision": 1} + queue.sort( + key=lambda item: ( + decision_rank.get(item.decision_type, 9), + item.project.lower(), + ) + ) + return [item.to_dict() for item in queue[:MAX_DECISION_QUEUE_ITEMS]] + + +def summarize_decision_queue(items: list[dict[str, Any]]) -> dict[str, Any]: + type_counts: dict[str, int] = {} + for item in items: + decision_type = _text(item.get("decision_type")) or "unknown" + type_counts[decision_type] = type_counts.get(decision_type, 0) + 1 + return { + "contract_version": CONTRACT_VERSION, + "decision_queue_count": len(items), + "decision_queue_type_counts": type_counts, + } diff --git a/src/weekly_command_center.py b/src/weekly_command_center.py index d9f481b..189a360 100644 --- a/src/weekly_command_center.py +++ b/src/weekly_command_center.py @@ -6,6 +6,7 @@ from typing import Any from src.portfolio_automation import select_automation_candidates +from src.portfolio_decision_queue import build_decision_queue, summarize_decision_queue from src.portfolio_truth_types import truth_latest_path from src.report_enrichment import build_weekly_review_pack @@ -95,6 +96,8 @@ def build_weekly_command_center_digest( ) truth = portfolio_truth or {} truth_summary = _build_truth_summary(truth) + decision_queue = build_decision_queue(truth) + decision_queue_summary = summarize_decision_queue(decision_queue) operator_decision = _operator_decision(operator_summary, operator_queue) operator_why = _safe_text(operator_summary.get("trend_summary")) or _safe_text( operator_summary.get("why_it_matters") @@ -133,7 +136,8 @@ def build_weekly_command_center_digest( or "Decision quality is not available yet.", "authority_cap": _safe_text(decision_quality.get("authority_cap")) or AUTHORITY_CAP, }, - "portfolio_truth": truth_summary, + "portfolio_truth": {**truth_summary, **decision_queue_summary}, + "decision_queue": decision_queue, "path_attention": _build_path_attention_items(truth), "automation_candidates": [ candidate.to_dict() @@ -193,13 +197,30 @@ def render_weekly_command_center_markdown(digest: dict[str, Any]) -> str: f"- Next Step: {_safe_text(digest.get('next_step'))}", 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_needed_count', 0)} decision-needed", + 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", f"- Risk Posture: {risk_posture.get('elevated_count', 0)} elevated, {tier_counts.get('moderate', 0)} moderate, {tier_counts.get('baseline', 0)} baseline", f"- Security Posture: {security_posture.get('scanned_count', 0)} scanned, {security_posture.get('repos_with_open_high_critical', 0)} with open high/critical Dependabot alerts ({security_posture.get('total_open_critical', 0)} critical, {security_posture.get('total_open_high', 0)} high)", "", - "## Path Attention", + "## Decision Queue", ] + decision_queue = list(digest.get("decision_queue") or []) + if not decision_queue: + lines.append("- No portfolio decisions clear the current evidence bar.") + else: + for item in decision_queue: + lines.append( + f"- **{item['project']}** [{item['decision_type']}]: " + f"{item['why_now']} Next: {item['recommended_action']}" + ) + + lines.extend( + [ + "", + "## Path Attention", + ] + ) + path_attention = list(digest.get("path_attention") or []) if not path_attention: lines.append("- No active path clarifications are currently surfaced.") diff --git a/tests/test_portfolio_decision_queue.py b/tests/test_portfolio_decision_queue.py new file mode 100644 index 0000000..67aac0f --- /dev/null +++ b/tests/test_portfolio_decision_queue.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from src.portfolio_decision_queue import build_decision_queue, summarize_decision_queue + + +def _project( + name: str, + *, + attention_state: str, + security_risk: bool = False, + dependabot_critical: int = 0, + dependabot_high: int = 0, +) -> dict: + return { + "identity": {"display_name": name, "path": name}, + "derived": { + "attention_state": attention_state, + "registry_status": "active", + "activity_status": "active", + }, + "risk": { + "risk_tier": "baseline", + "risk_summary": "No elevated risk factors.", + "security_risk": security_risk, + }, + "security": { + "dependabot_critical": dependabot_critical, + "dependabot_high": dependabot_high, + }, + } + + +def test_default_attention_without_decision_signal_stays_out_of_queue() -> None: + truth = { + "generated_at": "2026-06-19T04:36:19+00:00", + "projects": [ + _project("Product", attention_state="active-product"), + _project("Infra", attention_state="active-infra"), + _project("Manual", attention_state="manual-only"), + _project("Experiment", attention_state="experiment"), + ], + } + + assert build_decision_queue(truth) == [] + assert summarize_decision_queue([]) == { + "contract_version": "decision_queue_v1", + "decision_queue_count": 0, + "decision_queue_type_counts": {}, + } + + +def test_decision_needed_project_enters_queue() -> None: + truth = { + "generated_at": "2026-06-19T04:36:19+00:00", + "projects": [_project("NeedsDecision", attention_state="decision-needed")], + } + + [item] = build_decision_queue(truth) + assert item["project"] == "NeedsDecision" + assert item["decision_type"] == "owner or human decision" + assert item["source_freshness"] == "2026-06-19T04:36:19+00:00" + assert "attention_state=decision-needed" in item["evidence"] + + +def test_security_risk_enters_queue_even_when_manual_only() -> None: + truth = { + "generated_at": "2026-06-19T04:36:19+00:00", + "projects": [ + _project( + "ManualSecurity", + attention_state="manual-only", + security_risk=True, + dependabot_critical=1, + ) + ], + } + + [item] = build_decision_queue(truth) + assert item["project"] == "ManualSecurity" + assert item["decision_type"] == "security follow-up" + assert item["evidence"] == ["security_risk=true; dependabot critical=1, high=0"] + + +def test_archived_security_risk_stays_out_of_queue() -> None: + truth = { + "projects": [ + _project( + "ArchivedSecurity", + attention_state="archived", + security_risk=True, + dependabot_critical=1, + ) + ], + } + + assert build_decision_queue(truth) == [] diff --git a/tests/test_weekly_command_center.py b/tests/test_weekly_command_center.py index 8386e00..040f119 100644 --- a/tests/test_weekly_command_center.py +++ b/tests/test_weekly_command_center.py @@ -144,8 +144,16 @@ def test_build_weekly_command_center_digest_surfaces_truth_and_guardrails() -> N assert digest["portfolio_truth"]["active_project_count"] == 3 assert digest["portfolio_truth"]["default_attention_count"] == 2 assert digest["portfolio_truth"]["decision_needed_count"] == 2 + assert digest["portfolio_truth"]["decision_queue_count"] == 2 + assert digest["portfolio_truth"]["decision_queue_type_counts"] == { + "owner or human decision": 2 + } assert digest["portfolio_truth"]["investigate_override_count"] == 2 assert digest["portfolio_truth"]["attention_state_counts"]["manual-only"] == 1 + assert [item["project"] for item in digest["decision_queue"]] == [ + "GithubRepoAuditor", + "JobCommandCenter", + ] assert digest["path_attention"][0]["repo"] == "JobCommandCenter" assert digest["path_attention"][0]["headline"] == "Unspecified stable path" assert all(item["attention_state"] == "decision-needed" for item in digest["path_attention"]) @@ -159,6 +167,8 @@ def test_build_weekly_command_center_digest_surfaces_truth_and_guardrails() -> N assert digest["portfolio_truth"]["risk_tier_counts"]["deferred"] == 1 rendered_md = render_weekly_command_center_markdown(digest) + assert "## Decision Queue" in rendered_md + assert "owner or human decision" in rendered_md assert "## Risk Posture" in rendered_md assert "GithubRepoAuditor" in rendered_md assert "JobCommandCenter" in rendered_md @@ -353,6 +363,11 @@ def test_security_posture_surfaces_open_alerts_critical_first() -> None: top = posture["top_alerts"] assert [item["repo"] for item in top] == ["CriticalRepo", "HighRepo"] assert top[0]["dependabot_critical"] == 2 + assert [item["project"] for item in digest["decision_queue"]] == [ + "CriticalRepo", + "HighRepo", + ] + assert digest["portfolio_truth"]["decision_queue_count"] == 2 rendered = render_weekly_command_center_markdown(digest) assert "## Security Posture" in rendered @@ -384,3 +399,56 @@ def test_security_posture_reports_not_run_when_no_overlay() -> None: rendered = render_weekly_command_center_markdown(digest) assert "## Security Posture" in rendered assert "Security overlay not run" in rendered + + +def test_default_attention_watch_set_does_not_create_decision_queue() -> None: + portfolio_truth = { + "projects": [ + { + "identity": {"display_name": "ActiveProduct"}, + "declared": {"operating_path": "finish"}, + "derived": { + "registry_status": "active", + "attention_state": "active-product", + "activity_status": "active", + "path_override": "", + "path_confidence": "high", + "context_quality": "standard", + }, + "risk": { + "risk_tier": "baseline", + "risk_factors": [], + "risk_summary": "No elevated risk factors.", + "security_risk": False, + }, + }, + { + "identity": {"display_name": "ActiveInfra"}, + "declared": {"operating_path": "maintain"}, + "derived": { + "registry_status": "active", + "attention_state": "active-infra", + "activity_status": "active", + "path_override": "", + "path_confidence": "high", + "context_quality": "standard", + }, + "risk": { + "risk_tier": "baseline", + "risk_factors": [], + "risk_summary": "No elevated risk factors.", + "security_risk": False, + }, + }, + ] + } + + digest = _digest_for(portfolio_truth) + + assert digest["portfolio_truth"]["default_attention_count"] == 2 + assert digest["portfolio_truth"]["decision_queue_count"] == 0 + assert digest["decision_queue"] == [] + + rendered = render_weekly_command_center_markdown(digest) + assert "2 default attention, 0 decision queue" in rendered + assert "No portfolio decisions clear the current evidence bar." in rendered