🐾 feat(library): per-app managed_by replaces hardcoded Daedalus badge #7

Merged
r merged 1 commits from feature/managed-by into main 2026-08-03 01:55:16 +00:00
15 changed files with 523 additions and 66 deletions

View File

@@ -37,6 +37,7 @@ class LibrarySerializer(serializers.Serializer):
required=False, allow_blank=True, default="" required=False, allow_blank=True, default=""
) )
workspace_id = serializers.CharField(read_only=True) workspace_id = serializers.CharField(read_only=True)
managed_by = serializers.CharField(read_only=True)
created_at = serializers.DateTimeField(read_only=True) created_at = serializers.DateTimeField(read_only=True)
@@ -193,6 +194,7 @@ class WorkspaceStatusSerializer(serializers.Serializer):
name = serializers.CharField() name = serializers.CharField()
library_type = serializers.CharField() library_type = serializers.CharField()
description = serializers.CharField(allow_blank=True) description = serializers.CharField(allow_blank=True)
managed_by = serializers.CharField(allow_null=True, required=False)
item_count = serializers.IntegerField() item_count = serializers.IntegerField()
chunk_count = serializers.IntegerField() chunk_count = serializers.IntegerField()
created_at = serializers.DateTimeField() created_at = serializers.DateTimeField()

View File

@@ -17,6 +17,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response from rest_framework.response import Response
from library.content_types import get_library_type_config from library.content_types import get_library_type_config
from mcp_server.drf_auth import request_token_label
from .serializers import ( from .serializers import (
CollectionSerializer, CollectionSerializer,
@@ -84,6 +85,28 @@ def library_list_create(request):
serializer.is_valid(raise_exception=True) serializer.is_valid(raise_exception=True)
data = serializer.validated_data data = serializer.validated_data
# Library.name is globally unique; reject collisions with a clean 409
# (uid + managed_by included so the caller can say who owns the name)
# instead of letting the unique-index save raise a 500.
try:
existing = Library.nodes.get(name=data["name"])
except Library.DoesNotExist:
existing = None
if existing is not None:
logger.warning(
"library_create name_conflict name=%s existing_uid=%s caller=%s",
data["name"], existing.uid, request.user.username,
)
return Response(
{
"detail": f"A library named '{data['name']}' already exists.",
"code": "name_conflict",
"uid": existing.uid,
"managed_by": existing.managed_by_display or None,
},
status=status.HTTP_409_CONFLICT,
)
# Populate defaults from content-type config if not provided # Populate defaults from content-type config if not provided
library_type = data["library_type"] library_type = data["library_type"]
defaults = get_library_type_config(library_type) defaults = get_library_type_config(library_type)
@@ -92,6 +115,7 @@ def library_list_create(request):
name=data["name"], name=data["name"],
library_type=library_type, library_type=library_type,
description=data.get("description", ""), description=data.get("description", ""),
managed_by=request_token_label(request),
chunking_config=data.get("chunking_config") or defaults["chunking_config"], chunking_config=data.get("chunking_config") or defaults["chunking_config"],
embedding_instruction=( embedding_instruction=(
data.get("embedding_instruction") or defaults["embedding_instruction"] data.get("embedding_instruction") or defaults["embedding_instruction"]

View File

@@ -25,6 +25,7 @@ from rest_framework.response import Response
from library.content_types import get_library_type_config from library.content_types import get_library_type_config
from library.services.library_delete import delete_library_cascade from library.services.library_delete import delete_library_cascade
from mcp_server.drf_auth import request_token_label
from .serializers import WorkspaceCreateSerializer, WorkspaceStatusSerializer from .serializers import WorkspaceCreateSerializer, WorkspaceStatusSerializer
@@ -49,6 +50,7 @@ def _serialize_workspace(lib):
"name": lib.name, "name": lib.name,
"library_type": lib.library_type, "library_type": lib.library_type,
"description": lib.description or "", "description": lib.description or "",
"managed_by": lib.managed_by,
"item_count": item_count, "item_count": item_count,
"chunk_count": chunk_count, "chunk_count": chunk_count,
"created_at": lib.created_at, "created_at": lib.created_at,
@@ -104,6 +106,18 @@ def workspace_create(request):
}, },
status=status.HTTP_409_CONFLICT, status=status.HTTP_409_CONFLICT,
) )
# Lazy backfill: pre-managed_by libraries pick up the label from
# the first idempotent re-POST. Null-only — an already-stamped
# library never changes manager.
if not existing.managed_by:
label = request_token_label(request)
if label:
existing.managed_by = label
existing.save()
logger.info(
"Backfilled managed_by=%s workspace_id=%s library_uid=%s",
label, existing.workspace_id, existing.uid,
)
logger.info( logger.info(
"Workspace already exists workspace_id=%s library_uid=%s", "Workspace already exists workspace_id=%s library_uid=%s",
data["workspace_id"], existing.uid, data["workspace_id"], existing.uid,
@@ -120,6 +134,7 @@ def workspace_create(request):
description=data.get("description", ""), description=data.get("description", ""),
workspace_id=data["workspace_id"], workspace_id=data["workspace_id"],
owner_username=request.user.username, owner_username=request.user.username,
managed_by=request_token_label(request),
chunking_config=defaults["chunking_config"], chunking_config=defaults["chunking_config"],
embedding_instruction=defaults["embedding_instruction"], embedding_instruction=defaults["embedding_instruction"],
reranker_instruction=defaults["reranker_instruction"], reranker_instruction=defaults["reranker_instruction"],

View File

@@ -0,0 +1,111 @@
"""One-off backfill of ``Library.managed_by`` for pre-existing libraries.
``managed_by`` is stamped from the creating API token's name, so
libraries created before the property existed have it null. This
command labels them:
* Workspace-scoped libraries get :func:`infer_legacy_manager`'s answer —
``Kairos`` for ``kairos-mail-*`` workspace ids, ``Daedalus`` otherwise.
* Global libraries that Spelunker ingested into (any ``IngestJob`` with
``source="spelunker"``) get the Spelunker label.
* Everything else stays null (hand-made in the web UI).
Label flags let the operator match the *actual* production token names
so backfilled rows render identically to newly stamped ones.
Idempotent: only null ``managed_by`` is ever written, and the lazy fill
in ``workspace_create`` is also null-only, so re-runs and later API
traffic never overwrite these labels.
"""
from __future__ import annotations
import logging
from django.core.management.base import BaseCommand, CommandError
from library.models import IngestJob
logger = logging.getLogger(__name__)
class Command(BaseCommand):
help = (
"Backfill Library.managed_by for libraries created before the "
"property existed. Workspace libraries are labelled by inference "
"(kairos-mail-* → Kairos, else Daedalus); global libraries with "
"Spelunker ingest jobs get the Spelunker label."
)
def add_arguments(self, parser):
parser.add_argument(
"--daedalus-label", default="Daedalus",
help="Label for non-Kairos workspace libraries (default: Daedalus).",
)
parser.add_argument(
"--kairos-label", default="Kairos",
help="Label for kairos-mail-* workspace libraries (default: Kairos).",
)
parser.add_argument(
"--spelunker-label", default="Spelunker",
help="Label for global libraries with Spelunker ingest jobs "
"(default: Spelunker).",
)
parser.add_argument(
"--dry-run", action="store_true",
help="Report what would be labelled, don't persist.",
)
def handle(self, *args, **options):
try:
from library.models import Library, infer_legacy_manager
except Exception as exc: # pragma: no cover
raise CommandError(
f"Could not import library.models.Library (Neo4j unreachable?): {exc}"
) from exc
overrides = {
"Daedalus": options["daedalus_label"],
"Kairos": options["kairos_label"],
}
spelunker_uids = set(
IngestJob.objects
.filter(source="spelunker")
.values_list("library_uid", flat=True)
.distinct()
)
candidates = list(Library.nodes.filter(managed_by__isnull=True))
to_label = []
for lib in candidates:
label = infer_legacy_manager(lib.workspace_id)
if label:
label = overrides[label]
elif lib.uid in spelunker_uids:
label = options["spelunker_label"]
if label:
to_label.append((lib, label))
self.stdout.write(f"Libraries with null managed_by: {len(candidates)}")
self.stdout.write(
self.style.SUCCESS(f"Will label: {len(to_label)} "
f"(unmanaged, left null: {len(candidates) - len(to_label)})")
)
for lib, label in to_label:
self.stdout.write(f" {lib.uid} {lib.name!r}{label}")
if options["dry_run"]:
self.stdout.write(self.style.WARNING("--dry-run: nothing written."))
return
for lib, label in to_label:
lib.managed_by = label
lib.save()
logger.info(
"backfill_managed_by uid=%s name=%s label=%s",
lib.uid, lib.name, label,
)
self.stdout.write(self.style.SUCCESS("Done."))

View File

@@ -51,6 +51,18 @@ class NearbyImageRel(StructuredRel):
# --- Node models --- # --- Node models ---
def infer_legacy_manager(workspace_id):
"""Manager label for pre-``managed_by`` workspace libraries, else None.
Kairos mail workspaces are recognisable by their deterministic
``kairos-mail-`` id prefix; every other workspace id is a Daedalus
workspace UUID.
"""
if not workspace_id:
return None
return "Kairos" if workspace_id.startswith("kairos-mail-") else "Daedalus"
class Library(StructuredNode): class Library(StructuredNode):
""" """
Top-level container representing a content library. Top-level container representing a content library.
@@ -63,6 +75,11 @@ class Library(StructuredNode):
across the whole instance) or *workspace-scoped* (workspace_id set — across the whole instance) or *workspace-scoped* (workspace_id set —
visible only to agents inside that Daedalus workspace). Scoping is visible only to agents inside that Daedalus workspace). Scoping is
enforced structurally by every search query. enforced structurally by every search query.
Independently of scoping, a library may be *app-managed*
(``managed_by`` set — created through the API by an external app such
as Daedalus, Kairos, or Spelunker, which owns its content lifecycle)
or unmanaged (created by hand in the web UI).
""" """
uid = UniqueIdProperty() uid = UniqueIdProperty()
@@ -93,6 +110,12 @@ class Library(StructuredNode):
# this user. Null for global libraries. # this user. Null for global libraries.
owner_username = StringProperty(required=False, index=True) owner_username = StringProperty(required=False, index=True)
# Name of the API token that created this library ("Daedalus",
# "Kairos", "Spelunker", ...). Null for libraries created in the
# web UI. Stamped at create time only — token rotation or edits by
# another token never change it.
managed_by = StringProperty(required=False, index=True)
# Content-type configuration # Content-type configuration
chunking_config = JSONProperty(default={}) chunking_config = JSONProperty(default={})
embedding_instruction = StringProperty(default="") embedding_instruction = StringProperty(default="")
@@ -104,6 +127,16 @@ class Library(StructuredNode):
# Relationships # Relationships
collections = RelationshipTo("Collection", "CONTAINS") collections = RelationshipTo("Collection", "CONTAINS")
@property
def managed_by_display(self):
"""Managing-app label for display; empty string when unmanaged.
Falls back to inference for workspace libraries created before
``managed_by`` existed, so rendering is identical before and
after the backfill command runs.
"""
return self.managed_by or infer_legacy_manager(self.workspace_id) or ""
def __str__(self): def __str__(self):
return f"{self.name} ({self.library_type})" return f"{self.name} ({self.library_type})"

View File

@@ -12,15 +12,22 @@
<div class="alert alert-warning mb-6"> <div class="alert alert-warning mb-6">
<span>Are you sure you want to delete <strong>{{ library.name }}</strong>? This action cannot be undone.</span> <span>Are you sure you want to delete <strong>{{ library.name }}</strong>? This action cannot be undone.</span>
</div> </div>
{% if library.workspace_id %} {% if library.managed_by_display %}
<div class="alert alert-error mb-6"> <div class="alert alert-error mb-6">
<span> <span>
<strong>This Library is managed by Daedalus</strong> <strong>This Library is managed by {{ library.managed_by_display }}</strong>{% if library.workspace_id %}
(workspace <code>{{ library.workspace_id }}</code>). (workspace <code>{{ library.workspace_id }}</code>){% endif %}.
{% if library.workspace_id %}
Deleting it here removes its embedded content from Mnemosyne, but the Deleting it here removes its embedded content from Mnemosyne, but the
source files still live in Daedalus — it will be <strong>recreated and source files still live in {{ library.managed_by_display }} — it will
re-embedded on the next Daedalus sync</strong>. Use this to clear an be <strong>recreated and re-embedded on the next sync</strong>. Use
orphaned Library that is blocking workspace re-registration. this to clear an orphaned Library that is blocking workspace
re-registration.
{% else %}
Deleting it here removes its embedded content from Mnemosyne;
{{ library.managed_by_display }} may recreate and re-embed it on its
next sync.
{% endif %}
</span> </span>
</div> </div>
{% endif %} {% endif %}

View File

@@ -12,10 +12,10 @@
<h1 class="text-3xl font-bold">{{ library.name }}</h1> <h1 class="text-3xl font-bold">{{ library.name }}</h1>
<div class="flex flex-wrap gap-2 mt-2"> <div class="flex flex-wrap gap-2 mt-2">
<div class="badge badge-primary">{{ library.library_type }}</div> <div class="badge badge-primary">{{ library.library_type }}</div>
{% if library.workspace_id %} {% if library.managed_by_display %}
<div class="badge badge-warning gap-1" <div class="badge badge-warning gap-1"
title="Workspace {{ library.workspace_id }}"> {% if library.workspace_id %}title="Workspace {{ library.workspace_id }}"{% endif %}>
Daedalus workspace Managed by {{ library.managed_by_display }}
</div> </div>
{% endif %} {% endif %}
</div> </div>
@@ -29,18 +29,25 @@
</div> </div>
</div> </div>
{% if library.workspace_id %} {% if library.managed_by_display %}
<div class="alert alert-warning mb-6"> <div class="alert alert-warning mb-6">
<div> <div>
<div class="font-semibold">Managed by Daedalus</div> <div class="font-semibold">Managed by {{ library.managed_by_display }}</div>
<div class="text-sm opacity-80"> <div class="text-sm opacity-80">
This library was created for Daedalus workspace {% if library.workspace_id %}
This library was created for workspace
<code class="font-mono">{{ library.workspace_id }}</code>. <code class="font-mono">{{ library.workspace_id }}</code>.
Normally you manage it from Daedalus. Deleting it here removes its Normally you manage it from {{ library.managed_by_display }}.
embedded content from Mnemosyne, but the source files still live in Deleting it here removes its embedded content from Mnemosyne, but
Daedalus — it will be recreated and re-embedded on the next sync. the source files still live in {{ library.managed_by_display }} —
it will be recreated and re-embedded on the next sync.
Use Delete to clear an orphaned library that is blocking workspace Use Delete to clear an orphaned library that is blocking workspace
re-registration. re-registration.
{% else %}
Content in this library is pushed by
{{ library.managed_by_display }}. Edits made here may be
overwritten or re-created on its next sync.
{% endif %}
</div> </div>
</div> </div>
</div> </div>

View File

@@ -20,8 +20,8 @@
<label class="label"><span class="label-text">Scope</span></label> <label class="label"><span class="label-text">Scope</span></label>
<select name="scope" class="select select-bordered select-sm"> <select name="scope" class="select select-bordered select-sm">
<option value="all" {% if scope == "all" %}selected{% endif %}>All libraries</option> <option value="all" {% if scope == "all" %}selected{% endif %}>All libraries</option>
<option value="global" {% if scope == "global" %}selected{% endif %}>Global only</option> <option value="unmanaged" {% if scope == "unmanaged" %}selected{% endif %}>Unmanaged only</option>
<option value="daedalus" {% if scope == "daedalus" %}selected{% endif %}>Daedalus workspaces only</option> <option value="managed" {% if scope == "managed" %}selected{% endif %}>App-managed only</option>
</select> </select>
</div> </div>
<button type="submit" class="btn btn-sm btn-outline">Filter</button> <button type="submit" class="btn btn-sm btn-outline">Filter</button>
@@ -45,9 +45,9 @@
</h2> </h2>
<div class="flex flex-wrap gap-1"> <div class="flex flex-wrap gap-1">
<div class="badge badge-outline">{{ lib.library_type }}</div> <div class="badge badge-outline">{{ lib.library_type }}</div>
{% if lib.workspace_id %} {% if lib.managed_by_display %}
<div class="badge badge-warning gap-1" title="Managed by Daedalus workspace {{ lib.workspace_id }} — do not delete from Mnemosyne."> <div class="badge badge-warning gap-1" title="Managed by {{ lib.managed_by_display }}{% if lib.workspace_id %} (workspace {{ lib.workspace_id }}){% endif %} — do not delete from Mnemosyne.">
Daedalus workspace Managed by {{ lib.managed_by_display }}
</div> </div>
{% endif %} {% endif %}
</div> </div>

