system-prompts-and-models-of-ai-tools/salesflow-saas/backend/app/services/gstack_discipline.py

"""
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()