feat: background job system (tracked async jobs + jobs UI)

app/desktop/desktop_server.py

@@ -33,6 +33,8 @@
 from app.desktop.studio_server.eval_api import connect_evals_api
 from app.desktop.studio_server.finetune_api import connect_fine_tune_api
 from app.desktop.studio_server.import_api import connect_import_api
+from app.desktop.studio_server.jobs.api import connect_jobs_api
+from app.desktop.studio_server.jobs.registry import job_registry
 from app.desktop.studio_server.prompt_api import connect_prompt_api
 from app.desktop.studio_server.prompt_optimization_job_api import (
     connect_prompt_optimization_job_api,
@@ -111,6 +113,12 @@ async def lifespan(app: FastAPI):
         await _start_background_syncs()
         yield
     finally:
+        # End open SSE subscriptions so a UI holding the jobs stream open can't
+        # keep the worker alive (e.g. block a dev-server hot reload). Pure
+        # observer teardown — jobs keep running. Note uvicorn only reaches
+        # lifespan shutdown after its graceful-shutdown wait, so the dev server
+        # also sets timeout_graceful_shutdown to bound that wait.
+        job_registry.events.shutdown()
         try:
             await _stop_background_syncs()
         finally:
@@ -146,6 +154,7 @@ def make_app(tk_root: tk.Tk | None = None):
     connect_agent_api(app)
     connect_dev_tools(app)
     connect_chat_api(app)
+    connect_jobs_api(app)
     # Important: webhost must be last, it handles all other URLs
     connect_webhost(app)
     return app

app/desktop/dev_server.py

@@ -36,4 +36,10 @@
         reload=True,
         # Debounce when changing many files (changing branch)
         reload_delay=0.1,
+        # Bound the graceful-shutdown wait on reload. The UI holds the jobs SSE
+        # stream open; uvicorn waits for in-flight requests to finish BEFORE it
+        # runs lifespan shutdown (which closes the stream), so without a bound a
+        # reload would hang on the open SSE. After this many seconds uvicorn
+        # cancels the lingering request task instead.
+        timeout_graceful_shutdown=1,
     )

app/desktop/git_sync/middleware.py

@@ -357,7 +357,11 @@ def _resolve_endpoint(self, request: Request) -> Callable[..., Any] | None:
         return None
 
     def _get_manager_for_request(self, request: Request) -> GitSyncManager | None:
-        """Extract project_id from URL, resolve to path, return manager if auto-sync enabled."""
+        """Extract project_id from URL, resolve to path, return manager if auto-sync enabled.
+
+        Keep the project_id -> manager resolution below in sync with the request-free
+        copy in save_context.get_manager_for_project (used by background job workers).
+        """
         match = PROJECT_ID_PATTERN.match(request.url.path)
         if match is None:
             return None

app/desktop/git_sync/save_context.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from kiln_ai.utils.git_sync_protocols import SaveContext
+
+from app.desktop.git_sync.config import get_git_sync_config, project_path_from_id
+from app.desktop.git_sync.git_sync_manager import GitSyncManager
+from app.desktop.git_sync.registry import GitSyncRegistry
+
+
+def get_manager_for_project(project_id: str) -> GitSyncManager | None:
+    """Resolve a project_id to its GitSyncManager when auto-sync is active.
+
+    Request-free mirror of GitSyncMiddleware._get_manager_for_request (minus the
+    URL parsing). Returns None for every "not active" branch: the project has no
+    path, no git-sync config, sync_mode is not "auto", or no clone_path is set.
+
+    Config is keyed by project_path; the manager is keyed by clone_path. The
+    manager is always obtained via GitSyncRegistry.get_or_create so the single
+    per-clone-path manager (and its executor + non-reentrant write lock) is
+    shared with the HTTP path.
+    """
+    project_path = project_path_from_id(project_id)
+    if project_path is None:
+        return None
+
+    config = get_git_sync_config(project_path)
+    if config is None:
+        return None
+
+    if config["sync_mode"] != "auto":
+        return None
+
+    clone_path = config.get("clone_path")
+    if clone_path is None:
+        return None
+
+    return GitSyncRegistry.get_or_create(
+        repo_path=Path(clone_path),
+        remote_name=config["remote_name"],
+        pat_token=config.get("pat_token"),
+        oauth_token=config.get("oauth_token"),
+        auth_mode=config["auth_mode"],
+    )
+
+
+def save_context_for_project(project_id: str, context: str) -> SaveContext | None:
+    """Return a SaveContext wrapping writes in manager.atomic_write(context=...),
+    or None when git sync is not active for this project.
+
+    Mirrors build_save_context(request) for callers that have only a project_id
+    (e.g. background job workers). Runners coalesce None to a no-op context.
+    """
+    manager = get_manager_for_project(project_id)
+    if manager is None:
+        return None
+
+    bg_sync = GitSyncRegistry.get_background_sync(manager.repo_path)
+    if bg_sync is not None:
+        bg_sync.notify_request()
+
+    def factory():
+        return manager.atomic_write(context=context)
+
+    return factory

