fix(triggers): test-event 500, single-click test, dismiss drawer on create

api/oss/src/core/triggers/interfaces.py

@@ -145,6 +145,16 @@ async def query_subscriptions(
         windowing: Optional[Windowing] = None,
     ) -> List[TriggerSubscription]: ...
 
+    @abstractmethod
+    async def fetch_subscription_by_trigger_id(
+        self,
+        *,
+        project_id: UUID,
+        trigger_id: str,
+    ) -> Optional[TriggerSubscription]:
+        """Fetch the live (non-deleted) subscription owning ``trigger_id`` in this project."""
+        ...
+
     @abstractmethod
     async def get_project_and_subscription_by_trigger_id(
         self,

api/oss/src/core/triggers/service.py

@@ -59,6 +59,7 @@
 from oss.src.core.triggers.registry import TriggersGatewayRegistry
 from oss.src.core.triggers.utils import WebhookSecretResolver
 from oss.src.core.shared.dtos import Reference, Windowing
+from oss.src.core.shared.exceptions import EntityCreationConflict
 from oss.src.core.workflows.service import WorkflowsService
 
 
@@ -1175,20 +1176,51 @@ async def test_subscription(
         #
         subscription: TriggerSubscriptionCreate,
     ) -> Optional[TriggerDelivery]:
-        """One-shot test: create a test subscription, wait for the first captured
-        delivery (or a timeout), then tear it down.
+        """One-shot test: capture the first delivery for this (connection, event),
+        then tear down.
 
-        Pure convenience over the primitives — the same create / poll-deliveries /
-        delete lifecycle a client can drive itself. The test subscription is always
-        deleted, including on timeout and on any error.
+        Normally mints a short-lived test subscription and deletes it (including on
+        timeout and on any error). If a live subscription already owns the provider
+        trigger (the partial-unique index forbids a duplicate active row), capture
+        against that existing subscription instead and leave it in place.
         """
         subscription.flags.is_test = True
 
-        created = await self.create_subscription(
-            project_id=project_id,
-            user_id=user_id,
-            subscription=subscription,
-        )
+        owns_created = True
+        try:
+            created = await self.create_subscription(
+                project_id=project_id,
+                user_id=user_id,
+                subscription=subscription,
+            )
+        except EntityCreationConflict as err:
+            # Capture against the exact colliding subscription (project-scoped, non-deleted),
+            # not a fuzzy connection+event match that could land on a stale or inactive row.
+            trigger_id = (err.conflict or {}).get("trigger_id")
+            existing = (
+                await self.dao.fetch_subscription_by_trigger_id(
+                    project_id=project_id,
+                    trigger_id=trigger_id,
+                )
+                if trigger_id
+                else None
+            )
+            if existing is None:
+                raise
+            created = existing
+            owns_created = False
+
+        # A reused live subscription may already carry past deliveries; capture only
+        # a delivery newer than the current latest. A freshly minted test
+        # subscription has none, so no baseline is taken and the first delivery wins.
+        baseline_id = None
+        if not owns_created:
+            baseline = await self.dao.query_deliveries(
+                project_id=project_id,
+                delivery=TriggerDeliveryQuery(subscription_id=created.id),
+                windowing=Windowing(limit=1),
+            )
+            baseline_id = baseline[0].id if baseline else None
 
         try:
             deadline = asyncio.get_event_loop().time() + _TEST_TIMEOUT_SECONDS
@@ -1198,15 +1230,16 @@ async def test_subscription(
                     delivery=TriggerDeliveryQuery(subscription_id=created.id),
                     windowing=Windowing(limit=1),
                 )
-                if deliveries:
+                if deliveries and deliveries[0].id != baseline_id:
                     return deliveries[0]
                 await asyncio.sleep(1.0)
             return None
         finally:
-            await self.delete_subscription(
-                project_id=project_id,
-                subscription_id=created.id,
-            )
+            if owns_created:
+                await self.delete_subscription(
+                    project_id=project_id,
+                    subscription_id=created.id,
+                )
 
     async def _set_valid(
         self,

api/oss/src/dbs/postgres/triggers/dao.py

@@ -4,8 +4,10 @@
 
 from sqlalchemy import select
 from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.exc import IntegrityError
 
 from oss.src.core.shared.dtos import Windowing
+from oss.src.core.shared.exceptions import EntityCreationConflict
 from oss.src.core.triggers.dtos import (
     TriggerDelivery,
     TriggerDeliveryCreate,
@@ -70,12 +72,31 @@ async def create_subscription(
             trigger_id=trigger_id,
         )
 
-        async with self.engine.session() as session:
-            session.add(subscription_dbe)
-
-            await session.commit()
-
-            await session.refresh(subscription_dbe)
+        try:
+            async with self.engine.session() as session:
+                session.add(subscription_dbe)
+
+                await session.commit()
+
+                await session.refresh(subscription_dbe)
+        except IntegrityError as e:
+            # A live subscription already occupies this provider trigger; the partial-unique
+            # index forbids a second active row. Classify by the driver's constraint-name
+            # metadata, falling back to the message so a duplicate never slips through to a 500.
+            index_name = "ix_trigger_subscriptions_trigger_id"
+            orig = getattr(e, "orig", None)
+            constraint = getattr(
+                getattr(orig, "__cause__", None), "constraint_name", None
+            )
+            if constraint == index_name or index_name in (
+                str(orig) if orig else str(e)
+            ):
+                raise EntityCreationConflict(
+                    entity="Trigger subscription",
+                    message="A subscription for this connection and event already exists.",
+                    conflict={"trigger_id": trigger_id},
+                ) from e
+            raise
 
         return map_subscription_dbe_to_dto(
             subscription_dbe=subscription_dbe,
@@ -216,6 +237,32 @@ async def query_subscriptions(
                 for dbe in result.scalars().all()
             ]
 
+    async def fetch_subscription_by_trigger_id(
+        self,
+        *,
+        project_id: UUID,
+        trigger_id: str,
+    ) -> Optional[TriggerSubscription]:
+        async with self.engine.session() as session:
+            stmt = (
+                select(TriggerSubscriptionDBE)
+                .filter(
+                    TriggerSubscriptionDBE.project_id == project_id,
+                    TriggerSubscriptionDBE.trigger_id == trigger_id,
+                    TriggerSubscriptionDBE.deleted_at.is_(None),
+                )
+                .limit(1)
+            )
+
+            result = await session.execute(stmt)
+
+            subscription_dbe = result.scalars().first()
+
+            if not subscription_dbe:
+                return None
+
+            return map_subscription_dbe_to_dto(subscription_dbe=subscription_dbe)
+
     async def get_project_and_subscription_by_trigger_id(
         self,
         *,

api/oss/tests/pytest/unit/triggers/test_triggers_subscription_test_mode.py

@@ -15,6 +15,7 @@
 import pytest
 
 from oss.src.core.shared.dtos import Status
+from oss.src.core.shared.exceptions import EntityCreationConflict
 from oss.src.core.triggers.dtos import (
     TriggerDelivery,
     TriggerSubscription,
@@ -249,3 +250,45 @@ async def test_test_subscription_tears_down_on_error():
         )
 
     service.dao.delete_subscription.assert_awaited_once()
+
+
+async def test_test_subscription_reuses_existing_live_subscription_on_conflict():
+    """When a live subscription already owns the provider trigger (the mint hits
+    the partial-unique index and the DAO raises EntityCreationConflict), the test
+    captures against the existing subscription and leaves it in place."""
+    service = _make_service()
+
+    existing = _existing(is_test=False)  # a live, non-test subscription
+    service.dao.create_subscription = AsyncMock(
+        side_effect=EntityCreationConflict(
+            entity="Trigger subscription",
+            conflict={"trigger_id": "ti_existing"},
+        )
+    )
+    service.dao.fetch_subscription_by_trigger_id = AsyncMock(return_value=existing)
+
+    old_delivery = _delivery()  # pre-existing baseline delivery
+    new_delivery = _delivery()  # the newly captured one (different id)
+    service.dao.query_deliveries = AsyncMock(
+        side_effect=[[old_delivery], [new_delivery]]
+    )
+
+    result = await service.test_subscription(
+        project_id=uuid4(),
+        user_id=uuid4(),
+        subscription=_create(is_test=False),
+    )
+
+    # The freshly arrived delivery is returned, not the stale baseline.
+    assert result is new_delivery
+    # The existing live subscription must NOT be torn down.
+    service.dao.delete_subscription.assert_not_awaited()
+    # Only the single failed mint attempt — no duplicate persisted.
+    service.dao.create_subscription.assert_awaited_once()
+    # Reuse resolved via the exact colliding trigger_id, not a connection+event query.
+    service.dao.fetch_subscription_by_trigger_id.assert_awaited_once()
+    # Deliveries were polled against the existing subscription's id.
+    assert (
+        service.dao.query_deliveries.await_args.kwargs["delivery"].subscription_id
+        == existing.id
+    )

web/packages/agenta-entity-ui/src/DrillInView/SchemaControls/TriggerManagementSection.tsx

@@ -255,6 +255,7 @@ function SubscriptionRunPopover({
     return (
         <EventSourcePicker
             placement="bottomRight"
+            autoWaitOnOpen
             onOpenChange={(open) => open && refresh()}
             trigger={
                 <Tooltip title="Run in playground">

web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerScheduleDrawer.tsx

@@ -666,10 +666,15 @@ function ScheduleForm({
                 savedId = result.id ?? null
                 message.success("Schedule created")
             }
-            // Master-detail keeps the drawer open on the saved schedule; the single-form
-            // (settings) drawer closes.
-            if (onSaved && savedId) onSaved(savedId)
-            else onClose()
+            // A newly created schedule dismisses the drawer (it now shows in the triggers list);
+            // an edit keeps the master-detail open on the saved schedule (settings closes).
+            if (!isEdit) {
+                onClose()
+            } else if (onSaved && savedId) {
+                onSaved(savedId)
+            } else {
+                onClose()
+            }
         } catch (error) {
             message.error(triggerApiErrorMessage(error, "Failed to save schedule"))
         }

web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsx

@@ -759,8 +759,14 @@ function SubscriptionForm({
                 savedId = result.id ?? null
                 message.success("Trigger created")
             }
-            if (onSaved && savedId) onSaved(savedId)
-            else onClose()
+            if (!isEdit) {
+                // A newly created trigger dismisses the drawer; it now shows in the triggers list.
+                onClose()
+            } else if (onSaved && savedId) {
+                onSaved(savedId)
+            } else {
+                onClose()
+            }
         } catch (error) {
             message.error(triggerApiErrorMessage(error, "Failed to save trigger"))
         }

web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/shared/EventSourcePicker.tsx

@@ -1,4 +1,4 @@
-import {useState, type ReactNode} from "react"
+import {useRef, useState, type ReactNode} from "react"
 
 import {ClockCountdown, Lightning} from "@phosphor-icons/react"
 import {Empty, Popover, Spin} from "antd"
@@ -37,6 +37,7 @@ export function EventSourcePicker({
     placement = "bottomRight",
     waitLabel = "Wait for a new event",
     waitHint,
+    autoWaitOnOpen,
 }: {
     /** The element that opens the popover (a button). */
     trigger: ReactNode
@@ -49,33 +50,49 @@ export function EventSourcePicker({
     placement?: "bottomRight" | "topRight" | "bottomLeft" | "topLeft"
     waitLabel?: string
     waitHint?: string
+    /** Start the live capture immediately when the popover opens (single-click test). */
+    autoWaitOnOpen?: boolean
 }) {
     const [open, setOpen] = useState(false)
     const [waiting, setWaiting] = useState(false)
-    const handleOpenChange = (next: boolean) => {
-        setOpen(next)
-        onOpenChange?.(next)
-    }
+    const settledRef = useRef(false)
 
     const pick = (event: SampledEvent) => {
+        settledRef.current = true
         setOpen(false)
         onPick(event)
     }
 
     const wait = async () => {
-        if (!onWaitForEvent) return
+        if (!onWaitForEvent || waiting) return
         setWaiting(true)
         try {
             const event = await onWaitForEvent()
-            if (event) {
+            if (event && !settledRef.current) {
+                settledRef.current = true
                 setOpen(false)
                 onPick(event)
             }
+        } catch {
+            // Callers surface their own error before rejecting; swallow so the fire-and-forget
+            // `void wait()` never becomes an unhandled rejection.
         } finally {
             setWaiting(false)
         }
     }
 
+    const handleOpenChange = (next: boolean) => {
+        setOpen(next)
+        onOpenChange?.(next)
+        if (next) {
+            settledRef.current = false
+            if (autoWaitOnOpen && onWaitForEvent) void wait()
+        } else {
+            // A wait resolving after the popover closed must not fire onPick.
+            settledRef.current = true
+        }
+    }
+
     const content = (
         <div className="w-[280px]">
             {onWaitForEvent && (