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.

src/kaizen_agentic/engagement.py

@@ -7,10 +7,11 @@ phases — it does not invoke LLMs or touch production hosts.
 
 from __future__ import annotations
 
+import json
 import re
 import shutil
 from dataclasses import dataclass
-from datetime import date
+from datetime import date, datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Pattern, Sequence, Tuple
 
@@ -59,6 +60,22 @@ _SCRUB_PATTERNS: List[Tuple[str, Pattern]] = [
     ("password_assignment", re.compile(r"(?i)(password|passwd|secret)\s*[:=]\s*\S+")),
 ]
 
+# Kai duty catalog (base units before capability tier multiplier) — business model §5.4
+DUTY_BASE_KAI: Dict[str, int] = {
+    "short_assist": 100,
+    "standard_review": 400,
+    "deep_assessment": 1200,
+    "ramp_up_package": 5000,
+    "ramp_down_package": 3000,
+}
+ACCESS_SURCHARGE_KAI: Dict[str, int] = {
+    "read_only": 0,
+    "host_observe": 100,
+    "privileged_ops": 400,
+}
+# Package duties use half tier weight per business model
+DUTY_HALF_TIER = frozenset({"ramp_up_package", "ramp_down_package"})
+
 
 class EngagementError(Exception):
     """Raised when an engagement cannot be loaded or operated on."""
@@ -160,6 +177,60 @@ def today_iso() -> str:
     return date.today().isoformat()
 
 
+def utc_now_iso() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def text_has_sensitive_content(text: str) -> List[str]:
+    """Return labels of scrub patterns that match text (empty if clean)."""
+    hits: List[str] = []
+    for label, pattern in _SCRUB_PATTERNS:
+        if pattern.search(text):
+            hits.append(label)
+    return hits
+
+
+def capability_tier(eng: Engagement) -> int:
+    spec = eng.data.get("spec") or {}
+    commercial = spec.get("commercial") or {}
+    tier = commercial.get("capability_tier")
+    if tier is not None:
+        try:
+            return max(1, int(tier))
+        except (TypeError, ValueError):
+            pass
+    return 4
+
+
+def compute_duty_kai(
+    duty: str,
+    *,
+    tier: int = 4,
+    access_class: Optional[str] = None,
+    amount_override: Optional[int] = None,
+) -> int:
+    """Compute Kai charge for a duty (+ optional access surcharge)."""
+    if amount_override is not None:
+        return max(0, int(amount_override))
+    if duty not in DUTY_BASE_KAI:
+        raise EngagementError(
+            f"Unknown duty {duty!r}; expected one of: {', '.join(sorted(DUTY_BASE_KAI))}"
+        )
+    base = DUTY_BASE_KAI[duty]
+    if duty in DUTY_HALF_TIER:
+        amount = int(base * tier / 2)
+    else:
+        amount = base * tier
+    if access_class:
+        if access_class not in ACCESS_SURCHARGE_KAI:
+            raise EngagementError(
+                f"Unknown access class {access_class!r}; "
+                f"expected one of: {', '.join(sorted(ACCESS_SURCHARGE_KAI))}"
+            )
+        amount += ACCESS_SURCHARGE_KAI[access_class]
+    return amount
+
+
 def find_repo_root(start: Optional[Path] = None) -> Path:
     """Walk parents for engagements/ or roles/ or .git."""
     cur = (start or Path.cwd()).resolve()
@@ -483,10 +554,13 @@ def build_prepare_bundle(
         "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",
+            (
+                f"kaizen-agentic engagement close-session {eng.engagement_id} "
+                "--success --duty standard_review --summary '<one-line outcome>' "
+                "--time <seconds> --quality <0-1>"
+            ),
+            "Or manually: update vault memory, reports/, commercial/ledger.jsonl",
+            f"kaizen-agentic engagement checklist {eng.engagement_id}",
         ],
         "confidentiality": "client_owned",
     }
@@ -867,3 +941,240 @@ def load_quote(eng: Engagement) -> Optional[Dict[str, Any]]:
     except yaml.YAMLError as exc:
         raise EngagementError(f"Invalid quote YAML: {exc}") from exc
     return data if isinstance(data, dict) else None
