Automations: run each as its OWNER profile in the background (part 1 of per-profile sync)

Background automations had no session, so get_current_profile_id() fell back to
admin (1) — wrong for a non-admin's scheduled job. Now the engine declares the
automation's owner around handler execution via a contextvar
(core/profile_context.py), and get_current_profile_id() consults it only when
there's NO web request. So:
- a real logged-in request always wins (foreground unchanged),
- admin + system automations are profile 1 → resolve to admin exactly as before
  (the 8 admin-owned auto-sync pipelines behave identically),
- only non-admin-owned automations gain their correct identity, deep through the
  whole call chain (incl. the per-profile client resolvers) — no threading
  profile_id through dozens of signatures.

Reset in a finally so a pooled thread can't leak the override to the next job.

Tests: contextvar set/reset/nested; get_current_profile_id honours the override
only outside a request (a real session still wins); and end-to-end — the engine
runs a non-admin automation as profile 4, an admin one as 1, an explicit trigger
profile overrides the owner, and the context resets even when the handler raises.
27 + 4 tests pass.

Part 2 (next): point the sync handlers' source-playlist READ at
get_spotify_client_for_profile so a non-admin's auto-sync pulls THEIR playlist.

core/automation_engine.py

@@ -635,6 +635,11 @@ class AutomationEngine:
         action_config['_automation_name'] = auto.get('name', '')
         if profile_id is not None:
             action_config['_profile_id'] = profile_id
+        # The profile this run acts AS: an explicit trigger profile, else the
+        # automation's owner, else admin. System + admin automations are
+        # profile 1, so this is a no-op for them — only non-admin-owned
+        # automations gain their correct identity in the background.
+        _effective_profile_id = profile_id if profile_id is not None else (auto.get('profile_id') or 1)
 
         # Action delay (skipped for manual run_now)
         delay_minutes = action_config.get('delay', 0)
@@ -681,9 +686,14 @@ class AutomationEngine:
             except Exception as e:
                 logger.debug("scheduled progress init: %s", e)
 
-        # Execute the action
+        # Execute the action under the owner's profile so get_current_profile_id()
+        # (and the per-profile clients it resolves) act as the automation's owner
+        # in the background, not admin. Reset in finally so a pooled thread can't
+        # leak the override to the next job.
         error = None
         result = {}
+        from core.profile_context import set_background_profile, reset_background_profile
+        _bg_token = set_background_profile(_effective_profile_id)
         try:
             result = handler_info['handler'](action_config) or {}
             logger.info(f"Automation '{auto['name']}' (id={automation_id}) executed: {result.get('status', 'ok')}")
@@ -702,6 +712,8 @@ class AutomationEngine:
             error = str(e)
             result = {'status': 'error', 'error': error}
             logger.error(f"Automation '{auto['name']}' (id={automation_id}) failed: {e}")
+        finally:
+            reset_background_profile(_bg_token)
 
         # Finalize progress tracking
         if self._progress_finish_fn:

core/profile_context.py

@@ -0,0 +1,45 @@
+"""Background profile context.
+
+Work that runs OUTSIDE a web request — the automation engine, scheduled jobs —
+has no Flask session, so ``get_current_profile_id()`` falls back to admin
+(profile 1). That's wrong for an automation owned by a non-admin: their
+playlist pull, their per-profile writes, should act as THEM.
+
+This lets the engine declare "the work below is running for profile X" around a
+unit of background work (set/reset in a try/finally). ``get_current_profile_id``
+consults it only when there's no real request — so an actual logged-in session
+always wins, and nothing changes for foreground/admin paths. Built on a
+``ContextVar`` so the value is scoped to the running call and reset cleanly,
+even on thread-pool reuse.
+"""
+
+from __future__ import annotations
+
+import contextvars
+
+_background_profile_id: "contextvars.ContextVar[int | None]" = contextvars.ContextVar(
+    "background_profile_id", default=None
+)
+
+
+def set_background_profile(profile_id):
+    """Declare the profile for the current background unit of work. Returns a
+    token to pass to reset_background_profile (use in try/finally)."""
+    return _background_profile_id.set(profile_id)
+
+
+def reset_background_profile(token) -> None:
+    """Restore the previous background profile (clears the override)."""
+    try:
+        _background_profile_id.reset(token)
+    except Exception:
+        # Token from a different context — clear to the default rather than leak.
+        _background_profile_id.set(None)
+
+
+def get_background_profile():
+    """The background profile in effect, or None if none is set."""
+    return _background_profile_id.get()
+
+
+__all__ = ["set_background_profile", "reset_background_profile", "get_background_profile"]

tests/automation/test_engine_profile_context.py

