[feat] Unify eval loops

AGENTS.md

@@ -173,6 +173,28 @@ Concrete examples:
 - Legacy app storage marker (`WORKFLOW_MARKER_KEY`): `api/oss/src/core/applications/service.py`
 - Legacy dedup key normalization (`__dedup_id__` <-> `testcase_dedup_id`): `api/oss/src/apis/fastapi/testsets/router.py`
 
+### Alembic migration chains (OSS + EE)
+
+Migrations live in two separate, parallel chains that must each resolve to a
+single head:
+- OSS: `api/oss/databases/postgres/migrations/core/versions/`
+- EE: `api/ee/databases/postgres/migrations/core/versions/`
+
+Rules:
+- After adding/editing/renaming any migration, verify each chain has exactly ONE
+  head with the bundled tool: from each `.../migrations/` directory run
+  `python3 find_head.py core` and confirm the `Heads:` list has a single entry.
+  Run it for BOTH OSS and EE.
+- New migrations chain linearly after the existing head — never fork off an older
+  node (a fork produces two heads; alembic then can't resolve a linear upgrade).
+- Revision ids must be globally unique within a chain. A duplicate id makes
+  alembic silently skip one file (the migration never runs).
+- The EE chain extends past the shared OSS head with EE-only migrations, so an
+  OSS migration chains after the OSS head while its EE copy chains after the EE
+  head — same revision id, different `down_revision`.
+- `evaluation_runs.data` / `evaluation_queues.data` are `json` columns (not
+  `jsonb`); cast `data::jsonb` before using jsonb operators / `jsonb_array_elements`.
+
 ### Router and function style conventions
 
 Router style:

api/ee/databases/postgres/migrations/core/data_migrations/applications_workflow.py

@@ -254,9 +254,11 @@ def check_url_safety(cls, v: Any) -> Any:  # noqa: N805
             return v
 
     from oss.src.dbs.postgres.git.mappings import map_dto_to_dbe
-    from oss.src.dbs.postgres.shared.engine import engine as db_engine
+    from oss.src.dbs.postgres.shared.engine import get_transactions_engine
     from datetime import datetime, timezone
 
+    db_engine = get_transactions_engine()
+
     workflow_create = WorkflowCreate(
         **application_create.model_dump(mode="json"),
     )
@@ -267,7 +269,7 @@ def check_url_safety(cls, v: Any) -> Any:  # noqa: N805
 
     # Avoid slug collision with existing workflow artifacts (e.g. evaluators)
     artifact_slug = git_artifact_create.slug
-    async with db_engine.core_session() as session:
+    async with db_engine.session() as session:
         existing = (
             await session.execute(
                 select(WorkflowArtifactDBE).filter(
@@ -298,7 +300,7 @@ def check_url_safety(cls, v: Any) -> Any:  # noqa: N805
         dto=artifact_dto,
     )
 
-    async with db_engine.core_session() as session:
+    async with db_engine.session() as session:
         session.add(artifact_dbe)
         await session.commit()
 
@@ -364,7 +366,7 @@ def check_url_safety(cls, v: Any) -> Any:  # noqa: N805
             dto=variant_dto,
         )
 
-        async with db_engine.core_session() as session:
+        async with db_engine.session() as session:
             session.add(variant_dbe)
             await session.commit()
 
@@ -415,7 +417,7 @@ def check_url_safety(cls, v: Any) -> Any:  # noqa: N805
                 dto=revision_dto,
             )
 
-            async with db_engine.core_session() as session:
+            async with db_engine.session() as session:
                 session.add(revision_dbe)
                 await session.commit()
 

api/ee/databases/postgres/migrations/core/env.py

@@ -6,7 +6,7 @@
 
 from alembic import context
 
-from oss.src.dbs.postgres.shared.engine import engine
+from oss.src.utils.env import env
 from oss.src.dbs.postgres.shared.base import Base
 
 # Side-effect imports: register SQLAlchemy models with Base.metadata
@@ -29,7 +29,7 @@
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
-config.set_main_option("sqlalchemy.url", engine.postgres_uri_core)  # type: ignore
+config.set_main_option("sqlalchemy.url", env.postgres.uri_core)
 
 
 # Interpret the config file for Python logging.

api/ee/databases/postgres/migrations/core/versions/a1d2e3f4a5b6_add_default_evaluation_queues.py

@@ -0,0 +1,38 @@
+"""add default evaluation queues
+
+Revision ID: a1d2e3f4a5b6
+Revises: b2c3d4e5f7a8
+Create Date: 2026-05-15 00:00:00
+
+Previously shared revision id `a1b2c3d4e5f6` with
+`drop_corrupted_metrics_for_some_runs`, so alembic skipped it and the index
+below never ran. Renamed to `a1d2e3f4a5b6`. The EE chain extends past the shared
+OSS head `e6f7a8b9c0d1` with EE-only migrations
+(`9d3e8f0a1b2c -> a1b2c3d4e5f7 -> b2c3d4e5f7a8`), so this EE copy chains after
+`b2c3d4e5f7a8` while the OSS copy chains after `e6f7a8b9c0d1`.
+
+The partial unique index is scoped to ACTIVE default queues (`deleted_at IS
+NULL`) so a default can be archived then recreated/unarchived by reconcile.
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "a1d2e3f4a5b6"
+down_revision: Union[str, None] = "b2c3d4e5f7a8"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ux_evaluation_queues_default_per_run")
+    op.execute("""
+        CREATE UNIQUE INDEX ux_evaluation_queues_default_per_run
+        ON evaluation_queues (project_id, run_id)
+        WHERE (flags ->> 'is_default')::boolean = true AND deleted_at IS NULL
+    """)
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS ux_evaluation_queues_default_per_run")

api/ee/databases/postgres/migrations/core/versions/a2b3c4d5e6f8_backfill_default_evaluation_queues.py

@@ -0,0 +1,145 @@
+"""backfill default evaluation queues
+
+Revision ID: a2b3c4d5e6f8
+Revises: a1d2e3f4a5b6
+Create Date: 2026-05-15 00:10:00
+
+Backfills source-family flags (`has_traces` / `has_testcases` / `has_queries` /
+`has_testsets`) to match the runtime derivation rule, then mass-creates default
+queues per the runtime policy and recomputes `is_queue`. The query/testset
+recompute keys on exact reference-key presence, not a substring match.
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "a2b3c4d5e6f8"
+down_revision: Union[str, None] = "a1d2e3f4a5b6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Backfill source-family flags to match the runtime derivation rule in
+    # dbs/postgres/evaluations/utils.py. `data` is a `json` column, so cast to
+    # jsonb before navigating. Direct sources (`has_traces`/`has_testcases`) come
+    # from the exact step key on a reference-less input; reference-backed sources
+    # (`has_queries`/`has_testsets`) from exact-key presence (JSONB `?`), not a
+    # substring match that would misfire on `query_anchor` / `testset_metadata`.
+    op.execute("""
+        UPDATE evaluation_runs
+        SET flags = COALESCE(flags, '{}'::jsonb)
+            || jsonb_build_object(
+                'has_traces', EXISTS (
+                    SELECT 1
+                    FROM jsonb_array_elements(COALESCE(data::jsonb -> 'steps', '[]'::jsonb)) AS step
+                    WHERE step ->> 'type' = 'input'
+                      AND COALESCE(step -> 'references', '{}'::jsonb) = '{}'::jsonb
+                      AND lower(COALESCE(step ->> 'key', '')) IN ('traces', 'query-direct')
+                ),
+                'has_testcases', EXISTS (
+                    SELECT 1
+                    FROM jsonb_array_elements(COALESCE(data::jsonb -> 'steps', '[]'::jsonb)) AS step
+                    WHERE step ->> 'type' = 'input'
+                      AND COALESCE(step -> 'references', '{}'::jsonb) = '{}'::jsonb
+                      AND lower(COALESCE(step ->> 'key', '')) IN ('testcases', 'testset-direct')
+                ),
+                'has_queries', EXISTS (
+                    SELECT 1
+                    FROM jsonb_array_elements(COALESCE(data::jsonb -> 'steps', '[]'::jsonb)) AS step
+                    WHERE step ->> 'type' = 'input'
+                      AND COALESCE(step -> 'references', '{}'::jsonb) ? 'query_revision'
+                ),
+                'has_testsets', EXISTS (
+                    SELECT 1
+                    FROM jsonb_array_elements(COALESCE(data::jsonb -> 'steps', '[]'::jsonb)) AS step
+                    WHERE step ->> 'type' = 'input'
+                      AND COALESCE(step -> 'references', '{}'::jsonb) ? 'testset_revision'
+                )
+            )
+    """)
+
+    # Mass-create default queues, mirroring the runtime create policy in
+    # EvaluationsService._reconcile_default_queue: a default queue should exist
+    # only for runs that should have one. The runtime predicate is
+    # `EVALUATIONS_DEFAULT_QUEUES_FOR_ALL_RUNS or has_human`, with the env toggle
+    # currently hardcoded False, so the backfill condition is `has_human = true`.
+    # Existing default queues, active or archived, are preserved and block
+    # duplicates. The created queue carries the run's own status instead of a
+    # hardcoded 'running', so closed/successful runs are not misrepresented.
+    op.execute("""
+        INSERT INTO evaluation_queues (
+            project_id,
+            id,
+            created_at,
+            created_by_id,
+            flags,
+            data,
+            status,
+            run_id
+        )
+        SELECT
+            r.project_id,
+            gen_random_uuid(),
+            CURRENT_TIMESTAMP,
+            r.created_by_id,
+            jsonb_build_object('is_default', true, 'is_sequential', false),
+            '{}'::json,
+            COALESCE(r.status, 'running'),
+            r.id
+        FROM evaluation_runs r
+        WHERE COALESCE((r.flags ->> 'has_human')::boolean, false) = true
+          AND NOT EXISTS (
+            SELECT 1
+            FROM evaluation_queues q
+            WHERE q.project_id = r.project_id
+              AND q.run_id = r.id
+              AND (q.flags ->> 'is_default')::boolean = true
+        )
+    """)
+
+    # Reconcile the other direction: runs that should NOT have a default queue
+    # (has_human = false under the current policy) but carry a stale active
+    # default queue get that queue archived, matching the runtime archive branch
+    # in _reconcile_default_queue. This keeps the fleet consistent immediately
+    # instead of waiting for the first per-run edit to reconcile.
+    op.execute("""
+        UPDATE evaluation_queues q
+        SET deleted_at = CURRENT_TIMESTAMP,
+            deleted_by_id = r.created_by_id
+        FROM evaluation_runs r
+        WHERE q.project_id = r.project_id
+          AND q.run_id = r.id
+          AND (q.flags ->> 'is_default')::boolean = true
+          AND q.deleted_at IS NULL
+          AND COALESCE((r.flags ->> 'has_human')::boolean, false) = false
+    """)
+
+    # Recompute simple-queue eligibility under the new meaning. An already
+    # existing active default queue is as valid as one inserted above.
+    op.execute("""
+        UPDATE evaluation_runs r
+        SET flags = COALESCE(r.flags, '{}'::jsonb)
+            || jsonb_build_object(
+                'is_queue',
+                COALESCE((r.flags ->> 'has_human')::boolean, false)
+                AND EXISTS (
+                    SELECT 1
+                    FROM evaluation_queues q
+                    WHERE q.project_id = r.project_id
+                      AND q.run_id = r.id
+                      AND (q.flags ->> 'is_default')::boolean = true
+                      AND q.deleted_at IS NULL
+                )
+            )
+    """)
+
+
+def downgrade() -> None:
+    # Keep generated queues/results intact on downgrade. Remove only the newly
+    # inferred flags; old is_queue semantics cannot be reconstructed safely.
+    op.execute("""
+        UPDATE evaluation_runs
+        SET flags = COALESCE(flags, '{}'::jsonb) - 'has_traces' - 'has_testcases'
+    """)

api/ee/databases/postgres/migrations/tracing/env.py

@@ -7,7 +7,7 @@
 
 from alembic import context
 
-from oss.src.dbs.postgres.shared.engine import engine
+from oss.src.utils.env import env
 from oss.src.dbs.postgres.shared.base import Base
 
 # Side-effect import: register SQLAlchemy model with Base.metadata
@@ -19,7 +19,7 @@
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
-config.set_main_option("sqlalchemy.url", engine.postgres_uri_tracing)  # type: ignore
+config.set_main_option("sqlalchemy.url", env.postgres.uri_tracing)
 
 
 # Interpret the config file for Python logging.

api/ee/src/apis/fastapi/access/router.py

@@ -4,8 +4,8 @@
 
 from oss.src.utils.exceptions import intercept_exceptions
 
-from ee.src.core.entitlements.types import SCOPES, Tracker
-from ee.src.core.entitlements.controls import (
+from ee.src.core.access.entitlements.types import SCOPES, Tracker
+from ee.src.core.access.controls import (
     get_plans,
     get_plan_description,
     get_roles,
@@ -70,6 +70,6 @@ async def fetch_roles(self) -> Dict[str, List[Dict[str, Any]]]:
         verbatim from access-controls, including the `"*"` wildcard for
         `owner` — callers that need to render the full permission list
         should expand the wildcard themselves (see
-        `ee.src.services.converters._expand_permissions`).
+        `ee.src.services.db_manager_ee._expand_permissions`).
         """
         return {scope: list(get_roles(scope)) for scope in SCOPES}

api/ee/src/apis/fastapi/billing/router.py

@@ -14,7 +14,7 @@
 from oss.src.utils.caching import acquire_lock, release_lock, renew_lock
 from oss.src.utils.env import env
 
-from ee.src.utils.entitlements import period_from, scope_from
+from ee.src.core.access.entitlements.service import period_from, scope_from
 from ee.src.core.meters.types import Meters, MeterPeriod
 from oss.src.utils.context import get_auth_scope
 
@@ -24,10 +24,10 @@
 )
 
 from ee.src.services import db_manager_ee
-from ee.src.utils.permissions import check_action_access
-from ee.src.models.shared_models import Permission
-from ee.src.core.entitlements.types import Tracker, Quota, Period, Scope
-from ee.src.core.entitlements.controls import get_plan_entitlements, get_plans
+from ee.src.core.access.permissions.service import check_action_access
+from ee.src.core.access.permissions.types import Permission
+from ee.src.core.access.entitlements.types import Tracker, Quota, Period, Scope
+from ee.src.core.access.controls import get_plan_entitlements, get_plans
 from ee.src.core.subscriptions.settings import (
     get_catalog,
     get_pricing,

api/ee/src/apis/fastapi/organizations/router.py

@@ -4,16 +4,16 @@
 from fastapi import APIRouter, Request, HTTPException
 from fastapi.responses import JSONResponse, Response
 
-from ee.src.utils.permissions import check_user_org_access
-from ee.src.utils.entitlements import (
+from ee.src.core.access.permissions.service import check_user_org_access
+from ee.src.core.access.entitlements.service import (
     check_entitlements,
     NOT_ENTITLED_RESPONSE,
     Tracker,
     Flag,
 )
 
 from ee.src.services import db_manager_ee
-from ee.src.services.selectors import get_user_org_and_workspace_id
+from ee.src.services.db_manager_ee import get_user_org_and_workspace_id
 from ee.src.services.organization_service import (
     OrganizationDomainsService,
     OrganizationProvidersService,