+
+
+def append_session_log(eng: Engagement, line: str) -> Path:
+    """Append one session log line to vault memory under ## Session Log."""
+    sensitive = text_has_sensitive_content(line)
+    if sensitive:
+        raise EngagementError(
+            f"Session summary rejected — looks sensitive ({', '.join(sensitive)}). "
+            "Use a non-secret one-line outcome."
+        )
+    mem_path = eng.vault_memory_path()
+    mem_path.parent.mkdir(parents=True, exist_ok=True)
+    if not mem_path.exists():
+        mem_path.write_text(
+            f"---\nagent: {eng.role_id or 'agent'}\n"
+            f"engagement_id: {eng.engagement_id}\n"
+            f'last_updated: "{today_iso()}"\nsession_count: 0\n---\n\n'
+            f"# Memory\n\n## Session Log\n\n",
+            encoding="utf-8",
+        )
+    text = mem_path.read_text(encoding="utf-8")
+    # Bump session_count / last_updated in frontmatter when present
+    if text.startswith("---"):
+        parts = text.split("---", 2)
+        if len(parts) >= 3:
+            try:
+                fm = yaml.safe_load(parts[1]) or {}
+            except yaml.YAMLError:
+                fm = {}
+            if isinstance(fm, dict):
+                try:
+                    fm["session_count"] = int(fm.get("session_count") or 0) + 1
+                except (TypeError, ValueError):
+                    fm["session_count"] = 1
+                fm["last_updated"] = today_iso()
+                new_fm = yaml.safe_dump(fm, sort_keys=False, allow_unicode=True).strip()
+                text = f"---\n{new_fm}\n---{parts[2]}"
+
+    log_line = f"- {today_iso()} · {line.strip()}"
+    if "## Session Log" in text:
+        # Append after the heading (and any existing log lines)
+        idx = text.index("## Session Log")
+        after = text[idx + len("## Session Log") :]
+        # Find next ## heading or end
+        next_h = re.search(r"\n## ", after)
+        if next_h:
+            insert_at = idx + len("## Session Log") + next_h.start()
+            text = text[:insert_at].rstrip() + "\n" + log_line + "\n" + text[insert_at:]
+        else:
+            text = text.rstrip() + "\n" + log_line + "\n"
+    else:
+        text = text.rstrip() + "\n\n## Session Log\n\n" + log_line + "\n"
+    mem_path.write_text(text, encoding="utf-8")
+    return mem_path
+
+
+def append_kai_ledger_entry(eng: Engagement, entry: Dict[str, Any]) -> Path:
+    """Append one KaiLedgerEntry JSON line. Rejects sensitive payload fields."""
+    for key in ("summary", "notes", "detail", "transcript"):
+        val = entry.get(key)
+        if isinstance(val, str):
+            hits = text_has_sensitive_content(val)
+            if hits:
+                raise EngagementError(
+                    f"Ledger field {key!r} rejected — sensitive pattern ({', '.join(hits)})"
+                )
+    # Never allow raw vault dumps
+    forbidden = {"memory", "agent_prompt", "vault", "secret", "password", "token"}
+    for key in entry:
+        if key.lower() in forbidden:
+            raise EngagementError(f"Ledger must not include field {key!r}")
+
+    path = eng.ledger_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    payload = dict(entry)
+    payload.setdefault("apiVersion", "kaizen.agentic/v1")
+    payload.setdefault("kind", "KaiLedgerEntry")
+    payload.setdefault("engagement_id", eng.engagement_id)
+    payload.setdefault("currency", "KAI")
+    payload.setdefault("created_at", utc_now_iso())
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(payload, sort_keys=True))
+        handle.write("\n")
+    return path
+
+
+def write_session_report(
+    eng: Engagement,
+    *,
+    duty: str,
+    summary: str,
+    success: bool,
+    report_name: Optional[str] = None,
+) -> Path:
+    """Write a short client-visible duty report under reports/."""
+    hits = text_has_sensitive_content(summary)
+    if hits:
+        raise EngagementError(
+            f"Report summary rejected — sensitive pattern ({', '.join(hits)})"
+        )
+    reports = eng.path / "reports"
+    reports.mkdir(parents=True, exist_ok=True)
+    name = report_name or f"{today_iso()}-{duty.replace('_', '-')}.md"
+    path = reports / name
+    target_ids = ", ".join(
+        str(t.get("id")) for t in eng.targets if isinstance(t, dict) and t.get("id")
+    )
+    body = (
+        f"# Session report — {eng.engagement_id}\n\n"
+        f"- **Date:** {today_iso()}\n"
+        f"- **Duty:** {duty}\n"
+        f"- **Targets:** {target_ids or 'n/a'}\n"
+        f"- **Outcome:** {'success' if success else 'failure'}\n"
+        f"- **Phase:** {eng.phase}\n\n"
+        f"## Summary\n\n{summary.strip()}\n\n"
+        f"_Billing metadata only in commercial/ledger.jsonl; "
+        f"no secrets in this report._\n"
+    )
+    path.write_text(body, encoding="utf-8")
+    return path
+
+
+def close_session(
+    eng: Engagement,
+    *,
+    success: bool,
+    duty: str = "standard_review",
+    summary: str,
+    execution_time_s: Optional[float] = None,
+    quality: Optional[float] = None,
+    access_class: Optional[str] = "host_observe",
+    amount_kai: Optional[int] = None,
+    record_metrics: bool = True,
+    record_ledger: bool = True,
+    write_report: bool = True,
+    agent_name: Optional[str] = None,
+    idempotency_key: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Session-close: vault log + optional metrics + Kai duty charge + report.
+
+    Metrics are stored under the engagement path (``.kaizen/metrics/<agent>/``)
+    so they stay with the client vault tree. Ledger lines never include secrets.
+    """
+    from .metrics import MetricsStore  # local import avoids circular weight
+
+    agent = agent_name or eng.role_id or "host-operator"
+    tier = capability_tier(eng)
+    result: Dict[str, Any] = {
+        "engagement_id": eng.engagement_id,
+        "duty": duty,
+        "success": success,
+        "agent": agent,
+        "phase": eng.phase,
+    }
+
+    # 1) Vault session log
+    target_hint = ""
+    if eng.targets:
+        tid = eng.targets[0].get("id") if isinstance(eng.targets[0], dict) else None
+        if tid:
+            target_hint = f"{tid} · "
+    outcome = "ok" if success else "fail"
+    log_line = f"{target_hint}{duty} · {summary.strip()} · {outcome}"
+    mem_path = append_session_log(eng, log_line)
+    result["memory_path"] = str(mem_path)
+
+    # 2) Report
+    report_path = None
+    if write_report:
+        report_path = write_session_report(
+            eng, duty=duty, summary=summary, success=success
+        )
+        result["report_path"] = str(report_path)
+
+    # 3) Metrics (engagement-scoped)
+    if record_metrics:
+        store = MetricsStore(eng.path, agent)
+        payload: Dict[str, Any] = {
+            "success": success,
+            "engagement_id": eng.engagement_id,
+            "duty": duty,
+            "phase": eng.phase,
+        }
+        if execution_time_s is not None:
+            payload["execution_time_s"] = float(execution_time_s)
+        if quality is not None:
+            payload["quality_score"] = float(quality)
+        recorded = store.append(payload, idempotency_key=idempotency_key)
+        result["metrics_recorded"] = recorded
+        result["metrics_path"] = str(store.executions_path)
+    else:
+        result["metrics_recorded"] = False
+
+    # 4) Kai ledger
+    if record_ledger:
+        kai = compute_duty_kai(
+            duty,
+            tier=tier,
+            access_class=access_class,
+            amount_override=amount_kai,
+        )
+        entry_id = f"kai-{utc_now_iso().replace(':', '').replace('-', '')}-{duty}"
+        target_id = None
+        if eng.targets and isinstance(eng.targets[0], dict):
+            target_id = eng.targets[0].get("id")
+        ledger_entry: Dict[str, Any] = {
+            "id": entry_id,
+            "type": "duty_charge",
+            "product": duty,
+            "capability_tier": tier,
+            "amount_kai": kai,
+            "session_ref": (
+                str(Path(result["report_path"]).relative_to(eng.path))
+                if result.get("report_path")
+                else None
+            ),
+            "metadata": {
+                "target": target_id,
+                "phase": eng.phase,
+                "success": success,
+                "access_class": access_class,
+            },
+        }
+        if access_class:
+            ledger_entry["access_surcharge_product"] = access_class
+        # Billing account if present
+        client = (eng.data.get("spec") or {}).get("client") or {}
+        if client.get("billing_account"):
+            ledger_entry["account"] = client["billing_account"]
+        ledger_path = append_kai_ledger_entry(eng, ledger_entry)
+        result["ledger_path"] = str(ledger_path)
+        result["amount_kai"] = kai
+        result["ledger_entry_id"] = entry_id
+    else:
+        result["amount_kai"] = 0
+
+    return result