@@ -0,0 +1,64 @@
+"""The engine runs each automation AS its owner profile (per-profile automations).
+
+A non-admin's scheduled job must execute with their profile in the background
+context, so get_current_profile_id() / the per-profile clients act as them — not
+admin. Admin/system automations (profile 1) are unchanged. The context must be
+reset after every run so a pooled thread can't leak it.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from core.automation_engine import AutomationEngine
+from core.profile_context import get_background_profile
+
+
+def _engine(owner_profile_id):
+    db = MagicMock()
+    db.get_automation.return_value = {
+        'id': 1, 'name': 'Auto-Sync', 'enabled': True,
+        'action_type': 'sync_playlist', 'action_config': '{}',
+        'trigger_type': 'interval_hours', 'trigger_config': '{"hours": 1}',
+        'profile_id': owner_profile_id,
+    }
+    db.update_automation_run = MagicMock(return_value=True)
+    eng = AutomationEngine(db)
+    eng._running = True
+    seen = {}
+    eng._action_handlers['sync_playlist'] = {
+        'handler': lambda config: seen.update(profile=get_background_profile()) or {'status': 'completed'},
+        'guard': None,
+    }
+    return eng, seen
+
+
+def test_nonadmin_owned_automation_runs_as_owner():
+    eng, seen = _engine(owner_profile_id=4)
+    eng.run_automation(1, skip_delay=True)
+    assert seen['profile'] == 4                 # handler ran AS profile 4
+    assert get_background_profile() is None      # reset after the run
+
+
+def test_admin_owned_automation_runs_as_admin():
+    eng, seen = _engine(owner_profile_id=1)
+    eng.run_automation(1, skip_delay=True)
+    assert seen['profile'] == 1                 # unchanged for admin/system
+    assert get_background_profile() is None
+
+
+def test_explicit_trigger_profile_overrides_owner():
+    # A manual trigger (run_automation(profile_id=...)) wins over the owner.
+    eng, seen = _engine(owner_profile_id=4)
+    eng.run_automation(1, skip_delay=True, profile_id=9)
+    assert seen['profile'] == 9
+
+
+def test_context_reset_even_when_handler_raises():
+    eng, _ = _engine(owner_profile_id=4)
+    eng._action_handlers['sync_playlist'] = {
+        'handler': lambda config: (_ for _ in ()).throw(RuntimeError('boom')),
+        'guard': None,
+    }
+    eng.run_automation(1, skip_delay=True)       # error is caught + stored
+    assert get_background_profile() is None       # finally reset it

tests/test_credentials_endpoints.py

@@ -286,3 +286,32 @@ def test_listenbrainz_connection_status_and_disconnect(client, nonadmin_profile)
     # disconnect via the generic endpoint
     assert client.post('/api/profiles/me/connections/listenbrainz/disconnect').get_json()['success']
     assert client.get('/api/profiles/me/connections').get_json()['connections']['listenbrainz']['connected'] is False
+
+
+# ── Background profile context drives get_current_profile_id() (part 1) ────────
+
+def test_background_profile_override_when_no_request():
+    # Outside a web request, get_current_profile_id() honours the engine's
+    # background override; admin (default) and cleared state stay profile 1.
+    from core.profile_context import set_background_profile, reset_background_profile
+    assert web_server.get_current_profile_id() == 1     # no override → admin
+    tok = set_background_profile(7)
+    try:
+        assert web_server.get_current_profile_id() == 7  # acts as the owner
+    finally:
+        reset_background_profile(tok)
+    assert web_server.get_current_profile_id() == 1     # reset → admin
+
+
+def test_real_session_still_wins_over_background(client, nonadmin_profile):
+    # A genuine request's session profile must override any background context.
+    from core.profile_context import set_background_profile, reset_background_profile
+    with client.session_transaction() as sess:
+        sess['profile_id'] = nonadmin_profile
+    tok = set_background_profile(999)  # a bogus background override
+    try:
+        # the request resolves to the SESSION profile, not the background one
+        body = client.get('/api/profiles/me/connections').get_json()
+        assert body['is_admin'] is False  # it's the non-admin session, not 999/admin
+    finally:
+        reset_background_profile(tok)

tests/test_profile_context.py

@@ -0,0 +1,31 @@
+"""Background profile context (per-profile automations).
+
+Lets background work (the automation engine) declare which profile it's acting
+for, so get_current_profile_id() resolves to the automation's OWNER instead of
+admin when there's no web request. A real request must still win; admin/system
+(profile 1) and no-override must stay admin.
+"""
+
+from __future__ import annotations
+
+from core.profile_context import (
+    set_background_profile, reset_background_profile, get_background_profile,
+)
+
+
+def test_set_reset_roundtrip():
+    assert get_background_profile() is None
+    tok = set_background_profile(5)
+    assert get_background_profile() == 5
+    reset_background_profile(tok)
+    assert get_background_profile() is None
+
+
+def test_nested_set_reset():
+    t1 = set_background_profile(3)
+    t2 = set_background_profile(1)          # e.g. an admin/system sub-run
+    assert get_background_profile() == 1
+    reset_background_profile(t2)
+    assert get_background_profile() == 3    # back to the outer profile
+    reset_background_profile(t1)
+    assert get_background_profile() is None  # fully cleared (no leak)

web_server.py

@@ -529,11 +529,19 @@ def get_current_profile_id() -> int:
     scanner) have no request context, so `g.profile_id` raises
     `RuntimeError("Working outside of application context")` rather
     than `AttributeError`. Catch both so non-request callers degrade
-    to the admin profile instead of crashing the handler."""
+    to the admin profile instead of crashing the handler.
+
+    A real web request always wins. Only when there's NO request do we honour a
+    background-profile override (set by the automation engine to the automation's
+    owner) — so a non-admin's scheduled job acts as them, while admin/system jobs
+    (profile 1) and anything with no override resolve to admin exactly as before."""
     try:
         return g.profile_id
     except (AttributeError, RuntimeError):
-        return 1
+        pass
+    from core.profile_context import get_background_profile
+    pid = get_background_profile()
+    return pid if pid is not None else 1
 
 
 def admin_only(view_fn):