From 960b2822b551156808f7b1a04ff87887734112d9 Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh Date: Wed, 5 Aug 2026 00:29:09 +0530 Subject: [PATCH] Tell the planner which edges the policy denies, before it wastes a round on one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The planner's system prompt names the catalog, the START/END literals and the structural rules, and its own comments say why: stating a rule up front is cheaper than three wasted rounds. The rule models actually trip over was the one it never stated. Observed with qwen3:8b against the incident registry — the policy denied `*->deploy`, and the planner proposed an edge into `deploy` in all three rounds (`edge_denied`; `edge_denied` + `cycle`; `edge_denied`) until the loop stopped `admission_refused`. About 3.5 minutes of local inference spent discovering one sentence. The refusal came back every round, and `edge_denied` names the check, not the rule: "no edge may enter `deploy`, ever" was never on the page. `EdgePolicy.disclosure()` and `NodePolicy.disclosure()` render a policy's deny rules one line each — `edges into 'deploy' are denied by policy — do not propose them` — and `PlannerNode(edge_policy=..., node_policy=...)` puts them directly under the catalog, which is the other half of the same statement: here is what exists, here is what may not be wired. The shipped loop builders hand the planner the same policy *object* the checker holds, so the prompt cannot describe a policy the gate is not applying. Allow rules and the default are left out — they say what is permitted, which the catalog covers — and so is `ask`, whose remedy is an approval rather than a different proposal. The refusal side is enriched to match. `EdgeRule` carries the `reason` `NodeRule` already had, `PolicyEngine.edge_policy()` compiles it out of the document instead of dropping it, and `policy/edge_denied` quotes it, so a planner reads why and not only what. None of this is enforcement. No check consults a disclosure, the admission gate is byte-identical, and a model that ignores what it was told is refused exactly as one that was never told — pinned by a test comparing the rejections of a disclosed and an undisclosed planner field by field, and by the shipped demo, whose scripted round 1 still proposes the denied deploy and is still refused. Fixes #45 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 1 + docs/cookbook/05-governance.md | 43 +++++++ grapharc/examples/plan_incident.py | 18 ++- grapharc/planner/admission.py | 117 ++++++++++++++++-- grapharc/planner/proposal.py | 59 ++++++++- grapharc/policy/engine.py | 11 +- grapharc/stdlib.py | 16 ++- tests/test_admission.py | 189 +++++++++++++++++++++++++++++ tests/test_planner_loop.py | 62 +++++++++- tests/test_policy_engine.py | 12 ++ 10 files changed, 514 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0b15fcf..abc24be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ Entries are newest-last within a release, matching the order they were written. - a run **stopped for overspending reported spending nothing**. Tokens were attributed from `end` events, and a node the budget interrupts emits `error` instead — so `grapharc metrics` answered `tokens: 0` for a run whose own enforcement message named the figure that stopped it (`max_tokens reached (51/5)`). The audit trail lost precisely the number the stop was about, and per-node attribution dropped the most expensive node in the run. Every `error` event is now stamped with what its node spent, exactly as `end` is, and both `summarize` and the cost report count it; sub-events inside a node remain a breakdown of its total rather than an addition, so the disjointness that kept `ends + orphans` from double-counting is unchanged, and `RunCost.tokens == RunMetrics.tokens` still holds. - the `.env` credential loader **walked up parent directories to `/`**, while the config layer next door refuses exactly that on principle — so the file that *spends money* was discovered more eagerly than the one that *constrains* a run. A run started in a scratch subdirectory picked up an `OPENROUTER_API_KEY` from any ancestor: a `.env` in `$HOME` billed every user's experiment on a shared box to that key, a demo checked out under a client project quietly used the client's key, and since `redact()` is the only thing that ever prints a key, nothing in normal operation said *which file paid*. The rationale `cli/config.py` wrote down for `grapharc.toml` — "a run must never be silently governed by a file in a directory you didn't know about" — applies with more force to the file that pays than to the file that restrains, so `find_env_file` now reads the start directory (default: the working directory) and no ancestor of it. **This is a behaviour change:** anyone relying on a parent-directory `.env` must move it into the directory they run from, `export` the variable, or pass `env_file=` naming the file. Neither escape hatch moved — a real environment variable still beats any file, and an explicit `env_file=` still reads a file anywhere on disk — and no "search boundary" was added in place of the walk, because stopping at a git root is still an upward search. +- the planner's system prompt **withheld the edge policy**, so a model had to learn it one refusal at a time. The prompt states the catalog, the START/END literals and the structural rules, and its own comments say why — "stating the rule up front is cheaper than three wasted rounds" — but the rule models actually trip over was the one it never stated. Observed with qwen3:8b against the incident registry: the goal said "find the cause and propose a fix", the policy denied `*->deploy`, and the planner proposed an edge into `deploy` in all three rounds (`edge_denied`; `edge_denied` + `cycle`; `edge_denied`) until the loop stopped `admission_refused` — about 3.5 minutes of local inference spent discovering one sentence, and a run that reads as a model failure when it is an information failure. The refusal came back every round and `edge_denied` names the check, not the rule, so "no edge may enter `deploy`, ever" was never on the page. `EdgePolicy.disclosure()` and `NodePolicy.disclosure()` now render a policy's deny rules as one line each (`edges into 'deploy' are denied by policy — do not propose them`), `PlannerNode(edge_policy=…, node_policy=…)` puts them directly under the catalog, and the shipped loop builders hand the planner the same policy *object* the checker holds, so the prompt cannot describe a policy the gate is not applying. Allow rules and the default are left out — they say what is permitted, which the catalog already covers — and so is `ask`, whose remedy is an approval rather than a different proposal. The refusal side is enriched to match: `EdgeRule` carries the `reason` `NodeRule` already had, `PolicyEngine.edge_policy()` compiles it out of the document instead of dropping it on the floor, and `policy/edge_denied` quotes it, so a planner reads why and not only what. **None of this is enforcement.** No check consults the disclosure, the admission gate is byte-identical, and a model that ignores what it was told is refused exactly as one that was never told — pinned by a test that compares the rejections of a disclosed and an undisclosed planner field by field, and by the shipped demo, whose scripted round 1 still proposes the denied deploy and is still refused. ## 0.1.3 diff --git a/docs/cookbook/05-governance.md b/docs/cookbook/05-governance.md index 93f2eab..0ff362e 100644 --- a/docs/cookbook/05-governance.md +++ b/docs/cookbook/05-governance.md @@ -901,6 +901,49 @@ The default planner system prompt already tells the model that renaming a refused node is a wasted turn (`DEFAULT_PLANNER_SYSTEM_PROMPT`). That is a courtesy to save a round trip. It is not the enforcement — the gate is. +Hand the planner the gates as well and the same courtesy covers the policy. +Every `deny` rule is rendered under the catalog, carrying the rule's own +`reason` when the document wrote one: + +```python +from grapharc.harness.permissions import Decision +from grapharc.planner import EdgePolicy, EdgeRule, PlannerNode +from grapharc.testing import ScriptedChatModel + +policy = EdgePolicy( + rules=( + EdgeRule( + action=Decision.DENY, + target="deploy", + reason="a deploy is the operator's decision", + ), + EdgeRule(action=Decision.ALLOW), + ) +) +catalog = {"build": "compile the change", "deploy": "push to production"} +model = ScriptedChatModel(responses=['{"nodes": [], "edges": []}']) + +PlannerNode(model, catalog=catalog, edge_policy=policy).propose("ship it") + +system = str(model.calls[0][0].content) +print(system.split("Available node kinds:")[1].strip()) +``` + +``` +- build: compile the change +- deploy: push to production + +Denied by policy. The admission checker refuses these; it is deterministic code and this list is only telling you in advance: +- edges into 'deploy' are denied by policy — do not propose them: a deploy is the operator's decision +``` + +Without those two lines a model reads a registered-but-denied kind as an +invitation, proposes it, gets `edge_denied` back — a check name, which says +nothing about how wide the denial is — and proposes it again. One real run +spent all three of its rounds finding that out. The gate is untouched by the +disclosure: `PlannerNode` still decides nothing, and a proposal that walks into +a denial anyway is refused by exactly the code that refused it before. + ### With a real model Swap the scripted model for a real one; nothing else changes. diff --git a/grapharc/examples/plan_incident.py b/grapharc/examples/plan_incident.py index 17070b9..bf3f4a5 100644 --- a/grapharc/examples/plan_incident.py +++ b/grapharc/examples/plan_incident.py @@ -165,6 +165,11 @@ def build_loop( # Frozen: a driver that checks "against the same registry" every round means # the same object, and a node body could otherwise widen it between rounds. registry.freeze() + # Resolved once so the planner is *told* about exactly the policy the checker + # will *apply*. Two calls to `default_edge_policy()` would be two objects, + # and a disclosure describing a different object than the gate enforces is + # worse than no disclosure at all. + edge_policy = edge_policy or default_edge_policy() return GovernedLoop( # The planner and the materializer get the recorder too. Without it the # run's own trace held only `admission`/`round`/`stop`: no `plan` event @@ -174,11 +179,20 @@ def build_loop( # own start/end pairs" was true of a hand-wired loop and false of the # shipped one, which is the one `grapharc plan` drives. planner=PlannerNode( - model, name="incident", catalog=registry.catalog(), trace=trace + model, + name="incident", + catalog=registry.catalog(), + # Disclosure, not enforcement: the planner is shown the deny rules so + # it need not learn them one refusal at a time. The scripted planner + # below proposes a `deploy` anyway, and round 1 is still refused — + # which is the demo's whole point, and stays true with a real model. + edge_policy=edge_policy, + node_policy=node_policy, + trace=trace, ), checker=AdmissionChecker( registry=registry, - edge_policy=edge_policy or default_edge_policy(), + edge_policy=edge_policy, # There is no default node policy: this demo's registry *is* its # node allowlist. One arrives only when a policy document declares # node rules, and then it gates every kind the planner proposes. diff --git a/grapharc/planner/admission.py b/grapharc/planner/admission.py index dae6c58..8c01fe6 100644 --- a/grapharc/planner/admission.py +++ b/grapharc/planner/admission.py @@ -53,6 +53,14 @@ safe (`policy/unresolved_endpoint_kind`); pass `known_nodes` as a `{name: kind}` mapping to say what those nodes are. +**Disclosure is not enforcement.** `EdgePolicy.disclosure()` and +`NodePolicy.disclosure()` render a policy's deny rules as sentences a planner +can be *shown* before it proposes anything, which is what +`grapharc.planner.proposal.PlannerNode` puts in its system prompt. Nothing in +this module reads them back, no check consults them, and a model that ignores +them — or never saw them — is refused by byte-identical code. Telling a planner +the rule is a courtesy that saves rounds; the gate is what decides. + What this module does *not* do. It does not build a runnable graph — admission authorises a shape, and turning one into work is `grapharc.planner.materialize`, which takes the `AdmissionResult` this returns and refuses to build anything @@ -275,6 +283,12 @@ class EdgeRule(BaseModel): `source` and `target` are patterns over **registry kinds** — plus the literal `START`/`END` sentinels, which no node may be named. They are never matched against a planner's instance name. + + `reason` is the operator's own words, carried from the policy document that + compiled to this rule, exactly as `NodeRule.reason` is. A refusal quotes it + so a planner reads *why* rather than only `edge_denied`, and + `EdgePolicy.disclosure()` puts it in front of the model before the first + round. A rule without one still refuses; nothing decides on this string. """ model_config = ConfigDict(frozen=True) @@ -282,6 +296,7 @@ class EdgeRule(BaseModel): action: Decision source: str = "*" target: str = "*" + reason: str = "" class EdgePolicy(BaseModel): @@ -305,8 +320,8 @@ class EdgePolicy(BaseModel): rules: tuple[EdgeRule, ...] = () default: Decision = Decision.DENY - def decide(self, source_kind: str, target_kind: str) -> Decision: - """Decide one transition. Both arguments are kinds (or a sentinel).""" + def rule_for(self, source_kind: str, target_kind: str) -> EdgeRule | None: + """The rule that decides this transition, or None when the default applies.""" for tier in (Decision.DENY, Decision.ASK, Decision.ALLOW): for rule in self.rules: if ( @@ -314,8 +329,36 @@ def decide(self, source_kind: str, target_kind: str) -> Decision: and fnmatch(source_kind, rule.source) and fnmatch(target_kind, rule.target) ): - return tier - return self.default + return rule + return None + + def decide(self, source_kind: str, target_kind: str) -> Decision: + """Decide one transition. Both arguments are kinds (or a sentinel).""" + rule = self.rule_for(source_kind, target_kind) + return self.default if rule is None else rule.action + + def disclosure(self) -> tuple[str, ...]: + """The deny rules as sentences a planner can be shown before it proposes. + + **Disclosure, not enforcement.** Nothing reads this back: `decide` is + the only thing that decides, and a planner handed these lines and + ignoring them is refused exactly as one that never saw them. It exists + because `edge_denied` on round three is a fact the model could have had + on round one — the observed failure was a run that proposed an edge into + a denied kind every round until the loop gave up, unable to infer "no + edge may enter this, ever" from a check name. + + Deny rules only. An allow rule and the default say what is *permitted*, + which the catalog and the structural rules already cover, and listing + them would turn a short warning into a policy dump the model has to + read past. `ask` is left out for a different reason: its remedy is to + obtain approval, not to propose something else. + """ + return _denial_lines( + (_edge_subject(rule.source, rule.target), rule.reason) + for rule in self.rules + if rule.action is Decision.DENY + ) class NodeRule(BaseModel): @@ -371,6 +414,19 @@ def decide(self, kind: str) -> Decision: rule = self.rule_for(kind) return self.default if rule is None else rule.action + def disclosure(self) -> tuple[str, ...]: + """The deny rules as sentences a planner can be shown. See `EdgePolicy.disclosure`. + + A denied kind is worth stating for the same reason a denied edge is: the + registry lists it as proposable — it is registered — and the document + then forbids it, so the catalog alone reads as an invitation. + """ + return _denial_lines( + (_node_subject(rule.match), rule.reason) + for rule in self.rules + if rule.action is Decision.DENY + ) + class AdmissionLimits(BaseModel): """Structural limits, set by the operator and not by the proposal.""" @@ -684,7 +740,8 @@ def _check_policy(self, proposal: Subgraph) -> list[Rejection]: self._unresolved_endpoints(subject, edge, source_kind, target_kind) ) continue - decision = self.edge_policy.decide(source_kind, target_kind) + rule = self.edge_policy.rule_for(source_kind, target_kind) + decision = self.edge_policy.default if rule is None else rule.action if decision is Decision.ALLOW: continue denied = decision is Decision.DENY @@ -692,17 +749,23 @@ def _check_policy(self, proposal: Subgraph) -> list[Rejection]: f"{_describe(edge.source, source_kind)} -> " f"{_describe(edge.target, target_kind)}" ) + # The operator's own words, when the rule carried any — the same + # courtesy `_check_node_policy` extends. A planner told only + # `edge_denied` has to guess how wide the denial is; told "deploys + # are the operator's decision" it can stop proposing one. + because = f": {rule.reason}" if rule is not None and rule.reason else "" out.append( Rejection( check=Check.POLICY, code="edge_denied" if denied else "edge_needs_approval", subject=subject, detail=( - f"the edge policy denies this transition: {transition}" + f"the edge policy denies this transition: " + f"{transition}{because}" if denied else ( "the edge policy requires approval for this " - f"transition: {transition}" + f"transition: {transition}{because}" ) ), remedy=( @@ -971,6 +1034,46 @@ def _scoped(path: str, subject: str) -> str: return f"{path}/{subject}" if path else subject +def _pattern_text(pattern: str) -> str: + """A rule's pattern as prose: a bare kind is quoted, a glob is described as one.""" + return ( + f"kinds matching {pattern!r}" + if any(char in pattern for char in "*?[") + else repr(pattern) + ) + + +def _edge_subject(source: str, target: str) -> str: + """What one edge deny rule is about, in the plural so a line reads as a warning.""" + if source == "*" and target == "*": + return "all edges" + if source == "*": + return f"edges into {_pattern_text(target)}" + if target == "*": + return f"edges out of {_pattern_text(source)}" + return f"edges from {_pattern_text(source)} to {_pattern_text(target)}" + + +def _node_subject(match: str) -> str: + return "all node kinds" if match == "*" else f"nodes of kind {_pattern_text(match)}" + + +def _denial_lines(subjects: Iterable[tuple[str, str]]) -> tuple[str, ...]: + """`(subject, reason)` pairs -> one line each, in rule order, without repeats. + + Two rules can render the same sentence — a document scoped per tenant is the + ordinary way — and saying it twice would only cost the reader attention. + """ + lines: list[str] = [] + for subject, reason in subjects: + line = f"{subject} are denied by policy — do not propose them" + if reason.strip(): + line = f"{line}: {reason.strip()}" + if line not in lines: + lines.append(line) + return tuple(lines) + + def _describe(endpoint: str, kind: str) -> str: """An endpoint as the rejection should name it: what it is, then what it is called. diff --git a/grapharc/planner/proposal.py b/grapharc/planner/proposal.py index bd2b4d2..cfea5f0 100644 --- a/grapharc/planner/proposal.py +++ b/grapharc/planner/proposal.py @@ -34,7 +34,7 @@ import time import uuid from collections.abc import Callable, Iterator, Mapping, Sequence -from typing import Any +from typing import TYPE_CHECKING, Any from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage @@ -45,6 +45,9 @@ from grapharc.runtime.graph import END, START, RunContext from grapharc.runtime.parsing import extract_json +if TYPE_CHECKING: # `admission` imports this module, so the dependency is one-way + from grapharc.planner.admission import EdgePolicy, NodePolicy + # Node and kind names live in trace lines, Mermaid labels and fnmatch patterns. # The charset excludes `>` and whitespace so an "a -> b" edge rendering can # never be ambiguous about where the name ends. @@ -291,6 +294,16 @@ def fingerprint(self) -> str: _NO_CATALOG = "(no catalog supplied; the admission registry decides what is allowed)" +#: Introduces the compiled policies' deny rules, when a planner was given the +#: policies its proposals will be checked against. The catalog says which kinds +#: exist; this says which of them may not be wired, which is the other half of +#: the same question and the half a model cannot guess. The wording says who +#: enforces it, because the disclosure must not read as the rule itself. +_DENIED_HEADER = ( + "Denied by policy. The admission checker refuses these; it is deterministic " + "code and this list is only telling you in advance:" +) + #: Exception *names* that mean "the backend was not reached", matched by name #: so this module needs no provider SDK imported to recognise them. Substrings @@ -332,6 +345,26 @@ def _catalog_text(catalog: Mapping[str, str] | Sequence[str] | None) -> str: return "\n".join(f"- {name}: {desc}" if desc else f"- {name}" for name, desc in items) +def _denial_text(edge_policy: EdgePolicy | None, node_policy: NodePolicy | None) -> str: + """The policies' deny rules as a prompt section, or "" when there is nothing to say. + + Duck-typed on `disclosure()` rather than on the concrete classes: this module + is imported *by* `admission`, so it cannot import the policy types back, and + an operator's own policy object with the same method is disclosed the same + way. A policy with no deny rules contributes nothing — an empty header would + read as "nothing is denied", which for a default-deny policy is a lie. + """ + lines: list[str] = [] + for policy in (edge_policy, node_policy): + disclose = getattr(policy, "disclosure", None) + if disclose is None: + continue + lines.extend(line for line in disclose() if line not in lines) + if not lines: + return "" + return "\n".join([_DENIED_HEADER, *(f"- {line}" for line in lines)]) + + def _message_text(message: BaseMessage) -> str: text = getattr(message, "text", None) if isinstance(text, str): @@ -379,6 +412,13 @@ class PlannerNode: planner = PlannerNode(model, catalog=registry.catalog()) g.add_node("planner", planner, writes=planner.writes) + Pass the gates too — `edge_policy=`, `node_policy=` — and their deny rules + are rendered into the system prompt beside the catalog, so the model learns + "no edge may enter `deploy`" before round one instead of inferring it from + three `edge_denied` refusals. That is **disclosure and nothing else**: this + class still cannot decide anything, the checker is unchanged, and a model + that proposes the denied edge regardless is refused exactly as before. + **It cannot execute what it proposes**, and that is structural rather than promised: the only callables it holds are the chat model and the caller's own `prompt_fn` state reader. It is given no node registry, no harness and @@ -400,6 +440,8 @@ def __init__( *, name: str = "planner", catalog: Mapping[str, str] | Sequence[str] | None = None, + edge_policy: EdgePolicy | None = None, + node_policy: NodePolicy | None = None, system_prompt: str = DEFAULT_PLANNER_SYSTEM_PROMPT, instructions: str = "", task_field: str = "task", @@ -412,6 +454,13 @@ def __init__( self.model = model self.name = name self.catalog = catalog + # The gates this planner's proposals will be checked against, held only + # so their deny rules can be *disclosed* in the prompt. Nothing here + # consults them: `propose` never calls `decide`, and a proposal that + # walks straight into a denial is still produced and still refused by + # `AdmissionChecker`. Leave them None and the prompt is unchanged. + self.edge_policy = edge_policy + self.node_policy = node_policy self.system_prompt = system_prompt self.instructions = instructions self.task_field = task_field @@ -528,6 +577,14 @@ def propose( def _messages(self, task: str, feedback: str) -> list[BaseMessage]: system = f"{self.system_prompt}\n\nAvailable node kinds:\n{_catalog_text(self.catalog)}" + # Immediately after the catalog, because the two are one statement: here + # is what exists, and here is what may not be wired. A model shown only + # the first reads a registered-but-denied kind as an invitation, proposes + # it, is refused with a check name it cannot generalise from, and does it + # again — three rounds of real inference to learn one sentence. + denied = _denial_text(self.edge_policy, self.node_policy) + if denied: + system = f"{system}\n\n{denied}" if self.instructions: system = f"{system}\n\n{self.instructions}" messages: list[BaseMessage] = [SystemMessage(content=system), HumanMessage(content=task)] diff --git a/grapharc/policy/engine.py b/grapharc/policy/engine.py index 8a9ffe4..f23b3f2 100644 --- a/grapharc/policy/engine.py +++ b/grapharc/policy/engine.py @@ -285,7 +285,16 @@ def edge_policy(self, *, tenant: str = DEFAULT_TENANT) -> Any: continue source, _, target = rule.match.partition(EDGE_ARROW) rules.append( - EdgeRule(action=rule.effect, source=source.strip(), target=target.strip()) + EdgeRule( + action=rule.effect, + source=source.strip(), + target=target.strip(), + # Carried, as `node_policy()` carries it: the document's own + # words are what a refusal quotes and what a planner is told + # up front, and a rule that arrives without them leaves both + # saying only `edge_denied`. + reason=rule.reason, + ) ) return EdgePolicy(rules=tuple(rules), default=self._document.default) diff --git a/grapharc/stdlib.py b/grapharc/stdlib.py index 4d85d8d..e87fa97 100644 --- a/grapharc/stdlib.py +++ b/grapharc/stdlib.py @@ -414,13 +414,25 @@ def build_loop( registry = registry or build_registry(model) registry.freeze() + # One object, disclosed to the planner and applied by the checker. Resolving + # the default twice would build two, and a prompt describing a different + # object from the one the gate applies is worse than no disclosure at all. + edge_policy = edge_policy or default_edge_policy() return GovernedLoop( planner=PlannerNode( - model, name="stdlib", catalog=registry.catalog(), trace=trace + model, + name="stdlib", + catalog=registry.catalog(), + # The deny rules, in front of the model before round 1. Disclosure + # only — the checker below is what refuses, whether or not the model + # read this. + edge_policy=edge_policy, + node_policy=node_policy, + trace=trace, ), checker=AdmissionChecker( registry=registry, - edge_policy=edge_policy or default_edge_policy(), + edge_policy=edge_policy, # None unless a policy document declared node rules; the registry is # otherwise the only thing deciding which kinds may run. node_policy=node_policy, diff --git a/tests/test_admission.py b/tests/test_admission.py index 7160441..972cd7a 100644 --- a/tests/test_admission.py +++ b/tests/test_admission.py @@ -1133,6 +1133,195 @@ def test_the_catalog_is_put_in_front_of_the_model(): assert "summarise: summarise description" in system +# -- issue #45: the policy is disclosed to the planner, and only disclosed ----- +# +# The catalog said which kinds exist and nothing said which transitions were +# denied, so a model proposed an edge into a denied kind every round until the +# loop gave up: `edge_denied` names the check, not the rule, and "no edge may +# enter deploy, ever" is not inferable from it. These tests pin both halves — +# the disclosure is in the prompt, and it is *only* a disclosure. + + +def _system_prompt(planner: PlannerNode, model: ScriptedChatModel, task: str = "go") -> str: + planner.propose(task) + return str(model.calls[0][0].content) + + +def test_the_edge_policys_denials_are_put_in_front_of_the_model(): + reg = registry("build", "deploy") + model = ScriptedChatModel(responses=[json.dumps(RENAMED_PLAN_JSON)]) + + system = _system_prompt( + PlannerNode(model, catalog=reg.catalog(), edge_policy=DENY_DEPLOY), model, "ship it" + ) + + assert "edges into 'deploy' are denied by policy — do not propose them" in system + # And the catalog still lists the kind: it is registered, an operator did + # allow it to be proposed, and it is the *edge* that is refused. + assert "deploy: deploy description" in system + + +def test_a_disclosed_denial_carries_the_operators_own_words(): + policy = EdgePolicy( + rules=( + EdgeRule( + action="deny", target="deploy", reason="a deploy is the operator's decision" + ), + EdgeRule(action="allow"), + ) + ) + model = ScriptedChatModel(responses=[json.dumps(RENAMED_PLAN_JSON)]) + + system = _system_prompt( + PlannerNode(model, catalog=registry("build", "deploy").catalog(), edge_policy=policy), + model, + "ship it", + ) + + assert ( + "edges into 'deploy' are denied by policy — do not propose them: " + "a deploy is the operator's decision" in system + ) + + +def test_denied_node_kinds_are_disclosed_beside_the_denied_edges(): + node_policy = NodePolicy( + rules=( + NodeRule(action="deny", match="deploy", reason="not from a plan"), + NodeRule(action="allow"), + ) + ) + model = ScriptedChatModel(responses=[json.dumps(RENAMED_PLAN_JSON)]) + + system = _system_prompt( + PlannerNode( + model, + catalog=registry("build", "deploy").catalog(), + edge_policy=DENY_DEPLOY, + node_policy=node_policy, + ), + model, + "ship it", + ) + + assert ( + "nodes of kind 'deploy' are denied by policy — do not propose them: not from a plan" + in system + ) + assert "edges into 'deploy' are denied by policy" in system + + +def test_a_policy_that_denies_nothing_adds_nothing_to_the_prompt(): + """Allow rules and the default are the catalog's business, not a warning's.""" + reg = registry("fetch", "summarise") + told = ScriptedChatModel(responses=[json.dumps(PLAN_JSON)]) + untold = ScriptedChatModel(responses=[json.dumps(PLAN_JSON)]) + + with_policy = _system_prompt( + PlannerNode(told, catalog=reg.catalog(), edge_policy=ALLOW_ALL), told + ) + without = _system_prompt(PlannerNode(untold, catalog=reg.catalog()), untold) + + assert "denied by policy" not in with_policy + assert with_policy == without # the unpolicied prompt is byte-identical to before + + +def test_disclosure_does_not_move_the_enforcement_into_the_prompt(): + """A planner told about the denial and proposing it anyway is refused identically. + + The constraint the fix is subject to: enforcement stays in the checker. Two + planners, one shown the deny rule and one not, produce the same proposal + here — and the gate has to return the same rejections for both, down to the + text, or the disclosure has started deciding something. + """ + reg = registry("build", "deploy") + reply = json.dumps(RENAMED_PLAN_JSON) + told = PlannerNode( + ScriptedChatModel(responses=[reply]), catalog=reg.catalog(), edge_policy=DENY_DEPLOY + ) + untold = PlannerNode(ScriptedChatModel(responses=[reply]), catalog=reg.catalog()) + gate = checker(reg, edge_policy=DENY_DEPLOY) + + disclosed = gate.check(told.propose("ship it").proposal) + blind = gate.check(untold.propose("ship it").proposal) + + assert not disclosed.admitted + assert disclosed.failed_checks() == (Check.POLICY,) + assert [r.model_dump() for r in disclosed.rejections] == [ + r.model_dump() for r in blind.rejections + ] + assert reg.get("deploy").factory is _explode # it was there to be called + + +def test_a_denied_edge_is_refused_with_the_rules_reason_not_only_its_code(): + """`edge_denied` is what happened; the reason is why, and the planner needs both.""" + policy = EdgePolicy( + rules=( + EdgeRule(action="deny", target="deploy", reason="Deploy changes are dangerous"), + EdgeRule(action="allow"), + ) + ) + + result = checker(registry("build", "deploy"), edge_policy=policy).check( + linear("build", "deploy") + ) + + reason = result.reasons(Check.POLICY)[0] + assert reason.code == "edge_denied" + assert "Deploy changes are dangerous" in reason.detail + assert "Deploy changes are dangerous" in result.feedback() + + +def test_a_rule_without_a_reason_still_refuses_and_says_only_what_it_knows(): + result = checker(registry("build", "deploy"), edge_policy=DENY_DEPLOY).check( + linear("build", "deploy") + ) + + reason = result.reasons(Check.POLICY)[0] + assert reason.code == "edge_denied" + assert reason.detail.endswith("kind 'deploy' (proposed as 'deploy')") + + +def test_a_disclosure_describes_the_rule_it_was_compiled_from(): + """Every shape of deny rule renders as something a model can act on.""" + policy = EdgePolicy( + rules=( + EdgeRule(action="deny", target="deploy"), + EdgeRule(action="deny", source="deploy"), + EdgeRule(action="deny", source="triage", target="verify"), + EdgeRule(action="deny", target="risky_*"), + EdgeRule(action="ask", target="patch"), + EdgeRule(action="allow"), + ) + ) + + assert policy.disclosure() == ( + "edges into 'deploy' are denied by policy — do not propose them", + "edges out of 'deploy' are denied by policy — do not propose them", + "edges from 'triage' to 'verify' are denied by policy — do not propose them", + "edges into kinds matching 'risky_*' are denied by policy — do not propose them", + ) + # An `ask` is not a "do not propose": its remedy is an approval, not another + # proposal. And a policy with nothing denied has nothing to disclose. + assert ALLOW_ALL.disclosure() == () + assert NodePolicy(rules=(NodeRule(action="deny", match="*"),)).disclosure() == ( + "all node kinds are denied by policy — do not propose them", + ) + + +def test_a_repeated_denial_is_disclosed_once(): + """Two rules, one sentence: a per-tenant document renders duplicates.""" + policy = EdgePolicy( + rules=( + EdgeRule(action="deny", target="deploy"), + EdgeRule(action="deny", target="deploy"), + EdgeRule(action="allow"), + ) + ) + + assert len(policy.disclosure()) == 1 + + def test_a_rejection_can_be_fed_back_for_a_second_attempt(): reg = registry("fetch", "summarise") bad = { diff --git a/tests/test_planner_loop.py b/tests/test_planner_loop.py index 74c4977..bce66e0 100644 --- a/tests/test_planner_loop.py +++ b/tests/test_planner_loop.py @@ -233,13 +233,22 @@ def build_loop( goal_reached=None, on_exhausted: str = "raise", checker: AdmissionChecker | None = None, + disclose: bool = False, **loop_kwargs, ): """A whole cycle wired the way an operator would wire it.""" bodies = bodies if bodies is not None else Bodies() reg = reg if reg is not None else registry(bodies) model = ScriptedChatModel(responses=responses, on_exhausted=on_exhausted) - planner = PlannerNode(model, catalog=reg.catalog(), trace=trace) + planner = PlannerNode( + model, + catalog=reg.catalog(), + # Off unless a test asks, so every prompt assertion above stays about + # what it was written about. The shipped builders pass the policy here; + # `disclose=True` is that wiring, and it changes the prompt only. + edge_policy=policy if disclose else None, + trace=trace, + ) loop = GovernedLoop( planner=planner, checker=checker if checker is not None else gate(reg, policy=policy, trace=trace), @@ -1571,3 +1580,54 @@ def test_a_merely_bad_reply_still_gets_its_retries(): result = loop.run("goal") assert result.stop is LoopStop.PLANNING_FAILED assert len(result.rounds) == 3 # the full allowance, as before + + +# -- issue #45: the loop's planner is told the policy before round 1 ---------- + + +def test_the_shipped_loop_discloses_its_edge_policy_to_the_planner(): + """The observed failure, wired exactly as `grapharc plan` wires it. + + A real model burned three rounds proposing an edge into `deploy` — denied + by the policy every time — because the prompt listed `deploy` in the catalog + and never said no edge may enter it. The deny rule now travels with the + catalog. The scripted planner still proposes the deploy (it is a fixed + script), and round 1 is still refused, which is the point: the disclosure is + a courtesy and the checker is the gate. + """ + from grapharc.examples import plan_incident + + model = ScriptedChatModel(responses=plan_incident.scripted_planner_replies()) + result = plan_incident.build_loop(model).run( + "find the cause", plan_incident.IncidentState(goal="g") + ) + + system = str(model.calls[0][0].content) + assert "edges into 'deploy' are denied by policy — do not propose them" in system + assert "deploy: push to production" in system # the catalog is unchanged + + # Enforcement is where it was: round 1 proposed the denied edge anyway and + # was refused by the checker, round 2 replanned without it and ran. + assert [r.code for r in result.rejections()] == ["edge_denied"] + assert result.stop is LoopStop.GOAL_MET + + +def test_the_disclosure_is_not_what_refuses_the_edge(): + """Strip the disclosure and the verdict is byte-identical. + + The planner is handed the same policy object the checker holds, so the only + thing that could differ between these two runs is the prompt. If the + rejections ever diverge, enforcement has leaked into prompt text. + """ + denied = [plan(("ship", "deploy"))] * 3 + told, model_told, _ = build_loop(denied, policy=DENY_DEPLOY, disclose=True) + untold, model_untold, _ = build_loop(denied, policy=DENY_DEPLOY) + + with_disclosure = told.run("goal") + without = untold.run("goal") + + assert "denied by policy" in str(model_told.calls[0][0].content) + assert "denied by policy" not in str(model_untold.calls[0][0].content) + assert [r.model_dump() for r in with_disclosure.rejections()] == [ + r.model_dump() for r in without.rejections() + ] diff --git a/tests/test_policy_engine.py b/tests/test_policy_engine.py index 3b6a3ea..204607a 100644 --- a/tests/test_policy_engine.py +++ b/tests/test_policy_engine.py @@ -877,6 +877,18 @@ def test_each_side_of_an_edge_rule_is_matched_separately(): assert policy.decide("other", "triage") is Decision.ALLOW +def test_the_compiled_edge_policy_carries_the_documents_reason(engine): + """Issue #45: the words an operator wrote are what a refusal quotes and what + the planner is told up front. Dropped at compile time, both could only say + `edge_denied`.""" + rule = engine.edge_policy().rule_for("plan", "deploy_prod") + + assert rule is not None + assert rule.action is Decision.DENY + assert rule.reason == "the production deploy node is entered by an operator" + assert rule.reason in engine.edge_policy().disclosure()[0] + + def test_tool_rules_do_not_leak_into_the_edge_policy(): """Widening what an agent may call must not widen what a planner may wire.""" engine = PolicyEngine.from_toml(