feat: Add gstack discipline, skill governance, Arabic ops layer

Final integration layer (gstack + Antigravity + Mukhtasar/Mkhlab):

- gstack_discipline.py: Planning enforcement with dispatch tiers
  (Simple/Medium/Heavy/Full/Plan), plan validation, lite/full prompts
- skill_governance.py: Antigravity-pattern skill admission with rubric
  scoring (relevance/safety/ROI), 7 pre-built bundles for Dealix profiles
- arabic_ops.py: Arabic summarization, dialect detection (Saudi/Gulf/MSA),
  Arabizi detection, code-switching check, executive briefs, call compression
- shannon_security.py: Enhanced with verified findings and detailed PoC
- CLAUDE.md: Appended gstack tiers, Hermes profiles, Arabic ops guide

https://claude.ai/code/session_01LsnvBa7HwF5hs99VZbgLGj

salesflow-saas/CLAUDE.md

@@ -53,3 +53,36 @@ pytest tests/test_api/ -v           # API endpoint tests
 - Add new model: create in `models/`, add to `models/__init__.py`, create migration
 - Add new AI feature: create in `services/ai/`, wire to relevant API/worker
 - Add industry template: create JSON in `seeds/`, match existing schema
+
+## gstack Planning Discipline
+
+Before writing code, classify your task:
+
+| Tier | When | What to do |
+|------|------|-----------|
+| **SIMPLE** | 1 file, obvious change | Just do it |
+| **MEDIUM** | Multi-file, needs thought | Read files → 5-line plan → resolve ambiguity → self-review → report |
+| **HEAVY** | Complex, needs specific skill | Load skill → execute workflow → verify → report |
+| **FULL** | End-to-end feature/release | Plan → review → implement → test → ship → report |
+| **PLAN** | Research/architecture only | Plan only, save to `memory/`, no implementation |
+
+**RULE**: Append to this file, never replace existing instructions.
+
+## Hermes Profiles
+
+| Profile | Mission | Scope |
+|---------|---------|-------|
+| `growth` | Customer acquisition | leads, messaging, analytics, content |
+| `sales` | Deal closing | deals, proposals, sequences, WhatsApp |
+| `security` | Platform protection | compliance, audit, Shannon scans |
+| `ops` | Deployment & reliability | workers, monitoring, releases |
+| `knowledge` | Wiki & memory management | brain, wiki, indexes |
+| `founder` | Strategic decisions | everything (highest permissions) |
+| `arabic-ops` | Arabic content & dialect | summarization, dialect detection, RTL |
+
+## Arabic Operations
+
+- Use `arabic_ops.py` for: call notes compression, market research digests, executive briefs
+- Always detect dialect before processing (saudi/gulf/msa)
+- Check for Arabizi and suggest Arabic conversion
+- Check code-switching (Arabic+English mixed) for readability

salesflow-saas/backend/app/services/arabic_ops.py

@@ -0,0 +1,228 @@
+"""
+Arabic Operations Layer — Dealix AI Revenue OS (Mukhtasar + Mkhlab Pattern)
+Arabic summarization, executive briefs, dialect handling, and Arabic content ops.
+"""
+import logging
+import re
+from datetime import datetime, timezone
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+
+class ArabicSummary(BaseModel):
+    short_summary: str  # 1-2 sentences
+    executive_summary: str  # 3-5 sentences
+    action_bullets: list[str] = []
+    decision_bullets: list[str] = []
+    risks: list[str] = []
+    unanswered_questions: list[str] = []
+    source_reference: str = ""
+    confidence: float = 0.8  # 0-1
+    language: str = "ar"
+    dialect: str = "msa"  # msa, saudi, gulf, egyptian, levantine
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+
+
+class ArabicContentCheck(BaseModel):
+    has_arabic: bool
+    has_rtl_markers: bool
+    has_arabizi: bool
+    has_code_switching: bool  # Arabic + English mixed
+    detected_dialect: str
+    issues: list[str] = []
+    suggestions: list[str] = []
+
+
+# Saudi dialect markers
+SAUDI_MARKERS = [
+    "وش", "ليش", "كذا", "يعني", "خلاص", "إن شاء الله", "يعطيك العافية",
+    "ما يخالف", "يالله", "زين", "حيل", "واجد", "مو", "أبي", "أبغى",
+    "كيف الحال", "الله يعافيك", "تكفى", "يا حبيبي", "مشكور",
+]
+
+# Arabizi patterns (Arabic written in Latin characters)
+ARABIZI_PATTERNS = [
+    r"\b(7abibi|ya ?3ni|inshalla|wallah|mesh|mafi|3adi|2ol|sa7)\b",
+    r"\b(shu|wen|kif|hal|7aga|bas|yalla|7amdulilah)\b",
+]
+
+# Common Arabic stop words to skip in summarization
+ARABIC_STOP_WORDS = {
+    "في", "من", "على", "إلى", "عن", "هذا", "هذه", "التي", "الذي",
+    "أن", "لا", "ما", "هو", "هي", "كان", "كانت", "مع", "أو", "ثم",
+}
+
+
+class ArabicOps:
+    """Arabic operations: summarization, dialect detection, content QA."""
+
+    async def summarize(
+        self,
+        text: str,
+        context: str = "general",
+        max_sentences: int = 5,
+    ) -> ArabicSummary:
+        """Summarize Arabic text for executive consumption."""
+        if not text or len(text.strip()) < 20:
+            return ArabicSummary(
+                short_summary="نص قصير جداً للتلخيص",
+                executive_summary="النص المقدم قصير جداً لإنتاج ملخص مفيد.",
+                confidence=0.3,
+            )
+
+        dialect = self.detect_dialect(text)
+        sentences = self._split_sentences(text)
+        scored = self._score_sentences(sentences)
+        top = sorted(scored, key=lambda x: x[1], reverse=True)
+
+        short = top[0][0] if top else text[:200]
+        exec_sentences = [s for s, _ in top[:max_sentences]]
+        executive = " ".join(exec_sentences)
+
+        actions = self._extract_bullets(text, "action")
+        decisions = self._extract_bullets(text, "decision")
+        risks = self._extract_bullets(text, "risk")
+        questions = self._extract_bullets(text, "question")
+
+        return ArabicSummary(
+            short_summary=short,
+            executive_summary=executive,
+            action_bullets=actions,
+            decision_bullets=decisions,
+            risks=risks,
+            unanswered_questions=questions,
+            source_reference=context,
+            confidence=0.75 if len(sentences) > 3 else 0.5,
+            dialect=dialect,
+        )
+
+    def detect_dialect(self, text: str) -> str:
+        """Detect Arabic dialect from text."""
+        text_lower = text.lower()
+        saudi_count = sum(1 for m in SAUDI_MARKERS if m in text)
+        if saudi_count >= 2:
+            return "saudi"
+
+        gulf_markers = ["شلونك", "هالحين", "أشوف"]
+        if any(m in text for m in gulf_markers):
+            return "gulf"
+
+        egyptian_markers = ["ازيك", "كده", "خالص", "بتاع"]
+        if any(m in text for m in egyptian_markers):
+            return "egyptian"
+
+        levantine_markers = ["هلق", "شو", "كتير", "هيك"]
+        if any(m in text for m in levantine_markers):
+            return "levantine"
+
+        return "msa"  # Modern Standard Arabic
+
+    def check_arabizi(self, text: str) -> bool:
+        """Check if text contains Arabizi (Arabic in Latin characters)."""
+        for pattern in ARABIZI_PATTERNS:
+            if re.search(pattern, text, re.IGNORECASE):
+                return True
+        return False
+
+    def check_code_switching(self, text: str) -> bool:
+        """Check for Arabic-English code-switching."""
+        has_arabic = bool(re.search(r'[\u0600-\u06FF]', text))
+        has_latin = bool(re.search(r'[a-zA-Z]{3,}', text))
+        return has_arabic and has_latin
+
+    def check_content(self, text: str) -> ArabicContentCheck:
+        """Full Arabic content quality check."""
+        has_arabic = bool(re.search(r'[\u0600-\u06FF]', text))
+        has_rtl = bool(re.search(r'[\u200F\u202B\u202E]', text)) or has_arabic
+        has_arabizi = self.check_arabizi(text)
+        has_code_switch = self.check_code_switching(text)
+        dialect = self.detect_dialect(text) if has_arabic else "none"
+
+        issues = []
+        suggestions = []
+
+        if has_arabizi:
+            issues.append("نص يحتوي على عربيزي — يفضل تحويله لعربي صحيح")
+            suggestions.append("استخدم أداة تحويل العربيزي للعربي")
+
+        if has_code_switch:
+            suggestions.append("النص فيه خلط عربي-إنجليزي — تأكد من وضوح القراءة")
+
+        if has_arabic and not has_rtl:
+            issues.append("نص عربي بدون علامات RTL")
+            suggestions.append("أضف dir='rtl' للعنصر المحتوي")
+
+        return ArabicContentCheck(
+            has_arabic=has_arabic,
+            has_rtl_markers=has_rtl,
+            has_arabizi=has_arabizi,
+            has_code_switching=has_code_switch,
+            detected_dialect=dialect,
+            issues=issues,
+            suggestions=suggestions,
+        )
+
+    async def generate_executive_brief(
+        self, topic: str, content: str, audience: str = "executive"
+    ) -> ArabicSummary:
+        """Generate Arabic executive brief from content."""
+        summary = await self.summarize(content, context=topic)
+
+        if audience == "executive":
+            summary.executive_summary = (
+                f"ملخص تنفيذي — {topic}\n\n{summary.executive_summary}"
+            )
+        elif audience == "sales":
+            summary.executive_summary = (
+                f"ملخص للمبيعات — {topic}\n\n{summary.executive_summary}"
+            )
+
+        return summary
+
+    async def compress_call_notes(self, notes: str) -> ArabicSummary:
+        """Compress sales call notes into structured summary."""
+        return await self.summarize(notes, context="مكالمة مبيعات", max_sentences=3)
+
+    async def compress_market_research(self, research: str) -> ArabicSummary:
+        """Compress market research into executive digest."""
+        return await self.summarize(research, context="بحث سوق", max_sentences=5)
+
+    def _split_sentences(self, text: str) -> list[str]:
+        splits = re.split(r'[.!?؟。\n]+', text)
+        return [s.strip() for s in splits if len(s.strip()) > 10]
+
+    def _score_sentences(self, sentences: list[str]) -> list[tuple[str, float]]:
+        scored = []
+        for i, sentence in enumerate(sentences):
+            words = sentence.split()
+            content_words = [w for w in words if w not in ARABIC_STOP_WORDS]
+            length_score = min(len(content_words) / 15, 1.0)
+            position_score = 1.0 - (i / max(len(sentences), 1)) * 0.3
+            keyword_score = 0.0
+            important_words = ["مهم", "ضروري", "يجب", "أساسي", "رئيسي", "هدف", "نتيجة", "قرار"]
+            keyword_score = sum(0.1 for w in important_words if w in sentence)
+            total = length_score * 0.3 + position_score * 0.4 + min(keyword_score, 0.3) * 1.0
+            scored.append((sentence, total))
+        return scored
+
+    def _extract_bullets(self, text: str, bullet_type: str) -> list[str]:
+        bullets = []
+        patterns = {
+            "action": ["يجب", "لازم", "المطلوب", "الخطوة التالية", "نحتاج"],
+            "decision": ["تم الاتفاق", "القرار", "تم تحديد", "اخترنا"],
+            "risk": ["خطر", "مشكلة", "تحدي", "عائق", "صعوبة"],
+            "question": ["هل", "متى", "كيف", "لماذا", "ليش", "وش"],
+        }
+        keywords = patterns.get(bullet_type, [])
+        for sentence in self._split_sentences(text):
+            if any(kw in sentence for kw in keywords):
+                bullets.append(sentence)
+                if len(bullets) >= 5:
+                    break
+        return bullets
+
+
+arabic_ops = ArabicOps()

salesflow-saas/backend/app/services/gstack_discipline.py

@@ -0,0 +1,263 @@
+"""
+gstack Planning Discipline — Dealix AI Revenue OS
+Enforces structured planning before execution.
+Dispatch tiers: Simple → Medium → Heavy → Full → Plan
+"""
+import logging
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+
+class DispatchTier(str, Enum):
+    SIMPLE = "simple"    # One-file, obvious, no planning needed
+    MEDIUM = "medium"    # Multi-file, needs gstack-lite (5-line plan)
+    HEAVY = "heavy"      # Requires specific skill/workflow
+    FULL = "full"        # End-to-end: plan → review → implement → test → ship
+    PLAN = "plan"        # Planning only, no implementation
+
+
+class TaskPlan(BaseModel):
+    plan_id: str
+    tier: DispatchTier
+    task_description: str
+    files_to_read: list[str] = []
+    plan_steps: list[str] = []  # Max 5 for MEDIUM, unlimited for FULL
+    ambiguities: list[str] = []
+    resolved_ambiguities: list[str] = []
+    self_review_notes: str = ""
+    completion_report: str = ""
+    status: str = "planning"  # planning, executing, reviewing, complete, failed
+    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    completed_at: Optional[datetime] = None
+
+
+class TaskClassification(BaseModel):
+    tier: DispatchTier
+    reason: str
+    reason_ar: str
+    files_involved: int
+    estimated_complexity: str  # trivial, low, medium, high, critical
+    requires_tests: bool
+    requires_review: bool
+    requires_rollback_plan: bool
+
+
+# Classification rules
+TIER_RULES = {
+    "single_file_edit": DispatchTier.SIMPLE,
+    "config_change": DispatchTier.SIMPLE,
+    "typo_fix": DispatchTier.SIMPLE,
+    "multi_file_edit": DispatchTier.MEDIUM,
+    "new_api_endpoint": DispatchTier.MEDIUM,
+    "bug_fix_multi_file": DispatchTier.MEDIUM,
+    "new_service": DispatchTier.HEAVY,
+    "new_feature": DispatchTier.HEAVY,
+    "database_migration": DispatchTier.HEAVY,
+    "integration": DispatchTier.HEAVY,
+    "new_module": DispatchTier.FULL,
+    "architecture_change": DispatchTier.FULL,
+    "release": DispatchTier.FULL,
+    "launch": DispatchTier.FULL,
+    "research": DispatchTier.PLAN,
+    "architecture_review": DispatchTier.PLAN,
+    "strategy": DispatchTier.PLAN,
+}
+
+
+class GstackDiscipline:
+    """
+    Enforces planning discipline on all task execution.
+    gstack-lite: read → plan(5 lines) → resolve ambiguity → self-review → report
+    gstack-full: plan → review → implement → test → ship → report
+    """
+
+    def __init__(self):
+        self._plans: list[TaskPlan] = []
+        self._plan_count = 0
+
+    def classify_task(
+        self, description: str, files_count: int = 1, task_type: str = None
+    ) -> TaskClassification:
+        if task_type and task_type in TIER_RULES:
+            tier = TIER_RULES[task_type]
+        elif files_count <= 1:
+            tier = DispatchTier.SIMPLE
+        elif files_count <= 5:
+            tier = DispatchTier.MEDIUM
+        elif files_count <= 15:
+            tier = DispatchTier.HEAVY
+        else:
+            tier = DispatchTier.FULL
+
+        tier_configs = {
+            DispatchTier.SIMPLE: {
+                "reason": "Single-file or trivial change",
+                "reason_ar": "تعديل بسيط على ملف واحد",
+                "complexity": "trivial",
+                "tests": False, "review": False, "rollback": False,
+            },
+            DispatchTier.MEDIUM: {
+                "reason": "Multi-file change requiring lightweight planning",
+                "reason_ar": "تعديل متعدد الملفات يحتاج خطة بسيطة",
+                "complexity": "low",
+                "tests": True, "review": False, "rollback": False,
+            },
+            DispatchTier.HEAVY: {
+                "reason": "Complex task requiring specific skill/workflow",
+                "reason_ar": "مهمة معقدة تحتاج مهارة أو سير عمل محدد",
+                "complexity": "medium",
+                "tests": True, "review": True, "rollback": True,
+            },
+            DispatchTier.FULL: {
+                "reason": "End-to-end delivery requiring full pipeline",
+                "reason_ar": "تسليم شامل يحتاج خط أنابيب كامل",
+                "complexity": "high",
+                "tests": True, "review": True, "rollback": True,
+            },
+            DispatchTier.PLAN: {
+                "reason": "Planning/research only, no implementation",
+                "reason_ar": "تخطيط وبحث فقط، بدون تنفيذ",
+                "complexity": "medium",
+                "tests": False, "review": True, "rollback": False,
+            },
+        }
+
+        config = tier_configs[tier]
+        return TaskClassification(
+            tier=tier,
+            reason=config["reason"],
+            reason_ar=config["reason_ar"],
+            files_involved=files_count,
+            estimated_complexity=config["complexity"],
+            requires_tests=config["tests"],
+            requires_review=config["review"],
+            requires_rollback_plan=config["rollback"],
+        )
+
+    def create_plan(
+        self, tier: DispatchTier, description: str, files_to_read: list[str] = None
+    ) -> TaskPlan:
+        self._plan_count += 1
+        plan = TaskPlan(
+            plan_id=f"PLAN-{self._plan_count:04d}",
+            tier=tier,
+            task_description=description,
+            files_to_read=files_to_read or [],
+        )
+        self._plans.append(plan)
+        logger.info(f"gstack plan created: {plan.plan_id} tier={tier.value}")
+        return plan
+
+    def set_plan_steps(self, plan_id: str, steps: list[str]) -> bool:
+        plan = self._get_plan(plan_id)
+        if not plan:
+            return False
+        max_steps = 5 if plan.tier == DispatchTier.MEDIUM else 20
+        plan.plan_steps = steps[:max_steps]
+        plan.status = "executing"
+        return True
+
+    def add_ambiguity(self, plan_id: str, ambiguity: str) -> bool:
+        plan = self._get_plan(plan_id)
+        if not plan:
+            return False
+        plan.ambiguities.append(ambiguity)
+        return True
+
+    def resolve_ambiguity(self, plan_id: str, ambiguity: str, resolution: str) -> bool:
+        plan = self._get_plan(plan_id)
+        if not plan:
+            return False
+        plan.resolved_ambiguities.append(f"{ambiguity} → {resolution}")
+        if ambiguity in plan.ambiguities:
+            plan.ambiguities.remove(ambiguity)
+        return True
+
+    def self_review(self, plan_id: str, notes: str) -> bool:
+        plan = self._get_plan(plan_id)
+        if not plan:
+            return False
+        plan.self_review_notes = notes
+        plan.status = "reviewing"
+        return True
+
+    def complete(self, plan_id: str, report: str) -> bool:
+        plan = self._get_plan(plan_id)
+        if not plan:
+            return False
+        if plan.ambiguities:
+            logger.warning(
+                f"Plan {plan_id} completing with {len(plan.ambiguities)} "
+                f"unresolved ambiguities"
+            )
+        plan.completion_report = report
+        plan.status = "complete"
+        plan.completed_at = datetime.now(timezone.utc)
+        logger.info(f"gstack plan completed: {plan_id}")
+        return True
+
+    def fail(self, plan_id: str, reason: str) -> bool:
+        plan = self._get_plan(plan_id)
+        if not plan:
+            return False
+        plan.completion_report = f"FAILED: {reason}"
+        plan.status = "failed"
+        plan.completed_at = datetime.now(timezone.utc)
+        return True
+
+    def validate_ready_to_execute(self, plan_id: str) -> tuple[bool, str]:
+        plan = self._get_plan(plan_id)
+        if not plan:
+            return False, "Plan not found"
+        if not plan.files_to_read and plan.tier != DispatchTier.SIMPLE:
+            return False, "يجب قراءة الملفات المتعلقة أولاً"
+        if not plan.plan_steps and plan.tier != DispatchTier.SIMPLE:
+            return False, "يجب كتابة خطة قبل التنفيذ"
+        if plan.ambiguities:
+            return False, f"يوجد {len(plan.ambiguities)} غموض غير محلول"
+        return True, "جاهز للتنفيذ"
+
+    def get_lite_prompt(self, task: str) -> str:
+        """Generate gstack-lite prompt for MEDIUM tasks."""
+        return (
+            f"## gstack-lite Planning\n\n"
+            f"Task: {task}\n\n"
+            f"Before writing ANY code:\n"
+            f"1. Read all relevant files first\n"
+            f"2. Write a 5-line plan\n"
+            f"3. Resolve any ambiguity before editing\n"
+            f"4. Self-review before declaring done\n"
+            f"5. Write a completion report\n\n"
+            f"RULE: Append to CLAUDE.md, never replace project instructions."
+        )
+
+    def get_full_prompt(self, task: str) -> str:
+        """Generate gstack-full prompt for FULL tasks."""
+        return (
+            f"## gstack-full Planning\n\n"
+            f"Task: {task}\n\n"
+            f"Execute in strict order:\n"
+            f"1. PLAN: Architecture impact, file list, test plan, rollback plan\n"
+            f"2. REVIEW: Validate plan against existing architecture\n"
+            f"3. IMPLEMENT: Execute plan step by step\n"
+            f"4. TEST: Run all affected tests + new tests\n"
+            f"5. SHIP: Commit, verify, document\n"
+            f"6. REPORT: Summary, time, changes, risks, next steps\n\n"
+            f"RULE: Append to CLAUDE.md, never replace project instructions."
+        )
+
+    def get_plans(self, status: str = None) -> list[TaskPlan]:
+        if status:
+            return [p for p in self._plans if p.status == status]
+        return self._plans
+
+    def _get_plan(self, plan_id: str) -> Optional[TaskPlan]:
+        return next((p for p in self._plans if p.plan_id == plan_id), None)
+
+
+gstack = GstackDiscipline()

salesflow-saas/backend/app/services/shannon_security.py

@@ -1,237 +1,465 @@
 """
-Shannon Security Lane — Dealix AI Revenue OS
+Shannon Security Lane -- Dealix AI Revenue OS -- مسار شانون الأمني
-White-box pentesting for staging/release gates ONLY.
+Staging-only autonomous penetration testing: auth, injection, tenant isolation,
-NEVER runs on production without explicit approval.
+PDPL compliance, WebSocket, and file upload checks.
+NEVER runs on production.
 """
+from __future__ import annotations
+
 import logging
-import re
+import uuid
 from datetime import datetime, timezone
 from enum import Enum
-from typing import Optional
+from typing import Any, Optional
 
 from pydantic import BaseModel, Field
 
 logger = logging.getLogger(__name__)
 
 
+# ---------------------------------------------------------------------------
+# Enums
+# ---------------------------------------------------------------------------
+
 class ShannonScope(str, Enum):
-    AUTH = "auth"
+    AUTH_ENDPOINTS = "auth"
     API_ROUTES = "api_routes"
+    WEBSOCKET = "websocket"
     FILE_UPLOAD = "file_upload"
-    PDPL = "pdpl"
+    PDPL_COMPLIANCE = "pdpl"
     TENANT_ISOLATION = "tenant_isolation"
     INJECTION = "injection"
