120 lines
4.5 KiB
Python
Executable File
120 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import itertools
|
|
import json
|
|
import pathlib
|
|
from datetime import datetime, timezone
|
|
|
|
def _app_root(start):
|
|
# Plan-01: harness code lives in packages/casan-harness/ but runtime state
|
|
# (`.specify/logs`) stays with the app. `.resolve()` follows the compat symlink
|
|
# into packages, so walk UP for the `.specify` state marker instead of assuming
|
|
# a fixed parent depth.
|
|
d = pathlib.Path(start).resolve()
|
|
for p in (d, *d.parents):
|
|
if (p / ".specify").is_dir() or (p / "packages" / "casan-harness").is_dir():
|
|
return p
|
|
return d.parents[2]
|
|
|
|
|
|
ROOT = _app_root(__file__)
|
|
TRACE_DIR = ROOT / ".specify" / "logs" / "trace"
|
|
OUT = ROOT / "docs" / "output" / "output_logs" / "casan-demo" / "pipeline-context.yaml"
|
|
|
|
steps = [
|
|
("step-0", "detect-existing-spec", "agent_step"),
|
|
("step-1-srs", "casan.srs", "agent_step"),
|
|
("step-2-bd", "casan.bd", "agent_step"),
|
|
("step-3-spec", "speckit.specify", "agent_step"),
|
|
("step-4-clarify", "speckit.clarify", "agent_step"),
|
|
("step-5-review-spec", "casan.reviewspec", "agent_step"),
|
|
("step-6-plan", "speckit.plan", "agent_step"),
|
|
("step-7-review-plan", "casan.reviewplan", "agent_step"),
|
|
("step-8-dd", "casan.dd", "agent_step"),
|
|
("step-8b-testcases", "casan.testkit.gen-testcases", "agent_step"),
|
|
("step-9-tasks", "speckit.tasks", "agent_step"),
|
|
("step-10-implement", "speckit.implement", "write_code"),
|
|
("step-11-review-code", "casan.reviewcode", "agent_step"),
|
|
("step-12-testkit", "casan.testkit.run-tests", "agent_step"),
|
|
("step-13-launch", "boss.launch", "deploy"),
|
|
]
|
|
|
|
def load(path: pathlib.Path) -> dict:
|
|
with path.open(encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
security = [p for p in sorted(TRACE_DIR.glob("security-*.json")) if load(p).get("status") != "blocked"]
|
|
governance = [p for p in sorted(TRACE_DIR.glob("governance-*.json")) if load(p).get("decision") == "approved"]
|
|
agentops = [p for p in sorted(TRACE_DIR.glob("agentops-*.json")) if load(p).get("status") == "success"]
|
|
|
|
high_risk_approved = [
|
|
p for p in governance
|
|
if load(p).get("risk_level") == "high" and load(p).get("approval_status") == "human_approved"
|
|
]
|
|
low_risk_approved = [p for p in governance if load(p).get("risk_level") in {"low", "medium"}]
|
|
|
|
if not security or not governance or not agentops:
|
|
raise SystemExit("missing trace files; run run-casan4-harness-tests.sh first")
|
|
|
|
OUT.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
def rel(path: pathlib.Path) -> str:
|
|
return path.relative_to(ROOT).as_posix()
|
|
|
|
security_cycle = itertools.cycle(security)
|
|
low_governance_cycle = itertools.cycle(low_risk_approved or governance)
|
|
high_governance_cycle = itertools.cycle(high_risk_approved or governance)
|
|
agentops_cycle = itertools.cycle(agentops)
|
|
|
|
lines: list[str] = []
|
|
lines.extend(
|
|
[
|
|
"# CASAN4 demo pipeline context",
|
|
f"generated-at: {datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')}",
|
|
"feature-id: casan-demo",
|
|
"module-id: mod-casan",
|
|
"module-keyword: harness",
|
|
"mode: autonomous",
|
|
"language: Vietnamese",
|
|
"casan-harness:",
|
|
" level-target: 4",
|
|
" h4-security:",
|
|
" audit-log: .specify/logs/audit/security.jsonl",
|
|
" h5-governance:",
|
|
" audit-log: .specify/logs/audit/audit.jsonl",
|
|
" h6-agentops:",
|
|
" metrics-log: .specify/logs/cost/metrics.jsonl",
|
|
" alert-log: .specify/agentops/alerts.log",
|
|
"steps:",
|
|
]
|
|
)
|
|
|
|
for step_id, agent, action in steps:
|
|
h4 = next(security_cycle)
|
|
h5 = next(high_governance_cycle if action in {"write_code", "migration", "db_write", "deploy", "external_api"} else low_governance_cycle)
|
|
h6 = next(agentops_cycle)
|
|
lines.extend(
|
|
[
|
|
f" {step_id}:",
|
|
" status: COMPLETE",
|
|
f" agent: {agent}",
|
|
f" governance-action: {action}",
|
|
" casan:",
|
|
" h4-security:",
|
|
" status: PASS",
|
|
f" trace: {rel(h4)}",
|
|
" h5-governance:",
|
|
" decision: approved",
|
|
f" trace: {rel(h5)}",
|
|
" audit-log: .specify/logs/audit/audit.jsonl",
|
|
" h6-agentops:",
|
|
" status: success",
|
|
f" trace: {rel(h6)}",
|
|
" metrics-log: .specify/logs/cost/metrics.jsonl",
|
|
]
|
|
)
|
|
|
|
OUT.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
print(f"DEMO_CONTEXT_GENERATED {OUT}")
|