feat: add workflow context, escalation routing, semantic rules

Author: practicalmind-dev

examples/agent_pipeline/__init__.py

@@ -0,0 +1,73 @@
+from gateframe.core.contract import ValidationContract
+from gateframe.core.escalation import EscalationRoute
+from gateframe.core.failure import FailureMode
+from gateframe.rules.semantic import SemanticRule
+from gateframe.rules.structural import StructuralRule
+
+from .models import ActionPlanOutput, ClassificationOutput, ExecutionOutput, SummaryOutput
+
+classification_contract = ValidationContract(
+    name="classification",
+    rules=[
+        StructuralRule(schema=ClassificationOutput),
+        SemanticRule(
+            check=lambda output, **ctx: output.get("confidence", 0) >= 0.5,
+            name="minimum_confidence",
+            description="Classification confidence must be at least 0.5",
+            failure_mode=FailureMode.SOFT_FAIL,
+            failure_message="Classification confidence below 0.5 threshold.",
+        ),
+    ],
+    on_hard_fail=EscalationRoute.ABORT,
+    on_soft_fail=EscalationRoute.FLAG_AND_CONTINUE,
+)
+
+action_plan_contract = ValidationContract(
+    name="action_plan",
+    rules=[
+        StructuralRule(schema=ActionPlanOutput),
+        SemanticRule(
+            check=lambda output, **ctx: len(output.get("steps", [])) > 0,
+            name="has_steps",
+            description="Action plan must have at least one step",
+            failure_mode=FailureMode.HARD_FAIL,
+            failure_message="Action plan has no steps.",
+        ),
+        SemanticRule(
+            check=lambda output, **ctx: (
+                output.get("requires_approval", False)
+                if output.get("estimated_risk") == "high"
+                else True
+            ),
+            name="high_risk_needs_approval",
+            description="High risk actions must require approval",
+            failure_mode=FailureMode.SOFT_FAIL,
+            failure_message="High risk plan does not require approval.",
+        ),
+    ],
+    on_hard_fail=EscalationRoute.HUMAN_REVIEW,
+    on_soft_fail=EscalationRoute.FLAG_AND_CONTINUE,
+)
+
+execution_contract = ValidationContract(
+    name="execution",
+    rules=[
+        StructuralRule(schema=ExecutionOutput),
+        SemanticRule(
+            check=lambda output, **ctx: (
+                output.get("success", False) or len(output.get("details", "")) > 10
+            ),
+            name="failure_explained",
+            description="Failed executions must have detailed explanation",
+            failure_mode=FailureMode.SOFT_FAIL,
+            failure_message="Execution failed without sufficient explanation.",
+        ),
+    ],
+)
+
+summary_contract = ValidationContract(
+    name="summary",
+    rules=[
+        StructuralRule(schema=SummaryOutput),
+    ],
+)

examples/agent_pipeline/contracts.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+
+
+class ClassificationOutput(BaseModel):
+    category: str
+    confidence: float
+    reasoning: str
+
+
+class ActionPlanOutput(BaseModel):
+    steps: list[str]
+    estimated_risk: str
+    requires_approval: bool
+
+
+class ExecutionOutput(BaseModel):
+    action_taken: str
+    success: bool
+    details: str
+
+
+class SummaryOutput(BaseModel):
+    outcome: str
+    actions_completed: list[str]
+    follow_up_needed: bool

examples/agent_pipeline/models.py

