Merge pull request #19 from escoffier-labs/feat/brigade-work-recap

feat: recap brigade work sessions

Author: solomonneas

CHANGELOG.md

@@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `brigade work start` and `brigade work end` to create local `.brigade/work/` session artifacts for normal daily work loops.
 - `brigade work end --handoff` to write a Memory Handoff from closed work session artifacts.
 - `brigade work list`, `brigade work latest`, and `brigade work show` to inspect local work session artifacts.
+- `brigade work recap` to summarize recent or date-filtered work sessions.
 - Roster-level and per-agent `timeout_seconds` controls for bounded CLI calls.
 - `brigade run --read-only` prompt policy for planning and review runs that should inspect and recommend only, with native `codex exec --sandbox read-only` enforcement for Codex agents.
 

README.md

@@ -139,6 +139,7 @@ brigade work start "next slice"
 brigade work end --note "tests passed" --handoff
 brigade work list
 brigade work latest
+brigade work recap --since 2026-05-26
 ```
 
 `--dry-run` prints the planned assignments as JSON and stops before worker dispatch. `--show-plan` prints assignments before a normal run. `--verbose` prints the plan, worker statuses, and synthesis status. `--cwd` sets the working directory for the agent CLI calls and defaults to the current directory. `--handoff` writes a Memory Handoff for a successful non-dry run. `--inspect` prints the same readable artifact summary as `brigade runs show` after the run completes. `--read-only` tells the orchestrator and workers to inspect and recommend only, without modifying files or external state. For `codex` agents, Brigade also passes `codex exec --sandbox read-only`; other adapters receive the prompt policy only. The `cli` values are adapters for installed command-line tools: `codex`, `claude`, and `ollama:<model>`. Pick the ones you already use. Brigade shells out to those tools and keeps no provider keys. `brigade roster doctor` validates the roster syntax and reports which CLIs are present on `PATH`.
@@ -162,7 +163,7 @@ Use `--output-dir <path>` to pick the artifact directory, or `--no-artifacts` fo
 
 Use `brigade work status` as the quick daily dashboard for a repo. It reports the current branch, dirty files, dogfood readiness, configured dogfood paths, latest dogfood run, and extracted next step without starting a new orchestration. `brigade work start "title"` opens a local work session under `.brigade/work/<id>/`, records the starting git and dogfood context, and writes `start.md`. `brigade work end --note "what happened"` closes the active session, records ending context, and writes `end.md`. Add `--handoff` to also write a Memory Handoff for the closed session; it defaults to the configured dogfood handoff inbox or `.claude/memory-handoffs`.
 
-Inspect local work sessions with `brigade work list`, `brigade work latest`, or `brigade work show <session-id-or-path>`.
+Inspect local work sessions with `brigade work list`, `brigade work latest`, or `brigade work show <session-id-or-path>`. Use `brigade work recap` for a compact summary of recent sessions, or add `--since YYYY-MM-DD` for a day-range recap.
 
 Inspect a completed run without opening each JSON file:
 

src/brigade/cli.py

@@ -122,6 +122,10 @@ def _build_parser() -> argparse.ArgumentParser:
     p_work_show = work_sub.add_parser("show", help="Show one Brigade work session.")
     p_work_show.add_argument("session", help="Session id or path.")
     p_work_show.add_argument("--target", "-t", type=Path, default=Path("."), help="Repo or workspace to inspect.")
+    p_work_recap = work_sub.add_parser("recap", help="Summarize recent Brigade work sessions.")
+    p_work_recap.add_argument("--target", "-t", type=Path, default=Path("."), help="Repo or workspace to inspect.")
+    p_work_recap.add_argument("--limit", type=int, default=5, help="Maximum sessions to include.")
+    p_work_recap.add_argument("--since", default=None, help="Only include sessions since YYYY-MM-DD.")
     p_work_start = work_sub.add_parser("start", help="Start a local Brigade work session.")
     p_work_start.add_argument("title", nargs="*", help="Optional session title.")
     p_work_start.add_argument("--target", "-t", type=Path, default=Path("."), help="Repo or workspace for the session.")
@@ -414,6 +418,8 @@ def main(argv=None) -> int:
             return work_cmd.latest(target=args.target)
         if args.work_command == "show":
             return work_cmd.show(target=args.target, session=args.session)
+        if args.work_command == "recap":
+            return work_cmd.recap(target=args.target, limit=args.limit, since=args.since)
         if args.work_command == "start":
             title = " ".join(args.title) if args.title else None
             return work_cmd.start(target=args.target, title=title, force=args.force)

src/brigade/work_cmd.py

@@ -6,7 +6,7 @@
 import shutil
 import subprocess
 import sys
-from datetime import datetime, timezone
+from datetime import datetime, time, timezone
 from pathlib import Path
 from typing import Any
 from uuid import uuid4
@@ -125,6 +125,29 @@ def _session_sort_key(item: tuple[Path, dict[str, Any]]) -> str:
     return str(payload.get("ended_at") or payload.get("started_at") or path.name)
 
 
+def _parse_iso_datetime(value: object) -> datetime | None:
+    if not isinstance(value, str) or not value:
+        return None
+    normalized = value.replace("Z", "+00:00")
+    try:
+        parsed = datetime.fromisoformat(normalized)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _parse_since(value: str | None) -> datetime | None:
+    if value is None:
+        return None
+    try:
+        parsed_date = datetime.strptime(value, "%Y-%m-%d").date()
+    except ValueError as exc:
+        raise ValueError("--since must use YYYY-MM-DD") from exc
+    return datetime.combine(parsed_date, time.min, tzinfo=timezone.utc)
+
+
 def _collect_sessions(root: Path) -> tuple[list[tuple[Path, dict[str, Any]]], int]:
     sessions: list[tuple[Path, dict[str, Any]]] = []
     skipped = 0
@@ -157,6 +180,28 @@ def _dirty_count(snapshot: dict[str, Any]) -> int:
     return len(dirty) if isinstance(dirty, list) else 0
 
 
+def _snapshot(payload: dict[str, Any]) -> dict[str, Any]:
+    if isinstance(payload.get("end"), dict):
+        return payload["end"]
+    if isinstance(payload.get("start"), dict):
+        return payload["start"]
+    return {}
+
+
+def _branch(snapshot: dict[str, Any]) -> str | None:
+    git = snapshot.get("git")
+    if isinstance(git, dict) and isinstance(git.get("branch"), str):
+        return git["branch"]
+    return None
+
+
+def _next_step(snapshot: dict[str, Any]) -> str | None:
+    dogfood = snapshot.get("dogfood")
+    if isinstance(dogfood, dict) and isinstance(dogfood.get("next"), str):
+        return dogfood["next"]
+    return None
+
+
 def _display_session(path: Path, payload: dict[str, Any]) -> None:
     print(f"session: {path}")
     print(f"id: {payload.get('id', path.name)}")
@@ -465,6 +510,72 @@ def show(*, target: Path, session: str | Path) -> int:
     return 0
 
 
+def recap(*, target: Path, limit: int = 5, since: str | None = None) -> int:
+    if limit < 1:
+        print("error: --limit must be a positive integer", file=sys.stderr)
+        return 2
+    try:
+        since_dt = _parse_since(since)
+    except ValueError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+
+    target = target.expanduser().resolve()
+    if not target.is_dir():
+        print(f"error: --target is not a directory: {target}", file=sys.stderr)
+        return 2
+    root = _work_root(target)
+    sessions, skipped = _collect_sessions(root)
+    if since_dt is not None:
+        sessions = [
+            (path, payload)
+            for path, payload in sessions
+            if (_parse_iso_datetime(payload.get("ended_at") or payload.get("started_at")) or datetime.min.replace(tzinfo=timezone.utc))
+            >= since_dt
+        ]
+    sessions = sessions[:limit]
+
+    print(f"work recap: {target}")
+    if since:
+        print(f"since: {since}")
+    print(f"sessions: {len(sessions)}")
+    if skipped:
+        print(f"skipped: {skipped}", file=sys.stderr)
+    if not sessions:
+        print(f"no work sessions found in {root}")
+        return 0
+
+    branches = sorted({branch for _, payload in sessions if (branch := _branch(_snapshot(payload)))})
+    if branches:
+        print(f"branches: {', '.join(branches)}")
+    handoffs = [str(payload.get("handoff")) for _, payload in sessions if payload.get("handoff")]
+    if handoffs:
+        print(f"handoffs: {len(handoffs)}")
+
+    print("items:")
+    for path, payload in sessions:
+        snapshot = _snapshot(payload)
+        title = str(payload.get("title") or payload.get("id") or path.name)
+        print(f"- {title}")
+        print(f"  id: {payload.get('id', path.name)}")
+        print(f"  status: {payload.get('status', 'unknown')}")
+        print(f"  started: {payload.get('started_at', '')}")
+        if payload.get("ended_at"):
+            print(f"  ended: {payload['ended_at']}")
+        branch = _branch(snapshot)
+        if branch:
+            print(f"  branch: {branch}")
+        print(f"  dirty_files: {_dirty_count(snapshot)}")
+        if payload.get("note"):
+            print(f"  note: {_short(str(payload['note']))}")
+        if payload.get("handoff"):
+            print(f"  handoff: {payload['handoff']}")
+        next_text = _next_step(snapshot)
+        if next_text:
+            print(f"  next: {_short(next_text)}")
+    return 0
+
+
 def status(*, target: Path, limit: int = 12) -> int:
     if limit < 1:
         print("error: --limit must be a positive integer", file=sys.stderr)

tests/test_work_cmd.py

@@ -226,6 +226,48 @@ def test_work_latest_reports_no_sessions(tmp_path, capsys):
     assert "no work sessions found" in capsys.readouterr().err
 
 
+def test_work_recap_summarizes_recent_sessions(tmp_path, monkeypatch, capsys):
+    _init_git_repo(tmp_path)
+    times = iter(
+        [
+            datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc),
+            datetime(2026, 5, 25, 13, 0, 0, tzinfo=timezone.utc),
+            datetime(2026, 5, 26, 12, 0, 0, tzinfo=timezone.utc),
+            datetime(2026, 5, 26, 13, 0, 0, tzinfo=timezone.utc),
+        ]
+    )
+    monkeypatch.setattr(work_cmd, "_now", lambda: next(times))
+    dogfood_cmd.init(target=tmp_path)
+    run_dir = tmp_path / ".brigade" / "runs" / "latest"
+    run_dir.mkdir(parents=True)
+    _write_json(run_dir / "run.json", {"started_at": "2026-05-26T11:00:00Z", "status": "ok", "task": "review"})
+    (run_dir / "final.txt").write_text("Done.\n\nNext step: Build recap.\n")
+
+    assert work_cmd.start(target=tmp_path, title="Older Session") == 0
+    assert work_cmd.end(target=tmp_path, note="old note") == 0
+    assert work_cmd.start(target=tmp_path, title="Newer Session") == 0
+    assert work_cmd.end(target=tmp_path, note="new note", handoff=True, handoff_inbox=tmp_path / "handoffs") == 0
+
+    assert work_cmd.recap(target=tmp_path, since="2026-05-26", limit=5) == 0
+    out = capsys.readouterr().out
+    assert "work recap:" in out
+    assert "since: 2026-05-26" in out
+    assert "sessions: 1" in out
+    assert "branches:" in out
+    assert "handoffs: 1" in out
+    assert "Newer Session" in out
+    assert "Older Session" not in out
+    assert "note: new note" in out
+    assert "next: Build recap." in out
+
+
+def test_work_recap_rejects_bad_since(tmp_path, capsys):
+    _init_git_repo(tmp_path)
+
+    assert work_cmd.recap(target=tmp_path, since="05-26-2026") == 2
+    assert "--since must use YYYY-MM-DD" in capsys.readouterr().err
+
+
 def test_work_status_cli(tmp_path, monkeypatch):
     seen = {}
 
@@ -299,15 +341,22 @@ def fake_show(**kwargs):
         seen.append(("show", kwargs))
         return 0
 
+    def fake_recap(**kwargs):
+        seen.append(("recap", kwargs))
+        return 0
+
     monkeypatch.setattr(work_cmd, "list_sessions", fake_list)
     monkeypatch.setattr(work_cmd, "latest", fake_latest)
     monkeypatch.setattr(work_cmd, "show", fake_show)
+    monkeypatch.setattr(work_cmd, "recap", fake_recap)
 
     assert cli.main(["work", "list", "--target", str(tmp_path), "--limit", "2"]) == 0
     assert cli.main(["work", "latest", "--target", str(tmp_path)]) == 0
     assert cli.main(["work", "show", "abc123", "--target", str(tmp_path)]) == 0
+    assert cli.main(["work", "recap", "--target", str(tmp_path), "--since", "2026-05-26", "--limit", "3"]) == 0
     assert seen == [
         ("list", {"target": tmp_path, "limit": 2}),
         ("latest", {"target": tmp_path}),
         ("show", {"target": tmp_path, "session": "abc123"}),
+        ("recap", {"target": tmp_path, "limit": 3, "since": "2026-05-26"}),
     ]