fix: keep tests and one-off commands off other checkouts' Open WebUI

config/tests/test_runtime.py

@@ -8,6 +8,7 @@
 
 import os
 import unittest
+from unittest import mock
 
 from config import runtime
 from config.runtime import (
@@ -180,7 +181,18 @@ def test_chat_disable_flag_variants(self):
 
 
 class ChatPrecedenceTests(unittest.TestCase):
-    """Flag > explicit env value > mode default (chat on for single-process)."""
+    """Flag > explicit env value > mode default (chat on for single-process).
+
+    Each test starts with SIMPLEAUDIT_CHAT unset (the defaults are what is under
+    test) and the whole environment is restored afterwards, so pytest.ini's
+    SIMPLEAUDIT_CHAT=off guard survives for the tests that run after these.
+    """
+
+    def setUp(self):
+        env = mock.patch.dict(os.environ)
+        env.start()
+        self.addCleanup(env.stop)
+        os.environ.pop("SIMPLEAUDIT_CHAT", None)
 
     def test_mode_default_on_for_single_process(self):
         self.assertEqual(runtime.effective_chat_mode("dev"), "embedded")
@@ -196,10 +208,7 @@ def test_disable_flag_beats_env_and_default(self):
         # The flag is an explicit opt-out and wins over a .env that pins chat
         # on — otherwise --disable-chat would be a no-op on this machine.
         os.environ["SIMPLEAUDIT_CHAT"] = "embedded"
-        try:
-            self.assertEqual(runtime.effective_chat_mode("dev", disable_flag=True), "off")
-        finally:
-            os.environ.pop("SIMPLEAUDIT_CHAT", None)
+        self.assertEqual(runtime.effective_chat_mode("dev", disable_flag=True), "off")
 
     def test_disable_flag_beats_compose_default_on(self):
         # The flag wins over compose's on-default: off.
@@ -208,15 +217,9 @@ def test_disable_flag_beats_compose_default_on(self):
     def test_explicit_env_wins_over_default_without_flag(self):
         # No flag: an explicit env value is honored, on or off.
         os.environ["SIMPLEAUDIT_CHAT"] = "docker"
-        try:
-            self.assertEqual(runtime.effective_chat_mode("dev"), "docker")
-        finally:
-            os.environ.pop("SIMPLEAUDIT_CHAT", None)
+        self.assertEqual(runtime.effective_chat_mode("dev"), "docker")
         os.environ["SIMPLEAUDIT_CHAT"] = "off"
-        try:
-            self.assertEqual(runtime.effective_chat_mode("dev"), "off")
-        finally:
-            os.environ.pop("SIMPLEAUDIT_CHAT", None)
+        self.assertEqual(runtime.effective_chat_mode("dev"), "off")
 
 
 if __name__ == "__main__":

conftest.py

@@ -7,8 +7,49 @@
 slow"`` works the same way ``manage.py test --exclude-tag slow`` does.
 """
 
+import os
+
 import pytest
 
+from simpleaudit_studio import no_live_chat
+
+
+def pytest_configure(config):
+    # No test may reach a developer's live Open WebUI (an unmocked agent save
+    # used to overwrite their real workspace models). See no_live_chat.py.
+    no_live_chat.install()
+
+
+def pytest_unconfigure(config):
+    no_live_chat.uninstall()
+
+
+#: pytest.ini's SIMPLEAUDIT_CHAT=off, as seen before any test ran.
+_CHAT_ENV = os.environ.get("SIMPLEAUDIT_CHAT")
+
+
+@pytest.hookimpl(wrapper=True)
+def pytest_runtest_teardown(item, nextitem):
+    """A test that changes SIMPLEAUDIT_CHAT must restore it (``mock.patch.dict``).
+
+    Popping it in a ``finally`` used to delete pytest.ini's ``off`` for every
+    later test in the worker. Checked after every fixture (monkeypatch
+    included) has torn down; put it back and fail the test that leaked.
+    """
+    result = yield
+    leaked = os.environ.get("SIMPLEAUDIT_CHAT")
+    if leaked == _CHAT_ENV:
+        return result
+    if _CHAT_ENV is None:
+        os.environ.pop("SIMPLEAUDIT_CHAT", None)
+    else:
+        os.environ["SIMPLEAUDIT_CHAT"] = _CHAT_ENV
+    pytest.fail(
+        f"Test left SIMPLEAUDIT_CHAT={leaked!r} (expected {_CHAT_ENV!r}); "
+        "restore it with mock.patch.dict(os.environ) or monkeypatch.",
+        pytrace=False,
+    )
+
 
 @pytest.fixture(autouse=True)
 def _synchronous_otlp_span_writes(settings):

docs/chat.md

@@ -183,6 +183,24 @@ this cookie" (`SIMPLEAUDIT_CHAT_IDENTITY_TTL`, default 5s, 0 to disable). A page
 load pulls dozens of assets, and without it each one would ask Django again; a
 sign-out takes effect within that window.
 
+### Several checkouts on one machine
+
+`manage.py dev` gives each checkout its own ports (the main checkout's Open WebUI
+on `:8080`, a worktree's on `:8180`, `:8280`, ...) and exports
+`SIMPLEAUDIT_CHAT_UPSTREAM` to its own processes only. A one-off command
+(`manage.py shell`, `seed_platform`, `sync_chat_startup`, ...) reads the same
+`.env` but not that variable, so `manage.py` resolves it first: with chat on and
+no explicit `SIMPLEAUDIT_CHAT_UPSTREAM`, it uses this checkout's running dev
+stack's Open WebUI, or, when `:8080` belongs to another checkout's running stack,
+turns chat off for that command and says so. Agent ids (`studio.agent-<pk>`)
+collide between databases, so syncing into another checkout's Open WebUI would
+overwrite its workspace models.
+
+The test suite never talks to a live Open WebUI: `pytest.ini` (and `manage.py
+test`) force `SIMPLEAUDIT_CHAT=off`, chat tests enable it with mocks, and
+`simpleaudit_studio/no_live_chat.py` refuses every test connection to the chat
+upstream or to loopback `:8080`.
+
 Ollama is switched off (nothing in a Studio deployment serves it). Left on, Open
 WebUI polls it on every page load — a failing request in the browser console each
 time — and shows an empty Ollama section in its connection settings. The

infra/tests/test_connection_sharing.py

@@ -226,6 +226,12 @@ def test_owner_can_change_sharing_level(self):
 class PickerSharingTests(TestCase):
     """The New Experiment picker offers shared models with a sharing label."""
 
+    @staticmethod
+    def _picker_checkbox(model_id):
+        # Only the model pickers' checkboxes (partials/model_pickers.html): a
+        # bare value="<pk>" also matches scenario sets, prompt versions, etc.
+        return rf'name="\w+_model" value="{model_id}"'
+
     def setUp(self):
         self.owner_ws = ProjectFactory(name="Owner WS")
         self.consumer_ws = ProjectFactory(name="Consumer WS")
@@ -246,7 +252,7 @@ def test_picker_lists_shared_model_with_label(self):
         self.assertContains(page, "Shared GPT")
         self.assertContains(page, "shared</span>")
         # The model checkbox is present and usable.
-        self.assertContains(page, f'value="{self.model.id}"')
+        self.assertRegex(page.content.decode(), self._picker_checkbox(self.model.id))
 
     def test_private_connection_not_in_picker(self):
         private_conn = ModelConnectionFactory(
@@ -256,29 +262,29 @@ def test_private_connection_not_in_picker(self):
         private_model = RegisteredModelFactory(connection=private_conn, project=self.owner_ws, display_name="Secret", model_id="secret")
         page = self.client.get("/experiments/new/")
         self.assertNotContains(page, "Private Conn")
-        self.assertNotContains(page, f'value="{private_model.id}"')
+        self.assertNotRegex(page.content.decode(), self._picker_checkbox(private_model.id))
 
     def test_enabled_connection_still_in_picker(self):
         # Guards against over-filtering: a normal enabled connection is offered.
         page = self.client.get("/experiments/new/")
         self.assertContains(page, "Shared GPT")
-        self.assertContains(page, f'value="{self.model.id}"')
+        self.assertRegex(page.content.decode(), self._picker_checkbox(self.model.id))
 
     def test_disabled_connection_hidden_from_picker(self):
         # A disabled connection's models can't be picked for new runs.
         self.conn.enabled = False
         self.conn.save()
         page = self.client.get("/experiments/new/")
         self.assertNotContains(page, "Shared GPT")
-        self.assertNotContains(page, f'value="{self.model.id}"')
+        self.assertNotRegex(page.content.decode(), self._picker_checkbox(self.model.id))
 
     def test_disabled_shared_connection_hidden_from_picker(self):
         # A disabled *shared* connection is also hidden from the consumer workspace.
         self.conn.enabled = False
         self.conn.save()
         page = self.client.get("/experiments/new/")
         self.assertNotContains(page, "Shared GPT")
-        self.assertNotContains(page, f'value="{self.model.id}"')
+        self.assertNotRegex(page.content.decode(), self._picker_checkbox(self.model.id))
 
 
 class PingSharedConnectionTests(TestCase):

manage.py

@@ -71,6 +71,24 @@ def main() -> None:
             # OTLP tracing for OWUI spans must also be set before django.setup()
             # because chat/config.py freezes OTLP_ENABLED at import time.
             os.environ.setdefault("SIMPLEAUDIT_CHAT_OTLP", "true")
+    if command == "test":
+        # Tests never reach a live Open WebUI: chat off whatever `.env` says
+        # (chat tests switch it on themselves, with mocks) and every connect to
+        # the chat upstream refused. pytest does the same via pytest.ini and
+        # conftest.py.
+        os.environ["SIMPLEAUDIT_CHAT"] = "off"
+        from simpleaudit_studio import no_live_chat
+
+        no_live_chat.install()
+    elif command and not single_process:
+        # A one-off command must not sync into another checkout's Open WebUI.
+        from pathlib import Path
+
+        from simpleaudit_studio.lifecycle import route_one_off_chat
+
+        note = route_one_off_chat(Path(__file__).resolve().parent)
+        if note:
+            print(note, file=sys.stderr)
     # `dev` / `dev_server` are local dev commands — they imply DEBUG so the
     # production hardening block (SECURE_SSL_REDIRECT, HSTS, secure cookies)
     # stays off.

pytest.ini

@@ -5,11 +5,15 @@ DJANGO_SETTINGS_MODULE = config.settings
 ; e2e/test_ui.py is a standalone Playwright script (not a pytest test) that
 ; sys.exit()s at import, so it must not be collected.
 norecursedirs = e2e node_modules .venv static staticfiles
+; SIMPLEAUDIT_CHAT=off: chat stays off whatever the shell exports (a sourced
+; dev .env sets it); chat tests turn it on with mock.patch("chat.config.ENABLED").
+; conftest.py also refuses any connection to a live Open WebUI.
 env =
     SIMPLEAUDIT_LOCAL_SQLITE=1
     DJANGO_SECRET_KEY=test-secret-key-not-change-me
     DJANGO_DEBUG=false
     DJANGO_ALLOWED_HOSTS=*
+    SIMPLEAUDIT_CHAT=off
 ; Django @tag("...") values are mirrored onto pytest items by conftest.py, so
 ; -m selection works the same as manage.py test --exclude-tag.
 markers =

simpleaudit_studio/lifecycle.py

@@ -369,6 +369,50 @@ def _claimed_by_others(instance: Instance) -> tuple[set[int], set[str]]:
     return ports, data
 
 
+#: chat.config's default Open WebUI port (``http://127.0.0.1:8080/chat``): the
+#: chat port of a ``manage.py dev`` started from the main checkout on 8000.
+DEFAULT_CHAT_PORT = 8080
+#: Mirrors ``chat.config.DISABLED_VALUES``. Not imported from there: importing
+#: chat.config freezes its settings before this function can adjust them.
+_CHAT_DISABLED_VALUES = frozenset({"", "off", "disabled", "disable", "false", "no", "none", "0"})
+
+
+def route_one_off_chat(root: Path) -> str | None:
+    """Keep a one-off ``manage.py`` command on this checkout's own Open WebUI.
+
+    ``manage.py dev`` gives each checkout its own Open WebUI port and exports
+    ``SIMPLEAUDIT_CHAT_UPSTREAM`` to its children. A one-off command (``shell``,
+    ``seed_platform``, ``sync_chat_startup``...) loads the same ``.env`` but not
+    that variable, so chat.config falls back to ``127.0.0.1:8080``, which may be
+    another checkout's Open WebUI: an agent sync would then write this
+    checkout's rows over that instance's workspace models (ids collide).
+
+    With chat enabled and no explicit upstream: use this checkout's live dev
+    instance's chat port; failing that, turn chat off for the command when the
+    default port belongs to another live instance. Returns a note to print when
+    chat was turned off, else ``None``. Must run before ``django.setup()``.
+    """
+    if os.environ.get("SIMPLEAUDIT_CHAT_UPSTREAM"):
+        return None  # explicit: compose, or the dev stack's own children
+    if (os.environ.get("SIMPLEAUDIT_CHAT") or "").strip().lower() in _CHAT_DISABLED_VALUES:
+        return None
+    instance = Instance("dev", root)
+    own = instance.record()
+    chat_port = (own or {}).get("ports", {}).get("chat") if _live(own) else None
+    if chat_port:
+        os.environ["SIMPLEAUDIT_CHAT_UPSTREAM"] = f"http://127.0.0.1:{chat_port}/chat"
+        return None
+    claimed, _ = _claimed_by_others(instance)
+    if DEFAULT_CHAT_PORT not in claimed:
+        return None
+    os.environ["SIMPLEAUDIT_CHAT"] = "off"
+    return (
+        f"Chat is off for this command: the Open WebUI on 127.0.0.1:{DEFAULT_CHAT_PORT} "
+        "belongs to another checkout's `manage.py dev`. Start this checkout's own "
+        "stack (`uv run manage.py dev`) or set SIMPLEAUDIT_CHAT_UPSTREAM explicitly."
+    )
+
+
 def start(instance: Instance, *, requested_port: int = 8000, explicit_port: bool = False) -> bool:
     """Register a local stack and export ports before any Django import.
 

simpleaudit_studio/no_live_chat.py

@@ -0,0 +1,118 @@
+"""Keep test runs away from a developer's live Open WebUI.
+
+Chat-backed tests switch ``chat.config.ENABLED`` on while ``chat.config.UPSTREAM``
+keeps its default, ``http://127.0.0.1:8080/chat``. That is where a local
+``manage.py dev`` serves Open WebUI, so an unmocked push from a test (an agent
+save, a connection sync, a sign-in) used to overwrite that instance's real
+workspace models with test rows: test agent pk 1 becomes ``studio.agent-1``, the
+same id as the developer's real agent.
+
+``install()`` patches ``socket.socket`` so that every TCP connect to the chat
+upstream, or to any loopback address on Open WebUI's port, is refused, exactly
+as it is in CI where nothing listens there. Ports the test process listens on
+itself (stub servers) stay reachable. Both test runners install it: pytest via
+``conftest.py`` and ``manage.py test`` via ``manage.py``.
+
+Dependency-free on purpose: it runs before ``django.setup()``.
+"""
+from __future__ import annotations
+
+import ipaddress
+import socket
+from urllib.parse import urlsplit
+
+#: Open WebUI's port in the default upstream (embedded and compose alike).
+OPENWEBUI_PORT = 8080
+_LOCAL_HOSTNAMES = frozenset({"localhost", "localhost.localdomain"})
+
+_real_connect = socket.socket.connect
+_real_connect_ex = socket.socket.connect_ex
+_real_listen = socket.socket.listen
+
+#: Every chat-upstream address a connect was refused for, in order.
+refused: list[str] = []
+#: Ports this process listens on itself: those reach the test's own fake.
+_own_listener_ports: set[int] = set()
+
+
+class LiveOpenWebUIRefused(ConnectionRefusedError):
+    """A test tried to reach a real Open WebUI; mock ``chat.api.ChatAPI`` instead."""
+
+
+def _is_loopback(host: str) -> bool:
+    if host.lower() in _LOCAL_HOSTNAMES:
+        return True
+    try:
+        return ipaddress.ip_address(host.split("%", 1)[0]).is_loopback
+    except ValueError:
+        return False
+
+
+def _upstream_host_port() -> tuple[str, int] | None:
+    try:
+        from chat import config as chat_config
+
+        parts = urlsplit(chat_config.UPSTREAM)
+        port = parts.port or (443 if parts.scheme == "https" else 80)
+    except Exception:  # noqa: BLE001 - the guard itself must never break a test
+        return None
+    return (parts.hostname or "").lower(), port
+
+
+def is_chat_upstream(address: object) -> bool:
+    """Whether connecting to ``address`` could reach a live Open WebUI."""
+    if not (isinstance(address, tuple) and len(address) >= 2):
+        return False  # AF_UNIX paths and the like
+    host, port = str(address[0]).lower(), address[1]
+    if port in _own_listener_ports:
+        return False
+    if port == OPENWEBUI_PORT and _is_loopback(host):
+        return True
+    upstream = _upstream_host_port()
+    if upstream is None or port != upstream[1]:
+        return False
+    upstream_host = upstream[0]
+    return host == upstream_host or (_is_loopback(host) and _is_loopback(upstream_host))
+
+
+def _refuse(sock: socket.socket, address: object) -> None:
+    if sock.family in (socket.AF_INET, socket.AF_INET6) and is_chat_upstream(address):
+        target = f"{address[0]}:{address[1]}"
+        refused.append(target)
+        raise LiveOpenWebUIRefused(
+            f"refused test connection to a live Open WebUI at {target}; "
+            "mock chat.api.ChatAPI or httpx instead (simpleaudit_studio/no_live_chat.py)"
+        )
+
+
+def _guarded_connect(self, address):
+    _refuse(self, address)
+    return _real_connect(self, address)
+
+
+def _guarded_connect_ex(self, address):
+    _refuse(self, address)
+    return _real_connect_ex(self, address)
+
+
+def _recording_listen(self, *args):
+    result = _real_listen(self, *args)
+    try:
+        address = self.getsockname()
+    except OSError:
+        return result
+    if isinstance(address, tuple) and len(address) >= 2:
+        _own_listener_ports.add(address[1])
+    return result
+
+
+def install() -> None:
+    socket.socket.connect = _guarded_connect
+    socket.socket.connect_ex = _guarded_connect_ex
+    socket.socket.listen = _recording_listen
+
+
+def uninstall() -> None:
+    socket.socket.connect = _real_connect
+    socket.socket.connect_ex = _real_connect_ex
+    socket.socket.listen = _real_listen