acdl/demo/scripts/verify_phase03.sh

#!/usr/bin/env bash
# Phase 03 verification script.
# Confirms the 4 L2 modules and the 5 core scripts conform to their contracts.
#
# Usage: scripts/verify_phase03.sh
# Exit codes: 0 = all checks passed; 1 = one or more checks failed.

set -uo pipefail

ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "$ROOT"

fail_count=0
pass() { printf '  [PASS] %s\n' "$1"; }
fail() { printf '  [FAIL] %s\n' "$1"; fail_count=$((fail_count + 1)); }

# Expected L2 names per REQ-04.
EXPECTED_L2S=(
  l2-invoice-service
  l2-commodity-price-feed
  l2-energy-analytics-api
  l2-regulatory-reporting
)

echo "== Phase 03 verification =="
echo "Root: ${ROOT}"
echo

# --- Check 1: exactly 4 L2 folders with the expected names ---
echo "-- Check 1: 4 L2 folders with expected names --"
actual=$(ls modules/l2/ 2>/dev/null | sort | tr '\n' ' ')
expected=$(printf '%s\n' "${EXPECTED_L2S[@]}" | sort | tr '\n' ' ')
if [ "$actual" = "$expected" ]; then
  pass "exactly 4 L2 folders present and named correctly"
else
  fail "L2 folder list mismatch"
  echo "    expected: $expected"
  echo "    actual:   $actual"
fi

# --- Check 2: each L2 manifest.yaml validates + references 5 existing L1s ---
echo "-- Check 2: L2 manifests reference 5 existing L1s --"
l2_validate=$(python3 << 'PYEOF' || true
import yaml, glob, os, sys
ok = True
l1s = set(os.listdir('modules/l1'))
for f in sorted(glob.glob('modules/l2/*/manifest.yaml')):
    d = yaml.safe_load(open(f))
    folder = os.path.basename(os.path.dirname(f))
    problems = []
    if d.get('name') != folder: problems.append(f"name != {folder}")
    if d.get('kind') != 'l2': problems.append("kind != l2")
    refs = [x.get('name') for x in d.get('l1s', [])]
    if len(refs) != 5: problems.append(f"expected 5 l1s, got {len(refs)}")
    unknown = [r for r in refs if r not in l1s]
    if unknown: problems.append(f"unknown L1 refs: {unknown}")
    # each l1 entry must have an inputs: map
    for x in d.get('l1s', []):
        if not isinstance(x.get('inputs'), dict): problems.append(f"l1 {x.get('name')} missing inputs map")
    status = 'OK' if not problems else 'FAIL: ' + '; '.join(problems)
    print(f'  [{status}] {f}')
    if problems: ok = False
sys.exit(0 if ok else 1)
PYEOF
)
echo "$l2_validate"
if [ "$l2_validate" = "" ] || echo "$l2_validate" | grep -q FAIL; then
  if ! echo "$l2_validate" | grep -q PASS; then
    fail "one or more L2 manifests invalid (see above)"
  fi
else
  pass "all 4 L2 manifests valid"
fi
# Re-run for the explicit pass/fail count
python3 << 'PYEOF' > /tmp/l2_check.txt 2>&1 || true
import yaml, glob, os, sys
ok = True
l1s = set(os.listdir('modules/l1'))
for f in sorted(glob.glob('modules/l2/*/manifest.yaml')):
    d = yaml.safe_load(open(f))
    folder = os.path.basename(os.path.dirname(f))
    if d.get('name') != folder: ok = False
    if d.get('kind') != 'l2': ok = False
    refs = [x.get('name') for x in d.get('l1s', [])]
    if len(refs) != 5: ok = False
    if any(r not in l1s for r in refs): ok = False
    for x in d.get('l1s', []):
        if not isinstance(x.get('inputs'), dict): ok = False
sys.exit(0 if ok else 1)
PYEOF
if [ $? -eq 0 ]; then pass "all 4 L2 manifests pass structural + reference checks"; else fail "L2 manifest structural check"; fi

