647 lines
31 KiB
Python
647 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
"""context_package.py — 모든 spawn이 거치는 단일 context-package 컴파일러+validator (WP-5, finding #4).
|
|
|
|
cascade 커맨드(/decide·/ground·/design·/spec·/build)와 /run-wave가 워커(subagent)를 띄우기 전,
|
|
필수 context-package를 **이 한 곳에서** 만들고(compile) 검증(validate)한다. 예전엔 /run-wave만
|
|
패키지를 만들고 cascade는 objective/boundaries를 즉석 추론했다 — 그 구멍을 닫는다.
|
|
|
|
강제 필드 = context-package-spec.yaml `required-fields` + P0 신규 필수 5개:
|
|
workspace · target-repo · acceptance-tests · non-goals · evidence-plan
|
|
(spec을 SoT로 삼되, 신규 5개가 spec에 없는 구버전이어도 이 파일이 하한을 보장한다.)
|
|
|
|
--------------------------------------------------------------------------- #
|
|
Usage:
|
|
# VALIDATE mode — 패키지가 모든 필수 필드를 갖췄는지 검사. 통과 exit 0 / 위반 exit 1(목록 출력).
|
|
context_package.py <package.pkg.yaml>
|
|
echo '{"package_path": "<path>"}' | context_package.py # stdin JSON
|
|
cat pkg.yaml | context_package.py # stdin YAML
|
|
|
|
# COMPILE mode — 스켈레톤 패키지를 발급(derivable은 채우고 나머지는 placeholder). 경로를 출력.
|
|
context_package.py --compile --workflow WF --task T --role ROLE
|
|
[--mode divergent|converge] [--tier light|standard|heavy]
|
|
[--lens LENS] [--target-repo REPO] [--objective TEXT]
|
|
-> <state_dir>/context-packages/<WF>/<ROLE>-<UTCstamp>.pkg.yaml 를 만들고 경로를 stdout에 출력.
|
|
(남은 placeholder 목록은 stderr로 안내 — 채운 뒤 validate가 통과해야 spawn 가능.)
|
|
workspace 미설정 시(C1) WorkspaceNotSetError를 잡아 명확히 중단(traceback 없이 exit 1).
|
|
|
|
importable: from context_package import validate; violations = validate(pkg_dict)
|
|
-> validate는 예외를 던지지 않고 위반 사유 문자열 리스트를 반환한다(doctor/CI/테스트용).
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
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))
|
|
sys.path.insert(0, HERE)
|
|
import _workspace as W # noqa: E402 (경로 함수는 lazy 호출 — import는 workspace 없이 안전)
|
|
|
|
try: # P3-B: method-selection 게이트(공용 policy engine). 미가용 시 degrade(신규 게이트, 회귀 방지).
|
|
import method_contracts as _MC # noqa: E402
|
|
except Exception: # noqa: BLE001
|
|
_MC = None
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 계약: 필수 필드 (context-package-spec.yaml required-fields + P0 신규 5개)
|
|
# --------------------------------------------------------------------------- #
|
|
SPEC_REQUIRED = [
|
|
"workflow-id", "task-id", "mode", "tier", "target-role-agent",
|
|
"objective", "output-format", "allowed-tools", "task-boundaries",
|
|
"must-read", "inherited-decisions", "expected-output", "token-budget",
|
|
]
|
|
P0_REQUIRED = [
|
|
"workspace", "target-repo", "acceptance-tests", "non-goals", "evidence-plan",
|
|
]
|
|
|
|
# finding #13: required-fields 를 context-package-spec.yaml(SoT)에서 읽는다 — 예전엔 여기 하드코딩만
|
|
# 있어 spec을 고쳐도 검증이 안 바뀌었다(dead SSOT). spec을 소비하되 P0 하한(신규 5개)은 항상 보장
|
|
# (구버전 spec 방어). 파일 부재/파싱실패면 내장 하드코딩으로 폴백.
|
|
SPEC_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "context-package-spec.yaml")
|
|
|
|
|
|
def _required_from_spec():
|
|
try:
|
|
rf = ((yaml.safe_load(open(SPEC_PATH, encoding="utf-8")) or {})
|
|
.get("context-package-spec") or {}).get("required-fields")
|
|
if isinstance(rf, list) and rf:
|
|
return [str(x) for x in rf]
|
|
except (OSError, yaml.YAMLError):
|
|
pass
|
|
return None
|
|
|
|
|
|
# spec 소비 + P0 하한 보장(중복 제거, 순서 보존).
|
|
REQUIRED_FIELDS = list(dict.fromkeys(
|
|
(_required_from_spec() or (SPEC_REQUIRED + P0_REQUIRED)) + P0_REQUIRED))
|
|
|
|
# 키는 존재해야 하나 빈 컨테이너([])는 허용(예: 첫 cascade 단계는 상속 결정이 없다).
|
|
EMPTY_OK = {"inherited-decisions"}
|
|
|
|
MODES = {"divergent", "converge"}
|
|
TIERS = {"light", "standard", "heavy"}
|
|
|
|
# finding #17: tier -> model/effort. governance-tiers.yaml(model-effort-by-tier)이 SoT.
|
|
# 파일 부재/파싱 실패 시 이 내장 기본값으로 폴백(spawn 계약이 하드페일하지 않도록).
|
|
GOVTIERS = os.path.join(ROOT, "org-os", "06-agent-work", "governance-tiers.yaml")
|
|
_DEFAULT_ME = {
|
|
"light": {"model": "sonnet", "effort": "low"},
|
|
"standard": {"model": "sonnet", "effort": "medium"},
|
|
"heavy": {"model": "opus", "effort": "high"},
|
|
}
|
|
_EFFORT_ORDER = ["low", "medium", "high", "xhigh", "max"]
|
|
SYNTH_LEADS = {"consult-em", "doc-lead", "des-director"} # 교차 종합/방향 선택은 고 effort
|
|
|
|
|
|
def model_effort_for_tier(tier, role=None):
|
|
"""선언된 tier(+역할)로 {model, effort}를 SoT에서 해석. synthesis-lead면 effort 한 단계 상향."""
|
|
me_map = dict(_DEFAULT_ME)
|
|
bump = {}
|
|
try:
|
|
gt = (yaml.safe_load(open(GOVTIERS, encoding="utf-8")) or {}).get("governance-tiers") or {}
|
|
loaded = gt.get("model-effort-by-tier") or {}
|
|
for t in TIERS:
|
|
if isinstance(loaded.get(t), dict):
|
|
me_map[t] = {"model": loaded[t].get("model", me_map[t]["model"]),
|
|
"effort": loaded[t].get("effort", me_map[t]["effort"])}
|
|
if isinstance(loaded.get("synthesis-lead-bump"), dict):
|
|
bump = loaded["synthesis-lead-bump"]
|
|
except (OSError, yaml.YAMLError):
|
|
pass
|
|
me = dict(me_map.get(tier or "standard", me_map["standard"]))
|
|
if role and str(role).lower() in SYNTH_LEADS:
|
|
# synthesis-lead: effort를 bump 값과 현재 중 더 높은 쪽으로.
|
|
target = bump.get("effort", "high")
|
|
cur_i = _EFFORT_ORDER.index(me["effort"]) if me["effort"] in _EFFORT_ORDER else 1
|
|
tgt_i = _EFFORT_ORDER.index(target) if target in _EFFORT_ORDER else cur_i
|
|
me["effort"] = _EFFORT_ORDER[max(cur_i, tgt_i)]
|
|
return me
|
|
|
|
|
|
def _is_empty(v):
|
|
if v is None:
|
|
return True
|
|
if isinstance(v, str):
|
|
return v.strip() == ""
|
|
if isinstance(v, (list, tuple, dict, set)):
|
|
return len(v) == 0
|
|
return False
|
|
|
|
|
|
def _resolve_field(pkg, field):
|
|
"""(present, value) 반환. non-goals는 1급 top-level이 canonical이나,
|
|
기존 스키마의 (collaboration.)shared-constraints.non-goals 위치도 인정한다."""
|
|
if field == "non-goals":
|
|
if "non-goals" in pkg:
|
|
return True, pkg.get("non-goals")
|
|
for holder in (pkg.get("shared-constraints"),
|
|
(pkg.get("collaboration") or {}).get("shared-constraints")
|
|
if isinstance(pkg.get("collaboration"), dict) else None):
|
|
if isinstance(holder, dict) and "non-goals" in holder:
|
|
return True, holder.get("non-goals")
|
|
return False, None
|
|
return (field in pkg), pkg.get(field)
|
|
|
|
|
|
# finding P0-2: validator가 '비어있는가'만 보면 `must-read: none`·`acceptance-tests: none`·
|
|
# `evidence-plan: self-assertion`·`allowed-tools: ALL`·`token-budget: unlimited`·`target-repo: repo`
|
|
# 같은 placeholder 문자열이 그대로 통과한다(리뷰 재현). 아래 sentinel 값들을 '채워지지 않은 것'으로
|
|
# 간주해 거부한다. '비어있음'과 달리 이건 의미 검사(semantic) — 실제 계약을 우회하는 위장값 차단.
|
|
_SENTINELS_COMMON = {"none", "n/a", "na", "-", "tbd", "todo", "fill", "placeholder", "xxx", "..."}
|
|
_FIELD_SENTINELS = {
|
|
"must-read": _SENTINELS_COMMON,
|
|
"acceptance-tests": _SENTINELS_COMMON | {"self-assertion", "self-report", "trust-me"},
|
|
"evidence-plan": _SENTINELS_COMMON | {"self-assertion", "self-report", "trust-me"},
|
|
"target-repo": _SENTINELS_COMMON | {"repo", "the-repo", "some-repo"},
|
|
"objective": _SENTINELS_COMMON,
|
|
"task-boundaries": _SENTINELS_COMMON,
|
|
"non-goals": _SENTINELS_COMMON,
|
|
}
|
|
|
|
|
|
def _sentinel_hit(field, val):
|
|
"""field 값이 '위장 placeholder'면 그 값을 반환(아니면 None). 문자열/리스트 모두 검사."""
|
|
bad = _FIELD_SENTINELS.get(field)
|
|
if not bad:
|
|
return None
|
|
def is_bad(x):
|
|
return isinstance(x, str) and x.strip().lower() in bad
|
|
if is_bad(val):
|
|
return val
|
|
if isinstance(val, (list, tuple)):
|
|
for item in val:
|
|
if is_bad(item):
|
|
return item
|
|
return None
|
|
|
|
|
|
def _looks_like_path(s):
|
|
return isinstance(s, str) and ("/" in s or s.endswith((".md", ".yaml", ".yml", ".py",
|
|
".json", ".ts", ".tsx", ".js", ".txt")))
|
|
|
|
|
|
def _semantic_errors(pkg):
|
|
"""placeholder 위장값·미실존 참조를 잡는 의미 검사(finding P0-2). 파일시스템 검사는
|
|
workspace 해석 가능할 때만(테스트가 가짜 dict를 넘겨도 크래시하지 않도록)."""
|
|
errs = []
|
|
# Public-facing art direction is a judgment-heavy workflow even when the
|
|
# implementation slice is small. A light/low run optimizes plumbing and
|
|
# commonly falls back to generic component priors, so the design-direction
|
|
# plan has a hard standard-tier floor.
|
|
ledger = {}
|
|
try:
|
|
wf = str(pkg.get("workflow-id") or "")
|
|
ledger_path = os.path.join(W.state_dir(), wf, "workflow.yaml")
|
|
ledger = yaml.safe_load(open(ledger_path, encoding="utf-8")) or {}
|
|
is_design_direction = ledger.get("plan") == "design-direction"
|
|
except Exception:
|
|
role_hint = str(pkg.get("target-role-agent") or "").lower()
|
|
task_hint = str(pkg.get("task-id") or "").lower()
|
|
is_design_direction = (role_hint in {"des-director", "des-visual"}
|
|
and (task_hint.startswith("direction-")
|
|
or "divergence" in task_hint
|
|
or task_hint.startswith("review-")))
|
|
if is_design_direction and pkg.get("tier") == "light":
|
|
errs.append("design-direction은 tier=light 금지 — 최소 standard(발산/비교/선택의 시각 판단 예산 보장)")
|
|
if ledger.get("plan") == "cascade" and ledger.get("stage") == "discovery":
|
|
lens = str(pkg.get("assigned-lens") or "").upper()
|
|
role_id = str(pkg.get("target-role-agent") or "").upper()
|
|
if not lens:
|
|
errs.append("cascade discovery context-package는 assigned-lens 필수(--lens LENS-*)")
|
|
else:
|
|
try:
|
|
from orgos.planning.lens_policy import role_can_carry_lens
|
|
if not role_can_carry_lens(role_id, lens):
|
|
errs.append(f"target role {role_id}는 registry상 assigned-lens {lens}를 carry할 수 없다")
|
|
except Exception as exc:
|
|
errs.append(f"assigned-lens registry 검증 실패(fail-closed): {exc}")
|
|
# 1) sentinel 위장값
|
|
for field in ("must-read", "acceptance-tests", "evidence-plan", "target-repo",
|
|
"objective", "task-boundaries", "non-goals"):
|
|
present, val = _resolve_field(pkg, field)
|
|
if present:
|
|
hit = _sentinel_hit(field, val)
|
|
if hit is not None:
|
|
errs.append(f"필수 필드가 위장 placeholder 값이다: {field}={hit!r} (실제 값으로 채워라)")
|
|
# 2) allowed-tools: 'ALL'/'*' 금지(최소권한 — tool-permission-matrix 기반 명시 목록이어야)
|
|
at = pkg.get("allowed-tools")
|
|
at_items = at if isinstance(at, (list, tuple)) else [at]
|
|
for t in at_items:
|
|
if isinstance(t, str) and t.strip().lower() in ("all", "*", "everything"):
|
|
errs.append("allowed-tools에 'ALL'/'*' 금지 — 최소권한 명시 목록이어야 한다(tool-permission-matrix).")
|
|
break
|
|
# 2b) task allowlist는 concrete agent card의 정적 tool profile보다 넓을 수 없다.
|
|
role_for_tools = str(pkg.get("target-role-agent") or "").strip().lower()
|
|
if role_for_tools and not role_for_tools.startswith("fam-") and isinstance(at, (list, tuple)):
|
|
card_path = os.path.join(ROOT, ".claude", "agents", role_for_tools + ".md")
|
|
try:
|
|
card_text = open(card_path, encoding="utf-8").read()
|
|
card_meta = yaml.safe_load(card_text.split("---\n", 2)[1]) or {}
|
|
card_tools = card_meta.get("tools") or []
|
|
if isinstance(card_tools, str):
|
|
card_tools = [item.strip() for item in card_tools.split(",") if item.strip()]
|
|
excess = sorted({str(item) for item in at} - {str(item) for item in card_tools})
|
|
if excess:
|
|
errs.append(f"allowed-tools가 agent 정적 profile보다 넓다: {excess}")
|
|
except Exception:
|
|
pass # card existence error is reported below
|
|
# 3) token-budget: 'unlimited' 금지
|
|
tb = pkg.get("token-budget")
|
|
if isinstance(tb, str) and tb.strip().lower() in ("unlimited", "inf", "infinite"):
|
|
errs.append("token-budget에 'unlimited' 금지 — 구체 상한(max-input/output-tokens)이어야 한다.")
|
|
# 4) target-role-agent: 생성된 에이전트 카드가 실존해야 한다(가짜 역할 차단)
|
|
role = pkg.get("target-role-agent")
|
|
if isinstance(role, str) and role.strip() and not _sentinel_hit("objective", role):
|
|
if role.strip().lower().startswith("fam-"):
|
|
errs.append(
|
|
"family는 candidate metadata이며 실행 agent가 아니다. "
|
|
"state_engine.py resolve-family로 concrete role을 선택한 뒤 역할별 context-package를 컴파일하라.")
|
|
# 카드 존재는 case-무관(F2): role-id 는 대문자(DES-DIRECTOR), 카드 파일은 소문자(des-director.md).
|
|
# verbatim + lowercased 둘 다 확인 — validate_report 의 case-무관 대조(13d39a2)와 정합.
|
|
card = os.path.join(ROOT, ".claude", "agents", role.strip() + ".md")
|
|
card_ci = os.path.join(ROOT, ".claude", "agents", role.strip().lower() + ".md")
|
|
if not (os.path.exists(card) or os.path.exists(card_ci)):
|
|
errs.append(f"target-role-agent '{role}' 의 에이전트 카드(.claude/agents/{role}.md)가 없다 — 실존 역할이어야 spawn 가능.")
|
|
# 5) must-read: 경로처럼 보이는 항목은 실존해야 한다(workspace 해석 가능 시)
|
|
present, mr = _resolve_field(pkg, "must-read")
|
|
if present and isinstance(mr, (list, tuple)):
|
|
base = None
|
|
try:
|
|
base = W.work_root()
|
|
except Exception:
|
|
base = None
|
|
context_ids = set()
|
|
for entry in mr:
|
|
uri = entry.get("uri") if isinstance(entry, dict) else entry
|
|
if isinstance(entry, dict):
|
|
context_id = str(entry.get("context-id") or "").strip()
|
|
if not context_id:
|
|
errs.append("must-read object에는 context-id가 필요하다(context usage 계측 결속)")
|
|
elif context_id in context_ids:
|
|
errs.append(f"must-read context-id 중복: {context_id}")
|
|
context_ids.add(context_id)
|
|
if not str(entry.get("reason") or "").strip():
|
|
errs.append(f"must-read {context_id or uri!r}에 selection reason이 없다")
|
|
if _looks_like_path(uri) and base:
|
|
cand = uri if os.path.isabs(uri) else None
|
|
# workspace 상대 또는 repo 상대 둘 다 시도
|
|
for root in ([uri] if os.path.isabs(uri) else [os.path.join(base, uri), os.path.join(ROOT, uri)]):
|
|
if os.path.exists(root):
|
|
cand = root
|
|
break
|
|
if not cand:
|
|
errs.append(f"must-read 경로가 실존하지 않는다: {uri} (읽을 수 없는 파일은 계약 위반)")
|
|
allowed_paths = pkg.get("allowed-paths")
|
|
if allowed_paths is not None and not isinstance(allowed_paths, list):
|
|
errs.append("allowed-paths는 경로 목록이어야 한다")
|
|
return errs
|
|
|
|
|
|
def validate(pkg):
|
|
"""C-style validator. 위반 사유 리스트 반환(빈 리스트=통과). 예외를 던지지 않는다."""
|
|
errors = []
|
|
if not isinstance(pkg, dict):
|
|
return ["context-package가 dict가 아님(YAML 파싱 실패/형식 오류) — 통과 불가."]
|
|
for field in REQUIRED_FIELDS:
|
|
present, val = _resolve_field(pkg, field)
|
|
if not present:
|
|
errors.append(f"필수 필드 누락: {field}")
|
|
continue
|
|
if field in EMPTY_OK:
|
|
continue
|
|
if _is_empty(val):
|
|
errors.append(f"필수 필드 비어있음(placeholder 미충전): {field}")
|
|
mode = pkg.get("mode")
|
|
if mode is not None and not _is_empty(mode) and mode not in MODES:
|
|
errors.append(f"mode는 {sorted(MODES)} 중 하나여야 한다(got {mode!r}).")
|
|
tier = pkg.get("tier")
|
|
if tier is not None and not _is_empty(tier) and tier not in TIERS:
|
|
errors.append(f"tier는 {sorted(TIERS)} 중 하나여야 한다(got {tier!r}).")
|
|
errors.extend(_semantic_errors(pkg))
|
|
errors.extend(_method_selection_errors(pkg))
|
|
errors.extend(_handoff_spawn_errors(pkg))
|
|
return errors
|
|
|
|
|
|
def _method_selection_errors(pkg):
|
|
"""P3-B: worker spawn 시 method-selection 강제(standard/heavy 필수·light 유일후보).
|
|
|
|
v2 계약 역할에만 발효 — 현행 v1 역할은 policy engine 이 [] 반환(회귀 없음). fam-*는
|
|
실행 target 자체가 semantic gate에서 거부된다. policy engine 미가용/미대상이면 조용히 통과.
|
|
"""
|
|
if _MC is None:
|
|
return []
|
|
role = str(pkg.get("target-role-agent") or "")
|
|
if not role or role.startswith("fam-"):
|
|
return []
|
|
sel_cp = {
|
|
"role-id": role.upper(), # agent 이름(소문자)→role-id(대문자, role-working-methods 키)
|
|
"tier": pkg.get("tier"),
|
|
"method-selection": pkg.get("method-selection"),
|
|
}
|
|
try:
|
|
return _MC.validate_method_selection(sel_cp)
|
|
except Exception as exc: # noqa: BLE001
|
|
if pkg.get("tier") in ("standard", "heavy"):
|
|
return [f"method-selection policy 평가 실패(fail-closed): {exc}"]
|
|
return []
|
|
|
|
|
|
def _is_independent_design_review(pkg, ledger=None):
|
|
"""Return True only for the design-direction critique's isolated lens tasks.
|
|
|
|
``ledger`` is injectable so contract tests can exercise the boundary without a live workspace.
|
|
Runtime callers omit it and the canonical workflow ledger is loaded fail-closed.
|
|
"""
|
|
if not isinstance(pkg, dict) or not str(pkg.get("task-id") or "").startswith("review-"):
|
|
return False
|
|
if ledger is None:
|
|
workflow = str(pkg.get("workflow-id") or "")
|
|
if not workflow:
|
|
return False
|
|
try:
|
|
path = os.path.join(W.state_dir(), workflow, "workflow.yaml")
|
|
ledger = yaml.safe_load(open(path, encoding="utf-8")) or {}
|
|
except Exception:
|
|
return False
|
|
return (isinstance(ledger, dict)
|
|
and ledger.get("plan") == "design-direction"
|
|
and ledger.get("stage") == "design-direction-critique")
|
|
|
|
|
|
def _handoff_spawn_errors(pkg):
|
|
"""P3-B: consumer worker spawn 시 required-inputs handoff 엣지 충족 강제(hard=both-active).
|
|
|
|
아티팩트 수락은 acceptance_log 로 근사(from-role 최신 accepted 존재). v2 required-inputs 가
|
|
없으면 no-op. producer/consumer 둘 다 active 인 엣지만 차단, 한쪽 draft 는 debt(비차단).
|
|
"""
|
|
if _MC is None:
|
|
return []
|
|
# design-direction critique의 7개 lens는 workflow contract에서
|
|
# ``method-binding.mode=independent-review``로 명시된다. 이 리뷰들은 역할의 평상시 생산
|
|
# method를 실행하는 것이 아니라 이미 hash-bound 된 winner-prototype을 독립 감사한다.
|
|
# 따라서 DES-PLATFORM/tokenize의 design-decision-record, ENG-FE의 api-contract처럼
|
|
# *생산 method*에 필요한 upstream handoff를 여기서 강제하면 frontend-only prototype
|
|
# 리뷰가 가짜 API/백엔드 산출물을 만들기 전에는 시작조차 못 한다. 공식 critique stage와
|
|
# review-* task에만 좁게 면제하고 method-selection·context-package·report schema gate는
|
|
# 그대로 유지한다. 다른 stage/task의 handoff는 기존대로 fail-closed다.
|
|
if _is_independent_design_review(pkg):
|
|
return []
|
|
role = str(pkg.get("target-role-agent") or "")
|
|
if not role or role.startswith("fam-"):
|
|
return []
|
|
mid = (pkg.get("method-selection") or {}).get("method-id")
|
|
if not mid:
|
|
return []
|
|
wf = pkg.get("workflow-id")
|
|
try:
|
|
import acceptance_log as _AL
|
|
except Exception: # noqa: BLE001
|
|
_AL = None
|
|
|
|
def _accepted(edge):
|
|
frm = (edge.get("from") or {}).get("role-id")
|
|
kind = edge.get("artifact-type")
|
|
return bool(_AL and _AL.latest_accepted_artifact(
|
|
wf, producer_role=frm, artifact_kind=kind))
|
|
|
|
def _present(edge):
|
|
return _accepted(edge) # 근사: 수락된 upstream 산출물 존재 = present(Phase5 golden 에서 정밀화)
|
|
|
|
try:
|
|
errors, _debts = _MC.handoff_violations(role.upper(), mid, present=_present, accepted=_accepted)
|
|
return errors
|
|
except Exception as exc: # noqa: BLE001
|
|
if pkg.get("tier") in ("standard", "heavy"):
|
|
return [f"method handoff policy 평가 실패(fail-closed): {exc}"]
|
|
return []
|
|
|
|
|
|
def sha256_file(path):
|
|
"""파일의 sha256 hex(없으면 None). spawn ref binding 에 쓴다."""
|
|
try:
|
|
h = hashlib.sha256()
|
|
with open(path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def package_ref(path):
|
|
"""spawn 프롬프트에 넣을 바인딩 참조 2줄을 반환(검증 통과한 패키지의 경로+해시).
|
|
guard_tools 의 Agent 게이트가 이 참조를 파싱해 파일 실존·해시 일치·validate 통과를 확인한다."""
|
|
rel = os.path.relpath(path, ROOT)
|
|
h = sha256_file(path)
|
|
return f"context-package: {rel}\ncontext-package-sha256: {h}"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# COMPILE — 스켈레톤 패키지 발급 (new_report.py의 mint-and-print 패턴을 미러)
|
|
# --------------------------------------------------------------------------- #
|
|
def _infer_method_selection(role):
|
|
"""P3-B cutover: v2 계약 역할의 method-selection 을 컴파일 시점에 채운다(auto-infer 금지 정책의
|
|
준수 경로 — 컴파일러가 '유일 method 는 명시적으로 선택'해 패키지에 박아 넣는다).
|
|
단일 method → {method-id}. 복수(DES-DIRECTOR/DES-PROD/CONSULT-EM/DOC-LEAD) → placeholder(caller 선택).
|
|
v1/family-metadata/미가용 → None(필드 생략, 회귀 없음)."""
|
|
if _MC is None or not role or role.startswith("fam-"):
|
|
return None
|
|
try:
|
|
mids = _MC.role_method_ids(role.upper())
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
if not mids:
|
|
return None # v1 역할 — 계약 미적용
|
|
if len(mids) == 1:
|
|
return {"method-id": mids[0]}
|
|
return {"method-id": "", "candidates": mids} # FILL: 복수 method — applies-when 으로 골라 채운다
|
|
|
|
|
|
def _build_skeleton(workflow, task, role, mode, tier, lens, target_repo, objective, ws_name):
|
|
"""derivable 필드는 채우고, task-specific 필드는 빈 placeholder로 둔다.
|
|
빈 필드는 validate가 '비어있음'으로 잡아 채우도록 강제한다(스켈레톤은 일부러 미통과)."""
|
|
me = model_effort_for_tier(tier, role)
|
|
sk = {
|
|
"workflow-id": workflow,
|
|
"task-id": task,
|
|
"mode": mode,
|
|
"tier": tier,
|
|
# finding #17: 선언된 tier로 해석된 추론 강도. Orchestrator가 Agent 도구 model/effort 인자로 넘긴다.
|
|
# 에이전트 frontmatter의 model: inherit == spawn이 고른 이 model 사용.
|
|
"model": me["model"],
|
|
"effort": me["effort"],
|
|
"assigned-lens": lens or None,
|
|
"target-role-agent": role, # .claude/agents/<concrete-role>.md (fam-* 금지)
|
|
"workspace": ws_name, # 해석된 워크스페이스(.orgos-workspace / ORGOS_WORKSPACE)
|
|
"target-repo": target_repo or "", # FILL: 워커가 작업할 대상 저장소/폴더(company-context projects[].id)
|
|
"objective": objective or "", # FILL: 이 워커가 달성할 한 문장
|
|
"output-format":
|
|
"report.yaml (report-header BLUF로 시작) + 최종 메시지로 report-path + 1줄 bottom-line 반환",
|
|
"allowed-tools": [], # FILL: tool-permission-matrix.yaml 최소권한
|
|
"allowed-paths": [], # FILL: write/edit 가능한 repo/workspace 경로(최소 범위)
|
|
"task-boundaries": "", # FILL: 이 워커가 다루는/다루지 않는 범위
|
|
"non-goals": [], # FILL: 명시적 비목표(shared-constraints.non-goals의 1급 승격)
|
|
"must-read": [], # FILL: [{context-id, uri, reason, expected-use, estimated-tokens}]
|
|
"inherited-decisions": [], # 상속 결정(없으면 빈 목록 허용)
|
|
"acceptance-tests": [], # FILL: task-specific 수용 기준/명령(검증 가능)
|
|
"evidence-plan": [], # FILL: E4/E5 주장을 뒷받침할 receipt 계획(command/artifact)
|
|
"expected-output": {
|
|
"report-header": {
|
|
"bottom-line": "",
|
|
"decision-needed": {"needed": False, "approver": None},
|
|
"confidence": {"value": "Med", "derived-from": "evidence"},
|
|
"risks": [],
|
|
"evidence": [],
|
|
},
|
|
},
|
|
"token-budget": {
|
|
"max-input-tokens": 60000,
|
|
"max-output-tokens": 8000,
|
|
"max-tool-calls": 80,
|
|
"max-attempts": 2,
|
|
"max-cumulative-tokens": 136000,
|
|
"rehydration": {
|
|
"max-summary-input": 16000,
|
|
"full-read-triggers": ["tier-heavy", "critical-claim", "dissent-present",
|
|
"confidence-low", "reviewer-request", "projection-conflict"],
|
|
},
|
|
},
|
|
"rehydration-policy": (
|
|
"full-originals" if tier == "heavy" else
|
|
"structured-projection-only" if tier == "light" else
|
|
"projection-first-expand-on-trigger"
|
|
),
|
|
}
|
|
ms = _infer_method_selection(role)
|
|
if ms is not None: # P3-B: v2 역할이면 method-selection 을 컴파일 시점에 심는다
|
|
sk["method-selection"] = ms
|
|
return sk
|
|
|
|
|
|
def mint_path(workflow, role):
|
|
wdir = os.path.join(W.state_dir(), "context-packages", workflow)
|
|
os.makedirs(wdir, exist_ok=True)
|
|
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
base = f"{role}-{stamp}"
|
|
path = os.path.join(wdir, base + ".pkg.yaml")
|
|
n = 1
|
|
while os.path.exists(path): # never overwrite
|
|
path = os.path.join(wdir, f"{base}-{n}.pkg.yaml")
|
|
n += 1
|
|
return path
|
|
|
|
|
|
def compile_package(workflow, task, role, mode, tier, lens, target_repo, objective):
|
|
"""스켈레톤 파일을 만들고 (path, remaining_violations) 반환. workspace 미설정 시 WorkspaceNotSetError."""
|
|
ws_name = W.workspace_name() # C1: 미설정이면 WorkspaceNotSetError → caller가 잡는다.
|
|
path = mint_path(workflow, role)
|
|
skeleton = _build_skeleton(
|
|
workflow, task, role, mode, tier, lens, target_repo, objective, ws_name)
|
|
remaining = validate(skeleton)
|
|
header = [
|
|
"# context-package — 단일 컴파일러(context_package.py --compile)가 발급한 스켈레톤.",
|
|
f"# spawn 전: 아래 placeholder를 채우고 `python3 .claude/hooks/context_package.py {os.path.relpath(path, ROOT)}` 가",
|
|
"# exit 0(통과)이어야 워커를 spawn한다. 스키마·규칙: org-os/06-agent-work/context-package-spec.yaml",
|
|
"# 아직 채워야 하는 필드(발급 시점):",
|
|
]
|
|
header += [f"# - {v}" for v in remaining] or ["# (없음 — 모두 채워짐)"]
|
|
body = yaml.safe_dump(skeleton, sort_keys=False, allow_unicode=True,
|
|
default_flow_style=False)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write("\n".join(header) + "\n" + body)
|
|
return path, remaining
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# CLI
|
|
# --------------------------------------------------------------------------- #
|
|
def _parse_compile_args(args):
|
|
vals = {"workflow": None, "task": None, "role": None, "mode": "divergent",
|
|
"tier": "standard", "lens": None, "target-repo": None, "objective": None}
|
|
i = 0
|
|
while i < len(args):
|
|
a = args[i]
|
|
key = a[2:] if a.startswith("--") else None
|
|
if key in vals and i + 1 < len(args):
|
|
vals[key] = args[i + 1]; i += 2
|
|
else:
|
|
i += 1
|
|
return vals
|
|
|
|
|
|
def _load_package():
|
|
"""argv[1] 파일 또는 stdin(JSON {"package_path":...} | 원시 YAML)에서 패키지 dict를 로드."""
|
|
if len(sys.argv) > 1 and os.path.exists(sys.argv[1]):
|
|
with open(sys.argv[1], encoding="utf-8") as f:
|
|
return yaml.safe_load(f), sys.argv[1]
|
|
data = sys.stdin.read().strip() if not sys.stdin.isatty() else ""
|
|
if not data:
|
|
return None, None
|
|
try:
|
|
payload = json.loads(data)
|
|
if isinstance(payload, dict) and payload.get("package_path"):
|
|
p = payload["package_path"]
|
|
if os.path.exists(p):
|
|
with open(p, encoding="utf-8") as f:
|
|
return yaml.safe_load(f), p
|
|
return None, p
|
|
return payload, "<stdin>"
|
|
except json.JSONDecodeError:
|
|
return yaml.safe_load(data), "<stdin>"
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
if "--compile" in args:
|
|
v = _parse_compile_args([a for a in args if a != "--compile"])
|
|
if not (v["workflow"] and v["task"] and v["role"]):
|
|
sys.stderr.write(
|
|
"usage: context_package.py --compile --workflow WF --task T --role ROLE "
|
|
"[--mode ..] [--tier ..] [--lens ..] [--target-repo ..] [--objective ..]\n")
|
|
sys.exit(2)
|
|
try:
|
|
path, remaining = compile_package(
|
|
v["workflow"], v["task"], v["role"], v["mode"], v["tier"],
|
|
v["lens"], v["target-repo"], v["objective"])
|
|
except W.WorkspaceNotSetError as e:
|
|
sys.stderr.write(
|
|
"[context_package] 워크스페이스 미설정 — context-package를 발급할 수 없습니다.\n"
|
|
f" {e}\n")
|
|
sys.exit(1)
|
|
print(os.path.relpath(path, ROOT))
|
|
if remaining:
|
|
sys.stderr.write(
|
|
"[context_package] 채워야 할 필드(spawn 전 필수):\n"
|
|
+ "\n".join(f" - {r}" for r in remaining) + "\n")
|
|
sys.exit(0)
|
|
|
|
# VALIDATE mode
|
|
pkg, path = _load_package()
|
|
if pkg is None:
|
|
sys.stderr.write(
|
|
"usage: context_package.py <package.pkg.yaml> | "
|
|
"context_package.py --compile ... (--help 참고: 파일 헤더)\n")
|
|
sys.exit(2)
|
|
errors = validate(pkg)
|
|
label = path or "<package>"
|
|
if errors:
|
|
sys.stderr.write(
|
|
f"[context_package] INVALID {label} — 워커 spawn 금지:\n"
|
|
+ "\n".join(f" - {e}" for e in errors) + "\n")
|
|
sys.exit(1)
|
|
print(f"OK context-package valid: {label}")
|
|
# 바인딩 참조를 함께 출력한다(finding P0-2): 오케스트레이터가 이 2줄을 spawn 프롬프트
|
|
# 최상단에 넣어야 guard_tools 의 Agent 게이트가 spawn 을 허용한다(경로+해시+validate 일치).
|
|
if path and os.path.exists(path):
|
|
print(package_ref(path))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|