diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a3dd15..1b41c71 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,10 +12,6 @@ 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]'`) diff --git a/docs/CLI_CHEAT_SHEET.md b/docs/CLI_CHEAT_SHEET.md index b413e11..a9c6fd9 100644 --- a/docs/CLI_CHEAT_SHEET.md +++ b/docs/CLI_CHEAT_SHEET.md @@ -125,46 +125,6 @@ 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 diff --git a/engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/executions.jsonl b/engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/executions.jsonl deleted file mode 100644 index d7e011a..0000000 --- a/engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/executions.jsonl +++ /dev/null @@ -1 +0,0 @@ -{"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"} diff --git a/engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/summary.json b/engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/summary.json deleted file mode 100644 index 2443a7f..0000000 --- a/engagements/pilots/eng-coulomb-railiance01-ho-001/.kaizen/metrics/host-operator/summary.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "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" - } -} diff --git a/engagements/pilots/eng-coulomb-railiance01-ho-001/commercial/ledger.jsonl b/engagements/pilots/eng-coulomb-railiance01-ho-001/commercial/ledger.jsonl index 251d99e..a546024 100644 --- a/engagements/pilots/eng-coulomb-railiance01-ho-001/commercial/ledger.jsonl +++ b/engagements/pilots/eng-coulomb-railiance01-ho-001/commercial/ledger.jsonl @@ -1,2 +1 @@ {"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"} diff --git a/engagements/pilots/eng-coulomb-railiance01-ho-001/reports/2026-07-16-standard-review.md b/engagements/pilots/eng-coulomb-railiance01-ho-001/reports/2026-07-16-standard-review.md deleted file mode 100644 index 9ff9d2a..0000000 --- a/engagements/pilots/eng-coulomb-railiance01-ho-001/reports/2026-07-16-standard-review.md +++ /dev/null @@ -1,13 +0,0 @@ -# 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._ diff --git a/engagements/pilots/eng-coulomb-railiance01-ho-001/vault/memory.md b/engagements/pilots/eng-coulomb-railiance01-ho-001/vault/memory.md index 2ade042..bd438fb 100644 --- a/engagements/pilots/eng-coulomb-railiance01-ho-001/vault/memory.md +++ b/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: 1 +last_updated: "2026-07-16" +session_count: 0 confidentiality: client_owned --- @@ -72,4 +72,3 @@ _None yet._ ## Session Log -- 2026-07-16 · railiance01 · standard_review · T09 wire-up smoke: prepare+close-session path verified (no host access) · ok diff --git a/src/kaizen_agentic/cli.py b/src/kaizen_agentic/cli.py index 8c54f06..e191f83 100644 --- a/src/kaizen_agentic/cli.py +++ b/src/kaizen_agentic/cli.py @@ -24,27 +24,6 @@ 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, @@ -1433,485 +1412,6 @@ def protocols_show(agent_name: str, slug: str): click.echo(protocol_path.read_text()) -@cli.group() -def engagement(): - """Forward-deployed engagement lifecycle (KAIZEN-WP-0009 / DEC-FDA-001). - - Manage file-based engagements under engagements/pilots//: phase, - checklists, prepare bundles, staff from Role packages. Offline; no LLM invoke. - """ - pass - - -def _resolve_eng(engagement_ref: str, repo_root: Optional[str]): - root = Path(repo_root).resolve() if repo_root else find_repo_root() - try: - path = resolve_engagement_dir(engagement_ref, root) - return load_engagement(path), root - except EngagementError as exc: - click.echo(f"Error: {exc}", err=True) - sys.exit(1) - - -@engagement.command("list") -@click.option( - "--repo-root", - default=None, - help="Repo root containing engagements/ (default: auto-detect)", -) -@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") -def engagement_list(repo_root: Optional[str], as_json: bool): - """List engagement directories under engagements/pilots/.""" - root = Path(repo_root).resolve() if repo_root else find_repo_root() - dirs = list_engagement_dirs(root) - rows = [] - for d in dirs: - try: - eng = load_engagement(d) - rows.append( - { - "id": eng.engagement_id, - "phase": eng.phase, - "role": eng.role_id, - "path": str(d), - } - ) - except EngagementError as exc: - rows.append( - { - "id": d.name, - "phase": "?", - "role": None, - "path": str(d), - "error": str(exc), - } - ) - if as_json: - click.echo(json.dumps(rows, indent=2)) - return - if not rows: - click.echo(f"No engagements under {root / 'engagements'}") - return - for r in rows: - role = r.get("role") or "-" - click.echo(f" {r['id']}: phase={r['phase']} role={role}") - click.echo(f" {r['path']}") - - -@engagement.command("show") -@click.argument("engagement_ref") -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") -def engagement_show(engagement_ref: str, repo_root: Optional[str], as_json: bool): - """Show engagement summary from ENGAGEMENT.yaml.""" - eng, _root = _resolve_eng(engagement_ref, repo_root) - payload = { - "id": eng.engagement_id, - "phase": eng.phase, - "role": eng.role_id, - "targets": eng.targets, - "path": str(eng.path), - "validation_errors": validate_engagement(eng), - } - if as_json: - click.echo(json.dumps(payload, indent=2)) - return - click.echo(f"Engagement: {payload['id']}") - click.echo(f" Phase: {payload['phase']}") - click.echo(f" Role: {payload['role']}") - click.echo(f" Path: {payload['path']}") - for t in payload["targets"]: - click.echo(f" Target: {t.get('kind')} {t.get('id')}") - errs = payload["validation_errors"] - if errs: - click.echo(" Validation:") - for e in errs: - click.echo(f" - {e}") - else: - click.echo(" Validation: ok") - - -@engagement.command("validate") -@click.argument("engagement_ref") -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -def engagement_validate(engagement_ref: str, repo_root: Optional[str]): - """Validate engagement tree structure and required files.""" - eng, _root = _resolve_eng(engagement_ref, repo_root) - errs = validate_engagement(eng) - if errs: - click.echo(f"❌ {eng.engagement_id}: {len(errs)} issue(s)") - for e in errs: - click.echo(f" - {e}") - sys.exit(1) - click.echo(f"✅ {eng.engagement_id}: valid (phase={eng.phase})") - - -@engagement.command("checklist") -@click.argument("engagement_ref") -@click.option( - "--which", - type=click.Choice(["ramp_up", "ramp_down", "both"]), - default="both", - show_default=True, -) -@click.option( - "--mark", - metavar="ID=STATUS", - help="Update one item, e.g. RU-01=done (implies --which for that prefix)", -) -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") -def engagement_checklist( - engagement_ref: str, - which: str, - mark: Optional[str], - repo_root: Optional[str], - as_json: bool, -): - """Show or update ramp-up / ramp-down checklist status.""" - eng, _root = _resolve_eng(engagement_ref, repo_root) - - if mark: - if "=" not in mark: - click.echo("Error: --mark requires ID=STATUS (e.g. RU-01=done)", err=True) - sys.exit(1) - item_id, status = mark.split("=", 1) - item_id, status = item_id.strip(), status.strip() - which_mark = "ramp_up" if item_id.upper().startswith("RU") else "ramp_down" - path, _items = load_checklist(eng, which_mark) - if path is None: - click.echo(f"Error: no {which_mark} checklist", err=True) - sys.exit(1) - if not set_checklist_item_status(path, item_id, status): - click.echo(f"Error: item {item_id} not found in {path}", err=True) - sys.exit(1) - click.echo(f"Updated {item_id} → {status} in {path}") - - which_list = ["ramp_up", "ramp_down"] if which == "both" else [which] - result: dict = {"engagement_id": eng.engagement_id, "checklists": {}} - for w in which_list: - path, items = load_checklist(eng, w) - summary = checklist_summary(items) - result["checklists"][w] = { - "path": str(path) if path else None, - "summary": summary, - "items": [ - { - "id": i.item_id, - "criterion": i.criterion, - "status": i.status, - "evidence": i.evidence, - "done": i.done, - } - for i in items - ], - } - - if as_json: - click.echo(json.dumps(result, indent=2)) - return - - for w, block in result["checklists"].items(): - summary = block["summary"] - click.echo( - f"{w}: {summary['done']}/{summary['total']}" - + (" ✅ complete" if summary["complete"] else "") - ) - if not block["path"]: - click.echo(" (checklist file missing)") - continue - for it in block["items"]: - flag = "✅" if it["done"] else "☐ " - click.echo(f" {flag} {it['id']}: {it['criterion']} [{it['status']}]") - - -@engagement.command("phase") -@click.argument("engagement_ref") -@click.option( - "--to", - "to_phase", - required=True, - help=f"Target phase ({', '.join(VALID_PHASES)})", -) -@click.option("--force", is_flag=True, help="Skip transition graph checks") -@click.option("--notes", default=None, help="Optional status.notes update") -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -def engagement_phase( - engagement_ref: str, - to_phase: str, - force: bool, - notes: Optional[str], - repo_root: Optional[str], -): - """Transition engagement phase (updates ENGAGEMENT.yaml + agent frontmatter).""" - eng, _root = _resolve_eng(engagement_ref, repo_root) - prev = eng.phase - try: - eng = set_phase(eng, to_phase, force=force, notes=notes) - except EngagementError as exc: - click.echo(f"Error: {exc}", err=True) - sys.exit(1) - click.echo(f"{eng.engagement_id}: {prev} → {eng.phase}") - - -@engagement.command("prepare") -@click.argument("engagement_ref") -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -@click.option( - "--format", - "output_format", - type=click.Choice(["markdown", "json"]), - default="markdown", - show_default=True, -) -def engagement_prepare( - engagement_ref: str, repo_root: Optional[str], output_format: str -): - """Assemble session orientation bundle (agent + vault + protocols + access).""" - eng, root = _resolve_eng(engagement_ref, repo_root) - bundle = build_prepare_bundle(eng, root) - if output_format == "json": - # Drop large protocol bodies in json unless needed — keep presence flags - slim = {k: v for k, v in bundle.items() if k != "protocol_bodies"} - # Include protocol body sizes only - slim["protocol_chars"] = { - k: len(v) for k, v in (bundle.get("protocol_bodies") or {}).items() - } - click.echo(json.dumps(slim, indent=2)) - return - click.echo(render_prepare_markdown(bundle)) - - -@engagement.command("quote") -@click.argument("engagement_ref") -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") -def engagement_quote(engagement_ref: str, repo_root: Optional[str], as_json: bool): - """Show Kai quote snapshot for an engagement.""" - eng, _root = _resolve_eng(engagement_ref, repo_root) - try: - quote = load_quote(eng) - except EngagementError as exc: - click.echo(f"Error: {exc}", err=True) - sys.exit(1) - if quote is None: - click.echo(f"No quote file for {eng.engagement_id}") - sys.exit(1) - if as_json: - click.echo(json.dumps(quote, indent=2)) - return - spec = quote.get("spec") or {} - total = spec.get("total_kai") - click.echo(f"Quote: {eng.engagement_id}") - if total is not None: - click.echo(f" Total: {total} Kai") - for item in spec.get("line_items") or []: - if isinstance(item, dict): - click.echo( - f" - {item.get('product')}: {item.get('amount_kai')} Kai" - f" ({item.get('description', '')})" - ) - for note in spec.get("notes") or []: - click.echo(f" note: {note}") - - -@engagement.command("staff") -@click.option( - "--id", "engagement_id", required=True, help="Engagement id (directory name)" -) -@click.option("--role", "role_id", required=True, help="Role package id under roles/") -@click.option("--client", "client_id", required=True, help="Client id") -@click.option( - "--target", "target_id", required=True, help="Primary target id (e.g. host)" -) -@click.option( - "--target-kind", - default="host", - show_default=True, - help="Target kind", -) -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -@click.option("--force", is_flag=True, help="Overwrite scaffold if directory exists") -def engagement_staff( - engagement_id: str, - role_id: str, - client_id: str, - target_id: str, - target_kind: str, - repo_root: Optional[str], - force: bool, -): - """Scaffold a pilot engagement from a Role package.""" - root = Path(repo_root).resolve() if repo_root else find_repo_root() - try: - eng = staff_engagement( - engagement_id=engagement_id, - role_id=role_id, - client_id=client_id, - target_id=target_id, - target_kind=target_kind, - repo_root=root, - force=force, - ) - except EngagementError as exc: - click.echo(f"Error: {exc}", err=True) - sys.exit(1) - click.echo(f"Staffed engagement: {eng.engagement_id}") - click.echo(f" Phase: {eng.phase}") - click.echo(f" Path: {eng.path}") - click.echo(f" Next: kaizen-agentic engagement validate {eng.engagement_id}") - click.echo( - f" kaizen-agentic engagement phase {eng.engagement_id} --to ramp_up" - ) - - -@engagement.command("scrub") -@click.argument("engagement_ref") -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON") -def engagement_scrub(engagement_ref: str, repo_root: Optional[str], as_json: bool): - """Heuristic scan vault/reports for secrets before lesson contribution.""" - eng, _root = _resolve_eng(engagement_ref, repo_root) - report = scrub_engagement(eng) - if as_json: - click.echo(json.dumps(report, indent=2)) - return - click.echo( - f"Scrub {report['engagement_id']}: " - f"{report['files_scanned']} files, {len(report['hits'])} hit(s)" - ) - for h in report["hits"]: - click.echo(f" {h['file']}:{h['line']} [{h['pattern']}] {h['snippet']}") - if report["clean"]: - click.echo(" ✅ no heuristic hits (still review before contribute_lesson)") - else: - click.echo(" ⚠ review hits before any Role craft contribution") - sys.exit(1) - - -@engagement.command("export-handoff") -@click.argument("engagement_ref") -@click.option("--repo-root", default=None, help="Repo root (default: auto-detect)") -def engagement_export_handoff(engagement_ref: str, repo_root: Optional[str]): - """Ensure vault/handoff pack exists and copy baselines into it.""" - eng, _root = _resolve_eng(engagement_ref, repo_root) - handoff = export_handoff(eng) - click.echo(f"Handoff pack: {handoff}") - - -@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). diff --git a/src/kaizen_agentic/engagement.py b/src/kaizen_agentic/engagement.py deleted file mode 100644 index 79a23ef..0000000 --- a/src/kaizen_agentic/engagement.py +++ /dev/null @@ -1,1180 +0,0 @@ -"""Forward-deployed engagement lifecycle helpers (KAIZEN-WP-0009 / DEC-FDA-001). - -File-based Phase 1: ENGAGEMENT.yaml under engagements/pilots// is the source -of truth. This module loads, validates, prepares session bundles, and transitions -phases — it does not invoke LLMs or touch production hosts. -""" - -from __future__ import annotations - -import json -import re -import shutil -from dataclasses import dataclass -from datetime import date, datetime, timezone -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+")), -] - -# 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.""" - - -@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 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() - for candidate in [cur, *cur.parents]: - if (candidate / "engagements").is_dir() or (candidate / "roles").is_dir(): - return candidate - if (candidate / ".git").exists() and ( - (candidate / "pyproject.toml").exists() - or (candidate / "workplans").is_dir() - ): - return candidate - return cur - - -def list_engagement_dirs(repo_root: Path) -> List[Path]: - """Discover engagement directories (pilots first, then other trees).""" - found: List[Path] = [] - pilots = repo_root / DEFAULT_PILOTS_DIR - if pilots.is_dir(): - for child in sorted(pilots.iterdir()): - if child.is_dir() and (child / ENGAGEMENT_FILENAME).exists(): - found.append(child) - engagements = repo_root / "engagements" - if engagements.is_dir(): - for child in sorted(engagements.iterdir()): - if child.name == "pilots": - continue - if child.is_dir() and (child / ENGAGEMENT_FILENAME).exists(): - found.append(child) - return found - - -def resolve_engagement_dir( - engagement_ref: str, repo_root: Optional[Path] = None -) -> Path: - """Resolve an engagement id or path to its directory.""" - root = (repo_root or find_repo_root()).resolve() - ref = Path(engagement_ref) - - if ref.is_dir() and (ref / ENGAGEMENT_FILENAME).exists(): - return ref.resolve() - if ref.is_file() and ref.name == ENGAGEMENT_FILENAME: - return ref.parent.resolve() - - # id under pilots/ - pilot = root / DEFAULT_PILOTS_DIR / engagement_ref - if (pilot / ENGAGEMENT_FILENAME).exists(): - return pilot.resolve() - - # bare engagements/ - bare = root / "engagements" / engagement_ref - if (bare / ENGAGEMENT_FILENAME).exists(): - return bare.resolve() - - # search by metadata id - for d in list_engagement_dirs(root): - try: - eng = load_engagement(d) - except EngagementError: - continue - if eng.engagement_id == engagement_ref or d.name == engagement_ref: - return d - - raise EngagementError( - f"Engagement not found: {engagement_ref!r} " - f"(looked under {root / DEFAULT_PILOTS_DIR})" - ) - - -def load_engagement(path: Path) -> Engagement: - """Load ENGAGEMENT.yaml from a directory (or path to the file).""" - path = Path(path) - if path.is_file(): - source = path - base = path.parent - else: - source = path / ENGAGEMENT_FILENAME - base = path - if not source.exists(): - raise EngagementError(f"Missing {ENGAGEMENT_FILENAME}: {source}") - try: - data = yaml.safe_load(source.read_text(encoding="utf-8")) - except yaml.YAMLError as exc: - raise EngagementError(f"Invalid YAML in {source}: {exc}") from exc - if not isinstance(data, dict): - raise EngagementError(f"{source} must contain a mapping") - if data.get("kind") and data.get("kind") != "Engagement": - raise EngagementError( - f"{source}: kind must be Engagement (got {data.get('kind')})" - ) - return Engagement(path=base.resolve(), data=data, source_path=source.resolve()) - - -def save_engagement(eng: Engagement) -> None: - """Write ENGAGEMENT.yaml back (preserves structure via dump of data).""" - eng.source_path.write_text( - yaml.safe_dump(eng.data, sort_keys=False, allow_unicode=True), - encoding="utf-8", - ) - - -def parse_checklist_markdown(text: str) -> List[ChecklistItem]: - """Parse checklist tables with columns ID | Criterion | Status | Evidence.""" - items: List[ChecklistItem] = [] - for line in text.splitlines(): - line = line.strip() - if not line.startswith("|"): - continue - cells = [c.strip() for c in line.strip("|").split("|")] - if len(cells) < 4: - continue - item_id, criterion, status, evidence = cells[0], cells[1], cells[2], cells[3] - if item_id.lower() in ("id", "---") or set(item_id) <= {"-", ":"}: - continue - if not re.match(r"^[A-Z]{1,3}-\d+", item_id): - continue - items.append( - ChecklistItem( - item_id=item_id, - criterion=criterion, - status=status, - evidence=evidence, - ) - ) - return items - - -def load_checklist( - eng: Engagement, which: str -) -> Tuple[Optional[Path], List[ChecklistItem]]: - path = eng.checklist_path(which) - if path is None: - return None, [] - items = parse_checklist_markdown(path.read_text(encoding="utf-8")) - return path, items - - -def checklist_summary(items: Sequence[ChecklistItem]) -> Dict[str, Any]: - done = sum(1 for i in items if i.done) - total = len(items) - return { - "done": done, - "total": total, - "complete": total > 0 and done == total, - "pending": [i.item_id for i in items if not i.done], - } - - -def set_checklist_item_status(checklist_path: Path, item_id: str, status: str) -> bool: - """Update a checklist row status by item id. Returns True if updated.""" - text = checklist_path.read_text(encoding="utf-8") - lines = text.splitlines() - changed = False - out: List[str] = [] - for line in lines: - stripped = line.strip() - if stripped.startswith("|"): - cells = [c.strip() for c in stripped.strip("|").split("|")] - if len(cells) >= 4 and cells[0] == item_id: - cells[2] = status - line = "| " + " | ".join(cells) + " |" - changed = True - out.append(line) - if changed: - checklist_path.write_text( - "\n".join(out) + ("\n" if text.endswith("\n") else ""), encoding="utf-8" - ) - return changed - - -def set_phase( - eng: Engagement, - new_phase: str, - *, - force: bool = False, - notes: Optional[str] = None, -) -> Engagement: - """Transition engagement phase; updates YAML on disk.""" - new_phase = new_phase.strip() - if new_phase not in VALID_PHASES: - raise EngagementError( - f"Invalid phase {new_phase!r}; expected one of: {', '.join(VALID_PHASES)}" - ) - current = eng.phase - if not force and new_phase != current: - allowed = PHASE_TRANSITIONS.get(current, ()) - if new_phase not in allowed: - raise EngagementError( - f"Illegal phase transition {current!r} → {new_phase!r}. " - f"Allowed: {', '.join(allowed) or '(terminal)'}. Use --force to override." - ) - status = eng.data.setdefault("status", {}) - if not isinstance(status, dict): - eng.data["status"] = {} - status = eng.data["status"] - status["phase"] = new_phase - if notes is not None: - status["notes"] = notes - meta = eng.data.setdefault("metadata", {}) - if isinstance(meta, dict): - meta["updated"] = today_iso() - save_engagement(eng) - # Keep agent definition phase in sync if present - _sync_agent_definition_phase(eng, new_phase) - return load_engagement(eng.path) - - -def _sync_agent_definition_phase(eng: Engagement, phase: str) -> None: - ad_path = eng.agent_definition_path() - if ad_path is None or not ad_path.exists(): - return - text = ad_path.read_text(encoding="utf-8") - if not text.startswith("---"): - return - parts = text.split("---", 2) - if len(parts) < 3: - return - try: - fm = yaml.safe_load(parts[1]) or {} - except yaml.YAMLError: - return - if not isinstance(fm, dict): - return - fm["phase"] = phase - new_fm = yaml.safe_dump(fm, sort_keys=False, allow_unicode=True).strip() - ad_path.write_text(f"---\n{new_fm}\n---{parts[2]}", encoding="utf-8") - - -def validate_engagement(eng: Engagement) -> List[str]: - """Return list of validation errors (empty if ok).""" - errors: List[str] = [] - if eng.data.get("apiVersion") != "kaizen.agentic/v1": - errors.append("apiVersion should be kaizen.agentic/v1") - if eng.phase not in VALID_PHASES: - errors.append(f"unknown phase: {eng.phase}") - if not eng.role_id: - errors.append("spec.role.id missing") - if not eng.targets: - errors.append("spec.targets is empty") - ad = eng.agent_definition_path() - if ad is None: - errors.append("spec.agent_definition.path missing") - elif not ad.exists(): - errors.append(f"agent definition missing: {ad}") - mem = eng.vault_memory_path() - if not mem.exists(): - errors.append(f"vault memory missing: {mem}") - for which in ("ramp_up", "ramp_down"): - path, items = load_checklist(eng, which) - if path is None: - errors.append(f"checklist missing: {which}") - elif not items: - errors.append(f"checklist empty or unparseable: {path}") - access = eng.path / "access-plan.md" - if not access.exists(): - errors.append("access-plan.md missing") - return errors - - -def build_prepare_bundle( - eng: Engagement, repo_root: Optional[Path] = None -) -> Dict[str, Any]: - """Assemble an offline session orientation bundle for this engagement.""" - root = (repo_root or find_repo_root(eng.path)).resolve() - ad_path = eng.agent_definition_path() - agent_prompt = ( - ad_path.read_text(encoding="utf-8") if ad_path and ad_path.exists() else None - ) - mem_path = eng.vault_memory_path() - memory = mem_path.read_text(encoding="utf-8") if mem_path.exists() else None - access_path = eng.path / "access-plan.md" - access_plan = ( - access_path.read_text(encoding="utf-8") if access_path.exists() else None - ) - - phase = eng.phase - phase_instructions = None - role_dir = root / DEFAULT_ROLES_DIR / (eng.role_id or "") - if phase == "ramp_up" and (role_dir / "ramp-up.md").exists(): - phase_instructions = (role_dir / "ramp-up.md").read_text(encoding="utf-8") - elif phase == "ramp_down" and (role_dir / "ramp-down.md").exists(): - phase_instructions = (role_dir / "ramp-down.md").read_text(encoding="utf-8") - - protocols: Dict[str, Optional[str]] = {} - role_yaml = role_dir / "ROLE.yaml" - if role_yaml.exists(): - try: - role_data = yaml.safe_load(role_yaml.read_text(encoding="utf-8")) or {} - except yaml.YAMLError: - role_data = {} - for p in (role_data.get("spec") or {}).get("protocols") or []: - if not isinstance(p, dict): - continue - slug = p.get("slug") or p.get("path") - rel = p.get("path") - content = None - if rel: - full = root / rel - if full.exists(): - content = full.read_text(encoding="utf-8") - protocols[str(slug)] = content - - _, ramp_up_items = load_checklist(eng, "ramp_up") - _, ramp_down_items = load_checklist(eng, "ramp_down") - - return { - "engagement_id": eng.engagement_id, - "phase": phase, - "role_id": eng.role_id, - "targets": eng.targets, - "generated": today_iso(), - "path": str(eng.path), - "agent_prompt": agent_prompt, - "agent_prompt_found": agent_prompt is not None, - "memory": memory, - "access_plan": access_plan, - "phase_instructions": phase_instructions, - "protocols": {k: (v is not None) for k, v in protocols.items()}, - "protocol_bodies": {k: v for k, v in protocols.items() if v}, - "checklist_ramp_up": checklist_summary(ramp_up_items), - "checklist_ramp_down": checklist_summary(ramp_down_items), - "validation_errors": validate_engagement(eng), - "session_close": [ - ( - f"kaizen-agentic engagement close-session {eng.engagement_id} " - "--success --duty standard_review --summary '' " - "--time --quality <0-1>" - ), - "Or manually: update vault memory, reports/, commercial/ledger.jsonl", - f"kaizen-agentic engagement checklist {eng.engagement_id}", - ], - "confidentiality": "client_owned", - } - - -def render_prepare_markdown(bundle: Dict[str, Any]) -> str: - lines = [ - f"# Engagement Prepare: {bundle['engagement_id']}", - f"Phase: {bundle['phase']}", - f"Role: {bundle.get('role_id')}", - f"Generated: {bundle['generated']}", - f"Path: {bundle['path']}", - "", - "## Targets", - ] - for t in bundle.get("targets") or []: - lines.append(f"- {t.get('kind', 'target')}: `{t.get('id')}`") - lines.append("") - - errs = bundle.get("validation_errors") or [] - if errs: - lines.append("## Validation warnings") - for e in errs: - lines.append(f"- {e}") - lines.append("") - - ru = bundle.get("checklist_ramp_up") or {} - rd = bundle.get("checklist_ramp_down") or {} - lines.append("## Checklists") - lines.append( - f"- ramp_up: {ru.get('done', 0)}/{ru.get('total', 0)}" - + (f" pending={ru.get('pending')}" if ru.get("pending") else "") - ) - lines.append( - f"- ramp_down: {rd.get('done', 0)}/{rd.get('total', 0)}" - + (f" pending={rd.get('pending')}" if rd.get("pending") else "") - ) - lines.append("") - - lines.append("## Access Plan") - lines.append(bundle.get("access_plan") or "(missing access-plan.md)") - lines.append("") - - if bundle.get("phase_instructions"): - lines.append("## Phase Instructions") - lines.append(bundle["phase_instructions"]) - lines.append("") - - lines.append("## Agent Definition") - if bundle.get("agent_prompt_found"): - lines.append(bundle["agent_prompt"]) - else: - lines.append("(agent definition not found)") - lines.append("") - - lines.append("## Vault Memory") - lines.append(bundle.get("memory") or "(no memory yet)") - lines.append("") - - bodies = bundle.get("protocol_bodies") or {} - if bodies: - lines.append("## Protocols") - for slug, body in bodies.items(): - lines.append(f"### {slug}") - lines.append(body) - lines.append("") - else: - present = bundle.get("protocols") or {} - if present: - lines.append("## Protocols (paths only)") - for slug, ok in present.items(): - lines.append(f"- {slug}: {'loaded' if ok else 'missing'}") - lines.append("") - - lines.append("## Session Close") - for cmd in bundle.get("session_close") or []: - lines.append(f"- {cmd}") - lines.append("") - lines.append( - "_Confidentiality: operational knowledge is client-owned; " - "do not copy vault content into supplier public agents._" - ) - return "\n".join(lines) - - -def scrub_engagement(eng: Engagement) -> Dict[str, Any]: - """Heuristic scan of vault + reports for sensitive patterns.""" - hits: List[Dict[str, Any]] = [] - scan_roots = [ - eng.path / "vault", - eng.path / "reports", - eng.path / "access-plan.md", - ] - files: List[Path] = [] - for root in scan_roots: - if root.is_file(): - files.append(root) - elif root.is_dir(): - files.extend( - p - for p in root.rglob("*") - if p.is_file() - and p.suffix in {".md", ".txt", ".yml", ".yaml", ".json", ".jsonl"} - ) - - for fpath in files: - try: - text = fpath.read_text(encoding="utf-8", errors="replace") - except OSError: - continue - for label, pattern in _SCRUB_PATTERNS: - for match in pattern.finditer(text): - line_no = text[: match.start()].count("\n") + 1 - hits.append( - { - "file": str(fpath.relative_to(eng.path)), - "line": line_no, - "pattern": label, - "snippet": match.group(0)[:40] - + ("…" if len(match.group(0)) > 40 else ""), - } - ) - return { - "engagement_id": eng.engagement_id, - "files_scanned": len(files), - "hits": hits, - "clean": len(hits) == 0, - "note": "Heuristic only — human review required before contribute_lesson", - } - - -def export_handoff(eng: Engagement) -> Path: - """Ensure handoff pack exists; return path to handoff directory.""" - handoff = eng.path / "vault" / "handoff" - handoff.mkdir(parents=True, exist_ok=True) - readme = handoff / "README.md" - if not readme.exists(): - readme.write_text( - f"# Handoff — {eng.engagement_id}\n\nGenerated: {today_iso()}\n", - encoding="utf-8", - ) - risks = handoff / "risks.md" - if not risks.exists(): - risks.write_text( - "# Outstanding risks\n\n| Risk | Severity | Status |\n|------|----------|--------|\n", - encoding="utf-8", - ) - # Copy baseline files into handoff/baselines if present - baselines = eng.path / "vault" / "baselines" - if baselines.is_dir(): - dest = handoff / "baselines" - dest.mkdir(exist_ok=True) - for f in baselines.iterdir(): - if f.is_file(): - shutil.copy2(f, dest / f.name) - return handoff - - -def staff_engagement( - *, - engagement_id: str, - role_id: str, - client_id: str, - target_id: str, - target_kind: str = "host", - repo_root: Optional[Path] = None, - force: bool = False, -) -> Engagement: - """Scaffold a new pilot engagement directory from a Role package.""" - root = (repo_root or find_repo_root()).resolve() - role_dir = root / DEFAULT_ROLES_DIR / role_id - if not role_dir.is_dir(): - raise EngagementError(f"Role package not found: {role_dir}") - - dest = root / DEFAULT_PILOTS_DIR / engagement_id - if dest.exists() and not force: - if (dest / ENGAGEMENT_FILENAME).exists(): - return load_engagement(dest) - raise EngagementError(f"Directory exists without engagement: {dest}") - - dest.mkdir(parents=True, exist_ok=True) - (dest / "vault" / "baselines").mkdir(parents=True, exist_ok=True) - (dest / "vault" / "session-log").mkdir(parents=True, exist_ok=True) - (dest / "vault" / "findings").mkdir(parents=True, exist_ok=True) - (dest / "vault" / "handoff").mkdir(parents=True, exist_ok=True) - (dest / "checklists").mkdir(parents=True, exist_ok=True) - (dest / "commercial").mkdir(parents=True, exist_ok=True) - (dest / "reports").mkdir(parents=True, exist_ok=True) - - role_version = "0.1.0" - role_yaml = role_dir / "ROLE.yaml" - if role_yaml.exists(): - try: - rd = yaml.safe_load(role_yaml.read_text(encoding="utf-8")) or {} - role_version = str( - (rd.get("metadata") or {}).get("version") or role_version - ) - except yaml.YAMLError: - pass - - # Memory from template - mem_tmpl = role_dir / "memory-template.md" - mem_path = dest / "vault" / "memory.md" - if mem_tmpl.exists(): - mem_text = mem_tmpl.read_text(encoding="utf-8") - mem_text = mem_text.replace("", engagement_id) - mem_text = mem_text.replace( - "", f"{client_id}-{target_id}" - ) - mem_text = mem_text.replace("", today_iso()) - mem_path.write_text(mem_text, encoding="utf-8") - else: - mem_path.write_text( - f"---\nagent: {role_id}\nengagement_id: {engagement_id}\n" - f'last_updated: "{today_iso()}"\nsession_count: 0\n---\n\n# Memory\n', - encoding="utf-8", - ) - - # Agent definition from role template + binding frontmatter - ad_tmpl = role_dir / "agent-definition.md" - agent_name = f"agent-{role_id}.md" - if ad_tmpl.exists(): - body = ad_tmpl.read_text(encoding="utf-8") - # Replace or inject engagement binding frontmatter - if body.startswith("---"): - parts = body.split("---", 2) - fm = yaml.safe_load(parts[1]) if len(parts) >= 3 else {} - if not isinstance(fm, dict): - fm = {} - fm.update( - { - "engagement_id": engagement_id, - "role_id": role_id, - "role_version": role_version, - "phase": "staffing", - "memory_path": f"engagements/pilots/{engagement_id}/vault/memory.md", - "targets": [{"kind": target_kind, "id": target_id}], - "confidentiality": "client_owned", - "access_classes": ["host_observe", "privileged_ops"], - "human_approval_for": [ - "privileged_ops", - "package_upgrade", - "firewall_change", - "reboot", - ], - } - ) - new_fm = yaml.safe_dump(fm, sort_keys=False, allow_unicode=True).strip() - body = f"---\n{new_fm}\n---{parts[2] if len(parts) >= 3 else ''}" - (dest / agent_name).write_text(body, encoding="utf-8") - else: - (dest / agent_name).write_text( - f"---\nname: {role_id}\nengagement_id: {engagement_id}\nphase: staffing\n---\n\n# {role_id}\n", - encoding="utf-8", - ) - - # Checklists from role ramp docs (status tables) - for which, fname, prefix in ( - ("ramp_up", "ramp-up-status.md", "RU"), - ("ramp_down", "ramp-down-status.md", "RD"), - ): - role_check = role_dir / ("ramp-up.md" if which == "ramp_up" else "ramp-down.md") - dest_check = dest / "checklists" / fname - if role_check.exists(): - # Extract table rows from role checklist - items = parse_checklist_markdown(role_check.read_text(encoding="utf-8")) - lines = [ - f"# {which.replace('_', '-').title()} status — {engagement_id}", - "", - f"Source checklist: `roles/{role_id}/{role_check.name}`", - "", - "| ID | Criterion | Status | Evidence |", - "|----|-----------|--------|----------|", - ] - for it in items: - lines.append( - f"| {it.item_id} | {it.criterion} | todo | {it.evidence} |" - ) - if not items: - lines.append( - f"| {prefix}-01 | Complete {which} | todo | see role checklist |" - ) - dest_check.write_text("\n".join(lines) + "\n", encoding="utf-8") - else: - dest_check.write_text( - f"# {which}\n\n| ID | Criterion | Status | Evidence |\n" - f"|----|-----------|--------|----------|\n" - f"| {prefix}-01 | Complete | todo | |\n", - encoding="utf-8", - ) - - (dest / "access-plan.md").write_text( - f"# Access plan — {engagement_id}\n\n" - f"**Target:** {target_kind} `{target_id}`\n\n" - "Secrets: never stored in this tree.\n\n" - "## Verification log\n\n| Date | Result | Notes |\n|------|--------|-------|\n| _pending_ | | RU-01 |\n", - encoding="utf-8", - ) - (dest / "schedule.yml").write_text( - f"engagement_id: {engagement_id}\ntimezone: Europe/Berlin\nentries: []\n", - encoding="utf-8", - ) - (dest / "vault" / "baselines" / f"{target_id}.md").write_text( - f"# Baseline — {target_id}\n\n**Status:** pending\n", - encoding="utf-8", - ) - (dest / "commercial" / "ledger.jsonl").write_text("", encoding="utf-8") - (dest / "commercial" / "quote.yaml").write_text( - f"apiVersion: kaizen.agentic/v1\nkind: KaiQuote\nmetadata:\n" - f' engagement_id: {engagement_id}\n created: "{today_iso()}"\n' - f" currency: KAI\nspec:\n total_kai: 0\n notes: [scaffold — fill quote]\n", - encoding="utf-8", - ) - (dest / "request.yaml").write_text( - f"apiVersion: kaizen.agentic/v1\nkind: EngagementRequest\n" - f'metadata:\n id: engreq-{engagement_id}\n created: "{today_iso()}"\n' - f"spec:\n client:\n id: {client_id}\n role:\n id: {role_id}\n" - f" targets:\n - kind: {target_kind}\n id: {target_id}\n", - encoding="utf-8", - ) - - eng_doc = { - "apiVersion": "kaizen.agentic/v1", - "kind": "Engagement", - "metadata": { - "id": engagement_id, - "created": today_iso(), - "updated": today_iso(), - }, - "status": { - "phase": "staffing", - "notes": "Scaffolded by kaizen-agentic engagement staff", - }, - "spec": { - "client": {"id": client_id}, - "role": {"id": role_id, "version": role_version}, - "targets": [{"kind": target_kind, "id": target_id}], - "agent_definition": { - "path": agent_name, - "derived_from": f"roles/{role_id}/agent-definition.md", - }, - "vault": {"root": "vault/", "memory": "vault/memory.md"}, - "schedule": {"path": "schedule.yml"}, - "checklists": { - "ramp_up": "checklists/ramp-up-status.md", - "ramp_down": "checklists/ramp-down-status.md", - }, - "policy": { - "confidentiality": "client_owned", - "contribute_metrics": False, - "human_approval_for": [ - "privileged_ops", - "package_upgrade", - "firewall_change", - "reboot", - ], - }, - "commercial": { - "currency": "kai", - "ledger_ref": "commercial/ledger.jsonl", - "quote_ref": "commercial/quote.yaml", - }, - }, - } - (dest / ENGAGEMENT_FILENAME).write_text( - yaml.safe_dump(eng_doc, sort_keys=False, allow_unicode=True), - encoding="utf-8", - ) - return load_engagement(dest) - - -def load_quote(eng: Engagement) -> Optional[Dict[str, Any]]: - path = eng.quote_path() - if path is None: - return None - try: - data = yaml.safe_load(path.read_text(encoding="utf-8")) - except yaml.YAMLError as exc: - raise EngagementError(f"Invalid quote YAML: {exc}") from exc - return data if isinstance(data, dict) else None - - -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//``) - 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 diff --git a/tests/test_engagement_cli.py b/tests/test_engagement_cli.py deleted file mode 100644 index d41e61b..0000000 --- a/tests/test_engagement_cli.py +++ /dev/null @@ -1,476 +0,0 @@ -"""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: \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 diff --git a/workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md b/workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md index 3783f95..08e1a2e 100644 --- a/workplans/kaizen-agentic-WP-0009-forward-deployed-agency.md +++ b/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: done + status: todo title: CLI engagement command group (request, staff, prepare, checklist, scrub) - id: T09 - status: done + status: todo 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: done +status: todo priority: medium state_hub_task_id: "52f7d872-1087-4890-9439-8d6c704351fd" ``` -**Delivered:** `kaizen-agentic engagement` CLI group + `engagement.py` module: +Implement `kaizen-agentic engagement` subcommands (Phase 2 of architecture): -`list`, `show`, `validate`, `checklist` (+ `--mark`), `phase`, `prepare`, -`staff`, `quote`, `scrub`, `export-handoff`. +`request`, `quote`, `staff`, `render-agent`, `init-vault`, `phase`, `checklist`, +`prepare`, `scrub`, `export-handoff` — minimum viable subset for pilot may be +`staff`, `checklist`, `prepare`, `phase`. -Tests: `tests/test_engagement_cli.py`. Docs: CLI cheat sheet section. +Tests for schema validate and checklist status. ## Session wire-up ```task id: KAIZEN-WP-0009-T09 -status: done +status: todo priority: medium state_hub_task_id: "82ac0b5a-0309-4912-93bf-f05bee2abbc2" ``` -**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. +- `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 ## railiance01 pilot through ramp-up