View File

@@ -0,0 +1,179 @@
"""Tests for the per-app ``managed_by`` concept.
Covers the pure helpers (``infer_legacy_manager``,
``Library.managed_by_display``), the token-derived stamping and
duplicate-name rejection on the plain create endpoint (Neo4j stubbed
via ``sys.modules``, same style as ``test_views.py``), and the
``WorkspaceStatusSerializer`` surface. Cypher-touching paths are
covered by the manual end-to-end plan, not these unit tests.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import patch
from django.contrib.auth import get_user_model
from django.test import TestCase
from rest_framework.test import APIClient
from library.api.serializers import WorkspaceStatusSerializer
from library.models import Library, infer_legacy_manager
from mcp_server.models import UserToken
User = get_user_model()
class InferLegacyManagerTests(TestCase):
"""Truth table for the pre-``managed_by`` inference."""
def test_null_workspace_is_unmanaged(self):
self.assertIsNone(infer_legacy_manager(None))
self.assertIsNone(infer_legacy_manager(""))
def test_kairos_mail_prefix_is_kairos(self):
self.assertEqual(
infer_legacy_manager("kairos-mail-abc123-7"), "Kairos"
)
def test_other_workspace_is_daedalus(self):
self.assertEqual(
infer_legacy_manager("2f9c4a1e-uuid-ish"), "Daedalus"
)
class ManagedByDisplayTests(TestCase):
"""``managed_by_display`` on in-memory (unsaved) Library nodes."""
def test_stamped_label_wins(self):
lib = Library(name="x", managed_by="Spelunker")
self.assertEqual(lib.managed_by_display, "Spelunker")
def test_stamped_label_wins_over_inference(self):
lib = Library(
name="x", managed_by="My Token", workspace_id="kairos-mail-a-1"
)
self.assertEqual(lib.managed_by_display, "My Token")
def test_legacy_workspace_falls_back_to_inference(self):
lib = Library(name="x", workspace_id="ws-uuid")
self.assertEqual(lib.managed_by_display, "Daedalus")
def test_unmanaged_is_empty_string(self):
lib = Library(name="x")
self.assertEqual(lib.managed_by_display, "")
class _FakeLibrary:
"""Stand-in for the neomodel Library on the plain create endpoint."""
class DoesNotExist(Exception):
pass
existing = None # what nodes.get(name=...) returns
instances = [] # constructor kwargs, in order
def __init__(self, **kwargs):
type(self).instances.append(kwargs)
self.__dict__.update(kwargs)
self.uid = "lib-new"
self.workspace_id = None
self.created_at = None
def save(self):
return self
class _Nodes:
@staticmethod
def get(**kwargs):
if _FakeLibrary.existing is None:
raise _FakeLibrary.DoesNotExist()
return _FakeLibrary.existing
nodes = _Nodes()
class LibraryCreateStampingTests(TestCase):
"""POST /library/api/libraries/ stamps ``managed_by`` and rejects dupes."""
def setUp(self):
self.user = User.objects.create_user(username="op", password="pw")
self.client = APIClient()
_FakeLibrary.existing = None
_FakeLibrary.instances = []
def _post(self, token=None):
self.client.force_authenticate(user=self.user, token=token)
with patch.dict(
"sys.modules",
{"library.models": SimpleNamespace(Library=_FakeLibrary)},
):
return self.client.post(
"/library/api/libraries/",
{"name": "Docs", "library_type": "technical"},
format="json",
)
def test_token_create_stamps_token_name(self):
response = self._post(token=UserToken(name="Spelunker"))
self.assertEqual(response.status_code, 201)
self.assertEqual(_FakeLibrary.instances[0]["managed_by"], "Spelunker")
self.assertEqual(response.json()["managed_by"], "Spelunker")
def test_session_create_leaves_managed_by_null(self):
response = self._post(token=None)
self.assertEqual(response.status_code, 201)
self.assertIsNone(_FakeLibrary.instances[0]["managed_by"])
def test_duplicate_name_returns_409_name_conflict(self):
_FakeLibrary.existing = SimpleNamespace(
uid="lib-old", managed_by_display="Daedalus"
)
response = self._post(token=UserToken(name="Spelunker"))
self.assertEqual(response.status_code, 409)
body = response.json()
self.assertEqual(body["code"], "name_conflict")
self.assertIn("Docs", body["detail"])
self.assertEqual(body["uid"], "lib-old")
self.assertEqual(body["managed_by"], "Daedalus")
self.assertEqual(_FakeLibrary.instances, [])
def test_duplicate_of_unmanaged_reports_null_manager(self):
_FakeLibrary.existing = SimpleNamespace(
uid="lib-old", managed_by_display=""
)
response = self._post(token=None)
self.assertEqual(response.status_code, 409)
self.assertIsNone(response.json()["managed_by"])
class WorkspaceStatusSerializerManagedByTests(TestCase):
"""The workspace status payload carries ``managed_by`` (nullable)."""
BASE = {
"workspace_id": "ws_a",
"library_uid": "lib_1",
"name": "W",
"library_type": "technical",
"description": "",
"item_count": 0,
"chunk_count": 0,
"created_at": "2026-01-01T00:00:00Z",
}
def test_managed_by_value_round_trips(self):
s = WorkspaceStatusSerializer(data={**self.BASE, "managed_by": "Daedalus"})
self.assertTrue(s.is_valid(), s.errors)
self.assertEqual(s.validated_data["managed_by"], "Daedalus")
def test_managed_by_null_accepted(self):
s = WorkspaceStatusSerializer(data={**self.BASE, "managed_by": None})
self.assertTrue(s.is_valid(), s.errors)
def test_managed_by_absent_accepted(self):
s = WorkspaceStatusSerializer(data=self.BASE)
self.assertTrue(s.is_valid(), s.errors)

View File

@@ -1,13 +1,13 @@
"""Tests for the library CRUD HTML views. """Tests for the library CRUD HTML views.
Currently covers ``library_list``'s Daedalus-workspace scope filter. The Currently covers ``library_list``'s app-managed scope filter. The view
view loads every ``Library`` node from Neo4j and narrows it by a ``scope`` loads every ``Library`` node from Neo4j and narrows it in Python by a
GET param (``all`` / ``global`` / ``daedalus``). These tests stub out ``scope`` GET param (``all`` / ``unmanaged`` / ``managed``, with legacy
Neo4j entirely — patching ``neo4j_available`` and injecting a fake ``global`` / ``daedalus`` aliases). These tests stub out Neo4j entirely —
``Library`` class via ``sys.modules`` — so they assert on the queryset patching ``neo4j_available`` and injecting a fake ``Library`` class via
``.filter(...)`` call the view makes and the context it renders, not on ``sys.modules`` — so they assert on the filtering the view does and the
real graph behaviour. Mirrors the mocking style in context it renders, not on real graph behaviour. Mirrors the mocking
``test_search_views_admin_scope.py``. style in ``test_search_views_admin_scope.py``.
""" """
from __future__ import annotations from __future__ import annotations
@@ -22,6 +22,17 @@ from django.urls import reverse
User = get_user_model() User = get_user_model()
def _lib(name, managed_by_display):
return SimpleNamespace(
uid=f"uid-{name}",
name=name,
library_type="technical",
description="",
workspace_id=None,
managed_by_display=managed_by_display,
)
class LibraryListScopeFilterTests(TestCase): class LibraryListScopeFilterTests(TestCase):
"""Cover the ``scope`` filter branches of ``library_list``.""" """Cover the ``scope`` filter branches of ``library_list``."""
@@ -31,57 +42,64 @@ class LibraryListScopeFilterTests(TestCase):
) )
self.client.force_login(self.user) self.client.force_login(self.user)
self.url = reverse("library:library-list") self.url = reverse("library:library-list")
self.managed = _lib("Docs", "Spelunker")
self.unmanaged = _lib("Notes", "")
def _fake_library_cls(self): def _fake_library_cls(self):
"""Return (Library stub, nodes mock) where ``nodes`` chains fluently. """Return a Library stub whose ``nodes.order_by`` yields two libraries."""
``Library.nodes`` → ``.filter(...)`` → ``.order_by(...)`` all return
the same MagicMock so the view's queryset building works regardless
of which branch it takes, and ``.filter`` records its kwargs.
"""
fake_nodes = MagicMock() fake_nodes = MagicMock()
fake_nodes.filter.return_value = fake_nodes fake_nodes.order_by.return_value = [self.managed, self.unmanaged]
fake_nodes.order_by.return_value = [] return SimpleNamespace(nodes=fake_nodes)
return SimpleNamespace(nodes=fake_nodes), fake_nodes
def _get(self, fake_library_cls, **params): def _get(self, **params):
with patch("library.views.neo4j_available", return_value=True), \ with patch("library.views.neo4j_available", return_value=True), \
patch.dict( patch.dict(
"sys.modules", "sys.modules",
{"library.models": SimpleNamespace(Library=fake_library_cls)}, {"library.models": SimpleNamespace(Library=self._fake_library_cls())},
): ):
return self.client.get(self.url, params) return self.client.get(self.url, params)
def test_default_scope_is_all_and_does_not_filter(self): def test_default_scope_is_all_and_returns_everything(self):
fake_cls, fake_nodes = self._fake_library_cls() response = self._get()
response = self._get(fake_cls)
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["scope"], "all") self.assertEqual(response.context["scope"], "all")
fake_nodes.filter.assert_not_called() self.assertEqual(
fake_nodes.order_by.assert_called_once_with("name") list(response.context["libraries"]), [self.managed, self.unmanaged]
)
def test_global_scope_filters_workspace_isnull_true(self): def test_managed_scope_keeps_only_managed(self):
fake_cls, fake_nodes = self._fake_library_cls() response = self._get(scope="managed")
response = self._get(fake_cls, scope="global")
self.assertEqual(response.context["scope"], "global") self.assertEqual(response.context["scope"], "managed")
fake_nodes.filter.assert_called_once_with(workspace_id__isnull=True) self.assertEqual(list(response.context["libraries"]), [self.managed])
def test_daedalus_scope_filters_workspace_isnull_false(self): def test_unmanaged_scope_keeps_only_unmanaged(self):
fake_cls, fake_nodes = self._fake_library_cls() response = self._get(scope="unmanaged")
response = self._get(fake_cls, scope="daedalus")
self.assertEqual(response.context["scope"], "daedalus") self.assertEqual(response.context["scope"], "unmanaged")
fake_nodes.filter.assert_called_once_with(workspace_id__isnull=False) self.assertEqual(list(response.context["libraries"]), [self.unmanaged])
def test_legacy_daedalus_scope_aliases_to_managed(self):
response = self._get(scope="daedalus")
self.assertEqual(response.context["scope"], "managed")
self.assertEqual(list(response.context["libraries"]), [self.managed])
def test_legacy_global_scope_aliases_to_unmanaged(self):
response = self._get(scope="global")
self.assertEqual(response.context["scope"], "unmanaged")
self.assertEqual(list(response.context["libraries"]), [self.unmanaged])
def test_unknown_scope_does_not_filter(self): def test_unknown_scope_does_not_filter(self):
"""An unexpected scope value degrades to the unfiltered list.""" """An unexpected scope value degrades to the unfiltered list."""
fake_cls, fake_nodes = self._fake_library_cls() response = self._get(scope="bogus")
response = self._get(fake_cls, scope="bogus")
self.assertEqual(response.context["scope"], "bogus") self.assertEqual(response.context["scope"], "bogus")
fake_nodes.filter.assert_not_called() self.assertEqual(
list(response.context["libraries"]), [self.managed, self.unmanaged]
)
def test_neo4j_unavailable_sets_error_and_empty_list(self): def test_neo4j_unavailable_sets_error_and_empty_list(self):
with patch("library.views.neo4j_available", return_value=False): with patch("library.views.neo4j_available", return_value=False):

View File

@@ -31,20 +31,23 @@ logger = logging.getLogger(__name__)
@login_required @login_required
def library_list(request): def library_list(request):
"""List libraries, optionally filtered by Daedalus-workspace scope.""" """List libraries, optionally filtered by app-managed scope."""
scope = request.GET.get("scope", "all") scope = request.GET.get("scope", "all")
# Legacy bookmark values from before managed_by existed.
scope = {"daedalus": "managed", "global": "unmanaged"}.get(scope, scope)
libraries = [] libraries = []
error = None error = None
if neo4j_available(): if neo4j_available():
try: try:
from .models import Library from .models import Library
qs = Library.nodes libraries = list(Library.nodes.order_by("name"))
if scope == "daedalus": # managed_by_display covers legacy workspace libraries that
qs = qs.filter(workspace_id__isnull=False) # predate the managed_by property, so filter in Python.
elif scope == "global": if scope == "managed":
qs = qs.filter(workspace_id__isnull=True) libraries = [l for l in libraries if l.managed_by_display]
libraries = qs.order_by("name") elif scope == "unmanaged":
libraries = [l for l in libraries if not l.managed_by_display]
except Exception as e: except Exception as e:
error = f"Could not connect to Neo4j: {e}" error = f"Could not connect to Neo4j: {e}"
logger.error(error) logger.error(error)

View File

@@ -26,6 +26,20 @@ from rest_framework import authentication, exceptions
from .auth import MCPAuthError, resolve_mcp_user from .auth import MCPAuthError, resolve_mcp_user
def request_token_label(request):
"""Name of the ``UserToken`` authenticating this request, or None.
Session-authenticated requests (``request.auth`` is None) and blank
token names return None — the caller treats both as "no managing app".
"""
from .models import UserToken
token = getattr(request, "auth", None)
if isinstance(token, UserToken):
return token.name.strip() or None
return None
class UserTokenAuthentication(authentication.BaseAuthentication): class UserTokenAuthentication(authentication.BaseAuthentication):
"""Authenticate DRF requests with a ``UserToken`` bearer.""" """Authenticate DRF requests with a ``UserToken`` bearer."""

View File

@@ -57,6 +57,12 @@ class UserTokenCreateForm(forms.Form):
"class": "input input-bordered w-full", "class": "input input-bordered w-full",
"placeholder": "e.g. Claude Desktop, CI script", "placeholder": "e.g. Claude Desktop, CI script",
}), }),
help_text=(
"A friendly label so you can identify this token later. It also "
"labels any library the token creates (shown as “Managed by "
"<name>”) — for an app integration, use the app's name, e.g. "
"Daedalus, Kairos, Spelunker."
),
) )
expires_at = forms.DateTimeField( expires_at = forms.DateTimeField(
required=False, required=False,

View File

@@ -21,7 +21,7 @@
</label> </label>
{{ form.name }} {{ form.name }}
<label class="label"> <label class="label">
<span class="label-text-alt opacity-60">A friendly label so you can identify this token later (e.g. “Claude Desktop”).</span> <span class="label-text-alt opacity-60">{{ form.name.help_text }}</span>
</label> </label>
</div> </div>
<div class="form-control mt-4"> <div class="form-control mt-4">

View File

@@ -105,6 +105,44 @@ class UserTokenAuthenticationTest(TestCase):
resp = self._get(f"Bearer {self.plaintext} extra") resp = self._get(f"Bearer {self.plaintext} extra")
self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED) self.assertEqual(resp.status_code, status.HTTP_401_UNAUTHORIZED)
def test_request_token_label_reads_token_name(self):
from types import SimpleNamespace
from mcp_server.drf_auth import request_token_label
token = UserToken(name=" Spelunker ")
self.assertEqual(
request_token_label(SimpleNamespace(auth=token)), "Spelunker"
)
def test_request_token_label_none_for_session(self):
from types import SimpleNamespace
from mcp_server.drf_auth import request_token_label
self.assertIsNone(request_token_label(SimpleNamespace(auth=None)))
# A request object with no auth attribute at all (plain Django).
self.assertIsNone(request_token_label(SimpleNamespace()))
def test_request_token_label_none_for_blank_name(self):
from types import SimpleNamespace
from mcp_server.drf_auth import request_token_label
self.assertIsNone(
request_token_label(SimpleNamespace(auth=UserToken(name=" ")))
)
def test_request_token_label_none_for_foreign_auth_object(self):
from types import SimpleNamespace
from mcp_server.drf_auth import request_token_label
# e.g. a JWT dict from another auth class — not a UserToken.
self.assertIsNone(
request_token_label(SimpleNamespace(auth={"iss": "daedalus"}))
)
def test_request_auth_stashes_token(self): def test_request_auth_stashes_token(self):
# The auth class returns (user, token); DRF places the token on # The auth class returns (user, token); DRF places the token on
# request.auth. Re-use a UserToken-aware endpoint to verify. # request.auth. Re-use a UserToken-aware endpoint to verify.