Add validation-driven workflow decisions

backend/app/agents/decision.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class DecisionAgent:
+    """
+    Makes a deterministic workflow decision from validation results.
+
+    The decision controls which path the LangGraph workflow takes.
+    """
+
+    def decide(
+        self,
+        validation_results: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+
+        failures = [
+            result
+            for result in validation_results
+            if result.get("status") == "FAIL"
+        ]
+
+        warnings = [
+            result
+            for result in validation_results
+            if result.get("status") == "WARNING"
+        ]
+
+        affected_proposal_ids = sorted(
+            {
+                proposal_id
+                for result in validation_results
+                if result.get("status") in {"FAIL", "WARNING"}
+                for proposal_id in result.get("proposal_ids", [])
+            }
+        )
+
+        if failures:
+            return {
+                "decision": "REVIEW",
+                "reason": (
+                    f"{len(failures)} validation finding(s) "
+                    "require human review."
+                ),
+                "severity": "HIGH",
+                "failure_count": len(failures),
+                "warning_count": len(warnings),
+                "affected_proposal_ids": affected_proposal_ids,
+            }
+
+        if warnings:
+            return {
+                "decision": "REVIEW",
+                "reason": (
+                    f"{len(warnings)} validation warning(s) "
+                    "require human review."
+                ),
+                "severity": "MEDIUM",
+                "failure_count": 0,
+                "warning_count": len(warnings),
+                "affected_proposal_ids": affected_proposal_ids,
+            }
+
+        return {
+            "decision": "CONTINUE",
+            "reason": "All validation rules passed.",
+            "severity": "INFO",
+            "failure_count": 0,
+            "warning_count": 0,
+            "affected_proposal_ids": [],
+        }

backend/app/agents/validation.py

@@ -0,0 +1,337 @@
+from __future__ import annotations
+
+from typing import Any
+
+from app.models.proposal import Proposal, ProposalType
+from app.models.rule import Rule, RuleType
+
+
+class RuleValidationAgent:
+    """
+    Deterministic rule validation engine.
+
+    Supported operators:
+
+        required_evidence
+        min_confidence
+        allowed_proposal_types
+
+    Unknown operators produce a WARNING rather than silently passing.
+    """
+
+    def validate(
+        self,
+        *,
+        rules: list[Rule],
+        proposals: list[Proposal],
+    ) -> list[dict[str, Any]]:
+        results: list[dict[str, Any]] = []
+
+        if not rules:
+            return [
+                {
+                    "status": "PASS",
+                    "severity": "INFO",
+                    "rule_id": None,
+                    "rule_name": "No enabled rules",
+                    "rule_type": None,
+                    "operator": None,
+                    "message": "No enabled validation rules are configured.",
+                    "proposal_ids": [
+                        str(proposal.id) for proposal in proposals
+                    ],
+                }
+            ]
+
+        for rule in rules:
+            configuration = rule.configuration or {}
+            operator = configuration.get("operator")
+
+            if operator == "required_evidence":
+                results.extend(
+                    self._validate_required_evidence(
+                        rule=rule,
+                        proposals=proposals,
+                    )
+                )
+
+            elif operator == "min_confidence":
+                results.extend(
+                    self._validate_min_confidence(
+                        rule=rule,
+                        proposals=proposals,
+                    )
+                )
+
+            elif operator == "allowed_proposal_types":
+                results.extend(
+                    self._validate_allowed_proposal_types(
+                        rule=rule,
+                        proposals=proposals,
+                    )
+                )
+
+            else:
+                results.append(
+                    {
+                        "status": "WARNING",
+                        "severity": "MEDIUM",
+                        "rule_id": str(rule.id),
+                        "rule_name": rule.name,
+                        "rule_type": (
+                            rule.rule_type.value
+                            if isinstance(rule.rule_type, RuleType)
+                            else str(rule.rule_type)
+                        ),
+                        "operator": operator,
+                        "message": (
+                            f"Unsupported rule operator: {operator!r}. "
+                            "The rule was not evaluated."
+                        ),
+                        "proposal_ids": [
+                            str(proposal.id)
+                            for proposal in proposals
+                        ],
+                    }
+                )
+
+        return results
+
+    def _validate_required_evidence(
+        self,
+        *,
+        rule: Rule,
+        proposals: list[Proposal],
+    ) -> list[dict[str, Any]]:
+        failures = []
+
+        for proposal in proposals:
+            changes = proposal.proposed_changes or {}
+            proposed = changes.get("proposed", changes)
+            evidence = proposed.get("evidence")
+
+            if not evidence:
+                failures.append(
+                    {
+                        "status": "FAIL",
+                        "severity": "HIGH",
+                        "rule_id": str(rule.id),
+                        "rule_name": rule.name,
+                        "rule_type": (
+                            rule.rule_type.value
+                            if isinstance(rule.rule_type, RuleType)
+                            else str(rule.rule_type)
+                        ),
+                        "operator": "required_evidence",
+                        "message": (
+                            "Proposal does not contain evidence metadata "
+                            "in proposed_changes."
+                        ),
+                        "proposal_ids": [str(proposal.id)],
+                    }
+                )
+
+        if failures:
+            return failures
+
+        return [
+            {
+                "status": "PASS",
+                "severity": "INFO",
+                "rule_id": str(rule.id),
+                "rule_name": rule.name,
+                "rule_type": (
+                    rule.rule_type.value
+                    if isinstance(rule.rule_type, RuleType)
+                    else str(rule.rule_type)
+                ),
+                "operator": "required_evidence",
+                "message": "All proposals contain evidence metadata.",
+                "proposal_ids": [
+                    str(proposal.id) for proposal in proposals
+                ],
+            }
+        ]
+
+    def _validate_min_confidence(
+        self,
+        *,
+        rule: Rule,
+        proposals: list[Proposal],
+    ) -> list[dict[str, Any]]:
+        configuration = rule.configuration or {}
+        minimum = configuration.get("value")
+
+        if not isinstance(minimum, (int, float)):
+            return [
+                {
+                    "status": "WARNING",
+                    "severity": "HIGH",
+                    "rule_id": str(rule.id),
+                    "rule_name": rule.name,
+                    "rule_type": (
+                        rule.rule_type.value
+                        if isinstance(rule.rule_type, RuleType)
+                        else str(rule.rule_type)
+                    ),
+                    "operator": "min_confidence",
+                    "message": (
+                        "min_confidence requires a numeric "
+                        "'value' configuration."
+                    ),
+                    "proposal_ids": [],
+                }
+            ]
+
+        results = []
+
+        for proposal in proposals:
+            changes = proposal.proposed_changes or {}
+            proposed = changes.get("proposed", changes)
+            confidence = proposed.get("confidence")
+
+            if confidence is None:
+                results.append(
+                    {
+                        "status": "FAIL",
+                        "severity": "HIGH",
+                        "rule_id": str(rule.id),
+                        "rule_name": rule.name,
+                        "rule_type": (
+                            rule.rule_type.value
+                            if isinstance(rule.rule_type, RuleType)
+                            else str(rule.rule_type)
+                        ),
+                        "operator": "min_confidence",
+                        "message": (
+                            "Proposal does not contain a confidence value."
+                        ),
+                        "proposal_ids": [str(proposal.id)],
+                    }
+                )
+                continue
+
+            if confidence < minimum:
+                results.append(
+                    {
+                        "status": "FAIL",
+                        "severity": "HIGH",
+                        "rule_id": str(rule.id),
+                        "rule_name": rule.name,
+                        "rule_type": (
+                            rule.rule_type.value
+                            if isinstance(rule.rule_type, RuleType)
+                            else str(rule.rule_type)
+                        ),
+                        "operator": "min_confidence",
+                        "message": (
+                            f"Confidence {confidence:.3f} is below "
+                            f"required minimum {minimum:.3f}."
+                        ),
+                        "proposal_ids": [str(proposal.id)],
+                    }
+                )
+            else:
+                results.append(
+                    {
+                        "status": "PASS",
+                        "severity": "INFO",
+                        "rule_id": str(rule.id),
+                        "rule_name": rule.name,
+                        "rule_type": (
+                            rule.rule_type.value
+                            if isinstance(rule.rule_type, RuleType)
+                            else str(rule.rule_type)
+                        ),
+                        "operator": "min_confidence",
+                        "message": (
+                            f"Confidence {confidence:.3f} meets "
+                            f"minimum {minimum:.3f}."
+                        ),
+                        "proposal_ids": [str(proposal.id)],
+                    }
+                )
+
+        return results
+
+    def _validate_allowed_proposal_types(
+        self,
+        *,
+        rule: Rule,
+        proposals: list[Proposal],
+    ) -> list[dict[str, Any]]:
+        configuration = rule.configuration or {}
+        allowed = configuration.get("values")
+
+        if not isinstance(allowed, list):
+            return [
+                {
+                    "status": "WARNING",
+                    "severity": "HIGH",
+                    "rule_id": str(rule.id),
+                    "rule_name": rule.name,
+                    "rule_type": (
+                        rule.rule_type.value
+                        if isinstance(rule.rule_type, RuleType)
+                        else str(rule.rule_type)
+                    ),
+                    "operator": "allowed_proposal_types",
+                    "message": (
+                        "allowed_proposal_types requires a list "
+                        "under 'values'."
+                    ),
+                    "proposal_ids": [],
+                }
+            ]
+
+        allowed = {str(value).upper() for value in allowed}
+        results = []
+
+        for proposal in proposals:
+            proposal_type = (
+                proposal.proposal_type.value
+                if isinstance(proposal.proposal_type, ProposalType)
+                else str(proposal.proposal_type)
+            )
+
+            if proposal_type.upper() not in allowed:
+                results.append(
+                    {
+                        "status": "FAIL",
+                        "severity": "HIGH",
+                        "rule_id": str(rule.id),
+                        "rule_name": rule.name,
+                        "rule_type": (
+                            rule.rule_type.value
+                            if isinstance(rule.rule_type, RuleType)
+                            else str(rule.rule_type)
+                        ),
+                        "operator": "allowed_proposal_types",
+                        "message": (
+                            f"Proposal type {proposal_type!r} is not "
+                            f"allowed by this rule."
+                        ),
+                        "proposal_ids": [str(proposal.id)],
+                    }
+                )
+            else:
+                results.append(
+                    {
+                        "status": "PASS",
+                        "severity": "INFO",
+                        "rule_id": str(rule.id),
+                        "rule_name": rule.name,
+                        "rule_type": (
+                            rule.rule_type.value
+                            if isinstance(rule.rule_type, RuleType)
+                            else str(rule.rule_type)
+                        ),
+                        "operator": "allowed_proposal_types",
+                        "message": (
+                            f"Proposal type {proposal_type!r} is allowed."
+                        ),
+                        "proposal_ids": [str(proposal.id)],
+                    }
+                )
+
+        return results

backend/app/repositories/rule_repository.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from app.models.rule import Rule
+
+
+class RuleRepository:
+    """Repository for workspace-scoped validation rules."""
+
+    def __init__(self, db: Session):
+        self.db = db
+
+    def list_enabled(self, workspace_id) -> list[Rule]:
+        return (
+            self.db.query(Rule)
+            .filter(
+                Rule.workspace_id == workspace_id,
+                Rule.enabled.is_(True),
+            )
+            .order_by(Rule.created_at.asc())
+            .all()
+        )

backend/app/workflow/graph.py

@@ -1,4 +1,4 @@
-from __future__ import annotations
+from __future__ import annotations
 
 from sqlalchemy.orm import Session
 from langgraph.graph import END, START, StateGraph
@@ -7,6 +7,7 @@ from app.llm.client import LLMClient
 from app.repositories.knowledge_repository import KnowledgeRepository
 from app.services.chunking_service import ChunkingService
 from app.services.knowledge_link_service import KnowledgeLinkService
+from app.services.embedding_service import EmbeddingService
 
 from app.workflow.nodes.chunk import chunk
 from app.workflow.nodes.classify import classify
@@ -15,11 +16,19 @@ from app.workflow.nodes.extract import extract
 from app.workflow.nodes.knowledge import knowledge
 from app.workflow.nodes.link import link
 from app.workflow.nodes.reconcile import reconcile
-from app.workflow.state import WorkflowState
-from app.services.embedding_service import EmbeddingService
 from app.workflow.nodes.embed import embed
+from app.workflow.nodes.validate import validate
+from app.workflow.nodes.decision import decide
+
+from app.workflow.router import route_after_decision
+from app.workflow.state import WorkflowState
 
-def build_workflow(db: Session, checkpointer=None,interrupt_before=None,):
+
+def build_workflow(
+    db: Session,
+    checkpointer=None,
+    interrupt_before=None,
+):
     """
     Build the DocWeave document intelligence workflow.
     """
@@ -29,13 +38,17 @@ def build_workflow(db: Session, checkpointer=None,interrupt_before=None,):
     chunking_service = ChunkingService(db)
     knowledge_link_service = KnowledgeLinkService(db)
     embedding_service = EmbeddingService()
+
     graph = StateGraph(WorkflowState)
 
     # ------------------------------------------------------------------
     # Nodes
     # ------------------------------------------------------------------
 
-    graph.add_node("extract", extract)
+    graph.add_node(
+        "extract",
+        extract,
+    )
 
     graph.add_node(
         "chunk",
@@ -45,7 +58,19 @@ def build_workflow(db: Session, checkpointer=None,interrupt_before=None,):
         ),
     )
 
