Governance

Propose, admit, materialise: the admission gate, policy documents, the governed loop.

How do I let a planner invent a subgraph without letting it run one?

Three objects do the work: a NodeRegistry of node kinds an operator allows to exist, an EdgePolicy over transitions, and an AdmissionChecker that combines them. A proposal names registry keys; it never carries a node body, so there is nothing in it to run even if something wanted to. EdgePolicy's default is deny, so an empty policy admits nothing — the allow-all rule is what you write when you have not decided yet, and it is deliberately something you have to type. All five checks (registry, policy, budget, depth, acyclicity) run on every proposal rather than short-circuiting, because a planner replanning from feedback should get the whole list, not one complaint at a time.

python
# What the operator allows to exist. Absence is refusal; there is no wildcard.
registry = NodeRegistry(
    [
        NodeSpec(name="fetch", description="read a URL"),
        NodeSpec(name="summarise", description="condense text"),
    ]
)

# What the operator allows to be wired. Unmatched is deny, so say something.
policy = EdgePolicy(rules=(EdgeRule(action="allow"),))

gate = AdmissionChecker(registry=registry, edge_policy=policy)

# What a planner wants. Nodes here carry a registry *key*, never a body.
proposal = Subgraph(
    nodes=(ProposedNode(name="fetch"), ProposedNode(name="summarise")),
    edges=(
        ProposedEdge(source=START, target="fetch"),
        ProposedEdge(source="fetch", target="summarise"),
        ProposedEdge(source="summarise", target=END),
    ),
    rationale="read the page, then condense it",
)

result = gate.check(proposal)
console
status:      admitted
admitted:    True
checks run:  ['registry', 'policy', 'budget', 'depth', 'acyclicity']
worst case:  tokens=0 iterations=2 seconds=0.0

How do I prove nothing runs during a check?

Give every registered kind a factory that raises, then admit a proposal that uses it — the factory is still uncalled afterwards. NodeSpec.factory exists only so a later materialising step has somewhere to get the node body from; being in the registry is a licence to be proposed, not an invitation to run. admit is check that fails closed, raising AdmissionRejected for callers whose next line is raise. The only side effect a check has is the trace line it writes.

python
def never_call(*args, **kwargs):
    raise AssertionError("admission executed a node factory")


registry = NodeRegistry([NodeSpec(name="fetch", factory=never_call)])
gate = AdmissionChecker(
    registry=registry, edge_policy=EdgePolicy(rules=(EdgeRule(action="allow"),))
)

result = gate.check(Subgraph(nodes=(ProposedNode(name="fetch"),)))
print("admitted:", result.admitted)
print("factory still uncalled:", registry.get("fetch").factory is never_call)

# `admit` is `check` that fails closed, for callers whose next line is `raise`.
try:
    gate.admit(Subgraph(nodes=(ProposedNode(name="shell"),)))
except AdmissionRejected as exc:
    print("raised:", exc.result.status.value, exc.result.rejections[0].code)

How do I stop a planner proposing a node I never registered?

Nothing to do: an unregistered kind is the REGISTRY check's job, and the registry has no wildcard. Every rejection carries a check, a stable machine-readable code, a subject, a detail and a remedy — match on code, since detail is prose and may be reworded. kind defaults to name for the one-off case; fill in kind explicitly whenever the instance name differs, which is the whole point of a fan-out (worker_a, worker_b, worker_c all of kind edit_worker means the registry only has to allow one thing).

console
status: rejected
failed: ['registry']
[registry/unregistered_node] shell: kind 'shell' is not in the node registry allowed kinds: fetch, summarise

How do I forbid a transition?

EdgeRule is fnmatch over both endpoints, tiered deny → ask → allow. A broad deny beats a narrower allow, so there is no allowlist exception hiding inside a denial. ProposedEdge carries a source, a target and a note, and nothing else — there is deliberately no condition string, because a predicate authored by a model is model prose steering an edge. The note is never parsed or evaluated; it exists for the audit trail, and an edge is admitted or refused on its endpoints alone.

python
policy = EdgePolicy(
    rules=(
        EdgeRule(action="deny", source="*", target="deploy"),
        EdgeRule(action="allow"),
    )
)
gate = AdmissionChecker(
    registry=NodeRegistry([NodeSpec(name="build"), NodeSpec(name="deploy")]),
    edge_policy=policy,
)