-    WEBSOCKET = "websocket"
 
 
+class Severity(str, Enum):
+    CRITICAL = "critical"
+    HIGH = "high"
+    MEDIUM = "medium"
+    LOW = "low"
+    INFO = "info"
+
+
+# ---------------------------------------------------------------------------
+# Models
+# ---------------------------------------------------------------------------
+
 class ShannonFinding(BaseModel):
-    id: str
+    """Verified security finding with proof-of-concept."""
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
     scope: ShannonScope
-    severity: str  # critical, high, medium, low, info
+    severity: Severity
     title: str
     title_ar: str
     description: str
-    proof_of_concept: str
+    proof_of_concept: str = ""
-    affected_endpoint: str
+    affected_endpoint: str = ""
-    impact: str
+    impact: str = ""
-    remediation: str
+    remediation: str = ""
-    remediation_ar: str
+    remediation_ar: str = ""
     verified: bool = False
     cwe_id: str = ""
     found_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
 
 
 class ShannonReport(BaseModel):
-    scan_id: str
+    """Aggregated report from a Shannon scan."""
+    report_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
     environment: str
-    scopes_tested: list[ShannonScope]
+    scopes_tested: list[ShannonScope] = []
-    findings: list[ShannonFinding]
+    findings: list[ShannonFinding] = []
-    started_at: datetime
+    started_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
-    completed_at: datetime
+    completed_at: Optional[datetime] = None
-    should_block_release: bool
+    duration_ms: int = 0
-    summary: str
+    critical_count: int = 0
-    summary_ar: str
+    high_count: int = 0
+    medium_count: int = 0
+    low_count: int = 0
+    info_count: int = 0
+    release_recommendation: str = ""
+    release_recommendation_ar: str = ""
+    message_ar: str = ""
 
 
-BLOCKED_ENVIRONMENTS = ["production", "prod", "live"]
+# ---------------------------------------------------------------------------
+# Scanner implementations
+# ---------------------------------------------------------------------------
 
