sync: 145 file da Baida98/AI@0405a40d (2026-08-10 16:35 UTC) [deploy-all]

agents/engineering_state.py

@@ -0,0 +1,255 @@
+"""Versioned, bounded engineering lifecycle state for the unified agent loop.
+
+The module is deliberately dependency-free.  It mirrors the legacy lifecycle without
+being authoritative for recovery when the rollout mode is enabled, and it never stores
+raw prompts, credentials, or arbitrary tool output.
+"""
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Mapping
+
+SCHEMA_VERSION = 1
+MAX_HISTORY = 64
+MAX_DIAGNOSTICS = 24
+MAX_PREVIEW_CHARS = 256
+MAX_ID_CHARS = 180
+
+_SECRET_PATTERNS = (
+    re.compile(r"(?i)(bearer\s+)[A-Za-z0-9._~+/=-]{8,}"),
+    re.compile(r"(?i)(api[_-]?key\s*[:=]\s*)[^\s,;]+"),
+    re.compile(r"(?i)(token\s*[:=]\s*)[^\s,;]+"),
+    re.compile(r"(?i)\b(?:ghp|gho|github_pat|hf|sk|xoxb|xapp|r8)_[A-Za-z0-9_-]{8,}\b"),
+    re.compile(r"\beyJ[A-Za-z0-9_-]{20,}\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b"),
+)
+
+
+class EngineeringStateMode(str, Enum):
+    OFF = "off"
+    SHADOW = "shadow"
+    CANARY = "canary"
+    AUTHORITATIVE = "authoritative"
+
+
+@dataclass(frozen=True)
+class EngineeringStateConfig:
+    """Conservative rollout configuration read once per run."""
+
+    mode: EngineeringStateMode = EngineeringStateMode.OFF
+    canary_rate: float = 0.0
+
+    @classmethod
+    def from_env(cls) -> "EngineeringStateConfig":
+        raw_mode = os.getenv("ENGINEERING_STATE_MODE", "authoritative").strip().lower()  # P1 default; off remains an explicit rollback mode
+        try:
+            mode = EngineeringStateMode(raw_mode)
+        except ValueError:
+            mode = EngineeringStateMode.OFF
+        try:
+            rate = float(os.getenv("ENGINEERING_STATE_CANARY_RATE", "0"))
+        except (TypeError, ValueError):
+            rate = 0.0
+        return cls(mode=mode, canary_rate=max(0.0, min(rate, 1.0)))
+
+    @property
+    def enabled(self) -> bool:
+        return self.mode is not EngineeringStateMode.OFF
+
+    def selects_canary(self, run_id: str, session_id: str) -> bool:
+        if self.mode is not EngineeringStateMode.CANARY or not session_id:
+            return False
+        if self.canary_rate >= 1.0:
+            return True
+        if self.canary_rate <= 0.0:
+            return False
+        digest = hashlib.sha256(f"{run_id}:{session_id}".encode()).digest()
+        bucket = int.from_bytes(digest[:8], "big") / float(2**64)
+        return bucket < self.canary_rate
+
+
+def _bounded_id(value: str | None) -> str:
+    return re.sub(r"[^A-Za-z0-9_.:/-]", "_", str(value or ""))[:MAX_ID_CHARS]
+
+
+def redact_text(value: object, max_chars: int = MAX_PREVIEW_CHARS) -> str:
+    """Redact common credential forms before anything reaches a checkpoint."""
+    text = str(value or "")[: max_chars * 4]
+    for pattern in _SECRET_PATTERNS:
+        if pattern.groups:
+            text = pattern.sub(lambda match: f"{match.group(1)}[REDACTED]", text)
+        else:
+            text = pattern.sub("[REDACTED]", text)
+    return text[:max_chars]
+
+
+_ALLOWED_TRANSITIONS: dict[str, frozenset[str]] = {
+    "IDLE": frozenset({"CLASSIFYING", "FAILED"}),
+    "CLASSIFYING": frozenset({"TOOL_EXECUTING", "THINKING", "COMPLETED", "FAILED"}),
+    "TOOL_EXECUTING": frozenset({"THINKING", "COMPLETED", "FAILED"}),
+    "THINKING": frozenset({"COMPLETED", "FAILED"}),
+    "FAILED": frozenset({"IDLE"}),
+    "COMPLETED": frozenset({"IDLE", "FAILED"}),
+}
+
+
+@dataclass
+class EngineeringState:
+    """Bounded state envelope that can be persisted and safely restored."""
+
+    run_id: str
+    session_id: str
+    checkpoint_id: str
+    goal_digest: str
+    goal_preview: str
+    current_state: str = "IDLE"
+    history: list[dict[str, Any]] = field(default_factory=list)
+    diagnostics: list[str] = field(default_factory=list)
+    revision: int = 0
+    sequence: int = 0
+    created_at_ms: int = field(default_factory=lambda: int(time.time() * 1000))
+    updated_at_ms: int = field(default_factory=lambda: int(time.time() * 1000))
+
+    @classmethod
+    def start(
+        cls,
+        goal: str,
+        *,
+        run_id: str,
+        session_id: str = "",
+        checkpoint_id: str | None = None,
+        now_ms: int | None = None,
+    ) -> "EngineeringState":
+        now = int(time.time() * 1000) if now_ms is None else int(now_ms)
+        normalized_goal = str(goal or "")
+        return cls(
+            run_id=_bounded_id(run_id),
+            session_id=_bounded_id(session_id),
+            checkpoint_id=_bounded_id(checkpoint_id or session_id or run_id),
+            goal_digest=hashlib.sha256(normalized_goal.encode("utf-8", "replace")).hexdigest(),
+            goal_preview=redact_text(normalized_goal),
+            created_at_ms=now,
+            updated_at_ms=now,
+        )
+
+    @property
+    def status(self) -> str:
+        if self.current_state == "COMPLETED":
+            return "completed"
+        if self.current_state == "FAILED":
+            return "failed"
+        return "active"
+
+    def transition(self, next_state: str, *, now_ms: int | None = None) -> bool:
+        """Apply an idempotent transition; reject illegal transitions deterministically."""
+        target = str(next_state)
+        if target == self.current_state:
+            return False
+        allowed = _ALLOWED_TRANSITIONS.get(self.current_state, frozenset())
+        if target not in allowed:
+            raise ValueError(f"Invalid EngineeringState transition: {self.current_state} -> {target}")
+        now = int(time.time() * 1000) if now_ms is None else int(now_ms)
+        self.sequence += 1
+        self.revision += 1
+        self.history.append({
+            "sequence": self.sequence,
+            "from_state": self.current_state,
+            "to_state": target,
+            "at_ms": now,
+        })
+        if len(self.history) > MAX_HISTORY:
+            del self.history[:-MAX_HISTORY]
+        self.current_state = target
+        self.updated_at_ms = now
+        return True
+
+    def prepare_for_resume(self) -> None:
+        """Normalize a restored snapshot before a new loop execution."""
+        if self.current_state != "IDLE":
+            self.current_state = "IDLE"
+            self.revision += 1
+            self.updated_at_ms = int(time.time() * 1000)
+            self.diagnostic("resume normalized state to IDLE")
+
+    def diagnostic(self, message: str) -> None:
+        value = redact_text(message, 180)
+        if not value or value in self.diagnostics:
+            return
+        self.diagnostics.append(value)
+        if len(self.diagnostics) > MAX_DIAGNOSTICS:
+            del self.diagnostics[:-MAX_DIAGNOSTICS]
+        self.revision += 1
+        self.updated_at_ms = int(time.time() * 1000)
+
+    def snapshot(self) -> dict[str, Any]:
+        """Return a bounded JSON-compatible envelope; never expose the raw goal."""
+        return {
+            "schema_version": SCHEMA_VERSION,
+            "run_id": self.run_id,
+            "session_id": self.session_id,
+            "checkpoint_id": self.checkpoint_id,
+            "goal_digest": self.goal_digest,
+            "goal_preview": self.goal_preview,
+            "status": self.status,
+            "current_state": self.current_state,
+            "revision": self.revision,
+            "sequence": self.sequence,
+            "history": list(self.history[-MAX_HISTORY:]),
+            "diagnostics": list(self.diagnostics[-MAX_DIAGNOSTICS:]),
+            "created_at_ms": self.created_at_ms,
+            "updated_at_ms": self.updated_at_ms,
+        }
+
+    def projection(self) -> dict[str, Any]:
+        """Small read-only view safe for API/SSE consumers."""
+        return {
+            "schema_version": SCHEMA_VERSION,
+            "status": self.status,
+            "current_state": self.current_state,
+            "revision": self.revision,
+            "sequence": self.sequence,
+            "checkpoint_id": self.checkpoint_id,
+            "history": [dict(item) for item in self.history[-16:]],
+            "diagnostics": list(self.diagnostics[-MAX_DIAGNOSTICS:]),
+        }
+
+    @classmethod
+    def from_snapshot(cls, payload: Mapping[str, Any]) -> "EngineeringState":
+        if not isinstance(payload, Mapping):
+            raise ValueError("engineering state must be an object")
+        if int(payload.get("schema_version", -1)) != SCHEMA_VERSION:
+            raise ValueError("unsupported engineering state schema")
+        history = payload.get("history", [])
+        diagnostics = payload.get("diagnostics", [])
+        if not isinstance(history, list) or len(history) > MAX_HISTORY:
+            raise ValueError("invalid engineering state history")
+        if not isinstance(diagnostics, list) or len(diagnostics) > MAX_DIAGNOSTICS:
+            raise ValueError("invalid engineering state diagnostics")
+        current = str(payload.get("current_state", ""))
+        if current not in _ALLOWED_TRANSITIONS:
+            raise ValueError("invalid engineering state current state")
+        revision = int(payload.get("revision", -1))
+        sequence = int(payload.get("sequence", -1))
+        if revision < 0 or sequence < 0 or revision < sequence:
+            raise ValueError("invalid engineering state revision")
+        state = cls(
+            run_id=_bounded_id(str(payload.get("run_id", ""))),
+            session_id=_bounded_id(str(payload.get("session_id", ""))),
+            checkpoint_id=_bounded_id(str(payload.get("checkpoint_id", ""))),
+            goal_digest=str(payload.get("goal_digest", "")),
+            goal_preview=redact_text(payload.get("goal_preview", "")),
+            current_state=current,
+            history=[dict(item) for item in history if isinstance(item, Mapping)],
+            diagnostics=[redact_text(item, 180) for item in diagnostics],
+            revision=revision,
+            sequence=sequence,
+            created_at_ms=int(payload.get("created_at_ms", 0)),
+            updated_at_ms=int(payload.get("updated_at_ms", 0)),
+        )
+        if len(state.goal_digest) != 64 or not re.fullmatch(r"[0-9a-f]{64}", state.goal_digest):
+            raise ValueError("invalid engineering state goal digest")
+        return state

agents/unified_loop.py

@@ -63,6 +63,48 @@ from agents.unified_loop_types import (
 # I4.5: active state is scoped to the current asyncio task, not the loop instance.
 # This lets the public guard close unexpected exceptions without sharing state across runs.
 _ACTIVE_LOOP_STATE: ContextVar[UnifiedLoopState | None] = ContextVar("active_loop_state", default=None)
+# P0: EngineeringState is a shadow/canary projection of the legacy lifecycle.
+# Context-local storage keeps parallel runs isolated even when one loop instance is reused.
+from agents.engineering_state import EngineeringState, EngineeringStateConfig, EngineeringStateMode
+
+_ACTIVE_ENGINEERING_STATE: ContextVar[EngineeringState | None] = ContextVar(
+    "active_engineering_state", default=None
+)
+_ACTIVE_ENGINEERING_MODE: ContextVar[EngineeringStateMode | None] = ContextVar(
+    "active_engineering_mode", default=None
+)
+
+
+def _schedule_engineering_persist(engineering_state: EngineeringState) -> None:
+    """Persist a snapshot without blocking the loop or making observability fatal."""
+    snapshot = engineering_state.snapshot()
+
+    async def _persist() -> None:
+        try:
+            from api.persistence import sb_save_engineering_state
+            await sb_save_engineering_state(snapshot["checkpoint_id"], snapshot)
+        except Exception as exc:  # shadow state must never break the user task
+            _logger.debug("[engineering-state] persist silenced: %s", type(exc).__name__)
+
+    try:
+        task = asyncio.create_task(_persist())
+        task.add_done_callback(lambda done: done.exception() if not done.cancelled() else None)
+    except RuntimeError:
+        # No running event loop during defensive/test-only calls.
+        return
+
+
+async def _flush_engineering_persist(engineering_state: EngineeringState | None) -> None:
+    """Flush the terminal snapshot before returning a run result."""
+    if engineering_state is None:
+        return
+    snapshot = engineering_state.snapshot()
+    try:
+        from api.persistence import sb_save_engineering_state
+        await sb_save_engineering_state(snapshot["checkpoint_id"], snapshot, force=True)
+    except Exception as exc:  # persistence must not turn a completed task into a crash
+        engineering_state.diagnostic(f"final persist failed: {type(exc).__name__}")
+        _logger.debug("[engineering-state] final persist silenced: %s", type(exc).__name__)
 
 # S404: Error Classifier — import lazy per evitare circular import issues
 def _get_classifier():
@@ -143,15 +185,31 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
         """Validate and publish one per-run state transition."""
         previous = state.state_machine.current
         state.state_machine.transition(next_state)
+
+        # P0 adapter: mirror every legacy transition into the versioned state.
+        engineering_state = _ACTIVE_ENGINEERING_STATE.get()
+        if engineering_state is not None:
+            try:
+                engineering_state.transition(next_state.value)
+                _schedule_engineering_persist(engineering_state)
+            except Exception as exc:
+                engineering_state.diagnostic(f"transition adapter: {type(exc).__name__}")
+                if _ACTIVE_ENGINEERING_MODE.get() == EngineeringStateMode.AUTHORITATIVE:
+                    raise
+                _logger.debug("[engineering-state] transition silenced: %s", type(exc).__name__)
+
         if previous == next_state or on_step is None:
             return
         try:
-            await _maybe_await(on_step({
+            event = {
                 "action": "state_transition",
                 "status": "done",
                 "from_state": previous.value,
                 "to_state": next_state.value,
-            }))
+            }
+            if engineering_state is not None:
+                event["engineering_state"] = engineering_state.projection()
+            await _maybe_await(on_step(event))
         except Exception as _state_callback_error:
             _logger.debug("[unified_loop] state callback silenced: %s", _state_callback_error)
 
@@ -545,7 +603,12 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
                             "explanation": "Il pianificatore ha impiegato troppo — procedo senza piano",
                             "visibility": "progress",
                         }))
-            if plan is not None:
+            # P1-RECOVERY: check if plan already exists in steps
+            existing_plan_step = next((s for s in state.steps if s.get("action") == "plan"), None)
+            if existing_plan_step:
+                plan = existing_plan_step.get("result")
+                _logger.info("[P1-RECOVERY] Plan restored from steps")
+            elif plan is not None:
                 state.steps.append({"action": "plan", "result": plan})
             try:
                 from api.state import record_timing as _rtc_pl
@@ -952,6 +1015,15 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
                     tool_key_pair = _TOOL_MAP.get(_s_tool, (None, None))
                     reg_name, inp_builder = tool_key_pair
                     if reg_name and inp_builder is not None:
+                        # P1-RECOVERY: skip subtasks already completed in state.steps
+                        _st_id = subtask.get("id")
+                        _done_step = next((s for s in state.steps if s.get("subtask_id") == _st_id), None)
+                        if _done_step:
+                            _logger.info("[P1-RECOVERY] Skipping already completed subtask #%s", _st_id)
+                            # Ripristiniamo l'output nel buffer per i dipendenti
+                            _existing_out = _done_step.get("output", "")
+                            _subtask_outputs[str(_st_id)] = _existing_out
+                            continue
                         _pending_exec.append((subtask, reg_name, inp_builder))
                     elif _s_tool:
                         # COG-4: tool non in _TOOL_MAP — tenta generazione dinamica
@@ -3389,6 +3461,8 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
                   session_id: str = "") -> dict[str, Any]:
         """Run the loop and close unexpected exceptions as a controlled FAILED state."""
         previous_state = _ACTIVE_LOOP_STATE.get()
+        previous_engineering_state = _ACTIVE_ENGINEERING_STATE.get()
+        previous_engineering_mode = _ACTIVE_ENGINEERING_MODE.get()
         try:
             return await self._run_impl(goal, context, max_steps, on_step, session_id)
         except Exception as _run_error:
@@ -3406,17 +3480,14 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
             state.errors.append(error_text)
             previous = state.state_machine.current
             if previous != AgentState.FAILED:
-                state.state_machine.transition(AgentState.FAILED)
-                if on_step is not None:
-                    try:
-                        await _maybe_await(on_step({
-                            "action": "state_transition",
-                            "status": "done",
-                            "from_state": previous.value,
-                            "to_state": AgentState.FAILED.value,
-                        }))
-                    except Exception as _state_callback_error:
-                        _logger.debug("[unified_loop] failure callback silenced: %s", _state_callback_error)
+                try:
+                    await self._transition_state(state, AgentState.FAILED, on_step)
+                except Exception as _state_transition_error:
+                    _logger.debug(
+                        "[unified_loop] failure transition silenced: %s",
+                        _state_transition_error,
+                    )
+            await _flush_engineering_persist(_ACTIVE_ENGINEERING_STATE.get())
             return {
                 "success": False,
                 "goal": state.goal,
@@ -3427,6 +3498,8 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
             }
         finally:
             _ACTIVE_LOOP_STATE.set(previous_state)
+            _ACTIVE_ENGINEERING_STATE.set(previous_engineering_state)
+            _ACTIVE_ENGINEERING_MODE.set(previous_engineering_mode)
 
     async def _run_impl(self, goal: str, context: str = "", max_steps: int = 8,
                        on_step: StepCallback | None = None,
@@ -3489,16 +3562,83 @@ class UnifiedAgentLoop(DirectToolsMixin, PromptBuilderMixin, LLMSelectionMixin,
             max_steps = 12
 
         state = UnifiedLoopState(goal=goal, context=context, max_steps=max_steps, session_id=session_id)
+
+        # P1: EngineeringState is the recovery authority unless explicitly disabled.
+        engineering_config = EngineeringStateConfig.from_env()
+        _effective_mode = engineering_config.mode
+        _ACTIVE_ENGINEERING_MODE.set(_effective_mode)
+
+        engineering_state: EngineeringState | None = None
+        recovery_status = "disabled"
+        if _effective_mode != EngineeringStateMode.OFF:
+            engineering_state = EngineeringState.start(
+                goal,
+                run_id=self._run_task_id,
+                session_id=session_id,
+                checkpoint_id=session_id or self._run_task_id,
+            )
+            _ACTIVE_ENGINEERING_STATE.set(engineering_state)
+            recovery_status = "started"
+
+            # RECOV-P1.1/P1.2: load and validate EngineeringState before the first transition.
+            if _effective_mode.value in {"canary", "authoritative"} and engineering_state.checkpoint_id:
+                try:
+                    from api.persistence import sb_get_checkpoint
+                    legacy_checkpoint = await sb_get_checkpoint(engineering_state.checkpoint_id)
+                    candidate = (legacy_checkpoint or {}).get("engineering_state")
+                    if candidate:
+                        restored = EngineeringState.from_snapshot(candidate)
+                        if restored.session_id != engineering_state.session_id or restored.goal_digest != engineering_state.goal_digest:
+                            engineering_state.diagnostic("restore conflict: identity mismatch")
+                            recovery_status = "conflict"
+                        elif _effective_mode == EngineeringStateMode.AUTHORITATIVE:
+                            engineering_state = restored
+                            engineering_state.prepare_for_resume()
+                            _ACTIVE_ENGINEERING_STATE.set(engineering_state)
+                            if legacy_checkpoint:
+                                checkpoint_steps = legacy_checkpoint.get("steps")
+                                checkpoint_errors = legacy_checkpoint.get("errors")
+                                state.steps = list(checkpoint_steps)[-64:] if isinstance(checkpoint_steps, list) else []
+                                state.errors = [str(item)[:512] for item in checkpoint_errors][-24:] if isinstance(checkpoint_errors, list) else []
+                            recovery_status = "restored"
+                            _logger.info("[P1-RECOVERY] authoritative checkpoint restored revision=%d", restored.revision)
+                        else:
+                            engineering_state.diagnostic("restore validated read-only")
+                            recovery_status = "validated"
+                    else:
+                        recovery_status = "checkpoint_missing"
+                except Exception as restore_error:
+                    engineering_state.diagnostic(f"restore rejected: {type(restore_error).__name__}")
+                    recovery_status = "rejected"
+                    _logger.debug("[engineering-state] restore silenced: %s", type(restore_error).__name__)
+
         _ACTIVE_LOOP_STATE.set(state)
         await self._transition_state(state, AgentState.CLASSIFYING, on_step)
 
         def _with_state(result: dict[str, Any]) -> dict[str, Any]:
             result.update(state.state_machine.snapshot())
+            if engineering_state is not None:
+                result["engineering_state"] = engineering_state.projection()
             return result
 
+        if engineering_state is not None and on_step is not None:
+            try:
+                await _maybe_await(on_step({
+                    "action": "engineering_state",
+                    "status": recovery_status,
+                    "mode": _effective_mode.value,
+                    "engineering_state": engineering_state.projection(),
+                }))
+            except Exception as recovery_event_error:
+                _logger.debug("[engineering-state] recovery event silenced: %s", type(recovery_event_error).__name__)
+
         async def _finish(result: dict[str, Any]) -> dict[str, Any]:
             next_state = AgentState.COMPLETED if result.get("success", True) else AgentState.FAILED
-            await self._transition_state(state, next_state, on_step)
+            try:
+                await self._transition_state(state, next_state, on_step)
+            finally:
+                # P1 contract: persist the terminal state before returning to the caller.
+                await _flush_engineering_persist(_ACTIVE_ENGINEERING_STATE.get())
             return _with_state(result)
 
         # GAP-4: StrategicHealer — init + load past failures (LLM-based self-healing cognitivo)

agents/unified_loop_prompts.py

@@ -47,9 +47,10 @@ class PromptBuilderMixin:
         "8b. Per domande a scelta multipla (A/B/C/D): inizia la risposta con "
         "\'Risposta: X\' dove X è la lettera scelta, poi spiega il ragionamento.\n"
         "8c. OBBLIGO TypeScript: ogni snippet di codice TypeScript DEVE essere in blocchi "
-        "\'\'\'typescript\'\'\'...\'\'\'typescript. Mai inline, mai in blocchi generici. "
-        "Il codice deve compilare: nessun placeholder, nessun TODO, tipi espliciti.\n"
-        "9. Per decisioni architetturali: dai 3 opzioni con pro/contro e raccomandazione\n"
+        "```typescript```...```typescript. Mai inline, mai in blocchi generici. "
+        "Il codice deve compilare: nessun placeholder, nessun TODO, tipi espliciti. In caso di REFACTORING: sostituisci SEMPRE nomi di variabili a lettera singola (p, m, v) con nomi semantici e descrittivi, e usa interfacce o tipi per ogni oggetto complesso.\n"
+        "8d. REASONING: Per problemi complessi, scomponi il problema in sotto-task logici. Verifica la coerenza dei risultati intermedi prima di procedere al calcolo finale.
+9. Per decisioni architetturali: dai 3 opzioni con pro/contro e raccomandazione\n"
         "10. NON inventare mai informazioni su te stesso: token usati, context window, "
         "versione, architettura, parametri interni. Se non lo sai con certezza, "
         "di esplicitamente 'non ho accesso a questa informazione'.\n"
@@ -1829,3 +1830,4 @@ _CONTEXT_RULES_ADVANCED = [
 ]
 
 
+

api/agent.py

@@ -1067,6 +1067,15 @@ async def stream_agent_task(task_id: str, request: Request, resume: int = 0, rol
                     if _action == 'text_chunk':
                         _sse('text_chunk', {'taskId': task_id, 'token': _ss(step_data.get('token', ''))})
                         return
+                    # RECOV-P1: engineering_state event — forward projection to frontend
+                    if _action == 'engineering_state':
+                        _sse('engineering_state', {
+                            'taskId':            task_id,
+                            'status':            step_data.get('status'),
+                            'mode':              step_data.get('mode'),
+                            'engineering_state': step_data.get('engineering_state'),
+                        })
+                        return
 
                     # S363-Blueprint: Narrative Streaming — explanation lookup for ALL step_done events
                     # S376: _STEP_NARRATIONS espanso — aggiunge 12 tool mancanti
@@ -1409,7 +1418,7 @@ async def save_checkpoint(task_id: str, body: CheckpointIn, role: AuthRole = Dep
         'extra':      body.extra,
         'savedAt':    int(time.time() * 1000),
     }
-    asyncio.create_task(sb_save_checkpoint(task_id, _task_checkpoints[task_id])).add_done_callback(_log_task_exc)
+    asyncio.create_task(sb_save_checkpoint(task_id, body.step, _task_checkpoints[task_id])).add_done_callback(_log_task_exc)
     return {'saved': True, 'taskId': task_id, 'step': body.step}
 
 

api/agent_checkpoint.py

@@ -95,7 +95,7 @@ async def save_checkpoint_alias(body: CheckpointBody):
     }
     _task_checkpoints[task_id] = cp
     # Persist su Supabase — fire-and-forget (stesso pattern di agent.py)
-    asyncio.create_task(sb_save_checkpoint(task_id, cp))
+    asyncio.create_task(sb_save_checkpoint(task_id, body.step, cp))
     return {"saved": True, "taskId": task_id, "step": body.step}
 
 

api/auth_guard.py

@@ -96,6 +96,30 @@ _RATE_LIMITS: dict[int, int] = {
 _RATE_WINDOW_S = 60  # finestra sliding 60s
 _rate_store: dict[str, _col.deque] = {}  # token_hash → deque di timestamps
 
+# Lo store è usato anche quando Redis non è disponibile. Un client una tantum
+# lasciava una deque vuota nel dict per l'intera vita del processo. Eseguiamo uno
+# sweep ammortizzato: il lavoro resta O(1) per la quasi totalità delle richieste
+# e il numero di chiavi inattive rimane limitato al traffico tra due sweep.
+_RATE_STORE_SWEEP_EVERY = 128
+_rate_store_checks = 0
+
+
+def _prune_expired_rate_keys(now: float, window_s: float) -> None:
+    """Rimuove bucket in-memory senza timestamp ancora nella finestra corrente."""
+    global _rate_store_checks
+    _rate_store_checks += 1
+    if _rate_store_checks % _RATE_STORE_SWEEP_EVERY:
+        return
+
+    window_start = now - window_s
+    stale_keys = [
+        stored_key
+        for stored_key, timestamps in _rate_store.items()
+        if not timestamps or timestamps[-1] < window_start
+    ]
+    for stored_key in stale_keys:
+        _rate_store.pop(stored_key, None)
+
 
 def _rate_key(role: int, token_header: str | None, client_ip: str | None = None) -> str:
     """Chiave rate limiter: hash(role + discriminante) — non espone token né IP in chiaro.
@@ -123,6 +147,7 @@ def _inmem_rate_check(key: str, limit: int, window_s: float) -> tuple[bool, int]
     """
     now = _rl_time.monotonic()
     window_start = now - window_s
+    _prune_expired_rate_keys(now, window_s)
 
     if key not in _rate_store:
         _rate_store[key] = _col.deque()

api/persistence.py

@@ -27,6 +27,27 @@ MAX_EVENTS = 500  # max SSE frames persisted per task
 _MAX_RETRY   = 2    # GAP-P40D-FIX: tentativi massimi per write Supabase
 _RETRY_SLEEP = 0.3  # GAP-P40D-FIX: sleep tra tentativi (secondi)
 
+# P0: per-run locks serialize compatible envelope updates in one worker.
+_ENGINEERING_LOCKS: dict[str, asyncio.Lock] = {}
+_ENGINEERING_LOCK_LAST_USED: dict[str, float] = {}
+_ENGINEERING_LOCK_MAX = 256
+
+
+def _engineering_lock(task_id: str) -> asyncio.Lock:
+    lock = _ENGINEERING_LOCKS.get(task_id)
+    if lock is None:
+        lock = asyncio.Lock()
+        _ENGINEERING_LOCKS[task_id] = lock
+    _ENGINEERING_LOCK_LAST_USED[task_id] = time.monotonic()
+    if len(_ENGINEERING_LOCKS) > _ENGINEERING_LOCK_MAX:
+        for stale_id, _ in sorted(_ENGINEERING_LOCK_LAST_USED.items(), key=lambda item: item[1]):
+            stale_lock = _ENGINEERING_LOCKS.get(stale_id)
+            if stale_lock is not None and not stale_lock.locked() and stale_id != task_id:
+                _ENGINEERING_LOCKS.pop(stale_id, None)
+                _ENGINEERING_LOCK_LAST_USED.pop(stale_id, None)
+                break
+    return lock
+
 
 # ── Write helpers (fire-and-forget, never raise) ───────────────────────────────
 
@@ -177,22 +198,101 @@ async def sb_list_tasks(limit: int = 50) -> list[dict]:
 # ── Checkpoint helpers (S359: task state snapshots) ───────────────────────────
 
 async def sb_save_checkpoint(task_id: str, step: int, checkpoint_data: dict) -> None:
-    """Save a mid-task checkpoint for potential resume."""
+    """Save a legacy checkpoint while preserving a valid EngineeringState envelope."""
     from .state import _sb
     if not _sb:
         return
     now = int(time.time() * 1000)
     try:
-        await asyncio.to_thread(
-            lambda: _sb.table('agent_tasks')
-                .update({'checkpoint': _sjd(checkpoint_data)[:16000], 'updated_at': now})
-                .eq('task_id', task_id)
-                .execute()
-        )
+        payload = dict(checkpoint_data) if isinstance(checkpoint_data, dict) else {}
+        # A legacy save must not erase the shadow/canary envelope written by the
+        # adapter. Read/merge under the same per-task lock used by its writer.
+        lock = _engineering_lock(task_id)
+        async with lock:
+            current = await sb_get_checkpoint(task_id)
+            current_engineering = current.get('engineering_state') if isinstance(current, dict) else None
+            if isinstance(current_engineering, dict) and 'engineering_state' not in payload:
+                payload['engineering_state'] = current_engineering
+            serialized = _sjd(payload)
+            if len(serialized) > 16000:
+                _logger.debug('[persist] save_checkpoint %s#%d skipped: payload exceeds size limit', task_id, step)
+                return
+            await asyncio.to_thread(
+                lambda: _sb.table('agent_tasks')
+                    .update({'checkpoint': serialized, 'updated_at': now})
+                    .eq('task_id', task_id)
+                    .execute()
+            )
     except Exception as e:
         _logger.debug('[persist] save_checkpoint %s#%d: %s', task_id, step, e)
 
 
+_ENGINEERING_DEBOUNCE_CACHE: dict[str, dict[str, Any]] = {}
+_ENGINEERING_LAST_FLUSH_TS: dict[str, float] = {}
+DEBOUNCE_INTERVAL_SEC = 2.0
+
+async def sb_save_engineering_state(task_id: str, envelope: dict, force: bool = False) -> None:
+    """Merge a validated EngineeringState envelope with debouncing and monotone revision check."""
+    from .state import _sb
+    if not _sb or not task_id:
+        return
+    try:
+        from agents.engineering_state import EngineeringState
+        validated = EngineeringState.from_snapshot(envelope).snapshot()
+    except Exception as exc:
+        _logger.debug('[persist] engineering state rejected: %s', type(exc).__name__)
+        return
+
+    lock = _engineering_lock(task_id)
+    async with lock:
+        current = await sb_get_checkpoint(task_id)
+        current = current if isinstance(current, dict) else {}
+        current_engineering = current.get('engineering_state')
+        try:
+            current_revision = int(current_engineering.get('revision', -1)) if isinstance(current_engineering, dict) else -1
+        except (TypeError, ValueError):
+            current_revision = -1
+        incoming_revision = int(validated.get('revision', -1))
+        if current_revision > incoming_revision:
+            _logger.debug('[persist] engineering state conflict %s: remote revision %d > %d', task_id, current_revision, incoming_revision)
+            return
+    now_t = time.time()
+    _ENGINEERING_DEBOUNCE_CACHE[task_id] = validated
+    if not force and task_id in _ENGINEERING_LAST_FLUSH_TS:
+        if now_t - _ENGINEERING_LAST_FLUSH_TS[task_id] < DEBOUNCE_INTERVAL_SEC:
+            return
+
+    _ENGINEERING_LAST_FLUSH_TS[task_id] = now_t
+    to_flush = _ENGINEERING_DEBOUNCE_CACHE.get(task_id, validated)
+
+    async with lock:
+        current = await sb_get_checkpoint(task_id)
+        current = current if isinstance(current, dict) else {}
+        current_engineering = current.get('engineering_state')
+        try:
+            current_revision = int(current_engineering.get('revision', -1)) if isinstance(current_engineering, dict) else -1
+        except (TypeError, ValueError):
+            current_revision = -1
+        incoming_revision = int(to_flush.get('revision', -1))
+        if current_revision > incoming_revision and not force:
+            return
+        merged = dict(current)
+        merged['engineering_state'] = to_flush
+        serialized = _sjd(merged)
+        if len(serialized) > 16000:
+            return
+        now = int(time.time() * 1000)
+        try:
+            await asyncio.to_thread(
+                lambda: _sb.table('agent_tasks')
+                    .update({'checkpoint': serialized, 'updated_at': now})
+                    .eq('task_id', task_id)
+                    .execute()
+            )
+        except Exception as exc:
+            _logger.debug('[persist] save_engineering_state %s: %s', task_id, exc)
+
+
 async def sb_get_checkpoint(task_id: str) -> Optional[dict]:
     """Retrieve latest checkpoint for a task."""
     from .state import _sb

api/scheduler.py

@@ -26,6 +26,7 @@ Route:
 import asyncio
 import datetime
 import json
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 from .state import safe_json_dumps
 import os
 import time
@@ -198,6 +199,23 @@ def _is_due(task: dict, now_ms: int) -> bool:
     return False
 
 
+def _daily_timezone(trigger: dict) -> ZoneInfo | None:
+    """Ritorna il fuso IANA salvato dal browser, se disponibile e valido.
+
+    I task daily creati prima dell'introduzione del campo ``timeZone`` restano
+    compatibili: l'assenza o un valore non valido mantiene il calcolo nel fuso
+    locale del server invece di bloccare la pianificazione.
+    """
+    time_zone = trigger.get("timeZone")
+    if not isinstance(time_zone, str) or not time_zone:
+        return None
+    try:
+        return ZoneInfo(time_zone)
+    except ZoneInfoNotFoundError:
+        logger.warning("Scheduler: timezone daily non valida (%r), fallback server-local", time_zone)
+        return None
+
+
 def _advance_trigger(trigger: dict, now_ms: int) -> dict:
     t  = dict(trigger)
     tt = t.get("type")
@@ -206,14 +224,17 @@ def _advance_trigger(trigger: dict, now_ms: int) -> dict:
     elif tt == "daily":
         hour   = t.get("hour",   9)
         minute = t.get("minute", 0)
-        nxt    = datetime.datetime.now().replace(
-            hour=hour, minute=minute, second=0, microsecond=0
-        )
-        nxt_ms = int(nxt.timestamp() * 1000)
-        if nxt_ms <= now_ms:
-            nxt    = nxt + datetime.timedelta(days=1)
-            nxt_ms = int(nxt.timestamp() * 1000)
-        t["nextRun"] = nxt_ms
+        time_zone = _daily_timezone(t)
+        # Usa il timestamp dell'esecuzione, non l'orologio nel momento in cui
+        # il task termina: preserva la semantica esistente anche per task lunghi.
+        now = datetime.datetime.fromtimestamp(
+            now_ms / 1000,
+            tz=time_zone,
+        ) if time_zone else datetime.datetime.fromtimestamp(now_ms / 1000)
+        nxt = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
+        if nxt <= now:
+            nxt = nxt + datetime.timedelta(days=1)
+        t["nextRun"] = int(nxt.timestamp() * 1000)
     # once / on_open: nessun avanzamento
     return t
 

tests/test_auth_scheduler_regressions.py

@@ -0,0 +1,111 @@
+"""Regressioni auth/scheduler: cleanup rate limiter e timezone daily.
+
+Esegui con: python3 -m unittest backend.tests.test_auth_scheduler_regressions -v
+"""
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from collections import deque
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+_BACKEND = os.path.join(os.path.dirname(__file__), "..")
+if _BACKEND not in sys.path:
+    sys.path.insert(0, _BACKEND)
+
+
+class TestInMemoryRateStoreCleanup(unittest.TestCase):
+    """AUTH-RATE-LEAK: bucket inattivi non devono restare nel processo."""
+
+    def setUp(self) -> None:
+        try:
+            import api.auth_guard as auth_guard
+        except ImportError as exc:
+            self.skipTest(str(exc))
+        self.auth_guard = auth_guard
+        auth_guard._rate_store.clear()
+        auth_guard._rate_store_checks = 0
+
+    def tearDown(self) -> None:
+        self.auth_guard._rate_store.clear()
+        self.auth_guard._rate_store_checks = 0
+
+    def test_periodic_sweep_removes_expired_empty_bucket(self) -> None:
+        self.auth_guard._rate_store["expired-client"] = deque([1.0])
+        self.auth_guard._rate_store_checks = self.auth_guard._RATE_STORE_SWEEP_EVERY - 1
+
+        with patch.object(self.auth_guard._rl_time, "monotonic", return_value=120.0):
+            allowed, retry_after = self.auth_guard._inmem_rate_check(
+                "active-client", limit=10, window_s=60
+            )
+
+        self.assertTrue(allowed)
+        self.assertEqual(retry_after, 0)
+        self.assertNotIn(
+            "expired-client",
+            self.auth_guard._rate_store,
+            "AUTH-RATE-LEAK: il bucket inattivo resta nello store dopo lo sweep",
+        )
+        self.assertIn("active-client", self.auth_guard._rate_store)
+
+    def test_current_request_survives_its_own_sweep(self) -> None:
+        self.auth_guard._rate_store_checks = self.auth_guard._RATE_STORE_SWEEP_EVERY - 1
+
+        with patch.object(self.auth_guard._rl_time, "monotonic", return_value=120.0):
+            allowed, _ = self.auth_guard._inmem_rate_check(
+                "current-client", limit=1, window_s=60
+            )
+
+        self.assertTrue(allowed)
+        self.assertIn("current-client", self.auth_guard._rate_store)
+
+
+class TestDailyTriggerTimezone(unittest.TestCase):
+    """SCHED-TZ-DRIFT: il backend deve conservare l'ora civile scelta dal browser."""
+
+    def setUp(self) -> None:
+        try:
+            import api.scheduler as scheduler
+        except ImportError as exc:
+            self.skipTest(str(exc))
+        self.scheduler = scheduler
+
+    def _advance(self, iso_now: str) -> datetime:
+        now = datetime.fromisoformat(iso_now)
+        result = self.scheduler._advance_trigger(
+            {
+                "type": "daily",
+                "hour": 9,
+                "minute": 0,
+                "nextRun": int(now.timestamp() * 1000),
+                "timeZone": "Europe/Rome",
+            },
+            int(now.timestamp() * 1000),
+        )
+        return datetime.fromtimestamp(result["nextRun"] / 1000, tz=timezone.utc)
+
+    def test_daily_uses_browser_timezone_not_utc_server_timezone(self) -> None:
+        # 09:00 CEST è 07:00 UTC. Essendo già l'orario pianificato, il run successivo
+        # deve restare alle 09:00 civili del giorno seguente (07:00 UTC), non 09:00 UTC.
+        actual = self._advance("2026-06-01T07:00:00+00:00")
+        self.assertEqual(actual, datetime(2026, 6, 2, 7, 0, tzinfo=timezone.utc))
+
+    def test_daily_preserves_wall_clock_across_dst_transition(self) -> None:
+        # Il giorno dopo l'Europa passa da CET (UTC+1) a CEST (UTC+2): l'ora civile
+        # deve rimanere 09:00, quindi l'epoch UTC passa correttamente da 08:00 a 07:00.
+        actual = self._advance("2026-03-28T08:00:00+00:00")
+        self.assertEqual(actual, datetime(2026, 3, 29, 7, 0, tzinfo=timezone.utc))
+
+    def test_legacy_daily_trigger_without_timezone_remains_schedulable(self) -> None:
+        now = datetime(2026, 6, 1, 7, 0, tzinfo=timezone.utc)
+        result = self.scheduler._advance_trigger(
+            {"type": "daily", "hour": 9, "minute": 0, "nextRun": int(now.timestamp() * 1000)},
+            int(now.timestamp() * 1000),
+        )
+        self.assertGreater(result["nextRun"], int(now.timestamp() * 1000))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_engineering_state.py

@@ -0,0 +1,92 @@
+"""Focused P0 tests for EngineeringState's safety and rollout contract."""
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+_BACKEND = os.path.join(os.path.dirname(__file__), "..")
+if _BACKEND not in sys.path:
+    sys.path.insert(0, _BACKEND)
+
+from agents.engineering_state import (  # noqa: E402
+    EngineeringState,
+    EngineeringStateConfig,
+    EngineeringStateMode,
+    SCHEMA_VERSION,
+    redact_text,
+)
+
+
+class TestEngineeringState(unittest.TestCase):
+    def test_default_rollout_is_authoritative_and_invalid_mode_fails_closed(self) -> None:
+        with patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(EngineeringStateConfig.from_env().mode, EngineeringStateMode.AUTHORITATIVE)
+        with patch.dict(os.environ, {"ENGINEERING_STATE_MODE": "unsafe"}, clear=False):
+            self.assertEqual(EngineeringStateConfig.from_env().mode, EngineeringStateMode.OFF)
+
+    def test_redaction_removes_common_credentials(self) -> None:
+        value = "Authorization: Bearer abcdefghijkl token=ghp_1234567890abcdef hf_1234567890"
+        result = redact_text(value)
+        self.assertNotIn("abcdefghijkl", result)
+        self.assertNotIn("ghp_1234567890abcdef", result)
+        self.assertNotIn("hf_1234567890", result)
+        self.assertIn("[REDACTED]", result)
+
+    def test_transitions_are_validated_and_idempotent(self) -> None:
+        state = EngineeringState.start("build a safe agent", run_id="run-1", now_ms=100)
+        self.assertTrue(state.transition("CLASSIFYING", now_ms=101))
+        self.assertFalse(state.transition("CLASSIFYING", now_ms=102))
+        with self.assertRaises(ValueError):
+            state.transition("IDLE", now_ms=103)
+        self.assertEqual(state.revision, 1)
+        self.assertEqual(state.sequence, 1)
+
+    def test_round_trip_is_bounded_and_does_not_store_raw_goal(self) -> None:
+        goal = "use token=super-secret-value to build this agent"
+        state = EngineeringState.start(goal, run_id="run-2", session_id="session-2", now_ms=100)
+        for target in ("CLASSIFYING", "THINKING", "COMPLETED"):
+            state.transition(target, now_ms=101)
+        snapshot = state.snapshot()
+        restored = EngineeringState.from_snapshot(snapshot)
+        self.assertEqual(restored.snapshot(), snapshot)
+        self.assertEqual(snapshot["schema_version"], SCHEMA_VERSION)
+        self.assertNotIn("super-secret-value", str(snapshot))
+        self.assertLessEqual(len(snapshot["history"]), 64)
+
+    def test_corrupt_schema_and_revision_are_rejected(self) -> None:
+        state = EngineeringState.start("goal", run_id="run-3")
+        snapshot = state.snapshot()
+        snapshot["schema_version"] = 999
+        with self.assertRaises(ValueError):
+            EngineeringState.from_snapshot(snapshot)
+        snapshot = state.snapshot()
+        snapshot["revision"] = -1
+        with self.assertRaises(ValueError):
+            EngineeringState.from_snapshot(snapshot)
+
+    def test_canary_selection_is_deterministic_and_requires_session(self) -> None:
+        config = EngineeringStateConfig(EngineeringStateMode.CANARY, 0.5)
+        self.assertFalse(config.selects_canary("run", ""))
+        self.assertEqual(
+            config.selects_canary("run", "session"),
+            config.selects_canary("run", "session"),
+        )
+
+    def test_resume_normalizes_terminal_state_and_preserves_history(self) -> None:
+        state = EngineeringState.start("resume this task", run_id="run-4", session_id="session-4")
+        for target in ("CLASSIFYING", "THINKING", "COMPLETED"):
+            state.transition(target)
+        history_before_resume = list(state.history)
+
+        state.prepare_for_resume()
+
+        self.assertEqual(state.current_state, "IDLE")
+        self.assertEqual(state.status, "active")
+        self.assertEqual(state.history[:len(history_before_resume)], history_before_resume)
+        self.assertTrue(any("resume normalized state to IDLE" in item for item in state.diagnostics))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)