From fac8ba8db3ce5881a550047f00cdf43b980c7537 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Sun, 12 Jul 2026 16:55:12 +0200 Subject: [PATCH 1/6] feat(frontend): single-click starts the app-trigger test capture --- .../TriggerManagementSection.tsx | 1 + .../drawers/shared/EventSourcePicker.tsx | 25 +++++++++++++------ 2 files changed, 19 insertions(+), 7 deletions(-) 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/shared/EventSourcePicker.tsx b/web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/shared/EventSourcePicker.tsx index 801db43e65..9bc7a0bf4c 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,25 +50,26 @@ 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) } @@ -76,6 +78,15 @@ export function EventSourcePicker({ } } + const handleOpenChange = (next: boolean) => { + setOpen(next) + onOpenChange?.(next) + if (next) { + settledRef.current = false + if (autoWaitOnOpen && onWaitForEvent) void wait() + } + } + const content = (
{onWaitForEvent && ( From e64ba0058d3a42966ba60ecb5d4bfa44bd49815e Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Sun, 12 Jul 2026 17:10:21 +0200 Subject: [PATCH 2/6] fix(api): trigger test-event no longer 500s and reuses an existing live subscription --- api/oss/src/core/triggers/service.py | 60 ++++++++++++++----- api/oss/src/dbs/postgres/triggers/dao.py | 27 +++++++-- .../test_triggers_subscription_test_mode.py | 38 ++++++++++++ 3 files changed, 104 insertions(+), 21 deletions(-) diff --git a/api/oss/src/core/triggers/service.py b/api/oss/src/core/triggers/service.py index b7ef0159a7..39696424cd 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,48 @@ 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: + existing = await self.dao.query_subscriptions( + project_id=project_id, + subscription=TriggerSubscriptionQuery( + connection_id=subscription.connection_id, + event_key=subscription.data.event_key, + ), + windowing=Windowing(limit=1), + ) + if not existing: + raise + created = existing[0] + 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 +1227,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..0c43532d13 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,25 @@ 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: + error_str = str(e.orig) if e.orig else str(e) + if "ix_trigger_subscriptions_trigger_id" in error_str: + # A live subscription already occupies this provider trigger (same + # connection + event); the partial-unique index forbids a second + # active row. + 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, 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..16bbc97d49 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,40 @@ 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") + ) + service.dao.query_subscriptions = 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() + # Deliveries were polled against the existing subscription's id. + assert ( + service.dao.query_deliveries.await_args.kwargs["delivery"].subscription_id + == existing.id + ) From 93be917eaeb7212cb0057eeef1534fe3e89e66a0 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Sun, 12 Jul 2026 21:58:11 +0200 Subject: [PATCH 3/6] feat(frontend): dismiss the trigger drawer after a trigger is created --- .../drawers/TriggerScheduleDrawer.tsx | 13 +++++++++---- .../drawers/TriggerSubscriptionDrawer.tsx | 10 ++++++++-- 2 files changed, 17 insertions(+), 6 deletions(-) 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")) } From f25eb29b0c610d41676681cf5760049ce4f58f27 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Sun, 12 Jul 2026 23:06:17 +0200 Subject: [PATCH 4/6] fix(api): reuse the exact colliding subscription by trigger_id (deleted_at-guarded) --- api/oss/src/core/triggers/interfaces.py | 10 +++++++ api/oss/src/core/triggers/service.py | 23 +++++++++------- api/oss/src/dbs/postgres/triggers/dao.py | 26 +++++++++++++++++++ .../test_triggers_subscription_test_mode.py | 9 +++++-- 4 files changed, 56 insertions(+), 12 deletions(-) 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 39696424cd..39130fde30 100644 --- a/api/oss/src/core/triggers/service.py +++ b/api/oss/src/core/triggers/service.py @@ -1193,18 +1193,21 @@ async def test_subscription( user_id=user_id, subscription=subscription, ) - except EntityCreationConflict: - existing = await self.dao.query_subscriptions( - project_id=project_id, - subscription=TriggerSubscriptionQuery( - connection_id=subscription.connection_id, - event_key=subscription.data.event_key, - ), - windowing=Windowing(limit=1), + 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 not existing: + if existing is None: raise - created = existing[0] + created = existing owns_created = False # A reused live subscription may already carry past deliveries; capture only diff --git a/api/oss/src/dbs/postgres/triggers/dao.py b/api/oss/src/dbs/postgres/triggers/dao.py index 0c43532d13..61df1975b3 100644 --- a/api/oss/src/dbs/postgres/triggers/dao.py +++ b/api/oss/src/dbs/postgres/triggers/dao.py @@ -231,6 +231,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 16bbc97d49..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 @@ -260,9 +260,12 @@ async def test_test_subscription_reuses_existing_live_subscription_on_conflict() existing = _existing(is_test=False) # a live, non-test subscription service.dao.create_subscription = AsyncMock( - side_effect=EntityCreationConflict(entity="Trigger subscription") + side_effect=EntityCreationConflict( + entity="Trigger subscription", + conflict={"trigger_id": "ti_existing"}, + ) ) - service.dao.query_subscriptions = AsyncMock(return_value=[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) @@ -282,6 +285,8 @@ async def test_test_subscription_reuses_existing_live_subscription_on_conflict() 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 From 210eee19b7f6da5a277f2664f1be13b793edb936 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Sun, 12 Jul 2026 23:06:17 +0200 Subject: [PATCH 5/6] fix(frontend): guard EventSourcePicker wait() against unhandled rejection and late resolve --- .../src/gatewayTrigger/drawers/shared/EventSourcePicker.tsx | 6 ++++++ 1 file changed, 6 insertions(+) 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 9bc7a0bf4c..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 @@ -73,6 +73,9 @@ export function EventSourcePicker({ 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) } @@ -84,6 +87,9 @@ export function EventSourcePicker({ 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 } } From 1b012bcda37fc44169c6fe5f86a21d7d06d3e9c8 Mon Sep 17 00:00:00 2001 From: Arda Erzin Date: Mon, 13 Jul 2026 00:20:19 +0200 Subject: [PATCH 6/6] fix(api): classify the trigger-id unique violation by constraint name, not message text --- api/oss/src/dbs/postgres/triggers/dao.py | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/api/oss/src/dbs/postgres/triggers/dao.py b/api/oss/src/dbs/postgres/triggers/dao.py index 61df1975b3..da9f7efef6 100644 --- a/api/oss/src/dbs/postgres/triggers/dao.py +++ b/api/oss/src/dbs/postgres/triggers/dao.py @@ -80,11 +80,17 @@ async def create_subscription( await session.refresh(subscription_dbe) except IntegrityError as e: - error_str = str(e.orig) if e.orig else str(e) - if "ix_trigger_subscriptions_trigger_id" in error_str: - # A live subscription already occupies this provider trigger (same - # connection + event); the partial-unique index forbids a second - # active row. + # 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.",