feat(automation): bounded-automation candidate layer (Arc D phase 1)

src/portfolio_automation.py

@@ -0,0 +1,118 @@
+"""Bounded-automation eligibility + candidate selection (Arc D, phase 1).
+
+This module is strictly advisory and read-only. It answers a single question
+for each repo: *does it clear the automation trust bar?* A repo is eligible
+only when ALL of the following hold:
+
+* the portfolio's ``decision_quality_status`` is ``"trusted"`` (a portfolio-wide
+  gate — when calibration is noisy/mixed/insufficient, nothing is eligible);
+* the repo's ``path_confidence`` is ``"high"`` (its operating path is settled);
+* the repo's ``registry_status`` is active or candidate (not archived/parked);
+* the repo's ``context_quality`` is non-trivial (not boilerplate/none/unknown).
+
+Selecting candidates does NOT propose or apply any change — proposal creation
+(phase 2) and gated execution (phase 3) build on top of this signal layer.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+CONTRACT_VERSION = "automation_candidates_v1"
+
+# Trust-bar thresholds. Kept as module constants so later phases (proposal
+# creation, execution re-checks) reuse the exact same gate.
+TRUSTED_DECISION_QUALITY = "trusted"
+ELIGIBLE_PATH_CONFIDENCE = frozenset({"high"})
+ELIGIBLE_REGISTRY_STATUS = frozenset({"active", "candidate"})
+ELIGIBLE_CONTEXT_QUALITY = frozenset({"minimum-viable", "standard", "full"})
+
+MAX_AUTOMATION_CANDIDATES = 25
+
+
+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 AutomationEligibility:
+    """Whether a repo clears the automation trust bar, with reasons if not."""
+
+    eligible: bool
+    blockers: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class AutomationCandidate:
+    """An eligible repo surfaced as a bounded-automation candidate."""
+
+    display_name: str
+    repo_full_name: str
+    registry_status: str
+    path_confidence: str
+    context_quality: str
+
+    def to_dict(self) -> dict[str, str]:
+        return {
+            "repo": self.display_name,
+            "repo_full_name": self.repo_full_name,
+            "registry_status": self.registry_status,
+            "path_confidence": self.path_confidence,
+            "context_quality": self.context_quality,
+        }
+
+
+def evaluate_automation_eligibility(
+    project: dict[str, Any], *, decision_quality_status: str
+) -> AutomationEligibility:
+    """Evaluate a single truth project against the bounded-automation trust bar.
+
+    ``decision_quality_status`` is the portfolio-level value (the same for every
+    repo in a run); the remaining checks are per-repo. Blockers accumulate so
+    the operator sees every reason a repo is held back, not just the first.
+    """
+    derived = _mapping(project.get("derived"))
+    blockers: list[str] = []
+    if _text(decision_quality_status) != TRUSTED_DECISION_QUALITY:
+        blockers.append("decision-quality-not-trusted")
+    if _text(derived.get("registry_status")) not in ELIGIBLE_REGISTRY_STATUS:
+        blockers.append("registry-status-not-eligible")
+    if _text(derived.get("path_confidence")) not in ELIGIBLE_PATH_CONFIDENCE:
+        blockers.append("path-confidence-not-high")
+    if _text(derived.get("context_quality")) not in ELIGIBLE_CONTEXT_QUALITY:
+        blockers.append("context-quality-too-weak")
+    return AutomationEligibility(eligible=not blockers, blockers=tuple(blockers))
+
+
+def select_automation_candidates(
+    portfolio_truth: dict[str, Any], *, decision_quality_status: str
+) -> list[AutomationCandidate]:
+    """Return the eligible repos (sorted by display name, capped) as candidates."""
+    projects = portfolio_truth.get("projects") or []
+    candidates: list[AutomationCandidate] = []
+    for project in projects:
+        if not isinstance(project, dict):
+            continue
+        eligibility = evaluate_automation_eligibility(
+            project, decision_quality_status=decision_quality_status
+        )
+        if not eligibility.eligible:
+            continue
+        identity = _mapping(project.get("identity"))
+        derived = _mapping(project.get("derived"))
+        candidates.append(
+            AutomationCandidate(
+                display_name=_text(identity.get("display_name")) or "Repo",
+                repo_full_name=_text(identity.get("repo_full_name")),
+                registry_status=_text(derived.get("registry_status")),
+                path_confidence=_text(derived.get("path_confidence")),
+                context_quality=_text(derived.get("context_quality")),
+            )
+        )
+    candidates.sort(key=lambda candidate: candidate.display_name.lower())
+    return candidates[:MAX_AUTOMATION_CANDIDATES]

src/weekly_command_center.py

@@ -5,6 +5,7 @@
 from pathlib import Path
 from typing import Any
 
+from src.portfolio_automation import select_automation_candidates
 from src.report_enrichment import build_weekly_review_pack
 
 CONTRACT_VERSION = "weekly_command_center_digest_v1"
@@ -55,6 +56,9 @@ def build_weekly_command_center_digest(
     operator_summary = _mapping(snapshot.get("operator_summary"))
     repo_briefings = list(weekly_pack.get("repo_briefings") or [])
     decision_quality = _mapping(operator_summary.get("decision_quality_v1"))
+    decision_quality_status = (
+        _safe_text(decision_quality.get("decision_quality_status")) or "insufficient-data"
+    )
     truth = portfolio_truth or {}
     truth_summary = _build_truth_summary(truth)
 
@@ -78,8 +82,7 @@ def build_weekly_command_center_digest(
         "queue_pressure_summary": _safe_text(weekly_pack.get("queue_pressure_summary")),
         "operating_paths_summary": _safe_text(weekly_pack.get("operating_paths_summary")),
         "decision_quality": {
-            "status": _safe_text(decision_quality.get("decision_quality_status"))
-            or "insufficient-data",
+            "status": decision_quality_status,
             "human_skepticism_required": bool(
                 decision_quality.get("human_skepticism_required", True)
             ),
@@ -90,6 +93,13 @@ def build_weekly_command_center_digest(
         },
         "portfolio_truth": truth_summary,
         "path_attention": _build_path_attention_items(truth),
+        "automation_candidates": [
+            candidate.to_dict()
+            for candidate in select_automation_candidates(
+                truth,
+                decision_quality_status=decision_quality_status,
+            )
+        ],
         "risk_posture": {
             "elevated_count": truth_summary.get("elevated_risk_count", 0),
             "risk_tier_counts": truth_summary.get("risk_tier_counts", {}),
@@ -153,6 +163,17 @@ def render_weekly_command_center_markdown(digest: dict[str, Any]) -> str:
                 f"- {item['repo']} — {item['headline']} ({item['registry_status']}, {item['context_quality']} context)"
             )
 
+    lines.extend(["", "## Automation Candidates"])
+    automation_candidates = list(digest.get("automation_candidates") or [])
+    if not automation_candidates:
+        lines.append("- No repos currently clear the automation trust bar.")
+    else:
+        for item in automation_candidates:
+            lines.append(
+                f"- {item['repo']} ({item['registry_status']}, "
+                f"{item['path_confidence']} path confidence, {item['context_quality']} context)"
+            )
+
     lines.extend(["", "## Risk Posture"])
     risk_items = list(risk_posture.get("top_elevated") or [])
     if not risk_items: