diff --git a/api/oss/src/core/triggers/interfaces.py b/api/oss/src/core/triggers/interfaces.py index bcd9811070..9c3c748127 100644 --- a/api/oss/src/core/triggers/interfaces.py +++ b/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, diff --git a/api/oss/src/core/triggers/service.py b/api/oss/src/core/triggers/service.py index b7ef0159a7..39130fde30 100644 --- a/api/oss/src/core/triggers/service.py +++ b/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, diff --git a/api/oss/src/dbs/postgres/triggers/dao.py b/api/oss/src/dbs/postgres/triggers/dao.py index b2d4e7dd79..da9f7efef6 100644 --- a/api/oss/src/dbs/postgres/triggers/dao.py +++ b/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, *, diff --git a/api/oss/tests/pytest/unit/triggers/test_triggers_subscription_test_mode.py b/api/oss/tests/pytest/unit/triggers/test_triggers_subscription_test_mode.py index 0b5e7749af..f09935d595 100644 --- a/api/oss/tests/pytest/unit/triggers/test_triggers_subscription_test_mode.py +++ b/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 + ) diff --git a/web/packages/agenta-entity-ui/src/DrillInView/SchemaControls/TriggerManagementSection.tsx b/web/packages/agenta-entity-ui/src/DrillInView/SchemaControls/TriggerManagementSection.tsx index 98d501e087..e02feacb97 100644 --- a/web/packages/agenta-entity-ui/src/DrillInView/SchemaControls/TriggerManagementSection.tsx +++ b/web/packages/agenta-entity-ui/src/DrillInView/SchemaControls/TriggerManagementSection.tsx @@ -255,6 +255,7 @@ function SubscriptionRunPopover({ return ( open && refresh()} trigger={ diff --git a/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerScheduleDrawer.tsx b/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerScheduleDrawer.tsx index 30de24a110..ef59629a07 100644 --- a/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerScheduleDrawer.tsx +++ b/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")) } diff --git a/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsx b/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsx index e0690191c8..8094a4c10d 100644 --- a/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsx +++ b/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")) } diff --git a/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/shared/EventSourcePicker.tsx b/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/shared/EventSourcePicker.tsx index 801db43e65..ca3d338a75 100644 --- a/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/shared/EventSourcePicker.tsx +++ b/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 = (
{onWaitForEvent && (