​

Policy Binding — pull, enforcement, and per-run refresh for every agent surface

Addendum — post-review simplifications (2026-06-04). Phase 1 landed a deliberately smaller PolicyBinding than §3/§4 describe. Where this addendum and the body disagree, the addendum wins:

1. No auto-register in the binding. AutoRegisterSpec, manifest_payload, and the 404→register→retry flow are gone (SRP: registration is not policy resolution — and hexgate.cli.register already builds better manifests from the real agent object). A 404 propagates as HexgateError with .status == 404; callers that want register-on-miss catch it, call register_agent(agent), and resolve again. Adapter phases add this 4-line pattern at the wrap sites.
2. No fallback param. The plain constructor is the explicit static path: PolicyBinding(PolicyEnforcer(engine, agent_name=...)).
3. No prefetched param. The loader composes directly: build the pre-seeded PlatformPolicySource, decode the bundle (as it already does), and call the constructor. Same single round trip, no parameter.
4. No client / agent_name fields. The binding holds exactly enforcer + source; agent_name is read from the enforcer for logs, and the HexgateClient stays a caller concern (agent.hexgate_client, as today).
5. No _warn_uncovered_tools. YAGNI — revisit in the adapter phases if real usage shows the need.
6. The HEXGATE_LOCAL_POLICY helpers live in security/source.py (env var → PolicySource is source-construction), not in binding.py. The loader re-imports them from there.
7. The refresh lock lives in PlatformPolicySource, which owns the mutable cache it protects — the binding has no lock.

Net: binding.py is ~200 lines (resolve → local override → platform → raise; fail-soft refresh/refresh_async), and the client kept only HexgateError.status from its planned changes.

​

1. Problem statement

def build_policy_set(api_key, agent_name, tool_names) -> PolicySet:
    """Placeholder allow-all one-role bundle. TODO: cloud-fetch via HexgateClient."""

​

1.1 The structural fact this design exploits

enforcer.policy = new_bundle   # hot-swaps enforcement for every wrapped tool

​

2. Goals / non-goals

​

Goals

1. One implementation of policy resolution (precedence, verification, fallback) and refresh (ETag/304, fail-soft) shared by all six surfaces.
2. create_agent can bind to the platform at creation; refresh already fires per stream_agent / invoke_agent — no streaming changes.
3. All four adapters pull a real policy at wrap time and refresh at every run entry point; the allow-all placeholder is deleted.
4. Verification semantics are byte-identical to today’s loader: a tampered or unverifiable bundle is never silently downgraded.
5. load_hexgate_agent dedupes onto the same primitive (net code deletion).

​

Non-goals

• No new platform endpoints. GET /v1/agents/{name} (ETag/304, platform/api/main.py:857) and POST /v1/agents (auto-register with default role-aware policy + signed bundle, platform/api/main.py:829) already exist and suffice.
• No per-tool-call refresh. Refresh granularity stays per run/turn (matching stream_agent today). Mid-turn policy edits land next turn.
• No change to the enforcement decision pipeline (PolicyEnforcer.decide → engine evaluate → Decision), nor to the WASM evaluation path (hexgate/security/wasm_engine.py).
• No change to approval-handler semantics per adapter (LangChain resolves inline; OpenAI/Google/pydantic_ai render markered errors — unchanged).

​

3. Core design: PolicyBinding

@dataclass
class PolicyBinding:
    """Resolved policy for one agent: an enforcer plus an optional refresh source.

    The enforcer is the stable object every wrapped tool closes over;
    refresh() mutates `enforcer.policy` in place, so a binding created once
    keeps all previously wrapped tools current forever.
    """

    enforcer: PolicyEnforcer
    source: PolicySource | None          # None → static policy, refresh is a no-op
    client: HexgateClient | None         # kept for lazy User-scope attenuation
    agent_name: str
    _refresh_lock: threading.Lock        # serializes concurrent refreshes

​

3.1 PolicyBinding.resolve(...) — the pull

@classmethod
def resolve(
    cls,
    agent_name: str,
    *,
    api_key: str | None = None,          # explicit → HEXGATE_KEY env
    client: HexgateClient | None = None, # reuse an existing client (loader path)
    prefetched: tuple[dict, str | None] | None = None,  # (payload, etag) reuse
    fallback: PolicyEngine | None = None,
    auto_register: AutoRegisterSpec | None = None,
    approval_context: str | None = None,  # display only
) -> "PolicyBinding":

