sync: 175 file da Baida98/AI@d1881b9c (2026-08-25 07:39 UTC) [deploy-all] (#67)

- sync: 175 file da Baida98/AI@d1881b9c (2026-08-25 07:39 UTC) [deploy-all] (b85ebc9eb1d4d6c1e63c7e9b2c341f62c891faf3)

agents/workflow_engine.py

@@ -1,90 +1,112 @@
-import asyncio
 import logging
-import uuid
 import time
-from typing import List, Dict, Optional, Any
+import uuid
+from typing import Any, Dict, List, Optional
+
 from pydantic import BaseModel, Field
 
 _logger = logging.getLogger("agents.workflow_engine")
 
+
 class WorkflowStep(BaseModel):
     step_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
     tool_name: str
     args: Dict[str, Any]
-    status: str = "pending" # pending, running, completed, failed
+    status: str = "pending"  # pending, running, completed, failed
     result: Any = None
     error: Optional[str] = None
     started_at: Optional[float] = None
     finished_at: Optional[float] = None
 
+
 class Workflow(BaseModel):
     workflow_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
     name: str
     steps: List[WorkflowStep]
     status: str = "pending"
     created_at: float = Field(default_factory=time.time)
-    metadata: Dict[str, Any] = {}
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+
 
 class WorkflowExecutor:
     """
-    ARCH-I4.3: Workflow Engine
-    Coordina l'esecuzione di workflow persistenti orchestrati tramite l'Executor.
+    ARCH-I4.3: Workflow Engine.
+
+    Coordina workflow in-memory step-by-step tramite Kernel ed Executor. La
+    persistenza o il resume inter-processo non sono garantiti da questo motore;
+    i caller possono consultare lo stato del workflow corrente tramite
+    ``get_workflow``.
     """
-    def __init__(self, kernel, executor):
+
+    def __init__(self, kernel: Any, executor: Any):
         self.kernel = kernel
         self.executor = executor
         self.active_workflows: Dict[str, Workflow] = {}
 
+    def get_workflow(self, workflow_id: str) -> Optional[Workflow]:
+        """Ritorna il workflow noto, inclusi gli stati terminali in memoria."""
+        return self.active_workflows.get(workflow_id)
+
     async def execute_workflow(self, workflow: Workflow) -> Workflow:
-        """Esegue un workflow step-by-step."""
+        """Esegue un workflow step-by-step, mantenendo il fallback locale."""
         self.active_workflows[workflow.workflow_id] = workflow
         workflow.status = "running"
-        _logger.info(f"Avvio workflow: {workflow.name} ({workflow.workflow_id})")
+        _logger.info("Avvio workflow: %s (%s)", workflow.name, workflow.workflow_id)
 
         for step in workflow.steps:
             step.status = "running"
             step.started_at = time.time()
-            
-            _logger.info(f"Esecuzione step: {step.tool_name} in workflow {workflow.workflow_id}")
-            
+            _logger.info(
+                "Esecuzione step: %s in workflow %s",
+                step.tool_name,
+                workflow.workflow_id,
+            )
             try:
-                # ARCH-I4.3 Integration: Usa il Kernel per risolvere e sottomettere il task
-                # Risoluzione capability (ARCH-E3.2)
-                res = await self.kernel.resolve_capability(step.tool_name)
-                
-                if res.get("status") == "resolved":
-                    worker = res["worker"]
+                # ARCH-I4.3: il Kernel risolve la capability senza esporre
+                # l'infrastruttura al workflow.
+                resolution = await self.kernel.resolve_capability(step.tool_name)
+                if resolution.get("status") == "resolved":
+                    worker = resolution["worker"]
                     worker_id = worker.id if hasattr(worker, "id") else worker["id"]
-                    _logger.info(f"Step {step.tool_name} risolto su worker: {worker_id}")
-                    
-                    # Esecuzione via Executor (che ora usa il Kernel)
+                    _logger.info(
+                        "Step %s risolto su worker: %s",
+                        step.tool_name,
+                        worker_id,
+                    )
                     result = await self.executor.run_tool(
                         tool_name=step.tool_name,
-                        args=step.args,
-                        worker_hint=worker_id
+                        inputs=step.args,
+                        worker_hint=worker_id,
                     )
-                    
-                    step.result = result
-                    step.status = "completed"
                 else:
-                    # Fallback all'esecuzione locale se nessun worker è trovato
-                    _logger.warning(f"Nessun worker per {step.tool_name}, provo esecuzione locale")
-                    result = await self.executor.run_tool(step.tool_name, step.args)
-                    step.result = result
-                    step.status = "completed"
-                    
-            except Exception as e:
+                    # Nessun worker registrato: il comportamento storico resta
+                    # l'esecuzione locale tramite lo stesso Executor.
+                    _logger.warning(
+                        "Nessun worker per %s, provo esecuzione locale",
+                        step.tool_name,
+                    )
+                    result = await self.executor.run_tool(
+                        tool_name=step.tool_name,
+                        inputs=step.args,
+                    )
+
+                step.result = result
+                if isinstance(result, dict) and result.get("success") is False:
+                    step.status = "failed"
+                    step.error = str(result.get("error", "Tool execution failed"))
+                    workflow.status = "failed"
+                    break
+                step.status = "completed"
+            except Exception as exc:
                 step.status = "failed"
-                step.error = str(e)
+                step.error = str(exc)
                 workflow.status = "failed"
-                _logger.error(f"Step {step.tool_name} fallito: {e}")
+                _logger.error("Step %s fallito: %s", step.tool_name, exc)
                 break
-            
-            step.finished_at = time.time()
+            finally:
+                step.finished_at = time.time()
 
         if workflow.status == "running":
             workflow.status = "completed"
-        
-        _logger.info(f"Workflow {workflow.name} terminato con stato: {workflow.status}")
+        _logger.info("Workflow %s terminato con stato: %s", workflow.name, workflow.status)
         return workflow
-

api/resolver.py

@@ -1,11 +1,71 @@
-import time
 import logging
-from typing import List, Dict, Optional, Any
-from pydantic import BaseModel
+import re
+import time
+from typing import Optional, Tuple
+
+from pydantic import BaseModel, field_validator
+
 from .marketplace import WORKERS_REGISTRY, WorkerCapability
 
 _logger = logging.getLogger("api.resolver")
 
+_SEMVER_IDENTIFIER = r"(?:0|[1-9]\d*|\d*[A-Za-z-][0-9A-Za-z-]*)"
+_SEMVER_PATTERN = re.compile(
+    rf"^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)"
+    rf"(?:-(?P<prerelease>{_SEMVER_IDENTIFIER}(?:\.{_SEMVER_IDENTIFIER})*))?"
+    rf"(?:\+(?P<build>[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?$"
+)
+
+
+def _parse_semver(version: str) -> Tuple[int, int, int, Optional[Tuple[str, ...]]]:
+    """Parsa una versione Semantic Versioning 2.0.0 senza dipendenze esterne."""
+    match = _SEMVER_PATTERN.fullmatch(version)
+    if not match:
+        raise ValueError(f"Invalid Semantic Version: {version!r}")
+
+    prerelease = match.group("prerelease")
+    return (
+        int(match.group("major")),
+        int(match.group("minor")),
+        int(match.group("patch")),
+        tuple(prerelease.split(".")) if prerelease else None,
+    )
+
+
+def _compare_semver(left: str, right: str) -> int:
+    """Confronta due versioni SemVer, restituendo -1, 0 oppure 1."""
+    left_major, left_minor, left_patch, left_prerelease = _parse_semver(left)
+    right_major, right_minor, right_patch, right_prerelease = _parse_semver(right)
+
+    left_core = (left_major, left_minor, left_patch)
+    right_core = (right_major, right_minor, right_patch)
+    if left_core != right_core:
+        return -1 if left_core < right_core else 1
+
+    if left_prerelease is None and right_prerelease is None:
+        return 0
+    if left_prerelease is None:
+        return 1
+    if right_prerelease is None:
+        return -1
+
+    for left_identifier, right_identifier in zip(left_prerelease, right_prerelease):
+        if left_identifier == right_identifier:
+            continue
+
+        left_is_numeric = left_identifier.isdigit()
+        right_is_numeric = right_identifier.isdigit()
+        if left_is_numeric and right_is_numeric:
+            return -1 if int(left_identifier) < int(right_identifier) else 1
+        if left_is_numeric != right_is_numeric:
+            return -1 if left_is_numeric else 1
+        return -1 if left_identifier < right_identifier else 1
+
+    if len(left_prerelease) == len(right_prerelease):
+        return 0
+    return -1 if len(left_prerelease) < len(right_prerelease) else 1
+
+
 class ResolverConstraints(BaseModel):
     min_version: Optional[str] = None
     max_cost: Optional[float] = None
@@ -14,62 +74,99 @@ class ResolverConstraints(BaseModel):
     require_gpu: bool = False
     min_priority: int = 100
 
+    @field_validator("min_version")
+    @classmethod
+    def validate_min_version(cls, value: Optional[str]) -> Optional[str]:
+        if value is not None:
+            _parse_semver(value)
+        return value
+
+
 class CapabilityResolver:
     """
     ARCH-E3.2: Capability Resolver
     Mappa le capacità richieste dal Brain ai Worker disponibili tramite il Marketplace,
     scegliendo il migliore in base agli SLA.
     """
-    
+
     @staticmethod
     async def resolve(
-        capability: str, 
-        constraints: Optional[ResolverConstraints] = None
+        capability: str,
+        constraints: Optional[ResolverConstraints] = None,
     ) -> Optional[WorkerCapability]:
         """
         Risolve una capability in un Worker specifico.
         Strategia:
         1. Filtra per capability supportata.
         2. Filtra per worker attivi (last_seen < 300s).
-        3. Applica constraints (versione, costo, latenza, GPU).
-        4. Ordina per (priority ASC, cost ASC, latency ASC).
+        3. Applica constraints (versione, costo, latenza, GPU, priorità).
+        4. Se disponibile, preferisce la regione richiesta.
+        5. Ordina per (priority ASC, cost ASC, latency ASC).
         """
         now = int(time.time())
         candidates = []
-        
         from .health_manager import health_manager
 
         for worker in WORKERS_REGISTRY.values():
             # 1. & 2. Filtro base + Health Check (ARCH-P5.1)
-            is_alive = (now - worker.last_seen < 300)
+            is_alive = now - worker.last_seen < 300
             is_healthy = await health_manager.is_healthy(worker.id)
-            
-            if capability in worker.capabilities and is_alive and is_healthy:
-                # 3. Applica constraints
-                if constraints:
-                    if constraints.min_version and worker.version < constraints.min_version:
-                        continue
-                    if constraints.max_cost is not None and worker.cost > constraints.max_cost:
-                        continue
-                    if constraints.max_latency is not None and worker.latency > constraints.max_latency:
-                        continue
-                    if constraints.require_gpu and not worker.gpu:
+            if capability not in worker.capabilities or not is_alive or not is_healthy:
+                continue
+
+            # 3. Applica constraints
+            if constraints:
+                if constraints.min_version:
+                    try:
+                        if _compare_semver(worker.version, constraints.min_version) < 0:
+                            continue
+                    except ValueError:
+                        _logger.warning(
+                            "Worker %s escluso: versione non valida per il vincolo SemVer (%r)",
+                            worker.id,
+                            worker.version,
+                        )
                         continue
-                
-                candidates.append(worker)
-        
+                if constraints.max_cost is not None and worker.cost > constraints.max_cost:
+                    continue
+                if constraints.max_latency is not None and worker.latency > constraints.max_latency:
+                    continue
+                if constraints.require_gpu and not worker.gpu:
+                    continue
+                # Nel Marketplace una priorità più bassa è migliore; min_priority
+                # mantiene il nome del contratto esistente come soglia massima accettata.
+                if worker.priority > constraints.min_priority:
+                    continue
+
+            candidates.append(worker)
+
+        if constraints and constraints.preferred_region:
+            regional_candidates = [
+                worker
+                for worker in candidates
+                if worker.region == constraints.preferred_region
+            ]
+            if regional_candidates:
+                candidates = regional_candidates
+
         if not candidates:
             _logger.warning(f"Nessun worker trovato per capability: {capability}")
             return None
-            
-        # 4. Ordinamento per SLA
+
+        # 5. Ordinamento per SLA
         # Priorità: Priority (basso meglio), Cost (basso meglio), Latency (basso meglio)
-        candidates.sort(key=lambda w: (w.priority, w.cost, w.latency))
-        
+        candidates.sort(key=lambda worker: (worker.priority, worker.cost, worker.latency))
         best_worker = candidates[0]
-        _logger.info(f"Risolta capability '{capability}' su worker '{best_worker.id}' (score: p={best_worker.priority}, c={best_worker.cost}, l={best_worker.latency})")
-        
+        _logger.info(
+            "Risolta capability '%s' su worker '%s' (score: p=%s, c=%s, l=%s)",
+            capability,
+            best_worker.id,
+            best_worker.priority,
+            best_worker.cost,
+            best_worker.latency,
+        )
         return best_worker
 
+
 # Singleton instance
 resolver = CapabilityResolver()

api/workflows.py

@@ -0,0 +1,81 @@
+"""API in-memory per l'esecuzione esplicita di workflow tool-based."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any, Dict, List
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel, Field
+
+from .auth_guard import AuthRole, require_role
+from agents.workflow_engine import Workflow, WorkflowExecutor, WorkflowStep
+
+_logger = logging.getLogger("api.workflows")
+
+router = APIRouter(
+    prefix="/api/workflows",
+    tags=["workflows"],
+    dependencies=[Depends(require_role(AuthRole.MACHINE))],
+)
+
+
+class WorkflowStartIn(BaseModel):
+    name: str = Field(min_length=1, max_length=200)
+    steps: List[WorkflowStep] = Field(min_length=1, max_length=100)
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+
+
+_workflow_executor: WorkflowExecutor | None = None
+_workflow_tasks: Dict[str, asyncio.Task[None]] = {}
+
+
+def _get_workflow_executor() -> WorkflowExecutor:
+    """Costruisce una sola istanza con i singleton Kernel/Executor esistenti."""
+    global _workflow_executor
+    if _workflow_executor is None:
+        from .kernel import kernel
+        from .state import _get_executor
+
+        executor = _get_executor()
+        if executor is None:
+            raise RuntimeError("Executor non disponibile")
+        _workflow_executor = WorkflowExecutor(kernel=kernel, executor=executor)
+    return _workflow_executor
+
+
+async def _run_workflow(workflow: Workflow, executor: WorkflowExecutor) -> None:
+    """Esegue in background e conserva sempre uno stato terminale osservabile."""
+    try:
+        await executor.execute_workflow(workflow)
+    except Exception as exc:  # defensive: il task non deve fallire silenziosamente
+        workflow.status = "failed"
+        workflow.metadata["runtime_error"] = str(exc)[:500]
+        _logger.exception("Workflow %s terminato con errore inatteso", workflow.workflow_id)
+    finally:
+        _workflow_tasks.pop(workflow.workflow_id, None)
+
+
+@router.post("", response_model=Workflow, status_code=status.HTTP_202_ACCEPTED)
+async def start_workflow(body: WorkflowStartIn) -> Workflow:
+    """Avvia un workflow esplicito senza bloccare la richiesta HTTP."""
+    try:
+        executor = _get_workflow_executor()
+    except RuntimeError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+
+    workflow = Workflow(name=body.name, steps=body.steps, metadata=body.metadata)
+    executor.active_workflows[workflow.workflow_id] = workflow
+    task = asyncio.create_task(_run_workflow(workflow, executor))
+    _workflow_tasks[workflow.workflow_id] = task
+    return workflow
+
+
+@router.get("/{workflow_id}", response_model=Workflow)
+async def get_workflow(workflow_id: str) -> Workflow:
+    """Restituisce lo stato in-memory del workflow, inclusi gli esiti per step."""
+    workflow = _get_workflow_executor().get_workflow(workflow_id)
+    if workflow is None:
+        raise HTTPException(status_code=404, detail=f"Workflow {workflow_id} non trovato")
+    return workflow

main.py

@@ -130,6 +130,7 @@ _ROUTER_MAP = {
     "research": "research",
     "agent_memory": "agent_memory",
     "agent": "agent",
+    "workflows": "workflows",
     "exec": "exec",
     "vault": "vault",
     "browser": "browser",

tests/test_capability_resolver.py

@@ -0,0 +1,173 @@
+import asyncio
+import sys
+import time
+import types
+import unittest
+from enum import IntEnum
+from pathlib import Path
+
+from pydantic import ValidationError
+
+BACKEND_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BACKEND_ROOT))
+
+
+def _install_import_stubs() -> None:
+    auth_guard = types.ModuleType("api.auth_guard")
+
+    class AuthRole(IntEnum):
+        MACHINE = 1
+
+    auth_guard.AuthRole = AuthRole
+    auth_guard.require_role = lambda _role: (lambda: None)
+    sys.modules["api.auth_guard"] = auth_guard
+
+    health_module = types.ModuleType("api.health_manager")
+
+    class HealthyManager:
+        async def is_healthy(self, _worker_id: str) -> bool:
+            return True
+
+    health_module.health_manager = HealthyManager()
+    sys.modules["api.health_manager"] = health_module
+
+
+_install_import_stubs()
+
+from api.marketplace import WORKERS_REGISTRY, WorkerCapability
+from api.resolver import CapabilityResolver, ResolverConstraints
+
+
+class CapabilityResolverTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self._original_registry = dict(WORKERS_REGISTRY)
+        WORKERS_REGISTRY.clear()
+
+    def tearDown(self):
+        WORKERS_REGISTRY.clear()
+        WORKERS_REGISTRY.update(self._original_registry)
+
+    def register_worker(self, **overrides) -> WorkerCapability:
+        values = {
+            "id": "worker",
+            "name": "Worker",
+            "version": "1.0.0",
+            "last_seen": int(time.time()),
+            "capabilities": ["vision"],
+            "cost": 1.0,
+            "latency": 100.0,
+            "region": "global",
+            "gpu": False,
+            "priority": 10,
+        }
+        values.update(overrides)
+        worker = WorkerCapability(**values)
+        WORKERS_REGISTRY[worker.id] = worker
+        return worker
+
+    async def test_uses_semver_not_lexicographic_ordering(self):
+        self.register_worker(id="v2", version="2.0.0", priority=20)
+        self.register_worker(id="v10", version="10.0.0", priority=10)
+
+        worker = await CapabilityResolver.resolve(
+            "vision",
+            ResolverConstraints(min_version="2.0.0"),
+        )
+
+        self.assertIsNotNone(worker)
+        self.assertEqual(worker.id, "v10")
+
+    async def test_excludes_prerelease_below_stable_minimum(self):
+        self.register_worker(id="candidate", version="2.0.0-rc.1", priority=1)
+        self.register_worker(id="stable", version="2.0.0", priority=10)
+
+        worker = await CapabilityResolver.resolve(
+            "vision",
+            ResolverConstraints(min_version="2.0.0"),
+        )
+
+        self.assertIsNotNone(worker)
+        self.assertEqual(worker.id, "stable")
+
+    def test_rejects_malformed_minimum_semver(self):
+        with self.assertRaises(ValidationError):
+            ResolverConstraints(min_version="2.0")
+
+    async def test_excludes_malformed_worker_version_only_when_constrained(self):
+        self.register_worker(id="malformed", version="not-a-version", priority=1)
+        self.register_worker(id="valid", version="2.0.0", priority=10)
+
+        worker = await CapabilityResolver.resolve(
+            "vision",
+            ResolverConstraints(min_version="2.0.0"),
+        )
+
+        self.assertIsNotNone(worker)
+        self.assertEqual(worker.id, "valid")
+
+    async def test_prefers_requested_region_when_available(self):
+        self.register_worker(id="global", region="global", priority=1)
+        self.register_worker(id="eu", region="eu-west", priority=20)
+
+        worker = await CapabilityResolver.resolve(
+            "vision",
+            ResolverConstraints(preferred_region="eu-west"),
+        )
+
+        self.assertIsNotNone(worker)
+        self.assertEqual(worker.id, "eu")
+
+    async def test_falls_back_when_requested_region_is_unavailable(self):
+        self.register_worker(id="global", region="global", priority=1)
+
+        worker = await CapabilityResolver.resolve(
+            "vision",
+            ResolverConstraints(preferred_region="eu-west"),
+        )
+
+        self.assertIsNotNone(worker)
+        self.assertEqual(worker.id, "global")
+
+    async def test_applies_priority_threshold_with_lower_values_preferred(self):
+        self.register_worker(id="allowed", priority=100, cost=10.0)
+        self.register_worker(id="excluded", priority=101, cost=0.0)
+
+        worker = await CapabilityResolver.resolve(
+            "vision",
+            ResolverConstraints(min_priority=100),
+        )
+
+        self.assertIsNotNone(worker)
+        self.assertEqual(worker.id, "allowed")
+
+    async def test_preserves_existing_cost_latency_and_gpu_constraints(self):
+        self.register_worker(
+            id="cpu",
+            cost=1.0,
+            latency=50.0,
+            gpu=False,
+            priority=1,
+        )
+        self.register_worker(
+            id="gpu",
+            cost=2.0,
+            latency=100.0,
+            gpu=True,
+            priority=10,
+        )
+
+        worker = await CapabilityResolver.resolve(
+            "vision",
+            ResolverConstraints(
+                max_cost=2.0,
+                max_latency=100.0,
+                require_gpu=True,
+            ),
+        )
+
+        self.assertIsNotNone(worker)
+        self.assertEqual(worker.id, "gpu")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_workflow_integration.py

@@ -0,0 +1,175 @@
+import ast
+import asyncio
+import sys
+import types
+import unittest
+from enum import IntEnum
+from pathlib import Path
+
+from pydantic import ValidationError
+
+BACKEND_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BACKEND_ROOT))
+
+
+def _install_auth_stub() -> None:
+    auth_guard = types.ModuleType("api.auth_guard")
+
+    class AuthRole(IntEnum):
+        MACHINE = 1
+
+    auth_guard.AuthRole = AuthRole
+    auth_guard.require_role = lambda _role: (lambda: None)
+    sys.modules["api.auth_guard"] = auth_guard
+
+
+_install_auth_stub()
+
+from agents.workflow_engine import Workflow, WorkflowExecutor, WorkflowStep
+from api import workflows
+
+
+class FakeKernel:
+    def __init__(self, resolutions):
+        self.resolutions = resolutions
+        self.calls = []
+
+    async def resolve_capability(self, tool_name):
+        self.calls.append(tool_name)
+        return self.resolutions.get(tool_name, {"status": "error"})
+
+
+class FakeExecutor:
+    def __init__(self, results=None):
+        self.results = results or {}
+        self.calls = []
+
+    async def run_tool(self, tool_name, inputs, timeout=30.0, worker_hint=None):
+        self.calls.append({
+            "tool_name": tool_name,
+            "inputs": inputs,
+            "timeout": timeout,
+            "worker_hint": worker_hint,
+        })
+        result = self.results.get(tool_name, {"success": True, "output": tool_name})
+        if isinstance(result, Exception):
+            raise result
+        return result
+
+
+class WorkflowExecutorTests(unittest.IsolatedAsyncioTestCase):
+    async def test_uses_resolved_worker_hint_and_records_completed_step(self):
+        kernel = FakeKernel({"search": {"status": "resolved", "worker": {"id": "worker-eu"}}})
+        executor = FakeExecutor()
+        engine = WorkflowExecutor(kernel=kernel, executor=executor)
+        workflow = Workflow(name="ricerca", steps=[WorkflowStep(tool_name="search", args={"query": "AI"})])
+
+        result = await engine.execute_workflow(workflow)
+
+        self.assertEqual(result.status, "completed")
+        self.assertEqual(result.steps[0].status, "completed")
+        self.assertIsNotNone(result.steps[0].started_at)
+        self.assertIsNotNone(result.steps[0].finished_at)
+        self.assertEqual(executor.calls[0]["worker_hint"], "worker-eu")
+
+    async def test_preserves_local_fallback_when_no_worker_is_resolved(self):
+        kernel = FakeKernel({})
+        executor = FakeExecutor()
+        engine = WorkflowExecutor(kernel=kernel, executor=executor)
+        workflow = Workflow(name="fallback", steps=[WorkflowStep(tool_name="local", args={"value": 1})])
+
+        result = await engine.execute_workflow(workflow)
+
+        self.assertEqual(result.status, "completed")
+        self.assertIsNone(executor.calls[0]["worker_hint"])
+
+    async def test_stops_after_unsuccessful_tool_result(self):
+        kernel = FakeKernel({})
+        executor = FakeExecutor({"first": {"success": False, "error": "denied"}})
+        engine = WorkflowExecutor(kernel=kernel, executor=executor)
+        workflow = Workflow(
+            name="errore",
+            steps=[
+                WorkflowStep(tool_name="first", args={}),
+                WorkflowStep(tool_name="second", args={}),
+            ],
+        )
+
+        result = await engine.execute_workflow(workflow)
+
+        self.assertEqual(result.status, "failed")
+        self.assertEqual(result.steps[0].status, "failed")
+        self.assertEqual(result.steps[0].error, "denied")
+        self.assertEqual(result.steps[1].status, "pending")
+        self.assertEqual([call["tool_name"] for call in executor.calls], ["first"])
+
+
+class WorkflowApiTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.previous_executor = workflows._workflow_executor
+        self.previous_tasks = workflows._workflow_tasks.copy()
+        workflows._workflow_executor = None
+        workflows._workflow_tasks.clear()
+
+    async def asyncTearDown(self):
+        for task in list(workflows._workflow_tasks.values()):
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        workflows._workflow_tasks.clear()
+        workflows._workflow_executor = self.previous_executor
+        workflows._workflow_tasks.update(self.previous_tasks)
+
+    def test_router_exposes_start_and_status_paths(self):
+        routes = {
+            (route.path, method)
+            for route in workflows.router.routes
+            for method in (getattr(route, "methods", set()) or set())
+        }
+
+        self.assertEqual(workflows.router.prefix, "/api/workflows")
+        self.assertIn(("/api/workflows", "POST"), routes)
+        self.assertIn(("/api/workflows/{workflow_id}", "GET"), routes)
+
+    def test_main_router_map_mounts_workflows(self):
+        tree = ast.parse((BACKEND_ROOT / "main.py").read_text(encoding="utf-8"))
+        router_map = next(
+            node.value
+            for node in tree.body
+            if isinstance(node, ast.Assign)
+            and any(isinstance(target, ast.Name) and target.id == "_ROUTER_MAP" for target in node.targets)
+        )
+        routes = {
+            key.value: value.value
+            for key, value in zip(router_map.keys, router_map.values)
+            if isinstance(key, ast.Constant) and isinstance(value, ast.Constant)
+        }
+
+        self.assertEqual(routes.get("workflows"), "workflows")
+
+    def test_start_contract_rejects_empty_steps(self):
+        with self.assertRaises(ValidationError):
+            workflows.WorkflowStartIn(name="vuoto", steps=[])
+
+    async def test_start_and_get_workflow_use_background_execution(self):
+        engine = WorkflowExecutor(kernel=FakeKernel({}), executor=FakeExecutor())
+        workflows._workflow_executor = engine
+        body = workflows.WorkflowStartIn(
+            name="API workflow",
+            steps=[WorkflowStep(tool_name="local", args={"n": 1})],
+        )
+
+        started = await workflows.start_workflow(body)
+        task = workflows._workflow_tasks[started.workflow_id]
+        await task
+        fetched = await workflows.get_workflow(started.workflow_id)
+
+        self.assertEqual(fetched.workflow_id, started.workflow_id)
+        self.assertEqual(fetched.status, "completed")
+        self.assertEqual(fetched.steps[0].status, "completed")
+
+
+if __name__ == "__main__":
+    unittest.main()