refactor(cli): extract application flows

docs/architecture.md

@@ -500,6 +500,28 @@ output/
 
 ## Design Intent
 
+The 2026-07-10 elegance review ratified four design rules for all future work. They exist to
+stop the three debts that phase-by-phase growth created: callable-threaded extractions,
+template-stamped control modules, and lazy-import cycles.
+
+1. **No new callable-threaded extractions.** When logic is extracted from a large module,
+   the shared primitives it needs move to a leaf support module (for the operator layer,
+   `src/operator_trend_support.py`) and are imported by the extracted module. Threading the
+   parent's helpers back in as `Callable` parameters is no longer an accepted way to avoid a
+   circular import — fix the dependency direction instead.
+2. **No new stamped control modules.** A new closure-forecast or trend control may not be
+   added by copying the seven-function template (`*_side_from_status` / `*_path_label` /
+   `*_recovery_for_target` / `apply_*_control` / `*_hotspots` / `*_summary` / `*_reason`)
+   into a new module. New controls wait for (or drive) the parametrized control core; until
+   that core exists, adding one requires an explicit maintainer decision recorded here.
+3. **No new function-level internal imports outside the CLI dispatch layer.** `src/cli.py`
+   may defer imports for startup latency. Everywhere else, a deferred `from src...` import
+   inside a function body marks a layering bug; fix the direction rather than hiding the
+   import.
+4. **Overlays merge or die.** An advisory overlay that survives two further phases is either
+   merged into the core it overlays or deleted. Overlay status is a transition state, not a
+   destination — this keeps "bounded overlay" from being a one-way ratchet on module count.
+
 ## Automation Guidance
 
 Phase 92 adds one bounded execution-guidance layer on top of the existing Action Sync and historical stack: `Automation Guidance`.

src/app/__init__.py

@@ -0,0 +1 @@
+"""Application-layer mode handlers dispatched by :mod:`src.cli`."""

src/app/acknowledgments.py

