From 2f141ccee9d57e2c35de37ef355d34e0005d1b02 Mon Sep 17 00:00:00 2001
From: Kai <300677314+kai-openswarm@users.noreply.github.com>
Date: Mon, 17 Aug 2026 17:32:02 -0700
Subject: [PATCH 1/6] renderer: shared/config and backendConnection are
 import-safe without a window, so reducer tests run under node:test

Both modules touched window at import time (port/host derivation, the fetch
interceptor install, the debug handle), so any node:test file that imports a
reducer importing API_BASE died with 'window is not defined' before its first
assertion; fetchSessionsStrip.test.ts has been red that way since the
resilience work landed, unnoticed because nothing runs these tests in CI. In a
renderer (window present) nothing changes: same port/host, same interceptor,
same handle. Without one the module answers with the defaults and installs
nothing.
---
 frontend/src/shared/backendConnection.ts |  5 ++++-
 frontend/src/shared/config.ts            | 13 +++++++++----
 2 files changed, 13 insertions(+), 5 deletions(-)

diff --git a/frontend/src/shared/backendConnection.ts b/frontend/src/shared/backendConnection.ts
index e93237e10..c410bfc04 100644
--- a/frontend/src/shared/backendConnection.ts
+++ b/frontend/src/shared/backendConnection.ts
@@ -80,4 +80,7 @@ export function noteRequestStalled(): void {
 }
 
 // Harness/debug handle: lets a live session (CDP, support) read the signal without a store import.
-(window as unknown as { __OSW_CONN?: object }).__OSW_CONN = { backendReachable, onBackendReachability };
+// Guarded: reducers that import this module also run under node:test, where there is no window.
+if (typeof window !== 'undefined') {
+  (window as unknown as { __OSW_CONN?: object }).__OSW_CONN = { backendReachable, onBackendReachability };
+}
diff --git a/frontend/src/shared/config.ts b/frontend/src/shared/config.ts
index c0aec765f..f92bbe614 100644
--- a/frontend/src/shared/config.ts
+++ b/frontend/src/shared/config.ts
@@ -1,6 +1,9 @@
 import { noteBackendFailure, noteBackendSuccess, noteRequestStalled, setBackendProber } from '@/shared/backendConnection';
 
-const _w = window as any;
+// Import-safe outside a renderer: reducers that import API_BASE also run under node:test, where there
+// is no window. The module then answers with the defaults and installs nothing.
+const hasWindow = typeof window !== 'undefined';
+const _w = (hasWindow ? window : {}) as any;
 // Prefer the preload-injected port; if it's missing (preload raced the backend port being picked), re-query the live value before falling back to 8324. The bare 8324 guess is wrong on any machine where the backend landed on a fallback port (e.g. 8324 was held by a leftover backend); see the self-heal below.
 const port =
   _w.__OPENSWARM_PORT__ ||
@@ -8,7 +11,7 @@ const port =
     ? _w.openswarm.getBackendPortLive()
     : 0) ||
   8324;
-const host = window.location.hostname || 'localhost';
+const host = (hasWindow && window.location.hostname) || 'localhost';
 
 export const API_BASE = `http://${host}:${port}/api`;
 export const WS_BASE = `ws://${host}:${port}`;
@@ -217,5 +220,7 @@ function _installAuthFetchInterceptor() {
   };
 }
 
-_installAuthFetchInterceptor();
-ensureAuthToken();
+if (hasWindow) {
+  _installAuthFetchInterceptor();
+  ensureAuthToken();
+}

From 71475b66cd157cc33dee5c3801616cf17e182442 Mon Sep 17 00:00:00 2001
From: Kai <300677314+kai-openswarm@users.noreply.github.com>
Date: Mon, 17 Aug 2026 17:32:02 -0700
Subject: [PATCH 2/6] ci: run the backend, frontend and edge test suites on
 every pull request

Nothing ran any of them in CI: the 235-file backend pytest suite, the 22
renderer node:test files and the edge suite were run by hand, one file at a
time, so a regression only surfaced when someone happened to run the right
one. Three small workflows, hosted ubuntu, path-filtered, read-only token:

- backend-tests: pytest on Python 3.13 from the locked requirements, plus a
  completion assertion (junit testcase count == collect-only count) so a test
  process that dies mid-run can never read as green
- frontend-tests: tsc --noEmit + node:test via tsx over src/**/*.test.ts(x),
  through frontend/scripts/run-tests.mjs (the runner the tests already name)
- edge-tests: pytest for openswarm-edge

All three are green on the current tree: 2951 backend tests, 143 frontend
tests across 22 files, 14 edge tests.
---
 .github/workflows/backend-tests.yml  | 65 ++++++++++++++++++++++++++++
 .github/workflows/edge-tests.yml     | 44 +++++++++++++++++++
 .github/workflows/frontend-tests.yml | 43 ++++++++++++++++++
 frontend/scripts/run-tests.mjs       | 29 +++++++++++++
 4 files changed, 181 insertions(+)
 create mode 100644 .github/workflows/backend-tests.yml
 create mode 100644 .github/workflows/edge-tests.yml
 create mode 100644 .github/workflows/frontend-tests.yml
 create mode 100644 frontend/scripts/run-tests.mjs

diff --git a/.github/workflows/backend-tests.yml b/.github/workflows/backend-tests.yml
new file mode 100644
index 000000000..adb758a23
--- /dev/null
+++ b/.github/workflows/backend-tests.yml
@@ -0,0 +1,65 @@
+name: backend-tests
+
+# The backend pytest suite, run on every pull request and on pushes to the mainline branches. Until
+# now nothing ran it in CI, so a regression only surfaced when someone ran it by hand.
+on:
+  pull_request:
+    paths:
+      - 'backend/**'
+      - '.github/workflows/backend-tests.yml'
+  push:
+    branches: [main, dev]
+    paths:
+      - 'backend/**'
+      - '.github/workflows/backend-tests.yml'
+  workflow_dispatch:
+
+# Runs checked-out project code on pull_request: the token stays read-only.
+permissions:
+  contents: read
+
+concurrency:
+  group: backend-tests-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  pytest:
+    name: pytest (ubuntu)
+    runs-on: ubuntu-latest
+    timeout-minutes: 25
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.13'
+          cache: pip
+          cache-dependency-path: |
+            backend/requirements.lock
+            backend/requirements-dev.txt
+      - name: Install backend deps (locked runtime + dev)
+        run: |
+          python -m pip install --require-hashes --only-binary=:all: -r backend/requirements.lock
+          python -m pip install -r backend/requirements-dev.txt
+      # Two numbers that must agree. A test process that dies mid-run can exit 0 with no summary
+      # (a hard-exit shutdown path did exactly that once and silently skipped ~42% of the suite), so
+      # the job also asserts that every collected test was actually run: junit testcase count ==
+      # collect-only count. Green then means green, not "green as far as it got".
+      - name: Count the suite
+        run: |
+          python -m pytest backend/tests --ignore=backend/tests/formal --collect-only -q -p no:cacheprovider \
+            | tail -1 | tee collected.txt
+      - name: Run the backend suite
+        run: |
+          python -m pytest backend/tests --ignore=backend/tests/formal -q -p no:cacheprovider \
+            --junitxml "${RUNNER_TEMP}/pytest.xml"
+      - name: Every collected test ran
+        run: |
+          python - "${RUNNER_TEMP}/pytest.xml" collected.txt <<'PY'
+          import re, sys, xml.etree.ElementTree as ET
+          ran = sum(1 for _ in ET.parse(sys.argv[1]).getroot().iter('testcase'))
+          m = re.search(r'(\d+) tests? collected', open(sys.argv[2]).read())
+          collected = int(m.group(1)) if m else -1
+          print(f'collected={collected} ran={ran}')
+          if collected < 1 or ran != collected:
+              sys.exit(f'FAIL: {ran} of {collected} collected tests reached the report; the run was truncated')
+          PY
diff --git a/.github/workflows/edge-tests.yml b/.github/workflows/edge-tests.yml
new file mode 100644
index 000000000..82ef10d9d
--- /dev/null
+++ b/.github/workflows/edge-tests.yml
@@ -0,0 +1,44 @@
+name: edge-tests
+
+# The openswarm-edge pytest suite, on every pull request that touches it and on pushes to the
+# mainline branches.
+on:
+  pull_request:
+    paths:
+      - 'openswarm-edge/**'
+      - '.github/workflows/edge-tests.yml'
+  push:
+    branches: [main, dev]
+    paths:
+      - 'openswarm-edge/**'
+      - '.github/workflows/edge-tests.yml'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: edge-tests-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  pytest:
+    name: pytest (openswarm-edge)
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    defaults:
+      run:
+        working-directory: openswarm-edge
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.13'
+          cache: pip
+          cache-dependency-path: openswarm-edge/requirements.txt
+      - name: Install edge deps
+        run: |
+          python -m pip install -r requirements.txt
+          python -m pip install pytest pytest-asyncio
+      - name: Run the edge suite
+        run: python -m pytest tests -q -p no:cacheprovider
diff --git a/.github/workflows/frontend-tests.yml b/.github/workflows/frontend-tests.yml
new file mode 100644
index 000000000..00578bb8f
--- /dev/null
+++ b/.github/workflows/frontend-tests.yml
@@ -0,0 +1,43 @@
+name: frontend-tests
+
+# Typecheck plus the renderer's node:test suite, on every pull request and on pushes to the mainline
+# branches. Until now neither ran in CI; the tests were run by hand, one file at a time.
+on:
+  pull_request:
+    paths:
+      - 'frontend/**'
+      - '.github/workflows/frontend-tests.yml'
+  push:
+    branches: [main, dev]
+    paths:
+      - 'frontend/**'
+      - '.github/workflows/frontend-tests.yml'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: frontend-tests-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  typecheck-and-tests:
+    name: tsc + node:test
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    defaults:
+      run:
+        working-directory: frontend
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: '20.18.1'
+          cache: npm
+          cache-dependency-path: frontend/package-lock.json
+      - run: npm ci
+      - name: Typecheck
+        run: npx tsc --noEmit -p tsconfig.json
+      - name: Unit tests (node:test via tsx)
+        run: node scripts/run-tests.mjs
diff --git a/frontend/scripts/run-tests.mjs b/frontend/scripts/run-tests.mjs
new file mode 100644
index 000000000..8fad6fe2d
--- /dev/null
+++ b/frontend/scripts/run-tests.mjs
@@ -0,0 +1,29 @@
+#!/usr/bin/env node
+// Runs every renderer unit test (src/**/*.test.ts, *.test.tsx) under node:test, with tsx doing the
+// TypeScript. One command for CI and for a dev machine: `node scripts/run-tests.mjs`, optionally
+// followed by file paths to run a subset. Exits non-zero if any test fails or nothing was found.
+import { spawnSync } from 'node:child_process';
+import { readdirSync, statSync } from 'node:fs';
+import { join, dirname } from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const root = join(dirname(fileURLToPath(import.meta.url)), '..');
+const testFile = (p) => /\.test\.tsx?$/.test(p);
+
+function walk(dir, out) {
+  for (const name of readdirSync(dir)) {
+    if (name === 'node_modules' || name === 'dist') continue;
+    const p = join(dir, name);
+    if (statSync(p).isDirectory()) walk(p, out);
+    else if (testFile(p)) out.push(p);
+  }
+  return out;
+}
+
+const files = process.argv.length > 2 ? process.argv.slice(2) : walk(join(root, 'src'), []).sort();
+if (files.length === 0) {
+  console.error('run-tests: no test files found under src/');
+  process.exit(1);
+}
+const result = spawnSync(process.execPath, ['--import', 'tsx', '--test', ...files], { cwd: root, stdio: 'inherit' });
+process.exit(result.status ?? 1);

