diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7f40cb7a..1e89ec3d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,17 @@ Sync wave from `chat@4.31.0` to `chat@4.41.1` (tracking #184). `UPSTREAM_PARITY`
   - `--report-target` prints its delta against the report committed at `HEAD` (via `git show`), so a second run before committing no longer compares against the first run's output and prints `+0`.
   - `--check-docs` is case-insensitive and also catches `--branch=`, `-b`, line-continued and markdown-decorated pins.
   - `scripts/fidelity_target.json` is committed as the authoritative wave-wide missing list at `chat@4.41.1`: 282 missing of 1036 (130 in strict-tier files, 152 in target-tier files), including 16 `.each` templates.
+- **Breaking (security) — callback-URL button tokens are single-use, bound to their button and conversation, and expire after 7 days** (#194). Ports the callback-token part of upstream `b7c9316b` (vercel/chat#875, chat@4.40.0) and the button copy from `4a0b5c0c` (vercel/chat#895, chat@4.40.0).
+  - **Pre-upgrade tokens stop resolving.** A token is now stored as `{actionId, url, originalValue?, scope: {id, type}}`. Records written before the upgrade (`{url, originalValue}` or a bare URL string) are rejected: clicking such a button runs `on_action` handlers with the raw `__cb:…` value and POSTs nothing. The resolver also no longer reads a snake_case `original_value` key.
+  - **A repeat click no longer POSTs.** Resolving a token deletes it, under a 10-second per-token state lock. A second click, or one that lands while the first holds the lock, dispatches the raw `__cb:…` value without a POST. The record is deleted before the POST, so a failed POST is not retried.
+  - **Tokens expire after 7 days** (was 30).
+  - **The POST `actionId` is now the minted button's id**, read from the stored record, instead of the incoming event's `action_id`.
+  - Tokens resolve only for the button that minted them (`actionId`) and in the conversation they were posted to. Thread posts, schedules and edits bind to the thread. Channel posts, schedules and channel `SentMessage.edit`s bind to the channel. A `post_ephemeral` DM fallback binds to the DM channel, and with neither native ephemeral nor a DM fallback no token is minted. A click whose action or conversation does not match leaves the record in place.
+  - The token swap now keeps every button field except `callback_url` (it used to copy a fixed whitelist), so new fields such as `tooltip` (#202) survive.
+  - **Python-specific (divergence from upstream):** after deleting a matched record, the resolver checks with `extend_lock` that its 10-second lease never lapsed, and returns `None` if it did. Upstream returns the record regardless, so a state call that stalls past the lease can let a second click also resolve and POST. See `docs/UPSTREAM_SYNC.md`.
+  - **Python-specific (divergence from upstream):** a channel `SentMessage.edit` binds its tokens to the channel (the scope the original `channel.post` used), not to `{reported thread id, "thread"}`. The thread id an adapter reports for a channel post often never equals a click's thread id: Teams and Google Chat report the channel id, Slack reports the synthetic `slack:C…:` (a click carries the message ts, and a Slack DM click carries no ts even once #209 makes the post report one), and a chained edit drops the reported id. Upstream's thread scope would leave those edited buttons never POSTing.
+  - Known gap (see `docs/UPSTREAM_SYNC.md`): Google Chat cards `thread.post`ed into a DM thread never POST (upstream parity: the card click omits the `:dm` suffix).
+  - API: `process_card_callback_urls(card, state, scope)` takes a required `CallbackScope`. `resolve_callback_url(token, state, context=None)` takes a `CallbackContext` (a `None` context never matches). `ResolvedCallback` gains keyword-only `action_id` and `scope`. New constant: `CALLBACK_LOCK_TTL_MS = 10_000`.
 - **BREAKING (security) — Telegram: webhook verification is required by default; repeated updates are deduplicated** (#224; upstream vercel/chat#858, #799, #813).
   - Telegram webhook deployments without `TELEGRAM_WEBHOOK_SECRET_TOKEN` / `secret_token` now **fail to start or return 401**: `mode="webhook"` raises `ValidationError` in the constructor, `mode="auto"` raises from `initialize()` when it resolves to webhook mode (so `Chat` initialization fails and is retried on every webhook until the config is fixed), and `handle_webhook` returns 401 `"Webhook verification required"` before reading the body. Previously the adapter logged a warning and dispatched every update, including `callback_query` button actions.
   - **Escape hatch:** `TelegramAdapterConfig(allow_unverified_webhooks=True)` or `TELEGRAM_ALLOW_UNVERIFIED_WEBHOOKS=true` (only the exact string `"true"` counts; an explicit `allow_unverified_webhooks=False` wins over the env var; a non-`bool` value such as `"false"` raises `ValidationError` rather than being treated as truthy). Polling mode needs neither.
@@ -65,7 +76,7 @@ Sync wave from `chat@4.31.0` to `chat@4.41.1` (tracking #184). `UPSTREAM_PARITY`
   - Tests: the mock pool no longer reclaims expired rows under `DO NOTHING`, which is not what real PostgreSQL does. The old `test_succeeds_after_expired_key` passed against the mock while production failed. Expiry tests advance an injectable mock clock instead of sleeping. An opt-in live suite (ported from upstream `postgres.integration.test.ts`) runs when `POSTGRES_TEST_URL` is set.
 - **Cards & modals: `Chart`, `Table` options, button tooltips, `Card` width, `DateInput` / `NumberInput`, `dispatch_action`** (#202). Ports the core slice of upstream `4717a384` (chat@4.34.0), `0153a39f` (chat@4.36.0), `4a0b5c0c` (chat@4.40.0), `84219537` and `ad904325` (chat@4.41.0). Additive only: with the new options unset, existing card and modal output is unchanged.
   - `Chart(title=, chart=)` (alias `chart`) builds a pie, bar, area or line chart element; new `ChartSegment`, `ChartDataPoint`, `ChartSeries`, `PieChartDefinition`, `SeriesChartDefinition` (`x_label` / `y_label`), `ChartDefinition` and `ChartElement` types. `chart_element_to_fallback_text()` renders the title plus the data as an ASCII table, with values formatted as JS `String()` does (`45.0` → `45`). Card fallback text now includes charts, so adapters that fall back to it (Slack until #212, Teams, Google Chat, Discord, GitHub, Linear, Twilio) post them as text; Messenger and WhatsApp drop chart children, as upstream does.
-  - `Table()` gains `caption`, `page_size`, `widths`, `vertical_align` (`TableVerticalAlignment`), `grid_lines` and `grid_style` (`TableGridStyle`). `Button()` / `LinkButton()` gain `tooltip`; `Card()` gains `width` (`CardWidth`: `"default"` / `"full"`). Slack renders `caption` / `page_size` in #212; Teams renders `widths`, `vertical_align`, `grid_lines`, `grid_style`, `tooltip` and `width` in #220. Other adapters ignore them. Until #194, a `callback_url` button loses its `tooltip` when the URL is swapped for a token (no adapter renders `tooltip` before #220).
+  - `Table()` gains `caption`, `page_size`, `widths`, `vertical_align` (`TableVerticalAlignment`), `grid_lines` and `grid_style` (`TableGridStyle`). `Button()` / `LinkButton()` gain `tooltip`; `Card()` gains `width` (`CardWidth`: `"default"` / `"full"`). Slack renders `caption` / `page_size` in #212; Teams renders `widths`, `vertical_align`, `grid_lines`, `grid_style`, `tooltip` and `width` in #220. Other adapters ignore them. A `callback_url` button keeps its `tooltip` when the URL is swapped for a token (#194).
   - New modal children `DateInput()` / `NumberInput()` (aliases `date_input` / `number_input`), accepted by `filter_modal_children`. `Select()` / `RadioSelect()` gain `dispatch_action`. Falsy values (`initial_value=0`, `min=0`, `decimal=False`, `dispatch_action=False`, `grid_lines=False`) are kept. Until #212, a Slack modal containing `DateInput` / `NumberInput` raises `ValueError`.
   - JSX conversions of these props are not ported (no JSX runtime); see `docs/UPSTREAM_SYNC.md`.
 - **GitHub: `GITHUB_BOT_USER_ID` env var and bot id learned from the first posted comment** (#233; port of the generic part of upstream `6750d59e`, vercel/chat#650, chat@4.33.0). The GitHub adapter spots its own comments by comparing `sender.id` with the bot user id. That id used to come only from `config["bot_user_id"]` or from best-effort auto-detection (`GET /user`, then `GET /app`). When detection failed, `is_me` never matched and the bot could reply to its own comments in a loop.
diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md
index 5428c905..78bf6030 100644
--- a/docs/UPSTREAM_SYNC.md
+++ b/docs/UPSTREAM_SYNC.md
@@ -365,6 +365,96 @@ Regression coverage: `tests/test_twilio_adapter.py::TestThreadIds`
 (`test_uses_the_full_dm_thread_id_as_its_channel_id`,
 `test_isolates_concurrent_recipients_with_thread_scoped_locks`).
 
+### Callback-URL tokens (chat@4.40, #194)
+
+Parity, with two Python-only divergences recorded in the non-parity table:
+*Callback-token lease fence* and *Channel edit callback scope*. The
+callback-token part of upstream `b7c9316b`
+(vercel/chat#875, chat@4.40.0) plus the button copy from `4a0b5c0c`
+(vercel/chat#895):
+
+- **Stored record.** `chat:callback:<token>` now holds
+  `{"actionId", "url", "originalValue"?, "scope": {"id", "type"}}`. The keys
+  stay camelCase so either SDK can resolve the other's tokens.
+  `originalValue` is omitted when the button has no value. The TTL is 7 days
+  (was 30).
+- **Scope.** Thread posts, schedules and edits bind to
+  `{thread.id, "thread"}`. Channel posts and schedules bind to
+  `{channel.id, "channel"}`. A channel `SentMessage.edit` also binds to
+  `{channel.id, "channel"}` (a Python-only divergence; upstream binds it to
+  the reported thread id). The `post_ephemeral` DM fallback mints tokens only
+  after `open_dm`, bound to
+  `{adapter.channel_id_from_thread_id(dm_thread_id), "channel"}`. When neither
+  the native path nor the DM fallback runs, no token is minted.
+- **Resolve.** `resolve_callback_url(token, state, context)` acquires
+  `acquire_lock("chat:callback:<token>", 10_000)` and returns `None` if the
+  lock is taken. Otherwise it validates the record, requires `actionId` and
+  the scope id to match the click's `CallbackContext`, deletes the record,
+  checks with `extend_lock` that the lease never lapsed (Python-only), and
+  releases the lock in `finally`. The scope id is `channel_id` for a channel
+  scope and `thread_id` for a thread scope. The action's POST body carries the
+  stored `actionId`.
+- **Legacy records are rejected.** A bare URL string (the "legacy string
+  format" the old resolver accepted), a pre-upgrade `{url, originalValue}`
+  record, or any other object without `actionId` / `scope` no longer resolves. Such a click
+  dispatches the raw `__cb:` value and nothing is POSTed, as upstream.
+  Validation uses `isinstance` checks, not truthiness, so an empty-string
+  `actionId` still passes, and a present `originalValue` must be a `str`.
+- **Dropped Python-only fallback.** The resolver used to fall back to a
+  snake_case `original_value` key. No Python release wrote that key, and
+  upstream never read it, so it is gone. Only `originalValue` is read.
+- **Button copy.** The token swap copies every button key except
+  `callback_url` (it used to copy a whitelist), so fields such as `tooltip`
+  (#202) survive.
+
+Channel-scoped tokens resolve only when `adapter.channel_id_from_thread_id`
+of the clicked message's thread equals the `ChannelImpl.id` that minted the
+token. In every adapter, `thread.channel` derives its id with that same
+function (`derive_channel_id`). For a `chat.channel(id)` handle the ids match
+when `id` is canonical for the adapter (Slack `slack:C123`, Discord
+`discord:{guild}:{channel}`, Google Chat `gchat:spaces/X`, GitHub
+`github:owner/repo`, Linear `linear:{issueId}`, Telegram `telegram:{chatId}`,
+Teams: the thread id re-encoded with `;messageid=…` stripped from the
+conversation id; WhatsApp, Messenger and
+Twilio (#235): the thread id itself).
+
+Thread-scoped tokens resolve only when the adapter's click `thread_id`
+equals the thread id the card was posted or edited under. One known mismatch
+remains, and it is upstream behavior at chat@4.41.1, not a divergence: the
+click still runs `on_action` handlers with the raw `__cb:…` value, but
+nothing is POSTed.
+
+- **Google Chat cards posted to a DM thread** (`gchat:spaces/X:dm`) by
+  `thread.post`. `_handle_card_click` encodes the clicked message's thread
+  name without the `:dm` suffix, as upstream `handleCardClick` does. The
+  `post_ephemeral` DM fallback is unaffected, because it binds to the DM
+  *channel*, which both ids share.
+
+The channel-edit divergence exists because the thread id reported for a
+channel post often never equals a click's thread id:
+- Teams and Google Chat `post_channel_message` return the channel id as the
+  thread id (upstream too);
+- Slack `post_channel_message` returns the synthetic `slack:C…:` until #209
+  ports upstream `92530dd3` (vercel/chat#720), while a click reports
+  `slack:C…:<message_ts>`. After #209 a Slack *DM* post reports
+  `slack:D…:<ts>`, but a DM click reports `slack:D…:` (Python's DM
+  `_handle_block_actions` divergence), so a thread scope would still miss;
+- chained edits (`sent = await sent.edit(...)` twice), whose returned
+  `SentMessage` drops the thread-id override until #195 ports `16ea171e`.
+
+Binding to the channel resolves all of these, independent of #209's merge
+order, because every click on the message derives the same channel id.
+
+**Breaking for `callback_url` users:** tokens minted before the upgrade stop
+resolving, a repeat click no longer POSTs, tokens expire after 7 days, and the
+POST's `actionId` is the minted button's id. The record is deleted before the
+POST, so a failed POST is not retried, as upstream. Regression coverage:
+`tests/test_callback_url.py` (including the Python-specific
+`TestResolveCallbackUrlValidation` / `TestResolveCallbackUrlLocking`),
+`tests/test_chat_faithful.py::TestActionsCallbackTokenBinding`, and the
+`should bind fallback DM callbacks to the DM channel` ports in
+`tests/test_thread_faithful.py` and `tests/test_channel_faithful.py`.
+
 ### WhatsApp business-scoped user IDs (chat@4.37–4.39, #236)
 
 Parity apart from malformed-payload hardening (below) and one divergence:
@@ -563,12 +653,10 @@ separately (Slack #212, Teams #220); other adapters ignore the new fields.
   `SlackBlockError` for a `chart` child; the Slack adapter posts a chart as a
   mrkdwn section holding the fallback text; and `dispatch_action`, `caption`
   and `page_size` are ignored.
-- **Interim callback-URL gap (until #194).** When a `Button(callback_url=…)`
-  is posted, `callback_url.py` swaps the URL for a token and rebuilds the
-  button from a fixed key list, so its `tooltip` is dropped. Upstream
-  `4a0b5c0c` changed that copy to keep every field except `callbackUrl`; that
-  half of the commit is owned by #194 (PR #256). No adapter renders `tooltip`
-  until #220, so nothing visible is lost today.
+- **Callback-URL button copy (#194).** Upstream `4a0b5c0c` also changed the
+  callback-token swap to keep every button field except `callbackUrl`. That
+  half of the commit is ported by #194, so a `Button(callback_url=…)` keeps
+  its `tooltip` (see *Callback-URL tokens* above).
 - **Not ported (JSX):** `fromReactElement` / `fromReactModalElement` handling of
   `Chart`, `DateInput`, `NumberInput`, `tooltip`, `width` and `dispatchAction`,
   and `929878b5` (chat@4.39.0, link-button ids in JSX). See the jsx-runtime row
@@ -983,6 +1071,8 @@ stay explicit instead of being rediscovered in code review.
 | Area | Python behavior | TS behavior | Rationale |
 |------|----------------|-------------|-----------|
 | JSX Card/Modal elements | Not supported; tests skipped | `Card()` returns JSX element | Python has no JSX runtime |
+| Channel edit callback scope (4.41 wave, #194) | A channel `SentMessage.edit` binds new callback tokens to `{channel.id, "channel"}`, the scope the original `channel.post` used | `createSentMessage(...).edit` binds to `{threadId, "thread"}`, where `threadId` is the id the adapter reported for the post (or the channel id) | The reported thread id often never equals a click's thread id: Teams and Google Chat report the channel id (upstream as well; Teams clicks carry `;messageid=`, Google Chat clicks carry the thread name), Python's Slack reports the synthetic `slack:C…:` until #209 while clicks carry the message ts, a Slack DM click carries no ts even after #209 makes the post report one, and a chained edit drops the override until #195 ports `16ea171e`. Upstream's edited buttons never POST in those cases. Every click on the message derives the channel id, and the channel scope is no broader than the original post's. Regression tests: `tests/test_channel_faithful.py::TestCallbackUrlProcessing::test_edited_slack_channel_card_resolves_for_the_real_click` (real Slack id functions and `_handle_block_actions`, channel and DM, before and after #209), `::test_edited_teams_channel_card_resolves_for_a_click_in_that_channel` (real Teams id functions) and `::test_chained_edit_keeps_callback_tokens_resolvable`. To be filed as an upstream issue against vercel/chat (Teams and Google Chat edited channel cards never POST); upstream Slack is unaffected, since its `postChannelMessage` and DM clicks both carry the message ts. |
+| Callback-token lease fence (4.41 wave, #194) | After deleting a matched record, `resolve_callback_url` calls `extend_lock(lock, CALLBACK_LOCK_TTL_MS)`. If that fails, the 10 s lease lapsed mid-consume, and the call returns `None` instead of the record (fail closed: no POST, raw `__cb:` value to handlers) | `resolveCallbackUrl` returns the record after `delete` regardless of lease state, so if a `get`/`delete` stalls past the lease, a second click that takes the expired lock also resolves it and both POST | Keeps the single-use contract under state-backend stalls. `extend_lock` checks token ownership and expiry in every backend (Memory, Redis script, Postgres `WHERE token = $4 AND expires_at > now()`), and `Chat` already relies on it for lock heartbeats. The cost is one extra state call per resolved click. A stalled consume that loses its lease also burns the token without a POST. To be filed as an upstream issue against vercel/chat (a stalled `get`/`delete` past the 10 s lease lets a second click double-POST). Regression test: `tests/test_callback_url.py::TestResolveCallbackUrlLocking::test_lost_lease_fails_closed_instead_of_double_consuming`. |
 | Markdown parser | Subset of CommonMark (no setext headings, indented code, HTML, escaped chars, backtick spans >1) | Full CommonMark via remark | See [DECISIONS.md](DECISIONS.md#why-hand-rolled-markdown-parser) |
 | `_remend` streaming repair | Parity-based emphasis closing | `remend` npm package | Simplified; handles common cases |
 | `walkAst` | Deep-copies the tree (immutable) | Mutates the tree in place | Python convention; safer |
diff --git a/scripts/fidelity_target.json b/scripts/fidelity_target.json
index bcfe8e3f..2db5a449 100644
--- a/scripts/fidelity_target.json
+++ b/scripts/fidelity_target.json
@@ -5,10 +5,10 @@
   "totals": {
     "ts_tests": 1036,
     "each_templates": 16,
-    "matched_exact": 699,
-    "matched_fuzzy": 81,
-    "missing": 256,
-    "extra": 376
+    "matched_exact": 706,
+    "matched_fuzzy": 80,
+    "missing": 250,
+    "extra": 395
   },
   "absent_ts_files": [],
   "files": {
@@ -21,7 +21,7 @@
       "matched_exact": 134,
       "matched_fuzzy": 12,
       "missing_count": 31,
-      "extra_count": 13,
+      "extra_count": 16,
       "missing": [
         ["Chat", "should optionally propagate handler errors through waitUntil"],
         ["Chat", "aborts an active thread signal from another Chat instance"],
@@ -76,10 +76,10 @@
       "python_exists": true,
       "ts_tests": 159,
       "each_templates": 1,
-      "matched_exact": 133,
+      "matched_exact": 134,
       "matched_fuzzy": 8,
-      "missing_count": 18,
-      "extra_count": 28,
+      "missing_count": 17,
+      "extra_count": 29,
       "missing": [
         ["post with Plan", "should auto-complete all in_progress tasks when switching back to default addTask"],
         ["post with Plan", "should target the most recent in_progress task when updating without id"],
@@ -97,8 +97,7 @@
         ["reply()", "converts JSX cards before delegating"],
         ["reply()", "buffers streams into one markdown reply"],
         ["reply()", "resolves a message id against messages the thread knows"],
-        ["reply()", "leaves replyTo undefined for an unknown message id"],
-        ["callbackUrl processing", "should bind fallback DM callbacks to the DM channel"]
+        ["reply()", "leaves replyTo undefined for an unknown message id"]
       ],
       "fuzzy_matches": [
         ["should pass stream options from Slack current message context via $label", "test_should_pass_stream_options_from_current_message_context"],
@@ -117,14 +116,13 @@
       "python_exists": true,
       "ts_tests": 67,
       "each_templates": 0,
-      "matched_exact": 65,
-      "matched_fuzzy": 2,
+      "matched_exact": 66,
+      "matched_fuzzy": 1,
       "missing_count": 0,
-      "extra_count": 4,
+      "extra_count": 9,
       "missing": [],
       "fuzzy_matches": [
-        ["should resolve a JSX Card and its children before scheduling", "test_should_convert_jsx_card_elements_to_cardelement"],
-        ["should bind fallback DM callbacks to the DM channel", "test_should_leave_channel_unchanged_when_chat_rebind_lookup_fails"]
+        ["should resolve a JSX Card and its children before scheduling", "test_should_convert_jsx_card_elements_to_cardelement"]
       ]
     },
     "packages/chat/src/markdown.test.ts": {
@@ -290,17 +288,11 @@
       "python_exists": true,
       "ts_tests": 21,
       "each_templates": 0,
-      "matched_exact": 16,
+      "matched_exact": 21,
       "matched_fuzzy": 0,
-      "missing_count": 5,
-      "extra_count": 1,
-      "missing": [
-        ["processCardCallbackUrls", "keeps every other button field when replacing the callback URL"],
-        ["resolveCallbackUrl", "rejects legacy unbound callback records"],
-        ["resolveCallbackUrl", "allows a callback token to be consumed only once"],
-        ["resolveCallbackUrl", "rejects callback tokens outside their action and thread"],
-        ["resolveCallbackUrl", "rejects callback tokens outside their channel"]
-      ],
+      "missing_count": 0,
+      "extra_count": 11,
+      "missing": [],
       "fuzzy_matches": []
     },
     "packages/chat/src/thread-history.test.ts": {
diff --git a/src/chat_sdk/callback_url.py b/src/chat_sdk/callback_url.py
index c573928d..a6f5de1e 100644
--- a/src/chat_sdk/callback_url.py
+++ b/src/chat_sdk/callback_url.py
@@ -1,6 +1,6 @@
 """Callback URL handling for buttons and modals.
 
-Python port of callback-url.ts (vercel/chat#454).
+Python port of callback-url.ts (vercel/chat#454, vercel/chat#875).
 
 When a button (or modal) carries a ``callback_url``, the SDK stores the URL
 in the state adapter under a short random token at post time and rewrites
@@ -9,14 +9,20 @@
 the original value for handlers, and POSTs the action payload to the stored
 URL. Modal callback URLs are stored in the modal context and POSTed on
 submit.
+
+Since chat@4.40.0 (vercel/chat#875) a button token is bound to the button
+that minted it (``actionId``) and to the conversation it was posted in
+(``scope``: a thread or a channel). A click resolves the token only when
+both match, and a resolved token is deleted, so each token is consumed at
+most once.
 """
 
 from __future__ import annotations
 
 import json
 import secrets
-from dataclasses import dataclass
-from typing import Any
+from dataclasses import dataclass, field
+from typing import Any, Literal
 
 from chat_sdk.cards import ActionsElement, ButtonElement, CardChild, CardElement
 from chat_sdk.errors import ChatError
@@ -24,7 +30,8 @@
 
 CALLBACK_TOKEN_PREFIX = "__cb:"
 CALLBACK_CACHE_KEY_PREFIX = "chat:callback:"
-CALLBACK_TTL_MS = 30 * 24 * 60 * 60 * 1000  # 30 days
+CALLBACK_TTL_MS = 7 * 24 * 60 * 60 * 1000  # 7 days
+CALLBACK_LOCK_TTL_MS = 10_000
 
 
 # ---------------------------------------------------------------------------
@@ -39,12 +46,38 @@ class DecodedCallbackValue:
     callback_token: str | None
 
 
+@dataclass(frozen=True)
+class CallbackScope:
+    """The conversation a callback token is bound to.
+
+    Stored as ``{"id", "type"}``. A ``"thread"`` scope matches the clicked
+    message's thread id; a ``"channel"`` scope matches the channel id the
+    adapter derives from it (``channel_id_from_thread_id``).
+    """
+
+    id: str
+    type: Literal["channel", "thread"]
+
+
+@dataclass(frozen=True)
+class CallbackContext:
+    """Where a button click happened; matched against a stored callback."""
+
+    action_id: str
+    channel_id: str | None = None
+    thread_id: str | None = None
+
+
 @dataclass(frozen=True)
 class ResolvedCallback:
-    """A stored callback resolved from the state adapter."""
+    """A stored callback, resolved and consumed from the state adapter."""
 
     url: str
     original_value: str | None = None
+    # Keyword-only so positional ``ResolvedCallback(url, original_value)``
+    # keeps binding the same fields as before these were added.
+    action_id: str = field(kw_only=True)
+    scope: CallbackScope = field(kw_only=True)
 
 
 @dataclass(frozen=True)
@@ -87,6 +120,7 @@ def _generate_token() -> str:
 async def _process_actions_element(
     actions: ActionsElement,
     state_adapter: StateAdapter,
+    scope: CallbackScope,
 ) -> ActionsElement:
     children: list[Any] = []
     for el in actions.get("children", []):
@@ -95,25 +129,21 @@ async def _process_actions_element(
             continue
 
         token = _generate_token()
-        # Stored shape matches the TS SDK (`{url, originalValue}`) so state
-        # written by either SDK resolves in both. `originalValue` is omitted
-        # (not None) when the button has no value — hazard #7.
-        stored: dict[str, Any] = {"url": el["callback_url"]}
+        # Stored shape matches the TS SDK (`{actionId, url, originalValue?,
+        # scope: {id, type}}`) so state written by either SDK resolves in
+        # both. `originalValue` is omitted (not None) when the button has no
+        # value — hazard #7.
+        stored: dict[str, Any] = {"actionId": el["id"], "url": el["callback_url"]}
         original_value = el.get("value")
         if original_value is not None:
             stored["originalValue"] = original_value
+        stored["scope"] = {"id": scope.id, "type": scope.type}
         await state_adapter.set(f"{CALLBACK_CACHE_KEY_PREFIX}{token}", stored, CALLBACK_TTL_MS)
 
-        # Rebuild the button without `callback_url` (mirrors upstream's
-        # explicit-key copy; keys absent on the source stay absent).
-        processed: ButtonElement = {"type": "button", "id": el["id"], "label": el["label"]}
-        if "style" in el:
-            processed["style"] = el["style"]
-        if "disabled" in el:
-            processed["disabled"] = el["disabled"]
+        # Keep every other button field so new ones (like tooltip) are not
+        # silently dropped; only the callback URL is replaced by the token.
+        processed: ButtonElement = {k: v for k, v in el.items() if k != "callback_url"}  # type: ignore[assignment]
         processed["value"] = encode_callback_value(token)
-        if "action_type" in el:
-            processed["action_type"] = el["action_type"]
         children.append(processed)
     return {"type": "actions", "children": children}
 
@@ -134,13 +164,14 @@ def _has_callback_buttons(children: list[CardChild]) -> bool:
 async def _process_children(
     children: list[CardChild],
     state_adapter: StateAdapter,
+    scope: CallbackScope,
 ) -> list[CardChild]:
     result: list[CardChild] = []
     for child in children:
         if isinstance(child, dict) and child.get("type") == "actions":
-            result.append(await _process_actions_element(child, state_adapter))  # type: ignore[arg-type]
+            result.append(await _process_actions_element(child, state_adapter, scope))  # type: ignore[arg-type]
         elif isinstance(child, dict) and child.get("type") == "section" and "children" in child:
-            result.append({**child, "children": await _process_children(child["children"], state_adapter)})  # type: ignore[misc]
+            result.append({**child, "children": await _process_children(child["children"], state_adapter, scope)})  # type: ignore[misc]
         else:
             result.append(child)
     return result
@@ -149,16 +180,18 @@ async def _process_children(
 async def process_card_callback_urls(
     card: CardElement,
     state_adapter: StateAdapter,
+    scope: CallbackScope,
 ) -> CardElement:
     """Replace ``callback_url`` buttons with encoded token values.
 
+    Each minted token is bound to its button's ``id`` and to ``scope``.
     Returns the *same* card object when no button carries a callback URL;
     otherwise returns a new card (the original is never mutated).
     """
     if not _has_callback_buttons(card.get("children", [])):
         return card
 
-    return {**card, "children": await _process_children(card.get("children", []), state_adapter)}
+    return {**card, "children": await _process_children(card.get("children", []), state_adapter, scope)}
 
 
 # ---------------------------------------------------------------------------
@@ -166,19 +199,87 @@ async def process_card_callback_urls(
 # ---------------------------------------------------------------------------
 
 
+def _parse_stored_callback(stored: Any) -> ResolvedCallback | None:
+    """Validate a stored record strictly; ``None`` for any other shape.
+
+    Mirrors upstream's ``typeof`` checks (never truthiness): a legacy
+    bare-string record, or one without ``actionId`` / ``scope``, is
+    rejected. An empty-string ``actionId`` is still a string and passes.
+    """
+    if not isinstance(stored, dict):
+        return None
+    action_id = stored.get("actionId")
+    url = stored.get("url")
+    if not isinstance(action_id, str) or not isinstance(url, str):
+        return None
+    original_value = stored.get("originalValue")
+    if "originalValue" in stored and not isinstance(original_value, str):
+        return None
+    scope = stored.get("scope")
+    if not isinstance(scope, dict):
+        return None
+    scope_id = scope.get("id")
+    if not isinstance(scope_id, str):
+        return None
+    raw_type = scope.get("type")
+    scope_type: Literal["channel", "thread"]
+    if raw_type == "channel":
+        scope_type = "channel"
+    elif raw_type == "thread":
+        scope_type = "thread"
+    else:
+        return None
+    return ResolvedCallback(
+        url=url,
+        original_value=original_value,
+        action_id=action_id,
+        scope=CallbackScope(id=scope_id, type=scope_type),
+    )
+
+
 async def resolve_callback_url(
     token: str,
     state_adapter: StateAdapter,
+    context: CallbackContext | None = None,
 ) -> ResolvedCallback | None:
-    """Look up a stored callback by token. Returns ``None`` when unknown."""
-    stored = await state_adapter.get(f"{CALLBACK_CACHE_KEY_PREFIX}{token}")
-    if not stored:
+    """Resolve and consume a stored button callback.
+
+    Returns ``None`` when the token is unknown or malformed, when a
+    concurrent click holds the token's lock, or when the record does not
+    match ``context`` (the clicked ``action_id`` and, depending on the
+    stored scope, ``channel_id`` or ``thread_id``). A ``None`` context
+    never matches. A matching record is deleted before returning, so each
+    token resolves at most once; a mismatch leaves the record in place. If
+    the lock lease lapsed while the record was being consumed, the result is
+    ``None`` (fail closed) so a concurrent click cannot also POST.
+    """
+    key = f"{CALLBACK_CACHE_KEY_PREFIX}{token}"
+    # Lock keys live in their own namespace in every state backend, so
+    # locking the value key does not touch the stored record.
+    lock = await state_adapter.acquire_lock(key, CALLBACK_LOCK_TTL_MS)
+    if lock is None:
         return None
-    if isinstance(stored, str):
-        # Legacy format: the URL was stored as a bare string.
-        return ResolvedCallback(url=stored)
-    original_value = stored["originalValue"] if "originalValue" in stored else stored.get("original_value")
-    return ResolvedCallback(url=stored.get("url"), original_value=original_value)
+
+    try:
+        resolved = _parse_stored_callback(await state_adapter.get(key))
+        if resolved is None or context is None:
+            return None
+
+        scope_id = context.channel_id if resolved.scope.type == "channel" else context.thread_id
+        if resolved.action_id != context.action_id or resolved.scope.id != scope_id:
+            return None
+
+        await state_adapter.delete(key)
+        # Python-specific fence (divergence from upstream — see
+        # docs/UPSTREAM_SYNC.md): if a stalled state call outlived the lock
+        # lease, another click may have taken the lock and consumed the same
+        # record. `extend_lock` succeeds only while our lease never lapsed,
+        # so a lost lease fails closed instead of POSTing a second time.
+        if not await state_adapter.extend_lock(lock, CALLBACK_LOCK_TTL_MS):
+            return None
+        return resolved
+    finally:
+        await state_adapter.release_lock(lock)
 
 
 async def _fetch(url: str, *, method: str, headers: dict[str, str], body: str) -> tuple[int, str]:
diff --git a/src/chat_sdk/channel.py b/src/chat_sdk/channel.py
index 3bf57130..ac5c5b59 100644
--- a/src/chat_sdk/channel.py
+++ b/src/chat_sdk/channel.py
@@ -12,7 +12,7 @@
 from datetime import datetime, timezone
 from typing import Any
 
-from chat_sdk.callback_url import process_card_callback_urls
+from chat_sdk.callback_url import CallbackScope, process_card_callback_urls
 from chat_sdk.errors import ChatNotImplementedError
 from chat_sdk.plan import is_postable_object, post_postable_object
 from chat_sdk.thread import (
@@ -340,9 +340,10 @@ async def post_ephemeral(
         """Post an ephemeral message visible only to the specified user."""
         user_id = user if isinstance(user, str) else user.user_id
 
-        message = await self._process_callback_urls(message)  # type: ignore[assignment]
-
+        # Callback tokens are minted per delivery path so each is bound to
+        # the conversation the card actually lands in (vercel/chat#875).
         if hasattr(self.adapter, "post_ephemeral") and self.adapter.post_ephemeral:  # type: ignore[union-attr]
+            message = await self._process_callback_urls(message)  # type: ignore[assignment]
             return await self.adapter.post_ephemeral(self._id, user_id, message)  # type: ignore[union-attr]
 
         if not options.fallback_to_dm:
@@ -350,6 +351,10 @@ async def post_ephemeral(
 
         if hasattr(self.adapter, "open_dm") and self.adapter.open_dm:  # type: ignore[union-attr]
             dm_thread_id: str = await self.adapter.open_dm(user_id)  # type: ignore[union-attr]
+            message = await self._process_callback_urls(  # type: ignore[assignment]
+                message,
+                CallbackScope(id=self.adapter.channel_id_from_thread_id(dm_thread_id), type="channel"),
+            )
             result: RawMessage = await self.adapter.post_message(dm_thread_id, message)
             return EphemeralMessage(
                 id=result.id,
@@ -379,20 +384,27 @@ async def schedule(
     async def _process_callback_urls(
         self,
         postable: str | AdapterPostableMessage,
+        scope: CallbackScope | None = None,
     ) -> str | AdapterPostableMessage:
-        """Encode ``callback_url`` buttons in outgoing cards (vercel/chat#454)."""
+        """Encode ``callback_url`` buttons in outgoing cards (vercel/chat#454).
+
+        Tokens are bound to ``scope``, which defaults to this channel.
+        """
         if isinstance(postable, str):
             return postable
 
+        if scope is None:
+            scope = CallbackScope(id=self._id, type="channel")
+
         if isinstance(postable, dict) and postable.get("type") == "card":
-            return await process_card_callback_urls(postable, self._state_adapter)
+            return await process_card_callback_urls(postable, self._state_adapter, scope)
 
         if (
             isinstance(postable, PostableCard)
             and isinstance(postable.card, dict)
             and postable.card.get("type") == "card"
         ):
-            processed = await process_card_callback_urls(postable.card, self._state_adapter)
+            processed = await process_card_callback_urls(postable.card, self._state_adapter, scope)
             if processed is not postable.card:
                 return replace(postable, card=processed)
 
@@ -547,8 +559,18 @@ def _create_sent_message(
 
         plain_text, formatted, attachments = _extract_message_content(postable)
 
+        # Upstream binds edited-card tokens to `{thread_id, "thread"}`. Python
+        # binds them to the channel, the scope the original `channel.post`
+        # used, because the thread id reported for a channel post often never
+        # equals a click's thread id: Teams and Google Chat report the channel
+        # id; Slack reports `slack:C…:` while a channel click carries the
+        # message ts and a DM click carries none; a chained edit drops the
+        # override. Every click on this message derives this channel id.
+        # Divergence from upstream — see docs/UPSTREAM_SYNC.md.
+        edit_scope = CallbackScope(id=channel_impl._id, type="channel")
+
         async def _edit(new_content: Any) -> SentMessage:
-            new_content = await channel_impl._process_callback_urls(new_content)
+            new_content = await channel_impl._process_callback_urls(new_content, edit_scope)
             await adapter.edit_message(thread_id, message_id, new_content)
             return channel_impl._create_sent_message(message_id, new_content)
 
diff --git a/src/chat_sdk/chat.py b/src/chat_sdk/chat.py
index 5e20de46..cdf18f1f 100644
--- a/src/chat_sdk/chat.py
+++ b/src/chat_sdk/chat.py
@@ -19,6 +19,7 @@
 from typing import Any
 
 from chat_sdk.callback_url import (
+    CallbackContext,
     decode_callback_value,
     post_to_callback_url,
     resolve_callback_url,
@@ -1434,12 +1435,24 @@ async def _handle_action_event(self, event: ActionEvent) -> None:
         # Decode a callback token (`__cb:<token>`) planted at post time by
         # process_card_callback_urls. When one resolves, handlers see the
         # button's original value and the action payload is POSTed to the
-        # stored callback URL concurrently with the handlers.
+        # stored callback URL concurrently with the handlers. A token only
+        # resolves for the button and conversation that minted it, and only
+        # once (vercel/chat#875); otherwise handlers see the raw value and
+        # nothing is POSTed.
         callback_token = decode_callback_value(event.value).callback_token
 
         resolved = None
         if callback_token:
-            resolved = await resolve_callback_url(callback_token, self._state_adapter)
+            channel_id = event.adapter.channel_id_from_thread_id(event.thread_id) if event.thread_id else None
+            resolved = await resolve_callback_url(
+                callback_token,
+                self._state_adapter,
+                CallbackContext(
+                    action_id=event.action_id,
+                    channel_id=channel_id,
+                    thread_id=event.thread_id,
+                ),
+            )
 
         action_value = resolved.original_value if resolved is not None else event.value
 
@@ -1448,7 +1461,8 @@ async def _handle_action_event(self, event: ActionEvent) -> None:
             callback_url = resolved.url
             # Wire payload: camelCase keys, optional keys omitted (not None)
             # to mirror upstream's JSON.stringify semantics — hazard #7.
-            payload: dict[str, Any] = {"type": "action", "actionId": event.action_id}
+            # `actionId` is the minted button's id, as stored with the token.
+            payload: dict[str, Any] = {"type": "action", "actionId": resolved.action_id}
             if resolved.original_value is not None:
                 payload["value"] = resolved.original_value
             payload["user"] = {"id": event.user.user_id, "name": event.user.user_name}
diff --git a/src/chat_sdk/thread.py b/src/chat_sdk/thread.py
index 182e7829..99ee85ae 100644
--- a/src/chat_sdk/thread.py
+++ b/src/chat_sdk/thread.py
@@ -14,7 +14,7 @@
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Protocol, cast, runtime_checkable
 
-from chat_sdk.callback_url import process_card_callback_urls
+from chat_sdk.callback_url import CallbackScope, process_card_callback_urls
 from chat_sdk.errors import ChatNotImplementedError
 from chat_sdk.logger import Logger
 from chat_sdk.plan import is_postable_object, post_postable_object
@@ -613,10 +613,11 @@ async def post_ephemeral(
         """
         user_id = user if isinstance(user, str) else user.user_id
 
-        message = await self._process_callback_urls(message)  # type: ignore[assignment]
-
+        # Callback tokens are minted per delivery path so each is bound to
+        # the conversation the card actually lands in (vercel/chat#875).
         # Try native ephemeral
         if hasattr(self.adapter, "post_ephemeral") and self.adapter.post_ephemeral:  # type: ignore[union-attr]
+            message = await self._process_callback_urls(message)  # type: ignore[assignment]
             return await self.adapter.post_ephemeral(self._id, user_id, message)  # type: ignore[union-attr]
 
         if not options.fallback_to_dm:
@@ -625,6 +626,10 @@ async def post_ephemeral(
         # Fallback: send via DM
         if hasattr(self.adapter, "open_dm") and self.adapter.open_dm:  # type: ignore[union-attr]
             dm_thread_id: str = await self.adapter.open_dm(user_id)  # type: ignore[union-attr]
+            message = await self._process_callback_urls(  # type: ignore[assignment]
+                message,
+                CallbackScope(id=self.adapter.channel_id_from_thread_id(dm_thread_id), type="channel"),
+            )
             result: RawMessage = await self.adapter.post_message(dm_thread_id, message)
             return EphemeralMessage(
                 id=result.id,
@@ -638,20 +643,27 @@ async def post_ephemeral(
     async def _process_callback_urls(
         self,
         postable: str | AdapterPostableMessage,
+        scope: CallbackScope | None = None,
     ) -> str | AdapterPostableMessage:
-        """Encode ``callback_url`` buttons in outgoing cards (vercel/chat#454)."""
+        """Encode ``callback_url`` buttons in outgoing cards (vercel/chat#454).
+
+        Tokens are bound to ``scope``, which defaults to this thread.
+        """
         if isinstance(postable, str):
             return postable
 
+        if scope is None:
+            scope = CallbackScope(id=self._id, type="thread")
+
         if isinstance(postable, dict) and postable.get("type") == "card":
-            return await process_card_callback_urls(postable, self._state_adapter)
+            return await process_card_callback_urls(postable, self._state_adapter, scope)
 
         if (
             isinstance(postable, PostableCard)
             and isinstance(postable.card, dict)
             and postable.card.get("type") == "card"
         ):
-            processed = await process_card_callback_urls(postable.card, self._state_adapter)
+            processed = await process_card_callback_urls(postable.card, self._state_adapter, scope)
             if processed is not postable.card:
                 return replace(postable, card=processed)
 
diff --git a/tests/integration/test_replay_callback_url.py b/tests/integration/test_replay_callback_url.py
index 2a0edc0b..711985ce 100644
--- a/tests/integration/test_replay_callback_url.py
+++ b/tests/integration/test_replay_callback_url.py
@@ -153,7 +153,12 @@ async def on_mention(thread, message, context):
                 # done at post time.
                 await ctx.state.set(
                     f"chat:callback:{CALLBACK_TOKEN}",
-                    {"url": CALLBACK_BUTTON_URL, "originalValue": "order-99"},
+                    {
+                        "actionId": "approve",
+                        "url": CALLBACK_BUTTON_URL,
+                        "originalValue": "order-99",
+                        "scope": {"id": "slack:C00FAKECHAN1", "type": "channel"},
+                    },
                 )
 
                 # Synthesize a block_actions payload with the SDK's encoded
diff --git a/tests/test_callback_url.py b/tests/test_callback_url.py
index 6a4a683c..69faada0 100644
--- a/tests/test_callback_url.py
+++ b/tests/test_callback_url.py
@@ -1,4 +1,4 @@
-"""Faithful translation of callback-url.test.ts (17 tests).
+"""Faithful translation of callback-url.test.ts (21 tests at chat@4.41.1).
 
 Each ``it("...")`` block from the TypeScript test suite is translated
 to a corresponding ``async def test_...`` method, preserving the same
@@ -7,17 +7,28 @@
 TS stubs the global ``fetch``; Python patches the ``_fetch`` seam in
 ``chat_sdk.callback_url`` (the lazy aiohttp wrapper).
 
+Python-specific coverage for the token lock / consume path lives in
+``TestResolveCallbackUrlLocking`` at the bottom of the resolve section.
+
 TS file: packages/chat/src/callback-url.test.ts
 """
 
 from __future__ import annotations
 
+import asyncio
 import copy
 import json
 import re
 from unittest.mock import AsyncMock, call, patch
 
+import pytest
+
 from chat_sdk.callback_url import (
+    CALLBACK_LOCK_TTL_MS,
+    CALLBACK_TTL_MS,
+    CallbackContext,
+    CallbackScope,
+    ResolvedCallback,
     decode_callback_value,
     encode_callback_value,
     post_to_callback_url,
@@ -25,6 +36,7 @@
     resolve_callback_url,
 )
 from chat_sdk.cards import Actions, Button, Card, CardText, Section
+from chat_sdk.state.memory import MemoryStateAdapter
 from chat_sdk.testing import MockStateAdapter, create_mock_state
 
 CALLBACK_TOKEN_PATTERN = re.compile(r"^__cb:[a-f0-9]{16}$")
@@ -71,6 +83,9 @@ def test_roundtrips_encodedecode(self):
 # ===========================================================================
 
 
+CHANNEL_SCOPE = CallbackScope(id="slack:C1", type="channel")
+
+
 class TestProcessCardCallbackUrls:
     """describe("processCardCallbackUrls")"""
 
@@ -88,7 +103,7 @@ async def test_returns_card_unchanged_when_no_buttons_have_callbackurl(self):
             ],
         )
 
-        result = await process_card_callback_urls(card, state)
+        result = await process_card_callback_urls(card, state, CHANNEL_SCOPE)
         assert result is card
 
     # it("encodes callbackUrl into button value and stores in state")
@@ -109,7 +124,7 @@ async def test_encodes_callbackurl_into_button_value_and_stores_in_state(self):
             ],
         )
 
-        result = await process_card_callback_urls(card, state)
+        result = await process_card_callback_urls(card, state, CallbackScope(id="slack:C1:1.1", type="thread"))
 
         actions = next(c for c in result["children"] if c["type"] == "actions")
         button = actions["children"][0]
@@ -120,9 +135,14 @@ async def test_encodes_callbackurl_into_button_value_and_stores_in_state(self):
         decoded = decode_callback_value(button["value"])
         assert decoded.callback_token is not None
 
-        resolved = await resolve_callback_url(decoded.callback_token, state)
+        resolved = await resolve_callback_url(
+            decoded.callback_token,
+            state,
+            CallbackContext(action_id="approve", thread_id="slack:C1:1.1"),
+        )
         assert resolved is not None
         assert resolved.url == "https://example.com/webhook/123"
+        assert resolved.scope == CallbackScope(id="slack:C1:1.1", type="thread")
 
     # it("stores original value in state alongside callback URL")
     async def test_stores_original_value_in_state_alongside_callback_url(self):
@@ -143,13 +163,17 @@ async def test_stores_original_value_in_state_alongside_callback_url(self):
             ],
         )
 
-        result = await process_card_callback_urls(card, state)
+        result = await process_card_callback_urls(card, state, CHANNEL_SCOPE)
         button = next(c for c in result["children"] if c["type"] == "actions")["children"][0]
 
         assert CALLBACK_TOKEN_PATTERN.match(button["value"])
 
         decoded = decode_callback_value(button["value"])
-        resolved = await resolve_callback_url(decoded.callback_token or "", state)
+        resolved = await resolve_callback_url(
+            decoded.callback_token or "",
+            state,
+            CallbackContext(action_id="btn", channel_id="slack:C1"),
+        )
         assert resolved is not None
         assert resolved.url == "https://hook.example.com"
         assert resolved.original_value == "item-99"
@@ -173,7 +197,7 @@ async def test_only_processes_buttons_with_callbackurl_leaves_others_untouched(s
             ],
         )
 
-        result = await process_card_callback_urls(card, state)
+        result = await process_card_callback_urls(card, state, CHANNEL_SCOPE)
         actions = next(c for c in result["children"] if c["type"] == "actions")
         normal_btn = actions["children"][0]
         callback_btn = actions["children"][1]
@@ -181,6 +205,36 @@ async def test_only_processes_buttons_with_callbackurl_leaves_others_untouched(s
         assert normal_btn["value"] == "keep"
         assert CALLBACK_PREFIX_PATTERN.match(callback_btn["value"])
 
+    # it("keeps every other button field when replacing the callback URL")
+    async def test_keeps_every_other_button_field_when_replacing_the_callback_url(self):
+        state = self._state()
+        button_in = Button(
+            id="approve",
+            label="Approve",
+            style="primary",
+            disabled=True,
+            action_type="modal",
+            tooltip="Approve the request",
+            callback_url="https://example.com/hook",
+        )
+        card = Card(title="Test", children=[Actions([button_in])])
+
+        result = await process_card_callback_urls(card, state, CHANNEL_SCOPE)
+        actions = next(c for c in result["children"] if c["type"] == "actions")
+        button = actions["children"][0]
+
+        assert {k: v for k, v in button.items() if k != "value"} == {
+            "type": "button",
+            "id": "approve",
+            "label": "Approve",
+            "style": "primary",
+            "disabled": True,
+            "action_type": "modal",
+            "tooltip": "Approve the request",
+        }
+        assert "callback_url" not in button
+        assert CALLBACK_PREFIX_PATTERN.match(button["value"])
+
     # it("processes buttons nested inside sections")
     async def test_processes_buttons_nested_inside_sections(self):
         state = self._state()
@@ -204,7 +258,7 @@ async def test_processes_buttons_nested_inside_sections(self):
             ],
         )
 
-        result = await process_card_callback_urls(card, state)
+        result = await process_card_callback_urls(card, state, CHANNEL_SCOPE)
         section = next(c for c in result["children"] if c["type"] == "section")
         actions = next(c for c in section["children"] if c["type"] == "actions")
         button = actions["children"][0]
@@ -214,7 +268,11 @@ async def test_processes_buttons_nested_inside_sections(self):
         assert "callback_url" not in button
 
         decoded = decode_callback_value(button["value"])
-        resolved = await resolve_callback_url(decoded.callback_token or "", state)
+        resolved = await resolve_callback_url(
+            decoded.callback_token or "",
+            state,
+            CallbackContext(action_id="nested-btn", channel_id="slack:C1"),
+        )
         assert resolved is not None
         assert resolved.url == "https://example.com/nested"
 
@@ -237,10 +295,55 @@ async def test_does_not_mutate_the_original_card(self):
         )
 
         original = copy.deepcopy(card)
-        await process_card_callback_urls(card, state)
+        await process_card_callback_urls(card, state, CHANNEL_SCOPE)
         assert card == original
 
 
+class TestProcessCardCallbackUrlsStoredRecord:
+    """Python-specific: the exact record written to state (cross-SDK shape)."""
+
+    async def test_stores_camelcase_record_with_scope_and_seven_day_ttl(self):
+        state = create_mock_state()
+        state.set = AsyncMock(wraps=state.set)  # type: ignore[method-assign]
+        card = Card(
+            children=[
+                Actions(
+                    [
+                        Button(id="with-value", label="A", value="v1", callback_url="https://example.com/a"),
+                        Button(id="no-value", label="B", callback_url="https://example.com/b"),
+                    ]
+                )
+            ]
+        )
+
+        result = await process_card_callback_urls(card, state, CallbackScope(id="slack:C9:9.9", type="thread"))
+
+        tokens = [decode_callback_value(b["value"]).callback_token for b in result["children"][0]["children"]]
+        assert state.set.await_args_list == [
+            call(
+                f"chat:callback:{tokens[0]}",
+                {
+                    "actionId": "with-value",
+                    "url": "https://example.com/a",
+                    "originalValue": "v1",
+                    "scope": {"id": "slack:C9:9.9", "type": "thread"},
+                },
+                CALLBACK_TTL_MS,
+            ),
+            # `originalValue` is omitted, never written as None.
+            call(
+                f"chat:callback:{tokens[1]}",
+                {
+                    "actionId": "no-value",
+                    "url": "https://example.com/b",
+                    "scope": {"id": "slack:C9:9.9", "type": "thread"},
+                },
+                CALLBACK_TTL_MS,
+            ),
+        ]
+        assert CALLBACK_TTL_MS == 7 * 24 * 60 * 60 * 1000
+
+
 # ===========================================================================
 # resolveCallbackUrl
 # ===========================================================================
@@ -260,22 +363,287 @@ async def test_resolves_stored_callback_with_url_and_original_value(self):
         state = create_mock_state()
         await state.set(
             "chat:callback:test-token",
-            {"url": "https://example.com/hook", "originalValue": "item-42"},
+            {
+                "actionId": "approve",
+                "url": "https://example.com/hook",
+                "originalValue": "item-42",
+                "scope": {"id": "slack:C1:1.1", "type": "thread"},
+            },
+        )
+        result = await resolve_callback_url(
+            "test-token",
+            state,
+            CallbackContext(action_id="approve", thread_id="slack:C1:1.1"),
         )
-        result = await resolve_callback_url("test-token", state)
         assert result is not None
         assert result.url == "https://example.com/hook"
         assert result.original_value == "item-42"
+        assert result.scope == CallbackScope(id="slack:C1:1.1", type="thread")
+        assert await state.get("chat:callback:test-token") is None
+
+    # it("rejects legacy unbound callback records")
+    async def test_rejects_legacy_unbound_callback_records(self):
+        state = create_mock_state()
+        await state.set("chat:callback:legacy-token", "https://example.com/hook")
+        result = await resolve_callback_url("legacy-token", state, CallbackContext(action_id="approve"))
+        assert result is None
 
-    # it("handles legacy string format")
+    # it("handles legacy string format") — chat@4.31.0 title, behavior now
+    # reversed. Kept so strict fidelity at the 4.31.0 pin stays green until
+    # the pin moves to 4.41.1 (#203), where upstream replaced it with the
+    # test above. Asserts what the rejection above does not: even a context
+    # that would match any record resolves nothing, and the legacy record is
+    # left for its TTL rather than deleted.
     async def test_handles_legacy_string_format(self):
         state = create_mock_state()
         await state.set("chat:callback:legacy-token", "https://example.com/hook")
-        result = await resolve_callback_url("legacy-token", state)
+        state.delete = AsyncMock(wraps=state.delete)  # type: ignore[method-assign]
+
+        result = await resolve_callback_url(
+            "legacy-token",
+            state,
+            CallbackContext(action_id="approve", channel_id="slack:C1", thread_id="slack:C1:1.1"),
+        )
+
+        assert result is None
+        state.delete.assert_not_awaited()
+        assert await state.get("chat:callback:legacy-token") == "https://example.com/hook"
+        assert state._locks == {}
+
+    # it("allows a callback token to be consumed only once")
+    async def test_allows_a_callback_token_to_be_consumed_only_once(self):
+        state = create_mock_state()
+        await state.set(
+            "chat:callback:single-use",
+            {
+                "actionId": "approve",
+                "scope": {"id": "slack:C1", "type": "channel"},
+                "url": "https://example.com/hook",
+            },
+        )
+        context = CallbackContext(action_id="approve", channel_id="slack:C1")
+
+        first = await resolve_callback_url("single-use", state, context)
+        assert first is not None
+        assert first.url == "https://example.com/hook"
+        assert await resolve_callback_url("single-use", state, context) is None
+
+    # it("rejects callback tokens outside their action and thread")
+    async def test_rejects_callback_tokens_outside_their_action_and_thread(self):
+        state = create_mock_state()
+        await state.set(
+            "chat:callback:bound-token",
+            {
+                "actionId": "approve",
+                "scope": {"id": "slack:C1:1.1", "type": "thread"},
+                "url": "https://example.com/hook",
+            },
+        )
+
+        assert (
+            await resolve_callback_url(
+                "bound-token",
+                state,
+                CallbackContext(action_id="deny", thread_id="slack:C1:1.1"),
+            )
+            is None
+        )
+        assert (
+            await resolve_callback_url(
+                "bound-token",
+                state,
+                CallbackContext(action_id="approve", thread_id="slack:C1:2.2"),
+            )
+            is None
+        )
+        assert await state.get("chat:callback:bound-token") is not None
+
+    # it("rejects callback tokens outside their channel")
+    async def test_rejects_callback_tokens_outside_their_channel(self):
+        state = create_mock_state()
+        await state.set(
+            "chat:callback:channel-token",
+            {
+                "actionId": "approve",
+                "scope": {"id": "slack:C1", "type": "channel"},
+                "url": "https://example.com/hook",
+            },
+        )
+
+        assert (
+            await resolve_callback_url(
+                "channel-token",
+                state,
+                CallbackContext(action_id="approve", channel_id="slack:C2", thread_id="slack:C2:2.2"),
+            )
+            is None
+        )
+        resolved = await resolve_callback_url(
+            "channel-token",
+            state,
+            CallbackContext(action_id="approve", channel_id="slack:C1", thread_id="slack:C1:2.2"),
+        )
+        assert resolved is not None
+        assert resolved.url == "https://example.com/hook"
+
+
+# Python-specific: strict record validation (isinstance, never truthiness).
+_VALID_RECORD = {
+    "actionId": "approve",
+    "url": "https://example.com/hook",
+    "scope": {"id": "slack:C1", "type": "channel"},
+}
+
+
+class TestResolveCallbackUrlValidation:
+    """Python-specific: record shapes upstream's ``typeof`` checks reject."""
+
+    @pytest.mark.parametrize(
+        "record",
+        [
+            pytest.param({**_VALID_RECORD, "actionId": None}, id="actionId-none"),
+            pytest.param({k: v for k, v in _VALID_RECORD.items() if k != "actionId"}, id="actionId-missing"),
+            pytest.param({**_VALID_RECORD, "url": 42}, id="url-not-str"),
+            pytest.param({**_VALID_RECORD, "originalValue": None}, id="originalValue-none"),
+            pytest.param({**_VALID_RECORD, "originalValue": 7}, id="originalValue-not-str"),
+            pytest.param({k: v for k, v in _VALID_RECORD.items() if k != "scope"}, id="scope-missing"),
+            pytest.param({**_VALID_RECORD, "scope": "slack:C1"}, id="scope-not-dict"),
+            pytest.param({**_VALID_RECORD, "scope": {"id": 1, "type": "channel"}}, id="scope-id-not-str"),
+            pytest.param({**_VALID_RECORD, "scope": {"id": "slack:C1", "type": "dm"}}, id="scope-type-unknown"),
+            pytest.param(["https://example.com/hook"], id="list"),
+        ],
+    )
+    async def test_rejects_malformed_records_without_deleting(self, record):
+        state = create_mock_state()
+        state.cache["chat:callback:tok"] = record
+
+        result = await resolve_callback_url("tok", state, CallbackContext(action_id="approve", channel_id="slack:C1"))
+
+        assert result is None
+        assert state.cache["chat:callback:tok"] == record
+
+    async def test_accepts_empty_string_action_id_like_upstream(self):
+        state = create_mock_state()
+        state.cache["chat:callback:tok"] = {**_VALID_RECORD, "actionId": ""}
+
+        result = await resolve_callback_url("tok", state, CallbackContext(action_id="", channel_id="slack:C1"))
+
+        assert result is not None
+        assert result.action_id == ""
+        assert result.original_value is None
+
+    async def test_ignores_snake_case_original_value_key(self):
+        # The pre-4.41 Python-only `original_value` fallback read is gone:
+        # only the cross-SDK camelCase `originalValue` key is honored.
+        state = create_mock_state()
+        state.cache["chat:callback:tok"] = {**_VALID_RECORD, "original_value": "legacy"}
+
+        result = await resolve_callback_url("tok", state, CallbackContext(action_id="approve", channel_id="slack:C1"))
+
         assert result is not None
-        assert result.url == "https://example.com/hook"
         assert result.original_value is None
 
+    async def test_none_context_never_matches(self):
+        state = create_mock_state()
+        state.cache["chat:callback:tok"] = dict(_VALID_RECORD)
+
+        assert await resolve_callback_url("tok", state) is None
+        assert "chat:callback:tok" in state.cache
+
+
+class TestResolveCallbackUrlLocking:
+    """Python-specific: the consume path is serialized by a per-token lock."""
+
+    async def test_concurrent_resolves_yield_exactly_one_result(self):
+        state = MemoryStateAdapter()
+        await state.connect()
+        await state.set("chat:callback:race", dict(_VALID_RECORD))
+        context = CallbackContext(action_id="approve", channel_id="slack:C1")
+        real_get = state.get
+
+        async def yielding_get(key: str):
+            # Yield inside the critical section so the second click runs
+            # while the first still holds the token's lock.
+            await asyncio.sleep(0)
+            return await real_get(key)
+
+        state.get = yielding_get  # type: ignore[method-assign]
+
+        results = await asyncio.gather(
+            resolve_callback_url("race", state, context),
+            resolve_callback_url("race", state, context),
+        )
+
+        assert sum(r is not None for r in results) == 1
+        assert await state.get("chat:callback:race") is None
+
+    async def test_lost_lease_fails_closed_instead_of_double_consuming(self):
+        """Divergence from upstream (docs/UPSTREAM_SYNC.md): upstream returns
+        the record even when its 10 s lease lapsed mid-consume, so a second
+        click that took the expired lock also resolves and both POST.
+        """
+        state = MemoryStateAdapter()
+        await state.connect()
+        await state.set("chat:callback:stall", dict(_VALID_RECORD))
+        context = CallbackContext(action_id="approve", channel_id="slack:C1")
+        real_delete = state.delete
+        stalled = False
+        second: list[ResolvedCallback | None] = []
+
+        async def stalled_delete(key: str) -> None:
+            nonlocal stalled
+            if not stalled:
+                stalled = True
+                # The lease lapses while this delete is in flight, and a
+                # second click takes the lock and consumes the same record.
+                await state.force_release_lock(key)
+                second.append(await resolve_callback_url("stall", state, context))
+            await real_delete(key)
+
+        state.delete = stalled_delete  # type: ignore[method-assign]
+
+        first = await resolve_callback_url("stall", state, context)
+
+        assert first is None
+        assert second[0] is not None
+        assert second[0].url == "https://example.com/hook"
+        assert await state.get("chat:callback:stall") is None
+
+    async def test_returns_none_without_reading_when_lock_is_held(self):
+        state = create_mock_state()
+        state.cache["chat:callback:held"] = dict(_VALID_RECORD)
+        held = await state.acquire_lock("chat:callback:held", CALLBACK_LOCK_TTL_MS)
+        assert held is not None
+        state.get = AsyncMock(wraps=state.get)  # type: ignore[method-assign]
+
+        result = await resolve_callback_url("held", state, CallbackContext(action_id="approve", channel_id="slack:C1"))
+
+        assert result is None
+        state.get.assert_not_awaited()
+        assert state.cache["chat:callback:held"] == _VALID_RECORD
+
+    async def test_locks_the_record_key_with_a_ten_second_ttl(self):
+        state = create_mock_state()
+        state.cache["chat:callback:tok"] = dict(_VALID_RECORD)
+
+        await resolve_callback_url("tok", state, CallbackContext(action_id="approve", channel_id="slack:C1"))
+
+        assert state._acquire_lock_calls == [("chat:callback:tok", 10_000)]
+        # Released afterwards, so a later click is not locked out.
+        assert state._locks == {}
+
+    async def test_releases_lock_when_get_raises(self):
+        state = create_mock_state()
+        state.get = AsyncMock(side_effect=RuntimeError("state down"))  # type: ignore[method-assign]
+        state.release_lock = AsyncMock(wraps=state.release_lock)  # type: ignore[method-assign]
+
+        with pytest.raises(RuntimeError, match="state down"):
+            await resolve_callback_url("tok", state, CallbackContext(action_id="approve", channel_id="slack:C1"))
+
+        state.release_lock.assert_awaited_once()
+        assert state.release_lock.await_args.args[0].thread_id == "chat:callback:tok"
+        assert state._locks == {}
+
 
 # ===========================================================================
 # postToCallbackUrl
diff --git a/tests/test_channel_faithful.py b/tests/test_channel_faithful.py
index d14b1f2c..bc16b8f4 100644
--- a/tests/test_channel_faithful.py
+++ b/tests/test_channel_faithful.py
@@ -17,7 +17,7 @@
 
 import pytest
 
-from chat_sdk.callback_url import decode_callback_value
+from chat_sdk.callback_url import CallbackScope, decode_callback_value
 from chat_sdk.cards import Actions, Button, Card
 from chat_sdk.channel import ChannelImpl, _ChannelImplConfigWithAdapter, derive_channel_id
 from chat_sdk.errors import ChatNotImplementedError
@@ -1446,6 +1446,7 @@ async def test_should_encode_callbackurl_tokens_when_posting_a_card(self):
         stored = await state.get(f"chat:callback:{decoded.callback_token}")
         assert stored is not None
         assert stored["url"] == "https://example.com/hook"
+        assert stored["scope"] == {"id": "slack:C123", "type": "channel"}
 
     # it("should encode callbackUrl when posting via postEphemeral")
     @pytest.mark.asyncio
@@ -1484,6 +1485,59 @@ async def test_should_encode_callbackurl_when_posting_via_postephemeral(self):
         assert stored is not None
         assert stored["url"] == "https://example.com/eph"
 
+    # it("should bind fallback DM callbacks to the DM channel")
+    @pytest.mark.asyncio
+    async def test_should_bind_fallback_dm_callbacks_to_the_dm_channel(self):
+        adapter = create_mock_adapter()
+        state = create_mock_state()
+        channel = _make_channel(adapter, state)
+
+        await channel.post_ephemeral(
+            "U1",
+            Card(
+                children=[
+                    Actions(
+                        [
+                            Button(
+                                id="ack",
+                                label="Ack",
+                                callback_url="https://example.com/dm",
+                            )
+                        ]
+                    ),
+                ]
+            ),
+            PostEphemeralOptions(fallback_to_dm=True),
+        )
+
+        dm_thread_id, sent_card = adapter._post_calls[0]
+        assert dm_thread_id == "slack:DU1:"
+        button = sent_card["children"][0]["children"][0]
+        callback_token = decode_callback_value(button["value"]).callback_token
+        stored = await state.get(f"chat:callback:{callback_token}")
+        assert stored is not None
+        assert stored["scope"] == {"id": "slack:DU1", "type": "channel"}
+        assert len(self._callback_keys(state)) == 1
+
+    # Python-specific: tokens are minted per delivery path, so a channel
+    # ephemeral with no native path and no DM fallback mints nothing.
+    @pytest.mark.asyncio
+    async def test_post_ephemeral_without_delivery_path_mints_no_callback_token(self):
+        adapter = create_mock_adapter()
+        state = create_mock_state()
+        state.set = AsyncMock(wraps=state.set)  # type: ignore[method-assign]
+        channel = _make_channel(adapter, state)
+
+        result = await channel.post_ephemeral(
+            "U1",
+            Card(children=[Actions([Button(id="ack", label="Ack", callback_url="https://example.com/none")])]),
+            PostEphemeralOptions(fallback_to_dm=False),
+        )
+
+        assert result is None
+        state.set.assert_not_awaited()
+        assert self._callback_keys(state) == []
+
     # it("should encode callbackUrl when scheduling")
     @pytest.mark.asyncio
     async def test_should_encode_callbackurl_when_scheduling(self):
@@ -1555,6 +1609,135 @@ async def test_should_encode_callbackurl_when_editing_a_sent_card(self):
         assert stored is not None
         assert stored["url"] == "https://example.com/edit"
 
+    # Python-specific divergence (docs/UPSTREAM_SYNC.md): an edited channel
+    # card binds its tokens to the channel, not to the reported thread id.
+    # Round trip with the real Slack id functions and the real block_actions
+    # click, both for the synthetic `slack:C…:` post id Python reports today
+    # and for the `slack:C…:<ts>` id upstream 92530dd3 reports (#209). A DM
+    # click reports no ts, so a thread scope would miss it either way.
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("channel_id", "post_thread_id"),
+        [
+            ("slack:C123", "slack:C123:"),
+            ("slack:C123", "slack:C123:1700.1"),
+            ("slack:D123", "slack:D123:"),
+            ("slack:D123", "slack:D123:1700.1"),
+        ],
+    )
+    async def test_edited_slack_channel_card_resolves_for_the_real_click(self, channel_id: str, post_thread_id: str):
+        from chat_sdk.adapters.slack.adapter import SlackAdapter
+        from chat_sdk.adapters.slack.types import SlackAdapterConfig
+        from chat_sdk.callback_url import CallbackContext, resolve_callback_url
+
+        adapter = SlackAdapter(SlackAdapterConfig(bot_token="xoxb-test", signing_secret="s"))
+        clicks: list[Any] = []
+
+        class _ClickSink:
+            def process_action(self, event: Any, options: Any = None) -> None:
+                clicks.append(event)
+
+        adapter._chat = _ClickSink()  # type: ignore[assignment]
+        adapter.post_channel_message = AsyncMock(  # type: ignore[method-assign]
+            return_value=RawMessage(id="1700.1", thread_id=post_thread_id, raw={})
+        )
+        adapter.edit_message = AsyncMock(return_value=None)  # type: ignore[method-assign]
+        state = create_mock_state()
+        channel = ChannelImpl(_ChannelImplConfigWithAdapter(id=channel_id, adapter=adapter, state_adapter=state))
+
+        sent = await channel.post("Hello")
+        await sent.edit(Card(children=[Actions([Button(id="redo", label="Redo", callback_url="https://e.com/x")])]))
+
+        edit_thread_id, _, edited_card = adapter.edit_message.await_args.args
+        assert edit_thread_id == post_thread_id
+        token = decode_callback_value(edited_card["children"][0]["children"][0]["value"]).callback_token
+        assert token is not None
+
+        raw_channel = channel_id.split(":")[1]
+        adapter._handle_block_actions(
+            {
+                "type": "block_actions",
+                "channel": {"id": raw_channel},
+                "container": {"type": "message", "channel_id": raw_channel, "message_ts": "1700.1"},
+                "message": {"ts": "1700.1"},
+                "user": {"id": "U1", "username": "u"},
+                "actions": [{"action_id": "redo", "value": f"__cb:{token}"}],
+            }
+        )
+        click = clicks[0]
+        resolved = await resolve_callback_url(
+            token,
+            state,
+            CallbackContext(
+                action_id=click.action_id,
+                channel_id=adapter.channel_id_from_thread_id(click.thread_id),
+                thread_id=click.thread_id,
+            ),
+        )
+        assert resolved is not None
+        assert resolved.scope == CallbackScope(id=channel_id, type="channel")
+
+    # Python-specific divergence (docs/UPSTREAM_SYNC.md): when the adapter
+    # reports the channel id as the post's thread id (Teams, Google Chat),
+    # upstream's `{channel_id, "thread"}` edit scope can never match a click.
+    # Round trip with the real Teams id functions and the replayed click shape.
+    @pytest.mark.asyncio
+    async def test_edited_teams_channel_card_resolves_for_a_click_in_that_channel(self):
+        pytest.importorskip("microsoft_teams")
+        from chat_sdk.adapters.teams.adapter import TeamsAdapter
+        from chat_sdk.adapters.teams.types import TeamsAdapterConfig, TeamsThreadId
+        from chat_sdk.callback_url import CallbackContext, resolve_callback_url
+
+        adapter = TeamsAdapter(TeamsAdapterConfig(app_id="test-app-id", app_password="test-password"))
+        service_url = "https://smba.trafficmanager.net/teams/"
+        channel_id = adapter.encode_thread_id(
+            TeamsThreadId(conversation_id="19:abc@thread.tacv2", service_url=service_url)
+        )
+        # Real `post_channel_message` returns `thread_id=channel_id`.
+        adapter.post_channel_message = AsyncMock(  # type: ignore[method-assign]
+            return_value=RawMessage(id="1767297849909", thread_id=channel_id, raw={})
+        )
+        adapter.edit_message = AsyncMock(return_value=None)  # type: ignore[method-assign]
+        state = create_mock_state()
+        channel = ChannelImpl(_ChannelImplConfigWithAdapter(id=channel_id, adapter=adapter, state_adapter=state))
+
+        sent = await channel.post("Hello")
+        await sent.edit(Card(children=[Actions([Button(id="redo", label="Redo", callback_url="https://e.com/x")])]))
+
+        edited_card = adapter.edit_message.await_args.args[2]
+        token = decode_callback_value(edited_card["children"][0]["children"][0]["value"]).callback_token
+        assert token is not None
+        # A Teams card click reports the message's thread (`;messageid=`).
+        click_thread_id = adapter.encode_thread_id(
+            TeamsThreadId(conversation_id="19:abc@thread.tacv2;messageid=1767297849909", service_url=service_url)
+        )
+        resolved = await resolve_callback_url(
+            token,
+            state,
+            CallbackContext(
+                action_id="redo",
+                channel_id=adapter.channel_id_from_thread_id(click_thread_id),
+                thread_id=click_thread_id,
+            ),
+        )
+        assert resolved is not None
+        assert resolved.scope.type == "channel"
+        assert resolved.scope.id == channel_id
+
+    @pytest.mark.asyncio
+    async def test_chained_edit_keeps_callback_tokens_resolvable(self):
+        channel, adapter, state, post_calls = self._make_tracked_channel()
+
+        sent = await channel.post("Hello")
+        edited = await sent.edit("First edit")
+        await edited.edit(Card(children=[Actions([Button(id="redo", label="Redo", callback_url="https://e.com/x")])]))
+
+        edited_card = adapter._edit_calls[1][2]
+        token = decode_callback_value(edited_card["children"][0]["children"][0]["value"]).callback_token
+        stored = await state.get(f"chat:callback:{token}")
+        assert stored is not None
+        assert stored["scope"] == {"id": "slack:C123", "type": "channel"}
+
     # it("should pass plain string posts through unchanged")
     @pytest.mark.asyncio
     async def test_should_pass_plain_string_posts_through_unchanged(self):
diff --git a/tests/test_chat_faithful.py b/tests/test_chat_faithful.py
index 079cd80b..0a090dd2 100644
--- a/tests/test_chat_faithful.py
+++ b/tests/test_chat_faithful.py
@@ -17,6 +17,7 @@
 
 import pytest
 
+from chat_sdk.callback_url import CallbackContext, CallbackScope, ResolvedCallback
 from chat_sdk.chat import Chat
 from chat_sdk.emoji import get_emoji
 from chat_sdk.errors import ChatError, LockError
@@ -4271,8 +4272,10 @@ async def _handler(event):
         chat.on_action("approve", _handler)
 
         state.cache["chat:callback:testtoken123"] = {
+            "actionId": "approve",
             "url": "https://example.com/webhook/hook1",
             "originalValue": "order-789",
+            "scope": {"id": "slack:C123", "type": "channel"},
         }
 
         event = _make_action_event(adapter, action_id="approve", value="__cb:testtoken123")
@@ -4305,7 +4308,11 @@ async def test_should_decode_callbackurl_token_with_no_original_value(self):
 
         chat.on_action(lambda event: received.append(event))
 
-        state.cache["chat:callback:tok999"] = {"url": "https://example.com/webhook/hook2"}
+        state.cache["chat:callback:tok999"] = {
+            "actionId": "deny",
+            "url": "https://example.com/webhook/hook2",
+            "scope": {"id": "slack:C123:1234.5678", "type": "thread"},
+        }
 
         event = _make_action_event(adapter, action_id="deny", value="__cb:tok999")
 
@@ -4350,7 +4357,11 @@ async def _specific(event):
 
         chat.on_action("approve", _specific)
 
-        state.cache["chat:callback:tok555"] = {"url": "https://example.com/webhook/hook3"}
+        state.cache["chat:callback:tok555"] = {
+            "actionId": "approve",
+            "url": "https://example.com/webhook/hook3",
+            "scope": {"id": "slack:C123:1234.5678", "type": "thread"},
+        }
 
         event = _make_action_event(adapter, action_id="approve", value="__cb:tok555")
 
@@ -4362,6 +4373,75 @@ async def _specific(event):
         assert mock_fetch.await_count == 1
 
 
+class TestActionsCallbackTokenBinding:
+    """Python-specific: token binding and consumption through process_action."""
+
+    async def test_resolves_with_click_context_and_posts_the_stored_action_id(self):
+        chat, adapter, state = await _init_chat()
+        chat.on_action(lambda event: None)
+        resolved = ResolvedCallback(
+            url="https://example.com/webhook/ctx",
+            original_value="v",
+            action_id="minted-id",
+            scope=CallbackScope(id="slack:C123", type="channel"),
+        )
+        event = _make_action_event(adapter, action_id="approve", value="__cb:ctxtoken")
+
+        with (
+            patch("chat_sdk.chat.resolve_callback_url", new=AsyncMock(return_value=resolved)) as mock_resolve,
+            patch("chat_sdk.callback_url._fetch", new=AsyncMock(return_value=(200, "ok"))) as mock_fetch,
+        ):
+            await _process_action_and_wait(chat, event)
+
+        assert mock_resolve.await_args.args[0] == "ctxtoken"
+        assert mock_resolve.await_args.args[2] == CallbackContext(
+            action_id="approve",
+            channel_id="slack:C123",
+            thread_id="slack:C123:1234.5678",
+        )
+        body = json.loads(mock_fetch.await_args.kwargs["body"])
+        assert body["actionId"] == "minted-id"
+
+    async def test_repeat_click_dispatches_raw_value_without_posting(self):
+        chat, adapter, state = await _init_chat()
+        received: list[ActionEvent] = []
+        chat.on_action(lambda event: received.append(event))
+        state.cache["chat:callback:once"] = {
+            "actionId": "approve",
+            "url": "https://example.com/webhook/once",
+            "originalValue": "order-1",
+            "scope": {"id": "slack:C123:1234.5678", "type": "thread"},
+        }
+
+        with patch("chat_sdk.callback_url._fetch", new=AsyncMock(return_value=(200, "ok"))) as mock_fetch:
+            await _process_action_and_wait(chat, _make_action_event(adapter, value="__cb:once"))
+            await _process_action_and_wait(chat, _make_action_event(adapter, value="__cb:once"))
+
+        assert [e.value for e in received] == ["order-1", "__cb:once"]
+        assert mock_fetch.await_count == 1
+        assert "chat:callback:once" not in state.cache
+
+    async def test_click_in_another_thread_does_not_resolve_or_consume(self):
+        chat, adapter, state = await _init_chat()
+        received: list[ActionEvent] = []
+        chat.on_action(lambda event: received.append(event))
+        record = {
+            "actionId": "approve",
+            "url": "https://example.com/webhook/bound",
+            "scope": {"id": "slack:C123:1234.5678", "type": "thread"},
+        }
+        state.cache["chat:callback:bound"] = dict(record)
+
+        with patch("chat_sdk.callback_url._fetch", new=AsyncMock(return_value=(200, "ok"))) as mock_fetch:
+            await _process_action_and_wait(
+                chat, _make_action_event(adapter, value="__cb:bound", thread_id="slack:C999:1.1")
+            )
+
+        assert [e.value for e in received] == ["__cb:bound"]
+        assert mock_fetch.await_count == 0
+        assert state.cache["chat:callback:bound"] == record
+
+
 # ============================================================================
 # 25. Modal callbackUrl handling (vercel/chat#454)
 # ============================================================================
@@ -4607,7 +4687,11 @@ async def _handler(event):
 
         chat.on_action("approve", _handler)
 
-        state.cache["chat:callback:bad-token"] = {"url": "https://example.com/webhook/will-fail"}
+        state.cache["chat:callback:bad-token"] = {
+            "actionId": "approve",
+            "url": "https://example.com/webhook/will-fail",
+            "scope": {"id": "slack:C123:1234.5678", "type": "thread"},
+        }
 
         event = _make_action_event(adapter, action_id="approve", value="__cb:bad-token")
 
diff --git a/tests/test_thread_faithful.py b/tests/test_thread_faithful.py
index 9ac9e39a..ab03ebac 100644
--- a/tests/test_thread_faithful.py
+++ b/tests/test_thread_faithful.py
@@ -3808,6 +3808,7 @@ async def test_should_encode_callbackurl_when_posting_a_card(self):
         stored = await state.get(f"chat:callback:{callback_token}")
         assert stored is not None
         assert stored["url"] == "https://example.com/post-hook"
+        assert stored["scope"] == {"id": "slack:C123:1234.5678", "type": "thread"}
 
     # it("should encode callbackUrl when posting via postEphemeral with native support")
     @pytest.mark.asyncio
@@ -3840,6 +3841,46 @@ async def test_should_encode_callbackurl_when_posting_via_postephemeral_with_nat
         assert stored is not None
         assert stored["url"] == "https://example.com/eph"
 
+    # it("should bind fallback DM callbacks to the DM channel")
+    @pytest.mark.asyncio
+    async def test_should_bind_fallback_dm_callbacks_to_the_dm_channel(self):
+        adapter = create_mock_adapter()
+        state = create_mock_state()
+        thread = _make_thread(adapter, state)
+
+        await thread.post_ephemeral(
+            "U456",
+            _make_card_with_callback("https://example.com/dm"),
+            PostEphemeralOptions(fallback_to_dm=True),
+        )
+
+        dm_thread_id, sent_card = adapter._post_calls[0]
+        assert dm_thread_id == "slack:DU456:"
+        button = sent_card["children"][0]["children"][0]
+        callback_token = decode_callback_value(button["value"]).callback_token
+        stored = await state.get(f"chat:callback:{callback_token}")
+        assert stored is not None
+        assert stored["scope"] == {"id": "slack:DU456", "type": "channel"}
+
+    # Python-specific: with no native ephemeral and no DM fallback, nothing
+    # is delivered, so no callback token may be minted.
+    @pytest.mark.asyncio
+    async def test_post_ephemeral_without_delivery_path_mints_no_callback_token(self):
+        adapter = create_mock_adapter()
+        state = create_mock_state()
+        state.set = AsyncMock(wraps=state.set)  # type: ignore[method-assign]
+        thread = _make_thread(adapter, state)
+
+        result = await thread.post_ephemeral(
+            "U456",
+            _make_card_with_callback("https://example.com/none"),
+            PostEphemeralOptions(fallback_to_dm=False),
+        )
+
+        assert result is None
+        state.set.assert_not_awaited()
+        assert self._callback_keys(state) == []
+
     # it("should encode callbackUrl when scheduling a card")
     @pytest.mark.asyncio
     async def test_should_encode_callbackurl_when_scheduling_a_card(self):