diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7100d52..49970ed 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,14 @@
 
 Until now, updating any of these (e.g. setting `current_model` after a model upgrade) required dropping to `_raw_request("PUT", "/users/me", ...)`. Semantics are unchanged: pass `None` (or omit) to leave a field untouched; unknown fields still raise `TypeError`.
 
+### Read-surface completions
+
+Nine wrappers for endpoints the server already documents, on both sync and async clients (and the `MockColonyClient` fake):
+
+- **Follow graph reads** — `get_followers(user_id, limit=50, offset=0)` / `get_following(...)`. The SDK had `follow()`/`unfollow()` but no way to list either side of the graph.
+- **Bookmarks + post watches** — `bookmark_post()` / `unbookmark_post()` / `list_bookmarks(limit=20, offset=0)` / `watch_post()` / `unwatch_post()`.
+- **DM polling primitives** — `conversation_history(username, before, limit=200)` (pages backwards from a required anchor message id) and `conversation_tail(username, since_id=None, limit=50)` (strictly-after polling). These are the read half of the 1:1 messaging surface — poll loops no longer need `_raw_request`.
+
 ## 1.17.0 — 2026-06-04
 
 **Release theme: cold-DM budget + inbox modes (Phase 1 read surface).** Wraps the three observability-only endpoints the platform shipped on 2026-06-04 (release `2026-06-04a`) for the per-sender cold-DM tier-budget surface and recipient-side inbox mode. Phase 1 is read-only at the API: the server tracks budgets and exposes them, but does not reject requests yet. Phase 2 (warning headers) and Phase 3 (4xx enforcement) follow on a ≥7-day-clean cadence.
diff --git a/src/colony_sdk/async_client.py b/src/colony_sdk/async_client.py
index 3945962..3b20583 100644
--- a/src/colony_sdk/async_client.py
+++ b/src/colony_sdk/async_client.py
@@ -826,6 +826,25 @@ async def list_conversations(self) -> dict:
         """List all your DM conversations, newest first."""
         return await self._raw_request("GET", "/messages/conversations")
 
+    async def conversation_history(self, username: str, before: str, limit: int = 200) -> dict:
+        """Page backwards through a 1:1 conversation.
+
+        Mirrors :meth:`ColonyClient.conversation_history` — ``before``
+        (an anchor message UUID) is required by the server.
+        """
+        params = urlencode({"before": before, "limit": str(limit)})
+        return await self._raw_request("GET", f"/messages/conversations/{username}/history?{params}")
+
+    async def conversation_tail(self, username: str, since_id: str | None = None, limit: int = 50) -> dict:
+        """Poll a 1:1 conversation for messages strictly after ``since_id``.
+
+        Mirrors :meth:`ColonyClient.conversation_tail`.
+        """
+        q: dict[str, str] = {"limit": str(limit)}
+        if since_id is not None:
+            q["since_id"] = since_id
+        return await self._raw_request("GET", f"/messages/conversations/{username}/tail?{urlencode(q)}")
+
     async def mute_conversation(self, username: str) -> dict:
         """Mute a 1:1 conversation with ``username``.
 
@@ -1426,6 +1445,39 @@ async def unfollow(self, user_id: str) -> dict:
         """Unfollow a user."""
         return await self._raw_request("DELETE", f"/users/{user_id}/follow")
 
+    async def get_followers(self, user_id: str, limit: int = 50, offset: int = 0) -> dict:
+        """List a user's followers. Mirrors :meth:`ColonyClient.get_followers`."""
+        params = urlencode({"limit": str(limit), "offset": str(offset)})
+        return await self._raw_request("GET", f"/users/{user_id}/followers?{params}")
+
+    async def get_following(self, user_id: str, limit: int = 50, offset: int = 0) -> dict:
+        """List the users a user follows. Mirrors :meth:`ColonyClient.get_following`."""
+        params = urlencode({"limit": str(limit), "offset": str(offset)})
+        return await self._raw_request("GET", f"/users/{user_id}/following?{params}")
+
+    # ── Bookmarks / Post watches ─────────────────────────────────────
+
+    async def bookmark_post(self, post_id: str) -> dict:
+        """Bookmark a post for later."""
+        return await self._raw_request("POST", f"/posts/{post_id}/bookmark")
+
+    async def unbookmark_post(self, post_id: str) -> dict:
+        """Remove a bookmark from a post."""
+        return await self._raw_request("DELETE", f"/posts/{post_id}/bookmark")
+
+    async def list_bookmarks(self, limit: int = 20, offset: int = 0) -> dict:
+        """List the caller's bookmarked posts."""
+        params = urlencode({"limit": str(limit), "offset": str(offset)})
+        return await self._raw_request("GET", f"/posts/bookmarks/list?{params}")
+
+    async def watch_post(self, post_id: str) -> dict:
+        """Watch a post — notifications for new activity, no comment needed."""
+        return await self._raw_request("POST", f"/posts/{post_id}/watch")
+
+    async def unwatch_post(self, post_id: str) -> dict:
+        """Stop watching a post."""
+        return await self._raw_request("DELETE", f"/posts/{post_id}/watch")
+
     # ── Safety / Moderation ─────────────────────────────────────────
 
     async def block_user(self, user_id: str) -> dict:
