305 lines
16 KiB
Python
305 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""venture-bootstrap plan + predicate 단위테스트. standalone. exit 0 = all pass."""
|
|
import os, sys, shutil, tempfile, yaml
|
|
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
FIX = os.path.join(ROOT, ".claude", "tests", "fixtures")
|
|
os.makedirs(FIX, exist_ok=True)
|
|
WS = os.path.join(FIX, "venture-bootstrap-ws")
|
|
shutil.rmtree(WS, ignore_errors=True); os.makedirs(WS, exist_ok=True)
|
|
os.environ["CLAUDE_PROJECT_DIR"] = ROOT
|
|
os.environ["ORGOS_WORKSPACE"] = WS
|
|
sys.path.insert(0, HOOKS)
|
|
import state_engine as SE # noqa: E402
|
|
|
|
passed, failed = 0, 0
|
|
def check(name, ok):
|
|
global passed, failed
|
|
if ok: passed += 1; print(f" PASS {name}")
|
|
else: failed += 1; print(f" FAIL {name}")
|
|
|
|
# plan 로드에 venture-bootstrap 존재
|
|
plans = SE.load_plans().get("plans", {})
|
|
check("venture-bootstrap plan 존재", "venture-bootstrap" in plans)
|
|
vb = plans.get("venture-bootstrap", {})
|
|
check("stages 순서", vb.get("stages", [])[:3] == ["intake", "founder-setup", "opportunity-discovery"])
|
|
check("terminal-stage", vb.get("terminal-stage") == "bootstrap-complete")
|
|
|
|
# 전이 존재 + actor contract의 concrete executor는 OPS-ORCH 단독
|
|
t = SE._find_transition("founder-setup", "opportunity-discovery")
|
|
check("founder-setup->opportunity-discovery 전이 존재", bool(t))
|
|
check("전이 executor = [OPS-ORCH] 단독",
|
|
t and (t.get("allowed-by") or {}).get("executor") == ["OPS-ORCH"])
|
|
t2 = SE._find_transition("venture-decision", "company-context-commit")
|
|
check("venture-decision->company-context-commit 존재", bool(t2))
|
|
t3 = SE._find_transition("company-context-commit", "bootstrap-complete")
|
|
check("company-context-commit->bootstrap-complete 존재", bool(t3))
|
|
|
|
def led_at(stage, **extra):
|
|
d = {"workflow-id": "vb1", "stage": stage, "plan": "venture-bootstrap", "tier": "standard", "artifacts": []}
|
|
d.update(extra); return d
|
|
|
|
def temp_company_context(status):
|
|
fd, path = tempfile.mkstemp(suffix=".yaml")
|
|
os.close(fd)
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump({
|
|
"schema-version": 2,
|
|
"status": status,
|
|
"company": {
|
|
"facts": [],
|
|
"strategic-decisions": [],
|
|
"hypotheses": [],
|
|
"validation-state": {
|
|
"stage": "pre-traction",
|
|
"validated": [],
|
|
"open": [],
|
|
"refuted": [],
|
|
},
|
|
},
|
|
"projects": [],
|
|
}, fh, allow_unicode=True)
|
|
return path
|
|
|
|
# founder-context-present: 실제 파일 status 에 의존. filled 가정 불가하므로 조건 평가만 확인.
|
|
f = SE._facts("vb1", led_at("founder-setup"))
|
|
ok, _ = SE._eval_condition("founder-context-present", f)
|
|
check("founder-context-present 평가 가능", isinstance(ok, bool))
|
|
|
|
# opportunity-clusters-present: caller-supplied ledger 리스트는 신뢰하지 않는다.
|
|
f = SE._facts("vb1", led_at("opportunity-discovery", **{"opportunity-clusters": [{"id": "OC1"}]}))
|
|
ok, _ = SE._eval_condition("opportunity-clusters-present", f)
|
|
check("self-reported opportunity cluster 1개 -> 차단", ok is False)
|
|
f = SE._facts("vb1", led_at("opportunity-discovery", **{"opportunity-clusters": [{"id": "OC1"}, {"id": "OC2"}]}))
|
|
ok, _ = SE._eval_condition("opportunity-clusters-present", f)
|
|
check("self-reported opportunity clusters 2개도 -> 차단", ok is False)
|
|
check("동일 payload.id cluster 2개 복제는 distinct 1개",
|
|
SE._distinct_opportunity_cluster_count([
|
|
{"design-type": "opportunity-cluster", "opportunity-cluster": {"id": "OC1"}},
|
|
{"design-type": "opportunity-cluster", "opportunity-cluster": {"id": "OC1"}},
|
|
]) == 1)
|
|
check("서로 다른 payload.id cluster 2개는 distinct 2개",
|
|
SE._distinct_opportunity_cluster_count([
|
|
{"design-type": "opportunity-cluster", "opportunity-cluster": {"id": "OC1"}},
|
|
{"design-type": "opportunity-cluster", "opportunity-cluster": {"id": "OC2"}},
|
|
]) == 2)
|
|
|
|
# 상태별 predicate 테스트는 실제 운영 company-context와 격리한다. 운영 파일은
|
|
# bootstrap 완료 후 provisional/operating일 수 있으므로 template을 가정하면 안 된다.
|
|
_real_ctx_path = SE._COMPANY_CTX_PATH
|
|
_template_ctx_path = temp_company_context("template")
|
|
SE._COMPANY_CTX_PATH = _template_ctx_path
|
|
|
|
# company-context-lint-passed: 격리된 template 파일이 정상이면 통과
|
|
f = SE._facts("vb1", led_at("company-context-commit"))
|
|
ok, _ = SE._eval_condition("company-context-lint-passed", f)
|
|
check("company-context-lint-passed 평가 가능", isinstance(ok, bool))
|
|
|
|
# company-context-provisional-committed: 공식 status=template 이면 False
|
|
ok, _ = SE._eval_condition("company-context-provisional-committed", f)
|
|
check("template 상태 -> committed False", ok is False)
|
|
|
|
# human-acceptance-receipt-present: 이벤트 없으면 False(boolean 자기신고 거부)
|
|
f2 = SE._facts("vb1", led_at("venture-decision", **{"facts": {"human_acceptance_receipt_present": True}}))
|
|
ok, _ = SE._eval_condition("human-acceptance-receipt-present", f2)
|
|
check("human boolean 자기신고 -> 거부(receipt 없음)", ok is False)
|
|
|
|
# --- Task 11 fix: positive-case assertions (only pass when predicates are actually implemented) ---
|
|
|
|
# company-context-lint-passed: official file is template(empty blocks) -> lints clean -> True
|
|
f = SE._facts("vb1", led_at("company-context-commit"))
|
|
ok, _ = SE._eval_condition("company-context-lint-passed", f)
|
|
check("company-context-lint-passed on clean official -> True", ok is True)
|
|
|
|
# company-context-provisional-committed: monkeypatch official path to a provisional file -> True
|
|
_provisional_ctx_path = temp_company_context("provisional")
|
|
SE._COMPANY_CTX_PATH = _provisional_ctx_path
|
|
f = SE._facts("vb1", led_at("company-context-commit"))
|
|
ok, _ = SE._eval_condition("company-context-provisional-committed", f)
|
|
check("provisional official -> committed True", ok is True)
|
|
SE._COMPANY_CTX_PATH = _template_ctx_path
|
|
os.unlink(_provisional_ctx_path)
|
|
|
|
# company-context-artifact-recorded: artifact design-type=company-context -> True; empty -> False
|
|
f = SE._facts("vb1", led_at("company-context-commit", artifacts=[{"design-type": "company-context"}]))
|
|
ok, _ = SE._eval_condition("company-context-artifact-recorded", f)
|
|
check("company-context artifact present -> True", ok is True)
|
|
f = SE._facts("vb1", led_at("company-context-commit", artifacts=[]))
|
|
ok, _ = SE._eval_condition("company-context-artifact-recorded", f)
|
|
check("no company-context artifact -> False", ok is False)
|
|
|
|
# venture-decision-accepted: smoke (evaluable); full positive path is integration (accepted artifact + acceptance event) — covered by Task 16
|
|
f = SE._facts("vb1", led_at("venture-decision"))
|
|
ok, _ = SE._eval_condition("venture-decision-accepted", f)
|
|
check("venture-decision-accepted evaluable(bool)", isinstance(ok, bool))
|
|
|
|
# TODO(Task 15): add human-acceptance-receipt-present positive case (HUMAN-001 accepted event with matching report-sha256) once acceptance_log emits report-sha256.
|
|
|
|
# --- Task 12: company-context-ready seam predicate (advisory, product-cascade entry) ---
|
|
|
|
# company-context-ready: 공식 status=template 이면 not ready
|
|
f = SE._facts("vb1", led_at("intake"))
|
|
ok, _ = SE._eval_condition("company-context-ready", f)
|
|
check("template -> cascade not ready", ok is False)
|
|
|
|
# Proportional path: a trusted project-scoped workload may improve an existing
|
|
# repository without manufacturing company/market facts. Both explicit fields
|
|
# are required so a generic or malformed workload cannot open the seam.
|
|
project_scoped = led_at("intake", artifacts=[{
|
|
"artifact-kind": "workload-profile",
|
|
"workload-profile": {"context-scope": "project", "product-feature": True},
|
|
}])
|
|
f = SE._facts("vb1", project_scoped)
|
|
ok, _ = SE._eval_condition("company-context-ready", f)
|
|
check("template + trusted project-scoped workload -> cascade ready", ok is True)
|
|
|
|
not_product = led_at("intake", artifacts=[{
|
|
"artifact-kind": "workload-profile",
|
|
"workload-profile": {"context-scope": "project", "product-feature": False},
|
|
}])
|
|
f = SE._facts("vb1", not_product)
|
|
ok, _ = SE._eval_condition("company-context-ready", f)
|
|
check("project scope without product-feature=true -> not ready", ok is False)
|
|
|
|
# positive: provisional official + no blocker -> ready True
|
|
_provisional_ready_path = temp_company_context("provisional")
|
|
SE._COMPANY_CTX_PATH = _provisional_ready_path
|
|
f = SE._facts("vb1", led_at("intake"))
|
|
ok, _ = SE._eval_condition("company-context-ready", f)
|
|
check("provisional official -> cascade ready True", ok is True)
|
|
SE._COMPANY_CTX_PATH = _real_ctx_path
|
|
os.unlink(_provisional_ready_path)
|
|
os.unlink(_template_ctx_path)
|
|
|
|
# --- Task 16: commit_company_context.py — candidate → 공식 원자적 교체 ---
|
|
import subprocess, tempfile, yaml
|
|
|
|
def _write(path, obj):
|
|
with open(path, "w", encoding="utf-8") as fh: yaml.safe_dump(obj, fh, allow_unicode=True, sort_keys=False)
|
|
|
|
# lint 실패 candidate -> 공식 파일 무변경 + exit 1
|
|
official = os.path.join(ROOT, "org-os", "01-company", "company-context.yaml")
|
|
before = open(official, encoding="utf-8").read()
|
|
cand = os.path.join(WS, "bad.candidate.yaml")
|
|
_write(cand, {"schema-version": 2, "status": "provisional", "candidate-status": "bootstrap",
|
|
"company": {"facts": [], "strategic-decisions": [], "hypotheses": [],
|
|
"validation-state": {"stage": "pre-traction", "validated": [], "open": [], "refuted": []}}, "projects": []})
|
|
# provisional 인데 human decision 없음 -> lint hard fail
|
|
r = subprocess.run([sys.executable, os.path.join(HOOKS, "commit_company_context.py"), "--workflow", "vb1", "--candidate", cand],
|
|
capture_output=True, text=True, env={**os.environ})
|
|
check("lint 실패 candidate -> exit 1", r.returncode == 1)
|
|
check("공식 파일 무변경", open(official, encoding="utf-8").read() == before)
|
|
|
|
# --- Task 16 fix: success path (atomic replace) + require-human gate, in-process, temp OFFICIAL (real SoT untouched) ---
|
|
import commit_company_context as CC
|
|
import tempfile as _tf16, shutil as _sh16, glob as _gl16, yaml as _yaml16
|
|
|
|
_TEMPLATE_CAND = ("schema-version: 2\nstatus: template\ncandidate-status: bootstrap\n"
|
|
"company: {facts: [], strategic-decisions: [], hypotheses: [], "
|
|
"validation-state: {stage: pre-traction, validated: [], open: [], refuted: []}}\nprojects: []\n")
|
|
_OFFICIAL_SEED = ("schema-version: 2\nstatus: template\n"
|
|
"company: {facts: [], strategic-decisions: [], hypotheses: [], "
|
|
"validation-state: {stage: pre-traction, validated: [], open: [], refuted: []}}\nprojects: []\n")
|
|
|
|
_orig_off = CC.OFFICIAL
|
|
# success: valid template candidate -> exit 0, official replaced, candidate-status stripped, no tmp leak
|
|
_d = _tf16.mkdtemp()
|
|
_off = os.path.join(_d, "company-context.yaml"); open(_off, "w", encoding="utf-8").write(_OFFICIAL_SEED)
|
|
_cand = os.path.join(_d, "cand.yaml"); open(_cand, "w", encoding="utf-8").write(_TEMPLATE_CAND)
|
|
CC.OFFICIAL = _off
|
|
_rc = CC.main(["--workflow", "vb1", "--candidate", _cand])
|
|
_res = _yaml16.safe_load(open(_off, encoding="utf-8"))
|
|
check("commit success -> exit 0", _rc == 0)
|
|
check("official replaced + candidate-status stripped", ("candidate-status" not in _res) and _res.get("status") == "template")
|
|
check("no leftover .tmp in official dir", len([p for p in os.listdir(_d) if p != "company-context.yaml" and p != "cand.yaml"]) == 0)
|
|
CC.OFFICIAL = _orig_off
|
|
_sh16.rmtree(_d, ignore_errors=True)
|
|
|
|
# require-human without a matching acceptance receipt -> exit 1, official untouched
|
|
_d2 = _tf16.mkdtemp()
|
|
_off2 = os.path.join(_d2, "company-context.yaml"); open(_off2, "w", encoding="utf-8").write(_OFFICIAL_SEED)
|
|
_before2 = open(_off2, encoding="utf-8").read()
|
|
_cand2 = os.path.join(_d2, "cand.yaml"); open(_cand2, "w", encoding="utf-8").write(_TEMPLATE_CAND)
|
|
CC.OFFICIAL = _off2
|
|
_rc2 = CC.main(["--workflow", "no-such-wf", "--candidate", _cand2, "--require-human"])
|
|
check("require-human w/o receipt -> exit 1", _rc2 == 1)
|
|
check("require-human fail -> official untouched", open(_off2, encoding="utf-8").read() == _before2)
|
|
CC.OFFICIAL = _orig_off
|
|
_sh16.rmtree(_d2, ignore_errors=True)
|
|
|
|
# --- Critical fix: commit_company_context human-gate now mandatory for provisional/operating
|
|
# targets (driven by candidate's OWN target status, not just the --require-human flag) ---
|
|
import acceptance_log as AL, hashlib as _cf_hashlib, _workspace as _cf_W
|
|
|
|
_CF_PROVISIONAL_CAND = {
|
|
"schema-version": 2,
|
|
"status": "provisional",
|
|
"candidate-status": "bootstrap",
|
|
"company": {
|
|
"facts": [],
|
|
"strategic-decisions": [
|
|
{
|
|
"id": "sd-1",
|
|
"statement": "founder 가 벤처 방향을 확정했다(HUMAN-001 승인)",
|
|
"accepted-by": "HUMAN-001",
|
|
"accepted-at": "2026-07-12T00:00:00Z",
|
|
"source-decision-id": "vd-gate-1",
|
|
"supporting-evidence": [],
|
|
}
|
|
],
|
|
"hypotheses": [],
|
|
"validation-state": {"stage": "pre-traction", "validated": [], "open": [], "refuted": []},
|
|
},
|
|
"projects": [],
|
|
}
|
|
|
|
# (a) NEGATIVE regression — the bypass this fix closes: a provisional candidate that lints clean
|
|
# (hand-writable by an agent, incl. a HUMAN-001 string in accepted-by), committed with NEITHER
|
|
# --require-human NOR any real acceptance event for the workflow -> must now be refused (exit 1)
|
|
# and the official file must stay untouched. Before the fix this would have committed (exit 0),
|
|
# flipping company-context-provisional-committed with zero real human receipt.
|
|
_cf_d1 = _tf16.mkdtemp()
|
|
_cf_off1 = os.path.join(_cf_d1, "company-context.yaml"); open(_cf_off1, "w", encoding="utf-8").write(_OFFICIAL_SEED)
|
|
_cf_before1 = open(_cf_off1, encoding="utf-8").read()
|
|
_cf_cand1 = os.path.join(_cf_d1, "prov.yaml")
|
|
_write(_cf_cand1, _CF_PROVISIONAL_CAND)
|
|
CC.OFFICIAL = _cf_off1
|
|
_cf_rc1 = CC.main(["--workflow", "vb-bypass-neg", "--candidate", _cf_cand1]) # NO --require-human
|
|
check("provisional candidate w/o receipt + no --require-human -> exit 1 (bypass closed)", _cf_rc1 == 1)
|
|
check("bypass attempt -> official file untouched", open(_cf_off1, encoding="utf-8").read() == _cf_before1)
|
|
CC.OFFICIAL = _orig_off
|
|
_sh16.rmtree(_cf_d1, ignore_errors=True)
|
|
|
|
# (b) POSITIVE integration — the legit path still works: a real HUMAN-001 accepted event bound
|
|
# (report-sha256) to an actual venture-decision report file on disk -> committer succeeds with
|
|
# NO --require-human needed (mandatory-by-status makes the flag redundant for provisional).
|
|
_cf_wf, _cf_rid = "vbgate", "vd-gate-1"
|
|
_cf_report_dir = os.path.join(_cf_W.records_dir(), _cf_wf)
|
|
os.makedirs(_cf_report_dir, exist_ok=True)
|
|
_cf_report_path = os.path.join(_cf_report_dir, f"{_cf_rid}.report.yaml")
|
|
with open(_cf_report_path, "wb") as _fh:
|
|
_fh.write(b"report-id: vd-gate-1\nkind: venture-decision\nstub: true\n")
|
|
_cf_resolved = AL._resolve_report_path(_cf_rid, _cf_wf)
|
|
check("acceptance_log resolves the stub venture-decision report path", _cf_resolved == _cf_report_path)
|
|
_cf_sha = _cf_hashlib.sha256(open(_cf_report_path, "rb").read()).hexdigest()
|
|
_cf_ev = AL.build_event(_cf_rid, "accepted", workflow=_cf_wf, role="HUMAN-001", report_sha256=_cf_sha,
|
|
artifact_kind="venture-decision", producer_role_id="EXEC-CEO")
|
|
check("acceptance event appended", AL.append_event(_cf_ev) is True)
|
|
check("_venture_decision_receipt_ok(vbgate) is True (sanity: setup, not the predicate, must be right)",
|
|
SE._venture_decision_receipt_ok(_cf_wf) is True)
|
|
|
|
_cf_d2 = _tf16.mkdtemp()
|
|
_cf_off2 = os.path.join(_cf_d2, "company-context.yaml"); open(_cf_off2, "w", encoding="utf-8").write(_OFFICIAL_SEED)
|
|
_cf_cand2 = os.path.join(_cf_d2, "prov.yaml")
|
|
_write(_cf_cand2, _CF_PROVISIONAL_CAND)
|
|
CC.OFFICIAL = _cf_off2
|
|
_cf_rc2 = CC.main(["--workflow", _cf_wf, "--candidate", _cf_cand2]) # no --require-human needed
|
|
_cf_res2 = _yaml16.safe_load(open(_cf_off2, encoding="utf-8"))
|
|
check("legit human receipt -> commit succeeds exit 0", _cf_rc2 == 0)
|
|
check("official replaced (status=provisional, candidate-status stripped)",
|
|
("candidate-status" not in _cf_res2) and _cf_res2.get("status") == "provisional")
|
|
CC.OFFICIAL = _orig_off
|
|
_sh16.rmtree(_cf_d2, ignore_errors=True)
|
|
|
|
sys.exit(1 if failed else 0)
|