[feat] Gateway triggers: Composio event ingress, subscriptions, and UI#4749
[feat] Gateway triggers: Composio event ingress, subscriptions, and UI#4749jp-agenta wants to merge 5 commits into
Conversation
Inbound dual of webhooks: turn external provider events into Agenta workflow runs. Adds a shared routerless connections domain (core/gateway/connections), a triggers domain (event catalog, subscriptions, deliveries), a global Composio ingress endpoint with HMAC verification + async dispatch worker, and the web UI for catalog browse and subscription/delivery management. Includes design docs and unit/acceptance tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
dosubot
Bot
added
the
size:XXL
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (4)
🚧 Files skipped from review as they are similar to previous changes (3)
📝 WalkthroughSummary by CodeRabbit
WalkthroughThis PR adds shared gateway connections, trigger catalogs, subscriptions, deliveries, ingress, dispatch, worker/runtime wiring, trigger settings UI, webhook naming updates, and expanded GitButler workflow docs. ChangesGateway triggers and shared connections
Sequence Diagram(s)sequenceDiagram
participant Composio
participant TriggersRouter
participant TriggersWorker
participant TriggersDispatcher
participant WorkflowsService
Composio->>TriggersRouter: POST signed trigger event
TriggersRouter->>TriggersRouter: verify signature and parse metadata
TriggersRouter->>TriggersWorker: enqueue trigger_id, event_id, event
TriggersWorker->>TriggersDispatcher: dispatch(...)
TriggersDispatcher->>WorkflowsService: invoke_workflow(...)
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Railway Preview Environment
|
get_default_workspace_id no longer prefers owner-role (multi-org: an invitee owns their own empty personal workspace). Assert oldest membership wins regardless of role. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…g in tests Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
api/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.py (1)
182-227:⚠️ Potential issue | 🟠 Major | ⚡ Quick winGuarantee connection cleanup even on test failure.
If any assertion fails before Line 226, the connection cleanup is skipped, which can leak external/provider state and make later acceptance runs flaky. Move cleanup to a
finallyblock and assert the delete response.Suggested fix
def test_create_list_disable_delete_keeps_connection(self, triggers_api): connection_id = self._create_connection(triggers_api) - - create = triggers_api( - "POST", - "/triggers/subscriptions/", - json={ - "subscription": { - "name": f"sub-{uuid4().hex[:8]}", - "connection_id": connection_id, - "data": { - "event_key": "GITHUB_STAR_ADDED_EVENT", - "trigger_config": {}, - "inputs_fields": {"repo": "$.event.data.repository"}, - "references": {"workflow": {"slug": "triage"}}, - }, - } - }, - ) - assert create.status_code == 200, create.text - sub = create.json()["subscription"] - subscription_id = sub["id"] - assert sub["connection_id"] == connection_id - assert sub["data"]["ti_id"] is not None - - listing = triggers_api("GET", "/triggers/subscriptions/").json() - assert any(s["id"] == subscription_id for s in listing["subscriptions"]) - - revoke = triggers_api( - "POST", f"/triggers/subscriptions/{subscription_id}/revoke" - ) - assert revoke.status_code == 200, revoke.text - assert revoke.json()["subscription"]["enabled"] is False - - delete = triggers_api("DELETE", f"/triggers/subscriptions/{subscription_id}") - assert delete.status_code == 204 - - fetch = triggers_api("GET", f"/triggers/subscriptions/{subscription_id}") - assert fetch.status_code == 404 - - # C7: deleting the subscription must NOT delete/revoke the connection. - conn = triggers_api("GET", f"/tools/connections/{connection_id}") - assert conn.status_code == 200, conn.text - - triggers_api("DELETE", f"/tools/connections/{connection_id}") + try: + create = triggers_api( + "POST", + "/triggers/subscriptions/", + json={ + "subscription": { + "name": f"sub-{uuid4().hex[:8]}", + "connection_id": connection_id, + "data": { + "event_key": "GITHUB_STAR_ADDED_EVENT", + "trigger_config": {}, + "inputs_fields": {"repo": "$.event.data.repository"}, + "references": {"workflow": {"slug": "triage"}}, + }, + } + }, + ) + assert create.status_code == 200, create.text + sub = create.json()["subscription"] + subscription_id = sub["id"] + assert sub["connection_id"] == connection_id + assert sub["data"]["ti_id"] is not None + + listing = triggers_api("GET", "/triggers/subscriptions/").json() + assert any(s["id"] == subscription_id for s in listing["subscriptions"]) + + revoke = triggers_api("POST", f"/triggers/subscriptions/{subscription_id}/revoke") + assert revoke.status_code == 200, revoke.text + assert revoke.json()["subscription"]["enabled"] is False + + delete = triggers_api("DELETE", f"/triggers/subscriptions/{subscription_id}") + assert delete.status_code == 204 + + fetch = triggers_api("GET", f"/triggers/subscriptions/{subscription_id}") + assert fetch.status_code == 404 + + conn = triggers_api("GET", f"/tools/connections/{connection_id}") + assert conn.status_code == 200, conn.text + finally: + cleanup = triggers_api("DELETE", f"/tools/connections/{connection_id}") + assert cleanup.status_code == 204, cleanup.textapi/oss/tests/pytest/acceptance/triggers/test_triggers_subscriptions.py (1)
108-156:⚠️ Potential issue | 🟠 Major | ⚡ Quick winMake lifecycle cleanup unconditional.
Cleanup at Line 155 only runs on the happy path. A mid-test failure leaves shared connection state behind, which can pollute subsequent acceptance runs. Move deletion to
finallyand assert cleanup success.Suggested fix
def test_create_list_disable_delete_keeps_connection(self, authed_api): connection_id = self._create_connection(authed_api) - - # CREATE — binds the event to a workflow reference on the shared connection - create = authed_api( - "POST", - "/triggers/subscriptions/", - json={ - "subscription": { - "name": f"sub-{uuid4().hex[:8]}", - "connection_id": connection_id, - "data": { - "event_key": "GITHUB_STAR_ADDED_EVENT", - "trigger_config": {}, - "inputs_fields": {"repo": "$.event.data.repository"}, - "references": {"workflow": {"slug": "triage"}}, - }, - } - }, - ) - assert create.status_code == 200, create.text - sub = create.json()["subscription"] - subscription_id = sub["id"] - assert sub["connection_id"] == connection_id - assert sub["data"]["ti_id"] is not None - assert sub["enabled"] is True - - # LIST - listing = authed_api("GET", "/triggers/subscriptions/").json() - assert any(s["id"] == subscription_id for s in listing["subscriptions"]) - - # DISABLE (revoke the subscription, not the connection) - revoke = authed_api("POST", f"/triggers/subscriptions/{subscription_id}/revoke") - assert revoke.status_code == 200, revoke.text - assert revoke.json()["subscription"]["enabled"] is False - - # DELETE - delete = authed_api("DELETE", f"/triggers/subscriptions/{subscription_id}") - assert delete.status_code == 204 - - fetch = authed_api("GET", f"/triggers/subscriptions/{subscription_id}") - assert fetch.status_code == 404 - - # C7: deleting the subscription must NOT delete/revoke the connection. - conn = authed_api("GET", f"/tools/connections/{connection_id}") - assert conn.status_code == 200, conn.text - - authed_api("DELETE", f"/tools/connections/{connection_id}") + try: + create = authed_api( + "POST", + "/triggers/subscriptions/", + json={ + "subscription": { + "name": f"sub-{uuid4().hex[:8]}", + "connection_id": connection_id, + "data": { + "event_key": "GITHUB_STAR_ADDED_EVENT", + "trigger_config": {}, + "inputs_fields": {"repo": "$.event.data.repository"}, + "references": {"workflow": {"slug": "triage"}}, + }, + } + }, + ) + assert create.status_code == 200, create.text + sub = create.json()["subscription"] + subscription_id = sub["id"] + assert sub["connection_id"] == connection_id + assert sub["data"]["ti_id"] is not None + assert sub["enabled"] is True + + listing = authed_api("GET", "/triggers/subscriptions/").json() + assert any(s["id"] == subscription_id for s in listing["subscriptions"]) + + revoke = authed_api("POST", f"/triggers/subscriptions/{subscription_id}/revoke") + assert revoke.status_code == 200, revoke.text + assert revoke.json()["subscription"]["enabled"] is False + + delete = authed_api("DELETE", f"/triggers/subscriptions/{subscription_id}") + assert delete.status_code == 204 + + fetch = authed_api("GET", f"/triggers/subscriptions/{subscription_id}") + assert fetch.status_code == 404 + + conn = authed_api("GET", f"/tools/connections/{connection_id}") + assert conn.status_code == 200, conn.text + finally: + cleanup = authed_api("DELETE", f"/tools/connections/{connection_id}") + assert cleanup.status_code == 204, cleanup.textdocs/designs/gateway-triggers/proposal.md (1)
285-289:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winRemove the empty quoted line.
The blank
>line before the next heading trips MD028 and will keep docs lint failing.As per the markdownlint hint, this blockquote needs to stay contiguous.
🛠️ Proposed fix
> **Consequence — cross-domain revoke.** Because `ca_*` is shared, revoking it affects > both tools actions and trigger subscriptions on it. Lean: revoke-for-everyone + show > usage; deleting a subscription must not revoke the connection. Connect once, used > everywhere — the inverse of the connect-twice cost that rejected option B carried. -> ### The workflow dispatch seamSource: Linters/SAST tools
🟡 Minor comments (16)
web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerSubscriptions.ts-54-58 (1)
54-58:⚠️ Potential issue | 🟡 MinorGate loading state for empty
connectionIdto match the pattern inuseTriggerDeliveries.Line 57 exposes loading while the query is disabled. When
connectionIdis empty, the query is disabled (enabled: !!connectionIdat line 42) butisLoadingstill reflectsquery.isPending. Apply the same gating pattern used inuseTriggerDeliveriesto returnfalsewhen the query is disabled:Suggested fix
return { subscriptions, count: query.data?.count ?? 0, - isLoading: query.isPending, + isLoading: connectionId ? query.isPending : false, error: query.error, } }web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerConnections.ts-60-64 (1)
60-64:⚠️ Potential issue | 🟡 MinorGate loading state when
integrationKeyis empty.Line 63 can surface a loading state even when the query is intentionally disabled (line 48). Mirror the
useTriggerDeliverieshook pattern and gateisLoadingby the input key.Suggested fix
export const useTriggerIntegrationConnections = (integrationKey: string) => { const query = useAtomValue(triggerIntegrationConnectionsAtomFamily(integrationKey)) const connections = useMemo<TriggerConnection[]>( () => query.data?.connections ?? [], [query.data?.connections], ) return { connections, count: query.data?.count ?? 0, - isLoading: query.isPending, + isLoading: integrationKey ? query.isPending : false, error: query.error, } }web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerEvent.ts-32-35 (1)
32-35:⚠️ Potential issue | 🟡 MinorMake
isLoadingconditional on both required keys.Line 34 should be gated by
integrationKeyandeventKeyso consumers don't get a loading signal when the query is disabled due to missing prerequisites.Suggested fix
return { event: query.data?.event ?? null, - isLoading: query.isPending, + isLoading: integrationKey && eventKey ? query.isPending : false, error: query.error, } }web/oss/src/components/pages/settings/Triggers/components/GatewaySubscriptionsSection.tsx-248-248 (1)
248-248:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winAvoid non-unique
rowKeyfallback.Line 248 can return
"", which can produce duplicate keys and unstable row identity when multiple records miss these fields.Suggested fix
- rowKey={(record) => record.id ?? record.slug ?? record.data?.event_key ?? ""} + rowKey={(record) => + record.id ?? + record.slug ?? + `${record.connection_id ?? "unknown"}:${record.data?.event_key ?? "unknown"}:${record.created_at ?? "unknown"}` + }web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsx-97-100 (1)
97-100:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winEdit prefill uses raw workflow revision ID as UI label
Line 99 sets
workflowLabelto the revision ID, so edit mode can display an opaque ID instead of the workflow display name/version format used elsewhere.As per coding guidelines, display workflow names via
workflowMolecule.selectors.artifactName(entityId)and use the prescribed label rules for workflow/variant naming.Source: Coding guidelines
web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerDeliveriesDrawer.tsx-51-55 (1)
51-55:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winStatus tag color ignores the status-code fallback
Line 51 already falls back to
status.code, but Line 54 colors usingstatus.typeonly. Code-only statuses can display the wrong semantic color.Suggested fix
-const type = record.status?.type ?? record.status?.code +const type = record.status?.type ?? record.status?.code return ( <Tooltip title={record.status?.message ?? undefined}> - <Tag color={statusColor(record.status?.type)}>{type ?? "unknown"}</Tag> + <Tag color={statusColor(type)}>{type ?? "unknown"}</Tag> </Tooltip> )web/oss/src/components/pages/settings/Triggers/components/GatewayTriggersSection.tsx-56-57 (1)
56-57:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winName column can render blank for valid rows
Line 56 falls back to
slug, but if bothnameandslugare absent the cell is empty. Addintegration_keyas final fallback so every row remains identifiable.Suggested fix
-<Typography.Text>{record.name || record.slug}</Typography.Text> +<Typography.Text>{record.name || record.slug || record.integration_key}</Typography.Text>web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsx-138-146 (1)
138-146:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winReject non-object JSON for
inputs_fieldsLine 140 accepts any valid JSON (including arrays/primitives), but
inputs_fieldsis expected to be an object map. This can produce contract failures downstream.Suggested fix
let inputsFields: Record<string, unknown> = {} try { - inputsFields = inputsText.trim() ? JSON.parse(inputsText) : {} + const parsed = inputsText.trim() ? JSON.parse(inputsText) : {} + if (parsed === null || Array.isArray(parsed) || typeof parsed !== "object") { + throw new Error("Inputs mapping must be a JSON object") + } + inputsFields = parsed as Record<string, unknown> setInputsError(null) } catch { setInputsError("Invalid JSON") message.error("inputs mapping is not valid JSON") return }api/oss/src/core/gateway/connections/dtos.py-112-118 (1)
112-118:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winUse enum typing for
ConnectionRequest.auth_schemeto keep validation consistent.
ConnectionCreateData.auth_schemeis enum-validated, butConnectionRequest.auth_schemeis a plainstr, which lets invalid schemes bypass early validation and fail later in adapter calls.🔧 Proposed fix
class ConnectionRequest(BaseModel): @@ - auth_scheme: Optional[str] = None + auth_scheme: Optional[ConnectionAuthScheme] = Noneapi/oss/src/apis/fastapi/triggers/router.py-376-377 (1)
376-377:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winValidate
limitat the API boundary.Line 376 allows invalid values (e.g.,
0or negative), which can propagate upstream and surface as adapter failures instead of a clean client validation error.Suggested fix
- limit: Optional[int] = Query(default=None), + limit: Optional[int] = Query(default=None, ge=1, le=1000),docs/designs/gateway-triggers/wp/WP4-specs.md-20-25 (1)
20-25:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winUpdate WP4 ACK semantics from 200 to 202.
Line 22 and Line 54 document
200for no-op/unset-secret, but this PR’s ingress contract is202 Accepted. Keeping200here creates avoidable contract drift.Suggested doc fix
-- HMAC-SHA256 verify over `{id}.{ts}.{body}` with `COMPOSIO_WEBHOOK_SECRET`; 401 bad sig; - 200 no-op when secret unset; add `COMPOSIO_WEBHOOK_SECRET` to `env`. +- HMAC-SHA256 verify over `{id}.{ts}.{body}` with `COMPOSIO_WEBHOOK_SECRET`; 401 bad sig; + 202 no-op when secret unset; add `COMPOSIO_WEBHOOK_SECRET` to `env`. @@ -- Forged signature → 401; unset secret → 200 no-op. +- Forged signature → 401; unset secret → 202 no-op.Also applies to: 54-55
docs/designs/gateway-triggers/wp/WP4-status.md-7-25 (1)
7-25:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winRefresh WP4 status to match implemented work.
The document still marks WP4 as
NOT STARTED, but this stack includes ingress, dispatch runtime, worker wiring, and tests. This stale status will confuse follow-on planning.docs/designs/gateway-triggers/wp/WP0-status.md-15-18 (1)
15-18:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winAlign the connection path spelling.
This status file still uses the flat
core/connections/dbs/postgres/connectionsnames, but the spec and implementation cohort use thegateway/subpaths. Keeping both spellings will send the next lane to the wrong package.Based on the stack context, this PR already uses
core/gateway/connections/anddbs/postgres/gateway/connections/.🛠️ Suggested alignment
- [x] `core/connections/` service + DAO interface + `ConnectionsService` + `ConnectionsGatewayInterface` + [x] `core/gateway/connections/` service + DAO interface + `ConnectionsService` + `ConnectionsGatewayInterface` - [x] `dbs/postgres/connections/` DBE + DAO + mappings + [x] `dbs/postgres/gateway/connections/` DBE + DAO + mappingsAlso applies to: 50-59
docs/designs/gateway-triggers/wp/WL-runbook.md-128-154 (1)
128-154:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winLabel the markdown example fence.
The unlabeled fence trips MD040;
mdis sufficient here.As per the markdownlint hint, the docs PR body example needs a language tag.
🛠️ Proposed fix
-``` +```mdSource: Linters/SAST tools
docs/designs/gateway-triggers/wp/WP1-status.md-41-42 (1)
41-42:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winPermission note conflicts with the WP1 frozen contract.
This says no
VIEW_TRIGGERSand reusesVIEW_TOOLS, but WP1 specs freeze the opposite rule. Please reconcile this to avoid implementer confusion on authorization behavior.docs/designs/gateway-triggers/wp/WP2-specs.md-29-33 (1)
29-33:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winFrozen resolver return type is too narrow.
The spec locks
-> dict, but current trigger dispatch logic already handles scalar/non-dict results fromresolve_target_fields. Please widen the contract wording to avoid future mismatches.
🧹 Nitpick comments (9)
web/packages/agenta-entities/src/gatewayTrigger/api/client.ts (1)
5-19: ⚡ Quick winTrim block comments to “why”-only notes.
These docblocks narrate implementation details (“what”) across many lines; please reduce to terse invariant-level comments only, or rely on naming/extraction.
As per coding guidelines, “Keep AI-generated in-code comments minimal; comment only the non-obvious why, never the what.”
Source: Coding guidelines
web/packages/agenta-entities/src/gatewayTrigger/api/api.ts (1)
1-12: ⚡ Quick winReduce explanatory prose comments in this module.
The section/header comments mostly restate behavior; please keep only short “why” notes where needed.
As per coding guidelines, comments should be minimal and explain non-obvious rationale, not implementation narration.
Also applies to: 42-42, 111-111, 133-133, 243-243
Source: Coding guidelines
web/packages/agenta-entities/tests/unit/gatewayTriggerApi.test.ts (1)
1-14: ⚡ Quick winShorten the top-of-file comment block.
Please keep only the non-obvious invariant/intent and drop step-by-step prose.
As per coding guidelines, comments should be terse and explain “why,” not narrate “what.”
Source: Coding guidelines
web/packages/agenta-entity-ui/src/gatewayTrigger/index.ts (1)
1-8: ⚡ Quick winCondense the module docblock to a one-liner (or remove).
Current block explains behavior in detail; this can be inferred from exports and naming.
As per coding guidelines, keep in-code comments minimal and focused on subtle “why” context only.
Source: Coding guidelines
web/packages/agenta-entities/src/gatewayTrigger/core/types.ts (1)
1-15: ⚡ Quick winTrim non-obvious-comment blocks to terse intent notes only.
These large narrative blocks describe what and migration context; please reduce to minimal “why” comments (or rely on symbol names/module docs) to match repo standards.
As per coding guidelines, “Keep AI-generated in-code comments minimal; comment only the non-obvious why … never the what.”
Also applies to: 90-95, 134-140, 235-239
Source: Coding guidelines
web/packages/agenta-entities/src/gatewayTrigger/hooks/useCatalogEvents.ts (1)
14-16: ⚡ Quick winUnify catalog-search atom ownership to avoid split state.
eventsSearchAtomhere overlaps conceptually witheventSearchAtominstate/atoms.ts; keeping two similarly named atoms for the same concern invites accidental desync across consumers. Prefer exporting one shared atom from a single module and reusing it here.api/oss/tests/pytest/acceptance/triggers/test_triggers_ingress.py (1)
151-163: ⚡ Quick winMake the dedup assertion deterministic and cleanup-safe.
Line 151 currently allows a false positive (
<= 1) when the async dispatcher hasn’t written any row yet. Poll for completion (bounded) and then assert exactly one delivery; also run cleanup infinallywith response checks.Proposed test hardening
+from time import sleep @@ - # The dispatch is async; the dedup guard means at most one delivery row - # exists for this (subscription, event_id). - deliveries = authed_api( - "POST", - "/triggers/deliveries/query", - json={ - "delivery": {"subscription_id": subscription_id, "event_id": event_id} - }, - ).json()["deliveries"] - assert len(deliveries) <= 1 - - authed_api("DELETE", f"/triggers/subscriptions/{subscription_id}") - authed_api("DELETE", f"/tools/connections/{connection_id}") + try: + deliveries = [] + for _ in range(20): + query = authed_api( + "POST", + "/triggers/deliveries/query", + json={ + "delivery": {"subscription_id": subscription_id, "event_id": event_id} + }, + ) + assert query.status_code == 200, query.text + deliveries = query.json()["deliveries"] + if deliveries: + break + sleep(0.25) + + assert len(deliveries) == 1 + finally: + delete_sub = authed_api("DELETE", f"/triggers/subscriptions/{subscription_id}") + assert delete_sub.status_code in (200, 204), delete_sub.text + delete_conn = authed_api("DELETE", f"/tools/connections/{connection_id}") + assert delete_conn.status_code in (200, 204), delete_conn.textapi/oss/tests/pytest/unit/triggers/test_triggers_dispatcher.py (1)
129-149: ⚡ Quick winAdd coverage for the raised-exception branch in
dispatch.Current tests validate non-200 responses, but not the
invoke_workflowexception path that should write a failed delivery and re-raise.Suggested additional unit test
+import pytest @@ async def test_workflow_non_200_writes_failed_delivery(): @@ assert delivery.status.code == "500" + + +async def test_workflow_exception_writes_failed_delivery_and_reraises(): + project_id = uuid4() + reference = Reference(slug="wf-1") + subscription = _make_subscription(references={"workflow": reference}) + dao = _make_dao(resolved=(project_id, subscription)) + + workflows = MagicMock() + workflows.invoke_workflow = AsyncMock(side_effect=RuntimeError("boom")) + dispatcher = TriggersDispatcher(triggers_dao=dao, workflows_service=workflows) + + with pytest.raises(RuntimeError, match="boom"): + await dispatcher.dispatch(trigger_id="ti_1", event_id="e1", event=_EVENT) + + dao.write_delivery.assert_awaited_once() + delivery = dao.write_delivery.await_args.kwargs["delivery"] + assert delivery.status.code == "500"api/oss/src/tasks/asyncio/triggers/dispatcher.py (1)
1-9: ⚡ Quick winTrim narrative comments to “why-only” comments.
Several comments/docstrings here narrate behavior rather than a non-obvious invariant. Please reduce these to terse “why” notes (or rely on naming/extraction).
As per coding guidelines, "Keep AI-generated in-code comments minimal; comment only the non-obvious why ... never the what."
Also applies to: 109-130
Source: Coding guidelines
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: bb4b1c2d-a35f-4d60-ab97-8083ce9d0008
📒 Files selected for processing (125)
AGENTS.mdapi/ee/src/core/access/permissions/types.pyapi/ee/tests/pytest/acceptance/tools/__init__.pyapi/ee/tests/pytest/acceptance/tools/test_tools_connections.pyapi/ee/tests/pytest/acceptance/triggers/__init__.pyapi/ee/tests/pytest/acceptance/triggers/test_triggers_catalog.pyapi/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/entrypoints/routers.pyapi/entrypoints/worker_triggers.pyapi/oss/databases/postgres/migrations/core_oss/versions/oss000000002_rename_tool_connections_to_gateway_connections.pyapi/oss/databases/postgres/migrations/core_oss/versions/oss000000003_add_trigger_subscriptions_and_deliveries.pyapi/oss/src/apis/fastapi/tools/models.pyapi/oss/src/apis/fastapi/tools/router.pyapi/oss/src/apis/fastapi/triggers/__init__.pyapi/oss/src/apis/fastapi/triggers/models.pyapi/oss/src/apis/fastapi/triggers/router.pyapi/oss/src/core/gateway/__init__.pyapi/oss/src/core/gateway/connections/__init__.pyapi/oss/src/core/gateway/connections/dtos.pyapi/oss/src/core/gateway/connections/exceptions.pyapi/oss/src/core/gateway/connections/interfaces.pyapi/oss/src/core/gateway/connections/providers/__init__.pyapi/oss/src/core/gateway/connections/providers/composio/__init__.pyapi/oss/src/core/gateway/connections/providers/composio/adapter.pyapi/oss/src/core/gateway/connections/registry.pyapi/oss/src/core/gateway/connections/service.pyapi/oss/src/core/gateway/connections/utils.pyapi/oss/src/core/tools/dtos.pyapi/oss/src/core/tools/interfaces.pyapi/oss/src/core/tools/providers/composio/adapter.pyapi/oss/src/core/tools/service.pyapi/oss/src/core/triggers/__init__.pyapi/oss/src/core/triggers/dtos.pyapi/oss/src/core/triggers/exceptions.pyapi/oss/src/core/triggers/interfaces.pyapi/oss/src/core/triggers/providers/__init__.pyapi/oss/src/core/triggers/providers/composio/__init__.pyapi/oss/src/core/triggers/providers/composio/adapter.pyapi/oss/src/core/triggers/providers/composio/catalog.pyapi/oss/src/core/triggers/registry.pyapi/oss/src/core/triggers/service.pyapi/oss/src/core/webhooks/delivery.pyapi/oss/src/dbs/postgres/gateway/__init__.pyapi/oss/src/dbs/postgres/gateway/connections/__init__.pyapi/oss/src/dbs/postgres/gateway/connections/dao.pyapi/oss/src/dbs/postgres/gateway/connections/dbes.pyapi/oss/src/dbs/postgres/gateway/connections/mappings.pyapi/oss/src/dbs/postgres/triggers/__init__.pyapi/oss/src/dbs/postgres/triggers/dao.pyapi/oss/src/dbs/postgres/triggers/dbas.pyapi/oss/src/dbs/postgres/triggers/dbes.pyapi/oss/src/dbs/postgres/triggers/mappings.pyapi/oss/src/middlewares/auth.pyapi/oss/src/tasks/asyncio/triggers/__init__.pyapi/oss/src/tasks/asyncio/triggers/dispatcher.pyapi/oss/src/tasks/taskiq/triggers/__init__.pyapi/oss/src/tasks/taskiq/triggers/worker.pyapi/oss/src/utils/env.pyapi/oss/tests/pytest/acceptance/tools/test_tools_connections.pyapi/oss/tests/pytest/acceptance/triggers/__init__.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_catalog.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_ingress.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/oss/tests/pytest/unit/models/test_lifecycle_conventions.pyapi/oss/tests/pytest/unit/triggers/__init__.pyapi/oss/tests/pytest/unit/triggers/test_triggers_dispatcher.pyapi/oss/tests/pytest/unit/triggers/test_triggers_signature.pyapi/oss/tests/pytest/unit/webhooks/test_webhooks_tasks.pydocs/designs/gateway-triggers/gap.mddocs/designs/gateway-triggers/mapping.mddocs/designs/gateway-triggers/mimics.mddocs/designs/gateway-triggers/plan.mddocs/designs/gateway-triggers/proposal.mddocs/designs/gateway-triggers/research.mddocs/designs/gateway-triggers/wp/WL-runbook.mddocs/designs/gateway-triggers/wp/WP0-specs.mddocs/designs/gateway-triggers/wp/WP0-status.mddocs/designs/gateway-triggers/wp/WP1-specs.mddocs/designs/gateway-triggers/wp/WP1-status.mddocs/designs/gateway-triggers/wp/WP2-specs.mddocs/designs/gateway-triggers/wp/WP2-status.mddocs/designs/gateway-triggers/wp/WP3-specs.mddocs/designs/gateway-triggers/wp/WP3-status.mddocs/designs/gateway-triggers/wp/WP4-specs.mddocs/designs/gateway-triggers/wp/WP4-status.mddocs/designs/gateway-triggers/wp/WP5-specs.mddocs/designs/gateway-triggers/wp/WP5-status.mddocs/designs/gateway-triggers/wp/WP6-specs.mddocs/designs/gateway-triggers/wp/WP6-status.mdhosting/docker-compose/ee/docker-compose.dev.ymlhosting/docker-compose/ee/docker-compose.gh.local.ymlhosting/docker-compose/ee/docker-compose.gh.ymlhosting/docker-compose/oss/docker-compose.dev.ymlhosting/docker-compose/oss/docker-compose.gh.local.ymlhosting/docker-compose/oss/docker-compose.gh.ssl.ymlhosting/docker-compose/oss/docker-compose.gh.ymlsdks/python/agenta/sdk/utils/resolvers.pysdks/python/oss/tests/pytest/unit/test_resolvers.pyweb/oss/src/components/Sidebar/SettingsSidebar.tsxweb/oss/src/components/pages/settings/Triggers/Triggers.tsxweb/oss/src/components/pages/settings/Triggers/components/GatewaySubscriptionsSection.tsxweb/oss/src/components/pages/settings/Triggers/components/GatewayTriggersSection.tsxweb/oss/src/pages/w/[workspace_id]/p/[project_id]/settings/index.tsxweb/packages/agenta-entities/package.jsonweb/packages/agenta-entities/src/gatewayTrigger/api/api.tsweb/packages/agenta-entities/src/gatewayTrigger/api/client.tsweb/packages/agenta-entities/src/gatewayTrigger/api/index.tsweb/packages/agenta-entities/src/gatewayTrigger/core/index.tsweb/packages/agenta-entities/src/gatewayTrigger/core/types.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/index.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useCatalogEvents.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerConnections.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerDeliveries.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerEvent.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerSubscription.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerSubscriptions.tsweb/packages/agenta-entities/src/gatewayTrigger/index.tsweb/packages/agenta-entities/src/gatewayTrigger/state/atoms.tsweb/packages/agenta-entities/src/gatewayTrigger/state/index.tsweb/packages/agenta-entities/tests/unit/gatewayTriggerApi.test.tsweb/packages/agenta-entity-ui/package.jsonweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerDeliveriesDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerEventsDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/index.ts
💤 Files with no reviewable changes (1)
- api/oss/src/core/tools/dtos.py
| if self.dispatch_task is not None: | ||
| await self.dispatch_task.kiq( | ||
| trigger_id=str(trigger_id), | ||
| event_id=str(event_id), | ||
| event=envelope, | ||
| ) |
There was a problem hiding this comment.
Bound enqueue latency on the ingress request path.
Line 803 awaits broker enqueue directly with no timeout/error shaping. A slow or stuck broker call can pin request workers and degrade ingress availability.
Suggested fix
+import asyncio
@@
if self.dispatch_task is not None:
- await self.dispatch_task.kiq(
- trigger_id=str(trigger_id),
- event_id=str(event_id),
- event=envelope,
- )
+ try:
+ await asyncio.wait_for(
+ self.dispatch_task.kiq(
+ trigger_id=str(trigger_id),
+ event_id=str(event_id),
+ event=envelope,
+ ),
+ timeout=2.0,
+ )
+ except Exception as e:
+ log.error("Failed to enqueue trigger event: %s", e)
+ raise HTTPException(
+ status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+ detail="Failed to enqueue trigger event",
+ ) from e| async def get_connection_status( | ||
| self, | ||
| *, | ||
| provider_connection_id: str, | ||
| ) -> Dict[str, Any]: | ||
| """Poll provider for updated connection status.""" | ||
| ... | ||
|
|
||
| @abstractmethod | ||
| async def refresh_connection( | ||
| self, | ||
| *, | ||
| provider_connection_id: str, | ||
| force: bool = False, | ||
| callback_url: Optional[str] = None, | ||
| integration_key: Optional[str] = None, | ||
| user_id: Optional[str] = None, | ||
| ) -> Dict[str, Any]: ... |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | 🏗️ Heavy lift
Replace raw Dict[str, Any] returns with explicit DTOs in gateway interface methods.
Using untyped dicts for get_connection_status and refresh_connection weakens the core contract and makes downstream handling brittle. Define dedicated response DTOs and return those from the interface.
As per coding guidelines, api/oss/src/{core,clients}/**/*.py: “Do NOT use Dict[str, Any] as a return type when you can define a proper model”.
Source: Coding guidelines
| def __init__( | ||
| self, | ||
| *, | ||
| adapter_registry: TriggersGatewayRegistry, | ||
| triggers_dao: Optional[TriggersDAOInterface] = None, | ||
| connections_service: Optional[ConnectionsService] = None, | ||
| ): | ||
| self.adapter_registry = adapter_registry | ||
| self.dao = triggers_dao | ||
| self.connections_service = connections_service |
There was a problem hiding this comment.
Constructor allows invalid None dependencies and later crashes.
Line 48 and Line 49 store nullable dependencies, but methods call self.dao / self.connections_service unconditionally. Instantiating with defaults produces runtime AttributeError paths.
Proposed fix
- def __init__(
- self,
- *,
- adapter_registry: TriggersGatewayRegistry,
- triggers_dao: Optional[TriggersDAOInterface] = None,
- connections_service: Optional[ConnectionsService] = None,
- ):
+ def __init__(
+ self,
+ *,
+ adapter_registry: TriggersGatewayRegistry,
+ triggers_dao: TriggersDAOInterface,
+ connections_service: ConnectionsService,
+ ):
self.adapter_registry = adapter_registry
self.dao = triggers_dao
self.connections_service = connections_service| async def activate_connection_by_provider_id( | ||
| self, | ||
| *, | ||
| provider_connection_id: str, | ||
| project_id: Optional[UUID] = None, | ||
| ) -> Optional[Connection]: | ||
| """Set is_valid=True and is_active=True for the connection matching the provider ID.""" | ||
| async with self.engine.session() as session: | ||
| stmt = select(self.ConnectionDBE).filter( | ||
| self.ConnectionDBE.data["connected_account_id"].astext | ||
| == provider_connection_id | ||
| ) | ||
|
|
||
| if project_id is not None: | ||
| stmt = stmt.filter(self.ConnectionDBE.project_id == project_id) | ||
|
|
There was a problem hiding this comment.
Enforce tenant scope for provider-ID lookup and activation.
These methods allow read/update by provider_connection_id without mandatory project_id, which enables cross-project access patterns and weakens tenant isolation.
Suggested fix
async def activate_connection_by_provider_id(
self,
*,
provider_connection_id: str,
- project_id: Optional[UUID] = None,
+ project_id: UUID,
) -> Optional[Connection]:
@@
- stmt = select(self.ConnectionDBE).filter(
- self.ConnectionDBE.data["connected_account_id"].astext
- == provider_connection_id
- )
-
- if project_id is not None:
- stmt = stmt.filter(self.ConnectionDBE.project_id == project_id)
+ stmt = select(self.ConnectionDBE).filter(
+ self.ConnectionDBE.project_id == project_id,
+ self.ConnectionDBE.data["connected_account_id"].astext
+ == provider_connection_id,
+ )
@@
async def find_connection_by_provider_id(
self,
*,
+ project_id: UUID,
provider_connection_id: str,
) -> Optional[Connection]:
@@
stmt = (
select(self.ConnectionDBE)
.filter(
+ self.ConnectionDBE.project_id == project_id,
self.ConnectionDBE.data["connected_account_id"].astext
== provider_connection_id
)
.limit(1)
)As per coding guidelines, api/oss/src/dbs/postgres/**/dao.py: “Always enforce tenant scope (project_id minimum) in DAO reads and writes.”
Also applies to: 260-274
Source: Coding guidelines
| async def get_subscription_by_trigger_id( | ||
| self, | ||
| *, | ||
| trigger_id: str, | ||
| ) -> Optional[TriggerSubscription]: | ||
| async with self.engine.session() as session: | ||
| stmt = ( | ||
| select(TriggerSubscriptionDBE) | ||
| .filter( | ||
| TriggerSubscriptionDBE.data["ti_id"].astext == trigger_id, | ||
| ) | ||
| .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, | ||
| *, | ||
| trigger_id: str, | ||
| ) -> Optional[Tuple[UUID, TriggerSubscription]]: | ||
| async with self.engine.session() as session: | ||
| stmt = ( | ||
| select(TriggerSubscriptionDBE) | ||
| .filter( | ||
| TriggerSubscriptionDBE.data["ti_id"].astext == trigger_id, | ||
| ) | ||
| .limit(1) | ||
| ) |
There was a problem hiding this comment.
Enforce tenant scope for trigger-id subscription lookups.
Line 220 and Line 245 resolve subscriptions by ti_id without a tenant filter. This breaks the DAO-level tenant-scope invariant and can misroute data across projects if ids collide or are replayed.
As per coding guidelines, api/oss/src/dbs/postgres/**/dao.py: “Always enforce tenant scope (project_id minimum) in DAO reads and writes”.
Source: Coding guidelines
| # Preserve the provider ti_id even if the client omitted it on the full-PUT. | ||
| existing_ti_id = (subscription_dbe.data or {}).get("ti_id") | ||
| data = subscription.data | ||
| if data.ti_id is None and existing_ti_id is not None: | ||
| data = data.model_copy(update={"ti_id": existing_ti_id}) | ||
|
|
||
| subscription_dbe.data = data.model_dump(mode="json", exclude_none=True) |
There was a problem hiding this comment.
Prevent client-controlled overwrite of provider ti_id during edit.
On Line 110 and Line 113, a caller can submit data.ti_id and persist it. That allows rebinding a local subscription to an arbitrary provider trigger id, which is a cross-boundary identity integrity risk.
Proposed fix
- existing_ti_id = (subscription_dbe.data or {}).get("ti_id")
- data = subscription.data
- if data.ti_id is None and existing_ti_id is not None:
- data = data.model_copy(update={"ti_id": existing_ti_id})
+ existing_ti_id = (subscription_dbe.data or {}).get("ti_id")
+ data = subscription.data.model_copy(update={"ti_id": existing_ti_id})
subscription_dbe.data = data.model_dump(mode="json", exclude_none=True)| already_seen = await self.triggers_dao.dedup_seen( | ||
| project_id=project_id, | ||
| subscription_id=subscription.id, | ||
| event_id=event_id, | ||
| ) | ||
| if already_seen: | ||
| log.info( | ||
| "[TRIGGERS DISPATCHER] Duplicate event %s for subscription %s — skipping", | ||
| event_id, | ||
| subscription.id, | ||
| ) | ||
| return |
There was a problem hiding this comment.
Persisting failure before re-raise can short-circuit retries.
This flow records a failed delivery and then re-raises. With retry-on-error enabled in the TaskIQ worker, retry attempts can be skipped by the dedup guard, so transient failures may never get a real retry. Pick one policy: either retry first and persist only on terminal attempt, or persist and do not re-raise.
Also applies to: 166-176
| export function projectScopedParams(extra?: Record<string, unknown>) { | ||
| const projectId = getDefaultStore().get(projectIdAtom) | ||
| return { | ||
| params: { | ||
| ...(projectId ? {project_id: projectId} : {}), | ||
| ...(extra ?? {}), | ||
| }, |
There was a problem hiding this comment.
Prevent extra from overriding project_id.
projectScopedParams() currently lets extra.project_id overwrite the scoped value, which can bypass intended tenant scoping at call sites.
Suggested fix
export function projectScopedParams(extra?: Record<string, unknown>) {
const projectId = getDefaultStore().get(projectIdAtom)
return {
params: {
- ...(projectId ? {project_id: projectId} : {}),
...(extra ?? {}),
+ ...(projectId ? {project_id: projectId} : {}),
},
}
}As per coding guidelines, project-scoped/shared-state wiring should preserve expected boundary contracts and avoid fragile config/query patterns.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| export function projectScopedParams(extra?: Record<string, unknown>) { | |
| const projectId = getDefaultStore().get(projectIdAtom) | |
| return { | |
| params: { | |
| ...(projectId ? {project_id: projectId} : {}), | |
| ...(extra ?? {}), | |
| }, | |
| export function projectScopedParams(extra?: Record<string, unknown>) { | |
| const projectId = getDefaultStore().get(projectIdAtom) | |
| return { | |
| params: { | |
| ...(extra ?? {}), | |
| ...(projectId ? {project_id: projectId} : {}), | |
| }, | |
| } | |
| } |
Source: Coding guidelines
Normalize the inbound provider envelope in the dispatcher into a stable context (event.attributes + synthetic trigger_id/trigger_type/timestamp/ created_at), parallel to webhooks' event context. Resolve and complete the bound workflow reference on subscription create/edit (the /deploy pattern) so a variant id is resolved to a runnable revision. Align the drawer's mapping suggestions + live preview to the same normalized shape. Update trigger tests to the new shape and always-verify ingress; gate the create-roundtrip acceptance tests on an ACTIVE connected account. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| body = r.json() | ||
| print(f" ✅ id={body.get('id')}") | ||
| print("\n Set this in your env so signature verification passes:") | ||
| print(f" COMPOSIO_WEBHOOK_SECRET={body.get('secret')}") |
| def one(i: int) -> str: | ||
| with httpx.Client(timeout=20, base_url=base, headers=headers) as c: | ||
| secret = _resolve_secret(c, cache, url, force=False) | ||
| print(f" container#{i}: secret={secret[:12]}…") |
There was a problem hiding this comment.
Actionable comments posted: 14
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
web/packages/agenta-entity-ui/src/gatewayTool/drawers/ConnectDrawer.tsx (1)
80-90:⚠️ Potential issue | 🔴 CriticalDrawer mutation flow misses trigger cache invalidation.
After
createToolConnection, both mutation paths (lines 90 and 119) callinvalidateConnections()which only invalidates tool caches. Because connections are shared between tools and triggers (stored in the samegateway_connectionsrows), the trigger connection queries remain stale. Compare withTriggerConnectDrawer, which correctly invalidates both surfaces.Suggested fix
const invalidateConnections = useCallback(() => { queryClient.invalidateQueries({queryKey: ["tools", "connections"]}) queryClient.invalidateQueries({queryKey: ["tools", "catalog"]}) + queryClient.invalidateQueries({queryKey: ["triggers", "connections"]}) }, [])web/oss/src/components/pages/settings/Tools/hooks/useToolsConnections.ts (1)
22-35:⚠️ Potential issue | 🟠 MajorInvalidate trigger connection cache on tool-side mutations.
The
invalidate()callback (lines 22–35) and its use inhandleCreate(line 56–58),handleDelete, andhandleRefreshonly invalidate["tools", ...]keys. Since tools and triggers share the samegateway_connectionsrows, trigger screens can display stale connection state after these mutations until an unrelated refetch occurs. Other files in the codebase (useToolConnectionActions.ts,useTriggerConnectionActions.ts) correctly invalidate both surfaces.Suggested fix
const invalidate = useCallback(() => { queryClient.invalidateQueries({ queryKey: ["tools", "connections"], }) + queryClient.invalidateQueries({ + queryKey: ["triggers", "connections"], + }) }, [integrationKey])web/oss/src/styles/globals.css (1)
384-443:⚠️ Potential issue | 🟡 Minor | ⚡ Quick win
webhooks-tableCSS rules appear disconnected from the rendered table.These selectors only apply when the table (or an ancestor) has
webhooks-table, but the providedWebhooks.tsxtable markup doesn’t set that class. The column-width rules likely won’t take effect.Suggested fix outside this file
- <Table + <Table + className="webhooks-table" columns={columns} dataSource={webhooks ?? []} loading={isLoading} rowKey="id"api/oss/src/apis/fastapi/triggers/router.py (1)
111-117:⚠️ Potential issue | 🟠 MajorSupport both
/composio/eventsand/composio/events/to prevent POST redirect failures.FastAPI's default
redirect_slashes=Trueissues HTTP 307 redirects when requests don't match the defined route's trailing-slash format. For POST webhook ingestion, some providers do not properly resend the request body after a 307 redirect, causing lost events.Suggested fix
self.router.add_api_route( "/composio/events/", self.ingest_composio_event, methods=["POST"], operation_id="ingest_composio_event", response_model=TriggerEventAck, status_code=status.HTTP_202_ACCEPTED, ) + self.router.add_api_route( + "/composio/events", + self.ingest_composio_event, + methods=["POST"], + include_in_schema=False, + )
🧹 Nitpick comments (9)
api/ee/tests/pytest/acceptance/triggers/test_triggers_connections.py (1)
29-33: ⚡ Quick winUse the shared API env object instead of
os.getenvhere.This file introduces direct env access at Line 29; in API code/tests we should read config from the shared
envobject to keep config handling centralized.Suggested change
-import os +from oss.src.utils.env import env ... -_COMPOSIO_ENABLED = bool(os.getenv("COMPOSIO_API_KEY")) +_COMPOSIO_ENABLED = bool(env.composio.api_key)As per coding guidelines:
api/**/*.py: “Add new API environment variables toapi/oss/src/utils/env.pyand consume them via the sharedenvobject instead of callingos.getenv(...)directly”.Source: Coding guidelines
web/oss/src/components/Webhooks/WebhookDrawer.tsx (1)
58-67: ⚡ Quick winReuse a shared provider-detection helper instead of duplicating URL heuristics.
This edit path derives provider with inline hostname checks; the same concern is already handled elsewhere via shared helper logic. Consolidating avoids drift between pages.
hosting/docker-compose/ee/docker-compose.gh.local.yml (1)
516-545: 💤 Low valueConsider caching pip packages for faster container restarts.
The
composioservice runspip installon every container start, which adds latency on restarts. For a dev-only tunnel service withrestart: always, this is acceptable but could be optimized by building a custom image or using a volume-cached pip directory.api/oss/src/core/triggers/utils.py (1)
45-68: 💤 Low valueConsider returning cached value on Composio failure instead of None.
When
force_refresh=Falseand the cache is empty, if Composio fails (line 57-59), the method returnsNone. However, if there's a stale/expired cache entry that was evicted, users get no secret. Consider a fallback strategy or ensure callers handleNonegracefully.web/oss/src/services/webhooks/types.ts (1)
22-43: ⚡ Quick winSplit API payload types from draft form-state types.
WebhookFormValuescurrently models payload-style fields (event_types), while form consumers use draft fields (events,header_list), which is why downstream code needsas any. Define a dedicated draft type and use it in form-preview helpers to restore type safety.Proposed direction
interface WebhookFormValuesBase<P extends WebhookProvider = WebhookProvider> { provider: P name?: string event_types?: WebhookEventType[] } +export interface WebhookDraftFields { + events?: WebhookEventType[] + header_list?: {key: string; value: string}[] +} + export interface WebhookConfigFormValues extends WebhookFormValuesBase<"webhook"> { url?: string headers?: Record<string, string> auth_mode?: "signature" | "authorization" auth_value?: string } export interface GitHubFormValues extends WebhookFormValuesBase<"github"> { github_sub_type?: GitHubDispatchType github_repo?: string github_pat?: string github_workflow?: string github_branch?: string } export type WebhookFormValues = WebhookConfigFormValues | GitHubFormValues +export type WebhookDraftFormValues = + | (WebhookConfigFormValues & WebhookDraftFields) + | (GitHubFormValues & WebhookDraftFields)Then type form-preview paths with
WebhookDraftFormValuesinstead ofWebhookFormValues.web/packages/agenta-entities/src/gatewayTrigger/api/client.ts (1)
30-36: ⚡ Quick winTrim this block comment to a one-line “why” note (or remove it).
The current comment narrates behavior that is already clear from the function body, which adds maintenance noise.
As per coding guidelines, “Keep AI-generated in-code comments minimal; comment only the non-obvious why … never the what.”
Source: Coding guidelines
api/oss/src/apis/fastapi/triggers/router.py (1)
136-159: 🏗️ Heavy liftUse
POST /queryfor searchable catalog list endpoints.The new searchable/filterable catalog lists are exposed via GET query params; the router guideline requires query-style filtering via
POST /querypayload endpoints.As per coding guidelines, “
api/oss/src/apis/fastapi/**/router.py: UsePOST /queryfor filtering/search with payload support.”Also applies to: 603-614, 727-736
Source: Coding guidelines
api/entrypoints/dispatcher_composio.py (1)
70-70: 💤 Low valuehttpx.Client is never closed.
The
forwardclient is created but never explicitly closed. Since this script runs indefinitely, it's not a functional issue, but for resource hygiene consider using a context manager or callingforward.close()on exit. However, given this is a dev-only script that blocks forever onwait_forever(), this is acceptable as-is.web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerCatalogDrawer.tsx (1)
426-444: 💤 Low valueEvent cards are hoverable but not actionable.
The event
Cardhashoverableprop but noonClickhandler. If clicking an event should do something (e.g., show event details or select it for subscription), add the handler. If cards are purely informational, consider removinghoverableto avoid misleading UX affordance.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 2c857f25-e2d7-4af9-b561-b325285e1c10
📒 Files selected for processing (116)
api/ee/tests/pytest/acceptance/triggers/test_triggers_connections.pyapi/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/entrypoints/dispatcher_composio.pyapi/entrypoints/routers.pyapi/oss/src/apis/fastapi/tools/router.pyapi/oss/src/apis/fastapi/triggers/models.pyapi/oss/src/apis/fastapi/triggers/router.pyapi/oss/src/core/gateway/catalog/__init__.pyapi/oss/src/core/gateway/catalog/dtos.pyapi/oss/src/core/gateway/catalog/interfaces.pyapi/oss/src/core/gateway/catalog/providers/__init__.pyapi/oss/src/core/gateway/catalog/providers/composio/__init__.pyapi/oss/src/core/gateway/catalog/providers/composio/adapter.pyapi/oss/src/core/gateway/catalog/registry.pyapi/oss/src/core/gateway/catalog/service.pyapi/oss/src/core/gateway/connections/providers/composio/adapter.pyapi/oss/src/core/gateway/providers/__init__.pyapi/oss/src/core/gateway/providers/composio/__init__.pyapi/oss/src/core/gateway/providers/composio/errors.pyapi/oss/src/core/tools/dtos.pyapi/oss/src/core/tools/providers/composio/adapter.pyapi/oss/src/core/tools/service.pyapi/oss/src/core/triggers/dtos.pyapi/oss/src/core/triggers/interfaces.pyapi/oss/src/core/triggers/providers/composio/adapter.pyapi/oss/src/core/triggers/service.pyapi/oss/src/core/triggers/utils.pyapi/oss/src/middlewares/auth.pyapi/oss/src/tasks/asyncio/triggers/dispatcher.pyapi/oss/src/utils/env.pyapi/oss/tests/manual/triggers/try_composio_triggers.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_connections.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_ingress.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/oss/tests/pytest/unit/triggers/test_triggers_dispatcher.pyapi/oss/tests/pytest/unit/triggers/test_triggers_signature.pyhosting/docker-compose/ee/docker-compose.dev.ymlhosting/docker-compose/ee/docker-compose.gh.local.ymlhosting/docker-compose/ee/docker-compose.gh.ymlhosting/docker-compose/oss/docker-compose.dev.ymlhosting/docker-compose/oss/docker-compose.gh.local.ymlhosting/docker-compose/oss/docker-compose.gh.ssl.ymlhosting/docker-compose/oss/docker-compose.gh.ymlhosting/docker-compose/run.shweb/_reference/agenta-sdk/src/types.tsweb/oss/src/components/DrillInView/OSSdrillInUIProvider.tsxweb/oss/src/components/Playground/Components/PlaygroundVariantConfigPrompt/assets/GatewayToolsPanel.tsxweb/oss/src/components/Sidebar/SettingsSidebar.tsxweb/oss/src/components/Webhooks/Modals/DeleteWebhookModal.tsxweb/oss/src/components/Webhooks/Modals/SecretRevealModal.tsxweb/oss/src/components/Webhooks/RequestPreview.tsxweb/oss/src/components/Webhooks/WebhookDrawer.tsxweb/oss/src/components/Webhooks/WebhookFieldRenderer.tsxweb/oss/src/components/Webhooks/WebhookLogsTab.tsxweb/oss/src/components/Webhooks/assets/constants.tsweb/oss/src/components/Webhooks/assets/types.tsweb/oss/src/components/Webhooks/utils/buildPreviewRequest.tsweb/oss/src/components/Webhooks/utils/buildSubscription.tsweb/oss/src/components/Webhooks/utils/handleTestResult.tsweb/oss/src/components/Webhooks/widgets/AdvanceConfigWidget.tsxweb/oss/src/components/Webhooks/widgets/DispatchAlertWidget.tsxweb/oss/src/components/Webhooks/widgets/HeaderListWidget.tsxweb/oss/src/components/pages/settings/APIKeys/APIKeys.tsxweb/oss/src/components/pages/settings/Tools/components/GatewayToolsSection.tsxweb/oss/src/components/pages/settings/Tools/hooks/useIntegrationDetail.tsweb/oss/src/components/pages/settings/Tools/hooks/useToolsConnections.tsweb/oss/src/components/pages/settings/Tools/hooks/useToolsIntegrations.tsweb/oss/src/components/pages/settings/Triggers/components/GatewaySubscriptionsSection.tsxweb/oss/src/components/pages/settings/Triggers/components/GatewayTriggersSection.tsxweb/oss/src/components/pages/settings/Webhooks/Webhooks.tsxweb/oss/src/pages/w/[workspace_id]/p/[project_id]/settings/index.tsxweb/oss/src/services/webhooks/api.tsweb/oss/src/services/webhooks/types.tsweb/oss/src/state/automations/state.tsweb/oss/src/state/webhooks/atoms.tsweb/oss/src/state/webhooks/state.tsweb/oss/src/styles/globals.cssweb/packages/agenta-entities/src/gatewayTool/api/api.tsweb/packages/agenta-entities/src/gatewayTool/api/index.tsweb/packages/agenta-entities/src/gatewayTool/hooks/index.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolActionDetail.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolCatalogActions.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolCatalogIntegrations.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolConnectionActions.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolConnectionQuery.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolConnectionsQuery.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolIntegrationConnections.tsweb/packages/agenta-entities/src/gatewayTool/hooks/useToolIntegrationDetail.tsweb/packages/agenta-entities/src/gatewayTool/index.tsweb/packages/agenta-entities/src/gatewayTool/state/atoms.tsweb/packages/agenta-entities/src/gatewayTool/state/index.tsweb/packages/agenta-entities/src/gatewayTrigger/api/api.tsweb/packages/agenta-entities/src/gatewayTrigger/api/client.tsweb/packages/agenta-entities/src/gatewayTrigger/api/index.tsweb/packages/agenta-entities/src/gatewayTrigger/core/types.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/index.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerCatalogEvents.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerCatalogIntegrations.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerConnectionActions.tsweb/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerEvent.tsweb/packages/agenta-entities/src/gatewayTrigger/index.tsweb/packages/agenta-entities/src/gatewayTrigger/state/atoms.tsweb/packages/agenta-entities/src/gatewayTrigger/state/index.tsweb/packages/agenta-entities/src/index.tsweb/packages/agenta-entity-ui/src/gatewayTool/components/SchemaForm.tsxweb/packages/agenta-entity-ui/src/gatewayTool/drawers/CatalogDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTool/drawers/ConnectDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTool/drawers/ConnectionManagerDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTool/drawers/ToolExecutionDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerCatalogDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerConnectDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerDeliveriesDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerEventsDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerSubscriptionDrawer.tsxweb/packages/agenta-entity-ui/src/gatewayTrigger/index.tsweb/tests/tests/fixtures/base.fixture/providerHelpers/index.ts
💤 Files with no reviewable changes (1)
- web/oss/src/state/automations/state.ts
✅ Files skipped from review due to trivial changes (5)
- api/oss/src/core/gateway/catalog/providers/composio/init.py
- web/packages/agenta-entity-ui/src/gatewayTool/components/SchemaForm.tsx
- web/_reference/agenta-sdk/src/types.ts
- web/packages/agenta-entities/src/index.ts
- web/oss/src/components/pages/settings/APIKeys/APIKeys.tsx
🚧 Files skipped from review as they are similar to previous changes (11)
- api/oss/src/middlewares/auth.py
- web/packages/agenta-entities/src/gatewayTrigger/state/atoms.ts
- web/packages/agenta-entity-ui/src/gatewayTrigger/drawers/TriggerDeliveriesDrawer.tsx
- api/oss/src/core/triggers/interfaces.py
- web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerEvent.ts
- web/oss/src/components/pages/settings/Triggers/components/GatewaySubscriptionsSection.tsx
- api/oss/tests/pytest/acceptance/triggers/test_triggers_subscriptions.py
- api/entrypoints/routers.py
- api/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.py
- web/oss/src/components/Sidebar/SettingsSidebar.tsx
- api/oss/src/tasks/asyncio/triggers/dispatcher.py
| def test_create_revoke_roundtrip(self, connections_api): | ||
| slug = f"acc-{uuid4().hex[:8]}" | ||
| create = connections_api( | ||
| "POST", | ||
| "/triggers/connections/", | ||
| json={ | ||
| "connection": { | ||
| "slug": slug, | ||
| "provider_key": "composio", | ||
| "integration_key": "github", | ||
| "data": {"auth_scheme": "oauth"}, | ||
| } | ||
| }, | ||
| ) | ||
| assert create.status_code == 200, create.text | ||
| connection_id = create.json()["connection"]["id"] | ||
|
|
||
| revoke = connections_api( | ||
| "POST", f"/triggers/connections/{connection_id}/revoke" | ||
| ) | ||
| assert revoke.status_code == 200, revoke.text | ||
| assert revoke.json()["connection"]["flags"]["is_valid"] is False | ||
|
|
||
| delete = connections_api("DELETE", f"/triggers/connections/{connection_id}") | ||
| assert delete.status_code == 204, delete.text | ||
|
|
There was a problem hiding this comment.
Always clean up created connections with finally in lifecycle tests.
At Line 137/170 and Line 200, resources are created but deletion only happens on the success path. If an assertion fails before delete, the test leaks provider-backed state and can make later runs flaky.
Suggested pattern
- create = connections_api(...)
- assert create.status_code == 200, create.text
- connection_id = create.json()["connection"]["id"]
- ...
- delete = connections_api("DELETE", f"/triggers/connections/{connection_id}")
- assert delete.status_code == 204, delete.text
+ connection_id = None
+ try:
+ create = connections_api(...)
+ assert create.status_code == 200, create.text
+ connection_id = create.json()["connection"]["id"]
+ ...
+ finally:
+ if connection_id:
+ connections_api("DELETE", f"/triggers/connections/{connection_id}")Also applies to: 167-227
| async def list_integrations( | ||
| self, | ||
| *, | ||
| search: Optional[str] = None, | ||
| sort_by: Optional[str] = None, | ||
| limit: Optional[int] = None, | ||
| cursor: Optional[str] = None, | ||
| ) -> Tuple[List[CatalogIntegration], Optional[str], int]: | ||
| """Returns (items, next_cursor, total_items).""" |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | 🏗️ Heavy lift
Replace tuple pagination output with a named DTO contract.
Returning (items, next_cursor, total_items) as a tuple makes the contract positional and brittle across adapters/callers. Use a typed DTO (e.g., CatalogIntegrationsPage) for explicit fields and safer cross-layer evolution.
As per coding guidelines, "Do NOT return tuples like (data, trace_id) from service or client methods; use a named DTO instead."
Source: Coding guidelines
| adapter = self._adapters.get(provider_key) | ||
| if not adapter: | ||
| raise ProviderNotFoundError(provider_key) |
There was a problem hiding this comment.
Use explicit missing-key detection in registry lookup.
if not adapter can misclassify a valid adapter as missing if it is falsy; check for None explicitly.
Proposed fix
def get(self, provider_key: str) -> CatalogGatewayInterface:
adapter = self._adapters.get(provider_key)
- if not adapter:
+ if adapter is None:
raise ProviderNotFoundError(provider_key)
return adapter| async def list_integrations( | ||
| self, | ||
| *, | ||
| provider_key: str, | ||
| # | ||
| search: Optional[str] = None, | ||
| sort_by: Optional[str] = None, | ||
| limit: Optional[int] = None, | ||
| cursor: Optional[str] = None, | ||
| ) -> Tuple[List[CatalogIntegration], Optional[str], int]: |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | 🏗️ Heavy lift
Return a typed DTO from list_integrations instead of a tuple.
Core service methods should expose named DTO contracts; positional tuples are brittle at call sites and violate the service contract guideline.
As per coding guidelines, “Service methods must return typed DTOs … Do NOT return tuples like (data, trace_id) from service or client methods.”
Source: Coding guidelines
| existing = await self._get("/webhook_subscriptions") | ||
| items = existing.get("items", []) if isinstance(existing, dict) else [] | ||
| if items: | ||
| return items[0]["secret"] | ||
|
|
There was a problem hiding this comment.
Handle empty/malformed webhook subscription payloads before indexing.
items[0]["secret"] can raise IndexError/KeyError in both the initial read and 409 retry path, and those errors bypass the current httpx.HTTPError wrapper.
Suggested hardening
- if items:
- return items[0]["secret"]
+ if items and isinstance(items[0], dict) and items[0].get("secret"):
+ return items[0]["secret"]
@@
- if resp.status_code == 409:
- again = await self._get("/webhook_subscriptions")
- return again["items"][0]["secret"]
+ if resp.status_code == 409:
+ again = await self._get("/webhook_subscriptions")
+ again_items = again.get("items", []) if isinstance(again, dict) else []
+ if again_items and isinstance(again_items[0], dict) and again_items[0].get("secret"):
+ return again_items[0]["secret"]
+ raise AdapterError(
+ provider_key="composio",
+ operation="ensure_webhook_subscription",
+ detail="Webhook subscription conflict, but no readable secret was returned",
+ )Also applies to: 149-153
| _COMPOSIO_ENABLED = bool(os.getenv("COMPOSIO_API_KEY")) | ||
| _requires_composio = pytest.mark.skipif( | ||
| not _COMPOSIO_ENABLED, | ||
| reason="needs live Composio credentials (COMPOSIO_API_KEY)", | ||
| ) |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
Use the shared API env object instead of os.getenv in this test module.
Line 20 reads COMPOSIO_API_KEY directly via os.getenv(...); this should come from the shared env object for consistency with API config handling.
As per coding guidelines, "api/**/*.py: Add new API environment variables to api/oss/src/utils/env.py and consume them via the shared env object instead of calling os.getenv(...) directly."
Source: Coding guidelines
| def test_create_revoke_roundtrip(self, authed_api): | ||
| slug = f"acc-{uuid4().hex[:8]}" | ||
| create = authed_api( | ||
| "POST", | ||
| "/triggers/connections/", | ||
| json={ | ||
| "connection": { | ||
| "slug": slug, | ||
| "provider_key": "composio", | ||
| "integration_key": "github", | ||
| "data": {"auth_scheme": "oauth"}, | ||
| } | ||
| }, | ||
| ) | ||
| assert create.status_code == 200, create.text | ||
| connection_id = create.json()["connection"]["id"] | ||
|
|
||
| revoke = authed_api("POST", f"/triggers/connections/{connection_id}/revoke") | ||
| assert revoke.status_code == 200, revoke.text | ||
| assert revoke.json()["connection"]["flags"]["is_valid"] is False | ||
|
|
||
| delete = authed_api("DELETE", f"/triggers/connections/{connection_id}") | ||
| assert delete.status_code == 204, delete.text | ||
|
|
There was a problem hiding this comment.
Ensure created live connections are cleaned up even when assertions fail.
Lines 48-71, 78-110, and 111-140 only delete the connection on the happy path. If an assertion fails first, the provider resource is left behind and can make subsequent acceptance runs flaky. Wrap each create/use flow in try/finally and best-effort delete when connection_id is set.
Also applies to: 78-110, 111-140
| @_requires_connected_account | ||
| class TestTriggerIngressDedup: |
There was a problem hiding this comment.
Strengthen dedup test preconditions before signing events.
Line 112 gates only on COMPOSIO_TEST_CONNECTED_ACCOUNT, and Line 161 can return an empty secret. That can cause 401 failures unrelated to dedup behavior and make this acceptance test flaky.
Suggested patch
`@_requires_connected_account`
+@_requires_composio
class TestTriggerIngressDedup:
def test_duplicate_event_id_writes_single_delivery(self, authed_api, unauthed_api):
@@
timestamp = "1700000000"
secret = _resolve_webhook_secret()
+ assert secret, "Composio webhook secret is required for signed ingress tests"
headers = {
"Content-Type": "application/json",
"webhook-id": event_id,
"webhook-timestamp": timestamp,
"webhook-signature": _sign(secret, event_id, timestamp, body),
}Also applies to: 161-167
| image: python:3.13-slim-trixie | ||
| # === EXECUTION ============================================ # | ||
| command: | ||
| - bash | ||
| - -c | ||
| - "pip install --quiet --root-user-action=ignore composio httpx && python /app/dispatcher_composio.py" | ||
| # === STORAGE ============================================== # |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n hosting/docker-compose/oss/docker-compose.gh.yml | sed -n '540,560p'Repository: Agenta-AI/agenta
Length of output: 1120
🏁 Script executed:
# Check for profile definitions and tunnel service setup
rg "profiles:|composio|tunnel" hosting/docker-compose/oss/docker-compose.gh.yml -A 3 -B 1Repository: Agenta-AI/agenta
Length of output: 1740
🏁 Script executed:
# Check the full service definition for the tunnel/composio service
ast-grep --pattern 'composio-tunnel:' hosting/docker-compose/oss/docker-compose.gh.yml || rg "composio-tunnel:" -A 20 hosting/docker-compose/Repository: Agenta-AI/agenta
Length of output: 42
🏁 Script executed:
# Check if with-tunnel is enabled by default in run.sh
rg "with-tunnel|profiles" hosting/docker-compose/run.sh -B 2 -A 2Repository: Agenta-AI/agenta
Length of output: 361
🏁 Script executed:
# Check other docker-compose files for similar patterns or prebuilt images
fd "docker-compose" hosting/docker-compose/ -type f | head -10Repository: Agenta-AI/agenta
Length of output: 231
🏁 Script executed:
# Check if there's a Dockerfile for composio or other services
fd "Dockerfile" hosting/docker-compose/ -type fRepository: Agenta-AI/agenta
Length of output: 231
🏁 Script executed:
# Fix the fd command - check for docker-compose files
fd "docker-compose" hosting/docker-compose/ -t f | head -10Repository: Agenta-AI/agenta
Length of output: 469
🏁 Script executed:
# Check for Dockerfiles
fd "Dockerfile" hosting/ -t fRepository: Agenta-AI/agenta
Length of output: 509
🏁 Script executed:
# Check run.sh to see what WITH_TUNNEL default is
rg "WITH_TUNNEL" hosting/docker-compose/run.sh -B 3 -A 1Repository: Agenta-AI/agenta
Length of output: 542
🏁 Script executed:
# Check if there's a base image or Dockerfile for the composio service
rg "composio" hosting/docker-compose/ -lRepository: Agenta-AI/agenta
Length of output: 600
🏁 Script executed:
# Check if EE version has the same composio service
cat hosting/docker-compose/ee/docker-compose.gh.yml | grep -A 15 "composio:"Repository: Agenta-AI/agenta
Length of output: 808
🏁 Script executed:
# Search for other runtime pip install patterns in docker-compose files
rg "pip install" hosting/docker-compose/ -B 2 -A 1Repository: Agenta-AI/agenta
Length of output: 3064
🏁 Script executed:
# Check dispatcher_composio.py to understand dependencies
head -30 api/entrypoints/dispatcher_composio.pyRepository: Agenta-AI/agenta
Length of output: 1118
🏁 Script executed:
# Check if there's a requirements file for composio
fd "requirements.*composio" hosting/Repository: Agenta-AI/agenta
Length of output: 42
Avoid unpinned runtime pip install in the composio service across all docker-compose files.
Installing composio and httpx at startup makes the stack non-deterministic and brittle (network failures, PyPI index issues, or upstream releases can break runs). With the tunnel enabled by default, this impacts every normal run.sh invocation. This pattern exists across all OSS and EE docker-compose files.
Prefer a prebuilt image or pinned versions (e.g., composio==0.X.X httpx==0.Y.Y) to ensure reproducible, deterministic startup.
| useEffect(() => { | ||
| if (loadedPages < targetPages && query.hasNextPage && !query.isFetchingNextPage) { | ||
| query.fetchNextPage() | ||
| } | ||
| }, [loadedPages, targetPages, query.hasNextPage, query.isFetchingNextPage, query.fetchNextPage]) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cd /repo && find . -name "useTriggerCatalogIntegrations.ts" -o -name "useTriggerCatalogEvents.ts"Repository: Agenta-AI/agenta
Length of output: 116
🏁 Script executed:
cd /repo && head -100 web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerCatalogIntegrations.tsRepository: Agenta-AI/agenta
Length of output: 116
🏁 Script executed:
cd /repo && head -100 web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerCatalogEvents.tsRepository: Agenta-AI/agenta
Length of output: 116
🏁 Script executed:
find . -name "useTriggerCatalogIntegrations.ts" -o -name "useTriggerCatalogEvents.ts"Repository: Agenta-AI/agenta
Length of output: 232
🏁 Script executed:
cat web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerCatalogIntegrations.tsRepository: Agenta-AI/agenta
Length of output: 2986
🏁 Script executed:
cat web/packages/agenta-entities/src/gatewayTrigger/hooks/useTriggerCatalogEvents.tsRepository: Agenta-AI/agenta
Length of output: 3062
🌐 Web query:
TanStack Query useInfiniteQuery isError property API reference
💡 Result:
In TanStack Query, the useInfiniteQuery hook inherits the returned properties of the standard useQuery hook [1][2][3]. The isError property is a boolean that is true if the query has encountered an error [4]. When using useInfiniteQuery, you can access isError to determine if the query is in an error state. Because it shares the same base return type as useQuery, the standard pattern for error handling applies: const { isError, error } = useInfiniteQuery({... }); if (isError) { return
Citations:
- 1: https://tanstack.com/query/v5/docs/framework/react/reference/useInfiniteQuery
- 2: https://tanstack.com/query/latest/docs/framework/react/reference/useInfiniteQuery?from=reactQueryV3
- 3: https://tanstack.com/query/v4/docs/framework/react/reference/useInfiniteQuery
- 4: https://tanstack-query.mintlify.app/api/react/use-infinite-query
🏁 Script executed:
rg "query\.isError" web/packages/ --type ts --type tsx -A 2 -B 2Repository: Agenta-AI/agenta
Length of output: 88
🏁 Script executed:
rg "isError" web/packages/agenta-entities/ --type ts --type tsx -cRepository: Agenta-AI/agenta
Length of output: 88
🏁 Script executed:
rg "isError" web/packages/agenta-entities/ -A 2 -B 2Repository: Agenta-AI/agenta
Length of output: 40122
🏁 Script executed:
rg "isFetchNextPageError" web/packages/agenta-entities/ -A 2 -B 2Repository: Agenta-AI/agenta
Length of output: 42
🏁 Script executed:
rg "isFetchNextPageError|isFetchPreviousPageError" web/packages/agenta-entities/Repository: Agenta-AI/agenta
Length of output: 42
Guard auto-prefetch on error to avoid repeated failing fetch cycles.
When fetchNextPage() errors, this effect can keep re-triggering while loadedPages < targetPages, causing repeated failing requests. Add query.isError to prevent retries on error state.
Suggested fix
useEffect(() => {
- if (loadedPages < targetPages && query.hasNextPage && !query.isFetchingNextPage) {
- query.fetchNextPage()
+ if (
+ loadedPages >= targetPages ||
+ !query.hasNextPage ||
+ query.isFetchingNextPage ||
+ query.isError
+ ) {
+ return
}
+ void query.fetchNextPage()
}, [loadedPages, targetPages, query.hasNextPage, query.isFetchingNextPage, query.fetchNextPage])Apply the same guard to useTriggerCatalogEvents.
Resolve the bound reference via the canonical WorkflowsService.retrieve_workflow_revision (handles application/evaluator/ workflow + environment families) and rebuild the completed family with build_retrieval_info, so invoke_workflow finds the service uri. Raise TriggerReferenceInvalid when it cannot resolve. Skip soft-deleted subscriptions in the ti_* resolver. FE: scope the picker to application workflows and send the reference family by its true kind. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
3 participants
Context
The inbound dual of webhooks. Webhooks turn Agenta events into outbound HTTP calls; this turns inbound external provider events (via Composio) into Agenta workflow runs. Nothing in the platform previously turned an external event into a workflow invocation.
This PR collapses the former WP0–WP6 stack (+ design docs) into a single change.
What this adds
core/gateway/connections+dbs/postgres/gateway/connections, a routerless domain backed by the renamedgateway_connectionstable, reused by both tools and triggers (OAuth initiate/callback/refresh/revoke).core/triggers,dbs/postgres/triggers,apis/fastapi/triggers).POST /triggers/composio/eventsendpoint: HMAC-SHA256 signature verification, 202 ack-fast, enqueue to a TaskIQ worker that resolves the subscription, dedups onevent_id, maps inputs, and invokes the bound workflow, recording one delivery row.web/oss+@agentapackages).worker-triggersmounted in all compose stacks.docs/designs/gateway-triggers/.Notes
ProviderNotFoundErrormaps to 404 at the router boundary;connection_dataand catalog list fields useField(default_factory=...).What to QA
🤖 Generated with Claude Code