feat(pipeline): A35 and A36 canonical Antigravity workflow pipeline with role resolution and loops

src/antigravity_provider/hermes_plugin.py

@@ -27,17 +27,45 @@ def _error_message(exc: Exception) -> str:
 def antigravity_llm_execution(**kwargs: Any) -> Any:
     request = kwargs.get("request") or {}
     next_call = kwargs.get("next_call")
-    provider = kwargs.get("provider")
+    provider = kwargs.get("provider") or request.get("provider")
+    model = kwargs.get("model") or request.get("model")
+    session_id = kwargs.get("session_id") or request.get("session_id")
 
-    # 1. Try routing through Multi-Provider Account Router if enabled AND role is determined
+    # 1. Try routing through Multi-Provider Account Router if enabled
     try:
         from .router import get_router_engine
         engine = get_router_engine()
         if engine.config.enabled:
+            # Check if router has any configured and enabled profiles.
+            # Empty configuration must not break Hermes and falls through cleanly.
+            has_active_profiles = bool(
+                engine.config.profiles and any(p.enabled for p in engine.config.profiles.values())
+            )
+            if not has_active_profiles:
+                logger.info("Router has no active profiles; passing call downstream to Hermes")
+                if callable(next_call):
+                    return next_call(request)
+                return request
+
             role = kwargs.get("role") or request.get("role")
             if not role and isinstance(request.get("metadata"), dict):
                 role = request["metadata"].get("role")
-            resolved_role = engine.resolve_role(request, explicit_role=role)
+
+            if provider and not role and not any(p.provider == provider for p in engine.config.profiles.values()):
+                logger.info("Explicit provider %r not managed by router; passing call downstream to Hermes", provider)
+                if callable(next_call):
+                    return next_call(request)
+                return request
+
+            allow_default_fallback = not provider or provider == "antigravity"
+            resolved_role, resolution_source = engine.resolve_role_with_source(
+                request,
+                explicit_role=role,
+                model=model,
+                provider=provider,
+                session_id=session_id,
+                fallback_to_default=allow_default_fallback,
+            )
 
             if not resolved_role:
                 if callable(next_call):
@@ -45,8 +73,16 @@ def antigravity_llm_execution(**kwargs: Any) -> Any:
                 return request
 
             if resolved_role:
-                session_id = kwargs.get("session_id") or request.get("session_id")
+                source_labels = {
+                    "explicit": "явная роль",
+                    "model_match": "по модели и провайдеру",
+                    "session_affinity": "по устойчивости сессии",
+                    "default_fallback": "роль по умолчанию",
+                }
+                source_label = source_labels.get(resolution_source, resolution_source)
+
                 completion = engine.route_request(request, role=resolved_role, session_id=session_id)
+
                 # Исчерпанная цепочка — это отказ роутера, а не ответ модели.
                 # Возвращать её текст Гермесу нельзя: он подменит собой настоящий
                 # ответ провайдера, который Гермес выбрал бы сам, и пользователь
@@ -58,9 +94,41 @@ def antigravity_llm_execution(**kwargs: Any) -> Any:
                         resolved_role,
                         completion.get("failover_trail"),
                     )
+                    try:
+                        from antigravity_provider.router.unified_health import EventLogService
+                        EventLogService.get().log(
+                            category="routing",
+                            message=f"Цепочка для роли '{resolved_role}' исчерпана; вызов передан штатному обработчику Hermes",
+                            details=f"Признак выбора роли: {source_label}",
+                            level="warning",
+                        )
+                    except Exception:
+                        pass
                     if callable(next_call):
                         return next_call(request)
                     return openai_completion_object(completion)
+
+                # Log successful Hermes routing event with factual role origin and profile
+                try:
+                    from antigravity_provider.router.unified_health import EventLogService
+                    meta = completion.get("router_metadata", {}) if isinstance(completion, dict) else {}
+                    served_profile = meta.get("profile_id", "default")
+                    served_provider = meta.get("provider", provider or "unknown")
+                    served_model = (
+                        meta.get("selected_model")
+                        or (meta.get("selection_trace") or {}).get("selected_model")
+                        or model
+                        or "default"
+                    )
+                    EventLogService.get().log(
+                        category="routing",
+                        message=f"Запрос Hermes направлен роли '{resolved_role}' ({source_label})",
+                        details=f"Профиль: '{served_profile}' ({served_provider}), модель: {served_model}",
+                        level="info",
+                    )
+                except Exception:
+                    pass
+
                 if isinstance(completion, dict) and "error" in completion and not completion.get("choices"):
                     err_text = format_antigravity_error(completion.get("error"))
                     completion = {

src/antigravity_provider/router/model_registry.py

@@ -231,12 +231,57 @@ class ModelRegistry:
                 quota_bucket="antigravity.gemini",
                 quality_tier=5,
             ),
+            ModelDescriptor(
+                model_id="google-antigravity/gemini-3.1-pro-high",
+                display_name="Gemini 3.1 Pro (High)",
+                provider="antigravity",
+                family="gemini",
+                capabilities=["coding", "reasoning", "tools", "structured_output", "long_context", "planning", "security_analysis", "developer-2", "reviewer"],
+                context_window=1000000,
+                supports_tools=True,
+                supports_reasoning=True,
+                latency_class="medium",
+                cost_input_per_m=1.25,
+                cost_output_per_m=5.0,
+                quota_bucket="antigravity.gemini",
+                quality_tier=5,
+            ),
+            ModelDescriptor(
+                model_id="google-antigravity/gemini-3.1-pro-low",
+                display_name="Gemini 3.1 Pro (Low)",
+                provider="antigravity",
+                family="gemini",
+                capabilities=["coding", "reasoning", "tools", "structured_output", "long_context", "developer-2"],
+                context_window=1000000,
+                supports_tools=True,
+                supports_reasoning=True,
+                latency_class="low",
+                cost_input_per_m=1.25,
+                cost_output_per_m=5.0,
+                quota_bucket="antigravity.gemini",
+                quality_tier=4,
+            ),
             ModelDescriptor(
                 model_id="google-antigravity/gemini-3.7-flash",
                 display_name="Gemini 3.7 Flash",
                 provider="antigravity",
                 family="gemini",
-                capabilities=["coding", "reasoning", "tools", "structured_output", "classification", "routing", "long_context"],
+                capabilities=["coding", "reasoning", "tools", "structured_output", "classification", "routing", "long_context", "developer-1"],
+                context_window=1000000,
+                supports_tools=True,
+                supports_reasoning=True,
+                latency_class="ultra_low",
+                cost_input_per_m=0.15,
+                cost_output_per_m=0.6,
+                quota_bucket="antigravity.gemini",
+                quality_tier=4,
+            ),
+            ModelDescriptor(
+                model_id="google-antigravity/gemini-3.7-flash-high",
+                display_name="Gemini 3.7 Flash (High)",
+                provider="antigravity",
+                family="gemini",
+                capabilities=["coding", "reasoning", "tools", "structured_output", "classification", "routing", "long_context", "developer-1"],
                 context_window=1000000,
                 supports_tools=True,
                 supports_reasoning=True,
@@ -262,6 +307,51 @@ class ModelRegistry:
                 quality_tier=5,
             ),
             # Google Antigravity (Claude family inside AGY)
+            ModelDescriptor(
+                model_id="google-antigravity/claude-opus-4-6-thinking",
+                display_name="Claude Opus 4.6 (Thinking)",
+                provider="antigravity",
+                family="claude",
+                capabilities=["coding", "reasoning", "tools", "structured_output", "security_analysis", "planning", "code-reviewer", "reviewer"],
+                context_window=200000,
+                supports_tools=True,
+                supports_reasoning=True,
+                latency_class="high",
+                cost_input_per_m=15.0,
+                cost_output_per_m=75.0,
+                quota_bucket="antigravity.claude",
+                quality_tier=5,
+            ),
+            ModelDescriptor(
+                model_id="google-antigravity/claude-opus-4-6",
+                display_name="Claude Opus 4.6 (AGY)",
+                provider="antigravity",
+                family="claude",
+                capabilities=["coding", "reasoning", "tools", "structured_output", "security_analysis", "planning", "code-reviewer", "reviewer"],
+                context_window=200000,
+                supports_tools=True,
+                supports_reasoning=True,
+                latency_class="high",
+                cost_input_per_m=15.0,
+                cost_output_per_m=75.0,
+                quota_bucket="antigravity.claude",
+                quality_tier=5,
+            ),
+            ModelDescriptor(
+                model_id="google-antigravity/claude-sonnet-4-6",
+                display_name="Claude Sonnet 4.6 (Thinking)",
+                provider="antigravity",
+                family="claude",
+                capabilities=["coding", "reasoning", "tools", "structured_output", "security_analysis", "planning"],
+                context_window=200000,
+                supports_tools=True,
+                supports_reasoning=True,
+                latency_class="medium",
+                cost_input_per_m=3.0,
+                cost_output_per_m=15.0,
+                quota_bucket="antigravity.claude",
+                quality_tier=5,
+            ),
             ModelDescriptor(
                 model_id="google-antigravity/claude-3-7-sonnet",
                 display_name="Claude 3.7 Sonnet (AGY)",

src/antigravity_provider/router/role_registry.py

@@ -52,7 +52,8 @@ CANONICAL_ROLES: Dict[str, RoleDefinition] = {
         is_implemented=True,
         capabilities=["coding", "developer-1", "reasoning", "tools", "structured_output"],
         fallback_capabilities=["coding"],
-        default_preferred_chain=["codex-worker-1", "claude-worker-1", "ag-w1", "opengo-1"],
+        default_preferred_chain=["ag-w1", "codex-worker-1", "claude-worker-1", "opengo-1"],
+        default_model="gemini-3.7-flash",
         max_failover_attempts=3,
         tier="core",
     ),
@@ -64,7 +65,8 @@ CANONICAL_ROLES: Dict[str, RoleDefinition] = {
         is_implemented=True,
         capabilities=["coding", "developer-2", "reviewer", "tools"],
         fallback_capabilities=["coding", "reviewer"],
-        default_preferred_chain=["ag-w1", "grok-worker-1", "codex-worker-2", "opengo-3"],
+        default_preferred_chain=["ag-w2", "codex-worker-2", "grok-worker-1", "opengo-3"],
+        default_model="gemini-3.1-pro-high",
         max_failover_attempts=3,
         tier="core",
     ),
@@ -76,7 +78,8 @@ CANONICAL_ROLES: Dict[str, RoleDefinition] = {
         is_implemented=True,
         capabilities=["code-reviewer", "reviewer", "coding", "security_analysis"],
         fallback_capabilities=["reviewer", "coding"],
-        default_preferred_chain=["claude-worker-2", "codex-worker-2", "ag-w2", "opengo-2"],
+        default_preferred_chain=["ag-w3", "ag-w2", "claude-worker-2", "codex-worker-2", "opengo-2"],
+        default_model="claude-opus-4-6-thinking",
         max_failover_attempts=3,
         tier="core",
     ),

src/antigravity_provider/router/router_config.py

@@ -36,7 +36,7 @@ class RolePolicy:
 @dataclass
 class RouterConfig:
     enabled: bool = True
-    default_role: str = "orchestrator"
+    default_role: str = "manager"
     quota_cooldown_seconds: int = 1800  # 30 min default
     rate_limit_cooldown_seconds: int = 60  # 1 min default
     max_failover_attempts: int = 3
@@ -277,7 +277,7 @@ def get_default_router_config() -> RouterConfig:
 
     return RouterConfig(
         enabled=True,
-        default_role="orchestrator",
+        default_role="manager",
         roles=roles,
         profiles=profiles,
     )
@@ -343,7 +343,8 @@ def load_router_config(config_path: Optional[Path] = None) -> RouterConfig:
                     }
 
         enabled = bool(r_block.get("enabled", data.get("enabled", True)))
-        default_role = str(r_block.get("default_role", data.get("default_role", "orchestrator")))
+        raw_default_role = str(r_block.get("default_role", data.get("default_role", "manager"))).strip().lower()
+        default_role = RoleRegistry.resolve_canonical_role(raw_default_role) if raw_default_role else "manager"
         max_failover = int(r_block.get("max_failover_attempts", data.get("max_failover_attempts", 3)))
         cooldown_base = int(r_block.get("cooldown_base_seconds", data.get("cooldown_base_seconds", 300)))
         cooldown_max = int(r_block.get("cooldown_max_seconds", data.get("cooldown_max_seconds", 3600)))

src/antigravity_provider/router/router_engine.py

@@ -35,21 +35,117 @@ class RouterEngine:
         if self.affinity and hasattr(self.affinity, "ttl_seconds"):
             self.affinity.ttl_seconds = self.config.session_affinity_ttl_seconds
 
-    def resolve_role(self, request: Dict[str, Any], explicit_role: Optional[str] = None) -> Optional[str]:
-        """Determine logical role from explicit parameter, request payload, or metadata.
+    def resolve_role_with_source(
+        self,
+        request: Dict[str, Any],
+        explicit_role: Optional[str] = None,
+        model: Optional[str] = None,
+        provider: Optional[str] = None,
+        session_id: Optional[str] = None,
+        fallback_to_default: bool = False,
+    ) -> Tuple[Optional[str], str]:
+        """Determine logical role and factual reason from parameters, configuration, or session.
         
-        Returns None if role cannot be reliably determined (no guessing from prompts).
+        Resolution order:
+        1. Explicit role in parameters, request, or metadata -> ("role", "explicit")
+        2. By model and provider matched dynamically against router configuration -> ("role", "model_match")
+        3. By session affinity for session_id -> ("role", "session_affinity")
+        4. Configurable default role -> ("default_role", "default_fallback") if fallback_to_default else (None, "none")
+        
+        Zero prompt guessing.
         """
-        if explicit_role:
-            return explicit_role.strip().lower()
-        if "role" in request and request["role"]:
-            return str(request["role"]).strip().lower()
-        if "personality" in request and request["personality"]:
-            return str(request["personality"]).strip().lower()
-        metadata = request.get("metadata", {})
-        if isinstance(metadata, dict) and metadata.get("role"):
-            return str(metadata["role"]).strip().lower()
-        return None
+        from .role_registry import RoleRegistry
+
+        # 1. Explicit role
+        raw_role = explicit_role
+        if not raw_role and "role" in request and request["role"]:
+            raw_role = str(request["role"])
+        if not raw_role and "personality" in request and request["personality"]:
+            raw_role = str(request["personality"])
+        if not raw_role:
+            metadata = request.get("metadata", {})
+            if isinstance(metadata, dict) and metadata.get("role"):
+                raw_role = str(metadata["role"])
+        if raw_role and str(raw_role).strip():
+            canon_role = RoleRegistry.resolve_canonical_role(str(raw_role).strip().lower())
+            return canon_role, "explicit"
+
+        # 2. By model and provider (derived dynamically from router config, zero hardcoded literals)
+        req_model = str(model or request.get("model") or "").strip()
+        req_prov = str(provider or request.get("provider") or "").strip().lower()
+
+        if req_model:
+            import re
+
+            def _clean_m(m: str) -> str:
+                return m.split("/")[-1].strip().lower()
+
+            def _base_m(m: str) -> str:
+                short = _clean_m(m)
+                return re.sub(r"-(high|medium|low|none|thought|thinking)(?:-(high|medium|low|none))?$", "", short)
+
+            req_m_clean = _clean_m(req_model)
+            req_m_base = _base_m(req_model)
+
+            # 2a. Direct match with role default_model
+            for rname, rpolicy in self.config.roles.items():
+                if rpolicy.default_model:
+                    def_m_clean = _clean_m(rpolicy.default_model)
+                    def_m_base = _base_m(rpolicy.default_model)
+                    if req_m_clean == def_m_clean or (req_m_base and req_m_base == def_m_base):
+                        if not req_prov:
+                            return RoleRegistry.resolve_canonical_role(rname), "model_match"
+                        primary_pcfg = self.config.get_profile(rpolicy.preferred_chain[0]) if rpolicy.preferred_chain else None
+                        if primary_pcfg and primary_pcfg.provider.lower() == req_prov:
+                            return RoleRegistry.resolve_canonical_role(rname), "model_match"
+
+            # 2b. Match with preferred_models of profiles in role's chain
+            for rname, rpolicy in self.config.roles.items():
+                for pid in rpolicy.preferred_chain:
+                    pcfg = self.config.get_profile(pid)
+                    if pcfg and pcfg.preferred_models:
+                        if req_prov and pcfg.provider.lower() != req_prov:
+                            continue
+                        for pm in pcfg.preferred_models:
+                            pm_clean = _clean_m(pm)
+                            pm_base = _base_m(pm)
+                            if req_m_clean == pm_clean or (req_m_base and req_m_base == pm_base):
+                                return RoleRegistry.resolve_canonical_role(rname), "model_match"
+
+        # 3. By session affinity
+        target_session = session_id or self.resolve_session_id(request)
+        if target_session and self.affinity:
+            aff_rec = self.affinity.get_affinity(target_session)
+            if aff_rec and getattr(aff_rec, "role", None):
+                return RoleRegistry.resolve_canonical_role(aff_rec.role), "session_affinity"
+
+        # 4. Default fallback role
+        if fallback_to_default:
+            from .settings_service import get_hub_settings
+            def_role = get_hub_settings().get("default_role") or self.config.default_role or "manager"
+            return RoleRegistry.resolve_canonical_role(str(def_role).strip().lower()), "default_fallback"
+
+        return None, "none"
+
+    def resolve_role(
+        self,
+        request: Dict[str, Any],
+        explicit_role: Optional[str] = None,
+        model: Optional[str] = None,
+        provider: Optional[str] = None,
+        session_id: Optional[str] = None,
+        fallback_to_default: bool = False,
+    ) -> Optional[str]:
+        """Determine logical role from explicit parameter, request payload, metadata, model/provider, or session."""
+        role, _ = self.resolve_role_with_source(
+            request=request,
+            explicit_role=explicit_role,
+            model=model,
+            provider=provider,
+            session_id=session_id,
+            fallback_to_default=fallback_to_default,
+        )
+        return role
 
     def resolve_session_id(self, request: Dict[str, Any], explicit_session_id: Optional[str] = None) -> Optional[str]:
         if explicit_session_id:
@@ -69,7 +165,15 @@ class RouterEngine:
         session_id: Optional[str] = None,
     ) -> Dict[str, Any]:
         """Execute request with session affinity and role-aware failover."""
-        target_role = self.resolve_role(request, role) or self.config.default_role
+        target_role, _ = self.resolve_role_with_source(
+            request,
+            explicit_role=role,
+            model=request.get("model"),
+            provider=request.get("provider"),
+            session_id=session_id,
+            fallback_to_default=True,
+        )
+        target_role = target_role or self.config.default_role or "manager"
         target_session = self.resolve_session_id(request, session_id)
         role_policy = self.config.get_role_policy(target_role)
 

src/antigravity_provider/router/settings_service.py

@@ -23,6 +23,7 @@ DEFAULT_SETTINGS: Dict[str, Any] = {
     "quota_threshold_percent": 10.0,
     "quota_threshold_action": "notify",
     "email_masking_mode": "none",
+    "default_role": "manager",
 }
 
 
@@ -105,6 +106,9 @@ def get_hub_settings() -> Dict[str, Any]:
         email_mode = "none"
     merged["email_masking_mode"] = email_mode
 
+    default_role = str(merged.get("default_role", "manager")).strip().lower()
+    merged["default_role"] = default_role or "manager"
+
     _SETTINGS_CACHE = dict(merged)
     _SETTINGS_CACHE_MTIME = current_mtime
     _SETTINGS_CACHE_PATH = sfile_str

src/antigravity_provider/router/workflow_service.py

@@ -1,13 +1,9 @@
-"""Persistent agents, workflow graph and live execution for Hermes Hub.
-
-The router role registry remains the source of truth for logical agents and
-Provider -> Account -> Model assignment.  This module adds the pieces that do
-not fit the routing schema: Agent Files, editor layout, workflow transitions,
-execution checkpoints and a bounded event journal.
-"""
+"""Hermes Hub A30 / A36 workflow runtime and persistence layer."""
 from __future__ import annotations
 
+import datetime
 import json
+import logging
 import re
 import threading
 import time
@@ -17,75 +13,94 @@ from pathlib import Path
 from typing import Any, Optional
 
 from antigravity_provider import paths
-from antigravity_provider.router.router_config import RolePolicy, load_router_config, save_router_config
+from antigravity_provider.router.router_config import (
+    RolePolicy,
+    load_router_config,
+    save_router_config,
+)
+
+logger = logging.getLogger("hermes.router.workflow")
 
 
-AGENT_STATES = {"waiting", "working", "reviewing", "error", "completed", "not_implemented"}
-EDGE_CONDITIONS = {"SUCCESS", "REVIEW_PASSED", "REVIEW_FAILED", "NEXT", "ERROR", "ALWAYS"}
+_AUTH_BEARER_RE = re.compile(r"Bearer\s+[A-Za-z0-9._~+/-]+", re.IGNORECASE)
+_ACCESS_TOKEN_PARAM_RE = re.compile(r"(access_token=)[^&]+", re.IGNORECASE)
+
+
+def sanitize_run_data(data: Any) -> Any:
+    """Sanitize data for saving to workflow_run_state.json, masking secrets and tokens."""
+    if isinstance(data, dict):
+        result = {}
+        for k, v in data.items():
+            k_lower = str(k).lower()
+            if any(s in k_lower for s in ["api_key", "token", "password", "secret", "client_secret", "jwt"]):
+                result[k] = "***"
+            elif isinstance(v, (dict, list)):
+                result[k] = sanitize_run_data(v)
+            elif isinstance(v, str):
+                s_val = _AUTH_BEARER_RE.sub("Bearer ***", v)
+                s_val = _ACCESS_TOKEN_PARAM_RE.sub(r"\1***", s_val)
+                result[k] = s_val
+            else:
+                result[k] = v
+        return result
+    elif isinstance(data, list):
+        return [sanitize_run_data(item) for item in data]
+    elif isinstance(data, str):
+        s_val = _AUTH_BEARER_RE.sub("Bearer ***", data)
+        s_val = _ACCESS_TOKEN_PARAM_RE.sub(r"\1***", s_val)
+        return s_val
+    return data
+
+
+def get_last_run_state(path: Optional[Path] = None) -> Optional[dict[str, Any]]:
+    """Return the last saved run state from workflow_run_state.json."""
+    target_path = path or paths.get_workflow_run_state_path()
+    if not target_path.is_file():
+        return None
+    try:
+        data = json.loads(target_path.read_text(encoding="utf-8"))
+        return data if isinstance(data, dict) else None
+    except Exception:
+        return None
+
+EDGE_CONDITIONS = {
+    "SUCCESS",
+    "ALWAYS",
+    "NEXT",
+    "REVIEW_PASSED",
+    "REVIEW_FAILED",
+    "ERROR",
+    "COMPLETED",
+    "ACCEPTED",
+}
 
 
 def _utc_timestamp() -> str:
-    import datetime
-
-    return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
 
 
 def _slug(value: str) -> str:
-    result = re.sub(r"[^a-z0-9]+", "-", value.strip().lower()).strip("-")
-    return result or f"agent-{uuid.uuid4().hex[:8]}"
+    cleaned = re.sub(r"[^a-zA-Z0-9_-]+", "-", value.strip()).strip("-").lower()
+    return cleaned or f"agent-{uuid.uuid4().hex[:6]}"
 
 
-def sanitize_run_data(node: Any) -> Any:
-    """Recursively strip or mask any credentials or secret keys from workflow run state."""
-    secret_key_substrings = [
-        "api_key", "token", "password", "secret", "jwt", "bearer",
-        "access_token", "refresh_token", "client_secret", "authorization",
-    ]
-    if isinstance(node, dict):
-        sanitized: dict[str, Any] = {}
-        for k, v in node.items():
-            k_lower = str(k).lower()
-            if any(s in k_lower for s in secret_key_substrings) and k_lower not in ("auth_status", "author", "auth_required"):
-                sanitized[k] = "***"
-            else:
-                sanitized[k] = sanitize_run_data(v)
-        return sanitized
-    elif isinstance(node, list):
-        return [sanitize_run_data(x) for x in node]
-    elif isinstance(node, str):
-        val = node
-        val = re.sub(r'Bearer\s+[a-zA-Z0-9_\-\.]{8,}', 'Bearer ***', val, flags=re.IGNORECASE)
-        val = re.sub(r'sk-[a-zA-Z0-9_\-]{8,}', 'sk-***', val)
-        val = re.sub(r'gho_[a-zA-Z0-9_\-]{8,}', 'gho_***', val)
-        val = re.sub(r'((?:access_token|refresh_token|api_key|token|password|secret|key)=)([^\s&,"]+)', r'\g<1>***', val, flags=re.IGNORECASE)
-        return val
-    return node
+def _safe_agent_file(path_str: str, agent_id: str) -> tuple[Path, str]:
+    home = paths.get_hermes_home().resolve()
+    if not path_str or not path_str.strip():
+        relative = f"agents/{_slug(agent_id)}.md"
+        return (home / relative).resolve(), relative
 
+    raw = Path(path_str.strip())
+    if raw.is_absolute():
+        resolved = raw.resolve()
+    else:
+        resolved = (home / raw).resolve()
 
-def get_last_run_state(run_state_path: Optional[Path] = None) -> Optional[dict[str, Any]]:
-    """Return the last saved workflow run state from workflow_run_state.json, if any."""
-    p = run_state_path or paths.get_workflow_run_state_path()
-    if not p.is_file():
-        return None
     try:
-        data = json.loads(p.read_text(encoding="utf-8"))
-        if isinstance(data, dict):
-            return sanitize_run_data(data)
-    except Exception:
-        pass
-    return None
-
-
-def _safe_agent_file(value: str, agent_id: str) -> tuple[Path, str]:
-    """Resolve an Agent File below HERMES_HOME/agents and reject traversal."""
-    root = paths.get_agent_files_dir().resolve()
-    candidate_name = Path(value or f"{agent_id}.md").name
-    if not candidate_name.lower().endswith(".md"):
-        candidate_name += ".md"
-    target = (root / candidate_name).resolve()
-    if target.parent != root:
-        raise ValueError("Agent File должен находиться в каталоге agents")
-    return target, f"agents/{candidate_name}"
+        rel = resolved.relative_to(home).as_posix()
+    except ValueError as exc:
+        raise ValueError("Agent File должен находиться внутри каталога HERMES_HOME") from exc
+    return resolved, rel
 
 
 @dataclass
@@ -112,6 +127,7 @@ class WorkflowEdge:
     target: str
     condition: str = "SUCCESS"
     label: str = ""
+    max_iterations: Optional[int] = None
 
 
 @dataclass
@@ -124,6 +140,62 @@ class WorkflowDefinition:
     start_agent_id: Optional[str] = None
 
 
+def get_canonical_a36_pipeline() -> WorkflowDefinition:
+    """Return canonical Antigravity 4-agent workflow with nested feedback loops."""
+    return WorkflowDefinition(
+        id="a36-pipeline",
+        name="Конвейер Antigravity (Оркестратор, 2 кодера, ревьюер)",
+        start_agent_id="manager",
+        max_iterations=5,
+        edges=[
+            WorkflowEdge(
+                id="edge-manager-to-dev1",
+                source="manager",
+                target="developer-1",
+                condition="SUCCESS",
+                label="Постановка задачи",
+            ),
+            WorkflowEdge(
+                id="edge-dev1-to-dev2",
+                source="developer-1",
+                target="developer-2",
+                condition="SUCCESS",
+                label="Реализация на проверку",
+            ),
+            WorkflowEdge(
+                id="edge-dev2-to-dev1",
+                source="developer-2",
+                target="developer-1",
+                condition="REVIEW_FAILED",
+                label="Доработка Кодеру 1",
+                max_iterations=5,
+            ),
+            WorkflowEdge(
+                id="edge-dev2-to-reviewer",
+                source="developer-2",
+                target="code-reviewer",
+                condition="REVIEW_PASSED",
+                label="Одобрено Кодером 2",
+            ),
+            WorkflowEdge(
+                id="edge-reviewer-to-dev2",
+                source="code-reviewer",
+                target="developer-2",
+                condition="REVIEW_FAILED",
+                label="Переделка Кодеру 2",
+                max_iterations=5,
+            ),
+            WorkflowEdge(
+                id="edge-reviewer-to-manager",
+                source="code-reviewer",
+                target="manager",
+                condition="REVIEW_PASSED",
+                label="Приёмка",
+            ),
+        ],
+    )
+
+
 @dataclass
 class WorkflowEvent:
     timestamp: str
@@ -206,12 +278,6 @@ class WorkflowService:
             return
         try:
             raw = json.loads(self.state_path.read_text(encoding="utf-8"))
-
-            # Идентификаторы агентов повторяют идентификаторы ролей, а роли со
-            # старыми именами переименовываются в канонические при загрузке
-            # конфигурации. Без такого же переименования здесь сохранённый
-            # workflow ссылался бы на исчезнувших агентов, и граф падал бы с
-            # «Ребро ссылается на отсутствующего агента».
             from antigravity_provider.router.role_registry import RoleRegistry
 
             def _canon(agent_id: str) -> str:
@@ -228,12 +294,18 @@ class WorkflowService:
                 item["id"] = _canon(item["id"])
                 if item.get("role"):
                     item["role"] = _canon(item["role"])
-                # Первым выигрывает агент под старым именем: именно им
-                # пользовался владелец, канонический мог быть дописан пустым.
                 self.agents.setdefault(item["id"], AgentDefinition(**item))
 
             wf = raw.get("workflow") or {}
             edges = []
+            valid_edge_keys = {
+                "id",
+                "source",
+                "target",
+                "condition",
+                "label",
+                "max_iterations",
+            }
             for edge in wf.pop("edges", []):
                 if not isinstance(edge, dict):
                     continue
@@ -241,7 +313,8 @@ class WorkflowService:
                 for key in ("source", "target", "from_agent", "to_agent"):
                     if edge.get(key):
                         edge[key] = _canon(edge[key])
-                edges.append(WorkflowEdge(**edge))
+                filtered_edge = {k: v for k, v in edge.items() if k in valid_edge_keys}
+                edges.append(WorkflowEdge(**filtered_edge))
             self.workflow = WorkflowDefinition(edges=edges, **wf)
             self.events = [WorkflowEvent(**event) for event in raw.get("events", [])[-200:]]
             self.run = raw.get("run") or self._idle_run()
@@ -317,8 +390,8 @@ class WorkflowService:
                 from antigravity_provider.router.role_registry import get_role_definition
 
                 definition = get_role_definition(role_id)
-                name = getattr(definition, "name", None) or getattr(definition, "display_name", None) or name
-                description = getattr(definition, "description", "")
+                name = getattr(definition, "display_name_ru", None) or getattr(definition, "name", None) or name
+                description = getattr(definition, "description_ru", "") or getattr(definition, "description", "")
             except (ImportError, AttributeError, TypeError):
                 pass
             agent = AgentDefinition(
@@ -332,12 +405,28 @@ class WorkflowService:
             self.agents[role_id] = agent
             self._ensure_file(target, agent)
             changed = True
+
+        if not self.workflow.edges:
+            a36_roles = {"manager", "developer-1", "developer-2", "code-reviewer"}
+            if a36_roles.issubset(self.agents.keys()):
+                self.workflow = get_canonical_a36_pipeline()
+                if "manager" in self.agents:
+                    self.agents["manager"].position = {"x": 60.0, "y": 140.0}
+                if "developer-1" in self.agents:
+                    self.agents["developer-1"].position = {"x": 320.0, "y": 140.0}
+                if "developer-2" in self.agents:
+                    self.agents["developer-2"].position = {"x": 580.0, "y": 140.0}
+                if "code-reviewer" in self.agents:
+                    self.agents["code-reviewer"].position = {"x": 840.0, "y": 140.0}
+                changed = True
+
         if changed:
             self._save()
 
     @staticmethod
     def _ensure_file(target: Path, agent: AgentDefinition) -> None:
         if not target.exists():
+            target.parent.mkdir(parents=True, exist_ok=True)
             body = f"# {agent.name}\n\n## Роль\n\n{agent.role}\n\n## Назначение\n\n{agent.description or 'Инструкции ещё не заполнены.'}\n"
             target.write_text(body, encoding="utf-8")
 
@@ -390,6 +479,7 @@ class WorkflowService:
         with self._lock:
             agent = self._require_agent(agent_id)
             target, relative = _safe_agent_file(agent.agent_file, agent.id)
+            target.parent.mkdir(parents=True, exist_ok=True)
             temporary = target.with_suffix(".md.tmp")
             temporary.write_text(str(content), encoding="utf-8")
             temporary.replace(target)
@@ -404,6 +494,8 @@ class WorkflowService:
         if not name:
             raise ValueError("Укажите название агента")
         with self._lock:
+            if self.run.get("status") in {"running", "stopping"}:
+                raise ValueError("Нельзя менять конфигурацию агентов во время выполнения workflow")
             if agent_id in self.agents:
                 raise ValueError("Агент с таким идентификатором уже существует")
             profile_id = str(data.get("account") or data.get("profile_id") or "").strip()
@@ -446,6 +538,8 @@ class WorkflowService:
 
     def update_agent(self, agent_id: str, data: dict[str, Any]) -> AgentDefinition:
         with self._lock:
+            if self.run.get("status") in {"running", "stopping"}:
+                raise ValueError("Нельзя менять конфигурацию агентов во время выполнения workflow")
             agent = self._require_agent(agent_id)
             config = load_router_config()
             policy = config.roles.get(agent.role)
@@ -463,10 +557,11 @@ class WorkflowService:
                 policy.preferred_chain = [profile_id] + [item for item in policy.preferred_chain if item != profile_id]
             if "model" in data:
                 model = str(data.get("model") or "").strip() or None
-                if model and profile_id:
-                    profile = config.profiles[profile_id]
-                    if model not in profile.preferred_models:
-                        raise ValueError("Модель не доступна выбранному аккаунту")
+                if model and policy.preferred_chain:
+                    eff_profile_id = policy.preferred_chain[0]
+                    eff_profile = config.profiles.get(eff_profile_id)
+                    if eff_profile and eff_profile.preferred_models and model not in eff_profile.preferred_models:
+                        raise ValueError(f"Модель {model} не доступна для профиля {eff_profile_id}")
                 policy.default_model = model
             if not save_router_config(config):
                 raise OSError("Не удалось сохранить назначение агента")
@@ -487,6 +582,8 @@ class WorkflowService:
 
     def delete_agent(self, agent_id: str, force: bool = False) -> dict[str, Any]:
         with self._lock:
+            if self.run.get("status") in {"running", "stopping"}:
+                raise ValueError("Нельзя удалять агентов во время выполнения workflow")
             agent = self._require_agent(agent_id)
             edge_ids = [edge.id for edge in self.workflow.edges if edge.source == agent_id or edge.target == agent_id]
             route_used = bool(load_router_config().roles.get(agent.role))
@@ -498,7 +595,17 @@ class WorkflowService:
             if not save_router_config(config):
                 raise OSError("Не удалось удалить роль из маршрутизатора")
             self.workflow.edges = [edge for edge in self.workflow.edges if edge.id not in edge_ids]
-            self.agents.pop(agent_id)
+            if self.workflow.start_agent_id == agent_id:
+                self.workflow.start_agent_id = None
+            if self.workflow.escalation_agent_id == agent_id:
+                self.workflow.escalation_agent_id = None
+            target, _ = _safe_agent_file(agent.agent_file, agent.id)
+            if target.exists():
+                try:
+                    target.unlink()
+                except OSError:
+                    pass
+            self.agents.pop(agent_id, None)
             self._event("AGENT_DELETED", f"Удалён агент «{agent.name}»", agent_id=agent_id, level="warning")
             self._save()
             return {"deleted": True, "consequences": consequences}
@@ -509,18 +616,35 @@ class WorkflowService:
                 raise ValueError("Нельзя менять граф в режиме LIVE во время выполнения")
             edges: list[WorkflowEdge] = []
             seen: set[str] = set()
-            for raw in data.get("edges", []):
-                source, target = str(raw.get("source") or ""), str(raw.get("target") or "")
-                condition = str(raw.get("condition") or "SUCCESS").upper()
+            for raw_edge in data.get("edges", []):
+                source, target = str(raw_edge.get("source") or ""), str(raw_edge.get("target") or "")
+                condition = str(raw_edge.get("condition") or "SUCCESS").upper()
                 if source not in self.agents or target not in self.agents:
                     raise ValueError("Ребро ссылается на отсутствующего агента")
                 if condition not in EDGE_CONDITIONS:
                     raise ValueError(f"Неизвестное условие перехода: {condition}")
-                edge_id = str(raw.get("id") or f"edge-{uuid.uuid4().hex[:10]}")
+                edge_id = str(raw_edge.get("id") or f"edge-{uuid.uuid4().hex[:10]}")
                 if edge_id in seen:
                     raise ValueError("Идентификаторы рёбер должны быть уникальны")
                 seen.add(edge_id)
-                edges.append(WorkflowEdge(edge_id, source, target, condition, str(raw.get("label") or "")))
+                edge_max_it = raw_edge.get("max_iterations")
+                if edge_max_it is not None:
+                    try:
+                        edge_max_it = int(edge_max_it)
+                        if not 1 <= edge_max_it <= 100:
+                            edge_max_it = None
+                    except (ValueError, TypeError):
+                        edge_max_it = None
+                edges.append(
+                    WorkflowEdge(
+                        id=edge_id,
+                        source=source,
+                        target=target,
+                        condition=condition,
+                        label=str(raw_edge.get("label") or ""),
+                        max_iterations=edge_max_it,
+                    )
+                )
             max_iterations = int(data.get("max_iterations") or self.workflow.max_iterations)
             if not 1 <= max_iterations <= 100:
                 raise ValueError("Предел итераций должен быть от 1 до 100")
@@ -577,7 +701,13 @@ class WorkflowService:
             self._stop.set()
             self._event("WORKFLOW_STOP_REQUESTED", "Запрошена остановка workflow", level="warning")
             self._save()
-            self._save_run_state("STOPPED", step_index=len(self._completed_steps), current_agent=self.run.get("current_agent_id"), iteration=self.run.get("iteration", 1), interruption_reason="Остановлено пользователем")
+            self._save_run_state(
+                "STOPPED",
+                step_index=len(self._completed_steps),
+                current_agent=self.run.get("current_agent_id"),
+                iteration=self.run.get("iteration", 1),
+                interruption_reason="Остановлено пользователем",
+            )
             return dict(self.run)
 
     def _execute(self) -> None:
@@ -587,6 +717,7 @@ class WorkflowService:
         context = str(self.run.get("current_task") or "")
         current = str(self.run.get("current_agent_id") or "")
         visited: dict[str, int] = {}
+        edge_counts: dict[str, int] = {}
         try:
             engine = get_router_engine()
             engine.reload_config()
@@ -594,8 +725,14 @@ class WorkflowService:
                 with self._lock:
                     agent = self._require_agent(current)
                     visited[current] = visited.get(current, 0) + 1
-                    iteration = max(visited.values())
-                    self.run.update({"current_agent_id": current, "iteration": iteration})
+                    iteration = visited[current]
+                    global_iteration = sum(visited.values())
+                    self.run.update({
+                        "current_agent_id": current,
+                        "iteration": iteration,
+                        "global_iteration": global_iteration,
+                        "max_iterations": self.workflow.max_iterations,
+                    })
                     step_idx = len(self._completed_steps)
                     if iteration > self.workflow.max_iterations:
                         message = f"Достигнут предел итераций: {self.workflow.max_iterations}"
@@ -605,7 +742,7 @@ class WorkflowService:
                         break
                     file_data = self.read_agent_file(current)
                     if not file_data["exists"]:
-                        raise FileNotFoundError(f"{file_data['path']}: {file_data['reason']}")
+                        raise FileNotFoundError(f"{file_data['path']}: {file_data.get('reason')}")
                     self.run.setdefault("agent_states", {})[current] = (
                         "reviewing" if "review" in agent.role.lower() else "working"
                     )
@@ -665,7 +802,7 @@ class WorkflowService:
                         (item for item in self.workflow.edges if item.source == current and item.condition in {status, "ALWAYS"}),
                         None,
                     )
-                    if not edge and status not in {"ERROR", "REVIEW_FAILED"}:
+                    if not edge and status in {"SUCCESS", "NEXT"}:
                         edge = next(
                             (item for item in self.workflow.edges if item.source == current and item.condition in {"SUCCESS", "NEXT"}),
                             None,
@@ -703,6 +840,16 @@ class WorkflowService:
                             interruption_reason=self.run.get("error") if self.run["status"] == "failed" else None,
                         )
                         break
+
+                    edge_counts[edge.id] = edge_counts.get(edge.id, 0) + 1
+                    edge_limit = edge.max_iterations if edge.max_iterations is not None else self.workflow.max_iterations
+                    if edge_counts[edge.id] > edge_limit:
+                        message = f"Достигнут предел итераций для перехода {edge.source} → {edge.target}: {edge_limit}"
+                        self.run.update({"status": "failed", "error": message})
+                        self._event("WORKFLOW_MAX_ITERATIONS", message, level="error", agent_id=current, iteration=iteration)
+                        self._save_run_state("FAILED", step_index=step_idx + 1, current_agent=current, iteration=iteration, interruption_reason=message)
+                        break
+
                     self._event(
                         "WORKFLOW_TRANSITION",
                         f"Переход {current} → {edge.target}: {edge.condition}",
@@ -746,7 +893,6 @@ class WorkflowService:
                 self.run["current_agent_id"] = current or self.run.get("current_agent_id")
                 self._save()
 
-
     @staticmethod
     def _response_text(response: Any) -> str:
         if not isinstance(response, dict):
@@ -763,9 +909,26 @@ class WorkflowService:
             explicit = response.get("status") or response.get("structured_status")
             if explicit and str(explicit).upper() in EDGE_CONDITIONS:
                 return str(explicit).upper()
-        for status in ("REVIEW_FAILED", "REVIEW_PASSED", "SUCCESS", "ERROR"):
-            if re.search(rf"\b{status}\b", text.upper()):
+
+        text_upper = text.upper()
+        # 1. Match leading status keyword or STATUS: <keyword>
+        prefix_match = re.search(
+            r"^\s*(?:STATUS\s*:\s*|\[STATUS\s*:\s*)?(REVIEW_FAILED|REVIEW_PASSED|COMPLETED|ACCEPTED|SUCCESS|ERROR)\b",
+            text_upper,
+            re.MULTILINE,
+        )
+        if prefix_match:
+            return prefix_match.group(1)
+
+        # 2. Match multi-word distinct statuses anywhere
+        for status in ("REVIEW_FAILED", "REVIEW_PASSED", "COMPLETED", "ACCEPTED"):
+            if re.search(rf"\b{status}\b", text_upper):
                 return status
+
+        # 3. Explicit error status markers vs regular discussion of errors
+        if re.search(r"\bSTATUS\s*:\s*ERROR\b", text_upper) or re.search(r"\b\[ERROR\]\b", text_upper):
+            return "ERROR"
+
         return "SUCCESS"
 
     def _event(self, event_type: str, message: str, **kwargs: Any) -> None:

tests/test_a35_role_resolution.py

@@ -0,0 +1,271 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from antigravity_provider.router.router_config import (
+    RolePolicy,
+    RouterConfig,
+    RouterProfileConfig,
+    save_router_config,
+)
+from antigravity_provider.router.router_engine import RouterEngine, get_router_engine
+from antigravity_provider.hermes_plugin import antigravity_llm_execution
+from antigravity_provider.router.settings_service import save_hub_settings, invalidate_settings_cache
+
+
+class TestA35RoleResolution(unittest.TestCase):
+    """P0-1 & P0-2: Role resolution and failover safety for Hermes Hub integration."""
+
+    def setUp(self):
+        self.tmp_dir = tempfile.mkdtemp(prefix="hermes_test_a35_")
+        self.config_path = Path(self.tmp_dir) / "router_profiles.yaml"
+        self.settings_path = Path(self.tmp_dir) / "hub_settings.json"
+
+        self.profiles = {
+            "ag-orch-fallback": RouterProfileConfig(
+                profile_id="ag-orch-fallback",
+                provider="antigravity",
+                preferred_models=["gemini-3.7-flash"],
+            ),
+            "ag-w1": RouterProfileConfig(
+                profile_id="ag-w1",
+                provider="antigravity",
+                preferred_models=["gemini-3.7-flash"],
+            ),
+            "ag-w2": RouterProfileConfig(
+                profile_id="ag-w2",
+                provider="antigravity",
+                preferred_models=["gemini-3.1-pro-high"],
+            ),
+            "ag-w3": RouterProfileConfig(
+                profile_id="ag-w3",
+                provider="antigravity",
+                preferred_models=["claude-opus-4-6-thinking"],
+            ),
+        }
+
+        self.roles = {
+            "manager": RolePolicy(
+                role_name="manager",
+                preferred_chain=["ag-orch-fallback"],
+                default_model="gemini-3.7-flash",
+            ),
+            "developer-1": RolePolicy(
+                role_name="developer-1",
+                preferred_chain=["ag-w1"],
+                default_model="gemini-3.7-flash",
+            ),
+            "developer-2": RolePolicy(
+                role_name="developer-2",
+                preferred_chain=["ag-w2"],
+                default_model="gemini-3.1-pro-high",
+            ),
+            "code-reviewer": RolePolicy(
+                role_name="code-reviewer",
+                preferred_chain=["ag-w3"],
+                default_model="claude-opus-4-6-thinking",
+            ),
+        }
+
+        self.config = RouterConfig(
+            enabled=True,
+            default_role="manager",
+            roles=self.roles,
+            profiles=self.profiles,
+        )
+        save_router_config(self.config, self.config_path)
+
+        self.env_patcher = patch.dict(
+            "os.environ",
+            {
+                "HERMES_HOME": self.tmp_dir,
+                "HERMES_ROUTER_PROFILES": str(self.config_path),
+            },
+        )
+        self.env_patcher.start()
+        invalidate_settings_cache()
+        self.engine = RouterEngine(self.config)
+
+    def tearDown(self):
+        self.env_patcher.stop()
+        invalidate_settings_cache()
+
+    def test_explicit_role_resolution_level1(self):
+        """Level 1: Explicit role in arguments, request or metadata takes highest priority."""
+        # 1. Via explicit_role arg
+        role, source = self.engine.resolve_role_with_source({}, explicit_role="developer-2")
+        self.assertEqual(role, "developer-2")
+        self.assertEqual(source, "explicit")
+
+        # 2. Via request['role']
+        role, source = self.engine.resolve_role_with_source({"role": "code-reviewer"})
+        self.assertEqual(role, "code-reviewer")
+        self.assertEqual(source, "explicit")
+
+        # 3. Via request['metadata']['role']
+        role, source = self.engine.resolve_role_with_source({"metadata": {"role": "manager"}})
+        self.assertEqual(role, "manager")
+        self.assertEqual(source, "explicit")
+
+    def test_model_and_provider_resolution_level2(self):
+        """Level 2: Resolution by model and provider dynamically configured in router roles."""
+        # gemini-3.1-pro-high -> configured default_model for developer-2
+        role, source = self.engine.resolve_role_with_source({}, model="gemini-3.1-pro-high")
+        self.assertEqual(role, "developer-2")
+        self.assertEqual(source, "model_match")
+
+        # claude-opus-4-6-thinking -> configured default_model for code-reviewer
+        role, source = self.engine.resolve_role_with_source({}, model="claude-opus-4-6-thinking")
+        self.assertEqual(role, "code-reviewer")
+        self.assertEqual(source, "model_match")
+
+        # gemini-3.7-flash -> configured default_model for manager / developer-1
+        role, source = self.engine.resolve_role_with_source({}, model="gemini-3.7-flash")
+        self.assertIn(role, ["manager", "developer-1"])
+        self.assertEqual(source, "model_match")
+
+    def test_session_affinity_resolution_level3(self):
+        """Level 3: Resolution by session affinity for session_id."""
+        sess_id = "sess-affinity-test-123"
+        # Register affinity record in router engine
+        self.engine.affinity.set_affinity(sess_id, role="developer-2", profile_id="ag-w2")
+
+        role, source = self.engine.resolve_role_with_source({}, session_id=sess_id)
+        self.assertEqual(role, "developer-2")
+        self.assertEqual(source, "session_affinity")
+
+    def test_default_fallback_role_level4(self):
+        """Level 4: Default fallback role when no explicit, model, or session affinity matched."""
+        # By default, default_fallback gives 'manager'
+        role, source = self.engine.resolve_role_with_source(
+            {"messages": [{"role": "user", "content": "hello"}]},
+            fallback_to_default=True,
+        )
+        self.assertEqual(role, "manager")
+        self.assertEqual(source, "default_fallback")
+
+        # Configurable via hub_settings.json
+        save_hub_settings({"default_role": "code-reviewer"})
+        role, source = self.engine.resolve_role_with_source(
+            {"messages": [{"role": "user", "content": "hello"}]},
+            fallback_to_default=True,
+        )
+        self.assertEqual(role, "code-reviewer")
+        self.assertEqual(source, "default_fallback")
+
+    def test_no_prompt_guessing_returns_none_when_fallback_disabled(self):
+        """Zero prompt guessing: unspecified role returns None when fallback_to_default=False."""
+        req = {"messages": [{"role": "system", "content": "You are a senior coding agent developer"}]}
+        role, source = self.engine.resolve_role_with_source(req, fallback_to_default=False)
+        self.assertIsNone(role)
+        self.assertEqual(source, "none")
+
+    def test_router_error_never_returned_as_assistant_content_to_hermes(self):
+        """Safety Fuse: Router failover exhaustion (router_error) passes call downstream to next_call."""
+        downstream_calls = []
+
+        def mock_next(req):
+            downstream_calls.append(req)
+            return {
+                "choices": [
+                    {
+                        "message": {"role": "assistant", "content": "clean-downstream-response"},
+                        "finish_reason": "stop",
+                    }
+                ]
+            }
+
+        # Mock engine route_request to return failover exhaustion error payload
+        exhausted_payload = {
+            "router_error": True,
+            "error_type": "exhausted",
+            "message": "All 3 profiles in role 'manager' failed",
+            "failover_trail": ["ag-orch-fallback: timeout"],
+        }
+
+        with patch("antigravity_provider.router.get_router_engine", return_value=self.engine):
+            with patch.object(self.engine, "route_request", return_value=exhausted_payload):
+                res = antigravity_llm_execution(
+                    request={"messages": [{"role": "user", "content": "test safety"}]},
+                    next_call=mock_next,
+                    provider="antigravity",
+                    model="gemini-3.7-flash",
+                    session_id="sess-safety-1",
+                )
+
+        # Verified: Downstream call was executed and router error text did NOT become the assistant message
+        self.assertEqual(len(downstream_calls), 1)
+        content = res["choices"][0]["message"]["content"]
+        self.assertEqual(content, "clean-downstream-response")
+        self.assertNotIn("exhausted", content.lower())
+        self.assertNotIn("router_error", content.lower())
+
+    def test_empty_profile_config_passes_cleanly_to_next_call(self):
+        """Empty profile configuration must not break Hermes and falls through cleanly."""
+        empty_config = RouterConfig(enabled=True, roles={}, profiles={})
+        empty_engine = RouterEngine(empty_config)
+
+        downstream_calls = []
+
+        def mock_next(req):
+            downstream_calls.append(req)
+            return {
+                "choices": [
+                    {
+                        "message": {"role": "assistant", "content": "clean-passthrough-response"},
+                        "finish_reason": "stop",
+                    }
+                ]
+            }
+
+        with patch("antigravity_provider.router.get_router_engine", return_value=empty_engine):
+            res = antigravity_llm_execution(
+                request={"messages": [{"role": "user", "content": "test passthrough"}]},
+                next_call=mock_next,
+                provider="antigravity",
+                model="gemini-3.7-flash",
+            )
+
+        self.assertEqual(len(downstream_calls), 1)
+        self.assertEqual(res["choices"][0]["message"]["content"], "clean-passthrough-response")
+
+    def test_hermes_call_records_event_log_and_telemetry(self):
+        """Successful Hermes routing records chosen role, reason, profile, and telemetry."""
+        from antigravity_provider.router.unified_health import EventLogService
+
+        mock_completion = {
+            "choices": [{"message": {"role": "assistant", "content": "model-output-ok"}}],
+            "router_metadata": {
+                "provider": "antigravity",
+                "profile_id": "ag-w2",
+                "selected_model": "gemini-3.1-pro-high",
+            },
+        }
+
+        with patch("antigravity_provider.router.get_router_engine", return_value=self.engine):
+            with patch.object(self.engine, "route_request", return_value=mock_completion):
+                res = antigravity_llm_execution(
+                    request={"messages": [{"role": "user", "content": "test event logging"}]},
+                    provider="antigravity",
+                    model="gemini-3.1-pro-high",
+                    session_id="sess-log-1",
+                )
+
+        content = (
+            res.choices[0].message.content
+            if hasattr(res, "choices")
+            else res["choices"][0]["message"]["content"]
+        )
+        self.assertEqual(content, "model-output-ok")
+        events = EventLogService.get().get_events(limit=10, category="routing")
+        self.assertTrue(any("developer-2" in e.message for e in events))
+        matching_event = next(e for e in events if "developer-2" in e.message)
+        self.assertIn("по модели и провайдеру", matching_event.message)
+        self.assertIn("ag-w2", matching_event.details)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_a36_pipeline.py

@@ -0,0 +1,308 @@
+from __future__ import annotations
+
+import json
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from antigravity_provider.router.router_config import (
+    RolePolicy,
+    RouterConfig,
+    RouterProfileConfig,
+    save_router_config,
+)
+from antigravity_provider.router.workflow_service import (
+    WorkflowService,
+    WorkflowDefinition,
+    WorkflowEdge,
+    get_canonical_a36_pipeline,
+)
+
+
+class TestA36AntigravityPipeline(unittest.TestCase):
+    """P0-1 .. P0-5: Antigravity pipeline graph, loops, limits, models and execution."""
+
+    def setUp(self):
+        self.tmp_dir = tempfile.mkdtemp(prefix="hermes_test_a36_")
+        self.config_path = Path(self.tmp_dir) / "router_profiles.yaml"
+        self.state_path = Path(self.tmp_dir) / "workflow_state.json"
+
+        self.profiles = {
+            "ag-orch-fallback": RouterProfileConfig(
+                profile_id="ag-orch-fallback",
+                provider="antigravity",
+                preferred_models=["gemini-3.7-flash"],
+            ),
+            "ag-w1": RouterProfileConfig(
+                profile_id="ag-w1",
+                provider="antigravity",
+                preferred_models=["gemini-3.7-flash", "gemini-3.7-flash-high"],
+            ),
+            "ag-w2": RouterProfileConfig(
+                profile_id="ag-w2",
+                provider="antigravity",
+                preferred_models=["gemini-3.1-pro-high", "gemini-3.1-pro-low"],
+            ),
+            "ag-w3": RouterProfileConfig(
+                profile_id="ag-w3",
+                provider="antigravity",
+                preferred_models=["claude-opus-4-6-thinking"],
+            ),
+        }
+
+        self.roles = {
+            "manager": RolePolicy(
+                role_name="manager",
+                preferred_chain=["ag-orch-fallback"],
+                default_model="gemini-3.7-flash",
+            ),
+            "developer-1": RolePolicy(
+                role_name="developer-1",
+                preferred_chain=["ag-w1"],
+                default_model="gemini-3.7-flash",
+            ),
+            "developer-2": RolePolicy(
+                role_name="developer-2",
+                preferred_chain=["ag-w2"],
+                default_model="gemini-3.1-pro-high",
+            ),
+            "code-reviewer": RolePolicy(
+                role_name="code-reviewer",
+                preferred_chain=["ag-w3"],
+                default_model="claude-opus-4-6-thinking",
+            ),
+        }
+
+        self.config = RouterConfig(
+            enabled=True,
+            default_role="manager",
+            roles=self.roles,
+            profiles=self.profiles,
+        )
+        save_router_config(self.config, self.config_path)
+
+        self.env_patcher = patch.dict(
+            "os.environ",
+            {
+                "HERMES_HOME": self.tmp_dir,
+                "HERMES_ROUTER_PROFILES": str(self.config_path),
+            },
+        )
+        self.env_patcher.start()
+        self.wf_service = WorkflowService(self.state_path)
+
+    def tearDown(self):
+        self.env_patcher.stop()
+
+    def test_canonical_pipeline_graph_structure(self):
+        """P0-1: Canonical graph matches orchestrator, two coders, reviewer layout."""
+        snapshot = self.wf_service.snapshot()
+        agents = {a["id"]: a for a in snapshot["agents"]}
+
+        # 1. Check all 4 agents exist
+        self.assertIn("manager", agents)
+        self.assertIn("developer-1", agents)
+        self.assertIn("developer-2", agents)
+        self.assertIn("code-reviewer", agents)
+
+        # 2. Check model bindings
+        self.assertEqual(agents["manager"]["execution_config"]["model"], "gemini-3.7-flash")
+        self.assertEqual(agents["developer-1"]["execution_config"]["model"], "gemini-3.7-flash")
+        self.assertEqual(agents["developer-2"]["execution_config"]["model"], "gemini-3.1-pro-high")
+        self.assertEqual(agents["code-reviewer"]["execution_config"]["model"], "claude-opus-4-6-thinking")
+
+        # 3. Check account bindings
+        self.assertEqual(agents["manager"]["execution_config"]["account"], "ag-orch-fallback")
+        self.assertEqual(agents["developer-1"]["execution_config"]["account"], "ag-w1")
+        self.assertEqual(agents["developer-2"]["execution_config"]["account"], "ag-w2")
+        self.assertEqual(agents["code-reviewer"]["execution_config"]["account"], "ag-w3")
+
+        # 4. Check edges and feedback loops
+        definition = snapshot["definition"]
+        self.assertEqual(definition["start_agent_id"], "manager")
+        edges = {(e["source"], e["target"], e["condition"]) for e in definition["edges"]}
+
+        # Forward flow
+        self.assertIn(("manager", "developer-1", "SUCCESS"), edges)
+        self.assertIn(("developer-1", "developer-2", "SUCCESS"), edges)
+        self.assertIn(("developer-2", "code-reviewer", "REVIEW_PASSED"), edges)
+        self.assertIn(("code-reviewer", "manager", "REVIEW_PASSED"), edges)
+
+        # Inner feedback loop: developer-2 -> developer-1 on REVIEW_FAILED
+        self.assertIn(("developer-2", "developer-1", "REVIEW_FAILED"), edges)
+
+        # Outer feedback loop: code-reviewer -> developer-2 (NOT developer-1!) on REVIEW_FAILED
+        self.assertIn(("code-reviewer", "developer-2", "REVIEW_FAILED"), edges)
+
+    def test_live_execution_with_triggered_inner_loop(self):
+        """P0-3: Live execution run with Coder 2 returning work to Coder 1."""
+        step_call_count = {"developer-1": 0, "developer-2": 0, "code-reviewer": 0, "manager": 0}
+        step_history = []
+
+        class FakePipelineEngine:
+            def reload_config(self):
+                return None
+
+            def route_request(self, request, role=None, session_id=None):
+                step_call_count[role] = step_call_count.get(role, 0) + 1
+                step_history.append((role, step_call_count[role]))
+
+                if role == "manager":
+                    if step_call_count["manager"] == 1:
+                        return {
+                            "choices": [{"message": {"content": "SUCCESS: Task dispatched to Developer 1"}}],
+                            "router_metadata": {"profile_id": "ag-orch-fallback", "selected_model": "gemini-3.7-flash"},
+                        }
+                    else:
+                        return {
+                            "choices": [{"message": {"content": "ACCEPTED: Project verified and accepted by Orchestrator"}}],
+                            "router_metadata": {"profile_id": "ag-orch-fallback", "selected_model": "gemini-3.7-flash"},
+                        }
+                elif role == "developer-1":
+                    if step_call_count["developer-1"] == 1:
+                        return {
+                            "choices": [{"message": {"content": "SUCCESS: Initial code implementation"}}],
+                            "router_metadata": {"profile_id": "ag-w1", "selected_model": "gemini-3.7-flash"},
+                        }
+                    else:
+                        return {
+                            "choices": [{"message": {"content": "SUCCESS: Fixed error handling per Coder 2 feedback"}}],
+                            "router_metadata": {"profile_id": "ag-w1", "selected_model": "gemini-3.7-flash"},
+                        }
+                elif role == "developer-2":
+                    if step_call_count["developer-2"] == 1:
+                        # First check: Reject and return to developer-1
+                        return {
+                            "choices": [{"message": {"content": "REVIEW_FAILED: Missing error handling and edge cases"}}],
+                            "router_metadata": {"profile_id": "ag-w2", "selected_model": "gemini-3.1-pro-high"},
+                        }
+                    else:
+                        # Second check: Approve and advance to reviewer
+                        return {
+                            "choices": [{"message": {"content": "REVIEW_PASSED: Code approved by Coder 2"}}],
+                            "router_metadata": {"profile_id": "ag-w2", "selected_model": "gemini-3.1-pro-high"},
+                        }
+                elif role == "code-reviewer":
+                    return {
+                        "choices": [{"message": {"content": "REVIEW_PASSED: Security and architecture approved"}}],
+                        "router_metadata": {"profile_id": "ag-w3", "selected_model": "claude-opus-4-6-thinking"},
+                    }
+                return {"choices": [{"message": {"content": "SUCCESS"}}]}
+
+        with patch("antigravity_provider.router.router_engine.get_router_engine", return_value=FakePipelineEngine()):
+            self.wf_service.start("Создать модуль аутентификации с валидацией токенов")
+            thread = self.wf_service._thread
+            self.assertIsNotNone(thread)
+            thread.join(timeout=5)
+            self.assertFalse(thread.is_alive(), "Workflow thread hung during execution")
+
+        # Verify completed execution status
+        self.assertEqual(
+            self.wf_service.run["status"],
+            "completed",
+            f"Run failed with error: {self.wf_service.run.get('error')}, events: {[e.message for e in self.wf_service.events]}",
+        )
+
+        # Verify transition sequence
+        expected_sequence = [
+            ("manager", 1),
+            ("developer-1", 1),
+            ("developer-2", 1),  # Returns REVIEW_FAILED -> triggers loop back to dev-1
+            ("developer-1", 2),  # dev-1 fixes code
+            ("developer-2", 2),  # dev-2 approves -> REVIEW_PASSED
+            ("code-reviewer", 1),  # reviewer approves -> REVIEW_PASSED
+            ("manager", 2),  # manager acceptance -> completed
+        ]
+        self.assertEqual(step_history, expected_sequence)
+
+        # Verify transition events
+        transitions = [e.message for e in self.wf_service.events if e.type == "WORKFLOW_TRANSITION"]
+        self.assertIn("Переход developer-2 → developer-1: REVIEW_FAILED", transitions)
+        self.assertIn("Переход developer-2 → code-reviewer: REVIEW_PASSED", transitions)
+        self.assertIn("Переход code-reviewer → manager: REVIEW_PASSED", transitions)
+
+    def test_iteration_limit_cutoff_and_event(self):
+        """P0-2: Loop iteration cutoff emits WORKFLOW_MAX_ITERATIONS and sets failed status."""
+        self.wf_service.workflow.max_iterations = 2
+        self.wf_service._save()
+
+        class InfiniteLoopEngine:
+            def reload_config(self):
+                return None
+
+            def route_request(self, request, role=None, session_id=None):
+                if role == "manager":
+                    return {"choices": [{"message": {"content": "SUCCESS: Start task"}}]}
+                elif role == "developer-1":
+                    return {"choices": [{"message": {"content": "SUCCESS: Dev 1 draft"}}]}
+                elif role == "developer-2":
+                    # Always reject to simulate unending loop
+                    return {"choices": [{"message": {"content": "REVIEW_FAILED: Reject again"}}]}
+                return {"choices": [{"message": {"content": "SUCCESS"}}]}
+
+        with patch("antigravity_provider.router.router_engine.get_router_engine", return_value=InfiniteLoopEngine()):
+            self.wf_service.start("Тест предела итераций")
+            thread = self.wf_service._thread
+            self.assertIsNotNone(thread)
+            thread.join(timeout=5)
+            self.assertFalse(thread.is_alive())
+
+        self.assertEqual(self.wf_service.run["status"], "failed")
+        self.assertIn("Достигнут предел итераций", self.wf_service.run["error"])
+        self.assertTrue(any(e.type == "WORKFLOW_MAX_ITERATIONS" for e in self.wf_service.events))
+
+    def test_agent_model_reconfiguration(self):
+        """P0-4: Changing model on agent updates configuration and router policy."""
+        # Update developer-1 model to gemini-3.7-flash-high
+        updated = self.wf_service.update_agent("developer-1", {"model": "gemini-3.7-flash-high"})
+        self.assertEqual(updated.id, "developer-1")
+
+        snap = self.wf_service.snapshot()
+        dev1 = next(a for a in snap["agents"] if a["id"] == "developer-1")
+        self.assertEqual(dev1["execution_config"]["model"], "gemini-3.7-flash-high")
+
+    def test_multi_account_parallelism_without_global_mutex(self):
+        """P0-4: Requests to distinct Antigravity accounts execute concurrently without blocking."""
+        from antigravity_provider.router.router_engine import RouterEngine
+
+        engine = RouterEngine(self.config)
+        execution_times = {}
+
+        def slow_execution(profile_id, duration=0.2):
+            t0 = time.monotonic()
+            time.sleep(duration)
+            execution_times[profile_id] = round(time.monotonic() - t0, 3)
+            return {
+                "choices": [{"message": {"content": f"output from {profile_id}"}}],
+                "router_metadata": {"profile_id": profile_id, "provider": "antigravity"},
+            }
+
+        mock_adapter = MagicMock()
+        mock_adapter.invoke.side_effect = lambda profile, req: slow_execution(profile.profile_id)
+
+        with patch("antigravity_provider.router.router_engine.get_adapter", return_value=mock_adapter):
+            import concurrent.futures
+
+            start_t = time.monotonic()
+            with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
+                f1 = executor.submit(engine.route_request, {"model": "gemini-3.7-flash"}, role="developer-1")
+                f2 = executor.submit(engine.route_request, {"model": "gemini-3.1-pro-high"}, role="developer-2")
+                f3 = executor.submit(engine.route_request, {"model": "claude-opus-4-6-thinking"}, role="code-reviewer")
+
+                r1 = f1.result(timeout=2)
+                r2 = f2.result(timeout=2)
+                r3 = f3.result(timeout=2)
+
+            total_elapsed = time.monotonic() - start_t
+
+        # 3 calls taking 0.2s each running in parallel should take ~0.2-0.35s total, NOT 0.6s+
+        self.assertLess(total_elapsed, 0.55, f"Execution was serialized instead of parallel: {total_elapsed:.3f}s")
+        self.assertEqual(r1["choices"][0]["message"]["content"], "output from ag-w1")
+        self.assertEqual(r2["choices"][0]["message"]["content"], "output from ag-w2")
+        self.assertEqual(r3["choices"][0]["message"]["content"], "output from ag-w3")
+
+
+if __name__ == "__main__":
+    unittest.main()