diff --git a/pyproject.toml b/pyproject.toml
index c262b1b..14ce697 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -52,7 +52,8 @@ dependencies = [
     "pyyaml>=6.0.2",
     "textual>=0.50.0",
     "pydantic>=2.10.0",
-    "mcp>=1.6.0",
+    # <2.0: mcp 2.0.0 removes mcp.server.fastmcp, which aegis_memory/mcp_server.py is built on.
+    "mcp>=1.6.0,<2.0",
     # Transitive security floors (OpenSSF Scorecard / OSV) — pulled in transitively
     # (idna<-httpx, pygments<-rich); loose upstream bounds otherwise let known-vulnerable
     # versions resolve. Pinned to the first patched release of each advisory.
diff --git a/server/api/routers/decay.py b/server/api/routers/decay.py
index 4519273..263dd9d 100644
--- a/server/api/routers/decay.py
+++ b/server/api/routers/decay.py
@@ -7,12 +7,13 @@
 """
 
 from fastapi import APIRouter, Depends
+from memory_authz import effective_agent_id
 from memory_repository import MemoryRepository
 from pydantic import BaseModel, Field
 from sqlalchemy.ext.asyncio import AsyncSession
 from temporal_decay import DEFAULT_HALF_LIFE, HALF_LIVES
 
-from api.dependencies.auth import check_rate_limit
+from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
 from api.dependencies.database import get_db
 
 router = APIRouter()
@@ -69,6 +70,7 @@ async def get_decay_config(project_id: str = Depends(check_rate_limit)):
 async def archive_stale_memories(
     body: ArchiveRequest,
     project_id: str = Depends(check_rate_limit),
+    auth: AuthContext = Depends(get_auth_context),
     db: AsyncSession = Depends(get_db),
 ):
     """
@@ -77,13 +79,18 @@ async def archive_stale_memories(
     Uses the existing is_deprecated / deprecated_at soft-delete columns.
     Set dry_run=True to preview how many memories would be archived without
     actually modifying any rows.
+
+    A bound key acts as its agent only, so its sweep is confined to that agent's
+    memories; an unbound key represents the application and sweeps the project.
     """
+    acting_agent_id = effective_agent_id(auth, None)
     archived = await MemoryRepository.archive_stale(
         db,
         project_id=project_id,
         namespace=body.namespace,
         threshold=body.threshold,
         dry_run=body.dry_run,
+        agent_id=acting_agent_id,
     )
     return ArchiveResponse(
         archived=archived,
diff --git a/server/api/routers/handoffs.py b/server/api/routers/handoffs.py
index fa0ac61..1936ea8 100644
--- a/server/api/routers/handoffs.py
+++ b/server/api/routers/handoffs.py
@@ -1,18 +1,21 @@
 """
-Handoff Router (~60 lines)
+Handoff Router (~80 lines)
 
 Handles: /memories/handoff
 """
 
-from api.dependencies.auth import check_rate_limit
+from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
 from api.dependencies.database import get_db
+from config import get_settings
 from embedding_service import get_embedding_service
 from fastapi import APIRouter, BackgroundTasks, Depends
+from memory_authz import effective_agent_id, read_scope_restriction
 from memory_repository import MemoryRepository
 from pydantic import BaseModel, Field
 from sqlalchemy.ext.asyncio import AsyncSession
 
 router = APIRouter()
+_settings = get_settings()
 
 
 class HandoffRequest(BaseModel):
@@ -43,21 +46,33 @@ async def handoff(
     body: HandoffRequest,
     background_tasks: BackgroundTasks,
     project_id: str = Depends(check_rate_limit),
+    auth: AuthContext = Depends(get_auth_context),
     db: AsyncSession = Depends(get_db),
 ):
-    """Generate a structured handoff baton for agent-to-agent state transfer."""
+    """Generate a structured handoff baton for agent-to-agent state transfer.
+
+    The baton carries the source agent's memories — including agent-private ones — as
+    ``key_facts``, so a handoff is a push by the source, never a pull by an arbitrary key
+    naming a source: a bound key may only hand off its own memories, while an unbound key
+    represents the whole application and may orchestrate handoffs between any of its agents.
+    """
+    source_agent_id = effective_agent_id(auth, body.source_agent_id)
+    # Same principal-trust ceiling as the query routes: this is a bulk read that never
+    # passes through authorize_read.
+    scope_filter = read_scope_restriction(auth, enforce_principal_trust=_settings.enable_trust_levels)
     embed_service = get_embedding_service()
     task_embedding = None
     if body.task_context:
         task_embedding = await embed_service.embed_single(body.task_context, db)
     results = await MemoryRepository.get_agent_memories_for_handoff(
-        db, project_id=project_id, source_agent_id=body.source_agent_id,
+        db, project_id=project_id, source_agent_id=source_agent_id,
         namespace=body.namespace, user_id=body.user_id,
         task_embedding=task_embedding, max_memories=body.max_memories,
+        requested_scope=scope_filter,
     )
     memories = [mem for mem, _ in results]
     return HandoffBaton(
-        source_agent_id=body.source_agent_id, target_agent_id=body.target_agent_id,
+        source_agent_id=source_agent_id, target_agent_id=body.target_agent_id,
         namespace=body.namespace, user_id=body.user_id, task_context=body.task_context,
         summary=None, active_tasks=[], blocked_on=[], recent_decisions=[],
         key_facts=[mem.content for mem in memories],
diff --git a/server/memory_repository.py b/server/memory_repository.py
index c536487..b78db9a 100644
--- a/server/memory_repository.py
+++ b/server/memory_repository.py
@@ -583,12 +583,16 @@ async def get_agent_memories_for_handoff(
         user_id: str | None = None,
         task_embedding: list[float] | None = None,
         max_memories: int = 20,
+        requested_scope: str | None = None,
     ) -> list[tuple[Memory, float | None]]:
         """
         Get memories for agent handoff.
 
         If task_embedding provided, rank by relevance.
         Otherwise, return most recent.
+
+        ``requested_scope`` is the caller's principal-trust ceiling (``read_scope_restriction``):
+        a NULL scope means agent-private, so restricted principals never match those rows.
         """
         conditions = [
             Memory.project_id == project_id,
@@ -603,6 +607,9 @@ async def get_agent_memories_for_handoff(
         if user_id is not None:
             conditions.append(Memory.user_id == user_id)
 
+        if requested_scope is not None:
+            conditions.append(Memory.scope == requested_scope)
+
         if task_embedding is not None:
             # Rank by semantic similarity to task
             distance_expr = Memory.embedding.cosine_distance(task_embedding)
@@ -729,6 +736,7 @@ async def archive_stale(
         threshold: float = 0.1,
         batch_size: int = 500,
         dry_run: bool = False,
+        agent_id: str | None = None,
     ) -> int:
         """
         Soft-deprecate active memories whose relevance_score falls below threshold.
@@ -736,6 +744,9 @@ async def archive_stale(
         Uses existing is_deprecated / deprecated_at columns — no new schema needed.
         Relevance is computed in Python on fetched batches (chunked approach).
 
+        ``agent_id`` restricts the sweep to one agent's rows — set for bound keys, which act
+        as their agent only. None sweeps the whole project (unbound/application keys).
+
         Returns count of memories archived (or that would be archived for dry_run).
         """
         now = datetime.now(timezone.utc)
@@ -745,6 +756,9 @@ async def archive_stale(
             not_(Memory.is_deprecated),
         ]
 
+        if agent_id is not None:
+            conditions.append(Memory.agent_id == agent_id)
+
         stmt = (
             select(Memory)
             .where(and_(*conditions))
diff --git a/tests/test_authz_bypass.py b/tests/test_authz_bypass.py
index f1d8293..6ea2f92 100644
--- a/tests/test_authz_bypass.py
+++ b/tests/test_authz_bypass.py
@@ -435,6 +435,45 @@ def test_every_router_that_writes_memories_screens_and_authorizes(self):
             "these routers write to memory without the full gate set: " + "; ".join(offenders)
         )
 
+    def test_every_router_that_touches_the_repository_authorizes_an_identity(self):
+        """The read-side generalization of the ace_delta bug.
+
+        The write sweep above keys on ``MemoryRepository.add`` and so covered only writers.
+        ``handoffs.py`` — a pure read sink — inherited nothing from the authorization pass:
+        it took ``source_agent_id`` from the request body with no auth context and returned
+        that agent's memories, private ones included, as ``key_facts`` to any project key.
+        ``decay.py`` was the state-change analogue (project-wide archive with no agent authz).
+
+        So enumerate every router that touches ``MemoryRepository`` at all — read, write, or
+        sweep — and require it to resolve an authorized identity via a ``memory_authz`` helper
+        or sit behind the admin gate. Source-derived, not hand-maintained, so a new router
+        fails here on the day it is added.
+        """
+        import inspect
+        import pkgutil
+        import importlib
+        from api import routers as routers_pkg
+
+        gates = (
+            "effective_agent_id(", "authorize_read(", "authorize_write(",
+            "authorize_delete(", "read_scope_restriction(", "require_admin",
+        )
+        offenders = []
+        for mod_info in pkgutil.iter_modules(routers_pkg.__path__):
+            mod = importlib.import_module(f"api.routers.{mod_info.name}")
+            try:
+                src = inspect.getsource(mod)
+            except OSError:
+                continue
+            if "MemoryRepository." not in src:
+                continue
+            if not any(gate in src for gate in gates):
+                offenders.append(f"{mod_info.name}.py")
+        assert not offenders, (
+            "these routers reach MemoryRepository without resolving an authorized identity: "
+            + "; ".join(offenders)
+        )
+
     def test_authz_helpers_are_actually_called_by_the_routers(self):
         """Guards against the exact original failure: implemented, exported, never invoked."""
         import inspect
@@ -598,3 +637,153 @@ def test_omitted_agent_id_is_pinned_to_the_key_not_left_open(self, monkeypatch):
         assert canary.get("requesting_agent_id") == "agent-1", (
             "omitting agent_id left the ACL identity unset, widening the query beyond the key"
         )
+
+
+class TestHandoffHttpBypass:
+    """``POST /memories/handoff`` attempted the way an attacker would.
+
+    This route was missed by the first authorization pass because the source-derived sweep
+    keyed on ``MemoryRepository.add`` — a handoff writes nothing. Pre-fix, it took
+    ``source_agent_id`` from the request body with no auth context, and the repository applied
+    no scope filter, so any project key received another agent's memories — agent-private ones
+    included — as ``key_facts`` in one call.
+    """
+
+    @staticmethod
+    def _client(monkeypatch, bound_agent_id: str | None, canary: dict):
+        from fastapi import FastAPI
+        from fastapi.testclient import TestClient
+
+        from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
+        from api.dependencies.database import get_db
+        from api.routers import handoffs
+        from memory_repository import MemoryRepository
+
+        app = FastAPI()
+        # Same prefix as production (api/app.py:133) so the path under test is the real one.
+        app.include_router(handoffs.router, prefix="/memories")
+
+        async def _fake_db():
+            yield None
+
+        app.dependency_overrides[get_auth_context] = lambda: AuthContext(
+            project_id="proj-1", trust_level="internal", bound_agent_id=bound_agent_id
+        )
+        app.dependency_overrides[check_rate_limit] = lambda: "proj-1"
+        app.dependency_overrides[get_db] = _fake_db
+
+        async def _canary(*args, **kwargs):
+            canary["called"] = True
+            canary["source_agent_id"] = kwargs.get("source_agent_id")
+            return []
+
+        monkeypatch.setattr(
+            MemoryRepository, "get_agent_memories_for_handoff", staticmethod(_canary)
+        )
+        return TestClient(app, raise_server_exceptions=False)
+
+    def test_bound_key_cannot_pull_another_agents_baton(self, monkeypatch):
+        """The exfiltration path: agent-2's key names agent-1 as the handoff source.
+
+        Against the pre-fix code this returned 200 with agent-1's private memories in
+        key_facts.
+        """
+        canary: dict = {}
+        client = self._client(monkeypatch, "agent-2", canary)
+
+        resp = client.post("/memories/handoff", json={
+            "source_agent_id": "agent-1", "target_agent_id": "agent-2",
+        })
+
+        assert resp.status_code == 403, (
+            f"expected 403 for a handoff pull naming another agent, got {resp.status_code}. "
+            f"A handoff is a push by the source, never a pull by an arbitrary key."
+        )
+        assert not canary.get("called"), (
+            "authorization ran after the read: the memories were already fetched by the time "
+            "the request was denied"
+        )
+
+    def test_source_agent_may_hand_off_its_own_memories(self, monkeypatch):
+        """Positive control, and pins the identity handed to the repository to the key's."""
+        canary: dict = {}
+        client = self._client(monkeypatch, "agent-1", canary)
+
+        resp = client.post("/memories/handoff", json={
+            "source_agent_id": "agent-1", "target_agent_id": "agent-2",
+        })
+
+        assert resp.status_code == 200, f"a legitimate self-handoff was blocked: {resp.text}"
+        assert canary.get("source_agent_id") == "agent-1"
+
+    def test_unbound_key_may_orchestrate_handoffs(self, monkeypatch):
+        """Documented posture: an unbound project key represents the whole application."""
+        canary: dict = {}
+        client = self._client(monkeypatch, None, canary)
+
+        resp = client.post("/memories/handoff", json={
+            "source_agent_id": "agent-1", "target_agent_id": "agent-2",
+        })
+
+        assert resp.status_code == 200
+        assert canary.get("source_agent_id") == "agent-1"
+
+
+class TestDecayArchiveScoping:
+    """``POST /memories/decay/archive`` — a bound key's sweep must not touch other agents' rows.
+
+    The state-change analogue of the handoff read gap: pre-fix the route archived project-wide
+    with no agent authorization, so a key bound to one agent could soft-deprecate every other
+    agent's memories.
+    """
+
+    @staticmethod
+    def _client(monkeypatch, bound_agent_id: str | None, canary: dict):
+        from fastapi import FastAPI
+        from fastapi.testclient import TestClient
+
+        from api.dependencies.auth import AuthContext, check_rate_limit, get_auth_context
+        from api.dependencies.database import get_db
+        from api.routers import decay
+        from memory_repository import MemoryRepository
+
+        app = FastAPI()
+        app.include_router(decay.router, prefix="/memories/decay")
+
+        async def _fake_db():
+            yield None
+
+        app.dependency_overrides[get_auth_context] = lambda: AuthContext(
+            project_id="proj-1", trust_level="internal", bound_agent_id=bound_agent_id
+        )
+        app.dependency_overrides[check_rate_limit] = lambda: "proj-1"
+        app.dependency_overrides[get_db] = _fake_db
+
+        async def _canary(*args, **kwargs):
+            canary["called"] = True
+            canary["agent_id"] = kwargs.get("agent_id")
+            return 0
+
+        monkeypatch.setattr(MemoryRepository, "archive_stale", staticmethod(_canary))
+        return TestClient(app, raise_server_exceptions=False)
+
+    def test_bound_key_sweep_is_confined_to_its_agent(self, monkeypatch):
+        canary: dict = {}
+        client = self._client(monkeypatch, "agent-1", canary)
+
+        resp = client.post("/memories/decay/archive", json={})
+
+        assert resp.status_code == 200
+        assert canary.get("agent_id") == "agent-1", (
+            "a bound key's archive sweep ran project-wide instead of being confined to its agent"
+        )
+
+    def test_unbound_key_sweep_covers_the_project(self, monkeypatch):
+        canary: dict = {}
+        client = self._client(monkeypatch, None, canary)
+
+        resp = client.post("/memories/decay/archive", json={})
+
+        assert resp.status_code == 200
+        assert canary.get("called")
+        assert canary.get("agent_id") is None