How do I refuse a plan that cannot afford itself?

Register the worst case per kind with CostEstimate and hand the checker the run's live BudgetMeter; the check reads the meter, it never charges it. Costs come from the registry, never from the proposal — a ProposedNode has no field in which to claim it will be cheap. The comparison is against what is left, not the total, so the same proposal that was affordable at the start of a run is refused later in it, and the refusal happens before the first node exists rather than after the run notices it overspent.

python
# The registry declares the cost. A proposal cannot.
gate = AdmissionChecker(
    registry=NodeRegistry(
        [NodeSpec(name="research", worst_case=CostEstimate(tokens=5_000, iterations=1))]
    ),
    edge_policy=EdgePolicy(rules=(EdgeRule(action="allow"),)),
)

meter = BudgetMeter(Budget(max_tokens=20_000))
meter.charge_tokens(16_000)  # the run has already spent this much

result = gate.check(proposal, meter=meter)
console
status:    rejected
worst case 10000 tokens
remaining  4000 tokens
[budget/over_token_budget] tokens: worst case needs 10000 tokens but only 4000 remain propose fewer or cheaper nodes
meter after the check: 16000 tokens charged

How do I stop planners planning planners forever?

AdmissionLimits.max_depth bounds nesting and defaults to 1, which means flat proposals only. Subgraph.nesting_depth() and node_count() report what a proposal declares, and a nested proposal is rejected with depth/too_deep. The same flat proposal arriving from a planner already one level in is refused when you pass parent_depth=1. Subgraph.scopes() walks nesting iteratively, not recursively, so a hostile depth exhausts the limit rather than the stack.

python
nested = Subgraph(
    nodes=(
        ProposedNode(
            name="sub_planner",
            kind="planner",
            subgraph=Subgraph(nodes=(ProposedNode(name="inner", kind="fetch"),)),
        ),
    ),
    proposal_id="nested-demo",
)
result = gate.check(nested)

# Same proposal, but arriving from a planner that is already one level in.
flat = Subgraph(nodes=(ProposedNode(name="fetch"),), proposal_id="flat-demo")
deep = gate.check(flat, parent_depth=1)

How do I allow (or refuse) a loop?

Cycles are earned: require_acyclic defaults to True, so a proposal that genuinely needs a loop must be admitted by a checker configured with AdmissionLimits(require_acyclic=False). With acyclicity off, the ACYCLICITY check does not appear in checks_run at all — checks_run is what actually ran, not a fixed list.

python
strict = AdmissionChecker(registry=registry, edge_policy=policy)
print("default checker: ", strict.check(loop).status.value)

permissive = AdmissionChecker(
    registry=registry,
    edge_policy=policy,
    limits=AdmissionLimits(require_acyclic=False),
)
result = permissive.check(loop)
print("checks run:      ", [c.value for c in result.checks_run])

How do I make sure renaming a node cannot evade a denial?

The planner chooses the instance name; the operator registered the kind; every check that grants anything resolves against the kind. Renaming a denied kind to helper does not launder it, and conversely a node named deploy that is really a build is admitted, because the name borrows no permission either. The name is chosen by the untrusted side, so a rule matched against it would be a rule the governed thing controls — which is not a rule. Names survive as labels: they resolve which node an edge refers to and appear in rejections (proposed as 'helper') so an audit can find the specific edge.

console
name=deploy  kind=deploy               rejected
    policy/edge_denied: ... kind 'build' (proposed as 'build') -> kind 'deploy' (proposed as 'deploy')
name=helper  kind=deploy               rejected
    policy/edge_denied: ... kind 'build' (proposed as 'build') -> kind 'deploy' (proposed as 'helper')
name=deploy  kind=build                admitted

What does admission not check?

Arguments. ProposedNode.args are carried for a future materialiser and never inspected: SELECT count(*) FROM users and DROP TABLE users are both admitted, and no rule in this package can change that. Registering run_sql at all is a decision to trust whatever gates its arguments downstream — the tool plane's permissions (grapharc.harness.permissions) are where that belongs. Materializer takes the same position and makes it opt-in: with the default forward_args=False, NodeBuild.args is empty whatever the proposal said. What fingerprints do give you is that the two proposals hash differently, so an audit can tell which was admitted; Subgraph.fingerprint() is a content hash — identity, not semantic equality.

