From 01ba1865cef44be5f328f3700957adac65a61e0e Mon Sep 17 00:00:00 2001 From: tegwick Date: Thu, 6 Aug 2026 13:05:46 +0200 Subject: [PATCH] feat(activity): repo-scoped automation review CLI (WP-0028) Ship the activity console script for consumer-repo morning review: list, status, runs, deliverables, inbox, checkpoint, and ack. Offline-first with git + local defs; enriches from ops API and State Hub. Multi-source trust matrix never reports did-not-run when git has the artefact. Adds target_repo filter on GET /ops/automations. --- Makefile | 7 + docs/runbook.md | 26 + pyproject.toml | 3 + src/activity_core/ops_api.py | 27 +- src/activity_core/review_cli/__init__.py | 5 + src/activity_core/review_cli/__main__.py | 3 + src/activity_core/review_cli/builtins.py | 53 ++ src/activity_core/review_cli/checkpoint.py | 107 +++ src/activity_core/review_cli/defs.py | 112 +++ .../review_cli/git_deliverables.py | 104 +++ src/activity_core/review_cli/main.py | 709 ++++++++++++++++++ src/activity_core/review_cli/merge.py | 210 ++++++ src/activity_core/review_cli/repo.py | 83 ++ src/activity_core/review_cli/sources.py | 117 +++ src/activity_core/review_cli/timewin.py | 70 ++ tests/test_review_cli.py | 193 +++++ .../ACTIVITY-WP-0028-activity-review-cli.md | 8 + 17 files changed, 1836 insertions(+), 1 deletion(-) create mode 100644 src/activity_core/review_cli/__init__.py create mode 100644 src/activity_core/review_cli/__main__.py create mode 100644 src/activity_core/review_cli/builtins.py create mode 100644 src/activity_core/review_cli/checkpoint.py create mode 100644 src/activity_core/review_cli/defs.py create mode 100644 src/activity_core/review_cli/git_deliverables.py create mode 100644 src/activity_core/review_cli/main.py create mode 100644 src/activity_core/review_cli/merge.py create mode 100644 src/activity_core/review_cli/repo.py create mode 100644 src/activity_core/review_cli/sources.py create mode 100644 src/activity_core/review_cli/timewin.py create mode 100644 tests/test_review_cli.py diff --git a/Makefile b/Makefile index 93da91b..87b3932 100644 --- a/Makefile +++ b/Makefile @@ -3,6 +3,7 @@ export .PHONY: sync-event-types sync-activity-definitions sync-schedules test migrate sync-all \ automation-status automation-status-json automation-list automation-list-json \ + activity-review \ dev-up dev-down railiance-up railiance-down \ start-worker start-api start-event-router help @@ -52,6 +53,12 @@ automation-list: ## List configured scheduled automations from repo-owned defin automation-list-json: ## List configured scheduled automations as JSON @$(MAKE) --no-print-directory automation-list FORMAT=json +# Consumer-repo review CLI (ACTIVITY-WP-0028). Example: +# make activity-review ARGS='--cwd ~/freedom-intelligence status' +ARGS ?= +activity-review: ## Repo-scoped review CLI (activity status|list|inbox|…) + uv run activity $(ARGS) + # ── Infrastructure ───────────────────────────────────────────────────────────── dev-up: ## Start full dev stack (Temporal + PG + ES + NATS) diff --git a/docs/runbook.md b/docs/runbook.md index 9f1f518..ca39c1b 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -287,6 +287,32 @@ Compact human output looks like: - Daily State Hub WSJF Triage [enabled cron] schedule=activity-schedule-... trigger=20 7 * * * tz=Europe/Berlin source=files temporal=not_checked ``` +## Repo-scoped review CLI (`activity`) + +From a **consumer repo** (e.g. Freedom Intelligence), use the `activity` CLI for +morning review without AI tooling. Canon: `docs/repo-automation-review-cli.md` +(ACTIVITY-WP-0028). + +```bash +# Install once (from activity-core checkout) +uv tool install -e . +# or: uv run activity … + +cd ~/freedom-intelligence +activity status +activity inbox +activity runs --since today # needs ACTIVITY_CORE_URL for live ops API +activity ack briefs/2026/08/2026-08-06.md +``` + +| Env | Purpose | +| --- | ------- | +| `ACTIVITY_CORE_URL` | Ops API (e.g. `https://activity.coulomb.social` or ClusterIP) | +| `STATE_HUB_URL` | Hub progress for completion events | +| `ACTIVITY_REVIEW_STATE_DIR` | Override local checkpoint dir | + +Org-wide tools (`make automation-status`, prod SSH helper) remain for fleet view. + ## Automation status Use the repo-native status command to answer operator questions such as "how did diff --git a/pyproject.toml b/pyproject.toml index 3dc11ea..79c0999 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,9 @@ dependencies = [ "pyyaml>=6.0", ] +[project.scripts] +activity = "activity_core.review_cli.main:main" + [project.optional-dependencies] dev = [ "pytest>=8.0", diff --git a/src/activity_core/ops_api.py b/src/activity_core/ops_api.py index 1e664b7..8eb7826 100644 --- a/src/activity_core/ops_api.py +++ b/src/activity_core/ops_api.py @@ -75,13 +75,38 @@ class MutateBody(BaseModel): @router.get("/automations") async def list_automations( enabled: str = Query(default="all"), + target_repo: str | None = Query( + default=None, + description="Filter automations whose name/id/labels mention this repo slug (ACTIVITY-WP-0028)", + ), ) -> dict[str, Any]: - return await ops_inventory( + report = await ops_inventory( db_url=_db_url(), temporal_host=os.environ.get("TEMPORAL_HOST"), temporal_namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"), enabled=enabled, ) + if target_repo and isinstance(report, dict): + needle = target_repo.strip().lower() + autos = report.get("automations") or [] + if isinstance(autos, list) and needle: + filtered = [] + for a in autos: + if not isinstance(a, dict): + continue + blob = " ".join( + str(a.get(k) or "") + for k in ("name", "id", "activity_id", "slug", "labels") + ).lower() + if needle in blob: + filtered.append(a) + report = dict(report) + report["automations"] = filtered + report["filters"] = { + **(report.get("filters") or {}), + "target_repo": target_repo, + } + return report @router.get("/automations/status") diff --git a/src/activity_core/review_cli/__init__.py b/src/activity_core/review_cli/__init__.py new file mode 100644 index 0000000..41359e8 --- /dev/null +++ b/src/activity_core/review_cli/__init__.py @@ -0,0 +1,5 @@ +"""Repo-scoped automation review CLI (`activity`) — ACTIVITY-WP-0028.""" + +from activity_core.review_cli.main import main + +__all__ = ["main"] diff --git a/src/activity_core/review_cli/__main__.py b/src/activity_core/review_cli/__main__.py new file mode 100644 index 0000000..a230338 --- /dev/null +++ b/src/activity_core/review_cli/__main__.py @@ -0,0 +1,3 @@ +from activity_core.review_cli.main import main + +raise SystemExit(main()) diff --git a/src/activity_core/review_cli/builtins.py b/src/activity_core/review_cli/builtins.py new file mode 100644 index 0000000..d9497cb --- /dev/null +++ b/src/activity_core/review_cli/builtins.py @@ -0,0 +1,53 @@ +"""Built-in review metadata when definition frontmatter lacks a review: block.""" + +from __future__ import annotations + +from typing import Any + +# Keys: definition id and/or target_repo slug +_BUILTIN: dict[str, dict[str, Any]] = { + "fi-daily-research-brief": { + "deliverable_globs": ["briefs/**/*.md"], + "completion_event_type": "fi_daily_brief", + "path_in_event": "detail.path", + }, + "freedom-intelligence": { + "deliverable_globs": ["briefs/**/*.md"], + "completion_event_type": "fi_daily_brief", + "path_in_event": "detail.path", + }, + "binky-daily-rhythm": { + "deliverable_globs": ["briefs/**/*daily*", "briefs/**/*.md"], + "completion_event_type": "binky_daily_brief", + "path_in_event": "detail.path", + }, + "binky-control": { + "deliverable_globs": ["briefs/**/*daily*", "briefs/**/*.md"], + "completion_event_type": "binky_daily_brief", + "path_in_event": "detail.path", + }, +} + + +def review_meta_for( + *, + definition_id: str | None = None, + target_repo: str | None = None, + frontmatter_review: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Merge optional frontmatter review: block with builtins.""" + base: dict[str, Any] = { + "deliverable_globs": [], + "completion_event_type": None, + "path_in_event": "detail.path", + } + for key in (definition_id, target_repo): + if key and key in _BUILTIN: + for k, v in _BUILTIN[key].items(): + if v is not None: + base[k] = v + if isinstance(frontmatter_review, dict): + for k, v in frontmatter_review.items(): + if v is not None: + base[k] = v + return base diff --git a/src/activity_core/review_cli/checkpoint.py b/src/activity_core/review_cli/checkpoint.py new file mode 100644 index 0000000..91c4321 --- /dev/null +++ b/src/activity_core/review_cli/checkpoint.py @@ -0,0 +1,107 @@ +"""Local operator checkpoint / ack state (XDG).""" + +from __future__ import annotations + +import json +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +def default_state_dir() -> Path: + override = os.environ.get("ACTIVITY_REVIEW_STATE_DIR", "").strip() + if override: + return Path(override).expanduser() + xdg = os.environ.get("XDG_STATE_HOME", "").strip() + if xdg: + return Path(xdg).expanduser() / "activity" + return Path.home() / ".local" / "state" / "activity" + + +def checkpoint_path(repo_slug: str, state_dir: Path | None = None) -> Path: + base = state_dir or default_state_dir() + return base / repo_slug / "checkpoint.json" + + +def empty_checkpoint(repo_slug: str) -> dict[str, Any]: + return { + "repo": repo_slug, + "schema": 1, + "reviewed_at": None, + "reviewed_paths": [], + "reviewed_ops_run_ids": [], + "notes": "", + } + + +def load_checkpoint(repo_slug: str, state_dir: Path | None = None) -> dict[str, Any]: + path = checkpoint_path(repo_slug, state_dir) + if not path.is_file(): + return empty_checkpoint(repo_slug) + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return empty_checkpoint(repo_slug) + if not isinstance(data, dict): + return empty_checkpoint(repo_slug) + base = empty_checkpoint(repo_slug) + base.update({k: data[k] for k in base if k in data}) + base["reviewed_paths"] = list(base.get("reviewed_paths") or []) + base["reviewed_ops_run_ids"] = list(base.get("reviewed_ops_run_ids") or []) + return base + + +def save_checkpoint( + repo_slug: str, + data: dict[str, Any], + state_dir: Path | None = None, +) -> Path: + path = checkpoint_path(repo_slug, state_dir) + path.parent.mkdir(parents=True, exist_ok=True) + payload = empty_checkpoint(repo_slug) + payload.update(data) + payload["repo"] = repo_slug + payload["schema"] = 1 + path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") + return path + + +def ack( + repo_slug: str, + *, + paths: list[str] | None = None, + ops_run_ids: list[str] | None = None, + state_dir: Path | None = None, + all_paths: list[str] | None = None, +) -> dict[str, Any]: + """Mark paths and/or ops_run ids reviewed; optionally ack all_paths.""" + cp = load_checkpoint(repo_slug, state_dir) + now = datetime.now(timezone.utc).isoformat() + path_set = set(cp.get("reviewed_paths") or []) + id_set = set(cp.get("reviewed_ops_run_ids") or []) + if all_paths is not None: + path_set.update(all_paths) + if paths: + path_set.update(paths) + if ops_run_ids: + id_set.update(ops_run_ids) + cp["reviewed_paths"] = sorted(path_set) + cp["reviewed_ops_run_ids"] = sorted(id_set) + cp["reviewed_at"] = now + save_checkpoint(repo_slug, cp, state_dir) + return cp + + +def clear_checkpoint(repo_slug: str, state_dir: Path | None = None) -> Path: + path = checkpoint_path(repo_slug, state_dir) + if path.is_file(): + path.unlink() + return path + + +def set_checkpoint_now(repo_slug: str, state_dir: Path | None = None) -> dict[str, Any]: + cp = load_checkpoint(repo_slug, state_dir) + cp["reviewed_at"] = datetime.now(timezone.utc).isoformat() + save_checkpoint(repo_slug, cp, state_dir) + return cp diff --git a/src/activity_core/review_cli/defs.py b/src/activity_core/review_cli/defs.py new file mode 100644 index 0000000..7f2cb42 --- /dev/null +++ b/src/activity_core/review_cli/defs.py @@ -0,0 +1,112 @@ +"""Load and filter local activity definitions for a consumer repo.""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any + +import yaml + +from activity_core.definition_parser import ParseError, parse_file +from activity_core.review_cli.builtins import review_meta_for + +_FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL) + + +def _rel(path: Path, root: Path) -> str: + try: + return str(path.resolve().relative_to(root.resolve())) + except ValueError: + return str(path) + + +def _extract_target_repos(rules: list[dict[str, Any]]) -> list[str]: + """Explicit target_repo values from rule actions (not free-form labels).""" + found: list[str] = [] + for rule in rules: + action = rule.get("action") or {} + if not isinstance(action, dict): + continue + tr = action.get("target_repo") + if isinstance(tr, str) and tr.strip(): + found.append(tr.strip()) + return found + + +def _raw_frontmatter(path: Path) -> dict[str, Any]: + try: + text = path.read_text(encoding="utf-8") + except OSError: + return {} + m = _FRONTMATTER_RE.match(text) + if not m: + return {} + try: + data = yaml.safe_load(m.group(1)) + except yaml.YAMLError: + return {} + return data if isinstance(data, dict) else {} + + +def load_local_automations( + repo_root: Path, + repo_slug: str, +) -> tuple[list[dict[str, Any]], list[str]]: + """Return (automations, warnings). Always offline-safe.""" + defs_dir = repo_root / "activity-definitions" + warnings: list[str] = [] + items: list[dict[str, Any]] = [] + if not defs_dir.is_dir(): + warnings.append(f"no activity-definitions/ under {repo_root}") + return items, warnings + + for path in sorted(defs_dir.glob("*.md")): + try: + defn = parse_file(path) + except ParseError as exc: + warnings.append(str(exc)) + continue + fm = _raw_frontmatter(path) + targets = _extract_target_repos(defn.rules) + # context params repo= (e.g. fi_brief_status) + for cs in defn.context_sources: + params = cs.get("params") or {} + if isinstance(params, dict) and isinstance(params.get("repo"), str): + targets.append(params["repo"].strip()) + targets = list(dict.fromkeys(t for t in targets if t)) + if targets: + # Must explicitly target this slug + if repo_slug not in targets: + continue + primary_target = repo_slug + else: + # No target_repo: only apply when the definition tree *is* this consumer + # (repo root basename matches slug). Avoids activity-core fleet defs + # matching when --repo freedom-intelligence is set from wrong cwd. + if repo_root.name != repo_slug: + continue + primary_target = repo_slug + + review = review_meta_for( + definition_id=defn.id, + target_repo=primary_target, + frontmatter_review=fm.get("review") if isinstance(fm.get("review"), dict) else None, + ) + trig = defn.trigger_config + items.append( + { + "id": defn.id, + "name": defn.name, + "enabled": defn.enabled, + "status": defn.status, + "target_repo": primary_target, + "targets": targets, + "trigger_type": trig.get("trigger_type"), + "cron_expression": trig.get("cron_expression"), + "timezone": trig.get("timezone"), + "source_file": _rel(path, repo_root), + "review": review, + } + ) + return items, warnings diff --git a/src/activity_core/review_cli/git_deliverables.py b/src/activity_core/review_cli/git_deliverables.py new file mode 100644 index 0000000..64e0ab4 --- /dev/null +++ b/src/activity_core/review_cli/git_deliverables.py @@ -0,0 +1,104 @@ +"""Discover deliverable files via git and filesystem globs.""" + +from __future__ import annotations + +import subprocess +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + + +def _git_log_paths_since( + root: Path, + since: datetime | None, + globs: list[str], +) -> list[dict[str, Any]]: + """Return paths touched since `since` matching globs (via git log).""" + if not globs: + return [] + cmd = ["git", "-C", str(root), "log", "--name-only", "--pretty=format:"] + if since is not None: + cmd.append(f"--since={since.isoformat()}") + cmd.append("--") + # git pathspecs + for g in globs: + cmd.append(g) + try: + out = subprocess.run( + cmd, capture_output=True, text=True, timeout=60, check=False + ) + except (OSError, subprocess.TimeoutExpired): + return [] + if out.returncode != 0: + return [] + paths: list[str] = [] + for line in out.stdout.splitlines(): + line = line.strip() + if line and not line.startswith(" "): + paths.append(line) + # unique preserve order + seen: set[str] = set() + ordered: list[str] = [] + for p in paths: + if p not in seen: + seen.add(p) + ordered.append(p) + return [{"path": p, "source": "git"} for p in ordered] + + +def filesystem_glob_deliverables( + root: Path, + globs: list[str], + *, + since: datetime | None = None, +) -> list[dict[str, Any]]: + """List existing files matching globs; filter by mtime if since set.""" + found: list[dict[str, Any]] = [] + for pattern in globs: + for path in sorted(root.glob(pattern)): + if not path.is_file(): + continue + if path.name.startswith("_"): + continue + rel = str(path.relative_to(root)) + if since is not None: + mtime = datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc) + if mtime < since.astimezone(timezone.utc): + continue + found.append({"path": rel, "source": "fs"}) + # dedupe + seen: set[str] = set() + out: list[dict[str, Any]] = [] + for item in found: + if item["path"] not in seen: + seen.add(item["path"]) + out.append(item) + return out + + +def collect_git_deliverables( + root: Path, + globs: list[str], + since: datetime | None, +) -> tuple[list[dict[str, Any]], list[str]]: + warnings: list[str] = [] + if not (root / ".git").exists(): + warnings.append("no .git — filesystem globs only") + return filesystem_glob_deliverables(root, globs, since=since), warnings + items = _git_log_paths_since(root, since, globs) + # ensure files still exist + existing = [] + for item in items: + p = root / item["path"] + if p.is_file(): + existing.append(item) + else: + item = {**item, "missing_local": True} + existing.append(item) + # also pick up untracked/new files via fs if since is today-ish + fs_items = filesystem_glob_deliverables(root, globs, since=since) + seen = {i["path"] for i in existing} + for item in fs_items: + if item["path"] not in seen: + existing.append(item) + return existing, warnings diff --git a/src/activity_core/review_cli/main.py b/src/activity_core/review_cli/main.py new file mode 100644 index 0000000..f9a2d96 --- /dev/null +++ b/src/activity_core/review_cli/main.py @@ -0,0 +1,709 @@ +"""activity CLI — repo-scoped automation review (ACTIVITY-WP-0028).""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime +from pathlib import Path +from typing import Any + +from activity_core.review_cli import checkpoint as cp_mod +from activity_core.review_cli.defs import load_local_automations +from activity_core.review_cli.git_deliverables import collect_git_deliverables +from activity_core.review_cli.merge import ( + extract_ops_artifacts, + hub_paths_for_repo, + merge_deliverable_rows, + summarize_runs, +) +from activity_core.review_cli.repo import resolve_repo +from activity_core.review_cli.sources import ( + activity_core_url, + fetch_hub_progress, + fetch_ops_automations, + fetch_ops_runs_for_automation, + state_hub_url, +) +from activity_core.review_cli.timewin import parse_since + + +def _print_json(data: Any) -> None: + print(json.dumps(data, indent=2, default=str)) + + +def _collect_globs(automations: list[dict[str, Any]]) -> list[str]: + globs: list[str] = [] + for a in automations: + rev = a.get("review") or {} + for g in rev.get("deliverable_globs") or []: + if isinstance(g, str) and g not in globs: + globs.append(g) + return globs + + +def _completion_event_types(automations: list[dict[str, Any]]) -> list[str]: + types: list[str] = [] + for a in automations: + et = (a.get("review") or {}).get("completion_event_type") + if isinstance(et, str) and et and et not in types: + types.append(et) + return types + + +def cmd_list(args: argparse.Namespace) -> int: + repo = resolve_repo(cwd=Path(args.cwd) if args.cwd else None, explicit=args.repo) + autos, warnings = load_local_automations(repo["root"], repo["slug"]) + sources: dict[str, str] = {"defs": "ok" if autos or not warnings else "empty"} + api_url = activity_core_url() + if api_url: + live, err = fetch_ops_automations(api_url, target_repo=repo["slug"]) + sources["api"] = "ok" if err is None else "degraded" + if err: + warnings.append(f"api: {err}") + # annotate matching live defs + live_by_name = {str(a.get("name")): a for a in live} + for a in autos: + if a["name"] in live_by_name: + a["live"] = { + "id": live_by_name[a["name"]].get("id"), + "enabled": live_by_name[a["name"]].get("enabled"), + } + report = { + "repo": repo["slug"], + "root": str(repo["root"]), + "repo_source": repo["source"], + "automations": autos, + "count": len(autos), + "sources": sources, + "warnings": warnings, + } + if args.format == "json": + _print_json(report) + else: + print(f"Repo: {repo['slug']} (root={repo['root']}, via {repo['source']})") + if not autos: + print("(no local automations matched)") + for a in autos: + en = "enabled" if a.get("enabled") else "disabled" + cron = a.get("cron_expression") or a.get("trigger_type") or "?" + tz = a.get("timezone") or "" + print(f" {a['name']} [{en}] {cron} {tz}".rstrip()) + print(f" id={a['id']} target_repo={a.get('target_repo')}") + for w in warnings: + print(f"warn: {w}", file=sys.stderr) + return 2 if sources.get("api") == "degraded" else 0 + + +def _build_context(args: argparse.Namespace) -> dict[str, Any]: + repo = resolve_repo(cwd=Path(args.cwd) if args.cwd else None, explicit=args.repo) + state_dir = Path(args.state_dir).expanduser() if args.state_dir else None + checkpoint = cp_mod.load_checkpoint(repo["slug"], state_dir) + autos, warnings = load_local_automations(repo["root"], repo["slug"]) + since_raw = getattr(args, "since", None) + try: + since = parse_since(since_raw, checkpoint=checkpoint) + except ValueError as exc: + print(f"error: {exc}", file=sys.stderr) + raise SystemExit(2) from exc + return { + "repo": repo, + "state_dir": state_dir, + "checkpoint": checkpoint, + "automations": autos, + "warnings": warnings, + "since": since, + "since_raw": since_raw or ("checkpoint" if checkpoint.get("reviewed_at") else "today"), + } + + +def cmd_deliverables(args: argparse.Namespace) -> int: + ctx = _build_context(args) + repo = ctx["repo"] + globs = _collect_globs(ctx["automations"]) + if not globs: + globs = ["briefs/**/*.md"] + git_items, gw = collect_git_deliverables(repo["root"], globs, ctx["since"]) + ctx["warnings"].extend(gw) + + hub_by_path: dict[str, dict[str, Any]] = {} + sources = {"defs": "ok", "git": "ok"} + hub = state_hub_url() + if hub: + for et in _completion_event_types(ctx["automations"]) or ["fi_daily_brief"]: + events, err = fetch_hub_progress(hub, event_type=et) + if err: + sources["hub"] = "degraded" + ctx["warnings"].append(f"hub {et}: {err}") + else: + sources["hub"] = "ok" + hub_by_path.update( + hub_paths_for_repo( + events, repo_slug=repo["slug"], since=ctx["since"] + ) + ) + else: + sources["hub"] = "skipped" + + ops_arts: list[dict[str, Any]] = [] + api = activity_core_url() + if api: + sources["api"] = "ok" + for a in ctx["automations"]: + # try live id from name match + live, err = fetch_ops_automations(api, target_repo=repo["slug"]) + if err: + sources["api"] = "degraded" + ctx["warnings"].append(f"api: {err}") + break + def_id = None + for la in live: + if la.get("name") == a["name"] or str(la.get("id")) == a["id"]: + def_id = str(la.get("id")) + break + if not def_id: + continue + runs, rerr = fetch_ops_runs_for_automation( + api, def_id, since=ctx["since"] + ) + if rerr: + sources["api"] = "degraded" + ctx["warnings"].append(f"runs: {rerr}") + else: + ops_arts.extend(extract_ops_artifacts(runs)) + else: + sources["api"] = "skipped" + + rows = merge_deliverable_rows( + root=repo["root"], + git_items=git_items, + hub_by_path=hub_by_path, + ops_artifacts=ops_arts, + checkpoint=ctx["checkpoint"], + ) + report = { + "repo": repo["slug"], + "since": ctx["since"].isoformat() if ctx["since"] else None, + "deliverables": rows, + "count": len(rows), + "open_count": sum(1 for r in rows if r.get("open")), + "sources": sources, + "warnings": ctx["warnings"], + } + if args.format == "json": + _print_json(report) + else: + print( + f"Deliverables for {repo['slug']} since {ctx['since_raw']}" + f" ({ctx['since']})" + ) + for r in rows: + flag = "open" if r.get("open") else ("acked" if r.get("acked") else r["status"]) + print(f" [{flag}/{r['status']}] {r['path']}") + if r.get("url"): + print(f" {r['url']}") + for w in ctx["warnings"]: + print(f"warn: {w}", file=sys.stderr) + degraded = sources.get("api") == "degraded" or sources.get("hub") == "degraded" + return 2 if degraded else 0 + + +def cmd_inbox(args: argparse.Namespace) -> int: + # Collect deliverables in a time window, then filter by ack set (paths/ids). + # Do NOT use reviewed_at as the window after ack — that would hide unacked + # older items. Ack membership is the source of truth for "open". + args_copy = argparse.Namespace(**vars(args)) + if not getattr(args_copy, "since", None): + args_copy.since = "week" + # capture via shared builder + ctx = _build_context(args_copy) + # monkey: call deliverables logic + globs = _collect_globs(ctx["automations"]) or ["briefs/**/*.md"] + git_items, gw = collect_git_deliverables(repo := ctx["repo"]["root"], globs, ctx["since"]) + ctx["warnings"].extend(gw) + hub_by_path: dict[str, dict[str, Any]] = {} + hub = state_hub_url() + sources = {"git": "ok"} + if hub: + for et in _completion_event_types(ctx["automations"]) or ["fi_daily_brief"]: + events, err = fetch_hub_progress(hub, event_type=et) + if err: + sources["hub"] = "degraded" + ctx["warnings"].append(f"hub: {err}") + else: + sources["hub"] = "ok" + hub_by_path.update( + hub_paths_for_repo( + events, repo_slug=ctx["repo"]["slug"], since=ctx["since"] + ) + ) + rows = merge_deliverable_rows( + root=ctx["repo"]["root"], + git_items=git_items, + hub_by_path=hub_by_path, + ops_artifacts=[], + checkpoint=ctx["checkpoint"], + ) + open_rows = [r for r in rows if r.get("open")] + report = { + "repo": ctx["repo"]["slug"], + "inbox": open_rows, + "count": len(open_rows), + "checkpoint": ctx["checkpoint"].get("reviewed_at"), + "sources": sources, + "warnings": ctx["warnings"], + } + if args.format == "json": + _print_json(report) + else: + print(f"Inbox for {ctx['repo']['slug']} ({len(open_rows)} open)") + if not open_rows: + print(" (empty)") + for r in open_rows: + print(f" [ ] {r['path']} ({r['status']})") + for w in ctx["warnings"]: + print(f"warn: {w}", file=sys.stderr) + if getattr(args, "strict_review", False) and open_rows: + return 3 + if sources.get("hub") == "degraded": + return 2 + return 0 + + +def cmd_runs(args: argparse.Namespace) -> int: + ctx = _build_context(args) + repo = ctx["repo"] + sources: dict[str, str] = {"defs": "ok"} + all_runs: list[dict[str, Any]] = [] + api = activity_core_url() + if not api: + sources["api"] = "skipped" + if args.format == "json": + _print_json( + { + "repo": repo["slug"], + "runs": [], + "sources": sources, + "warnings": ["ACTIVITY_CORE_URL not set — runs require API"], + } + ) + else: + print( + "runs: ACTIVITY_CORE_URL not set; set it to ops API " + "(e.g. https://activity.coulomb.social or ClusterIP)", + file=sys.stderr, + ) + return 2 + + live, err = fetch_ops_automations(api, target_repo=repo["slug"]) + if err: + sources["api"] = "degraded" + ctx["warnings"].append(err) + if args.format == "json": + _print_json( + { + "repo": repo["slug"], + "runs": [], + "sources": sources, + "warnings": ctx["warnings"], + } + ) + else: + print(f"api error: {err}", file=sys.stderr) + return 2 + sources["api"] = "ok" + + # map local autos to live ids + for a in ctx["automations"]: + def_id = None + for la in live: + if la.get("name") == a["name"] or str(la.get("id")) == a["id"]: + def_id = str(la.get("id")) + break + if not def_id and live: + # try any live auto for this repo + for la in live: + if repo["slug"].lower() in str(la.get("name") or "").lower(): + def_id = str(la.get("id")) + break + if not def_id: + continue + runs, rerr = fetch_ops_runs_for_automation(api, def_id, since=ctx["since"]) + if rerr: + sources["api"] = "degraded" + ctx["warnings"].append(rerr) + continue + for run in runs: + run = dict(run) + run["automation_name"] = a["name"] + run["definition_id"] = def_id + all_runs.append(run) + + # hub annotations + hub = state_hub_url() + hub_by_path: dict[str, dict[str, Any]] = {} + if hub: + for et in _completion_event_types(ctx["automations"]) or ["fi_daily_brief"]: + events, herr = fetch_hub_progress(hub, event_type=et) + if herr: + sources["hub"] = "degraded" + ctx["warnings"].append(herr) + else: + sources["hub"] = "ok" + hub_by_path.update( + hub_paths_for_repo( + events, repo_slug=repo["slug"], since=ctx["since"] + ) + ) + else: + sources["hub"] = "skipped" + + # attach trust per run from first artefact path + for run in all_runs: + arts = extract_ops_artifacts([run]) + path = arts[0]["path"] if arts else None + git_ok = bool(path and (repo["root"] / path).is_file()) if path else None + hub_ok = bool(path and path in hub_by_path) if path else None + ops_failed = any( + (o.get("state") == "failed") + for o in (run.get("ops_runs") or []) + if isinstance(o, dict) + ) + from activity_core.review_cli.merge import classify_row + + run["trust"] = classify_row( + git_present=git_ok, + hub_present=hub_ok if sources.get("hub") == "ok" else None, + ops_failed=ops_failed, + ops_ok=not ops_failed and bool(run.get("ops_runs")), + ) + run["artefact_path"] = path + + summary = summarize_runs(all_runs) + report = { + "repo": repo["slug"], + "since": ctx["since"].isoformat() if ctx["since"] else None, + "runs": all_runs, + "summary": summary, + "sources": sources, + "warnings": ctx["warnings"], + } + if args.format == "json": + _print_json(report) + else: + print(f"Runs for {repo['slug']} since {ctx['since_raw']}") + if not all_runs: + print(" (none)") + for run in all_runs: + fired = run.get("fired_at") or "" + trust = run.get("trust") or "?" + path = run.get("artefact_path") or "" + name = run.get("automation_name") or "" + print(f" {fired} {trust:8} {name} {path}") + for op in run.get("ops_runs") or []: + if isinstance(op, dict): + print( + f" ops_run={op.get('id')} state={op.get('state')} " + f"title={op.get('title')}" + ) + print( + f"summary: ok={summary['ok']} failed={summary['failed']} " + f"total_runs={summary['total']}" + ) + for w in ctx["warnings"]: + print(f"warn: {w}", file=sys.stderr) + if summary["failed"]: + return 1 + if sources.get("api") == "degraded" or sources.get("hub") == "degraded": + return 2 + return 0 + + +def cmd_status(args: argparse.Namespace) -> int: + ctx = _build_context(args) + repo = ctx["repo"] + # deliverables/inbox counts offline + globs = _collect_globs(ctx["automations"]) or ["briefs/**/*.md"] + git_items, gw = collect_git_deliverables(repo["root"], globs, ctx["since"]) + ctx["warnings"].extend(gw) + hub_by_path: dict[str, dict[str, Any]] = {} + sources: dict[str, str] = { + "defs": "ok" if ctx["automations"] else "empty", + "git": "ok", + } + hub = state_hub_url() + if hub: + for et in _completion_event_types(ctx["automations"]) or ["fi_daily_brief"]: + events, err = fetch_hub_progress(hub, event_type=et, limit=30) + if err: + sources["hub"] = "degraded" + ctx["warnings"].append(err) + else: + sources["hub"] = "ok" + hub_by_path.update( + hub_paths_for_repo( + events, repo_slug=repo["slug"], since=ctx["since"] + ) + ) + else: + sources["hub"] = "skipped" + + rows = merge_deliverable_rows( + root=repo["root"], + git_items=git_items, + hub_by_path=hub_by_path, + ops_artifacts=[], + checkpoint=ctx["checkpoint"], + ) + open_n = sum(1 for r in rows if r.get("open")) + enabled_n = sum(1 for a in ctx["automations"] if a.get("enabled")) + + run_summary = {"ok": 0, "failed": 0, "total": 0} + api = activity_core_url() + if api: + # lightweight runs summary + args_runs = argparse.Namespace(**vars(args)) + # don't recurse fully — fetch briefly + live, err = fetch_ops_automations(api, target_repo=repo["slug"]) + if err: + sources["api"] = "degraded" + ctx["warnings"].append(err) + else: + sources["api"] = "ok" + all_runs: list[dict[str, Any]] = [] + for a in ctx["automations"]: + def_id = None + for la in live: + if la.get("name") == a["name"] or str(la.get("id")) == a["id"]: + def_id = str(la.get("id")) + break + if not def_id: + continue + runs, rerr = fetch_ops_runs_for_automation( + api, def_id, since=ctx["since"], limit=20 + ) + if rerr: + sources["api"] = "degraded" + break + all_runs.extend(runs) + run_summary = summarize_runs(all_runs) + else: + sources["api"] = "skipped" + + report = { + "repo": repo["slug"], + "root": str(repo["root"]), + "automations_enabled": enabled_n, + "automations_total": len(ctx["automations"]), + "automations": ctx["automations"], + "since": ctx["since"].isoformat() if ctx["since"] else None, + "checkpoint_at": ctx["checkpoint"].get("reviewed_at"), + "deliverables": len(rows), + "inbox": open_n, + "runs": run_summary, + "sources": sources, + "warnings": ctx["warnings"], + } + if args.format == "json": + _print_json(report) + else: + print(f"Repo: {repo['slug']}") + print( + f"Automations: {enabled_n} enabled / {len(ctx['automations'])} local" + ) + for a in ctx["automations"]: + en = "on" if a.get("enabled") else "off" + print( + f" - {a['name']} [{en}] " + f"{a.get('cron_expression') or a.get('trigger_type')} " + f"{a.get('timezone') or ''}".rstrip() + ) + print(f"Since: {ctx['since_raw']} ({ctx['since']})") + print(f"Checkpoint: {ctx['checkpoint'].get('reviewed_at') or '(none)'}") + print( + f"Runs: ok={run_summary['ok']} failed={run_summary['failed']} " + f"total={run_summary['total']}" + ) + print(f"Deliverables: {len(rows)} inbox open: {open_n}") + print( + "Sources: " + + " ".join(f"{k}={v}" for k, v in sorted(sources.items())) + ) + for w in ctx["warnings"]: + print(f"warn: {w}", file=sys.stderr) + if run_summary.get("failed"): + return 1 + if "degraded" in sources.values(): + return 2 + if getattr(args, "strict_review", False) and open_n: + return 3 + return 0 + + +def cmd_checkpoint(args: argparse.Namespace) -> int: + repo = resolve_repo(cwd=Path(args.cwd) if args.cwd else None, explicit=args.repo) + state_dir = Path(args.state_dir).expanduser() if args.state_dir else None + action = args.checkpoint_action + if action == "show": + data = cp_mod.load_checkpoint(repo["slug"], state_dir) + path = cp_mod.checkpoint_path(repo["slug"], state_dir) + if args.format == "json": + _print_json({"path": str(path), "checkpoint": data}) + else: + print(f"path: {path}") + print(json.dumps(data, indent=2)) + return 0 + if action == "clear": + path = cp_mod.clear_checkpoint(repo["slug"], state_dir) + print(f"cleared {path}") + return 0 + if action == "set": + data = cp_mod.set_checkpoint_now(repo["slug"], state_dir) + if args.format == "json": + _print_json(data) + else: + print(f"checkpoint set reviewed_at={data.get('reviewed_at')}") + return 0 + print("usage: activity checkpoint show|set|clear", file=sys.stderr) + return 2 + + +def cmd_ack(args: argparse.Namespace) -> int: + repo = resolve_repo(cwd=Path(args.cwd) if args.cwd else None, explicit=args.repo) + state_dir = Path(args.state_dir).expanduser() if args.state_dir else None + paths: list[str] = [] + ops_ids: list[str] = [] + all_paths: list[str] | None = None + if args.all: + # collect current open deliverable paths + ns = argparse.Namespace( + repo=args.repo, + cwd=args.cwd, + state_dir=args.state_dir, + since="checkpoint", + format="json", + strict_review=False, + ) + ctx = _build_context(ns) + globs = _collect_globs(ctx["automations"]) or ["briefs/**/*.md"] + git_items, _ = collect_git_deliverables( + ctx["repo"]["root"], globs, ctx["since"] + ) + all_paths = [i["path"] for i in git_items] + for target in args.targets or []: + # uuid-ish → ops_run id + if len(target) >= 32 and "-" in target: + ops_ids.append(target) + else: + paths.append(target) + data = cp_mod.ack( + repo["slug"], + paths=paths or None, + ops_run_ids=ops_ids or None, + all_paths=all_paths, + state_dir=state_dir, + ) + if args.format == "json": + _print_json(data) + else: + print(f"acked repo={repo['slug']} reviewed_at={data.get('reviewed_at')}") + if paths: + print(" paths:", ", ".join(paths)) + if ops_ids: + print(" ops_runs:", ", ".join(ops_ids)) + if all_paths is not None: + print(f" all: {len(all_paths)} paths") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="activity", + description=( + "Repo-scoped automation review CLI (ACTIVITY-WP-0028). " + "Run from a consumer repo (e.g. freedom-intelligence)." + ), + ) + p.add_argument("--repo", help="Repo slug (default: auto-detect from cwd)") + p.add_argument("--cwd", help="Working directory (default: process cwd)") + p.add_argument( + "--state-dir", + help="Override checkpoint state directory", + ) + p.add_argument( + "--format", + choices=("text", "json"), + default="text", + help="Output format", + ) + p.add_argument( + "--strict-review", + action="store_true", + help="Exit 3 when inbox has open items", + ) + sub = p.add_subparsers(dest="command", required=True) + + for name, help_text in ( + ("list", "List automations for this repo"), + ("status", "One-screen dashboard"), + ("runs", "What ran / failed in a window"), + ("deliverables", "What was produced"), + ("inbox", "Open deliverables for review"), + ): + sp = sub.add_parser(name, help=help_text) + if name in {"runs", "deliverables", "status", "inbox"}: + sp.add_argument( + "--since", + default=None, + help="today|yesterday|week|sunday|checkpoint|ISO (default: checkpoint or today)", + ) + + sp_cp = sub.add_parser("checkpoint", help="Show/set/clear local review cursor") + sp_cp.add_argument( + "checkpoint_action", + choices=("show", "set", "clear"), + help="Action", + ) + + sp_ack = sub.add_parser("ack", help="Mark path(s) or ops_run id(s) reviewed") + sp_ack.add_argument( + "targets", + nargs="*", + help="Relative paths or ops_run UUIDs", + ) + sp_ack.add_argument( + "--all", + action="store_true", + help="Ack all deliverable paths since checkpoint", + ) + return p + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + # propagate format/strict onto subcommands that share globals + handlers = { + "list": cmd_list, + "status": cmd_status, + "runs": cmd_runs, + "deliverables": cmd_deliverables, + "inbox": cmd_inbox, + "checkpoint": cmd_checkpoint, + "ack": cmd_ack, + } + handler = handlers.get(args.command) + if handler is None: + parser.print_help() + return 2 + try: + return handler(args) + except BrokenPipeError: + return 0 + except KeyboardInterrupt: + return 130 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/activity_core/review_cli/merge.py b/src/activity_core/review_cli/merge.py new file mode 100644 index 0000000..f4adb13 --- /dev/null +++ b/src/activity_core/review_cli/merge.py @@ -0,0 +1,210 @@ +"""Merge multi-source evidence with trust matrix.""" + +from __future__ import annotations + +from datetime import datetime +from pathlib import Path +from typing import Any + + +def path_exists(root: Path, rel: str) -> bool: + return (root / rel).is_file() + + +def classify_row( + *, + git_present: bool | None, + hub_present: bool | None, + ops_failed: bool = False, + ops_ok: bool | None = None, +) -> str: + """Return ok|partial|lag|missing|failed.""" + if ops_failed: + return "failed" + if git_present is True and hub_present is True: + return "ok" + if git_present is True and hub_present is False: + return "partial" + if git_present is False and hub_present is True: + return "lag" + if git_present is True and hub_present is None: + return "ok" if ops_ok is not False else "partial" + if git_present is None and hub_present is True: + return "lag" + if ops_ok is True: + return "partial" + if ops_ok is False: + return "failed" + return "missing" + + +def hub_paths_for_repo( + events: list[dict[str, Any]], + *, + repo_slug: str, + since: datetime | None, +) -> dict[str, dict[str, Any]]: + """Map path -> event for completion events matching repo.""" + out: dict[str, dict[str, Any]] = {} + for ev in events: + detail = ev.get("detail") or {} + if not isinstance(detail, dict): + continue + if detail.get("repo") and str(detail.get("repo")) != repo_slug: + continue + created = ev.get("created_at") + if since and created: + try: + dt = datetime.fromisoformat(str(created).replace("Z", "+00:00")) + if dt < since: + continue + except ValueError: + pass + path = detail.get("path") + if isinstance(path, str) and path.strip(): + out[path.strip()] = ev + return out + + +def merge_deliverable_rows( + *, + root: Path, + git_items: list[dict[str, Any]], + hub_by_path: dict[str, dict[str, Any]], + ops_artifacts: list[dict[str, Any]], + checkpoint: dict[str, Any], +) -> list[dict[str, Any]]: + """Build deliverable rows with trust status and inbox flag.""" + reviewed_paths = set(checkpoint.get("reviewed_paths") or []) + reviewed_ops = set(checkpoint.get("reviewed_ops_run_ids") or []) + by_path: dict[str, dict[str, Any]] = {} + + for item in git_items: + path = item["path"] + git_ok = not item.get("missing_local") and path_exists(root, path) + by_path[path] = { + "path": path, + "git": git_ok, + "hub": path in hub_by_path, + "ops_run_ids": [], + "sources": [item.get("source") or "git"], + } + + for path, ev in hub_by_path.items(): + row = by_path.setdefault( + path, + { + "path": path, + "git": path_exists(root, path), + "hub": True, + "ops_run_ids": [], + "sources": ["hub"], + }, + ) + row["hub"] = True + if "hub" not in row["sources"]: + row["sources"].append("hub") + row["hub_event"] = { + "id": ev.get("id"), + "summary": ev.get("summary"), + "created_at": ev.get("created_at"), + } + + for art in ops_artifacts: + path = art.get("path") + if not isinstance(path, str): + continue + row = by_path.setdefault( + path, + { + "path": path, + "git": path_exists(root, path), + "hub": path in hub_by_path, + "ops_run_ids": [], + "sources": ["ops"], + }, + ) + if "ops" not in row["sources"]: + row["sources"].append("ops") + oid = art.get("ops_run_id") + if oid and oid not in row["ops_run_ids"]: + row["ops_run_ids"].append(oid) + if art.get("url"): + row["url"] = art["url"] + + rows: list[dict[str, Any]] = [] + for path, row in sorted(by_path.items(), key=lambda x: x[0], reverse=True): + status = classify_row( + git_present=row.get("git"), + hub_present=row.get("hub") if "hub" in row.get("sources", []) or row.get("hub") else ( + True if row.get("hub") else False if hub_by_path is not None else None + ), + ) + # refine hub_present: if we never queried hub, leave None + if not hub_by_path and "hub" not in row.get("sources", []): + status = classify_row( + git_present=row.get("git"), + hub_present=None, + ops_ok=bool(row.get("ops_run_ids")), + ) + acked = path in reviewed_paths or any( + oid in reviewed_ops for oid in row.get("ops_run_ids") or [] + ) + rows.append( + { + **row, + "status": status, + "acked": acked, + "open": not acked and status in {"ok", "partial", "lag"}, + } + ) + return rows + + +def extract_ops_artifacts(runs: list[dict[str, Any]]) -> list[dict[str, Any]]: + out: list[dict[str, Any]] = [] + for run in runs: + for op in run.get("ops_runs") or []: + if not isinstance(op, dict): + continue + oid = op.get("id") + result = op.get("result") or {} + path = result.get("path") if isinstance(result, dict) else None + for art in op.get("artifacts") or []: + if not isinstance(art, dict): + continue + out.append( + { + "path": path or art.get("label"), + "url": art.get("url"), + "ops_run_id": oid, + "state": op.get("state"), + "fired_at": run.get("fired_at"), + } + ) + if path and not (op.get("artifacts") or []): + out.append( + { + "path": path, + "ops_run_id": oid, + "state": op.get("state"), + "fired_at": run.get("fired_at"), + "ok": result.get("ok") if isinstance(result, dict) else None, + } + ) + return out + + +def summarize_runs(runs: list[dict[str, Any]]) -> dict[str, int]: + ok = failed = 0 + for run in runs: + ops = run.get("ops_runs") or [] + if not ops: + # activity run with tasks but no ops: neutral + continue + states = [o.get("state") for o in ops if isinstance(o, dict)] + if any(s == "failed" for s in states): + failed += 1 + elif any(s == "succeeded" for s in states): + ok += 1 + return {"ok": ok, "failed": failed, "total": len(runs)} diff --git a/src/activity_core/review_cli/repo.py b/src/activity_core/review_cli/repo.py new file mode 100644 index 0000000..5298943 --- /dev/null +++ b/src/activity_core/review_cli/repo.py @@ -0,0 +1,83 @@ +"""Resolve consumer repo slug and root from cwd.""" + +from __future__ import annotations + +import re +import subprocess +from pathlib import Path +from typing import Any + +import yaml + + +def find_repo_root(start: Path | None = None) -> Path: + """Walk up for .git or activity-definitions/.""" + cur = (start or Path.cwd()).resolve() + for p in [cur, *cur.parents]: + if (p / ".git").exists() or (p / "activity-definitions").is_dir(): + return p + return cur + + +def _slug_from_remote_url(url: str) -> str | None: + url = url.strip() + if not url: + return None + # git@host:org/repo.git or https://host/org/repo.git + m = re.search(r"[:/]([^/]+/)?([^/]+?)(?:\.git)?$", url) + if not m: + return None + return m.group(2) + + +def _slug_from_classification(root: Path) -> str | None: + path = root / ".repo-classification.yaml" + if not path.is_file(): + return None + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return None + if not isinstance(data, dict): + return None + for key in ("repo", "slug", "name", "repo_slug"): + val = data.get(key) + if isinstance(val, str) and val.strip(): + return val.strip() + return None + + +def git_remote_url(root: Path) -> str | None: + try: + out = subprocess.run( + ["git", "-C", str(root), "remote", "get-url", "origin"], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return None + if out.returncode != 0: + return None + return out.stdout.strip() or None + + +def resolve_repo( + *, + cwd: Path | None = None, + explicit: str | None = None, +) -> dict[str, Any]: + """Return {slug, root, source}.""" + root = find_repo_root(cwd) + if explicit and explicit.strip(): + return {"slug": explicit.strip(), "root": root, "source": "flag"} + class_slug = _slug_from_classification(root) + if class_slug: + return {"slug": class_slug, "root": root, "source": "classification"} + remote = git_remote_url(root) + if remote: + slug = _slug_from_remote_url(remote) + if slug: + return {"slug": slug, "root": root, "source": "git-remote"} + return {"slug": root.name, "root": root, "source": "dirname"} diff --git a/src/activity_core/review_cli/sources.py b/src/activity_core/review_cli/sources.py new file mode 100644 index 0000000..703788f --- /dev/null +++ b/src/activity_core/review_cli/sources.py @@ -0,0 +1,117 @@ +"""Optional live sources: ops API and State Hub progress.""" + +from __future__ import annotations + +import os +from datetime import datetime +from typing import Any + +import httpx + + +def activity_core_url() -> str | None: + raw = (os.environ.get("ACTIVITY_CORE_URL") or "").strip().rstrip("/") + return raw or None + + +def state_hub_url() -> str | None: + raw = ( + os.environ.get("STATE_HUB_URL") + or os.environ.get("STATEHUB_URL") + or "" + ).strip().rstrip("/") + return raw or None + + +def fetch_ops_automations( + base_url: str, + *, + target_repo: str | None = None, + timeout: float = 10.0, +) -> tuple[list[dict[str, Any]], str | None]: + """Return automations list and error message if any.""" + params: dict[str, str] = {} + if target_repo: + params["target_repo"] = target_repo + try: + with httpx.Client(timeout=timeout) as client: + r = client.get(f"{base_url}/ops/automations", params=params or None) + r.raise_for_status() + data = r.json() + except (httpx.HTTPError, ValueError) as exc: + return [], str(exc) + items = data.get("automations") or data.get("items") or [] + if not isinstance(items, list): + return [], "unexpected automations payload" + if target_repo: + # Client-side filter as well (older servers ignore the query param) + filtered = [] + needle = target_repo.lower() + for a in items: + if not isinstance(a, dict): + continue + blob = " ".join( + str(a.get(k) or "") + for k in ("name", "id", "target_repo", "labels", "slug") + ).lower() + if needle in blob: + filtered.append(a) + continue + labels = a.get("labels") or [] + if isinstance(labels, list) and target_repo in labels: + filtered.append(a) + items = filtered + return [a for a in items if isinstance(a, dict)], None + + +def fetch_ops_runs_for_automation( + base_url: str, + definition_id: str, + *, + since: datetime | None = None, + limit: int = 50, + timeout: float = 15.0, +) -> tuple[list[dict[str, Any]], str | None]: + params: dict[str, Any] = {"limit": limit} + if since is not None: + params["since"] = since.astimezone().isoformat() + try: + with httpx.Client(timeout=timeout) as client: + r = client.get( + f"{base_url}/ops/automations/{definition_id}/runs", + params=params, + ) + r.raise_for_status() + data = r.json() + except (httpx.HTTPError, ValueError) as exc: + return [], str(exc) + runs = data.get("runs") or [] + if not isinstance(runs, list): + return [], "unexpected runs payload" + return [r for r in runs if isinstance(r, dict)], None + + +def fetch_hub_progress( + base_url: str, + *, + event_type: str, + limit: int = 50, + timeout: float = 10.0, +) -> tuple[list[dict[str, Any]], str | None]: + try: + with httpx.Client(timeout=timeout) as client: + r = client.get( + f"{base_url}/progress/", + params={"event_type": event_type, "limit": limit}, + ) + r.raise_for_status() + data = r.json() + except (httpx.HTTPError, ValueError) as exc: + return [], str(exc) + if isinstance(data, list): + items = data + elif isinstance(data, dict): + items = data.get("items") or data.get("results") or [] + else: + return [], "unexpected progress payload" + return [i for i in items if isinstance(i, dict)], None diff --git a/src/activity_core/review_cli/timewin.py b/src/activity_core/review_cli/timewin.py new file mode 100644 index 0000000..84bc2e9 --- /dev/null +++ b/src/activity_core/review_cli/timewin.py @@ -0,0 +1,70 @@ +"""Parse --since shortcuts for the review CLI.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any +from zoneinfo import ZoneInfo + +DEFAULT_TZ = "Europe/Berlin" + + +def parse_since( + value: str | None, + *, + checkpoint: dict[str, Any] | None = None, + tz_name: str = DEFAULT_TZ, + now: datetime | None = None, +) -> datetime | None: + """Return inclusive lower bound as aware datetime, or None for 'all'.""" + if value is None or value.strip() == "": + value = "checkpoint" if (checkpoint or {}).get("reviewed_at") else "week" + raw = value.strip().lower() + try: + tz = ZoneInfo(tz_name) + except Exception: + tz = timezone.utc + now = now or datetime.now(tz) + if now.tzinfo is None: + now = now.replace(tzinfo=tz) + else: + now = now.astimezone(tz) + + if raw in {"all", "forever", "0"}: + return None + if raw == "checkpoint": + ra = (checkpoint or {}).get("reviewed_at") + if ra: + try: + dt = datetime.fromisoformat(str(ra).replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=tz) + return dt + except ValueError: + pass + raw = "today" + if raw == "today": + return now.replace(hour=0, minute=0, second=0, microsecond=0) + if raw == "yesterday": + day = (now - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + return day + if raw in {"week", "7d"}: + return now - timedelta(days=7) + if raw == "sunday": + # most recent Sunday 00:00 in local tz + days_back = (now.weekday() + 1) % 7 + sunday = (now - timedelta(days=days_back)).replace( + hour=0, minute=0, second=0, microsecond=0 + ) + return sunday + # ISO date or datetime + try: + if len(raw) == 10 and raw[4] == "-" and raw[7] == "-": + dt = datetime.fromisoformat(raw).replace(tzinfo=tz) + return dt + dt = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=tz) + return dt + except ValueError as exc: + raise ValueError(f"invalid --since value: {value!r}") from exc diff --git a/tests/test_review_cli.py b/tests/test_review_cli.py new file mode 100644 index 0000000..e967c98 --- /dev/null +++ b/tests/test_review_cli.py @@ -0,0 +1,193 @@ +"""Tests for activity review CLI (ACTIVITY-WP-0028).""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from activity_core.review_cli.checkpoint import ack, load_checkpoint, save_checkpoint +from activity_core.review_cli.defs import load_local_automations +from activity_core.review_cli.main import main +from activity_core.review_cli.merge import classify_row +from activity_core.review_cli.repo import resolve_repo, _slug_from_remote_url +from activity_core.review_cli.timewin import parse_since + + +def test_slug_from_remote() -> None: + assert ( + _slug_from_remote_url("forgejo-remote:coulomb/freedom-intelligence.git") + == "freedom-intelligence" + ) + assert ( + _slug_from_remote_url("https://forgejo.example/coulomb/binky-control.git") + == "binky-control" + ) + + +def test_resolve_repo_explicit(tmp_path: Path) -> None: + r = resolve_repo(cwd=tmp_path, explicit="freedom-intelligence") + assert r["slug"] == "freedom-intelligence" + assert r["source"] == "flag" + + +def test_parse_since_today() -> None: + now = datetime(2026, 8, 6, 15, 0, tzinfo=timezone.utc) + s = parse_since("today", now=now, tz_name="UTC") + assert s is not None + assert s.day == 6 + assert s.hour == 0 + + +def test_parse_since_checkpoint() -> None: + cp = {"reviewed_at": "2026-08-05T14:00:00+00:00"} + s = parse_since("checkpoint", checkpoint=cp) + assert s is not None + assert s.day == 5 + + +def test_classify_trust_matrix() -> None: + assert classify_row(git_present=True, hub_present=True) == "ok" + assert classify_row(git_present=True, hub_present=False) == "partial" + assert classify_row(git_present=False, hub_present=True) == "lag" + assert classify_row(git_present=False, hub_present=False) == "missing" + assert classify_row(git_present=None, hub_present=None, ops_failed=True) == "failed" + + +def test_checkpoint_ack_roundtrip(tmp_path: Path) -> None: + save_checkpoint("freedom-intelligence", {"reviewed_paths": []}, tmp_path) + ack( + "freedom-intelligence", + paths=["briefs/2026/08/2026-08-06.md"], + state_dir=tmp_path, + ) + cp = load_checkpoint("freedom-intelligence", tmp_path) + assert "briefs/2026/08/2026-08-06.md" in cp["reviewed_paths"] + assert cp["reviewed_at"] + + +def test_load_local_fi_definition(tmp_path: Path) -> None: + defs = tmp_path / "activity-definitions" + defs.mkdir() + (defs / "fi-daily-research-brief.md").write_text( + """--- +id: fi-daily-research-brief +name: Freedom Intelligence Daily Research Brief +enabled: true +trigger: + type: cron + cron_expression: "30 7 * * 1-5" + timezone: Europe/Berlin +--- + +```rule +id: emit +action: + target_repo: freedom-intelligence + labels: ["freedom-intelligence"] +``` +""", + encoding="utf-8", + ) + items, warnings = load_local_automations(tmp_path, "freedom-intelligence") + assert not warnings + assert len(items) == 1 + assert items[0]["id"] == "fi-daily-research-brief" + assert "briefs/**/*.md" in items[0]["review"]["deliverable_globs"] + + +def test_cli_list_json(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + defs = tmp_path / "activity-definitions" + defs.mkdir() + (defs / "fi-daily-research-brief.md").write_text( + """--- +id: fi-daily-research-brief +name: Freedom Intelligence Daily Research Brief +enabled: true +trigger: + type: cron + cron_expression: "30 7 * * 1-5" + timezone: Europe/Berlin +--- + +```rule +id: emit +action: + target_repo: freedom-intelligence +``` +""", + encoding="utf-8", + ) + code = main( + [ + "--cwd", + str(tmp_path), + "--repo", + "freedom-intelligence", + "--format", + "json", + "list", + ] + ) + assert code in (0, 2) + out = capsys.readouterr().out + data = json.loads(out) + assert data["repo"] == "freedom-intelligence" + assert data["count"] == 1 + + +def test_cli_status_offline(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + defs = tmp_path / "activity-definitions" + defs.mkdir() + briefs = tmp_path / "briefs" / "2026" / "08" + briefs.mkdir(parents=True) + (briefs / "2026-08-06.md").write_text("# brief\n", encoding="utf-8") + (defs / "fi-daily-research-brief.md").write_text( + """--- +id: fi-daily-research-brief +name: Freedom Intelligence Daily Research Brief +enabled: true +trigger: + type: cron + cron_expression: "30 7 * * 1-5" + timezone: Europe/Berlin +--- + +```rule +id: emit +action: + target_repo: freedom-intelligence +``` +""", + encoding="utf-8", + ) + # init git for deliverables + import subprocess + + subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True) + subprocess.run(["git", "add", "."], cwd=tmp_path, check=True, capture_output=True) + subprocess.run( + ["git", "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-m", "init"], + cwd=tmp_path, + check=True, + capture_output=True, + ) + code = main( + [ + "--cwd", + str(tmp_path), + "--repo", + "freedom-intelligence", + "--state-dir", + str(tmp_path / "state"), + "status", + "--since", + "week", + ] + ) + assert code in (0, 2) + out = capsys.readouterr().out + assert "freedom-intelligence" in out + assert "Automations:" in out diff --git a/workplans/ACTIVITY-WP-0028-activity-review-cli.md b/workplans/ACTIVITY-WP-0028-activity-review-cli.md index cebf373..6bee0bf 100644 --- a/workplans/ACTIVITY-WP-0028-activity-review-cli.md +++ b/workplans/ACTIVITY-WP-0028-activity-review-cli.md @@ -18,6 +18,7 @@ related: - ACTIVITY-WP-0019 - ACTIVITY-WP-0021 - ACT-ADR-005 +state_hub_workstream_id: "a405ce6a-2f1d-47b0-8ab8-85ae9cb8bfb2" --- # ACTIVITY-WP-0028 — `activity` CLI (repo-scoped automation review) @@ -79,6 +80,7 @@ Canon: `docs/repo-automation-review-cli.md`. id: ACTIVITY-WP-0028-T01 status: todo priority: high +state_hub_task_id: "9a780335-d97a-4c39-9845-7654d3bdf206" ``` 1. Add `activity_core/review_cli/` package with `main()` argparse dispatcher. @@ -101,6 +103,7 @@ consumer cwd in tests or documented local install. id: ACTIVITY-WP-0028-T02 status: todo priority: high +state_hub_task_id: "c5ebb26f-2494-4fac-8540-d9a9d74e350b" ``` Depends on T01. @@ -121,6 +124,7 @@ definition without API. id: ACTIVITY-WP-0028-T03 status: todo priority: high +state_hub_task_id: "3e39e349-c43e-406a-b86d-f7dc704707c1" ``` Depends on T02. @@ -140,6 +144,7 @@ Depends on T02. id: ACTIVITY-WP-0028-T04 status: todo priority: high +state_hub_task_id: "a3203d03-5f6e-4ddf-875f-9d5c31ea9aeb" ``` Depends on T03. @@ -162,6 +167,7 @@ not “did not run”; exit `1` only on real failures. id: ACTIVITY-WP-0028-T05 status: todo priority: medium +state_hub_task_id: "f687c2af-80ef-44f3-94da-5fb663b035c3" ``` Depends on T04. @@ -184,6 +190,7 @@ and Forgejo/path artefacts without SSH. id: ACTIVITY-WP-0028-T06 status: todo priority: medium +state_hub_task_id: "a6cd4704-1284-4859-8cee-3ea49aae4312" ``` Depends on T03 (docs can land with P1). @@ -205,6 +212,7 @@ Depends on T03 (docs can land with P1). id: ACTIVITY-WP-0028-T07 status: todo priority: high +state_hub_task_id: "84b090ce-168c-4c3b-9390-1de8a4d5f488" ``` Depends on T04–T05.