From 941dfb73be0b97225654b868bd755e601ec17c51 Mon Sep 17 00:00:00 2001
From: Kai <300677314+kai-openswarm@users.noreply.github.com>
Date: Mon, 17 Aug 2026 19:07:12 -0700
Subject: [PATCH 3/6] ci: cap each backend test at 300s so a stall fails by
 name

Two of five hosted runs of the backend suite stalled at 99% until the job cap
with no summary and no junit: one test blocked forever on a bare
ws.receive_json() (fixed on its own in a separate change). A CI lane should
never depend on every test being unable to hang, so add pytest-timeout to the
dev requirements and run the suite with --timeout=300. On Linux the default
signal method fails just the offending test and the run continues, so the
report and the "every collected test ran" assertion stay meaningful.
---
 .github/workflows/backend-tests.yml | 6 ++++++
 backend/requirements-dev.txt        | 3 +++
 2 files changed, 9 insertions(+)

diff --git a/.github/workflows/backend-tests.yml b/.github/workflows/backend-tests.yml
index adb758a23..39972626f 100644
--- a/.github/workflows/backend-tests.yml
+++ b/.github/workflows/backend-tests.yml
@@ -48,11 +48,17 @@ jobs:
         run: |
           python -m pytest backend/tests --ignore=backend/tests/formal --collect-only -q -p no:cacheprovider \
             | tail -1 | tee collected.txt
+      # --timeout: a test that blocks forever (a bare ws.receive_json() waiting for an event that never
+      # comes, say) otherwise stalls the run at 99% until timeout-minutes with no summary and no junit.
+      # With the cap it fails by name, the rest of the suite runs, and the assertion below still holds.
       - name: Run the backend suite
         run: |
           python -m pytest backend/tests --ignore=backend/tests/formal -q -p no:cacheprovider \
+            --timeout=300 \
             --junitxml "${RUNNER_TEMP}/pytest.xml"
       - name: Every collected test ran
+        # Runs after a red suite too, so a failure report also says whether the run was complete.
+        if: ${{ !cancelled() }}
         run: |
           python - "${RUNNER_TEMP}/pytest.xml" collected.txt <<'PY'
           import re, sys, xml.etree.ElementTree as ET
diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt
index fe1535065..710b74dc3 100644
--- a/backend/requirements-dev.txt
+++ b/backend/requirements-dev.txt
@@ -9,6 +9,9 @@
 
 pytest==8.3.4
 pytest-asyncio==0.25.2
+# Per-test wall-clock cap for CI (see .github/workflows/backend-tests.yml): a test that
+# blocks forever fails by name instead of stalling the whole run until the job cap.
+pytest-timeout==2.4.0
 
 # Used by linter/lint.py (the vulture dead-code check). watchfiles, also
 # needed by lint.py, already comes in transitively via uvicorn[standard].

From 5ced7185c0cf4bc644d17b486b82d004d08fc7a3 Mon Sep 17 00:00:00 2001
From: Kai <300677314+kai-openswarm@users.noreply.github.com>
Date: Mon, 17 Aug 2026 19:05:42 -0700
Subject: [PATCH 4/6] tests: the WS end-to-end turn test cannot hang the suite

test_ws_endpoint_streams_a_full_turn_end_to_end read the socket with a bare
ws.receive_json() in a 40-iteration loop and broke only on the assistant reply.
When the loop ends early for any reason (fewer than 40 events, no reply), the
next receive blocks forever and the whole pytest run stalls at 99% until the
job cap. On hosted runners it does exactly that intermittently, on Linux and
Windows alike: the turn path's configure_provider_env decides whether 9Router
needs reviving from provider evidence earlier tests may leave behind, and that
revival spawns/installs the router behind a module-level asyncio.Lock; the
background turn-label aux call reaches the same machinery. Neither is part of
this test's contract ("SDK and WS auth mocked, everything else real").

Pin both out with monkeypatch, bound every receive at 5s (a regression now
fails this test instead of hanging the runner), and wait for the turn's
completed status before asserting on session.messages so the assertion cannot
race the loop's tail.
---
 backend/tests/test_ws_integration.py | 48 ++++++++++++++++++++++++++--
 1 file changed, 46 insertions(+), 2 deletions(-)

diff --git a/backend/tests/test_ws_integration.py b/backend/tests/test_ws_integration.py
index 741cc223f..33f86210e 100644
--- a/backend/tests/test_ws_integration.py
+++ b/backend/tests/test_ws_integration.py
@@ -6,6 +6,8 @@
 The SDK and WS auth are mocked; everything else is the real running stack."""
 
 import asyncio
+import queue
+import threading
 
 import pytest
 
@@ -17,6 +19,29 @@
 import backend.main as main_mod
 from backend.apps.agents.agent_manager import agent_manager
 from backend.apps.agents.core.models import AgentSession
+import backend.apps.agents.manager.run.RunOptions as run_options_mod
+
+
+def p_receive_json(ws, timeout: float = 5.0):
+    """ws.receive_json() blocks forever when the event never comes; a regression in the loop then
+    hangs the whole pytest run instead of failing this test. Bound every receive."""
+    out = queue.Queue(maxsize=1)
+
+    def p_recv():
+        try:
+            out.put((True, ws.receive_json()))
+        except BaseException as exc:
+            out.put((False, exc))
+
+    thread = threading.Thread(target=p_recv, daemon=True)
+    thread.start()
+    try:
+        ok, value = out.get(timeout=timeout)
+    except queue.Empty as exc:
+        raise AssertionError(f"timed out waiting for websocket event after {timeout}s") from exc
+    if ok:
+        return value
+    raise value
 
 
 def p_assistant():
@@ -33,6 +58,20 @@ def p_result():
 def test_ws_endpoint_streams_a_full_turn_end_to_end(monkeypatch):
     monkeypatch.setattr(main_mod, "p_ws_auth_ok", lambda ws: True, raising=True)
 
+    # The contract of this test is "SDK and WS auth mocked, everything else real", but two things on
+    # the turn path reach outside the process and must not decide the outcome: configure_provider_env
+    # can wander into 9Router revival (spawn/npm install, serialized on a module-level lock) whenever
+    # earlier tests left provider evidence behind, and the background turn-label aux call does the
+    # same. Pin both out; the persistent-client path is pinned off suite-wide in conftest.
+    async def p_noop_provider_env(*args, **kwargs):
+        return None
+
+    async def p_noop_turn_label(*args, **kwargs):
+        return None
+
+    monkeypatch.setattr(run_options_mod, "configure_provider_env", p_noop_provider_env, raising=True)
+    monkeypatch.setattr(agent_manager, "generate_turn_label", p_noop_turn_label, raising=True)
+
     async def fake_query(*args, **kwargs):
         yield p_assistant()
         yield p_result()
@@ -49,10 +88,15 @@ async def fake_query(*args, **kwargs):
             ws.send_json({"event": "agent:send_message", "data": {"prompt": "hi"}})
             seen = []
             for _ in range(40):
-                ev = ws.receive_json()
+                ev = p_receive_json(ws)
                 seen.append(ev.get("event"))
-                if ev.get("event") == "agent:message" and "hello from the loop" in str(ev.get("data", {})):
+                if (
+                    ev.get("event") == "agent:status"
+                    and ev.get("data", {}).get("status") == "completed"
+                ):
                     break
+            else:
+                raise AssertionError(f"did not receive completed status; saw events={seen}")
         # the real loop's assistant reply made it all the way back over the WS
         assert "agent:message" in seen
         assert any(m.role == "assistant" and "hello from the loop" in str(m.content)

From ec41b84aa265f0a2fecdbc8c943a7649c0e13446 Mon Sep 17 00:00:00 2001
From: Kai <300677314+kai-openswarm@users.noreply.github.com>
Date: Mon, 17 Aug 2026 20:34:54 -0700
Subject: [PATCH 5/6] renderer: shared/safeMode is import-safe without a window

Same class as the config/backendConnection change: safeMode.ts read `window`
at import, and dashboardLayoutSlice imports it, so any reducer test that
imports the slice died under node:test before it ran. Guard the read; in a
renderer nothing changes.
---
 frontend/src/shared/safeMode.ts | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/frontend/src/shared/safeMode.ts b/frontend/src/shared/safeMode.ts
index c8c3f6e93..432fb7691 100644
--- a/frontend/src/shared/safeMode.ts
+++ b/frontend/src/shared/safeMode.ts
@@ -11,7 +11,10 @@ export interface SafeModeInfo {
 
 let cached: SafeModeInfo = { safeMode: false, dirtyCount: 0, fingerprint: null };
 
-const api = (window as unknown as { openswarm?: { getSafeMode?: () => Promise<SafeModeInfo> } }).openswarm;
+// Import-safe outside a renderer: the layout slice imports this, and its reducer tests run under node:test.
+const api = typeof window === 'undefined'
+  ? undefined
+  : (window as unknown as { openswarm?: { getSafeMode?: () => Promise<SafeModeInfo> } }).openswarm;
 if (api?.getSafeMode) {
   void api.getSafeMode().then((info) => {
     if (info && typeof info.safeMode === 'boolean') cached = info;

From 1b28da63f869c76166cb98ffdf89fcfc0d5418f1 Mon Sep 17 00:00:00 2001
From: Kai <300677314+kai-openswarm@users.noreply.github.com>
Date: Tue, 18 Aug 2026 12:02:16 -0700
Subject: [PATCH 6/6] dashboards: one lock around every dashboard file
 read-modify-write, a duplicate that fails half-way rolls its copied sessions
 back, and the routes reach sibling apps through an injected boundary
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three changes to backend/apps/dashboards, no product behaviour changes:

- p_dashboard_lifecycle_lock (an RLock) wraps save/load/delete/update and the
  duplicate's final write. Layout saves, renames, thumbnail writes, delete and
  duplicate arrive concurrently from the renderer (and from threads FastAPI runs
  sync routes on); two writers interleaving on the same file lost one of them.
- duplicate copies the source dashboard's sessions first and writes the new
  dashboard file last; if that write fails, the copied sessions were left as
  orphans owned by a dashboard that does not exist. p_rollback_duplicated_sessions
  deletes them (and purges memory + the session file for one that refuses to
  delete), then re-raises.
- dashboard_runtime.py: the routes' calls into the agents app (live sessions,
  session store, delete/duplicate/purge), the analytics client and the aux-naming
  primitives go through three small Protocols with default adapters
  (SessionAuthority, DashboardTelemetry, AuxNaming) instead of function-local
  imports. Behaviour is identical (the adapters look the siblings up dynamically,
  so existing test seams keep working); it is what makes the routes testable in
  isolation — the tests below patch the ports and assert the sibling apps are
  never touched.

Tests: the runtime-port contract (default adapters and fakes conform; create /
delete / duplicate / generate-name / pruning consult only the injected ports;
the two rollback paths; no function-local sibling imports can return), and the
lifecycle + naming characterization suites (telemetry emission and failure
swallowing, owned-session removal on delete, session copy/remap on duplicate,
orphan-card pruning never touches disk, naming fallbacks). Full backend suite on
3.14: 2974 passed / 15 skipped.
---
 backend/apps/dashboards/dashboard_runtime.py  | 105 +++++++
 backend/apps/dashboards/dashboards.py         | 139 +++++----
 backend/tests/test_dashboard_runtime_port.py  | 278 ++++++++++++++++++
 ...t_dashboards_lifecycle_characterization.py | 208 +++++++++++++
 ...test_dashboards_naming_characterization.py | 239 +++++++++++++++
 5 files changed, 912 insertions(+), 57 deletions(-)
 create mode 100644 backend/apps/dashboards/dashboard_runtime.py
 create mode 100644 backend/tests/test_dashboard_runtime_port.py
 create mode 100644 backend/tests/test_dashboards_lifecycle_characterization.py
 create mode 100644 backend/tests/test_dashboards_naming_characterization.py

diff --git a/backend/apps/dashboards/dashboard_runtime.py b/backend/apps/dashboards/dashboard_runtime.py
new file mode 100644
index 000000000..2b3781701
--- /dev/null
+++ b/backend/apps/dashboards/dashboard_runtime.py
@@ -0,0 +1,105 @@
+"""Injected cross-app runtime boundary for the dashboards app."""
+
+from __future__ import annotations
+
+from typing import Any, Optional, Protocol, runtime_checkable
+
+from backend.apps.agents import agent_manager as agents_runtime
+from backend.apps.agents.core import aux_llm
+from backend.apps.agents.manager.session import session_store
+from backend.apps.agents.providers import registry
+from backend.apps.service.analytics import client as analytics
+from backend.apps.settings import credentials
+from backend.apps.settings import settings as settings_app
+
+
+@runtime_checkable
+class DashboardTelemetry(Protocol):
+    """Fire-and-forget analytics the dashboard routes emit."""
+
+    def dashboard_event(self, *, dashboard_id: str, action: str) -> None: ...
+
+
+@runtime_checkable
+class SessionAuthority(Protocol):
+    """Live/persisted agent-session operations the dashboard routes need."""
+
+    def live_sessions(self) -> dict[str, Any]: ...
+
+    def load_session_data(self, session_id: str) -> Optional[dict]: ...
+
+    def save_session(self, session_id: str, data: dict) -> None: ...
+
+    async def delete_session(self, session_id: str) -> None: ...
+
+    async def duplicate_session(self, session_id: str, *, dashboard_id: str) -> Any: ...
+
+    def purge_session_memory(self, session_id: str) -> None: ...
+
+
+@runtime_checkable
+class AuxNaming(Protocol):
+    """Primitive lookups behind auto-naming; prompt/stream logic stays in dashboards.py."""
+
+    def load_settings(self) -> Any: ...
+
+    async def resolve_aux_model(self, settings: Any, *, preferred_tier: str) -> tuple[str, Any]: ...
+
+    def client_for_model(self, settings: Any, model: str) -> Any: ...
+
+    def clean_short_label(self, text: str) -> str: ...
+
+    def aux_max_tokens_for(self, model: str) -> int: ...
+
+
+class DefaultDashboardTelemetry:
+    """Production adapter; dynamic lookups preserve established test seams."""
+
+    def dashboard_event(self, *, dashboard_id: str, action: str) -> None:
+        analytics.track_dashboard_event(dashboard_id=dashboard_id, action=action)
+
+
+class DefaultSessionAuthority:
+    """Production adapter; dynamic lookups preserve established test seams."""
+
+    def live_sessions(self) -> dict[str, Any]:
+        return agents_runtime.agent_manager.sessions
+
+    def load_session_data(self, session_id: str) -> Optional[dict]:
+        return session_store.load_session_data(session_id)
+
+    def save_session(self, session_id: str, data: dict) -> None:
+        session_store.save_session(session_id, data)
+
+    async def delete_session(self, session_id: str) -> None:
+        await agents_runtime.agent_manager.delete_session(session_id)
+
+    async def duplicate_session(self, session_id: str, *, dashboard_id: str) -> Any:
+        return await agents_runtime.agent_manager.duplicate_session(session_id, dashboard_id=dashboard_id)
+
+    def purge_session_memory(self, session_id: str) -> None:
+        agents_runtime.agent_manager.purge_session_memory(session_id)
+
+
+class DefaultAuxNaming:
+    """Production adapter; dynamic lookups preserve established test seams."""
+
+    def load_settings(self) -> Any:
+        return settings_app.load_settings()
+
+    async def resolve_aux_model(self, settings: Any, *, preferred_tier: str) -> tuple[str, Any]:
+        return await registry.resolve_aux_model(settings, preferred_tier=preferred_tier)
+
+    def client_for_model(self, settings: Any, model: str) -> Any:
+        return credentials.get_anthropic_client_for_model(settings, model)
+
+    def clean_short_label(self, text: str) -> str:
+        return aux_llm.clean_short_label(text)
+
+    def aux_max_tokens_for(self, model: str) -> int:
+        return aux_llm.aux_max_tokens_for(model)
+
+
+DEFAULT_DASHBOARD_TELEMETRY: DashboardTelemetry = DefaultDashboardTelemetry()
+DEFAULT_SESSION_AUTHORITY: SessionAuthority = DefaultSessionAuthority()
+DEFAULT_AUX_NAMING: AuxNaming = DefaultAuxNaming()
diff --git a/backend/apps/dashboards/dashboards.py b/backend/apps/dashboards/dashboards.py
index 13d3b3e68..1fad3a950 100644
--- a/backend/apps/dashboards/dashboards.py
+++ b/backend/apps/dashboards/dashboards.py
@@ -1,11 +1,13 @@
 import json
 import os
 import logging
+import threading
 from contextlib import asynccontextmanager
 from datetime import datetime
 from uuid import uuid4
 
 from backend.config.Apps import SubApp
+from backend.apps.dashboards import dashboard_runtime
 from backend.apps.dashboards.models import (
     Dashboard,
     DashboardCreate,
@@ -23,6 +25,25 @@
 from backend.config.json_store import read_json_or_none, atomic_write_json
 
 OLD_LAYOUT_FILE = os.path.join(OLD_LAYOUT_DIR, "layout.json")
+# One lock around every read-modify-write of a dashboard file: layout saves, renames, thumbnail writes, delete and duplicate arrive concurrently from the renderer, and two writers interleaving on the same file lost one of them.
+p_dashboard_lifecycle_lock = threading.RLock()
+
+
+async def p_rollback_duplicated_sessions(
+    authority: dashboard_runtime.SessionAuthority, duplicated_sessions
+) -> None:
+    for _, session in reversed(duplicated_sessions):
+        try:
+            await authority.delete_session(session.id)
+        except Exception:
+            logger.exception("Failed to roll back duplicated session %s", session.id)
+            try:
+                authority.purge_session_memory(session.id)
+                session_path = os.path.join(SESSIONS_DIR, f"{session.id}.json")
+                if os.path.exists(session_path):
+                    os.remove(session_path)
+            except Exception:
+                logger.exception("Fallback rollback failed for duplicated session %s", session.id)
 
 
 def load_all() -> list[Dashboard]:
@@ -43,21 +64,24 @@ def load_all() -> list[Dashboard]:
 
 
 def save(dashboard: Dashboard):
-    atomic_write_json(os.path.join(DATA_DIR, f"{dashboard.id}.json"), dashboard.model_dump(mode="json"))
+    with p_dashboard_lifecycle_lock:
+        atomic_write_json(os.path.join(DATA_DIR, f"{dashboard.id}.json"), dashboard.model_dump(mode="json"))
 
 
 def load(dashboard_id: str) -> Dashboard:
-    path = os.path.join(DATA_DIR, f"{dashboard_id}.json")
-    data = read_json_or_none(path)
-    if data is None:
-        raise HTTPException(status_code=404, detail="Dashboard not found")
-    return Dashboard(**data)
+    with p_dashboard_lifecycle_lock:
+        path = os.path.join(DATA_DIR, f"{dashboard_id}.json")
+        data = read_json_or_none(path)
+        if data is None:
+            raise HTTPException(status_code=404, detail="Dashboard not found")
+        return Dashboard(**data)
 
 
 def p_delete(dashboard_id: str):
-    path = os.path.join(DATA_DIR, f"{dashboard_id}.json")
-    if os.path.exists(path):
-        os.remove(path)
+    with p_dashboard_lifecycle_lock:
+        path = os.path.join(DATA_DIR, f"{dashboard_id}.json")
+        if os.path.exists(path):
+            os.remove(path)
 
 
 def migrate_if_needed():
@@ -135,8 +159,7 @@ async def create_dashboard(body: DashboardCreate):
     dashboard = Dashboard(name=body.name)
     save(dashboard)
     try:
-        from backend.apps.service.analytics.client import track_dashboard_event
-        track_dashboard_event(dashboard_id=dashboard.id, action="create")
+        dashboard_runtime.DEFAULT_DASHBOARD_TELEMETRY.dashboard_event(dashboard_id=dashboard.id, action="create")
     except Exception:
         pass
     return dashboard.model_dump(mode="json")
@@ -306,10 +329,8 @@ async def generate_name(dashboard_id: str):
     if not dashboard.auto_named and dashboard.name != "Untitled Dashboard":
         return {"name": dashboard.name, "auto_named": dashboard.auto_named}
 
-    from backend.apps.agents.agent_manager import agent_manager
-
     prompts = []
-    for session in agent_manager.sessions.values():
+    for session in dashboard_runtime.DEFAULT_SESSION_AUTHORITY.live_sessions().values():
         if getattr(session, "dashboard_id", None) != dashboard_id:
             continue
         for msg in session.messages:
@@ -321,13 +342,11 @@ async def generate_name(dashboard_id: str):
         return {"name": dashboard.name, "auto_named": dashboard.auto_named}
 
     fallback = " ".join(prompts[0].split()[:4])[:36] or "Untitled Dashboard"
+    naming = dashboard_runtime.DEFAULT_AUX_NAMING
     try:
-        from backend.apps.settings.settings import load_settings
-        from backend.apps.settings.credentials import get_anthropic_client_for_model
-        from backend.apps.agents.providers.registry import resolve_aux_model
-        global_settings = load_settings()
-        aux_model, p_aux_base = await resolve_aux_model(global_settings, preferred_tier="haiku")
-        client = get_anthropic_client_for_model(global_settings, aux_model)
+        global_settings = naming.load_settings()
+        aux_model, p_aux_base = await naming.resolve_aux_model(global_settings, preferred_tier="haiku")
+        client = naming.client_for_model(global_settings, aux_model)
 
         # Mirrors generate_title's hardening: the tasks are inert text to LABEL, never answer, or the aux model happily replies with a markdown essay that becomes the title.
         system = (
@@ -342,17 +361,16 @@ async def generate_name(dashboard_id: str):
             "<tasks>\n" + "\n".join(f"- {p}" for p in prompts) + "\n</tasks>"
         )
 
-        from backend.apps.agents.core.aux_llm import clean_short_label, aux_max_tokens_for
         chunks: list[str] = []
         async with client.messages.stream(
             model=aux_model,
-            max_tokens=aux_max_tokens_for(aux_model),
+            max_tokens=naming.aux_max_tokens_for(aux_model),
             system=system,
             messages=[{"role": "user", "content": user_content}],
         ) as stream:
             async for text in stream.text_stream:
                 chunks.append(text)
-        generated = clean_short_label("".join(chunks))
+        generated = naming.clean_short_label("".join(chunks))
         if generated:
             fallback = generated
     except Exception as e:
@@ -365,7 +383,9 @@ async def generate_name(dashboard_id: str):
     return {"name": dashboard.name, "auto_named": True}
 
 
-def strip_orphan_session_cards(data: dict) -> None:
+def strip_orphan_session_cards(
+    data: dict, authority: dashboard_runtime.SessionAuthority | None = None
+) -> None:
     """Drop layout cards (and expanded ids) whose agent session no longer exists
     anywhere, in memory OR on disk. The frontend mounts an AgentChat per card and
     GETs its session; a card pointing at a vanished session (e.g. an empty
@@ -375,8 +395,7 @@ def strip_orphan_session_cards(data: dict) -> None:
     cards and nothing else. Filtering the RESPONSE (never the stored file) is
     non-destructive: a wrong check can only hide a card for one response, not
     delete it. Drafts have no backend session yet, so they're always kept."""
-    from backend.apps.agents.agent_manager import agent_manager
-    from backend.apps.agents.manager.session.session_store import load_session_data
+    live = authority or dashboard_runtime.DEFAULT_SESSION_AUTHORITY
     layout = data.get("layout")
     if not isinstance(layout, dict):
         return
@@ -385,9 +404,9 @@ def strip_orphan_session_cards(data: dict) -> None:
         return
 
     def gone(sid: str) -> bool:
-        if sid.startswith("draft-") or sid in agent_manager.sessions:
+        if sid.startswith("draft-") or sid in live.live_sessions():
             return False
-        return load_session_data(sid) is None
+        return live.load_session_data(sid) is None
 
     orphans = [sid for sid in cards if gone(sid)]
     for sid in orphans:
@@ -408,26 +427,29 @@ async def get_dashboard(dashboard_id: str):
 
 @dashboards.router.put("/{dashboard_id}")
 async def update_dashboard(dashboard_id: str, body: DashboardUpdate):
-    dashboard = load(dashboard_id)
-    if body.name is not None:
-        dashboard.name = body.name
-        dashboard.auto_named = False
-    if body.layout is not None:
-        dashboard.layout = body.layout
-    now = datetime.now()
-    if body.thumbnail is not None:
-        dashboard.thumbnail = body.thumbnail
-        dashboard.preview_signature = body.preview_signature
-        # Only a real screenshot write moves the sort key; layout/rename saves don't reorder.
-        dashboard.preview_updated_at = now
-    dashboard.updated_at = now
-    save(dashboard)
+    with p_dashboard_lifecycle_lock:
+        dashboard = load(dashboard_id)
+        if body.name is not None:
+            dashboard.name = body.name
+            dashboard.auto_named = False
+        if body.layout is not None:
+            dashboard.layout = body.layout
+        now = datetime.now()
+        if body.thumbnail is not None:
+            dashboard.thumbnail = body.thumbnail
+            dashboard.preview_signature = body.preview_signature
+            # Only a real screenshot write moves the sort key; layout/rename saves don't reorder.
+            dashboard.preview_updated_at = now
+        dashboard.updated_at = now
+        save(dashboard)
     return dashboard.model_dump(mode="json")
 
 
 @dashboards.router.delete("/{dashboard_id}")
 async def delete_dashboard(dashboard_id: str):
-    load(dashboard_id)
+    with p_dashboard_lifecycle_lock:
+        load(dashboard_id)
+        p_delete(dashboard_id)
 
     if os.path.exists(SESSIONS_DIR):
         for fname in os.listdir(SESSIONS_DIR):
@@ -442,21 +464,20 @@ async def delete_dashboard(dashboard_id: str):
             except Exception:
                 logger.warning(f"Failed to read/delete session file {fname}")
 
-    from backend.apps.agents.agent_manager import agent_manager
+    authority = dashboard_runtime.DEFAULT_SESSION_AUTHORITY
     to_remove = [
-        sid for sid, sess in agent_manager.sessions.items()
+        sid for sid, sess in authority.live_sessions().items()
         if getattr(sess, "dashboard_id", None) == dashboard_id
     ]
     for sid in to_remove:
         try:
-            await agent_manager.delete_session(sid)
+            await authority.delete_session(sid)
         except Exception:
             logger.warning(f"Failed to delete active session {sid} during dashboard deletion")
 
     p_delete(dashboard_id)
     try:
-        from backend.apps.service.analytics.client import track_dashboard_event
-        track_dashboard_event(dashboard_id=dashboard_id, action="delete")
+        dashboard_runtime.DEFAULT_DASHBOARD_TELEMETRY.dashboard_event(dashboard_id=dashboard_id, action="delete")
     except Exception:
         pass
     return {"ok": True}
@@ -469,8 +490,7 @@ async def duplicate_dashboard(dashboard_id: str):
     new_id = uuid4().hex
     now = datetime.now().isoformat()
 
-    from backend.apps.agents.agent_manager import agent_manager
-    from backend.apps.agents.manager.session.session_store import save_session
+    authority = dashboard_runtime.DEFAULT_SESSION_AUTHORITY
 
     source_layout = source_data.get("layout", {}) or {}
     source_browser_cards = source_layout.get("browser_cards", {}) or {}
@@ -484,7 +504,7 @@ async def duplicate_dashboard(dashboard_id: str):
         new_browser_cards[new_bid] = new_card
 
     candidate_ids: set[str] = set()
-    for sid, sess in agent_manager.sessions.items():
+    for sid, sess in authority.live_sessions().items():
         if getattr(sess, "dashboard_id", None) == dashboard_id:
             candidate_ids.add(sid)
     if os.path.exists(SESSIONS_DIR):
@@ -499,7 +519,7 @@ async def duplicate_dashboard(dashboard_id: str):
     duplicated_sessions = []  # (old_id, new_session)
     for old_sid in candidate_ids:
         try:
-            new_sess = await agent_manager.duplicate_session(old_sid, dashboard_id=new_id)
+            new_sess = await authority.duplicate_session(old_sid, dashboard_id=new_id)
         except Exception:
             logger.warning(f"Failed to duplicate session {old_sid} during dashboard duplication", exc_info=True)
             continue
@@ -507,7 +527,7 @@ async def duplicate_dashboard(dashboard_id: str):
         duplicated_sessions.append((old_sid, new_sess))
 
     for old_sid, new_sess in duplicated_sessions:
-        source_sess = agent_manager.sessions.get(old_sid)
+        source_sess = authority.live_sessions().get(old_sid)
         old_browser_id = getattr(source_sess, "browser_id", None) if source_sess else None
         old_parent_sid = getattr(source_sess, "parent_session_id", None) if source_sess else None
         if old_browser_id is None or old_parent_sid is None:
@@ -520,7 +540,7 @@ async def duplicate_dashboard(dashboard_id: str):
             new_sess.browser_id = browser_id_remap[old_browser_id]
         if old_parent_sid and old_parent_sid in session_id_remap:
             new_sess.parent_session_id = session_id_remap[old_parent_sid]
-        save_session(new_sess.id, new_sess.model_dump(mode="json"))
+        authority.save_session(new_sess.id, new_sess.model_dump(mode="json"))
 
     new_cards: dict[str, dict] = {}
     for old_sid, card in source_cards.items():
@@ -558,11 +578,16 @@ async def duplicate_dashboard(dashboard_id: str):
         "updated_at": now,
         "layout": new_layout,
     }
