fix(api): bind /recall to the authenticated agent the way /search does

core-api/src/core_api/routes/memories.py

@@ -1989,14 +1989,33 @@ async def recall_endpoint(
     """
     # Read endpoint — readable set widening applies (see /search).
     auth.enforce_readable_tenant(body.tenant_id)
+    # Same identity binding as /search, for the same two reasons. An agent
+    # credential may only filter to itself, and the effective identity is
+    # body-first with a fall back to the authenticated agent. Before this,
+    # /recall ran its trust<2 fleet forcing against whatever
+    # ``filter_agent_id`` the caller asserted, and an agent that omitted the
+    # filter passed ``caller_agent_id=None`` to the search, which is the
+    # tenant-wide visibility a tenant credential gets. Either way a trust-1
+    # agent read another fleet's scope_team rows through /recall while
+    # /search refused the same request. The MCP twin (``caura_recall``)
+    # already binds to the authenticated agent.
+    if auth.agent_id and body.filter_agent_id and body.filter_agent_id != auth.agent_id:
+        raise HTTPException(
+            status_code=403,
+            detail=(
+                f"filter_agent_id '{body.filter_agent_id}' does not match the "
+                f"authenticated agent identity '{auth.agent_id}'."
+            ),
+        )
+    eff_agent_id = body.filter_agent_id or auth.agent_id
     if auth.tenant_id:
-        if body.filter_agent_id:
+        if eff_agent_id:
             fleet_id_hint = body.fleet_ids[0] if body.fleet_ids and len(body.fleet_ids) == 1 else None
-            _agent = await get_or_create_agent(body.tenant_id, body.filter_agent_id, fleet_id_hint)
+            _agent = await get_or_create_agent(body.tenant_id, eff_agent_id, fleet_id_hint)
             if not body.fleet_ids and _agent.get("fleet_id") and _agent.get("trust_level", 0) < 2:
                 body.fleet_ids = [_agent["fleet_id"]]
             if body.fleet_ids and len(body.fleet_ids) == 1:
-                await enforce_fleet_read(body.tenant_id, body.filter_agent_id, body.fleet_ids[0])
+                await enforce_fleet_read(body.tenant_id, eff_agent_id, body.fleet_ids[0])
         # D13 — a recall is a recall, not a search: plans meter them separately
         # and the recalls counter never moved because this site (and the MCP
         # twin) billed "search". Flag-gated; see ``recall_operation``.
@@ -2021,7 +2040,8 @@ async def recall_endpoint(
         query=body.query,
         fleet_ids=body.fleet_ids,
         filter_agent_id=body.filter_agent_id,
-        caller_agent_id=body.filter_agent_id,
+        # Visibility identity is the authenticated agent, as in /search.
+        caller_agent_id=eff_agent_id,
         memory_type_filter=body.memory_type_filter,
         status_filter=body.status_filter,
         top_k=body.top_k,

tests/test_route_authz_gaps.py

@@ -256,10 +256,10 @@ async def test_stm_promote_rejects_peer_agent(client, as_auth, _stm_enabled):
 async def test_delete_audit_attributes_gateway_agent(client, as_auth, sc):
     """A gateway agent credential deleting WITHOUT the agent_id query param
     must be attributed to its verified identity, not None."""
+    from core_storage_api.services.postgres_service import get_read_session
     from sqlalchemy import select
 
     from common.models.audit import AuditLog
-    from core_storage_api.services.postgres_service import get_read_session
 
     tenant = f"tenant-{_uid()}"
     await _seed_agent(sc, tenant, "deleter-agent", 3)
@@ -551,6 +551,92 @@ async def test_a_tenant_credential_may_still_filter_search_by_any_agent(
     assert resp.status_code == 200, resp.text
 
 
+# ---------------------------------------------------------------------------
+# POST /recall — the sibling of /search never received the H-14 binding
+# ---------------------------------------------------------------------------
+
+
+async def test_recall_cannot_borrow_a_peer_identity_via_filter_agent_id(
+    client, as_auth
+):
+    """``/recall`` ran the H-14 path unbound: ``filter_agent_id`` was both the
+    visibility identity and the subject of the trust<2 fleet forcing, and an
+    omitted filter passed ``caller_agent_id=None``, the tenant-wide identity.
+    A trust-1 agent read another fleet's scope_team rows through ``/recall``
+    while ``/search`` refused the same request.
+    """
+    tenant = f"tenant-{_uid()}"
+    as_auth(tenant, agent_id="agent-a")
+    resp = await client.post(
+        "/api/v1/recall",
+        json={
+            "tenant_id": tenant,
+            "query": "anything",
+            "top_k": 5,
+            "filter_agent_id": "agent-b",
+        },
+    )
+    assert resp.status_code == 403, resp.text
+    assert "does not match the authenticated agent identity" in resp.text
+
+
+async def test_recall_filtering_to_own_agent_id_is_allowed(client, as_auth):
+    tenant = f"tenant-{_uid()}"
+    as_auth(tenant, agent_id="agent-a")
+    resp = await client.post(
+        "/api/v1/recall",
+        json={
+            "tenant_id": tenant,
+            "query": "anything",
+            "top_k": 5,
+            "filter_agent_id": "agent-a",
+        },
+    )
+    assert resp.status_code == 200, resp.text
+
+
+async def test_recall_binds_an_omitted_filter_to_the_authenticated_agent(
+    client, as_auth, monkeypatch
+):
+    """An agent credential that omits ``filter_agent_id`` must still recall as
+    itself, not as the tenant. The search call is the seam: its
+    ``caller_agent_id`` is what decides which scope_agent rows are visible.
+    """
+    from core_api.services import memory_service
+
+    seen: dict = {}
+
+    async def _fake_search(**kwargs):
+        seen.update(kwargs)
+        return []
+
+    monkeypatch.setattr(memory_service, "search_memories", _fake_search)
+    tenant = f"tenant-{_uid()}"
+    as_auth(tenant, agent_id="agent-a")
+    resp = await client.post(
+        "/api/v1/recall",
+        json={"tenant_id": tenant, "query": "anything", "top_k": 5},
+    )
+    assert resp.status_code == 200, resp.text
+    assert seen["caller_agent_id"] == "agent-a"
+    assert seen["filter_agent_id"] is None
+
+
+async def test_a_tenant_credential_may_still_recall_by_any_agent(client, as_auth):
+    tenant = f"tenant-{_uid()}"
+    as_auth(tenant)  # no agent_id
+    resp = await client.post(
+        "/api/v1/recall",
+        json={
+            "tenant_id": tenant,
+            "query": "anything",
+            "top_k": 5,
+            "filter_agent_id": "agent-b",
+        },
+    )
+    assert resp.status_code == 200, resp.text
+
+
 # ---------------------------------------------------------------------------
 # H-17 — /stm/promote owed the LTM write gates
 #