refactor(backend): split into api/ schemas/ services/ clients/ layers

4-phase restructuring of the previously flat backend/ package. Each
phase lands as a single squash commit so future bisects stay readable
per phase if needed.

## Phase 1 — move + shim (location-only, zero behavior change)
* git mv 14 files into api/ schemas/ services/ clients/ subpackages
  (history preserved via RM/R renames)
* New files: api/{admin,auth,dependencies,jupyter,platform,resources,
  scripts,storage}.py + api/schedules/{schedules,runs}.py
* New files: schemas/{auth,common,jupyter,platform,resources,
  schedules,scripts}.py
* New files: clients/{rclone,runtime,scheduler}.py
* Old paths kept as 1-line `from backend.<new> import *` shims so
  tests/main.py/importers kept working untouched
* schemas/__init__.py now re-exports from backend.schemas.<domain>

## Phase 2 — APIRouter prefix consolidation
* Every APIRouter() now carries its prefix (e.g. prefix="/api/v1/auth")
  and decorators are stripped of the redundant path prefix
* URL paths exposed to the frontend are byte-identical to before
* Affected: api/{auth,jupyter,admin,platform,resources,scripts,
  storage}.py + api/schedules/{schedules,runs}.py

## Phase 3 — first service-layer extraction
* backend.services.schedules.validate_dag moved out of api/
  (pure DAG validator, no Request/BackgroundTasks/DB)
* api/schedules/schedules.py now re-exports the symbol so existing
  4 callsites keep working unchanged
* Added backend/tests/test_validate_dag.py: 8 unit tests covering
  DAG_EMPTY, linear chain, diamond, cycle, self-edge, duplicate
  edge, orphan edge, multi-root ordering

## Phase 4 — delete shims + unify test imports
* Removed 14 flat shim files + schemas/__init__.py
* Migrated 5 test files (32 import sites) to new paths:
  backend.scripts.* → backend.api.scripts.*
  backend.resources.* → backend.api.resources.*
  backend.jupyter.* → backend.api.jupyter.*
  backend.runtime_client.* → backend.clients.runtime.*
  backend.schemas.UpdateScriptRequest → backend.schemas.scripts.*
* audit.py kept at backend.audit (main.py references it; not a
  shim, real code)

## Final structure
backend/src/backend/
  main.py, audit.py, __init__.py
  api/            (10 files: routes + 2 subpackage)
  schemas/        (7 files: Pydantic contracts)
  services/       (storage + schedules)
  clients/        (rclone, runtime, scheduler)

## Verification
* uv run python -m compileall backend/src backend/tests — clean
* uv run --package backend pytest backend/tests -q — 122 passed
  (114 → 114 → 122 → 122 across phases)
* grep -r 'from backend\.\(scripts\|resources\|...\)' backend/ — 0 hits
* git blame --follow still traces file origins through the renames

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

backend/tests/test_count_scripts.py