# --- Check 3: typecheck (bash -n + py_compile + yaml load) ---
echo "-- Check 3: typecheck --"
if bash -n scripts/mock_executor.sh; then pass "bash -n mock_executor.sh"; else fail "bash -n mock_executor.sh"; fi
if python3 -m py_compile scripts/policy_checker.py scripts/confidence_signal.py scripts/evidence_writer.py scripts/l3b_agent_stub.py 2>/dev/null; then
  pass "py_compile all 4 python scripts"
else
  fail "py_compile"
fi
if python3 -c "import yaml, glob; [yaml.safe_load(open(f)) for f in glob.glob('modules/l2/*/manifest.yaml')]" 2>/dev/null; then
  pass "yaml load all L2 manifests"
else
  fail "yaml load L2 manifests"
fi

# --- Check 4: policy_checker (D-025) ---
echo "-- Check 4: policy_checker behavior (D-025) --"
WORK="$(mktemp -d)"
trap 'rm -rf "$WORK" "$ROOT/tmp_pass_contract.yaml" "$ROOT/tmp_fail_contract.yaml" "$ROOT/state.json" 2>/dev/null || true' EXIT
printf 'stack: l2-commodity-price-feed\npublic-ingress: false\n' > "$WORK/pass.yaml"
printf 'stack: l2-regulatory-reporting\npublic-ingress: true\n' > "$WORK/fail.yaml"
out=$(python3 scripts/policy_checker.py "$WORK/pass.yaml" 2>&1); rc=$?
if [ "$out" = "POLICY_PASS" ] && [ "$rc" = "0" ]; then
  pass "policy_checker pass contract -> POLICY_PASS exit 0"
else
  fail "policy_checker pass contract: got '$out' exit=$rc"
fi
out=$(python3 scripts/policy_checker.py "$WORK/fail.yaml" 2>&1); rc=$?
if [ "$out" = "POLICY_VIOLATION:PUBLIC_INGRESS" ] && [ "$rc" = "1" ]; then
  pass "policy_checker fail contract -> POLICY_VIOLATION:PUBLIC_INGRESS exit 1"
else
  fail "policy_checker fail contract: got '$out' exit=$rc"
fi

# --- Check 5: confidence_signal (D-024) ---
echo "-- Check 5: confidence_signal behavior (D-024) --"
out=$(python3 scripts/confidence_signal.py "$WORK/pass.yaml" 2>&1); rc=$?
if echo "$out" | grep -q '"score": 0.90' && [ "$rc" = "0" ]; then
  pass "confidence_signal pass -> score 0.90 exit 0"
else
  fail "confidence_signal pass: got '$out' exit=$rc"
fi
out=$(python3 scripts/confidence_signal.py "$WORK/fail.yaml" 2>&1); rc=$?
if echo "$out" | grep -q '"score": 0.40' && [ "$rc" = "0" ]; then
  pass "confidence_signal fail -> score 0.40 exit 0"
else
  fail "confidence_signal fail: got '$out' exit=$rc"
fi

# --- Check 6: evidence_writer hash chain (D-023) ---
echo "-- Check 6: evidence_writer hash chain (D-023) --"
rm -f "$WORK/audit.json"
python3 scripts/evidence_writer.py --stage dev --event "dev start" --audit "$WORK/audit.json" > /dev/null
python3 scripts/evidence_writer.py --stage qa --event "qa approved" --audit "$WORK/audit.json" > /dev/null
python3 scripts/evidence_writer.py --stage prod --event "prod approved" --audit "$WORK/audit.json" > /dev/null
chain_ok=$(python3 << PYEOF
import json, hashlib, sys
try:
    events = json.load(open("$WORK/audit.json"))
    assert len(events) == 4, f"expected 4 (genesis + 3), got {len(events)}"
    assert events[0]['prev_hash'] == 'GENESIS', "genesis prev_hash"
    for i in range(1, len(events)):
        assert events[i]['prev_hash'] == events[i-1]['hash'], f"chain break at {i}"
        e = dict(events[i]); h = e.pop('hash'); e['hash'] = ''
        canon = json.dumps(e, sort_keys=True, separators=(',',':'))
        assert hashlib.sha256(canon.encode()).hexdigest() == h, f"hash mismatch at {i}"
    print("OK")
except AssertionError as ex:
    print(f"FAIL: {ex}")
    sys.exit(1)
PYEOF
)
if [ "$chain_ok" = "OK" ]; then
  pass "evidence_writer: 4 events, GENESIS + 3, chain links + hashes valid"
