fix: source authorization, silent retrieval failures, and prompt structure

Source access control
---------------------
`active_docs` is client-supplied and reached the retriever unchecked, and the
retriever queries `WHERE source_id = <id>` with no owner predicate — so any
caller could pass any source id to /stream or /api/answer and have another
tenant's documents quoted back, while /api/sources/<id>/search correctly
refused the same id. Gate it through `can_access`, the helper the guarded
endpoints already use, and filter `self.source` down to the authorized set.
Fails closed: no principal, or a check that errors, drops the source.

Three sibling paths had the same gap:

- workflow agent nodes: `AgentNodeConfig.sources` is written verbatim from
  client JSON at save time and nothing validated it, so a node could name any
  tenant's source. Gate against the workflow owner, so shared workflows keep
  reading their owner's sources like shared agents do.
- /api/share: `_resolve_source_pg_id` resolved any id with no ownership
  predicate and baked it into the agent the share creates; /api/search then
  searched it. Authorize before attaching.
- search_service: re-resolve the ids stored on an agent row instead of
  trusting them, so a row written by any future path with the same gap cannot
  be read back.

Team grantees previously lost their source's retrieval config: the post-check
read was still owner-scoped, so it missed and fell back to defaults (an
`agentic_tool` source was bulk-prefetched for every grantee). Read unscoped
after `can_access` passes.

Retrieval
---------
`PGVectorStore._ensure_table_exists` created an IVFFlat index on the empty
table it had just created. IVFFlat computes centroids at build time, so those
centroids were random, and combined with the `source_id` post-filter a source
with hundreds of embedded chunks returned zero rows — retrieval reported no
documents, the model answered from memory, and nothing was logged. Stop
creating the index (exact search is correct and fast well past the sizes most
deployments reach); raise `ivfflat.probes` to sqrt(lists) where an index still
exists; and re-run a short indexed search exactly, since post-filtering means
no index setting can guarantee a full result. `graphrag` had the same
empty-table index with no fallback at all.

Also: bound `chunks` to 0-500 on both the request and agent paths (0 still
means "skip retrieval"), let a source's configured `retrieval.chunks` outrank
the request body, and cap ClassicRAG's per-source floor at
max(top_k, n_sources) so attaching sources cannot inflate the result set.

Silent failures
---------------
An empty retrieval was invisible to both the model and the client: the `source`
event was suppressed when the list was empty, so "searched and found nothing"
looked identical to "no source attached", and the prompt said nothing at all.
Emit the event always, and tell the model when a search ran and returned
nothing. A file that parses to nothing now fails ingest with a message naming
the cause instead of storing an embedding of the empty string. `score_threshold`
returns warnings when the active store or retriever cannot honour it.

Prompt structure
----------------
Retrieved documents move from the system prompt into the user turn, with the
injection guard restated next to them: they change every turn (defeating prefix
caching), they are third-party text that should not carry system authority, and
routing them through the query budget makes them truncatable rather than
silently crowding it out. Documents are shed lowest-ranked-first before the
question is touched.

The six chat presets (3 tones x 2 retrieval modes) differed only in their
Answering section; they are now composed from single-source fragments at load
time, not through Jinja inheritance, which would have opened a file-read
surface in the template sandbox and broken the tool-prefetch parser. Per-tool
guidance moves out of the prompt into tool schemas, so it travels with the tool
and cannot render when the tool is absent. A plain-text custom prompt is staged
as a persona value inside the skeleton instead of replacing it wholesale — it
used to silently lose the injection guard, platform block, memory and
attachments, and its braces are now inert.

Other fixes
-----------
- agents/base: an oversized system prompt drove the query budget negative and
  dispatched a full-price request with an empty question; raise instead.
- llm/anthropic: migrate off the retired Text Completions API. It flattened
  history to first+last message and ignored tools entirely. Adds the missing
  Anthropic handler, without which every tool call was silently dropped.
- sources/upload: `sitemap` had no branch, so every sitemap ingest died on a
  TypeError; `validate_url` now rejects a falsy URL cleanly.
- workflow nodes: retrieved documents never reached the node agent, so a
  classic node with a source and an ordinary prompt answered "I have no
  documents" while the run reported completed.
- parser/bulk: copy the metadata dict, or every chunk reports the last chunk's
  token_count.