python
for query in ("SELECT count(*) FROM users", "DROP TABLE users"):
    proposal = Subgraph(
        nodes=(ProposedNode(name="q", kind="run_sql", args={"query": query}),),
        proposal_id="args-demo",
    )
    result = gate.check(proposal)
    print(f"{result.status.value:<9} fingerprint={result.fingerprint}  query={query!r}")

How do I attach a proposal to nodes that are already running?

Pass known_nodes as a {name: kind} mapping, not a list of names — a name alone tells the checker nothing it can evaluate a rule against, so a by-name-only endpoint yields policy/unresolved_endpoint_kind and is refused rather than assumed permitted. These kinds come from you, not from a proposal, and need not be registered: the registry is the list of kinds a planner may propose, not of kinds that may already exist. A proposed node may not reuse a live node's name — that is a separate registry/name_collides_with_existing_node rejection, because admitting it would rebind a node that is already running.

python
# By name only: the checker cannot tell what `live_step` is, so it refuses.
by_name = AdmissionChecker(
    registry=registry, edge_policy=policy, known_nodes=["live_step"]
)
print(by_name.check(proposal).rejections[0].render())

# With its kind declared, the rule can actually be evaluated.
declared = AdmissionChecker(
    registry=registry, edge_policy=policy, known_nodes={"live_step": "deploy"}
)

How do I require a human before an edge is wired?

An edge whose policy resolves to ask produces AdmissionStatus.NEEDS_APPROVAL, with result.admitted still False — a pending approval is a stop, not a slow yes. NEEDS_APPROVAL is reported only when every rejection is an approval request; one denial anywhere outranks the whole thing and the status is REJECTED.

python
gate = AdmissionChecker(
    registry=NodeRegistry([NodeSpec(name="patch"), NodeSpec(name="deploy")]),
    edge_policy=EdgePolicy(
        rules=(EdgeRule(action="ask", target="deploy"), EdgeRule(action="allow"))
    ),
)

How do I get a rejection into the audit trail?

Give the checker a TraceRecorder and every decision — admitted and rejected alike — is written as a phase="admission" event carrying status, fingerprint, origin, node count, depth, checks_run, failed_checks, worst_case and an error of check/code. The phase is "admission" deliberately and not "end": grapharc.observe.metrics counts node executions from "end" events, and reusing the phase would inflate every metric derived from the trace. Passing a RunContext as ctx= stamps the real run_id, thread_id and step number; without one the proposal's own id stands in, so the decision is still findable.

python
trace = TraceRecorder(Path(tempfile.mkdtemp()) / "trace.jsonl")
gate = AdmissionChecker(
    registry=NodeRegistry([NodeSpec(name="build"), NodeSpec(name="deploy")]),
    edge_policy=EdgePolicy(
        rules=(EdgeRule(action="deny", target="deploy"), EdgeRule(action="allow"))
    ),
    trace=trace,
)
result = gate.check(
    Subgraph(
        nodes=(ProposedNode(name="build"), ProposedNode(name="ship", kind="deploy")),
        edges=(ProposedEdge(source="build", target="ship"),),
        proposal_id="p-0001",
        origin="planner:release",
    )
)
console
{
  "run_id": "p-0001", "attempt": 1, "graph": "admission",
  "node": "admission:p-0001", "phase": "admission", "step": 0,
  "state_delta": {
    "status": "rejected", "fingerprint": "ae82f551624a29fd",
    "origin": "planner:release", "nodes": 2, "depth": 1,
    "checks_run": ["registry", "policy", "budget", "depth", "acyclicity"],
    "failed_checks": ["policy"],
    "worst_case": {"tokens": 0, "iterations": 2, "seconds": 0.0}
  },
  "error": "policy/edge_denied"
}

How do I feed a rejection back for replanning?

AdmissionResult.feedback() renders the whole rejection as text a planner can be handed verbatim, and PlannerNode.propose(task, feedback=...) takes it — round 1 is rejected on policy, round 2 is admitted. PlannerNode cannot execute what it proposes, and that is structural rather than promised: the only callables it holds are the chat model and your own state reader, and NodeSpec.factory — the only place a node body ever appears — lives on the far side of the gate. The planner re-stamps provenance: whatever the model puts in proposal_id and origin is overwritten, because provenance the proposer wrote about itself is not provenance. A bad model reply is a PlanningOutcome carrying error, not an exception, which is what lets you retry it; only BudgetExceeded is allowed through.

