Adds PostgreSQL-level enforcement for agent safety rules, closing the gap where direct SQL could bypass the Python safety layer.
Changes
New trigger functionassert_agent_safety(): reads app.agent_id from transaction context via SET LOCAL, blocks status escalation to verified/published when agent context is present
Applied to 52 governed tables with status columns (excluding ai_summary and agent_task which already have CHECK constraints)
set_agent_context(conn, agent_id) helper in services/agents/safety.py for Python callers
Coverage
Tables include: impact_claim, carbon_credit, stakeholder_feedback, data_stream_post, crisp_risk_assessment, threat, delphi_recommendation, metric_proposal, stakeholder_outcome, credit_retirement, credit_class, and 41 more.
Tests
17 tests (8 unit + 9 DB integration) covering trigger creation, agent blocking, and human operation allowance.
- Create assert_agent_safety() trigger function reading app.agent_id
- Block status escalation to verified/published when agent context is set
- Apply trigger to 52 governed tables with status columns
- Add set_agent_context() helper to safety.py
- Add 17 tests (8 unit + 9 DB integration)
Summary
Adds PostgreSQL-level enforcement for agent safety rules, closing the gap where direct SQL could bypass the Python safety layer.
Changes
assert_agent_safety(): readsapp.agent_idfrom transaction context viaSET LOCAL, blocks status escalation toverified/publishedwhen agent context is presentset_agent_context(conn, agent_id)helper inservices/agents/safety.pyfor Python callersCoverage
Tables include: impact_claim, carbon_credit, stakeholder_feedback, data_stream_post, crisp_risk_assessment, threat, delphi_recommendation, metric_proposal, stakeholder_outcome, credit_retirement, credit_class, and 41 more.
Tests
17 tests (8 unit + 9 DB integration) covering trigger creation, agent blocking, and human operation allowance.