else
  fail "evidence_writer chain: $chain_ok"
fi

# --- Check 7: mock_executor (D-022) ---
echo "-- Check 7: mock_executor writes state.json (D-022) --"
rm -f "$ROOT/state.json"
out=$(bash scripts/mock_executor.sh "$WORK/pass.yaml" 2>&1); rc=$?
if [ "$rc" != "0" ]; then
  fail "mock_executor exit $rc (expected 0)"
else
  me_ok=$(python3 << PYEOF
import json, sys
try:
    s = json.load(open("$ROOT/state.json"))
    assert s['l2'] == 'l2-commodity-price-feed', f"l2 mismatch: {s.get('l2')}"
    assert 'l1s' in s and len(s['l1s']) == 5, f"expected 5 l1s, got {len(s.get('l1s', []))}"
    assert all(x['applied'] is True and x['exit_code'] == 0 for x in s['l1s']), "l1 not all applied+0"
    assert 'contract' in s, "missing contract field"
    print("OK")
except Exception as ex:
    print(f"FAIL: {ex}")
    sys.exit(1)
PYEOF
)
  if [ "$me_ok" = "OK" ]; then
    pass "mock_executor: state.json with l2 + 5 l1s (all exit 0) + contract"
  else
    fail "mock_executor state.json: $me_ok"
  fi
fi
rm -f "$ROOT/state.json"

# --- Check 8: l3b_agent_stub D-008 keyword map ---
echo "-- Check 8: l3b_agent_stub keyword map (D-008) --"
act3=$(python3 scripts/l3b_agent_stub.py "We need to ingest natural gas prices from Platts and report on compliance." 2>&1)
if echo "$act3" | grep -q 'stack: l2-commodity-price-feed'; then
  pass "l3b Act 3 example -> l2-commodity-price-feed"
else
  fail "l3b Act 3 example: got '$act3'"
fi
fallback=$(python3 scripts/l3b_agent_stub.py "please deploy something" 2>&1)
if echo "$fallback" | grep -q 'stack: l2-invoice-service'; then
  pass "l3b fallback (no keywords) -> l2-invoice-service"
else
  fail "l3b fallback: got '$fallback'"
fi
regulatory=$(python3 scripts/l3b_agent_stub.py "regulatory compliance reporting for trading desk" 2>&1)
if echo "$regulatory" | grep -q 'stack: l2-regulatory-reporting'; then
  pass "l3b regulatory keywords -> l2-regulatory-reporting"
else
  fail "l3b regulatory: got '$regulatory'"
fi
invoice=$(python3 scripts/l3b_agent_stub.py "monthly invoice and billing reconciliation" 2>&1)
if echo "$invoice" | grep -q 'stack: l2-invoice-service'; then
  pass "l3b invoice keywords -> l2-invoice-service"
else
  fail "l3b invoice: got '$invoice'"
fi
analytics=$(python3 scripts/l3b_agent_stub.py "historical analytics and query API" 2>&1)
if echo "$analytics" | grep -q 'stack: l2-energy-analytics-api'; then
  pass "l3b analytics keywords -> l2-energy-analytics-api"
else
  fail "l3b analytics: got '$analytics'"
fi

echo
echo "== Summary =="
if [ "$fail_count" -eq 0 ]; then
  echo "Phase 03 verification PASSED (4 L2s + 5 core scripts, all checks ok)"
  exit 0
else
  echo "Phase 03 verification FAILED (${fail_count} check(s) failed)"
  exit 1
fi