From fe2403f810aa755ddaab5d7d7acd504a2add8a5b Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Tue, 16 Jun 2026 03:38:00 -0700 Subject: [PATCH] refactor(attachments): retire the vestigial reservation scaffolding MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 2 +- turnstone/api/console_spec.py | 8 ++--- turnstone/api/server_schemas.py | 6 ++-- turnstone/api/server_spec.py | 2 +- turnstone/console/server.py | 4 +-- turnstone/core/session.py | 9 +++-- turnstone/core/session_routes.py | 57 ++++++++++++-------------------- turnstone/core/session_worker.py | 2 +- turnstone/sdk/console.py | 6 ++-- turnstone/sdk/server.py | 2 +- turnstone/server.py | 5 +-- 11 files changed, 44 insertions(+), 59 deletions(-) diff --git a/turnstone/api/console_schemas.py b/turnstone/api/console_schemas.py index 10a63212..1edb9023 100644 --- a/turnstone/api/console_schemas.py +++ b/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." ), diff --git a/turnstone/api/console_spec.py b/turnstone/api/console_spec.py index d601497b..5588f956 100644 --- a/turnstone/api/console_spec.py +++ b/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], diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index c99b5668..48b9bbc1 100644 --- a/turnstone/api/server_schemas.py +++ b/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." ), diff --git a/turnstone/api/server_spec.py b/turnstone/api/server_spec.py index 48db74ed..0b8890b5 100644 --- a/turnstone/api/server_spec.py +++ b/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`." ), diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 3eafb953..5326a3fd 100644 --- a/turnstone/console/server.py +++ b/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 diff --git a/turnstone/core/session.py b/turnstone/core/session.py index fecdfd61..78f211c7 100644 --- a/turnstone/core/session.py +++ b/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: diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index 4458f434..c1f03077 100644 --- a/turnstone/core/session_routes.py +++ b/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, } ) diff --git a/turnstone/core/session_worker.py b/turnstone/core/session_worker.py index c57d5dd3..982eb87b 100644 --- a/turnstone/core/session_worker.py +++ b/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. """ diff --git a/turnstone/sdk/console.py b/turnstone/sdk/console.py index cb58980c..c4c717bd 100644 --- a/turnstone/sdk/console.py +++ b/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: diff --git a/turnstone/sdk/server.py b/turnstone/sdk/server.py index 216b6c73..1ef885c9 100644 --- a/turnstone/sdk/server.py +++ b/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. """ diff --git a/turnstone/server.py b/turnstone/server.py index cbdc7c37..41b8773d 100644 --- a/turnstone/server.py +++ b/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``.