feat(case): KERNEL-2/3 lane resolution and mode-aware admission; CATALOG-1 draft lanes
nk48 1.3 owner review (owner-review-ops-warden.json): - KERNEL-2: Executor.admit parses catalog_source and resolves descriptor.lane; refuses a missing, non-active, undelegated or ungraded lane, an owner outside owner_repo, and a mint/change effect on a lane not graded risk high. catalog-routing-acceptance still covers the whole catalog digest. - KERNEL-3: dispatch mode is read from the type's follow_through step (warden_executes/attended). The step must declare both booleans and a lane and agree with the descriptor's lane and owner. Attended: human actor, WARDEN_AGENT_ID unset, parent-session revocation scope, caller lane equal to the step lane. Owner: native delegation, owner_repo == descriptor.owner, no Warden-executed lane and no Warden-brokered mint. - CATALOG-1: draft entries keycape-attended-case-change (OW-1.3-C1), platform-secret-delivery, consumer-owner-deployment, apps-consumer-handover. Not promoted; promotion waits for owner confirmation. - Fixture catalogs and type steps in tests; published routing_lanes coverage and the workload-join count follow the four new not-applicable lanes. PARAMETER_PINS untouched (PIN-1 waits for founder acknowledgment). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 903097@bnt-lap001 Assistant-Session: aa320c8f-c3fa-46e7-8541-0b58e1e45d23
This commit is contained in:
parent
117fec9241
commit
a14376cf62
8 changed files with 363 additions and 13 deletions
|
|
@ -553,6 +553,67 @@ class Journal:
|
|||
raise
|
||||
|
||||
|
||||
def _catalog_lane(catalog_source: bytes, descriptor: Descriptor):
|
||||
"""KERNEL-2: the descriptor's lane must be an active, fitting catalog entry.
|
||||
|
||||
catalog-routing-acceptance still covers the whole catalog digest; this
|
||||
ties the one lane the descriptor names to an entry inside that digest.
|
||||
"""
|
||||
from .routing.catalog import CatalogError, _parse_entry
|
||||
|
||||
try:
|
||||
raw = yaml.safe_load(catalog_source)
|
||||
items = raw["entries"]
|
||||
if not isinstance(items, list):
|
||||
raise TypeError
|
||||
entries = {}
|
||||
for index, item in enumerate(items):
|
||||
parsed = _parse_entry(item, index)
|
||||
if parsed.id in entries:
|
||||
raise ValueError
|
||||
entries[parsed.id] = parsed
|
||||
except (yaml.YAMLError, CatalogError, KeyError, TypeError, ValueError):
|
||||
raise Refused("invalid catalog source shape") from None
|
||||
entry = entries.get(descriptor.lane)
|
||||
if entry is None:
|
||||
raise Refused("descriptor lane is not in the catalog")
|
||||
if not entry.is_active:
|
||||
raise Refused("descriptor lane is not active in the catalog")
|
||||
if entry.delegation is None:
|
||||
raise Refused("descriptor lane has no catalog delegation record")
|
||||
if not entry.is_graded:
|
||||
raise Refused("descriptor lane carries no risk grade")
|
||||
if descriptor.owner not in entry.owner_repo.split("/"):
|
||||
raise Refused("descriptor owner does not fit catalog lane owner_repo")
|
||||
if descriptor.effect in ("mint", "change") and entry.risk != "high":
|
||||
raise Refused("catalog lane risk does not cover the descriptor effect")
|
||||
return entry
|
||||
|
||||
|
||||
def _step_mode(case_type: dict, case: dict, descriptor: Descriptor) -> str:
|
||||
"""KERNEL-3: warden, attended or owner, from the type step's own text."""
|
||||
steps = [
|
||||
step
|
||||
for step in case_type.get("follow_through") or []
|
||||
if isinstance(step, dict) and step.get("id") == descriptor.step
|
||||
]
|
||||
step = steps[0] if len(steps) == 1 else {}
|
||||
executes, attended = step.get("warden_executes"), step.get("attended")
|
||||
if (
|
||||
type(executes) is not bool
|
||||
or type(attended) is not bool
|
||||
or (executes and attended)
|
||||
or not isinstance(step.get("lane"), str)
|
||||
):
|
||||
raise Refused("type step does not declare how the kernel dispatches it")
|
||||
owner = step.get("owner")
|
||||
if isinstance(owner, str) and re.fullmatch(r"\{[a-z_]+\}", owner):
|
||||
owner = (case.get("parameters") or {}).get(owner[1:-1])
|
||||
if step["lane"] != descriptor.lane or owner != descriptor.owner:
|
||||
raise Refused("type step lane or owner disagrees with descriptor")
|
||||
return "warden" if executes else "attended" if attended else "owner"
|
||||
|
||||
|
||||
class Executor:
|
||||
"""Injected fixed owner adapters and provenance verifier; none ship by default.
|
||||
|
||||
|
|
@ -650,6 +711,21 @@ class Executor:
|
|||
actor != "human" or os.getenv("WARDEN_AGENT_ID")
|
||||
):
|
||||
raise Refused("agent mint is forbidden")
|
||||
entry = _catalog_lane(catalog_source, descriptor)
|
||||
mode = _step_mode(case_type, case, descriptor)
|
||||
if mode == "attended" and (
|
||||
actor != "human"
|
||||
or os.getenv("WARDEN_AGENT_ID")
|
||||
or "parent-session" not in descriptor.revocation_scopes
|
||||
):
|
||||
raise Refused("attended step requires a human parent session")
|
||||
if mode == "owner" and (
|
||||
entry.effective_delegation.mode != "native"
|
||||
or entry.owner_repo != descriptor.owner
|
||||
or entry.warden_executes
|
||||
or descriptor.effect == "mint"
|
||||
):
|
||||
raise Refused("owner step lane is not the owner's native path")
|
||||
parameters = case.get("parameters", {})
|
||||
declared = {
|
||||
item["name"]: item
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue