sync: 204 file da Baida98/AI@843961e3 (2026-08-31 07:48 UTC) [deploy-all] (#154)

- sync: 204 file da Baida98/AI@843961e3 (2026-08-31 07:48 UTC) [deploy-all] (9f9d3dc314be6b399d4fe2afb217579f98304139)

agents/executor.py

@@ -292,15 +292,39 @@ class Executor:
                 _t0          = _time_mod.monotonic()
                 result       = await asyncio.wait_for(fn(**inputs), timeout=_adaptive_to)
                 _timeout_tracker.record(tool_name, _time_mod.monotonic() - _t0)
+
+                # Il tool ha già prodotto il side effect: la persistenza memoria
+                # è osservabilità e non deve riaprire il retry del tool.
+                _memory_persisted = True
+                _memory_error = None
                 if self.memory:
-                    # S577→S600: inputs 100→500 — parity con altri handler
-                    await self.memory.save_episode(
-                        "tool",
-                        f"{tool_name}: {str(inputs)[:500]}",
-                        str(result)[:500],
-                        True,
-                    )
-                return {"success": True, "tool": tool_name, "output": result, "attempt": attempt + 1}
+                    try:
+                        # S577→S600: inputs 100→500 — parity con altri handler
+                        await self.memory.save_episode(
+                            "tool",
+                            f"{tool_name}: {str(inputs)[:500]}",
+                            str(result)[:500],
+                            True,
+                        )
+                    except Exception as _memory_exc:
+                        _memory_persisted = False
+                        _memory_error = f"{type(_memory_exc).__name__}: {str(_memory_exc)[:240]}"
+                        _logger.warning(
+                            "[executor] tool %s completato ma save_episode fallito; "
+                            "nessun retry del side effect: %s",
+                            tool_name,
+                            _memory_error,
+                        )
+                response = {
+                    "success": True,
+                    "tool": tool_name,
+                    "output": result,
+                    "attempt": attempt + 1,
+                    "memory_persisted": _memory_persisted,
+                }
+                if _memory_error:
+                    response["memory_error"] = _memory_error
+                return response
 
             except asyncio.TimeoutError:
                 # FIX-GAP2: registra il timeout come durata massima per shrink futuro

agents/unified_loop.py

@@ -214,37 +214,46 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
             _logger.debug("[unified_loop] state callback silenced: %s", _state_callback_error)
 
     async def _rollback_writes(self, on_step=None) -> None:
-        """
-        GAP-3: ripristina i file sovrascritti se il loop si interrompe a metà.
-        Chiama dopo un errore grave che ha lasciato il progetto in stato inconsistente.
-        Ogni file in _write_snapshots viene ripristinato al suo contenuto originale.
-        File che non esistevano (snapshot=None) vengono ignorati (non possiamo eliminarli in modo sicuro).
+        """Restore all writes from this run or report an incomplete rollback.
+
+        A None snapshot means the file did not exist and must be removed.
+        Failed restores remain tracked so a supervisor can retry or block the run.
         """
         if not self._write_snapshots or not self.executor:
             return
         if on_step:
             await _maybe_await(on_step({
                 "action": "text_chunk",
-                "token":  f"\u23ea Rollback di {len(self._write_snapshots)} file modificati...\n",
+                "token": f"\u23ea Rollback di {len(self._write_snapshots)} file modificati...\n",
                 "status": "streaming",
             }))
-        _rolled = 0
-        for path, original in self._write_snapshots.items():
-            if original is None:
-                continue  # file non esisteva prima — saltiamo (non eliminiamo)
+
+        remaining: dict[str, str | None] = {}
+        rolled = 0
+        for path, original in list(self._write_snapshots.items()):
+            tool_name = "delete_file" if original is None else "write_file"
+            inputs = {"path": path} if original is None else {"path": path, "content": original}
             try:
-                await asyncio.wait_for(
-                    self.executor.run_tool("write_file", {"path": path, "content": original}),
+                result = await asyncio.wait_for(
+                    self.executor.run_tool(tool_name, inputs),
                     timeout=10.0,
                 )
-                _rolled += 1
-            except Exception:
-                pass  # non-fatal — best effort rollback
-        _total = len(self._write_snapshots)  # salva prima del clear
-        self._write_snapshots = {}
-        _logger.info("GAP-3 rollback: %d/%d file ripristinati", _rolled, _total)
+                payload = result.get("output") if isinstance(result, dict) else None
+                nested_failed = isinstance(payload, dict) and payload.get("ok") is False
+                if not isinstance(result, dict) or not result.get("success") or nested_failed:
+                    error = (payload or {}).get("error") if isinstance(payload, dict) else None
+                    raise RuntimeError(error or result.get("error", "rollback tool failed"))
+                rolled += 1
+            except Exception as exc:
+                remaining[path] = original
+                _logger.error("GAP-3 rollback fallito per %s: %s", path, str(exc)[:240])
+
+        total = len(self._write_snapshots)
+        self._write_snapshots = remaining
+        _logger.info("GAP-3 rollback: %d/%d file ripristinati", rolled, total)
+        if remaining:
+            raise RuntimeError(f"VFS rollback incompleto: {len(remaining)}/{total} file non ripristinati")
 
-    # ── GAP-NEW-4: Git VFS auto-snapshot ────────────────────────────────────────
     async def _vfs_git_backup(self) -> None:
         """GAP-NEW-4: Push _session_files al branch vfs-backup su GitHub.
 
@@ -1319,18 +1328,28 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
                                 {"path": _wf_path, "content": _wf_generated} if rn == "write_file"
                                 else {"path": _wf_path, "patch": _wf_generated}
                             )
-                            # GAP-3: snapshot pre-write — cattura originale per rollback atomico
+                            # GAP-3: snapshot pre-write; read errors are not "file absent".
                             if rn == "write_file" and _wf_path not in self._write_snapshots:
-                                try:
-                                    _snap_r = await asyncio.wait_for(
-                                        self.executor.run_tool("read_file", {"path": _wf_path}),
-                                        timeout=4.0,
-                                    )
-                                    self._write_snapshots[_wf_path] = (
-                                        _snap_r.get("output") if _snap_r.get("success") else None
+                                _snap_r = await asyncio.wait_for(
+                                    self.executor.run_tool("read_file", {"path": _wf_path}),
+                                    timeout=4.0,
+                                )
+                                _snap_payload = _snap_r.get("output") if isinstance(_snap_r, dict) else None
+                                _snap_failed = isinstance(_snap_payload, dict) and _snap_payload.get("ok") is False
+                                if isinstance(_snap_payload, dict):
+                                    _snap_content = _snap_payload.get("content")
+                                    _snap_error = str(_snap_payload.get("error", ""))
+                                else:
+                                    _snap_content = _snap_payload
+                                    _snap_error = str(_snap_r.get("error", "")) if isinstance(_snap_r, dict) else ""
+                                if isinstance(_snap_content, str) and not _snap_failed:
+                                    self._write_snapshots[_wf_path] = _snap_content
+                                elif "File non trovato" in _snap_error or "File not found" in _snap_error:
+                                    self._write_snapshots[_wf_path] = None
+                                else:
+                                    raise RuntimeError(
+                                        f"Snapshot VFS non disponibile per {_wf_path}: {_snap_error[:240]}"
                                     )
-                                except Exception:
-                                    self._write_snapshots[_wf_path] = None  # file non esisteva
                             # GAP-VFS: lock per-path — serializza scritture parallele sullo stesso file
                             _vfs_lock = self._get_vfs_lock(_wf_path)
                             async with _vfs_lock:
@@ -3557,6 +3576,12 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
                 }
 
             state.errors.append(error_text)
+            if getattr(self, "_write_snapshots", None):
+                try:
+                    await self._rollback_writes(on_step)
+                except Exception as rollback_error:
+                    state.errors.append(str(rollback_error)[:500])
+                    _logger.error("[unified_loop] rollback inatteso incompleto: %s", rollback_error)
             previous = state.state_machine.current
             if previous != AgentState.FAILED:
                 try:
@@ -3714,6 +3739,12 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
 
         async def _finish(result: dict[str, Any]) -> dict[str, Any]:
             next_state = AgentState.COMPLETED if result.get("success", True) else AgentState.FAILED
+            if next_state == AgentState.FAILED and getattr(self, "_write_snapshots", None):
+                try:
+                    await self._rollback_writes(on_step)
+                except Exception as rollback_error:
+                    result.setdefault("errors", []).append(str(rollback_error)[:500])
+                    result["rollback_incomplete"] = True
             try:
                 await self._transition_state(state, next_state, on_step)
             finally:

api/background_tasks.py

@@ -0,0 +1,53 @@
+"""Supervision utilities for long-lived background asyncio tasks."""
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Awaitable
+from typing import Any
+
+_logger = logging.getLogger("agente_ai.background_tasks")
+_tasks: dict[str, asyncio.Task[Any]] = {}
+
+
+def spawn_background_task(coro: Awaitable[Any], *, name: str) -> asyncio.Task[Any]:
+    """Start one named background task and retain it for lifecycle shutdown.
+
+    A live task with the same name is reused. The passed coroutine is closed in
+    that case so duplicate startup calls do not leak an un-awaited coroutine.
+    """
+    current = _tasks.get(name)
+    if current is not None and not current.done():
+        close = getattr(coro, "close", None)
+        if close is not None:
+            close()
+        return current
+
+    task = asyncio.create_task(coro, name=name)
+    _tasks[name] = task
+
+    def _report(task_result: asyncio.Task[Any]) -> None:
+        if task_result.cancelled():
+            return
+        try:
+            error = task_result.exception()
+        except asyncio.CancelledError:
+            return
+        if error is not None:
+            _logger.error("background task %s failed: %s", name, error, exc_info=error)
+
+    task.add_done_callback(_report)
+    return task
+
+
+async def shutdown_background_tasks() -> None:
+    """Cancel and await all supervised background tasks."""
+    tasks = [task for task in _tasks.values() if not task.done()]
+    for task in tasks:
+        task.cancel()
+    if tasks:
+        await asyncio.gather(*tasks, return_exceptions=True)
+    _tasks.clear()
+
+
+__all__ = ["spawn_background_task", "shutdown_background_tasks"]

api/job_queue.py

@@ -355,11 +355,11 @@ async def _hands_consumer_loop() -> None:
                 continue
 
             # Esegui in background — non blocca il loop consumer
-            t = asyncio.create_task(_execute_queued_job(job))
-            t.add_done_callback(lambda task: (
-                _logger.error("[jq] job task crashed: %s", task.exception(), exc_info=task.exception())
-                if not task.cancelled() and task.exception() else None
-            ))
+            from api.background_tasks import spawn_background_task
+            task_id = str(job.get("taskId") or uuid.uuid4())
+            spawn_background_task(
+                _execute_queued_job(job), name=f"job-queue-execution:{task_id}"
+            )
 
         except Exception as exc:
             _logger.debug("[jq] consumer tick error: %s", exc)
@@ -377,18 +377,15 @@ async def start_job_queue_consumer() -> None:
         return
 
     # Load publisher su tutti gli Space
-    def _log_jq_exc(t):
-        if not t.cancelled() and t.exception():
-            _logger.warning("[job_queue] bg loop raised: %s", t.exception())
-
-    asyncio.create_task(_load_publisher_loop()).add_done_callback(_log_jq_exc)
+    from api.background_tasks import spawn_background_task
+    spawn_background_task(_load_publisher_loop(), name="job-queue-load-publisher")
 
     # Consumer per tutti i ruoli worker o legacy 'hands'
     _IS_WORKER = _SPACE_ROLE.startswith("worker-") or _SPACE_ROLE in ("hands", "unknown")
     
     if _IS_WORKER:
         _logger.info("[jq] Avvio consumer loop per ruolo worker: %s", _SPACE_ROLE)
-        asyncio.create_task(_hands_consumer_loop()).add_done_callback(_log_jq_exc)
+        spawn_background_task(_hands_consumer_loop(), name="job-queue-hands-consumer")
     else:
         _logger.info("[jq] SPACE_ROLE=%s — consumer non avviato (ruolo non worker)", _SPACE_ROLE)
 

api/state.py

@@ -139,6 +139,11 @@ _CHECKPOINT_TTL_MS = 2 * 60 * 60 * 1000
 _CHECKPOINT_MAX    = 100
 _AGENT_TASK_TTL_MS = 2 * 60 * 60 * 1000
 _AGENT_TASK_MAX    = 200
+# All statuses that represent a finished task and can therefore expire from the
+# bounded in-memory store. Keep this set aligned with API lifecycle writers.
+_AGENT_TASK_TERMINAL_STATES: frozenset[str] = frozenset({
+    'SUCCESS', 'COMPLETED', 'ERROR', 'CANCELLED', 'RATE_LIMITED',
+})
 
 # ── Telemetry & Health ────────────────────────────────────────────────────────
 _ai_health_cache: dict = {"data": None, "at": 0.0}
@@ -244,7 +249,7 @@ def _prune_checkpoints() -> None:
 def _prune_agent_tasks() -> None:
     now = int(time.time() * 1000)
     expired = [k for k, v in list(_agent_tasks.items())
-               if v.get('status') in ('SUCCESS', 'ERROR', 'CANCELLED')
+               if v.get('status') in _AGENT_TASK_TERMINAL_STATES
                and now - v.get('created_at', 0) > _AGENT_TASK_TTL_MS]
     for k in expired:
         _agent_tasks.pop(k, None)

cors_policy.py

@@ -0,0 +1,61 @@
+"""CORS policy shared by the FastAPI entrypoint.
+
+The backend authenticates browser calls with explicit bearer/internal headers,
+not with ambient browser cookies. Credentials therefore remain disabled unless
+an explicit future policy requires them.
+"""
+from __future__ import annotations
+
+import os
+from urllib.parse import urlsplit
+
+PUBLIC_FRONTEND_ORIGIN = "https://agente-ai.pages.dev"
+_DEVELOPMENT_ORIGINS = frozenset(
+    {
+        "http://localhost:3000",
+        "http://localhost:5173",
+        "http://127.0.0.1:3000",
+        "http://127.0.0.1:5173",
+    }
+)
+_DEVELOPMENT_ENVS = frozenset({"dev", "development", "local", "test"})
+
+
+def _is_valid_origin(origin: str) -> bool:
+    """Return True only for an exact HTTP(S) origin without path or wildcard."""
+    if not origin or origin == "*" or any(char.isspace() for char in origin):
+        return False
+    parsed = urlsplit(origin)
+    return bool(
+        parsed.scheme in {"http", "https"}
+        and parsed.netloc
+        and not parsed.username
+        and not parsed.password
+        and not parsed.path
+        and not parsed.query
+        and not parsed.fragment
+    )
+
+
+def allowed_origins(raw: str | None = None, environment: str | None = None) -> list[str]:
+    """Build the exact CORS allow-list from environment and safe defaults.
+
+    ``CORS_ALLOWED_ORIGINS`` is a comma-separated list of exact origins. An
+    invalid value, including ``*``, is ignored rather than widening access.
+    Local development origins are enabled only for an explicit development or
+    test environment. The production default is the canonical Pages origin.
+    """
+    configured = raw if raw is not None else os.getenv("CORS_ALLOWED_ORIGINS", "")
+    origins = [
+        origin.strip().rstrip("/")
+        for origin in configured.split(",")
+        if origin.strip()
+    ]
+    valid = {origin for origin in origins if _is_valid_origin(origin)}
+
+    env_name = (environment if environment is not None else os.getenv("APP_ENV", os.getenv("ENVIRONMENT", ""))).strip().lower()
+    if not valid:
+        valid.add(PUBLIC_FRONTEND_ORIGIN)
+    if env_name in _DEVELOPMENT_ENVS:
+        valid.update(_DEVELOPMENT_ORIGINS)
+    return sorted(valid)

main.py

@@ -10,6 +10,7 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
 from api.version import RUNTIME_VERSION
+from cors_policy import allowed_origins
 
 # Configurazione Logging
 logging.basicConfig(
@@ -25,13 +26,14 @@ app = FastAPI(
     version=RUNTIME_VERSION,
 )
 
-# CORS
+# CORS: whitelist esatta e fail-closed. Il frontend usa bearer/internal headers,
+# non cookie cross-origin; le credenziali browser restano quindi disabilitate.
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
+    allow_origins=allowed_origins(),
+    allow_credentials=False,
+    allow_methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
+    allow_headers=["Authorization", "Content-Type", "X-Internal-Token", "X-Requested-With"],
 )
 
 # ── P17-F1: RLS Fix & Auto-Migration ──────────────────────────────────────────
@@ -244,6 +246,20 @@ async def run_cli_task(task_description: str):
         sys.exit(1)
 
 # ── Startup ───────────────────────────────────────────────────────────────────
+async def _snapshot_heartbeat() -> None:
+    """Keep the public dashboard snapshot fresh without affecting request handling."""
+    from api.public_snapshot import refresh_public_dashboard_snapshot
+
+    while True:
+        await asyncio.sleep(30)
+        try:
+            await asyncio.wait_for(refresh_public_dashboard_snapshot(), timeout=10.0)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            _logger.debug("Public snapshot heartbeat failed (non-blocking): %s", exc)
+
+
 @app.on_event("startup")
 async def startup_event():
     _logger.info("Server starting up...")
@@ -265,6 +281,14 @@ async def startup_event():
             _logger.warning("⚠️ BOOT: public dashboard snapshot writer non ha persistito lo snapshot.")
     except Exception as e:
         _logger.warning(f"⚠️ BOOT: avvio public snapshot writer fallito (non bloccante): {e}")
+    try:
+        from api.background_tasks import spawn_background_task
+        from api.telemetry import telemetry_alert_loop
+        spawn_background_task(telemetry_alert_loop(), name="telemetry-alert-loop")
+        spawn_background_task(_snapshot_heartbeat(), name="public-snapshot-heartbeat")
+        _logger.info("✅ BOOT: telemetry e snapshot loops supervisionati.")
+    except Exception as e:
+        _logger.warning(f"⚠️ BOOT: avvio telemetry/snapshot loops fallito (non bloccante): {e}")
     try:
         from api.providers import start_heartbeat
         start_heartbeat()
@@ -273,9 +297,24 @@ async def startup_event():
         _logger.warning(f"⚠️ BOOT: avvio provider heartbeat fallito (non bloccante): {e}")
     if not any(arg in sys.argv for arg in ["--task", "-t"]):
         try:
+            from api.background_tasks import spawn_background_task
             from api.job_queue import start_job_queue_consumer
-            asyncio.create_task(start_job_queue_consumer())
-        except Exception: pass
+            spawn_background_task(
+                start_job_queue_consumer(), name="job-queue-consumer-supervisor"
+            )
+        except Exception as e:
+            _logger.warning("⚠️ BOOT: avvio job queue fallito (non bloccante): %s", e)
+
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    """Cancel and await all registered background tasks before loop shutdown."""
+    try:
+        from api.background_tasks import shutdown_background_tasks
+        await shutdown_background_tasks()
+        _logger.info("✅ SHUTDOWN: background tasks supervisionate arrestate.")
+    except Exception as e:
+        _logger.warning("⚠️ SHUTDOWN: arresto background tasks incompleto: %s", e)
 
 # ── SPA Hosting ───────────────────────────────────────────────────────────────
 _STATIC_DIR = os.getenv('FRONTEND_DIST', '/app/backend/static')

scripts/lib/report-engine.mjs

@@ -0,0 +1,36 @@
+import { mkdirSync, renameSync, writeFileSync } from "node:fs";
+import { join, resolve } from "node:path";
+
+const DEFAULT_REPORT_DIRECTORY = "/tmp/agente-ai";
+
+function safeSegment(value) {
+  return String(value ?? "benchmark")
+    .trim()
+    .replace(/[^a-zA-Z0-9._-]+/g, "-")
+    .replace(/^-+|-+$/g, "")
+    .slice(0, 96) || "benchmark";
+}
+
+/**
+ * Persiste un artefatto diagnostico del runner senza incidere sui calcoli.
+ * Il write-then-rename evita file JSON parziali in caso di interruzione.
+ */
+export function saveBenchmarkReport(reportName, payload) {
+  const reportDirectory = resolve(process.env.BENCHMARK_REPORT_DIR || DEFAULT_REPORT_DIRECTORY);
+  mkdirSync(reportDirectory, { recursive: true });
+
+  const safeName = safeSegment(reportName);
+  const stamp = new Date().toISOString().replace(/[:.]/g, "-");
+  const finalPath = join(reportDirectory, `${safeName}-${stamp}.json`);
+  const temporaryPath = `${finalPath}.${process.pid}.tmp`;
+  const document = {
+    schemaVersion: "benchmark-report-v1",
+    generatedAt: new Date().toISOString(),
+    reportName: safeName,
+    ...payload,
+  };
+
+  writeFileSync(temporaryPath, `${JSON.stringify(document, null, 2)}\n`, { encoding: "utf8", mode: 0o600 });
+  renameSync(temporaryPath, finalPath);
+  return finalPath;
+}

tests/test_agent_task_pruning.py

@@ -0,0 +1,70 @@
+import unittest
+from unittest.mock import patch
+
+import api.state as state
+
+
+_TERMINAL_STATES = (
+    "SUCCESS",
+    "COMPLETED",
+    "ERROR",
+    "CANCELLED",
+    "RATE_LIMITED",
+)
+
+
+class AgentTaskPruningTests(unittest.TestCase):
+    def test_prune_expires_every_terminal_state_and_cleans_byok(self):
+        now_ms = 10_000_000
+        old_ms = now_ms - state._AGENT_TASK_TTL_MS - 1
+        task_ids = {f"old-{status.lower()}" for status in _TERMINAL_STATES}
+        task_ids.update({"running", "queued"})
+        tasks = {
+            task_id: {
+                "status": (
+                    task_id.removeprefix("old-").upper()
+                    if task_id.startswith("old-")
+                    else task_id.upper()
+                ),
+                "created_at": old_ms,
+            }
+            for task_id in task_ids
+        }
+        byok_clients = {task_id: object() for task_id in task_ids}
+
+        with patch.object(state, "_agent_tasks", tasks), patch.object(
+            state, "_task_ai_clients", byok_clients
+        ), patch.object(state.time, "time", return_value=now_ms / 1000):
+            state._prune_agent_tasks()
+            self.assertEqual(set(state._agent_tasks), {"running", "queued"})
+            self.assertEqual(set(state._task_ai_clients), {"running", "queued"})
+
+    def test_prune_keeps_recent_terminal_tasks(self):
+        now_ms = 10_000_000
+        recent_ms = now_ms - state._AGENT_TASK_TTL_MS + 1
+        tasks = {
+            status.lower(): {"status": status, "created_at": recent_ms}
+            for status in _TERMINAL_STATES
+        }
+
+        with patch.object(state, "_agent_tasks", tasks), patch.object(
+            state, "_task_ai_clients", {}
+        ), patch.object(state.time, "time", return_value=now_ms / 1000):
+            state._prune_agent_tasks()
+            self.assertEqual(
+                set(state._agent_tasks),
+                {status.lower() for status in _TERMINAL_STATES},
+            )
+
+    def test_terminal_state_set_matches_pruning_contract(self):
+        self.assertEqual(
+            state._AGENT_TASK_TERMINAL_STATES,
+            frozenset(_TERMINAL_STATES),
+        )
+        self.assertNotIn("QUEUED", state._AGENT_TASK_TERMINAL_STATES)
+        self.assertNotIn("RUNNING", state._AGENT_TASK_TERMINAL_STATES)
+        self.assertNotIn("CREATING", state._AGENT_TASK_TERMINAL_STATES)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_background_tasks.py