-    graph.add_node("classify", classify)
+    graph.add_node(
+        "embedding",
+        lambda state: embed(
+            state,
+            db,
+            embedding_service,
+        ),
+    )
+
+    graph.add_node(
+        "classify",
+        classify,
+    )
 
     graph.add_node(
         "knowledge",
@@ -64,6 +89,19 @@ def build_workflow(db: Session, checkpointer=None,interrupt_before=None,):
         ),
     )
 
+    graph.add_node(
+        "validation",
+        lambda state: validate(
+            state,
+            db,
+        ),
+    )
+
+    graph.add_node(
+        "decision",
+        decide,
+    )
+
     graph.add_node(
         "link",
         lambda state: link(
@@ -72,29 +110,94 @@ def build_workflow(db: Session, checkpointer=None,interrupt_before=None,):
         ),
     )
 
-    graph.add_node("complete", complete)
+    # Temporary branch target.
+    # This will become the real durable human gate later.
+    def human_review(state: WorkflowState) -> WorkflowState:
+        state.current_node = "HUMAN_REVIEW"
+        return state
 
     graph.add_node(
-        "embedding",
-        lambda state: embed(
-            state,
-            db,
-            embedding_service,
-        ),
+        "human_review",
+        human_review,
     )
+
+    graph.add_node(
+        "complete",
+        complete,
+    )
+
     # ------------------------------------------------------------------
     # Workflow edges
     # ------------------------------------------------------------------
 