@@ -0,0 +1,85 @@
+"""Agent pipeline example: 4-step workflow with confidence degradation and escalation."""
+
+from gateframe.audit.log import AuditLog
+from gateframe.core.context import WorkflowContext
+from gateframe.core.escalation import EscalationRouter
+
+from .contracts import (
+    action_plan_contract,
+    classification_contract,
+    execution_contract,
+    summary_contract,
+)
+
+
+def main() -> None:
+    ctx = WorkflowContext(
+        workflow_id="incident_response_001",
+        escalation_threshold=0.5,
+    )
+    router = EscalationRouter()
+    audit = AuditLog()
+
+    # Step 1: Classification -- passes cleanly
+    step1_output = {
+        "category": "infrastructure",
+        "confidence": 0.92,
+        "reasoning": "CPU usage spike correlated with deployment at 14:32 UTC",
+    }
+    result1 = classification_contract.validate(step1_output)
+    ctx.update(result1)
+    audit.record(result1, workflow_context=ctx)
+    print(f"Step 1 (classification): passed={result1.passed}, confidence={ctx.confidence:.2f}")
+
+    # Step 2: Action plan -- soft fail (high risk without approval)
+    step2_output = {
+        "steps": ["Roll back deployment", "Check error logs", "Notify team"],
+        "estimated_risk": "high",
+        "requires_approval": False,  # this triggers soft fail
+    }
+    result2 = action_plan_contract.validate(step2_output)
+    ctx.update(result2)
+    audit.record(result2, workflow_context=ctx)
+    print(f"Step 2 (action_plan): passed={result2.passed}, confidence={ctx.confidence:.2f}")
+
+    # Step 3: Execution -- soft fail (failed without explanation)
+    step3_output = {
+        "action_taken": "Attempted rollback",
+        "success": False,
+        "details": "Timeout",  # too short, triggers soft fail
+    }
+    result3 = execution_contract.validate(step3_output)
+    ctx.update(result3)
+    audit.record(result3, workflow_context=ctx)
+    print(f"Step 3 (execution): passed={result3.passed}, confidence={ctx.confidence:.2f}")
+
+    # Step 4: Summary -- passes structurally
+    step4_output = {
+        "outcome": "Partial resolution",
+        "actions_completed": ["Rollback attempted", "Logs checked"],
+        "follow_up_needed": True,
+    }
+    result4 = summary_contract.validate(step4_output)
+    ctx.update(result4)
+    audit.record(result4, workflow_context=ctx)
+    print(f"Step 4 (summary): passed={result4.passed}, confidence={ctx.confidence:.2f}")
+
+    # Check threshold
+    print(f"\nThreshold breached: {ctx.threshold_breached}")
+    if ctx.threshold_breached:
+        escalation = router.route_threshold_breach(ctx)
+        if escalation:
+            print(f"Escalation: {escalation.route.value} -- {escalation.reason}")
+
+    # Audit trail
+    print(f"\nAudit entries: {len(audit.entries)}")
+    for entry in audit.entries:
+        data = entry.to_dict()
+        print(f"  [{data['contract_name']}] passed={data['passed']}", end="")
+        if data.get("confidence") is not None:
+            print(f", confidence={data['confidence']:.2f}", end="")
+        print()
+
+
+if __name__ == "__main__":
+    main()

examples/agent_pipeline/run.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
 from datetime import datetime, timezone
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import structlog
 
 from gateframe.core.contract import ValidationResult
 
+if TYPE_CHECKING:
+    from gateframe.core.context import WorkflowContext
+
 logger = structlog.get_logger()
 
 
@@ -18,9 +21,15 @@ class AuditEntry:
         "rules_applied",
         "rules_failed",
         "failures",
+        "workflow_id",
+        "confidence",
     )
 
-    def __init__(self, result: ValidationResult) -> None:
+    def __init__(
+        self,
+        result: ValidationResult,
+        workflow_context: WorkflowContext | None = None,
+    ) -> None:
         self.timestamp = datetime.now(timezone.utc)
         self.contract_name = result.contract_name
         self.passed = result.passed
@@ -34,32 +43,45 @@ def __init__(self, result: ValidationResult) -> None:
             }
             for f in result.failures
         ]
+        self.workflow_id = workflow_context.workflow_id if workflow_context else None
+        self.confidence = workflow_context.confidence if workflow_context else None
 
     def to_dict(self) -> dict[str, Any]:
-        return {
+        data: dict[str, Any] = {
             "timestamp": self.timestamp.isoformat(),
             "contract_name": self.contract_name,
             "passed": self.passed,
             "rules_applied": self.rules_applied,
             "rules_failed": self.rules_failed,
             "failures": self.failures,
         }
+        if self.workflow_id is not None:
+            data["workflow_id"] = self.workflow_id
+            data["confidence"] = self.confidence
+        return data
 
 
 class AuditLog:
     def __init__(self) -> None:
         self._entries: list[AuditEntry] = []
 
-    def record(self, result: ValidationResult) -> None:
-        entry = AuditEntry(result)
+    def record(
+        self,
+        result: ValidationResult,
+        workflow_context: WorkflowContext | None = None,
+    ) -> None:
+        entry = AuditEntry(result, workflow_context=workflow_context)
         self._entries.append(entry)
-        logger.info(
-            "validation_event",
-            contract=entry.contract_name,
-            passed=entry.passed,
-            rules_applied=entry.rules_applied,
-            rules_failed=entry.rules_failed,
-        )
+        log_kwargs: dict[str, Any] = {
+            "contract": entry.contract_name,
+            "passed": entry.passed,
+            "rules_applied": entry.rules_applied,
+            "rules_failed": entry.rules_failed,
+        }
+        if entry.workflow_id is not None:
+            log_kwargs["workflow_id"] = entry.workflow_id
+            log_kwargs["confidence"] = entry.confidence
+        logger.info("validation_event", **log_kwargs)
 
     @property
     def entries(self) -> list[AuditEntry]: