Skip to main content
This walkthrough adds a custom predicate to the policy engine and surfaces it in replay.

1. Define the predicate in policy.py

Extend PolicyEngine.validate() with your invariant. Example — block plans when gas exceeds a ceiling:
Add max_gas_wei to PolicyConfig with a sensible default.

2. Extend GuardrailService (optional)

For checks that need observation context beyond PolicyEngine, add logic in apps/worker/ameo_worker/services/guardrail_service.pycheck_plan(). Existing examples: observation quality, balance sufficiency, gas spike, protocol whitelist.

3. Emit structured violations

The graph guardrail node already emits:
Keep violation strings stable — they appear in replay JSON.

4. Surface in replay

cycle_store.py maps guardrail_evaluated events to PolicyCheck rows. New violation codes appear automatically in the Policy validation node when checks fail.

5. Test

Add a unit test in tests/test_policy.py (or extend an existing test module) that asserts your predicate blocks an invalid plan.

6. Document

Add a row to Policy specification predicate table. Rule: predicates must be deterministic — no LLM calls inside the guardrail layer.