-    graph.add_edge(START, "extract")
-    graph.add_edge("extract", "chunk")
-    graph.add_edge("chunk", "embedding")
-    graph.add_edge("embedding", "classify")
-    graph.add_edge("classify", "knowledge")
-    graph.add_edge("knowledge", "reconciliation")
-    graph.add_edge("reconciliation", "link")
-    graph.add_edge("link", "complete")
-    graph.add_edge("complete", END)
-
-    
-    return graph.compile(checkpointer=checkpointer, interrupt_before=interrupt_before,)
+    graph.add_edge(
+        START,
+        "extract",
+    )
+
+    graph.add_edge(
+        "extract",
+        "chunk",
+    )
+
+    graph.add_edge(
+        "chunk",
+        "embedding",
+    )
+
+    graph.add_edge(
+        "embedding",
+        "classify",
+    )
+
+    graph.add_edge(
+        "classify",
+        "knowledge",
+    )
+
+    graph.add_edge(
+        "knowledge",
+        "reconciliation",
+    )
+
+    graph.add_edge(
+        "reconciliation",
+        "validation",
+    )
+
+    graph.add_edge(
+        "validation",
+        "decision",
+    )
+
+    # Decision controls the workflow path.
+    graph.add_conditional_edges(
+        "decision",
+        route_after_decision,
+        {
+            "link": "link",
+            "human_review": "human_review",
+        },
+    )
+
+    graph.add_edge(
+        "link",
+        "complete",
+    )
+
+    # Temporary review branch returns to completion.
+    # We will replace this with the real human gate later.
+    graph.add_edge(
+        "human_review",
+        "complete",
+    )
+
+    graph.add_edge(
+        "complete",
+        END,
+    )
+
+    return graph.compile(
+        checkpointer=checkpointer,
+        interrupt_before=interrupt_before,
+    )