@@ -0,0 +1,57 @@
+import asyncio
+import unittest
+
+from api import background_tasks
+
+
+class BackgroundTaskSupervisorTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        await background_tasks.shutdown_background_tasks()
+
+    async def asyncTearDown(self):
+        await background_tasks.shutdown_background_tasks()
+
+    async def test_duplicate_name_reuses_task_and_closes_duplicate_coroutine(self):
+        started = asyncio.Event()
+        release = asyncio.Event()
+
+        async def worker():
+            started.set()
+            await release.wait()
+
+        first = background_tasks.spawn_background_task(worker(), name="duplicate")
+        await started.wait()
+        duplicate = background_tasks.spawn_background_task(worker(), name="duplicate")
+        self.assertIs(first, duplicate)
+        release.set()
+        await background_tasks.shutdown_background_tasks()
+        self.assertTrue(first.done())
+
+    async def test_shutdown_cancels_and_awaits_running_task(self):
+        cancelled = asyncio.Event()
+
+        async def worker():
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                cancelled.set()
+                raise
+
+        background_tasks.spawn_background_task(worker(), name="cancellable")
+        await asyncio.sleep(0)
+        await background_tasks.shutdown_background_tasks()
+        self.assertTrue(cancelled.is_set())
+        self.assertFalse(background_tasks._tasks)
+
+    async def test_completed_task_is_not_left_in_registry_after_shutdown(self):
+        async def worker():
+            return "ok"
+
+        task = background_tasks.spawn_background_task(worker(), name="completed")
+        await task
+        await background_tasks.shutdown_background_tasks()
+        self.assertFalse(background_tasks._tasks)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_cors_policy.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import unittest
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.testclient import TestClient
+
+from cors_policy import PUBLIC_FRONTEND_ORIGIN, allowed_origins
+
+
+class CorsPolicyTests(unittest.TestCase):
+    def test_production_defaults_to_canonical_frontend_only(self) -> None:
+        self.assertEqual(allowed_origins(raw="", environment="production"), [PUBLIC_FRONTEND_ORIGIN])
+
+    def test_custom_origins_are_exact_and_wildcards_are_ignored(self) -> None:
+        origins = allowed_origins(
+            raw=" https://agente-ai.pages.dev/ , * , https://evil.example/path , http://localhost:5173 ",
+            environment="production",
+        )
+        self.assertEqual(origins, ["http://localhost:5173", PUBLIC_FRONTEND_ORIGIN])
+
+    def test_local_origins_are_available_only_in_development(self) -> None:
+        production = allowed_origins(raw="", environment="production")
+        development = allowed_origins(raw="", environment="development")
+        self.assertNotIn("http://localhost:5173", production)
+        self.assertIn("http://localhost:5173", development)
+
+    def test_invalid_configuration_fails_closed_to_public_default(self) -> None:
+        self.assertEqual(allowed_origins(raw="*", environment="production"), [PUBLIC_FRONTEND_ORIGIN])
+
+    def test_middleware_rejects_untrusted_origin_and_does_not_allow_credentials(self) -> None:
+        app = FastAPI()
+        app.add_middleware(
+            CORSMiddleware,
+            allow_origins=allowed_origins(raw=PUBLIC_FRONTEND_ORIGIN, environment="production"),
+            allow_credentials=False,
+            allow_methods=["GET", "OPTIONS"],
+            allow_headers=["Authorization", "Content-Type"],
+        )
+
+        @app.get("/health")
+        def health() -> dict[str, str]:
+            return {"status": "ok"}
+
+        client = TestClient(app)
+        trusted = client.get("/health", headers={"Origin": PUBLIC_FRONTEND_ORIGIN})
+        untrusted = client.get("/health", headers={"Origin": "https://audit-origin.invalid"})
+        preflight = client.options(
+            "/health",
+            headers={
+                "Origin": "https://audit-origin.invalid",
+                "Access-Control-Request-Method": "GET",
+                "Access-Control-Request-Headers": "authorization",
+            },
+        )
+
+        self.assertEqual(trusted.headers.get("access-control-allow-origin"), PUBLIC_FRONTEND_ORIGIN)
+        self.assertNotIn("access-control-allow-credentials", trusted.headers)
+        self.assertNotIn("access-control-allow-origin", untrusted.headers)
+        self.assertNotEqual(preflight.headers.get("access-control-allow-origin"), "https://audit-origin.invalid")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_executor_side_effect_retry.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from agents import executor as executor_module
+from agents.executor import Executor
+
+
+class FailingMemory:
+    async def save_episode(self, *_args, **_kwargs) -> None:
+        raise RuntimeError("memory unavailable")
+
+
+class RecordingMemory:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def save_episode(self, *_args, **_kwargs) -> None:
+        self.calls += 1
+
+
+class ExecutorSideEffectRetryTests(unittest.IsolatedAsyncioTestCase):
+    async def test_memory_failure_does_not_retry_completed_side_effect(self) -> None:
+        calls = 0
+
+        async def non_idempotent_tool(**_inputs):
+            nonlocal calls
+            calls += 1
+            return {"created_id": "resource-1"}
+
+        with patch.dict(
+            executor_module.TOOL_REGISTRY,
+            {"non_idempotent_tool": {"required_inputs": [], "fallbacks": [], "_fn": non_idempotent_tool}},
+            clear=False,
+        ):
+            result = await Executor(llm_client=object(), memory=FailingMemory(), max_retries=2).run_tool(
+                "non_idempotent_tool", {}, timeout=2, worker_hint="test",
+            )
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["attempt"], 1)
+        self.assertEqual(calls, 1)
+        self.assertFalse(result["memory_persisted"])
+        self.assertIn("memory unavailable", result["memory_error"])
+
+    async def test_tool_failure_still_retries_before_any_side_effect_result(self) -> None:
+        calls = 0
+
+        async def flaky_tool(**_inputs):
+            nonlocal calls
+            calls += 1
+            if calls == 1:
+                raise RuntimeError("transient tool failure")
+            return {"ok": True}
+
+        with patch.dict(
+            executor_module.TOOL_REGISTRY,
+            {"flaky_tool": {"required_inputs": [], "fallbacks": [], "_fn": flaky_tool}},
+            clear=False,
+        ):
+            result = await Executor(llm_client=object(), memory=None, max_retries=2).run_tool(
+                "flaky_tool", {}, timeout=2, worker_hint="test",
+            )
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["attempt"], 2)
+        self.assertEqual(calls, 2)
+
+    async def test_successful_memory_persistence_is_reported(self) -> None:
+        memory = RecordingMemory()
+
+        async def safe_tool(**_inputs):
+            return "done"
+
+        with patch.dict(
+            executor_module.TOOL_REGISTRY,
+            {"safe_tool": {"required_inputs": [], "fallbacks": [], "_fn": safe_tool}},
+            clear=False,
+        ):
+            result = await Executor(llm_client=object(), memory=memory, max_retries=0).run_tool(
+                "safe_tool", {}, timeout=2, worker_hint="test",
+            )
+
+        self.assertTrue(result["success"])
+        self.assertTrue(result["memory_persisted"])
+        self.assertEqual(memory.calls, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_model_watch_adapter.py

@@ -171,14 +171,14 @@ class GeminiModelsAdapterTests(unittest.IsolatedAsyncioTestCase):
             provider="gemini",
             profile=profile,
             base_url="https://generativelanguage.googleapis.com/v1beta",
-            api_key="gemini-secret-not-logged",
+            api_key="x",
             default_model=default_model,
             auth_mode="query_key",
         )
 
     async def test_native_models_payload_is_parsed_and_prefix_removed(self):
         async def handler(request):
