diff --git a/.github/plans/aiograpi-direct-readonly-design.md b/.github/plans/aiograpi-direct-readonly-design.md
new file mode 100644
index 0000000..355f1b9
--- /dev/null
+++ b/.github/plans/aiograpi-direct-readonly-design.md
@@ -0,0 +1,113 @@
+# Read-only aiograpi Direct Design
+
+Issue: https://github.com/subzeroid/insto/issues/4
+
+## Goal
+
+Expose the safe read-only subset of aiograpi Direct through insto without widening the product boundary into account automation.
+
+This first PR implements only:
+
+- `/direct [N]`: list recent Direct threads for the authenticated aiograpi account.
+- `/direct-thread <thread_id> [N]`: show recent messages in one Direct thread.
+- JSON export for both commands.
+
+The PR does not implement Direct search, saved collections, personal feed, private GraphQL rewrites, polling, notifications, or any write operation.
+
+## Product Boundary
+
+Direct support is read-only. The CLI must not expose commands for sending messages, reacting, deleting, unsending, marking seen, muting, approving requests, updating titles, uploading attachments, or sharing media/profile/story objects.
+
+This matters because insto is an OSINT read tool. Direct write actions turn it into an account automation tool with much higher account-risk and abuse potential.
+
+## Command UX
+
+`/direct [N]` lists threads. Default count is 20. It renders a compact table with:
+
+- thread id
+- title or participant usernames
+- participant usernames
+- last activity timestamp
+- message count returned by aiograpi for the thread preview
+- pending, archived, muted, group flags
+
+`/direct-thread <thread_id> [N]` lists messages from a specific thread. Default count is 20. It renders:
+
+- timestamp
+- sender user id
+- item type
+- text preview when text exists
+- shared media/code markers when the SDK exposes them cleanly
+
+Both commands support `--json`. Both reject `--csv` through the existing non-flat export guard unless we explicitly add flat rows in a later PR.
+
+## Data Model
+
+Add DTOs to `insto.models`:
+
+- `DirectThread`: stable thread metadata plus participants and preview messages.
+- `DirectMessage`: stable message metadata plus a read-only summary of content references.
+
+DTOs store only plain Python values. They never expose raw aiograpi objects above the backend layer.
+
+The message body is still useful OSINT data, so JSON export includes text. The backend and commands must not log raw Direct payloads or message text.
+
+## Backend Contract
+
+Add optional methods to `OSINTBackend`, not abstract methods:
+
+- `iter_direct_threads(limit: int | None = None)`
+- `iter_direct_messages(thread_id: str, *, limit: int | None = None)`
+
+Default behavior raises `BackendError("needs aiograpi backend")`. This keeps HikerAPI and future non-Instagram backends from inheriting raw `NotImplementedError` behavior.
+
+`AiograpiBackend` implements the methods via:
+
+- `client.direct_threads(amount=limit, thread_message_limit=1)`
+- `client.direct_messages(thread_id, amount=limit)`
+
+The implementation keeps lazy login and `_translate` error mapping unchanged.
+
+## Capability Gate
+
+Introduce a backend capability token for Direct reads, for example `direct_read`.
+
+The aiograpi backend advertises it. HikerAPI and fake production backend do not. Command dispatch then rejects `/direct` and `/direct-thread` on unsupported backends before making any backend call.
+
+Unit tests may enable the capability on `tests.fakes.FakeBackend` when testing command rendering.
+
+## Tests
+
+All normal tests stay offline. Live Instagram checks remain opt-in only.
+
+Coverage required in this PR:
+
+- DTO dataclass shape and `dataclasses.asdict` behavior.
+- aiograpi mapper behavior for text messages, media-share messages, and empty/non-text messages.
+- backend optional-method defaults raise typed `BackendError`.
+- command capability gate rejects unsupported backends clearly.
+- `/direct` renders thread rows using `FakeBackend`.
+- `/direct-thread` renders message rows using `FakeBackend`.
+- JSON export writes the expected envelope and stdout form.
+- CSV rejection remains clear.
+
+## Docs
+
+Update:
+
+- `docs/cli-reference.md`
+- `docs/backends.md`
+- `docs/roadmap.md`
+
+Docs must state that Direct support requires `insto[aiograpi]`, logs into a real Instagram account, carries account-risk, and is read-only.
+
+README only changes if the public command table needs to include Direct in the top-level command surface.
+
+## Acceptance Criteria
+
+- `/direct` works in REPL and one-shot dispatch on aiograpi.
+- `/direct-thread` works in REPL and one-shot dispatch on aiograpi.
+- Unsupported backends fail with a clear user-facing missing-capability message.
+- JSON export works for both commands.
+- No Direct write operation appears in command registration, facade API, docs, or tests.
+- `ruff check`, `ruff format --check`, `mypy insto`, `pytest --cov=insto --cov-fail-under=75`, and `mkdocs build --strict` pass.
diff --git a/.github/plans/aiograpi-direct-readonly-plan.md b/.github/plans/aiograpi-direct-readonly-plan.md
new file mode 100644
index 0000000..0960e4c
--- /dev/null
+++ b/.github/plans/aiograpi-direct-readonly-plan.md
@@ -0,0 +1,550 @@
+# Read-only aiograpi Direct Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add read-only Direct thread and message commands for the aiograpi backend.
+
+**Architecture:** Add Direct DTOs in `insto.models`, optional read methods on `OSINTBackend`, mapper helpers for aiograpi Direct payloads, thin facade methods, and a new `insto.commands.direct` module registered from `insto.commands.__init__`. The command layer exports JSON and renders tables, while capability gating prevents unsupported backends from executing the commands.
+
+**Tech Stack:** Python dataclasses, async iterators, aiograpi 0.9.6, Rich rendering, pytest-asyncio, strict mypy, ruff.
+
+---
+
+## File Structure
+
+- `insto/models.py`: add `DirectThread` and `DirectMessage` DTOs.
+- `insto/backends/_base.py`: add optional Direct methods and import DTOs.
+- `insto/backends/_aiograpi_map.py`: add mapper helpers for aiograpi Direct objects.
+- `insto/backends/aiograpi.py`: advertise `direct_read`, call aiograpi Direct read methods, map results.
+- `insto/service/facade.py`: add `direct_threads()` and `direct_messages()`.
+- `insto/commands/direct.py`: add `/direct` and `/direct-thread`.
+- `insto/commands/__init__.py`: import the new command module.
+- `tests/fakes.py`: add Direct fixtures, errors, and iterators.
+- `tests/test_models.py`: cover DTO shape.
+- `tests/test_backend_contract.py`: cover optional default errors.
+- `tests/test_aiograpi_map.py`: cover Direct mappers.
+- `tests/test_commands_direct.py`: cover command render/export/capability behavior.
+- `docs/cli-reference.md`, `docs/backends.md`, `docs/roadmap.md`, optionally `README.md`: document the new read-only commands.
+
+## Task 1: Direct DTOs
+
+**Files:**
+- Modify: `insto/models.py`
+- Test: `tests/test_models.py`
+
+- [ ] **Step 1: Write failing DTO tests**
+
+Add tests that import `DirectThread`, `DirectMessage`, and `User`, construct representative values, and assert `dataclasses.asdict()` returns stable plain dictionaries.
+
+Expected fields:
+
+```python
+DirectMessage(
+    pk="m1",
+    thread_id="t1",
+    sender_pk="100",
+    timestamp=1_700_000_000,
+    item_type="text",
+    text="hello",
+    media_pk=None,
+    media_code=None,
+    link_url=None,
+)
+
+DirectThread(
+    pk="t1",
+    title="Alice",
+    users=[User(pk="100", username="alice")],
+    last_activity_at=1_700_000_000,
+    message_count=1,
+    is_group=False,
+    is_pending=False,
+    is_archived=False,
+    is_muted=False,
+    messages=[message],
+)
+```
+
+- [ ] **Step 2: Run the DTO tests and verify RED**
+
+Run:
+
+```bash
+uv run pytest tests/test_models.py -q
+```
+
+Expected: import failure for `DirectThread` / `DirectMessage`.
+
+- [ ] **Step 3: Implement DTOs**
+
+Add dataclasses near the other Instagram DTOs in `insto/models.py`:
+
+```python
+@dataclass(slots=True)
+class DirectMessage:
+    """Read-only Direct message summary."""
+
+    pk: str
+    thread_id: str
+    sender_pk: str
+    timestamp: int
+    item_type: str = ""
+    text: str | None = None
+    media_pk: str | None = None
+    media_code: str | None = None
+    link_url: str | None = None
+
+
+@dataclass(slots=True)
+class DirectThread:
+    """Read-only Direct thread summary."""
+
+    pk: str
+    title: str
+    users: list[User] = field(default_factory=list)
+    last_activity_at: int = 0
+    message_count: int = 0
+    is_group: bool = False
+    is_pending: bool = False
+    is_archived: bool = False
+    is_muted: bool = False
+    messages: list[DirectMessage] = field(default_factory=list)
+```
+
+- [ ] **Step 4: Run DTO tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/test_models.py -q
+uv run mypy insto
+```
+
+Commit:
+
+```bash
+git add insto/models.py tests/test_models.py
+git commit -m "feat: add direct read DTOs"
+```
+
+## Task 2: Backend Contract and FakeBackend
+
+**Files:**
+- Modify: `insto/backends/_base.py`
+- Modify: `tests/fakes.py`
+- Test: `tests/test_backend_contract.py`
+
+- [ ] **Step 1: Write failing contract tests**
+
+Add tests that call `OSINTBackend.iter_direct_threads()` and `iter_direct_messages()` through a minimal concrete test backend and assert `BackendError` with `needs aiograpi backend`.
+
+Add fake-backend tests that configure `direct_threads` and `direct_messages` data and assert paging respects `limit`.
+
+- [ ] **Step 2: Run contract tests and verify RED**
+
+Run:
+
+```bash
+uv run pytest tests/test_backend_contract.py -q
+```
+
+Expected: missing methods or raw `NotImplementedError`.
+
+- [ ] **Step 3: Implement optional backend methods**
+
+In `OSINTBackend`, import `DirectMessage` and `DirectThread`, then add:
+
+```python
+def iter_direct_threads(self, *, limit: int | None = None) -> AsyncIterator[DirectThread]:
+    """Iterate read-only Direct threads. Default requires aiograpi."""
+    raise BackendError("needs aiograpi backend")
+
+
+def iter_direct_messages(
+    self, thread_id: str, *, limit: int | None = None
+) -> AsyncIterator[DirectMessage]:
+    """Iterate read-only Direct messages in one thread. Default requires aiograpi."""
+    raise BackendError("needs aiograpi backend")
+```
+
+- [ ] **Step 4: Extend `tests.fakes.FakeBackend`**
+
+Add `iter_direct_threads` and `iter_direct_messages` error slots, fixture fields, request logging, and paged iterators.
+
+Use dictionaries keyed by thread id:
+
+```python
+direct_threads: list[DirectThread] = field(default_factory=list)
+direct_messages: dict[str, list[DirectMessage]] = field(default_factory=dict)
+```
+
+- [ ] **Step 5: Run contract tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/test_backend_contract.py -q
+uv run pytest tests/test_commands_base.py::test_dispatch_rejects_command_requiring_missing_capability -q
+uv run mypy insto
+```
+
+Commit:
+
+```bash
+git add insto/backends/_base.py tests/fakes.py tests/test_backend_contract.py
+git commit -m "feat: add direct backend contract"
+```
+
+## Task 3: aiograpi Direct Mappers
+
+**Files:**
+- Modify: `insto/backends/_aiograpi_map.py`
+- Test: `tests/test_aiograpi_map.py`
+
+- [ ] **Step 1: Write failing mapper tests**
+
+Use lightweight objects with attributes matching aiograpi Direct models. Cover:
+
+- text message maps `text`.
+- media share maps `media_pk` and `media_code` when present.
+- non-text message maps `item_type` and leaves text fields `None`.
+- thread maps users, title, flags, last activity timestamp, and preview messages.
+
+- [ ] **Step 2: Run mapper tests and verify RED**
+
+Run:
+
+```bash
+uv run pytest tests/test_aiograpi_map.py -q
+```
+
+Expected: mapper function import failure.
+
+- [ ] **Step 3: Implement mapper helpers**
+
+Add:
+
+```python
+def _direct_ts(raw: Any, attr: str) -> int:
+    value = getattr(raw, attr, None)
+    if hasattr(value, "timestamp"):
+        return int(value.timestamp())
+    if isinstance(value, int | float):
+        return int(value)
+    return 0
+
+
+def _maybe_str(value: Any) -> str | None:
+    if value is None:
+        return None
+    return str(value)
+
+
+def map_direct_message(raw: Any, *, thread_id: str | None = None) -> DirectMessage:
+    media = getattr(raw, "media_share", None) or getattr(raw, "clip", None)
+    link = getattr(raw, "link", None)
+    return DirectMessage(
+        pk=str(getattr(raw, "id", "")),
+        thread_id=str(getattr(raw, "thread_id", None) or thread_id or ""),
+        sender_pk=str(getattr(raw, "user_id", "") or ""),
+        timestamp=_direct_ts(raw, "timestamp"),
+        item_type=str(getattr(raw, "item_type", "") or ""),
+        text=getattr(raw, "text", None),
+        media_pk=_maybe_str(getattr(media, "pk", None) or getattr(media, "id", None))
+        if media is not None
+        else None,
+        media_code=_maybe_str(getattr(media, "code", None)) if media is not None else None,
+        link_url=_maybe_str(getattr(link, "url", None)) if link is not None else None,
+    )
+
+
+def map_direct_thread(raw: Any) -> DirectThread:
+    messages = [map_direct_message(msg, thread_id=str(getattr(raw, "id", ""))) for msg in getattr(raw, "messages", [])]
+    users = [
+        User(
+            pk=str(getattr(user, "pk", "")),
+            username=str(getattr(user, "username", "") or ""),
+            full_name=str(getattr(user, "full_name", "") or ""),
+            is_private=bool(getattr(user, "is_private", False) or False),
+        )
+        for user in getattr(raw, "users", [])
+    ]
+    return DirectThread(
+        pk=str(getattr(raw, "id", None) or getattr(raw, "pk", "")),
+        title=str(getattr(raw, "thread_title", "") or ""),
+        users=users,
+        last_activity_at=_direct_ts(raw, "last_activity_at"),
+        message_count=len(messages),
+        is_group=bool(getattr(raw, "is_group", False)),
+        is_pending=bool(getattr(raw, "pending", False)),
+        is_archived=bool(getattr(raw, "archived", False)),
+        is_muted=bool(getattr(raw, "muted", False)),
+        messages=messages,
+    )
+```
+
+Rules:
+
+- Convert `datetime` values to Unix seconds with `int(dt.timestamp())`.
+- Prefer `raw.thread_id`, fallback to explicit `thread_id`, fallback to empty string.
+- Convert `raw.user_id` to `sender_pk`, fallback to empty string.
+- Extract `media_pk` and `media_code` from `media_share` or `clip` only when attributes exist.
+- Extract `link_url` from `link.url` when present.
+- Do not retain raw aiograpi objects.
+
+- [ ] **Step 4: Run mapper tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/test_aiograpi_map.py -q
+uv run mypy insto
+```
+
+Commit:
+
+```bash
+git add insto/backends/_aiograpi_map.py tests/test_aiograpi_map.py
+git commit -m "feat: map aiograpi direct payloads"
+```
+
+## Task 4: Aiograpi Backend Methods
+
+**Files:**
+- Modify: `insto/backends/aiograpi.py`
+- Test: `tests/test_aiograpi_map.py` and existing backend import tests
+
+- [ ] **Step 1: Write backend behavior tests if a client stub exists**
+
+If the current test suite has aiograpi backend client stubs, add tests proving:
+
+- `iter_direct_threads(limit=3)` calls `client.direct_threads(amount=3, thread_message_limit=1)`.
+- `iter_direct_messages("123", limit=5)` calls `client.direct_messages(123, amount=5)`.
+- errors pass through `_translate`.
+
+If there is no existing client-stub pattern, keep this coverage in mapper and command tests to avoid inventing a large test harness in this PR.
+
+- [ ] **Step 2: Implement capability and methods**
+
+In `AiograpiBackend`:
+
+```python
+capabilities = frozenset({"followed", "direct_read"})
+```
+
+Add async generator methods:
+
+```python
+async def iter_direct_threads(
+    self, *, limit: int | None = None
+) -> AsyncIterator[DirectThread]:
+    amount = limit if limit is not None and limit > 0 else 20
+    raws = await self._call(
+        lambda: self._client.direct_threads(amount=amount, thread_message_limit=1)
+    )
+    for raw in raws:
+        yield map_direct_thread(raw)
+
+
+async def iter_direct_messages(
+    self, thread_id: str, *, limit: int | None = None
+) -> AsyncIterator[DirectMessage]:
+    amount = limit if limit is not None and limit > 0 else 20
+    raws = await self._call(lambda: self._client.direct_messages(int(thread_id), amount=amount))
+    for raw in raws:
+        yield map_direct_message(raw, thread_id=thread_id)
+```
+
+Reject non-numeric thread ids with `BackendError(f"invalid direct thread id: {thread_id!r}")`.
+
+- [ ] **Step 3: Run backend-adjacent tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/test_aiograpi_map.py tests/test_backend_contract.py -q
+uv run mypy insto
+```
+
+Commit:
+
+```bash
+git add insto/backends/aiograpi.py tests/test_aiograpi_map.py
+git commit -m "feat: wire aiograpi direct read methods"
+```
+
+## Task 5: Facade and Commands
+
+**Files:**
+- Modify: `insto/service/facade.py`
+- Create: `insto/commands/direct.py`
+- Modify: `insto/commands/__init__.py`
+- Test: `tests/test_commands_direct.py`
+
+- [ ] **Step 1: Write failing command tests**
+
+Create `tests/test_commands_direct.py` with tests for:
+
+- `/direct 2` renders two threads.
+- `/direct-thread t1 2` renders two messages.
+- `/direct --json` writes `output/direct/direct.json` or equivalent default path chosen by `default_export_path`.
+- `/direct --json -` writes a valid JSON envelope to stdout.
+- `/direct --csv -` is rejected by the existing CSV guard.
+- unsupported backend without `direct_read` fails before backend calls.
+
+- [ ] **Step 2: Run command tests and verify RED**
+
+Run:
+
+```bash
+uv run pytest tests/test_commands_direct.py -q
+```
+
+Expected: unknown command `/direct`.
+
+- [ ] **Step 3: Add facade methods**
+
+Add to `OsintFacade`:
+
+```python
+async def direct_threads(self, *, limit: int = 20) -> list[DirectThread]:
+    return [t async for t in self.backend.iter_direct_threads(limit=limit)]
+
+
+async def direct_messages(self, thread_id: str, *, limit: int = 20) -> list[DirectMessage]:
+    return [m async for m in self.backend.iter_direct_messages(thread_id, limit=limit)]
+```
+
+- [ ] **Step 4: Add direct command module**
+
+Implement `insto/commands/direct.py` with:
+
+- `_add_direct_args(parser)`: optional count default 20.
+- `_add_direct_thread_args(parser)`: `thread_id` plus optional count default 20.
+- `_resolve_count(ctx, default=20)`: global `--limit` wins.
+- `direct_cmd`: `@command("direct", "List read-only Direct threads (aiograpi only)", add_args=_add_direct_args, requires=("direct_read",))`.
+- `direct_thread_cmd`: `@command("direct-thread", "Show read-only Direct messages for one thread (aiograpi only)", add_args=_add_direct_thread_args, requires=("direct_read",))`.
+
+Use `dataclasses.asdict()` for JSON export.
+
+- [ ] **Step 5: Register command module**
+
+Add to `insto/commands/__init__.py`:
+
+```python
+from insto.commands import direct as _direct  # noqa: F401  (registers commands)
+```
+
+- [ ] **Step 6: Run command tests and commit**
+
+Run:
+
+```bash
+uv run pytest tests/test_commands_direct.py tests/test_commands_base.py -q
+uv run mypy insto
+```
+
+Commit:
+
+```bash
+git add insto/service/facade.py insto/commands/direct.py insto/commands/__init__.py tests/test_commands_direct.py
+git commit -m "feat: add read-only direct commands"
+```
+
+## Task 6: Docs and Final Verification
+
+**Files:**
+- Modify: `docs/cli-reference.md`
+- Modify: `docs/backends.md`
+- Modify: `docs/roadmap.md`
+- Modify: `README.md` if the command table should include Direct
+
+- [ ] **Step 1: Update docs**
+
+Document:
+
+- `/direct [N]`
+- `/direct-thread <thread_id> [N]`
+- aiograpi-only requirement
+- read-only boundary
+- account-ban risk
+- no Direct write operations
+
+- [ ] **Step 2: Run docs build**
+
+Run:
+
+```bash
+uv run --extra docs mkdocs build --strict --site-dir /tmp/insto-mkdocs-site
+```
+
+Expected: exit code 0. Existing Material for MkDocs warning is acceptable if the build exits 0.
+
+- [ ] **Step 3: Run full verification**
+
+Run:
+
+```bash
+uv run ruff check
+uv run ruff format --check
+uv run mypy insto
+uv run pytest --cov=insto --cov-fail-under=75
+uv run --extra docs mkdocs build --strict --site-dir /tmp/insto-mkdocs-site
+```
+
+- [ ] **Step 4: Guardrail scan for Direct write operations**
+
+Run:
+
+```bash
+rg -n "direct_(send|answer|delete|unsend|like|unlike|seen|mute|approve|hide|create|update|upload|share)|message_seen|mark_unread|send_seen" insto docs tests
+```
+
+Expected: no production command exposure. Mapper tests may mention forbidden method names only if asserting they are absent.
+
+- [ ] **Step 5: Commit docs**
+
+Commit:
+
+```bash
+git add docs/cli-reference.md docs/backends.md docs/roadmap.md README.md
+git commit -m "docs: document read-only direct commands"
+```
+
+## Task 7: PR Prep
+
+**Files:**
+- No source changes unless verification finds issues.
+
+- [ ] **Step 1: Push branch**
+
+Run:
+
+```bash
+git push -u origin feat/aiograpi-direct-readonly
+```
+
+- [ ] **Step 2: Create PR**
+
+Use title:
+
+```text
+feat: add read-only aiograpi direct commands
+```
+
+PR body must include:
+
+- Closes #4 partially.
+- Scope: `/direct`, `/direct-thread`, JSON export, read-only aiograpi Direct.
+- Non-goals: Direct writes, Direct search, saved/feed, private GraphQL rewrites.
+- Verification commands and outputs.
+
+- [ ] **Step 3: Wait for CI**
+
+Run:
+
+```bash
+gh pr checks --watch --fail-fast
+```
+
+Expected: CI success.
diff --git a/README.md b/README.md
index 3545e36..2a5f085 100644
--- a/README.md
+++ b/README.md
@@ -231,6 +231,7 @@ echo 'fpath+=~/.insto && autoload -Uz compinit && compinit' >> ~/.zshrc
 | **Content** | `hashtags` `mentions` `captions` `likes` `timeline` 🔥 | top hashtags / @mentions / captions, like-count stats, **`timeline` = posting-cadence histogram (hour-of-day + day-of-week)** |
 | **Interactions** | `comments` `wcommented` `wliked` `wtagged` `fans` 🔥 | per-post + aggregated comments, top commenters / likers / taggers, **`fans` = weighted superfan ranking (likes + 3×comments)** |
 | **Discovery** | `resolve` | expand `instagram.com/share/...` short-links to canonical URLs (aiograpi only) |
