tegwick
4532196546
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.
476 lines
15 KiB
Python
476 lines
15 KiB
Python
"""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
|