backend/app/workflow/nodes/__init__.py

@@ -1,9 +1,11 @@
 from .extract import extract
 from .classify import classify
 from .complete import complete
+from .validate import validate
 
 __all__ = [
     "extract",
     "classify",
     "complete",
+    "validate",
 ]

backend/app/workflow/nodes/decision.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from app.agents.decision import DecisionAgent
+from app.workflow.state import WorkflowState
+
+
+def decide(state: WorkflowState) -> WorkflowState:
+    """
+    Convert validation findings into a workflow decision.
+    """
+
+    state.current_node = "DECISION"
+
+    agent = DecisionAgent()
+
+    state.decision = agent.decide(
+        state.validation_results
+    )
+
+    return state

backend/app/workflow/nodes/validate.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from app.agents.validation import RuleValidationAgent
+from app.models.proposal import Proposal, ProposalType, ProposalStatus
+from app.repositories.knowledge_repository import KnowledgeRepository
+from app.repositories.rule_repository import RuleRepository
+from app.workflow.state import WorkflowState
+
+
+def validate(
+    state: WorkflowState,
+    db,
+) -> WorkflowState:
+    """
+    Validate proposals generated by reconciliation against
+    enabled workspace rules.
+    """
+
+    state.current_node = "RULE_VALIDATION"
+
+    knowledge_repository = KnowledgeRepository(db)
+    rule_repository = RuleRepository(db)
+
+    new_items = knowledge_repository.list_by_document_version(
+        state.document_version_id
+    )
+
+    new_item_ids = {str(item.id) for item in new_items}
+
+    if not new_item_ids:
+        state.validation_results = []
+        state.metadata["validation_summary"] = {
+            "status": "PASS",
+            "rules_evaluated": 0,
+            "proposals_evaluated": 0,
+            "failures": 0,
+            "warnings": 0,
+        }
+        return state
+
+    proposals = (
+        db.query(Proposal)
+        .filter(
+            Proposal.workspace_id == state.workspace_id,
+            Proposal.status == ProposalStatus.PENDING,
+        )
+        .all()
+    )
+
+    relevant_proposals = []
+
+    for proposal in proposals:
+        changes = proposal.proposed_changes or {}
+
+        if proposal.proposal_type == ProposalType.CREATE:
+            if proposal.knowledge_item_id is not None:
+                if str(proposal.knowledge_item_id) in new_item_ids:
+                    relevant_proposals.append(proposal)
+
+        elif proposal.proposal_type == ProposalType.UPDATE:
+            source_id = changes.get("source_knowledge_item_id")
+
+            if source_id and str(source_id) in new_item_ids:
+                relevant_proposals.append(proposal)
+
+    rules = rule_repository.list_enabled(state.workspace_id)
+
+    agent = RuleValidationAgent()
+
+    results = agent.validate(
+        rules=rules,
+        proposals=relevant_proposals,
+    )
+
+    state.validation_results = results
+
+    failures = sum(
+        1 for result in results if result["status"] == "FAIL"
+    )
+
+    warnings = sum(
+        1 for result in results if result["status"] == "WARNING"
+    )
+
+    state.metadata["validation_summary"] = {
+        "status": (
+            "FAIL"
+            if failures
+            else "WARNING"
+            if warnings
+            else "PASS"
+        ),
+        "rules_evaluated": len(rules),
+        "proposals_evaluated": len(relevant_proposals),
+        "failures": failures,
+        "warnings": warnings,
+    }
+
+    return state