+| **Direct** | `direct` `direct-thread` | read-only Direct threads and messages (aiograpi only) |
 | **Watch / diff** | `watch` `unwatch` `watching` `diff` `history` | poll-based snapshot diffing; cli-history |
 | **Operational** | `quota` `health` `config` `purge` | balance + p50/p95 latency + error breakdown, effective config with origins, sqlite/cache cleanup |
 | **Session** | `target` `current` `clear` | active-target plumbing for the REPL |
diff --git a/docs/backends.md b/docs/backends.md
index cf9dd13..9727a88 100644
--- a/docs/backends.md
+++ b/docs/backends.md
@@ -11,7 +11,7 @@ The contract lives in `insto/backends/_base.py:OSINTBackend`. Two implementation
 | Account ban risk | None | Real |
 | Stability | High | Brittle (Instagram churn) |
 | Sees private accounts you follow | No | Yes |
-| Sees DMs / saved feed | No | aiograpi has SDK methods; insto does not expose CLI commands yet |
+| Sees DMs / saved feed | No | Read-only Direct threads/messages exposed; saved/feed still planned |
 | Quota visibility | Yes (`/sys/balance`) | No |
 | Install footprint | base | `pip install 'insto[aiograpi]'` |
 
@@ -86,11 +86,13 @@ Then run `insto setup`, pick `aiograpi`, paste your Instagram username + passwor
 
 What works on aiograpi (>= 0.9.6):
 