@@ -15,7 +15,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from backend.scripts import count_scripts
+from backend.api.scripts import count_scripts
 
 
 def _ctx(
@@ -139,7 +139,7 @@ async def test_count_scripts_route_declared_before_script_id_route() -> None:
     """Static check: the `/api/v1/scripts/count` route MUST be declared in
     scripts.py before `/api/v1/scripts/{script_id}/...`, otherwise FastAPI's
     declaration-order matching will interpret `count` as a script_id."""
-    from backend.scripts import count_scripts, get_script
+    from backend.api.scripts import count_scripts, get_script
 
     assert callable(count_scripts)
     assert callable(get_script)

backend/tests/test_jupyter_auth_cache.py

@@ -18,12 +18,12 @@ from unittest.mock import AsyncMock
 import pytest
 from fastapi import HTTPException
 
-import backend.jupyter as jupyter_module
-from backend.jupyter import (
+import backend.api.jupyter as jupyter_module
+from backend.api.jupyter import (
     _JUPYTER_AUTH_CACHE,
     verify_jupyter_access,
 )
-from backend.runtime_client import RuntimeClientError
+from backend.clients.runtime import RuntimeClientError
 
 WS_ID = "01WS0000000000000000000A"
 USER_ID = "01USR0000000000000000000A"

backend/tests/test_list_scripts_parent_path.py

@@ -27,7 +27,7 @@ from fastapi import HTTPException
 from sqlalchemy import Column, MetaData, String, Table, create_engine, select, text
 from sqlalchemy.dialects import mysql as mysql_dialect
 
-from backend.scripts import (
+from backend.api.scripts import (
     _build_list_scripts_descendant_prefix,
     _build_list_scripts_workspace_descendant_prefix,
     _escape_like_pattern,
@@ -179,7 +179,7 @@ def _compile_sql(stmt) -> str:
 
 
 async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper() -> None:
-    from backend.scripts import list_scripts
+    from backend.api.scripts import list_scripts
 
     captured_sql: list[str] = []
 
@@ -205,7 +205,7 @@ async def test_list_scripts_where_clause_uses_like_prefix_and_excludes_deeper()
 async def test_list_scripts_where_clause_escapes_pattern_literal() -> None:
     """Regression: parent_path containing ``_`` MUST be escaped in the
     compiled LIKE pattern, otherwise sibling-path leak returns to bite."""
-    from backend.scripts import list_scripts
+    from backend.api.scripts import list_scripts
 
     captured_sql: list[str] = []
 
@@ -237,7 +237,7 @@ async def test_list_scripts_where_clause_escapes_pattern_literal() -> None:
 
 async def test_list_scripts_where_clause_escapes_percent_pattern() -> None:
     """Same regression for ``%``."""
-    from backend.scripts import list_scripts
+    from backend.api.scripts import list_scripts
 
     captured_sql: list[str] = []
 
@@ -264,7 +264,7 @@ async def test_list_scripts_non_admin_adds_visibility_filter() -> None:
     """Workspace-wide listing is narrowed by visibility for non-admin:
     owner_user_id = me OR visibility IN (workspace, public) — exactly like
     list_resources. The workspace prefix contains NO user_id (cross-owner)."""
-    from backend.scripts import list_scripts
+    from backend.api.scripts import list_scripts
 
     captured_sql: list[str] = []
 
@@ -288,7 +288,7 @@ async def test_list_scripts_non_admin_adds_visibility_filter() -> None:
 
 async def test_list_scripts_admin_skips_visibility_filter() -> None:
     """Admin short-circuits the visibility predicate and sees everything."""
-    from backend.scripts import list_scripts
+    from backend.api.scripts import list_scripts
 
     captured_sql: list[str] = []
 
@@ -316,7 +316,7 @@ async def test_list_scripts_admin_skips_visibility_filter() -> None:
 async def test_list_workspace_directories_where_clause_escapes_pattern() -> None:
     """list_workspace_directories must escape user input too (was
     pre-existing debt)."""
-    from backend.scripts import list_workspace_directories
+    from backend.api.scripts import list_workspace_directories
 
     captured_sql: list[str] = []
 

backend/tests/test_resources.py

@@ -13,7 +13,7 @@ import pytest
 from fastapi import HTTPException
 from sqlalchemy import Column, MetaData, String, Table, create_engine, select
 from sqlalchemy.dialects import mysql as mysql_dialect
-from backend.resources import (
+from backend.api.resources import (
     _build_list_resources_descendant_prefix,
     can_view,
     compute_jupyter_relative_path,
@@ -126,7 +126,7 @@ def _bind_payload():
 @pytest.mark.asyncio
 async def test_bind_resource_rejects_duplicate_name_in_same_directory() -> None:
     """Same resource_name in the same directory raises 409."""
-    from backend.resources import bind_resource
+    from backend.api.resources import bind_resource
 
     existing_rows = [
         (
@@ -153,7 +153,7 @@ async def test_bind_resource_rejects_duplicate_name_in_same_directory() -> None:
 @pytest.mark.asyncio
 async def test_bind_resource_allows_same_name_in_different_directory() -> None:
     """Same resource_name in a different directory binds successfully."""
-    from backend.resources import bind_resource
+    from backend.api.resources import bind_resource
 
     existing_rows = [
         (
@@ -179,7 +179,7 @@ async def test_bind_resource_allows_same_name_in_different_directory() -> None:
 @pytest.mark.asyncio
 async def test_bind_resource_allows_same_name_when_workspace_empty() -> None:
     """No same-name rows at all: bind succeeds (root directory)."""
-    from backend.resources import bind_resource
+    from backend.api.resources import bind_resource
 
     session = _BindSessionMock(
         new_object_key=f"{_BIND_WS}/{_BIND_USER}/data.csv",
@@ -197,7 +197,7 @@ async def test_bind_resource_allows_same_name_when_workspace_empty() -> None:
 @pytest.mark.asyncio
 async def test_bind_resource_allows_same_name_for_different_owner() -> None:
     """其他用户在同目录下的同名资源不阻塞当前用户的绑定。"""
-    from backend.resources import bind_resource
+    from backend.api.resources import bind_resource
 
     other_user = "01USR0000000000000000000B"
     existing_rows = [
@@ -223,7 +223,7 @@ async def test_bind_resource_allows_same_name_for_different_owner() -> None:
 @pytest.mark.asyncio
 async def test_bind_resource_allows_rebinding_same_storage_object() -> None:
     """重新绑定同一 upload_id 应走 idempotent 复用路径,不触发 409。"""
-    from backend.resources import bind_resource
+    from backend.api.resources import bind_resource
 
     new_object_key = f"{_BIND_WS}/{_BIND_USER}/data.csv"
     existing_resource = _make_resource(_BIND_WS, _BIND_USER)
@@ -251,7 +251,7 @@ async def test_bind_resource_allows_rebinding_same_storage_object() -> None:
 @pytest.mark.asyncio
 async def test_bind_resource_rejects_non_data_resource_upload() -> None:
     """其他用途(如 working_copy)的 upload session 不能 bind 成数据资源。"""
-    from backend.resources import bind_resource
+    from backend.api.resources import bind_resource
 
     session = _BindSessionMock(
         new_object_key=f"{_BIND_WS}/{_BIND_USER}/data.csv",
@@ -566,7 +566,7 @@ def _list_resources_capturing_session(captured_sql: list[str]) -> MagicMock:
 
 
 async def test_list_resources_where_clause_uses_like_prefix_and_excludes_deeper() -> None:
-    from backend.resources import list_resources
+    from backend.api.resources import list_resources
 
     captured_sql: list[str] = []
     mock_session = _list_resources_capturing_session(captured_sql)
@@ -589,7 +589,7 @@ async def test_list_resources_where_clause_uses_like_prefix_and_excludes_deeper(
 async def test_list_resources_where_clause_escapes_underscore() -> None:
     """Regression: parent_path containing ``_`` MUST be escaped in the
     compiled LIKE pattern, otherwise sibling-path leak (``fooXbar``) returns."""
-    from backend.resources import list_resources
+    from backend.api.resources import list_resources
 
     captured_sql: list[str] = []
     mock_session = _list_resources_capturing_session(captured_sql)
@@ -614,7 +614,7 @@ async def test_list_resources_where_clause_escapes_underscore() -> None:
 async def test_list_resources_without_parent_path_adds_no_like_clause() -> None:
     """Empty parent_path keeps the legacy workspace-wide behaviour — no
     object_key LIKE filter at all."""
-    from backend.resources import list_resources
+    from backend.api.resources import list_resources
 
     captured_sql: list[str] = []
     mock_session = _list_resources_capturing_session(captured_sql)
@@ -634,7 +634,7 @@ async def test_list_resources_without_parent_path_adds_no_like_clause() -> None:
 async def test_list_resources_joins_users_for_display_name() -> None:
     """list_resources must OUTER JOIN users and SELECT users.display_name so
     every resource carries owner_display_name (frontend displayName chain)."""
-    from backend.resources import list_resources
+    from backend.api.resources import list_resources
 
     captured_sql: list[str] = []
     mock_session = _list_resources_capturing_session(captured_sql)

backend/tests/test_runtime_client_directories.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 import httpx
 import pytest
 import respx
-from backend.runtime_client import RuntimeClient, RuntimeClientError
+from backend.clients.runtime import RuntimeClient, RuntimeClientError
 
 WORKSPACE_ID = "01HWS0000000000000000000A"
 BASE_URL = "http://runtime"

backend/tests/test_scripts.py

@@ -103,7 +103,7 @@ class _AsyncSessionMock:
 @pytest.mark.asyncio
 async def test_create_script_record_flushes_storage_object_before_script() -> None:
     """StorageObjects must flush first so path conflicts surface early."""
-    from backend.scripts import create_script_record
+    from backend.api.scripts import create_script_record
 
     session = _AsyncSessionMock()
     request = _make_request()
@@ -133,7 +133,7 @@ async def test_create_script_record_flushes_storage_object_before_script() -> No
 @pytest.mark.asyncio
 async def test_create_script_record_storage_object_flush_failure_does_not_add_script() -> None:
     """If the StorageObjects flush fails, the Scripts row must never be added."""
-    from backend.scripts import create_script_record
+    from backend.api.scripts import create_script_record
 
     class FailingSession(_AsyncSessionMock):
         async def flush(self) -> None:
@@ -172,7 +172,7 @@ async def test_create_script_record_allows_reupload_after_delete() -> None:
     """Without uk_scripts_workspace_name_active, re-uploading a script with
     the same name after the previous one was soft-deleted succeeds.
     """
-    from backend.scripts import create_script_record
+    from backend.api.scripts import create_script_record
 
     session = _AsyncSessionMock()
     request = _make_request()
@@ -230,7 +230,7 @@ async def test_create_script_record_allows_same_name_different_parent() -> None:
     must coexist — they correspond to different Jupyter paths
     (/user/foo.ipynb vs /user/test/foo.ipynb).
     """
-    from backend.scripts import create_script_record
+    from backend.api.scripts import create_script_record
 
     session = _AsyncSessionMock()
     request = _make_request()
@@ -286,7 +286,7 @@ async def test_create_script_after_soft_delete_does_not_conflict() -> None:
     raise IntegrityError — the generated column is NULL for the deleted row,
     so it does not occupy the UNIQUE slot.
     """
-    from backend.scripts import create_script_record
+    from backend.api.scripts import create_script_record
 
     session = _AsyncSessionMock()
     request = _make_request()
@@ -376,7 +376,7 @@ def _storage_object_row() -> StorageObjects:
 @pytest.mark.asyncio
 async def test_delete_script_route_sets_is_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
     """Soft-deleting a script via the route handler flips is_deleted=1."""
-    from backend.scripts import delete_script
+    from backend.api.scripts import delete_script
 
     script = _script_row()
     storage_object = _storage_object_row()
@@ -395,7 +395,7 @@ async def test_delete_script_route_sets_is_deleted(monkeypatch: pytest.MonkeyPat
     ) -> tuple[Scripts, StorageObjects]:
         return script, storage_object
 
-    monkeypatch.setattr("backend.scripts.get_script_row", _fake_get_script_row)
+    monkeypatch.setattr("backend.api.scripts.get_script_row", _fake_get_script_row)
     mock_soft_delete = AsyncMock(
         return_value={
             "data": {
@@ -406,7 +406,7 @@ async def test_delete_script_route_sets_is_deleted(monkeypatch: pytest.MonkeyPat
             }
         }
     )
-    monkeypatch.setattr("backend.scripts.soft_delete_object", mock_soft_delete)
+    monkeypatch.setattr("backend.api.scripts.soft_delete_object", mock_soft_delete)
 
     result = await delete_script(
         script_id=script.script_id,
@@ -430,7 +430,7 @@ async def test_delete_script_route_sets_is_deleted(monkeypatch: pytest.MonkeyPat
 @pytest.mark.asyncio
 async def test_delete_resource_sets_is_deleted(monkeypatch: pytest.MonkeyPatch) -> None:
     """Soft-deleting a data resource must write is_deleted=1 on the row."""
-    from backend.resources import delete_resource
+    from backend.api.resources import delete_resource
 
     resource = DataResources(
         resource_id="01RES0000000000000000000A",
@@ -455,7 +455,7 @@ async def test_delete_resource_sets_is_deleted(monkeypatch: pytest.MonkeyPatch)
 
     with monkeypatch.context() as mp:
         mp.setattr(
-            "backend.resources.soft_delete_object",
+            "backend.api.resources.soft_delete_object",
             AsyncMock(return_value={"data": {}}),
         )
         result = await delete_resource(
@@ -558,7 +558,7 @@ async def test_soft_delete_object_streams_via_get_stream() -> None:
 @pytest.mark.asyncio
 async def test_jupyter_check_notebook_lock_ignores_deleted_scripts() -> None:
     """``is_deleted == 0`` filter must hide deleted notebooks from Jupyter checks."""
-    from backend.jupyter import check_notebook_is_locked
+    from backend.api.jupyter import check_notebook_is_locked
 
     session = AsyncMock()
     session.execute = AsyncMock()
@@ -593,8 +593,8 @@ async def test_update_script_writes_back_storage_object_metadata(
     """
     import hashlib
 
-    from backend.schemas import UpdateScriptRequest
-    from backend.scripts import update_script
+    from backend.schemas.scripts import UpdateScriptRequest
+    from backend.api.scripts import update_script
 
     script = _script_row()
     storage_object = _storage_object_row()
@@ -624,7 +624,7 @@ async def test_update_script_writes_back_storage_object_metadata(
     ) -> tuple[Scripts, StorageObjects]:
         return script, storage_object
 
-    monkeypatch.setattr("backend.scripts.get_script_row", _fake_get_script_row)
+    monkeypatch.setattr("backend.api.scripts.get_script_row", _fake_get_script_row)
 
     new_content = '{"cells": [{"cell_type": "code", "source": ["print(1)"]}]}\n'
     payload = UpdateScriptRequest(content=new_content)
@@ -670,8 +670,8 @@ async def test_update_script_jupyter_only_uses_dict_fallback(
     """
     import hashlib
 
-    from backend.schemas import UpdateScriptRequest
-    from backend.scripts import update_script
+    from backend.schemas.scripts import UpdateScriptRequest
+    from backend.api.scripts import update_script
 
     script = _script_row()
     user_id = "01USR0000000000000000000A"
@@ -695,7 +695,7 @@ async def test_update_script_jupyter_only_uses_dict_fallback(
     ) -> tuple[Scripts, None]:
         return script, None
 
-    monkeypatch.setattr("backend.scripts.get_script_row", _fake_get_script_row)
+    monkeypatch.setattr("backend.api.scripts.get_script_row", _fake_get_script_row)
 
     payload = UpdateScriptRequest(content='{"cells": []}\n')
     result = await update_script(

backend/tests/test_validate_dag.py

@@ -0,0 +1,143 @@
+"""Unit tests for backend.services.schedules.validate_dag.
+
+Pure function — no DB, no FastAPI, no fixtures beyond SimpleNamespace
+stand-ins for the SQLAlchemy rows. The function only reads five
+attributes: ``node_id``, ``node_key``, ``edge_id``, ``source_node_id``,
+``target_node_id``.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from backend.services.schedules import validate_dag
+
+
+def _node(node_id: str, node_key: str) -> SimpleNamespace:
+    return SimpleNamespace(node_id=node_id, node_key=node_key)
+
+
+def _edge(edge_id: str, source: str, target: str) -> SimpleNamespace:
+    return SimpleNamespace(
+        edge_id=edge_id,
+        source_node_id=source,
+        target_node_id=target,
+    )
+
+
+def test_empty_nodes_is_rejected_as_dag_empty() -> None:
+    result = validate_dag(nodes=[], edges=[])
+    assert result["valid"] is False
+    assert result["node_count"] == 0
+    assert result["edge_count"] == 0
+    assert result["topological_order"] == []
+    codes = [err["code"] for err in result["errors"]]
+    assert "DAG_EMPTY" in codes
+
+
+def test_linear_chain_orders_by_node_key() -> None:
+    nodes = [_node("n1", "A"), _node("n2", "B"), _node("n3", "C")]
+    edges = [_edge("e1", "n1", "n2"), _edge("e2", "n2", "n3")]
+    result = validate_dag(nodes, edges)
+    assert result["valid"] is True
+    assert result["root_node_ids"] == ["n1"]
+    assert result["leaf_node_ids"] == ["n3"]
+    assert result["topological_order"] == ["n1", "n2", "n3"]
+
+
+def test_diamond_topology_is_valid() -> None:
+    #   A -> B -> D
+    #   A -> C -> D
+    nodes = [
+        _node("a", "A"),
+        _node("b", "B"),
+        _node("c", "C"),
+        _node("d", "D"),
+    ]
+    edges = [
+        _edge("e1", "a", "b"),
+        _edge("e2", "a", "c"),
+        _edge("e3", "b", "d"),
+        _edge("e4", "c", "d"),
+    ]
+    result = validate_dag(nodes, edges)
+    assert result["valid"] is True
+    assert result["root_node_ids"] == ["a"]
+    assert result["leaf_node_ids"] == ["d"]
+    # Kahn's algorithm with node_key tie-breaking: starting at A, then B
+    # and C both become ready (B alphabetically first), then D.
+    assert result["topological_order"] == ["a", "b", "c", "d"]
+
+
+def test_cycle_is_rejected_with_dag_cycle() -> None:
+    # n1 -> n2 -> n3 -> n1
+    nodes = [_node("n1", "A"), _node("n2", "B"), _node("n3", "C")]
+    edges = [
+        _edge("e1", "n1", "n2"),
+        _edge("e2", "n2", "n3"),
+        _edge("e3", "n3", "n1"),
+    ]
+    result = validate_dag(nodes, edges)
+    assert result["valid"] is False
+    codes = [err["code"] for err in result["errors"]]
+    assert "DAG_CYCLE" in codes
+    cycle_err = next(err for err in result["errors"] if err["code"] == "DAG_CYCLE")
+    # The cycle should list every node in the cycle (sorted by node_key).
+    assert set(cycle_err["node_ids"]) == {"n1", "n2", "n3"}
+
+
+def test_self_edge_is_rejected_but_does_not_count_as_cycle() -> None:
+    nodes = [_node("n1", "A"), _node("n2", "B")]
+    edges = [
+        _edge("e_self", "n1", "n1"),
+        _edge("e_real", "n1", "n2"),
+    ]
+    result = validate_dag(nodes, edges)
+    codes = [err["code"] for err in result["errors"]]
+    assert "DAG_SELF_EDGE" in codes
+    # The A->B edge still makes the DAG valid overall except for the self-edge.
+    assert "DAG_CYCLE" not in codes
+    # One node remains reachable (B), so cycle detection must not fire.
+    assert result["topological_order"] == ["n1", "n2"]
+
+
+def test_duplicate_edge_is_rejected_with_dag_duplicate_edge() -> None:
+    nodes = [_node("n1", "A"), _node("n2", "B")]
+    edges = [
+        _edge("e1", "n1", "n2"),
+        _edge("e1_dup", "n1", "n2"),
+    ]
+    result = validate_dag(nodes, edges)
+    codes = [err["code"] for err in result["errors"]]
+    assert "DAG_DUPLICATE_EDGE" in codes
+    # The first edge still counts toward edge_count, the second is rejected.
+    assert result["edge_count"] == 2
+
+
+def test_edge_to_unknown_node_is_dag_edge_node_missing() -> None:
+    nodes = [_node("n1", "A")]
+    edges = [
+        _edge("e1", "n1", "ghost"),
+        _edge("e2", "ghost", "n1"),
+    ]
+    result = validate_dag(nodes, edges)
+    codes = [err["code"] for err in result["errors"]]
+    assert codes.count("DAG_EDGE_NODE_MISSING") == 2
+    # No cycle should be reported for orphan edges.
+    assert "DAG_CYCLE" not in codes
+
+
+def test_multiple_roots_are_sorted_by_node_key() -> None:
+    nodes = [
+        _node("z", "Z"),
+        _node("a", "A"),
+        _node("m", "M"),
+    ]
+    edges = []
+    result = validate_dag(nodes, edges)
+    assert result["valid"] is True
+    # All three nodes are roots (no indegree) and leaves (no outgoing).
+    assert result["root_node_ids"] == ["a", "m", "z"]
+    assert result["leaf_node_ids"] == ["a", "m", "z"]
+    # Topological order picks the smallest node_key first.
+    assert result["topological_order"] == ["a", "m", "z"]