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

@@ -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]'`)

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

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/<id>/: 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).

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/<id>/ 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/<id>
+    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/<date>-*.md for the duty",
+            "Append commercial/ledger.jsonl Kai duty charge (no secrets)",
+            "kaizen-agentic engagement checklist <id>   # 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("<set on init>", engagement_id)
+        mem_text = mem_text.replace(
+            "<engagement or client slug>", f"{client_id}-{target_id}"
+        )
+        mem_text = mem_text.replace("<ISO date>", 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

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: <set on init>\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

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