backend/app/workflow/router.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def route_after_decision(state: Any) -> str:
+    """
+    Route the workflow based on the decision stored in WorkflowState.
+    """
+
+    decision = state.decision.get("decision")
+
+    if decision == "CONTINUE":
+        return "link"
+
+    if decision == "REVIEW":
+        return "human_review"
+
+    raise ValueError(
+        f"Unknown workflow decision: {decision!r}"
+    )

backend/app/workflow/state.py

@@ -29,6 +29,11 @@ class WorkflowState:
 
     metadata: dict[str, Any] = field(default_factory=dict)
 
+    validation_results: list[dict[str, Any]] = field(
+        default_factory=list
+    )
+
+    decision: dict[str, Any] = field(default_factory=dict)
     current_node: str = "START"
 
     completed: bool = False

backend/test_decision.py

@@ -0,0 +1,82 @@
+from app.agents.decision import DecisionAgent
+
+
+agent = DecisionAgent()
+
+
+print("\nTEST 1 - ALL PASS")
+
+result = agent.decide([
+    {
+        "status": "PASS",
+        "severity": "INFO",
+        "proposal_ids": ["proposal-1"],
+    }
+])
+
+print(result)
+
+assert result["decision"] == "CONTINUE"
+
+
+print("\nTEST 2 - FAIL")
+
+result = agent.decide([
+    {
+        "status": "FAIL",
+        "severity": "HIGH",
+        "proposal_ids": ["proposal-2"],
+    }
+])
+
+print(result)
+
+assert result["decision"] == "REVIEW"
+assert result["failure_count"] == 1
+assert "proposal-2" in result["affected_proposal_ids"]
+
+
+print("\nTEST 3 - WARNING")
+
+result = agent.decide([
+    {
+        "status": "WARNING",
+        "severity": "MEDIUM",
+        "proposal_ids": ["proposal-3"],
+    }
+])
+
+print(result)
+
+assert result["decision"] == "REVIEW"
+assert result["warning_count"] == 1
+assert "proposal-3" in result["affected_proposal_ids"]
+
+
+print("\nTEST 4 - FAIL + WARNING")
+
+result = agent.decide([
+    {
+        "status": "FAIL",
+        "severity": "HIGH",
+        "proposal_ids": ["proposal-4"],
+    },
+    {
+        "status": "WARNING",
+        "severity": "MEDIUM",
+        "proposal_ids": ["proposal-5"],
+    }
+])
+
+print(result)
+
+assert result["decision"] == "REVIEW"
+assert result["failure_count"] == 1
+assert result["warning_count"] == 1
+assert "proposal-4" in result["affected_proposal_ids"]
+assert "proposal-5" in result["affected_proposal_ids"]
+
+
+print("\n===================================")
+print("ALL DECISION TESTS PASSED")
+print("===================================")

