generated from coulomb/repo-seed
tegwick
0d05dfcc5d
The analysis half of the weekly coding retrospection. retro/build.py: windowed detect+measure -> top-3 improvement suggestions per repo (cross-flavor first, recommendations pulled from the Pattern Catalog) + fleet snapshot. retro/publish.py: publishes the report to the hub as the coding_retro read model (event_type= coding_retro progress event) + local JSON/md, graceful degrade. retro entrypoint with --window-days/--publish/--json. Live verify over real sessions surfaced per-repo suggestions with catalog recommendations. 13 new tests; suite 152/152. Consumed by activity-core ACTIVITY-WP-0008 (Weekly Coding Retrospection, Sat 19:00). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
"""Windowed weekly retro report (AGENTIC-WP-0010 T01).
|
|
|
|
Runs the existing detect pipeline over a date window, ranks the recurring problem
|
|
patterns into **per-repo improvement suggestions** (top 3, cross-flavor first),
|
|
attaches a recommendation from the Pattern Catalog where one exists, and bundles a
|
|
fleet measure snapshot for context. Pure function over digests — the entrypoint
|
|
(T03) handles store/publish.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import collections
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from ..curate.schema import SolutionPattern
|
|
from ..detect.cluster import cluster
|
|
from ..detect.quality import QualityConfig, filter_real
|
|
from ..detect.signals import extract_signals
|
|
from ..measure.metrics import aggregate
|
|
|
|
# score at/above which a suggestion is "high" priority even when single-flavor
|
|
_HIGH_SCORE = 100.0
|
|
|
|
|
|
def _parse(ts: str) -> datetime:
|
|
return datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
|
|
|
|
|
def _iso(dt: datetime) -> str:
|
|
return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
@dataclass
|
|
class Suggestion:
|
|
repo: str
|
|
title: str
|
|
recommendation: str
|
|
priority: str # high | medium
|
|
score: float
|
|
signal_type: str
|
|
cross_flavor: bool
|
|
pattern_key: str
|
|
|
|
|
|
def _recommendation(pattern_key: str, catalog) -> Optional[str]:
|
|
if catalog is None:
|
|
return None
|
|
sp = catalog.load(SolutionPattern.make_id(pattern_key))
|
|
if sp and sp.resolutions:
|
|
return sp.resolutions[0].summary
|
|
return None
|
|
|
|
|
|
def weekly_retro(digests: list[dict], catalog=None, *, since: Optional[str] = None,
|
|
until: Optional[str] = None, window_days: int = 7,
|
|
max_per_repo: int = 3, min_frequency: int = 2,
|
|
quality: Optional[QualityConfig] = None) -> dict:
|
|
"""Build the ranked weekly retro report over a date window."""
|
|
until_dt = _parse(until) if until else _now()
|
|
since_dt = _parse(since) if since else until_dt - timedelta(days=window_days)
|
|
|
|
windowed = [d for d in digests
|
|
if d.get("started_at") and since_dt <= _parse(d["started_at"]) < until_dt]
|
|
real = filter_real(windowed, quality or QualityConfig())
|
|
|
|
patterns = cluster(extract_signals(real), min_frequency=min_frequency)
|
|
|
|
by_repo: dict[str, list[Suggestion]] = collections.defaultdict(list)
|
|
for p in patterns:
|
|
if p.polarity != "problem":
|
|
continue # improvements come from problems
|
|
rec = (_recommendation(p.key, catalog)
|
|
or f"Investigate {p.signal_type.replace('_', ' ')} on {p.locus}")
|
|
priority = "high" if (p.cross_flavor or p.score >= _HIGH_SCORE) else "medium"
|
|
for repo in (p.repos or ["(unknown)"]):
|
|
by_repo[repo].append(Suggestion(
|
|
repo=repo, title=p.title, recommendation=rec, priority=priority,
|
|
score=p.score, signal_type=p.signal_type, cross_flavor=p.cross_flavor,
|
|
pattern_key=p.key))
|
|
|
|
suggestions: list[Suggestion] = []
|
|
for repo in sorted(by_repo):
|
|
items = sorted(by_repo[repo], key=lambda s: -s.score)
|
|
suggestions.extend(items[:max_per_repo])
|
|
# cross-flavor first, then by score (global ordering for the report)
|
|
suggestions.sort(key=lambda s: (not s.cross_flavor, -s.score))
|
|
|
|
return {
|
|
"window": {"since": _iso(since_dt), "until": _iso(until_dt), "days": window_days},
|
|
"generated_at": _iso(_now()),
|
|
"n_sessions": len(real),
|
|
"suggestions": [asdict(s) for s in suggestions],
|
|
"measure": aggregate(real),
|
|
}
|