3596         updated_at_col = func.coalesce(Project.updated_at, _KEYSET_PAGINATION_EPOCH)
  3597         if updated_after is not None:
  3598             query = query.filter(updated_at_col >= updated_after)
  3599         if keyset_after is not None or limit is not None:
! 3600             if keyset_after is not None:
! 3601                 after_updated_at, after_name = keyset_after
  3602                 # Portable keyset predicate (works identically on SQLite/Postgres/MySQL)
  3603                 # — avoids row-value comparison syntax some dialects handle inconsistently.
! 3604                 query = query.filter(
  3605                     or_(
  3606                         updated_at_col > after_updated_at,
  3607                         and_(
  3608                             updated_at_col == after_updated_at,

  3612                 )
  3613             # A stable order is required for keyset pagination to mean anything — the
  3614             # very first page (keyset_after=None) still needs it, so the boundary row
  3615             # a caller computes its next cursor from is consistent across pages.
! 3616             query = query.order_by(updated_at_col, Project.name)
  3617         if limit is not None:
! 3618             query = query.limit(limit)
  3619         if for_update:
! 3620             query = query.with_for_update()
  3621 
  3622         project_records = query.all()
  3623         return mlrun.common.schemas.ProjectsOutput(
  3624             projects=self._format_projects(project_records, format_)

  193         Re-routes the /follower/projects commit-delete call to chief. Blocks until
  194         the full purge finishes, so the timeout needs to cover that, not the legacy
  195         chief-proxy default.
  196         """
! 197         return await self._proxy_request_to_chief(
  198             "DELETE",
  199             f"follower/projects/{name}",
  200             request,
  201             timeout=int(

  367 
  368         if response.status_code == requests.codes.accepted:
  369             # 202: accepted, still converging - poll for completion
  370             if not wait_for_completion:
! 371                 return True
  372             op_id = response.json()["status"]["opId"]
  373             self._wait_for_op(name, op_id, auth_info)
  374 
  375         return False

  467                 auth_info,
  468                 fatal_exceptions=(_OrcaActionFailedError,),
  469             )
  470         except _OrcaActionFailedError as exc:
! 471             raise mlrun.errors.MLRunRuntimeError(str(exc)) from exc
  472 
  473     def _verify_op_terminal(
  474         self,
  475         name: str,

  489             },
  490         )
  491         items = response.json().get("items", [])
  492         if not items:
! 493             raise mlrun.errors.MLRunRuntimeError(
  494                 f"No Orca sync-project action observed yet for project {name} (op_id={op_id})"
  495             )
  496         state = items[0].get("status", {}).get("state")
  497         if state == "failed":

  511         wire = {"name": project.metadata.name}
  512         if project.spec.owner:
  513             wire["owner"] = project.spec.owner
  514         if project.spec.description:
! 515             wire["description"] = project.spec.description
  516         if project.metadata.labels:
! 517             wire["labels"] = project.metadata.labels
  518         if project.metadata.annotations:
! 519             wire["annotations"] = project.metadata.annotations
  520         return wire
  521 
  522     @staticmethod
  523     def _update_project_wire(

  530             "prevOpId": str(prev_op_id) if prev_op_id else None,
  531             "owner": project.spec.owner,
  532         }
  533         if project.spec.description:
! 534             wire["description"] = project.spec.description
  535         if project.metadata.labels:
! 536             wire["labels"] = project.metadata.labels
  537         if project.metadata.annotations:
! 538             wire["annotations"] = project.metadata.annotations
  539         return wire
  540 
  541     @staticmethod
  542     def _wire_to_project(body: dict) -> mlrun.common.schemas.Project:

  44 def _to_follower_state(
  45     name: str, snapshot: mlrun.common.schemas.Project | None
  46 ) -> follower_schemas.FollowerProjectState:
  47     if snapshot is None:
! 48         return follower_schemas.FollowerProjectState(name=name)
  49     return follower_schemas.FollowerProjectState(
  50         name=snapshot.metadata.name,
  51         op_id=snapshot.status.op_id,
  52         sync_status=snapshot.status.state,

  60 async def prepare_create_project(
  61     name: str,
  62     body: follower_schemas.FollowerPrepareCreateRequest,
  63 ) -> follower_schemas.FollowerProjectState:
! 64     if body.metadata.name != name:
! 65         raise mlrun.errors.MLRunBadRequestError(
  66             "Path project name does not match metadata.name"
  67         )
! 68     project = mlrun.common.schemas.Project(
  69         metadata=body.metadata,
  70         spec=body.spec,
  71         status=mlrun.common.schemas.ProjectStatus(op_id=body.status.op_id),
  72     )
! 73     result = await mlrun.utils.run_in_threadpool(
  74         services.api.crud.Projects().prepare_create_project,
  75         project,
  76         body.status.op_id,
  77     )
! 78     return _to_follower_state(name, result)
  79 
  80 
  81 @router.post(
  82     "/follower/projects/{name}/commit-create",

  85 async def commit_create_project(
  86     name: str,
  87     body: follower_schemas.FollowerCommitCreateRequest,
  88 ) -> follower_schemas.FollowerProjectState:
! 89     result = await mlrun.utils.run_in_threadpool(
  90         services.api.crud.Projects().commit_create_project, name, body.status.op_id
  91     )
! 92     return _to_follower_state(name, result)
  93 
  94 
  95 @router.put(
  96     "/follower/projects/{name}",

   99 async def update_project(
  100     name: str,
  101     body: follower_schemas.FollowerUpdateRequest,
  102 ) -> follower_schemas.FollowerProjectState:
! 103     if body.metadata.name != name:
! 104         raise mlrun.errors.MLRunBadRequestError(
  105             "Path project name does not match metadata.name"
  106         )
! 107     project = mlrun.common.schemas.Project(
  108         metadata=body.metadata,
  109         spec=body.spec,
  110         status=mlrun.common.schemas.ProjectStatus(op_id=body.status.op_id),
  111     )
! 112     result = await mlrun.utils.run_in_threadpool(
  113         services.api.crud.Projects().update_project_follower,
  114         name,
  115         project,
  116         op_id=body.status.op_id,

  115         project,
  116         op_id=body.status.op_id,
  117         prev_op_id=body.prev_op_id,
  118     )
! 119     return _to_follower_state(name, result)
  120 
  121 
  122 @router.post(
  123     "/follower/projects/{name}/prepare-delete",

  126 async def prepare_delete_project(
  127     name: str,
  128     body: follower_schemas.FollowerPrepareDeleteRequest,
  129 ) -> follower_schemas.FollowerProjectState:
! 130     result = await mlrun.utils.run_in_threadpool(
  131         services.api.crud.Projects().prepare_delete_project,
  132         name,
  133         op_id=body.status.op_id,
  134         prev_op_id=body.prev_op_id,

  132         name,
  133         op_id=body.status.op_id,
  134         prev_op_id=body.prev_op_id,
  135     )
! 136     return _to_follower_state(name, result)
  137 
  138 
  139 @router.delete(
  140     "/follower/projects/{name}",

  146     request: fastapi.Request,
  147 ) -> follower_schemas.FollowerProjectState:
  148     # delete_project_resources deletes schedules, which run only on chief, so we
  149     # re-route to chief — same reason the legacy (non-follower) delete endpoint does.
! 150     if (
  151         mlrun.mlconf.httpdb.clusterization.role
  152         != mlrun.common.schemas.ClusterizationRole.chief
  153     ):
! 154         logger.info(
  155             "Requesting to commit-delete follower project, re-routing to chief",
  156             project=name,
  157         )
! 158         return (
  159             await framework.utils.clients.chief.Client().commit_delete_follower_project(
  160                 name=name, request=request
  161             )
  162         )

  160                 name=name, request=request
  161             )
  162         )
  163 
! 164     op_id = body.status.op_id
  165     # Validates internally (CAS/ordering/state) and, on success, purges the project's
  166     # resources and its row — a no-op if it's already gone (a previous call already
  167     # removed it) or a genuine retry with the same op_id (e.g. after a dropped
  168     # connection re-runs the purge rather than skipping it).
! 169     await mlrun.utils.run_in_threadpool(
  170         services.api.crud.Projects().commit_delete_project, name, op_id
  171     )
  172     # The row is gone at this point — sync_status reports the state it was in right
  173     # before removal, matching every other op's {name, op_id, sync_status} response.
! 174     return follower_schemas.FollowerProjectState(
  175         name=name,
  176         op_id=op_id,
  177         sync_status=mlrun.common.schemas.ProjectState.deleting,
  178     )

  189     db_session: sqlalchemy.orm.Session = fastapi.Depends(
  190         framework.api.deps.get_db_session
  191     ),
  192 ) -> follower_schemas.FollowerProjectStatesPage:
! 193     projects, next_cursor = await mlrun.utils.run_in_threadpool(
  194         services.api.crud.Projects().list_project_states,
  195         db_session,
  196         updated_after,
  197         cursor,

  196         updated_after,
  197         cursor,
  198         page_size,
  199     )
! 200     return follower_schemas.FollowerProjectStatesPage(
  201         projects=[
  202             _to_follower_state(project.metadata.name, project) for project in projects
  203         ],
  204         next_cursor=next_cursor,

  1163             # Not on the wire for this op (Orca's request has no updated_at field at
  1164             # all) — MLRun stamps its own local time for when it applied the change.
  1165             project.status.updated_at = datetime.datetime.now(datetime.UTC)
  1166             if existing:
! 1167                 self.store_project(session, name, project)
  1168             else:
  1169                 self.create_project(session, project)
  1170             return project
  1171         finally:

  1231                 incoming_op_id=op_id,
  1232                 prev_op_id=prev_op_id,
  1233             )
  1234             if outcome == follower_contract.ReplayOutcome.replay:
! 1235                 return existing
  1236             # No project payload on this call, and updated_at isn't on the wire at
  1237             # all — MLRun stamps its own local time for when it applied the change.
  1238             updated_at = datetime.datetime.now(datetime.UTC)
  1239             self.patch_project(

  1315                 incoming_op_id=op_id,
  1316                 prev_op_id=prev_op_id,
  1317             )
  1318             if outcome == follower_contract.ReplayOutcome.replay:
! 1319                 return existing
  1320             # No updated_at on the wire for this op either — MLRun stamps its own
  1321             # local time for when it applied the change.
  1322             updated_at = datetime.datetime.now(datetime.UTC)
  1323             # The common set only — labels/annotations/owner/description. Everything else

  1394     def _decode_project_states_cursor(cursor: str) -> tuple[datetime.datetime, str]:
  1395         updated_at_str, _, name = cursor.partition("|")
  1396         try:
  1397             updated_at = datetime.datetime.fromisoformat(updated_at_str)
! 1398         except ValueError as exc:
! 1399             raise mlrun.errors.MLRunBadRequestError(
  1400                 f"Invalid list-states cursor: {cursor}"
  1401             ) from exc
  1402         return updated_at, name