Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 21 additions & 5 deletions src/automation_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -159,9 +159,15 @@ def execute_context_pr(
try:
plan.apply_change(plan.repo_path)
except Exception as error: # noqa: BLE001 - surface as a failed result, not a strand
# The change failed after the branch was created; best-effort return to
# the default branch so the repo is not left stranded on a partial branch.
runner(["git", "checkout", default_branch], plan.repo_path)
# The change failed after the branch was created. apply_change may have
# left partial writes, so a plain checkout would be blocked ("local
# changes would be overwritten") and we'd be stranded on the orphan with
# the branch -D refused ("currently checked out"). Force-checkout the
# default branch — the worktree was verified clean before apply_change,
# so -f only discards its partial garbage — then delete the orphan so a
# retry isn't blocked by an "already exists" branch.
runner(["git", "checkout", "-f", default_branch], plan.repo_path)
runner(["git", "branch", "-D", branch], plan.repo_path)
return ExecutionResult(proposal.proposal_id, "failed", f"apply-change: {error}".strip())

for args in (
Expand Down Expand Up @@ -194,11 +200,21 @@ def execute_context_pr(
if pr.returncode != 0:
return ExecutionResult(proposal.proposal_id, "failed", f"gh pr create: {pr.stderr}".strip())

# gh prints the PR URL on success. If it somehow reported success with no
# URL, the PR was still created (rc 0) so we must not fail and re-run (that
# would duplicate it) — but surface the missing reference in the detail
# rather than silently recording an empty audit ref.
pr_url = pr.stdout.strip()
detail = (
f"Opened PR on branch {branch}."
if pr_url
else f"Opened PR on branch {branch} (gh returned no PR URL)."
)
return ExecutionResult(
proposal.proposal_id,
"applied",
f"Opened PR on branch {branch}.",
reference=pr.stdout.strip(),
detail,
reference=pr_url,
)


Expand Down
17 changes: 11 additions & 6 deletions src/automation_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,8 +53,10 @@
CONTRACT_VERSION = "automation_workflow_v1"

# Auto-PR branch defaults. ``default_branch`` is the PR base / the branch the
# executor refuses to commit onto; it is a parameter (not a buried literal) so
# callers can override per-repo, defaulting to the portfolio convention.
# executor refuses to commit onto. It resolves in precedence order: an explicit
# caller override, then the repo's own detected default branch
# (``identity.default_branch``), then this portfolio-wide fallback. Passing ""
# (the new default) means "auto-detect per repo".
DEFAULT_BRANCH_PREFIX = "auto/context-"
DEFAULT_DEFAULT_BRANCH = "main"

Expand All @@ -75,15 +77,18 @@ def build_context_pr_plan(
*,
workspace_root: Path,
branch_prefix: str = DEFAULT_BRANCH_PREFIX,
default_branch: str = DEFAULT_DEFAULT_BRANCH,
default_branch: str = "",
) -> ExecutionPlan:
"""Build the ``ExecutionPlan`` for a context-improvement auto-PR.

The plan's ``apply_change`` regenerates the managed context block from the
project's current repository signals; the executor handles all git/gh
mechanics (and every safety rail) around it.
mechanics (and every safety rail) around it. The PR base resolves in
precedence order: explicit ``default_branch`` arg, the repo's detected
default branch, then the portfolio-wide fallback.
"""
repo_path = workspace_root / project.identity.path
resolved_branch = default_branch or project.identity.default_branch or DEFAULT_DEFAULT_BRANCH
display = project.identity.display_name
commit_message = f"docs(context): refresh managed context block for {display}"
pr_body = (
Expand All @@ -95,7 +100,7 @@ def build_context_pr_plan(
)
return ExecutionPlan(
repo_path=repo_path,
default_branch=default_branch,
default_branch=resolved_branch,
branch_name=f"{branch_prefix}{_branch_slug(project)}",
commit_message=commit_message,
pr_title=commit_message,
Expand Down Expand Up @@ -148,7 +153,7 @@ def execute_approved_proposals(
dry_run: bool = True,
runner: CommandRunner = default_command_runner,
branch_prefix: str = DEFAULT_BRANCH_PREFIX,
default_branch: str = DEFAULT_DEFAULT_BRANCH,
default_branch: str = "",
) -> list[ExecutionResult]:
"""Execute every APPROVED proposal in the queue, behind the executor's rails.

Expand Down
23 changes: 12 additions & 11 deletions src/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
from src.cloner import clone_workspace
from src.github_client import GitHubClient
from src.models import AnalyzerResult, AuditReport, RepoAudit, RepoMetadata
from src.portfolio_truth_types import TRUTH_LATEST_FILENAME, truth_latest_path
from src.recurring_review import FULL_REFRESH_DAYS
from src.report_enrichment import build_run_change_counts, build_run_change_summary
from src.reporter import (
Expand Down Expand Up @@ -2742,7 +2743,7 @@ def _run_plan_campaign_mode(args) -> None:
reviewer: str = getattr(args, "approval_reviewer", None) or _default_reviewer()

# ── Load audit results from portfolio-truth-latest.json ───────────────────
truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if not truth_path.exists():
print_info(
f"portfolio-truth-latest.json not found in {output_dir}. "
Expand Down Expand Up @@ -2870,7 +2871,7 @@ def _run_draft_readmes_mode(args) -> None:

# ── Load audit results (portfolio-truth-latest.json or warehouse) ─────────
audit_results: list[dict] = []
truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if truth_path.exists():
try:
raw = json.loads(truth_path.read_text(encoding="utf-8"))
Expand Down Expand Up @@ -3036,7 +3037,7 @@ def _run_set_initiative_mode(args) -> None:
# Load portfolio-truth to validate repo and check current tier
import json as _json

pt_candidates = sorted(output_dir.glob("portfolio-truth-latest.json"))
pt_candidates = sorted(output_dir.glob(TRUTH_LATEST_FILENAME))
if not pt_candidates:
pt_candidates = sorted(output_dir.glob("portfolio-truth-*.json"))
if not pt_candidates:
Expand All @@ -3046,7 +3047,7 @@ def _run_set_initiative_mode(args) -> None:
)
sys.exit(2)

pt_path = Path(str(output_dir / "portfolio-truth-latest.json"))
pt_path = truth_latest_path(output_dir)
if not pt_path.exists():
pt_path = pt_candidates[-1]

Expand Down Expand Up @@ -3109,7 +3110,7 @@ def _run_list_initiatives_mode(args) -> None:

# Load portfolio-truth for current-tier lookup (best-effort)
projects_by_name: dict[str, dict] = {}
pt_path = output_dir / "portfolio-truth-latest.json"
pt_path = truth_latest_path(output_dir)
if pt_path.exists():
try:
pt_data = _json.loads(pt_path.read_text(encoding="utf-8"))
Expand Down Expand Up @@ -3194,7 +3195,7 @@ def _run_suggest_initiatives_mode(args) -> None:
from src.maturity_tiers import tier_name
from src.suggest_initiatives import generate_suggestions

truth_path = _Path(args.output_dir) / "portfolio-truth-latest.json"
truth_path = truth_latest_path(_Path(args.output_dir))
if not truth_path.exists():
print_warning(
"portfolio-truth-latest.json not found. "
Expand Down Expand Up @@ -3238,7 +3239,7 @@ def _run_accept_suggestion_mode(args) -> None:
from src.suggest_initiatives import accept_suggestion

output_dir = Path(args.output_dir)
truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if not truth_path.exists():
print_warning(
"portfolio-truth-latest.json not found. Run `audit run --portfolio-truth` first."
Expand Down Expand Up @@ -3369,7 +3370,7 @@ def _run_tier_gaps_export_mode(args) -> None:
from src.maturity_tiers import compute_tier, tier_gap, tier_name

output_dir = Path(args.output_dir)
truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if not truth_path.exists():
print_warning(
"portfolio-truth-latest.json not found. Run `audit run --portfolio-truth` first."
Expand Down Expand Up @@ -5114,7 +5115,7 @@ def _run_auto_apply_approved_mode(args, output_dir: Path) -> None:
print_info("No existing audit report found in output directory. Run a normal audit first.")
return

truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if not truth_path.exists():
print_info("No portfolio truth snapshot found. Run --portfolio-truth first.")
return
Expand Down Expand Up @@ -5586,7 +5587,7 @@ def _run_tier_recalibration_report_mode(args) -> None:
from src.tier_recalibration import tier_distribution_report

output_dir = Path(args.output_dir)
truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if not truth_path.exists():
print_warning(
"portfolio-truth-latest.json not found. Run `audit run --portfolio-truth` first."
Expand Down Expand Up @@ -5626,7 +5627,7 @@ def _run_context_triage_mode(args) -> None:
from src.portfolio_context_triage import run_triage

output_dir = Path(args.output_dir)
truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if not truth_path.exists():
print_warning(
"portfolio-truth-latest.json not found. Run `audit run --portfolio-truth` first."
Expand Down
4 changes: 3 additions & 1 deletion src/excel_export_truth_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,12 +5,14 @@
import json
from pathlib import Path

from src.portfolio_truth_types import truth_latest_path


def load_risk_truth(truth_dir: Path | None) -> tuple[dict[str, str], dict[str, int]]:
if not truth_dir:
return {}, {}

truth_path = truth_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(truth_dir)
if not truth_path.is_file():
return {}, {}

Expand Down
3 changes: 2 additions & 1 deletion src/portfolio_truth_publish.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@

from src.portfolio_truth_reconcile import build_portfolio_truth_snapshot
from src.portfolio_truth_render import render_portfolio_report_markdown, render_registry_markdown
from src.portfolio_truth_types import truth_latest_path
from src.portfolio_truth_validate import (
validate_portfolio_report_markdown,
validate_publish_targets,
Expand Down Expand Up @@ -56,7 +57,7 @@ def publish_portfolio_truth(

snapshot_stamp = build_result.snapshot.generated_at.strftime("%Y-%m-%dT%H%M%SZ")
snapshot_path = output_dir / f"portfolio-truth-{snapshot_stamp}.json"
latest_path = output_dir / "portfolio-truth-latest.json"
latest_path = truth_latest_path(output_dir)
latest_name = latest_path.name
snapshot_json = json.dumps(build_result.snapshot.to_dict(), indent=2) + "\n"
registry_markdown = render_registry_markdown(build_result.snapshot)
Expand Down
1 change: 1 addition & 0 deletions src/portfolio_truth_reconcile.py
Original file line number Diff line number Diff line change
Expand Up @@ -348,6 +348,7 @@ def _build_truth_project(
section_label=_resolve_section_label(group_entry, raw_project),
has_git=bool(raw_project["has_git"]),
repo_full_name=str(raw_project.get("repo_full_name") or ""),
default_branch=str(raw_project.get("default_branch") or ""),
)

declared_values = {
Expand Down
63 changes: 45 additions & 18 deletions src/portfolio_truth_sources.py
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,7 @@ def _inspect_project_dir(
"group_entry": group_entry,
"has_git": bool(git_facts.get("has_git")),
"repo_full_name": str(git_facts.get("repo_full_name", "") or "").strip(),
"default_branch": str(git_facts.get("default_branch", "") or "").strip(),
"context_files": context_files,
"context_quality": context_analysis.context_quality,
"primary_context_file": context_analysis.primary_context_file,
Expand Down Expand Up @@ -405,7 +406,21 @@ def _read_small_json(path: Path) -> dict[str, Any]:
def _gather_git_facts(project_path: Path) -> dict[str, Any]:
git_dir = project_path / ".git"
if not git_dir.exists():
return {"has_git": False, "last_commit_at": None, "repo_full_name": ""}
return {
"has_git": False,
"last_commit_at": None,
"repo_full_name": "",
"default_branch": "",
}

# Computed once; ``last_commit_at`` is the only field the git-log probe below
# can refine, so every error path returns this base unchanged.
base = {
"has_git": True,
"last_commit_at": None,
"repo_full_name": _git_remote_full_name(project_path),
"default_branch": _git_default_branch(project_path),
}

try:
result = subprocess.run(
Expand All @@ -416,31 +431,43 @@ def _gather_git_facts(project_path: Path) -> dict[str, Any]:
check=False,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return {
"has_git": True,
"last_commit_at": None,
"repo_full_name": _git_remote_full_name(project_path),
}
return base

if result.returncode != 0 or not result.stdout.strip():
return {
"has_git": True,
"last_commit_at": None,
"repo_full_name": _git_remote_full_name(project_path),
}
return base

try:
return {
"has_git": True,
**base,
"last_commit_at": datetime.fromisoformat(result.stdout.strip().replace("Z", "+00:00")),
"repo_full_name": _git_remote_full_name(project_path),
}
except ValueError:
return {
"has_git": True,
"last_commit_at": None,
"repo_full_name": _git_remote_full_name(project_path),
}
return base


def _git_default_branch(project_path: Path) -> str:
"""The repo's default branch from the local ``origin/HEAD`` ref, if set.

Resolves only local refs (no network). Returns "" when ``origin/HEAD`` is
not set locally (common for repos that were ``git init``'d rather than
cloned) — callers fall back to the portfolio default.
"""
try:
result = subprocess.run(
["git", "-C", str(project_path), "symbolic-ref", "--short", "refs/remotes/origin/HEAD"],
capture_output=True,
text=True,
timeout=5,
check=False,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return ""

if result.returncode != 0:
return ""
# e.g. "origin/main" -> "main"; partition keeps multi-segment branch names
# like "origin/release/v1" -> "release/v1" intact.
return result.stdout.strip().partition("/")[2].strip()


def _git_remote_full_name(project_path: Path) -> str:
Expand Down
15 changes: 15 additions & 0 deletions src/portfolio_truth_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,22 @@
import dataclasses
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any

SCHEMA_VERSION = "0.5.0"

# The published "latest" portfolio-truth artifact. The producer
# (portfolio_truth_publish) writes it; every reader resolves it through
# truth_latest_path() so the filename lives in exactly one place.
TRUTH_LATEST_FILENAME = "portfolio-truth-latest.json"


def truth_latest_path(output_dir: Path) -> Path:
"""Resolve the canonical portfolio-truth-latest.json under an output dir."""
return output_dir / TRUTH_LATEST_FILENAME


VALID_CONTEXT_QUALITY = {"full", "standard", "minimum-viable", "boilerplate", "none"}
VALID_ACTIVITY_STATUS = {"active", "recent", "stale", "archived"}
VALID_REGISTRY_STATUS = {"active", "recent", "parked", "archived"}
Expand Down Expand Up @@ -47,6 +59,9 @@ class IdentityFields:
# metadata.name) and not only the local-dir display_name, which often differ
# (e.g. "Signal & Noise" vs "signal-noise").
repo_full_name: str = ""
# The repo's default branch (from local ``origin/HEAD``), when detectable.
# Empty when not set locally; consumers fall back to the portfolio default.
default_branch: str = ""

def to_dict(self) -> dict[str, Any]:
return dataclasses.asdict(self)
Expand Down
3 changes: 2 additions & 1 deletion src/report_enrichment.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from pathlib import Path
from typing import Any

from src.portfolio_truth_types import truth_latest_path
from src.terminology import ACTION_SYNC_CANONICAL_LABELS
from src.weekly_packaging import finalize_weekly_pack
from src.weekly_scheduling_overlay import apply_weekly_scheduling_overlay
Expand Down Expand Up @@ -164,7 +165,7 @@ def build_risk_lookup(output_dir: Path | None) -> dict[str, dict[str, str]]:
"""
if not output_dir:
return {}
truth_path = output_dir / "portfolio-truth-latest.json"
truth_path = truth_latest_path(output_dir)
if not truth_path.is_file():
return {}
try:
Expand Down
Loading