address self-review findings on PR #60

- **Reject `max_concurrent` with non-concurrent strategy**: if a user
  sets `ConcurrencyConfig(strategy="queue", max_concurrent=5)`, they
  have a misconfig — the field is only honored under `"concurrent"`.
  Upstream accepts silently but we fail-fast so it's never invisible.
  New test `test_max_concurrent_with_non_concurrent_strategy_raises`.
- **Tighten error message**: raw input via `!r` repr instead of private
  `self._concurrency_max_concurrent` attribute; message now also tells
  the user how to fix ("pass None for unbounded").
- **Test robustness**: replace `for _ in range(20): await asyncio.sleep(0)`
  yield-counting loops with `asyncio.wait_for(_reach_cap(), timeout=1.0)`
  polling until the condition holds. Fixed counts can be flaky on slow
  CI when handlers have multiple internal await points.
- **Document divergence in UPSTREAM_SYNC.md**: added row to Known
  Non-Parity table with rationale (upstream accepts the config field
  but never enforces it; we do).
- **Inline breadcrumb comment** `# Divergence from upstream — see
  docs/UPSTREAM_SYNC.md` at the `_concurrent_semaphore` construction
  site, matching the divergence discipline in CLAUDE.md.

Author: patrick-chinchill

docs/UPSTREAM_SYNC.md

