Upload state.py

state.py

@@ -0,0 +1,463 @@
+"""Глобальное состояние PinkSky"""
+
+import os
+import json
+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:
+    def __init__(self):
+        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.load_all()
+
+    def load_all(self):
+        self._load_models()
+        self._load_roles()
+        self._load_conductors()
+        self._load_history()
+
+    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 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()