@@ -0,0 +1,59 @@
+"""Acknowledgment-capture application flow."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from src.cli_output import print_info
+from src.diff import diff_reports
+from src.history import find_previous
+from src.operator_acknowledgments import (
+    build_acknowledgment_record,
+    find_matching_change,
+    find_sibling_changes,
+    load_acknowledgments,
+    save_acknowledgment,
+)
+from src.recurring_review import MATERIALITY_THRESHOLDS, evaluate_material_changes
+from src.report_state import load_latest_report
+
+
+def run_acknowledgment_capture_mode(args: Any, parser: Any) -> None:
+    if not args.acknowledge_target:
+        parser.error("--acknowledge-target is required for acknowledgment capture")
+    if not args.acknowledge_kind:
+        parser.error("--acknowledge-kind is required for acknowledgment capture")
+    if not (args.acknowledge_note or "").strip():
+        parser.error("--acknowledge-note is required and must explain the acknowledgment")
+    output_dir = Path(args.output_dir)
+    report_path, report_data = load_latest_report(output_dir)
+    if not report_path or not report_data:
+        parser.error("No existing audit report found in output directory")
+    diff_dict = None
+    previous_path = find_previous(report_path.name)
+    if previous_path:
+        diff_dict = diff_reports(
+            previous_path, report_path, portfolio_profile=args.portfolio_profile,
+            collection_name=args.collection,
+        ).to_dict()
+    material_changes = evaluate_material_changes(
+        report_data, diff_data=diff_dict, thresholds=MATERIALITY_THRESHOLDS["standard"]
+    )
+    acknowledgments = load_acknowledgments(output_dir, args.username)
+    matched = find_matching_change(
+        repo_name=args.acknowledge_target, change_kind=args.acknowledge_kind,
+        material_changes=material_changes, acknowledgments=acknowledgments,
+    )
+    if not matched:
+        parser.error(f"No open '{args.acknowledge_kind}' change found for '{args.acknowledge_target}' in the latest report")
+    reviewer = args.acknowledge_reviewer or args.approval_reviewer
+    record = build_acknowledgment_record(matched, reviewer=reviewer, note=args.acknowledge_note)
+    saved_path = save_acknowledgment(output_dir, args.username, record)
+    print_info(f"Acknowledged {args.acknowledge_kind} for {args.acknowledge_target} (change_key={record['change_key'][:12]}…, reviewer={reviewer})")
+    for sibling in find_sibling_changes(matched, material_changes):
+        sibling_record = build_acknowledgment_record(sibling, reviewer=reviewer, note=args.acknowledge_note)
+        save_acknowledgment(output_dir, args.username, sibling_record)
+        print_info(f"Acknowledged sibling {sibling.get('change_type')} for {sibling.get('repo_name')} (change_key={sibling_record['change_key'][:12]}…)")
+    print_info(f"Acknowledgment store: {saved_path}")
+    print_info("Run --control-center to confirm the item is filtered from the queue.")

src/app/approval_center.py

@@ -0,0 +1,102 @@
+"""Approval-center application flow."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from src.cli_output import print_info
+from src.control_center_report_state import refresh_latest_report_state
+from src.operator_approval_artifacts import write_approval_center_artifacts
+from src.operator_approval_artifacts import (
+    write_approval_receipt,
+    write_followup_review_receipt,
+)
+from src.operator_prefs import load_rejection_events, post_process_approval_session
+from src.approval_ledger import (
+    build_approval_followup_record,
+    build_approval_record,
+    load_approval_ledger_bundle,
+)
+from src.report_shared_artifacts import refresh_shared_artifacts_from_report
+from src.warehouse import save_approval_followup_event, save_approval_record
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _post_process_approval_center_prefs(payload: dict, output_dir: Path) -> None:
+    try:
+        rejection_records = load_rejection_events(output_dir)
+        total, newly_added = post_process_approval_session(rejection_records, output_dir)
+        print_info(f"Suppressions: {total} action type(s) suppressed ({newly_added} newly added).")
+    except Exception as exc:  # noqa: BLE001
+        logging.getLogger(__name__).warning(
+            "operator_prefs post-process failed (non-fatal): %s", exc
+        )
+
+
+def run_approval_center_mode(args: Any, parser: Any) -> None:
+    report_output_dir = Path(args.output_dir)
+    try:
+        _report_path, _diff_dict, report = refresh_latest_report_state(report_output_dir, args)
+    except FileNotFoundError:
+        parser.error("No existing audit report found in output directory")
+    approval_json, approval_md, payload = write_approval_center_artifacts(
+        report, report_output_dir, approval_view=args.approval_view
+    )
+    print_info(payload.get("approval_workflow_summary", {}).get("summary", "No current approval needs review yet."))
+    print_info(payload.get("next_approval_review", {}).get("summary", "Stay local for now; no current approval needs review."))
+    print_info(f"Approval center JSON: {approval_json}")
+    print_info(f"Approval center Markdown: {approval_md}")
+    _post_process_approval_center_prefs(payload, report_output_dir)
+
+
+def run_approval_capture_mode(args: Any, parser: Any) -> None:
+    report_output_dir = Path(args.output_dir)
+    try:
+        _report_path, diff_dict, report = refresh_latest_report_state(report_output_dir, args)
+    except FileNotFoundError:
+        parser.error("No existing audit report found in output directory")
+    bundle = load_approval_ledger_bundle(report_output_dir, report.to_dict(), list(report.operator_queue or []), approval_view="all")
+    ledger = {str(item.get("approval_id") or ""): item for item in bundle.get("approval_ledger", [])}
+    approval_id = f"governance:{args.governance_scope}" if args.approve_governance or args.review_governance else f"campaign:{args.campaign}"
+    ledger_record = ledger.get(approval_id)
+    if not ledger_record:
+        parser.error("No matching approval subject is surfaced in the latest report.")
+    if args.approve_governance or args.approve_packet:
+        if ledger_record.get("approval_state") == "blocked":
+            parser.error("That approval subject is blocked by non-approval prerequisites and cannot be approved yet.")
+        if ledger_record.get("approval_state") == "not-applicable":
+            parser.error("That approval subject is not part of the current approval workflow.")
+        approval_record = build_approval_record(ledger_record, reviewer=args.approval_reviewer, note=args.approval_note or "")
+        save_approval_record(report_output_dir, approval_record)
+    else:
+        if ledger_record.get("approval_state") in {"ready-for-review", "needs-reapproval", "blocked", "not-applicable"}:
+            parser.error("That approval subject is not currently eligible for a recurring local follow-up review.")
+        if str(ledger_record.get("follow_up_command") or "").strip() == "":
+            parser.error("That approval subject does not currently expose a follow-up review command.")
+        followup_event = build_approval_followup_record(ledger_record, reviewer=args.approval_reviewer, note=args.approval_note or "")
+        save_approval_followup_event(report_output_dir, followup_event)
+    _report_path, diff_dict, report = refresh_latest_report_state(report_output_dir, args)
+    refresh_shared_artifacts_from_report(report, report_output_dir, args, diff_dict=diff_dict)
+    approval_json, approval_md, _payload = write_approval_center_artifacts(report, report_output_dir, approval_view="all")
+    updated_bundle = load_approval_ledger_bundle(report_output_dir, report.to_dict(), list(report.operator_queue or []), approval_view="all")
+    updated_record = next((item for item in updated_bundle.get("approval_ledger", []) if item.get("approval_id") == approval_id), ledger_record)
+    if args.approve_governance or args.approve_packet:
+        receipt_payload = {**updated_record, **approval_record}
+        receipt_json, receipt_md = write_approval_receipt(report_output_dir, report.username, generated_at=_utcnow(), receipt=receipt_payload)
+        print_info(receipt_payload.get("summary", "Local approval captured."))
+        print_info(f"Approval receipt JSON: {receipt_json}")
+        print_info(f"Approval receipt Markdown: {receipt_md}")
+    else:
+        receipt_payload = {**updated_record, **followup_event}
+        receipt_json, receipt_md = write_followup_review_receipt(report_output_dir, report.username, generated_at=_utcnow(), receipt=receipt_payload)
+        print_info(receipt_payload.get("summary", "Local follow-up review captured."))
+        print_info(f"Approval follow-up receipt JSON: {receipt_json}")
+        print_info(f"Approval follow-up receipt Markdown: {receipt_md}")
+    print_info(f"Approval center JSON: {approval_json}")
+    print_info(f"Approval center Markdown: {approval_md}")

