🐾 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=""
)
workspace_id = serializers.CharField(read_only=True)
managed_by = serializers.CharField(read_only=True)
created_at = serializers.DateTimeField(read_only=True)
@@ -193,6 +194,7 @@ class WorkspaceStatusSerializer(serializers.Serializer):
name = serializers.CharField()
library_type = serializers.CharField()
description = serializers.CharField(allow_blank=True)
managed_by = serializers.CharField(allow_null=True, required=False)
item_count = serializers.IntegerField()
chunk_count = serializers.IntegerField()
created_at = serializers.DateTimeField()

View File

@@ -17,6 +17,7 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from library.content_types import get_library_type_config
from mcp_server.drf_auth import request_token_label
from .serializers import (
CollectionSerializer,
@@ -84,6 +85,28 @@ def library_list_create(request):
serializer.is_valid(raise_exception=True)
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
library_type = data["library_type"]
defaults = get_library_type_config(library_type)
@@ -92,6 +115,7 @@ def library_list_create(request):
name=data["name"],
library_type=library_type,
description=data.get("description", ""),
managed_by=request_token_label(request),
chunking_config=data.get("chunking_config") or defaults["chunking_config"],
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.services.library_delete import delete_library_cascade
from mcp_server.drf_auth import request_token_label
from .serializers import WorkspaceCreateSerializer, WorkspaceStatusSerializer
@@ -49,6 +50,7 @@ def _serialize_workspace(lib):
"name": lib.name,
"library_type": lib.library_type,
"description": lib.description or "",
"managed_by": lib.managed_by,
"item_count": item_count,
"chunk_count": chunk_count,
"created_at": lib.created_at,
@@ -104,6 +106,18 @@ def workspace_create(request):
},
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(
"Workspace already exists workspace_id=%s library_uid=%s",
data["workspace_id"], existing.uid,
@@ -120,6 +134,7 @@ def workspace_create(request):
description=data.get("description", ""),
workspace_id=data["workspace_id"],
owner_username=request.user.username,
managed_by=request_token_label(request),
chunking_config=defaults["chunking_config"],
embedding_instruction=defaults["embedding_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 ---
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):
"""
Top-level container representing a content library.
@@ -63,6 +75,11 @@ class Library(StructuredNode):
across the whole instance) or *workspace-scoped* (workspace_id set —
visible only to agents inside that Daedalus workspace). Scoping is
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()
@@ -93,6 +110,12 @@ class Library(StructuredNode):
# this user. Null for global libraries.
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
chunking_config = JSONProperty(default={})
embedding_instruction = StringProperty(default="")
@@ -104,6 +127,16 @@ class Library(StructuredNode):
# Relationships
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):
return f"{self.name} ({self.library_type})"

View File

@@ -12,15 +12,22 @@
<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>
</div>
{% if library.workspace_id %}
{% if library.managed_by_display %}
<div class="alert alert-error mb-6">
<span>
<strong>This Library is managed by Daedalus</strong>
(workspace <code>{{ library.workspace_id }}</code>).
<strong>This Library is managed by {{ library.managed_by_display }}</strong>{% if library.workspace_id %}
(workspace <code>{{ library.workspace_id }}</code>){% endif %}.
{% if library.workspace_id %}
Deleting it here removes its embedded content from Mnemosyne, but the
source files still live in Daedalus — it will be <strong>recreated and
re-embedded on the next Daedalus sync</strong>. Use this to clear an
orphaned Library that is blocking workspace re-registration.
source files still live in {{ library.managed_by_display }} — it will
be <strong>recreated and re-embedded on the next sync</strong>. Use
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>
</div>
{% endif %}

View File

@@ -12,10 +12,10 @@
<h1 class="text-3xl font-bold">{{ library.name }}</h1>
<div class="flex flex-wrap gap-2 mt-2">
<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"
title="Workspace {{ library.workspace_id }}">
Daedalus workspace
{% if library.workspace_id %}title="Workspace {{ library.workspace_id }}"{% endif %}>
Managed by {{ library.managed_by_display }}
</div>
{% endif %}
</div>
@@ -29,18 +29,25 @@
</div>
</div>
{% if library.workspace_id %}
{% if library.managed_by_display %}
<div class="alert alert-warning mb-6">
<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">
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>.
Normally you manage it from Daedalus. Deleting it here removes its
embedded content from Mnemosyne, but the source files still live in
Daedalus — it will be recreated and re-embedded on the next sync.
Normally you manage it from {{ library.managed_by_display }}.
Deleting it here removes its embedded content from Mnemosyne, but
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
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>

View File

@@ -20,8 +20,8 @@
<label class="label"><span class="label-text">Scope</span></label>
<select name="scope" class="select select-bordered select-sm">
<option value="all" {% if scope == "all" %}selected{% endif %}>All libraries</option>
<option value="global" {% if scope == "global" %}selected{% endif %}>Global only</option>
<option value="daedalus" {% if scope == "daedalus" %}selected{% endif %}>Daedalus workspaces only</option>
<option value="unmanaged" {% if scope == "unmanaged" %}selected{% endif %}>Unmanaged only</option>
<option value="managed" {% if scope == "managed" %}selected{% endif %}>App-managed only</option>
</select>
</div>
<button type="submit" class="btn btn-sm btn-outline">Filter</button>
@@ -45,9 +45,9 @@
</h2>
<div class="flex flex-wrap gap-1">
<div class="badge badge-outline">{{ lib.library_type }}</div>
{% if lib.workspace_id %}
<div class="badge badge-warning gap-1" title="Managed by Daedalus workspace {{ lib.workspace_id }} — do not delete from Mnemosyne.">
Daedalus workspace
{% if lib.managed_by_display %}
<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.">
Managed by {{ lib.managed_by_display }}
</div>
{% endif %}
</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.
Currently covers ``library_list``'s Daedalus-workspace scope filter. The
view loads every ``Library`` node from Neo4j and narrows it by a ``scope``
GET param (``all`` / ``global`` / ``daedalus``). These tests stub out
Neo4j entirely — patching ``neo4j_available`` and injecting a fake
``Library`` class via ``sys.modules`` — so they assert on the queryset
``.filter(...)`` call the view makes and the context it renders, not on
real graph behaviour. Mirrors the mocking style in
``test_search_views_admin_scope.py``.
Currently covers ``library_list``'s app-managed scope filter. The view
loads every ``Library`` node from Neo4j and narrows it in Python by a
``scope`` GET param (``all`` / ``unmanaged`` / ``managed``, with legacy
``global`` / ``daedalus`` aliases). These tests stub out Neo4j entirely —
patching ``neo4j_available`` and injecting a fake ``Library`` class via
``sys.modules`` — so they assert on the filtering the view does and the
context it renders, not on real graph behaviour. Mirrors the mocking
style in ``test_search_views_admin_scope.py``.
"""
from __future__ import annotations
@@ -22,6 +22,17 @@ from django.urls import reverse
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):
"""Cover the ``scope`` filter branches of ``library_list``."""
@@ -31,57 +42,64 @@ class LibraryListScopeFilterTests(TestCase):
)
self.client.force_login(self.user)
self.url = reverse("library:library-list")
self.managed = _lib("Docs", "Spelunker")
self.unmanaged = _lib("Notes", "")
def _fake_library_cls(self):
"""Return (Library stub, nodes mock) where ``nodes`` chains fluently.
``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.
"""
"""Return a Library stub whose ``nodes.order_by`` yields two libraries."""
fake_nodes = MagicMock()
fake_nodes.filter.return_value = fake_nodes
fake_nodes.order_by.return_value = []
return SimpleNamespace(nodes=fake_nodes), fake_nodes
fake_nodes.order_by.return_value = [self.managed, self.unmanaged]
return SimpleNamespace(nodes=fake_nodes)
def _get(self, fake_library_cls, **params):
def _get(self, **params):
with patch("library.views.neo4j_available", return_value=True), \
patch.dict(
"sys.modules",
{"library.models": SimpleNamespace(Library=fake_library_cls)},
{"library.models": SimpleNamespace(Library=self._fake_library_cls())},
):
return self.client.get(self.url, params)
def test_default_scope_is_all_and_does_not_filter(self):
fake_cls, fake_nodes = self._fake_library_cls()
response = self._get(fake_cls)
def test_default_scope_is_all_and_returns_everything(self):
response = self._get()
self.assertEqual(response.status_code, 200)
self.assertEqual(response.context["scope"], "all")
fake_nodes.filter.assert_not_called()
fake_nodes.order_by.assert_called_once_with("name")
self.assertEqual(
list(response.context["libraries"]), [self.managed, self.unmanaged]
)
def test_global_scope_filters_workspace_isnull_true(self):
fake_cls, fake_nodes = self._fake_library_cls()
response = self._get(fake_cls, scope="global")
def test_managed_scope_keeps_only_managed(self):
response = self._get(scope="managed")
self.assertEqual(response.context["scope"], "global")
fake_nodes.filter.assert_called_once_with(workspace_id__isnull=True)
self.assertEqual(response.context["scope"], "managed")
self.assertEqual(list(response.context["libraries"]), [self.managed])
def test_daedalus_scope_filters_workspace_isnull_false(self):
fake_cls, fake_nodes = self._fake_library_cls()
response = self._get(fake_cls, scope="daedalus")
def test_unmanaged_scope_keeps_only_unmanaged(self):
response = self._get(scope="unmanaged")
self.assertEqual(response.context["scope"], "daedalus")
fake_nodes.filter.assert_called_once_with(workspace_id__isnull=False)
self.assertEqual(response.context["scope"], "unmanaged")
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):
"""An unexpected scope value degrades to the unfiltered list."""
fake_cls, fake_nodes = self._fake_library_cls()
response = self._get(fake_cls, scope="bogus")
response = self._get(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):
with patch("library.views.neo4j_available", return_value=False):

View File

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

View File

@@ -26,6 +26,20 @@ from rest_framework import authentication, exceptions
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):
"""Authenticate DRF requests with a ``UserToken`` bearer."""

View File

@@ -57,6 +57,12 @@ class UserTokenCreateForm(forms.Form):
"class": "input input-bordered w-full",
"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(
required=False,

View File

@@ -21,7 +21,7 @@
</label>
{{ form.name }}
<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>
</div>
<div class="form-control mt-4">

View File

@@ -105,6 +105,44 @@ class UserTokenAuthenticationTest(TestCase):
resp = self._get(f"Bearer {self.plaintext} extra")
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):
# The auth class returns (user, token); DRF places the token on
# request.auth. Re-use a UserToken-aware endpoint to verify.