diff --git a/src/portfolio_automation.py b/src/portfolio_automation.py new file mode 100644 index 0000000..24635c1 --- /dev/null +++ b/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] diff --git a/src/weekly_command_center.py b/src/weekly_command_center.py index 912e65b..215deca 100644 --- a/src/weekly_command_center.py +++ b/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: diff --git a/tests/test_portfolio_automation.py b/tests/test_portfolio_automation.py new file mode 100644 index 0000000..467c603 --- /dev/null +++ b/tests/test_portfolio_automation.py @@ -0,0 +1,237 @@ +"""Tests for the Arc D bounded-automation eligibility + candidate layer. + +This layer is strictly advisory/read-only: it identifies which repos clear the +automation trust bar (high path confidence + trusted portfolio decision quality ++ non-trivial context + eligible registry status). It opens no PRs and applies +no changes — proposal creation and execution land in later Arc D phases. +""" + +from __future__ import annotations + +from src.portfolio_automation import ( + AutomationCandidate, + AutomationEligibility, + evaluate_automation_eligibility, + select_automation_candidates, +) +from src.weekly_command_center import ( + build_weekly_command_center_digest, + render_weekly_command_center_markdown, +) + + +def _project( + *, + display_name: str = "Repo", + repo_full_name: str = "owner/Repo", + registry_status: str = "active", + path_confidence: str = "high", + context_quality: str = "standard", +) -> dict: + return { + "identity": {"display_name": display_name, "repo_full_name": repo_full_name}, + "declared": {"operating_path": "maintain"}, + "derived": { + "registry_status": registry_status, + "path_confidence": path_confidence, + "context_quality": context_quality, + }, + } + + +# --- evaluate_automation_eligibility --------------------------------------- + + +def test_fully_eligible_repo_has_no_blockers() -> None: + result = evaluate_automation_eligibility(_project(), decision_quality_status="trusted") + assert isinstance(result, AutomationEligibility) + assert result.eligible is True + assert result.blockers == () + + +def test_non_trusted_decision_quality_blocks() -> None: + result = evaluate_automation_eligibility(_project(), decision_quality_status="needs-skepticism") + assert result.eligible is False + assert "decision-quality-not-trusted" in result.blockers + + +def test_low_path_confidence_blocks() -> None: + result = evaluate_automation_eligibility( + _project(path_confidence="low"), decision_quality_status="trusted" + ) + assert result.eligible is False + assert "path-confidence-not-high" in result.blockers + + +def test_legacy_path_confidence_blocks() -> None: + result = evaluate_automation_eligibility( + _project(path_confidence="legacy"), decision_quality_status="trusted" + ) + assert result.eligible is False + assert "path-confidence-not-high" in result.blockers + + +def test_boilerplate_context_blocks() -> None: + result = evaluate_automation_eligibility( + _project(context_quality="boilerplate"), decision_quality_status="trusted" + ) + assert result.eligible is False + assert "context-quality-too-weak" in result.blockers + + +def test_none_context_blocks() -> None: + result = evaluate_automation_eligibility( + _project(context_quality="none"), decision_quality_status="trusted" + ) + assert result.eligible is False + assert "context-quality-too-weak" in result.blockers + + +def test_unknown_context_blocks_conservatively() -> None: + result = evaluate_automation_eligibility( + _project(context_quality=""), decision_quality_status="trusted" + ) + assert result.eligible is False + assert "context-quality-too-weak" in result.blockers + + +def test_archived_registry_status_blocks() -> None: + result = evaluate_automation_eligibility( + _project(registry_status="archived"), decision_quality_status="trusted" + ) + assert result.eligible is False + assert "registry-status-not-eligible" in result.blockers + + +def test_candidate_registry_status_is_eligible() -> None: + result = evaluate_automation_eligibility( + _project(registry_status="candidate"), decision_quality_status="trusted" + ) + assert result.eligible is True + + +def test_minimum_viable_context_is_eligible() -> None: + result = evaluate_automation_eligibility( + _project(context_quality="minimum-viable"), decision_quality_status="trusted" + ) + assert result.eligible is True + + +def test_multiple_blockers_accumulate() -> None: + result = evaluate_automation_eligibility( + _project(registry_status="archived", path_confidence="low", context_quality="none"), + decision_quality_status="needs-skepticism", + ) + assert result.eligible is False + assert set(result.blockers) == { + "decision-quality-not-trusted", + "registry-status-not-eligible", + "path-confidence-not-high", + "context-quality-too-weak", + } + + +# --- select_automation_candidates ------------------------------------------ + + +def test_selects_only_eligible_repos_sorted_by_name() -> None: + truth = { + "projects": [ + _project(display_name="Zebra", repo_full_name="o/Zebra"), + _project(display_name="Alpha", repo_full_name="o/Alpha"), + _project(display_name="LowPath", path_confidence="low"), + _project(display_name="Boiler", context_quality="boilerplate"), + _project(display_name="Archived", registry_status="archived"), + ] + } + candidates = select_automation_candidates(truth, decision_quality_status="trusted") + assert [c.display_name for c in candidates] == ["Alpha", "Zebra"] + assert all(isinstance(c, AutomationCandidate) for c in candidates) + + +def test_non_trusted_decision_quality_yields_no_candidates() -> None: + # The portfolio-level gate kills every candidate regardless of per-repo state. + truth = {"projects": [_project(), _project(display_name="Other", repo_full_name="o/Other")]} + assert select_automation_candidates(truth, decision_quality_status="needs-skepticism") == [] + + +def test_candidate_to_dict_shape() -> None: + truth = {"projects": [_project(display_name="Solo", repo_full_name="acme/Solo")]} + [candidate] = select_automation_candidates(truth, decision_quality_status="trusted") + assert candidate.to_dict() == { + "repo": "Solo", + "repo_full_name": "acme/Solo", + "registry_status": "active", + "path_confidence": "high", + "context_quality": "standard", + } + + +def test_empty_or_missing_projects_is_safe() -> None: + assert select_automation_candidates({}, decision_quality_status="trusted") == [] + assert select_automation_candidates({"projects": None}, decision_quality_status="trusted") == [] + + +def test_non_dict_projects_are_skipped() -> None: + truth = {"projects": ["not-a-dict", _project(display_name="Real", repo_full_name="o/Real")]} + candidates = select_automation_candidates(truth, decision_quality_status="trusted") + assert [c.display_name for c in candidates] == ["Real"] + + +def test_eligible_repo_with_missing_identity_falls_back_safely() -> None: + # An eligible repo whose identity block is absent must not crash; it falls + # back to a placeholder name and an empty (honest) slug. + project = _project() + del project["identity"] + candidates = select_automation_candidates( + {"projects": [project]}, decision_quality_status="trusted" + ) + assert len(candidates) == 1 + assert candidates[0].display_name == "Repo" + assert candidates[0].repo_full_name == "" + + +# --- weekly digest integration --------------------------------------------- + + +def _digest_for(portfolio_truth: dict, decision_quality_status: str) -> dict: + operator_summary = {"decision_quality_v1": {"decision_quality_status": decision_quality_status}} + report_data = { + "username": "testuser", + "generated_at": "2026-04-14T12:00:00+00:00", + "operator_summary": operator_summary, + "audits": [], + } + snapshot = {"operator_summary": operator_summary, "operator_queue": []} + return build_weekly_command_center_digest( + report_data, + snapshot, + portfolio_truth=portfolio_truth, + generated_at="2026-04-14T12:00:00+00:00", + ) + + +def test_digest_surfaces_automation_candidates_when_trusted() -> None: + truth = { + "projects": [ + _project(display_name="EligibleOne", repo_full_name="o/EligibleOne"), + _project(display_name="WeakContext", context_quality="boilerplate"), + ] + } + digest = _digest_for(truth, "trusted") + candidates = digest["automation_candidates"] + assert [c["repo"] for c in candidates] == ["EligibleOne"] + + rendered = render_weekly_command_center_markdown(digest) + assert "## Automation Candidates" in rendered + assert "EligibleOne" in rendered + + +def test_digest_has_no_automation_candidates_when_not_trusted() -> None: + truth = {"projects": [_project(display_name="EligibleOne", repo_full_name="o/EligibleOne")]} + digest = _digest_for(truth, "needs-skepticism") + assert digest["automation_candidates"] == [] + + rendered = render_weekly_command_center_markdown(digest) + assert "## Automation Candidates" in rendered + assert "No repos currently clear the automation trust bar." in rendered