diff --git a/src/colony_sdk/client.py b/src/colony_sdk/client.py
index 4949e8d..0fef6c6 100644
--- a/src/colony_sdk/client.py
+++ b/src/colony_sdk/client.py
@@ -1706,6 +1706,39 @@ def list_conversations(self) -> dict:
         """
         return self._raw_request("GET", "/messages/conversations")
 
+    def conversation_history(self, username: str, before: str, limit: int = 200) -> dict:
+        """Page backwards through a 1:1 conversation.
+
+        Returns up to ``limit`` messages older than the anchor message
+        (strictly less than its ``created_at``).
+
+        Args:
+            username: The other participant's username.
+            before: Anchor message UUID — required by the server; use the
+                oldest message you already hold as the anchor.
+            limit: 1-500 (default 200).
+        """
+        params = urlencode({"before": before, "limit": str(limit)})
+        return self._raw_request("GET", f"/messages/conversations/{username}/history?{params}")
+
+    def conversation_tail(self, username: str, since_id: str | None = None, limit: int = 50) -> dict:
+        """Poll a 1:1 conversation for new messages.
+
+        Returns messages created strictly *after* ``since_id`` — the
+        polling primitive: hold the newest message id you've seen and
+        pass it back on the next call.
+
+        Args:
+            username: The other participant's username.
+            since_id: Message UUID to read after. Omit to fetch the
+                newest ``limit`` messages.
+            limit: 1-200 (default 50).
+        """
+        q: dict[str, str] = {"limit": str(limit)}
+        if since_id is not None:
+            q["since_id"] = since_id
+        return self._raw_request("GET", f"/messages/conversations/{username}/tail?{urlencode(q)}")
+
     def mute_conversation(self, username: str) -> dict:
         """Mute a 1:1 conversation with ``username``.
 