backend/test_router.py

@@ -0,0 +1,53 @@
+from types import SimpleNamespace
+
+from app.workflow.router import route_after_decision
+
+
+print("\nTEST 1 - CONTINUE")
+
+state = SimpleNamespace(
+    decision={
+        "decision": "CONTINUE",
+    }
+)
+
+result = route_after_decision(state)
+
+print("ROUTE:", result)
+
+assert result == "link"
+
+
+print("\nTEST 2 - REVIEW")
+
+state = SimpleNamespace(
+    decision={
+        "decision": "REVIEW",
+    }
+)
+
+result = route_after_decision(state)
+
+print("ROUTE:", result)
+
+assert result == "human_review"
+
+
+print("\nTEST 3 - UNKNOWN DECISION")
+
+state = SimpleNamespace(
+    decision={
+        "decision": "SOMETHING_ELSE",
+    }
+)
+
+try:
+    route_after_decision(state)
+    raise AssertionError("Expected ValueError")
+except ValueError as exc:
+    print("EXPECTED ERROR:", exc)
+
+
+print("\n===================================")
+print("ALL ROUTER TESTS PASSED")
+print("===================================")

backend/test_validation.py

@@ -0,0 +1,168 @@
+from types import SimpleNamespace
+from uuid import uuid4
+
+from app.agents.validation import RuleValidationAgent
+
+
+agent = RuleValidationAgent()
+
+
+def make_rule(name, configuration):
+    return SimpleNamespace(
+        id=uuid4(),
+        name=name,
+        description="test rule",
+        rule_type="COMPLIANCE",
+        configuration=configuration,
+        enabled=True,
+    )
+
+
+def make_proposal(confidence=0.90, evidence=True):
+    proposed = {
+        "confidence": confidence,
+    }
+
+    if evidence:
+        proposed["evidence"] = {
+            "source": "test.pdf",
+            "page": 1,
+        }
+
+    return SimpleNamespace(
+        id=uuid4(),
+        proposal_type="CREATE",
+        proposed_changes=proposed,
+    )
+
+
+# TEST 1: No rules
+result = agent.validate(
+    rules=[],
+    proposals=[make_proposal()],
+)
+
+print("\nTEST 1 - NO RULES")
+print(result)
+
+
+# TEST 2: Confidence passes
+result = agent.validate(
+    rules=[
+        make_rule(
+            "Minimum confidence",
+            {
+                "operator": "min_confidence",
+                "value": 0.80,
+            },
+        )
+    ],
+    proposals=[
+        make_proposal(confidence=0.90),
+    ],
+)
+
+print("\nTEST 2 - CONFIDENCE PASS")
+print(result)
+
+
+# TEST 3: Confidence fails
+result = agent.validate(
+    rules=[
+        make_rule(
+            "Minimum confidence",
+            {
+                "operator": "min_confidence",
+                "value": 0.80,
+            },
+        )
+    ],
+    proposals=[
+        make_proposal(confidence=0.50),
+    ],
+)
+
+print("\nTEST 3 - CONFIDENCE FAIL")
+print(result)
+
+
+# TEST 4: Evidence passes
+result = agent.validate(
+    rules=[
+        make_rule(
+            "Required evidence",
+            {
+                "operator": "required_evidence",
+            },
+        )
+    ],
+    proposals=[
+        make_proposal(evidence=True),
+    ],
+)
+
+print("\nTEST 4 - EVIDENCE PASS")
+print(result)
+
+
+# TEST 5: Evidence fails
+result = agent.validate(
+    rules=[
+        make_rule(
+            "Required evidence",
+            {
+                "operator": "required_evidence",
+            },
+        )
+    ],
+    proposals=[
+        make_proposal(evidence=False),
+    ],
+)
+
+print("\nTEST 5 - EVIDENCE FAIL")
+print(result)
+
+
+# TEST 6: Malformed rule
+result = agent.validate(
+    rules=[
+        make_rule(
+            "Broken confidence rule",
+            {
+                "operator": "min_confidence",
+                "value": "not-a-number",
+            },
+        )
+    ],
+    proposals=[
+        make_proposal(),
+    ],
+)
+
+print("\nTEST 6 - MALFORMED RULE")
+print(result)
+
+
+# TEST 7: Unsupported operator
+result = agent.validate(
+    rules=[
+        make_rule(
+            "Unknown rule",
+            {
+                "operator": "unsupported_operator",
+            },
+        )
+    ],
+    proposals=[
+        make_proposal(),
+    ],
+)
+
+print("\nTEST 7 - UNKNOWN OPERATOR")
+print(result)
+
+
+print("\n===================================")
+print("VALIDATION TEST RUN COMPLETE")
+print("===================================")