app/desktop/git_sync/test_save_context.py

@@ -0,0 +1,219 @@
+from __future__ import annotations
+
+from contextlib import ExitStack, asynccontextmanager
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.desktop.git_sync.config import GitSyncProjectConfig
+from app.desktop.git_sync.save_context import (
+    get_manager_for_project,
+    save_context_for_project,
+)
+
+PROJECT_ID = "project_abc"
+PROJECT_PATH = "/tmp/test/project.kiln"
+CLONE_PATH = "/tmp/test/clone"
+
+
+def _auto_config(clone_path: str | None = CLONE_PATH) -> GitSyncProjectConfig:
+    return GitSyncProjectConfig(
+        sync_mode="auto",
+        auth_mode="system_keys",
+        remote_name="origin",
+        branch="main",
+        clone_path=clone_path,
+        git_url=None,
+        pat_token=None,
+        oauth_token=None,
+    )
+
+
+def _manual_config() -> GitSyncProjectConfig:
+    return GitSyncProjectConfig(
+        sync_mode="manual",
+        auth_mode="system_keys",
+        remote_name="origin",
+        branch="main",
+        clone_path=CLONE_PATH,
+        git_url=None,
+        pat_token=None,
+        oauth_token=None,
+    )
+
+
+class _FakeManager:
+    """Minimal AtomicWriteCapable stand-in that records atomic_write calls."""
+
+    def __init__(self, repo_path: Path = Path(CLONE_PATH)):
+        self.repo_path = repo_path
+        self.calls: list[str] = []
+        self.entered = False
+
+    @asynccontextmanager
+    async def atomic_write(self, context: str):
+        self.calls.append(context)
+        self.entered = True
+        yield
+
+
+def _patch_resolution(project_path, config, manager=None, bg_sync=None):
+    """Patch the config + registry calls used by the helper.
+
+    project_path_from_id and get_git_sync_config are looked up in the
+    save_context module namespace, so patch them there.
+    """
+    stack = ExitStack()
+    stack.enter_context(
+        patch(
+            "app.desktop.git_sync.save_context.project_path_from_id",
+            return_value=project_path,
+        )
+    )
+    stack.enter_context(
+        patch(
+            "app.desktop.git_sync.save_context.get_git_sync_config",
+            return_value=config,
+        )
+    )
+    stack.enter_context(
+        patch(
+            "app.desktop.git_sync.save_context.GitSyncRegistry.get_or_create",
+            return_value=manager,
+        )
+    )
+    stack.enter_context(
+        patch(
+            "app.desktop.git_sync.save_context.GitSyncRegistry.get_background_sync",
+            return_value=bg_sync,
+        )
+    )
+    return stack
+
+
+# -- None branches -----------------------------------------------------------
+
+
+def test_returns_none_when_no_project_path():
+    with _patch_resolution(project_path=None, config=None):
+        assert save_context_for_project(PROJECT_ID, context="ctx") is None
+        assert get_manager_for_project(PROJECT_ID) is None
+
+
+def test_returns_none_when_no_git_sync_config():
+    with _patch_resolution(project_path=PROJECT_PATH, config=None):
+        assert save_context_for_project(PROJECT_ID, context="ctx") is None
+        assert get_manager_for_project(PROJECT_ID) is None
+
+
+def test_returns_none_when_sync_mode_not_auto():
+    with _patch_resolution(project_path=PROJECT_PATH, config=_manual_config()):
+        assert save_context_for_project(PROJECT_ID, context="ctx") is None
+        assert get_manager_for_project(PROJECT_ID) is None
+
+
+def test_returns_none_when_clone_path_missing():
+    with _patch_resolution(
+        project_path=PROJECT_PATH, config=_auto_config(clone_path=None)
+    ):
+        assert save_context_for_project(PROJECT_ID, context="ctx") is None
+        assert get_manager_for_project(PROJECT_ID) is None
+
+
+# -- active branches ---------------------------------------------------------
+
+
+def test_get_manager_uses_registry_with_config_values():
+    manager = _FakeManager()
+    with (
+        patch(
+            "app.desktop.git_sync.save_context.project_path_from_id",
+            return_value=PROJECT_PATH,
+        ),
+        patch(
+            "app.desktop.git_sync.save_context.get_git_sync_config",
+            return_value=_auto_config(),
+        ),
+        patch(
+            "app.desktop.git_sync.save_context.GitSyncRegistry.get_or_create",
+            return_value=manager,
+        ) as mock_get_or_create,
+    ):
+        result = get_manager_for_project(PROJECT_ID)
+
+    assert result is manager
+    mock_get_or_create.assert_called_once_with(
+        repo_path=Path(CLONE_PATH),
+        remote_name="origin",
+        pat_token=None,
+        oauth_token=None,
+        auth_mode="system_keys",
+    )
+
+
+async def test_save_context_enters_atomic_write_with_label():
+    manager = _FakeManager()
+    with _patch_resolution(
+        project_path=PROJECT_PATH, config=_auto_config(), manager=manager
+    ):
+        save_context = save_context_for_project(PROJECT_ID, context="eval job e1/r1")
+
+    assert save_context is not None
+    assert manager.entered is False  # built lazily, not yet entered
+
+    async with save_context():
+        pass
+
+    assert manager.calls == ["eval job e1/r1"]
+
+
+def test_save_context_notifies_background_sync():
+    manager = _FakeManager()
+    bg_sync = MagicMock()
+    with _patch_resolution(
+        project_path=PROJECT_PATH,
+        config=_auto_config(),
+        manager=manager,
+        bg_sync=bg_sync,
+    ):
+        save_context = save_context_for_project(PROJECT_ID, context="ctx")
+
+    assert save_context is not None
+    bg_sync.notify_request.assert_called_once()
+
+
+def test_save_context_no_background_sync_is_fine():
+    manager = _FakeManager()
+    with _patch_resolution(
+        project_path=PROJECT_PATH,
+        config=_auto_config(),
+        manager=manager,
+        bg_sync=None,
+    ):
+        save_context = save_context_for_project(PROJECT_ID, context="ctx")
+
+    assert save_context is not None
+
+
+# -- error propagation -------------------------------------------------------
+
+
+def test_propagates_when_config_lookup_raises():
+    # A corrupt/raising config lookup must surface (failing the job) rather than
+    # be swallowed to None, which would silently skip commits for an auto-sync
+    # project — the very bug this resolver exists to prevent.
+    with (
+        patch(
+            "app.desktop.git_sync.save_context.project_path_from_id",
+            return_value=PROJECT_PATH,
+        ),
+        patch(
+            "app.desktop.git_sync.save_context.get_git_sync_config",
+            side_effect=RuntimeError("corrupt config"),
+        ),
+    ):
+        with pytest.raises(RuntimeError, match="corrupt config"):
+            get_manager_for_project(PROJECT_ID)
+        with pytest.raises(RuntimeError, match="corrupt config"):
+            save_context_for_project(PROJECT_ID, context="ctx")