-- Every command — `/info`, `/posts`, `/reels`, `/stories`, `/highlights`, `/followers`, `/followings`, `/mutuals`, `/comments`, `/captions`, `/likes`, `/wcommented`, `/hashtags`, `/mentions`, `/locations`, `/tagged`, `/similar`, `/dossier`.
+- Every command — `/info`, `/posts`, `/reels`, `/stories`, `/highlights`, `/followers`, `/followings`, `/mutuals`, `/comments`, `/captions`, `/likes`, `/wcommented`, `/hashtags`, `/mentions`, `/locations`, `/tagged`, `/similar`, `/direct`, `/direct-thread`, `/dossier`.
 - Reads private profiles you follow.
 - Login is **lazy** — the constructor stores credentials, the actual `client.login()` fires on the first network call. The session is then dumped to `~/.insto/aiograpi.session.json` (mode `0600`); subsequent runs reuse it without re-authenticating.
 
-aiograpi 0.9.x also exposes a much larger Direct, private GraphQL, music, archive, and collection surface. Insto intentionally exposes only read-oriented OSINT commands today. Read-only Direct inbox and saved-collection support are tracked as follow-up work in the [Roadmap](roadmap.md).
+aiograpi 0.9.x also exposes a much larger Direct, private GraphQL, music, archive, and collection surface. Insto intentionally exposes only read-oriented OSINT commands today: Direct support is limited to `/direct` and `/direct-thread`; saved-collection, feed, and private GraphQL audits remain tracked in the [Roadmap](roadmap.md).
+
+Direct commands are read-only. Insto does not expose send, reaction, seen, unsend, mute, approve, upload, or title-update operations.
 
 ### Account-ban risk
 
