server/state.py

"""Глобальное состояние PinkSky"""
import os
import json
import threading
import logging
from datetime import datetime
from typing import Dict, List, Any, Optional
from .models import ModelConfig, Role, Conductor
from .model_ranking import MODEL_RANKING
from .config import ROLES_FILE, MODELS_FILE, CONDUCTORS_FILE, HISTORY_FILE

class PinkSkyState:
    _instance = None
    _initialized = False
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def load_all(self):
        """Загружает модели, роли и кондукторы из файлов."""
        self._load_models()
        self._load_roles()
        self._load_conductors()
        self._load_history()

    def _load_models(self):
        defaults = {
            name: self._build_model_config(name, data)
            for name, data in MODEL_RANKING.items()
        }
        if os.path.exists(MODELS_FILE):
            try:
                with open(MODELS_FILE, "r", encoding="utf-8") as f:
                    custom = json.load(f)
                for k, v in custom.items():
                    if k not in defaults:
                        defaults[k] = ModelConfig(**v)
            except Exception as e:
                self.logger.error(f"Ошибка загрузки models.json: {e}")
        self.models = defaults
        
    def __init__(self):
        if PinkSkyState._initialized:
            return
        PinkSkyState._initialized = True
        self.logger = logging.getLogger(__name__)
        self.models: Dict[str, ModelConfig] = {}
        self.roles: Dict[str, Role] = {}
        self.conductors: Dict[str, Conductor] = {}
        self.current_mode: str = "chat"
        self.current_conductor: str = "default"
        self.current_role: str = "universal"
        self.current_model: str = "deepseek-v4-pro"
        self.chat_history: List[Dict[str, str]] = []
        self.skill_history: List[Dict[str, str]] = []
        self.build_history: List[Dict[str, str]] = []
        self.build_context: Dict[str, Any] = {
            "spec": "",
            "agents": 3,
            "models_tier": "tier1",
            "skills_count": 2,
            "files_count": 3,
            "role": "universal",
            "strategy": "parallel",
            "use_interpreter": True,
            "notifications": True,
            "internet_access": True
        }
        self.cancel_flag: bool = False
        self.last_history_save: Optional[datetime] = None
        self.load_all()

    def add_to_history(self, mode: str, role: str, content: str):
        with PinkSkyState._lock:
            entry = {
                "role": role,
                "content": content,
                "timestamp": datetime.now().isoformat()
            }
            history_attr = f"{mode}_history"
            history = getattr(self, history_attr, [])
            history.append(entry)
            setattr(self, history_attr, history)

            # Сохраняем не чаще раза в 30 секунд
            now = datetime.now()
            if (self.last_history_save is None or
                (now - self.last_history_save).total_seconds() > 30):
                self.save_history()
                self.last_history_save = now

    def _build_model_config(self, name: str, data: dict) -> ModelConfig:
        return ModelConfig(
            name=name, provider="openai", endpoint=data["endpoint"],
            api_key_env="NVIDIA_API_KEY",
            context_window=data.get("context_window", 32000),
            max_tokens=data.get("max_tokens", 8000),
            cost_per_1k_input=data.get("cost_per_1k_input", 0.0),
            cost_per_1k_output=data.get("cost_per_1k_output", 0.0),
            coding_rank=data.get("coding_rank", 50),
            speed_rank=data.get("speed_rank", 50),
            reasoning_rank=data.get("reasoning_rank", 50),
            tags=data.get("tags", [])
        )

    def _load_models(self):
        defaults = {name: self._build_model_config(name, data) for name, data in MODEL_RANKING.items()}
        if os.path.exists(MODELS_FILE):
            try:
                with open(MODELS_FILE, "r", encoding="utf-8") as f:
                    custom = json.load(f)
                    for k, v in custom.items():
                        if k not in defaults:
                            defaults[k] = ModelConfig(**v)
            except Exception as e:
                print(f"⚠️ Ошибка загрузки models.json: {e}")
        self.models = defaults

    def _load_roles(self):
        defaults = {
            "universal": Role(
                name="universal",
                prompt="You are PinkSky -- a universal AI assistant and autonomous developer. You help users with any tasks, scripts, theory, and project creation from scratch.",
                description="Universal assistant for any tasks",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "qwen3.5-397b"],
                complexity="medium",
                tags=["general"]
            ),
            "guru": Role(
                name="guru",
                prompt="You are Guru Programmer PinkSky. 15+ years experience. Write elegant, production-ready code. Principles: KISS, explicit > implicit, composition > inheritance, PEP8, type hints, docstrings. Format: analysis -> code -> explanations -> edge cases.",
                description="Guru programmer. Elegant code with deep explanations.",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"],
                complexity="high",
                tags=["coding", "senior", "mentor", "python"]
            ),
            "hacker": Role(
                name="hacker",
                prompt="You are Hacker PinkSky. Code virtuoso. Find elegant and unconventional solutions. Use __slots__, descriptors, metaclasses. Optimize time complexity, memory layout. Love functional: itertools, functools, operator.",
                description="Hacker-coder. Optimization and unconventional solutions.",
                preferred_models=["deepseek-v4-pro", "deepseek-v4-flash", "llama-4-maverick", "nemotron-super-49b"],
                complexity="high",
                tags=["coding", "optimization", "hacks", "performance"]
            ),
            "architect": Role(
                name="architect",
                prompt="You are Software Architect PinkSky. Design systems that last years. Bounded contexts, aggregates, CQRS, Event Sourcing. API: REST, gRPC, GraphQL, WebSocket. Observability: logs, metrics, tracing from the start.",
                description="Software Architect. High-level system design.",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "nemotron-3-super", "qwen3.5-397b"],
                complexity="high",
                tags=["architecture", "design", "system", "ddd"]
            ),
            "principal": Role(
                name="principal",
                prompt="You are Principal Engineer PinkSky. Solve problems no one else can. Refactor legacy without downtime. Platform-level: CI/CD, observability, service mesh. Engineering culture: code review, RFC process. ADR for all decisions.",
                description="Principal engineer. Strategy, mentorship, hard problems.",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"],
                complexity="high",
                tags=["leadership", "strategy", "mentoring", "legacy"]
            ),
            "evangelist": Role(
                name="evangelist",
                prompt="You are Quality Evangelist PinkSky. TDD, BDD, property-based testing, mutation testing. pytest, hypothesis, coverage, mypy, ruff, bandit. Test pyramid: unit -> integration -> e2e. CI/CD gates: coverage threshold, mutation score.",
                description="Quality evangelist. Testing and quality culture.",
                preferred_models=["kimi-k2.6", "deepseek-v4-pro", "mistral-medium-3.5"],
                complexity="high",
                tags=["quality", "testing", "tdd", "ci-cd"]
            ),
            "techlead": Role(
                name="techlead",
                prompt="You are Tech Lead PinkSky. Code review: correctness, readability, maintainability, security, performance. Find race conditions, memory leaks, injection points, N+1. must-fix vs should-fix vs nitpick. Code review = teaching, not tribunal.",
                description="Tech Lead. Code review and team direction.",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"],
                complexity="high",
                tags=["review", "leadership", "team", "mentoring"]
            ),
            "qa": Role(
                name="qa",
                prompt="You are QA Engineer PinkSky. Test cases: positive, negative, boundary, exploratory. Equivalence partitioning, boundary value analysis. Automation: Selenium, Playwright, Postman. Performance: k6, Locust. Security: OWASP Top 10.",
                description="QA engineer. Bug hunting and test strategy.",
                preferred_models=["mistral-small-4", "step-3.7-flash", "llama-3.3-70b", "deepseek-v4-flash"],
                complexity="medium",
                tags=["qa", "testing", "automation", "manual"]
            ),
            "sdet": Role(
                name="sdet",
                prompt="You are SDET PinkSky. Test frameworks: pytest plugins, custom matchers. CI/CD: parallel execution, test sharding. Test data: factories, fixtures, seeding, cleanup. Mocks/stubs/fakes: wiremock, mockserver. Test code = production code.",
                description="SDET. Autotests and test infrastructure at dev level.",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "llama-4-maverick", "mistral-medium-3.5"],
                complexity="high",
                tags=["sdet", "automation", "framework", "infrastructure"]
            ),
            "qe": Role(
                name="qe",
                prompt="You are Quality Engineer (QE) PinkSky. Analyze SDLC: where quality is lost. Shift-left testing: quality gates at every stage. Metrics: DORA, SPACE, custom KPIs. Root cause analysis: 5 Whys, Fishbone, FMEA. Every production bug = learning opportunity.",
                description="Quality engineer. Processes, metrics, and quality culture.",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "nemotron-3-super"],
                complexity="high",
                tags=["qe", "process", "metrics", "culture", "sdlc"]
            ),
            "researcher": Role(
                name="researcher",
                prompt="You are Researcher PinkSky. Deep topic analysis. Compare approaches: trade-offs, limitations. Structure: executive summary -> details -> sources. Identify trends. Evidence > opinions. Numbers > words.",
                description="Researcher and analyst. Deep topic analysis.",
                preferred_models=["deepseek-v4-pro", "qwen3.5-397b", "kimi-k2.6", "gpt-oss-120b"],
                complexity="high",
                tags=["research", "analysis", "comparison"]
            ),
            "critic": Role(
                name="critic",
                prompt="You are Critic and Auditor PinkSky. correctness, security, performance, maintainability. race conditions, injection points, memory leaks, N+1. code smells, technical debt, architecture risks. Every issue with severity. Suggest fixes.",
                description="Critic and auditor. Bug and issue hunting.",
                preferred_models=["deepseek-v4-pro", "kimi-k2.6", "mistral-large-3", "gpt-oss-120b"],
                complexity="medium",
                tags=["audit", "security", "review", "critic"]
            ),
        }
        if os.path.exists(ROLES_FILE):
            try:
                with open(ROLES_FILE, "r", encoding="utf-8") as f:
                    custom = json.load(f)
                    for k, v in custom.items():
                        if k not in defaults:
                            defaults[k] = Role(**v)
            except Exception as e:
                print(f"⚠️ Ошибка загрузки roles.json: {e}")
        self.roles = defaults

    def _load_conductors(self):
        defaults = {
            "default": Conductor(
                name="default",
                prompt="""You are Conductor PinkSky (Default). Analyze request and choose optimal roles and models.

RULES:
1. Simple questions -- 1 role, 1 model.
2. Complex tasks -- decompose, assign roles.
3. Consider cost: cheap for simple, powerful for complex.
4. If code -- add critic.
5. If architecture -- add architect.

AVAILABLE ROLES: guru, hacker, architect, principal, evangelist, techlead, qa, sdet, qe, researcher, critic, universal.

AVAILABLE MODELS (by coding rank, best to worst):
TIER 1 (Elite): deepseek-v4-pro, kimi-k2.6, qwen3.5-397b, mistral-large-3, gpt-oss-120b
TIER 2 (Strong): deepseek-v4-flash, llama-4-maverick, nemotron-3-super, mistral-medium-3.5, dracarys-llama-70b, llama-3.3-70b, nemotron-super-49b
TIER 3 (Good): step-3.7-flash, mistral-small-4, minimax-m2.7, nemotron-super-49b-v1, llama-3.2-90b-vision
TIER 4 (Fast): nemotron-nano-12b, nemotron-3-nano-30b, nemotron-nano-9b, nemotron-content-safety
TIER 5 (Specialized): nemotron-3-nano-omni, diffusiongemma

FORMAT (STRICT JSON):
{"strategy": "single|sequential|parallel", "tasks": [{"role": "role_name", "model": "model_name", "prompt": "subtask"}], "synthesis_prompt": "how to combine"}""",
                description="Standard conductor -- balance of quality and speed",
                strategy="selective",
                max_agents=3,
                cost_aware=True,
                auto_rank_by="balanced"
            ),
            "strict": Conductor(
                name="strict",
                prompt="""You are Strict Conductor PinkSky. Minimum agents, maximum efficiency.

RULES:
1. ONLY one role and one model.
2. Cheapest model capable of solving the task.
3. Only sequential.

FORMAT (STRICT JSON):
{"strategy": "single", "tasks": [{"role": "name", "model": "name", "prompt": "task"}], "synthesis_prompt": ""}""",
                description="Minimum agents, minimum cost",
                strategy="single",
                max_agents=1,
                cost_aware=True,
                auto_rank_by="coding"
            ),
            "creative": Conductor(
                name="creative",
                prompt="""You are Creative Conductor PinkSky. Maximum perspectives, brainstorm.

RULES:
1. Multiple roles from different angles.
2. Parallel strategy.
3. guru + hacker + researcher + critic.
4. Do not save on models -- use the best.

FORMAT (STRICT JSON):
{"strategy": "parallel", "tasks": [...], "synthesis_prompt": "synthesize creative ideas"}""",
                description="Maximum roles, creative brainstorm",
                strategy="parallel",
                max_agents=5,
                cost_aware=False,
                auto_rank_by="coding"
            ),
            "economy": Conductor(
                name="economy",
                prompt="""You are Economy Conductor PinkSky. Solve task for minimum cost.

RULES:
1. Start with TIER 4 (fast/cheap): nemotron-nano-9b, nemotron-nano-12b, nemotron-3-nano-30b.
2. Only if it fails -- escalate to TIER 3/2.
3. One role, one model.

FORMAT (STRICT JSON):
{"strategy": "single", "tasks": [{"role": "name", "model": "name", "prompt": "task"}], "synthesis_prompt": ""}""",
                description="Cheap models, budget saving",
                strategy="single",
                max_agents=1,
                cost_aware=True,
                auto_rank_by="speed"
            ),
            "review": Conductor(
                name="review",
                prompt="""You are Code Review Conductor PinkSky. Maximum quality code review.

RULES:
1. techlead (architectural review) + critic (bugs/vulnerabilities) + guru (best practices).
2. Parallel review.
3. Synthesize into structured report.

FORMAT (STRICT JSON):
{"strategy": "parallel", "tasks": [{"role": "techlead", "model": "deepseek-v4-pro", "prompt": "architectural review"}, {"role": "critic", "model": "kimi-k2.6", "prompt": "bug hunting"}, {"role": "guru", "model": "mistral-large-3", "prompt": "best practices"}], "synthesis_prompt": "structured report with severity"}""",
                description="Focus on code review. Multi-angle code check.",
                strategy="parallel",
                max_agents=4,
                cost_aware=True,
                auto_rank_by="coding"
            ),
            "build": Conductor(
                name="build",
                prompt="""You are Project Build Conductor PinkSky. Build full project from spec.

RULES:
1. Sequential: architect -> guru/hacker -> sdet -> critic.
2. Each stage -- separate call.

FORMAT (STRICT JSON):
{"strategy": "sequential", "tasks": [{"role": "architect", "model": "deepseek-v4-pro", "prompt": "architecture"}, {"role": "guru", "model": "kimi-k2.6", "prompt": "code"}, {"role": "sdet", "model": "mistral-medium-3.5", "prompt": "tests"}, {"role": "critic", "model": "gpt-oss-120b", "prompt": "audit"}], "synthesis_prompt": "assemble into single project"}""",
                description="Project build. Architecture -> code -> tests -> audit.",
                strategy="sequential",
                max_agents=5,
                cost_aware=True,
                auto_rank_by="coding"
            ),
        }
        if os.path.exists(CONDUCTORS_FILE):
            try:
                with open(CONDUCTORS_FILE, "r", encoding="utf-8") as f:
                    custom = json.load(f)
                    for k, v in custom.items():
                        if k not in defaults:
                            defaults[k] = Conductor(**v)
            except Exception as e:
                print(f"⚠️ Ошибка загрузки conductors.json: {e}")
        self.conductors = defaults

    def _load_history(self):
        if os.path.exists(HISTORY_FILE):
            try:
                with open(HISTORY_FILE, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    self.chat_history = data.get("chat", [])
                    self.skill_history = data.get("skill", [])
                    self.build_history = data.get("build", [])
            except Exception as e:
                print(f"⚠️ Ошибка загрузки истории: {e}")
    def __init__(self):
        if PinkSkyState._initialized:
            return
        PinkSkyState._initialized = True
        self.logger = logging.getLogger(__name__)
        self.models: Dict[str, ModelConfig] = {}
        self.roles: Dict[str, Role] = {}
        self.conductors: Dict[str, Conductor] = {}
        self.current_mode: str = "chat"
        self.current_conductor: str = "default"
        self.current_role: str = "universal"
        self.current_model: str = "deepseek-v4-pro"
        self.chat_history: List[Dict[str, str]] = []
        self.skill_history: List[Dict[str, str]] = []
        self.build_history: List[Dict[str, str]] = []
        self.build_context: Dict[str, Any] = {
            "spec": "",
            "agents": 3,
            "models_tier": "tier1",
            "skills_count": 2,
            "files_count": 3,
            "role": "universal",
            "strategy": "parallel",
            "use_interpreter": True,
            "notifications": True,
            "internet_access": True,
        }
        self.cancel_flag: bool = False
        self.last_history_save: Optional[datetime] = None
        self.load_all()  # Теперь метод load_all доступен
        
    def save_roles(self):
        data = {k: {"name": v.name, "prompt": v.prompt, "description": v.description,
                    "preferred_models": v.preferred_models, "complexity": v.complexity, "tags": v.tags}
                for k, v in self.roles.items()}
        with open(ROLES_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def save_models(self):
        data = {k: {"name": v.name, "provider": v.provider, "endpoint": v.endpoint,
                    "api_key_env": v.api_key_env, "context_window": v.context_window,
                    "max_tokens": v.max_tokens, "cost_per_1k_input": v.cost_per_1k_input,
                    "cost_per_1k_output": v.cost_per_1k_output,
                    "coding_rank": v.coding_rank, "speed_rank": v.speed_rank, "reasoning_rank": v.reasoning_rank,
                    "tags": v.tags}
                for k, v in self.models.items()}
        with open(MODELS_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def save_conductors(self):
        data = {k: {"name": v.name, "prompt": v.prompt, "description": v.description,
                    "strategy": v.strategy, "max_agents": v.max_agents, "cost_aware": v.cost_aware,
                    "auto_rank_by": v.auto_rank_by}
                for k, v in self.conductors.items()}
        with open(CONDUCTORS_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def save_history(self):
        data = {"chat": self.chat_history, "skill": self.skill_history, "build": self.build_history}
        with open(HISTORY_FILE, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    def add_to_history(self, mode: str, role: str, content: str):
        entry = {"role": role, "content": content, "timestamp": datetime.now().isoformat()}
        if mode == "chat":
            self.chat_history.append(entry)
        elif mode == "skill":
            self.skill_history.append(entry)
        elif mode == "build":
            self.build_history.append(entry)
        self.save_history()

    def get_best_model(self, rank_by: str = "coding", min_tier: int = 1, max_tier: int = 5, exclude: List[str] = None) -> str:
        exclude = exclude or []
        candidates = []
        for name, model in self.models.items():
            if name in exclude or name == "hf_fallback":
                continue
            tier = 5
            if model.coding_rank <= 5: tier = 1
            elif model.coding_rank <= 12: tier = 2
            elif model.coding_rank <= 18: tier = 3
            elif model.coding_rank <= 24: tier = 4
            if min_tier <= tier <= max_tier:
                candidates.append((name, model))
        if not candidates:
            return "deepseek-v4-pro"
        if rank_by == "coding":
            candidates.sort(key=lambda x: x[1].coding_rank)
        elif rank_by == "speed":
            candidates.sort(key=lambda x: x[1].speed_rank)
        elif rank_by == "reasoning":
            candidates.sort(key=lambda x: x[1].reasoning_rank)
        elif rank_by == "balanced":
            candidates.sort(key=lambda x: (x[1].coding_rank + x[1].speed_rank + x[1].reasoning_rank) / 3)
        else:
            candidates.sort(key=lambda x: x[1].coding_rank)
        return candidates[0][0]

    def get_model_for_role(self, role_name: str, preference: str = None, rank_by: str = None) -> str:
        role = self.roles.get(role_name)
        if not role:
            return preference or self.current_model
        conductor = self.conductors.get(self.current_conductor, self.conductors["default"])
        rank_criteria = rank_by or conductor.auto_rank_by
        max_tier = 5
        if role.complexity == "high":
            max_tier = 2
        elif role.complexity == "medium":
            max_tier = 3
        if preference and preference in self.models:
            return preference
        available = [m for m in role.preferred_models if m in self.models and m != "hf_fallback"]
        if available:
            if conductor.cost_aware and rank_criteria != "coding":
                available.sort(key=lambda m: self.models[m].cost_per_1k_output)
            else:
                if rank_criteria == "coding":
                    available.sort(key=lambda m: self.models[m].coding_rank)
                elif rank_criteria == "speed":
                    available.sort(key=lambda m: self.models[m].speed_rank)
                elif rank_criteria == "reasoning":
                    available.sort(key=lambda m: self.models[m].reasoning_rank)
                else:
                    available.sort(key=lambda m: (self.models[m].coding_rank + self.models[m].speed_rank + self.models[m].reasoning_rank) / 3)
            return available[0]
        return self.get_best_model(rank_by=rank_criteria, max_tier=max_tier)

    def get_models_by_tier(self, tier: int) -> List[str]:
        result = []
        for name, model in self.models.items():
            if name == "hf_fallback":
                continue
            model_tier = 5
            if model.coding_rank <= 5: model_tier = 1
            elif model.coding_rank <= 12: model_tier = 2
            elif model.coding_rank <= 18: model_tier = 3
            elif model.coding_rank <= 24: model_tier = 4
            if model_tier == tier:
                result.append(name)
        return result

    def get_next_tier_model(self, current_model_name: str) -> Optional[str]:
        if current_model_name not in self.models:
            return None
        current = self.models[current_model_name]
        current_tier = 5
        if current.coding_rank <= 5: current_tier = 1
        elif current.coding_rank <= 12: current_tier = 2
        elif current.coding_rank <= 18: current_tier = 3
        elif current.coding_rank <= 24: current_tier = 4
        next_tier = current_tier + 1
        if next_tier > 5:
            return None
        models_in_tier = self.get_models_by_tier(next_tier)
        if models_in_tier:
            return models_in_tier[0]
        return None

    def export_history_json(self) -> str:
        return json.dumps({"exported_at": datetime.now().isoformat(), "chat": self.chat_history, "skill": self.skill_history, "build": self.build_history}, ensure_ascii=False, indent=2)

    def export_history_md(self) -> str:
        lines = ["# PinkSky History Export", ""]
        lines.append("*Exported: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S") + "*")
        lines.append("")
        for mode, history in [("Chat", self.chat_history), ("Skill", self.skill_history), ("Build", self.build_history)]:
            lines.append("## " + mode + " Mode")
            lines.append("")
            for entry in history:
                ts = entry.get("timestamp", "unknown")
                role = entry.get("role", "unknown")
                content = entry.get("content", "")
                lines.append("### " + role + " (" + ts + ")")
                lines.append("")
                lines.append("```")
                lines.append(content[:500])
                lines.append("```")
                lines.append("")
        return "\n".join(lines)

STATE = PinkSkyState()