From f3b240af6fdedab5c55925371e386061a1062f0b Mon Sep 17 00:00:00 2001 From: Dhinesh Ponnarasan Date: Thu, 1 Oct 2026 07:50:28 -0400 Subject: [PATCH 1/3] policy: govern grapharc agent by policy document --- grapharc/cli/agent.py | 280 +++++++++- grapharc/cli/main.py | 20 + grapharc/harness/__init__.py | 5 +- grapharc/harness/agent.py | 10 +- grapharc/harness/core.py | 37 ++ grapharc/harness/permissions.py | 60 ++- tests/test_cli.py | 921 +++++++++++++++++++++++++++++++- tests/test_delegate_tiers.py | 2 +- tests/test_harness_gate.py | 175 ++++++ 9 files changed, 1495 insertions(+), 15 deletions(-) diff --git a/grapharc/cli/agent.py b/grapharc/cli/agent.py index 17da119..ca67b82 100644 --- a/grapharc/cli/agent.py +++ b/grapharc/cli/agent.py @@ -39,6 +39,13 @@ DEFAULT_MAX_TOKENS = 100_000 DEFAULT_MAX_SECONDS = 300.0 +#: Sibling of the run's trace: the policy-document enforcement behind an +#: `agent --policy` run lands here as JSONL, via `grapharc.policy.audit` — +#: denials, and the decision-plus-outcome pair behind every approval request. +#: Allowances leave no record: the compiled policy grants them without +#: consulting the engine, so there is no document decision to write down. +POLICY_AUDIT_FILENAME = "policy-audit.jsonl" + def _accepts(fn: Any, param: str) -> bool: """Whether `fn` can be called with `param=`; unknown signatures say no.""" @@ -88,14 +95,25 @@ def build_registry(workspace: Path) -> tuple[Any, str]: ) -def build_policy(allow: list[str], deny: list[str], ask: list[str]) -> Any: - """Flags to a `PermissionPolicy`. Unmatched tools keep the policy's DENY default.""" +def build_policy( + allow: list[str], deny: list[str], ask: list[str], *, default: Any = None +) -> Any: + """Flags to a `PermissionPolicy`. + + Unmatched tools keep the `PermissionPolicy` DENY default — unless a + caller passes `default=ALLOW`, which is what the policy-document path + does: beside a document the flags are refinements, and an unmatched tool + is "no flag opinion", decided by the document alone. The default is `None` + meaning DENY rather than a `Decision` so this module keeps its lazy + `grapharc.harness` import. + """ from grapharc.harness import Decision, PermissionPolicy, PermissionRule rules = [PermissionRule(action=Decision.DENY, pattern=p) for p in deny] rules += [PermissionRule(action=Decision.ASK, pattern=p) for p in ask] rules += [PermissionRule(action=Decision.ALLOW, pattern=p) for p in allow] - return PermissionPolicy(rules=rules) + unmatched = Decision.DENY if default is None else default + return PermissionPolicy(rules=rules, default=unmatched) def _approval(as_json: bool, *, stream: Any = None) -> Any: @@ -117,6 +135,129 @@ def ask(tool_name: str, args: dict[str, Any]) -> bool: return ask +def _load_agent_document( + policy_path: Path, *, tenant: str | None, audit_path: Path +) -> tuple[Any, str]: + """Load and validate the document governing an agent run. + + Returns `(engine, tenant_name)`. Raises `PolicyError` — the shape a + missing or malformed file already fails with — for the two agent-specific + refusals as well: a document declaring no tool rules cannot govern tools, + and a tenant the document does not declare would deny every tool. Both + are refused before the audit log is built, so a refused run writes + nothing, not even an empty audit file. + """ + from grapharc.policy import AuditLog, PolicyEngine, PolicyError, load_document + from grapharc.policy.document import DEFAULT_TENANT, ResourceKind + + document = load_document(policy_path) + if not document.rules_for(ResourceKind.TOOL): + raise PolicyError( + f"policy document {policy_path} declares no tool rules, so it cannot " + "govern an agent run — it constrains nothing this command does" + ) + tenant_name = DEFAULT_TENANT if tenant is None else tenant + if not document.declares_tenant(tenant_name): + if tenant is None: + hint = "pass --tenant to name one" + else: + hint = "check the spelling of --tenant" + raise PolicyError( + f"tenant {tenant_name!r} is not declared by policy " + f"{document.version!r}; declared: {document.tenants!r} — {hint}" + ) + return PolicyEngine(document, audit=AuditLog(audit_path)), tenant_name + + +def combined_approval( + *, + engine: Any, + doc_policy: Any, + flag_policy: Any, + tenant: str, + as_json: bool, + stream: Any = None, + context: dict[str, Any] | None = None, +) -> Any: + """Approval honoring both a policy document and CLI `--ask` flags. + + Called by `Harness` when the combined policy answers ASK — so at least + one side asks and neither denies — and granted only when every asking + side grants: the document's approver role through `ApprovalRouter`, the + flags through the same terminal prompt `--ask` has always used. A role + handler that is missing, refuses, fails, or has nobody behind it (JSON + mode, redirected stdin) denies, exactly as the router already fails + closed; the role, rule and reason travel into the prompt so the human + approves a named thing, not a bare tool call. + """ + from grapharc.harness import Decision + from grapharc.policy.document import ResourceKind + + roles = sorted( + { + rule.approver_role + for rule in engine.document.rules_for(ResourceKind.TOOL) + if rule.effect is Decision.ASK and rule.approver_role + } + ) + + def handle_role(role: str) -> Any: + def handle(request: Any) -> bool: + # Same fail-closed guard `_approval` stands on: a prompt has no + # business in a pipe, and piped bytes are not consent. + source = stream or sys.stdin + if as_json or not getattr(source, "isatty", lambda: False)(): + return False + answer = input( + f"allow {request.subject} as {role} " + f"(rule {request.rule_id}: {request.reason})? [y/N] " + ).strip().lower() + return answer in ("y", "yes") + + return handle + + router = engine.approval_router( + {role: handle_role(role) for role in roles}, tenant=tenant + ) + flag_prompt = _approval(as_json, stream=stream) + + def approve(tool_name: str, args: dict[str, Any]) -> bool: + doc_asks = doc_policy.decide(tool_name) is Decision.ASK + flag_asks = flag_policy.decide(tool_name) is Decision.ASK + granted = True + if doc_asks: + decision = engine.check_tool(tool_name, tenant=tenant, context=context) + granted = router.route(decision, args=args).granted + if granted and flag_asks: + granted = flag_prompt(tool_name, args) + # Both asking sides granted; neither asking at all is a refusal, not + # an approval — unreachable from Harness, which only calls back on ASK. + return granted and (doc_asks or flag_asks) + + return approve + + +def document_denial_recorder( + *, engine: Any, doc_policy: Any, tenant: str, context: dict[str, Any] | None = None +) -> Any: + """An `on_denial` hook recording document-side denials, and only those. + + `Harness` fires the hook for every policy denial, including ones the CLI + flags caused. Recording a flag-side denial through the engine would write + ALLOW beside a refusal whenever the document permits the tool — a + contradiction in the audit log — so those stay trace-only, exactly as + before this command learned about documents. + """ + from grapharc.harness import Decision + + def record(tool_name: str) -> None: + if doc_policy.decide(tool_name) is not Decision.DENY: + return + engine.check_tool(tool_name, tenant=tenant, context=context) + + return record + + def run_agent( task: str, *, @@ -132,6 +273,8 @@ def run_agent( executor: str = "sandbox", system_prompt: str | None = None, run_id: str | None = None, + policy_path: Path | None = None, + tenant: str | None = None, as_json: bool = False, ) -> int: """Run one agent loop and report it. Returns the process exit code. @@ -140,10 +283,38 @@ def run_agent( an exhausted budget, an error — exits 1, because a script that ran an agent needs to know the task was not finished without parsing the reason first. + `policy_path` names a TOML policy document whose tool rules govern the run + beside the CLI flags: the document is the ceiling and the flags can only + narrow it — most restrictive wins, so a document DENY beats a flag allow + and a document ASK stays asked. Denials the document causes are recorded + to a policy audit file next to the trace; `--tenant` compiles the + document for one tenant and is refused without `--policy`, as `--policy` + itself is refused for delegated execution — `--executor claude-cli` or a + `claude-cli/*` model — which cannot enforce it. + `max_tokens=None` means the default ceiling on the governed path — and is the only value the delegated path accepts, because a ceiling it cannot enforce must be refused rather than silently unapplied. """ + if policy_path is not None and executor == "claude-cli": + # The delegated loop runs inside Claude Code, outside this process's + # policy, approval routing and audit — mapping a document onto CLI + # flags would claim an enforcement that is not there. Refused, like + # the token ceiling the same path cannot honor below. + return fail( + "--policy cannot be enforced under --executor claude-cli: the delegated " + "loop runs outside this process's policy and audit. Drop --executor " + "claude-cli for a governed run", + as_json=as_json, + command="agent", + ) + if tenant is not None and policy_path is None: + return fail( + "--tenant names whose rules a policy document enforces, so it needs " + "--policy to mean anything", + as_json=as_json, + command="agent", + ) if executor == "claude-cli": # The whole loop is Claude Code's; nothing below (registry, harness, # gateway model) applies. `--model` semantics shift too: the delegated @@ -184,7 +355,14 @@ def run_agent( from grapharc.runtime.budget import Budget, BudgetExceeded, BudgetMeter, deadline_guard from grapharc.runtime.graph import RunContext - allow = allow or ["*"] + if policy_path is None: + allow = allow or ["*"] + else: + # No implicit allow-all beside a document: the document's default + # governs unmatched tools, and an implicit `*` would permit what a + # default-deny document refuses. Explicit --allow still votes allow — + # it just cannot outvote the document (see CombinedPolicy). + allow = allow or [] deny = deny or [] ask = ask or [] workspace = Path(workspace).expanduser().resolve() @@ -202,15 +380,74 @@ def run_agent( except optional.Unavailable as exc: return fail(str(exc), as_json=as_json, command="agent") - policy = build_policy(allow, deny, ask) + governed: tuple[Any, str, Path] | None = None + if policy_path is not None: + from grapharc.policy import PolicyError + + try: + audit_path = trace_path.parent / POLICY_AUDIT_FILENAME + engine, tenant_name = _load_agent_document( + Path(policy_path), + tenant=tenant, + audit_path=audit_path, + ) + except PolicyError as exc: + return fail(str(exc), as_json=as_json, command="agent", task=task) + governed = (engine, tenant_name, audit_path) + + if governed is None: + policy = build_policy(allow, deny, ask) + approval: Any = _approval(as_json) + on_denial: Any = None + else: + from grapharc.harness import CombinedPolicy, Decision + + engine, tenant_name, _ = governed + doc_policy = engine.permission_policy(tenant=tenant_name) + # Flags beside a document vote no opinion by default (ALLOW): the + # document decides unmatched tools, and --deny/--ask narrow it. + flag_policy = build_policy(allow, deny, ask, default=Decision.ALLOW) + policy = CombinedPolicy(policies=[doc_policy, flag_policy]) + audit_context = {"command": "agent", "run_id": run_id} + approval = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_policy, + tenant=tenant_name, + as_json=as_json, + context=audit_context, + ) + on_denial = document_denial_recorder( + engine=engine, doc_policy=doc_policy, tenant=tenant_name, context=audit_context + ) harness = Harness( registry, policy, executor=LocalExecutor() if executor == "local" else None, workspace=str(workspace), - approval=_approval(as_json), + approval=approval, + on_denial=on_denial, ) visible = [spec.name for spec in harness.visible_tools()] + # Present only when a document governed the run: the no-document payload + # keeps exactly the keys it has always had. + governance_extra: dict[str, Any] = {} + if governed is not None: + from grapharc.policy.document import ResourceKind + + _engine, _tenant_name, _audit_path = governed + _document = _engine.document + governance_extra = { + "policy_document": { + "source": "flag", + "path": str(policy_path), + "version": _document.version, + "digest": _engine.digest, + "tenant": _tenant_name, + "tool_rules": len(_document.rules_for(ResourceKind.TOOL)), + }, + "policy_audit": str(_audit_path), + } try: from grapharc.gateway import get_model @@ -221,6 +458,22 @@ def run_agent( f"could not build model {model_spec!r}: {exc}", as_json=as_json, command="agent" ) + if governed is not None: + from grapharc.harness.agent import is_claude_cli + + # Delegation triggers on the model, not on --executor: a claude-cli + # model would hand the loop to a subprocess outside the harness this + # policy was just compiled for. Same predicate AgentNode delegates on, + # so the two cannot disagree. + if is_claude_cli(model): + return fail( + "--policy cannot be enforced for a claude-cli model: the loop " + "runs inside Claude Code, outside this process's policy and " + "audit. Use a tool-calling backend for a governed run", + as_json=as_json, + command="agent", + ) + trace = TraceRecorder(trace_path) # The loop's own turn cap bounds iterations, so the meter is left to bound # the two things it alone can see: spend and wall clock. Setting both would @@ -255,6 +508,7 @@ def run_agent( "tools_from": f"grapharc.tools.{entry_point}", "policy": {"allow": allow, "ask": ask, "deny": deny}, "tools_visible": visible, + **governance_extra, } try: @@ -303,6 +557,16 @@ def count(number: int) -> str: """ return style.err(str(number)) if number else str(number) + policy_value = ( + f"{style.dim('allow=')}{allow} {style.dim('ask=')}{ask} {style.dim('deny=')}{deny}" + ) + if governed is not None: + _, _tenant_name, _audit_path = governed + policy_value += ( + f" {style.dim('document=')}{policy_path}" + f" {style.dim('tenant=')}{_tenant_name}" + f" {style.dim('audit=')}{_audit_path}" + ) lines = [ style.kv("task", task, width=width), style.kv("model", model_spec, width=width, tint=style.accent), @@ -314,7 +578,7 @@ def count(number: int) -> str: ), style.kv( "policy", - f"{style.dim('allow=')}{allow} {style.dim('ask=')}{ask} {style.dim('deny=')}{deny}", + policy_value, width=width, ), "", @@ -357,5 +621,7 @@ def count(number: int) -> str: "CORE_TOOL_ENTRY_POINTS", "build_policy", "build_registry", + "combined_approval", + "document_denial_recorder", "run_agent", ] diff --git a/grapharc/cli/main.py b/grapharc/cli/main.py index 0640b72..6a8280a 100644 --- a/grapharc/cli/main.py +++ b/grapharc/cli/main.py @@ -566,6 +566,8 @@ def _cmd_agent(args: argparse.Namespace) -> int: max_seconds=args.max_seconds, executor=args.executor, system_prompt=args.system_prompt, + policy_path=args.policy, + tenant=args.tenant, run_id=args.run_id, as_json=args.json, ) @@ -1130,6 +1132,24 @@ def build_parser() -> argparse.ArgumentParser: "subscription (default: sandbox)" ), ) + agent.add_argument( + "--policy", + type=Path, + default=None, + metavar="PATH", + help=( + "TOML policy document governing this agent's tools. The document " + "is the ceiling: --deny/--ask narrow it further, --allow cannot " + "widen it, and denials are recorded to policy-audit.jsonl next to " + "the trace" + ), + ) + agent.add_argument( + "--tenant", + default=None, + metavar="NAME", + help="tenant to compile --policy for (needs --policy)", + ) agent.add_argument("--system-prompt", default=None) agent.set_defaults(handler=_cmd_agent) diff --git a/grapharc/harness/__init__.py b/grapharc/harness/__init__.py index 3cc6566..1549327 100644 --- a/grapharc/harness/__init__.py +++ b/grapharc/harness/__init__.py @@ -8,10 +8,11 @@ tool_schema, tool_schemas, ) -from grapharc.harness.core import ApprovalCallback, Harness +from grapharc.harness.core import ApprovalCallback, DenialCallback, Harness from grapharc.harness.executor import LocalExecutor, SandboxedExecutor, SandboxViolation from grapharc.harness.hooks import HookAction, HookDecision, PostHook, PreHook from grapharc.harness.permissions import ( + CombinedPolicy, Decision, PermissionDenied, PermissionPolicy, @@ -25,6 +26,8 @@ "AgentNode", "AgentResult", "ApprovalCallback", + "DenialCallback", + "CombinedPolicy", "Decision", "Harness", "HookAction", diff --git a/grapharc/harness/agent.py b/grapharc/harness/agent.py index 8cd58db..a24338c 100644 --- a/grapharc/harness/agent.py +++ b/grapharc/harness/agent.py @@ -355,8 +355,12 @@ class DelegatedToolUseWarning(UserWarning): ) -def _is_claude_cli(model: Any) -> bool: - """Is this the Claude CLI backend? +def is_claude_cli(model: Any) -> bool: + """Is this the Claude CLI backend — the loop that delegates? + + `AgentNode` delegates on this predicate, and `grapharc agent --policy` + refuses on it; one shared test so the two can never disagree about what + "delegated" means. Matched on `_llm_type` rather than `isinstance`, so this module does not import the gateway, and rather than "does it lack bind_tools" — which is @@ -419,7 +423,7 @@ def __init__( #: True when the backend is the Claude CLI, which has no tool-calling #: wire format and therefore cannot be driven as a raw model. The loop #: is handed to Claude Code instead — see `_run_delegated`. - self.delegated = _is_claude_cli(model) + self.delegated = is_claude_cli(model) self.delegated_mode = delegated_mode if self.delegated: if delegated_mode == "bypass": diff --git a/grapharc/harness/core.py b/grapharc/harness/core.py index be3266a..80a075c 100644 --- a/grapharc/harness/core.py +++ b/grapharc/harness/core.py @@ -9,6 +9,8 @@ from __future__ import annotations +import logging +import threading from collections.abc import Callable from typing import Any @@ -17,8 +19,14 @@ from grapharc.harness.permissions import Decision, PermissionDenied, PermissionPolicy from grapharc.harness.tools import ToolRegistry, ToolSpec +_log = logging.getLogger(__name__) + # (tool_name, args) -> approved? Human checkpoints implement this. ApprovalCallback = Callable[[str, dict[str, Any]], bool] +# (tool_name) -> None. Fired when the policy denies a call, so a caller that +# records enforcement elsewhere (a policy document's audit log) hears about +# the denials this object otherwise answers silently. +DenialCallback = Callable[[str], None] class Harness: @@ -31,6 +39,7 @@ def __init__( pre_hooks: tuple[PreHook, ...] = (), post_hooks: tuple[PostHook, ...] = (), approval: ApprovalCallback | None = None, + on_denial: DenialCallback | None = None, workspace: str | None = None, ) -> None: self.registry = registry @@ -50,6 +59,11 @@ def __init__( self.pre_hooks = pre_hooks self.post_hooks = post_hooks self.approval = approval + self.on_denial = on_denial + #: Denial-hook exceptions swallowed so far. Non-zero means enforcement + #: records are missing somewhere downstream — see `_notify_denial`. + self.denial_hook_errors = 0 + self._denial_lock = threading.Lock() def visible_tools(self) -> list[ToolSpec]: """The tool schemas a model may see — policy-filtered before exposure.""" @@ -62,6 +76,7 @@ def call(self, tool_name: str, args: dict[str, Any]) -> Any: decision = self.policy.decide(tool_name) if decision is Decision.DENY: + self._notify_denial(tool_name) raise PermissionDenied(f"tool {tool_name!r} denied by policy") if decision is Decision.ASK: # Fail closed: no approval channel means no approval. @@ -87,3 +102,25 @@ def call(self, tool_name: str, args: dict[str, Any]) -> Any: for post in self.post_hooks: result = post(tool_name, dict(args), result) return result + + def _notify_denial(self, tool_name: str) -> None: + """Tell `on_denial` about a refused call, without fail. + + The hook runs before the `PermissionDenied` it annotates, and its + exceptions are counted in `denial_hook_errors` and swallowed: a + recorder must never turn a denied call into a tool error, which is + what an exception here would become one frame up in `AgentNode`. + Only the policy-DENY branch notifies — an approval refused by a + human is the approval path's record to write, and an unknown tool + is not a policy decision at all. + """ + if self.on_denial is None: + return + try: + self.on_denial(tool_name) + except Exception: + with self._denial_lock: + self.denial_hook_errors += 1 + _log.exception( + "denial hook raised for tool %r; the denial stands", tool_name + ) diff --git a/grapharc/harness/permissions.py b/grapharc/harness/permissions.py index 5b082c7..1e0e8a9 100644 --- a/grapharc/harness/permissions.py +++ b/grapharc/harness/permissions.py @@ -18,7 +18,7 @@ from enum import StrEnum from fnmatch import fnmatch -from pydantic import BaseModel +from pydantic import BaseModel, field_validator, model_validator class Decision(StrEnum): @@ -90,3 +90,61 @@ def decide(self, tool_name: str) -> Decision: if rule.action == tier and rule.matches(tool_name): return tier return self.default + + +class CombinedPolicy(PermissionPolicy): + """Two or more policies consulted together; the most restrictive wins. + + Tiered semantics lifted from rules to policies: a tool is DENY when any + side denies it, ASK when any side asks and none denies, and ALLOW only + when every side allows. The order of `policies` does not matter — unlike + rule order within one policy there is no first match here, only the + strictest verdict. + + The use case is a ceiling plus refinements: a policy document compiled by + `PolicyEngine.permission_policy()` sets the maximum authority, and a + flag-built policy can only narrow it. A CLI `--allow` votes ALLOW exactly + as before, but it can never outvote a document DENY or quietly demote a + document ASK — most-restrictive is precisely "flags cannot widen the + document". The flag side therefore carries default ALLOW: with no flag + opinion on a tool, the document decides it alone. + + `rules` and `default` are inherited and rejected, not merged: this answers + only from `policies`, and a rule written on the combination itself would + be silently ignored, which for a DENY rule fails open. `Harness` and + `ToolRegistry.visible` speak `PermissionPolicy`, so this subclasses it and + drops into either unchanged. + """ + + policies: list[PermissionPolicy] = [] + + @field_validator("policies") + @classmethod + def _require_policies( + cls, policies: list[PermissionPolicy] + ) -> list[PermissionPolicy]: + if not policies: + raise ValueError("CombinedPolicy needs at least one policy to combine") + return policies + + @model_validator(mode="after") + def _forbid_own_rules(self) -> CombinedPolicy: + if self.rules or self.default is not Decision.DENY: + raise ValueError( + "CombinedPolicy answers from `policies`, not from its own rules: " + "append another PermissionPolicy instead of writing rules here" + ) + return self + + def decide(self, tool_name: str) -> Decision: + if not self.policies: + # Unreachable from the constructor — the validator refuses an + # empty list — but reachable by post-construction mutation, and + # an empty verdict set must deny, never fall through to allow. + return Decision.DENY + decisions = {policy.decide(tool_name) for policy in self.policies} + if Decision.DENY in decisions: + return Decision.DENY + if Decision.ASK in decisions: + return Decision.ASK + return Decision.ALLOW diff --git a/tests/test_cli.py b/tests/test_cli.py index c997320..0e8f00a 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -23,6 +23,7 @@ import subprocess import sys import time +from io import StringIO from pathlib import Path from types import ModuleType @@ -30,11 +31,17 @@ from langchain_core.messages import AIMessage from pydantic import BaseModel -from grapharc.cli.agent import _approval +from grapharc.cli.agent import ( + _approval, + build_policy, + combined_approval, + document_denial_recorder, +) from grapharc.cli.main import build_parser, main from grapharc.examples.stage0_dag import DEMO_DOC, build_stage0 -from grapharc.harness import ToolSpec +from grapharc.harness import Decision, ToolSpec from grapharc.observe.trace import TraceRecorder +from grapharc.policy import PolicyEngine # -- harness ------------------------------------------------------------------ @@ -2481,3 +2488,913 @@ def test_default_flag_forces_the_builtin_kinds(tmp_path, monkeypatch, capsys): payload = json.loads(capsys.readouterr().out) assert code == 0 assert payload["registry"] == "grapharc.stdlib:build_registry" + + +# -- agent --policy (issue #6) -------------------------------------------------- + + +DOC_DENY_SHELL = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "no-shell" +resource = "tool" +match = "run_command" +effect = "deny" +reason = "the shell is never permitted" + +[[rule]] +id = "notes-ok" +resource = "tool" +match = "write_note" +effect = "allow" +""" + +DOC_ALLOW_WRITES = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "notes-ok" +resource = "tool" +match = "write_*" +effect = "allow" +""" + +DOC_ASK_WRITES = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "notes-ask" +resource = "tool" +match = "write_note" +effect = "ask" +approver_role = "reviewer" +reason = "writes need a human" +""" + +DOC_TENANTED = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" +tenants = ["acme", "globex"] + +[[rule]] +id = "acme-notes" +resource = "tool" +match = "write_*" +effect = "allow" +tenant = "acme" +""" + +DOC_DENY_WRITES = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "no-writes" +resource = "tool" +match = "write_note" +effect = "deny" +reason = "read-only run" +""" + +DOC_NO_TOOLS = """ +version = "1.0.0" +name = "agent-cli-test" +default = "deny" + +[[rule]] +id = "chain" +resource = "edge" +match = "triage->fix" +effect = "allow" +""" + + +def _write_doc(path: Path, text: str) -> Path: + path.write_text(text.strip() + "\n", encoding="utf-8") + return path + + +def _read_jsonl(path) -> list[dict]: + """Audit/trace lines as dicts. A run that recorded nothing has no file yet — + the same contract TraceRecorder keeps — which reads as zero records.""" + target = Path(path) + if not target.exists(): + return [] + return [ + json.loads(line) + for line in target.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +class _TtyInput(StringIO): + """Piped bytes with a tty face, for the interactive approval path.""" + + def isatty(self) -> bool: + return True + + +def _refuse_to_prompt(*args, **kwargs): + raise AssertionError("prompted for approval in a non-interactive run") + + +def test_agent_policy_document_denies_a_tool_end_to_end( + tmp_path, capsys, scripted_model, stub_tools +): + """Issue #6: the document governs the run — denied tools are undescribed, + unexecuted, and recorded in both the audit log and the trace.""" + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + model = scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "do the thing", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--run-id", + "doc-test", + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to what this test + # proves (policy visibility, denial, audit, trace). + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + # Selectively governed: the denied tool is undescribed, the allowed one runs. + assert model.bound_tools == ["write_note"] + assert payload["tools_visible"] == ["write_note"] + assert payload["policy"] == {"allow": [], "ask": [], "deny": []} + assert (workspace / "note.txt").read_text(encoding="utf-8") == "hello" + assert payload["tool_calls"][0]["tool"] == "run_command" + assert payload["tool_calls"][0]["status"] == "denied" + assert payload["tool_calls"][0]["refused_by"] == "policy" + assert payload["tool_calls"][1]["status"] == "ok" + + # The denial is attributable: rule, version, digest, tenant, effect. + audit = _read_jsonl(payload["policy_audit"]) + decision = next( + e for e in audit if e["kind"] == "decision" and e["subject"] == "run_command" + ) + assert decision["effect"] == "deny" + assert decision["rule_id"] == "no-shell" + assert decision["policy_version"] == "1.0.0" + assert decision["tenant"] == "default" + assert decision["policy_digest"] == payload["policy_document"]["digest"] + + # ...and the same refusal is on the run trace, not only in the audit log. + trace = _read_jsonl(payload["trace"]) + refusal = next( + e + for e in trace + if e.get("phase") == "tool" + and (e.get("state_delta") or {}).get("tool") == "run_command" + ) + assert refusal["state_delta"]["status"] == "denied" + assert refusal["state_delta"]["refused_by"] == "policy" + assert "PERMISSION_DENIED" in (refusal.get("error") or "") + + +def test_agent_flag_allow_cannot_widen_a_document_deny( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + model = scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--allow", + "run_command", + ], + capsys, + ) + assert code == 0 + assert payload["policy"]["allow"] == ["run_command"] + assert "run_command" not in model.bound_tools + assert payload["tool_calls"][0]["status"] == "denied" + audit = _read_jsonl(payload["policy_audit"]) + assert any( + e["kind"] == "decision" + and e["subject"] == "run_command" + and e["effect"] == "deny" + and e["rule_id"] == "no-shell" + for e in audit + ) + + +def test_agent_flag_deny_narrows_a_document_allow( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ALLOW_WRITES) + model = scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--deny", + "write_note", + ], + capsys, + ) + assert code == 0 + assert "write_note" not in (model.bound_tools or []) + assert payload["tool_calls"][0]["status"] == "denied" + assert not (workspace / "note.txt").exists() + # The flag caused this refusal, not the document — recording it through + # the engine would write ALLOW beside a denial, so the audit stays empty. + assert _read_jsonl(payload["policy_audit"]) == [] + + +def test_agent_flag_allow_reaffirms_a_document_allow( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ALLOW_WRITES) + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--allow", + "write_note", + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to the reaffirmed + # allow this test proves. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert payload["tool_calls"][0]["status"] == "ok" + assert (workspace / "note.txt").read_text(encoding="utf-8") == "hello" + + +def test_agent_flag_allow_cannot_bypass_a_document_ask( + tmp_path, capsys, scripted_model, stub_tools, monkeypatch +): + """Document ASK plus flag allow stays asked — and in JSON mode the human + it asks for does not exist, so the call is refused, not granted.""" + monkeypatch.setattr("builtins.input", _refuse_to_prompt) + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ASK_WRITES) + model = scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--allow", + "write_note", + ], + capsys, + ) + assert code == 0 + # Asked tools stay visible — ask gates the call, it does not hide the schema. + assert "write_note" in model.bound_tools + assert payload["tool_calls"][0]["status"] == "denied" + assert not (workspace / "note.txt").exists() + audit = _read_jsonl(payload["policy_audit"]) + decision = next(e for e in audit if e["kind"] == "decision") + assert decision["effect"] == "ask" + assert decision["rule_id"] == "notes-ask" + assert decision["approver_role"] == "reviewer" + approval = next(e for e in audit if e["kind"] == "approval") + assert approval["granted"] is False + assert approval["approver_role"] == "reviewer" + # The guard refused — had `input()` been reached, the router would have + # recorded the stub's AssertionError as a handler failure instead. + assert approval["reason"] == "reviewer refused approval" + + +def test_agent_document_deny_plus_flag_deny_is_still_denied( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + "--deny", + "run_command", + ], + capsys, + ) + assert code == 0 + assert payload["tool_calls"][0]["status"] == "denied" + audit = _read_jsonl(payload["policy_audit"]) + assert any(e["kind"] == "decision" and e["rule_id"] == "no-shell" for e in audit) + + +def test_agent_denied_tool_never_reaches_the_executor( + tmp_path, capsys, scripted_model, stub_tools +): + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_WRITES) + model = scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 0 + assert "write_note" not in (model.bound_tools or []) + assert payload["tool_calls"][0]["status"] == "denied" + assert payload["tool_calls"][0]["refused_by"] == "policy" + assert not (workspace / "note.txt").exists() + + +def test_agent_policy_tenant_allow_and_deny( + tmp_path, capsys, scripted_model, stub_tools +): + """One tenant's grant is not another's; the ungranted tenant is denied.""" + stub_tools() + policy = _write_doc(tmp_path / "policy.toml", DOC_TENANTED) + + granted_ws = tmp_path / "granted" + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + code, granted, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(granted_ws), + "--policy", + str(policy), + "--tenant", + "acme", + # Local execution on both halves of this comparison: the + # sandbox executor forks, which Windows cannot do. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert granted["tool_calls"][0]["status"] == "ok" + assert (granted_ws / "note.txt").read_text(encoding="utf-8") == "hello" + assert granted["policy_document"]["tenant"] == "acme" + + refused_ws = tmp_path / "refused" + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, refused, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(refused_ws), + "--policy", + str(policy), + "--tenant", + "globex", + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert refused["tool_calls"][0]["status"] == "denied" + assert not (refused_ws / "note.txt").exists() + audit = _read_jsonl(refused["policy_audit"]) + decision = next(e for e in audit if e["kind"] == "decision") + assert decision["effect"] == "deny" + assert decision["tenant"] == "globex" + assert decision["rule_id"] is None # the document default, not a rule + + +def test_agent_policy_unknown_tenant_is_refused_upfront(tmp_path, capsys): + policy = _write_doc(tmp_path / "policy.toml", DOC_TENANTED) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + "--tenant", + "nope", + ], + capsys, + ) + assert code == 2 + assert payload["ok"] is False + assert "nope" in payload["error"] + assert "declared" in payload["error"] + + +def test_agent_policy_missing_tenant_is_refused_when_tenants_are_listed( + tmp_path, capsys +): + policy = _write_doc(tmp_path / "policy.toml", DOC_TENANTED) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 2 + assert "'default'" in payload["error"] + assert "pass --tenant" in payload["error"] + + +def test_agent_policy_ask_fails_closed_in_json_mode( + tmp_path, capsys, scripted_model, stub_tools, monkeypatch +): + """JSON means no human is reading: ask denies, never prompts, and the + refusal is structured — the run still completes so the document's shape + is visible instead of a traceback.""" + monkeypatch.setattr("builtins.input", _refuse_to_prompt) + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ASK_WRITES) + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "blocked"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 0 + assert payload["denied"] == 1 + assert payload["tool_calls"][0]["refused_by"] == "policy" + assert not (workspace / "note.txt").exists() + audit = _read_jsonl(payload["policy_audit"]) + assert any(e["kind"] == "decision" and e["effect"] == "ask" for e in audit) + approval = next(e for e in audit if e["kind"] == "approval") + assert approval["granted"] is False + assert approval["reason"] == "reviewer refused approval" + + +def test_agent_policy_ask_grants_interactively( + tmp_path, capsys, scripted_model, stub_tools, monkeypatch +): + """On a terminal the role, rule and reason reach the human, and a yes runs.""" + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_ASK_WRITES) + scripted_model( + [ + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "wrote the note"}, + ] + ) + monkeypatch.setattr("sys.stdin", _TtyInput("y\n")) + code, out, _ = call( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to the granted + # approval this test proves. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + assert (workspace / "note.txt").read_text(encoding="utf-8") == "hello" + assert "reviewer" in out + assert "notes-ask" in out + + +def test_agent_policy_refuses_the_delegated_executor(tmp_path, capsys): + """A document mapped onto Claude Code flags would claim an enforcement + that is not there — refused before the (missing) file is even read.""" + code, payload, _ = call_json( + [ + "agent", + "t", + "--executor", + "claude-cli", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(tmp_path / "nope.toml"), + ], + capsys, + ) + assert code == 2 + assert "--policy" in payload["error"] + assert "claude-cli" in payload["error"] + + +def test_agent_tenant_without_policy_is_refused(tmp_path, capsys): + code, payload, _ = call_json( + ["agent", "t", "--workspace", str(tmp_path / "ws"), "--tenant", "acme"], + capsys, + ) + assert code == 2 + assert "--tenant" in payload["error"] + assert "--policy" in payload["error"] + + +def test_agent_policy_missing_file_is_a_structured_error(tmp_path, capsys): + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(tmp_path / "nope.toml"), + ], + capsys, + ) + assert code == 2 + assert payload["ok"] is False + assert payload["command"] == "agent" + assert "nope.toml" in payload["error"] + + +def test_agent_policy_without_tool_rules_is_refused(tmp_path, capsys): + """A planner-only document constrains nothing an agent does — running + under it would report a policy that is not in force.""" + policy = _write_doc(tmp_path / "policy.toml", DOC_NO_TOOLS) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 2 + assert "tool rules" in payload["error"] + + +def test_agent_policy_provenance_is_reported( + tmp_path, capsys, scripted_model, stub_tools +): + """The payload says which document governed the run — and the digest it + names is the one the audit records carry.""" + stub_tools() + workspace = tmp_path / "ws" + policy = _write_doc(tmp_path / "policy.toml", DOC_DENY_SHELL) + scripted_model( + [ + {"tools": [("run_command", {"argv": "id"})]}, + {"tools": [("write_note", {"path": "note.txt", "content": "hello"})]}, + {"content": "done"}, + ] + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(workspace), + "--policy", + str(policy), + # Local execution: the sandbox executor forks, which Windows + # cannot do — and confinement is orthogonal to the provenance + # this test proves. + "--executor", + "local", + ], + capsys, + ) + assert code == 0 + document = payload["policy_document"] + assert document["source"] == "flag" + assert document["path"] == str(policy) + assert document["version"] == "1.0.0" + assert document["tenant"] == "default" + assert document["tool_rules"] == 2 + assert len(document["digest"]) == 64 + audit = _read_jsonl(payload["policy_audit"]) + assert audit, "a governed run that denied a tool must have audit records" + assert {e["policy_digest"] for e in audit} == {document["digest"]} + + text_ws = tmp_path / "text-ws" + scripted_model([{"content": "nothing to do"}]) + code, out, _ = call( + [ + "agent", + "t", + "--model", + "mock/x", + "--workspace", + str(text_ws), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 0 + assert "document=" in out + assert "tenant=default" in out + assert "audit=" in out + + +def test_agent_combined_approval_grants_only_when_every_asking_side_grants(monkeypatch): + engine = PolicyEngine.from_toml(DOC_ASK_WRITES) + doc_policy = engine.permission_policy(tenant="default") + flag_open = build_policy(["*"], [], []) + flag_ask = build_policy([], [], ["write_note"]) + + # Answers travel through `builtins.input`, the way the handlers read + # them: `stream` only decides whether a terminal is behind stdin. + prompts: list[str] = [] + script: list[str] = [] + + def respond(prompt: str) -> str: + prompts.append(prompt) + return script.pop(0) + + monkeypatch.setattr("builtins.input", respond) + + approve = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_open, + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + script.append("y") + assert approve("write_note", {"path": "n"}) is True + assert script == [] # the answer was consumed, exactly once + assert len(prompts) == 1 + assert "reviewer" in prompts[0] + assert "notes-ask" in prompts[0] + + both = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_ask, + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + script.extend(["y", "y"]) + assert both("write_note", {}) is True + assert script == [] + + vetoed = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_ask, + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + script.extend(["y", "n"]) + assert vetoed("write_note", {}) is False + assert script == [] + + asked = len(prompts) + refused = combined_approval( + engine=engine, + doc_policy=doc_policy, + flag_policy=flag_open, + tenant="default", + as_json=True, + stream=_TtyInput(), + ) + assert refused("write_note", {}) is False + assert len(prompts) == asked # JSON mode never prompts, not even on a tty + + open_engine = PolicyEngine.from_toml(DOC_ALLOW_WRITES) + nobody_asks = combined_approval( + engine=open_engine, + doc_policy=open_engine.permission_policy(tenant="default"), + flag_policy=build_policy(["*"], [], []), + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + assert nobody_asks("write_note", {}) is False + assert len(prompts) == asked + + +def test_agent_combined_approval_skips_the_second_prompt_after_a_refusal(monkeypatch): + engine = PolicyEngine.from_toml(DOC_ASK_WRITES) + prompts: list[str] = [] + + def refuse(prompt: str) -> str: + prompts.append(prompt) + return "n" + + monkeypatch.setattr("builtins.input", refuse) + approve = combined_approval( + engine=engine, + doc_policy=engine.permission_policy(tenant="default"), + flag_policy=build_policy([], [], ["write_note"]), + tenant="default", + as_json=False, + stream=_TtyInput(), + ) + assert approve("write_note", {}) is False + # The document's role refused first, so the flag prompt never fired. + assert len(prompts) == 1 + assert "reviewer" in prompts[0] + + +def test_agent_document_denial_recorder_skips_flag_side_denials(): + """Recording a flag-side denial through the engine would write ALLOW + beside a refusal — so the hook records document denials only.""" + engine = PolicyEngine.from_toml(DOC_ALLOW_WRITES) + record = document_denial_recorder( + engine=engine, + doc_policy=engine.permission_policy(tenant="default"), + tenant="default", + ) + record("write_note") # the document allows it; some flag denied it + assert len(engine.audit) == 0 + + strict = PolicyEngine.from_toml(DOC_DENY_SHELL) + record_doc_deny = document_denial_recorder( + engine=strict, + doc_policy=strict.permission_policy(tenant="default"), + tenant="default", + context={"command": "agent", "run_id": "r1"}, + ) + record_doc_deny("run_command") + assert len(strict.audit) == 1 + entry = strict.audit.entries()[0] + assert entry.effect is Decision.DENY + assert entry.rule_id == "no-shell" + assert entry.tenant == "default" + assert entry.context == {"command": "agent", "run_id": "r1"} + + +def test_agent_policy_refuses_a_claude_cli_model( + tmp_path, capsys, monkeypatch, stub_tools +): + """Delegation triggers on the model, not on --executor: a claude-cli model + would hand the loop to a subprocess outside the governed harness.""" + from types import SimpleNamespace + + stub_tools() + policy = _write_doc(tmp_path / "policy.toml", DOC_ALLOW_WRITES) + monkeypatch.setattr( + "grapharc.gateway.get_model", + lambda *args, **kwargs: SimpleNamespace(_llm_type="grapharc-claude-cli"), + ) + code, payload, _ = call_json( + [ + "agent", + "t", + "--model", + "claude-cli/opus-4-6", + "--workspace", + str(tmp_path / "ws"), + "--policy", + str(policy), + ], + capsys, + ) + assert code == 2 + assert "--policy" in payload["error"] + assert "claude-cli" in payload["error"] diff --git a/tests/test_delegate_tiers.py b/tests/test_delegate_tiers.py index d80bc51..9cf7f0f 100644 --- a/tests/test_delegate_tiers.py +++ b/tests/test_delegate_tiers.py @@ -29,7 +29,7 @@ class ClaudeCliDouble(ScriptedChatModel): - """Looks like the Claude CLI backend to `_is_claude_cli`, runs nothing.""" + """Looks like the Claude CLI backend to `is_claude_cli`, runs nothing.""" @property def _llm_type(self) -> str: diff --git a/tests/test_harness_gate.py b/tests/test_harness_gate.py index 4d84294..92f5cae 100644 --- a/tests/test_harness_gate.py +++ b/tests/test_harness_gate.py @@ -22,8 +22,10 @@ import uuid import pytest +from pydantic import ValidationError from grapharc.harness import ( + CombinedPolicy, Decision, Harness, HookAction, @@ -1220,3 +1222,176 @@ def signal_self() -> str: with pytest.raises(SandboxViolation, match="signal another process"): harness.call("signal_pid", {"pid": os.getpid()}) assert harness.call("signal_self", {}) == "signalled" + + +# -- CombinedPolicy ----------------------------------------------------------- + + +def test_combined_policy_denies_when_either_side_denies(): + """Most restrictive wins: one DENY outvotes any number of allows.""" + doc = _policy([{"action": "deny", "pattern": "run_command"}]) + flags = _policy([{"action": "allow", "pattern": "*"}]) + assert CombinedPolicy(policies=[doc, flags]).decide("run_command") is Decision.DENY + # Order does not matter: there is no first match between policies. + assert CombinedPolicy(policies=[flags, doc]).decide("run_command") is Decision.DENY + + +def test_combined_policy_asks_when_either_side_asks_and_neither_denies(): + doc = _policy([{"action": "ask", "pattern": "deploy_*"}]) + flags = _policy([{"action": "allow", "pattern": "*"}]) + assert CombinedPolicy(policies=[doc, flags]).decide("deploy_prod") is Decision.ASK + # An ask on the flag side gates a document allow, the way --ask narrows. + doc_allow = _policy([{"action": "allow", "pattern": "write_*"}]) + flag_ask = _policy([{"action": "ask", "pattern": "write_note"}]) + assert ( + CombinedPolicy(policies=[doc_allow, flag_ask]).decide("write_note") + is Decision.ASK + ) + + +def test_combined_policy_allows_only_when_every_side_allows(): + doc = _policy([{"action": "allow", "pattern": "read_*"}]) + flags = PermissionPolicy(rules=[], default=Decision.ALLOW) + assert ( + CombinedPolicy(policies=[doc, flags]).decide("read_file") is Decision.ALLOW + ) + # ...and a default-deny on either side holds for unmatched tools. + strict_flags = PermissionPolicy(rules=[], default=Decision.DENY) + assert ( + CombinedPolicy(policies=[doc, strict_flags]).decide("read_file") + is Decision.DENY + ) + + +def test_combined_policy_needs_at_least_one_policy(): + with pytest.raises(ValidationError, match="at least one policy"): + CombinedPolicy(policies=[]) + + +def test_combined_policy_denies_when_its_sides_are_emptied_after_construction(): + """Defense in depth: the validator refuses an empty list, and `decide` + refuses to allow on one — so post-construction mutation fails closed.""" + combined = CombinedPolicy(policies=[_policy([{"action": "allow", "pattern": "*"}])]) + combined.policies = [] + assert combined.decide("anything") is Decision.DENY + + +def test_combined_policy_rejects_rules_written_on_the_combination_itself(): + """Own rules would be silently ignored — for a DENY that fails open.""" + side = _policy([{"action": "allow", "pattern": "*"}]) + with pytest.raises(ValidationError, match="answers from `policies`"): + CombinedPolicy( + policies=[side], + rules=[PermissionRule(action=Decision.DENY, pattern="x")], + ) + with pytest.raises(ValidationError, match="answers from `policies`"): + CombinedPolicy(policies=[side], default=Decision.ALLOW) + + +def test_combined_policy_hides_what_either_side_denies(): + """Policy-before-schema survives combination: denied means undescribed. + + The document side allows by default here, so the one hidden tool is + hidden by its explicit DENY — which a flag-side `ALLOW *` cannot widen. + """ + registry = ToolRegistry() + registry.register(ToolSpec(name="read_file", description="read", fn=_echo)) + registry.register(ToolSpec(name="run_command", description="shell", fn=_echo)) + doc = PermissionPolicy( + rules=[PermissionRule(action=Decision.DENY, pattern="run_command")], + default=Decision.ALLOW, + ) + flags = _policy([{"action": "allow", "pattern": "*"}]) + visible = registry.visible(CombinedPolicy(policies=[doc, flags])) + assert [spec.name for spec in visible] == ["read_file"] + + +def test_combined_policy_default_deny_side_is_not_widened_by_an_allow_all(): + """The opposite invariant: a default-deny document denies an unmatched + tool even beside `ALLOW *` — the default is the side's verdict, and + flags narrow a document, never widen it.""" + registry = ToolRegistry() + registry.register(ToolSpec(name="read_file", description="read", fn=_echo)) + registry.register(ToolSpec(name="run_command", description="shell", fn=_echo)) + doc = _policy([{"action": "deny", "pattern": "run_command"}]) # default DENY + flags = _policy([{"action": "allow", "pattern": "*"}]) + combined = CombinedPolicy(policies=[doc, flags]) + assert combined.decide("read_file") is Decision.DENY + assert [spec.name for spec in registry.visible(combined)] == [] + + +# -- denial hook -------------------------------------------------------------- + + +class _EchoExecutor: + """Runs everything: denials are the policy's business, not the executor's.""" + + def run(self, spec, args): + return {"ran": spec.name} + + +def _hook_harness(policy, *, approval=None): + registry = ToolRegistry() + registry.register(ToolSpec(name="write_note", description="write", fn=_echo)) + registry.register(ToolSpec(name="list_notes", description="list", fn=_echo)) + fired: list[str] = [] + harness = Harness( + registry, + policy, + executor=_EchoExecutor(), + approval=approval, + on_denial=fired.append, + ) + return harness, fired + + +def test_harness_notifies_on_denial_for_a_denied_tool(): + harness, fired = _hook_harness(_policy([{"action": "deny", "pattern": "write_*"}])) + with pytest.raises(PermissionDenied, match="denied by policy"): + harness.call("write_note", {"path": "n.txt"}) + assert fired == ["write_note"] + + +def test_harness_denial_hook_stays_silent_otherwise(): + """Allowances, granted approvals and unknown tools are not policy denials; + a refused approval already belongs to the approval path's own record.""" + harness, fired = _hook_harness( + _policy( + [ + {"action": "ask", "pattern": "write_note"}, + {"action": "allow", "pattern": "*"}, + ] + ), + approval=lambda tool, args: True, + ) + assert harness.call("list_notes", {}) == {"ran": "list_notes"} + assert harness.call("write_note", {}) == {"ran": "write_note"} + with pytest.raises(PermissionDenied, match="unknown tool"): + harness.call("no_such_tool", {}) + assert fired == [] + + refused, fired_refused = _hook_harness( + _policy( + [ + {"action": "ask", "pattern": "write_note"}, + {"action": "allow", "pattern": "*"}, + ] + ), + approval=lambda tool, args: False, + ) + with pytest.raises(PermissionDenied, match="requires approval"): + refused.call("write_note", {}) + assert fired_refused == [] + + +def test_a_failing_denial_hook_cannot_break_the_denial(): + """A broken recorder must not turn a denial into a tool error one frame up.""" + harness, _ = _hook_harness(_policy([{"action": "deny", "pattern": "*"}])) + + def broken(tool_name): + raise OSError("audit disk is gone") + + harness.on_denial = broken + with pytest.raises(PermissionDenied, match="denied by policy"): + harness.call("write_note", {}) + assert harness.denial_hook_errors == 1 From bf424a61dd334dceefd77a9027f4c84a25e17b95 Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh <123410790+Shashankss1205@users.noreply.github.com> Date: Fri, 9 Oct 2026 21:42:03 +0530 Subject: [PATCH 2/3] fix(cli): apply configured policy to agent execution Resolve agent policy and tenant with the shared flag, environment and config precedence before model setup or delegation. Preserve config-relative paths and report the actual policy source. Add six regression cases and update governance documentation and selected-test figures. This follow-up is based on contributor PR #131; the original contributor commit remains its parent. --- README.md | 2 +- ROADMAP.md | 18 ++++-- docs/cookbook/03-agents-and-tools.md | 37 +++++++++++ docs/deep-dive.md | 7 ++- grapharc/cli/agent.py | 25 ++++++-- grapharc/cli/main.py | 4 +- tests/test_cli.py | 91 ++++++++++++++++++++++++++++ 7 files changed, 169 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 2266a16..3852a7d 100644 --- a/README.md +++ b/README.md @@ -220,7 +220,7 @@ The edges are documented, not denied — the full list with mechanisms is in the - The in-process sandbox is defense in depth; `ContainerExecutor` is the real boundary. `run_command` children are unconfined. - The HTTP API does not yet use the durable session layer. - On the Claude CLI backend an agent node is *delegated*, not governed: by default it runs under an allowlist mapped from the node's own tools, but enforcement there is Claude Code's, and the `bypass` tier — explicit opt-in — has no checks at all. -- Policy documents govern planning; the tool plane still reads CLI flags. +- Agent tool policies are enforced on tool-calling backends. A policy document refuses delegated Claude CLI runs; allowed tool calls are not written to the document audit. - The MCP gate binds the MCP surface, not the host: an agent with its own file tools in the run directory could forge the approval decision. The trust boundary is the working directory, as it is for the Slack workspace. Version `0.1.8` · [changelog](CHANGELOG.md) · [roadmap](ROADMAP.md) · [website](https://codegraphcontext.github.io/GraphARC/) · MIT diff --git a/ROADMAP.md b/ROADMAP.md index d35b8f7..93ca052 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -324,9 +324,11 @@ The component with no prior art to copy. It exists, and the cycle runs. runner from claiming a session, and nothing reclaims one whose runner died holding it. That is a claim, not a lease. -## 7. Policy engine — `[~] ~75% built, 0% wired` +## 7. Policy engine — `[~] node, edge and tool paths wired` -Everything here works and nothing calls it. +The planner's admission gate and the agent CLI compile policy documents into +their existing permission boundaries. Compiled rules do not audit ordinary +allowances; the agent records document-driven denials and approval decisions. - [x] **7.1 — Declarative policy config** over nodes, edges, tools and spend. TOML in, `PolicyEngine` out; a commented example ships at @@ -339,7 +341,7 @@ Everything here works and nothing calls it. `engine.approval_router(handlers, tenant=…)` produces the callback a `Harness` already obeys, and `engine.permission_policy(tenant=…)` produces a real `PermissionPolicy`. -- [x] **7.3 — Policy versioning and decision audit.** Every decision lands in a +- [x] **7.3 — Policy versioning and decision audit.** Every direct engine check lands in a JSONL record naming the resource, subject, tenant, effect, the rule id and reason that produced it, the policy version, and a digest of the document — so a decision can be tied to the exact policy text that made it. @@ -355,9 +357,13 @@ Everything here works and nothing calls it. longer imported by nothing. What the compiled object still cannot carry is what `permission_policy()` cannot either: the approver role and the audit record, because `EdgePolicy.decide` returns a bare `Decision`. Admission - treats `ask` as not-yet-permitted. Still open: no call from `AgentNode` or - `grapharc agent` to `permission_policy()`, so the tool plane is still - governed by Python objects rather than by the document. + treats `ask` as not-yet-permitted. `grapharc agent --policy` now compiles + tool rules through `permission_policy()`, pairs document `ask` with + `approval_router()`, and records document denials through `check_tool()`. + Flags can narrow the document but cannot widen it. Policy and tenant + follow the shared flag/environment/config precedence, and a configured + policy refuses delegated execution just as an explicit one does. Ordinary + allowed tool calls still have no document-audit record. ## 8. Memory & artifacts — `[~] ~85%` diff --git a/docs/cookbook/03-agents-and-tools.md b/docs/cookbook/03-agents-and-tools.md index e48aa0b..5c499b4 100644 --- a/docs/cookbook/03-agents-and-tools.md +++ b/docs/cookbook/03-agents-and-tools.md @@ -1601,3 +1601,40 @@ grapharc agent --workspace ./scratch --deny 'run_command' --ask 'write_file' \ A trace lands at `/trace.jsonl` either way; `grapharc trace`, `grapharc metrics` and `grapharc viz` read it. + +### How does a policy document govern the CLI agent? + +`grapharc agent --policy policy.toml --tenant acme` applies the document's +`resource = "tool"` rules through the same harness that filters tool schemas +before the model sees them. Flags can narrow that authority: any denial wins, +then any approval requirement, and a tool runs freely only when both sides +allow it. `--allow '*'` cannot override a document denial. + +Policy and tenant can also come from configuration: + +```toml +# grapharc.toml +[grapharc] +policy = "policy.toml" +tenant = "acme" +``` + +The document must declare `acme` when it lists tenants. Resolution is +`flag > GRAPHARC_POLICY / GRAPHARC_TENANT > grapharc.toml > default`. +`--config PATH` selects another file; a relative policy path in that file is +anchored to its directory. Parent directories are not searched. This agent +config path resolves policy and tenant only; model, budgets and the other agent +options retain their CLI/Python-argument defaults. + +A governed JSON result includes `config_file`, `sources`, `policy_source`, and +`policy_document` with the document's path, source, version, digest and tenant. +The human view includes the document's source beside its path. The +`policy_audit` path points to `policy-audit.jsonl` beside the trace, recording +document-driven denials and approval decisions. Ordinary allowed calls and +flag-only denials are not document-audited; attempted calls and their outcomes +remain on the run trace. + +Document approval requests fail closed under `--json` or redirected stdin. +A policy selected from flags, environment or config also refuses +`--executor claude-cli` and Claude CLI models: that delegated loop cannot +enforce the document. Bad config is refused before model setup or delegation. diff --git a/docs/deep-dive.md b/docs/deep-dive.md index 1338f90..7c20b2d 100644 --- a/docs/deep-dive.md +++ b/docs/deep-dive.md @@ -139,7 +139,9 @@ Three things that phrase over-promises if left alone. **An interrupt does not st **The HTTP API is FastAPI plus SSE** — create a session, list, get, post an event, stream the trace, fetch it as NDJSON, healthz. A request may name a registered graph and supply input and a budget; it may not *describe* a graph, because topology comes from a registry the operator fills in Python. But note the seam: **it does not use the session layer above.** It ships its own in-process runtime whose sessions die with the process, never evict, and record `message` and `approval` events without delivering them into a running graph. Two session layers that have not been joined ([ROADMAP.md](../ROADMAP.md) §12.3). -**Policy is a TOML document** over nodes, edges, tools and spend, with tiered evaluation — every `deny` before every `ask` before every `allow`, so a broad deny beats a narrow allow including one scoped to a single tenant. Every decision lands in an audit record naming the rule id, the reason, the policy version and a digest of the document, so a decision can be tied to the exact text that made it. And the seam, now narrowed to the tool plane: **the planner half is wired and the tool half is not.** `PolicyEngine.edge_policy()` and `PolicyEngine.node_policy()` compile the document into the `EdgePolicy` and `NodePolicy` the admission checker consults, and `grapharc plan --policy` is a real caller — so what may run, and what may connect to what, *is* governed by a document you can read. (A `resource = "node"` rule used to be dropped by the compiler and enforced by nothing; [issue #66](https://github.com/CodeGraphContext/GraphARC/issues/66).) But `permission_policy()`, `check_tool()` and `approval_router()` have no caller outside `grapharc/policy/`, so `grapharc agent` still assembles its tool gating from `--allow` / `--deny` / `--ask` globs. The most dangerous surface in the package is the one the document cannot reach yet; [issue #6](https://github.com/CodeGraphContext/GraphARC/issues/6) is that work, and the precedence question it has to settle is what happens when a flag `allow` meets a document `deny`. +**Policy is a TOML document** over nodes, edges, tools and spend, with tiered evaluation: every `deny` before every `ask` before every `allow`. The planner compiles node and edge rules into its admission gate. `grapharc agent --policy` compiles tool rules into the harness, combines them with CLI restrictions using `DENY > ASK > ALLOW`, and routes document approval requests through the policy's named role. An `--allow` flag cannot widen a document denial or bypass its approval requirement. JSON mode and redirected stdin refuse approval requests, and a selected policy document refuses delegated Claude CLI execution. + +Agent policy and tenant settings follow `flag > GRAPHARC_* environment > grapharc.toml > default`, including `--config` and paths anchored to the config file. Governed results report the source, document version and digest, tenant, and policy audit path. Document-driven denials and approval decisions are audited; ordinary allowed tool calls and flag-only denials are not written to the document audit. The run trace still records attempted tool calls and their outcomes. These limits matter when assessing the remaining work in [issue #6](https://github.com/CodeGraphContext/GraphARC/issues/6). ## Independent verification @@ -254,7 +256,7 @@ A stable system is not one that claims to have no edges — it is one whose edge - **`.env` and `grapharc.toml` follow the same discovery rule: the working directory, and nowhere else.** Neither searches parent directories — a run must not be governed by a file you did not know about, and must not be *billed* to one either. **This is a behaviour change:** the credential loader used to walk up to `/`, so a `.env` in an ancestor directory (a `$HOME` one on a shared box, a client project one above a demo checkout) was picked up silently. If you relied on that, move the file into the directory you run from, `export` the variable, or pass `env_file=` to name it explicitly. A real environment variable still beats any file. - **`grapharc run` has no budget unless you give it one.** Set any of `--max-tokens`, `--max-iterations`, `--max-seconds`, or `--max-concurrency`; without them each dimension is unlimited and the gate admits a topology of any worst-case cost. -**Verified this pass:** `pytest` → green, 2,214 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. +**Verified this pass:** `pytest` → green, 2,252 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. [ROADMAP.md](../ROADMAP.md) tracks what is built and what is not, item by item. @@ -265,4 +267,3 @@ Defects that have been **closed** — each with what broke, how it was found and Architecturally *inspired by* systems studied from public documentation: OpenClaw (policy-before-schema tool gating, file-first state, and its security post-mortems), Hermes Agent (budgeted tiered memory, ephemeral subagents), Claude Code (advisory-vs-enforced split, subagent context isolation, verification-centered loops), and OpenRouter (routing semantics, budget-scoped accounting). - diff --git a/grapharc/cli/agent.py b/grapharc/cli/agent.py index ca67b82..b033ae0 100644 --- a/grapharc/cli/agent.py +++ b/grapharc/cli/agent.py @@ -2,7 +2,7 @@ This is the first CLI command that is not a demo: the task comes from the caller, the tools come from `grapharc.tools`, and what the agent was permitted -to do comes from flags rather than from a hard-coded example. The output is +to do comes from a policy document and flags. The output is built so a run answers the three questions the architecture is graded on — what it did (`tool_calls`), what it was allowed to do (`policy`, `tools_visible`), and why it stopped (`termination_reason`, `note`). @@ -21,6 +21,8 @@ from typing import Any from grapharc.cli import optional, style +from grapharc.cli.config import ConfigError +from grapharc.cli.config import load as load_settings from grapharc.cli.output import EXIT_FAILED, EXIT_OK, emit, fail from grapharc.cli.runid import refuse_reused_run_id @@ -275,6 +277,7 @@ def run_agent( run_id: str | None = None, policy_path: Path | None = None, tenant: str | None = None, + config_path: Path | None = None, as_json: bool = False, ) -> int: """Run one agent loop and report it. Returns the process exit code. @@ -295,7 +298,18 @@ def run_agent( `max_tokens=None` means the default ceiling on the governed path — and is the only value the delegated path accepts, because a ceiling it cannot enforce must be refused rather than silently unapplied. + + Policy and tenant resolve from flags, environment, `grapharc.toml`, then + defaults, using the same path anchoring and provenance as the planner. + Other agent options retain their flag/Python-argument defaults. """ + try: + settings = load_settings(config_path) + policy_path = settings.resolve_path("policy", policy_path) + tenant = settings.resolve("tenant", tenant) + except ConfigError as exc: + return fail(str(exc), as_json=as_json, command="agent", task=task) + if policy_path is not None and executor == "claude-cli": # The delegated loop runs inside Claude Code, outside this process's # policy, approval routing and audit — mapping a document onto CLI @@ -429,8 +443,8 @@ def run_agent( on_denial=on_denial, ) visible = [spec.name for spec in harness.visible_tools()] - # Present only when a document governed the run: the no-document payload - # keeps exactly the keys it has always had. + # Present only when a document governed the run, with the same provenance + # used to resolve it. Runs without a document retain their existing payload. governance_extra: dict[str, Any] = {} if governed is not None: from grapharc.policy.document import ResourceKind @@ -438,8 +452,10 @@ def run_agent( _engine, _tenant_name, _audit_path = governed _document = _engine.document governance_extra = { + "policy_source": settings.sources["policy"], + **settings.provenance(), "policy_document": { - "source": "flag", + "source": settings.sources["policy"], "path": str(policy_path), "version": _document.version, "digest": _engine.digest, @@ -564,6 +580,7 @@ def count(number: int) -> str: _, _tenant_name, _audit_path = governed policy_value += ( f" {style.dim('document=')}{policy_path}" + f" {style.dim('source=')}{settings.sources['policy']}" f" {style.dim('tenant=')}{_tenant_name}" f" {style.dim('audit=')}{_audit_path}" ) diff --git a/grapharc/cli/main.py b/grapharc/cli/main.py index 6a8280a..75980ab 100644 --- a/grapharc/cli/main.py +++ b/grapharc/cli/main.py @@ -568,6 +568,7 @@ def _cmd_agent(args: argparse.Namespace) -> int: system_prompt=args.system_prompt, policy_path=args.policy, tenant=args.tenant, + config_path=args.config, run_id=args.run_id, as_json=args.json, ) @@ -1058,7 +1059,8 @@ def build_parser() -> argparse.ArgumentParser: ap.set_defaults(handler=_cmd_approve) agent = sub.add_parser( - "agent", parents=[common], help="run an agent node against a task with the core tools" + "agent", parents=[common, configurable], + help="run an agent node against a task with the core tools" ) agent.add_argument("task", help="what the agent should do") agent.add_argument( diff --git a/tests/test_cli.py b/tests/test_cli.py index 0e8f00a..225fe5e 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3398,3 +3398,94 @@ def test_agent_policy_refuses_a_claude_cli_model( assert code == 2 assert "--policy" in payload["error"] assert "claude-cli" in payload["error"] + + +@pytest.mark.parametrize("source", ["config", "env", "flag"]) +def test_agent_configuration_governs_tools_and_records_its_source( + tmp_path, monkeypatch, capsys, scripted_model, stub_tools, source +): + """Issue #6: configured rules must reach the same boundary as explicit flags.""" + monkeypatch.chdir(tmp_path) + stub_tools() + policy = _write_doc(tmp_path / "deny.toml", DOC_DENY_WRITES) + config = _write_doc( + tmp_path / "grapharc.toml", '[grapharc]\npolicy = "deny.toml"\ntenant = "default"' + ) + args = ["agent", "t", "--model", "mock/x", "--workspace", str(tmp_path / "ws")] + if source in ("env", "flag"): + monkeypatch.setenv("GRAPHARC_POLICY", str(policy)) + monkeypatch.setenv("GRAPHARC_TENANT", "default") + if source == "flag": + args += ["--policy", str(policy), "--tenant", "default"] + model = scripted_model([ + {"tools": [("write_note", {"path": "note.txt", "content": "must not run"})]}, + {"content": "done"}, + ]) + + code, payload, _ = call_json(args, capsys) + + assert code == 0 + assert payload["denied"] == 1 + assert not (tmp_path / "ws" / "note.txt").exists() + assert "write_note" not in (model.bound_tools or []) + assert payload["policy_document"]["source"] == source + assert payload["policy_document"]["path"] == str(policy) + assert payload["sources"] == {"policy": source, "tenant": source} + assert payload["config_file"] == str(config) + audit = _read_jsonl(payload["policy_audit"]) + assert audit[0]["rule_id"] == "no-writes" + + +def test_agent_explicit_config_anchors_policy_and_supplies_tenant( + tmp_path, monkeypatch, capsys, scripted_model, stub_tools +): + project = tmp_path / "project" + project.mkdir() + policy = _write_doc(project / "policy.toml", DOC_TENANTED) + config = _write_doc( + project / "grapharc.toml", '[grapharc]\npolicy = "policy.toml"\ntenant = "acme"' + ) + monkeypatch.chdir(tmp_path) + stub_tools() + scripted_model([{"content": "done"}]) + + code, payload, _ = call_json([ + "agent", "t", "--model", "mock/x", "--workspace", str(tmp_path / "ws"), + "--config", str(config), + ], capsys) + + assert code == 0 + assert payload["policy_document"]["path"] == str(policy) + assert payload["policy_document"]["tenant"] == "acme" + assert payload["sources"] == {"policy": "config", "tenant": "config"} + + +def test_agent_configured_policy_cannot_be_bypassed_by_delegation( + tmp_path, monkeypatch, capsys +): + monkeypatch.chdir(tmp_path) + _write_doc(tmp_path / "grapharc.toml", '[grapharc]\npolicy = "deny.toml"') + + def delegate(*args, **kwargs): + pytest.fail("delegation bypassed the configured policy") + + monkeypatch.setattr("grapharc.cli.delegate.run_delegated", delegate) + code, payload, _ = call_json(["agent", "t", "--executor", "claude-cli"], capsys) + + assert code == 2 + assert "--policy cannot be enforced" in payload["error"] + + +def test_agent_bad_config_is_refused_before_model_construction(tmp_path, monkeypatch, capsys): + monkeypatch.chdir(tmp_path) + _write_doc(tmp_path / "grapharc.toml", '[grapharc]\npollicy = "deny.toml"') + + def build_model(*args, **kwargs): + pytest.fail("a model was constructed under an invalid governance config") + + monkeypatch.setattr("grapharc.gateway.get_model", build_model) + code, payload, _ = call_json(["agent", "t", "--workspace", str(tmp_path / "ws")], capsys) + + assert code == 2 + assert "unknown key(s) pollicy" in payload["error"] + assert not (tmp_path / "ws").exists() From a76199cf1b4f91677979c93e1ab6bf1a11c264b0 Mon Sep 17 00:00:00 2001 From: Shashank Shekhar Singh <123410790+Shashankss1205@users.noreply.github.com> Date: Fri, 9 Oct 2026 22:02:49 +0530 Subject: [PATCH 3/3] fix(slack): require finite timing settings and approval waits Reject NaN, infinity and overflowing environment durations before bot startup, and reject NaN approval waits at the command gate. Positive fractional values and existing command budgets remain configurable. Update timing documentation and the selected-test figure. Fixes #135. --- docs/cookbook/07-slack.md | 6 +++ docs/deep-dive.md | 2 +- grapharc/slack/command.py | 5 ++- grapharc/slack/config.py | 13 ++++--- tests/test_slack_finite_timings.py | 59 ++++++++++++++++++++++++++++++ 5 files changed, 76 insertions(+), 9 deletions(-) create mode 100644 tests/test_slack_finite_timings.py diff --git a/docs/cookbook/07-slack.md b/docs/cookbook/07-slack.md index 18a5691..0703b6e 100644 --- a/docs/cookbook/07-slack.md +++ b/docs/cookbook/07-slack.md @@ -116,6 +116,12 @@ Configuration is environment-only, read once at startup: | `GRAPHARC_SLACK_LIVE_INTERVAL` | `2.5` | seconds between two edits of the status message | | `GRAPHARC_SLACK_LIVE_URL` | unset | base URL of a `grapharc serve --live-root` the requester can reach; posts a "watch live" link | +The timeout and live-interval environment values must be finite and positive; +fractions of a second are accepted. `NaN`, infinity and overflowing values +such as `1e309` are startup errors naming the variable. A requester-supplied +`--approval-timeout` must also be finite and positive and fit within the +command's existing timeout ceiling. + The bot reads tokens from the process environment only. The model gateway's `.env` loader is deliberately not used here — even though it now reads the working directory alone rather than searching upward: a bot that a whole diff --git a/docs/deep-dive.md b/docs/deep-dive.md index 1338f90..21d1092 100644 --- a/docs/deep-dive.md +++ b/docs/deep-dive.md @@ -254,7 +254,7 @@ A stable system is not one that claims to have no edges — it is one whose edge - **`.env` and `grapharc.toml` follow the same discovery rule: the working directory, and nowhere else.** Neither searches parent directories — a run must not be governed by a file you did not know about, and must not be *billed* to one either. **This is a behaviour change:** the credential loader used to walk up to `/`, so a `.env` in an ancestor directory (a `$HOME` one on a shared box, a client project one above a demo checkout) was picked up silently. If you relied on that, move the file into the directory you run from, `export` the variable, or pass `env_file=` to name it explicitly. A real environment variable still beats any file. - **`grapharc run` has no budget unless you give it one.** Set any of `--max-tokens`, `--max-iterations`, `--max-seconds`, or `--max-concurrency`; without them each dimension is unlimited and the gate admits a topology of any worst-case cost. -**Verified this pass:** `pytest` → green, 2,214 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. +**Verified this pass:** `pytest` → green, 2,229 selected and 13 deselected (the live ones); `ruff check .` clean; all eight `grapharc demo` stages green, plus the `trace` / `metrics` / `viz` / `replay` tour against a freshly recorded demo trace; the wheel builds and imports all submodules in a clean virtualenv with `[all]`, and `0.1.8` on PyPI is that wheel. The counts are a snapshot, not a property of the project — `pytest` re-derives them in one command, which is the only reason they are quoted, and `tests/test_deep_dive.py` fails this line rather than letting it drift. [ROADMAP.md](../ROADMAP.md) tracks what is built and what is not, item by item. diff --git a/grapharc/slack/command.py b/grapharc/slack/command.py index 5d47ee3..b58ee88 100644 --- a/grapharc/slack/command.py +++ b/grapharc/slack/command.py @@ -40,6 +40,7 @@ from __future__ import annotations import importlib +import math import shlex import tomllib import uuid @@ -530,10 +531,10 @@ def parse_command( raise SlackCommandError( f"`--approval-timeout` wants a number of seconds, got {supplied!r}" ) from None - if asked <= 0 or asked > ceiling: + if not math.isfinite(asked) or asked <= 0 or asked > ceiling: raise SlackCommandError( f"`--approval-timeout {supplied}` does not fit this command's " - f"budget: the wait must be between 1 and {ceiling:.0f} seconds, " + f"budget: the wait must be finite, positive and at most {ceiling:.0f} seconds, " "so that a run nobody answers ends by reporting a timeout " "rather than by being killed mid-wait" ) diff --git a/grapharc/slack/config.py b/grapharc/slack/config.py index 91ba9e2..0c9d60b 100644 --- a/grapharc/slack/config.py +++ b/grapharc/slack/config.py @@ -11,6 +11,7 @@ from __future__ import annotations +import math import os from dataclasses import dataclass, field from pathlib import Path @@ -86,8 +87,8 @@ def from_env(cls, environ: dict[str, str] | None = None) -> SlackBotConfig: raise SlackConfigError( f"GRAPHARC_SLACK_TIMEOUT must be a number of seconds, got {raw_timeout!r}" ) from None - if timeout <= 0: - raise SlackConfigError("GRAPHARC_SLACK_TIMEOUT must be positive") + if not math.isfinite(timeout) or timeout <= 0: + raise SlackConfigError("GRAPHARC_SLACK_TIMEOUT must be positive and finite") raw_work_timeout = env.get("GRAPHARC_SLACK_WORK_TIMEOUT", "1800") try: @@ -97,8 +98,8 @@ def from_env(cls, environ: dict[str, str] | None = None) -> SlackBotConfig: "GRAPHARC_SLACK_WORK_TIMEOUT must be a number of seconds, " f"got {raw_work_timeout!r}" ) from None - if work_timeout <= 0: - raise SlackConfigError("GRAPHARC_SLACK_WORK_TIMEOUT must be positive") + if not math.isfinite(work_timeout) or work_timeout <= 0: + raise SlackConfigError("GRAPHARC_SLACK_WORK_TIMEOUT must be positive and finite") # A work budget under the reader budget is almost certainly a typo, and # the failure it produces is confusing: `plan --go` would be killed # sooner than `metrics`. Take the larger rather than obeying literally. @@ -112,8 +113,8 @@ def from_env(cls, environ: dict[str, str] | None = None) -> SlackBotConfig: "GRAPHARC_SLACK_LIVE_INTERVAL must be a number of seconds, " f"got {raw_interval!r}" ) from None - if live_interval <= 0: - raise SlackConfigError("GRAPHARC_SLACK_LIVE_INTERVAL must be positive") + if not math.isfinite(live_interval) or live_interval <= 0: + raise SlackConfigError("GRAPHARC_SLACK_LIVE_INTERVAL must be positive and finite") live_url_base = env.get("GRAPHARC_SLACK_LIVE_URL", "").rstrip("/") or None if live_url_base is not None and not live_url_base.startswith(("http://", "https://")): diff --git a/tests/test_slack_finite_timings.py b/tests/test_slack_finite_timings.py new file mode 100644 index 0000000..f4aa856 --- /dev/null +++ b/tests/test_slack_finite_timings.py @@ -0,0 +1,59 @@ +"""Non-finite time settings must not reach Slack's timers or command runner.""" + +from __future__ import annotations + +import pytest + +from grapharc.slack.command import SlackCommandError, parse_command +from grapharc.slack.config import SlackBotConfig, SlackConfigError + + +@pytest.mark.parametrize("value", ["nan", "inf", "-inf", "1e309"]) +@pytest.mark.parametrize( + "key", ["GRAPHARC_SLACK_TIMEOUT", "GRAPHARC_SLACK_WORK_TIMEOUT", "GRAPHARC_SLACK_LIVE_INTERVAL"] +) +def test_environment_timing_values_must_be_finite(tmp_path, key, value): + env = { + "SLACK_BOT_TOKEN": "test-bot", + "SLACK_APP_TOKEN": "test-app", + "GRAPHARC_SLACK_WORKDIR": str(tmp_path), + key: value, + } + + with pytest.raises(SlackConfigError, match=key): + SlackBotConfig.from_env(env) + + +@pytest.mark.parametrize("form", ["--approval-timeout nan", "--approval-timeout=NaN"]) +def test_a_nan_approval_wait_is_refused(tmp_path, form): + with pytest.raises(SlackCommandError, match="does not fit this command's budget"): + parse_command( + f"plan goal --scripted --go {form}", + workdir=tmp_path, + timeout_seconds=60, + work_timeout_seconds=180, + ) + + +def test_finite_fractional_timings_remain_configurable(tmp_path): + config = SlackBotConfig.from_env( + { + "SLACK_BOT_TOKEN": "test-bot", + "SLACK_APP_TOKEN": "test-app", + "GRAPHARC_SLACK_WORKDIR": str(tmp_path), + "GRAPHARC_SLACK_TIMEOUT": "12.5", + "GRAPHARC_SLACK_WORK_TIMEOUT": "25.5", + "GRAPHARC_SLACK_LIVE_INTERVAL": "0.25", + } + ) + + assert config.timeout_seconds == 12.5 + assert config.work_timeout_seconds == 25.5 + assert config.live_interval_seconds == 0.25 + argv = parse_command( + "plan goal --scripted --approval-timeout 1.5", + workdir=tmp_path, + timeout_seconds=config.timeout_seconds, + work_timeout_seconds=config.work_timeout_seconds, + ) + assert argv[argv.index("--approval-timeout") + 1] == "1.5"