Pipeline Architecture

Edictum is a single pipeline that every adapter calls. Contracts written in YAML compile to the same runtime objects as Python-defined contracts. The pipeline is deterministic -- same input, same decision, every time.

Pipeline Overview

Every tool call passes through one pipeline, regardless of which framework adapter triggers it.

In enforce mode, if any precondition fails, the tool call is denied. The tool never executes. There is no "soft deny" -- either every check passes or the call does not happen. (In observe mode, failures are logged as CALL_WOULD_DENY but the call proceeds.)

Pre-Execution Detail

GovernancePipeline.pre_execute() runs seven checks in order. The first failure in steps 1-5.5 short-circuits -- remaining checks are skipped. Step 6 (observed evaluation) always runs when the decision is "allow."

If the PreDecision.action is "allow", the adapter lets the tool execute.

Post-Execution Detail

Once a tool has executed, Edictum checks its output. Postcondition behavior depends on the declared effect and the tool's side-effect classification.

For warn, the pipeline warns the agent and lets it decide how to proceed. For redact and deny on READ/PURE tools, the pipeline modifies the response before it reaches the agent. WRITE/IRREVERSIBLE tools always get warn because the action already happened -- hiding the result only removes context the agent needs.

YAML Compilation

YAML contract bundles go through a multi-stage pipeline that produces the same runtime objects as hand-written Python contracts. When multiple files are passed to from_yaml(), a composition step merges them before compilation.

There is no separate "YAML execution path." A precondition compiled from YAML and a precondition written as a Python function are indistinguishable to the pipeline. They produce the same Verdict objects, appear in the same contracts_evaluated audit records, and are subject to the same observe-mode behavior.

Adapter Pattern

Adapters are thin translation layers between framework-specific hook APIs and the pipeline. Each adapter:

1. Intercepts the framework's tool-call lifecycle event
2. Builds a ToolEnvelope via create_envelope()
3. Calls pipeline.pre_execute() and translates the PreDecision into the framework's expected format
4. If allowed, lets the tool execute
5. Calls pipeline.post_execute() and forwards any findings

Adapters never contain enforcement logic. They translate formats. If you need to add a new contract, add it as a contract or hook -- not adapter code.

Design Decisions

Envelope Immutability

ToolEnvelope is a frozen dataclass. Once created, no field can be modified.

This is enforced at two levels: @dataclass(frozen=True) raises FrozenInstanceError on assignment, and create_envelope() deep-copies args and metadata via json.loads(json.dumps(...)) so the caller cannot mutate the original dicts. create_envelope() also validates tool_name -- rejecting empty strings, null bytes, newlines, and path separators that could corrupt session keys or audit records.

Always create envelopes through create_envelope(), never by constructing ToolEnvelope(...) directly. The Principal dataclass is also frozen.

Session and Storage Model

Sessions track execution state across multiple tool calls within an agent run.

All counter operations go through the StorageBackend protocol:

class StorageBackend(Protocol):
    async def get(self, key: str) -> str | None: ...
    async def set(self, key: str, value: str) -> None: ...
    async def delete(self, key: str) -> None: ...
    async def increment(self, key: str, amount: float = 1) -> float: ...

increment() must be atomic. This is the fundamental requirement for correctness under concurrent access.

MemoryBackend enforces atomicity via asyncio.Lock on increment() and delete(). Single dict reads (get(), set()) are not locked -- they do not have read-modify-write patterns.

MemoryBackend stores counters in a Python dict -- one process, one agent. This covers the vast majority of use cases: a single agent process enforcing session limits on its own tool calls. For multi-agent coordination across processes, edictum-console handles centralized session tracking. See the roadmap.

Operation Limits

OperationLimits defines three cap types:

max_attempts fires first because it counts denied calls too. An agent stuck in a denial loop hits the attempt cap without ever incrementing the execution counter. The denial message tells the agent to stop and reassess rather than keep retrying.

Claude Agent SDK: Intentional Decoupling

to_hook_callables() returns callables using Edictum's own calling convention (snake_case keys, (tool_name, tool_input, tool_use_id) signature) rather than the SDK-native HookCallback signature ((input_data, tool_use_id, context) with PascalCase event keys and HookMatcher wrappers).

This is intentional. The claude-agent-sdk package is pre-1.0 and its types (HookMatcher, HookContext, HookCallback) may change. Importing them would add a runtime dependency to Edictum's zero-dep core and couple releases to SDK breaking changes. The ~10-line bridge recipe in the Claude SDK adapter docs lives in user-land where the claude-agent-sdk coupling already exists.

If the SDK stabilizes at 1.0, a to_native_hooks() convenience method that returns dict[HookEvent, list[HookMatcher]] directly could be added without breaking to_hook_callables().

Error Handling: Fail-Closed

Edictum follows a fail-closed default with explicit opt-in to permissive behavior:

• Unregistered tools default to SideEffect.IRREVERSIBLE (most restrictive classification)
• Contract evaluation errors deny the tool call rather than silently allowing it
• Observe mode is opt-in per-contract or per-pipeline, never the default
• Postconditions default to warn; redact and deny effects are enforced for READ/PURE tools but fall back to warn for WRITE/IRREVERSIBLE tools
• Backend errors propagate as exceptions. ServerBackend.get() returns None only for HTTP 404 (key not found); all other errors (connection refused, timeout, 500) propagate to the pipeline which converts them to deny decisions
• Input validation in create_envelope() rejects tool names with null bytes, newlines, or path separators before any pipeline processing
• Concurrent access is safe: MemoryBackend.increment() and delete() use asyncio.Lock to prevent lost updates under asyncio.gather

Audit events record policy_error: true when contract loading fails, ensuring that broken contract bundles are visible in monitoring even when the system falls back to a safe default.

Where It's Heading

Edictum is currently an in-process library -- contracts are loaded and enforced within the same process as the agent. This covers single-agent deployments and most production use cases today.

The server SDK (pip install edictum[server]) is shipped and provides the client-side connectivity for agents to talk to edictum-console. It implements the core protocols (ApprovalBackend, AuditSink, StorageBackend) over HTTP, letting agents use server-managed approvals, centralized audit ingestion, distributed session state, and SSE-pushed contract updates. Edictum.from_server() wires all five server components from a single URL and API key. Edictum.reload() atomically swaps contracts from a new YAML bundle (fail-closed on errors). The SSE watcher passes env, bundle_name, and policy_version as query params so the server can filter events per-bundle and detect drift.

Edictum Console (v0.1.0) is the self-hostable operations server. docker compose up → dashboard at localhost:8000. It provides the coordination infrastructure that cannot run in-process: contract management with Ed25519 signing, HITL approval workflows with 6 notification channels, audit event feeds, fleet monitoring with drift detection, and hot-reload via SSE. See the console docs for setup.

The Boundary Principle

The split between core and server follows one principle: evaluation = core library, coordination = server.

The pipeline that takes a tool call and returns allow/deny/warn runs entirely in-process with zero external dependencies. Anything that requires coordination across processes, networking to external systems, or centralized management is server scope.

src/edictum/
  __init__.py              Edictum facade (registers contracts, hooks, sinks)

  envelope.py              ToolEnvelope, Principal, ToolRegistry, BashClassifier
  contracts.py             @precondition, @postcondition, @session_contract, Verdict
  pipeline.py              GovernancePipeline -- PreDecision, PostDecision
  evaluation.py            ContractResult, EvaluationResult (dry-run evaluation API)
  findings.py              Finding, PostCallResult, classify_finding (postcondition output)
  hooks.py                 HookResult, HookDecision (allow/deny)
  session.py               Session (atomic counters via StorageBackend)
  storage.py               StorageBackend protocol, MemoryBackend
  limits.py                OperationLimits (max_attempts, max_tool_calls, per-tool)
  audit.py                 AuditEvent, AuditAction, AuditSink, RedactionPolicy, CollectingAuditSink
  approval.py              ApprovalBackend protocol, ApprovalRequest/Decision, LocalApprovalBackend
  telemetry.py             GovernanceTelemetry (OTel spans + metrics, no-op fallback)
  builtins.py              deny_sensitive_reads() built-in precondition
  types.py                 HookRegistration, ToolConfig (internal shared types)

  yaml_engine/
    loader.py              Parse YAML, validate against JSON Schema, SHA-256 hash
    evaluator.py           Condition evaluation (match, principal checks, etc.)
    compiler.py            YAML contracts -> @precondition/@postcondition/@sandbox objects
    composer.py            Bundle composition (compose_bundles, observe_alongside)

  otel.py                  configure_otel(), has_otel(), get_tracer() (OTel spans)

  cli/
    main.py                Click CLI entry point (validate, check, diff, replay, test)

  adapters/
    langchain.py           LangChain tool-calling middleware
    crewai.py              CrewAI before/after hooks
    agno.py                Agno async hook wrapper
    semantic_kernel.py     Semantic Kernel filter pattern
    openai_agents.py       OpenAI Agents guardrails
    claude_agent_sdk.py    Anthropic Claude Agent SDK hooks
    nanobot.py             Nanobot governed ToolRegistry
    google_adk.py          Google ADK plugin and agent callback adapter

  server/                  pip install edictum[server]
    client.py              EdictumServerClient (async HTTP, auth, retries, env, bundle_name)
    approval_backend.py    ServerApprovalBackend (ApprovalBackend via HTTP)
    audit_sink.py          ServerAuditSink (batched event ingestion, bundle_name in payload)
    backend.py             ServerBackend (StorageBackend via HTTP)
    contract_source.py     ServerContractSource (SSE with env/bundle_name/policy_version params)