From ad5965be91ed1b086079a2955293438020e85416 Mon Sep 17 00:00:00 2001 From: tegwick Date: Thu, 16 Jul 2026 11:20:52 +0200 Subject: [PATCH] feat: engagement CLI for forward-deployed agency (WP-0009 T08) Add kaizen-agentic engagement subcommands (list, show, validate, checklist, phase, prepare, staff, quote, scrub, export-handoff) backed by engagement.py for file-based pilot lifecycle. Tests cover staff/validate/phase/prepare and the railiance01 pilot smoke path. --- CHANGELOG.md | 3 + docs/CLI_CHEAT_SHEET.md | 33 + src/kaizen_agentic/cli.py | 385 ++++++++ src/kaizen_agentic/engagement.py | 869 ++++++++++++++++++ tests/test_engagement_cli.py | 365 ++++++++ ...agentic-WP-0009-forward-deployed-agency.md | 13 +- 6 files changed, 1661 insertions(+), 7 deletions(-) create mode 100644 src/kaizen_agentic/engagement.py create mode 100644 tests/test_engagement_cli.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b41c71..986385e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `keepaTodofile` agent remains available for other projects ### Added +- **`engagement` CLI (WP-0009 T08)** — forward-deployed engagement lifecycle: + `list`/`show`/`validate`/`checklist`/`phase`/`prepare`/`staff`/`quote`/ + `scrub`/`export-handoff` over file-based pilots (`engagements/pilots/`) - **`metrics record --emit-event`** — publishes `kaizen.metrics.recorded` NATS envelope for activity-core event-driven definitions (optional `nats-py` via `pip install 'kaizen-agentic[events]'`) diff --git a/docs/CLI_CHEAT_SHEET.md b/docs/CLI_CHEAT_SHEET.md index a9c6fd9..b2d88ac 100644 --- a/docs/CLI_CHEAT_SHEET.md +++ b/docs/CLI_CHEAT_SHEET.md @@ -125,6 +125,39 @@ kaizen-agentic schedule prepare optimization --format json activity-core fires the schedule and creates a task per (repo, agent); the task runs `schedule prepare`. kaizen-agentic does not run cron or invoke Claude. +### Forward-deployed engagements (WP-0009 / DEC-FDA-001) +```bash +# List / inspect pilot engagements under engagements/pilots/ +kaizen-agentic engagement list +kaizen-agentic engagement show eng-coulomb-railiance01-ho-001 +kaizen-agentic engagement validate eng-coulomb-railiance01-ho-001 + +# Checklists (ramp-up / ramp-down) +kaizen-agentic engagement checklist eng-coulomb-railiance01-ho-001 +kaizen-agentic engagement checklist eng-… --mark RU-01=done + +# Phase transitions (staffing → ramp_up → operating → ramp_down → closed) +kaizen-agentic engagement phase eng-… --to ramp_up +kaizen-agentic engagement phase eng-… --to operating --force # override graph + +# Session orientation bundle (agent + vault + protocols + access plan) +kaizen-agentic engagement prepare eng-coulomb-railiance01-ho-001 +kaizen-agentic engagement prepare eng-… --format json + +# Staff a new pilot from a Role package +kaizen-agentic engagement staff \ + --id eng-example-001 --role host-operator \ + --client coulomb --target railiance01 + +# Kai quote, scrub before lesson export, handoff pack +kaizen-agentic engagement quote eng-coulomb-railiance01-ho-001 +kaizen-agentic engagement scrub eng-… +kaizen-agentic engagement export-handoff eng-… +``` + +Does not invoke LLMs or touch production hosts. Spec: +`docs/forward-deployed-engagement-architecture.md`. + ### Information ```bash # List templates diff --git a/src/kaizen_agentic/cli.py b/src/kaizen_agentic/cli.py index e191f83..8135480 100644 --- a/src/kaizen_agentic/cli.py +++ b/src/kaizen_agentic/cli.py @@ -24,6 +24,25 @@ from .integrations.helix import HelixCorrelationAdapter, enrich_helix_correlatio from .metrics import MetricsStore, OptimizerStore, performance_summary_markdown from .optimization import OptimizationLoop, MIN_SAMPLES_FOR_RECOMMENDATIONS from .engagement_promote import promote_engagement +from .engagement import ( + EngagementError, + build_prepare_bundle, + export_handoff, + find_repo_root, + list_engagement_dirs, + load_checklist, + load_engagement, + load_quote, + checklist_summary, + render_prepare_markdown, + resolve_engagement_dir, + scrub_engagement, + set_checklist_item_status, + set_phase, + staff_engagement, + validate_engagement, + VALID_PHASES, +) from .schedule import ( ScheduleError, default_schedule_yaml, @@ -1412,6 +1431,372 @@ def protocols_show(agent_name: str, slug: str): click.echo(protocol_path.read_text()) +@cli.group() +def engagement(): + """Forward-deployed engagement lifecycle (KAIZEN-WP-0009 / DEC-FDA-001). + + Manage file-based engagements under engagements/pilots//: phase, + checklists, prepare bundles, staff from Role packages. Offline; no LLM invoke. + """ + pass + + +def _resolve_eng(engagement_ref: str, repo_root: Optional[str]): + root = Path(repo_root).resolve() if repo_root else find_repo_root() + try: + path = resolve_engagement_dir(engagement_ref, root) + return load_engagement(path), root + except EngagementError as exc: + click.echo(f"Error: {exc}", err=True) + sys.exit(1) + + +@engagement.command("list") +@click.option( + "--repo-root", + default=None, + help="Repo root containing engagements/ (default: auto-detect)", +) +@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") +def engagement_list(repo_root: Optional[str], as_json: bool): + """List engagement directories under engagements/pilots/.""" + root = Path(repo_root).resolve() if repo_root else find_repo_root() + dirs = list_engagement_dirs(root) + rows = [] + for d in dirs: + try: + eng = load_engagement(d) + rows.append( + { + "id": eng.engagement_id, + "phase": eng.phase, + "role": eng.role_id, + "path": str(d), + } + ) + except EngagementError as exc: + rows.append( + { + "id": d.name, + "phase": "?", + "role": None, + "path": str(d), + "error": str(exc), + } + ) + if as_json: + click.echo(json.dumps(rows, indent=2)) + return + if not rows: + click.echo(f"No engagements under {root / 'engagements'}") + return + for r in rows: + role = r.get("role") or "-" + click.echo(f" {r['id']}: phase={r['phase']} role={role}") + click.echo(f" {r['path']}") + + +@engagement.command("show") +@click.argument("engagement_ref") +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") +def engagement_show(engagement_ref: str, repo_root: Optional[str], as_json: bool): + """Show engagement summary from ENGAGEMENT.yaml.""" + eng, _root = _resolve_eng(engagement_ref, repo_root) + payload = { + "id": eng.engagement_id, + "phase": eng.phase, + "role": eng.role_id, + "targets": eng.targets, + "path": str(eng.path), + "validation_errors": validate_engagement(eng), + } + if as_json: + click.echo(json.dumps(payload, indent=2)) + return + click.echo(f"Engagement: {payload['id']}") + click.echo(f" Phase: {payload['phase']}") + click.echo(f" Role: {payload['role']}") + click.echo(f" Path: {payload['path']}") + for t in payload["targets"]: + click.echo(f" Target: {t.get('kind')} {t.get('id')}") + errs = payload["validation_errors"] + if errs: + click.echo(" Validation:") + for e in errs: + click.echo(f" - {e}") + else: + click.echo(" Validation: ok") + + +@engagement.command("validate") +@click.argument("engagement_ref") +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +def engagement_validate(engagement_ref: str, repo_root: Optional[str]): + """Validate engagement tree structure and required files.""" + eng, _root = _resolve_eng(engagement_ref, repo_root) + errs = validate_engagement(eng) + if errs: + click.echo(f"❌ {eng.engagement_id}: {len(errs)} issue(s)") + for e in errs: + click.echo(f" - {e}") + sys.exit(1) + click.echo(f"✅ {eng.engagement_id}: valid (phase={eng.phase})") + + +@engagement.command("checklist") +@click.argument("engagement_ref") +@click.option( + "--which", + type=click.Choice(["ramp_up", "ramp_down", "both"]), + default="both", + show_default=True, +) +@click.option( + "--mark", + metavar="ID=STATUS", + help="Update one item, e.g. RU-01=done (implies --which for that prefix)", +) +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") +def engagement_checklist( + engagement_ref: str, + which: str, + mark: Optional[str], + repo_root: Optional[str], + as_json: bool, +): + """Show or update ramp-up / ramp-down checklist status.""" + eng, _root = _resolve_eng(engagement_ref, repo_root) + + if mark: + if "=" not in mark: + click.echo("Error: --mark requires ID=STATUS (e.g. RU-01=done)", err=True) + sys.exit(1) + item_id, status = mark.split("=", 1) + item_id, status = item_id.strip(), status.strip() + which_mark = "ramp_up" if item_id.upper().startswith("RU") else "ramp_down" + path, _items = load_checklist(eng, which_mark) + if path is None: + click.echo(f"Error: no {which_mark} checklist", err=True) + sys.exit(1) + if not set_checklist_item_status(path, item_id, status): + click.echo(f"Error: item {item_id} not found in {path}", err=True) + sys.exit(1) + click.echo(f"Updated {item_id} → {status} in {path}") + + which_list = ["ramp_up", "ramp_down"] if which == "both" else [which] + result: dict = {"engagement_id": eng.engagement_id, "checklists": {}} + for w in which_list: + path, items = load_checklist(eng, w) + summary = checklist_summary(items) + result["checklists"][w] = { + "path": str(path) if path else None, + "summary": summary, + "items": [ + { + "id": i.item_id, + "criterion": i.criterion, + "status": i.status, + "evidence": i.evidence, + "done": i.done, + } + for i in items + ], + } + + if as_json: + click.echo(json.dumps(result, indent=2)) + return + + for w, block in result["checklists"].items(): + summary = block["summary"] + click.echo( + f"{w}: {summary['done']}/{summary['total']}" + + (" ✅ complete" if summary["complete"] else "") + ) + if not block["path"]: + click.echo(" (checklist file missing)") + continue + for it in block["items"]: + flag = "✅" if it["done"] else "☐ " + click.echo(f" {flag} {it['id']}: {it['criterion']} [{it['status']}]") + + +@engagement.command("phase") +@click.argument("engagement_ref") +@click.option( + "--to", + "to_phase", + required=True, + help=f"Target phase ({', '.join(VALID_PHASES)})", +) +@click.option("--force", is_flag=True, help="Skip transition graph checks") +@click.option("--notes", default=None, help="Optional status.notes update") +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +def engagement_phase( + engagement_ref: str, + to_phase: str, + force: bool, + notes: Optional[str], + repo_root: Optional[str], +): + """Transition engagement phase (updates ENGAGEMENT.yaml + agent frontmatter).""" + eng, _root = _resolve_eng(engagement_ref, repo_root) + prev = eng.phase + try: + eng = set_phase(eng, to_phase, force=force, notes=notes) + except EngagementError as exc: + click.echo(f"Error: {exc}", err=True) + sys.exit(1) + click.echo(f"{eng.engagement_id}: {prev} → {eng.phase}") + + +@engagement.command("prepare") +@click.argument("engagement_ref") +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +@click.option( + "--format", + "output_format", + type=click.Choice(["markdown", "json"]), + default="markdown", + show_default=True, +) +def engagement_prepare( + engagement_ref: str, repo_root: Optional[str], output_format: str +): + """Assemble session orientation bundle (agent + vault + protocols + access).""" + eng, root = _resolve_eng(engagement_ref, repo_root) + bundle = build_prepare_bundle(eng, root) + if output_format == "json": + # Drop large protocol bodies in json unless needed — keep presence flags + slim = {k: v for k, v in bundle.items() if k != "protocol_bodies"} + # Include protocol body sizes only + slim["protocol_chars"] = { + k: len(v) for k, v in (bundle.get("protocol_bodies") or {}).items() + } + click.echo(json.dumps(slim, indent=2)) + return + click.echo(render_prepare_markdown(bundle)) + + +@engagement.command("quote") +@click.argument("engagement_ref") +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") +def engagement_quote(engagement_ref: str, repo_root: Optional[str], as_json: bool): + """Show Kai quote snapshot for an engagement.""" + eng, _root = _resolve_eng(engagement_ref, repo_root) + try: + quote = load_quote(eng) + except EngagementError as exc: + click.echo(f"Error: {exc}", err=True) + sys.exit(1) + if quote is None: + click.echo(f"No quote file for {eng.engagement_id}") + sys.exit(1) + if as_json: + click.echo(json.dumps(quote, indent=2)) + return + spec = quote.get("spec") or {} + total = spec.get("total_kai") + click.echo(f"Quote: {eng.engagement_id}") + if total is not None: + click.echo(f" Total: {total} Kai") + for item in spec.get("line_items") or []: + if isinstance(item, dict): + click.echo( + f" - {item.get('product')}: {item.get('amount_kai')} Kai" + f" ({item.get('description', '')})" + ) + for note in spec.get("notes") or []: + click.echo(f" note: {note}") + + +@engagement.command("staff") +@click.option( + "--id", "engagement_id", required=True, help="Engagement id (directory name)" +) +@click.option("--role", "role_id", required=True, help="Role package id under roles/") +@click.option("--client", "client_id", required=True, help="Client id") +@click.option( + "--target", "target_id", required=True, help="Primary target id (e.g. host)" +) +@click.option( + "--target-kind", + default="host", + show_default=True, + help="Target kind", +) +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +@click.option("--force", is_flag=True, help="Overwrite scaffold if directory exists") +def engagement_staff( + engagement_id: str, + role_id: str, + client_id: str, + target_id: str, + target_kind: str, + repo_root: Optional[str], + force: bool, +): + """Scaffold a pilot engagement from a Role package.""" + root = Path(repo_root).resolve() if repo_root else find_repo_root() + try: + eng = staff_engagement( + engagement_id=engagement_id, + role_id=role_id, + client_id=client_id, + target_id=target_id, + target_kind=target_kind, + repo_root=root, + force=force, + ) + except EngagementError as exc: + click.echo(f"Error: {exc}", err=True) + sys.exit(1) + click.echo(f"Staffed engagement: {eng.engagement_id}") + click.echo(f" Phase: {eng.phase}") + click.echo(f" Path: {eng.path}") + click.echo(f" Next: kaizen-agentic engagement validate {eng.engagement_id}") + click.echo( + f" kaizen-agentic engagement phase {eng.engagement_id} --to ramp_up" + ) + + +@engagement.command("scrub") +@click.argument("engagement_ref") +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") +def engagement_scrub(engagement_ref: str, repo_root: Optional[str], as_json: bool): + """Heuristic scan vault/reports for secrets before lesson contribution.""" + eng, _root = _resolve_eng(engagement_ref, repo_root) + report = scrub_engagement(eng) + if as_json: + click.echo(json.dumps(report, indent=2)) + return + click.echo( + f"Scrub {report['engagement_id']}: " + f"{report['files_scanned']} files, {len(report['hits'])} hit(s)" + ) + for h in report["hits"]: + click.echo(f" {h['file']}:{h['line']} [{h['pattern']}] {h['snippet']}") + if report["clean"]: + click.echo(" ✅ no heuristic hits (still review before contribute_lesson)") + else: + click.echo(" ⚠ review hits before any Role craft contribution") + sys.exit(1) + + +@engagement.command("export-handoff") +@click.argument("engagement_ref") +@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") +def engagement_export_handoff(engagement_ref: str, repo_root: Optional[str]): + """Ensure vault/handoff pack exists and copy baselines into it.""" + eng, _root = _resolve_eng(engagement_ref, repo_root) + handoff = export_handoff(eng) + click.echo(f"Handoff pack: {handoff}") + + @cli.group() def schedule(): """Prepare and validate scheduled agent runs (.kaizen/schedule.yml, ADR-005). diff --git a/src/kaizen_agentic/engagement.py b/src/kaizen_agentic/engagement.py new file mode 100644 index 0000000..39089a2 --- /dev/null +++ b/src/kaizen_agentic/engagement.py @@ -0,0 +1,869 @@ +"""Forward-deployed engagement lifecycle helpers (KAIZEN-WP-0009 / DEC-FDA-001). + +File-based Phase 1: ENGAGEMENT.yaml under engagements/pilots// is the source +of truth. This module loads, validates, prepares session bundles, and transitions +phases — it does not invoke LLMs or touch production hosts. +""" + +from __future__ import annotations + +import re +import shutil +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import Any, Dict, List, Optional, Pattern, Sequence, Tuple + +import yaml + +ENGAGEMENT_FILENAME = "ENGAGEMENT.yaml" +DEFAULT_PILOTS_DIR = Path("engagements") / "pilots" +DEFAULT_ROLES_DIR = Path("roles") + +VALID_PHASES = ( + "requested", + "quoting", + "funded", + "staffing", + "ramp_up", + "operating", + "renewing", + "ramp_down", + "closed", + "cancelled", +) + +# Allowed forward transitions (plus renewing ↔ operating, and cancelled from most) +PHASE_TRANSITIONS: Dict[str, Tuple[str, ...]] = { + "requested": ("quoting", "cancelled"), + "quoting": ("funded", "requested", "cancelled"), + "funded": ("staffing", "cancelled"), + "staffing": ("ramp_up", "cancelled"), + "ramp_up": ("operating", "cancelled"), + "operating": ("renewing", "ramp_down", "cancelled"), + "renewing": ("operating", "ramp_down", "cancelled"), + "ramp_down": ("closed", "cancelled"), + "closed": (), + "cancelled": (), +} + +# Sensitive patterns for scrub (heuristic; not a security scanner) +_SCRUB_PATTERNS: List[Tuple[str, Pattern]] = [ + ("private_key_block", re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----")), + ("aws_access_key", re.compile(r"AKIA[0-9A-Z]{16}")), + ("bearer_token", re.compile(r"(?i)bearer\s+[a-z0-9\-._~+/]+=*")), + ( + "ipv4_privateish", + re.compile(r"\b(?:10|192\.168|172\.(?:1[6-9]|2\d|3[01]))\.\d{1,3}\.\d{1,3}\b"), + ), + ("password_assignment", re.compile(r"(?i)(password|passwd|secret)\s*[:=]\s*\S+")), +] + + +class EngagementError(Exception): + """Raised when an engagement cannot be loaded or operated on.""" + + +@dataclass +class ChecklistItem: + item_id: str + criterion: str + status: str + evidence: str + + @property + def done(self) -> bool: + return self.status.strip().lower() in ("done", "complete", "yes", "ok") + + +@dataclass +class Engagement: + """Parsed engagement record plus filesystem paths.""" + + path: Path + data: Dict[str, Any] + source_path: Path + + @property + def engagement_id(self) -> str: + meta = self.data.get("metadata") or {} + return str(meta.get("id") or self.path.name) + + @property + def phase(self) -> str: + status = self.data.get("status") or {} + return str(status.get("phase") or "requested") + + @property + def role_id(self) -> Optional[str]: + spec = self.data.get("spec") or {} + role = spec.get("role") or {} + rid = role.get("id") + return str(rid) if rid else None + + @property + def targets(self) -> List[Dict[str, Any]]: + spec = self.data.get("spec") or {} + targets = spec.get("targets") or [] + return list(targets) if isinstance(targets, list) else [] + + def rel(self, *parts: str) -> Path: + return self.path.joinpath(*parts) + + def agent_definition_path(self) -> Optional[Path]: + spec = self.data.get("spec") or {} + ad = spec.get("agent_definition") or {} + rel = ad.get("path") + if not rel: + return None + p = self.path / rel + return p if p.exists() else p + + def vault_memory_path(self) -> Path: + spec = self.data.get("spec") or {} + vault = spec.get("vault") or {} + rel = vault.get("memory") or "vault/memory.md" + return self.path / rel + + def checklist_path(self, which: str) -> Optional[Path]: + spec = self.data.get("spec") or {} + checks = spec.get("checklists") or {} + key = "ramp_up" if which in ("ramp_up", "ramp-up", "up") else "ramp_down" + if which in ("ramp_down", "ramp-down", "down"): + key = "ramp_down" + rel = checks.get(key) + if not rel: + default = ( + "checklists/ramp-up-status.md" + if key == "ramp_up" + else "checklists/ramp-down-status.md" + ) + rel = default + p = self.path / rel + return p if p.exists() else None + + def quote_path(self) -> Optional[Path]: + spec = self.data.get("spec") or {} + commercial = spec.get("commercial") or {} + rel = commercial.get("quote_ref") or "commercial/quote.yaml" + p = self.path / rel + return p if p.exists() else None + + def ledger_path(self) -> Path: + spec = self.data.get("spec") or {} + commercial = spec.get("commercial") or {} + rel = commercial.get("ledger_ref") or "commercial/ledger.jsonl" + return self.path / rel + + +def today_iso() -> str: + return date.today().isoformat() + + +def find_repo_root(start: Optional[Path] = None) -> Path: + """Walk parents for engagements/ or roles/ or .git.""" + cur = (start or Path.cwd()).resolve() + for candidate in [cur, *cur.parents]: + if (candidate / "engagements").is_dir() or (candidate / "roles").is_dir(): + return candidate + if (candidate / ".git").exists() and ( + (candidate / "pyproject.toml").exists() + or (candidate / "workplans").is_dir() + ): + return candidate + return cur + + +def list_engagement_dirs(repo_root: Path) -> List[Path]: + """Discover engagement directories (pilots first, then other trees).""" + found: List[Path] = [] + pilots = repo_root / DEFAULT_PILOTS_DIR + if pilots.is_dir(): + for child in sorted(pilots.iterdir()): + if child.is_dir() and (child / ENGAGEMENT_FILENAME).exists(): + found.append(child) + engagements = repo_root / "engagements" + if engagements.is_dir(): + for child in sorted(engagements.iterdir()): + if child.name == "pilots": + continue + if child.is_dir() and (child / ENGAGEMENT_FILENAME).exists(): + found.append(child) + return found + + +def resolve_engagement_dir( + engagement_ref: str, repo_root: Optional[Path] = None +) -> Path: + """Resolve an engagement id or path to its directory.""" + root = (repo_root or find_repo_root()).resolve() + ref = Path(engagement_ref) + + if ref.is_dir() and (ref / ENGAGEMENT_FILENAME).exists(): + return ref.resolve() + if ref.is_file() and ref.name == ENGAGEMENT_FILENAME: + return ref.parent.resolve() + + # id under pilots/ + pilot = root / DEFAULT_PILOTS_DIR / engagement_ref + if (pilot / ENGAGEMENT_FILENAME).exists(): + return pilot.resolve() + + # bare engagements/ + bare = root / "engagements" / engagement_ref + if (bare / ENGAGEMENT_FILENAME).exists(): + return bare.resolve() + + # search by metadata id + for d in list_engagement_dirs(root): + try: + eng = load_engagement(d) + except EngagementError: + continue + if eng.engagement_id == engagement_ref or d.name == engagement_ref: + return d + + raise EngagementError( + f"Engagement not found: {engagement_ref!r} " + f"(looked under {root / DEFAULT_PILOTS_DIR})" + ) + + +def load_engagement(path: Path) -> Engagement: + """Load ENGAGEMENT.yaml from a directory (or path to the file).""" + path = Path(path) + if path.is_file(): + source = path + base = path.parent + else: + source = path / ENGAGEMENT_FILENAME + base = path + if not source.exists(): + raise EngagementError(f"Missing {ENGAGEMENT_FILENAME}: {source}") + try: + data = yaml.safe_load(source.read_text(encoding="utf-8")) + except yaml.YAMLError as exc: + raise EngagementError(f"Invalid YAML in {source}: {exc}") from exc + if not isinstance(data, dict): + raise EngagementError(f"{source} must contain a mapping") + if data.get("kind") and data.get("kind") != "Engagement": + raise EngagementError( + f"{source}: kind must be Engagement (got {data.get('kind')})" + ) + return Engagement(path=base.resolve(), data=data, source_path=source.resolve()) + + +def save_engagement(eng: Engagement) -> None: + """Write ENGAGEMENT.yaml back (preserves structure via dump of data).""" + eng.source_path.write_text( + yaml.safe_dump(eng.data, sort_keys=False, allow_unicode=True), + encoding="utf-8", + ) + + +def parse_checklist_markdown(text: str) -> List[ChecklistItem]: + """Parse checklist tables with columns ID | Criterion | Status | Evidence.""" + items: List[ChecklistItem] = [] + for line in text.splitlines(): + line = line.strip() + if not line.startswith("|"): + continue + cells = [c.strip() for c in line.strip("|").split("|")] + if len(cells) < 4: + continue + item_id, criterion, status, evidence = cells[0], cells[1], cells[2], cells[3] + if item_id.lower() in ("id", "---") or set(item_id) <= {"-", ":"}: + continue + if not re.match(r"^[A-Z]{1,3}-\d+", item_id): + continue + items.append( + ChecklistItem( + item_id=item_id, + criterion=criterion, + status=status, + evidence=evidence, + ) + ) + return items + + +def load_checklist( + eng: Engagement, which: str +) -> Tuple[Optional[Path], List[ChecklistItem]]: + path = eng.checklist_path(which) + if path is None: + return None, [] + items = parse_checklist_markdown(path.read_text(encoding="utf-8")) + return path, items + + +def checklist_summary(items: Sequence[ChecklistItem]) -> Dict[str, Any]: + done = sum(1 for i in items if i.done) + total = len(items) + return { + "done": done, + "total": total, + "complete": total > 0 and done == total, + "pending": [i.item_id for i in items if not i.done], + } + + +def set_checklist_item_status(checklist_path: Path, item_id: str, status: str) -> bool: + """Update a checklist row status by item id. Returns True if updated.""" + text = checklist_path.read_text(encoding="utf-8") + lines = text.splitlines() + changed = False + out: List[str] = [] + for line in lines: + stripped = line.strip() + if stripped.startswith("|"): + cells = [c.strip() for c in stripped.strip("|").split("|")] + if len(cells) >= 4 and cells[0] == item_id: + cells[2] = status + line = "| " + " | ".join(cells) + " |" + changed = True + out.append(line) + if changed: + checklist_path.write_text( + "\n".join(out) + ("\n" if text.endswith("\n") else ""), encoding="utf-8" + ) + return changed + + +def set_phase( + eng: Engagement, + new_phase: str, + *, + force: bool = False, + notes: Optional[str] = None, +) -> Engagement: + """Transition engagement phase; updates YAML on disk.""" + new_phase = new_phase.strip() + if new_phase not in VALID_PHASES: + raise EngagementError( + f"Invalid phase {new_phase!r}; expected one of: {', '.join(VALID_PHASES)}" + ) + current = eng.phase + if not force and new_phase != current: + allowed = PHASE_TRANSITIONS.get(current, ()) + if new_phase not in allowed: + raise EngagementError( + f"Illegal phase transition {current!r} → {new_phase!r}. " + f"Allowed: {', '.join(allowed) or '(terminal)'}. Use --force to override." + ) + status = eng.data.setdefault("status", {}) + if not isinstance(status, dict): + eng.data["status"] = {} + status = eng.data["status"] + status["phase"] = new_phase + if notes is not None: + status["notes"] = notes + meta = eng.data.setdefault("metadata", {}) + if isinstance(meta, dict): + meta["updated"] = today_iso() + save_engagement(eng) + # Keep agent definition phase in sync if present + _sync_agent_definition_phase(eng, new_phase) + return load_engagement(eng.path) + + +def _sync_agent_definition_phase(eng: Engagement, phase: str) -> None: + ad_path = eng.agent_definition_path() + if ad_path is None or not ad_path.exists(): + return + text = ad_path.read_text(encoding="utf-8") + if not text.startswith("---"): + return + parts = text.split("---", 2) + if len(parts) < 3: + return + try: + fm = yaml.safe_load(parts[1]) or {} + except yaml.YAMLError: + return + if not isinstance(fm, dict): + return + fm["phase"] = phase + new_fm = yaml.safe_dump(fm, sort_keys=False, allow_unicode=True).strip() + ad_path.write_text(f"---\n{new_fm}\n---{parts[2]}", encoding="utf-8") + + +def validate_engagement(eng: Engagement) -> List[str]: + """Return list of validation errors (empty if ok).""" + errors: List[str] = [] + if eng.data.get("apiVersion") != "kaizen.agentic/v1": + errors.append("apiVersion should be kaizen.agentic/v1") + if eng.phase not in VALID_PHASES: + errors.append(f"unknown phase: {eng.phase}") + if not eng.role_id: + errors.append("spec.role.id missing") + if not eng.targets: + errors.append("spec.targets is empty") + ad = eng.agent_definition_path() + if ad is None: + errors.append("spec.agent_definition.path missing") + elif not ad.exists(): + errors.append(f"agent definition missing: {ad}") + mem = eng.vault_memory_path() + if not mem.exists(): + errors.append(f"vault memory missing: {mem}") + for which in ("ramp_up", "ramp_down"): + path, items = load_checklist(eng, which) + if path is None: + errors.append(f"checklist missing: {which}") + elif not items: + errors.append(f"checklist empty or unparseable: {path}") + access = eng.path / "access-plan.md" + if not access.exists(): + errors.append("access-plan.md missing") + return errors + + +def build_prepare_bundle( + eng: Engagement, repo_root: Optional[Path] = None +) -> Dict[str, Any]: + """Assemble an offline session orientation bundle for this engagement.""" + root = (repo_root or find_repo_root(eng.path)).resolve() + ad_path = eng.agent_definition_path() + agent_prompt = ( + ad_path.read_text(encoding="utf-8") if ad_path and ad_path.exists() else None + ) + mem_path = eng.vault_memory_path() + memory = mem_path.read_text(encoding="utf-8") if mem_path.exists() else None + access_path = eng.path / "access-plan.md" + access_plan = ( + access_path.read_text(encoding="utf-8") if access_path.exists() else None + ) + + phase = eng.phase + phase_instructions = None + role_dir = root / DEFAULT_ROLES_DIR / (eng.role_id or "") + if phase == "ramp_up" and (role_dir / "ramp-up.md").exists(): + phase_instructions = (role_dir / "ramp-up.md").read_text(encoding="utf-8") + elif phase == "ramp_down" and (role_dir / "ramp-down.md").exists(): + phase_instructions = (role_dir / "ramp-down.md").read_text(encoding="utf-8") + + protocols: Dict[str, Optional[str]] = {} + role_yaml = role_dir / "ROLE.yaml" + if role_yaml.exists(): + try: + role_data = yaml.safe_load(role_yaml.read_text(encoding="utf-8")) or {} + except yaml.YAMLError: + role_data = {} + for p in (role_data.get("spec") or {}).get("protocols") or []: + if not isinstance(p, dict): + continue + slug = p.get("slug") or p.get("path") + rel = p.get("path") + content = None + if rel: + full = root / rel + if full.exists(): + content = full.read_text(encoding="utf-8") + protocols[str(slug)] = content + + _, ramp_up_items = load_checklist(eng, "ramp_up") + _, ramp_down_items = load_checklist(eng, "ramp_down") + + return { + "engagement_id": eng.engagement_id, + "phase": phase, + "role_id": eng.role_id, + "targets": eng.targets, + "generated": today_iso(), + "path": str(eng.path), + "agent_prompt": agent_prompt, + "agent_prompt_found": agent_prompt is not None, + "memory": memory, + "access_plan": access_plan, + "phase_instructions": phase_instructions, + "protocols": {k: (v is not None) for k, v in protocols.items()}, + "protocol_bodies": {k: v for k, v in protocols.items() if v}, + "checklist_ramp_up": checklist_summary(ramp_up_items), + "checklist_ramp_down": checklist_summary(ramp_down_items), + "validation_errors": validate_engagement(eng), + "session_close": [ + "Update vault memory (session log, findings, envelope)", + "Write reports/-*.md for the duty", + "Append commercial/ledger.jsonl Kai duty charge (no secrets)", + "kaizen-agentic engagement checklist # refresh status", + ], + "confidentiality": "client_owned", + } + + +def render_prepare_markdown(bundle: Dict[str, Any]) -> str: + lines = [ + f"# Engagement Prepare: {bundle['engagement_id']}", + f"Phase: {bundle['phase']}", + f"Role: {bundle.get('role_id')}", + f"Generated: {bundle['generated']}", + f"Path: {bundle['path']}", + "", + "## Targets", + ] + for t in bundle.get("targets") or []: + lines.append(f"- {t.get('kind', 'target')}: `{t.get('id')}`") + lines.append("") + + errs = bundle.get("validation_errors") or [] + if errs: + lines.append("## Validation warnings") + for e in errs: + lines.append(f"- {e}") + lines.append("") + + ru = bundle.get("checklist_ramp_up") or {} + rd = bundle.get("checklist_ramp_down") or {} + lines.append("## Checklists") + lines.append( + f"- ramp_up: {ru.get('done', 0)}/{ru.get('total', 0)}" + + (f" pending={ru.get('pending')}" if ru.get("pending") else "") + ) + lines.append( + f"- ramp_down: {rd.get('done', 0)}/{rd.get('total', 0)}" + + (f" pending={rd.get('pending')}" if rd.get("pending") else "") + ) + lines.append("") + + lines.append("## Access Plan") + lines.append(bundle.get("access_plan") or "(missing access-plan.md)") + lines.append("") + + if bundle.get("phase_instructions"): + lines.append("## Phase Instructions") + lines.append(bundle["phase_instructions"]) + lines.append("") + + lines.append("## Agent Definition") + if bundle.get("agent_prompt_found"): + lines.append(bundle["agent_prompt"]) + else: + lines.append("(agent definition not found)") + lines.append("") + + lines.append("## Vault Memory") + lines.append(bundle.get("memory") or "(no memory yet)") + lines.append("") + + bodies = bundle.get("protocol_bodies") or {} + if bodies: + lines.append("## Protocols") + for slug, body in bodies.items(): + lines.append(f"### {slug}") + lines.append(body) + lines.append("") + else: + present = bundle.get("protocols") or {} + if present: + lines.append("## Protocols (paths only)") + for slug, ok in present.items(): + lines.append(f"- {slug}: {'loaded' if ok else 'missing'}") + lines.append("") + + lines.append("## Session Close") + for cmd in bundle.get("session_close") or []: + lines.append(f"- {cmd}") + lines.append("") + lines.append( + "_Confidentiality: operational knowledge is client-owned; " + "do not copy vault content into supplier public agents._" + ) + return "\n".join(lines) + + +def scrub_engagement(eng: Engagement) -> Dict[str, Any]: + """Heuristic scan of vault + reports for sensitive patterns.""" + hits: List[Dict[str, Any]] = [] + scan_roots = [ + eng.path / "vault", + eng.path / "reports", + eng.path / "access-plan.md", + ] + files: List[Path] = [] + for root in scan_roots: + if root.is_file(): + files.append(root) + elif root.is_dir(): + files.extend( + p + for p in root.rglob("*") + if p.is_file() + and p.suffix in {".md", ".txt", ".yml", ".yaml", ".json", ".jsonl"} + ) + + for fpath in files: + try: + text = fpath.read_text(encoding="utf-8", errors="replace") + except OSError: + continue + for label, pattern in _SCRUB_PATTERNS: + for match in pattern.finditer(text): + line_no = text[: match.start()].count("\n") + 1 + hits.append( + { + "file": str(fpath.relative_to(eng.path)), + "line": line_no, + "pattern": label, + "snippet": match.group(0)[:40] + + ("…" if len(match.group(0)) > 40 else ""), + } + ) + return { + "engagement_id": eng.engagement_id, + "files_scanned": len(files), + "hits": hits, + "clean": len(hits) == 0, + "note": "Heuristic only — human review required before contribute_lesson", + } + + +def export_handoff(eng: Engagement) -> Path: + """Ensure handoff pack exists; return path to handoff directory.""" + handoff = eng.path / "vault" / "handoff" + handoff.mkdir(parents=True, exist_ok=True) + readme = handoff / "README.md" + if not readme.exists(): + readme.write_text( + f"# Handoff — {eng.engagement_id}\n\nGenerated: {today_iso()}\n", + encoding="utf-8", + ) + risks = handoff / "risks.md" + if not risks.exists(): + risks.write_text( + "# Outstanding risks\n\n| Risk | Severity | Status |\n|------|----------|--------|\n", + encoding="utf-8", + ) + # Copy baseline files into handoff/baselines if present + baselines = eng.path / "vault" / "baselines" + if baselines.is_dir(): + dest = handoff / "baselines" + dest.mkdir(exist_ok=True) + for f in baselines.iterdir(): + if f.is_file(): + shutil.copy2(f, dest / f.name) + return handoff + + +def staff_engagement( + *, + engagement_id: str, + role_id: str, + client_id: str, + target_id: str, + target_kind: str = "host", + repo_root: Optional[Path] = None, + force: bool = False, +) -> Engagement: + """Scaffold a new pilot engagement directory from a Role package.""" + root = (repo_root or find_repo_root()).resolve() + role_dir = root / DEFAULT_ROLES_DIR / role_id + if not role_dir.is_dir(): + raise EngagementError(f"Role package not found: {role_dir}") + + dest = root / DEFAULT_PILOTS_DIR / engagement_id + if dest.exists() and not force: + if (dest / ENGAGEMENT_FILENAME).exists(): + return load_engagement(dest) + raise EngagementError(f"Directory exists without engagement: {dest}") + + dest.mkdir(parents=True, exist_ok=True) + (dest / "vault" / "baselines").mkdir(parents=True, exist_ok=True) + (dest / "vault" / "session-log").mkdir(parents=True, exist_ok=True) + (dest / "vault" / "findings").mkdir(parents=True, exist_ok=True) + (dest / "vault" / "handoff").mkdir(parents=True, exist_ok=True) + (dest / "checklists").mkdir(parents=True, exist_ok=True) + (dest / "commercial").mkdir(parents=True, exist_ok=True) + (dest / "reports").mkdir(parents=True, exist_ok=True) + + role_version = "0.1.0" + role_yaml = role_dir / "ROLE.yaml" + if role_yaml.exists(): + try: + rd = yaml.safe_load(role_yaml.read_text(encoding="utf-8")) or {} + role_version = str( + (rd.get("metadata") or {}).get("version") or role_version + ) + except yaml.YAMLError: + pass + + # Memory from template + mem_tmpl = role_dir / "memory-template.md" + mem_path = dest / "vault" / "memory.md" + if mem_tmpl.exists(): + mem_text = mem_tmpl.read_text(encoding="utf-8") + mem_text = mem_text.replace("", engagement_id) + mem_text = mem_text.replace( + "", f"{client_id}-{target_id}" + ) + mem_text = mem_text.replace("", today_iso()) + mem_path.write_text(mem_text, encoding="utf-8") + else: + mem_path.write_text( + f"---\nagent: {role_id}\nengagement_id: {engagement_id}\n" + f'last_updated: "{today_iso()}"\nsession_count: 0\n---\n\n# Memory\n', + encoding="utf-8", + ) + + # Agent definition from role template + binding frontmatter + ad_tmpl = role_dir / "agent-definition.md" + agent_name = f"agent-{role_id}.md" + if ad_tmpl.exists(): + body = ad_tmpl.read_text(encoding="utf-8") + # Replace or inject engagement binding frontmatter + if body.startswith("---"): + parts = body.split("---", 2) + fm = yaml.safe_load(parts[1]) if len(parts) >= 3 else {} + if not isinstance(fm, dict): + fm = {} + fm.update( + { + "engagement_id": engagement_id, + "role_id": role_id, + "role_version": role_version, + "phase": "staffing", + "memory_path": f"engagements/pilots/{engagement_id}/vault/memory.md", + "targets": [{"kind": target_kind, "id": target_id}], + "confidentiality": "client_owned", + "access_classes": ["host_observe", "privileged_ops"], + "human_approval_for": [ + "privileged_ops", + "package_upgrade", + "firewall_change", + "reboot", + ], + } + ) + new_fm = yaml.safe_dump(fm, sort_keys=False, allow_unicode=True).strip() + body = f"---\n{new_fm}\n---{parts[2] if len(parts) >= 3 else ''}" + (dest / agent_name).write_text(body, encoding="utf-8") + else: + (dest / agent_name).write_text( + f"---\nname: {role_id}\nengagement_id: {engagement_id}\nphase: staffing\n---\n\n# {role_id}\n", + encoding="utf-8", + ) + + # Checklists from role ramp docs (status tables) + for which, fname, prefix in ( + ("ramp_up", "ramp-up-status.md", "RU"), + ("ramp_down", "ramp-down-status.md", "RD"), + ): + role_check = role_dir / ("ramp-up.md" if which == "ramp_up" else "ramp-down.md") + dest_check = dest / "checklists" / fname + if role_check.exists(): + # Extract table rows from role checklist + items = parse_checklist_markdown(role_check.read_text(encoding="utf-8")) + lines = [ + f"# {which.replace('_', '-').title()} status — {engagement_id}", + "", + f"Source checklist: `roles/{role_id}/{role_check.name}`", + "", + "| ID | Criterion | Status | Evidence |", + "|----|-----------|--------|----------|", + ] + for it in items: + lines.append( + f"| {it.item_id} | {it.criterion} | todo | {it.evidence} |" + ) + if not items: + lines.append( + f"| {prefix}-01 | Complete {which} | todo | see role checklist |" + ) + dest_check.write_text("\n".join(lines) + "\n", encoding="utf-8") + else: + dest_check.write_text( + f"# {which}\n\n| ID | Criterion | Status | Evidence |\n" + f"|----|-----------|--------|----------|\n" + f"| {prefix}-01 | Complete | todo | |\n", + encoding="utf-8", + ) + + (dest / "access-plan.md").write_text( + f"# Access plan — {engagement_id}\n\n" + f"**Target:** {target_kind} `{target_id}`\n\n" + "Secrets: never stored in this tree.\n\n" + "## Verification log\n\n| Date | Result | Notes |\n|------|--------|-------|\n| _pending_ | | RU-01 |\n", + encoding="utf-8", + ) + (dest / "schedule.yml").write_text( + f"engagement_id: {engagement_id}\ntimezone: Europe/Berlin\nentries: []\n", + encoding="utf-8", + ) + (dest / "vault" / "baselines" / f"{target_id}.md").write_text( + f"# Baseline — {target_id}\n\n**Status:** pending\n", + encoding="utf-8", + ) + (dest / "commercial" / "ledger.jsonl").write_text("", encoding="utf-8") + (dest / "commercial" / "quote.yaml").write_text( + f"apiVersion: kaizen.agentic/v1\nkind: KaiQuote\nmetadata:\n" + f' engagement_id: {engagement_id}\n created: "{today_iso()}"\n' + f" currency: KAI\nspec:\n total_kai: 0\n notes: [scaffold — fill quote]\n", + encoding="utf-8", + ) + (dest / "request.yaml").write_text( + f"apiVersion: kaizen.agentic/v1\nkind: EngagementRequest\n" + f'metadata:\n id: engreq-{engagement_id}\n created: "{today_iso()}"\n' + f"spec:\n client:\n id: {client_id}\n role:\n id: {role_id}\n" + f" targets:\n - kind: {target_kind}\n id: {target_id}\n", + encoding="utf-8", + ) + + eng_doc = { + "apiVersion": "kaizen.agentic/v1", + "kind": "Engagement", + "metadata": { + "id": engagement_id, + "created": today_iso(), + "updated": today_iso(), + }, + "status": { + "phase": "staffing", + "notes": "Scaffolded by kaizen-agentic engagement staff", + }, + "spec": { + "client": {"id": client_id}, + "role": {"id": role_id, "version": role_version}, + "targets": [{"kind": target_kind, "id": target_id}], + "agent_definition": { + "path": agent_name, + "derived_from": f"roles/{role_id}/agent-definition.md", + }, + "vault": {"root": "vault/", "memory": "vault/memory.md"}, + "schedule": {"path": "schedule.yml"}, + "checklists": { + "ramp_up": "checklists/ramp-up-status.md", + "ramp_down": "checklists/ramp-down-status.md", + }, + "policy": { + "confidentiality": "client_owned", + "contribute_metrics": False, + "human_approval_for": [ + "privileged_ops", + "package_upgrade", + "firewall_change", + "reboot", + ], + }, + "commercial": { + "currency": "kai", + "ledger_ref": "commercial/ledger.jsonl", + "quote_ref": "commercial/quote.yaml", + }, + }, + } + (dest / ENGAGEMENT_FILENAME).write_text( + yaml.safe_dump(eng_doc, sort_keys=False, allow_unicode=True), + encoding="utf-8", + ) + return load_engagement(dest) + + +def load_quote(eng: Engagement) -> Optional[Dict[str, Any]]: + path = eng.quote_path() + if path is None: + return None + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + except yaml.YAMLError as exc: + raise EngagementError(f"Invalid quote YAML: {exc}") from exc + return data if isinstance(data, dict) else None diff --git a/tests/test_engagement_cli.py b/tests/test_engagement_cli.py new file mode 100644 index 0000000..718437b --- /dev/null +++ b/tests/test_engagement_cli.py @@ -0,0 +1,365 @@ +"""CLI + module tests for forward-deployed engagements (KAIZEN-WP-0009 T08).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +import yaml +from click.testing import CliRunner + +from kaizen_agentic.cli import cli +from kaizen_agentic.engagement import ( + EngagementError, + build_prepare_bundle, + checklist_summary, + load_checklist, + load_engagement, + parse_checklist_markdown, + set_checklist_item_status, + set_phase, + staff_engagement, + validate_engagement, +) + + +@pytest.fixture +def runner() -> CliRunner: + return CliRunner() + + +@pytest.fixture +def mini_repo(tmp_path: Path) -> Path: + """Minimal repo with host-operator Role package.""" + root = tmp_path / "kaizen-root" + role = root / "roles" / "host-operator" + role.mkdir(parents=True) + (role / "ROLE.yaml").write_text( + yaml.safe_dump( + { + "apiVersion": "kaizen.agentic/v1", + "kind": "Role", + "metadata": {"id": "host-operator", "version": "0.1.0"}, + "spec": { + "protocols": [ + { + "agent": "host-operator", + "slug": "load-workload-review", + "path": "roles/host-operator/protocols/load-workload-review.md", + } + ] + }, + } + ), + encoding="utf-8", + ) + (role / "agent-definition.md").write_text( + "---\nname: host-operator\ncategory: infrastructure\nmemory: enabled\n---\n\n# Host Operator\n\nDo ops.\n", + encoding="utf-8", + ) + (role / "memory-template.md").write_text( + "---\nagent: host-operator\nengagement_id: \n---\n\n# Memory\n", + encoding="utf-8", + ) + (role / "ramp-up.md").write_text( + "# Ramp-up\n\n| ID | Criterion | Status | Evidence |\n" + "|----|-----------|--------|----------|\n" + "| RU-01 | Access path verified | todo | access-plan.md |\n" + "| RU-02 | Host baseline documented | todo | vault/baselines |\n", + encoding="utf-8", + ) + (role / "ramp-down.md").write_text( + "# Ramp-down\n\n| ID | Criterion | Status | Evidence |\n" + "|----|-----------|--------|----------|\n" + "| RD-01 | Open threads triaged | todo | memory |\n", + encoding="utf-8", + ) + proto = role / "protocols" + proto.mkdir() + (proto / "load-workload-review.md").write_text( + "---\nslug: load-workload-review\n---\n\n# Load review\n", + encoding="utf-8", + ) + (root / "engagements" / "pilots").mkdir(parents=True) + return root + + +class TestChecklistParse: + def test_parse_checklist_rows(self): + text = ( + "| ID | Criterion | Status | Evidence |\n" + "|----|-----------|--------|----------|\n" + "| RU-01 | Access | done | access-plan.md |\n" + "| RU-02 | Baseline | todo | vault |\n" + ) + items = parse_checklist_markdown(text) + assert len(items) == 2 + assert items[0].done is True + assert items[1].done is False + summary = checklist_summary(items) + assert summary == { + "done": 1, + "total": 2, + "complete": False, + "pending": ["RU-02"], + } + + def test_set_checklist_item_status(self, tmp_path: Path): + p = tmp_path / "c.md" + p.write_text( + "| ID | Criterion | Status | Evidence |\n" + "|----|-----------|--------|----------|\n" + "| RU-01 | Access | todo | a |\n", + encoding="utf-8", + ) + assert set_checklist_item_status(p, "RU-01", "done") + assert "RU-01" in p.read_text() + assert "done" in p.read_text() + items = parse_checklist_markdown(p.read_text()) + assert items[0].done + + +class TestStaffAndLifecycle: + def test_staff_validate_phase_prepare(self, mini_repo: Path): + eng = staff_engagement( + engagement_id="eng-test-host-001", + role_id="host-operator", + client_id="coulomb", + target_id="railiance01", + repo_root=mini_repo, + ) + assert eng.phase == "staffing" + assert eng.engagement_id == "eng-test-host-001" + assert (eng.path / "vault" / "memory.md").exists() + assert (eng.path / "agent-host-operator.md").exists() + + errs = validate_engagement(eng) + assert errs == [] + + path, items = load_checklist(eng, "ramp_up") + assert path is not None + assert any(i.item_id == "RU-01" for i in items) + + eng = set_phase(eng, "ramp_up") + assert eng.phase == "ramp_up" + ad = (eng.path / "agent-host-operator.md").read_text(encoding="utf-8") + assert "phase: ramp_up" in ad or "phase: ramp_up" in ad.replace('"', "") + + bundle = build_prepare_bundle(eng, mini_repo) + assert bundle["engagement_id"] == "eng-test-host-001" + assert bundle["agent_prompt_found"] is True + assert bundle["phase"] == "ramp_up" + assert "Access" in (bundle.get("phase_instructions") or "") + + with pytest.raises(EngagementError): + set_phase(eng, "closed") # illegal without force + + eng = set_phase(eng, "closed", force=True) + assert eng.phase == "closed" + + +class TestEngagementCli: + def test_staff_list_checklist_prepare(self, runner: CliRunner, mini_repo: Path): + result = runner.invoke( + cli, + [ + "engagement", + "staff", + "--id", + "eng-cli-001", + "--role", + "host-operator", + "--client", + "coulomb", + "--target", + "railiance01", + "--repo-root", + str(mini_repo), + ], + ) + assert result.exit_code == 0, result.output + assert "Staffed engagement" in result.output + + result = runner.invoke( + cli, ["engagement", "list", "--repo-root", str(mini_repo), "--json"] + ) + assert result.exit_code == 0, result.output + rows = json.loads(result.output) + assert any(r["id"] == "eng-cli-001" for r in rows) + + result = runner.invoke( + cli, + [ + "engagement", + "validate", + "eng-cli-001", + "--repo-root", + str(mini_repo), + ], + ) + assert result.exit_code == 0, result.output + + result = runner.invoke( + cli, + [ + "engagement", + "checklist", + "eng-cli-001", + "--repo-root", + str(mini_repo), + "--mark", + "RU-01=done", + ], + ) + assert result.exit_code == 0, result.output + assert "RU-01" in result.output + + result = runner.invoke( + cli, + [ + "engagement", + "checklist", + "eng-cli-001", + "--repo-root", + str(mini_repo), + "--json", + ], + ) + assert result.exit_code == 0, result.output + data = json.loads(result.output) + ru = data["checklists"]["ramp_up"]["items"] + assert any(i["id"] == "RU-01" and i["done"] for i in ru) + + result = runner.invoke( + cli, + [ + "engagement", + "phase", + "eng-cli-001", + "--to", + "ramp_up", + "--repo-root", + str(mini_repo), + ], + ) + assert result.exit_code == 0, result.output + assert "ramp_up" in result.output + + result = runner.invoke( + cli, + [ + "engagement", + "prepare", + "eng-cli-001", + "--repo-root", + str(mini_repo), + "--format", + "json", + ], + ) + assert result.exit_code == 0, result.output + bundle = json.loads(result.output) + assert bundle["phase"] == "ramp_up" + assert bundle["agent_prompt_found"] is True + + def test_phase_rejects_illegal_transition(self, runner: CliRunner, mini_repo: Path): + runner.invoke( + cli, + [ + "engagement", + "staff", + "--id", + "eng-cli-002", + "--role", + "host-operator", + "--client", + "c", + "--target", + "h1", + "--repo-root", + str(mini_repo), + ], + ) + result = runner.invoke( + cli, + [ + "engagement", + "phase", + "eng-cli-002", + "--to", + "closed", + "--repo-root", + str(mini_repo), + ], + ) + assert result.exit_code != 0 + assert "Illegal phase" in result.output + + def test_real_pilot_if_present(self, runner: CliRunner): + """Smoke against repo pilot when checked out as kaizen-agentic.""" + repo = Path.cwd() + pilot = ( + repo + / "engagements" + / "pilots" + / "eng-coulomb-railiance01-ho-001" + / "ENGAGEMENT.yaml" + ) + if not pilot.exists(): + pytest.skip("pilot engagement not in cwd") + result = runner.invoke( + cli, + [ + "engagement", + "show", + "eng-coulomb-railiance01-ho-001", + "--repo-root", + str(repo), + "--json", + ], + ) + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["id"] == "eng-coulomb-railiance01-ho-001" + assert data["role"] == "host-operator" + + result = runner.invoke( + cli, + [ + "engagement", + "checklist", + "eng-coulomb-railiance01-ho-001", + "--repo-root", + str(repo), + "--which", + "ramp_up", + ], + ) + assert result.exit_code == 0, result.output + assert "RU-01" in result.output + + result = runner.invoke( + cli, + [ + "engagement", + "validate", + "eng-coulomb-railiance01-ho-001", + "--repo-root", + str(repo), + ], + ) + assert result.exit_code == 0, result.output + + result = runner.invoke( + cli, + [ + "engagement", + "quote", + "eng-coulomb-railiance01-ho-001", + "--repo-root", + str(repo), + ], + ) + assert result.exit_code == 0, result.output + assert "72800" in result.output or "Kai" in result.output diff --git a/workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md b/workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md index 08e1a2e..19056f8 100644 --- a/workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md +++ b/workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md @@ -36,7 +36,7 @@ tasks: status: done title: JSONL Kai ledger convention and quote snapshot for pilot - id: T08 - status: todo + status: done title: CLI engagement command group (request, staff, prepare, checklist, scrub) - id: T09 status: todo @@ -186,18 +186,17 @@ with quote-open note, commercial README. ```task id: KAIZEN-WP-0009-T08 -status: todo +status: done priority: medium state_hub_task_id: "52f7d872-1087-4890-9439-8d6c704351fd" ``` -Implement `kaizen-agentic engagement` subcommands (Phase 2 of architecture): +**Delivered:** `kaizen-agentic engagement` CLI group + `engagement.py` module: -`request`, `quote`, `staff`, `render-agent`, `init-vault`, `phase`, `checklist`, -`prepare`, `scrub`, `export-handoff` — minimum viable subset for pilot may be -`staff`, `checklist`, `prepare`, `phase`. +`list`, `show`, `validate`, `checklist` (+ `--mark`), `phase`, `prepare`, +`staff`, `quote`, `scrub`, `export-handoff`. -Tests for schema validate and checklist status. +Tests: `tests/test_engagement_cli.py`. Docs: CLI cheat sheet section. ## Session wire-up