src/app/auto_apply.py

@@ -0,0 +1,143 @@
+"""Approved-campaign auto-apply flow.
+
+Imports remain local to the flow so optional operator dependencies keep their
+existing mode-specific loading behavior.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from src.cli_output import print_info
+from src.control_center_report_state import refresh_latest_report_state
+from src.portfolio_truth_types import truth_latest_path
+
+
+def run_auto_apply_approved_mode(args, output_dir: Path) -> None:
+    """Apply approved campaign packets for repos that pass the automation trust bar."""
+    from src.approval_ledger import load_approval_ledger_bundle
+    from src.auto_apply import (
+        build_trust_bar_index,
+        filter_safe_actions,
+        filter_trusted_repo_actions,
+        get_approved_manual_campaigns,
+        summarize_trust_bar,
+    )
+    from src.github_client import GitHubClient
+    from src.ops_writeback import (
+        apply_github_writeback,
+        build_campaign_bundle,
+        summarize_writeback_results,
+    )
+    from src.warehouse import load_latest_campaign_state
+
+    cache = None if getattr(args, "no_cache", False) else None
+    client: GitHubClient | None = (
+        GitHubClient(token=args.token, cache=cache) if getattr(args, "token", None) else None
+    )
+
+    try:
+        _report_path, _diff_dict, report = refresh_latest_report_state(output_dir, args)
+    except FileNotFoundError:
+        print_info("No existing audit report found in output directory. Run a normal audit first.")
+        return
+
+    truth_path = truth_latest_path(output_dir)
+    if not truth_path.exists():
+        print_info("No portfolio truth snapshot found. Run --portfolio-truth first.")
+        return
+
+    truth_snapshot = json.loads(truth_path.read_text())
+    decision_quality_status = (
+        (report.operator_summary or {})
+        .get("decision_quality_v1", {})
+        .get("decision_quality_status", "")
+    )
+    trust_bar_index = build_trust_bar_index(truth_snapshot, decision_quality_status)
+    trust_bar_summary = summarize_trust_bar(truth_snapshot, decision_quality_status)
+    print_info(
+        "Automation trust bar: "
+        f"{trust_bar_summary['automation_eligible_count']} opted-in repos; "
+        f"{trust_bar_summary['baseline_eligible_count']} baseline opted-in repos; "
+        f"{trust_bar_summary['trusted_repo_count']} repos pass the full trust bar "
+        f"(decision quality: {trust_bar_summary['decision_quality_status']})."
+    )
+    if trust_bar_summary["automation_eligible_repos"]:
+        print_info(
+            "Automation-eligible repos: "
+            + ", ".join(trust_bar_summary["automation_eligible_repos"])
+        )
+
+    bundle = load_approval_ledger_bundle(
+        output_dir,
+        report.to_dict(),
+        list(report.operator_queue or []),
+        approval_view="all",
+    )
+    approved_campaigns = get_approved_manual_campaigns(bundle)
+
+    if not approved_campaigns:
+        print_info("No approved-manual campaign packets found.")
+        return
+
+    total_applied = 0
+    total_skipped = 0
+    for record in approved_campaigns:
+        campaign_type = str(record.get("subject_key") or "")
+        if not campaign_type:
+            continue
+        _campaign_summary, actions = build_campaign_bundle(
+            report.to_dict(),
+            campaign_type=campaign_type,
+            portfolio_profile=getattr(args, "portfolio_profile", None),
+            collection_name=getattr(args, "collection", None),
+            max_actions=None,
+            writeback_target="github",
+        )
+        safe_actions = filter_safe_actions(actions)
+        trusted_actions = filter_trusted_repo_actions(safe_actions, trust_bar_index)
+
+        if not trusted_actions:
+            skipped_repos = {str(a.get("repo") or "") for a in actions} - {
+                str(a.get("repo") or "") for a in trusted_actions
+            }
+            print_info(
+                f"Campaign {campaign_type!r}: 0 eligible actions "
+                f"(skipped repos: {', '.join(sorted(skipped_repos)) or 'none'})"
+            )
+            total_skipped += len(actions)
+            continue
+
+        if getattr(args, "dry_run", False):
+            print_info(
+                f"Campaign {campaign_type!r}: {len(trusted_actions)} eligible actions "
+                "but dry-run mode is enabled; no GitHub writes were attempted."
+            )
+            continue
+
+        if client is None:
+            print_info(
+                f"Campaign {campaign_type!r}: {len(trusted_actions)} eligible actions "
+                "but no GitHub client available (dry run)."
+            )
+            continue
+
+        previous_state = load_latest_campaign_state(output_dir, campaign_type)
+        github_results, _refs, _drift, _closure = apply_github_writeback(
+            client,
+            trusted_actions,
+            previous_state=previous_state,
+            sync_mode=str(record.get("sync_mode") or "reconcile"),
+            campaign_summary=_campaign_summary,
+            github_projects_config=None,
+            operator_context={},
+        )
+        summary = summarize_writeback_results(github_results, "github", apply=True)
+        applied_count = int(summary.get("applied_count", 0))
+        total_applied += applied_count
+        print_info(
+            f"Campaign {campaign_type!r}: applied {applied_count} / {len(trusted_actions)} actions."
+        )
+
+    print_info(f"Auto-apply complete: {total_applied} applied, {total_skipped} skipped.")