525 lines
24 KiB
Python
525 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""Trusted workflow-artifact envelope helpers.
|
|
|
|
The state engine must never accept gate facts as CLI arguments. This module
|
|
loads a report snapshot, validates its identity and schema, and derives every
|
|
materialized fact from the immutable bytes that were submitted.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from functools import lru_cache
|
|
|
|
import yaml
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.environ.get("CLAUDE_PROJECT_DIR") or os.path.dirname(os.path.dirname(HERE))
|
|
CONTRACT_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "workflow-contracts.yaml")
|
|
VOCABULARY_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "artifact-type-vocabulary.yaml")
|
|
ARTIFACT_REGISTRY_PATH = os.path.join(
|
|
ROOT, "org-os", "06-agent-work", "generated", "artifact-registry.yaml")
|
|
|
|
_GRADE = {f"E{i}": i for i in range(6)}
|
|
|
|
|
|
@lru_cache(maxsize=1)
|
|
def load_contract():
|
|
try:
|
|
with open(CONTRACT_PATH, encoding="utf-8") as fh:
|
|
contract = (yaml.safe_load(fh) or {}).get("workflow-contracts", {}) or {}
|
|
with open(ARTIFACT_REGISTRY_PATH, encoding="utf-8") as fh:
|
|
registry = (yaml.safe_load(fh) or {}).get("artifact-registry", {}) or {}
|
|
definitions = registry.get("artifact-kinds")
|
|
if not isinstance(definitions, dict) or not definitions:
|
|
raise ValueError("generated artifact registry is empty")
|
|
return {**contract, "artifact-kinds": definitions}
|
|
except Exception as exc:
|
|
raise RuntimeError(
|
|
"artifact registry unavailable; run compile_artifact_registry.py and preflight --check: "
|
|
f"{exc}") from exc
|
|
|
|
|
|
def absolute_path(path):
|
|
if not path:
|
|
return None
|
|
path = os.path.expandvars(str(path))
|
|
if os.path.isabs(path):
|
|
return os.path.normpath(path)
|
|
candidates = [os.path.join(ROOT, path), os.path.join(os.getcwd(), path)]
|
|
try:
|
|
import _workspace as workspace
|
|
candidates.insert(0, os.path.join(workspace.work_root(), path))
|
|
except Exception:
|
|
pass
|
|
for candidate in candidates:
|
|
if os.path.exists(candidate):
|
|
return os.path.normpath(candidate)
|
|
return os.path.normpath(candidates[0])
|
|
|
|
|
|
def sha256_file(path):
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as fh:
|
|
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def load_report(path):
|
|
ap = absolute_path(path)
|
|
if not ap or not os.path.isfile(ap):
|
|
raise ValueError(f"report 파일 없음: {path}")
|
|
try:
|
|
with open(ap, encoding="utf-8") as fh:
|
|
report = yaml.safe_load(fh)
|
|
except Exception as exc:
|
|
raise ValueError(f"report YAML 로드 실패: {exc}") from exc
|
|
if not isinstance(report, dict):
|
|
raise ValueError("report는 YAML object여야 한다")
|
|
return ap, report
|
|
|
|
|
|
def identity(report):
|
|
ident = report.get("identity") if isinstance(report.get("identity"), dict) else {}
|
|
return {
|
|
"artifact-id": ident.get("artifact-id") or report.get("artifact-id") or report.get("report-id"),
|
|
"workflow-id": ident.get("workflow-id") or report.get("workflow-id"),
|
|
"stage": ident.get("stage") or report.get("stage"),
|
|
"producer-role-id": ident.get("producer-role-id") or report.get("producer-role-id") or report.get("role-id"),
|
|
}
|
|
|
|
|
|
def payload(report):
|
|
value = report.get("payload")
|
|
return value if isinstance(value, dict) else report
|
|
|
|
|
|
def artifact_kind(report):
|
|
kind = report.get("artifact-kind")
|
|
if isinstance(kind, str) and kind.strip():
|
|
return kind.strip()
|
|
# Narrow read-compatibility for unambiguous legacy snapshots. Ambiguous
|
|
# report-type=decision/design/spec/work must declare artifact-kind.
|
|
return {
|
|
"completion": "completion-record",
|
|
"blocked": "blocked-report",
|
|
}.get(str(report.get("report-type") or "").strip())
|
|
|
|
|
|
def option_set(report):
|
|
body = payload(report)
|
|
opts = body.get("options")
|
|
if not isinstance(opts, list):
|
|
opts = body.get("option-set")
|
|
return list(opts) if isinstance(opts, list) else []
|
|
|
|
|
|
def max_evidence_grade(report):
|
|
header = report.get("report-header") or {}
|
|
evidence = header.get("evidence") if isinstance(header, dict) else []
|
|
grades = [e.get("grade") for e in (evidence or []) if isinstance(e, dict)]
|
|
valid = [g for g in grades if g in _GRADE]
|
|
return max(valid, key=lambda g: _GRADE[g]) if valid else None
|
|
|
|
|
|
def _ledger_path_for(report_path):
|
|
if report_path:
|
|
ap = os.path.abspath(report_path)
|
|
parts = ap.split(os.sep)
|
|
if "completion-records" in parts:
|
|
idx = len(parts) - 1 - parts[::-1].index("completion-records")
|
|
work_root = os.sep.join(parts[:idx]) or os.sep
|
|
return os.path.join(work_root, "evidence", "ledger.jsonl")
|
|
try:
|
|
import _workspace as workspace
|
|
return os.path.join(workspace.evidence_dir(), "ledger.jsonl")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def load_receipts(report_path=None):
|
|
"""Read typed tool receipts. Malformed lines are ignored here and surfaced by doctor."""
|
|
path = _ledger_path_for(report_path)
|
|
if not path or not os.path.isfile(path):
|
|
return []
|
|
receipts = []
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
for line in fh:
|
|
try:
|
|
value = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
if isinstance(value, dict):
|
|
receipts.append(value)
|
|
except Exception:
|
|
return []
|
|
return receipts
|
|
|
|
|
|
def receipt_id(receipt):
|
|
return receipt.get("receipt_id") or receipt.get("tool_use_id")
|
|
|
|
|
|
def validate_receipt_ids(report_path, workflow_id, ids, *, require_success=True,
|
|
since=None, require_context=True):
|
|
"""Resolve receipt ids to the same workflow/run context.
|
|
|
|
Unscoped workspace-wide receipts are intentionally rejected. ``since`` is an
|
|
ISO-8601 stage epoch; a prior run cannot be reused for a later quality gate.
|
|
"""
|
|
requested = [str(value) for value in (ids or []) if str(value or "").strip()]
|
|
by_id = {str(receipt_id(r)): r for r in load_receipts(report_path) if receipt_id(r)}
|
|
errors, resolved = [], []
|
|
for rid in requested:
|
|
receipt = by_id.get(rid)
|
|
if not receipt:
|
|
errors.append(f"evidence receipt 없음: {rid}")
|
|
continue
|
|
if str(receipt.get("workflow_id") or "") != str(workflow_id):
|
|
errors.append(f"receipt {rid}: workflow_id가 현재 workflow와 불일치/누락")
|
|
continue
|
|
if require_context and (not receipt.get("session_id") or not receipt.get("agent_id")):
|
|
errors.append(f"receipt {rid}: session_id/agent_id 결속 누락")
|
|
continue
|
|
if since and str(receipt.get("ts") or "") < str(since):
|
|
errors.append(f"receipt {rid}: 현재 stage 시작 이전의 stale receipt")
|
|
continue
|
|
if require_success:
|
|
exit_code = receipt.get("exit_code")
|
|
artifact_sha = receipt.get("artifact_sha256")
|
|
if exit_code not in (0, "0") and not artifact_sha:
|
|
errors.append(f"receipt {rid}: 성공 exit_code=0 또는 artifact_sha256 증거 없음")
|
|
continue
|
|
resolved.append(receipt)
|
|
if len(set(requested)) != len(requested):
|
|
errors.append("evidence receipt id 중복")
|
|
return errors, resolved
|
|
|
|
|
|
def _completion_errors(report, report_path):
|
|
body = payload(report)
|
|
ident = identity(report)
|
|
errors = []
|
|
for index, artifact in enumerate(body.get("primary-artifacts") or []):
|
|
if not isinstance(artifact, dict):
|
|
continue
|
|
path = absolute_path(artifact.get("path"))
|
|
if not path or not os.path.isfile(path):
|
|
errors.append(f"completion primary-artifacts[{index}] 파일 없음: {artifact.get('path')}")
|
|
continue
|
|
live_sha = sha256_file(path)
|
|
if artifact.get("sha256") != live_sha:
|
|
errors.append(f"completion primary-artifacts[{index}] live SHA 불일치")
|
|
receipt_ids = list(body.get("verification-receipt-ids") or [])
|
|
for coverage in body.get("acceptance-criteria-coverage") or []:
|
|
if isinstance(coverage, dict):
|
|
receipt_ids.extend(coverage.get("evidence-receipt-ids") or [])
|
|
if coverage.get("status") == "Failed":
|
|
errors.append(f"completion criterion {coverage.get('criterion-id')}가 Failed")
|
|
receipt_errors, _ = validate_receipt_ids(
|
|
report_path, ident.get("workflow-id"), list(dict.fromkeys(receipt_ids)),
|
|
require_success=True, require_context=True,
|
|
)
|
|
errors.extend(f"completion {error}" for error in receipt_errors)
|
|
return errors
|
|
|
|
|
|
def _data_execution_errors(report, kind, report_path):
|
|
body = payload(report)
|
|
ident = identity(report)
|
|
errors = []
|
|
receipt_ids = body.get("evidence-receipt-ids") or []
|
|
if kind == "metrics-analysis":
|
|
snapshot = body.get("dataset-snapshot") or {}
|
|
snapshot_path = absolute_path(snapshot.get("path"))
|
|
if not snapshot_path or not os.path.isfile(snapshot_path):
|
|
errors.append("metrics-analysis dataset-snapshot.path 파일 없음")
|
|
elif sha256_file(snapshot_path) != snapshot.get("sha256"):
|
|
errors.append("metrics-analysis dataset-snapshot live SHA 불일치")
|
|
receipt_ids = (body.get("analysis-run") or {}).get("evidence-receipt-ids") or []
|
|
receipt_errors, _ = validate_receipt_ids(
|
|
report_path, ident.get("workflow-id"), receipt_ids,
|
|
require_success=True, require_context=True,
|
|
)
|
|
errors.extend(f"{kind} {error}" for error in receipt_errors)
|
|
return errors
|
|
|
|
|
|
def _experience_contract_errors(body, kind):
|
|
errors = []
|
|
if kind == "competitive-experience-benchmark":
|
|
references = body.get("references") or []
|
|
names = [str(item.get("name") or "").strip() for item in references if isinstance(item, dict)]
|
|
if len(names) != len(set(names)):
|
|
errors.append("competitive benchmark reference name 중복")
|
|
classes = {item.get("class") for item in references if isinstance(item, dict)}
|
|
if len(classes & {"direct", "adjacent", "substitute"}) < 2:
|
|
errors.append("competitive benchmark는 direct/adjacent/substitute 중 최소 2개 class를 혼합해야 한다")
|
|
now = datetime.now(timezone.utc)
|
|
for index, item in enumerate(references):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
try:
|
|
captured = datetime.fromisoformat(str(item.get("captured-at") or "").replace("Z", "+00:00"))
|
|
if captured.tzinfo is None:
|
|
captured = captured.replace(tzinfo=timezone.utc)
|
|
age_days = (now - captured.astimezone(timezone.utc)).days
|
|
if age_days < -1 or age_days > 365:
|
|
errors.append(f"competitive benchmark references[{index}] 캡처 freshness 365일 초과/미래")
|
|
except Exception:
|
|
errors.append(f"competitive benchmark references[{index}].captured-at ISO date-time 오류")
|
|
screenshots = item.get("screenshots") or {}
|
|
for viewport in ("desktop", "mobile"):
|
|
for shot_index, shot in enumerate(screenshots.get(viewport) or []):
|
|
if not isinstance(shot, dict):
|
|
continue
|
|
path = absolute_path(shot.get("path"))
|
|
if not path or not os.path.isfile(path):
|
|
errors.append(
|
|
f"competitive benchmark references[{index}].screenshots.{viewport}[{shot_index}] 파일 없음")
|
|
elif sha256_file(path) != shot.get("sha256"):
|
|
errors.append(
|
|
f"competitive benchmark references[{index}].screenshots.{viewport}[{shot_index}] live SHA 불일치")
|
|
if kind == "design-system-release":
|
|
path = absolute_path(body.get("source-ref"))
|
|
if not path or not os.path.isfile(path):
|
|
errors.append("design-system-release source-ref 파일 없음")
|
|
elif sha256_file(path) != body.get("source-sha256"):
|
|
errors.append("design-system-release source-ref live SHA 불일치")
|
|
if kind == "first-draft-evaluation":
|
|
output_path = absolute_path(body.get("output-ref"))
|
|
if not output_path or not os.path.isfile(output_path):
|
|
errors.append("first-draft-evaluation output-ref 파일 없음")
|
|
elif sha256_file(output_path) != body.get("output-sha256"):
|
|
errors.append("first-draft-evaluation output-ref live SHA 불일치")
|
|
for viewport in ("desktop", "mobile"):
|
|
evidence = (body.get("screenshots") or {}).get(viewport) or {}
|
|
screenshot_path = absolute_path(evidence.get("path"))
|
|
if not screenshot_path or not os.path.isfile(screenshot_path):
|
|
errors.append(f"first-draft-evaluation screenshot {viewport} 파일 없음")
|
|
continue
|
|
if sha256_file(screenshot_path) != evidence.get("sha256"):
|
|
errors.append(f"first-draft-evaluation screenshot {viewport} live SHA 불일치")
|
|
continue
|
|
try:
|
|
with open(screenshot_path, "rb") as handle:
|
|
signature = handle.read(8)
|
|
if signature != b"\x89PNG\r\n\x1a\n" or os.path.getsize(screenshot_path) <= 1000:
|
|
errors.append(f"first-draft-evaluation screenshot {viewport} 실제 PNG 증거 아님")
|
|
except OSError:
|
|
errors.append(f"first-draft-evaluation screenshot {viewport} 읽기 실패")
|
|
return errors
|
|
|
|
|
|
def _semantic_errors(report, kind, report_path=None):
|
|
body = payload(report)
|
|
errors = []
|
|
if not kind:
|
|
return ["artifact-kind 누락: submit-report는 산출물 종류를 report 본문에서만 받는다"]
|
|
defs = load_contract().get("artifact-kinds", {}) or {}
|
|
definition = defs.get(kind)
|
|
if not isinstance(definition, dict):
|
|
return [f"등록되지 않은 artifact-kind: {kind}"]
|
|
enforcement = load_contract().get("payload-enforcement", {}) or {}
|
|
tiered = enforcement.get("tiered-kinds", {}) or {}
|
|
tier = report.get("tier") or "standard"
|
|
light_contract = tiered.get(kind) if kind in tiered else None
|
|
strict_payload = light_contract is None or tier in set(enforcement.get("strict-tiers") or [])
|
|
required_fields = (definition.get("required-payload-fields", [])
|
|
if strict_payload else light_contract)
|
|
allow_empty = set(definition.get("allow-empty-payload-fields") or [])
|
|
for field in required_fields or []:
|
|
if field not in body or (body.get(field) in (None, "", []) and field not in allow_empty):
|
|
errors.append(f"artifact-kind={kind}: payload 필수 필드 누락/빈값: {field}")
|
|
min_options = definition.get("option-count-min")
|
|
if min_options is not None and len(option_set(report)) < int(min_options):
|
|
errors.append(f"artifact-kind={kind}: options는 최소 {min_options}개여야 한다")
|
|
if kind == "blocked-report" and not body.get("resume-condition"):
|
|
errors.append("artifact-kind=blocked-report: resume-condition 필수")
|
|
if kind == "decision-brief":
|
|
try:
|
|
from orgos.planning.lens_policy import candidate_family_errors
|
|
errors.extend(candidate_family_errors(
|
|
body.get("candidate-families"),
|
|
tier=str(body.get("tier") or report.get("tier") or "standard"),
|
|
mode=str(body.get("mode") or "converge"),
|
|
enforce_lens_floor=True,
|
|
))
|
|
except Exception as exc:
|
|
errors.append(f"decision-brief candidate-family policy 평가 실패(fail-closed): {exc}")
|
|
if kind == "workload-profile":
|
|
try:
|
|
from orgos.planning.coverage_model import CAPABILITY_ROLE_HINTS
|
|
requested = {str(value).strip().lower()
|
|
for value in body.get("required-capabilities", []) or []
|
|
if str(value or "").strip()}
|
|
unknown = sorted(requested - set(CAPABILITY_ROLE_HINTS))
|
|
if unknown:
|
|
errors.append(f"workload-profile required-capabilities 미등록: {unknown}")
|
|
except Exception as exc:
|
|
errors.append(f"workload-profile capability policy 평가 실패(fail-closed): {exc}")
|
|
if kind == "competitive-market-grounding":
|
|
entries = body.get("competitors-and-substitutes") or []
|
|
names = [str(item.get("name") or "").strip().lower()
|
|
for item in entries if isinstance(item, dict)]
|
|
if len(names) != len(set(names)):
|
|
errors.append("competitive-market-grounding competitor/substitute name 중복")
|
|
types = {item.get("type") for item in entries if isinstance(item, dict)}
|
|
if not {"competitor", "substitute"}.issubset(types):
|
|
errors.append("competitive-market-grounding은 named competitor와 substitute를 각각 포함해야 한다")
|
|
if kind == "completion-record":
|
|
errors.extend(_completion_errors(report, report_path))
|
|
if kind == "venture-validation":
|
|
expected_gates = {
|
|
"problem-intensity", "competition-alternatives", "willingness-to-pay",
|
|
"revenue-unit-economics", "tech-feasibility-moat", "operability",
|
|
"distribution", "founder-fit", "kill-criteria",
|
|
}
|
|
seen_option_ids = set()
|
|
for index, option in enumerate(body.get("option-evaluations") or []):
|
|
if not isinstance(option, dict):
|
|
continue # JSON Schema가 구조 오류를 보고한다.
|
|
option_id = str(option.get("id") or "").strip()
|
|
if option_id in seen_option_ids:
|
|
errors.append(f"venture-validation option-evaluations[{index}] 중복 option id: {option_id}")
|
|
elif option_id:
|
|
seen_option_ids.add(option_id)
|
|
results = option.get("validation-results") or []
|
|
gate_counts = {}
|
|
for result_index, result in enumerate(results):
|
|
if not isinstance(result, dict):
|
|
continue
|
|
result_option_id = str(result.get("option-id") or "").strip()
|
|
if option_id and result_option_id != option_id:
|
|
errors.append(
|
|
f"venture-validation option '{option_id}' validation-results[{result_index}] "
|
|
f"option-id 불일치: {result_option_id!r}")
|
|
gate = str(result.get("gate") or "").strip()
|
|
if gate:
|
|
gate_counts[gate] = gate_counts.get(gate, 0) + 1
|
|
missing = sorted(expected_gates - set(gate_counts))
|
|
duplicate = sorted(gate for gate, count in gate_counts.items() if count > 1)
|
|
unexpected = sorted(set(gate_counts) - expected_gates)
|
|
if missing:
|
|
errors.append(f"venture-validation option '{option_id}' 9-gate 누락: {missing}")
|
|
if duplicate:
|
|
errors.append(f"venture-validation option '{option_id}' gate 중복: {duplicate}")
|
|
if unexpected:
|
|
errors.append(f"venture-validation option '{option_id}' 미등록 gate: {unexpected}")
|
|
if kind in ("metrics-analysis", "data-pipeline", "bigdata-pipeline"):
|
|
errors.extend(_data_execution_errors(report, kind, report_path))
|
|
errors.extend(_experience_contract_errors(body, kind))
|
|
schema_ref = (definition.get("payload-schema-ref") if strict_payload else None)
|
|
schema_ref = schema_ref or load_contract().get("default-payload-schema-ref")
|
|
if schema_ref:
|
|
schema_path = os.path.join(ROOT, ".claude", "schemas", schema_ref)
|
|
try:
|
|
import json
|
|
import jsonschema
|
|
with open(schema_path, encoding="utf-8") as fh:
|
|
schema = json.load(fh)
|
|
for err in jsonschema.Draft7Validator(schema).iter_errors(body):
|
|
loc = "/".join(str(item) for item in err.path) or "payload"
|
|
errors.append(f"artifact-kind={kind} {loc}: {err.message}")
|
|
except Exception as exc:
|
|
errors.append(f"artifact-kind={kind} payload schema 검증 실패: {exc}")
|
|
return errors
|
|
|
|
|
|
def validate_snapshot(path, expected_workflow=None):
|
|
"""Return a trusted artifact record or raise ValueError.
|
|
|
|
Validation binds the immutable report bytes to its in-document identity;
|
|
no caller-supplied kind/id/count/grade participates in derivation.
|
|
"""
|
|
ap, report = load_report(path)
|
|
kind = artifact_kind(report)
|
|
ident = identity(report)
|
|
current_artifact = {
|
|
"artifact-id": ident.get("artifact-id"),
|
|
"artifact-kind": kind,
|
|
"artifact-sha256": sha256_file(ap),
|
|
}
|
|
try:
|
|
import validate_report
|
|
validation_errors = validate_report.validate(
|
|
report, report_path=ap, current_artifact=current_artifact)
|
|
except Exception as exc:
|
|
raise ValueError(f"report validator 실행 실패: {exc}") from exc
|
|
validation_errors = list(validation_errors or []) + _semantic_errors(report, kind, ap)
|
|
for key in ("artifact-id", "workflow-id", "producer-role-id"):
|
|
if not str(ident.get(key) or "").strip():
|
|
validation_errors.append(f"identity.{key} 누락")
|
|
declared_stages = allowed_stages(kind)
|
|
if declared_stages and ident.get("stage") not in declared_stages:
|
|
validation_errors.append(
|
|
f"artifact-kind={kind}는 stage {sorted(declared_stages)} output이다"
|
|
f"(report stage={ident.get('stage')})"
|
|
)
|
|
if expected_workflow is not None and str(ident.get("workflow-id")) != str(expected_workflow):
|
|
validation_errors.append(
|
|
f"workflow-id 불일치(report={ident.get('workflow-id')}, command={expected_workflow})"
|
|
)
|
|
if validation_errors:
|
|
raise ValueError("report 계약 위반:\n" + "\n".join(f" - {e}" for e in validation_errors[:20]))
|
|
rel = ap
|
|
try:
|
|
import _workspace as workspace
|
|
work_root = os.path.abspath(workspace.work_root())
|
|
if ap.startswith(work_root + os.sep):
|
|
rel = os.path.relpath(ap, work_root)
|
|
elif ap.startswith(os.path.abspath(ROOT) + os.sep):
|
|
rel = os.path.relpath(ap, ROOT)
|
|
except Exception:
|
|
if ap.startswith(os.path.abspath(ROOT) + os.sep):
|
|
rel = os.path.relpath(ap, ROOT)
|
|
return {
|
|
"artifact-id": str(ident["artifact-id"]),
|
|
"report-id": str(ident["artifact-id"]),
|
|
"workflow-id": str(ident["workflow-id"]),
|
|
"artifact-kind": kind,
|
|
"design-type": kind, # read compatibility; never accepted as caller input
|
|
"artifact-version": report.get("artifact-version", 1),
|
|
"stage": ident.get("stage"),
|
|
"producer-role-id": str(ident["producer-role-id"]),
|
|
"path": rel,
|
|
"artifact-sha256": sha256_file(ap),
|
|
"report-sha256": sha256_file(ap),
|
|
"option-set": option_set(report),
|
|
"max-evidence-grade": max_evidence_grade(report),
|
|
"payload": payload(report),
|
|
}
|
|
|
|
|
|
def producer_allowed(kind, role_id):
|
|
definition = (load_contract().get("artifact-kinds", {}) or {}).get(kind, {}) or {}
|
|
allowed = definition.get("producer-roles") or []
|
|
return not allowed or str(role_id).upper() in {str(x).upper() for x in allowed}
|
|
|
|
|
|
def reviewer_capability(kind):
|
|
definition = (load_contract().get("artifact-kinds", {}) or {}).get(kind, {}) or {}
|
|
return definition.get("reviewer-capability") or "artifact-reviewer"
|
|
|
|
|
|
def allowed_stages(kind):
|
|
"""Stages that declare ``kind`` as a direct or dynamic-bundle output."""
|
|
contract = load_contract()
|
|
bundles = contract.get("artifact-bundles", {}) or {}
|
|
stages = set()
|
|
for workflow in (contract.get("workflows", {}) or {}).values():
|
|
for stage_name, stage in (workflow.get("stages", {}) or {}).items():
|
|
outputs = (stage or {}).get("outputs") or {}
|
|
declared = set(outputs.get("bundle") or [])
|
|
dynamic = outputs.get("dynamic-bundle")
|
|
if dynamic:
|
|
bundle = bundles.get(dynamic, {}) or {}
|
|
declared.update(bundle.get("always") or [])
|
|
for conditional in bundle.get("conditional") or []:
|
|
declared.update(conditional.get("require") or [])
|
|
if kind in declared:
|
|
stages.add(stage_name)
|
|
return stages
|