-            self.assertEqual(request.url.params.get("key"), "gemini-secret-not-logged")
+            self.assertEqual(request.url.params.get("key"), "x")
             return httpx.Response(200, json={"models": [
                 {"name": "models/gemini-3.6-flash"},
                 {"name": "models/gemini-3.5-flash"},

tests/test_telemetry_alert_lifecycle.py

@@ -0,0 +1,45 @@
+import ast
+import pathlib
+import unittest
+
+
+_MAIN = pathlib.Path(__file__).parents[1] / "main.py"
+
+
+class TelemetryAlertLifecycleTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.tree = ast.parse(_MAIN.read_text(encoding="utf-8"))
+        cls.source = _MAIN.read_text(encoding="utf-8")
+        cls.functions = {
+            node.name: node
+            for node in cls.tree.body
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        }
+
+    def test_startup_registers_named_telemetry_task(self):
+        startup = self.functions["startup_event"]
+        source = ast.get_source_segment(self.source, startup)
+        self.assertIn("telemetry_alert_loop", source)
+        self.assertIn('name="telemetry-alert-loop"', source)
+        self.assertIn("_telemetry_alert_task", source)
+
+    def test_shutdown_cancels_and_awaits_telemetry_task(self):
+        shutdown = self.functions["shutdown_event"]
+        source = ast.get_source_segment(self.source, shutdown)
+        self.assertIn("task.cancel()", source)
+        self.assertIn("await task", source)
+        self.assertIn("asyncio.CancelledError", source)
+
+    def test_task_reference_is_module_scoped(self):
+        assignment_names = {
+            target.id
+            for node in self.tree.body
+            if isinstance(node, ast.AnnAssign)
+            and isinstance(target := node.target, ast.Name)
+        }
+        self.assertIn("_telemetry_alert_task", assignment_names)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_vfs_atomic_rollback.py

@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import asyncio
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from agents.unified_loop import UnifiedAgentLoop
+from tools.registry import _delete_file, _read_file, _write_file
+
+
+class FakeExecutor:
+    def __init__(self, failures: set[str] | None = None) -> None:
+        self.calls: list[tuple[str, dict]] = []
+        self.failures = failures or set()
+
+    async def run_tool(self, name: str, inputs: dict, timeout: float = 30.0) -> dict:
+        self.calls.append((name, inputs))
+        if name in self.failures:
+            return {"success": False, "error": f"forced failure: {name}", "output": None}
+        return {"success": True, "output": {"ok": True}}
+
+
+class VfsAtomicRollbackTests(unittest.IsolatedAsyncioTestCase):
+    def make_loop(self, executor: FakeExecutor) -> UnifiedAgentLoop:
+        loop = UnifiedAgentLoop.__new__(UnifiedAgentLoop)
+        loop.executor = executor
+        loop._write_snapshots = {}
+        return loop
+
+    async def test_rollback_restores_existing_and_deletes_new_files(self) -> None:
+        executor = FakeExecutor()
+        loop = self.make_loop(executor)
+        loop._write_snapshots = {"existing.txt": "before", "created.txt": None}
+
+        await loop._rollback_writes()
+
+        self.assertEqual(
+            [(name, inputs) for name, inputs in executor.calls],
+            [
+                ("write_file", {"path": "existing.txt", "content": "before"}),
+                ("delete_file", {"path": "created.txt"}),
+            ],
+        )
+        self.assertEqual(loop._write_snapshots, {})
+
+    async def test_failed_restore_is_not_marked_clean(self) -> None:
+        executor = FakeExecutor({"write_file"})
+        loop = self.make_loop(executor)
+        loop._write_snapshots = {"existing.txt": "before", "created.txt": None}
+
+        with self.assertRaisesRegex(RuntimeError, "rollback incompleto"):
+            await loop._rollback_writes()
+
+        self.assertEqual(loop._write_snapshots, {"existing.txt": "before"})
+        self.assertEqual(executor.calls[1][0], "delete_file")
+
+    async def test_delete_file_respects_fs_jail_and_is_idempotent(self) -> None:
+        with tempfile.TemporaryDirectory() as root:
+            with patch.dict("os.environ", {"FS_TOOL_ROOT": root}, clear=False):
+                target = Path(root) / "created.txt"
+                target.write_text("created", encoding="utf-8")
+                deleted = await _delete_file("created.txt")
+                repeated = await _delete_file("created.txt")
+                outside = await _delete_file("../outside.txt")
+
+        self.assertTrue(deleted["ok"])
+        self.assertTrue(deleted["deleted"])
+        self.assertTrue(repeated["ok"])
+        self.assertFalse(repeated["deleted"])
+        self.assertFalse(outside["ok"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/registry.py

@@ -685,6 +685,24 @@ async def _read_file(path: str, encoding: str = "utf-8") -> dict:
         return {"ok": False, "error": str(exc)}
 
 
+async def _delete_file(path: str) -> dict:
+    """Rimuove un file dalla jail VFS in modo idempotente."""
+    try:
+        _p, _err = _safe_fs_path(path)
+        if _err:
+            return {"ok": False, "error": _err}
+        if not _p.exists():
+            return {"ok": True, "path": path, "deleted": False}
+        if not _p.is_file():
+            return {"ok": False, "error": f"Non è un file: {path}"}
+        _p.unlink()
+        return {"ok": True, "path": path, "deleted": True}
+    except PermissionError:
+        return {"ok": False, "error": f"Accesso negato: {path}"}
+    except Exception as exc:
+        return {"ok": False, "error": str(exc)}
+
+
 async def _write_file(path: str, content: str, encoding: str = "utf-8") -> dict:
     """S666/GAP-2/SEC-FS-JAIL: Scrive/sovrascrive un file nel filesystem del backend, confinato a _fs_jail_root().
     Manus Gap fix: read-back verifica completezza scrittura.
@@ -2299,6 +2317,16 @@ TOOL_REGISTRY: dict[str, dict] = {
         "fallbacks": [],
         "_fn": _read_file,
     },
+    "delete_file": {
+        "name": "delete_file",
+        "goal": "Rimuove un file dal filesystem del backend",
+        "description": "Rimuove un file locale confinato alla jail VFS; idempotente se il file è già assente.",
+        "required_inputs": ["path"],
+        "optional_inputs": {},
+        "risk_level": "medium",
+        "fallbacks": [],
+        "_fn": _delete_file,
+    },
     "write_file": {
         "name": "write_file",
         "goal": "Scrive o sovrascrive un file nel filesystem del backend",