|
| 1 | +"""Decision-queue compression for portfolio truth. |
| 2 | +
|
| 3 | +This layer is intentionally narrower than default attention. ``active-product`` |
| 4 | +and ``active-infra`` form the watch set; the decision queue is only for current |
| 5 | +truth entries that already carry a concrete decision signal. |
| 6 | +""" |
| 7 | + |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +from dataclasses import dataclass |
| 11 | +from typing import Any |
| 12 | + |
| 13 | +CONTRACT_VERSION = "decision_queue_v1" |
| 14 | +MAX_DECISION_QUEUE_ITEMS = 5 |
| 15 | + |
| 16 | +NON_DEFAULT_STATES = frozenset( |
| 17 | + {"parked", "archived", "experiment", "evidence-history", "manual-only"} |
| 18 | +) |
| 19 | + |
| 20 | + |
| 21 | +def _mapping(value: Any) -> dict[str, Any]: |
| 22 | + return value if isinstance(value, dict) else {} |
| 23 | + |
| 24 | + |
| 25 | +def _text(value: Any) -> str: |
| 26 | + return value.strip() if isinstance(value, str) else "" |
| 27 | + |
| 28 | + |
| 29 | +@dataclass(frozen=True) |
| 30 | +class DecisionQueueItem: |
| 31 | + project: str |
| 32 | + path: str |
| 33 | + attention_state: str |
| 34 | + decision_type: str |
| 35 | + why_now: str |
| 36 | + evidence: tuple[str, ...] |
| 37 | + source_freshness: str |
| 38 | + recommended_action: str |
| 39 | + do_not_refresh_docs_unless: str |
| 40 | + |
| 41 | + def to_dict(self) -> dict[str, Any]: |
| 42 | + return { |
| 43 | + "project": self.project, |
| 44 | + "path": self.path, |
| 45 | + "attention_state": self.attention_state, |
| 46 | + "decision_type": self.decision_type, |
| 47 | + "why_now": self.why_now, |
| 48 | + "evidence": list(self.evidence), |
| 49 | + "source_freshness": self.source_freshness, |
| 50 | + "recommended_action": self.recommended_action, |
| 51 | + "do_not_refresh_docs_unless": self.do_not_refresh_docs_unless, |
| 52 | + } |
| 53 | + |
| 54 | + |
| 55 | +def _decision_for_project( |
| 56 | + project: dict[str, Any], *, generated_at: str |
| 57 | +) -> DecisionQueueItem | None: |
| 58 | + identity = _mapping(project.get("identity")) |
| 59 | + derived = _mapping(project.get("derived")) |
| 60 | + risk = _mapping(project.get("risk")) |
| 61 | + security = _mapping(project.get("security")) |
| 62 | + |
| 63 | + attention_state = _text(derived.get("attention_state")) or "manual-only" |
| 64 | + project_name = _text(identity.get("display_name")) or "Repo" |
| 65 | + path = _text(identity.get("path")) or project_name |
| 66 | + |
| 67 | + if attention_state in {"archived", "evidence-history"}: |
| 68 | + return None |
| 69 | + |
| 70 | + evidence: list[str] = [] |
| 71 | + decision_type = "" |
| 72 | + why_now = "" |
| 73 | + recommended_action = "" |
| 74 | + |
| 75 | + if bool(risk.get("security_risk")): |
| 76 | + critical = int(security.get("dependabot_critical") or 0) |
| 77 | + high = int(security.get("dependabot_high") or 0) |
| 78 | + decision_type = "security follow-up" |
| 79 | + why_now = "Current portfolio truth marks this project with security risk." |
| 80 | + evidence.append(f"security_risk=true; dependabot critical={critical}, high={high}") |
| 81 | + recommended_action = "Decide whether to run the repo's security follow-up lane." |
| 82 | + elif attention_state in NON_DEFAULT_STATES: |
| 83 | + return None |
| 84 | + elif attention_state == "decision-needed": |
| 85 | + decision_type = "owner or human decision" |
| 86 | + why_now = "Current portfolio truth marks this project as decision-needed." |
| 87 | + evidence.append("attention_state=decision-needed") |
| 88 | + risk_summary = _text(risk.get("risk_summary")) |
| 89 | + if risk_summary: |
| 90 | + evidence.append(risk_summary) |
| 91 | + recommended_action = "Resolve the explicit portfolio decision before expanding scope." |
| 92 | + else: |
| 93 | + return None |
| 94 | + |
| 95 | + return DecisionQueueItem( |
| 96 | + project=project_name, |
| 97 | + path=path, |
| 98 | + attention_state=attention_state, |
| 99 | + decision_type=decision_type, |
| 100 | + why_now=why_now, |
| 101 | + evidence=tuple(evidence), |
| 102 | + source_freshness=generated_at or "unknown", |
| 103 | + recommended_action=recommended_action, |
| 104 | + do_not_refresh_docs_unless=( |
| 105 | + "Do not refresh context, roadmap, handoff, AGENTS, or docs unless " |
| 106 | + "that work directly resolves this decision." |
| 107 | + ), |
| 108 | + ) |
| 109 | + |
| 110 | + |
| 111 | +def build_decision_queue(portfolio_truth: dict[str, Any]) -> list[dict[str, Any]]: |
| 112 | + """Return the small decision queue from current portfolio truth. |
| 113 | +
|
| 114 | + This is deliberately stricter than the watch set: active product or active |
| 115 | + infrastructure projects are ignored unless current truth also contains a |
| 116 | + concrete decision signal. |
| 117 | + """ |
| 118 | + projects = portfolio_truth.get("projects") or [] |
| 119 | + generated_at = _text(portfolio_truth.get("generated_at")) |
| 120 | + queue: list[DecisionQueueItem] = [] |
| 121 | + for project in projects: |
| 122 | + if not isinstance(project, dict): |
| 123 | + continue |
| 124 | + item = _decision_for_project(project, generated_at=generated_at) |
| 125 | + if item is not None: |
| 126 | + queue.append(item) |
| 127 | + |
| 128 | + decision_rank = {"security follow-up": 0, "owner or human decision": 1} |
| 129 | + queue.sort( |
| 130 | + key=lambda item: ( |
| 131 | + decision_rank.get(item.decision_type, 9), |
| 132 | + item.project.lower(), |
| 133 | + ) |
| 134 | + ) |
| 135 | + return [item.to_dict() for item in queue[:MAX_DECISION_QUEUE_ITEMS]] |
| 136 | + |
| 137 | + |
| 138 | +def summarize_decision_queue(items: list[dict[str, Any]]) -> dict[str, Any]: |
| 139 | + type_counts: dict[str, int] = {} |
| 140 | + for item in items: |
| 141 | + decision_type = _text(item.get("decision_type")) or "unknown" |
| 142 | + type_counts[decision_type] = type_counts.get(decision_type, 0) + 1 |
| 143 | + return { |
| 144 | + "contract_version": CONTRACT_VERSION, |
| 145 | + "decision_queue_count": len(items), |
| 146 | + "decision_queue_type_counts": type_counts, |
| 147 | + } |
0 commit comments