369 lines
19 KiB
Python
369 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""Real standard-tier ENG-BE checkpoint submissions do not require a self SHA."""
|
|
import hashlib
|
|
import json
|
|
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", "method-checkpoint-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
|
|
|
|
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):
|
|
return hashlib.sha256(open(path, "rb").read()).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 report(wf, artifact_id, kind, stage, payload, method_execution):
|
|
directory = os.path.join(WS, "completion-records", wf)
|
|
os.makedirs(directory, exist_ok=True)
|
|
path = os.path.join(directory, f"{artifact_id}.report.yaml")
|
|
value = {
|
|
"report-type": "workflow-artifact", "artifact-kind": kind, "artifact-version": 1,
|
|
"tier": "standard",
|
|
"identity": {"artifact-id": artifact_id, "workflow-id": wf, "stage": stage,
|
|
"producer-role-id": "ENG-BE"},
|
|
"payload": payload, "method-execution": method_execution,
|
|
"report-header": {
|
|
"bottom-line": f"{kind} checkpoint", "decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med", "derived-from": "evidence"}, "risks": [],
|
|
"evidence": [{"source-uri": "README.md", "grade": "E3"}],
|
|
},
|
|
}
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump(value, fh, allow_unicode=True, sort_keys=False)
|
|
return path
|
|
|
|
|
|
wf = "eng-checkpoint"
|
|
SE.init_ledger(wf, tier="standard")
|
|
check("open spec stage", open_stage(wf, "spec"))
|
|
|
|
# Seed a real immutable trusted design basis; the behavior under test begins at ENG-BE submit.
|
|
basis_dir = os.path.join(WS, "completion-records", wf)
|
|
os.makedirs(basis_dir, exist_ok=True)
|
|
basis_path = os.path.join(basis_dir, "design-basis.report.yaml")
|
|
with open(basis_path, "w", encoding="utf-8") as fh:
|
|
fh.write("design-basis: true\n")
|
|
basis_sha = sha(basis_path)
|
|
SE._atomic_event_transaction(wf, artifact_event={
|
|
"artifact-event-id": "fixture-design-basis", "event-type": "artifact-submitted",
|
|
"artifact-id": "design-basis", "report-id": "design-basis", "workflow-id": wf,
|
|
"artifact-kind": "overall-design", "design-type": "overall-design", "artifact-version": 1,
|
|
"stage": "design", "producer-role-id": "ARCH-SOLUTION",
|
|
"path": os.path.relpath(basis_path, WS), "artifact-sha256": basis_sha,
|
|
"report-sha256": basis_sha, "effective-at": SE._now(),
|
|
})
|
|
|
|
profile = MC.resolve_method_profile("ENG-BE", "backend-implementation")
|
|
contract_sha = MC.canonical_contract_hash(profile)
|
|
api_me = {
|
|
"role-id": "ENG-BE", "method-id": "backend-implementation", "contract-sha256": contract_sha,
|
|
"step-results": [{"step-id": "design-api", "status": "completed",
|
|
"output-binding": "current-artifact"}],
|
|
}
|
|
api_payload = {
|
|
"basis-artifact-id": "design-basis", "basis-artifact-sha256": basis_sha,
|
|
"summary": "contract-first API", "protocol": "HTTP", "version": "v1",
|
|
"operations": [{"operation-id": "get-item", "method": "GET", "path": "/items/{id}"}],
|
|
"schemas": {"Item": {"type": "object"}},
|
|
"errors": [{"code": "not_found", "status": 404}],
|
|
"compatibility": {"policy": "backward-compatible"},
|
|
}
|
|
api_path = report(wf, "api-v1", "api-contract", "spec", api_payload, api_me)
|
|
api_ok, api_result = SE.submit_report(wf, api_path, "ENG-BE")
|
|
check("intermediate api-contract submits at its checkpoint without future completion", api_ok, api_result)
|
|
|
|
api_event = next((item for item in SE._trusted_artifacts(wf) if item.get("artifact-id") == "api-v1"), None)
|
|
check("api checkpoint is now trusted", bool(api_event))
|
|
check("open build stage", open_stage(wf, "build"))
|
|
|
|
implementation = os.path.join(WS, "implemented.py")
|
|
with open(implementation, "w", encoding="utf-8") as fh:
|
|
fh.write("def get_item(item_id): return {'id': item_id}\n")
|
|
receipt_dir = os.path.join(WS, "evidence")
|
|
os.makedirs(receipt_dir, exist_ok=True)
|
|
with open(os.path.join(receipt_dir, "ledger.jsonl"), "a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps({
|
|
"tool_use_id": "verify-api-v1", "tool_name": "Bash", "command": "pytest",
|
|
"exit_code": 0, "workflow_id": wf, "session_id": "fixture-session",
|
|
"agent_id": "ENG-BE", "ts": SE._now(),
|
|
}) + "\n")
|
|
|
|
completion_me = {
|
|
"role-id": "ENG-BE", "method-id": "backend-implementation", "contract-sha256": contract_sha,
|
|
"step-results": [
|
|
{"step-id": "design-api", "status": "completed", "output-binding": "trusted-artifact",
|
|
"artifact-refs": [{"report-id": "api-v1", "sha256": api_event["artifact-sha256"]}]},
|
|
{"step-id": "implement-verify", "status": "completed", "output-binding": "current-artifact"},
|
|
],
|
|
"self-check-results": [{"step-id": "implement-verify", "gate-id": "contract-verified",
|
|
"verdict": "Passed", "evidence-refs": ["verify-api-v1"]}],
|
|
}
|
|
completion_payload = {
|
|
"summary": "implemented against api-v1", "source-revision": {"kind": "workspace-tree", "sha256": "b" * 64},
|
|
"primary-artifacts": [{"path": implementation, "kind": "code", "sha256": sha(implementation)}],
|
|
"acceptance-criteria-coverage": [{"criterion-id": "AC-1", "status": "Passed",
|
|
"evidence-receipt-ids": ["verify-api-v1"]}],
|
|
"verification-receipt-ids": ["verify-api-v1"], "remaining-risks": [],
|
|
}
|
|
completion_path = report(wf, "completion-v1", "completion-record", "build",
|
|
completion_payload, completion_me)
|
|
completion_ok, completion_result = SE.submit_report(wf, completion_path, "ENG-BE")
|
|
check("final completion submits with prior exact ref and current-artifact binding", completion_ok, completion_result)
|
|
|
|
self_ref = {**completion_me, "step-results": [completion_me["step-results"][0], {
|
|
"step-id": "implement-verify", "status": "completed", "output-binding": "current-artifact",
|
|
"artifact-refs": [{"report-id": "completion-v1", "sha256": "c" * 64}],
|
|
}]}
|
|
self_errors = MC.validate_method_execution(
|
|
{"role-id": "ENG-BE", "tier": "standard", "method-execution": self_ref},
|
|
current_artifact={"artifact-kind": "completion-record"})
|
|
check("self artifact-ref remains explicitly rejected", any("자기 자신" in error for error in self_errors))
|
|
|
|
|
|
def write_generic(wf_id, artifact_id, kind, producer, stage, payload, execution=None):
|
|
directory = os.path.join(WS, "completion-records", wf_id)
|
|
os.makedirs(directory, exist_ok=True)
|
|
path = os.path.join(directory, f"{artifact_id}.report.yaml")
|
|
value = {
|
|
"report-type": "workflow-artifact", "artifact-kind": kind, "artifact-version": 1,
|
|
"tier": "standard",
|
|
"identity": {"artifact-id": artifact_id, "workflow-id": wf_id, "stage": stage,
|
|
"producer-role-id": producer},
|
|
"payload": payload,
|
|
"report-header": {
|
|
"bottom-line": f"{kind} independent judgment fixture",
|
|
"decision-needed": {"needed": False},
|
|
"confidence": {"value": "Med", "derived-from": "evidence"}, "risks": [],
|
|
"evidence": [{"source-uri": "README.md", "grade": "E3"}],
|
|
},
|
|
}
|
|
if execution is not None:
|
|
value["method-execution"] = execution
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
yaml.safe_dump(value, fh, allow_unicode=True, sort_keys=False)
|
|
return path
|
|
|
|
|
|
# Independent judgment is post-submit/pre-acceptance, avoiding a second circular dependency.
|
|
judge_wf = "independent-judgment"
|
|
SE.init_ledger(judge_wf, tier="standard")
|
|
check("open design stage for independent judgment", open_stage(judge_wf, "design"))
|
|
brief_path = os.path.join(WS, "completion-records", judge_wf, "brief-1.report.yaml")
|
|
os.makedirs(os.path.dirname(brief_path), exist_ok=True)
|
|
with open(brief_path, "w", encoding="utf-8") as fh:
|
|
fh.write("brief: concrete signals\n")
|
|
brief_sha = sha(brief_path)
|
|
SE._atomic_event_transaction(judge_wf, artifact_event={
|
|
"artifact-event-id": "fixture-method-brief", "event-type": "artifact-submitted",
|
|
"artifact-id": "brief-1", "report-id": "brief-1", "workflow-id": judge_wf,
|
|
"artifact-kind": "design-brief", "design-type": "design-brief", "artifact-version": 1,
|
|
"stage": "design", "producer-role-id": "DES-PROD",
|
|
"path": os.path.relpath(brief_path, WS), "artifact-sha256": brief_sha,
|
|
"report-sha256": brief_sha, "effective-at": SE._now(),
|
|
})
|
|
des_profile = MC.resolve_method_profile("DES-PROD", "pre-direction")
|
|
des_sha = MC.canonical_contract_hash(des_profile)
|
|
constraints_execution = {
|
|
"role-id": "DES-PROD", "method-id": "pre-direction", "contract-sha256": des_sha,
|
|
"step-results": [
|
|
{"step-id": "frame-brief", "status": "completed", "output-binding": "trusted-artifact",
|
|
"artifact-refs": [{"report-id": "brief-1", "sha256": brief_sha}]},
|
|
{"step-id": "discover", "status": "completed", "output-binding": "current-artifact"},
|
|
],
|
|
"decisions": [{"decision-id": "constraint-scope", "selected-option-id": "observed",
|
|
"alternatives": [{"option-id": "observed"}, {"option-id": "assumed"}]}],
|
|
}
|
|
constraints_path = write_generic(
|
|
judge_wf, "constraints-1", "experience-constraints", "DES-PROD", "design",
|
|
{"constraints": ["operators need dense comparison"]}, constraints_execution)
|
|
constraints_ok, constraints_result = SE.submit_report(judge_wf, constraints_path, "DES-PROD")
|
|
check("artifact with independent judgment gate can be submitted first", constraints_ok, constraints_result)
|
|
target = next(item for item in SE._trusted_artifacts(judge_wf)
|
|
if item.get("artifact-id") == "constraints-1")
|
|
premature_ok, premature_error = SE.review_artifact(
|
|
judge_wf, constraints_path, "accepted", "HUMAN-001")
|
|
check("target cannot be Accepted before typed independent judgment",
|
|
not premature_ok and "UX-RESEARCHER" in str(premature_error), premature_error)
|
|
judgment_path = write_generic(judge_wf, "judgment-1", "method-judgment-review",
|
|
"UX-RESEARCHER", "design", {
|
|
"method-role-id": "DES-PROD", "method-id": "pre-direction", "step-id": "discover",
|
|
"gate-id": "evidence-grounded", "criterion": "constraints are grounded in user signals",
|
|
"reviewed-artifact-id": "constraints-1", "reviewed-artifact-sha256": target["artifact-sha256"],
|
|
"reviewer-role-id": "UX-RESEARCHER", "verdict": "Passed", "findings": [],
|
|
})
|
|
judgment_ok, judgment_error = SE.submit_report(judge_wf, judgment_path, "UX-RESEARCHER")
|
|
check("independent reviewer submits exact typed judgment", judgment_ok, judgment_error)
|
|
accepted_ok, accepted_error = SE.review_artifact(
|
|
judge_wf, constraints_path, "accepted", "HUMAN-001")
|
|
check("target becomes acceptable after exact typed judgment", accepted_ok, accepted_error)
|
|
|
|
|
|
# Canonical stage envelopes explicitly aggregate craft-method outputs instead of
|
|
# pretending their artifact-kind is itself a method required-output.
|
|
core_wf = "core-aggregate-binding"
|
|
SE.init_ledger(core_wf, tier="standard")
|
|
check("open discovery stage for aggregate binding", open_stage(core_wf, "discovery"))
|
|
strategy_profile = MC.resolve_method_profile("STR-ANALYST", "strategy-analysis")
|
|
strategy_execution = {
|
|
"role-id": "STR-ANALYST", "method-id": "strategy-analysis",
|
|
"contract-sha256": MC.canonical_contract_hash(strategy_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": ["ground-evidence"]}],
|
|
"decisions": [{"decision-id": "strategic-options", "selected-option-id": "defer",
|
|
"alternatives": [{"option-id": "guided"}, {"option-id": "reference"}]}],
|
|
}
|
|
ground_payload = {
|
|
"problem-structure": {"core-question": "how should deep technical learning be structured"},
|
|
"analysis-synthesis": {"insight": "guided practice and reference depth solve different jobs"},
|
|
"evidence": [{"source": "README.md", "claim": "local product intent"}],
|
|
"options": [
|
|
{"id": "guided", "problem": "passive reading", "tradeoffs": ["authoring cost"],
|
|
"evidence-refs": ["README.md"]},
|
|
{"id": "reference", "problem": "fragmented lookup", "tradeoffs": ["less guidance"],
|
|
"evidence-refs": ["README.md"]},
|
|
],
|
|
# This suite exercises aggregate method binding, not the discovery gate.
|
|
# Exact source/context verification is covered by test_grounding_lens_coverage.py.
|
|
"source-contributions": [
|
|
{"report-id": f"fixture-source-{index}", "report-ref": f"fixture/source-{index}.report.yaml",
|
|
"report-sha256": str(index) * 64, "producer-role-id": "EXEC-CEO",
|
|
"context-package-ref": f"fixture/context-{index}.yaml",
|
|
"context-package-sha256": str(index + 3) * 64,
|
|
"assigned-lens": lens, "producer-run-id": f"fixture-run-{index}"}
|
|
for index, lens in enumerate(
|
|
["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH", "LENS-FINANCE", "LENS-CONTRARIAN"], 1)
|
|
],
|
|
"lens-coverage": {"required-min": 5,
|
|
"covered": ["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH", "LENS-FINANCE", "LENS-CONTRARIAN"],
|
|
"contrarian-report-id": "fixture-source-5"},
|
|
}
|
|
ground_path = write_generic(core_wf, "ground-core", "grounding-package", "STR-ANALYST",
|
|
"discovery", ground_payload, strategy_execution)
|
|
ground_ok, ground_result = SE.submit_report(core_wf, ground_path, "OPS-ORCH")
|
|
check("grounding aggregate submits with all method outputs embedded", ground_ok, ground_result)
|
|
|
|
bad_ground_execution = dict(strategy_execution)
|
|
bad_ground_execution["step-results"] = [dict(item) for item in strategy_execution["step-results"]]
|
|
bad_ground_execution["step-results"][0]["output-binding"] = "trusted-artifact"
|
|
bad_ground_path = write_generic(core_wf, "ground-bad-binding", "grounding-package",
|
|
"STR-ANALYST", "discovery", ground_payload, bad_ground_execution)
|
|
bad_ground_ok, bad_ground_result = SE.submit_report(
|
|
core_wf, bad_ground_path, "OPS-ORCH")
|
|
check("aggregate rejects a self-contained step falsely bound as prior artifact",
|
|
not bad_ground_ok and "current-artifact" in str(bad_ground_result), bad_ground_result)
|
|
|
|
ground_event = next(item for item in SE._trusted_artifacts(core_wf)
|
|
if item.get("artifact-id") == "ground-core")
|
|
check("open decide stage for aggregate binding", open_stage(core_wf, "decide"))
|
|
decision_profile = MC.resolve_method_profile("EXEC-CEO", "decide-direction")
|
|
decision_execution = {
|
|
"role-id": "EXEC-CEO", "method-id": "decide-direction",
|
|
"contract-sha256": MC.canonical_contract_hash(decision_profile),
|
|
"step-results": [
|
|
{"step-id": "read-evidence", "status": "completed",
|
|
"output-binding": "current-artifact"},
|
|
{"step-id": "evaluate-options", "status": "completed",
|
|
"output-binding": "current-artifact"},
|
|
{"step-id": "converge-decision", "status": "completed",
|
|
"output-binding": "current-artifact"},
|
|
],
|
|
"self-check-results": [{"step-id": "converge-decision", "gate-id": "single-direction",
|
|
"verdict": "Passed", "evidence-refs": ["ground-core"]}],
|
|
"decisions": [{"decision-id": "product-direction", "selected-option-id": "guided",
|
|
"alternatives": [{"option-id": "guided"}, {"option-id": "reference"}]}],
|
|
}
|
|
decision_payload = {
|
|
"recommendation": "guided learning with deep reference layers",
|
|
"selected-option-id": "guided", "evaluation-criteria": ["learning-depth", "usability"],
|
|
"option-evaluations": [
|
|
{"option-id": "guided", "scores": {"learning-depth": 5},
|
|
"evidence-refs": ["ground-core"]},
|
|
{"option-id": "reference", "scores": {"learning-depth": 3},
|
|
"evidence-refs": ["ground-core"]},
|
|
],
|
|
"tradeoffs": ["higher content-model complexity"], "dissent": [],
|
|
"kill-criteria": ["learners cannot complete a guided path"],
|
|
"revisit-conditions": ["reference usage dominates guided usage"],
|
|
"evidence-refs": ["ground-core"],
|
|
}
|
|
decision_path = write_generic(core_wf, "decision-core", "executive-decision-packet",
|
|
"EXEC-CEO", "decide", decision_payload, decision_execution)
|
|
decision_ok, decision_result = SE.submit_report(core_wf, decision_path, "OPS-ORCH")
|
|
check("decision aggregate submits with all method outputs embedded", decision_ok, decision_result)
|
|
|
|
decision_event = next(item for item in SE._trusted_artifacts(core_wf)
|
|
if item.get("artifact-id") == "decision-core")
|
|
check("open design stage for trusted synthesis", open_stage(core_wf, "design"))
|
|
overall_payload = {
|
|
"basis-artifact-id": "decision-core",
|
|
"basis-artifact-sha256": decision_event["artifact-sha256"],
|
|
"source-artifact-refs": [
|
|
{"artifact-id": "ground-core", "artifact-sha256": ground_event["artifact-sha256"]},
|
|
{"artifact-id": "decision-core", "artifact-sha256": decision_event["artifact-sha256"]},
|
|
],
|
|
"summary": "layered technical learning architecture",
|
|
"architecture-boundaries": ["content", "learning-path", "progress"],
|
|
"quality-attributes": ["explainability", "accessibility"],
|
|
"decisions": [{"id": "ADR-1", "choice": "content-first modular architecture"}],
|
|
"dependencies": [], "compatibility-assumptions": ["modern evergreen browser"],
|
|
}
|
|
overall_path = write_generic(core_wf, "overall-core", "overall-design", "ARCH-SOLUTION",
|
|
"design", overall_payload)
|
|
overall_ok, overall_result = SE.submit_report(core_wf, overall_path, "OPS-ORCH")
|
|
check("stage synthesis submits only with exact trusted source refs", overall_ok, overall_result)
|
|
|
|
bad_overall_payload = dict(overall_payload)
|
|
bad_overall_payload["source-artifact-refs"] = [
|
|
{"artifact-id": "ground-core", "artifact-sha256": "f" * 64}]
|
|
bad_overall_path = write_generic(core_wf, "overall-bad-source", "overall-design",
|
|
"ARCH-SOLUTION", "design", bad_overall_payload)
|
|
bad_overall_ok, bad_overall_result = SE.submit_report(
|
|
core_wf, bad_overall_path, "OPS-ORCH")
|
|
check("stage synthesis rejects untrusted source hash",
|
|
not bad_overall_ok and "trusted registry" in str(bad_overall_result), bad_overall_result)
|
|
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
raise SystemExit(1 if failed else 0)
|