feat: egress schema and constitution checks at high bind (#78)

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Egress schema and constitution checks ([#78](https://github.com/ARPAHLS/aura/issues/78))** — profile `variables.schema_refs` (named JSON Schema specs per tool) and explicit `schema_check` constraint rules validate `tool.call` args and/or `tool.result` payloads at egress; spectrum `high` / `full` auto-wires refs when `schema_enforcement` is not false; audit finding `SCHEMA_VIOLATION`; conformance `schema` check on close; `session.open` and `aura agent show` spectrum summary include `schema` block; `merged_profile_rules()` helper for session/CLI parity; stress sim `schema_high_bind` scenario; depends on `jsonschema`.
 - **Capability broker ([#48](https://github.com/ARPAHLS/aura/issues/48))** — profile `capabilities[]` declares named intents and secret **refs** only; constraint `capability_scope` on `tool.call`; SecretBroker (`EnvSecretBroker`, `MapSecretBroker`, `CallableSecretBroker`, `ChainSecretBroker`) injects the live token after allow; spine/summary/OTel redact secret-like keys and injected values; spectrum `low` records misses (`audit_only`, finding `CAPABILITY_AUDIT`) while `mid`+ blocks (`CAPABILITY_DENIED`); unresolved refs emit `tool.error` / `SECRET_BROKER_ERROR`; `capability.injected` records ref not value; CLI `aura agent set --capabilities-json` / `--capabilities-file`; `session(secret_broker=...)`; example `examples/14-capability-broker/`; stress sims `scripts/aura_capability_stress_sim.py` (16 scenarios) and six host-sim coats (33 host scenarios total).
 - **Escalation playbooks ([#47](https://github.com/ARPAHLS/aura/issues/47))** — profile `escalations[]` (agent registry / SDK — not manifest bindings) maps trigger events (`slo.missed`, `conformance.drift`, `observer.alert`, `constraint.violated`) to actions (`log`, `alert`, `nudge`, `pause`, `email`, `wake`, `custom`); spine events `escalation.fired`, `membrane.nudge`, `escalation.email`; destructive `pause` requires spectrum ≥ mid and uses `escalation_pause` + `approve()`; session-only `escalation_handler` callback; audit finding `ESCALATION_FIRED`; example `examples/12-escalation-playbooks/` (six scenarios); stress sim `escalation_slo_playbook` scenario (27 total).
 - **Verified identity via spectrum ([#73](https://github.com/ARPAHLS/aura/issues/73))** — `spectrum.identity_required` and level defaults (`high` / `full` → verified operator mandatory when a profile has a `spectrum` block); session open fails with `IdentityRequiredError` when policy requires verified IdP identity and none resolves; lite `aura_id` / `agent_ref` unchanged; `session.open` spectrum summary includes `verified_identity_required` + source; audit finding `VERIFIED_IDENTITY_REQUIRED`; `aura run --require-identity` session override; `aura agent show` / `aura config show` document policy.

CONTRIBUTING.md

@@ -157,7 +157,7 @@ Pure internal refactors with no user-visible effect may omit CHANGELOG; ask on t
 | `spectrum.level` / enforcement rules | `aura/core/spectrum_enforcement.py`, `docs/aura-levels.md`, `docs/comparison.md`, `docs/ROADMAP.md`, `tests/test_spectrum_enforcement.py`, stress sim / flow report scripts, CHANGELOG |
 | `spectrum.services[]` / field-service wiring | `aura/core/spectrum_services.py`, `aura/observers/presets/limit.py`, `spec/manifest.schema.json`, `docs/observers.md`, `docs/field-services.md`, `docs/using-aura.md`, `tests/test_spectrum_services.py`, stress sim / flow report scripts, CHANGELOG |
 | `escalations[]` playbooks | `aura/core/escalations.py`, `aura/core/constraints.py` (`escalation_pause`), `aura/agents/profile.py`, `docs/observers.md`, `docs/outputs.md`, `examples/12-escalation-playbooks/`, `tests/test_escalations.py`, `tests/test_example_12_escalation_playbooks.py`, stress sim, CHANGELOG |
-| Constraint rule types | `docs/concepts.md`, `docs/capabilities.md` (for `capability_scope`), `aura/core/constraints.py` tests, CHANGELOG |
+| Constraint rule types | `docs/concepts.md`, `docs/capabilities.md` (for `capability_scope`), `docs/aura-levels.md` (for `schema_check`), `aura/core/constraints.py`, `aura/core/schema_validation.py`, `tests/test_schema_checks.py`, stress sim, CHANGELOG |
 | Sequencer step model | `spec/sequencer.schema.json`, `docs/sequencer.md`, `tests/test_v02.py`, CHANGELOG |
 | Skillware host / egress | `integrations/skillware/` (when shipped), `docs/skillware-integration.md` redirect, CHANGELOG |
 | Integration example (Ollama, API, framework) | `integrations/<stack>/`, `docs/integrations/README.md`, `.env.example`, CHANGELOG |

aura/api.py

@@ -147,6 +147,25 @@ def session(
             _finalize_session_run(run, session, do_export=do_export)
 
 
+def merged_profile_rules(
+    profile: AgentProfile,
+    *,
+    session_rules: list[dict[str, Any]] | None = None,
+) -> list[dict[str, Any]]:
+    """Profile rules + spectrum enforcement + capability + schema merge."""
+    from aura.core.capabilities import merge_capability_rules
+    from aura.core.schema_validation import merge_schema_rules
+    from aura.core.spectrum_enforcement import enforcement_rules
+
+    merged = list(profile.rules or [])
+    merged.extend(enforcement_rules(profile))
+    merged = merge_capability_rules(profile, merged)
+    merged = merge_schema_rules(profile, merged)
+    if session_rules:
+        merged.extend(session_rules)
+    return merged
+
+
 def _build_session(
     agent: AgentHandle,
     mode: str | None,
@@ -158,15 +177,7 @@ def _build_session(
         session_mode = SessionMode(mode_str)
     except ValueError:
         session_mode = SessionMode.SCRIPT
-    merged_rules = list(agent.profile.rules)
-    from aura.core.spectrum_enforcement import enforcement_rules
-
-    merged_rules.extend(enforcement_rules(agent.profile))
-    from aura.core.capabilities import merge_capability_rules
-
-    merged_rules = merge_capability_rules(agent.profile, merged_rules)
-    if rules:
-        merged_rules.extend(rules)
+    merged_rules = merged_profile_rules(agent.profile, session_rules=rules)
     from aura.sequencer.spec import merge_sequencer_spec
 
     seq_spec = merge_sequencer_spec(agent.profile.sequencer, sequencer)

aura/cli/commands.py

@@ -98,8 +98,10 @@ def cmd_agent_show(name: str, *, console: Console | None = None) -> int:
         else:
             console.print(message, style="bold #FF9AA2")
         return 1
+    from aura.api import merged_profile_rules
     from aura.core.capabilities import capabilities_summary, parse_capabilities
     from aura.core.escalations import escalation_summary
+    from aura.core.schema_validation import parse_schema_refs, schema_enforcement_enabled
     from aura.core.spectrum_enforcement import effective_spectrum, enforcement_rules
     from aura.core.spectrum_identity import identity_policy_summary
 
@@ -108,6 +110,17 @@ def cmd_agent_show(name: str, *, console: Console | None = None) -> int:
     payload["effective_spectrum"] = spec.summary()
     payload["effective_spectrum"].update(identity_policy_summary(profile))
     payload["effective_spectrum"]["enforcement_rules"] = enforcement_rules(profile)
+    effective_rules = merged_profile_rules(profile)
+    schema_rules = [
+        r for r in effective_rules if (r.get("type") or r.get("kind")) == "schema_check"
+    ]
+    schema_refs = parse_schema_refs(profile.variables)
+    if schema_refs or schema_rules:
+        payload["effective_spectrum"]["schema"] = {
+            "refs": len(schema_refs),
+            "active_rules": len(schema_rules),
+            "auto_enforced": schema_enforcement_enabled(profile),
+        }
     payload["escalations"] = escalation_summary(profile)
     caps = parse_capabilities(profile.capabilities, strict=False)
     payload["capabilities_summary"] = capabilities_summary(caps)

aura/core/audit_report.py

@@ -63,7 +63,13 @@ def build(
                     1
                     for e in tool_denied
                     if (e.payload.get("rule") or {}).get("type")
-                    in ("deny_tools", "allow_tools", "capability_scope")
+                    in ("deny_tools", "allow_tools", "capability_scope", "schema_check")
+                    and not e.payload.get("audit_only")
+                ),
+                "schema_violations": sum(
+                    1
+                    for e in tool_denied
+                    if (e.payload.get("rule") or {}).get("type") == "schema_check"
                     and not e.payload.get("audit_only")
                 ),
                 "capability_audit": sum(
@@ -113,6 +119,22 @@ def build(
                         "whose allowed fields match the request, or update profile.capabilities."
                     )
                 continue
+            if rtype == "schema_check":
+                findings.append(
+                    {
+                        "severity": "high",
+                        "code": "SCHEMA_VIOLATION",
+                        "message": event.payload.get("message", "Schema validation failed"),
+                        "rule_type": rtype,
+                        "rule_ref": (rule.get("ref")),
+                        "event_id": event.event_id,
+                    }
+                )
+                recommendations.append(
+                    "Tool args or result did not match the declared JSON Schema — "
+                    "fix the payload or update variables.schema_refs / schema_check rules."
+                )
+                continue
             findings.append(
                 {
                     "severity": "high",

aura/core/conformance.py

@@ -67,6 +67,14 @@ def summarize(
                 for item in goal_slo_check.get("violations", []):
                     violations.append(item)
 
+        schema_check = self._check_schema(spine, declared_rules)
+        if schema_check:
+            checks.append(schema_check)
+            if not schema_check.get("passed", True):
+                passed = False
+                for item in schema_check.get("violations", []):
+                    violations.append(item)
+
         return ConformanceReport(
             passed=passed,
             violations=violations,
@@ -141,3 +149,45 @@ def _check_goal_slo(self, spine: AuditSpine) -> dict[str, Any] | None:
             "miss_count": len(misses),
             "violations": violations,
         }
+
+    def _check_schema(
+        self,
+        spine: AuditSpine,
+        declared_rules: list[dict[str, Any]],
+    ) -> dict[str, Any] | None:
+        schema_rules = [
+            r for r in declared_rules if (r.get("type") or r.get("kind")) == "schema_check"
+        ]
+        if not schema_rules:
+            return None
+
+        schema_violations = [
+            e
+            for e in spine.stream()
+            if e.kind == "constraint.violated"
+            and (e.payload.get("rule") or {}).get("type") == "schema_check"
+            and not e.payload.get("audit_only")
+        ]
+        tool_calls = [e for e in spine.stream() if e.kind == "tool.call"]
+        tool_results = [e for e in spine.stream() if e.kind == "tool.result"]
+
+        passed = not schema_violations
+        result: dict[str, Any] = {
+            "type": "schema",
+            "declared_schema_rules": len(schema_rules),
+            "tool_calls": len(tool_calls),
+            "tool_results": len(tool_results),
+            "schema_violations": len(schema_violations),
+            "passed": passed,
+        }
+        if not passed:
+            result["violations"] = [
+                {
+                    "kind": "schema.violation",
+                    "message": evt.payload.get("message", "Schema validation failed"),
+                    "event_id": evt.event_id,
+                    "rule_ref": (evt.payload.get("rule") or {}).get("ref"),
+                }
+                for evt in schema_violations
+            ]
+        return result

aura/core/constraints.py

@@ -202,6 +202,47 @@ def _rule_capability_scope(ctx: ConstraintContext, rule: dict[str, Any]) -> Cons
     return evaluate_capability_scope(ctx, rule)
 
 
+def _rule_schema_check(ctx: ConstraintContext, rule: dict[str, Any]) -> ConstraintResult | None:
+    from aura.core.schema_validation import (
+        payload_subject,
+        resolve_schema,
+        rule_applies_to_event,
+        validate_payload,
+    )
+
+    if not rule_applies_to_event(rule, ctx.event_kind, ctx.payload):
+        return None
+
+    schema = resolve_schema(rule, ctx.session_state)
+    if not schema:
+        ref = rule.get("ref")
+        return ConstraintResult(
+            passed=False,
+            rule=rule,
+            message=f"Schema ref not found or invalid: {ref or '(inline schema missing)'}",
+            blocked=True,
+        )
+
+    subject = payload_subject(ctx.payload, ctx.event_kind)
+    errors = validate_payload(schema, subject)
+    if errors:
+        tool = ctx.payload.get("tool") or ctx.payload.get("skill_id") or "unknown"
+        detail = "; ".join(errors[:3])
+        if len(errors) > 3:
+            detail += f" (+{len(errors) - 3} more)"
+        return ConstraintResult(
+            passed=False,
+            rule=rule,
+            message=f"Schema validation failed for {tool} on {ctx.event_kind}: {detail}",
+            blocked=True,
+        )
+    return ConstraintResult(
+        passed=True,
+        rule=rule,
+        message=f"Schema valid for {ctx.event_kind}",
+    )
+
+
 def _rule_escalation_pause(ctx: ConstraintContext, rule: dict[str, Any]) -> ConstraintResult | None:
     if ctx.event_kind not in ("tool.call", "action.request"):
         return None
@@ -231,4 +272,5 @@ def _rule_escalation_pause(ctx: ConstraintContext, rule: dict[str, Any]) -> Cons
     "sequencer_required": _rule_sequencer_required,
     "escalation_pause": _rule_escalation_pause,
     "capability_scope": _rule_capability_scope,
+    "schema_check": _rule_schema_check,
 }