diff --git a/backend/alembic/versions/v1_0_0_f061_default_tenant_timezone.py b/backend/alembic/versions/v1_0_0_f061_default_tenant_timezone.py
new file mode 100644
index 000000000..f0119623b
--- /dev/null
+++ b/backend/alembic/versions/v1_0_0_f061_default_tenant_timezone.py
@@ -0,0 +1,51 @@
+"""F061: Use Beijing as the required default tenant timezone.
+
+Revision ID: f061_default_tenant_timezone
+Revises: f060_tenant_id_backfill
+Create Date: 2026-08-05 12:00:00
+
+Background:
+  Agent scheduling inherits its timezone from the Tenant when the Agent has no
+  override, so new Tenants need a stable platform default.
+
+Scope:
+  Require the Tenant timezone column and change its server default to
+  Asia/Shanghai.
+
+Idempotent:
+  Reapplying the same nullability and server-default metadata is safe.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+
+revision: str = "f061_default_tenant_timezone"
+down_revision: str | None = "f060_tenant_id_backfill"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.alter_column(
+        "tenants",
+        "timezone",
+        existing_type=sa.String(length=50),
+        nullable=False,
+        server_default="Asia/Shanghai",
+    )
+
+
+def downgrade() -> None:
+    op.alter_column(
+        "tenants",
+        "timezone",
+        existing_type=sa.String(length=50),
+        nullable=True,
+        server_default="UTC",
+    )
diff --git a/backend/app/api/agents.py b/backend/app/api/agents.py
index e7da1a80d..98cc1079f 100644
--- a/backend/app/api/agents.py
+++ b/backend/app/api/agents.py
@@ -24,6 +24,7 @@
 from app.models.user import User
 from app.schemas.schemas import AgentCreate, AgentOut, AgentUpdate
 from app.services.storage import get_storage_backend
+from app.services.timezone_utils import DEFAULT_TIMEZONE
 from app.services.access_relationships import ensure_access_granted_platform_relationships
 from app.services.quota_guard import check_agent_creation_quota, QuotaExceeded
 from app.models.tenant import Tenant
@@ -592,13 +593,15 @@ async def get_agent(
         creator = await user_dao.get_with_identity(agent.creator_id)
         out["creator_username"] = creator.username if creator else None
 
-    # Resolve effective timezone (agent → tenant → UTC)
+    # Resolve effective timezone (agent → tenant → platform default)
     effective_tz = agent.timezone
     if not effective_tz and agent.tenant_id:
         tenant = await tenant_dao.get(agent.tenant_id)
         if tenant:
-            effective_tz = tenant.timezone or "UTC"
-    out["effective_timezone"] = effective_tz or "UTC"
+            effective_tz = tenant.timezone
+    if not effective_tz:
+        effective_tz = DEFAULT_TIMEZONE
+    out["effective_timezone"] = effective_tz
 
     return out
 
diff --git a/backend/app/api/tenants.py b/backend/app/api/tenants.py
index 7206fc585..898ffe0e5 100644
--- a/backend/app/api/tenants.py
+++ b/backend/app/api/tenants.py
@@ -13,7 +13,7 @@
 from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
 from fastapi.responses import FileResponse
 from PIL import Image
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, field_validator
 from sqlalchemy import func as sqla_func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -24,6 +24,7 @@
 from app.models.tenant import Tenant
 from app.models.user import User
 from app.services.storage import ensure_local_path, get_storage_backend, normalize_storage_key
+from app.services.timezone_utils import validate_timezone_name
 
 router = APIRouter(prefix="/tenants", tags=["tenants"])
 
@@ -39,7 +40,7 @@ class TenantOut(BaseModel):
     name: str
     slug: str
     im_provider: str
-    timezone: str = "UTC"
+    timezone: str = "Asia/Shanghai"
     country_region: str = "001"
     is_active: bool
     sso_enabled: bool = False
@@ -62,6 +63,13 @@ class TenantUpdate(BaseModel):
     sso_domain: str | None = None
     a2a_async_enabled: bool | None = None
 
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, value: str | None) -> str:
+        if value is None:
+            raise ValueError("Tenant timezone is required")
+        return validate_timezone_name(value)
+
 
 def _tenant_logo_key(tenant_id: uuid.UUID) -> str:
     return normalize_storage_key(f"_tenant_logos/{tenant_id}.png")
@@ -265,7 +273,7 @@ async def join_company(
     ic_result = await query_dao.execute(db, 
         select(InvitationCode).where(
             InvitationCode.code == data.invitation_code,
-            InvitationCode.is_active == True,
+            InvitationCode.is_active.is_(True),
             InvitationCode.tenant_id.is_not(None),
         )
     )
diff --git a/backend/app/api/triggers.py b/backend/app/api/triggers.py
index 02c7ff98d..215e48ec6 100644
--- a/backend/app/api/triggers.py
+++ b/backend/app/api/triggers.py
@@ -2,6 +2,7 @@
 
 import uuid
 
+from croniter import croniter
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlalchemy import select
@@ -91,6 +92,20 @@ async def update_trigger(
             raise HTTPException(404, "Trigger not found")
 
         if body.config is not None:
+            if trigger.type == "cron":
+                expr = body.config.get("expr")
+                if not isinstance(expr, str) or not expr.strip():
+                    raise HTTPException(
+                        400,
+                        "cron trigger requires config.expr",
+                    )
+                try:
+                    croniter(expr)
+                except Exception as exc:
+                    raise HTTPException(
+                        400,
+                        f"Invalid cron expression: '{expr}'.",
+                    ) from exc
             trigger.config = body.config
         if body.reason is not None:
             trigger.reason = body.reason
diff --git a/backend/app/models/tenant.py b/backend/app/models/tenant.py
index 01d47af11..f8743d130 100644
--- a/backend/app/models/tenant.py
+++ b/backend/app/models/tenant.py
@@ -38,7 +38,11 @@ class Tenant(Base):
     min_heartbeat_interval_minutes: Mapped[int] = mapped_column(Integer, default=240)
 
     # Default timezone for all agents in this company (IANA format, e.g. "Asia/Shanghai")
-    timezone: Mapped[str] = mapped_column(String(50), default="UTC")
+    timezone: Mapped[str] = mapped_column(
+        String(50),
+        default="Asia/Shanghai",
+        nullable=False,
+    )
     # Company country/region code used to derive default timezone and business calendar.
     country_region: Mapped[str] = mapped_column(String(10), default="001")
 
diff --git a/backend/app/schemas/schemas.py b/backend/app/schemas/schemas.py
index 11b354a01..3369ae7d2 100644
--- a/backend/app/schemas/schemas.py
+++ b/backend/app/schemas/schemas.py
@@ -3,7 +3,9 @@
 import uuid
 from datetime import datetime
 
-from pydantic import BaseModel, EmailStr, Field
+from pydantic import BaseModel, EmailStr, Field, field_validator
+
+from app.services.timezone_utils import validate_timezone_name
 
 
 # ─── Auth ───────────────────────────────────────────────
@@ -323,6 +325,13 @@ class AgentUpdate(BaseModel):
     timezone: str | None = None
     expires_at: datetime | None = None  # Admin only — extend agent expiry
 
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        return validate_timezone_name(value)
+
 
 class AgentStatusOut(BaseModel):
     """Agent status from state.json."""
diff --git a/backend/app/services/agent_tools.py b/backend/app/services/agent_tools.py
index 430579801..38b2ef6c6 100644
--- a/backend/app/services/agent_tools.py
+++ b/backend/app/services/agent_tools.py
@@ -31,6 +31,7 @@
 from typing import Optional, Any, cast
 import re
 
+from croniter import croniter
 from loguru import logger
 from sqlalchemy import select, or_
 
@@ -10494,7 +10495,6 @@ async def _handle_set_trigger_outcome(
                 "invalid_tool_arguments",
             )
         try:
-            from croniter import croniter
             croniter(expr)
         except Exception:
             return _typed_failure(
@@ -10790,7 +10790,22 @@ async def _handle_update_trigger_outcome(
                     for key, value in new_config.items()
                     if key != "token" and not key.startswith("_")
                 }
-                trigger.config = {**old_config, **user_patch, **protected}
+                updated_config = {**old_config, **user_patch, **protected}
+                if trigger.type == "cron":
+                    expr = updated_config.get("expr")
+                    if not isinstance(expr, str) or not expr.strip():
+                        return _typed_failure(
+                            "cron trigger requires config.expr.",
+                            "invalid_tool_arguments",
+                        )
+                    try:
+                        croniter(expr)
+                    except Exception:
+                        return _typed_failure(
+                            f"Invalid cron expression: '{expr}'.",
+                            "invalid_tool_arguments",
+                        )
+                trigger.config = updated_config
                 changes.append(f"config fields patched: {sorted(user_patch)}")
             if new_reason is not None:
                 if not isinstance(new_reason, str) or not new_reason.strip():
diff --git a/backend/app/services/timezone_utils.py b/backend/app/services/timezone_utils.py
index 58e2fe1ed..db87f8bb1 100644
--- a/backend/app/services/timezone_utils.py
+++ b/backend/app/services/timezone_utils.py
@@ -2,7 +2,7 @@
 
 import uuid
 from datetime import datetime
-from zoneinfo import ZoneInfo
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 
 from sqlalchemy import select
@@ -32,11 +32,22 @@
     "Pacific/Auckland",
 ]
 
+DEFAULT_TIMEZONE = "Asia/Shanghai"
+
+
+def validate_timezone_name(value: str) -> str:
+    """Return a valid IANA timezone name or raise a validation error."""
+    try:
+        ZoneInfo(value)
+    except (ValueError, ZoneInfoNotFoundError) as error:
+        raise ValueError(f"Invalid IANA timezone: {value}") from error
+    return value
+
 
 async def get_agent_timezone(agent_id: uuid.UUID) -> str:
     """Resolve effective timezone for an agent.
 
