[feat] Keep automation session history linked - #5929
Conversation
dosubot
Bot
added
the
size:XXL
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe PR updates session query contracts, pagination, stream attribution, trigger expansion, trigger soft deletion, atomic trigger-session claims, and related worker, router, persistence, and test paths. ChangesSession query and stream handling
Trigger lifecycle and dispatch
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant SessionRouter
participant SessionsService
participant SessionStreamsDAO
participant TriggerConfigurations
SessionRouter->>SessionsService: submit normalized session query
SessionsService->>SessionStreamsDAO: query sessions and optional expansions
SessionStreamsDAO->>TriggerConfigurations: resolve trigger details
TriggerConfigurations-->>SessionStreamsDAO: return trigger names
SessionStreamsDAO-->>SessionsService: return paged session streams
SessionsService-->>SessionRouter: return sanitized page and windowing
sequenceDiagram
participant TriggerWorker
participant TriggersDispatcher
participant SessionStreamsDAO
participant WorkflowsService
TriggerWorker->>TriggersDispatcher: dispatch resolved trigger
TriggersDispatcher->>SessionStreamsDAO: claim attributed delivery session
SessionStreamsDAO-->>TriggersDispatcher: return claim result
TriggersDispatcher->>WorkflowsService: invoke workflow
WorkflowsService-->>TriggersDispatcher: return completion or failure
TriggersDispatcher->>SessionStreamsDAO: update delivery and abandon session if needed
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 2📝 Generate docstrings 💡
🛠️ Fix failing CI checks 💡
🧪 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 |
| ) | ||
|
|
||
| claimed = await self.triggers_dao.claim_delivery( | ||
| claimed = await self.session_claims_dao.claim_trigger_delivery( |
There was a problem hiding this comment.
Review focus: this claim is the only authorization boundary for invocation. The DAO must keep the delivery insert and session attribution in one transaction, and a duplicate claim must return false before mapping or workflow invocation. The real-Postgres duplicate and rollback tests cover that invariant.
Signed: OpenCode
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
api/oss/src/apis/fastapi/triggers/router.py (1)
1092-1101: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winAlign the EE acceptance test with historical trigger reads
The shared route now returns
200for soft-deleted subscriptions, butapi/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pystill asserts404after deletion. Update the EE assertion. List and query routes can continue to omit deleted records. The generated Python SDK does not use404as a deletion signal.
🧹 Nitpick comments (14)
api/oss/src/tasks/asyncio/triggers/dispatcher.py (1)
418-441: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider logging when the delivery write is skipped.
write_subscription_delivery_if_livereturnsOptional[TriggerDelivery]and returnsNonewhen the parent subscription is no longer live or active. This helper discards that value. The test-capture path at Line 137 and the invalid-subscription path at Line 160 then return with no record and no log entry. A user who polls for a test delivery sees only a timeout, with nothing in the logs to explain it.Add a debug or info log when the DAO returns
None.♻️ Proposed change to record the skipped write
- await self.triggers_dao.write_subscription_delivery_if_live( + written = await self.triggers_dao.write_subscription_delivery_if_live( project_id=project_id, user_id=user_id, delivery=TriggerDeliveryCreate( id=delivery_id, subscription_id=subscription_id, schedule_id=schedule_id, event_id=event_id, status=status, data=data, ), ) + if written is None: + log.info( + "[TRIGGERS DISPATCHER] delivery write skipped — subscription %s " + "is no longer live/active (event=%s)", + subscription_id, + event_id, + )api/oss/src/tasks/taskiq/triggers/worker.py (1)
101-118: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueConsider guarding the UUID conversions.
Line 108 and Line 111 call
UUID(...)without error handling. A malformedproject_idor a malformedidinside a legacyschedulepayload raisesValueError. The task is registered withretry_on_error=True, so taskiq retries the messageTRIGGER_MAX_RETRIEStimes before it fails permanently. The failure is deterministic, so the retries cannot succeed.Lines 102-106 already log and skip when the id is missing. Handle a malformed id the same way.
♻️ Proposed change to skip malformed identifiers
- resolved_project_id = UUID(project_id) + try: + resolved_project_id = UUID(project_id) + resolved_schedule_id = UUID(str(queued_schedule_id)) + except ValueError: + log.warning( + "[TASK] triggers.dispatch_schedule Malformed identifiers " + "project=%s schedule=%s — skipping", + project_id, + queued_schedule_id, + ) + return + entity = await self.triggers_dao.fetch_schedule( project_id=resolved_project_id, - schedule_id=UUID(str(queued_schedule_id)), + schedule_id=resolved_schedule_id, )api/oss/tests/pytest/unit/triggers/test_triggers_worker_lifecycle.py (1)
41-116: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider covering the missing-identifier branch.
The tests cover the new
schedule_idpayload, the legacyschedulepayload, a deleted schedule, an inactive schedule, and an unknown subscription. No test callsdispatch_schedulewith neitherschedule_idnorschedule. That branch atapi/oss/src/tasks/taskiq/triggers/worker.pyLines 101-106 logs and returns without touching the DAO.Add one test that asserts
fetch_scheduleis not awaited in that case.♻️ Proposed test to add
+async def test_schedule_task_skips_payload_without_any_identifier(): + worker, dao, dispatcher = _worker(resolved=_schedule()) + + await worker.dispatch_schedule( + project_id=str(uuid4()), + event_id="event-1", + event={}, + ) + + dao.fetch_schedule.assert_not_awaited() + dispatcher.dispatch_schedule.assert_not_awaited()api/oss/tests/pytest/unit/triggers/conftest.py (1)
10-25: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winIncrease the PostgreSQL probe timeout or fail explicitly in CI.
_postgres_reachable()caches the first result and uses a 0.5-second timeout. A slow initial connection can skip all marked trigger integration tests for the process.api/pytest.inialready registers theintegrationmarker.api/oss/src/dbs/postgres/triggers/dao.py (2)
326-341: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueLog the ambiguous-match case.
The lookup now returns
Nonewhen two live subscriptions sharetrigger_id. Inbound provider events are then dropped with no trace. Add a warning log before returningNoneso operators can detect the duplicate rows.♻️ Proposed change
result = await session.execute(stmt) subscription_dbes = result.scalars().all() if len(subscription_dbes) != 1: + if len(subscription_dbes) > 1: + log.warning( + "[TRIGGERS DAO] Ambiguous trigger_id across projects — failing closed", + trigger_id=trigger_id, + ) return None
414-480: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider extracting the shared delivery-upsert builder.
write_delivery,write_subscription_delivery_if_live, andSessionStreamsDAO.claim_trigger_deliveryeach rebuild the same column-values dict,index_elements, andindex_where. Extract one helper so the conflict target and the skipped-column rules cannot drift.api/oss/src/core/triggers/service.py (2)
1110-1124: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueLog the skipped provider cleanup.
If
connectionisNone, the method returns without deleting the provider trigger. The local row is then deleted while the providerti_*stays live, and nothing records this. Add a warning log in that branch.
1126-1147: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winGuard the purge on
flags.is_test.
cleanup_test_subscriptionphysically deletes a subscription and its delivery history. The method is public and does not check that the subscription is a test row. A caller that passes a productionsubscription_iddestroys real history with no recovery path. Reject non-test subscriptions before the purge.🛡️ Proposed guard
if existing is None: return False + if not existing.flags.is_test: + raise ValueError( + "cleanup_test_subscription only purges test subscriptions." + ) + await self._delete_provider_subscription(api/oss/src/core/sessions/dtos.py (1)
9-15: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMark the re-exports so Flake8 stops reporting F401.
Flake8 reports lines 9, 11, 12, and 15 as unused imports. The
import X as Xform is honored by mypy, but pyflakes only honors it inside__init__.py. The symbols are real re-exports;api/oss/tests/pytest/unit/sessions/test_sessions_query_contract.pyline 30 importsSessionTriggerKindfrom this module. Add an__all__entry list or a# noqa: F401marker so the lint signal stays clean.♻️ Proposed change
from oss.src.core.sessions.types import SessionTriggerKind as SessionTriggerKind + +__all__ = [ + "SessionDelivery", + "SessionOrigin", + "SessionTrigger", + "SessionTriggerAttribution", + "SessionTriggerKind", + "SessionExpansion", + "SessionListItem", + "SessionQuery", + "SessionQueryLifecycle", + "SessionQueryOptions", + "SessionQueryPage", +]Source: Linters/SAST tools
api/oss/src/core/sessions/service.py (2)
157-168: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBuild the stream query call with explicit keyword arguments.
stream_kwargs = dict(...)followed by a conditional key insert removes static type checking on thequery_streamscall. Passread_optionsdirectly so a signature change is caught by the type checker.♻️ Proposed change
- stream_kwargs = dict( - project_id=project_id, - filter=_stream_filter(query, lifecycle), - windowing=windowing, - session_ids=session_ids, - exclude_session_ids=query.exclude_session_ids if query else None, - ) - if SessionExpansion.trigger in options.expand: - stream_kwargs["read_options"] = SessionStreamReadOptions( - include_trigger_details=True - ) - streams = await self.streams_service.query_streams(**stream_kwargs) + streams = await self.streams_service.query_streams( + project_id=project_id, + filter=_stream_filter(query, lifecycle), + windowing=windowing, + session_ids=session_ids, + exclude_session_ids=query.exclude_session_ids if query else None, + read_options=SessionStreamReadOptions( + include_trigger_details=SessionExpansion.trigger in options.expand + ), + )
217-227: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winLog the traceback when the preview lookup fails.
The handler swallows every
Exceptionand logs one line without the exception. A repeated failure of the records engine then looks like sessions that simply have no messages. Attach the exception so the cause is visible.♻️ Proposed change
- except Exception: - log.warning( - "[SESSIONS] latest-message lookup failed", - project_id=str(project_id), - ) - return {} + except Exception as exception: + log.warning( + "[SESSIONS] latest-message lookup failed", + project_id=str(project_id), + exc_info=exception, + ) + return {}api/oss/src/dbs/postgres/sessions/streams/dao.py (1)
286-295: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoffConsider expression indexes for the tag-based predicates.
The origin filter and the trigger join both read JSONB text expressions:
tags->>'ag.origin',tags->>'ag.trigger.id', andtags->>'ag.trigger.kind'. PostgreSQL cannot use a plain btree index for these expressions. Each filtered session list therefore scans the project's stream rows and evaluates the expressions per row.The PR states that no index is included. The queries are scoped by
project_id, so this is acceptable at current per-project sizes. If session counts per project grow, add expression indexes, for example on(project_id, (tags->>'ag.origin')).Also applies to: 389-424
api/oss/tests/pytest/unit/sessions/test_session_trigger_expansion.py (1)
43-61: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueTwo assertions cannot fail.
Line 53 asserts
"schedule" in sqland line 54 asserts"subscription" in sql. Lines 46 and 47 already assert thattrigger_schedulesandtrigger_subscriptionsappear in the SQL, and those table names contain both substrings. The two assertions therefore always pass and give no signal.Assert the compared literal in the
CASEpredicate instead.♻️ Proposed refactor for the kind assertions
assert "ag.trigger.kind" in sql - assert "schedule" in sql - assert "subscription" in sql + assert "= 'schedule'" in sql + assert "= 'subscription'" in sql assert "ag.trigger.id" in sqlapi/oss/src/dbs/postgres/sessions/streams/mappings.py (1)
23-37: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCentralize the reserved session tag keys. The sanitizer duplicates the four keys from
mappings.pyand addsag.trigger.name. Move the keys to one shared definition and use it in both layers.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: CHILL
Plan: Pro Plus
Run ID: 2ae3f07f-7d83-47e2-b5f0-584724284749
📒 Files selected for processing (46)
api/entrypoints/routers.pyapi/entrypoints/worker_queues.pyapi/oss/src/apis/fastapi/sessions/models.pyapi/oss/src/apis/fastapi/sessions/router.pyapi/oss/src/apis/fastapi/sessions/utils.pyapi/oss/src/apis/fastapi/triggers/router.pyapi/oss/src/core/sessions/dtos.pyapi/oss/src/core/sessions/records/dtos.pyapi/oss/src/core/sessions/service.pyapi/oss/src/core/sessions/streams/dtos.pyapi/oss/src/core/sessions/streams/interfaces.pyapi/oss/src/core/sessions/streams/service.pyapi/oss/src/core/sessions/types.pyapi/oss/src/core/triggers/interfaces.pyapi/oss/src/core/triggers/providers/composio/adapter.pyapi/oss/src/core/triggers/service.pyapi/oss/src/dbs/postgres/sessions/records/dao.pyapi/oss/src/dbs/postgres/sessions/streams/dao.pyapi/oss/src/dbs/postgres/sessions/streams/mappings.pyapi/oss/src/dbs/postgres/triggers/dao.pyapi/oss/src/tasks/asyncio/triggers/dispatcher.pyapi/oss/src/tasks/taskiq/triggers/worker.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_schedules.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/oss/tests/pytest/unit/sessions/test_query_sessions_filters.pyapi/oss/tests/pytest/unit/sessions/test_query_sessions_windowing.pyapi/oss/tests/pytest/unit/sessions/test_session_last_message.pyapi/oss/tests/pytest/unit/sessions/test_session_response_sanitization.pyapi/oss/tests/pytest/unit/sessions/test_session_trigger_expansion.pyapi/oss/tests/pytest/unit/sessions/test_session_trigger_stamp.pyapi/oss/tests/pytest/unit/sessions/test_sessions_query_contract.pyapi/oss/tests/pytest/unit/sessions/test_sessions_root_service.pyapi/oss/tests/pytest/unit/sessions/test_trigger_session_claim_postgres.pyapi/oss/tests/pytest/unit/sessions/test_wp5_dao_fanout.pyapi/oss/tests/pytest/unit/sessions/test_wp5_root_router.pyapi/oss/tests/pytest/unit/triggers/conftest.pyapi/oss/tests/pytest/unit/triggers/test_triggers_composio_adapter.pyapi/oss/tests/pytest/unit/triggers/test_triggers_dao_lifecycle.pyapi/oss/tests/pytest/unit/triggers/test_triggers_dao_update_delivery.pyapi/oss/tests/pytest/unit/triggers/test_triggers_dispatcher.pyapi/oss/tests/pytest/unit/triggers/test_triggers_lifecycle.pyapi/oss/tests/pytest/unit/triggers/test_triggers_schedules_refresh.pyapi/oss/tests/pytest/unit/triggers/test_triggers_soft_delete_postgres.pyapi/oss/tests/pytest/unit/triggers/test_triggers_subscription_test_mode.pyapi/oss/tests/pytest/unit/triggers/test_triggers_worker_lifecycle.pyapi/oss/tests/pytest/unit/workflows/test_invoke_detached.py
💤 Files with no reviewable changes (2)
- api/oss/src/core/sessions/records/dtos.py
- api/oss/src/dbs/postgres/sessions/records/dao.py
| @pytest_asyncio.fixture(autouse=True) | ||
| async def _fresh_engine_per_test(): | ||
| engine_module._transactions_engine = None | ||
| yield | ||
| if engine_module._transactions_engine is not None: | ||
| await engine_module._transactions_engine.close() | ||
| engine_module._transactions_engine = None |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Find modules and fixtures that cache the transactions engine.
set -euo pipefail
rg -nP --type=py -C4 '_transactions_engine|get_transactions_engine\s*\('Repository: Agenta-AI/agenta
Length of output: 154
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file='api/oss/tests/pytest/unit/sessions/test_trigger_session_claim_postgres.py'
printf '%s\n' '--- target file ---'
sed -n '1,180p' "$file"
printf '%s\n' '--- related engine symbols ---'
rg -n -P -g '*.py' -C3 '_transactions_engine|get_transactions_engine\s*\(' api || trueRepository: Agenta-AI/agenta
Length of output: 50373
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- engine implementation ---'
engine_file="$(fd -t f -i '^engine\.py$' api/oss/src api/ee/src | head -n1)"
printf 'file: %s\n' "$engine_file"
sed -n '1,240p' "$engine_file"
printf '%s\n' '--- target test imports and fixture usage ---'
rg -n -P '^(import|from) |^`@pytest`|^async def test_|get_transactions_engine|engine_module\._transactions_engine' \
api/oss/tests/pytest/unit/sessions/test_trigger_session_claim_postgres.py
printf '%s\n' '--- conftest files in the relevant test tree ---'
fd -t f 'conftest\.py$' api/oss/tests/pytest | sortRepository: Agenta-AI/agenta
Length of output: 5886
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- pytest setup imports ---'
for file in \
api/oss/tests/pytest/conftest.py \
api/oss/tests/pytest/unit/conftest.py \
api/oss/tests/pytest/unit/sessions/conftest.py \
api/oss/tests/pytest/unit/triggers/conftest.py
do
if [ -f "$file" ]; then
printf '\n--- %s ---\n' "$file"
sed -n '1,220p' "$file"
fi
done
printf '%s\n' '--- routers imports and engine holder definitions ---'
rg -n -P -C3 'entrypoints\.routers|from .*routers|import .*routers|_transactions_engine|get_transactions_engine' \
api/oss/tests/pytest api/entrypoints/routers.py | head -n 260Repository: Agenta-AI/agenta
Length of output: 26239
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import ast
from pathlib import Path
target = Path("api/oss/tests/pytest/unit/sessions/test_trigger_session_claim_postgres.py")
routers = Path("api/entrypoints/routers.py")
target_tree = ast.parse(target.read_text())
fixture = next(
node for node in target_tree.body
if isinstance(node, ast.AsyncFunctionDef)
and node.name == "_fresh_engine_per_test"
)
print("fixture statements:")
for node in fixture.body:
print(f" {type(node).__name__}: {ast.unparse(node)}")
router_tree = ast.parse(routers.read_text())
assignments = []
for node in ast.walk(router_tree):
if isinstance(node, (ast.Assign, ast.AnnAssign)):
text = ast.unparse(node)
if "_transactions_engine" in text:
assignments.append(text)
print("\nrouter assignments containing _transactions_engine:")
for text in assignments:
print(f" {text}")
print("\nidentity simulation:")
old_engine = object()
global_engine = old_engine
global_engine = None
new_engine = object()
global_engine = new_engine
print(f" old engine retained by external holder: {old_engine is not global_engine}")
print(f" new singleton differs from old engine: {new_engine is not old_engine}")
PYRepository: Agenta-AI/agenta
Length of output: 2095
Avoid resetting the shared transactions engine in this fixture.
The reset discards an existing TransactionsEngine without closing its pool, then creates a second singleton. A pre-yield close() does not make this safe because module-level DAOs in api/entrypoints/routers.py retain the old engine. Use a test-scoped engine and close only that engine.
✅ Addressed in commit 703338e
There was a problem hiding this comment.
Valid observation, but pre-existing: this exact _fresh_engine_per_test pattern is copied verbatim across every DB-backed integration test in the repo (test_turns_dao.py, test_triggers_soft_delete_postgres.py, etc.), not introduced by this PR. A systemic fix (test-scoped engine) belongs in its own repo-wide cleanup rather than a one-off change in this file. Filing a follow-up to track it.
There was a problem hiding this comment.
@mmabrouk, acknowledged. The issue is pre-existing and affects the shared integration-test infrastructure.
A repository-wide follow-up is the correct scope. Please link the follow-up issue here when available. I will not require a one-file change in this PR.
You are interacting with an AI system.
| @pytest_asyncio.fixture | ||
| async def trigger_scope(): | ||
| engine = get_transactions_engine() | ||
| user_id = uuid.uuid4() | ||
| organization_id = uuid.uuid4() | ||
| workspace_id = uuid.uuid4() | ||
| project_id = uuid.uuid4() | ||
| connection_id = uuid.uuid4() | ||
|
|
||
| async with engine.session() as session: | ||
| await session.execute( | ||
| text( | ||
| "INSERT INTO users (id, uid, username, email) " | ||
| "VALUES (:id, :uid, :username, :email)" | ||
| ), | ||
| { | ||
| "id": user_id, | ||
| "uid": str(user_id), | ||
| "username": "trigger-soft-delete-test", | ||
| "email": f"trigger-soft-delete-{user_id.hex[:8]}@example.com", | ||
| }, | ||
| ) | ||
| await session.execute( | ||
| text( | ||
| "INSERT INTO organizations (id, name, owner_id) " | ||
| "VALUES (:id, :name, :owner_id)" | ||
| ), | ||
| { | ||
| "id": organization_id, | ||
| "name": "trigger-soft-delete-org", | ||
| "owner_id": user_id, | ||
| }, | ||
| ) | ||
| await session.execute( | ||
| text( | ||
| "INSERT INTO workspaces (id, name, organization_id) " | ||
| "VALUES (:id, :name, :organization_id)" | ||
| ), | ||
| { | ||
| "id": workspace_id, | ||
| "name": "trigger-soft-delete-workspace", | ||
| "organization_id": organization_id, | ||
| }, | ||
| ) | ||
| await session.execute( | ||
| text( | ||
| "INSERT INTO projects " | ||
| "(id, project_name, workspace_id, organization_id) " | ||
| "VALUES (:id, :name, :workspace_id, :organization_id)" | ||
| ), | ||
| { | ||
| "id": project_id, | ||
| "name": "trigger-soft-delete-project", | ||
| "workspace_id": workspace_id, | ||
| "organization_id": organization_id, | ||
| }, | ||
| ) | ||
| await session.execute( | ||
| text( | ||
| "INSERT INTO gateway_connections " | ||
| "(id, project_id, created_by_id, slug, provider_key, integration_key) " | ||
| "VALUES (:id, :project_id, :user_id, :slug, :provider_key, :integration_key)" | ||
| ), | ||
| { | ||
| "id": connection_id, | ||
| "project_id": project_id, | ||
| "user_id": user_id, | ||
| "slug": "trigger-soft-delete-connection", | ||
| "provider_key": "composio", | ||
| "integration_key": "github", | ||
| }, | ||
| ) | ||
| await session.commit() | ||
|
|
||
| yield { | ||
| "project_id": project_id, | ||
| "user_id": user_id, | ||
| "connection_id": connection_id, | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Reuse the shared account fixtures.
The trigger_scope fixture hand-rolls users, organizations, workspaces, and projects rows with raw SQL, then tears them down manually. The repository provides foo_account, cls_account, and mod_account in api/oss/tests/pytest/utils/accounts.py for this. Build only the gateway_connections row on top of one of those fixtures.
As per coding guidelines: "For tests requiring accounts or API keys, reuse the foo_account, cls_account, and mod_account fixtures from api/oss/tests/pytest/utils/accounts.py rather than hand-rolling account creation."
Source: Coding guidelines
There was a problem hiding this comment.
Not a drop-in swap: foo_account/cls_account/mod_account POST to a live HTTP API and hand back API credentials, while this file (like every other DAO-level Postgres integration test, including the new ones this PR adds) needs a project_id UUID in hand to seed rows via engine.session() directly, which the HTTP-level fixtures don't expose. Filing a follow-up to track a proper DAO-level account-fixture helper.
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '=== Account fixture definitions ==='
ast-grep outline api/oss/tests/pytest/utils/accounts.py --items all --type function
sed -n '1,260p' api/oss/tests/pytest/utils/accounts.py
echo '=== DAO-level Postgres fixture patterns ==='
rg -n -C 5 --glob 'api/oss/tests/pytest/unit/**/*.py' \
'`@pytest_asyncio`\.fixture|async def .*scope|INSERT INTO (users|organizations|workspaces|projects)' \
api/oss/tests/pytest/unit
echo '=== Direct uses of shared account fixtures in unit tests ==='
rg -n -C 3 --glob 'api/oss/tests/pytest/unit/**/*.py' \
'\b(foo_account|cls_account|mod_account)\b' \
api/oss/tests/pytest/unitLength of output: 68610
@mmabrouk, verification confirms your explanation.
foo_account, cls_account, and mod_account call the HTTP account endpoint. They return only api_url and API-key credentials. They do not expose project_id or user_id.
Comparable DAO-level PostgreSQL integration tests create the account hierarchy through engine.session() and use the generated database identifiers directly. A dedicated DAO-level account fixture is the correct follow-up.
You are interacting with an AI system.
Railway Preview Environment
Updated at 2026-08-11T11:48:28.058Z |
|
Follow-up on the rest of CodeRabbit's review (the items without individual inline threads): Fixed:
Rejected (no individual thread to reply on — these were nitpicks bundled in the review body):
|
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
api/oss/src/core/triggers/service.py (1)
1140-1155: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConfirm the default-provider fallback cannot target the wrong provider.
When the connection row is missing, this code assumes
composioand calls that adapter withtrigger_id. If a subscription of another provider reaches this path, the delete targets the wrong provider API. Store or derive the provider key on the subscription if more providers become supported.Run the following script to check which providers can own a subscription today:
#!/bin/bash set -euo pipefail rg -n -C3 'class TriggerProviderKind' api/oss/src rg -n -C3 'adapter_registry.get\(' api/oss/src/core/triggers
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: CHILL
Plan: Pro Plus
Run ID: 8f520eab-40ef-4ccc-90a0-1d1890b03aa3
📒 Files selected for processing (36)
api/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/oss/src/apis/fastapi/sessions/models.pyapi/oss/src/apis/fastapi/sessions/router.pyapi/oss/src/apis/fastapi/sessions/utils.pyapi/oss/src/core/sessions/records/dtos.pyapi/oss/src/core/sessions/service.pyapi/oss/src/core/sessions/streams/dtos.pyapi/oss/src/core/sessions/streams/interfaces.pyapi/oss/src/core/sessions/turns/interfaces.pyapi/oss/src/core/sessions/turns/service.pyapi/oss/src/core/triggers/interfaces.pyapi/oss/src/core/triggers/service.pyapi/oss/src/dbs/postgres/sessions/records/dao.pyapi/oss/src/dbs/postgres/sessions/streams/dao.pyapi/oss/src/dbs/postgres/sessions/streams/mappings.pyapi/oss/src/dbs/postgres/sessions/turns/dao.pyapi/oss/src/dbs/postgres/shared/utils.pyapi/oss/src/dbs/postgres/triggers/dao.pyapi/oss/src/tasks/asyncio/triggers/dispatcher.pyapi/oss/tests/pytest/integration/sessions/__init__.pyapi/oss/tests/pytest/integration/sessions/conftest.pyapi/oss/tests/pytest/integration/sessions/test_sessions_query_postgres.pyapi/oss/tests/pytest/integration/sessions/test_trigger_session_claim_postgres.pyapi/oss/tests/pytest/unit/sessions/test_query_sessions_filters.pyapi/oss/tests/pytest/unit/sessions/test_query_sessions_references.pyapi/oss/tests/pytest/unit/sessions/test_query_sessions_search.pyapi/oss/tests/pytest/unit/sessions/test_session_last_message.pyapi/oss/tests/pytest/unit/sessions/test_session_trigger_expansion.pyapi/oss/tests/pytest/unit/sessions/test_session_trigger_stamp.pyapi/oss/tests/pytest/unit/sessions/test_sessions_query_contract.pyapi/oss/tests/pytest/unit/sessions/test_sessions_root_service.pyapi/oss/tests/pytest/unit/sessions/test_turns_dao.pyapi/oss/tests/pytest/unit/sessions/test_wp5_root_router.pyapi/oss/tests/pytest/unit/triggers/test_triggers_dispatcher.pyapi/oss/tests/pytest/unit/triggers/test_triggers_lifecycle.pyapi/oss/tests/pytest/unit/triggers/test_triggers_soft_delete_postgres.py
💤 Files with no reviewable changes (1)
- api/oss/src/core/sessions/streams/dtos.py
🚧 Files skipped from review as they are similar to previous changes (10)
- api/oss/tests/pytest/unit/triggers/test_triggers_soft_delete_postgres.py
- api/oss/tests/pytest/unit/sessions/test_wp5_root_router.py
- api/oss/tests/pytest/unit/sessions/test_query_sessions_filters.py
- api/oss/src/apis/fastapi/sessions/utils.py
- api/oss/tests/pytest/unit/sessions/test_session_trigger_expansion.py
- api/oss/tests/pytest/unit/sessions/test_sessions_query_contract.py
- api/oss/src/core/sessions/service.py
- api/oss/tests/pytest/unit/sessions/test_session_last_message.py
- api/oss/src/tasks/asyncio/triggers/dispatcher.py
- api/oss/src/dbs/postgres/sessions/streams/dao.py
| from oss.src.core.sessions.mounts.dtos import SessionMount, SessionMountQuery | ||
| from oss.src.core.sessions.turns.dtos import HarnessKind, SessionTurn, SessionTurnQuery | ||
| from oss.src.core.shared.dtos import OTelSpanId, Reference, Windowing | ||
| from oss.src.dbs.postgres.sessions.streams.dao import MAX_SESSION_QUERY_LIMIT |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Move the query limit constant out of the PostgreSQL DAO.
This request schema imports MAX_SESSION_QUERY_LIMIT from a concrete PostgreSQL DAO. Put the shared limit in a DB-independent core contract or configuration module, then import it from there.
As per coding guidelines, “Follow the required dependency direction: router -> service -> DAO interface -> DAO implementation -> DB.”
Source: Coding guidelines
| async def query_session_ids_by_references( | ||
| self, | ||
| *, | ||
| project_id: UUID, | ||
| references: List[Reference], | ||
| limit: int, | ||
| ) -> List[str]: | ||
| return await self._dao.query_session_ids_by_references( | ||
| project_id=project_id, | ||
| references=references, | ||
| limit=limit, | ||
| ) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Return typed DTOs from the reference session-ID lookup.
The new lookup exposes List[str] across the core service and DAO contract. Define a named Pydantic DTO for session IDs and use List[DTO] at this boundary. Update the session query caller to extract IDs for the stream query.
api/oss/src/core/sessions/turns/service.py#L86-L97: returnList[SessionIdDTO]instead ofList[str].api/oss/src/core/sessions/turns/interfaces.py#L54-L67: change the DAO contract to the same DTO collection.api/oss/src/dbs/postgres/sessions/turns/dao.py#L132-L153: map selected session IDs into the DTO collection.
As per coding guidelines, “Service methods must return typed DTOs (Pydantic BaseModel subclasses)” and “use List[DTO] for collections.”
📍 Affects 3 files
api/oss/src/core/sessions/turns/service.py#L86-L97(this comment)api/oss/src/core/sessions/turns/interfaces.py#L54-L67api/oss/src/dbs/postgres/sessions/turns/dao.py#L132-L153
Source: Coding guidelines
| # `json.dumps(None) == "null"` — a JSON-null scalar, not a SQL | ||
| # NULL bind — sidesteps asyncpg's ambiguous-type-for-NULL error | ||
| # on a bound parameter feeding a `CAST(:x AS jsonb)` expression. | ||
| # It round-trips to Python `None` on read either way. | ||
| "tags": json.dumps(tags), |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Seed SQL NULL for manual-session tags.
Lines 136-140 store JSONB null, not SQL NULL. These values have different PostgreSQL semantics. The test can pass when the origin filter still fails for legacy rows where session_streams.tags IS NULL.
Use a conditional SQL expression that inserts NULL::jsonb for manual rows. Keep JSONB objects for trigger rows.
Proposed fix
- # `json.dumps(None) == "null"` — a JSON-null scalar, not a SQL
- # NULL bind — sidesteps asyncpg's ambiguous-type-for-NULL error
- # on a bound parameter feeding a `CAST(:x AS jsonb)` expression.
- # It round-trips to Python `None` on read either way.
+ "is_trigger": is_trigger,
"tags": json.dumps(tags),- "VALUES (:id, :project_id, :user_id, :session_id, :created_at, "
- "CAST(:tags AS jsonb))"
+ "VALUES (:id, :project_id, :user_id, :session_id, :created_at, "
+ "CASE WHEN :is_trigger THEN CAST(:tags AS jsonb) ELSE NULL::jsonb END)"📝 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.
| # `json.dumps(None) == "null"` — a JSON-null scalar, not a SQL | |
| # NULL bind — sidesteps asyncpg's ambiguous-type-for-NULL error | |
| # on a bound parameter feeding a `CAST(:x AS jsonb)` expression. | |
| # It round-trips to Python `None` on read either way. | |
| "tags": json.dumps(tags), | |
| "is_trigger": is_trigger, | |
| "tags": json.dumps(tags), |
🧰 Tools
🪛 ast-grep (0.45.1)
[info] 139-139: use jsonify instead of json.dumps for JSON output
Context: json.dumps(tags)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
…eserved ag. namespace
703338e to
62d2916
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository YAML (base), Organization UI (inherited)
Review profile: CHILL
Plan: Pro Plus
Run ID: 8838dc5f-2abd-44ca-ab35-1ee1c5508ba3
📒 Files selected for processing (12)
api/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/oss/src/apis/fastapi/sessions/utils.pyapi/oss/src/apis/fastapi/triggers/router.pyapi/oss/src/core/triggers/dtos.pyapi/oss/src/dbs/postgres/sessions/streams/dao.pyapi/oss/src/dbs/postgres/sessions/streams/mappings.pyapi/oss/src/dbs/postgres/triggers/dao.pyapi/oss/tests/pytest/acceptance/triggers/test_triggers_subscriptions.pyapi/oss/tests/pytest/integration/sessions/test_trigger_session_claim_postgres.pyapi/oss/tests/pytest/unit/sessions/test_session_response_sanitization.pyapi/oss/tests/pytest/unit/sessions/test_session_trigger_stamp.pyapi/oss/tests/pytest/unit/sessions/test_sessions_query_contract.py
🚧 Files skipped from review as they are similar to previous changes (8)
- api/oss/tests/pytest/acceptance/triggers/test_triggers_subscriptions.py
- api/oss/src/apis/fastapi/triggers/router.py
- api/oss/tests/pytest/unit/sessions/test_sessions_query_contract.py
- api/oss/tests/pytest/unit/sessions/test_session_trigger_stamp.py
- api/oss/src/dbs/postgres/sessions/streams/mappings.py
- api/ee/tests/pytest/acceptance/triggers/test_triggers_subscriptions.py
- api/oss/src/dbs/postgres/sessions/streams/dao.py
- api/oss/src/dbs/postgres/triggers/dao.py
| from oss.src.dbs.postgres.sessions.streams.mappings import ( | ||
| SESSION_RESERVED_TAG_NAMESPACE, | ||
| ) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Move SESSION_RESERVED_TAG_NAMESPACE out of the Postgres mapping module.
This API-layer module imports a constant from oss.src.dbs.postgres.sessions.streams.mappings, which is a concrete DB implementation. The reserved tag namespace is a domain concept, not a persistence detail. Define it in the sessions core layer (for example oss/src/core/sessions/types.py) and let both the API layer and the Postgres mappings import it from there.
As per coding guidelines: "Follow the required dependency direction: router -> service -> DAO interface -> DAO implementation -> DB; routers/services must not depend on concrete DB implementations or return DBE objects."
#!/bin/bash
# Description: Locate the reserved tag namespace/key definitions and all importers.
set -euo pipefail
rg -nP --type=py -C3 'SESSION_RESERVED_TAG_NAMESPACE|SESSION_RESERVED_TAG_KEYS'
1 participant
Context
Automation-created sessions needed a reliable link to the configuration and exact delivery that started them. The previous path wrote attribution separately and exposed private
ag.*tag keys, so failures, concurrent updates, or deletion could leave incomplete history.Changes
The dispatcher now claims a delivery and creates or merges its attributed session in one Postgres transaction before invoking the workflow. The session query accepts nested typed predicates, returns typed trigger and delivery relationships plus terminal windowing, and runs message or trigger enrichment only when requested. Normal schedule and subscription deletion is now soft deletion, while public session responses strip reserved attribution tags.
Before, a caller had to interpret
tags["ag.trigger.id"]. After, it receivestrigger: {id, kind, name}anddelivery: {id}.Tests / notes
Stack: 1 of 4. Base:
release/v0.112.0.