Skip to main content

Governed agents, in-process

The Quickstart governs individual tools with the @govern.tool decorator from pip install watchlight. This page goes a level deeper: governing a real agent's lifecycle — authorize each action, preflight before it runs, attenuate what a sub-agent may do, and record local lineage — all in-process, with the same SDK your production agents use. No database, no network, no services.

# The lifecycle SDK (InProcessClient) + the compiled engine it drives:
pip install watchlight-agent-sdk

# Governing a framework agent? The plugin extra pulls the plugin, the SDK, and
# the govern decorator together — one install, nothing else to wire:
pip install "watchlight[pydantic-ai]"     # or [langgraph], [claude-agent]

watchlight-agent-sdk (imported as watchlight_core) is the open-source glue your agents already call; the compiled watchlight-engine wheel — pulled in automatically — makes the decisions. In the DE you point that glue at the in-process engine with one object: InProcessClient — no apdp_url, no network.

Same code, one line different

In production the SDK talks to the governed control plane over the network. In the Developer Edition you swap in InProcessClient — the agent and plugin code is identical. Graduating to Enterprise is a configuration change, never a rewrite.

Govern a decision​

InProcessClient answers the same calls the networked backend does, but every decision is a real, fail-closed evaluation by the compiled engine:

import asyncio
from watchlight_core import InProcessClient

POLICIES = [
    {"name": "reader",
     "code": 'permit(principal == User::"research-agent", action == Action::"read", resource);'},
]

async def main():
    gov = InProcessClient(POLICIES)  # real engine decisions, in-process — no network

    print((await gov.authorize("research-agent", "read", "dataset"))["decision"])    # Allow
    print((await gov.authorize("research-agent", "delete", "dataset"))["decision"])  # Deny

asyncio.run(main())
Allow
Deny

Preflight a step​

Preflight is a read-only "would this be allowed?" check — no audit weight, ideal for a plan step or a UI affordance. It authorizes under the session's agent:

gov = InProcessClient(POLICIES)
session = await gov.create_session(agent_id="research-agent")
sid = session["session"]["id"]

pf = await gov.preflight_step(sid, "read", "dataset")
print(pf["preflight"]["decision"])   # Allow

pf = await gov.preflight_step(sid, "delete", "dataset")
print(pf["preflight"]["decision"])   # Deny

Govern a framework agent​

Because the SDK reaches its backend through one seam, an existing framework plugin runs governed in-process by pointing it at InProcessClient — the agent code doesn't change:

from watchlight_pydantic_ai import WatchlightPydanticAIPlugin  # or -langgraph, -claude-agent, …
from watchlight_core import InProcessClient

# Pass the in-process backend at construction — the one line that differs from
# production, where you would give `apdp_url=` instead. Same plugin, same agent
# code; going to production is that one config flip, not a rewrite.
plugin = WatchlightPydanticAIPlugin(governance=InProcessClient(POLICIES))

async with await plugin.start_run("research-agent") as handle:
    if not await handle.authorize_action("read", "dataset"):
        raise PermissionError("denied by policy")
    # ... your agent does its work, every action governed ...
Requires the plugin version that ships governance=

watchlight-langgraph >= 0.4, watchlight-pydantic-ai >= 0.3, watchlight-claude-agent >= 0.2. On earlier versions, set the backend after construction instead: plugin.apdp = InProcessClient(POLICIES).

Install the plugin for your framework from the Framework Plugins section; the in-process backend above is the only difference from the networked setup.

Attenuate a sub-agent's scope​

When an agent spawns a sub-agent, the sub-agent must get a strict subset of the parent's authority — never more. The engine runs the real strict-subset validator in-process:

Requires watchlight-engine >= 0.1.1
import json, watchlight_engine

engine = watchlight_engine.PolicyEngine()

parent = {"allowed_tools": ["read", "write"], "allowed_resources": [],
          "allowed_intents": ["research"], "max_depth": 2,
          "time_budget_seconds": 3600, "depth": 0}

# A subset is granted, and the engine returns the CLAMPED scope to hand the child.
child_ok = {"allowed_tools": ["read"], "allowed_resources": [],
            "allowed_intents": ["research"], "max_depth": 1, "time_budget_seconds": 1800}
ok = json.loads(engine.attenuate_scope(json.dumps(parent), json.dumps(child_ok)))
print(ok["decision"], "->", ok["granted_scope"]["allowed_tools"])
# -> Allow -> ['read']

# … a request for authority the parent lacks is denied, and told exactly why.
child_bad = {"allowed_tools": ["read", "delete"], "allowed_resources": [],
             "allowed_intents": ["research"], "max_depth": 1, "time_budget_seconds": 1800}
bad = json.loads(engine.attenuate_scope(json.dumps(parent), json.dumps(child_bad)))
print(bad["decision"], "->", bad["reason"], bad["violations"])
# -> Deny -> 1 tool(s) not in parent.allowed_tools (e.g. ["delete"]) ['AllowedTools']

granted_scope is what you pass to the sub-agent — never the child's request, always the engine's clamped result. On a Deny, violations names which dimension overreached (AllowedTools, AllowedIntents, MaxDepth, TimeBudget, …) so you can surface a precise error.

This is the same M43 attenuation the platform enforces — a sub-agent can only ever narrow authority, never widen it.

Local lineage​

Every governed run writes canonical, value-free lineage events to .watchlight/audit.jsonl — greppable, and useful for local debugging:

await gov.publish_events([{"event_type": "execution_started", "agent": "research-agent"}])
# appended to .watchlight/audit.jsonl

Growing up: Enterprise​

The Developer Edition governs an agent on your machine. Watchlight Beacon (Enterprise) is the same code pointed at the governed control plane, adding what a fleet in production needs:

  • Signed, tamper-evident lineage & audit — instead of local JSONL, every event cryptographically signed (KMS-backed) and court-defensible.
  • Drift & anomaly detection → automatic quarantine — stop a misbehaving agent before its next action.
  • Fleet-wide revocation, multi-tenant administration, and the global execution graph — govern many agents across many environments, centrally.

See Developer Edition vs Enterprise for the full comparison, or talk to us about Enterprise when you're ready for production.