1. HEXGATE_LOCAL_POLICY override (dev loop) — wins outright. BundleDirPolicySource (mtime-refreshed pre-built bundle) or YamlPolicySource (auto-recompile on save), exactly as today (hexgate/security/source.py:165,256). The platform is not contacted.
2. Platform — when an api_key/HEXGATE_KEY/client is available:
   1. Build or reuse the HexgateClient (Biscuit signature verified lazily on first use, cloud/client.py:177).
   2. payload, etag = client.get_agent(agent_name) — or use prefetched when the caller already fetched (loader path; avoids a double fetch).
   3. decode_and_verify_platform_bundle(payload, client.public_key_bytes()) (source.py:117): Ed25519 signature over the exact manifest bytes and sha256(wasm) == manifest.wasm_hash. Any failure raises — identical to today.
   4. Bundle present → it is the policy. Bundle absent (platform couldn’t compile, e.g. no opa) → fall back to load_policy_set_from_dict(payload["policy_yaml"]) (pydantic engine), unless HEXGATE_BUNDLE_REQUIRE_SIGNATURE is set → raise (today’s rule, loader.py:615-621).
   5. Attach PlatformPolicySource(client, agent_name, initial_bundle=..., initial_etag=...) pre-seeded so the first refresh is a 304 (source.py:78-93).
   6. 404 — agent unknown to the platform:
      • if auto_register is provided → POST /v1/agents with a minimal manifest (name + tool names), which the platform answers with a default role-aware policy and a signed bundle (platform/api/main.py:829, services.py:1290); then re-fetch and proceed. Registration failure → raise.
      • else → raise PolicyBindingError with a message pointing at hexgate agents register / auto_register=.
3. fallback engine — used only when neither a local override nor any API key is in play. None (the default) → raise. This is the explicit opt-out that replaces the adapters’ silent allow-all; callers who truly want ungoverned behavior must write it down: fallback=PolicySet.allow_all(tool_names).

​

AutoRegisterSpec

@dataclass(frozen=True)
class AutoRegisterSpec:
    tool_names: list[str]
    model: str | None = None
    description: str | None = None

​

3.2 refresh() / refresh_async() — the per-run pull

def refresh(self) -> None:
    """Pull the current policy and swap it in. Cheap when nothing changed.

    No-op when source is None. Failures are logged and swallowed —
    a transient network blip never crashes a turn; the previous verified
    policy stays in force (fail-open to STALENESS, never to NO-POLICY).
    """
    if self.source is None:
        return
    with self._refresh_lock:               # collapse concurrent refreshes
        try:
            new_policy = self.source.fetch()
        except Exception as exc:
            logger.warning("policy refresh failed: %s — keeping previous policy", exc)
            return
        if new_policy is None or new_policy is self.enforcer.policy:
            return                          # 304 path: same object identity
        self.enforcer.policy = new_policy   # atomic rebind — all tools see it

async def refresh_async(self) -> None:
    await asyncio.to_thread(self.refresh)

• fail-soft moves into the binding (one place instead of per-caller), and
• a threading.Lock serializes concurrent refreshes. Adapters’ proxies are documented multi-user objects; without the lock, two concurrent runs both miss the ETag cache and double-fetch. enforcer.policy = itself is an atomic rebind and WasmPolicy already serializes evaluation (wasm_engine.py:130-133), so the lock is purely an efficiency/etag- coherence measure, not a correctness one.

​

3.3 Refresh granularity and timing contract

• Refresh fires once at the top of every run (turn), before any model or tool execution for that run. All surfaces below uphold this.
• Tool calls within a single run see one consistent policy: the swap only happens at the run boundary; mid-run a concurrent runner’s swap can land (shared enforcer), which is acceptable — both policies involved are platform-verified, and per-tool-call atomicity is guaranteed by decide() reading self.policy once.
• Sync entry points call refresh() directly (blocking HTTP, ~ms, on the caller’s thread — same as any sync tool I/O). Async entry points call await refresh_async().

​

4. Surface integration

​

4.1 HexgateAgent — create_agent (hexgate/agents/factory.py)

def create_agent(
    model, tools, system_prompt=DEFAULT_SYSTEM_PROMPT,
    *,
    name: str | None = None,             # exists today
    bind_policy: bool | None = None,     # NEW — None = auto
    approval_handler: ApprovalHandler | None = None,   # NEW (threaded to enforce)
    ...
) -> tuple[AgentGraph, CallbackHandler]:

binding = PolicyBinding.resolve(name, auto_register=AutoRegisterSpec(tool_names))
enforced = agent.enforce_policy(binding.enforcer.policy, approval_handler=approval_handler)
enforced._binding = binding
enforced._policy_source = binding.source     # back-compat alias, see 4.2
enforced.hexgate_client = binding.client

​

4.2 Loader dedupe (hexgate/agents/loader.py)

binding = PolicyBinding.resolve(
    resolved_name,
    client=client,
    prefetched=(payload, initial_etag),   # no second fetch
)
enforced = enforce_policy(agent, binding.enforcer.policy, approval_handler=...)
enforced._binding = binding
enforced.hexgate_client = binding.client

​

4.3 LangChain BYO-graph (adapters/langchain/wrapper.py, agent.py)

binding = PolicyBinding.resolve(
    agent_name,
    api_key=resolved_key,
    auto_register=AutoRegisterSpec(tool_names),
)
install_enforcer_on_tools(tools, enforcer=binding.enforcer)   # unchanged mechanics
return HexgateLangchainAgent(agent=agent, binding=binding, tool_names=tool_names)

​

4.4 OpenAI Agents (adapters/openai/wrapper.py, runner.py)

• HexgateRunner gains a binding cache: self._bindings: dict[str, PolicyBinding].
• New private helper:

def _binding_for(self, agent: Agent) -> PolicyBinding:
    name = getattr(agent, "name", "default")
    binding = self._bindings.get(name)
    if binding is None:
        binding = PolicyBinding.resolve(
            name, api_key=self.api_key,
            auto_register=AutoRegisterSpec([t.name for t in agent.tools]),
        )
        self._bindings[name] = binding
    return binding

• wrap_openai_agent changes signature to accept the enforcer (or binding) instead of building its own placeholder: wrap_openai_agent(agent, enforcer=binding.enforcer) → tool copies via wrap_tools (adapters/openai/tools.py:53), mechanics unchanged. Re- wrapping per call stays (it’s cheap copy.copy of FunctionTools and the agent arrives per call anyway); the enforcer is the cached, shared object, so refresh reaches every copy.
• Run methods:
  • run (async): binding = self._binding_for(agent) → await binding.refresh_async() → wrap → run inside the User scope.
  • run_sync: same with binding.refresh().
  • run_streamed: refresh synchronously in the setup body, before Runner.run_streamed is called. Tools execute later, during stream_events iteration, but they hold the enforcer reference fixed at wrap time — and the policy they consult is whatever the enforcer holds when the tool actually fires. Refreshing at setup satisfies the “refresh before the run’s first tool call” contract. (A second refresh inside _stream_events_with_scope is not added: one refresh per run.)

​

4.5 Google ADK (adapters/google/wrapper.py, runner.py)

• __init__: self._binding = PolicyBinding.resolve(agent_name, api_key=self.api_key, auto_register=AutoRegisterSpec(tool_names)); wrap_google_agent(agent, enforcer=self._binding.enforcer); ADK Runner built once, as today. Construction becomes the loud-failure point (network, signature, 404-without-register all raise here).
• run (sync generator): self._binding.refresh() first line, before user.sync_scope().
• run_async (async generator): await self._binding.refresh_async() first line.

​

4.6 pydantic_ai (adapters/pydantic_ai/wrapper.py, agent.py)

• wrap_pydantic_agent resolves the binding (auto-register from the extracted tool names), wraps the cloned toolset with binding.enforcer, deletes its build_policy_set, and passes the binding into HexgatePydanticAgent.
• HexgatePydanticAgent run methods: await self._binding.refresh_async() (async) / self._binding.refresh() (sync) as their first line.

​

5. Failure semantics — one matrix, all surfaces

​

6. Security invariants (must hold after the refactor)

1. Single trust root. The platform’s Ed25519 root key signs both Biscuits (HEXGATE_KEY) and bundle manifests; the SDK verifies both against the same key (explicit → HEXGATE_PUBLIC_KEY → JWKS TOFU).
2. No silent downgrade. A served-but-unverifiable bundle is fatal at resolve and inert at refresh. The pydantic fallback is only reachable when the platform affirmatively served no bundle, and HEXGATE_BUNDLE_REQUIRE_SIGNATURE closes even that.
3. No silent allow-all. Removing build_policy_set removes the last construction path that runs ungoverned with a HEXGATE_KEY present. Ungoverned operation requires an explicit fallback= or bind_policy=False in the caller’s code.
4. Verification happens before caching. PlatformPolicySource only caches bundles that passed signature + integrity; the 304 path can only return previously verified objects.
5. Role resolution stays call-time. PolicyEnforcer.decide re-reads the User contextvar per tool call (enforcer.py:42-44); binding/refresh is user-agnostic. One binding safely serves many concurrent users.
6. Refresh can deny freshness, never grant access. The worst a compromised refresh channel can do is keep an old (verified) policy in force.