@@ -2917,6 +2950,73 @@ def unfollow(self, user_id: str) -> dict:
         """
         return self._raw_request("DELETE", f"/users/{user_id}/follow")
 
+    def get_followers(self, user_id: str, limit: int = 50, offset: int = 0) -> dict:
+        """List a user's followers.
+
+        Args:
+            user_id: The UUID of the user whose followers to list.
+            limit: 1-100 (default 50).
+            offset: Pagination offset.
+        """
+        params = urlencode({"limit": str(limit), "offset": str(offset)})
+        return self._raw_request("GET", f"/users/{user_id}/followers?{params}")
+
+    def get_following(self, user_id: str, limit: int = 50, offset: int = 0) -> dict:
+        """List the users a user follows.
+
+        Args:
+            user_id: The UUID of the user whose follows to list.
+            limit: 1-100 (default 50).
+            offset: Pagination offset.
+        """
+        params = urlencode({"limit": str(limit), "offset": str(offset)})
+        return self._raw_request("GET", f"/users/{user_id}/following?{params}")
+
+    # ── Bookmarks / Post watches ─────────────────────────────────────
+
+    def bookmark_post(self, post_id: str) -> dict:
+        """Bookmark a post for later.
+
+        Args:
+            post_id: The UUID of the post to bookmark.
+        """
+        return self._raw_request("POST", f"/posts/{post_id}/bookmark")
+
+    def unbookmark_post(self, post_id: str) -> dict:
+        """Remove a bookmark from a post.
+
+        Args:
+            post_id: The UUID of the post to unbookmark.
+        """
+        return self._raw_request("DELETE", f"/posts/{post_id}/bookmark")
+
+    def list_bookmarks(self, limit: int = 20, offset: int = 0) -> dict:
+        """List the caller's bookmarked posts.
+
+        Args:
+            limit: 1-100 (default 20).
+            offset: Pagination offset.
+        """
+        params = urlencode({"limit": str(limit), "offset": str(offset)})
+        return self._raw_request("GET", f"/posts/bookmarks/list?{params}")
+
+    def watch_post(self, post_id: str) -> dict:
+        """Watch a post — subscribe to notifications for its new activity
+        without commenting on it.
+
+        Args:
+            post_id: The UUID of the post to watch.
+        """
+        return self._raw_request("POST", f"/posts/{post_id}/watch")
+
+    def unwatch_post(self, post_id: str) -> dict:
+        """Stop watching a post.
+
+        Args:
+            post_id: The UUID of the post to unwatch.
+        """
+        return self._raw_request("DELETE", f"/posts/{post_id}/watch")
+
     # ── Safety / Moderation ─────────────────────────────────────────
 
     def block_user(self, user_id: str) -> dict:
diff --git a/src/colony_sdk/testing.py b/src/colony_sdk/testing.py
index b71bded..8897f76 100644
--- a/src/colony_sdk/testing.py
+++ b/src/colony_sdk/testing.py
@@ -262,6 +262,12 @@ def get_conversation(self, username: str) -> dict:
     def list_conversations(self) -> dict:
         return self._respond("list_conversations", {})
 
+    def conversation_history(self, username: str, before: str, **kwargs: Any) -> dict:
+        return self._respond("conversation_history", {"username": username, "before": before, **kwargs})
+
+    def conversation_tail(self, username: str, **kwargs: Any) -> dict:
+        return self._respond("conversation_tail", {"username": username, **kwargs})
+
     def mute_conversation(self, username: str) -> dict:
         return self._respond("mute_conversation", {"username": username})
 
@@ -545,6 +551,29 @@ def follow(self, user_id: str) -> dict:
     def unfollow(self, user_id: str) -> dict:
         return self._respond("unfollow", {"user_id": user_id})
 
+    def get_followers(self, user_id: str, **kwargs: Any) -> dict:
+        return self._respond("get_followers", {"user_id": user_id, **kwargs})
+
+    def get_following(self, user_id: str, **kwargs: Any) -> dict:
+        return self._respond("get_following", {"user_id": user_id, **kwargs})
+
+    # ── Bookmarks / Post watches ──
+
+    def bookmark_post(self, post_id: str) -> dict:
+        return self._respond("bookmark_post", {"post_id": post_id})
+
+    def unbookmark_post(self, post_id: str) -> dict:
+        return self._respond("unbookmark_post", {"post_id": post_id})
+
+    def list_bookmarks(self, **kwargs: Any) -> dict:
+        return self._respond("list_bookmarks", kwargs)
+
+    def watch_post(self, post_id: str) -> dict:
+        return self._respond("watch_post", {"post_id": post_id})
+
+    def unwatch_post(self, post_id: str) -> dict:
+        return self._respond("unwatch_post", {"post_id": post_id})
+
     # ── Safety / Moderation ──
 
     def block_user(self, user_id: str) -> dict:
diff --git a/tests/test_api_methods.py b/tests/test_api_methods.py
index 15b9ab6..ffbfcdb 100644
--- a/tests/test_api_methods.py
+++ b/tests/test_api_methods.py
@@ -1059,6 +1059,114 @@ def test_unfollow(self, mock_urlopen: MagicMock) -> None:
         assert req.get_method() == "DELETE"
         assert req.full_url == f"{BASE}/users/u1/follow"
 
+    @patch("colony_sdk.client.urlopen")
+    def test_get_followers(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({"items": []})
+        client = _authed_client()
+
+        client.get_followers("u1")
+
+        req = _last_request(mock_urlopen)
+        assert req.get_method() == "GET"
+        assert req.full_url == f"{BASE}/users/u1/followers?limit=50&offset=0"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_get_following_pagination(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({"items": []})
+        client = _authed_client()
+
+        client.get_following("u1", limit=10, offset=20)
+
+        req = _last_request(mock_urlopen)
+        assert req.full_url == f"{BASE}/users/u1/following?limit=10&offset=20"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_bookmark_post(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({})
+        client = _authed_client()
+
+        client.bookmark_post("p1")
+
+        req = _last_request(mock_urlopen)
+        assert req.get_method() == "POST"
+        assert req.full_url == f"{BASE}/posts/p1/bookmark"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_unbookmark_post(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({})
+        client = _authed_client()
+
+        client.unbookmark_post("p1")
+
+        req = _last_request(mock_urlopen)
+        assert req.get_method() == "DELETE"
+        assert req.full_url == f"{BASE}/posts/p1/bookmark"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_list_bookmarks(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({"items": []})
+        client = _authed_client()
+
+        client.list_bookmarks()
+
+        req = _last_request(mock_urlopen)
+        assert req.full_url == f"{BASE}/posts/bookmarks/list?limit=20&offset=0"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_watch_post(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({})
+        client = _authed_client()
+
+        client.watch_post("p1")
+
+        req = _last_request(mock_urlopen)
+        assert req.get_method() == "POST"
+        assert req.full_url == f"{BASE}/posts/p1/watch"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_unwatch_post(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({})
+        client = _authed_client()
+
+        client.unwatch_post("p1")
+
+        req = _last_request(mock_urlopen)
+        assert req.get_method() == "DELETE"
+        assert req.full_url == f"{BASE}/posts/p1/watch"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_conversation_history(self, mock_urlopen: MagicMock) -> None:
+        """``before`` is a required anchor and lands in the query string."""
+        mock_urlopen.return_value = _mock_response({"items": []})
+        client = _authed_client()
+
+        client.conversation_history("alice", before="m9", limit=100)
+
+        req = _last_request(mock_urlopen)
+        assert req.get_method() == "GET"
+        assert req.full_url == f"{BASE}/messages/conversations/alice/history?before=m9&limit=100"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_conversation_tail_with_since_id(self, mock_urlopen: MagicMock) -> None:
+        mock_urlopen.return_value = _mock_response({"items": []})
+        client = _authed_client()
+
+        client.conversation_tail("alice", since_id="m42")
+
+        req = _last_request(mock_urlopen)
+        assert req.full_url == f"{BASE}/messages/conversations/alice/tail?limit=50&since_id=m42"
+
+    @patch("colony_sdk.client.urlopen")
+    def test_conversation_tail_omits_absent_since_id(self, mock_urlopen: MagicMock) -> None:
+        """Without ``since_id`` the param is left out, not sent as null."""
+        mock_urlopen.return_value = _mock_response({"items": []})
+        client = _authed_client()
+
+        client.conversation_tail("alice")
+
+        req = _last_request(mock_urlopen)
+        assert req.full_url == f"{BASE}/messages/conversations/alice/tail?limit=50"
+
 
 # ---------------------------------------------------------------------------
 # Safety / Moderation
diff --git a/tests/test_async_client.py b/tests/test_async_client.py
index 45f6373..f99e527 100644
--- a/tests/test_async_client.py
+++ b/tests/test_async_client.py
@@ -1033,6 +1033,64 @@ def handler(request: httpx.Request) -> httpx.Response:
         await client.unfollow("u2")
         assert seen["method"] == "DELETE"
 
+    async def test_get_followers_and_following(self) -> None:
+        seen: list = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen.append(str(request.url))
+            return _json_response({"items": []})
+
+        client = _make_client(handler)
+        await client.get_followers("u2")
+        await client.get_following("u2", limit=5, offset=10)
+        assert seen[0].endswith("/users/u2/followers?limit=50&offset=0")
+        assert seen[1].endswith("/users/u2/following?limit=5&offset=10")
+
+    async def test_bookmark_watch_roundtrip(self) -> None:
+        seen: list = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen.append((request.method, request.url.path))
+            return _json_response({})
+
+        client = _make_client(handler)
+        await client.bookmark_post("p1")
+        await client.unbookmark_post("p1")
+        await client.watch_post("p1")
+        await client.unwatch_post("p1")
+        assert seen == [
+            ("POST", "/api/v1/posts/p1/bookmark"),
+            ("DELETE", "/api/v1/posts/p1/bookmark"),
+            ("POST", "/api/v1/posts/p1/watch"),
+            ("DELETE", "/api/v1/posts/p1/watch"),
+        ]
+
+    async def test_list_bookmarks(self) -> None:
+        seen: dict = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen["url"] = str(request.url)
+            return _json_response({"items": []})
+
+        client = _make_client(handler)
+        await client.list_bookmarks(limit=5)
+        assert seen["url"].endswith("/posts/bookmarks/list?limit=5&offset=0")
+
+    async def test_conversation_history_and_tail(self) -> None:
+        seen: list = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen.append(str(request.url))
+            return _json_response({"items": []})
+
+        client = _make_client(handler)
+        await client.conversation_history("alice", before="m9")
+        await client.conversation_tail("alice", since_id="m42", limit=10)
+        await client.conversation_tail("alice")
+        assert seen[0].endswith("/messages/conversations/alice/history?before=m9&limit=200")
+        assert seen[1].endswith("/messages/conversations/alice/tail?limit=10&since_id=m42")
+        assert seen[2].endswith("/messages/conversations/alice/tail?limit=50")
+
     async def test_block_user(self) -> None:
         seen: dict = {}
 
diff --git a/tests/test_testing.py b/tests/test_testing.py
index 3def08e..5aea378 100644
--- a/tests/test_testing.py
+++ b/tests/test_testing.py
@@ -168,6 +168,41 @@ def test_directory(self) -> None:
         assert "items" in result
         assert client.calls[-1] == ("directory", {"query": "test"})
 
+    def test_follow_graph_reads(self) -> None:
+        client = MockColonyClient()
+        client.get_followers("u1", limit=5)
+        assert client.calls[-1] == ("get_followers", {"user_id": "u1", "limit": 5})
+        client.get_following("u1")
+        assert client.calls[-1] == ("get_following", {"user_id": "u1"})
+
+    def test_bookmarks_and_watches(self) -> None:
+        client = MockColonyClient()
+        client.bookmark_post("p1")
+        client.unbookmark_post("p1")
+        client.list_bookmarks(limit=5)
+        client.watch_post("p1")
+        client.unwatch_post("p1")
+        assert [name for name, _ in client.calls[-5:]] == [
+            "bookmark_post",
+            "unbookmark_post",
+            "list_bookmarks",
+            "watch_post",
+            "unwatch_post",
+        ]
+
+    def test_conversation_history_and_tail(self) -> None:
+        client = MockColonyClient()
+        client.conversation_history("alice", before="m9")
+        assert client.calls[-1] == (
+            "conversation_history",
+            {"username": "alice", "before": "m9"},
+        )
+        client.conversation_tail("alice", since_id="m42")
+        assert client.calls[-1] == (
+            "conversation_tail",
+            {"username": "alice", "since_id": "m42"},
+        )
+
     def test_last_rate_limit_is_none(self) -> None:
         client = MockColonyClient()
         assert client.last_rate_limit is None