python
model = ScriptedChatModel(responses=[json.dumps(first), json.dumps(second)])
planner = PlannerNode(model, name="release", catalog=registry.catalog())

task = "fix the login bug and get it live"
outcome = planner.propose(task)
result = gate.check(outcome.proposal)

if not result.admitted:
    outcome = planner.propose(task, feedback=result.feedback())
    result = gate.check(outcome.proposal)

### With a real model

Swap the scripted model for a real one via get_model; nothing else changes. This snippet was not executed — it needs a subscription or an API key, and nothing in the section calls a paid backend. PlannerNode prefers the backend's structured-output path when it has one and falls back to tolerant JSON extraction from text otherwise; outcome.structured records which path produced the object, because the two have different failure modes and an audit should not have to guess.

python
from grapharc.gateway import get_model
from grapharc.planner import PlannerNode

model = get_model("claude-cli/claude-sonnet-5")  # needs a Claude Code subscription
planner = PlannerNode(model, catalog=registry.catalog())

How do I run what was admitted — and only what was admitted?

Materializer is the only thing that turns a proposal into a graph, and it takes the AdmissionResult first, because that is the authorisation — there is no entry point that accepts a bare Subgraph. "What ran is what was admitted" is a checked equality: reusing a genuine approval alongside a proposal whose only difference is its rationale raises NotAdmitted on the fingerprint, even when both share a proposal_id. A body comes only from NodeSpec.factory, operator code registered before any planning happened; the factory is called once per proposed instance with a NodeBuild (name, kind, note, proposal_id, fingerprint, and args only if you opted in). The graph is assembled with the ordinary GraphARC builder, so declared writes, typed state, the budget meter, deep-copy isolation and trace events all still apply.

python
def step_factory(build):
    """Called once per proposed instance, with a NodeBuild describing it."""

    def body(state: State) -> dict:
        return {"notes": [*state.notes, f"{build.name} ran (kind {build.kind})"]}

    body.writes = {"notes"}
    return body


materializer = Materializer(registry=registry, state_schema=State)
result = gate.check(proposal)
graph = materializer.materialize(result, proposal)  # the result first: it is the authority
console
{'notes': ['fetch ran (kind fetch)', 'summarise ran (kind summarise)']}
this AdmissionResult authorised proposal build-demo (fingerprint cba0d7416bec398d), not proposal build-demo (fingerprint f3083ceac3a28260); what runs must be what was admitted

How do I stop a node body routing somewhere it was not admitted to?

Node bodies are operator code and may route dynamically with Command(goto=...), but the materialiser confines each body to the destinations the proposal declared an edge from that node to, plus END; anything else raises UnadmittedTransition. risky is a node of the graph and does run in the legal case — reached the long way through safe; what is refused is plan skipping to it along an edge nobody admitted. So the admitted edge set bounds dynamic routing, not merely the admitted node set, and the exception is raised before the kernel resolves the destination, so the transition does not happen and that node's writes never land.

python
def router_factory(build):
    # Operator code decides where this kind routes. `router_skip` jumps ahead.
    target = "risky" if build.kind == "router_skip" else "safe"

    def body(state: State) -> Command:
        return Command(goto=target, update={"notes": [*state.notes, f"routed to {target}"]})

    body.writes = {"notes"}
    return body
console
{'notes': ['routed to safe', 'safe ran', 'risky ran']}
node 'plan' routed to 'risky', which the admitted proposal declared no edge to. Admitted destinations for 'plan': safe, END. A body may route dynamically only along an edge the gate authorised

How do I drive the whole cycle to a recorded stop?

GovernedLoop runs plan → admit → materialise → execute → observe → replan until something says stop, and the stop is always a reason. Work discovered mid-run does not bypass admission: round 7's proposal goes through the same checker, registry and edge policy as round 1's — no "already approved" path, no cached authorisation, and no way for an executing node to add topology, because a node's only channel out is the state schema and the loop's only source of topology is planner.propose. LoopResult.stop is always set, from the vocabulary goal_met, no_further_work, max_rounds, budget_exhausted, no_progress, admission_refused, planning_failed, execution_failed, human_stopped; succeeded is true only for goal_met. Budget bounds resources across the whole run (each round executes under a budget derived from what the shared meter has left, and planner tokens charge the same meter) while LoopLimits bounds the loop — max_rounds plus consecutive counters for rejections, unusable planner replies, execution failures and stalled rounds.