- crawler_loader: carry the page title, or citations render the whole chunk
  body as the label.

tests/api/answer/services/test_stream_processor_utils.py

@@ -384,14 +384,61 @@ class TestConfigureSource:
         sp._configure_source()
         assert sp.source == {}
 
-    def test_request_active_docs_used(self):
+    def test_request_active_docs_used(self, monkeypatch):
+        import application.api.answer.services.stream_processor as sp_mod
         from application.api.answer.services.stream_processor import (
             StreamProcessor,
         )
+
+        monkeypatch.setattr(sp_mod, "can_access", lambda *a, **k: True)
         sp = StreamProcessor({"active_docs": "abc"}, {"sub": "u"})
         sp._configure_source()
         assert sp.source == {"active_docs": "abc"}
 
+    def test_request_active_docs_the_caller_cannot_read_is_dropped(self, monkeypatch):
+        """``active_docs`` is client input; the retriever has no owner predicate.
+
+        An unchecked id read another tenant's documents straight into the
+        answer, while /api/sources/<id>/search correctly refused the same id.
+        """
+        import application.api.answer.services.stream_processor as sp_mod
+        from application.api.answer.services.stream_processor import (
+            StreamProcessor,
+        )
+
+        monkeypatch.setattr(sp_mod, "can_access", lambda *a, **k: False)
+        sp = StreamProcessor({"active_docs": "someone-elses-id"}, {"sub": "u"})
+        sp._configure_source()
+        assert sp.source == {}
+        assert sp.all_sources == []
+
+    def test_mixed_access_keeps_only_the_readable_ids(self, monkeypatch):
+        import application.api.answer.services.stream_processor as sp_mod
+        from application.api.answer.services.stream_processor import (
+            StreamProcessor,
+        )
+
+        monkeypatch.setattr(
+            sp_mod, "can_access", lambda conn, kind, sid, user: sid == "mine"
+        )
+        sp = StreamProcessor({"active_docs": ["mine", "theirs"]}, {"sub": "u"})
+        sp._configure_source()
+        assert sp.source == {"active_docs": ["mine"]}
+
+    def test_access_check_failure_fails_closed(self, monkeypatch):
+        import application.api.answer.services.stream_processor as sp_mod
+        from application.api.answer.services.stream_processor import (
+            StreamProcessor,
+        )
+
+        def _boom(*a, **k):
+            raise RuntimeError("db down")
+
+        monkeypatch.setattr(sp_mod, "can_access", _boom)
+        sp = StreamProcessor({"active_docs": "abc"}, {"sub": "u"})
+        sp._configure_source()
+        assert sp.source == {}
+
     def test_request_active_docs_default(self):
         from application.api.answer.services.stream_processor import (
             StreamProcessor,
@@ -509,7 +556,31 @@ class TestGetPromptContent:
 
         with _patch_db(pg_conn):
             content = sp._get_prompt_content()
-        assert content == "My prompt content"
+        # A plain-text custom prompt is staged as a persona value inside the
+        # composed skeleton so it keeps Boundaries/platform/memory rather than
+        # replacing them wholesale.
+        assert sp._persona == "My prompt content"
+        assert "## Boundaries" in content
+        assert "{{ system.persona }}" in content
+
+    def test_templated_custom_prompt_is_left_alone(self, pg_conn):
+        from application.api.answer.services.stream_processor import (
+            StreamProcessor,
+        )
+        from application.storage.db.repositories.prompts import (
+            PromptsRepository,
+        )
+
+        prompt = PromptsRepository(pg_conn).create(
+            "u", "p2", "Answer as of {{ system.date }}.",
+        )
+        sp = StreamProcessor({}, {"sub": "u"})
+        sp.agent_config = {"prompt_id": str(prompt["id"])}
+
+        with _patch_db(pg_conn):
+            content = sp._get_prompt_content()
+        assert content == "Answer as of {{ system.date }}."
+        assert sp._persona is None
 
     def test_returns_none_on_missing(self, pg_conn):
         from application.api.answer.services.stream_processor import (
@@ -559,7 +630,9 @@ class TestGetPromptContent:
         sp = StreamProcessor({}, {"sub": "u"})
         sp.agent_config = {"prompt_id": "default", "agent_type": "classic"}
         content = sp._get_prompt_content()
-        assert "source.summaries" in content
+        # Classic presets ground on documents supplied with the question;
+        # agentic ones instruct the model to call the search tool.
+        assert "documents are provided with the question" in content
 
     def test_null_prompt_id_agentic_agent_gets_agentic_preset(self):
         from application.api.answer.services.stream_processor import (
@@ -582,7 +655,9 @@ class TestGetPromptContent:
         sp.agent_config = {"prompt_id": None}
         content = sp._get_prompt_content()
         assert content is not None
-        assert "source.summaries" in content
+        # Classic presets ground on documents supplied with the question;
+        # agentic ones instruct the model to call the search tool.
+        assert "documents are provided with the question" in content
 
 
 class TestPreFetchDocs:

tests/api/answer/test_empty_retrieval_visibility.py

@@ -0,0 +1,58 @@
+"""An empty retrieval must be visible to both the model and the client.
+
+Two independent suppressions made "searched your sources, found nothing"
+indistinguishable from "no source was attached": the ``source`` SSE event was
+skipped when the list was empty, and the prompt said nothing at all. Together
+they let a retrieval failure surface as a confident, ungrounded answer.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from application.agents.classic_agent import ClassicAgent
+
+
+def _agent(**kwargs):
+    with patch("application.llm.llm_creator.LLMCreator.create_llm"), patch(
+        "application.llm.handlers.handler_creator.LLMHandlerCreator.create_handler"
+    ):
+        return ClassicAgent(
+            endpoint="stream",
+            llm_name="openai",
+            model_id="gpt-4o",
+            api_key="k",
+            prompt="SYSTEM",
+            decoded_token={"sub": "u"},
+            tool_executor=MagicMock(),
+            **kwargs,
+        )
+
+
+@pytest.mark.unit
+class TestEmptyRetrievalNote:
+    def test_searched_sources_with_no_hits_tells_the_model(self):
+        agent = _agent(retrieved_docs=[], sources_were_searched=True)
+        user = agent._build_messages("SYSTEM", "What is the rebate?")[-1]["content"]
+
+        assert "were searched" in user
+        assert user.rstrip().endswith("What is the rebate?")
+
+    def test_no_sources_attached_stays_silent(self):
+        """Nothing was searched, so there is nothing to report."""
+        agent = _agent(retrieved_docs=[], sources_were_searched=False)
+        user = agent._build_messages("SYSTEM", "What is the rebate?")[-1]["content"]
+
+        assert user == "What is the rebate?"
+
+    def test_documents_found_suppresses_the_note(self):
+        agent = _agent(
+            retrieved_docs=[{"filename": "a.pdf", "text": "the rebate is 4250"}],
+            sources_were_searched=True,
+        )
+        user = agent._build_messages("SYSTEM", "What is the rebate?")[-1]["content"]
+
+        assert "were searched" not in user
+        assert "<documents>" in user

tests/api/answer/test_retrieved_docs_reach_the_model.py

@@ -0,0 +1,157 @@
+"""Retrieved documents must survive the whole request path to the user turn.
+
+The document-placement change was originally verified by constructing a
+``ClassicAgent`` with ``retrieved_docs`` already populated, which skipped the
+seam that actually carries them: retrieval -> ``StreamProcessor`` ->
+``agent_kwargs`` -> ``BaseAgent._build_messages``. A break anywhere along it
+looks exactly like "the model ignored my source", so it is pinned here.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from application.agents.classic_agent import ClassicAgent
+from application.api.answer.services.stream_processor import StreamProcessor
+
+DOCS = [
+    {"text": "Clause 4: reporting is due within 30 days.", "filename": "aml.pdf"},
+    {"text": "Clause 9: records are kept for five years.", "filename": "aml.pdf"},
+]
+
+
+def _processor(**data) -> StreamProcessor:
+    """A processor with only the fields the retrieval seam touches."""
+    sp = StreamProcessor.__new__(StreamProcessor)
+    sp.data = {"question": "Summarize current context", **data}
+    sp.agent_id = None
+    sp.agent_config = {"prompt_id": "default", "agent_type": "classic"}
+    sp.source = {"active_docs": "src-1"}
+    sp.all_sources = [{"id": "src-1", "retrieval": None}]
+    sp.retriever_config = {
+        "retriever_name": "classic",
+        "chunks": 2,
+        "doc_token_limit": 50000,
+    }
+    sp.retrieved_docs = []
+    return sp
+
+
+@pytest.mark.unit
+class TestRetrievalReachesTheAgent:
+    def test_prefetch_populates_retrieved_docs(self):
+        sp = _processor()
+        retriever = MagicMock()
+        retriever.search.return_value = DOCS
+        retriever.chunks = 2
+        retriever.doc_token_limit = 50000
+
+        with patch.object(sp, "create_retriever", return_value=retriever):
+            docs_together, docs = sp.pre_fetch_docs("Summarize current context")
+
+        assert docs == DOCS
+        assert docs_together and "Clause 4" in docs_together
+        # This is the attribute agent_kwargs forwards; empty here means the
+        # model silently answers with no source material.
+        assert sp.retrieved_docs == DOCS
+
+    def test_no_active_docs_retrieves_nothing(self):
+        """The signature of a request that forgot to attach its source."""
+        sp = _processor()
+        sp.source = {}
+        sp.all_sources = []
+
+        docs_together, docs = sp.pre_fetch_docs("Summarize current context")
+
+        assert docs is None and docs_together is None
+        assert sp.retrieved_docs == []
+
+
+@pytest.mark.unit
+class TestDocumentsLandInTheUserTurn:
+    def _agent(self, **kwargs):
+        with patch("application.llm.llm_creator.LLMCreator.create_llm"), patch(
+            "application.llm.handlers.handler_creator.LLMHandlerCreator.create_handler"
+        ):
+            return ClassicAgent(
+                endpoint="stream",
+                llm_name="openai",
+                model_id="gpt-4o",
+                api_key="k",
+                prompt="SYSTEM",
+                decoded_token={"sub": "u"},
+                tool_executor=MagicMock(),
+                **kwargs,
+            )
+
+    def test_documents_reach_the_user_turn(self):
+        agent = self._agent(retrieved_docs=DOCS)
+        messages = agent._build_messages("SYSTEM", "Summarize current context")
+
+        system, user = messages[0]["content"], messages[-1]["content"]
+        assert "Clause 4" not in system, "documents must not sit in the system prompt"
+        assert "<documents>" in user and "Clause 4" in user and "Clause 9" in user
+        assert user.rstrip().endswith("Summarize current context")
+
+    def test_empty_retrieval_leaves_the_question_alone(self):
+        agent = self._agent(retrieved_docs=[])
+        user = agent._build_messages("SYSTEM", "Summarize current context")[-1]["content"]
+        assert user == "Summarize current context"
+
+
+@pytest.mark.unit
+class TestChunksPrecedence:
+    """A source that tuned its own top-k outranks the request body.
+
+    The agentless path took ``chunks`` straight from the request, unbounded, so
+    a client could both override an owner's tuning and ask for any number.
+    """
+
+    def _sp(self, request_chunks=None, source_chunks=None):
+        from application.storage.db.source_config import RetrievalConfig
+
+        sp = _processor()
+        sp._agent_data = None
+        sp.model_id = "gpt-4o"
+        sp.model_user_id = None
+        sp.agent_key = None
+        if request_chunks is not None:
+            sp.data["chunks"] = request_chunks
+        retrieval = (
+            RetrievalConfig(chunks=source_chunks) if source_chunks else RetrievalConfig()
+        )
+        sp.all_sources = [{"id": "src-1", "retrieval": retrieval}]
+        sp._configure_retriever()
+        return sp.retriever_config["chunks"]
+
+    def test_request_applies_when_source_is_unconfigured(self):
+        assert self._sp(request_chunks="7") == 7
+
+    def test_configured_source_beats_the_request(self):
+        assert self._sp(request_chunks="100", source_chunks=5) == 5
+
+    @pytest.mark.parametrize(
+        "sent,expected",
+        [
+            ("0", 0),  # 0 means "suppress retrieval" — must survive clamping
+            ("-5", 0),
+            ("100000", 500),
+            ("501", 500),
+            ("abc", 2),
+        ],
+    )
+    def test_request_chunks_is_clamped(self, sent, expected):
+        assert self._sp(request_chunks=sent) == expected
+
+    def test_agent_chunks_is_clamped_too(self):
+        """The agent path was unbounded even after the request path was fixed."""
+        sp = _processor()
+        sp._agent_data = {"chunks": 100000}
+        sp.model_id = "gpt-4o"
+        sp.model_user_id = None
+        sp.agent_key = None
+        sp.all_sources = []
+        sp._configure_retriever()
+        assert sp.retriever_config["chunks"] == 500

tests/api/answer/test_source_access_control.py

@@ -0,0 +1,130 @@
+"""``active_docs`` is client input and must be authorization-gated.
+
+The retriever runs ``WHERE source_id = <id>`` with no owner predicate, so an
+unchecked id read another tenant's documents straight into the answer while
+``/api/sources/<id>/search`` correctly refused the same id.
+
+The gate must not break the three legitimate ways a caller reaches a source
+they do not own: a direct team grant, running an agent someone shared with
+them, and calling that agent with its API key. The first goes through
+``can_access``; the other two resolve from agent config and deliberately never
+reach this gate — transitive access *through* a shared agent is a separate
+run-time concept (see ``can_access``'s docstring).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import application.api.answer.services.stream_processor as sp_mod
+from application.api.answer.services.stream_processor import StreamProcessor
+
+OWNED = "src-owned"
+TEAM_SHARED = "src-team-shared"
+FOREIGN = "src-foreign"
+
+
+@pytest.fixture
+def access_model(monkeypatch):
+    """``can_access`` semantics: own it, or hold a direct team grant."""
+
+    def _can_access(conn, resource_type, resource_id, user_id):
+        if user_id == "owner":
+            return resource_id in (OWNED, TEAM_SHARED)
+        if user_id == "teammate":
+            return resource_id == TEAM_SHARED
+        return False
+
+    monkeypatch.setattr(sp_mod, "can_access", _can_access)
+
+
+def _processor(data, user):
+    sp = StreamProcessor.__new__(StreamProcessor)
+    sp.data = data
+    sp.decoded_token = {"sub": user}
+    sp.initial_user_id = user
+    sp._agent_data = data.get("_agent_data")
+    sp.source = {}
+    sp.all_sources = []
+    return sp
+
+
+@pytest.mark.unit
+class TestRequestSourceAccess:
+    def test_owner_reaches_their_own_source(self, access_model):
+        sp = _processor({"active_docs": OWNED}, "owner")
+        sp._configure_source()
+        assert sp.source == {"active_docs": OWNED}
+
+    def test_teammate_reaches_a_team_shared_source(self, access_model):
+        """A direct team grant is exactly what ``can_access`` allows."""
+        sp = _processor({"active_docs": TEAM_SHARED}, "teammate")
+        sp._configure_source()
+        assert sp.source == {"active_docs": TEAM_SHARED}
+
+    def test_teammate_cannot_reach_an_unshared_source(self, access_model):
+        sp = _processor({"active_docs": OWNED}, "teammate")
+        sp._configure_source()
+        assert sp.source == {}
+
+    def test_stranger_cannot_reach_anything(self, access_model):
+        sp = _processor({"active_docs": FOREIGN}, "stranger")
+        sp._configure_source()
+        assert sp.source == {}
+
+    def test_list_keeps_only_the_permitted_ids(self, access_model):
+        sp = _processor({"active_docs": [TEAM_SHARED, OWNED]}, "teammate")
+        sp._configure_source()
+        assert sp.source == {"active_docs": [TEAM_SHARED]}
+
+    def test_all_sources_matches_the_permitted_set(self, access_model):
+        sp = _processor({"active_docs": [TEAM_SHARED, OWNED]}, "teammate")
+        sp._configure_source()
+        assert [e["id"] for e in sp.all_sources] == [TEAM_SHARED]
+
+
+@pytest.mark.unit
+class TestSharedAgentAccessIsUnaffected:
+    """Agent-resolved sources must keep working for a non-owner.
+
+    Someone running a shared agent, or calling it with its API key, never
+    supplied the source id — it comes from the agent's own config, which was
+    gated when the owner attached it. Re-gating here against the *caller*
+    would break every shared agent that has a source.
+    """
+
+    def test_shared_agent_multi_source_survives(self, access_model):
+        sp = _processor(
+            {
+                "_agent_data": {
+                    "user_id": "owner",
+                    "sources": [{"id": OWNED}, {"id": FOREIGN}],
+                }
+            },
+            "teammate",
+        )
+        sp._configure_source()
+        assert sp.source == {"active_docs": [OWNED, FOREIGN]}
+        assert len(sp.all_sources) == 2
+
+    def test_shared_agent_legacy_single_source_survives(self, access_model):
+        sp = _processor(
+            {"_agent_data": {"user_id": "owner", "source": OWNED}}, "teammate"
+        )
+        sp._configure_source()
+        assert sp.source == {"active_docs": OWNED}
+
+    def test_api_key_caller_reaches_the_agents_source(self, access_model):
+        """An API-key call carries no user of its own; the agent owns access."""
+        sp = _processor(
+            {"_agent_data": {"user_id": "owner", "source": OWNED}}, None
+        )
+        sp._configure_source()
+        assert sp.source == {"active_docs": OWNED}
+
+    def test_agent_default_placeholder_still_means_no_source(self, access_model):
+        sp = _processor(
+            {"_agent_data": {"user_id": "owner", "source": "default"}}, "teammate"
+        )
+        sp._configure_source()
+        assert sp.source == {}

tests/api/answer/test_stream_processor_exposure.py

@@ -165,10 +165,15 @@ class TestNonAgentSourceConfig:
         }
         with patch(
             "application.api.answer.services.stream_processor.db_readonly"
+        ), patch(
+            "application.api.answer.services.stream_processor.can_access",
+            return_value=True,
         ), patch(
             "application.api.answer.services.stream_processor.SourcesRepository"
         ) as repo:
-            repo.return_value.get.return_value = fake_source
+            # Read unscoped after the access check, so a team grantee gets the
+            # source's real config instead of silently falling back to defaults.
+            repo.return_value.get_by_id.return_value = fake_source
             sp._configure_source()
         assert sp.source == {"active_docs": "s1"}
         assert len(sp.all_sources) == 1

tests/api/user/test_prompts.py

@@ -1,6 +1,6 @@
 import uuid
 from contextlib import contextmanager
-from unittest.mock import mock_open, patch
+from unittest.mock import patch
 
 import pytest
 from flask import Flask
@@ -85,41 +85,44 @@ class TestGetSinglePrompt:
     def test_returns_default_prompt(self, app):
         from application.api.user.prompts.routes import GetSinglePrompt
 
-        with patch("builtins.open", mock_open(read_data="Default prompt content")):
-            with app.test_request_context("/api/get_single_prompt?id=default"):
-                from flask import request
+        from application.prompts.composer import compose_preset
 
-                request.decoded_token = {"sub": "user1"}
-                response = GetSinglePrompt().get()
+        with app.test_request_context("/api/get_single_prompt?id=default"):
+            from flask import request
+
+            request.decoded_token = {"sub": "user1"}
+            response = GetSinglePrompt().get()
 
         assert response.status_code == 200
-        assert response.json["content"] == "Default prompt content"
+        assert response.json["content"] == compose_preset("default")
 
     def test_returns_creative_prompt(self, app):
         from application.api.user.prompts.routes import GetSinglePrompt
 
-        with patch("builtins.open", mock_open(read_data="Creative content")):
-            with app.test_request_context("/api/get_single_prompt?id=creative"):
-                from flask import request
+        from application.prompts.composer import compose_preset
 
-                request.decoded_token = {"sub": "user1"}
-                response = GetSinglePrompt().get()
+        with app.test_request_context("/api/get_single_prompt?id=creative"):
+            from flask import request
+
+            request.decoded_token = {"sub": "user1"}
+            response = GetSinglePrompt().get()
 
         assert response.status_code == 200
-        assert response.json["content"] == "Creative content"
+        assert response.json["content"] == compose_preset("creative")
 
     def test_returns_strict_prompt(self, app):
         from application.api.user.prompts.routes import GetSinglePrompt
 
-        with patch("builtins.open", mock_open(read_data="Strict content")):
-            with app.test_request_context("/api/get_single_prompt?id=strict"):
-                from flask import request
+        from application.prompts.composer import compose_preset
 
-                request.decoded_token = {"sub": "user1"}
-                response = GetSinglePrompt().get()
+        with app.test_request_context("/api/get_single_prompt?id=strict"):
+            from flask import request
+
+            request.decoded_token = {"sub": "user1"}
+            response = GetSinglePrompt().get()
 
         assert response.status_code == 200
-        assert response.json["content"] == "Strict content"
+        assert response.json["content"] == compose_preset("strict")
 
 
     def test_returns_400_missing_id(self, app):
@@ -315,8 +318,12 @@ class TestGetSinglePromptHappyPath:
     def test_file_read_exception_returns_400(self, app):
         from application.api.user.prompts.routes import GetSinglePrompt
 
-        with patch("builtins.open", side_effect=OSError("boom")), \
-             app.test_request_context("/api/get_single_prompt?id=default"):
+        # Presets are composed in-process now, so the failure this covers is a
+        # repository error on the custom-prompt path.
+        with patch(
+            "application.api.user.prompts.routes.PromptsRepository",
+            side_effect=OSError("boom"),
+        ), app.test_request_context("/api/get_single_prompt?id=some-custom-id"):
             from flask import request
 
             request.decoded_token = {"sub": "u1"}

tests/api/user/test_sharing.py

@@ -379,18 +379,63 @@ class TestResolveSourcePgId:
     def test_returns_none_for_falsy(self, pg_conn):
         from application.api.user.sharing.routes import _resolve_source_pg_id
 
-        assert _resolve_source_pg_id(pg_conn, None) is None
-        assert _resolve_source_pg_id(pg_conn, "") is None
+        assert _resolve_source_pg_id(pg_conn, None, "u") is None
+        assert _resolve_source_pg_id(pg_conn, "", "u") is None
 
     def test_returns_none_for_unknown_uuid(self, pg_conn):
         from application.api.user.sharing.routes import _resolve_source_pg_id
 
         assert (
-            _resolve_source_pg_id(pg_conn, "00000000-0000-0000-0000-000000000000")
+            _resolve_source_pg_id(
+                pg_conn, "00000000-0000-0000-0000-000000000000", "u"
+            )
             is None
         )
 
     def test_returns_none_for_unknown_legacy(self, pg_conn):
         from application.api.user.sharing.routes import _resolve_source_pg_id
 
-        assert _resolve_source_pg_id(pg_conn, "507f1f77bcf86cd799439011") is None
+        assert _resolve_source_pg_id(pg_conn, "507f1f77bcf86cd799439011", "u") is None
+
+
+@pytest.mark.unit
+class TestShareSourceAuthorization:
+    """A share must not attach a source the sharer cannot read.
+
+    ``_resolve_source_pg_id`` resolved any id with no ownership predicate, and
+    the id was baked into the agent the share creates — which ``/api/search``
+    then searched, returning another tenant's documents.
+    """
+
+    def test_unauthorized_source_is_refused(self, monkeypatch, pg_conn):
+        """A resolvable id the caller cannot read must not reach the agent."""
+        import application.api.user.team_sharing as ts
+
+        from application.api.user.sharing.routes import _resolve_source_pg_id
+
+        monkeypatch.setattr(ts, "can_access", lambda *a, **k: False)
+        assert (
+            _resolve_source_pg_id(
+                pg_conn, "00000000-0000-0000-0000-000000000000", "attacker"
+            )
+            is None
+        )
+
+    def test_missing_principal_resolves_nothing(self, pg_conn):
+        from application.api.user.sharing.routes import _resolve_source_pg_id
+
+        assert _resolve_source_pg_id(pg_conn, "any-id", None) is None
+
+    def test_authorized_source_passes_through(self, monkeypatch):
+        import application.api.user.team_sharing as ts
+        from application.api.user.sharing.routes import _authorized_source
+
+        monkeypatch.setattr(ts, "can_access", lambda *a, **k: True)
+        assert _authorized_source(None, ("src-1",), "owner") == "src-1"
+
+    def test_denied_source_returns_none(self, monkeypatch):
+        import application.api.user.team_sharing as ts
+        from application.api.user.sharing.routes import _authorized_source
+
+        monkeypatch.setattr(ts, "can_access", lambda *a, **k: False)
+        assert _authorized_source(None, ("src-1",), "stranger") is None