-    atomic_write_json(os.path.join(DATA_DIR, f"{new_id}.json"), new_dashboard)
+    try:
+        with p_dashboard_lifecycle_lock:
+            load(dashboard_id)
+            atomic_write_json(os.path.join(DATA_DIR, f"{new_id}.json"), new_dashboard)
+    except Exception:
+        await p_rollback_duplicated_sessions(authority, duplicated_sessions)
+        raise
 
     try:
-        from backend.apps.service.analytics.client import track_dashboard_event
-        track_dashboard_event(dashboard_id=new_id, action="create")
+        dashboard_runtime.DEFAULT_DASHBOARD_TELEMETRY.dashboard_event(dashboard_id=new_id, action="create")
     except Exception:
         pass
 
diff --git a/backend/tests/test_dashboard_runtime_port.py b/backend/tests/test_dashboard_runtime_port.py
new file mode 100644
index 000000000..a55b8a3c7
--- /dev/null
+++ b/backend/tests/test_dashboard_runtime_port.py
@@ -0,0 +1,278 @@
+"""Contract tests for the dashboards app's injected runtime boundary.
+
+Proves the injected runtime boundary through the public surface only: the
+dashboard routes and the strip_orphan_session_cards helper consult the
+injected SessionAuthority/DashboardTelemetry/AuxNaming ports, injected
+fakes fully control behavior while the underlying sibling apps are patched
+to reject any access, and the lazy sibling imports cannot silently return.
+
+Run:
+    python -m pytest backend/tests/test_dashboard_runtime_port.py -v
+"""
+
+from __future__ import annotations
+
+import ast
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+
+import backend.apps.agents.agent_manager as agent_manager_module
+import backend.apps.agents.manager.session.session_store as session_store_module
+import backend.apps.service.analytics.client as analytics_module
+from backend.apps.dashboards import dashboard_runtime
+from backend.apps.dashboards import dashboards as dashboard_routes
+from backend.apps.dashboards.dashboard_runtime import (
+    AuxNaming,
+    DashboardTelemetry,
+    DefaultAuxNaming,
+    DefaultDashboardTelemetry,
+    DefaultSessionAuthority,
+    SessionAuthority,
+)
+from backend.apps.dashboards.models import Dashboard
+
+
+class FakeSession:
+    def __init__(self, id, dashboard_id):
+        self.id = id
+        self.dashboard_id = dashboard_id
+        self.messages = []
+        self.browser_id = None
+        self.parent_session_id = None
+
+    def model_dump(self, mode="json"):
+        return {"id": self.id, "dashboard_id": self.dashboard_id}
+
+
+class FakeAuthority:
+    def __init__(self, sessions=None, disk=None):
+        self.sessions = dict(sessions or {})
+        self.disk = dict(disk or {})
+        self.deleted = []
+        self.saved = []
+        self.purged = []
+
+    def live_sessions(self):
+        return self.sessions
+
+    def load_session_data(self, session_id):
+        return self.disk.get(session_id)
+
+    def save_session(self, session_id, data):
+        self.saved.append((session_id, data))
+
+    async def delete_session(self, session_id):
+        self.deleted.append(session_id)
+        self.sessions.pop(session_id, None)
+
+    async def duplicate_session(self, session_id, *, dashboard_id):
+        return FakeSession(f"dup-{session_id}", dashboard_id)
+
+    def purge_session_memory(self, session_id):
+        self.purged.append(session_id)
+
+
+class FakeTelemetry:
+    def __init__(self):
+        self.events = []
+
+    def dashboard_event(self, *, dashboard_id, action):
+        self.events.append((action, dashboard_id))
+
+
+class FakeNaming:
+    """Naming port whose settings load always fails, forcing the fallback path."""
+
+    def load_settings(self):
+        raise RuntimeError("settings off limits")
+
+    async def resolve_aux_model(self, settings, *, preferred_tier):
+        raise AssertionError("must not be reached after load_settings fails")
+
+    def client_for_model(self, settings, model):
+        raise AssertionError("must not be reached")
+
+    def clean_short_label(self, text):
+        return text
+
+    def aux_max_tokens_for(self, model):
+        return 16
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    data_dir = tmp_path / "dashboards"
+    sessions_dir = tmp_path / "sessions"
+    data_dir.mkdir()
+    sessions_dir.mkdir()
+    monkeypatch.setattr(dashboard_routes, "DATA_DIR", str(data_dir))
+    monkeypatch.setattr(dashboard_routes, "SESSIONS_DIR", str(sessions_dir))
+
+    def deny(*args, **kwargs):
+        raise AssertionError("sibling app must not be touched when ports are injected")
+    monkeypatch.setattr(agent_manager_module, "agent_manager", None)
+    monkeypatch.setattr(analytics_module, "track_dashboard_event", deny)
+    monkeypatch.setattr(session_store_module, "load_session_data", deny)
+    monkeypatch.setattr(session_store_module, "save_session", deny)
+
+    authority = FakeAuthority()
+    telemetry = FakeTelemetry()
+    monkeypatch.setattr(dashboard_runtime, "DEFAULT_SESSION_AUTHORITY", authority)
+    monkeypatch.setattr(dashboard_runtime, "DEFAULT_DASHBOARD_TELEMETRY", telemetry)
+    monkeypatch.setattr(dashboard_runtime, "DEFAULT_AUX_NAMING", FakeNaming())
+
+    app = FastAPI()
+    app.include_router(dashboard_routes.dashboards.router, prefix="/api/dashboards")
+    return SimpleNamespace(app=app, data_dir=data_dir, authority=authority, telemetry=telemetry)
+
+
+def request(app, method, path, json_payload=None):
+    async def go():
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            return await client.request(method, path, json=json_payload)
+    return asyncio.run(go())
+
+
+# --- port conformance ---------------------------------------------------------
+
+def test_default_adapters_conform_to_protocols():
+    assert isinstance(DefaultSessionAuthority(), SessionAuthority)
+    assert isinstance(DefaultDashboardTelemetry(), DashboardTelemetry)
+    assert isinstance(DefaultAuxNaming(), AuxNaming)
+
+
+def test_fakes_conform_to_protocols():
+    assert isinstance(FakeAuthority(), SessionAuthority)
+    assert isinstance(FakeTelemetry(), DashboardTelemetry)
+    assert isinstance(FakeNaming(), AuxNaming)
+
+
+# --- routes consume only the injected ports -----------------------------------
+
+def test_create_reports_through_injected_telemetry(env):
+    response = request(env.app, "POST", "/api/dashboards/create", {"name": "Ported"})
+    assert response.status_code == 200
+    assert env.telemetry.events == [("create", response.json()["id"])]
+
+
+def test_delete_uses_injected_authority_and_telemetry(env):
+    dashboard_routes.save(Dashboard(id="d1", name="Doomed"))
+    env.authority.sessions["mine"] = FakeSession("mine", "d1")
+    env.authority.sessions["other"] = FakeSession("other", "d2")
+
+    response = request(env.app, "DELETE", "/api/dashboards/d1")
+    assert response.status_code == 200
+    assert env.authority.deleted == ["mine"]
+    assert "other" in env.authority.sessions
+    assert env.telemetry.events == [("delete", "d1")]
+
+
+def test_duplicate_uses_injected_authority_for_copy_and_save(env):
+    dashboard_routes.save(Dashboard(
+        id="d1", name="Source",
+        layout={"cards": {"s1": {"session_id": "s1"}}, "expanded_session_ids": ["s1"]},
+    ))
+    env.authority.sessions["s1"] = FakeSession("s1", "d1")
+
+    response = request(env.app, "POST", "/api/dashboards/d1/duplicate")
+    assert response.status_code == 200
+    body = response.json()
+    assert set(body["layout"]["cards"]) == {"dup-s1"}
+    assert body["layout"]["expanded_session_ids"] == ["dup-s1"]
+    assert [sid for sid, _ in env.authority.saved] == ["dup-s1"]
+    assert env.telemetry.events == [("create", body["id"])]
+
+
+def test_duplicate_rolls_back_copied_sessions_when_the_dashboard_write_fails(env, monkeypatch):
+    """The sessions are copied before the new dashboard file is written; if that write fails, the copies must not survive as orphans."""
+    dashboard_routes.save(Dashboard(
+        id="d1", name="Source",
+        layout={"cards": {"s1": {"session_id": "s1"}, "s2": {"session_id": "s2"}}},
+    ))
+    env.authority.sessions["s1"] = FakeSession("s1", "d1")
+    env.authority.sessions["s2"] = FakeSession("s2", "d1")
+
+    def fail_write(path, payload):
+        raise OSError("disk full")
+    monkeypatch.setattr(dashboard_routes, "atomic_write_json", fail_write)
+
+    with pytest.raises(OSError, match="disk full"):
+        request(env.app, "POST", "/api/dashboards/d1/duplicate")
+    assert sorted(env.authority.deleted) == ["dup-s1", "dup-s2"]
+    assert env.telemetry.events == []
+    assert sorted(item.name for item in env.data_dir.iterdir()) == ["d1.json"]
+
+
+def test_duplicate_rollback_falls_back_to_purging_a_session_that_will_not_delete(env, monkeypatch):
+    dashboard_routes.save(Dashboard(id="d1", name="Source", layout={"cards": {"s1": {"session_id": "s1"}}}))
+    env.authority.sessions["s1"] = FakeSession("s1", "d1")
+
+    async def refuse_delete(session_id):
+        raise RuntimeError("delete refused")
+    env.authority.delete_session = refuse_delete
+    monkeypatch.setattr(dashboard_routes, "atomic_write_json", lambda path, payload: (_ for _ in ()).throw(OSError("disk full")))
+
+    with pytest.raises(OSError, match="disk full"):
+        request(env.app, "POST", "/api/dashboards/d1/duplicate")
+    assert env.authority.purged == ["dup-s1"]
+
+
+def test_generate_name_uses_injected_authority_and_naming(env):
+    dashboard_routes.save(Dashboard(id="d1", name="Untitled Dashboard"))
+    session = FakeSession("s1", "d1")
+    session.messages = [SimpleNamespace(role="user", content="Review the launch checklist today")]
+    env.authority.sessions["s1"] = session
+
+    response = request(env.app, "POST", "/api/dashboards/d1/generate-name")
+    assert response.status_code == 200
+    assert response.json() == {"name": "Review the launch checklist", "auto_named": True}
+
+
+def test_get_pruning_accepts_injected_authority_argument(env):
+    authority = FakeAuthority(sessions={"live1": FakeSession("live1", "d1")}, disk={"disk1": {}})
+    data = {"layout": {
+        "cards": {
+            "live1": {"session_id": "live1"},
+            "disk1": {"session_id": "disk1"},
+            "gone1": {"session_id": "gone1"},
+            "draft-x": {"session_id": "draft-x"},
+        },
+        "expanded_session_ids": ["live1", "gone1"],
+    }}
+    dashboard_routes.strip_orphan_session_cards(data, authority)
+    assert set(data["layout"]["cards"]) == {"live1", "disk1", "draft-x"}
+    assert data["layout"]["expanded_session_ids"] == ["live1"]
+
+
+# --- the lazy sibling imports must not come back ------------------------------
+
+def function_local_sibling_imports(module_path: Path) -> list[int]:
+    tree = ast.parse(module_path.read_text(encoding="utf-8"))
+    offenders: list[int] = []
+
+    def walk(node: ast.AST, in_function: bool) -> None:
+        for child in ast.iter_child_nodes(node):
+            inner = in_function or isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef))
+            if in_function and isinstance(child, (ast.Import, ast.ImportFrom)):
+                names = [alias.name for alias in child.names] if isinstance(child, ast.Import) else [child.module or ""]
+                if any(
+                    name.startswith("backend.apps.") and not name.startswith("backend.apps.dashboards")
+                    for name in names
+                ):
+                    offenders.append(child.lineno)
+            walk(child, inner)
+
+    walk(tree, False)
+    return offenders
+
+
+def test_dashboards_has_no_function_local_sibling_imports():
+    repo = Path(__file__).resolve().parents[2]
+    for rel in ("backend/apps/dashboards/dashboards.py", "backend/apps/dashboards/dashboard_runtime.py"):
+        assert function_local_sibling_imports(repo / rel) == [], f"{rel} regressed to lazy sibling imports"
diff --git a/backend/tests/test_dashboards_lifecycle_characterization.py b/backend/tests/test_dashboards_lifecycle_characterization.py
new file mode 100644
index 000000000..47e95cf2c
--- /dev/null
+++ b/backend/tests/test_dashboards_lifecycle_characterization.py
@@ -0,0 +1,208 @@
+"""Characterization tests for dashboards lifecycle cross-app behavior.
+
+Pins the observable behavior of the create/delete/duplicate routes in
+backend/apps/dashboards/dashboards.py across the move of their sibling-app
+calls behind the injected dashboard_runtime boundary: telemetry emission and
+failure swallowing, owned-session removal on delete, and session copy/remap on
+duplicate. Exercised via a lifespan-free FastAPI harness; the seams patched here
+(module attributes on agent_manager, session_store, analytics client) keep
+working identically through the default adapters.
+
+Run:
+    python -m pytest backend/tests/test_dashboards_lifecycle_characterization.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+
+import backend.apps.agents.agent_manager as agent_manager_module
+import backend.apps.agents.manager.session.session_store as session_store_module
+import backend.apps.service.analytics.client as analytics_module
+from backend.apps.dashboards import dashboards as dashboard_routes
+from backend.apps.dashboards.models import Dashboard
+
+
+class FakeSession:
+    def __init__(self, id, dashboard_id, browser_id=None, parent_session_id=None):
+        self.id = id
+        self.dashboard_id = dashboard_id
+        self.browser_id = browser_id
+        self.parent_session_id = parent_session_id
+        self.messages = []
+
+    def model_dump(self, mode="json"):
+        return {
+            "id": self.id,
+            "dashboard_id": self.dashboard_id,
+            "browser_id": self.browser_id,
+            "parent_session_id": self.parent_session_id,
+        }
+
+
+class FakeAgentManager:
+    def __init__(self):
+        self.sessions = {}
+        self.deleted = []
+        self.duplicated = []
+        self.purged = []
+
+    async def delete_session(self, sid):
+        self.deleted.append(sid)
+        self.sessions.pop(sid, None)
+
+    async def duplicate_session(self, sid, *, dashboard_id):
+        self.duplicated.append(sid)
+        return FakeSession(f"dup-{sid}", dashboard_id)
+
+    def purge_session_memory(self, sid):
+        self.purged.append(sid)
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    data_dir = tmp_path / "dashboards"
+    sessions_dir = tmp_path / "sessions"
+    data_dir.mkdir()
+    sessions_dir.mkdir()
+    monkeypatch.setattr(dashboard_routes, "DATA_DIR", str(data_dir))
+    monkeypatch.setattr(dashboard_routes, "SESSIONS_DIR", str(sessions_dir))
+    manager = FakeAgentManager()
+    monkeypatch.setattr(agent_manager_module, "agent_manager", manager)
+    events = []
+    monkeypatch.setattr(
+        analytics_module, "track_dashboard_event",
+        lambda *, dashboard_id, action: events.append((action, dashboard_id)),
+    )
+    saved_sessions = []
+    monkeypatch.setattr(
+        session_store_module, "save_session",
+        lambda sid, data: saved_sessions.append((sid, data)),
+    )
+    app = FastAPI()
+    app.include_router(dashboard_routes.dashboards.router, prefix="/api/dashboards")
+    return SimpleNamespace(
+        app=app, data_dir=data_dir, sessions_dir=sessions_dir,
+        manager=manager, events=events, saved_sessions=saved_sessions,
+        monkeypatch=monkeypatch,
+    )
+
+
+def request(app, method, path, json_payload=None):
+    async def go():
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            return await client.request(method, path, json=json_payload)
+    return asyncio.run(go())
+
+
+def write_session_file(sessions_dir, sid, dashboard_id):
+    (sessions_dir / f"{sid}.json").write_text(json.dumps({"id": sid, "dashboard_id": dashboard_id}))
+
+
+# --- create -------------------------------------------------------------------
+
+def test_create_emits_create_telemetry_and_persists(env):
+    response = request(env.app, "POST", "/api/dashboards/create", {"name": "Fresh"})
+    assert response.status_code == 200
+    created_id = response.json()["id"]
+    assert env.events == [("create", created_id)]
+    assert (env.data_dir / f"{created_id}.json").exists()
+
+
+def test_create_survives_telemetry_failure(env):
+    def explode(*, dashboard_id, action):
+        raise RuntimeError("telemetry down")
+    env.monkeypatch.setattr(analytics_module, "track_dashboard_event", explode)
+    response = request(env.app, "POST", "/api/dashboards/create", {"name": "Fresh"})
+    assert response.status_code == 200
+    assert (env.data_dir / f"{response.json()['id']}.json").exists()
+
+
+# --- delete -------------------------------------------------------------------
+
+def test_delete_removes_owned_disk_and_live_sessions_only(env):
+    dashboard_routes.save(Dashboard(id="d1", name="Doomed"))
+    write_session_file(env.sessions_dir, "mine", "d1")
+    write_session_file(env.sessions_dir, "other", "d2")
+    env.manager.sessions["live1"] = FakeSession("live1", "d1")
+    env.manager.sessions["live2"] = FakeSession("live2", "d2")
+
+    response = request(env.app, "DELETE", "/api/dashboards/d1")
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}
+    assert not (env.sessions_dir / "mine.json").exists()
+    assert (env.sessions_dir / "other.json").exists()
+    assert env.manager.deleted == ["live1"]
+    assert not (env.data_dir / "d1.json").exists()
+    assert env.events == [("delete", "d1")]
+
+
+def test_delete_missing_dashboard_is_404_and_silent(env):
+    response = request(env.app, "DELETE", "/api/dashboards/ghost")
+    assert response.status_code == 404
+    assert env.events == []
+
+
+def test_delete_survives_telemetry_failure(env):
+    dashboard_routes.save(Dashboard(id="d1", name="Doomed"))
+
+    def explode(*, dashboard_id, action):
+        raise RuntimeError("telemetry down")
+    env.monkeypatch.setattr(analytics_module, "track_dashboard_event", explode)
+    response = request(env.app, "DELETE", "/api/dashboards/d1")
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}
+
+
+# --- duplicate ----------------------------------------------------------------
+
+def test_duplicate_copies_sessions_and_remaps_layout(env):
+    dashboard_routes.save(Dashboard(
+        id="d1", name="Source",
+        layout={
+            "cards": {"s1": {"session_id": "s1", "x": 10, "y": 20}},
+            "browser_cards": {"b1": {"browser_id": "b1", "spawned_by": "s1"}},
+            "expanded_session_ids": ["s1", "vanished"],
+        },
+    ))
+    env.manager.sessions["s1"] = FakeSession("s1", "d1", browser_id="b1")
+    write_session_file(env.sessions_dir, "s2", "d1")
+    write_session_file(env.sessions_dir, "elsewhere", "d9")
+
+    response = request(env.app, "POST", "/api/dashboards/d1/duplicate")
+    assert response.status_code == 200
+    body = response.json()
+
+    assert body["name"] == "Source (copy)"
+    assert body["id"] != "d1"
+    assert sorted(env.manager.duplicated) == ["s1", "s2"]
+    assert sorted(sid for sid, _ in env.saved_sessions) == ["dup-s1", "dup-s2"]
+
+    assert set(body["layout"]["cards"]) == {"dup-s1"}
+    assert body["layout"]["cards"]["dup-s1"]["session_id"] == "dup-s1"
+    assert body["layout"]["cards"]["dup-s1"]["x"] == 10
+
+    new_browser_cards = body["layout"]["browser_cards"]
+    assert len(new_browser_cards) == 1
+    new_bid, new_browser_card = next(iter(new_browser_cards.items()))
+    assert new_bid != "b1"
+    assert new_browser_card["browser_id"] == new_bid
+    assert new_browser_card["spawned_by"] == "dup-s1"
+
+    assert body["layout"]["expanded_session_ids"] == ["dup-s1"]
+    assert (env.data_dir / f"{body['id']}.json").exists()
+    assert env.events == [("create", body["id"])]
+
+
+def test_duplicate_missing_dashboard_is_404(env):
+    response = request(env.app, "POST", "/api/dashboards/ghost/duplicate")
+    assert response.status_code == 404
+    assert env.manager.duplicated == []
+    assert env.events == []
diff --git a/backend/tests/test_dashboards_naming_characterization.py b/backend/tests/test_dashboards_naming_characterization.py
new file mode 100644
index 000000000..0f8451dfa
--- /dev/null
+++ b/backend/tests/test_dashboards_naming_characterization.py
@@ -0,0 +1,239 @@
+"""Characterization tests for dashboard card pruning and auto-naming.
+
+Pins the observable behavior of GET /{id} orphan-card pruning and
+POST /{id}/generate-name in backend/apps/dashboards/dashboards.py across the
+move of their sibling-app calls behind the injected dashboard_runtime
+boundary: draft cards always survive, the stored file is never modified by
+pruning, naming falls back to the first four prompt words on aux failure, and
+a successful aux stream is cleaned and persisted. Seams patched here (module
+attributes on agent_manager, session_store, settings, credentials, registry,
+aux_llm) keep working identically through the default adapters.
+
+Run:
+    python -m pytest backend/tests/test_dashboards_naming_characterization.py -v
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+
+import backend.apps.agents.agent_manager as agent_manager_module
+import backend.apps.agents.core.aux_llm as aux_llm_module
+import backend.apps.agents.manager.session.session_store as session_store_module
+import backend.apps.agents.providers.registry as registry_module
+import backend.apps.settings.credentials as credentials_module
+import backend.apps.settings.settings as settings_module
+from backend.apps.dashboards import dashboards as dashboard_routes
+from backend.apps.dashboards.models import Dashboard
+
+
+class FakeMessage:
+    def __init__(self, role, content):
+        self.role = role
+        self.content = content
+
+
+class FakeSession:
+    def __init__(self, id, dashboard_id, messages=()):
+        self.id = id
+        self.dashboard_id = dashboard_id
+        self.messages = list(messages)
+
+
+class FakeAgentManager:
+    def __init__(self):
+        self.sessions = {}
+
+
+class FakeStream:
+    def __init__(self, chunks):
+        self.chunks = chunks
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    @property
+    def text_stream(self):
+        async def gen():
+            for chunk in self.chunks:
+                yield chunk
+        return gen()
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    data_dir = tmp_path / "dashboards"
+    data_dir.mkdir()
+    monkeypatch.setattr(dashboard_routes, "DATA_DIR", str(data_dir))
+    manager = FakeAgentManager()
+    monkeypatch.setattr(agent_manager_module, "agent_manager", manager)
+    app = FastAPI()
+    app.include_router(dashboard_routes.dashboards.router, prefix="/api/dashboards")
+    return SimpleNamespace(
+        app=app, data_dir=data_dir, manager=manager, monkeypatch=monkeypatch,
+    )
+
+
+def request(app, method, path, json_payload=None):
+    async def go():
+        transport = ASGITransport(app=app)
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            return await client.request(method, path, json=json_payload)
+    return asyncio.run(go())
+
+
+def card(session_id):
+    return {"session_id": session_id, "x": 1, "y": 2}
+
+
+# --- GET orphan-card pruning --------------------------------------------------
+
+def test_get_prunes_only_vanished_cards_and_never_touches_disk(env):
+    dashboard_routes.save(Dashboard(
+        id="d1", name="Board",
+        layout={
+            "cards": {
+                "live1": card("live1"),
+                "disk1": card("disk1"),
+                "gone1": card("gone1"),
+                "draft-x": card("draft-x"),
+            },
+            "expanded_session_ids": ["live1", "gone1"],
+        },
+    ))
+    env.manager.sessions["live1"] = FakeSession("live1", "d1")
+    probed = []
+
+    def fake_load_session_data(sid):
+        probed.append(sid)
+        return {} if sid == "disk1" else None
+    env.monkeypatch.setattr(session_store_module, "load_session_data", fake_load_session_data)
+    stored_before = (env.data_dir / "d1.json").read_bytes()
+
+    response = request(env.app, "GET", "/api/dashboards/d1")
+    assert response.status_code == 200
+    layout = response.json()["layout"]
+    assert set(layout["cards"]) == {"live1", "disk1", "draft-x"}
+    assert layout["expanded_session_ids"] == ["live1"]
+    assert sorted(probed) == ["disk1", "gone1"]
+    assert (env.data_dir / "d1.json").read_bytes() == stored_before
+
+
+# --- generate-name ------------------------------------------------------------
+
+def seed_named_dashboard(auto_named, name):
+    dashboard_routes.save(Dashboard(id="d1", name=name, auto_named=auto_named))
+
+
+def generate_name(env):
+    return request(env.app, "POST", "/api/dashboards/d1/generate-name")
+
+
+def test_custom_named_dashboard_is_returned_untouched(env):
+    seed_named_dashboard(auto_named=False, name="My Board")
+    response = generate_name(env)
+    assert response.status_code == 200
+    assert response.json() == {"name": "My Board", "auto_named": False}
+
+
+def test_no_matching_prompts_keeps_current_name(env):
+    seed_named_dashboard(auto_named=True, name="Old Auto Name")
+    env.manager.sessions["s9"] = FakeSession(
+        "s9", "other-dashboard", [FakeMessage("user", "Unrelated prompt")],
+    )
+    response = generate_name(env)
+    assert response.status_code == 200
+    assert response.json() == {"name": "Old Auto Name", "auto_named": True}
+
+
+def test_aux_failure_falls_back_to_first_four_prompt_words(env):
+    seed_named_dashboard(auto_named=False, name="Untitled Dashboard")
+    env.manager.sessions["s1"] = FakeSession(
+        "s1", "d1", [FakeMessage("user", "Plan a big trip to Tokyo next week")],
+    )
+
+    def explode():
+        raise RuntimeError("no settings")
+    env.monkeypatch.setattr(settings_module, "load_settings", explode)
+
+    response = generate_name(env)
+    assert response.status_code == 200
+    assert response.json() == {"name": "Plan a big trip", "auto_named": True}
+    reloaded = request(env.app, "GET", "/api/dashboards/d1").json()
+    assert reloaded["name"] == "Plan a big trip"
+    assert reloaded["auto_named"] is True
+
+
+def test_streamed_label_is_cleaned_and_persisted(env):
+    seed_named_dashboard(auto_named=True, name="Old Auto Name")
+    env.manager.sessions["s1"] = FakeSession(
+        "s1", "d1", [FakeMessage("user", "Compare hotel options for the offsite")],
+    )
+    stream_kwargs = {}
+
+    class FakeMessages:
+        def stream(self, **kwargs):
+            stream_kwargs.update(kwargs)
+            return FakeStream(["Offsite ", "Logistics"])
+
+    class FakeClient:
+        messages = FakeMessages()
+
+    async def fake_resolve_aux_model(settings, *, preferred_tier):
+        assert preferred_tier == "haiku"
+        return "haiku-model", None
+
+    env.monkeypatch.setattr(settings_module, "load_settings", lambda: SimpleNamespace())
+    env.monkeypatch.setattr(registry_module, "resolve_aux_model", fake_resolve_aux_model)
+    env.monkeypatch.setattr(
+        credentials_module, "get_anthropic_client_for_model",
+        lambda settings, model: FakeClient(),
+    )
+    env.monkeypatch.setattr(aux_llm_module, "clean_short_label", lambda text: f"Cleaned {text.strip()}")
+    env.monkeypatch.setattr(aux_llm_module, "aux_max_tokens_for", lambda model: 16)
+
+    response = generate_name(env)
+    assert response.status_code == 200
+    assert response.json() == {"name": "Cleaned Offsite Logistics", "auto_named": True}
+    assert stream_kwargs["model"] == "haiku-model"
+    assert stream_kwargs["max_tokens"] == 16
+    reloaded = request(env.app, "GET", "/api/dashboards/d1").json()
+    assert reloaded["name"] == "Cleaned Offsite Logistics"
+
+
+def test_empty_cleaned_label_keeps_first_words_fallback(env):
+    seed_named_dashboard(auto_named=True, name="Old Auto Name")
+    env.manager.sessions["s1"] = FakeSession(
+        "s1", "d1", [FakeMessage("user", "Draft the quarterly budget review deck")],
+    )
+
+    class FakeMessages:
+        def stream(self, **kwargs):
+            return FakeStream(["   "])
+
+    class FakeClient:
+        messages = FakeMessages()
+
+    async def fake_resolve_aux_model(settings, *, preferred_tier):
+        return "haiku-model", None
+
+    env.monkeypatch.setattr(settings_module, "load_settings", lambda: SimpleNamespace())
+    env.monkeypatch.setattr(registry_module, "resolve_aux_model", fake_resolve_aux_model)
+    env.monkeypatch.setattr(
+        credentials_module, "get_anthropic_client_for_model",
+        lambda settings, model: FakeClient(),
+    )
+    env.monkeypatch.setattr(aux_llm_module, "clean_short_label", lambda text: text.strip())
+    env.monkeypatch.setattr(aux_llm_module, "aux_max_tokens_for", lambda model: 16)
+
+    response = generate_name(env)
+    assert response.status_code == 200
+    assert response.json() == {"name": "Draft the quarterly budget", "auto_named": True}