​

7. Platform contract (existing, consumed as-is)

​

8. Behavior changes, migration, and known gaps

​

8.1 Breaking-ish changes (changelog required)

1. Adapters: allow-all → real policy. Wrapped OpenAI/Google/pydantic_ai/ LangChain agents go from “everything allowed” to “whatever the platform says”, with deny-by-absence for unlisted tools. Mitigated by auto-register’s generated default policy covering the agent’s actual tool names; surfaced by the §5 uncovered-tools warning.
2. Adapters: missing/unreachable platform now raises at wrap/construct instead of silently running open. Escape hatch: fallback=.
3. create_agent with HEXGATE_KEY set + a name now binds by default (auto mode). Programmatic callers who want a bare graph in a keyed environment pass bind_policy=False.

​

8.2 Non-changes

• stream_agent / invoke_agent / hexgate serve behavior is identical.
• Loaders’ public signatures are identical.
• Enforcement outcomes (ALLOW / NEEDS_APPROVAL / DENY rendering per adapter) are identical.

​

8.3 Known gaps carried forward (explicitly out of scope, tracked)

1. Unbounded staleness on refresh failure. No max-staleness TTL; a platform outage keeps the last verified policy in force indefinitely (one warning per run). A future HEXGATE_POLICY_MAX_STALENESS knob can harden revocation scenarios.
2. _WasmPolicyCache is unwired. WasmPolicy.from_bytes_cached (wasm_engine.py:169) has no production call sites; PolicyBundle.policy() calls from_bytes directly (bundle.py:275). A 200 whose wasm bytes are unchanged (e.g. a non-policy field edit) pays a fresh ~50–100 ms wasmtime instantiation. Fix (small, recommended rider): policy() → WasmPolicy.from_bytes_cached(self.wasm_bytes, self.wasm_hash) when wasm_hash is present.
3. Non-BaseTool specs pass enforce_policy unguarded (factory.py:415-426) — pre-existing; unchanged by this spec but worth a warning log in a follow-up.

​

9. Test plan

​

9.1 Core (tests/security/test_policy_binding.py)

• precedence: local override beats platform beats fallback; no key + no fallback raises.
• resolve: 200 → verified bundle, source pre-seeded (next fetch sends If-None-Match); bundle-less payload → pydantic engine; REQUIRE_SIGNATURE blocks the fallback; bad signature raises; 404 + auto-register → registers then binds; 404 without → raises.
• refresh: 304 → no swap (object identity), one HTTP call; 200 → swap, old decisions change on next decide; fetch exception → warn + keep policy; tampered 200 → warn + keep policy; concurrent refreshes serialize (lock).
• prefetched short-circuits the resolve fetch (call count == 0).

​

9.2 Factory/loader

• create_agent(name=..., bind_policy=True) with mocked client: tools are GuardedTools, _policy_source/_binding present, stream_agent issues the conditional GET.
• auto mode: no key → bare agent unchanged, refresh_policy() no-ops.
• bind_policy=True without name → ValueError.
• load_hexgate_agent regression suite passes unmodified (dedupe is behavior-preserving); single round-trip asserted via mock call count.
• with_tools rebuild keeps _binding reachable.

​

9.3 Per adapter (×4, same skeleton)

• wrap pulls platform policy: a platform deny rule blocks the tool (assert the framework-appropriate error rendering).
• second run sends If-None-Match; 304 keeps enforcer.policy identity.
• platform policy updated between runs → next run enforces the new policy.
• platform down at refresh → run proceeds on previous policy + warning.
• wrap with unreachable platform raises; fallback= allows it.
• OpenAI-specific: binding cached across run calls (resolve once); run_streamed refreshes before setup; per-call rewrap shares the cached enforcer.
• Google-specific: construct-time failure raises; refresh fires per run/run_async without rebuilding the ADK Runner.
• concurrency smoke: two concurrent runs on one proxy/runner, single fetch.

​

9.4 End-to-end (against the platform test app)

​

10. Implementation phases

​

11. Resolved decisions (rationale recorded)