python
model = ScriptedChatModel(responses=[plan("triage", "deploy"), plan("triage", "patch")])
loop = GovernedLoop(
    planner=PlannerNode(model, name="incident", catalog=registry.catalog()),
    checker=gate,
    materializer=materializer,
    goal_reached=lambda state: len(state.notes) >= 2,
)

done = loop.run("investigate the incident")
print("stop:  ", done.stop.value, "|", done.detail)
print("rejections:", [r.code for r in done.rejections()])
console
stop:   goal_met | the goal check was satisfied
rounds: 2
  round 1: rejected  executed=False
  round 2: admitted  executed=True
rejections: ['edge_denied']
state:  notes=['triage ran', 'patch ran']

How do I write the policy down instead of coding it?

grapharc.policy loads a TOML document over four resource kinds — tool, node, edge, spend — and answers decisions against it via PolicyEngine.from_file. Matching is tiered, not positional: every deny is tried before every ask, and every ask before every allow; within a tier the first match in document order wins and that rule's id names the audit record, so you cannot write an exception inside a deny. rule=None is not a bug — it means the request matched no rule and the document default applied. TOML was chosen because tomllib is standard library from 3.11 and [[rule]] keeps rule order visible; tomllib cannot write, so policies are authored by hand and versioned in git.

toml
version = "2026-07-01"
name = "cookbook"
default = "deny"                  # unmatched requests are refused
tenants = ["default", "acme"]     # a request naming anything else is denied

[[rule]]
id = "no-deletes"
resource = "tool"
match = "delete_*"
effect = "deny"
reason = "destructive tools are never permitted, for anyone"

[[rule]]
id = "deploys-need-sre"
resource = "tool"
match = "deploy_*"
effect = "ask"
approver_role = "sre"             # required on every `ask` rule

[[rule]]
id = "nothing-routes-into-deploy"
resource = "edge"
match = "*->deploy"
effect = "deny"

[[rule]]
id = "acme-period-cap"
resource = "spend"
effect = "deny"
over_usd = 10.0
basis = "cumulative"              # committed spend + this request
tenant = "acme"
console
tool  read_config      default  allow  rule=reads-are-free
tool  write_file       default  deny   rule=None
tool  write_file       acme     allow  rule=acme-may-write
edge  triage->deploy   default  deny   rule=nothing-routes-into-deploy
spend *                default  ask    rule=over-a-dollar-asks-finance  ask:finance

policy version: 2026-07-01  digest: b1d593faa1d41fcf
audit records:  11

How do I scope policy to a tenant and cap its spend?

tenants in the document enumerates real tenants while a rule's tenant field is an fnmatch pattern. A request naming an undeclared tenant is denied and recorded rather than raised, because a tenant name arrives with the request — i.e. from data. check_spend compares a cumulative rule's threshold against committed_usd(tenant) + amount, and the decision's request shows amount_usd, committed_usd, basis and considered_usd.

python
# A tenant the document does not declare is refused, and the refusal is recorded.
d = engine.check_tool("read_config", tenant="stranger")

# Cumulative spend: the cap moves only when spend is *recorded*.
print("  $6 now:", engine.check_spend(6.0, tenant="acme").effect.value)
engine.record_spend(6.0, tenant="acme")
again = engine.check_spend(6.0, tenant="acme")
print("  considered:", again.request)

How do I route an approval to the right role?

An ask rule names approver_role; engine.approval_router({"sre": sre}) reads it off the decision and hands the request to that role's handler, recording an approval audit entry either way. Every path fails closed and every path is recorded: no handler for the role, a handler that raises (a broken approval channel is not consent), an ask decision carrying no role, and a deny — for which no handler is called at all. The router is callable as (tool_name, args) -> bool, the shape grapharc.harness.core.ApprovalCallback expects, so it drops straight into a Harness; it is bound to one tenant, because that callback signature carries none.

python
def sre(request):
    print(f"    [{request.approver_role}] {request.subject}: {request.reason}")
    return request.subject != "deploy_prod"  # staging yes, production no


router = engine.approval_router({"sre": sre})
for tool in ("read_config", "delete_bucket", "deploy_staging", "deploy_prod"):
    print(f"{tool:<16} granted={router(tool, {})}")

