#!/usr/bin/env python3 """Front-of-funnel grounding coverage and exact-binding regression tests.""" import hashlib import os import shutil import sys import yaml ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) HOOKS = os.path.join(ROOT, ".claude", "hooks") WS = os.path.join(ROOT, ".claude", "tests", "fixtures", "grounding-lens-ws") os.environ["CLAUDE_PROJECT_DIR"] = ROOT os.environ["ORGOS_WORKSPACE"] = WS sys.path.insert(0, HOOKS) import method_contracts as MC # noqa: E402 import state_engine as SE # noqa: E402 from orgos.planning.role_selector import select_minimum_sufficient_roles # noqa: E402 shutil.rmtree(WS, ignore_errors=True) os.makedirs(WS, exist_ok=True) passed = failed = 0 def check(name, condition, detail=""): global passed, failed if condition: passed += 1 print(f" PASS {name}") else: failed += 1 print(f" FAIL {name}: {detail}") def sha(path): digest = hashlib.sha256() with open(path, "rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def open_stage(wf, stage): current = SE.read_ledger(wf).get("stage") return SE._atomic_event_transaction(wf, workflow_event={ "state-event-id": f"fixture-{wf}-{stage}", "event-type": "state-transition", "workflow-id": wf, "from": current, "to": stage, "actor": "OPS-ORCH", "effective-at": SE._now(), })[0] def write_report(wf, artifact_id, kind, producer, stage, payload, execution=None): directory = os.path.join(WS, "completion-records", wf) os.makedirs(directory, exist_ok=True) path = os.path.join(directory, f"{artifact_id}.report.yaml") report = { "report-type": "workflow-artifact", "artifact-kind": kind, "artifact-version": 1, "tier": (SE.read_ledger(wf) or {}).get("tier", "standard"), "identity": {"artifact-id": artifact_id, "workflow-id": wf, "stage": stage, "producer-role-id": producer}, "payload": payload, "report-header": { "bottom-line": f"{kind} grounding fixture", "decision-needed": {"needed": False}, "confidence": {"value": "Med", "derived-from": "evidence"}, "risks": [], "evidence": [{"source-uri": "README.md", "grade": "E3"}], }, } if execution is not None: report["method-execution"] = execution with open(path, "w", encoding="utf-8") as handle: yaml.safe_dump(report, handle, allow_unicode=True, sort_keys=False) return path def context_package(wf, run_id, role, lens, *, context_lens=None): directory = os.path.join(WS, "context-packages", wf) os.makedirs(directory, exist_ok=True) path = os.path.join(directory, f"{run_id}.yaml") value = { "workflow-id": wf, "task-id": run_id, "mode": "divergent", "tier": "standard", "assigned-lens": context_lens or lens, "target-role-agent": role.lower(), "objective": f"independently investigate {lens}", } with open(path, "w", encoding="utf-8") as handle: yaml.safe_dump(value, handle, allow_unicode=True, sort_keys=False) return path def contribution(wf, index, role, lens, *, context_lens=None, market=False): run_id = f"{wf}-run-{index}" package = context_package(wf, run_id, role, lens, context_lens=context_lens) payload = { "assigned-lens": lens, "producer-run-id": run_id, "context-package-ref": package, "context-package-sha256": sha(package), } kind = "competitive-market-grounding" if market else "grounding-contribution" if market: payload.update({ "competitors-and-substitutes": [ {"name": "Named Competitor", "type": "competitor", "evidence-urls": ["https://example.com/competitor"]}, {"name": "Manual Workflow", "type": "substitute", "evidence-urls": ["https://example.com/substitute"]}, ], "current-alternatives": ["keep the manual workflow"], "strengths-weaknesses": [{"subject": "Named Competitor", "strengths": ["distribution"], "weaknesses": ["workflow depth"]}], "differentiation-hypotheses": ["deeper guided workflow"], "evidence-urls": ["https://example.com/competitor", "https://example.com/substitute"], }) else: payload.update({"findings": [f"independent {lens} finding"], "evidence-urls": [f"https://example.com/{index}"]}) artifact_id = f"{wf}-contribution-{index}" path = write_report(wf, artifact_id, kind, role, "discovery", payload) ok, detail = SE.submit_report(wf, path, role) if not ok: raise AssertionError(detail) return { "report-id": artifact_id, "report-ref": path, "report-sha256": sha(path), "producer-role-id": role, "context-package-ref": package, "context-package-sha256": sha(package), "assigned-lens": lens, "producer-run-id": run_id, }, path def strategy_execution(): profile = MC.resolve_method_profile("STR-ANALYST", "strategy-analysis") return { "role-id": "STR-ANALYST", "method-id": "strategy-analysis", "contract-sha256": MC.canonical_contract_hash(profile), "step-results": [ {"step-id": "structure-problem", "status": "completed", "output-binding": "current-artifact"}, {"step-id": "analyze-environment", "status": "completed", "output-binding": "current-artifact"}, {"step-id": "diverge-options", "status": "completed", "output-binding": "current-artifact"}, ], "self-check-results": [{"step-id": "diverge-options", "gate-id": "options-diverge", "verdict": "Passed", "evidence-refs": ["source-contributions"]}], "decisions": [{"decision-id": "grounding-options", "selected-option-id": "a", "alternatives": [{"option-id": "a"}, {"option-id": "b"}]}], } def setup(wf, specs, *, public=False, required_capabilities=None, context_mismatch_index=None): SE.init_ledger(wf, tier="standard") brief = write_report(wf, f"{wf}-brief", "decision-brief", "EXEC-CEO", "intake", { "mode": "divergent", "tier": "standard", "candidate-families": ["FAM-CEO", "FAM-CPO", "FAM-CTO", "FAM-CFO", "FAM-COO", "FAM-QA", "FAM-GTM-GROWTH", "FAM-STRATEGY"], }) ok, detail = SE.submit_report(wf, brief, "EXEC-CEO") if not ok: raise AssertionError(detail) if public or required_capabilities: profile = write_report(wf, f"{wf}-profile", "workload-profile", "EXEC-CEO", "intake", { "surfaces": {"ui": True, "public-api": False, "persistence": False, "infrastructure": False}, "risk": {"security-bearing": False, "data-migration": False, "external-side-effect": False, "risk-level": "Med", "reversibility": "two-way-door", "blast-radius": "single-role", "privacy": False, "regulatory": False, "slo-impact": False}, "required-capabilities": required_capabilities or ["competitive-intelligence"], "product-feature": True, "surface-archetype": "public-website" if public else "internal-tool", "experience-change": "new-product" if public else "incremental", }) ok, detail = SE.submit_report(wf, profile, "EXEC-CEO") if not ok: raise AssertionError(detail) if not open_stage(wf, "discovery"): raise AssertionError("failed to open discovery") refs, paths = [], [] for index, (role, lens, market) in enumerate(specs, start=1): ref, path = contribution( wf, index, role, lens, market=market, context_lens=("LENS-OPS" if context_mismatch_index == index else None), ) refs.append(ref) paths.append(path) return refs, paths def submit_ground(wf, refs, *, market_ref=None): covered = sorted({ref["assigned-lens"] for ref in refs}) contrarian = [ref["report-id"] for ref in refs if ref["assigned-lens"] == "LENS-CONTRARIAN"] payload = { "problem-structure": {"question": "which grounded direction is defensible"}, "analysis-synthesis": {"insight": "independent lenses expose different failure modes"}, "evidence": [ref["report-id"] for ref in refs], "options": [ {"id": "a", "problem": "narrow value", "tradeoffs": ["reach"], "evidence-refs": [refs[0]["report-id"]]}, {"id": "b", "problem": "broad value", "tradeoffs": ["cost"], "evidence-refs": [refs[-1]["report-id"]]}, ], "source-contributions": refs, "lens-coverage": {"required-min": 5, "covered": covered, "contrarian-report-id": contrarian[0] if len(contrarian) == 1 else None}, } if market_ref: payload["competitive-market-grounding-ref"] = { key: market_ref[key] for key in ("report-id", "report-ref", "report-sha256") } path = write_report(wf, f"{wf}-ground", "grounding-package", "STR-ANALYST", "discovery", payload, strategy_execution()) ok, detail = SE.submit_report(wf, path, "STR-ANALYST") if not ok: raise AssertionError(detail) return SE.can_transition(wf, "decide", actor="OPS-ORCH") VALUE = ("EXEC-CEO", "LENS-VALUE", False) PRODUCT = ("EXEC-CPO", "LENS-PRODUCT", False) TECH = ("EXEC-CTO", "LENS-TECH", False) FINANCE = ("EXEC-CFO", "LENS-FINANCE", False) OPS = ("EXEC-COO", "LENS-OPS", False) CONTRARIAN = ("QA", "LENS-CONTRARIAN", False) MARKET = ("GTM-CI", "LENS-REVENUE", True) print("== tier lens floor and contrarian ==") refs, _ = setup("ground-one-lens", [VALUE] * 5) ok, reasons = submit_ground("ground-one-lens", refs) check("1 standard one distinct lens blocks", not ok and any("distinct lens" in reason for reason in reasons), reasons) refs, _ = setup("ground-four-lenses", [VALUE, PRODUCT, TECH, FINANCE]) ok, reasons = submit_ground("ground-four-lenses", refs) check("2 standard four unique lenses block", not ok and any("distinct lens" in reason for reason in reasons), reasons) refs, _ = setup("ground-no-contrarian", [VALUE, PRODUCT, TECH, FINANCE, OPS]) ok, reasons = submit_ground("ground-no-contrarian", refs) check("3 standard five lenses without contrarian block", not ok and any("CONTRARIAN" in reason for reason in reasons), reasons) refs, _ = setup("ground-valid", [VALUE, PRODUCT, TECH, FINANCE, CONTRARIAN]) ok, reasons = submit_ground("ground-valid", refs) check("4 standard five lenses including contrarian allow", ok, reasons) print("== exact source/context identity ==") refs, _ = setup("ground-reused", [VALUE, PRODUCT, TECH, FINANCE, CONTRARIAN]) reused = [dict(refs[0]) for _ in range(5)] ok, reasons = submit_ground("ground-reused", reused) check("5 same report repeated five times blocks", not ok and any("중복" in reason for reason in reasons), reasons) refs, _ = setup("ground-context-mismatch", [VALUE, PRODUCT, TECH, FINANCE, CONTRARIAN], context_mismatch_index=1) ok, reasons = submit_ground("ground-context-mismatch", refs) check("6 report lens and context assigned-lens mismatch blocks", not ok and any("assigned-lens 불일치" in reason for reason in reasons), reasons) print("== candidate families and capability selection ==") SE.init_ledger("unknown-family", tier="standard") unknown = write_report("unknown-family", "unknown-brief", "decision-brief", "EXEC-CEO", "intake", { "mode": "divergent", "tier": "standard", "candidate-families": ["FAM-DOES-NOT-EXIST"], }) ok, detail = SE.submit_report("unknown-family", unknown, "EXEC-CEO") check("7 unknown candidate family hard-fails submission", not ok and "미등록 family" in str(detail), detail) SE.init_ledger("duplicate-family", tier="standard") duplicate = write_report("duplicate-family", "duplicate-brief", "decision-brief", "EXEC-CEO", "intake", { "mode": "divergent", "tier": "standard", "candidate-families": ["FAM-CEO", "FAM-CPO", "FAM-CTO", "FAM-CFO", "FAM-QA", "FAM-CEO"], }) ok, detail = SE.submit_report("duplicate-family", duplicate, "EXEC-CEO") check("candidate family duplicates hard-fail submission", not ok and "중복" in str(detail), detail) plan = select_minimum_sufficient_roles({ "tier": "standard", "mode": "divergent", "candidate-families": ["FAM-STRATEGY"], })["selection-plan"] check("8 theoretically insufficient candidate lens coverage blocks role plan", plan["status"] == "blocked" and any("이론 렌즈" in item for item in plan.get("errors", [])), plan) plan = select_minimum_sufficient_roles({ "tier": "standard", "mode": "converge", "candidate-families": ["FAM-GTM-GROWTH"], "required-capabilities": ["competitive-intelligence"], "excluded-role-ids": ["GTM-CI"], })["selection-plan"] check("9 FAM-GTM-GROWTH without GTM-CI cannot satisfy competitive intelligence", plan["status"] == "blocked" and "capability:competitive-intelligence" in plan["coverage"]["missing"], plan) refs, _ = setup("ground-capability-missing", [VALUE, PRODUCT, TECH, FINANCE, CONTRARIAN], required_capabilities=["competitive-intelligence"]) ok, reasons = submit_ground("ground-capability-missing", refs) check("9b competitive-intelligence workload without GTM-CI contribution blocks discovery", not ok and any("GTM-CI" in reason for reason in reasons), reasons) print("== stale source and public market grounding ==") refs, paths = setup("ground-stale", [VALUE, PRODUCT, TECH, FINANCE, CONTRARIAN]) old_path = paths[0] check("stale setup accepts old contribution", SE.review_artifact("ground-stale", old_path, "accepted", "HUMAN-001")[0]) replacement, replacement_path = contribution("ground-stale", 99, "EXEC-CEO", "LENS-VALUE") check("stale setup supersedes old contribution", SE.review_artifact("ground-stale", replacement_path, "accepted", "HUMAN-001", supersedes=refs[0]["report-id"])[0]) ok, reasons = submit_ground("ground-stale", refs) check("10 superseded contribution blocks", not ok and any("stale" in reason for reason in reasons), reasons) refs, _ = setup("ground-public-missing", [VALUE, PRODUCT, TECH, FINANCE, CONTRARIAN], public=True) ok, reasons = submit_ground("ground-public-missing", refs) check("public/new ground requires GTM-CI competitive market artifact", not ok and any("GTM-CI" in reason for reason in reasons), reasons) refs, _ = setup("ground-public-valid", [VALUE, PRODUCT, TECH, CONTRARIAN, MARKET], public=True) market_ref = next(ref for ref in refs if ref["producer-role-id"] == "GTM-CI") ok, reasons = submit_ground("ground-public-valid", refs, market_ref=market_ref) check("public/new ground accepts exact GTM-CI competitive market evidence", ok, reasons) print(f"\n{passed} passed, {failed} failed") sys.exit(1 if failed else 0)