feat: engagement close-session wires vault, metrics, Kai ledger (WP-0009 T09)

Add close_session helpers and CLI to append vault session logs, engagement-
scoped metrics, duty Kai charges, and client reports. Rejects sensitive
summaries. Pilot smoke-closed once; prepare points operators at close-session.

CHANGELOG.md

@@ -12,6 +12,10 @@ 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–T09)** — forward-deployed engagement lifecycle:
+  `list`/`show`/`validate`/`checklist`/`phase`/`prepare`/`staff`/`quote`/
+  `scrub`/`export-handoff`/`close-session` over file-based pilots
+  (`engagements/pilots/`); close-session wires vault log, metrics, and Kai ledger
 - **`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,46 @@ 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
+
+# Session close — vault log + metrics + Kai ledger + report (no secrets)
+kaizen-agentic engagement close-session eng-coulomb-railiance01-ho-001 \
+  --success --duty standard_review \
+  --summary "daily health: watch-level disk, load OK" \
+  --time 120 --quality 0.85
+# Flags: --no-metrics --no-ledger --no-report --kai-amount N --access-class host_observe
+
+# 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

engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/executions.jsonl

@@ -0,0 +1 @@
+{"agent": "host-operator", "duty": "standard_review", "engagement_id": "eng-coulomb-railiance01-ho-001", "execution_time_s": 5.0, "phase": "staffing", "quality_score": 0.7, "success": true, "timestamp": "2026-07-16T10:08:57Z"}

engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/summary.json

@@ -0,0 +1,12 @@
+{
+  "agent": "host-operator",
+  "avg_execution_time_s": 5.0,
+  "avg_quality_score": 0.7,
+  "execution_count": 1,
+  "last_execution": "2026-07-16T10:08:57Z",
+  "success_rate": 1.0,
+  "trend": {
+    "quality_score": "stable",
+    "success_rate": "stable"
+  }
+}

engagements/pilots/eng-coulomb-railiance01-ho-001/commercial/ledger.jsonl

@@ -1 +1,2 @@
 {"apiVersion":"kaizen.agentic/v1","kind":"KaiLedgerEntry","id":"kai-20260716-quote-open","account":"coulomb-ops-kai","engagement_id":"eng-coulomb-railiance01-ho-001","type":"note","product":"quote_snapshot","capability_tier":4,"amount_kai":0,"currency":"KAI","created_at":"2026-07-16T08:00:00Z","metadata":{"total_quoted_kai":72800,"phase":"staffing","note":"Month-1 estimate recorded; no charge until fund/ramp"}}
+{"access_surcharge_product": "read_only", "account": "coulomb-ops-kai", "amount_kai": 1600, "apiVersion": "kaizen.agentic/v1", "capability_tier": 4, "created_at": "2026-07-16T10:08:57Z", "currency": "KAI", "engagement_id": "eng-coulomb-railiance01-ho-001", "id": "kai-20260716T100857Z-standard_review", "kind": "KaiLedgerEntry", "metadata": {"access_class": "read_only", "phase": "staffing", "success": true, "target": "railiance01"}, "product": "standard_review", "session_ref": "reports/2026-07-16-standard-review.md", "type": "duty_charge"}

engagements/pilots/eng-coulomb-railiance01-ho-001/reports/2026-07-16-standard-review.md

@@ -0,0 +1,13 @@
+# Session report — eng-coulomb-railiance01-ho-001
+
+- **Date:** 2026-07-16
+- **Duty:** standard_review
+- **Targets:** railiance01
+- **Outcome:** success
+- **Phase:** staffing
+
+## Summary
+
+T09 wire-up smoke: prepare+close-session path verified (no host access)
+
+_Billing metadata only in commercial/ledger.jsonl; no secrets in this report._

engagements/pilots/eng-coulomb-railiance01-ho-001/vault/memory.md

@@ -2,8 +2,8 @@
 agent: host-operator
 engagement_id: eng-coulomb-railiance01-ho-001
 project: coulomb-railiance01
-last_updated: "2026-07-16"
-session_count: 0
+last_updated: '2026-07-16'
+session_count: 1
 confidentiality: client_owned
 ---
 
@@ -72,3 +72,4 @@ _None yet._
 ## Session Log
 
 <!-- YYYY-MM-DD · host(s) · key finding · outcome -->
+- 2026-07-16 · railiance01 · standard_review · T09 wire-up smoke: prepare+close-session path verified (no host access) · ok

src/kaizen_agentic/cli.py

@@ -24,6 +24,27 @@ 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 (
+    DUTY_BASE_KAI,
+    EngagementError,
+    build_prepare_bundle,
+    close_session,
+    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 +1433,485 @@ 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}")
+
+
+@engagement.command("close-session")
+@click.argument("engagement_ref")
+@click.option("--success", "outcome_success", is_flag=True, help="Session succeeded")
+@click.option("--failure", "outcome_failure", is_flag=True, help="Session failed")
+@click.option(
+    "--duty",
+    type=click.Choice(sorted(DUTY_BASE_KAI.keys())),
+    default="standard_review",
+    show_default=True,
+    help="Duty product for Kai charge",
+)
+@click.option(
+    "--summary",
+    required=True,
+    help="One-line non-secret outcome (session log + report)",
+)
+@click.option("--time", "execution_time", type=float, help="Wall time seconds")
+@click.option("--quality", type=float, help="Quality score 0.0–1.0")
+@click.option(
+    "--access-class",
+    default="host_observe",
+    show_default=True,
+    type=click.Choice(["read_only", "host_observe", "privileged_ops", "none"]),
+    help="Access surcharge for Kai (none = no surcharge)",
+)
+@click.option(
+    "--kai-amount",
+    type=int,
+    default=None,
+    help="Override Kai charge (skip catalog formula)",
+)
+@click.option(
+    "--no-metrics", is_flag=True, help="Skip .kaizen/metrics under engagement"
+)
+@click.option("--no-ledger", is_flag=True, help="Skip commercial/ledger.jsonl charge")
+@click.option("--no-report", is_flag=True, help="Skip reports/ duty report")
+@click.option("--idempotency-key", help="Skip metrics append if key already recorded")
+@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)")
+@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON result")
+def engagement_close_session(
+    engagement_ref: str,
+    outcome_success: bool,
+    outcome_failure: bool,
+    duty: str,
+    summary: str,
+    execution_time: Optional[float],
+    quality: Optional[float],
+    access_class: str,
+    kai_amount: Optional[int],
+    no_metrics: bool,
+    no_ledger: bool,
+    no_report: bool,
+    idempotency_key: Optional[str],
+    repo_root: Optional[str],
+    as_json: bool,
+):
+    """Close a duty session: vault log, metrics, Kai ledger, optional report.
+
+    Does not invoke LLMs. Rejects summaries that look like secrets. Metrics land
+    under the engagement tree; ledger lines are billing metadata only.
+    """
+    eng, _root = _resolve_eng(engagement_ref, repo_root)
+    if outcome_success and outcome_failure:
+        click.echo("Error: use only one of --success or --failure", err=True)
+        sys.exit(1)
+    if not outcome_success and not outcome_failure:
+        click.echo("Error: specify --success or --failure", err=True)
+        sys.exit(1)
+    if quality is not None and not (0.0 <= quality <= 1.0):
+        click.echo("Error: --quality must be between 0.0 and 1.0", err=True)
+        sys.exit(1)
+
+    access = None if access_class == "none" else access_class
+    try:
+        result = close_session(
+            eng,
+            success=outcome_success,
+            duty=duty,
+            summary=summary,
+            execution_time_s=execution_time,
+            quality=quality,
+            access_class=access,
+            amount_kai=kai_amount,
+            record_metrics=not no_metrics,
+            record_ledger=not no_ledger,
+            write_report=not no_report,
+            idempotency_key=idempotency_key,
+        )
+    except EngagementError as exc:
+        click.echo(f"Error: {exc}", err=True)
+        sys.exit(1)
+    except ValueError as exc:
+        click.echo(f"Error: {exc}", err=True)
+        sys.exit(1)
+
+    if as_json:
+        click.echo(json.dumps(result, indent=2))
+        return
+
+    click.echo(f"Closed session: {result['engagement_id']} ({result['duty']})")
+    click.echo(f"  Memory:  {result.get('memory_path')}")
+    if result.get("report_path"):
+        click.echo(f"  Report:  {result['report_path']}")
+    if result.get("metrics_recorded"):
+        click.echo(f"  Metrics: recorded → {result.get('metrics_path')}")
+    elif not no_metrics:
+        click.echo("  Metrics: skipped (duplicate idempotency key)")
+    if not no_ledger:
+        click.echo(
+            f"  Ledger:  {result.get('amount_kai', 0)} Kai → {result.get('ledger_path')}"
+        )
+
+
 @cli.group()
 def schedule():
     """Prepare and validate scheduled agent runs (.kaizen/schedule.yml, ADR-005).

tests/test_engagement_cli.py

@@ -0,0 +1,476 @@
+"""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,
+    close_session,
+    compute_duty_kai,
+    load_checklist,
+    load_engagement,
+    parse_checklist_markdown,
+    set_checklist_item_status,
+    set_phase,
+    staff_engagement,
+    text_has_sensitive_content,
+    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 TestCloseSession:
+    def test_compute_duty_kai(self):
+        assert (
+            compute_duty_kai("standard_review", tier=4, access_class="host_observe")
+            == 1700
+        )
+        assert compute_duty_kai("ramp_up_package", tier=4) == 10000
+        assert compute_duty_kai("short_assist", tier=1, amount_override=50) == 50
+
+    def test_sensitive_summary_rejected(self, mini_repo: Path):
+        eng = staff_engagement(
+            engagement_id="eng-close-sens",
+            role_id="host-operator",
+            client_id="c",
+            target_id="h1",
+            repo_root=mini_repo,
+        )
+        with pytest.raises(EngagementError, match="sensitive"):
+            close_session(
+                eng,
+                success=True,
+                summary="password: hunter2 leaked",
+                record_metrics=False,
+                record_ledger=False,
+                write_report=False,
+            )
+        assert text_has_sensitive_content("-----BEGIN RSA PRIVATE KEY-----")
+
+    def test_close_session_writes_all_artifacts(self, mini_repo: Path):
+        eng = staff_engagement(
+            engagement_id="eng-close-001",
+            role_id="host-operator",
+            client_id="coulomb",
+            target_id="railiance01",
+            repo_root=mini_repo,
+        )
+        result = close_session(
+            eng,
+            success=True,
+            duty="standard_review",
+            summary="first health pass watch-level disk",
+            execution_time_s=90.0,
+            quality=0.85,
+            access_class="host_observe",
+        )
+        assert result["amount_kai"] == 1700
+        assert result["metrics_recorded"] is True
+        mem = Path(result["memory_path"]).read_text(encoding="utf-8")
+        assert "first health pass" in mem
+        assert "session_count: 1" in mem or "session_count: 1" in mem.replace('"', "")
+        report = Path(result["report_path"])
+        assert report.exists()
+        assert "first health pass" in report.read_text(encoding="utf-8")
+        ledger = (
+            Path(result["ledger_path"]).read_text(encoding="utf-8").strip().splitlines()
+        )
+        last = json.loads(ledger[-1])
+        assert last["amount_kai"] == 1700
+        assert last["product"] == "standard_review"
+        assert "password" not in last
+        metrics_path = Path(result["metrics_path"])
+        assert metrics_path.exists()
+        assert "engagement_id" in metrics_path.read_text(encoding="utf-8")
+
+
+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
+
+    def test_close_session_cli(self, runner: CliRunner, mini_repo: Path):
+        runner.invoke(
+            cli,
+            [
+                "engagement",
+                "staff",
+                "--id",
+                "eng-close-cli",
+                "--role",
+                "host-operator",
+                "--client",
+                "coulomb",
+                "--target",
+                "railiance01",
+                "--repo-root",
+                str(mini_repo),
+            ],
+        )
+        result = runner.invoke(
+            cli,
+            [
+                "engagement",
+                "close-session",
+                "eng-close-cli",
+                "--success",
+                "--duty",
+                "standard_review",
+                "--summary",
+                "load review healthy",
+                "--time",
+                "60",
+                "--quality",
+                "0.9",
+                "--repo-root",
+                str(mini_repo),
+                "--json",
+            ],
+        )
+        assert result.exit_code == 0, result.output
+        data = json.loads(result.output)
+        assert data["amount_kai"] == 1700
+        assert data["metrics_recorded"] is True

workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md

@@ -36,10 +36,10 @@ 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
+    status: done
     title: Wire prepare/session-close to vault paths, metrics, and ledger entries
   - id: T10
     status: todo
@@ -186,31 +186,31 @@ 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
 
 ```task
 id: KAIZEN-WP-0009-T09
-status: todo
+status: done
 priority: medium
 state_hub_task_id: "82ac0b5a-0309-4912-93bf-f05bee2abbc2"
 ```
 
-- `engagement prepare` bundles definition + vault + protocols + access plan
-- Session close path updates vault, `metrics record`, optional Kai duty charge
-- No secrets in prepare output or ledger
+**Delivered:** `engagement close-session` wires vault session log, engagement-scoped
+metrics (`.kaizen/metrics/` under the pilot tree), Kai `ledger.jsonl` duty charge,
+and `reports/` stub. Rejects sensitive summaries. `prepare` session-close section
+points at close-session. Tests cover formula, scrub, and CLI.
 
 ## railiance01 pilot through ramp-up