-    Priority: agent.timezone → tenant.timezone → 'UTC'
+    Priority: agent.timezone → tenant.timezone → default timezone.
     """
     from app.models.agent import Agent
     from app.models.tenant import Tenant
@@ -51,7 +62,7 @@ async def get_agent_timezone(agent_id: uuid.UUID) -> str:
         )
         agent = result.scalar_one_or_none()
         if not agent:
-            return "UTC"
+            return DEFAULT_TIMEZONE
 
         # Agent-level override
         if agent.timezone:
@@ -64,19 +75,19 @@ async def get_agent_timezone(agent_id: uuid.UUID) -> str:
             if tenant and tenant.timezone:
                 return tenant.timezone
 
-        return "UTC"
+        return DEFAULT_TIMEZONE
 
 
 def get_agent_timezone_sync(agent, tenant=None) -> str:
     """Synchronous version — when agent and tenant objects are already loaded.
 
-    Priority: agent.timezone → tenant.timezone → 'UTC'
+    Priority: agent.timezone → tenant.timezone → default timezone.
     """
     if agent.timezone:
         return agent.timezone
     if tenant and hasattr(tenant, 'timezone') and tenant.timezone:
         return tenant.timezone
-    return "UTC"
+    return DEFAULT_TIMEZONE
 
 
 def now_in_timezone(tz_name: str) -> datetime:
diff --git a/backend/app/services/trigger_daemon.py b/backend/app/services/trigger_daemon.py
index 2445fc2f7..c4c42cd14 100644
--- a/backend/app/services/trigger_daemon.py
+++ b/backend/app/services/trigger_daemon.py
@@ -99,7 +99,7 @@ async def _handle_okr_report_trigger(trigger: AgentTrigger, now: datetime) -> bo
 async def _handle_okr_collection_trigger(trigger: AgentTrigger, now: datetime) -> bool:
     return await handle_okr_collection_trigger_runtime(trigger, now)
 
-async def _evaluate_trigger(trigger: AgentTrigger, now: datetime) -> bool:
+async def _evaluate_trigger(trigger: AgentTrigger, now: datetime) -> datetime | None:
     return await evaluate_trigger_runtime(trigger, now)
 
 # ── Main Tick Loop ──────────────────────────────────────────────────
@@ -139,7 +139,8 @@ async def _tick():
             continue
 
         try:
-            if await _evaluate_trigger(trigger, now):
+            scheduled_at = await _evaluate_trigger(trigger, now)
+            if scheduled_at is not None:
                 handled = await _handle_okr_report_trigger(trigger, now)
                 if not handled:
                     handled = await _handle_okr_collection_trigger(trigger, now)
@@ -166,7 +167,7 @@ async def _tick():
                             continue
                         recent.append(now)
                         _on_msg_fire_log[trigger.agent_id] = recent
-                    await enqueue_due_trigger(trigger, now)
+                    await enqueue_due_trigger(trigger, scheduled_at)
         except Exception as e:
             logger.warning(f"Error evaluating trigger {trigger.name}: {e}")
 
diff --git a/backend/app/services/trigger_runtime/dispatch.py b/backend/app/services/trigger_runtime/dispatch.py
index 0fd9d6ca9..eccc491cf 100644
--- a/backend/app/services/trigger_runtime/dispatch.py
+++ b/backend/app/services/trigger_runtime/dispatch.py
@@ -4,6 +4,8 @@
 
 from datetime import datetime
 
+from loguru import logger
+
 from app.dao import query_dao
 from app.models.trigger import AgentTrigger
 from app.services.trigger_runtime.keys import build_scheduled_execution_key
@@ -31,12 +33,22 @@ def runtime_execution_payload(trigger: AgentTrigger) -> dict:
     return payload
 
 
-async def enqueue_due_trigger(trigger: AgentTrigger, now: datetime) -> None:
+async def enqueue_due_trigger(trigger: AgentTrigger, scheduled_at: datetime) -> None:
     async with query_dao.session() as db:
-        await enqueue_trigger_execution(
-            db,
-            trigger=trigger,
-            source=trigger.type,
-            idempotency_key=build_scheduled_execution_key(trigger, now),
-            payload_obj=runtime_execution_payload(trigger),
-        )
+        try:
+            await enqueue_trigger_execution(
+                db,
+                trigger=trigger,
+                source=trigger.type,
+                idempotency_key=build_scheduled_execution_key(trigger, scheduled_at),
+                scheduled_at=scheduled_at,
+                payload_obj=runtime_execution_payload(trigger),
+            )
+        except Exception as error:
+            logger.bind(
+                trigger_id=str(trigger.id),
+                trigger_name=trigger.name,
+                trigger_type=trigger.type,
+                scheduled_at=scheduled_at.isoformat(),
+            ).error("Trigger occurrence registration failed: {}", error)
+            raise
diff --git a/backend/app/services/trigger_runtime/evaluator.py b/backend/app/services/trigger_runtime/evaluator.py
index 711ac5ee0..a4307fe1c 100644
--- a/backend/app/services/trigger_runtime/evaluator.py
+++ b/backend/app/services/trigger_runtime/evaluator.py
@@ -12,10 +12,11 @@
 from sqlalchemy import select
 
 from app.dao import query_dao
-async_session = query_dao.session
 from app.models.agent import Agent
 from app.models.trigger import AgentTrigger
 
+async_session = query_dao.session
+
 MIN_POLL_INTERVAL_MINUTES = 5
 
 
@@ -171,18 +172,27 @@ def is_private_url(url: str) -> bool:
         return True
 
 
-async def evaluate_trigger(trigger: AgentTrigger, now: datetime) -> bool:
+MISFIRE_GRACE = timedelta(seconds=30)
+
+
+def _as_utc(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+async def evaluate_trigger(trigger: AgentTrigger, now: datetime) -> datetime | None:
     if not trigger.is_enabled:
-        return False
+        return None
     if trigger.expires_at and now >= trigger.expires_at:
-        return False
+        return None
     if trigger.max_fires is not None and trigger.fire_count >= trigger.max_fires:
-        return False
+        return None
 
     if trigger.last_fired_at:
         cooldown = timedelta(seconds=trigger.cooldown_seconds)
         if (now - trigger.last_fired_at) < cooldown:
-            return False
+            return None
 
     cfg = trigger.config or {}
     if isinstance(cfg, str):
@@ -195,63 +205,73 @@ async def evaluate_trigger(trigger: AgentTrigger, now: datetime) -> bool:
 
     if t == "cron":
         expr = cfg.get("expr", "* * * * *")
-        base = trigger.last_fired_at or trigger.created_at
         try:
-            tz_name = cfg.get("timezone")
-            if not tz_name:
-                from app.services.timezone_utils import get_agent_timezone
-                tz_name = await get_agent_timezone(trigger.agent_id)
+            from app.services.timezone_utils import get_agent_timezone
+
+            tz_name = await get_agent_timezone(trigger.agent_id)
             from zoneinfo import ZoneInfo
-            try:
-                tz = ZoneInfo(tz_name)
-            except (KeyError, Exception):
-                tz = ZoneInfo("UTC")
+
+            tz = ZoneInfo(tz_name)
             local_now = now.astimezone(tz)
-            local_base = base.astimezone(tz) if base.tzinfo else base.replace(tzinfo=tz)
-            cron = croniter(expr, local_base)
-            next_run = cron.get_next(datetime)
-            if local_now >= next_run:
-                if await should_skip_non_workday(trigger, local_now):
-                    await mark_trigger_skipped(trigger.id, now)
-                    logger.info(f"[Trigger] Skipped {trigger.name} on non-workday {local_now.date()}")
-                    return False
-                return True
-            return False
-        except Exception as e:
-            logger.warning(f"Invalid cron expr '{expr}' for trigger {trigger.name}: {e}")
-            return False
+            scheduled_at = croniter(
+                expr,
+                local_now + timedelta(microseconds=1),
+            ).get_prev(datetime)
+            scheduled_at_utc = _as_utc(scheduled_at)
+            now_utc = _as_utc(now)
+            created_at_utc = _as_utc(trigger.created_at)
+            if scheduled_at_utc <= created_at_utc:
+                return None
+            if scheduled_at_utc > now_utc:
+                return None
+            if now_utc - scheduled_at_utc > MISFIRE_GRACE:
+                return None
+            if await should_skip_non_workday(trigger, local_now):
+                await mark_trigger_skipped(trigger.id, now)
+                logger.info(f"[Trigger] Skipped {trigger.name} on non-workday {local_now.date()}")
+                return None
+            return scheduled_at
+        except Exception as error:
+            logger.bind(
+                trigger_id=str(trigger.id),
+                trigger_name=trigger.name,
+                trigger_type=trigger.type,
+                cron_expr=expr,
+            ).warning("Trigger occurrence evaluation failed: {}", error)
+            return None
 
     if t == "once":
         at_str = cfg.get("at")
         if not at_str:
-            return False
+            return None
         try:
             at = datetime.fromisoformat(at_str)
             if at.tzinfo is None:
                 at = at.replace(tzinfo=timezone.utc)
-            return now >= at and trigger.fire_count == 0
+            return at if now >= at and trigger.fire_count == 0 else None
         except Exception:
-            return False
+            return None
 
     if t == "interval":
         minutes = cfg.get("minutes", 30)
         base = trigger.last_fired_at or trigger.created_at
-        return (now - base) >= timedelta(minutes=minutes)
+        scheduled_at = base + timedelta(minutes=minutes)
+        return scheduled_at if now >= scheduled_at else None
 
     if t == "poll":
         interval_min = max(cfg.get("interval_min", 5), MIN_POLL_INTERVAL_MINUTES)
         base = trigger.last_fired_at or trigger.created_at
         if (now - base) < timedelta(minutes=interval_min):
-            return False
-        return await poll_check(trigger)
+            return None
+        return now if await poll_check(trigger) else None
 
     if t == "on_message":
-        return await check_new_agent_messages(trigger)
+        return now if await check_new_agent_messages(trigger) else None
 
     if t == "webhook":
-        return False
+        return None
 
-    return False
+    return None
 
 
 async def poll_check(trigger: AgentTrigger) -> bool:
diff --git a/backend/app/services/trigger_runtime/keys.py b/backend/app/services/trigger_runtime/keys.py
index 4fbb3178a..4261f256a 100644
--- a/backend/app/services/trigger_runtime/keys.py
+++ b/backend/app/services/trigger_runtime/keys.py
@@ -3,14 +3,15 @@
 from __future__ import annotations
 
 import hashlib
-from datetime import datetime, timedelta, timezone
-
-from croniter import croniter
+from datetime import datetime, timezone
 
 from app.models.trigger import AgentTrigger
 
 
-def build_scheduled_execution_key(trigger: AgentTrigger, now: datetime) -> str:
+def build_scheduled_execution_key(
+    trigger: AgentTrigger,
+    scheduled_at: datetime,
+) -> str:
     """Build a deterministic idempotency key for non-webhook trigger runs."""
     cfg = trigger.config or {}
     trigger_type = trigger.type
@@ -19,19 +20,16 @@ def build_scheduled_execution_key(trigger: AgentTrigger, now: datetime) -> str:
         return f"once:{trigger.id}:{cfg.get('at', '')}"
 
     if trigger_type == "interval":
-        minutes = int(cfg.get("minutes", 30) or 30)
-        base = trigger.last_fired_at or trigger.created_at
-        due_at = base + timedelta(minutes=minutes)
-        return f"interval:{trigger.id}:{due_at.astimezone(timezone.utc).isoformat()}"
+        return (
+            f"interval:{trigger.id}:"
+            f"{scheduled_at.astimezone(timezone.utc).isoformat()}"
+        )
 
     if trigger_type == "cron":
-        expr = cfg.get("expr", "* * * * *")
-        base = trigger.last_fired_at or trigger.created_at
-        cron = croniter(expr, base)
-        due_at = cron.get_next(datetime)
-        if due_at.tzinfo is None:
-            due_at = due_at.replace(tzinfo=timezone.utc)
-        return f"cron:{trigger.id}:{due_at.astimezone(timezone.utc).isoformat()}"
+        return (
+            f"cron:{trigger.id}:"
+            f"{scheduled_at.astimezone(timezone.utc).isoformat()}"
+        )
 
     if trigger_type == "on_message":
         matched_from = str(cfg.get("_matched_from") or "")
@@ -44,4 +42,7 @@ def build_scheduled_execution_key(trigger: AgentTrigger, now: datetime) -> str:
         digest = hashlib.sha256(current_value.encode("utf-8")).hexdigest()
         return f"poll:{trigger.id}:{digest}"
 
-    return f"{trigger_type}:{trigger.id}:{now.replace(microsecond=0).isoformat()}"
+    return (
+        f"{trigger_type}:{trigger.id}:"
+        f"{scheduled_at.replace(microsecond=0).isoformat()}"
+    )
diff --git a/backend/app/services/trigger_runtime/queue.py b/backend/app/services/trigger_runtime/queue.py
index 5aed37903..f6604578f 100644
--- a/backend/app/services/trigger_runtime/queue.py
+++ b/backend/app/services/trigger_runtime/queue.py
@@ -57,18 +57,39 @@ def _fail_runtime_execution(
     execution.last_error = f"{error.code}: {error}"[:2000]
 
 
+async def _handle_intake_failure(
+    db: AsyncSession,
+    *,
+    execution: TriggerExecution,
+    error: TriggerRuntimeIntakeError,
+    now: datetime,
+    persist_intake_failure: bool,
+) -> None:
+    if not persist_intake_failure:
+        await db.rollback()
+        raise error
+    _fail_runtime_execution(execution, error, now)
+
+
 async def enqueue_trigger_execution(
     db: AsyncSession,
     *,
     trigger: AgentTrigger,
     source: str,
     idempotency_key: str,
+    scheduled_at: datetime | None = None,
+    persist_intake_failure: bool = False,
     payload_text: str = "",
     payload_obj: dict | None = None,
 ) -> tuple[TriggerExecution | None, bool]:
     """Atomically insert an occurrence and its required Runtime command."""
     normalized_key = idempotency_key[:255]
     now = datetime.now(timezone.utc)
+    scheduled_at_utc = scheduled_at or now
+    if scheduled_at_utc.tzinfo is None:
+        scheduled_at_utc = scheduled_at_utc.replace(tzinfo=timezone.utc)
+    else:
+        scheduled_at_utc = scheduled_at_utc.astimezone(timezone.utc)
     execution = TriggerExecution(
         id=uuid.uuid4(),
         trigger_id=trigger.id,
@@ -78,7 +99,7 @@ async def enqueue_trigger_execution(
         idempotency_key=normalized_key,
         payload=payload_obj if isinstance(payload_obj, dict) else {},
         payload_text=payload_text[:8000],
-        scheduled_at=now,
+        scheduled_at=scheduled_at_utc,
     )
     try:
         async with db.begin_nested():
@@ -106,25 +127,29 @@ async def enqueue_trigger_execution(
             "Trigger disappeared while its execution was being registered",
         )
     if not stored_trigger.is_enabled:
-        _fail_runtime_execution(
-            execution,
-            TriggerRuntimeIntakeError(
+        await _handle_intake_failure(
+            db,
+            execution=execution,
+            error=TriggerRuntimeIntakeError(
                 "trigger_disabled",
                 "Trigger was disabled before its execution was accepted",
             ),
-            now,
+            now=now,
+            persist_intake_failure=persist_intake_failure,
         )
         await db.commit()
         return execution, True
     agent: Agent | None = await load_trigger_agent(db, trigger=stored_trigger)
     if agent is None:
-        _fail_runtime_execution(
-            execution,
-            TriggerRuntimeIntakeError(
+        await _handle_intake_failure(
+            db,
+            execution=execution,
+            error=TriggerRuntimeIntakeError(
                 "agent_not_found",
                 "Runtime Trigger Agent does not exist",
             ),
-            now,
+            now=now,
+            persist_intake_failure=persist_intake_failure,
         )
     else:
         try:
@@ -147,7 +172,13 @@ async def enqueue_trigger_execution(
                 _mark_trigger_fired(stored_trigger, now)
                 await db.flush()
         except TriggerRuntimeIntakeError as error:
-            _fail_runtime_execution(execution, error, now)
+            await _handle_intake_failure(
+                db,
+                execution=execution,
+                error=error,
+                now=now,
+                persist_intake_failure=persist_intake_failure,
+            )
 
     await db.commit()
     return execution, True
@@ -180,6 +211,7 @@ async def enqueue_webhook_execution(
         trigger=trigger,
         source="webhook",
         idempotency_key=delivery_key,
+        persist_intake_failure=True,
         payload_text=payload_text,
         payload_obj=payload_obj,
     )
diff --git a/backend/tests/test_timezone_validation.py b/backend/tests/test_timezone_validation.py
new file mode 100644
index 000000000..951f9d493
--- /dev/null
+++ b/backend/tests/test_timezone_validation.py
@@ -0,0 +1,92 @@
+"""Timezone defaults and write-boundary validation."""
+
+from __future__ import annotations
+
+import uuid
+from types import SimpleNamespace
+
+import pytest
+from pydantic import ValidationError
+
+from app.api import agents as agents_api
+from app.api.tenants import TenantOut, TenantUpdate
+from app.models.tenant import Tenant
+from app.schemas.schemas import AgentUpdate
+
+
+def test_tenant_timezone_defaults_to_beijing() -> None:
+    assert Tenant.__table__.c.timezone.default.arg == "Asia/Shanghai"
+    assert TenantOut.model_fields["timezone"].default == "Asia/Shanghai"
+
+
+@pytest.mark.parametrize("timezone_name", ["Asia/Shanghai", "America/New_York"])
+def test_tenant_update_accepts_iana_timezone(timezone_name: str) -> None:
+    assert TenantUpdate(timezone=timezone_name).timezone == timezone_name
+
+
+@pytest.mark.parametrize("timezone_name", [None, "", "UTC+8", "Invalid/Timezone"])
+def test_tenant_update_rejects_missing_or_invalid_timezone(
+    timezone_name: str | None,
+) -> None:
+    with pytest.raises(ValidationError):
+        TenantUpdate(timezone=timezone_name)
+
+
+def test_tenant_update_allows_timezone_to_be_omitted() -> None:
+    update = TenantUpdate(name="Renamed")
+
+    assert "timezone" not in update.model_dump(exclude_unset=True)
+
+
+@pytest.mark.parametrize("timezone_name", [None, "Asia/Shanghai", "America/New_York"])
+def test_agent_update_accepts_inheritance_or_iana_timezone(
+    timezone_name: str | None,
+) -> None:
+    assert AgentUpdate(timezone=timezone_name).timezone == timezone_name
+
+
+@pytest.mark.parametrize("timezone_name", ["", "UTC+8", "Invalid/Timezone"])
+def test_agent_update_rejects_invalid_timezone(timezone_name: str) -> None:
+    with pytest.raises(ValidationError):
+        AgentUpdate(timezone=timezone_name)
+
+
+@pytest.mark.asyncio
+async def test_agent_detail_uses_platform_timezone_when_agent_and_tenant_missing(
+    monkeypatch,
+) -> None:
+    agent = SimpleNamespace(
+        id=uuid.uuid4(),
+        creator_id=None,
+        tenant_id=None,
+        timezone=None,
+    )
+
+    async def fake_check_agent_access(*_args, **_kwargs):
+        return agent, "manage"
+
+    async def fake_lazy_reset(*_args, **_kwargs):
+        return False
+
+    async def fake_agent_to_out(*_args, **_kwargs):
+        return SimpleNamespace(model_dump=lambda: {})
+
+    monkeypatch.setattr(
+        agents_api,
+        "check_agent_access",
+        fake_check_agent_access,
+    )
+    monkeypatch.setattr(
+        agents_api,
+        "_lazy_reset_token_counters",
+        fake_lazy_reset,
+    )
+    monkeypatch.setattr(agents_api, "_agent_to_out", fake_agent_to_out)
+
+    result = await agents_api.get_agent(
+        agent.id,
+        current_user=SimpleNamespace(id=uuid.uuid4()),
+        db=SimpleNamespace(),
+    )
+
+    assert result["effective_timezone"] == "Asia/Shanghai"
diff --git a/backend/tests/test_trigger_config_updates.py b/backend/tests/test_trigger_config_updates.py
new file mode 100644
index 000000000..06c9b7cb4
--- /dev/null
+++ b/backend/tests/test_trigger_config_updates.py
@@ -0,0 +1,151 @@
+"""Validation at the existing Trigger update boundaries."""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+import uuid
+
+from fastapi import HTTPException
+import pytest
+
+from app.api import triggers as triggers_api
+from app.models.trigger import AgentTrigger
+from app.services import agent_tools, audit_logger
+
+
+class _ScalarResult:
+    def __init__(self, value: AgentTrigger) -> None:
+        self._value = value
+
+    def scalar_one_or_none(self) -> AgentTrigger:
+        return self._value
+
+
+class _TriggerSession:
+    def __init__(self, trigger: AgentTrigger) -> None:
+        self._trigger = trigger
+        self.commit_count = 0
+
+    async def execute(self, _statement) -> _ScalarResult:
+        return _ScalarResult(self._trigger)
+
+    async def commit(self) -> None:
+        self.commit_count += 1
+
+
+def _cron_trigger() -> AgentTrigger:
+    return AgentTrigger(
+        id=uuid.uuid4(),
+        agent_id=uuid.uuid4(),
+        name="daily-check",
+        type="cron",
+        config={"expr": "0 9 * * *"},
+        reason="Daily check",
+        is_enabled=True,
+        fire_count=0,
+        cooldown_seconds=60,
+    )
+
+
+@pytest.mark.asyncio
+async def test_rest_update_rejects_invalid_cron_before_commit(monkeypatch) -> None:
+    trigger = _cron_trigger()
+    session = _TriggerSession(trigger)
+
+    @asynccontextmanager
+    async def fake_session():
+        yield session
+
+    monkeypatch.setattr(triggers_api.query_dao, "session", fake_session)
+
+    with pytest.raises(HTTPException) as error:
+        await triggers_api.update_trigger(
+            trigger.agent_id,
+            trigger.id,
+            triggers_api.TriggerUpdate(config={"expr": "not-a-cron"}),
+            user=object(),
+        )
+
+    assert error.value.status_code == 400
+    assert trigger.config == {"expr": "0 9 * * *"}
+    assert session.commit_count == 0
+
+
+@pytest.mark.asyncio
+async def test_rest_update_accepts_valid_cron(monkeypatch) -> None:
+    trigger = _cron_trigger()
+    session = _TriggerSession(trigger)
+
+    @asynccontextmanager
+    async def fake_session():
+        yield session
+
+    monkeypatch.setattr(triggers_api.query_dao, "session", fake_session)
+
+    result = await triggers_api.update_trigger(
+        trigger.agent_id,
+        trigger.id,
+        triggers_api.TriggerUpdate(config={"expr": "30 9 * * 1-5"}),
+        user=object(),
+    )
+
+    assert result == {"ok": True}
+    assert trigger.config == {"expr": "30 9 * * 1-5"}
+    assert session.commit_count == 1
+
+
+@pytest.mark.asyncio
+async def test_agent_tool_update_rejects_invalid_cron_before_commit(
+    monkeypatch,
+) -> None:
+    trigger = _cron_trigger()
+    session = _TriggerSession(trigger)
+
+    @asynccontextmanager
+    async def fake_session():
+        yield session
+
+    monkeypatch.setattr(agent_tools, "async_session", fake_session)
+
+    outcome = await agent_tools._handle_update_trigger_outcome(
+        trigger.agent_id,
+        {"name": trigger.name, "config": {"expr": "not-a-cron"}},
+    )
+
+    assert outcome.status == "failed"
+    assert outcome.error_code == "invalid_tool_arguments"
+    assert trigger.config == {"expr": "0 9 * * *"}
+    assert session.commit_count == 0
+
+
+@pytest.mark.asyncio
+async def test_agent_tool_partial_update_keeps_valid_existing_cron(
+    monkeypatch,
+) -> None:
+    trigger = _cron_trigger()
+    session = _TriggerSession(trigger)
+
+    @asynccontextmanager
+    async def fake_session():
+        yield session
+
+    async def fake_audit_log(*_args, **_kwargs) -> None:
+        return None
+
+    monkeypatch.setattr(agent_tools, "async_session", fake_session)
+    monkeypatch.setattr(audit_logger, "write_audit_log", fake_audit_log)
+
+    outcome = await agent_tools._handle_update_trigger_outcome(
+        trigger.agent_id,
+        {
+            "name": trigger.name,
+            "config": {"timezone": "America/New_York"},
+        },
+    )
+
+    assert outcome.status == "succeeded"
+    assert trigger.config == {
+        "expr": "0 9 * * *",
+        "timezone": "America/New_York",
+    }
+    assert session.commit_count == 1
diff --git a/backend/tests/test_trigger_runtime_queue.py b/backend/tests/test_trigger_runtime_queue.py
index cae556e28..5c6552e0d 100644
--- a/backend/tests/test_trigger_runtime_queue.py
+++ b/backend/tests/test_trigger_runtime_queue.py
@@ -4,6 +4,7 @@
 
 from unittest.mock import AsyncMock, patch
 import uuid
+from datetime import UTC, datetime, timedelta, timezone
 
 import pytest
 from app.models.agent import Agent
@@ -37,6 +38,7 @@ def __init__(self, stored_trigger: AgentTrigger) -> None:
         self.nested = 0
         self.flushes = 0
         self.commits = 0
+        self.rollbacks = 0
 
     def begin_nested(self) -> _Nested:
         self.nested += 1
@@ -54,6 +56,9 @@ async def execute(self, _statement) -> _ScalarResult:
     async def commit(self) -> None:
         self.commits += 1
 
+    async def rollback(self) -> None:
+        self.rollbacks += 1
+
 
 def _records() -> tuple[AgentTrigger, Agent]:
     agent_id = uuid.uuid4()
@@ -108,11 +113,20 @@ async def accept_runtime(*_args, **kwargs):
             side_effect=accept_runtime,
         ),
     ):
+        scheduled_at = datetime(
+            2026,
+            8,
+            5,
+            9,
+            0,
+            tzinfo=timezone(timedelta(hours=8)),
+        )
         execution, created = await enqueue_trigger_execution(
             db,  # type: ignore[arg-type]
             trigger=trigger,
             source="poll",
             idempotency_key="poll:2026-07-13T16:00",
+            scheduled_at=scheduled_at,
         )
 
     assert created is True
@@ -123,10 +137,11 @@ async def accept_runtime(*_args, **kwargs):
     assert db.added == [execution]
     assert trigger.fire_count == 1
     assert trigger.last_fired_at is not None
+    assert execution.scheduled_at == scheduled_at.astimezone(UTC)
 
 
 @pytest.mark.asyncio
-async def test_runtime_intake_rejection_settles_occurrence_without_legacy_fallback() -> None:
+async def test_runtime_intake_rejection_rolls_back_scheduled_occurrence() -> None:
     trigger, agent = _records()
     db = _QueueSession(trigger)
     error = TriggerRuntimeIntakeError(
@@ -144,24 +159,23 @@ async def test_runtime_intake_rejection_settles_occurrence_without_legacy_fallba
             new=AsyncMock(side_effect=error),
         ),
     ):
-        execution, created = await enqueue_trigger_execution(
-            db,  # type: ignore[arg-type]
-            trigger=trigger,
-            source="poll",
-            idempotency_key="poll:2026-07-13T16:00",
-        )
-
-    assert created is True
-    assert execution is not None
-    assert execution.status == "failed"
-    assert execution.last_error == "agent_model_missing: Runtime Trigger Agent has no primary model"
-    assert execution.finished_at is not None
+        with pytest.raises(TriggerRuntimeIntakeError) as raised:
+            await enqueue_trigger_execution(
+                db,  # type: ignore[arg-type]
+                trigger=trigger,
+                source="poll",
+                idempotency_key="poll:2026-07-13T16:00",
+            )
+
+    assert raised.value.code == "agent_model_missing"
     assert trigger.fire_count == 0
-    assert db.commits == 1
+    assert trigger.last_fired_at is None
+    assert db.commits == 0
+    assert db.rollbacks == 1
 
 
 @pytest.mark.asyncio
-async def test_runtime_disabled_settles_occurrence_without_legacy_claiming() -> None:
+async def test_runtime_disabled_rolls_back_scheduled_occurrence() -> None:
     trigger, agent = _records()
     db = _QueueSession(trigger)
 
@@ -174,18 +188,56 @@ async def test_runtime_disabled_settles_occurrence_without_legacy_claiming() ->
             "app.services.trigger_runtime.queue.enqueue_trigger_runtime",
             new=AsyncMock(return_value=None),
         ),
+    ):
+        with pytest.raises(TriggerRuntimeIntakeError) as raised:
+            await enqueue_trigger_execution(
+                db,  # type: ignore[arg-type]
+                trigger=trigger,
+                source="poll",
+                idempotency_key="poll:2026-07-13T16:00",
+            )
+
+    assert raised.value.code == "runtime_v2_disabled"
+    assert trigger.fire_count == 0
+    assert trigger.last_fired_at is None
+    assert db.commits == 0
+    assert db.rollbacks == 1
+
+
+@pytest.mark.asyncio
+async def test_webhook_intake_rejection_keeps_failure_receipt() -> None:
+    trigger, agent = _records()
+    trigger.type = "webhook"
+    db = _QueueSession(trigger)
+    error = TriggerRuntimeIntakeError(
+        "agent_model_missing",
+        "Runtime Trigger Agent has no primary model",
+    )
+
+    with (
+        patch(
+            "app.services.trigger_runtime.queue.load_trigger_agent",
+            new=AsyncMock(return_value=agent),
+        ),
+        patch(
+            "app.services.trigger_runtime.queue.enqueue_trigger_runtime",
+            new=AsyncMock(side_effect=error),
+        ),
     ):
         execution, created = await enqueue_trigger_execution(
             db,  # type: ignore[arg-type]
             trigger=trigger,
-            source="poll",
-            idempotency_key="poll:2026-07-13T16:00",
+            source="webhook",
+            idempotency_key="delivery-1",
+            persist_intake_failure=True,
         )
 
     assert created is True
     assert execution is not None
     assert execution.status == "failed"
-    assert execution.last_error is not None
-    assert execution.last_error.startswith("runtime_v2_disabled:")
+    assert execution.last_error == (
+        "agent_model_missing: Runtime Trigger Agent has no primary model"
+    )
     assert trigger.fire_count == 0
     assert db.commits == 1
+    assert db.rollbacks == 0
diff --git a/backend/tests/test_trigger_runtime_scheduling.py b/backend/tests/test_trigger_runtime_scheduling.py
new file mode 100644
index 000000000..7bbb24e92
--- /dev/null
+++ b/backend/tests/test_trigger_runtime_scheduling.py
@@ -0,0 +1,250 @@
+"""Scheduled occurrence ownership across evaluator and dispatch."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from unittest.mock import AsyncMock, MagicMock, patch
+import uuid
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from app.models.trigger import AgentTrigger
+from app.services.trigger_runtime.dispatch import enqueue_due_trigger
+from app.services.trigger_runtime.evaluator import evaluate_trigger
+from app.services.trigger_runtime.keys import build_scheduled_execution_key
+
+
+def _cron_trigger(
+    *,
+    created_at: datetime,
+    last_fired_at: datetime | None = None,
+    config: dict | None = None,
+) -> AgentTrigger:
+    return AgentTrigger(
+        id=uuid.uuid4(),
+        agent_id=uuid.uuid4(),
+        name="daily-check",
+        type="cron",
+        config=config or {"expr": "0 9 * * *"},
+        reason="Daily check",
+        is_enabled=True,
+        created_at=created_at,
+        last_fired_at=last_fired_at,
+        fire_count=0,
+        cooldown_seconds=60,
+    )
+
+
+@pytest.mark.asyncio
+async def test_cron_evaluator_returns_agent_local_occurrence() -> None:
+    now = datetime(2026, 8, 5, 1, 0, 8, tzinfo=UTC)
+    trigger = _cron_trigger(created_at=now - timedelta(days=2))
+
+    with patch(
+        "app.services.timezone_utils.get_agent_timezone",
+        new=AsyncMock(return_value="Asia/Shanghai"),
+    ):
+        scheduled_at = await evaluate_trigger(trigger, now)
+
+    assert scheduled_at == datetime(
+        2026,
+        8,
+        5,
+        9,
+        0,
+        tzinfo=ZoneInfo("Asia/Shanghai"),
+    )
+
+
+@pytest.mark.asyncio
+async def test_cron_evaluator_ignores_trigger_timezone_override() -> None:
+    now = datetime(2026, 8, 5, 1, 0, 8, tzinfo=UTC)
+    trigger = _cron_trigger(
+        created_at=now - timedelta(days=2),
+        config={"expr": "0 9 * * *", "timezone": "America/New_York"},
+    )
+
+    with patch(
+        "app.services.timezone_utils.get_agent_timezone",
+        new=AsyncMock(return_value="Asia/Shanghai"),
+    ):
+        scheduled_at = await evaluate_trigger(trigger, now)
+
+    assert scheduled_at is not None
+    assert scheduled_at.tzinfo == ZoneInfo("Asia/Shanghai")
+    assert scheduled_at.astimezone(UTC) == datetime(2026, 8, 5, 1, 0, tzinfo=UTC)
+
+
+@pytest.mark.asyncio
+async def test_cron_occurrence_does_not_drift_with_last_fired_at() -> None:
+    now = datetime(2026, 8, 5, 1, 0, 8, tzinfo=UTC)
+    trigger = _cron_trigger(
+        created_at=now - timedelta(days=3),
+        last_fired_at=datetime(2026, 8, 4, 1, 5, tzinfo=UTC),
+    )
+
+    with patch(
+        "app.services.timezone_utils.get_agent_timezone",
+        new=AsyncMock(return_value="Asia/Shanghai"),
+    ):
+        scheduled_at = await evaluate_trigger(trigger, now)
+
+    assert scheduled_at is not None
+    assert scheduled_at.astimezone(UTC) == datetime(2026, 8, 5, 1, 0, tzinfo=UTC)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("delay_seconds, expected_due", [(30, True), (31, False)])
+async def test_cron_evaluator_applies_thirty_second_grace(
+    delay_seconds: int,
+    expected_due: bool,
+) -> None:
+    now = datetime(2026, 8, 5, 1, 0, delay_seconds, tzinfo=UTC)
+    trigger = _cron_trigger(created_at=now - timedelta(days=2))
+
+    with patch(
+        "app.services.timezone_utils.get_agent_timezone",
+        new=AsyncMock(return_value="Asia/Shanghai"),
+    ):
+        scheduled_at = await evaluate_trigger(trigger, now)
+
+    assert (scheduled_at is not None) is expected_due
+
+
+@pytest.mark.asyncio
+async def test_cron_evaluator_rejects_occurrence_before_trigger_creation() -> None:
+    now = datetime(2026, 8, 5, 1, 0, 8, tzinfo=UTC)
+    trigger = _cron_trigger(
+        created_at=datetime(2026, 8, 5, 1, 0, 5, tzinfo=UTC),
+    )
+
+    with patch(
+        "app.services.timezone_utils.get_agent_timezone",
+        new=AsyncMock(return_value="Asia/Shanghai"),
+    ):
+        scheduled_at = await evaluate_trigger(trigger, now)
+
+    assert scheduled_at is None
+
+
+@pytest.mark.asyncio
+async def test_cron_evaluator_does_not_fallback_for_invalid_timezone() -> None:
+    now = datetime(2026, 8, 5, 1, 0, 8, tzinfo=UTC)
+    trigger = _cron_trigger(created_at=now - timedelta(days=2))
+    bound_logger = MagicMock()
+
+    with (
+        patch(
+            "app.services.timezone_utils.get_agent_timezone",
+            new=AsyncMock(return_value="Invalid/Timezone"),
+        ),
+        patch(
+            "app.services.trigger_runtime.evaluator.logger.bind",
+            return_value=bound_logger,
+        ) as bind,
+    ):
+        scheduled_at = await evaluate_trigger(trigger, now)
+
+    assert scheduled_at is None
+    bind.assert_called_once_with(
+        trigger_id=str(trigger.id),
+        trigger_name=trigger.name,
+        trigger_type=trigger.type,
+        cron_expr="0 9 * * *",
+    )
+    bound_logger.warning.assert_called_once()
+
+
+def test_cron_execution_key_uses_supplied_occurrence() -> None:
+    scheduled_at = datetime(
+        2026,
+        8,
+        5,
+        9,
+        0,
+        tzinfo=ZoneInfo("Asia/Shanghai"),
+    )
+    trigger = _cron_trigger(created_at=scheduled_at - timedelta(days=2))
+
+    key = build_scheduled_execution_key(trigger, scheduled_at)
+
+    assert key == f"cron:{trigger.id}:2026-08-05T01:00:00+00:00"
+
+
+class _SessionContext:
+    async def __aenter__(self):
+        return MagicMock()
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+
+@pytest.mark.asyncio
+async def test_dispatch_passes_occurrence_to_queue_unchanged() -> None:
+    scheduled_at = datetime(
+        2026,
+        8,
+        5,
+        9,
+        0,
+        tzinfo=ZoneInfo("Asia/Shanghai"),
+    )
+    trigger = _cron_trigger(created_at=scheduled_at - timedelta(days=2))
+
+    with (
+        patch(
+            "app.services.trigger_runtime.dispatch.query_dao.session",
+            return_value=_SessionContext(),
+        ),
+        patch(
+            "app.services.trigger_runtime.dispatch.enqueue_trigger_execution",
+            new=AsyncMock(),
+        ) as enqueue,
+    ):
+        await enqueue_due_trigger(trigger, scheduled_at)
+
+    assert enqueue.await_args.kwargs["scheduled_at"] is scheduled_at
+    assert enqueue.await_args.kwargs["idempotency_key"] == (
+        f"cron:{trigger.id}:2026-08-05T01:00:00+00:00"
+    )
+
+
+@pytest.mark.asyncio
+async def test_dispatch_logs_scheduled_occurrence_registration_failure() -> None:
+    scheduled_at = datetime(
+        2026,
+        8,
+        5,
+        9,
+        0,
+        tzinfo=ZoneInfo("Asia/Shanghai"),
+    )
+    trigger = _cron_trigger(created_at=scheduled_at - timedelta(days=2))
+    error = RuntimeError("database unavailable")
+    bound_logger = MagicMock()
+
+    with (
+        patch(
+            "app.services.trigger_runtime.dispatch.query_dao.session",
+            return_value=_SessionContext(),
+        ),
+        patch(
+            "app.services.trigger_runtime.dispatch.enqueue_trigger_execution",
+            new=AsyncMock(side_effect=error),
+        ),
+        patch(
+            "app.services.trigger_runtime.dispatch.logger.bind",
+            return_value=bound_logger,
+        ) as bind,
+        pytest.raises(RuntimeError, match="database unavailable"),
+    ):
+        await enqueue_due_trigger(trigger, scheduled_at)
+
+    bind.assert_called_once_with(
+        trigger_id=str(trigger.id),
+        trigger_name=trigger.name,
+        trigger_type=trigger.type,
+        scheduled_at=scheduled_at.isoformat(),
+    )
+    bound_logger.error.assert_called_once()