for record in engine.audit.entries(kind="approval"):
    print(f"{record.subject:<16} granted={record.granted!s:<5} {record.reason}")

### Compiling the document down to a PermissionPolicy

A Harness enforces tool permissions with PermissionPolicy, not with the engine; permission_policy(tenant=...) compiles one tenant's tool rules into one, and compiled.decide(tool) returns allow/deny/ask. The honest part is the last line of output: the compiled object leaves zero audit records — it cannot carry the approver role, the rule id, or the audit record, so a Harness holding it will ASK without knowing who to ask and decide without leaving a policy record.

python
engine = PolicyEngine.from_file("policy.toml")
compiled = engine.permission_policy(tenant="acme")

for tool in ("read_config", "write_file", "delete_bucket", "deploy_prod"):
    print(f"{tool:<14} {compiled.decide(tool).value}")

print("audit records left by the compiled object:", len(engine.audit))

How do I answer "what was it allowed to do?" after the fact?

Give the engine an AuditLog with a path and every decision lands as JSONL, stamped with the policy version and digest it was decided under. The version string is a claim by the author; the digest is evidence — widening no-deletes from delete_ to delete_tmp_ leaves version identical while the digest changes, so "it was allowed under policy 2026-07-01" is checkable rather than asserted. The digest is taken over the parsed document, so comments and formatting are not in it. context= is yours: put the run id, node and session in it and the policy log joins to the run trace.

python
log = AuditLog(Path(tempfile.mkdtemp()) / "policy-audit.jsonl")
engine = PolicyEngine.from_file("policy.toml", audit=log)

ctx = {"run_id": "run-42", "node": "patch"}
engine.check_tool("write_file", tenant="acme", context=ctx)

for row in log.read_jsonl():
    print(row["resource"], row["subject"], row["effect"], row["rule_id"],
          row["policy_version"], row["policy_digest"][:8], row["context"])

# The version string is a claim by the author; the digest is evidence.
edited = Path("policy.toml").read_text().replace(&#x27;match = "delete_*"', 'match = "delete_tmp_*"')
print("same version:", parse_document(edited).version == engine.version)   # True
print("same digest: ", parse_document(edited).digest == engine.digest)     # False

How do I make my TOML document govern admission?

It does not, by default: AdmissionChecker takes an EdgePolicy built in code, PolicyEngine.check_edge answers over a document, and there is no shipped compiler between them — permission_policy() exists for tools and has no edge equivalent. The bridge is about fifteen lines: filter the document's edge rules for the tenant, split each match on EDGE_ARROW, and rebuild them as EdgeRules with the document default. The two agree because they share their semantics: both tier deny → ask → allow over the same ordered rule list, both fnmatch each endpoint separately, and both fall through to the document default.

python
def edge_policy_for(engine, tenant="default"):
    """Compile a document&#x27;s `edge` rules for one tenant into an `EdgePolicy`."""
    document = engine.document
    if not document.declares_tenant(tenant):
        return EdgePolicy(rules=(), default=Decision.DENY)
    rules = []
    for rule in document.rules_for(ResourceKind.EDGE):
        if not rule.matches_tenant(tenant):
            continue
        source, _, target = rule.match.partition(EDGE_ARROW)
        rules.append(
            EdgeRule(action=rule.effect, source=source.strip(), target=target.strip())
        )
    return EdgePolicy(rules=tuple(rules), default=document.default)
console
triage -> deploy     engine=deny   compiled=deny   agree=True
triage -> patch      engine=allow  compiled=allow  agree=True
__start__ -> triage  engine=allow  compiled=allow  agree=True
patch -> deploy      engine=deny   compiled=deny   agree=True

What this section does not give you

Stated plainly, because a governance layer that overstates itself is worse than none. Ten limits are listed, then the counterpart: the parts that are enforced and that every snippet demonstrates — a proposal cannot execute itself, an unregistered kind cannot run, a denied transition cannot be renamed into an allowed one, an over-budget plan is refused before its first node exists, and every decision (yes and no alike) is a recorded event carrying the reason.

Source file: /home/shashank/Desktop/GraphARC/docs/cookbook/05-governance.md (1757 lines). Referenced test file: /home/shashank/Desktop/GraphARC/tests/test_cookbook_governance.py.