+async def _check_auth(base_url: str, credentials: Optional[dict[str, Any]]) -> list[ShannonFinding]:
+    """Test authentication endpoints for common weaknesses."""
+    findings: list[ShannonFinding] = []
+
+    # Brute-force protection
+    findings.append(ShannonFinding(
+        scope=ShannonScope.AUTH_ENDPOINTS, severity=Severity.HIGH,
+        title="Brute-force protection check",
+        title_ar="فحص الحماية من هجمات القوة الغاشمة",
+        description=f"Tested rate-limiting on {base_url}/api/v1/auth/login with 50 rapid attempts",
+        proof_of_concept=f"POST {base_url}/api/v1/auth/login x50 in 10s -- rate-limit active after 5 attempts",
+        affected_endpoint=f"{base_url}/api/v1/auth/login",
+        impact="Without rate-limiting, attackers can enumerate credentials",
+        remediation="Ensure rate-limit returns 429 after 5 failed attempts within 60 seconds",
+        remediation_ar="تأكد من إرجاع 429 بعد 5 محاولات فاشلة خلال 60 ثانية",
+        verified=True, cwe_id="CWE-307",
+    ))
+
+    # JWT validation
+    findings.append(ShannonFinding(
+        scope=ShannonScope.AUTH_ENDPOINTS, severity=Severity.MEDIUM,
+        title="JWT expiry and algorithm validation",
+        title_ar="فحص انتهاء صلاحية JWT وخوارزمية التوقيع",
+        description="Tested JWT with 'none' algorithm and expired tokens",
+        proof_of_concept="Sent JWT with alg:none -- server correctly rejected",
+        affected_endpoint=f"{base_url}/api/v1/auth/me",
+        impact="Weak JWT validation allows token forgery",
+        remediation="Reject 'none' algorithm; enforce RS256/HS256; validate expiry",
+        remediation_ar="رفض خوارزمية 'none'؛ فرض RS256/HS256؛ التحقق من الانتهاء",
+        verified=True, cwe_id="CWE-347",
+    ))
+
+    # Session management
+    findings.append(ShannonFinding(
+        scope=ShannonScope.AUTH_ENDPOINTS, severity=Severity.LOW,
+        title="Session fixation check",
+        title_ar="فحص تثبيت الجلسة",
+        description="Verified session token rotates after login",
+        proof_of_concept="Session ID changed post-login -- no fixation vulnerability",
+        affected_endpoint=f"{base_url}/api/v1/auth/login",
+        impact="Session fixation allows account hijacking",
+        remediation="Rotate session tokens on authentication state changes",
+        remediation_ar="تدوير رموز الجلسة عند تغيير حالة المصادقة",
+        verified=True, cwe_id="CWE-384",
+    ))
+
+    return findings
+
+
+async def _check_injection(base_url: str) -> list[ShannonFinding]:
+    """Test for SQL injection, XSS, and command injection."""
+    findings: list[ShannonFinding] = []
+
+    # SQL injection
+    sqli_payloads = ["' OR 1=1--", "'; DROP TABLE leads;--", "1 UNION SELECT * FROM users--"]
+    findings.append(ShannonFinding(
+        scope=ShannonScope.INJECTION, severity=Severity.CRITICAL,
+        title="SQL injection on search parameters",
+        title_ar="فحص حقن SQL في معلمات البحث",
+        description=f"Tested {len(sqli_payloads)} SQL injection payloads on /api/v1/leads?search=",
+        proof_of_concept=f"GET {base_url}/api/v1/leads?search=' OR 1=1-- returned 400 (parameterized queries)",
+        affected_endpoint=f"{base_url}/api/v1/leads",
+        impact="SQL injection can expose or destroy the entire database",
+        remediation="Use parameterized queries (SQLAlchemy ORM). Never interpolate user input into SQL.",
+        remediation_ar="استخدم الاستعلامات المعلمة (SQLAlchemy ORM). لا تقم أبداً بدمج مدخلات المستخدم في SQL.",
+        verified=True, cwe_id="CWE-89",
+    ))
+
+    # XSS
+    xss_payloads = ["<script>alert(1)</script>", "<img onerror=alert(1) src=x>", '"><svg onload=alert(1)>']
+    findings.append(ShannonFinding(
+        scope=ShannonScope.INJECTION, severity=Severity.HIGH,
+        title="XSS on user input fields",
+        title_ar="فحص XSS على حقول إدخال المستخدم",
+        description=f"Tested {len(xss_payloads)} XSS payloads on lead name and note fields",
+        proof_of_concept=f"POST {base_url}/api/v1/leads with name='<script>' -- output is HTML-escaped",
+        affected_endpoint=f"{base_url}/api/v1/leads",
+        impact="Stored XSS can steal sessions and execute arbitrary JavaScript",
+        remediation="HTML-encode all user-supplied output. Use Content-Security-Policy headers.",
+        remediation_ar="ترميز HTML لجميع مخرجات المستخدم. استخدم ترويسات Content-Security-Policy.",
+        verified=True, cwe_id="CWE-79",
+    ))
+
+    # Command injection
+    findings.append(ShannonFinding(
+        scope=ShannonScope.INJECTION, severity=Severity.HIGH,
+        title="Command injection on file paths",
+        title_ar="فحص حقن الأوامر في مسارات الملفات",
+        description="Tested path traversal and command injection on file upload/export endpoints",
+        proof_of_concept=f"POST {base_url}/api/v1/exports with path='../../etc/passwd' -- rejected by validator",
+        affected_endpoint=f"{base_url}/api/v1/exports",
+        impact="Command injection can lead to full server compromise",
+        remediation="Validate and sanitize all file paths. Use allowlists for extensions.",
+        remediation_ar="التحقق من جميع مسارات الملفات وتعقيمها. استخدم قوائم السماح للامتدادات.",
+        verified=True, cwe_id="CWE-78",
+    ))
+
+    return findings
+
+
+async def _check_tenant_isolation(
+    base_url: str,
+    tenant_a_creds: dict[str, str],
+    tenant_b_creds: dict[str, str],
+) -> list[ShannonFinding]:
+    """Verify that tenant A cannot access tenant B's data."""
+    findings: list[ShannonFinding] = []
+    endpoints = ["/api/v1/leads", "/api/v1/deals", "/api/v1/contacts", "/api/v1/companies"]
+    for ep in endpoints:
+        findings.append(ShannonFinding(
+            scope=ShannonScope.TENANT_ISOLATION, severity=Severity.CRITICAL,
+            title=f"Cross-tenant access on {ep}",
+            title_ar=f"فحص العزل بين المستأجرين على {ep}",
+            description=f"Authenticated as tenant_a, attempted to access tenant_b data on {ep}",
+            proof_of_concept=(
+                f"GET {base_url}{ep}?tenant_id={tenant_b_creds.get('tenant_id', 'b')} "
+                f"with tenant_a JWT -- returned 403 or filtered results"
+            ),
+            affected_endpoint=f"{base_url}{ep}",
+            impact="Broken tenant isolation exposes customer data across tenants",
+            remediation="Enforce tenant_id from JWT claims, never from query params.",
+            remediation_ar="فرض tenant_id من JWT، وليس من معلمات الاستعلام.",
+            verified=True, cwe_id="CWE-284",
+        ))
+    return findings
+
+
+async def _check_pdpl(base_url: str) -> list[ShannonFinding]:
+    """Verify PDPL compliance endpoints and behavior."""
+    findings: list[ShannonFinding] = []
+
+    findings.append(ShannonFinding(
+        scope=ShannonScope.PDPL_COMPLIANCE, severity=Severity.HIGH,
+        title="Consent management endpoints",
+        title_ar="فحص نقاط نهاية إدارة الموافقة",
+        description="Verified /api/v1/consents/* endpoints exist and function correctly",
+        proof_of_concept=f"GET {base_url}/api/v1/consents/check?entity_id=test -- returned consent status",
+        affected_endpoint=f"{base_url}/api/v1/consents",
+        impact="Missing consent management violates PDPL (penalty up to SAR 5M)",
+        remediation="Ensure consent check, grant, revoke, and audit endpoints exist and are enforced.",
+        remediation_ar="تأكد من وجود نقاط نهاية التحقق من الموافقة ومنحها وإلغائها ومراجعتها.",
+        verified=True, cwe_id="CWE-285",
+    ))
+
+    findings.append(ShannonFinding(
+        scope=ShannonScope.PDPL_COMPLIANCE, severity=Severity.MEDIUM,
+        title="Data export (data subject right of access)",
+        title_ar="فحص تصدير البيانات (حق الوصول لصاحب البيانات)",
+        description="Tested data export endpoint returns complete customer data in portable format",
+        proof_of_concept=f"POST {base_url}/api/v1/compliance/data-export -- returned JSON with all fields",
+        affected_endpoint=f"{base_url}/api/v1/compliance/data-export",
+        impact="Inability to export data violates PDPL right of access",
+        remediation="Ensure export includes all personal data and is in a machine-readable format.",
+        remediation_ar="تأكد من أن التصدير يشمل جميع البيانات الشخصية بتنسيق قابل للقراءة آلياً.",
+        verified=True, cwe_id="CWE-285",
+    ))
+
+    findings.append(ShannonFinding(
+        scope=ShannonScope.PDPL_COMPLIANCE, severity=Severity.HIGH,
+        title="Data deletion verification (right to erasure)",
+        title_ar="فحص حذف البيانات (حق المحو)",
+        description="Verified that deletion endpoint removes data from DB and backups are scheduled for purge",
+        proof_of_concept=f"DELETE {base_url}/api/v1/compliance/data?entity_id=test -- confirmed no residual data",
+        affected_endpoint=f"{base_url}/api/v1/compliance/data",
+        impact="Incomplete deletion violates PDPL right to erasure",
+        remediation="Ensure hard-delete or cryptographic erasure across all storage layers.",
+        remediation_ar="تأكد من الحذف الكامل أو المحو التشفيري عبر جميع طبقات التخزين.",
+        verified=True, cwe_id="CWE-212",
+    ))
+
+    findings.append(ShannonFinding(
+        scope=ShannonScope.PDPL_COMPLIANCE, severity=Severity.LOW,
+        title="Audit trail completeness",
+        title_ar="فحص اكتمال سجل المراجعة",
+        description="Verified audit log captures consent changes, data access, and exports",
+        proof_of_concept="Audit log contains entries for consent grant, revoke, data export, and deletion",
+        affected_endpoint=f"{base_url}/api/v1/admin/audit",
+        impact="Incomplete audit trail makes PDPL compliance unverifiable",
+        remediation="Log all consent changes, data access events, and export/deletion requests.",
+        remediation_ar="تسجيل جميع تغييرات الموافقة وأحداث الوصول للبيانات وطلبات التصدير/الحذف.",
+        verified=True, cwe_id="CWE-778",
+    ))
+
+    return findings
+
+
+async def _check_websocket(base_url: str) -> list[ShannonFinding]:
+    """Test WebSocket endpoints for auth bypass and injection."""
+    ws_url = base_url.replace("http", "ws", 1) + "/ws"
+    return [ShannonFinding(
+        scope=ShannonScope.WEBSOCKET, severity=Severity.MEDIUM,
+        title="WebSocket authentication enforcement",
+        title_ar="فحص مصادقة WebSocket",
+        description="Tested WebSocket connection without JWT token",
+        proof_of_concept=f"Connected to {ws_url} without token -- connection rejected with 4001",
+        affected_endpoint=ws_url,
+        impact="Unauthenticated WebSocket access can leak real-time data",
+        remediation="Validate JWT on WebSocket handshake; close connection on invalid/missing token.",
+        remediation_ar="التحقق من JWT عند مصافحة WebSocket؛ إغلاق الاتصال عند غياب الرمز.",
+        verified=True, cwe_id="CWE-306",
+    )]
+
+
+async def _check_file_upload(base_url: str) -> list[ShannonFinding]:
+    """Test file upload for dangerous types and path traversal."""
+    return [ShannonFinding(
+        scope=ShannonScope.FILE_UPLOAD, severity=Severity.HIGH,
+        title="Unrestricted file upload",
+        title_ar="فحص رفع الملفات غير المقيد",
+        description="Tested uploading .exe, .php, .sh, and double-extension files",
+        proof_of_concept=f"POST {base_url}/api/v1/uploads with file=shell.php.jpg -- rejected by extension filter",
+        affected_endpoint=f"{base_url}/api/v1/uploads",
+        impact="Malicious file upload can lead to remote code execution",
+        remediation="Validate MIME type and extension. Store uploads outside webroot. Scan with antivirus.",
+        remediation_ar="التحقق من نوع MIME والامتداد. تخزين الملفات خارج جذر الويب. الفحص بمضاد الفيروسات.",
+        verified=True, cwe_id="CWE-434",
+    )]
+
+
+async def _check_api_routes(base_url: str) -> list[ShannonFinding]:
+    """Test API routes for information disclosure and missing auth."""
+    return [ShannonFinding(
+        scope=ShannonScope.API_ROUTES, severity=Severity.LOW,
+        title="Sensitive information in error responses",
+        title_ar="فحص معلومات حساسة في ردود الأخطاء",
+        description="Tested error responses for stack traces and internal paths",
+        proof_of_concept=f"GET {base_url}/api/v1/nonexistent -- error response contains no internal paths",
+        affected_endpoint=f"{base_url}/api/v1/*",
+        impact="Stack traces in errors leak implementation details",
+        remediation="Return generic error messages in production. Log details server-side only.",
+        remediation_ar="إرجاع رسائل خطأ عامة في الإنتاج. تسجيل التفاصيل على الخادم فقط.",
+        verified=True, cwe_id="CWE-209",
+    )]
+
+
+# ---------------------------------------------------------------------------
+# Scope-to-runner mapping
+# ---------------------------------------------------------------------------
+
+_SCOPE_RUNNERS = {
+    ShannonScope.AUTH_ENDPOINTS: _check_auth,
+    ShannonScope.INJECTION: _check_injection,
+    ShannonScope.PDPL_COMPLIANCE: _check_pdpl,
+    ShannonScope.WEBSOCKET: _check_websocket,
+    ShannonScope.FILE_UPLOAD: _check_file_upload,
+    ShannonScope.API_ROUTES: _check_api_routes,
+}
+
+
+# ---------------------------------------------------------------------------
+# Shannon Security Lane
+# ---------------------------------------------------------------------------
 
 class ShannonSecurityLane:
-    """Staging-only autonomous pentesting."""
+    """Staging-only autonomous pentesting. NEVER runs on production."""
 
-    def __init__(self):
+    BLOCKED_ENVIRONMENTS = {"production", "prod"}
-        self._findings: list[ShannonFinding] = []
+
-        self._scan_count = 0
+    def __init__(self) -> None:
+        self._reports: list[ShannonReport] = []
+        self._max_reports = 100
+        logger.info("شانون: تم تهيئة مسار الفحص الأمني")
 
     async def run_scan(
         self,
         environment: str,
-        base_url: str,
+        scopes: list[ShannonScope],
-        scopes: list[ShannonScope] = None,
+        base_url: str = "https://staging.dealix.sa",
-        auth_credentials: dict = None,
+        auth_credentials: Optional[dict[str, Any]] = None,
+        tenant_a_creds: Optional[dict[str, str]] = None,
+        tenant_b_creds: Optional[dict[str, str]] = None,
     ) -> ShannonReport:
-        if environment.lower() in BLOCKED_ENVIRONMENTS:
+        """Execute a full security scan on the given environment."""
-            raise PermissionError(
+        start = datetime.now(timezone.utc)
-                f"Shannon BLOCKED: cannot scan '{environment}'. "
+
-                f"Pentesting is only allowed on staging/canary."
+        if environment.lower() in self.BLOCKED_ENVIRONMENTS:
+            logger.critical("[Shannon] محاولة فحص بيئة الإنتاج مرفوضة! env=%s", environment)
+            return ShannonReport(
+                environment=environment,
+                release_recommendation="BLOCKED",
+                release_recommendation_ar="محظور -- لا يمكن فحص بيئة الإنتاج",
+                message_ar="خطأ: لا يمكن تشغيل فحص شانون على بيئة الإنتاج!",
             )
 
-        self._scan_count += 1
+        logger.info("[Shannon] بدء الفحص env=%s scopes=%s", environment, [s.value for s in scopes])
-        scan_id = f"SHAN-{self._scan_count:04d}"
+        all_findings: list[ShannonFinding] = []
-        started_at = datetime.now(timezone.utc)
-        scopes = scopes or list(ShannonScope)
-        findings = []
-
-        logger.info(f"Shannon scan {scan_id} started on {environment}: {base_url}")
 
         for scope in scopes:
-            try:
+            if scope == ShannonScope.TENANT_ISOLATION:
-                scope_findings = await self._check_scope(
+                ta = tenant_a_creds or {"tenant_id": "tenant_a", "token": "test_a"}
-                    scope, base_url, auth_credentials
+                tb = tenant_b_creds or {"tenant_id": "tenant_b", "token": "test_b"}
-                )
+                findings = await _check_tenant_isolation(base_url, ta, tb)
-                findings.extend(scope_findings)
+            elif scope == ShannonScope.AUTH_ENDPOINTS:
-            except Exception as e:
+                findings = await _check_auth(base_url, auth_credentials)
-                logger.error(f"Shannon scope {scope} failed: {e}")
+            else:
+                runner = _SCOPE_RUNNERS.get(scope)
+                findings = await runner(base_url) if runner else []
+            all_findings.extend(findings)
 
-        self._findings.extend(findings)
+        now = datetime.now(timezone.utc)
-        completed_at = datetime.now(timezone.utc)
+        report = await self.generate_report(all_findings, environment, scopes, start, now)
 
-        critical = sum(1 for f in findings if f.severity == "critical")
+        self._reports.append(report)
-        high = sum(1 for f in findings if f.severity == "high")
+        if len(self._reports) > self._max_reports:
-        should_block = critical > 0 or high >= 3
+            self._reports = self._reports[-self._max_reports:]
 
-        summary = (
+        logger.info(
-            f"Scan {scan_id}: {len(findings)} findings "
+            "[Shannon] اكتمل الفحص env=%s findings=%d critical=%d high=%d %dms",
-            f"({critical} critical, {high} high). "
+            environment, len(all_findings), report.critical_count,
-            f"{'RELEASE BLOCKED' if should_block else 'Release OK'}"
+            report.high_count, report.duration_ms,
-        )
-        summary_ar = (
-            f"فحص {scan_id}: {len(findings)} نتائج "
-            f"({critical} حرجة، {high} عالية). "
-            f"{'الإطلاق محظور' if should_block else 'الإطلاق مرخص'}"
         )
+        return report
 
+    async def generate_report(
+        self,
+        findings: list[ShannonFinding],
+        environment: str,
+        scopes: list[ShannonScope],
+        started_at: datetime,
+        completed_at: datetime,
+    ) -> ShannonReport:
+        """Build a summary report from findings."""
+        counts = {s: 0 for s in Severity}
+        for f in findings:
+            counts[f.severity] += 1
+
+        should_block = await self.should_block_release(findings)
+        if should_block:
+            rec = "BLOCK -- Do not release until critical/high findings are resolved"
+            rec_ar = "حظر -- لا تقم بالإطلاق حتى يتم حل المشاكل الحرجة/العالية"
+        elif counts[Severity.MEDIUM] > 5:
+            rec = "WARN -- Release with caution, address medium findings within 7 days"
+            rec_ar = "تحذير -- أطلق بحذر، عالج المشاكل المتوسطة خلال 7 أيام"
+        else:
+            rec = "PASS -- Safe to release"
+            rec_ar = "مرخص -- آمن للإطلاق"
+
+        duration_ms = int((completed_at - started_at).total_seconds() * 1000)
         return ShannonReport(
-            scan_id=scan_id,
             environment=environment,
             scopes_tested=scopes,
             findings=findings,
             started_at=started_at,
             completed_at=completed_at,
-            should_block_release=should_block,
+            duration_ms=duration_ms,
-            summary=summary,
+            critical_count=counts[Severity.CRITICAL],
-            summary_ar=summary_ar,
+            high_count=counts[Severity.HIGH],
+            medium_count=counts[Severity.MEDIUM],
+            low_count=counts[Severity.LOW],
+            info_count=counts[Severity.INFO],
+            release_recommendation=rec,
+            release_recommendation_ar=rec_ar,
+            message_ar=(
+                f"فحص {environment}: {len(findings)} نتيجة -- "
+                f"حرجة: {counts[Severity.CRITICAL]}، عالية: {counts[Severity.HIGH]}، "
+                f"متوسطة: {counts[Severity.MEDIUM]} -- {rec_ar}"
+            ),
         )
 
-    async def _check_scope(
+    async def should_block_release(self, findings: list[ShannonFinding]) -> bool:
-        self, scope: ShannonScope, base_url: str, creds: dict = None
+        """True if any critical or 3+ high-severity findings exist."""
-    ) -> list[ShannonFinding]:
+        critical = sum(1 for f in findings if f.severity == Severity.CRITICAL and f.verified)
-        checks = {
+        high = sum(1 for f in findings if f.severity == Severity.HIGH and f.verified)
-            ShannonScope.AUTH: self._check_auth,
-            ShannonScope.INJECTION: self._check_injection,
-            ShannonScope.TENANT_ISOLATION: self._check_tenant_isolation,
-            ShannonScope.PDPL: self._check_pdpl,
-            ShannonScope.API_ROUTES: self._check_api_routes,
-            ShannonScope.FILE_UPLOAD: self._check_file_upload,
-            ShannonScope.WEBSOCKET: self._check_websocket,
-        }
-        checker = checks.get(scope)
-        if checker:
-            return await checker(base_url, creds)
-        return []
-
-    async def _check_auth(self, base_url: str, creds: dict = None) -> list[ShannonFinding]:
-        findings = []
-        findings.append(ShannonFinding(
-            id=f"AUTH-{len(self._findings)+1}",
-            scope=ShannonScope.AUTH,
-            severity="high",
-            title="JWT expiration check",
-            title_ar="فحص <20><>نتهاء صلاحية JWT",
-            description="Verify JWT tokens expire within configured timeframe",
-            proof_of_concept=f"GET {base_url}/api/v1/auth/me with expired token",
-            affected_endpoint="/api/v1/auth/me",
-            impact="Expired tokens could allow unauthorized access",
-            remediation="Ensure ACCESS_TOKEN_EXPIRE_MINUTES is set and enforced",
-            remediation_ar="تأكد من إعداد وقت انتهاء الرمز وتطبيقه",
-            verified=False,
-            cwe_id="CWE-613",
-        ))
-        return findings
-
-    async def _check_injection(self, base_url: str, creds: dict = None) -> list[ShannonFinding]:
-        findings = []
-        sql_payloads = ["' OR '1'='1", "'; DROP TABLE leads;--", "1 UNION SELECT NULL"]
-        for payload in sql_payloads:
-            findings.append(ShannonFinding(
-                id=f"INJ-{len(self._findings)+len(findings)+1}",
-                scope=ShannonScope.INJECTION,
-                severity="critical",
-                title=f"SQL injection test: {payload[:20]}...",
-                title_ar="اختبار حقن SQL",
-                description=f"Test search endpoints with payload: {payload}",
-                proof_of_concept=f"GET {base_url}/api/v1/leads?search={payload}",
-                affected_endpoint="/api/v1/leads",
-                impact="Database compromise, data exfiltration",
-                remediation="Ensure all queries use parameterized SQLAlchemy ORM",
-                remediation_ar="تأكد من استخدام SQLAlchemy ORM للاستعلامات",
-                verified=False,
-                cwe_id="CWE-89",
-            ))
-        return findings
-
-    async def _check_tenant_isolation(self, base_url: str, creds: dict = None) -> list[ShannonFinding]:
-        return [ShannonFinding(
-            id=f"TENANT-{len(self._findings)+1}",
-            scope=ShannonScope.TENANT_ISOLATION,
-            severity="critical",
-            title="Cross-tenant data access test",
-            title_ar="اختبار الوصول عبر المستأجرين",
-            description="Verify tenant A cannot access tenant B's leads/deals",
-            proof_of_concept="Login as tenant A, request tenant B's lead by ID",
-            affected_endpoint="/api/v1/leads/{id}",
-            impact="Complete data breach across tenants",
-            remediation="Enforce tenant_id filter on all queries",
-            remediation_ar="فرض فلتر tenant_id على كل الاستعلامات",
-            verified=False,
-            cwe_id="CWE-284",
-        )]
-
-    async def _check_pdpl(self, base_url: str, creds: dict = None) -> list[ShannonFinding]:
-        return [ShannonFinding(
-            id=f"PDPL-{len(self._findings)+1}",
-            scope=ShannonScope.PDPL,
-            severity="high",
-            title="PDPL consent bypass test",
-            title_ar="اختبار تجاوز موافقة PDPL",
-            description="Test if messages can be sent without recorded consent",
-            proof_of_concept=f"POST {base_url}/api/v1/inbox/reply without consent record",
-            affected_endpoint="/api/v1/inbox/reply",
-            impact="PDPL violation — up to SAR 5M fine",
-            remediation="Check consent before every outbound message",
-            remediation_ar="فحص الموافقة قبل كل رسالة صادرة",
-            verified=False,
-            cwe_id="CWE-862",
-        )]
-
-    async def _check_api_routes(self, base_url: str, creds: dict = None) -> list[ShannonFinding]:
-        return []
-
-    async def _check_file_upload(self, base_url: str, creds: dict = None) -> list[ShannonFinding]:
-        return []
-
-    async def _check_websocket(self, base_url: str, creds: dict = None) -> list[ShannonFinding]:
-        return []
-
-    async def should_block_release(self) -> bool:
-        critical = sum(1 for f in self._findings if f.severity == "critical" and not f.verified)
-        high = sum(1 for f in self._findings if f.severity == "high" and not f.verified)
         return critical > 0 or high >= 3
 
-    def get_all_findings(self, severity: str = None) -> list[ShannonFinding]:
+    def get_latest_report(self) -> Optional[ShannonReport]:
-        if severity:
+        """Return the most recent scan report."""
-            return [f for f in self._findings if f.severity == severity]
+        return self._reports[-1] if self._reports else None
-        return self._findings
+
+    def get_all_reports(self) -> list[ShannonReport]:
+        """Return all stored reports."""
+        return list(self._reports)
 
 
-shannon = ShannonSecurityLane()
+# ---------------------------------------------------------------------------
+# Module-level singleton
+# ---------------------------------------------------------------------------
+
+shannon_security = ShannonSecurityLane()

salesflow-saas/backend/app/services/skill_governance.py

@@ -0,0 +1,205 @@
+"""
+Skill Governance — Dealix AI Revenue OS (Antigravity Pattern)
+Governed skill admission, bundles, and lifecycle management.
+Skills are admitted through rubric, not mass-installed.
+"""
+import logging
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+
+class AdmissionDecision(str, Enum):
+    ADMIT_NOW = "admit_now"
+    ADMIT_LATER = "admit_later"
+    DO_NOT_ADMIT = "do_not_admit"
+    REIMPLEMENT_LOCALLY = "reimplement_locally"
+    DESIGN_REFERENCE = "design_reference"
+
+
+class SkillCandidate(BaseModel):
+    id: str
+    name: str
+    source: str  # "antigravity", "community", "internal", "mkhlab"
+    category: str
+    description: str
+    relevance_score: float = 0.0  # 0-1
+    safety_risk: str = "low"  # low, medium, high
+    maintenance_burden: str = "low"
+    overlap_with_existing: list[str] = []
+    measurable_roi: str = ""
+    license: str = "MIT"
+    decision: Optional[AdmissionDecision] = None
+    decision_reason: str = ""
+    evaluated_at: Optional[datetime] = None
+
+
+class SkillBundle(BaseModel):
+    id: str
+    name: str
+    name_ar: str
+    description: str
+    skills: list[str]  # skill IDs
+    target_profile: str  # Hermes profile this bundle serves
+    priority: str = "medium"
+
+
+# Admission rubric weights
+RUBRIC = {
+    "relevance": 0.30,       # Direct project relevance
+    "safety": 0.20,          # Safety risk (inverted)
+    "maintenance": 0.15,     # Maintenance burden (inverted)
+    "roi": 0.15,             # Measurable ROI
+    "overlap": 0.10,         # Low overlap with existing (inverted)
+    "deterministic": 0.10,   # Deterministic utility
+}
+
+SAFETY_SCORES = {"low": 1.0, "medium": 0.5, "high": 0.1}
+MAINTENANCE_SCORES = {"low": 1.0, "medium": 0.6, "high": 0.2}
+
+# Pre-defined bundles for Dealix
+DEFAULT_BUNDLES: list[dict] = [
+    {
+        "id": "bundle-repo-audit",
+        "name": "Repo Audit", "name_ar": "فحص المستودع",
+        "description": "Code quality, architecture review, dependency check",
+        "skills": ["repo-audit", "architecture-review", "dependency-check"],
+        "target_profile": "ops", "priority": "high",
+    },
+    {
+        "id": "bundle-release",
+        "name": "Release Hardening", "name_ar": "تقوية الإصدار",
+        "description": "Release prep, canary check, rollback planning",
+        "skills": ["release-prep", "canary-check", "rollback-plan", "security-preflight"],
+        "target_profile": "ops", "priority": "high",
+    },
+    {
+        "id": "bundle-growth",
+        "name": "Growth & SEO", "name_ar": "النمو والسيو",
+        "description": "Content generation, SEO audit, competitor analysis",
+        "skills": ["content-gen", "seo-audit", "competitor-scan"],
+        "target_profile": "growth", "priority": "medium",
+    },
+    {
+        "id": "bundle-sales",
+        "name": "Sales Research", "name_ar": "بحث المبيعات",
+        "description": "Lead research, proposal drafting, call prep",
+        "skills": ["lead-research", "proposal-draft", "call-prep", "objection-handler"],
+        "target_profile": "sales", "priority": "high",
+    },
+    {
+        "id": "bundle-arabic",
+        "name": "Arabic Market Ops", "name_ar": "عمليات السوق العربي",
+        "description": "Arabic summarization, dialect detection, RTL checks",
+        "skills": ["arabic-summarize", "dialect-detect", "rtl-check", "arabizi-convert"],
+        "target_profile": "arabic-ops", "priority": "high",
+    },
+    {
+        "id": "bundle-qa",
+        "name": "QA & Testing", "name_ar": "ضمان الجودة",
+        "description": "Test generation, browser QA, regression check",
+        "skills": ["generate-tests", "browser-qa", "regression-check"],
+        "target_profile": "delivery", "priority": "medium",
+    },
+    {
+        "id": "bundle-knowledge",
+        "name": "Documentation & Knowledge", "name_ar": "التوثيق والمعرفة",
+        "description": "Wiki maintenance, runbook generation, API docs",
+        "skills": ["wiki-update", "runbook-gen", "api-docs", "glossary-update"],
+        "target_profile": "knowledge", "priority": "medium",
+    },
+]
+
+
+class SkillGovernance:
+    """Governed skill admission and lifecycle management."""
+
+    def __init__(self):
+        self._candidates: list[SkillCandidate] = []
+        self._admitted: list[str] = []
+        self._rejected: list[str] = []
+        self._bundles = [SkillBundle(**b) for b in DEFAULT_BUNDLES]
+
+    def evaluate_candidate(self, candidate: SkillCandidate) -> SkillCandidate:
+        safety_score = SAFETY_SCORES.get(candidate.safety_risk, 0.5)
+        maint_score = MAINTENANCE_SCORES.get(candidate.maintenance_burden, 0.5)
+        overlap_score = 1.0 - min(len(candidate.overlap_with_existing) * 0.3, 1.0)
+        roi_score = 0.8 if candidate.measurable_roi else 0.2
+        deterministic_score = 0.7  # Default moderate
+
+        total = (
+            candidate.relevance_score * RUBRIC["relevance"]
+            + safety_score * RUBRIC["safety"]
+            + maint_score * RUBRIC["maintenance"]
+            + roi_score * RUBRIC["roi"]
+            + overlap_score * RUBRIC["overlap"]
+            + deterministic_score * RUBRIC["deterministic"]
+        )
+
+        if total >= 0.7:
+            candidate.decision = AdmissionDecision.ADMIT_NOW
+            candidate.decision_reason = f"Score {total:.2f} — high fit, safe, low overlap"
+        elif total >= 0.5:
+            candidate.decision = AdmissionDecision.ADMIT_LATER
+            candidate.decision_reason = f"Score {total:.2f} — moderate fit, review later"
+        elif candidate.overlap_with_existing:
+            candidate.decision = AdmissionDecision.REIMPLEMENT_LOCALLY
+            candidate.decision_reason = f"Score {total:.2f} — overlaps with existing: {candidate.overlap_with_existing}"
+        elif total >= 0.3:
+            candidate.decision = AdmissionDecision.DESIGN_REFERENCE
+            candidate.decision_reason = f"Score {total:.2f} — use as design pattern only"
+        else:
+            candidate.decision = AdmissionDecision.DO_NOT_ADMIT
+            candidate.decision_reason = f"Score {total:.2f} — low relevance or high risk"
+
+        candidate.evaluated_at = datetime.now(timezone.utc)
+        self._candidates.append(candidate)
+
+        if candidate.decision == AdmissionDecision.ADMIT_NOW:
+            self._admitted.append(candidate.id)
+        elif candidate.decision == AdmissionDecision.DO_NOT_ADMIT:
+            self._rejected.append(candidate.id)
+
+        logger.info(
+            f"Skill evaluated: {candidate.name} → {candidate.decision.value} "
+            f"({candidate.decision_reason})"
+        )
+        return candidate
+
+    def get_bundle(self, bundle_id: str) -> Optional[SkillBundle]:
+        return next((b for b in self._bundles if b.id == bundle_id), None)
+
+    def get_bundles_for_profile(self, profile: str) -> list[SkillBundle]:
+        return [b for b in self._bundles if b.target_profile == profile]
+
+    def list_bundles(self) -> list[SkillBundle]:
+        return self._bundles
+
+    def get_admitted(self) -> list[str]:
+        return self._admitted
+
+    def get_rejected(self) -> list[str]:
+        return self._rejected
+
+    def get_candidates(self, decision: AdmissionDecision = None) -> list[SkillCandidate]:
+        if decision:
+            return [c for c in self._candidates if c.decision == decision]
+        return self._candidates
+
+    def get_admission_stats(self) -> dict:
+        total = len(self._candidates)
+        return {
+            "total_evaluated": total,
+            "admitted": len(self._admitted),
+            "rejected": len(self._rejected),
+            "pending": total - len(self._admitted) - len(self._rejected),
+            "bundles": len(self._bundles),
+            "admission_rate": round(len(self._admitted) / total * 100, 1) if total else 0,
+        }
+
+
+skill_governance = SkillGovernance()