feat: auth

backend/src/backend/auth.py

@@ -0,0 +1,256 @@
+"""Cookie+JWT authentication endpoints.
+
+The user-facing flow is:
+  1. POST /api/v1/auth/login  — verify password, set HttpOnly cookie
+  2. every other /api/ request reads the cookie via
+     ``backend.dependencies.request_context``
+  3. POST /api/v1/auth/logout — clear the cookie
+  4. GET  /api/v1/auth/me     — return the current user
+
+Service-to-service calls do not use these endpoints — they live on the
+shared Docker network and have no application-layer auth. See
+``docker-compose.yml`` and ``schedule/service.py``.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.dependencies import database_session
+from common.auth.jwt import JwtError, issue_jwt, verify_jwt_token
+from common.auth.passwords import verify_password
+from common.db.models import Roles, Users, WorkspaceMembers, Workspaces
+from common.ids import new_ulid
+
+
+router = APIRouter(tags=["auth"])
+
+# Cookie config. ``secure=True`` requires HTTPS — the only safe
+# assumption in production. Dev environments running on plain HTTP
+# should reverse-proxy with TLS termination or set the env knob
+# (future extension).
+COOKIE_NAME = "access_token"
+COOKIE_TTL_SECONDS = 24 * 60 * 60
+COOKIE_SECURE = True
+COOKIE_SAMESITE = "lax"
+
+
+def _set_session_cookie(response: Response, token: str) -> None:
+    response.set_cookie(
+        key=COOKIE_NAME,
+        value=token,
+        max_age=COOKIE_TTL_SECONDS,
+        path="/",
+        httponly=True,
+        secure=COOKIE_SECURE,
+        samesite=COOKIE_SAMESITE,
+    )
+
+
+def _clear_session_cookie(response: Response) -> None:
+    response.delete_cookie(key=COOKIE_NAME, path="/")
+
+
+def _user_payload(user: Users, role_code: str | None = None) -> dict[str, Any]:
+    return {
+        "user_id": user.user_id,
+        "username": user.username,
+        "display_name": user.display_name,
+        "email": user.email,
+        "status": user.status,
+        "role_code": role_code,
+    }
+
+
+def _workspace_payload(
+    workspace: Workspaces,
+    role: Roles,
+) -> dict[str, Any]:
+    return {
+        "workspace_id": workspace.workspace_id,
+        "workspace_code": workspace.workspace_code,
+        "workspace_name": workspace.workspace_name,
+        "role_code": role.role_code,
+        "role_name": role.role_name,
+    }
+
+
+@router.post("/api/v1/auth/login")
+async def login(
+    request: Request,
+    response: Response,
+    session: AsyncSession = Depends(database_session),
+) -> dict[str, Any]:
+    """Verify username/password and issue a session cookie.
+
+    Returns the user record and the workspaces they are an active
+    member of (joined earliest first, which doubles as the default
+    selection until the user picks a different one in the UI). The
+    workspace list is informational — the JWT itself does not bind a
+    workspace; each request specifies its own ``?workspace_id=``.
+    """
+    body = await request.json()
+    username = (body or {}).get("username", "").strip()
+    password = (body or {}).get("password", "")
+    if not username or not password:
+        raise HTTPException(
+            status.HTTP_400_BAD_REQUEST,
+            "username and password are required",
+        )
+
+    user = await session.scalar(
+        select(Users).where(
+            Users.username == username,
+            Users.status == "active",
+        )
+    )
+    if user is None or not verify_password(password, user.password_hash):
+        # Unified 401 to prevent username enumeration.
+        raise HTTPException(
+            status.HTTP_401_UNAUTHORIZED,
+            "invalid username or password",
+        )
+
+    # Pull all active memberships. The earliest join wins as default
+    # because there is no `is_default` column on `workspace_members`.
+    rows = (
+        await session.execute(
+            select(Workspaces, Roles, WorkspaceMembers.joined_at)
+            .join(
+                WorkspaceMembers,
+                WorkspaceMembers.workspace_id == Workspaces.workspace_id,
+            )
+            .join(Roles, Roles.role_id == WorkspaceMembers.role_id)
+            .where(
+                WorkspaceMembers.user_id == user.user_id,
+                WorkspaceMembers.member_status == "active",
+                Workspaces.status == "active",
+            )
+            .order_by(WorkspaceMembers.joined_at.asc())
+        )
+    ).all()
+    if not rows:
+        raise HTTPException(
+            status.HTTP_403_FORBIDDEN,
+            "user is not a member of any active workspace",
+        )
+
+    workspaces = []
+    default_workspace_id: str | None = None
+    for workspace, role, joined_at in rows:
+        workspaces.append(_workspace_payload(workspace, role))
+        if default_workspace_id is None:
+            default_workspace_id = workspace.workspace_id
+
+    # Pick a default role_code for the user payload: prefer admin if
+    # the user has it in any workspace, otherwise use the first one
+    # returned. This is only for UI greeting; access control checks
+    # run on a per-request basis via the chosen workspace_id.
+    user_role_code: str | None = None
+    for workspace, role, _ in rows:
+        if role.role_code == "admin":
+            user_role_code = "admin"
+            break
+    if user_role_code is None:
+        user_role_code = rows[0][1].role_code
+
+    token = issue_jwt(user.user_id, ttl_seconds=COOKIE_TTL_SECONDS)
+    _set_session_cookie(response, token)
+
+    return {
+        "request_id": new_ulid(),
+        "data": {
+            "user": _user_payload(user, user_role_code),
+            "workspaces": workspaces,
+            "default_workspace_id": default_workspace_id,
+        },
+        "meta": {},
+    }
+
+
+@router.post("/api/v1/auth/logout")
+async def logout(response: Response) -> dict[str, Any]:
+    """Clear the session cookie. Idempotent."""
+    _clear_session_cookie(response)
+    return {
+        "request_id": new_ulid(),
+        "data": {"logged_out": True},
+        "meta": {},
+    }
+
+
+@router.get("/api/v1/auth/me")
+async def me(
+    request: Request,
+    session: AsyncSession = Depends(database_session),
+) -> dict[str, Any]:
+    """Return the current user record based on the session cookie.
+
+    Does not require a workspace_id — useful for the frontend to
+    bootstrap identity on app load before any workspace has been
+    selected. Workspace list is included so the login screen can be
+    skipped on subsequent visits.
+    """
+    token = request.cookies.get(COOKIE_NAME)
+    if not token:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated")
+    try:
+        payload = verify_jwt_token(token)
+    except JwtError as exc:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(exc)) from exc
+
+    user_id = payload.get("sub")
+    if not user_id:
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token")
+
+    user = await session.get(Users, user_id)
+    if user is None or user.status != "active":
+        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not found")
+
+    rows = (
+        await session.execute(
+            select(Workspaces, Roles, WorkspaceMembers.joined_at)
+            .join(
+                WorkspaceMembers,
+                WorkspaceMembers.workspace_id == Workspaces.workspace_id,
+            )
+            .join(Roles, Roles.role_id == WorkspaceMembers.role_id)
+            .where(
+                WorkspaceMembers.user_id == user.user_id,
+                WorkspaceMembers.member_status == "active",
+                Workspaces.status == "active",
+            )
+            .order_by(WorkspaceMembers.joined_at.asc())
+        )
+    ).all()
+
+    workspaces = [_workspace_payload(ws, role) for ws, role, _ in rows]
+    default_workspace_id = workspaces[0]["workspace_id"] if workspaces else None
+    user_role_code: str | None = None
+    for _ws, role, _ in rows:
+        if role.role_code == "admin":
+            user_role_code = "admin"
+            break
+    if user_role_code is None and rows:
+        user_role_code = rows[0][1].role_code
+
+    return {
+        "request_id": new_ulid(),
+        "data": {
+            "user": _user_payload(user, user_role_code),
+            "workspaces": workspaces,
+            "default_workspace_id": default_workspace_id,
+        },
+        "meta": {},
+    }
+
+
+__all__ = [
+    "COOKIE_NAME",
+    "COOKIE_TTL_SECONDS",
+    "router",
+]

common/src/common/auth/__init__.py

@@ -0,0 +1,14 @@
+"""Authentication primitives shared across services.
+
+Houses the HS256 JWT issuer/verifier, bcrypt password helpers, and the
+single canonical active-membership loader. The previous implementation
+lived inline inside ``backend/jupyter.py`` and ``backend/dependencies.py``;
+moving it here lets the schedule service verify the same tokens (when
+service-to-service auth is reintroduced) and keeps the dependency
+inversion clean.
+
+Service-to-service HTTP calls in this repository do NOT currently
+authenticate at the application layer (see ``docker-compose.yml``: only
+the gateway exposes a host port). The JWT and membership helpers are
+used exclusively by user-facing endpoints.
+"""

common/src/common/auth/jwt.py

@@ -0,0 +1,141 @@
+"""HS256 JWT issuance and verification.
+
+The implementation is intentionally minimal: a hand-rolled HS256
+signer/verifier so the project does not depend on PyJWT. It deliberately
+ignores the ``alg`` header on the verify side and always recomputes
+HMAC-SHA256, which means a forged ``"alg":"none"`` token still fails
+signature validation.
+
+Token payload contract:
+    sub  - user_id (CHAR(26) ULID)
+    exp  - unix seconds; mandatory
+    iat  - unix seconds; mandatory (used for last_logout_at checks if added later)
+
+Any other claim is preserved by the verifier but not interpreted here.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import time
+from typing import Any
+
+from common.config import settings
+
+
+JWT_SECRET: str = settings.jwt_secret
+JWT_ALGORITHM: str = "HS256"
+DEFAULT_TTL_SECONDS: int = 24 * 60 * 60
+
+
+def _b64encode(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
+
+
+def _b64decode(value: str) -> bytes:
+    padding = "=" * (-len(value) % 4)
+    return base64.urlsafe_b64decode(value + padding)
+
+
+def issue_jwt(
+    user_id: str,
+    *,
+    ttl_seconds: int = DEFAULT_TTL_SECONDS,
+    extra_claims: dict[str, Any] | None = None,
+    now: int | None = None,
+) -> str:
+    """Sign a JWT for ``user_id`` and return the compact serialization.
+
+    The header is fixed ``{"alg":"HS256","typ":"JWT"}``; the signature
+    uses HMAC-SHA256 over ``{header_b64}.{payload_b64}`` keyed by
+    ``settings.jwt_secret``.
+
+    ``ttl_seconds`` defaults to 24h. ``extra_claims`` is merged into the
+    payload after ``sub``/``iat``/``exp`` are populated and would
+    overwrite those if callers passed the same keys — kept simple on
+    purpose so we never accidentally bypass the contract.
+    """
+    if not user_id:
+        raise ValueError("user_id is required")
+
+    issued_at = int(time.time()) if now is None else int(now)
+    payload: dict[str, Any] = {
+        "sub": user_id,
+        "iat": issued_at,
+        "exp": issued_at + int(ttl_seconds),
+    }
+    if extra_claims:
+        payload.update(extra_claims)
+
+    header = {"alg": JWT_ALGORITHM, "typ": "JWT"}
+    header_b64 = _b64encode(json.dumps(header, separators=(",", ":")).encode())
+    payload_b64 = _b64encode(json.dumps(payload, separators=(",", ":")).encode())
+    signing_input = f"{header_b64}.{payload_b64}".encode()
+    signature = hmac.new(
+        JWT_SECRET.encode(),
+        signing_input,
+        hashlib.sha256,
+    ).digest()
+    signature_b64 = _b64encode(signature)
+    return f"{header_b64}.{payload_b64}.{signature_b64}"
+
+
+class JwtError(Exception):
+    """Raised on missing / malformed / expired / wrong-signature tokens."""
+
+
+def verify_jwt_token(token: str | None) -> dict[str, Any]:
+    """Verify an HS256-signed JWT and return its payload.
+
+    The function is the inverse of :func:`issue_jwt`. The ``alg`` header
+    is read for completeness but the signature is always recomputed
+    under HS256 — a token claiming ``alg":"none"`` is rejected because
+    its signature segment will not match a recomputed HMAC.
+
+    Raises :class:`JwtError` on every failure mode; callers translate to
+    401 in HTTP contexts.
+    """
+    if not token:
+        raise JwtError("missing authentication token")
+
+    try:
+        header_b64, payload_b64, signature_b64 = token.split(".", 2)
+    except ValueError as exc:
+        raise JwtError("malformed token") from exc
+
+    signing_input = f"{header_b64}.{payload_b64}".encode()
+    expected = hmac.new(
+        JWT_SECRET.encode(),
+        signing_input,
+        hashlib.sha256,
+    ).digest()
+    try:
+        signature = _b64decode(signature_b64)
+    except Exception as exc:  # pragma: no cover - malformed b64
+        raise JwtError("malformed signature") from exc
+    if not hmac.compare_digest(expected, signature):
+        raise JwtError("signature mismatch")
+
+    try:
+        payload = json.loads(_b64decode(payload_b64))
+    except Exception as exc:
+        raise JwtError("malformed payload") from exc
+
+    exp = payload.get("exp")
+    if not isinstance(exp, (int, float)) or exp < time.time():
+        raise JwtError("token expired")
+
+    return payload
+
+
+__all__ = [
+    "DEFAULT_TTL_SECONDS",
+    "JWT_ALGORITHM",
+    "JWT_SECRET",
+    "JwtError",
+    "issue_jwt",
+    "verify_jwt_token",
+]

common/src/common/auth/membership.py

@@ -0,0 +1,69 @@
+"""Active workspace-membership loader.
+
+Single canonical implementation that verifies a user is currently a
+member of the given workspace. Used by the dependency-injection layer
+in the backend; ``jupyter.py`` previously carried its own near-duplicate
+that is now reduced to a thin caller.
+
+The query joins Users / WorkspaceMembers / Workspaces / Roles so
+callers receive the role they need for ``is_admin``-style checks
+without a second round-trip.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.db.models import Roles, Users, WorkspaceMembers, Workspaces
+
+
+class MembershipError(Exception):
+    """Raised when the user is not an active member of the workspace."""
+
+
+async def load_active_membership(
+    session: AsyncSession,
+    user_id: str,
+    workspace_id: str,
+) -> tuple[Users, Workspaces, Roles]:
+    """Return ``(user, workspace, role)`` for an active membership.
+
+    All four conditions must hold: ``Users.status == 'active'``,
+    ``WorkspaceMembers.member_status == 'active'``,
+    ``Workspaces.status == 'active'``, and the row exists at all.
+    Raises :class:`MembershipError` otherwise.
+    """
+    statement = (
+        select(Users, Workspaces, Roles)
+        .join(
+            WorkspaceMembers,
+            WorkspaceMembers.user_id == Users.user_id,
+        )
+        .join(
+            Workspaces,
+            Workspaces.workspace_id == WorkspaceMembers.workspace_id,
+        )
+        .join(
+            Roles,
+            Roles.role_id == WorkspaceMembers.role_id,
+        )
+        .where(
+            Users.user_id == user_id,
+            Users.status == "active",
+            WorkspaceMembers.workspace_id == workspace_id,
+            WorkspaceMembers.member_status == "active",
+            Workspaces.status == "active",
+        )
+    )
+    row = (await session.execute(statement)).one_or_none()
+    if row is None:
+        raise MembershipError("active workspace membership is required")
+    user, workspace, role = row
+    return user, workspace, role
+
+
+__all__ = [
+    "MembershipError",
+    "load_active_membership",
+]

common/src/common/auth/passwords.py

@@ -0,0 +1,55 @@
+"""Bcrypt password hashing helpers.
+
+Uses passlib's :class:`CryptContext` so the algorithm choice stays in
+one place — when (not if) we move to argon2 we change the ``schemes``
+list and existing hashes still verify.
+
+The :data:`make_unusable_password` helper returns a bcrypt hash of a
+random 32-byte secret. It is intentionally verifiable (to keep the
+``verify_password`` path symmetric) but cannot be matched by any
+human-supplied plaintext, so it is safe to assign to service accounts
+that should never log in interactively.
+"""
+
+from __future__ import annotations
+
+import secrets
+
+from passlib.context import CryptContext
+
+
+_crypt_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_password(plain: str) -> str:
+    """Hash ``plain`` with bcrypt and return the encoded digest."""
+    if not plain:
+        raise ValueError("plain must be a non-empty string")
+    return _crypt_context.hash(plain)
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    """Return True iff ``plain`` matches ``hashed`` under the active scheme."""
+    if not plain or not hashed:
+        return False
+    try:
+        return _crypt_context.verify(plain, hashed)
+    except ValueError:
+        return False
+
+
+def make_unusable_password() -> str:
+    """Return a bcrypt hash of a random 32-byte secret.
+
+    Service accounts store this so the ``Users.password_hash`` column is
+    populated and any stray login attempt is rejected by the password
+    check (random secret → impossible to brute force offline).
+    """
+    return _crypt_context.hash(secrets.token_urlsafe(32))
+
+
+__all__ = [
+    "hash_password",
+    "make_unusable_password",
+    "verify_password",
+]

common/src/common/scheduler/trigger.py

@@ -0,0 +1,370 @@
+"""Shared schedule-trigger logic.
+
+Both the user-facing manual run endpoint (``backend.schedule_runs``)
+and the schedule service's cron tick handler call into this module to
+materialize a ``ScheduleRuns`` row plus the corresponding
+``schedule.run.requested`` outbox event. The outbox is the single
+source of truth for run dispatch — the schedule executor polls MySQL
+and picks the row up.
+
+The schedule executor does NOT take any application-layer auth from
+this codebase. Service-to-service calls on the shared Docker network
+are intentionally unauthenticated; the ``triggered_by`` field stores
+the user_id that originated the run (a human for manual runs, the
+fixed ``_system_cron`` user for cron ticks) and the executor
+re-verifies that user against ``Users.status='active'`` before doing
+work.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Any, Literal
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.db.models import (
+    ScheduleEdges,
+    ScheduleNodes,
+    ScheduleRuns,
+    Schedules,
+    Scripts,
+    Versions,
+)
+from common.eventing import add_outbox_event, utcnow
+from common.ids import new_ulid
+
+
+# A stable user_id used for cron-triggered runs. The corresponding
+# ``Users`` row is seeded by the auth-bootstrap migration so any audit
+# query joining on ``ScheduleRuns.triggered_by`` still resolves.
+SYSTEM_CRON_USER_ID = "01HZZZZZZZZZZZZZZZZZZZZZZCR"
+
+
+TriggerType = Literal["manual", "cron", "api"]
+
+
+class TriggerError(Exception):
+    """Raised when a run cannot be created. Subclasses carry the
+    appropriate HTTP status when surfaced from the backend router."""
+
+
+class ScheduleNotFound(TriggerError):
+    pass
+
+
+class InvalidDag(TriggerError):
+    def __init__(self, errors: list[dict[str, Any]]) -> None:
+        super().__init__("schedule must contain a valid non-empty DAG")
+        self.errors = errors
+
+
+class DagTooLarge(TriggerError):
+    pass
+
+
+class InvalidNodeArguments(TriggerError):
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.message = message
+
+
+def normalize_idempotency_key(
+    workspace_id: str,
+    schedule_id: str,
+    value: str,
+    *,
+    min_length: int = 8,
+) -> str:
+    """SHA-256 the (workspace, schedule, header) triple and prefix a
+    version tag. Centralized so the schedule service and the backend
+    router produce the same key.
+    """
+    normalized = value.strip()
+    if len(normalized) < min_length:
+        raise TriggerError(
+            f"Idempotency-Key must contain at least {min_length} characters"
+        )
+    digest = hashlib.sha256(
+        f"{workspace_id}:{schedule_id}:{normalized}".encode("utf-8")
+    ).hexdigest()
+    return f"run:v1:{digest}"
+
+
+def parse_node_arguments(value: dict[str, Any] | None) -> list[str]:
+    """Turn a node's ``arguments_json`` dict into a list of CLI args.
+
+    Mirrors the backend's old ``_arguments`` helper but raises
+    :class:`InvalidNodeArguments` instead of an HTTP exception, so
+    the schedule service can use it without importing FastAPI.
+    """
+    payload = value or {}
+    raw = payload.get("_args")
+    result: list[str] = [str(item) for item in raw] if isinstance(raw, list) else []
+    for key, item in payload.items():
+        if key == "_args":
+            continue
+        option = f"--{key.replace('_', '-')}"
+        if item is True:
+            result.append(option)
+        elif item is False or item is None:
+            continue
+        elif isinstance(item, list):
+            for list_item in item:
+                result.extend((option, str(list_item)))
+        elif isinstance(item, (str, int, float)):
+            result.extend((option, str(item)))
+        else:
+            raise InvalidNodeArguments(
+                f"node argument {key!r} must be a scalar or list"
+            )
+    return result
+
+
+async def _load_schedule(
+    session: AsyncSession,
+    schedule_id: str,
+    workspace_id: str,
+    *,
+    for_update: bool = False,
+) -> Schedules:
+    statement = select(Schedules).where(
+        Schedules.schedule_id == schedule_id,
+        Schedules.workspace_id == workspace_id,
+        Schedules.deleted_at.is_(None),
+    )
+    if for_update:
+        statement = statement.with_for_update()
+    item = await session.scalar(statement)
+    if item is None:
+        raise ScheduleNotFound("schedule not found")
+    return item
+
+
+async def _load_graph(
+    session: AsyncSession,
+    schedule_id: str,
+) -> tuple[
+    list[tuple[ScheduleNodes, Versions, Scripts]],
+    list[ScheduleEdges],
+]:
+    node_rows = (
+        await session.execute(
+            select(ScheduleNodes, Versions, Scripts)
+            .join(Versions, Versions.versions_id == ScheduleNodes.versions_id)
+            .join(Scripts, Scripts.script_id == Versions.script_id)
+            .where(ScheduleNodes.schedule_id == schedule_id)
+            .order_by(ScheduleNodes.created_at, ScheduleNodes.node_key)
+        )
+    ).all()
+    edges = list(
+        (
+            await session.scalars(
+                select(ScheduleEdges)
+                .where(ScheduleEdges.schedule_id == schedule_id)
+                .order_by(ScheduleEdges.created_at, ScheduleEdges.edge_id)
+            )
+        ).all()
+    )
+    return list(node_rows), edges
+
+
+def _validate_dag(
+    nodes: list[ScheduleNodes],
+    edges: list[ScheduleEdges],
+    *,
+    max_nodes: int = 100,
+    max_edges: int = 500,
+) -> list[dict[str, Any]]:
+    errors: list[dict[str, Any]] = []
+    if not nodes:
+        errors.append(
+            {
+                "code": "DAG_EMPTY",
+                "message": "schedule must contain at least one node",
+            }
+        )
+        return errors
+    if len(nodes) > max_nodes or len(edges) > max_edges:
+        raise DagTooLarge(
+            f"schedule exceeds the v1 execution size limit "
+            f"({len(nodes)} nodes / {len(edges)} edges > {max_nodes}/{max_edges})"
+        )
+
+    # Cycle detection via Kahn's algorithm.
+    in_degree: dict[str, int] = {n.node_id: 0 for n in nodes}
+    adjacency: dict[str, list[str]] = {n.node_id: [] for n in nodes}
+    for edge in edges:
+        if edge.source_node_id not in in_degree or edge.target_node_id not in in_degree:
+            errors.append(
+                {
+                    "code": "DAG_EDGE_REFERENCES_MISSING_NODE",
+                    "message": f"edge {edge.edge_id} references unknown node",
+                    "edge_id": edge.edge_id,
+                }
+            )
+            continue
+        adjacency[edge.source_node_id].append(edge.target_node_id)
+        in_degree[edge.target_node_id] += 1
+    queue = [nid for nid, d in in_degree.items() if d == 0]
+    ordered: list[str] = []
+    while queue:
+        queue.sort()
+        current = queue.pop(0)
+        ordered.append(current)
+        for neighbor in adjacency[current]:
+            in_degree[neighbor] -= 1
+            if in_degree[neighbor] == 0:
+                queue.append(neighbor)
+    if len(ordered) != len(nodes):
+        cycle_nodes = [nid for nid, d in in_degree.items() if d > 0]
+        errors.append(
+            {
+                "code": "DAG_CYCLE",
+                "message": "schedule graph contains a directed cycle",
+                "node_ids": cycle_nodes,
+            }
+        )
+    return errors
+
+
+def _build_snapshot(
+    schedule: Schedules,
+    node_rows: list[tuple[ScheduleNodes, Versions, Scripts]],
+    edges: list[ScheduleEdges],
+) -> dict[str, Any]:
+    return {
+        "schedule_name": schedule.schedule_name,
+        "workflow_version": schedule.workflow_version,
+        "max_concurrency": schedule.max_concurrency,
+        "failure_policy": schedule.failure_policy,
+        "nodes": [
+            {
+                "node_id": node.node_id,
+                "node_key": node.node_key,
+                "versions_id": version.versions_id,
+                "script_type": script.script_type,
+                "artifact_object_id": version.artifact_object_id,
+                "artifact_path": version.artifact_path,
+                "timeout_seconds": node.timeout_seconds,
+                "retry_count": node.retry_count,
+                "retry_interval_sec": node.retry_interval_sec,
+                "arguments": parse_node_arguments(node.arguments_json),
+            }
+            for node, version, script in node_rows
+        ],
+        "edges": [
+            {
+                "source_node_id": edge.source_node_id,
+                "target_node_id": edge.target_node_id,
+            }
+            for edge in edges
+        ],
+    }
+
+
+async def create_scheduled_run(
+    session: AsyncSession,
+    *,
+    schedule_id: str,
+    workspace_id: str,
+    triggered_by_user_id: str,
+    trigger_type: TriggerType,
+    idempotency_key: str,
+    trace_id: str | None = None,
+) -> tuple[ScheduleRuns, bool]:
+    """Create a ``ScheduleRuns`` row + outbox event in this session.
+
+    Returns ``(run, is_new)``. ``is_new=False`` means the
+    idempotency_key was already used and the existing run is returned
+    unchanged. ``triggered_by_user_id`` is stored as-is — for cron
+    triggers pass :data:`SYSTEM_CRON_USER_ID`.
+
+    The caller owns the transaction: ``create_scheduled_run`` flushes
+    the new row to surface the unique-constraint violation on
+    ``idempotency_key`` deterministically, then leaves the commit to
+    the caller's session lifecycle. Both the backend's
+    ``request_context`` (which uses ``session_scope``) and the
+    schedule service's own session scope can wrap this call.
+    """
+    # 1) Existing run short-circuit (re-using a known idempotency key).
+    existing = await session.scalar(
+        select(ScheduleRuns).where(ScheduleRuns.idempotency_key == idempotency_key)
+    )
+    if existing is not None:
+        if (
+            existing.workspace_id != workspace_id
+            or existing.schedule_id != schedule_id
+        ):
+            raise TriggerError("Idempotency-Key belongs to another schedule run")
+        return existing, False
+
+    # 2) Lock + load schedule for the duration of this transaction.
+    schedule = await _load_schedule(
+        session, schedule_id, workspace_id, for_update=True,
+    )
+
+    # 3) Build snapshot (validates node arguments eagerly).
+    node_rows, edges = await _load_graph(session, schedule_id)
+    snapshot = _build_snapshot(schedule, node_rows, edges)
+
+    # 4) Validate DAG after snapshot so parse_node_arguments errors
+    # surface first.
+    errors = _validate_dag(
+        [n for n, _v, _s in node_rows], edges,
+    )
+    if errors:
+        raise InvalidDag(errors)
+
+    # 5) Persist run + outbox.
+    now = utcnow()
+    run = ScheduleRuns(
+        run_id=new_ulid(),
+        schedule_id=schedule.schedule_id,
+        workspace_id=schedule.workspace_id,
+        workflow_version=schedule.workflow_version,
+        trigger_type=trigger_type,
+        idempotency_key=idempotency_key,
+        run_status="queued",
+        state_version=0,
+        schedule_snapshot=snapshot,
+        queued_at=now,
+        triggered_by=triggered_by_user_id,
+    )
+    session.add(run)
+    schedule.last_run_at = now
+
+    await add_outbox_event(
+        session,
+        event_type="schedule.run.requested",
+        producer="platform-api",
+        trace_id=trace_id,
+        aggregate_type="schedule_run",
+        aggregate_id=run.run_id,
+        idempotency_key=idempotency_key,
+        payload={
+            "workspace_id": run.workspace_id,
+            "schedule_id": run.schedule_id,
+            "run_id": run.run_id,
+            "workflow_version": run.workflow_version,
+            "trigger_type": run.trigger_type,
+            "triggered_by": run.triggered_by,
+            "schedule_snapshot": snapshot,
+        },
+    )
+    await session.flush()
+    return run, True
+
+
+__all__ = [
+    "DagTooLarge",
+    "InvalidDag",
+    "InvalidNodeArguments",
+    "SYSTEM_CRON_USER_ID",
+    "ScheduleNotFound",
+    "TriggerError",
+    "create_scheduled_run",
+    "normalize_idempotency_key",
+    "parse_node_arguments",
+]