diff --git a/src/agentic_cli/workflow/adk/manager.py b/src/agentic_cli/workflow/adk/manager.py
index a31eb8d..d488fe1 100644
--- a/src/agentic_cli/workflow/adk/manager.py
+++ b/src/agentic_cli/workflow/adk/manager.py
@@ -1092,23 +1092,25 @@ def _job_result_payload(record, result: Any) -> dict:
     # Sessions (native — DatabaseSessionService persists events continuously)
     # -------------------------------------------------------------------------
 
-    async def session_exists(self, session_id: str) -> bool:
-        """True if the store holds this session with any events."""
+    async def session_exists(
+        self, session_id: str, *, user_id: str | None = None
+    ) -> bool:
+        """True if the store holds this session (for the given user) with any events."""
         if not self._session_service:
             return False
         session = await self._session_service.get_session(
             app_name=self.app_name,
-            user_id=self._settings.default_user,
+            user_id=user_id or self._settings.default_user,
             session_id=session_id,
         )
         return session is not None and bool(getattr(session, "events", None))
 
-    async def list_sessions(self) -> list[dict]:
-        """List persisted sessions for the current user (most recent first)."""
+    async def list_sessions(self, *, user_id: str | None = None) -> list[dict]:
+        """List persisted sessions for a user (default: settings.default_user)."""
         if not self._session_service:
             return []
         resp = await self._session_service.list_sessions(
-            app_name=self.app_name, user_id=self._settings.default_user,
+            app_name=self.app_name, user_id=user_id or self._settings.default_user,
         )
         sessions = [
             {
@@ -1121,24 +1123,28 @@ async def list_sessions(self) -> list[dict]:
         sessions.sort(key=lambda x: x["last_update"] or 0, reverse=True)
         return sessions
 
-    async def delete_session(self, session_id: str) -> bool:
-        """Delete a persisted session from the store."""
+    async def delete_session(
+        self, session_id: str, *, user_id: str | None = None
+    ) -> bool:
+        """Delete a persisted session (for the given user) from the store."""
         if not self._session_service:
             return False
         await self._session_service.delete_session(
             app_name=self.app_name,
-            user_id=self._settings.default_user,
+            user_id=user_id or self._settings.default_user,
             session_id=session_id,
         )
         return True
 
-    async def recent_messages(self, session_id: str, limit: int = 20) -> list[dict]:
+    async def recent_messages(
+        self, session_id: str, limit: int = 20, *, user_id: str | None = None
+    ) -> list[dict]:
         """Recent text messages from the stored session (for fact extraction)."""
         if not self._session_service:
             return []
         session = await self._session_service.get_session(
             app_name=self.app_name,
-            user_id=self._settings.default_user,
+            user_id=user_id or self._settings.default_user,
             session_id=session_id,
         )
         if session is None or not getattr(session, "events", None):
diff --git a/tests/workflow/test_session_store.py b/tests/workflow/test_session_store.py
index 5995329..a462909 100644
--- a/tests/workflow/test_session_store.py
+++ b/tests/workflow/test_session_store.py
@@ -137,3 +137,68 @@ async def test_persists_across_fresh_manager(self, tmp_path: Path):
         # A second manager over the same sqlite file sees the session.
         mgr2, _ = self._manager(tmp_path)
         assert await mgr2.session_exists("sess-z") is True
+
+
+class TestAdkSessionUserScope:
+    """Session APIs accept an explicit user_id, defaulting to settings.default_user.
+
+    process() always accepted arbitrary user_id (and job-resume threads
+    record.user_id), but the query/manage APIs hard-coded default_user —
+    sessions created for another user were invisible to them.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _require_adk(self):
+        pytest.importorskip("google.adk")
+
+    def _manager(self, tmp_path: Path):
+        from agentic_cli.workflow.adk.manager import GoogleADKWorkflowManager
+
+        settings = _settings(tmp_path, session_store="sqlite")
+        mgr = GoogleADKWorkflowManager.__new__(GoogleADKWorkflowManager)
+        mgr._settings = settings
+        mgr._app_name = "test_app"
+        mgr.session_id = "default_session"
+        mgr._session_service = mgr._make_session_service()
+        return mgr, settings
+
+    async def _seed_as(self, mgr, user_id: str, sid: str, text: str):
+        from google.adk.events import Event
+        from google.genai import types
+
+        s = await mgr._session_service.create_session(
+            app_name=mgr.app_name, user_id=user_id, session_id=sid
+        )
+        await mgr._session_service.append_event(
+            session=s,
+            event=Event(
+                author="user",
+                content=types.Content(
+                    role="user", parts=[types.Part.from_text(text=text)]
+                ),
+            ),
+        )
+
+    async def test_session_apis_scope_to_given_user(self, tmp_path: Path):
+        mgr, settings = self._manager(tmp_path)
+        await self._seed_as(mgr, "alice", "sess-a", "alice message")
+        await self._seed_as(mgr, settings.default_user, "sess-d", "default message")
+
+        # Default scope: unchanged behavior, sees only default_user's sessions
+        assert await mgr.session_exists("sess-d") is True
+        assert await mgr.session_exists("sess-a") is False
+
+        # Explicit user scope reaches alice's session through every API
+        assert await mgr.session_exists("sess-a", user_id="alice") is True
+
+        listed = await mgr.list_sessions(user_id="alice")
+        assert [s["session_id"] for s in listed] == ["sess-a"]
+
+        recent = await mgr.recent_messages("sess-a", user_id="alice")
+        assert recent and recent[-1]["content"] == "alice message"
+
+        assert await mgr.delete_session("sess-a", user_id="alice") is True
+        assert await mgr.session_exists("sess-a", user_id="alice") is False
+
+        # Default user's session untouched by alice-scoped operations
+        assert await mgr.session_exists("sess-d") is True