feat(G): G-WORK/G-HUMAN-DEC/G-UI-TEMPL/G-DOC — workflow workstream, decision guard, template gallery (3 templates), API docs, 43 tests passing

tests/test_phase_g_workflows.py

@@ -370,3 +370,169 @@ class TestWorkflowModelDurableFields:
         assert hasattr(WorkflowInstance, "error_message")
         assert hasattr(WorkflowInstance, "retry_count")
         assert hasattr(WorkflowInstance, "max_retries")
+
+
+# ─── G-HUMAN-DEC: Automated-Decision Guard ────────────────────────────────────
+
+
+class TestDecisionGuard:
+    """Test the automated-decision guard (G-HUMAN-DEC)."""
+
+    def test_no_metadata_allows_everything(self):
+        """No AI use case metadata means no guard — action allowed."""
+        from app.workflows.decision_guard import requires_human_review
+
+        assert requires_human_review(None, "send_email") is False
+        assert requires_human_review({}, "delete_entity") is False
+
+    def test_auto_execute_false_requires_review(self):
+        """auto_execute=False always requires review."""
+        from app.workflows.decision_guard import requires_human_review
+
+        metadata = {"auto_execute": False}
+        assert requires_human_review(metadata, "any_action") is True
+
+    def test_requires_approval_true_requires_review(self):
+        """requires_approval=True always requires review."""
+        from app.workflows.decision_guard import requires_human_review
+
+        metadata = {"requires_approval": True}
+        assert requires_human_review(metadata, "any_action") is True
+
+    def test_medium_risk_requires_review(self):
+        """Medium or higher risk level always requires review."""
+        from app.workflows.decision_guard import requires_human_review
+
+        assert requires_human_review({"risk_level": "medium"}, "any_action") is True
+        assert requires_human_review({"risk_level": "high"}, "any_action") is True
+        assert requires_human_review({"risk_level": "critical"}, "any_action") is True
+
+    def test_low_risk_allows_normal_actions(self):
+        """Low risk allows normal actions."""
+        from app.workflows.decision_guard import requires_human_review
+
+        assert requires_human_review({"risk_level": "low"}, "noop") is False
+        assert requires_human_review({"risk_level": "none"}, "noop") is False
+
+    def test_low_risk_blocks_high_risk_actions(self):
+        """Low risk still blocks inherently high-risk actions."""
+        from app.workflows.decision_guard import requires_human_review
+
+        assert requires_human_review({"risk_level": "low"}, "send_email") is True
+        assert requires_human_review({"risk_level": "low"}, "delete_entity") is True
+        assert requires_human_review({"risk_level": "low"}, "execute_payment") is True
+
+    def test_reviewed_actions_list_checked(self):
+        """Specific actions in reviewed_actions list require review."""
+        from app.workflows.decision_guard import requires_human_review
+
+        metadata = {"risk_level": "none", "reviewed_actions": ["custom_action"]}
+        assert requires_human_review(metadata, "custom_action") is True
+        assert requires_human_review(metadata, "other_action") is False
+
+    @pytest.mark.asyncio
+    async def test_check_decision_guard_allows_low_risk(self):
+        """check_decision_guard allows low-risk actions without metadata."""
+        from app.workflows.decision_guard import check_decision_guard
+
+        result = await check_decision_guard(
+            db=MagicMock(),
+            tenant_id=uuid.uuid4(),
+            instance_id=uuid.uuid4(),
+            step_config={},
+            action="noop",
+            ai_use_case_metadata=None,
+        )
+        assert result["allowed"] is True
+        assert result["requires_approval"] is False
+
+    @pytest.mark.asyncio
+    async def test_check_decision_guard_blocks_high_risk(self):
+        """check_decision_guard blocks high-risk actions."""
+        from app.workflows.decision_guard import check_decision_guard
+
+        result = await check_decision_guard(
+            db=MagicMock(),
+            tenant_id=uuid.uuid4(),
+            instance_id=uuid.uuid4(),
+            step_config={"ai_use_case_metadata": {"risk_level": "high"}},
+            action="send_email",
+            ai_use_case_metadata=None,
+        )
+        assert result["allowed"] is False
+        assert result["requires_approval"] is True
+        assert "send_email" in result["reason"]
+
+
+# ─── G-WORK: Workflow Workstream ─────────────────────────────────────────────
+
+
+class TestWorkflowWorkstream:
+    """Test the workflow workstream integration (G-WORK)."""
+
+    @pytest.mark.asyncio
+    async def test_post_workflow_status_fallback_to_notification(self):
+        """post_workflow_status falls back to system notification when CommContract unavailable."""
+        from app.workflows.workstream import post_workflow_status
+
+        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
+            result = await post_workflow_status(
+                db=MagicMock(),
+                tenant_id=uuid.uuid4(),
+                instance_id=uuid.uuid4(),
+                workflow_name="Test Workflow",
+                status="in_progress",
+                user_id=uuid.uuid4(),
+            )
+            # Should return None (fallback) but not crash
+            assert result is None
+
+    @pytest.mark.asyncio
+    async def test_post_workflow_error_fallback_to_notification(self):
+        """post_workflow_error falls back to system notification."""
+        from app.workflows.workstream import post_workflow_error
+
+        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
+            result = await post_workflow_error(
+                db=MagicMock(),
+                tenant_id=uuid.uuid4(),
+                instance_id=uuid.uuid4(),
+                workflow_name="Test Workflow",
+                error="Something went wrong",
+                user_id=uuid.uuid4(),
+            )
+            assert result is None
+
+    @pytest.mark.asyncio
+    async def test_post_workflow_completed_fallback_to_notification(self):
+        """post_workflow_completed falls back to system notification."""
+        from app.workflows.workstream import post_workflow_completed
+
+        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
+            result = await post_workflow_completed(
+                db=MagicMock(),
+                tenant_id=uuid.uuid4(),
+                instance_id=uuid.uuid4(),
+                workflow_name="Test Workflow",
+                result={"output": "done"},
+                user_id=uuid.uuid4(),
+            )
+            assert result is None
+
+    @pytest.mark.asyncio
+    async def test_post_workflow_handoff_fallback_to_notification(self):
+        """post_workflow_handoff falls back to system notification."""
+        from app.workflows.workstream import post_workflow_handoff
+
+        with patch("app.core.notifications.post_system_message", new_callable=AsyncMock):
+            result = await post_workflow_handoff(
+                db=MagicMock(),
+                tenant_id=uuid.uuid4(),
+                instance_id=uuid.uuid4(),
+                workflow_name="Test Workflow",
+                handoff_type="review_needed",
+                assignee_id=uuid.uuid4(),
+                description="Please review",
+                user_id=uuid.uuid4(),
+            )
+            assert result is None