diff --git a/src/operator_control_center_artifacts.py b/src/operator_control_center_artifacts.py index 6f8309a..57364d3 100644 --- a/src/operator_control_center_artifacts.py +++ b/src/operator_control_center_artifacts.py @@ -69,6 +69,7 @@ def write_control_center_artifacts( snapshot, diff_data=diff_dict, portfolio_truth=portfolio_truth, + portfolio_truth_history_dir=portfolio_truth_path.parent if portfolio_truth_path else None, portfolio_truth_reference=str(portfolio_truth_path) if portfolio_truth_path else "", control_center_reference=str(json_path), report_reference=report_reference, diff --git a/src/portfolio_truth_trends.py b/src/portfolio_truth_trends.py new file mode 100644 index 0000000..ec182ff --- /dev/null +++ b/src/portfolio_truth_trends.py @@ -0,0 +1,373 @@ +"""Read-side trends for the canonical portfolio-truth snapshot history. + +The producer owns the truth artifact schema. This module only reads existing +``portfolio-truth-*.json`` snapshots and derives movement, so it cannot change +scores, verdicts, rollups, or the producer's output contract. +""" + +from __future__ import annotations + +import json +from collections import defaultdict +from pathlib import Path +from typing import Any, Mapping, Sequence + +DEFAULT_HISTORY_DIR = Path("output/history") +DEFAULT_MAX_SNAPSHOTS = 8 +CONTRACT_VERSION = "portfolio_truth_trends_v1" +_TREND_FIELDS = ("attention_state", "activity_status", "risk_tier") + + +def load_portfolio_truth_history( + history_dir: Path = DEFAULT_HISTORY_DIR, + *, + max_snapshots: int = DEFAULT_MAX_SNAPSHOTS, +) -> list[dict[str, Any]]: + """Load the last ``max_snapshots`` valid truth artifacts chronologically. + + The current producer has emitted dated artifacts directly under ``output`` + while the original history contract names ``output/history``. Both layouts + are read, with the conventional ``portfolio-truth-latest.json`` pointer + excluded because it is not a historical observation. + """ + if max_snapshots < 1: + return [] + + candidates: dict[Path, dict[str, Any]] = {} + for directory in _history_directories(history_dir): + if not directory.is_dir(): + continue + for path in directory.glob("portfolio-truth-*.json"): + if path.name == "portfolio-truth-latest.json": + continue + payload = _read_mapping(path) + if payload is None: + continue + generated_at = _text(payload.get("generated_at")) + candidates[path.resolve()] = { + "path": str(path), + "generated_at": generated_at, + "projects": list(payload.get("projects") or []) + if isinstance(payload.get("projects"), list) + else [], + } + + snapshots = sorted( + candidates.values(), + key=lambda item: (item["generated_at"], item["path"]), + ) + return snapshots[-max_snapshots:] + + +def build_verdict_transition_ledger( + history_dir: Path = DEFAULT_HISTORY_DIR, + *, + max_snapshots: int = DEFAULT_MAX_SNAPSHOTS, + current_snapshot: Mapping[str, Any] | None = None, + current_path: Path | None = None, +) -> dict[str, Any]: + """Derive truth-layer transitions from existing snapshot history. + + ``current_snapshot`` is useful to callers that have already loaded the + canonical ``portfolio-truth-latest.json`` pointer. It is merged into the + read window by generated timestamp and does not get written anywhere. + """ + snapshots = load_portfolio_truth_history(history_dir, max_snapshots=max_snapshots) + if current_snapshot is not None and max_snapshots > 0: + snapshots = _with_current_snapshot( + snapshots, + current_snapshot, + current_path=current_path, + max_snapshots=max_snapshots, + ) + + observations = [_snapshot_observations(snapshot) for snapshot in snapshots] + transitions = _build_transitions(snapshots, observations) + activity_streaks = _build_activity_streaks(snapshots, observations) + attention_transitions = [ + item for item in transitions if item["kind"] == "attention_state" + ] + activity_transitions = [ + item for item in transitions if item["kind"] == "activity_status" + ] + risk_changes = [item for item in transitions if item["kind"] == "risk_tier"] + lifecycle_events = [ + item for item in transitions if item["kind"] == "repo_lifecycle" + ] + + return { + "contract_version": CONTRACT_VERSION, + "snapshot_count": len(snapshots), + "window_size": max_snapshots, + "from_generated_at": snapshots[0]["generated_at"] if snapshots else "", + "to_generated_at": snapshots[-1]["generated_at"] if snapshots else "", + "transitions": transitions, + "attention_state_transitions": attention_transitions, + "activity_status_transitions": activity_transitions, + "activity_status_streaks": activity_streaks, + "risk_tier_changes": risk_changes, + "repo_lifecycle_events": lifecycle_events, + "summary": { + "transition_count": len(transitions), + "attention_state_transition_count": len(attention_transitions), + "activity_status_transition_count": len(activity_transitions), + "risk_tier_change_count": len(risk_changes), + "repo_appeared_count": sum( + item["to"] == "present" for item in lifecycle_events + ), + "repo_disappeared_count": sum( + item["to"] == "absent" for item in lifecycle_events + ), + }, + } + + +def render_movement_summary(ledger: Mapping[str, Any]) -> str: + """Render a compact human-readable summary for the weekly digest.""" + transitions = list(ledger.get("transitions") or []) + if not transitions: + return "No verdict movement is recorded across the available truth snapshots." + + grouped: dict[tuple[str, str, str], int] = defaultdict(int) + for item in transitions: + kind = _text(item.get("kind")) + if kind == "repo_lifecycle": + grouped[(kind, _text(item.get("from")), _text(item.get("to")))] += 1 + else: + grouped[(kind, _text(item.get("from")), _text(item.get("to")))] += 1 + + clauses: list[str] = [] + for (kind, previous, current), count in sorted(grouped.items()): + transition = f"{previous}→{current}" + if kind == "repo_lifecycle": + verb = "appeared" if current == "present" else "disappeared" + clauses.append(f"{count} repo{'s' if count != 1 else ''} {verb}") + elif kind == "activity_status": + verb = _movement_verb(previous, current) + clauses.append(f"{count} repo{'s' if count != 1 else ''} {verb} {transition}") + elif kind == "attention_state": + verb = _movement_verb(previous, current) + clauses.append(f"{count} repo{'s' if count != 1 else ''} {verb} {transition}") + elif kind == "risk_tier": + clauses.append( + f"{count} risk-tier change{'s' if count != 1 else ''} {transition}" + ) + + return "; ".join(clauses) + "." + + +def _history_directories(history_dir: Path) -> list[Path]: + directories = [history_dir] + if history_dir.name != "history": + directories.append(history_dir / "history") + else: + directories.append(history_dir.parent) + return list(dict.fromkeys(directory for directory in directories)) + + +def _with_current_snapshot( + snapshots: Sequence[dict[str, Any]], + current_snapshot: Mapping[str, Any], + *, + current_path: Path | None, + max_snapshots: int, +) -> list[dict[str, Any]]: + generated_at = _text(current_snapshot.get("generated_at")) + if not generated_at: + return list(snapshots)[-max_snapshots:] + current = { + "path": str(current_path) if current_path else "", + "generated_at": generated_at, + "projects": list(current_snapshot.get("projects") or []) + if isinstance(current_snapshot.get("projects"), list) + else [], + } + merged = [snapshot for snapshot in snapshots if snapshot["generated_at"] != generated_at] + merged.append(current) + merged.sort(key=lambda item: (item["generated_at"], item["path"])) + return merged[-max_snapshots:] + + +def _snapshot_observations(snapshot: Mapping[str, Any]) -> dict[str, dict[str, Any]]: + observations: dict[str, dict[str, Any]] = {} + for project in snapshot.get("projects") or []: + if not isinstance(project, Mapping): + continue + identity = _mapping(project.get("identity")) + derived = _mapping(project.get("derived")) + risk = _mapping(project.get("risk")) + project_key = ( + _text(identity.get("project_key")) + or _text(identity.get("repo_full_name")) + or _text(identity.get("display_name")) + ) + if not project_key: + continue + observations[project_key] = { + "repo": _text(identity.get("display_name")) or project_key, + "project_key": project_key, + "attention_state": _value(derived.get("attention_state")), + "activity_status": _value(derived.get("activity_status")), + "risk_tier": _value(risk.get("risk_tier")), + } + return observations + + +def _build_transitions( + snapshots: Sequence[Mapping[str, Any]], + observations: Sequence[dict[str, dict[str, Any]]], +) -> list[dict[str, Any]]: + transitions: list[dict[str, Any]] = [] + for index in range(1, len(snapshots)): + previous = observations[index - 1] + current = observations[index] + previous_at = _text(snapshots[index - 1].get("generated_at")) + current_at = _text(snapshots[index].get("generated_at")) + for project_key in sorted(set(previous) | set(current)): + before = previous.get(project_key) + after = current.get(project_key) + if before is None or after is None: + transitions.append( + _transition( + project_key=project_key, + repo=(after or before)["repo"], + kind="repo_lifecycle", + previous="present" if before else "absent", + current="present" if after else "absent", + previous_at=previous_at, + current_at=current_at, + ) + ) + continue + for field in _TREND_FIELDS: + if before[field] == after[field]: + continue + transitions.append( + _transition( + project_key=project_key, + repo=after["repo"], + kind=field, + previous=before[field], + current=after[field], + previous_at=previous_at, + current_at=current_at, + ) + ) + transitions.sort( + key=lambda item: ( + item["to_generated_at"], + item["repo"].lower(), + item["kind"], + ) + ) + return transitions + + +def _build_activity_streaks( + snapshots: Sequence[Mapping[str, Any]], + observations: Sequence[dict[str, dict[str, Any]]], +) -> list[dict[str, Any]]: + keys = sorted({project_key for snapshot in observations for project_key in snapshot}) + streaks: list[dict[str, Any]] = [] + for project_key in keys: + present = [ + (index, snapshot[project_key]) + for index, snapshot in enumerate(observations) + if project_key in snapshot + ] + if not present: + continue + latest_index, latest = present[-1] + if latest_index != len(observations) - 1: + status = "disappeared" + run_count = 1 + since_index = latest_index + else: + status = latest["activity_status"] or "unknown" + run_count = 0 + since_index = latest_index + for index, observation in reversed(present): + if index != since_index or observation["activity_status"] != status: + break + run_count += 1 + since_index = index - 1 + since_index += 1 + streaks.append( + { + "repo": latest["repo"], + "project_key": project_key, + "status": status, + "run_count": run_count, + "consecutive_snapshots": run_count, + "since_generated_at": ( + _text(snapshots[since_index].get("generated_at")) + if snapshots + else "" + ), + "through_generated_at": ( + _text(snapshots[-1].get("generated_at")) if snapshots else "" + ), + } + ) + return streaks + + +def _transition( + *, + project_key: str, + repo: str, + kind: str, + previous: str | None, + current: str | None, + previous_at: str, + current_at: str, +) -> dict[str, Any]: + return { + "repo": repo, + "project_key": project_key, + "kind": kind, + "from": previous, + "to": current, + "from_date": previous_at[:10], + "to_date": current_at[:10], + "from_generated_at": previous_at, + "to_generated_at": current_at, + } + + +def _movement_verb(previous: str, current: str) -> str: + if previous in {"active", "recent", "active-product", "active-infra"} and current in { + "stale", + "archived", + "manual-only", + }: + return "slid" + if previous in {"stale", "archived", "manual-only", "decision-needed"} and current in { + "active", + "recent", + "active-product", + "active-infra", + }: + return "recovered" + return "changed" + + +def _read_mapping(path: Path) -> dict[str, Any] | None: + try: + payload = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError): + return None + return payload if isinstance(payload, dict) else None + + +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _text(value: Any) -> str: + return value.strip() if isinstance(value, str) else "" + + +def _value(value: Any) -> str | None: + text = _text(value) + return text or None diff --git a/src/weekly_command_center.py b/src/weekly_command_center.py index 55fd17e..bcaaf9d 100644 --- a/src/weekly_command_center.py +++ b/src/weekly_command_center.py @@ -8,6 +8,10 @@ from src.portfolio_automation import select_automation_candidates from src.portfolio_decision_queue import build_decision_queue, summarize_decision_queue from src.portfolio_truth_types import truth_latest_path +from src.portfolio_truth_trends import ( + build_verdict_transition_ledger, + render_movement_summary, +) from src.report_enrichment import build_weekly_review_pack CONTRACT_VERSION = "weekly_command_center_digest_v1" @@ -118,6 +122,7 @@ def build_weekly_command_center_digest( *, diff_data: dict[str, Any] | None = None, portfolio_truth: dict[str, Any] | None = None, + portfolio_truth_history_dir: Path | None = None, portfolio_truth_reference: str = "", control_center_reference: str = "", report_reference: str = "", @@ -136,6 +141,13 @@ def build_weekly_command_center_digest( freshness = _source_freshness(report_data, truth) source_is_stale = freshness["status"] != "current" truth_summary = _build_truth_summary(truth) + movement = build_verdict_transition_ledger( + portfolio_truth_history_dir, + max_snapshots=8, + current_snapshot=truth, + current_path=Path(portfolio_truth_reference) if portfolio_truth_reference else None, + ) if portfolio_truth_history_dir else build_verdict_transition_ledger(max_snapshots=0) + movement["summary_text"] = render_movement_summary(movement) decision_queue = build_decision_queue(truth) decision_queue_summary = summarize_decision_queue(decision_queue) operator_decision = _operator_decision(operator_summary, operator_queue) @@ -215,6 +227,7 @@ def build_weekly_command_center_digest( "authority_cap": _safe_text(decision_quality.get("authority_cap")) or AUTHORITY_CAP, }, "portfolio_truth": {**truth_summary, **decision_queue_summary}, + "movement": movement, "decision_queue": decision_queue, "path_attention": _build_path_attention_items(truth), "automation_candidates": [ @@ -260,6 +273,7 @@ def render_weekly_command_center_markdown(digest: dict[str, Any]) -> str: risk_posture = _mapping(digest.get("risk_posture")) tier_counts = _mapping(risk_posture.get("risk_tier_counts")) security_posture = _mapping(digest.get("security_posture")) + movement = _mapping(digest.get("movement")) lines = [ f"# Weekly Command Center: {_safe_text(digest.get('username')) or 'unknown'}", "", @@ -345,6 +359,19 @@ def render_weekly_command_center_markdown(digest: dict[str, Any]) -> str: "(re-run with `--portfolio-truth-include-security`)." ) + lines.extend(["", "## Movement"]) + lines.append( + f"- {_safe_text(movement.get('summary_text')) or 'No verdict movement is recorded across the available truth snapshots.'}" + ) + for item in list(movement.get("transitions") or []): + if item.get("kind") == "repo_lifecycle": + continue + lines.append( + f"- **{_safe_text(item.get('repo'))}** [{_safe_text(item.get('kind'))}]: " + f"{_safe_text(item.get('from')) or 'unknown'}→{_safe_text(item.get('to')) or 'unknown'} " + f"({_safe_text(item.get('to_date')) or 'undated'})" + ) + lines.extend(["", "## Weekly Sections"]) for section in list(digest.get("section_digest") or []): lines.append( diff --git a/tests/test_portfolio_truth_trends.py b/tests/test_portfolio_truth_trends.py new file mode 100644 index 0000000..791122d --- /dev/null +++ b/tests/test_portfolio_truth_trends.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from src.portfolio_truth_trends import ( + build_verdict_transition_ledger, + render_movement_summary, +) +from src.weekly_command_center import ( + build_weekly_command_center_digest, + render_weekly_command_center_markdown, +) + + +def _project( + name: str, + *, + attention: str = "active-product", + activity: str = "active", + risk: str = "baseline", +) -> dict: + return { + "identity": {"project_key": name.lower(), "display_name": name}, + "derived": { + "attention_state": attention, + "activity_status": activity, + }, + "risk": {"risk_tier": risk}, + } + + +def _write_snapshot(directory: Path, stamp: str, projects: list[dict]) -> dict: + snapshot = { + "generated_at": f"{stamp}T12:00:00+00:00", + "projects": projects, + } + (directory / f"portfolio-truth-{stamp.replace('-', '')}T120000Z.json").write_text( + json.dumps(snapshot) + ) + return snapshot + + +def test_transition_ledger_reports_no_movement_and_activity_streaks(tmp_path: Path) -> None: + _write_snapshot(tmp_path, "2026-07-10", [_project("Steady")]) + _write_snapshot(tmp_path, "2026-07-11", [_project("Steady")]) + + ledger = build_verdict_transition_ledger(tmp_path) + + assert ledger["snapshot_count"] == 2 + assert ledger["transitions"] == [] + assert ledger["summary"]["transition_count"] == 0 + assert ledger["activity_status_streaks"] == [ + { + "repo": "Steady", + "project_key": "steady", + "status": "active", + "run_count": 2, + "consecutive_snapshots": 2, + "since_generated_at": "2026-07-10T12:00:00+00:00", + "through_generated_at": "2026-07-11T12:00:00+00:00", + } + ] + assert render_movement_summary(ledger).startswith("No verdict movement") + + +def test_transition_ledger_reports_attention_activity_and_risk_changes(tmp_path: Path) -> None: + _write_snapshot( + tmp_path, + "2026-07-10", + [_project("Moving", attention="decision-needed", activity="stale", risk="elevated")], + ) + current = _write_snapshot( + tmp_path, + "2026-07-11", + [_project("Moving", attention="active-infra", activity="active", risk="baseline")], + ) + + ledger = build_verdict_transition_ledger( + tmp_path, + current_snapshot=current, + current_path=tmp_path / "portfolio-truth-latest.json", + ) + + assert [(item["kind"], item["from"], item["to"]) for item in ledger["transitions"]] == [ + ("activity_status", "stale", "active"), + ("attention_state", "decision-needed", "active-infra"), + ("risk_tier", "elevated", "baseline"), + ] + assert all(item["from_date"] == "2026-07-10" for item in ledger["transitions"]) + assert all(item["to_date"] == "2026-07-11" for item in ledger["transitions"]) + assert "recovered stale→active" in render_movement_summary(ledger) + assert "recovered decision-needed→active-infra" in render_movement_summary(ledger) + + +def test_transition_ledger_reports_repos_appearing_and_disappearing(tmp_path: Path) -> None: + _write_snapshot(tmp_path, "2026-07-10", [_project("Stays"), _project("Leaves")]) + _write_snapshot(tmp_path, "2026-07-11", [_project("Stays")]) + _write_snapshot(tmp_path, "2026-07-12", [_project("Stays"), _project("Arrives")]) + + ledger = build_verdict_transition_ledger(tmp_path) + + lifecycle = { + (item["repo"], item["from"], item["to"]) + for item in ledger["repo_lifecycle_events"] + } + assert lifecycle == { + ("Leaves", "present", "absent"), + ("Arrives", "absent", "present"), + } + streaks = {item["repo"]: item for item in ledger["activity_status_streaks"]} + assert streaks["Leaves"]["status"] == "disappeared" + assert streaks["Arrives"]["run_count"] == 1 + assert ledger["summary"]["repo_appeared_count"] == 1 + assert ledger["summary"]["repo_disappeared_count"] == 1 + + +def test_weekly_digest_surfaces_movement_section(tmp_path: Path) -> None: + _write_snapshot( + tmp_path, + "2026-07-10", + [_project("Moving", attention="decision-needed", activity="stale")], + ) + current = _write_snapshot( + tmp_path, + "2026-07-11", + [_project("Moving", attention="active-infra", activity="active")], + ) + report_data = { + "username": "testuser", + "generated_at": "2026-07-11T12:00:00+00:00", + "operator_summary": {"decision_quality_v1": {}}, + "audits": [], + } + snapshot = {"operator_summary": report_data["operator_summary"], "operator_queue": []} + + digest = build_weekly_command_center_digest( + report_data, + snapshot, + portfolio_truth=current, + portfolio_truth_history_dir=tmp_path, + portfolio_truth_reference=str(tmp_path / "portfolio-truth-latest.json"), + ) + rendered = render_weekly_command_center_markdown(digest) + + assert digest["movement"]["attention_state_transitions"][0]["repo"] == "Moving" + assert "recovered decision-needed→active-infra" in digest["movement"]["summary_text"] + assert "## Movement" in rendered + assert "Moving" in rendered + assert "2026-07-11" in rendered