Skip to content

rag-policy.md

Latest commit

 

History

History
212 lines (163 loc) · 7.59 KB

rag-policy.md

File metadata and controls

212 lines (163 loc) · 7.59 KB
Raw

Reference — rag-policy

The rag-policy package houses the PolicyEngine SPI — the single Policy Decision Point (PDP) for AgentContextOS governance. See ADR-0005 for the decision and docs/architecture/policy-engine.md for the design.

This page focuses on the package's public surface as it lands in Step 1.1c. Consumers (gateway, ingest pipeline) will arrive in later phases; the coverage linter described below already guarantees that those consumers cannot land without consulting the PDP.

Overview

Five governance touchpoints in the V1 plan (PII at ingest, quotas, ACL push-down, ACL egress verifier, PII at egress) are consolidated behind one SPI:

class PolicyEngine(HealthCheckMixin, ABC):
    async def evaluate(ctx, decision, subject) -> PolicyResult: ...
    async def filter_pushdown(ctx, decision) -> FilterExpr: ...

Every retrieval / ingest / egress code path consults an engine instance. A coverage linter (tests/policy/coverage.py) fails CI when a known governance-relevant SPI call site is found without an adjacent PolicyEngine / PolicyWriter consultation.

Usage

from rag_policy import (
    NoopPolicyEngine,
    PolicyDecision,
    PolicyResult,
    PolicyWriter,
)

policy = PolicyWriter(NoopPolicyEngine())

# 1. Push-down: PolicyEngine returns a FilterExpr injected into the backend.
filter_expr = await policy.filter_pushdown(ctx, PolicyDecision.read_chunk)
candidates = await backend.retrieve_ids(ctx, query_vec, top_k=200, corpus_ids=[])

# 2. Per-item evaluation (rare — most filtering happens push-down).
allowed = []
for ref in candidates:
    result = await policy.evaluate(ctx, PolicyDecision.read_chunk, ref)
    if result.is_allow():
        allowed.append(ref)
    elif result.is_transform():
        allowed.append(result.transformed)
    # DENY: skipped, logged, audited via the writer's structured-log entry.

Reach for PolicyWriter rather than calling PolicyEngine directly — it emits the policy.decision structured log entry every governance decision should produce.

PolicyDecision values

Decision Subject Typical results
read_chunk Chunk or ChunkRef allow / deny (ACL mismatch)
ingest_doc Document allow / deny (size, MIME) / transform (PII redaction)
egress_text str or Chunk allow / transform (PII redaction)
quota_check QuotaSubject(tenant_id, kind, amount) allow / deny (over-quota)
rate_limit RateLimitSubject(tenant_id, endpoint) allow / deny (rate exceeded)
execute_plan QueryPlan allow / transform (degraded plan under cost cap)

Adding a new decision requires updating NoopPolicyEngine, writing conformance tests, and amending this page.

FilterExpr mini-language

Returned by filter_pushdown; consumed by retrieval backends. Minimal shape at 1.1c — enough for tenant + ACL scoping:

from rag_policy import and_, any_in, eq, FilterExpr

expr: FilterExpr = and_(
    eq("tenant_id", str(ctx.tenant_id)),
    any_in("acl_labels", list(ctx.principal.acl_labels)),
)

Nodes: Eq, AnyIn, And, Or, Not, TrueExpr. Each is a frozen Pydantic model; the union is discriminated on the kind tag so backends switch over the shape rather than introspecting fields.

Step 2.1 relocation: the AST and the evaluate() reference evaluator now live in rag_core.filter so the read-layer SPIs (in rag-core) can accept FilterExpr without inverting the dependency graph. rag_policy.filter and the top-level rag_policy imports continue to re-export every symbol unchanged. See docs/architecture/retrieval-read-layer.md for the consumer contract and per-backend translator details.

Internals

Why a separate package

Three reasons:

  1. Lifecycle independence. Policy will evolve (new decisions, new backends like an OPA adapter) on a different cadence from the core types.
  2. Dependency surface. rag-policy depends only on rag-core and rag-observability. Production deployments that disable the noop and load an external PDP do not need to drop in a heavier package.
  3. CLAUDE.md graph. The standing constraint records policy → core; keeping it in its own package makes the boundary mechanical, not just social.

PolicyWriter vs PolicyEngine

PolicyEngine is the SPI authors implement. PolicyWriter is the facade consumers should use. The split mirrors AuditStore ↔ AuditWriter in rag-core — the writer adds the cross-cutting concern (structured-log emission) that every consumer needs but no SPI implementation should duplicate.

Writer logs go through rag_observability.logging.get_logger(__name__) and land in the standard 7-field JSON envelope under the message policy.decision. Fields: rag_decision, rag_outcome, rag_tenant_id, rag_principal_id, rag_request_id, rag_reason.

Coverage linter

tests/policy/coverage.py greps for direct calls to governance-relevant SPI methods (retrieve_ids, hydrate, bulk_index, stream_index, bulk_embed, complete) in files outside an allowlist. If a file contains such a call but no PolicyEngine / PolicyWriter marker, CI fails.

The allowlist at the top of coverage.py contains:

  • SPI abstracts and noop reference impls (the call surface itself).
  • Real backend impls (governance happens at the call site, not inside the backend).
  • Tests (fixtures exercise the SPI directly).
  • The rag-policy package (its writer/engine are the policed surface).

As Steps 1.10, 3.1, etc. add consumers, their files come off the allowlist as PolicyEngine is wired in. A future Step 1.1f or 3.x tightens the linter from file-allowlist to call-pattern matching.

Performance

  • NoopPolicyEngine.evaluate is essentially free; filter_pushdown constructs one And(Eq(...)) per call.
  • Production impls are expected to cache decisions keyed by (ctx.principal, decision, subject_hash) for the lifetime of a RequestContext. See docs/architecture/performance.md for the published p99 budget (PolicyEngine.evaluate noop ≤ 100 µs).

Extension points

Implement rag_policy.PolicyEngine:

from rag_policy import (
    FilterExpr,
    PolicyDecision,
    PolicyEngine,
    PolicyResult,
    and_,
    any_in,
    eq,
)

class MyOrgPolicyEngine(PolicyEngine):
    async def evaluate(self, ctx, decision, subject) -> PolicyResult:
        if decision is PolicyDecision.read_chunk:
            if subject.acl_labels & ctx.principal.acl_labels:
                return PolicyResult.allow()
            return PolicyResult.deny("acl-mismatch")
        return PolicyResult.allow()

    async def filter_pushdown(self, ctx, decision) -> FilterExpr:
        return and_(
            eq("tenant_id", str(ctx.tenant_id)),
            any_in("acl_labels", list(ctx.principal.acl_labels)),
        )

    async def health(self) -> bool:
        return True

Register at the gateway composition root. Future built-ins on the roadmap: OpaPolicyEngine (sidecar delegation), CedarPolicyEngine (in-process AWS Cedar).

Related