fix: the eight high-severity defects a full API sweep turned up

Seventy of the seventy-six operations were driven against the live stack and
every finding was reproduced by a second agent before it was believed. These
are the eight rated high.

**An identity:manage key could mint itself an admin key.** The create route is
guarded by `manage_pool`, which admits `identity:manage`, and never looked at
the scopes being minted - so a key refused `GET /admin/users` created a key with
`scopes:["admin"]` and used it one request later. The guard asked whether the
caller may create keys, never whether it may create *this* key. A key may now
only mint scopes it holds; a console session is still bounded by its role.

**Sensitive scalar settings were never masked.** `redact_setting` decided on the
value's shape: it masked by field name inside a mapping or a list and returned
any top-level scalar untouched. `notify.channels` was masked correctly on every
surface while `security.webhook_secret` - a bare string, declared SENSITIVE -
came back verbatim from the listing, was echoed by the write, and was written in
cleartext into an audit row retention never trims. Shape cannot answer that
question, so the setting key is passed in and the declaration decides.

**A JSON body with the wrong Content-Type returned 500 and logged the body.**
Pydantic puts the offending value in every error, and for a body that never
parsed that value is the raw request bytes; serializing them raised, the
catch-all turned it into a 500, and the traceback wrote the whole body to the
container log - an imported cookie jar on one endpoint, a plaintext password on
another, reachable unauthenticated through `POST /auth/login`, and triggered by
plain `curl -d`. The input never travels now, and it answers 400.

**Pinning did not protect files.** The layout has nothing per-download in it, so
several rows share one directory, and the sidecar deletes directories
recursively. Deciding row by row scheduled a directory because one of its rows
was unpinned - destroying the pinned row's files while reporting those same
bytes as protected - and counted the same bytes once per row, so the sweep
believed it had freed twice what it had. Eviction now groups by directory,
refuses one if any row on it is pinned, and counts it once.

**archive:read alone could spend the identity pool.** `/archive/recheck` and
`/archive/backfill` required only `archive:read` while making a real request per
post, against the rule the downloads module states outright and the archive
module's own promise that reading it spends no identity. They now ask for the
platform read scope, which is what they actually cost.

**include_raw was not in the cache key.** The cached value is the shaped body,
so whichever value the first caller used was frozen for the whole TTL in both
directions - a later `include_raw=false` got a body that still carried `raw`,
persisted into `tasks.result` for a caller who opted out.

**A withheld page looked like a parser bug.** Douyin answers author_posts with
any non-zero cursor as `{"status_code":0}`; the withheld-payload rule fires only
on a payload key that is present and empty, so an absent one fell through to
`default.ok`, reached the parser, and came back as a non-retryable
UPSTREAM_CHANGED asking the operator to report a parser bug no parser change can
fix - while the request log recorded it as a clean 200 that pool health never
saw. A 200 whose envelope carries a status and nothing else is now read as
withheld, keyed on "nothing but metadata" rather than on payload names that
differ per endpoint.

**The MCP server was unreachable at its own documented path.** The console's
SPA catch-all full-matched `GET /mcp` and answered the console shell with 200,
so a probing client saw success; `POST /mcp` got a 405. Only `/mcp/` worked and
nothing said so. The prefix is reserved now, and the bare path 307s.

Each verified against the running stack after the fix, not only in tests.

src/dtk/api/app.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import contextlib
 from collections.abc import AsyncIterator
+from typing import Any, Final
 
 from fastapi import FastAPI, Request
 from fastapi.exceptions import RequestValidationError
@@ -65,6 +66,27 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         log.info("api.stopped")
 
 
+#: Keys pydantic attaches to a validation error that must never leave the
+#: process. ``input`` is the offending value, and for a body that failed to
+#: parse at all it is the raw request bytes; ``ctx`` can carry the same value
+#: again inside an exception it wrapped; ``url`` is a docs link, not an error.
+_UNSAFE_ERROR_KEYS: Final[frozenset[str]] = frozenset({"input", "ctx", "url"})
+
+
+def _safe_validation_fields(exc: RequestValidationError) -> list[dict[str, Any]]:
+    """The field path and the reason, and nothing the caller sent.
+
+    FastAPI's ``errors()`` takes no arguments - it hands back pydantic's list
+    verbatim - so the stripping happens here rather than at the call.
+    """
+    fields: list[dict[str, Any]] = []
+    for error in list(exc.errors())[:10]:
+        if not isinstance(error, dict):
+            continue
+        fields.append({k: v for k, v in error.items() if k not in _UNSAFE_ERROR_KEYS})
+    return fields
+
+
 def create_app(settings: BootstrapSettings | None = None) -> FastAPI:
     settings = settings or BootstrapSettings()
 
@@ -110,11 +132,26 @@ def _install_error_handlers(app: FastAPI) -> None:
 
     @app.exception_handler(RequestValidationError)
     async def _validation(request: Request, exc: RequestValidationError) -> JSONResponse:
+        """Answer a malformed request without quoting it back.
+
+        ``include_input=False`` is the whole point. Pydantic puts the offending
+        value in every error, and for a body that failed to parse at all - a
+        JSON document sent with `Content-Type: text/plain`, which is what plain
+        `curl -d` does - that value is the raw request bytes. Serializing them
+        raised `TypeError: Object of type bytes is not JSON serializable`, the
+        catch-all turned it into a 500, and the traceback wrote the whole body
+        into the container log: an imported cookie jar on one endpoint, a
+        plaintext password on another, and reachable unauthenticated through
+        `POST /auth/login`.
+
+        So the input never travels. The field path and the reason are what a
+        caller needs; the value is something they already have.
+        """
         return envelope.failure(
             ErrorCode.INVALID_PARAM,
             getattr(request.state, "request_id", "unknown"),
             language=response_language(request),
-            details={"fields": exc.errors()[:10]},
+            details={"fields": _safe_validation_fields(exc)},
         )
 
     @app.exception_handler(StarletteHTTPException)

src/dtk/api/console.py

@@ -55,7 +55,21 @@ def _candidates() -> list[Path]:
 #: Paths the fallback must never claim. Without this a typo in an API path would
 #: return the console's HTML with status 200, and a client parsing JSON would
 #: report something entirely unrelated to the actual mistake.
-RESERVED_PREFIXES = ("/api/", "/healthz", "/readyz", "/docs", "/redoc", "/openapi.json")
+#:
+#: ``/mcp`` is here because leaving it out was worse than a wrong body: the
+#: catch-all full-matched ``GET /mcp`` and answered the console shell with 200,
+#: so a probing client saw apparent success, while ``POST /mcp`` partial-matched
+#: and got 405 before Starlette's trailing-slash redirect could run. Only
+#: ``/mcp/`` worked, and nothing in either answer said so.
+RESERVED_PREFIXES = (
+    "/api/",
+    "/healthz",
+    "/readyz",
+    "/docs",
+    "/redoc",
+    "/openapi.json",
+    "/mcp",
+)
 
 
 def localize(html: str, language: Language) -> str:

src/dtk/api/routes/admin/api_keys.py

@@ -12,6 +12,7 @@ time, so there is no cache to wait out.
 from __future__ import annotations
 
 import uuid
+from collections.abc import Sequence
 from datetime import UTC, datetime
 from typing import Any
 
@@ -22,8 +23,9 @@ from dtk.api.deps import Principal
 from dtk.api.routes.schemas import ApiKeyCreate
 from dtk.api.routes.support import audit, iso, manage_pool, ok, read_admin
 from dtk.core.crypto import new_api_key
-from dtk.core.errors import InvalidParam, NotFound
+from dtk.core.errors import ForbiddenScope, InvalidParam, NotFound
 from dtk.core.logging import get_logger
+from dtk.core.types import Scope
 from dtk.db.models import ApiKey
 from dtk.db.repositories import ApiKeyRepository
 
@@ -80,6 +82,32 @@ async def list_keys(
     return ok(request, [_row(row) for row in rows])
 
 
+def _refuse_escalation(principal: Principal, requested: Sequence[Scope]) -> None:
+    """A key may not mint scopes its creator does not hold.
+
+    This route is guarded by ``manage_pool``, which admits ``identity:manage``
+    as well as ``admin`` - and it did not look at the scopes being minted. So a
+    key holding only ``identity:manage``, refused ``GET /admin/users`` and
+    refused a write to ``api.public_endpoints``, could create a key with
+    ``scopes: ["admin"]`` and use it to do both, one request apart. The guard
+    was asking whether the caller may create keys, never whether it may create
+    *this* key.
+
+    A console session is bounded by its role rather than by scopes, and the
+    guard above already requires at least OPERATOR, so a session is left to
+    mint what its role allows. It is a credential minting a credential that has
+    to be bounded by what the first one holds.
+    """
+    if principal.api_key_id is None:
+        return
+    excess = sorted(scope.value for scope in requested if scope not in principal.scopes)
+    if excess:
+        raise ForbiddenScope(
+            "an API key cannot create a key with scopes it does not hold itself",
+            details={"refused": excess, "held": sorted(s.value for s in principal.scopes)},
+        )
+
+
 @router.post("", summary="Create an API key")
 async def create_key(
     request: Request,
@@ -97,6 +125,7 @@ async def create_key(
             "expires_at must be in the future",
             details={"field": "expires_at"},
         )
+    _refuse_escalation(principal, body.scopes)
     full, prefix, digest = new_api_key()
     key = await ApiKeyRepository(request.state.db).create(
         user_id=principal.user_id,

src/dtk/api/routes/admin/maintenance.py

@@ -99,7 +99,9 @@ async def list_audit(
                 # were masked at the source are still in this table - nothing
                 # trims it - so they are masked again on the way out.
                 "detail": (
-                    redact_setting(row.detail) if row.target_type == "setting" else row.detail
+                    redact_setting(row.detail, row.target_id)
+                    if row.target_type == "setting"
+                    else row.detail
                 ),
                 "ip": row.ip,
                 "user_agent": row.user_agent,

src/dtk/api/routes/admin/settings.py

@@ -100,7 +100,7 @@ async def list_settings(request: Request, principal: Principal = Depends(read_ad
     for key, spec in sorted(RUNTIME_SETTINGS.items()):
         stored = rows.get(key)
         value = config.get(key)
-        redacted = redact_setting(value)
+        redacted = redact_setting(value, spec.key)
         items.append(
             {
                 "key": key,
@@ -159,7 +159,7 @@ async def update_setting(
     try:
         # Unmasked before coercion, so what is validated is what will be stored:
         # a value that arrived as a mask must not reach the table unchecked.
-        value = coerce(key, unredact_setting(body.value, previous))
+        value = coerce(key, unredact_setting(body.value, previous, key))
     except (ValueError, TypeError) as exc:
         # The message is replaced by a translated one for the caller's language,
         # so anything the operator needs in order to fix the value has to travel
@@ -181,7 +181,7 @@ async def update_setting(
     # AuditRepository.record: never a credential in detail. The audit trail is
     # read by humans, returned verbatim by GET /admin/audit and never trimmed by
     # retention, so a bot token written here outlives the channel it belongs to.
-    change = {"from": redact_setting(previous), "to": redact_setting(value)}
+    change = {"from": redact_setting(previous, key), "to": redact_setting(value, key)}
     if spec.scope is Scope.SENSITIVE:
         # Doc 08 lists a SENSITIVE change beside importing an identity and
         # creating a key: the operations that widen the attack surface.
@@ -207,7 +207,7 @@ async def update_setting(
         request,
         {
             "key": key,
-            "value": redact_setting(value),
+            "value": redact_setting(value, key),
             "version": request.app.state.config.version,
         },
     )
@@ -236,15 +236,15 @@ async def reset_setting(
         target_type="setting",
         target_id=key,
         detail={
-            "from": redact_setting(previous),
-            "to": redact_setting(request.app.state.config.get(key)),
+            "from": redact_setting(previous, key),
+            "to": redact_setting(request.app.state.config.get(key), key),
         },
     )
     return ok(
         request,
         {
             "key": key,
-            "value": redact_setting(request.app.state.config.get(key)),
+            "value": redact_setting(request.app.state.config.get(key), key),
             "source": _source(key, None),
             "version": request.app.state.config.version,
         },

src/dtk/api/routes/archive.py

@@ -20,7 +20,7 @@ from __future__ import annotations
 import json
 from collections.abc import AsyncIterator
 from datetime import datetime
-from typing import Any
+from typing import Any, Final
 
 from fastapi import APIRouter, Depends, Path, Query, Request
 from fastapi.responses import StreamingResponse
@@ -33,7 +33,7 @@ from dtk.api.routes.support import ok
 from dtk.core.db import session_scope
 from dtk.core.errors import NotFound
 from dtk.core.logging import get_logger
-from dtk.core.types import Scope
+from dtk.core.types import Platform, Scope
 from dtk.db.models import ArchivedContent
 from dtk.services import archive
 
@@ -45,6 +45,13 @@ router = APIRouter(prefix="/api/v1/archive", tags=["archive"])
 #: who needs more pages through with `cursor` on the list endpoint.
 EXPORT_LIMIT = 50_000
 
+#: Which read scope a walk of one platform's history costs. Both of these
+#: endpoints look like archive reads and behave like platform reads.
+_READ_SCOPE: Final[dict[Platform, Scope]] = {
+    Platform.DOUYIN: Scope.DOUYIN_READ,
+    Platform.TIKTOK: Scope.TIKTOK_READ,
+}
+
 PLATFORM_QUERY = Query(default=None, description="Only this platform.")
 AUTHOR_QUERY = Query(default=None, max_length=256, description="Only this author's posts.")
 TAG_QUERY = Query(default=None, max_length=128, description="Only posts carrying this tag.")
@@ -285,7 +292,12 @@ async def recheck(
     `202` with a task id. The result carries how many were checked and how many
     turned out to be gone.
     """
-    principal.require(Scope.ARCHIVE_READ)
+    # Not archive:read. This spends the identity pool on a real request per
+    # post, which is exactly what the downloads module says no read scope may
+    # imply - and the archive module's own docstring promises that reading it
+    # spends no identity. A recheck is a platform read wearing an archive name,
+    # so it asks for a platform read scope.
+    principal.require(Scope.DOUYIN_READ, Scope.TIKTOK_READ)
     return await operations.submit_and_wait(
         request,
         principal,
@@ -326,7 +338,9 @@ async def backfill(
     `202` with a task id. The result says how many pages were walked, how many
     posts were archived, and why it stopped.
     """
-    principal.require(Scope.ARCHIVE_READ)
+    # The platform's own read scope, for the same reason as the recheck above -
+    # and this one names its platform, so it can ask for exactly that one.
+    principal.require(_READ_SCOPE[body.platform])
     return await operations.submit_and_wait(
         request,
         principal,

src/dtk/cli/settings_cmd.py

@@ -31,12 +31,12 @@ def redact(key: str, value: Any) -> Any:
     """Mask credential-bearing parts of a setting value.
 
     The rules moved to :mod:`dtk.ops.masking` when the admin API needed them
-    too; this keeps the terminal's spelling of them. ``key`` is not consulted:
-    what makes a value a credential is the field it sits in, not the setting it
-    belongs to.
+    too; this keeps the terminal's spelling of them. ``key`` IS consulted: a
+    field name is what makes a nested value a credential, but a top-level
+    scalar like ``security.webhook_secret`` is a credential because the registry
+    says so, and nothing about the string itself reveals that.
     """
-    del key
-    return redact_setting(value)
+    return redact_setting(value, key)
 
 
 def render(value: Any) -> str:

src/dtk/mcp/http.py

@@ -27,6 +27,7 @@ from mcp.server.mcpserver import MCPServer
 from mcp.server.transport_security import TransportSecuritySettings
 from starlette.applications import Starlette
 from starlette.requests import Request
+from starlette.responses import RedirectResponse
 from starlette.types import Receive, Send
 from starlette.types import Scope as ASGIScope
 
@@ -238,6 +239,35 @@ def _chain_lifespan(app: FastAPI, sub: Starlette) -> None:
     app.router.lifespan_context = combined
 
 
+def _redirect_bare_path(app: FastAPI, path: str) -> None:
+    """Send ``/mcp`` to ``/mcp/`` rather than answering 404.
+
+    A Mount matches only the paths beneath it, so the bare prefix is left for
+    whatever comes next - and for a while that was the console's SPA catch-all,
+    which answered ``GET /mcp`` with the console shell and **200**. A client
+    probing the documented path saw apparent success and nothing that hinted at
+    the missing slash; ``POST /mcp`` got a 405 from the same partial match.
+    Adding the prefix to the console's reserved list stopped the wrong body but
+    left a bare 404, which is honest and still unhelpful.
+
+    Registered after the mount so the mount keeps precedence for everything
+    under it.
+    """
+
+    async def _to_slash(_: Request) -> RedirectResponse:
+        # 307, not 302: the method and body have to survive, and the client
+        # that hits this is usually POSTing an initialize call.
+        return RedirectResponse(url=f"{path}/", status_code=307)
+
+    app.add_api_route(
+        path,
+        _to_slash,
+        methods=["GET", "POST"],
+        include_in_schema=False,
+        name="mcp_trailing_slash",
+    )
+
+
 def mount(
     app: FastAPI,
     *,
@@ -255,6 +285,7 @@ def mount(
     server = build_server(context or app_context(app))
     http_app = create_http_app(server, stateless=stateless)
     app.mount(path, ApiKeyGuard(http_app, scopes=scopes))
+    _redirect_bare_path(app, path)
     _chain_lifespan(app, http_app)
     log.info("mcp.http.mounted", path=path, stateless=stateless)
     return server

src/dtk/ops/masking.py

@@ -214,22 +214,55 @@ def is_credential_field(name: str) -> bool:
     return lowered in SECRET_FIELDS or lowered in URL_FIELDS
 
 
-def redact_setting(value: Any) -> Any:
+def is_sensitive_setting(key: str | None) -> bool:
+    """Whether this setting is declared SENSITIVE in the registry.
+
+    Read from the declaration rather than guessed from the name. A setting is
+    sensitive because someone said so in :mod:`dtk.core.config`, and a name
+    heuristic would both miss ``api.public_endpoints`` and mask things that are
+    not secret at all.
+    """
+    if not key:
+        return False
+    from dtk.core.config import RUNTIME_SETTINGS, Scope
+
+    spec = RUNTIME_SETTINGS.get(key)
+    return spec is not None and spec.scope is Scope.SENSITIVE
+
+
+def redact_setting(value: Any, key: str | None = None) -> Any:
     """Mask the credential-bearing parts of a setting value.
 
     Walks lists and mappings because the values that matter are nested: the
     channel descriptors in ``notify.channels`` are where a bot token lives. Every
     surface that shows a setting goes through here - the CLI's tables, the admin
     API's responses, and the audit row a settings write leaves behind.
+
+    ``key`` is what makes a *scalar* secret maskable, and its absence was a real
+    leak. This function used to decide purely on shape: it masked by field name
+    inside a mapping or a list and returned any top-level scalar untouched. So
+    ``notify.channels`` - a list of records - was masked correctly on every
+    surface, while ``security.webhook_secret`` - a bare string, declared
+    SENSITIVE - was returned verbatim by the settings listing, echoed by the
+    write, and written in cleartext into an audit row that retention never
+    trims. A caller with only ``identity:manage`` could read it while being
+    refused permission to change it.
+
+    Shape cannot answer this question; only the declaration can, which is why
+    the key is passed in.
     """
     if isinstance(value, Mapping):
         return {field: _redact_field(str(field), item) for field, item in value.items()}
     if isinstance(value, list):
         return [redact_setting(item) for item in value]
+    if is_sensitive_setting(key) and isinstance(value, str):
+        # An empty string is "not set", which is not a secret and is worth
+        # being able to see.
+        return mask_secret(value) if value else value
     return value
 
 
-def unredact_setting(value: Any, stored: Any) -> Any:
+def unredact_setting(value: Any, stored: Any, key: str | None = None) -> Any:
     """Put back the stored values that the masks in ``value`` stand in for.
 
     Masking on read breaks an editable setting, because the console reads the
@@ -271,6 +304,18 @@ def unredact_setting(value: Any, stored: Any) -> Any:
         ]
     if isinstance(value, Mapping):
         return _merge_record(value, stored if isinstance(stored, Mapping) else None)
+    if (
+        is_sensitive_setting(key)
+        and isinstance(value, str)
+        and isinstance(stored, str)
+        and stored
+        and value == mask_secret(stored)
+    ):
+        # The console reads the masked value and writes the whole form back.
+        # Without this, saving any other field on the page would store the mask
+        # as the secret. Matched against the mask *of the stored value*, so a
+        # caller cannot invent one and have a credential handed back to them.
+        return stored
     return value
 
 
@@ -384,6 +429,7 @@ __all__ = [
     "SECRET_FIELDS",
     "URL_FIELDS",
     "is_credential_field",
+    "is_sensitive_setting",
     "mask_api_key",
     "mask_cookies",
     "mask_endpoint",

src/dtk/services/cache.py

@@ -24,11 +24,24 @@ CACHE_KEY = "cache:{digest}"
 INFLIGHT_KEY = "inflight:{digest}"
 
 
-def cache_key(endpoint: str, params: dict[str, Any]) -> str:
+def cache_key(endpoint: str, params: dict[str, Any], *, include_raw: bool = False) -> str:
     """Digest of the normalized business parameters, not of the raw URL.
 
     Two callers who reach the same content by different URL spellings must land
     on the same entry.
+
+    ``include_raw`` is part of the key because the cached value is the *shaped*
+    body, not the platform's answer: ``_dump`` strips ``raw`` before it is
+    stored. Leaving it out froze whichever value the first caller used for the
+    whole TTL, in both directions - a later ``include_raw=true`` got a body with
+    no ``raw`` in it, and a later ``include_raw=false`` got one that still had
+    it. The second is the one that matters: the raw payload is the largest
+    thing this instance stores, and it was being handed to, and persisted in
+    ``tasks.result`` for, a caller who explicitly opted out.
+
+    It is not part of the *upstream* request, so this does mean two fetches
+    where a smarter cache would do one. That is the honest trade until the
+    cache stores the unshaped body.
     """
     canonical = json.dumps(
         {k: v for k, v in sorted(params.items()) if v is not None},
@@ -36,7 +49,8 @@ def cache_key(endpoint: str, params: dict[str, Any]) -> str:
         ensure_ascii=False,
         default=str,
     )
-    return hashlib.sha256(f"{endpoint}\x00{canonical}".encode()).hexdigest()[:32]
+    suffix = "\x00raw" if include_raw else ""
+    return hashlib.sha256(f"{endpoint}\x00{canonical}{suffix}".encode()).hexdigest()[:32]
 
 
 async def get(digest: str) -> dict[str, Any] | None:

src/dtk/services/downloads.py

@@ -26,8 +26,8 @@ whole of doc 08's fourth constraint, restated where it is enforced.
 from __future__ import annotations
 
 import uuid
-from collections.abc import Sequence
-from dataclasses import dataclass
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta
 from typing import Any, Final
 
@@ -363,6 +363,11 @@ class Eviction:
 
     paths: tuple[str, ...] = ()
     ids: tuple[uuid.UUID, ...] = ()
+    #: Which rows live on each directory. Several rows routinely share one -
+    #: the layout has nothing per-download in it - and every one of them loses
+    #: its files when that directory goes, so the caller needs the grouping to
+    #: mark the right rows after a partial delete.
+    ids_by_path: Mapping[str, tuple[uuid.UUID, ...]] = field(default_factory=dict)
     bytes_freed: int = 0
     #: Bytes still on the volume that the sweep cannot touch, because they are
     #: pinned. Surfaced rather than swallowed: an operator who pins more than
@@ -378,27 +383,54 @@ def choose_evictions(rows: Sequence[Any], *, over_by: int) -> Eviction:
     part with a policy in it, and a policy that deletes files should be
     testable without a database.
 
-    ``rows`` arrives oldest first. Pinned rows are counted and skipped rather
-    than filtered out beforehand, so the caller can say how much of the volume
-    the sweep is unable to touch - an operator who has pinned more than their
-    own ceiling has silently disabled their cleanup, and should hear about it.
+    ``rows`` arrives oldest first.
+
+    **The unit is the directory, not the row.** ``directory_of`` is
+    ``platform/author/content`` with nothing per-download in it, and nothing
+    stops the same post being downloaded twice, so several rows routinely share
+    one directory - and what the sidecar deletes is a directory, recursively.
+    Deciding row by row got both halves wrong: it scheduled a directory for
+    deletion because *one* of its rows was unpinned, destroying the pinned
+    row's files while reporting those same bytes as protected; and it added the
+    same directory's bytes to ``bytes_freed`` once per row, so the sweep
+    believed it had freed twice what it had and stopped short of the ceiling.
+
+    So: group by directory, refuse the whole directory if **any** row on it is
+    pinned, and count its bytes once. A pin protects the files, which is what
+    the pin endpoint promises and what an operator would assume.
     """
+    groups: dict[str, dict[str, Any]] = {}
+    for row in rows:
+        group = groups.setdefault(
+            row.directory, {"pinned": False, "bytes": 0, "ids": [], "order": len(groups)}
+        )
+        group["ids"].append(row.id)
+        group["pinned"] = group["pinned"] or bool(row.pinned)
+        # The largest claim wins rather than the sum: two rows for one post
+        # describe the same files, and adding them counts the bytes twice.
+        group["bytes"] = max(int(group["bytes"]), int(row.bytes_total))
+
     freed = 0
     paths: list[str] = []
     ids: list[uuid.UUID] = []
+    by_path: dict[str, tuple[uuid.UUID, ...]] = {}
     pinned_bytes = 0
-    for row in rows:
-        if row.pinned:
-            pinned_bytes += int(row.bytes_total)
+    for directory, group in sorted(groups.items(), key=lambda item: item[1]["order"]):
+        if group["pinned"]:
+            pinned_bytes += int(group["bytes"])
             continue
         if freed >= over_by:
             continue
-        paths.append(row.directory)
-        ids.append(row.id)
-        freed += int(row.bytes_total)
+        paths.append(directory)
+        # Every row on the directory is marked evicted, because every one of
+        # them loses its files when the directory goes.
+        ids.extend(group["ids"])
+        by_path[directory] = tuple(group["ids"])
+        freed += int(group["bytes"])
     return Eviction(
         paths=tuple(paths),
         ids=tuple(ids),
+        ids_by_path=by_path,
         bytes_freed=freed,
         pinned_bytes=pinned_bytes,
         over_by=over_by,

src/dtk/services/fetch.py

@@ -354,7 +354,7 @@ class FetchService:
         ctx = ctx or FetchContext()
         adapter = get_adapter(platform)
         started = time.monotonic()
-        digest = cache.cache_key(endpoint, params)
+        digest = cache.cache_key(endpoint, params, include_raw=ctx.include_raw)
 
         hit = await cache.get(digest) if cache_ttl > 0 else None
         if hit is not None:

src/dtk/transport/classify.py

@@ -443,6 +443,44 @@ def _withheld_payload(view: ResponseView) -> RuleResult:
     return False
 
 
+#: Keys an envelope may carry that say nothing about the content: the status,
+#: its message, and the tracing crumbs both platforms attach to everything.
+_ENVELOPE_META_KEYS: Final[frozenset[str]] = frozenset(
+    {*ENVELOPE_STATUS_KEYS, *ENVELOPE_MESSAGE_KEYS, "extra", "log_pb", "logid", "log_id"}
+)
+
+
+def _bare_envelope(view: ResponseView) -> RuleResult:
+    """A 200 whose envelope carries a status and nothing else.
+
+    Measured on 2026-09-08: Douyin answers ``douyin.author_posts`` with any
+    non-zero ``max_cursor`` as HTTP 200 and exactly ``{"status_code":0}`` - 17
+    bytes, no ``aweme_list``, no reason - while the same request with
+    ``max_cursor=0`` returns 612 KB. It is not the end of the feed; older posts
+    demonstrably exist.
+
+    :func:`_withheld_payload` could not see it, because that rule fires on a
+    payload key that is *present and empty*, and here the key is absent
+    entirely. So the response was classified ``default.ok``, handed to the
+    parser, and came back as ``UPSTREAM_CHANGED`` - a non-retryable error whose
+    message asks the operator to report a parser bug that no parser change can
+    fix. It also went into ``request_log`` as ``outcome=ok, http_status=200``,
+    so a systematic refusal never reached pool health or the endpoint breaker.
+
+    Keyed on "nothing but metadata" rather than on a list of payload names,
+    because the payload key differs per endpoint - ``aweme_list``, ``comments``,
+    ``user`` - and a rule that had to know them all would miss the next one.
+    """
+    if not view.response.ok:
+        return False
+    envelope = view.envelope
+    if not isinstance(envelope, dict) or not envelope:
+        return False
+    if any(key not in _ENVELOPE_META_KEYS for key in envelope):
+        return False
+    return "envelope carries a status and no payload"
+
+
 def _server_error(view: ResponseView) -> RuleResult:
     if view.status >= 500:
         return f"http {view.status}"
@@ -469,6 +507,9 @@ DEFAULT_RULES: Final[tuple[ClassificationRule, ...]] = (
     ClassificationRule("body.empty", Outcome.RISK_CONTROL, _empty_body),
     ClassificationRule("payload.explained", Outcome.BUSINESS_ERROR, _explained_absence),
     ClassificationRule("payload.withheld", Outcome.RISK_CONTROL, _withheld_payload),
+    # After payload.withheld and payload.explained, so a refusal that names its
+    # reason is still reported as the business answer it is.
+    ClassificationRule("payload.bare_envelope", Outcome.RISK_CONTROL, _bare_envelope),
     ClassificationRule("http.server_error", Outcome.NETWORK_ERROR, _server_error),
     ClassificationRule("http.client_error", Outcome.BUSINESS_ERROR, _client_error),
 )

src/dtk/worker/maintenance.py

@@ -282,10 +282,14 @@ class Maintenance:
         # Only what the sidecar confirmed it removed is marked evicted. Marking
         # optimistically would produce a row claiming the file is gone while it
         # is still on the disk, and the next sweep would then never reclaim it.
+        # Grouped by directory rather than zipped against it: several rows can
+        # share one directory, so the two tuples are not parallel and zipping
+        # them raised as soon as a post had been downloaded twice.
         evicted_ids = [
             download_id
-            for path, download_id in zip(eviction.paths, eviction.ids, strict=True)
+            for path in eviction.paths
             if path in confirmed
+            for download_id in eviction.ids_by_path.get(path, ())
         ]
         async with self._session_factory() as session:
             report.media_evicted = await downloads.mark_evicted(session, evicted_ids)

tests/unit/test_downloads.py

@@ -9,6 +9,7 @@ exception worth pinning down precisely.
 
 from __future__ import annotations
 
+import itertools
 import uuid
 from datetime import UTC, datetime, timedelta
 from types import SimpleNamespace
@@ -21,10 +22,20 @@ from dtk.services import downloads
 MB = 1024 * 1024
 
 
+_ROWS = itertools.count()
+
+
 def row(bytes_total: int, *, pinned: bool = False, name: str = "") -> SimpleNamespace:
+    """One download row.
+
+    Each gets its OWN directory unless `name` says otherwise. The default used
+    to be derived from the byte count, so two rows of the same size silently
+    shared a directory - which is the exact collision `choose_evictions` groups
+    on, and it made this helper hide the thing the tests below now pin.
+    """
     return SimpleNamespace(
         id=uuid.uuid4(),
-        directory=name or f"douyin/a/{bytes_total}",
+        directory=name or f"douyin/a/{next(_ROWS)}",
         bytes_total=bytes_total,
         pinned=pinned,
     )
@@ -153,3 +164,44 @@ def test_a_media_download_is_not_exempt_from_the_queue_ceiling() -> None:
 
     assert Maintenance.MEDIA_DOWNLOAD.value not in _OPERATOR_TRIGGERED
     assert Maintenance.DIAGNOSE.value in _OPERATOR_TRIGGERED
+
+
+class TestSharedDirectories:
+    """Several downloads of one post land in one directory.
+
+    `directory_of` is platform/author/content with nothing per-download in it,
+    and nothing dedupes, so this is the normal case rather than an edge one -
+    and what the sidecar deletes is the whole directory, recursively.
+    """
+
+    def test_a_pin_protects_every_row_on_its_directory(self):
+        """The defect this replaced: the sweep scheduled a directory for
+        deletion because one of its rows was unpinned, destroying the pinned
+        row's files while reporting those same bytes as protected."""
+        shared = "douyin/MS4wA/7408"
+        pinned = row(240 * MB, pinned=True, name=shared)
+        duplicate = row(240 * MB, name=shared)
+        plan = downloads.choose_evictions([pinned, duplicate], over_by=100 * MB)
+
+        assert plan.paths == ()
+        assert plan.pinned_bytes == 240 * MB
+
+    def test_one_directory_is_counted_once_not_once_per_row(self):
+        """Two rows describing the same files are not twice the disk. Counting
+        them twice made the sweep believe it had freed enough and stop short."""
+        shared = "douyin/MS4wA/7408"
+        plan = downloads.choose_evictions(
+            [row(240 * MB, name=shared), row(240 * MB, name=shared)], over_by=100 * MB
+        )
+        assert plan.paths == (shared,)
+        assert plan.bytes_freed == 240 * MB
+
+    def test_every_row_on_a_removed_directory_is_marked(self):
+        """All of them lose their files, so all of them have to be recorded as
+        evicted - otherwise a row keeps claiming on_disk for files that are gone."""
+        shared = "douyin/MS4wA/7408"
+        first, second = row(240 * MB, name=shared), row(240 * MB, name=shared)
+        plan = downloads.choose_evictions([first, second], over_by=100 * MB)
+
+        assert set(plan.ids) == {first.id, second.id}
+        assert plan.ids_by_path[shared] == (first.id, second.id)

tests/unit/test_fetch_classification.py

@@ -634,3 +634,48 @@ async def test_the_signature_and_the_request_describe_the_same_visitor() -> None
     # half a bare fingerprint would not have carried.
     assert signed_as[0].cookies == {"ttwid": "x"}
     assert signed_as[0].proxy_url == "http://user:pass@exit.example:8080"
+
+
+class TestBareEnvelope:
+    """A 200 carrying a status and no payload.
+
+    Measured on 2026-09-08: Douyin answers author_posts with any non-zero
+    max_cursor as exactly `{"status_code":0}` - 17 bytes - while max_cursor=0
+    returns 612 KB. It is a refusal, not the end of the feed. It used to reach
+    the parser and come back as a non-retryable UPSTREAM_CHANGED telling the
+    operator to report a parser bug, and it was logged as outcome=ok, so the
+    pool never learned anything from it.
+    """
+
+    def _classify(self, body: bytes, status: int = 200):
+        return classify_detailed(response(status=status, body=body))
+
+    def test_a_status_only_envelope_is_withheld_not_ok(self):
+        result = self._classify(b'{"status_code":0}')
+        assert result.outcome is Outcome.RISK_CONTROL
+        assert result.rule == "payload.bare_envelope"
+
+    def test_tracing_crumbs_do_not_make_it_look_like_a_payload(self):
+        result = self._classify(b'{"status_code":0,"status_msg":"","log_pb":{"impr_id":"x"}}')
+        assert result.outcome is Outcome.RISK_CONTROL
+
+    def test_a_real_answer_is_untouched(self):
+        """The rule must key on 'nothing but metadata', not on a payload name -
+        the payload key differs per endpoint."""
+        result = self._classify(b'{"status_code":0,"aweme_list":[{"aweme_id":"1"}],"has_more":1}')
+        assert result.outcome is Outcome.OK
+
+    def test_an_empty_list_is_still_a_real_answer(self):
+        """An author with no posts answers with the key present and empty; that
+        is the end of a feed, not a refusal."""
+        result = self._classify(b'{"status_code":0,"aweme_list":[],"has_more":0}')
+        assert result.outcome is Outcome.OK
+
+    def test_a_named_refusal_still_reports_its_own_reason(self):
+        """payload.explained runs first, so a platform that says why keeps
+        being reported as the business answer it is."""
+        body = (
+            b'{"status_code":0,"aweme_detail":null,'
+            b'"filter_detail":{"filter_reason":"self_see","detail_msg":"only you"}}'
+        )
+        assert self._classify(body).outcome is Outcome.BUSINESS_ERROR