@@ -455,6 +455,7 @@ stay explicit instead of being rediscovered in code review.
 | Google Chat image rendering | Images emit as `{alt} ({url})` or bare `url` | No image branch — falls through to default which concatenates children only, dropping the URL | Upstream silently drops image URLs when rendering to Google Chat text. We preserve the URL so the message content isn't lost. |
 | Fallback streaming stream-exception capture | `_fallback_stream` captures exceptions from the stream iterator, flushes whatever content was already rendered, awaits `pending_edit`, and re-raises after cleanup | `try/finally` only — exception propagates immediately, `pendingEdit` is un-awaited, and the placeholder is stranded as `"..."` | Upstream leaves a hard UX failure when streams crash mid-flight (common: LLM connection drops): placeholder visible forever, orphan background task. We flush + clean up before re-raising so the caller still sees the original error and users see the partial content instead of a spinner. |
 | Fallback streaming final SentMessage content | SentMessage + final edit carry `final_content` (remend'd — inline markers auto-closed) | SentMessage + final edit carry raw `accumulated` | Narrow UX refinement. If a stream ends with an unclosed `*`/`~~`/etc., upstream ships the unclosed marker; we run `_remend` so the user sees a clean final message. Not observable in the common case where streams close their own markers. |
+| `ConcurrencyConfig.max_concurrent` | Enforced via `asyncio.Semaphore` in the `"concurrent"` strategy path; rejects `<= 0` and rejects the field under non-`"concurrent"` strategies | Accepted into the config type with docstring "Default: Infinity" but never read (3 writes, 0 reads) | Silent correctness bug upstream — consumers setting `max_concurrent=N` with `strategy="concurrent"` reasonably expect an N-way bound on in-flight handlers. We honor the documented contract via a semaphore and fail-fast on misconfiguration so it's never silent. |
 
 ### Platform-specific gaps
 

src/chat_sdk/chat.py

@@ -38,6 +38,7 @@
     Author,
     ChannelVisibility,
     ChatConfig,
+    ConcurrencyConfig,
     ConcurrencyStrategy,
     EmojiValue,
     Lock,
@@ -297,19 +298,33 @@ def __init__(self, config: ChatConfig | None = None, **kwargs: Any) -> None:
             self._concurrency_queue_entry_ttl_ms = concurrency.queue_entry_ttl_ms
 
         # -- Concurrent-strategy semaphore ------------------------------------
+        # Divergence from upstream — see docs/UPSTREAM_SYNC.md.
         # `max_concurrent` bounds in-flight handler dispatches when using the
         # `"concurrent"` strategy. `None` means unbounded (matches the upstream
         # TS default of `Infinity`). A positive integer caps parallel handler
-        # runs. Divergence from upstream: upstream accepts the config field
-        # but doesn't enforce it; we do.
+        # runs. Upstream accepts the config field but never enforces it
+        # (3 writes, 0 reads); we enforce it via `asyncio.Semaphore`.
+        #
+        # Only construct the semaphore when the strategy actually uses it —
+        # if a user sets `max_concurrent=5` with `strategy="queue"`, they
+        # have a misconfiguration that we surface as a `ValueError` rather
+        # than silently allocating an unused primitive.
         #
         # Reject `<= 0` explicitly rather than silently ignoring — a user
         # passing `max_concurrent=0` likely means "pause all processing"
         # (not supported) or has a typo. Either way, silently falling back
         # to unbounded concurrency would surprise them.
+        raw_max = concurrency.max_concurrent if isinstance(concurrency, ConcurrencyConfig) else None
         if self._concurrency_max_concurrent is not None and self._concurrency_max_concurrent <= 0:
             raise ValueError(
-                f"ConcurrencyConfig.max_concurrent must be > 0 or None, got {self._concurrency_max_concurrent}"
+                f"ConcurrencyConfig.max_concurrent must be a positive integer or None; "
+                f"got {raw_max!r}. Pass None for unbounded concurrency."
+            )
+        if self._concurrency_max_concurrent is not None and self._concurrency_strategy != "concurrent":
+            raise ValueError(
+                f"ConcurrencyConfig.max_concurrent is only honored when strategy='concurrent'; "
+                f"got strategy={self._concurrency_strategy!r}. Either switch to strategy='concurrent' "
+                "or drop max_concurrent."
             )
         self._concurrent_semaphore: asyncio.Semaphore | None = (
             asyncio.Semaphore(self._concurrency_max_concurrent)

tests/test_chat_faithful.py

@@ -2303,11 +2303,19 @@ async def handler(thread, message, context=None):
             for i in range(5)
         ]
 
-        # Let the first two handlers enter and block on the gate. Others
-        # must be blocked on the semaphore, so `in_flight` is capped at 2
-        # and the observed peak is <= 2 (not `== 2` — we'd narrow to 2
-        # only after the final drain below, when the peak is meaningful).
-        for _ in range(20):
+        # Poll until the first 2 handlers have reached the gate. Fixed
+        # yield counts can be flaky on slow CI — a handler that needs to
+        # traverse a few internal `await`s before reaching `gate.wait()`
+        # may not show up in N cycles. Poll with a short timeout instead.
+        async def _reach_cap() -> None:
+            while in_flight < 2:
+                await asyncio.sleep(0.001)
+
+        await asyncio.wait_for(_reach_cap(), timeout=1.0)
+
+        # Let any would-be extra tasks finish racing to the gate; if the
+        # bound leaks, in_flight would climb above 2 here.
+        for _ in range(10):
             await asyncio.sleep(0)
         assert in_flight == 2
         assert max_observed <= 2
@@ -2330,13 +2338,30 @@ async def test_max_concurrent_zero_or_negative_raises(self):
         for bad_value in (0, -1, -100):
             import pytest
 
-            with pytest.raises(ValueError, match="max_concurrent must be > 0 or None"):
+            with pytest.raises(ValueError, match="max_concurrent must be a positive integer or None"):
                 await _init_chat(
                     adapter=adapter,
                     state=state,
                     concurrency=ConcurrencyConfig(strategy="concurrent", max_concurrent=bad_value),
                 )
 
+    # Python-specific: setting `max_concurrent` with a non-concurrent strategy
+    # is a misconfiguration — the field is only honored under `"concurrent"`.
+    # Fail loudly instead of silently allocating an unused semaphore.
+    async def test_max_concurrent_with_non_concurrent_strategy_raises(self):
+        import pytest
+
+        state = create_mock_state()
+        adapter = create_mock_adapter("slack")
+
+        for bad_strategy in ("queue", "debounce", "drop"):
+            with pytest.raises(ValueError, match="only honored when strategy='concurrent'"):
+                await _init_chat(
+                    adapter=adapter,
+                    state=state,
+                    concurrency=ConcurrencyConfig(strategy=bad_strategy, max_concurrent=5),
+                )
+
     # Python-specific: None / missing max_concurrent must keep the
     # unbounded behavior (matches upstream TS default of Infinity).
     async def test_max_concurrent_none_allows_unbounded(self):
@@ -2367,9 +2392,14 @@ async def handler(thread, message, context=None):
             )
             for i in range(5)
         ]
-        for _ in range(20):
-            await asyncio.sleep(0)
-        # All 5 should be in flight simultaneously (no bound).
+
+        # Poll until all 5 are in flight; with no semaphore they should
+        # all reach the gate.
+        async def _reach_five() -> None:
+            while in_flight < 5:
+                await asyncio.sleep(0.001)
+
+        await asyncio.wait_for(_reach_five(), timeout=1.0)
         assert in_flight == 5
         gate.set()
         await asyncio.gather(*tasks)