refactor(attachments): retire the vestigial reservation scaffolding

The by-ref change replaced the send_id reservation model with the per-node
upload buffer (peek-then-drain at write time), but the surrounding narration was
never swept and a no-op stub was retained to make the send handler "read like"
the old flow — which is what made a recent diagnosis assume reservations still
existed.

- Delete the no-op _release_reservation_on_fail() and its 5 call sites in the
  send handler (behaviour-preserving — it did nothing).
- Rename ordered_reserved / reserved_set -> ordered_taken / taken_set (the values
  are the "taken" subset from resolve_staged_attachments, not reservations).
- Sweep the stale "reserve/reservation" wording across the create/send
  docstrings, the API schemas/specs, and the SDK docstrings to the staged-buffer
  vocabulary (resolve / attach / drain). The canonical docs in attachment_buffer
  and attachments already stated the reservation token is gone.

No behaviour change; no tests exercised the removed scaffolding (the
migration-060 test correctly pins the reserved_at column removal and stays).

turnstone/api/console_schemas.py

@@ -1224,7 +1224,7 @@ class CoordinatorSendResponse(BaseModel):
     attached_ids: list[str] = Field(
         default_factory=list,
         description=(
-            "Attachment ids actually reserved onto this turn. Subset of "
+            "Attachment ids actually attached to this turn. Subset of "
             "the request's `attachment_ids` (or the auto-consumed pending "
             "set). Empty when the send carries no attachments."
         ),

turnstone/api/console_spec.py

@@ -1261,10 +1261,10 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
         "Queue a user message onto the coordinator session",
         description=(
             "Worker thread picks up the message via the session's queue. "
-            "Optional ``attachment_ids`` reserve attachments under the "
-            "message's send_id token (parity with the interactive surface). "
+            "Optional ``attachment_ids`` select staged uploads to attach to "
+            "the message (parity with the interactive surface). "
             "Response carries ``attached_ids`` / ``dropped_attachment_ids`` "
-            "so callers can detect partial reservations and ``priority`` / "
+            "so callers can detect partial attaches and ``priority`` / "
             "``msg_id`` on the queued path. "
             "``status: queue_full`` when the worker queue is full — caller "
             "should back off."
@@ -1284,7 +1284,7 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
             "the interactive surface: magic-byte image sniff, UTF-8 text "
             "decode, per-kind size cap, per-(ws,user) pending cap. "
             "Attachments stay pending until a subsequent ``/send`` "
-            "reserves them under its ``send_id`` token."
+            "attaches them to a message."
         ),
         response_model=UploadAttachmentResponse,
         error_codes=[400, 403, 404, 409, 413, 503],

turnstone/api/server_schemas.py

@@ -45,7 +45,7 @@ class SendResponse(BaseModel):
     attached_ids: list[str] = Field(
         default_factory=list,
         description=(
-            "Attachment ids actually reserved onto this turn. Subset of "
+            "Attachment ids actually attached to this turn. Subset of "
             "the request's `attachment_ids` (or the auto-consumed pending "
             "set). Empty when the send carries no attachments."
         ),
@@ -159,7 +159,7 @@ class CreateWorkstreamRequest(BaseModel):
         description=(
             "Optional first user message dispatched as a background turn after "
             "the workstream is created. When attachments are also provided "
-            "(via the multipart variant), they are reserved onto this turn."
+            "(via the multipart variant), they are attached to this turn."
         ),
     )
     ws_id: str = Field(
@@ -201,7 +201,7 @@ class CreateWorkstreamResponse(BaseModel):
         default_factory=list,
         description=(
             "Ids of attachments saved by this request (multipart variant only). "
-            "Already reserved onto the initial_message turn when one was provided; "
+            "Already attached to the initial_message turn when one was provided; "
             "otherwise left pending for a follow-up POST "
             "/v1/api/workstreams/{ws_id}/send."
         ),

turnstone/api/server_spec.py

@@ -75,7 +75,7 @@ SERVER_ENDPOINTS: list[EndpointSpec] = [
             "with one `meta` field (JSON-encoded `CreateWorkstreamRequest` shape) "
             "plus zero-or-more `file` parts saves each file as an attachment "
             "under the new workstream. When `initial_message` is also set, "
-            "attachments are reserved onto that turn before the worker thread "
+            "attachments are resolved onto that turn before the worker thread "
             "dispatches; otherwise they remain pending for a follow-up "
             "`POST /v1/api/workstreams/{ws_id}/send`."
         ),

turnstone/console/server.py

@@ -3408,8 +3408,8 @@ async def _coord_create_post_install(
     Wired onto :attr:`SessionEndpointConfig.create_post_install`. When
     an ``initial_message`` is provided, dispatches via
     :meth:`CoordinatorAdapter.send`; any uploaded ``attachment_ids``
-    are reserved onto the same ``send_id`` token so the worker's
-    first turn picks them up exactly the way interactive's
+    are resolved from the buffer onto the first turn (and drained) so
+    the worker picks them up exactly the way interactive's
     ``post_install`` worker thread does.
 
     Returns ``{}`` — coord's response carries only the always-include

turnstone/core/session.py

@@ -6166,7 +6166,7 @@ class ChatSession:
         the current turn to finish before allowing an attached send.
 
         ``queue_msg_id`` lets the caller supply the id (so it matches the
-        attachment-reservation token already taken server-side) — when
+        ``send_id`` tracking token threaded through the send) — when
         omitted, an id is generated.
         """
         from turnstone.core.tool_advisory import parse_priority
@@ -6181,10 +6181,9 @@ class ChatSession:
         # Cap individual message length to prevent context bloat
         if len(cleaned) > 2000:
             cleaned = cleaned[:2000] + "..."
-        # Full UUID hex (128 bits) rather than a truncated prefix — this
-        # id doubles as a cross-table reservation token on
-        # workstream_attachments, and a 48-bit truncation narrows the
-        # birthday bound unnecessarily.
+        # Full UUID hex (128 bits) rather than a truncated prefix — this id is
+        # the ``send_id`` tracking token threaded through the turn, so the wide
+        # space keeps the birthday bound comfortable.
         msg_id = queue_msg_id or uuid.uuid4().hex
         with self._queued_lock:
             if len(self._queued_messages) >= self._QUEUE_MAX:

turnstone/core/session_routes.py

@@ -327,7 +327,7 @@ class SessionEndpointConfig:
     Capability flags (added with the P1.5 ``send`` body lift):
 
     - ``supports_attachments``: when ``True``, the lifted ``send``
-      handler resolves attachment_ids, reserves under a send_id token,
+      handler resolves attachment_ids from the per-node upload buffer
       and threads them through ``ChatSession.send`` /
       ``ChatSession.queue_message``. Both kinds wire ``True`` post-P1.5
       (the storage layer was always kind-agnostic; the gate stays
@@ -2065,8 +2065,8 @@ def make_create_handler(
       the lifted body parses multipart bodies on coord and saves
       attachments through the kind-agnostic storage layer (§ Post-P3
       reckoning item #1). When the same request supplies an
-      ``initial_message``, the uploads are reserved onto the
-      dispatched first turn via ``CoordinatorAdapter.send`` (which
+      ``initial_message``, the uploads are resolved from the buffer onto
+      the dispatched first turn via ``CoordinatorAdapter.send`` (which
       gained ``attachments`` + ``send_id`` kwargs in the same
       release).
     - **No phantom create→close pair on coord rollback.** The lifted
@@ -3416,9 +3416,9 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
     Capability flags on ``cfg`` toggle the kind-specific behaviour:
 
     - ``supports_attachments``: when ``False``, the entire
-      attachment-resolution block (reservation, fetch, scope-check)
+      attachment-resolution block (buffer peek + scope-check)
       short-circuits and any ``attachment_ids`` in the body are
-      silently ignored — no reservation, no error. Both kinds wire
+      silently ignored — no resolution, no error. Both kinds wire
       ``True`` post-P1.5; the flag exists so a kind that hasn't
       lit up its UI surface yet can defer.
     - ``spawn_metrics``: when set, fires once on the spawn path with
@@ -3490,8 +3490,8 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
         # ----- Attachment resolution (from the per-node upload buffer) -----
         send_id = ""
         requested_ids: list[str] = []
-        ordered_reserved: list[str] = []
-        reserved_set: set[str] = set()
+        ordered_taken: list[str] = []
+        taken_set: set[str] = set()
         resolved_atts: list[Any] = []
         attach_user_id = ""
 
@@ -3524,18 +3524,10 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
             # turn the queue rejects can still be retried; the committing
             # ``send`` drains them at write time.  ``resolved`` carries the
             # bytes the session persists content-addressed.
-            resolved_atts, ordered_reserved, _dropped_resolve = resolve_staged_attachments(
+            resolved_atts, ordered_taken, _dropped_resolve = resolve_staged_attachments(
                 requested_ids, ws_id, attach_user_id
             )
-            reserved_set = set(ordered_reserved)
-
-        def _release_reservation_on_fail() -> None:
-            """No-op: the upload buffer is a peek, not a lock.
-
-            Retained as the worker-failure hook so the call sites below read
-            the same as the pre-cutover reservation flow; there is nothing to
-            release — undrained staged bytes simply expire on the buffer TTL.
-            """
+            taken_set = set(ordered_taken)
 
         # If a cancel was just issued, briefly poll for the worker to
         # exit before dispatching — avoids spawning into a stale
@@ -3548,7 +3540,6 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
                 if not ws._worker_running:
                     break
         if ws.session is None:
-            _release_reservation_on_fail()
             return JSONResponse({"error": "No session"}, status_code=500)
 
         session = ws.session
@@ -3560,7 +3551,7 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
             try:
                 cleaned, priority, msg_id = session.queue_message(
                     message,
-                    attachment_ids=list(ordered_reserved),
+                    attachment_ids=list(ordered_taken),
                     queue_msg_id=send_id or None,
                 )
             except AttachmentsNotQueueableError:
@@ -3606,16 +3597,13 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
                 # Safety net — send() normally handles this internally.
                 # If this thread was force-abandoned, ws.worker_thread
                 # was set to None — don't emit spurious events.
-                _release_reservation_on_fail()
                 if ws.worker_thread is me:
                     _emit_ui("on_stream_end")
                     _emit_ui("on_state_change", "idle")
             except Exception:
-                # Release the reservation so attachments don't stay
-                # soft-locked forever on a worker crash before the
-                # consume step. Idempotent: once consume cleared the
-                # token, a follow-up unreserve is a no-op.
-                _release_reservation_on_fail()
+                # Undrained staged uploads aren't locked (the buffer is a peek,
+                # not a reservation) — they expire on the buffer TTL — so the
+                # only cleanup owed here is the UI streaming hook.
                 if ws.worker_thread is me:
                     # ``session.send()`` already fired ``on_error``
                     # (with sanitized text), persisted ``last_error``,
@@ -3634,12 +3622,10 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
         )
         if not ok:
             # queue.Full or session-disappeared race — surface as
-            # queue_full so clients retry rather than 500. Reservations
-            # released above; ``attached_ids`` is always empty on this
-            # path (the dispatch never took ownership). The empty
-            # arrays preserve the response-shape guarantee so SDK
-            # consumers don't branch on status.
-            _release_reservation_on_fail()
+            # queue_full so clients retry rather than 500. ``attached_ids``
+            # is always empty on this path (the dispatch never took
+            # ownership); the empty arrays preserve the response-shape
+            # guarantee so SDK consumers don't branch on status.
             return JSONResponse(
                 {
                     "status": "queue_full",
@@ -3651,9 +3637,8 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
         if queue_outcome.get("rejected") == "attachments_busy":
             # Attachments can't ride a queued user turn (see
             # AttachmentsNotQueueableError for the role-ordering reason).
-            # Release reservations and surface to the caller so the
+            # The staged uploads stay in the buffer (peek, not drain) so the
             # client can hold the file and retry once the worker idles.
-            _release_reservation_on_fail()
             return JSONResponse(
                 {
                     "status": "attachments_busy",
@@ -3662,7 +3647,7 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
                 }
             )
 
-        dropped = [aid for aid in requested_ids if aid not in reserved_set]
+        dropped = [aid for aid in requested_ids if aid not in taken_set]
         if queue_outcome:
             # Reused a live worker; ``queue_message`` succeeded.
             if cfg.emit_message_queued and hasattr(ui, "_enqueue"):
@@ -3679,7 +3664,7 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
                     "status": "queued",
                     "priority": queue_outcome["priority"],
                     "msg_id": queue_outcome["msg_id"],
-                    "attached_ids": list(ordered_reserved),
+                    "attached_ids": list(ordered_taken),
                     "dropped_attachment_ids": dropped,
                 }
             )
@@ -3697,7 +3682,7 @@ def make_send_handler(cfg: SessionEndpointConfig) -> Handler:
         return JSONResponse(
             {
                 "status": "ok",
-                "attached_ids": list(ordered_reserved),
+                "attached_ids": list(ordered_taken),
                 "dropped_attachment_ids": dropped,
             }
         )

turnstone/core/session_worker.py

@@ -20,7 +20,7 @@ fix.
 
 This module owns ONLY the dispatch decision and the
 ``_worker_running`` lifecycle. Per-kind concerns — session resolution,
-attachments reservation, error surfacing, UI callbacks,
+attachment resolution, error surfacing, UI callbacks,
 ``GenerationCancelled`` handling — live in the caller's
 ``enqueue`` / ``run`` no-arg closures.
 """

turnstone/sdk/console.py

@@ -346,9 +346,9 @@ class AsyncTurnstoneConsole(_BaseClient):
     ) -> dict[str, Any]:
         """Send a message to a coordinator workstream.
 
-        ``attachment_ids`` reserves attachments under the message's
-        ``send_id`` token; pass ``None`` to auto-consume the caller's
-        pending attachments, or ``[]`` to disable auto-consume.
+        ``attachment_ids`` selects which staged uploads to attach to the
+        message; pass ``None`` to auto-consume the caller's pending
+        attachments, or ``[]`` to disable auto-consume.
         """
         body: dict[str, Any] = {"message": message}
         if attachment_ids is not None:

turnstone/sdk/server.py

@@ -120,7 +120,7 @@ class AsyncTurnstoneServer(_BaseClient):
         field and one ``file`` part per attachment.  A ws_id is
         auto-generated client-side when not supplied so cluster-routed
         callers can bind the body to the owning node up front.  When
-        *initial_message* is also set, the server reserves the
+        *initial_message* is also set, the server resolves the staged
         attachments onto that turn before its background worker
         dispatches.
         """

turnstone/server.py

@@ -1917,8 +1917,9 @@ async def _interactive_create_post_install(
     8. Pin the workstream's routing to this node when no caller-
        supplied ``ws_id`` was provided (direct creates).
     9. Spawn the initial-message worker thread when ``initial_message``
-       is set, reserving any uploaded attachments for that first
-       turn.
+       is set, resolving any staged uploads from the buffer onto that
+       first turn (then draining them so a freshly-opened pane's
+       rehydrate can't observe them as still-pending).
 
     Returns ``{resumed, message_count}`` for the response. On the
     no-resume path both default to ``False`` / ``0``.