diff --git a/docs/cli-reference.md b/docs/cli-reference.md
index ea568ab..389c90d 100644
--- a/docs/cli-reference.md
+++ b/docs/cli-reference.md
@@ -120,6 +120,15 @@ All take a bounded post window (default 50). `/wliked` and `/fans` make `N` (or
 |---|---|
 | `/resolve <url>` | Expand `instagram.com/share/...` short-link → canonical URL. aiograpi only. |
 
+### Direct
+
+| Command | Purpose |
+|---|---|
+| `/direct [N]` | List recent Direct threads for the logged-in aiograpi account (default 20). |
+| `/direct-thread <thread_id> [N]` | Show recent messages in one Direct thread (default 20). |
+
+Direct commands are aiograpi-only, read-only, and support JSON export. They intentionally expose no send, reaction, seen, unsend, mute, approve, upload, or title-update flows.
+
 ### Batch
 
 ```sh
diff --git a/docs/index.md b/docs/index.md
index 437f971..37fa41a 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -65,6 +65,7 @@ See [Backends](backends.md) for the full breakdown.
 | **Content** | `hashtags` `mentions` `captions` `likes` `timeline` 🔥 |
 | **Interactions** | `comments` `wcommented` `wliked` `wtagged` `fans` 🔥 |
 | **Discovery** | `resolve` |
+| **Direct** | `direct` `direct-thread` |
 | **Watch / diff** | `watch` `unwatch` `watching` `diff` `history` |
 | **Operational** | `quota` `health` `config` `purge` |
 | **Session** | `target` `current` `clear` |
diff --git a/docs/roadmap.md b/docs/roadmap.md
index 241e0b8..aa93ce7 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -4,9 +4,9 @@ Deferred work that is still relevant after the aiograpi 0.9.x update.
 
 ## aiograpi follow-ups
 
-### Read-only Direct inbox
+### Direct inbox search
 
-Add read-only commands for Direct: list threads, show recent messages in a thread, and search threads by participant. Do not add send, reaction, or unsend commands to the core CLI.
+`/direct` and `/direct-thread` are shipped. Follow-up: add read-only search/filter by participant if aiograpi's search surface is stable enough. Do not add send, reaction, or unsend commands to the core CLI.
 
 Why: aiograpi 0.9.x synced the current Direct API surface, including message requests, single-message lookup, reactions, title updates, unsend, and voice/video attachments. For insto's OSINT surface, only the read-only subset fits the product boundary.
 
diff --git a/insto/backends/_aiograpi_map.py b/insto/backends/_aiograpi_map.py
index ff002c0..d05b2c0 100644
--- a/insto/backends/_aiograpi_map.py
+++ b/insto/backends/_aiograpi_map.py
@@ -38,6 +38,8 @@
 from insto.exceptions import SchemaDrift
 from insto.models import (
     Comment,
+    DirectMessage,
+    DirectThread,
     Highlight,
     HighlightItem,
     Post,
@@ -184,6 +186,74 @@ def map_user_short(user: Any) -> User:
     )
 
 
+# ---- direct ----------------------------------------------------------------
+
+
+def _direct_media_ref(message: Any) -> tuple[str | None, str | None]:
+    media = _opt(message, "media_share") or _opt(message, "clip")
+    if media is None:
+        return None, None
+    return _opt_str(_opt(media, "pk") or _opt(media, "id")), _opt_str(_opt(media, "code"))
+
+
+def map_direct_message(message: Any, *, thread_id: str | None = None) -> DirectMessage:
+    """Map an aiograpi `DirectMessage` to a read-only DTO."""
+    pk = _require(message, "id", "direct_message")
+    timestamp = _to_unix(
+        _require(message, "timestamp", "direct_message"),
+        endpoint="direct_message",
+        field="timestamp",
+    )
+    media_pk, media_code = _direct_media_ref(message)
+    link = _opt(message, "link")
+    return DirectMessage(
+        pk=str(pk),
+        thread_id=str(_opt(message, "thread_id") or thread_id or ""),
+        sender_pk=str(_opt(message, "user_id") or ""),
+        timestamp=timestamp,
+        item_type=str(_opt(message, "item_type") or ""),
+        text=_opt_str(_opt(message, "text")),
+        media_pk=media_pk,
+        media_code=media_code,
+        link_url=_opt_str(_opt(link, "url")) if link is not None else None,
+    )
+
+
+def map_direct_thread(thread: Any) -> DirectThread:
+    """Map an aiograpi `DirectThread` preview to a read-only DTO."""
+    pk = _require(thread, "id", "direct_thread")
+    messages = [
+        map_direct_message(message, thread_id=str(pk))
+        for message in (_opt(thread, "messages") or [])
+    ]
+    users = [
+        User(
+            pk=str(_opt(user, "pk") or ""),
+            username=str(_opt(user, "username") or ""),
+            full_name=str(_opt(user, "full_name") or ""),
+            is_private=bool(_opt(user, "is_private")),
+            is_verified=bool(_opt(user, "is_verified")),
+        )
+        for user in (_opt(thread, "users") or [])
+    ]
+    return DirectThread(
+        pk=str(pk),
+        title=str(_opt(thread, "thread_title") or ""),
+        users=users,
+        last_activity_at=_to_unix(
+            _require(thread, "last_activity_at", "direct_thread"),
+            endpoint="direct_thread",
+            field="last_activity_at",
+        ),
+        message_count=len(messages),
+        is_group=bool(_opt(thread, "is_group")),
+        is_pending=bool(_opt(thread, "pending")),
+        is_archived=bool(_opt(thread, "archived")),
+        is_muted=bool(_opt(thread, "muted")),
+        messages=messages,
+    )
+
+
 # ---- media (post) ----------------------------------------------------------
 
 
diff --git a/insto/backends/_base.py b/insto/backends/_base.py
index 68f1245..ba016c4 100644
--- a/insto/backends/_base.py
+++ b/insto/backends/_base.py
@@ -19,8 +19,11 @@
 from collections.abc import AsyncIterator
 from typing import Any
 
+from insto.exceptions import BackendError
 from insto.models import (
     Comment,
+    DirectMessage,
+    DirectThread,
     Highlight,
     HighlightItem,
     Place,
@@ -169,6 +172,16 @@ def iter_place_posts(self, place_pk: str, *, limit: int | None = None) -> AsyncI
         """Iterate top posts at a given Instagram location pk. Default raises."""
         raise NotImplementedError("this backend does not implement place media listing")
 
+    def iter_direct_threads(self, *, limit: int | None = None) -> AsyncIterator[DirectThread]:
+        """Iterate read-only Direct threads. Default requires aiograpi."""
+        raise BackendError("needs aiograpi backend")
+
+    def iter_direct_messages(
+        self, thread_id: str, *, limit: int | None = None
+    ) -> AsyncIterator[DirectMessage]:
+        """Iterate read-only Direct messages in one thread. Default requires aiograpi."""
+        raise BackendError("needs aiograpi backend")
+
     @abstractmethod
     def get_quota(self) -> Quota:
         """Return the last-known quota state for the backend."""
diff --git a/insto/backends/aiograpi.py b/insto/backends/aiograpi.py
index bc20277..2e2f533 100644
--- a/insto/backends/aiograpi.py
+++ b/insto/backends/aiograpi.py
@@ -47,7 +47,19 @@
     SchemaDrift,
     Transient,
 )
-from insto.models import Comment, Highlight, HighlightItem, Place, Post, Profile, Quota, Story, User
+from insto.models import (
+    Comment,
+    DirectMessage,
+    DirectThread,
+    Highlight,
+    HighlightItem,
+    Place,
+    Post,
+    Profile,
+    Quota,
+    Story,
+    User,
+)
 from insto.service.metrics import Metrics, MetricsSnapshot
 
 log = logging.getLogger("insto.backends.aiograpi")
@@ -141,6 +153,7 @@ class AiograpiBackend(OSINTBackend):
     """
 
     name = "aiograpi"
+    capabilities = frozenset({"followed", "direct_read"})
 
     def __init__(
         self,
@@ -333,6 +346,42 @@ async def iter_highlight_items(
         for raw in items:
             yield map_highlight_item(raw, highlight_pk=str(highlight_id))
 
+    # ------------------------------------------------------------------ direct
+
+    async def iter_direct_threads(self, *, limit: int | None = None) -> AsyncIterator[DirectThread]:
+        from insto.backends._aiograpi_map import map_direct_thread
+
+        amount = int(limit) if limit is not None and limit > 0 else 20
+        items = await self._call(
+            lambda: self._client.direct_threads(amount=amount, thread_message_limit=1)
+        )
+        for raw in items or []:
+            try:
+                yield map_direct_thread(raw)
+            except SchemaDrift as drift:
+                self._drift_count += 1
+                self._last_error = drift
+                raise
+
+    async def iter_direct_messages(
+        self, thread_id: str, *, limit: int | None = None
+    ) -> AsyncIterator[DirectMessage]:
+        from insto.backends._aiograpi_map import map_direct_message
+
+        try:
+            thread_pk = int(thread_id)
+        except (TypeError, ValueError) as exc:
+            raise BackendError(f"invalid direct thread id: {thread_id!r}") from exc
+        amount = int(limit) if limit is not None and limit > 0 else 20
+        items = await self._call(lambda: self._client.direct_messages(thread_pk, amount=amount))
+        for raw in items or []:
+            try:
+                yield map_direct_message(raw, thread_id=str(thread_pk))
+            except SchemaDrift as drift:
+                self._drift_count += 1
+                self._last_error = drift
+                raise
+
     # ------------------------------------------------------------------ network
 
     async def iter_user_followers(
diff --git a/insto/commands/__init__.py b/insto/commands/__init__.py
index 3e7fbd6..36543a3 100644
--- a/insto/commands/__init__.py
+++ b/insto/commands/__init__.py
@@ -14,6 +14,7 @@
 
 from insto.commands import batch as _batch  # noqa: F401  (registers commands)
 from insto.commands import content as _content  # noqa: F401  (registers commands)
+from insto.commands import direct as _direct  # noqa: F401  (registers commands)
 from insto.commands import discovery as _discovery  # noqa: F401  (registers commands)
 from insto.commands import dossier as _dossier  # noqa: F401  (registers commands)
 from insto.commands import interactions as _interactions  # noqa: F401  (registers commands)
diff --git a/insto/commands/direct.py b/insto/commands/direct.py
new file mode 100644
index 0000000..8db9929
--- /dev/null
+++ b/insto/commands/direct.py
@@ -0,0 +1,175 @@
+"""Read-only Direct inbox commands.
+
+These commands are aiograpi-only and intentionally expose no write surface:
+no send, reaction, seen, unsend, mute, approve, upload, or title-update flows.
+"""
+
+from __future__ import annotations
+
+import argparse
+import dataclasses
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import IO
+
+from rich.table import Table
+
+from insto.commands._base import CommandContext, command, resolve_export_dest
+from insto.models import DirectMessage, DirectThread
+
+
+def _add_direct_args(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "count",
+        nargs="?",
+        type=int,
+        default=20,
+        help="number of threads to fetch (default 20)",
+    )
+
+
+def _add_direct_thread_args(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("thread_id", help="Direct thread id")
+    parser.add_argument(
+        "count",
+        nargs="?",
+        type=int,
+        default=20,
+        help="number of messages to fetch (default 20)",
+    )
+
+
+def _resolve_count(ctx: CommandContext, default: int = 20) -> int:
+    if ctx.limit is not None:
+        return int(ctx.limit) if ctx.limit > 0 else default
+    return int(getattr(ctx.args, "count", default))
+
+
+def _resolve_dest(ctx: CommandContext) -> Path | IO[bytes] | None:
+    return resolve_export_dest(ctx.args.json if ctx.args.json is not None else "")
+
+
+def _format_ts(timestamp: int) -> str:
+    if timestamp <= 0:
+        return ""
+    return datetime.fromtimestamp(timestamp, UTC).strftime("%Y-%m-%d %H:%M:%SZ")
+
+
+def _participants(thread: DirectThread) -> str:
+    names = [user.username for user in thread.users if user.username]
+    return ", ".join(names)
+
+
+def _thread_flags(thread: DirectThread) -> str:
+    flags: list[str] = []
+    if thread.is_group:
+        flags.append("group")
+    if thread.is_pending:
+        flags.append("pending")
+    if thread.is_archived:
+        flags.append("archived")
+    if thread.is_muted:
+        flags.append("muted")
+    return ", ".join(flags)
+
+
+def _message_preview(message: DirectMessage) -> str:
+    if message.text:
+        return message.text
+    refs: list[str] = []
+    if message.media_code:
+        refs.append(f"media:{message.media_code}")
+    elif message.media_pk:
+        refs.append(f"media:{message.media_pk}")
+    if message.link_url:
+        refs.append(message.link_url)
+    return " ".join(refs)
+
+
+def _render_threads(threads: list[DirectThread]) -> Table:
+    table = Table(title=f"Direct threads ({len(threads)})")
+    table.add_column("Thread ID", style="cyan", no_wrap=True)
+    table.add_column("Title")
+    table.add_column("Participants")
+    table.add_column("Last activity", no_wrap=True)
+    table.add_column("Messages", justify="right")
+    table.add_column("Flags")
+    for thread in threads:
+        table.add_row(
+            thread.pk,
+            thread.title or _participants(thread),
+            _participants(thread),
+            _format_ts(thread.last_activity_at),
+            str(thread.message_count),
+            _thread_flags(thread),
+        )
+    return table
+
+
+def _render_messages(thread_id: str, messages: list[DirectMessage]) -> Table:
+    table = Table(title=f"Direct thread {thread_id} ({len(messages)} messages)")
+    table.add_column("Time", no_wrap=True)
+    table.add_column("Sender", style="cyan", no_wrap=True)
+    table.add_column("Type", no_wrap=True)
+    table.add_column("Text / ref")
+    for message in messages:
+        table.add_row(
+            _format_ts(message.timestamp),
+            message.sender_pk,
+            message.item_type,
+            _message_preview(message),
+        )
+    return table
+
+
+@command(
+    "direct",
+    "List read-only Direct threads (aiograpi only)",
+    add_args=_add_direct_args,
+    requires=("direct_read",),
+)
+async def direct_cmd(ctx: CommandContext) -> list[DirectThread]:
+    count = _resolve_count(ctx)
+    threads = await ctx.facade.direct_threads(limit=count)
+
+    if ctx.output_format() == "json":
+        ctx.facade.export_json(
+            [dataclasses.asdict(thread) for thread in threads],
+            command="direct",
+            target=None,
+            dest=_resolve_dest(ctx),
+        )
+        return threads
+
+    if not threads:
+        ctx.print("no Direct threads found")
+        return threads
+    ctx.print(_render_threads(threads))
+    return threads
+
+
+@command(
+    "direct-thread",
+    "Show read-only Direct messages for one thread (aiograpi only)",
+    add_args=_add_direct_thread_args,
+    requires=("direct_read",),
+)
+async def direct_thread_cmd(ctx: CommandContext) -> list[DirectMessage]:
+    thread_id = str(getattr(ctx.args, "thread_id", "") or "").strip()
+    count = _resolve_count(ctx)
+    messages = await ctx.facade.direct_messages(thread_id, limit=count)
+
+    if ctx.output_format() == "json":
+        ctx.facade.export_json(
+            [dataclasses.asdict(message) for message in messages],
+            command="direct-thread",
+            target=thread_id,
+            dest=_resolve_dest(ctx),
+        )
+        return messages
+
+    if not messages:
+        ctx.print(f"thread {thread_id} has no messages")
+        return messages
+    ctx.print(_render_messages(thread_id, messages))
+    return messages
diff --git a/insto/models.py b/insto/models.py
index 79c51a1..f29d801 100644
--- a/insto/models.py
+++ b/insto/models.py
@@ -74,6 +74,37 @@ class User:
     is_verified: bool = False
 
 
+@dataclass(slots=True)
+class DirectMessage:
+    """Read-only Direct message summary."""
+
+    pk: str
+    thread_id: str
+    sender_pk: str
+    timestamp: int
+    item_type: str = ""
+    text: str | None = None
+    media_pk: str | None = None
+    media_code: str | None = None
+    link_url: str | None = None
+
+
+@dataclass(slots=True)
+class DirectThread:
+    """Read-only Direct thread summary."""
+
+    pk: str
+    title: str
+    users: list[User] = field(default_factory=list)
+    last_activity_at: int = 0
+    message_count: int = 0
+    is_group: bool = False
+    is_pending: bool = False
+    is_archived: bool = False
+    is_muted: bool = False
+    messages: list[DirectMessage] = field(default_factory=list)
+
+
 @dataclass(slots=True)
 class Post:
     """Instagram media item (image, video, carousel)."""
diff --git a/insto/service/facade.py b/insto/service/facade.py
index 81361e4..6405cd5 100644
--- a/insto/service/facade.py
+++ b/insto/service/facade.py
@@ -39,6 +39,8 @@
 from insto.exceptions import BackendError
 from insto.models import (
     Comment,
+    DirectMessage,
+    DirectThread,
     Highlight,
     HighlightItem,
     Place,
@@ -258,6 +260,14 @@ async def recommended(self, username: str) -> list[User]:
         pk = await self.resolve_pk(username)
         return await self.backend.get_recommended(pk)
 
+    async def direct_threads(self, *, limit: int = 20) -> list[DirectThread]:
+        """Read-only Direct threads for the logged-in aiograpi account."""
+        return [t async for t in self.backend.iter_direct_threads(limit=limit)]
+
+    async def direct_messages(self, thread_id: str, *, limit: int = 20) -> list[DirectMessage]:
+        """Read-only Direct messages for one thread."""
+        return [m async for m in self.backend.iter_direct_messages(thread_id, limit=limit)]
+
     async def mutuals(
         self, username: str, *, follower_limit: int = 1000, following_limit: int = 1000
     ) -> analytics.MutualsResult:
diff --git a/mkdocs.yml b/mkdocs.yml
index f0f93af..fd098df 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -37,6 +37,11 @@ not_in_nav: |
   plans/
   superpowers/
 
+exclude_docs: |
+  internal/
+  plans/
+  superpowers/
+
 nav:
   - Home: index.md
   - Killer features 🔥: killer-features.md
diff --git a/tests/fakes.py b/tests/fakes.py
index 3189d6f..6879e38 100644
--- a/tests/fakes.py
+++ b/tests/fakes.py
@@ -34,6 +34,8 @@
 )
 from insto.models import (
     Comment,
+    DirectMessage,
+    DirectThread,
     Highlight,
     HighlightItem,
     Place,
@@ -77,6 +79,8 @@ class FakeErrors:
     get_post_by_ref: BackendError | None = None
     search_places: BackendError | None = None
     iter_place_posts: BackendError | None = None
+    iter_direct_threads: BackendError | None = None
+    iter_direct_messages: BackendError | None = None
 
 
 @dataclass
@@ -107,6 +111,8 @@ class FakeBackend(OSINTBackend):
     posts_by_ref: dict[str, Post] = field(default_factory=dict)
     places: dict[str, list[Place]] = field(default_factory=dict)
     place_posts: dict[str, list[Post]] = field(default_factory=dict)
+    direct_threads: list[DirectThread] = field(default_factory=list)
+    direct_messages: dict[str, list[DirectMessage]] = field(default_factory=dict)
 
     quota: Quota = field(default_factory=Quota.unknown)
     errors: FakeErrors = field(default_factory=FakeErrors)
@@ -330,6 +336,22 @@ async def iter_place_posts(
         ):
             yield item
 
+    async def iter_direct_threads(self, *, limit: int | None = None) -> AsyncIterator[DirectThread]:
+        self.request_log.append(("iter_direct_threads", (limit,)))
+        self._consume_error("iter_direct_threads")
+        async for item in self._paged("iter_direct_threads", self.direct_threads, limit):
+            yield item
+
+    async def iter_direct_messages(
+        self, thread_id: str, *, limit: int | None = None
+    ) -> AsyncIterator[DirectMessage]:
+        self.request_log.append(("iter_direct_messages", (thread_id, limit)))
+        self._consume_error("iter_direct_messages")
+        async for item in self._paged(
+            "iter_direct_messages", self.direct_messages.get(thread_id, []), limit
+        ):
+            yield item
+
     def get_quota(self) -> Quota:
         return self.quota
 
diff --git a/tests/test_aiograpi_backend_direct.py b/tests/test_aiograpi_backend_direct.py
new file mode 100644
index 0000000..106ec70
--- /dev/null
+++ b/tests/test_aiograpi_backend_direct.py
@@ -0,0 +1,126 @@
+"""AiograpiBackend Direct read-method wiring tests.
+
+These tests install a tiny fake ``aiograpi`` module into ``sys.modules`` so
+the optional dependency is not required in CI. They only verify insto's wiring
+to read-only SDK methods; mapper behavior is covered in ``test_aiograpi_map``.
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime
+from types import ModuleType, SimpleNamespace
+from typing import Any
+
+import pytest
+
+from insto.backends.aiograpi import AiograpiBackend
+from insto.exceptions import BackendError
+
+
+class _FakeClient:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, tuple[Any, ...], dict[str, Any]]] = []
+
+    async def direct_threads(
+        self,
+        *,
+        amount: int = 20,
+        selected_filter: str = "",
+        box: str = "",
+        thread_message_limit: int | None = None,
+    ) -> list[Any]:
+        self.calls.append(
+            (
+                "direct_threads",
+                (),
+                {
+                    "amount": amount,
+                    "selected_filter": selected_filter,
+                    "box": box,
+                    "thread_message_limit": thread_message_limit,
+                },
+            )
+        )
+        return [
+            SimpleNamespace(
+                id="123",
+                thread_title="Alice",
+                users=[SimpleNamespace(pk="100", username="alice")],
+                last_activity_at=datetime(2026, 4, 17, 17, 45, 12, tzinfo=UTC),
+                messages=[],
+                is_group=False,
+                pending=False,
+                archived=False,
+                muted=False,
+            )
+        ]
+
+    async def direct_messages(self, thread_id: int, *, amount: int = 20) -> list[Any]:
+        self.calls.append(("direct_messages", (thread_id,), {"amount": amount}))
+        return [
+            SimpleNamespace(
+                id="m1",
+                user_id="100",
+                thread_id=thread_id,
+                timestamp=datetime(2026, 4, 17, 17, 45, 12, tzinfo=UTC),
+                item_type="text",
+                text="hello",
+            )
+        ]
+
+
+@pytest.fixture
+def fake_aiograpi(monkeypatch: pytest.MonkeyPatch) -> type[_FakeClient]:
+    module = ModuleType("aiograpi")
+    module.Client = _FakeClient  # type: ignore[attr-defined]
+    monkeypatch.setitem(sys.modules, "aiograpi", module)
+    return _FakeClient
+
+
+def _backend() -> AiograpiBackend:
+    backend = AiograpiBackend(username="user", password="pass")
+    backend._logged_in = True
+    return backend
+
+
+async def test_direct_threads_calls_read_only_sdk_method(fake_aiograpi: type[_FakeClient]) -> None:
+    backend = _backend()
+    client = backend._client
+
+    threads = [thread async for thread in backend.iter_direct_threads(limit=3)]
+
+    assert threads[0].pk == "123"
+    assert threads[0].title == "Alice"
+    assert client.calls == [
+        (
+            "direct_threads",
+            (),
+            {
+                "amount": 3,
+                "selected_filter": "",
+                "box": "",
+                "thread_message_limit": 1,
+            },
+        )
+    ]
+
+
+async def test_direct_messages_calls_read_only_sdk_method(fake_aiograpi: type[_FakeClient]) -> None:
+    backend = _backend()
+    client = backend._client
+
+    messages = [message async for message in backend.iter_direct_messages("123", limit=5)]
+
+    assert messages[0].pk == "m1"
+    assert messages[0].thread_id == "123"
+    assert client.calls == [("direct_messages", (123,), {"amount": 5})]
+
+
+async def test_direct_messages_rejects_non_numeric_thread_id(
+    fake_aiograpi: type[_FakeClient],
+) -> None:
+    backend = _backend()
+
+    with pytest.raises(BackendError, match="invalid direct thread id"):
+        [message async for message in backend.iter_direct_messages("not-a-number", limit=5)]
diff --git a/tests/test_aiograpi_map.py b/tests/test_aiograpi_map.py
index b9f45a2..7bb185f 100644
--- a/tests/test_aiograpi_map.py
+++ b/tests/test_aiograpi_map.py
@@ -23,6 +23,8 @@
 from insto.backends._aiograpi_map import (
     about_payload,
     map_comment,
+    map_direct_message,
+    map_direct_thread,
     map_highlight,
     map_highlight_item,
     map_post,
@@ -366,3 +368,106 @@ def test_about_payload_serialises_extended_fields() -> None:
     assert out["public_phone_country_code"] == "+1"
     assert out["is_business"] is True
     assert out["is_private"] is False
+
+
+# ---- Direct ---------------------------------------------------------------
+
+
+def test_map_direct_message_text() -> None:
+    msg = map_direct_message(
+        _bag(
+            id="m1",
+            user_id="100",
+            thread_id=123,
+            timestamp=datetime(2026, 4, 17, 17, 45, 12, tzinfo=UTC),
+            item_type="text",
+            text="hello",
+        )
+    )
+
+    assert msg.pk == "m1"
+    assert msg.thread_id == "123"
+    assert msg.sender_pk == "100"
+    assert msg.timestamp == 1776447912
+    assert msg.item_type == "text"
+    assert msg.text == "hello"
+    assert msg.media_pk is None
+    assert msg.media_code is None
+    assert msg.link_url is None
+
+
+def test_map_direct_message_uses_explicit_thread_id_when_raw_omits_it() -> None:
+    msg = map_direct_message(
+        _bag(
+            id="m2",
+            user_id="100",
+            thread_id=None,
+            timestamp=1_700_000_000,
+            item_type="text",
+            text="fallback",
+        ),
+        thread_id="t1",
+    )
+
+    assert msg.thread_id == "t1"
+
+
+def test_map_direct_message_media_share_and_link_refs() -> None:
+    msg = map_direct_message(
+        _bag(
+            id="m3",
+            user_id="100",
+            thread_id=123,
+            timestamp=datetime(2026, 4, 17, 17, 45, 12, tzinfo=UTC),
+            item_type="media_share",
+            text=None,
+            media_share=_bag(pk="media1", code="ABC123"),
+            link=_bag(url="https://example.test/story"),
+        )
+    )
+
+    assert msg.media_pk == "media1"
+    assert msg.media_code == "ABC123"
+    assert msg.link_url == "https://example.test/story"
+    assert msg.text is None
+
+
+def test_map_direct_thread_summary_maps_users_flags_and_preview_messages() -> None:
+    thread = map_direct_thread(
+        _bag(
+            id="t1",
+            pk="thread-pk",
+            thread_title="Alice, Bob",
+            users=[
+                _bag(pk="100", username="alice", full_name="Alice", is_private=False),
+                _bag(pk="101", username="bob", full_name="Bob", is_private=True),
+            ],
+            last_activity_at=datetime(2026, 4, 17, 17, 45, 12, tzinfo=UTC),
+            messages=[
+                _bag(
+                    id="m1",
+                    user_id="100",
+                    thread_id=None,
+                    timestamp=datetime(2026, 4, 17, 17, 45, 12, tzinfo=UTC),
+                    item_type="text",
+                    text="hello",
+                )
+            ],
+            is_group=True,
+            pending=True,
+            archived=False,
+            muted=True,
+        )
+    )
+
+    assert thread.pk == "t1"
+    assert thread.title == "Alice, Bob"
+    assert [user.username for user in thread.users] == ["alice", "bob"]
+    assert thread.users[1].is_private is True
+    assert thread.last_activity_at == 1776447912
+    assert thread.message_count == 1
+    assert thread.is_group is True
+    assert thread.is_pending is True
+    assert thread.is_archived is False
+    assert thread.is_muted is True
+    assert thread.messages[0].thread_id == "t1"
diff --git a/tests/test_backend_contract.py b/tests/test_backend_contract.py
index bbd3eaf..098a9b2 100644
--- a/tests/test_backend_contract.py
+++ b/tests/test_backend_contract.py
@@ -17,6 +17,7 @@
 
 import pytest
 
+from insto.backends._base import OSINTBackend
 from insto.exceptions import (
     AuthInvalid,
     BackendError,
@@ -32,7 +33,7 @@
     SchemaDrift,
     Transient,
 )
-from insto.models import Post, Profile
+from insto.models import DirectMessage, DirectThread, Post, Profile
 from tests.fakes import FakeBackend, FakeErrors
 
 
@@ -40,6 +41,17 @@ def _make_post(pk: str) -> Post:
     return Post(pk=pk, code=f"c{pk}", taken_at=0, media_type="image")
 
 
+def _make_direct_message(pk: str, thread_id: str = "t1") -> DirectMessage:
+    return DirectMessage(
+        pk=pk,
+        thread_id=thread_id,
+        sender_pk="100",
+        timestamp=1_700_000_000,
+        item_type="text",
+        text=f"message {pk}",
+    )
+
+
 @pytest.mark.asyncio
 async def test_iter_user_posts_respects_limit_and_stops_early() -> None:
     """500 posts, limit=25 → exactly 25 emitted, only ⌈25/12⌉ pages fetched."""
@@ -61,6 +73,49 @@ async def test_iter_user_posts_respects_limit_and_stops_early() -> None:
     assert backend.page_requests["iter_user_posts"] == 3
 
 
+def test_base_direct_threads_default_requires_aiograpi() -> None:
+    backend = FakeBackend()
+
+    with pytest.raises(BackendError, match="needs aiograpi backend"):
+        OSINTBackend.iter_direct_threads(backend, limit=1)
+
+
+def test_base_direct_messages_default_requires_aiograpi() -> None:
+    backend = FakeBackend()
+
+    with pytest.raises(BackendError, match="needs aiograpi backend"):
+        OSINTBackend.iter_direct_messages(backend, "t1", limit=1)
+
+
+@pytest.mark.asyncio
+async def test_fake_direct_threads_respects_limit_and_stops_early() -> None:
+    backend = FakeBackend(
+        direct_threads=[
+            DirectThread(pk=f"t{i}", title=f"Thread {i}", last_activity_at=i) for i in range(10)
+        ],
+        page_size=2,
+    )
+
+    collected = [thread async for thread in backend.iter_direct_threads(limit=5)]
+
+    assert [thread.pk for thread in collected] == ["t0", "t1", "t2", "t3", "t4"]
+    assert backend.page_requests["iter_direct_threads"] == 3
+
+
+@pytest.mark.asyncio
+async def test_fake_direct_messages_respects_limit_and_stops_early() -> None:
+    backend = FakeBackend(
+        direct_messages={"t1": [_make_direct_message(str(i)) for i in range(10)]},
+        page_size=2,
+    )
+
+    collected = [message async for message in backend.iter_direct_messages("t1", limit=5)]
+
+    assert [message.pk for message in collected] == ["0", "1", "2", "3", "4"]
+    assert backend.page_requests["iter_direct_messages"] == 3
+    assert backend.request_log == [("iter_direct_messages", ("t1", 5))]
+
+
 @pytest.mark.asyncio
 async def test_iter_user_posts_unbounded_when_limit_none() -> None:
     backend = FakeBackend(
diff --git a/tests/test_commands_direct.py b/tests/test_commands_direct.py
new file mode 100644
index 0000000..12a8db3
--- /dev/null
+++ b/tests/test_commands_direct.py
@@ -0,0 +1,200 @@
+"""Tests for read-only Direct commands."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Generator
+from pathlib import Path
+
+import pytest
+from rich.console import Console
+
+# Importing the package registers all command modules.
+import insto.commands  # noqa: F401  (side-effect import)
+from insto.commands._base import CommandUsageError, Session, dispatch
+from insto.config import Config
+from insto.models import DirectMessage, DirectThread, User
+from insto.service.facade import OsintFacade
+from insto.service.history import HistoryStore
+from insto.ui.theme import INSTO_THEME
+from tests.fakes import FakeBackend
+
+
+@pytest.fixture
+def history(tmp_path: Path) -> Generator[HistoryStore, None, None]:
+    store = HistoryStore(tmp_path / "store.db")
+    yield store
+    store.close()
+
+
+@pytest.fixture
+def config(tmp_path: Path) -> Config:
+    return Config(output_dir=tmp_path / "output", db_path=tmp_path / "store.db")
+
+
+@pytest.fixture
+def session() -> Session:
+    return Session()
+
+
+@pytest.fixture
+def recording_console() -> Console:
+    return Console(
+        theme=INSTO_THEME,
+        width=120,
+        force_terminal=True,
+        color_system="truecolor",
+        record=True,
+    )
+
+
+def _captured(console: Console) -> str:
+    return console.export_text(styles=False)
+
+
+def _message(pk: str, *, thread_id: str = "t1", text: str = "hello") -> DirectMessage:
+    return DirectMessage(
+        pk=pk,
+        thread_id=thread_id,
+        sender_pk="100",
+        timestamp=1_700_000_000,
+        item_type="text",
+        text=text,
+    )
+
+
+def _thread(pk: str, *, title: str, username: str, message_count: int = 1) -> DirectThread:
+    return DirectThread(
+        pk=pk,
+        title=title,
+        users=[User(pk="100", username=username)],
+        last_activity_at=1_700_000_000,
+        message_count=message_count,
+        messages=[_message("m1", thread_id=pk, text="preview")],
+    )
+
+
+def _direct_backend() -> FakeBackend:
+    backend = FakeBackend(
+        direct_threads=[
+            _thread("t1", title="Alice", username="alice"),
+            _thread("t2", title="Bob", username="bob"),
+            _thread("t3", title="Carol", username="carol"),
+        ],
+        direct_messages={
+            "t1": [
+                _message("m1", text="first"),
+                _message("m2", text="second"),
+                DirectMessage(
+                    pk="m3",
+                    thread_id="t1",
+                    sender_pk="101",
+                    timestamp=1_700_000_100,
+                    item_type="media_share",
+                    media_pk="p1",
+                    media_code="ABC123",
+                ),
+            ]
+        },
+    )
+    backend.capabilities = frozenset({"direct_read"})
+    return backend
+
+
+async def test_direct_lists_threads(
+    history: HistoryStore,
+    config: Config,
+    session: Session,
+    recording_console: Console,
+) -> None:
+    backend = _direct_backend()
+    facade = OsintFacade(backend=backend, history=history, config=config)
+
+    out = await dispatch("/direct 2", facade=facade, session=session, console=recording_console)
+
+    assert [thread.pk for thread in out] == ["t1", "t2"]
+    captured = _captured(recording_console)
+    assert "Direct threads" in captured
+    assert "Alice" in captured
+    assert "Bob" in captured
+    assert "Carol" not in captured
+
+
+async def test_direct_thread_lists_messages(
+    history: HistoryStore,
+    config: Config,
+    session: Session,
+    recording_console: Console,
+) -> None:
+    backend = _direct_backend()
+    facade = OsintFacade(backend=backend, history=history, config=config)
+
+    out = await dispatch(
+        "/direct-thread t1 2", facade=facade, session=session, console=recording_console
+    )
+
+    assert [message.pk for message in out] == ["m1", "m2"]
+    captured = _captured(recording_console)
+    assert "Direct thread t1" in captured
+    assert "first" in captured
+    assert "second" in captured
+    assert "ABC123" not in captured
+
+
+async def test_direct_json_export(
+    history: HistoryStore,
+    config: Config,
+    session: Session,
+) -> None:
+    backend = _direct_backend()
+    facade = OsintFacade(backend=backend, history=history, config=config)
+
+    await dispatch("/direct 2 --json", facade=facade, session=session)
+
+    out_path = config.output_dir / "_" / "direct.json"
+    payload = json.loads(out_path.read_text())
+    assert payload["command"] == "direct"
+    assert payload["target"] is None
+    assert [thread["pk"] for thread in payload["data"]] == ["t1", "t2"]
+
+
+async def test_direct_thread_json_stdout(
+    history: HistoryStore,
+    config: Config,
+    session: Session,
+    capsysbinary: pytest.CaptureFixture[bytes],
+) -> None:
+    backend = _direct_backend()
+    facade = OsintFacade(backend=backend, history=history, config=config)
+
+    await dispatch("/direct-thread t1 2 --json -", facade=facade, session=session)
+
+    payload = json.loads(capsysbinary.readouterr().out)
+    assert payload["command"] == "direct-thread"
+    assert payload["target"] == "t1"
+    assert [message["pk"] for message in payload["data"]] == ["m1", "m2"]
+
+
+async def test_direct_rejects_csv(
+    history: HistoryStore,
+    config: Config,
+    session: Session,
+) -> None:
+    facade = OsintFacade(backend=_direct_backend(), history=history, config=config)
+
+    with pytest.raises(CommandUsageError, match="cannot be exported as CSV"):
+        await dispatch("/direct --csv -", facade=facade, session=session)
+
+
+async def test_direct_requires_direct_read_capability(
+    history: HistoryStore,
+    config: Config,
+    session: Session,
+) -> None:
+    backend = FakeBackend(direct_threads=[_thread("t1", title="Alice", username="alice")])
+    facade = OsintFacade(backend=backend, history=history, config=config)
+
+    with pytest.raises(CommandUsageError, match="missing capability: direct_read"):
+        await dispatch("/direct", facade=facade, session=session)
+
+    assert backend.request_log == []
diff --git a/tests/test_models.py b/tests/test_models.py
index 661bab8..2afe56a 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -8,6 +8,8 @@
 
 from insto.models import (
     Comment,
+    DirectMessage,
+    DirectThread,
     Highlight,
     HighlightItem,
     Post,
@@ -22,6 +24,8 @@
 ALL_MODELS = [
     Profile,
     User,
+    DirectMessage,
+    DirectThread,
     Post,
     Comment,
     Story,
@@ -172,12 +176,95 @@ def test_user_construction() -> None:
     assert u.is_verified is False
 
 
+def test_direct_message_asdict_shape() -> None:
+    message = DirectMessage(
+        pk="m1",
+        thread_id="t1",
+        sender_pk="100",
+        timestamp=1_700_000_000,
+        item_type="text",
+        text="hello",
+    )
+
+    assert asdict(message) == {
+        "pk": "m1",
+        "thread_id": "t1",
+        "sender_pk": "100",
+        "timestamp": 1_700_000_000,
+        "item_type": "text",
+        "text": "hello",
+        "media_pk": None,
+        "media_code": None,
+        "link_url": None,
+    }
+
+
+def test_direct_thread_asdict_shape() -> None:
+    message = DirectMessage(
+        pk="m1",
+        thread_id="t1",
+        sender_pk="100",
+        timestamp=1_700_000_000,
+        item_type="text",
+        text="hello",
+    )
+    thread = DirectThread(
+        pk="t1",
+        title="Alice",
+        users=[User(pk="100", username="alice")],
+        last_activity_at=1_700_000_000,
+        message_count=1,
+        is_group=False,
+        is_pending=False,
+        is_archived=False,
+        is_muted=False,
+        messages=[message],
+    )
+
+    assert asdict(thread) == {
+        "pk": "t1",
+        "title": "Alice",
+        "users": [
+            {
+                "pk": "100",
+                "username": "alice",
+                "full_name": "",
+                "is_private": False,
+                "is_verified": False,
+            }
+        ],
+        "last_activity_at": 1_700_000_000,
+        "message_count": 1,
+        "is_group": False,
+        "is_pending": False,
+        "is_archived": False,
+        "is_muted": False,
+        "messages": [
+            {
+                "pk": "m1",
+                "thread_id": "t1",
+                "sender_pk": "100",
+                "timestamp": 1_700_000_000,
+                "item_type": "text",
+                "text": "hello",
+                "media_pk": None,
+                "media_code": None,
+                "link_url": None,
+            }
+        ],
+    }
+
+
 def _make_sample(cls: type) -> object:
     """Build a minimum-viable instance of any DTO for slot-attribute checks."""
     if cls is Profile:
         return Profile(pk="1", username="a", access="public")
     if cls is User:
         return User(pk="1", username="a")
+    if cls is DirectMessage:
+        return DirectMessage(pk="m1", thread_id="t1", sender_pk="u1", timestamp=0)
+    if cls is DirectThread:
+        return DirectThread(pk="t1", title="thread")
     if cls is Post:
         return Post(pk="1", code="A", taken_at=0, media_type="image")
     if cls is Comment: