Concurrency hardening for the async SDK (#60)

* test(concurrency): regression suite for six verified races

Adds deterministic concurrency tests for the async client and its
generated sync twin. Interleaving is driven by Event/Barrier handshakes
rather than wall-clock sleeps, so the ordering under test is fixed.

Tests assert the invariant (a failure surfaces as CodesphereError), not
the exception class, so they encode what callers depend on and survive
the fixes that follow.

Eleven of these fail against the current client, each for a distinct
defect: unrefcounted close(), a captured client across retry backoff,
cross-thread TOCTOU in the sync open(), a lost flags invalidation, a
retried DELETE whose 404 masks success, and in-place model mutation
that diverges from server write ordering.

* fix(client): reference-count transport scopes, map closed-client errors

Sharing one SDK across tasks was unsafe: open() was idempotent but
close() was not, so the first scope to exit tore the transport down for
everyone else. open()/close() are now reference counted behind
_compat.Lock, which also closes the check-then-act window that let two
threads in the sync twin each build a transport and orphan one.

The teardown awaits __aexit__ outside the lock so a slow drain cannot
block a concurrent open() of the next transport.

request() now re-resolves the client on every retry attempt instead of
capturing it once, so a close during backoff is caught by the SDK rather
than surfacing as httpx's bare RuntimeError. That error is remapped to
the new ClientStateError(CodesphereError, RuntimeError) only when the
client is actually gone; unrelated httpx internals still propagate. The
dual base keeps existing `except RuntimeError` handlers working.

test_open_is_idempotent asserted the old (buggy) teardown semantics and
is replaced by tests for reuse, balanced counting, and close underflow.

* fix(flags): honor invalidate() against an in-flight fetch

invalidate() cleared the snapshot outside the lock, so a get() already
awaiting its request would overwrite the None with data fetched before
the invalidation. The invalidation evaporated and the next read served
the stale snapshot.

A fetch now stamps the generation it started under and installs its
result only if invalidate() has not bumped it meanwhile. invalidate()
stays synchronous: taking the lock there would block the caller for the
duration of an in-flight flags request.

require() also read _legacy_platform outside the lock after get()
returned, so a concurrent refresh could pair a snapshot with the legacy
flag of a different fetch. _fetch() now returns both, and the new
_get_with_legacy() hands them to callers from one lock acquisition.

* fix(retry): treat a 404 on a retried DELETE as success

When a DELETE reached the platform and only its response was lost (a
gateway 503), the retry saw 404 and the SDK raised NotFoundError for an
operation that had in fact succeeded. Tearing down a landscape reported
"no landscape deployed" precisely because the teardown worked.

A 404 is now treated as success when the request is a DELETE and at
least one retry has already run. The attempt > 0 guard matters: a 404 on
the first attempt is a genuine miss and still raises. The rule is scoped
to DELETE, so a retried PUT that 404s is unaffected.

Safe for every current DELETE operation: all five declare
response_model=NoneType, so the response body is never parsed.

The RetryConfig docstring now states the residual risk this does not
solve. PUT and DELETE are idempotent as methods but not always in
effect, and without idempotency keys the SDK cannot stop a retried
teardown from executing twice. Callers who cannot tolerate that are
pointed at max_retries=0.

* feat(models)!: invalidate models after a write instead of guessing

BREAKING CHANGE: reading a field on a model after update() raises
StaleModelError until refresh() is called.

Writes copied values back into the local model, which is wrong by
construction under concurrency: the platform orders writes by arrival
while the client sees them by response. Two tasks updating the same
workspace could leave the model reporting "first" while the server held
"second", permanently and silently.

Workspace.update() wrote back the request payload; the Domain methods
wrote back the server's response, which is authoritative for the moment
it was produced but can still be overtaken by a concurrent write. Both
are now handled the same way: the instance is marked stale.

Staleness drops the field values from __dict__ so reads route through
__getattr__, which costs nothing until a model actually goes stale.
Identity fields survive, so a stale instance can still be logged and
re-fetched, and repr() marks it. model_dump/model_dump_json are guarded
too: pydantic serializes straight from __dict__, so without that a stale
model would quietly dump only its identity.

Workspace.refresh() and Domain.refresh() re-read the entity in place.
Domain writes still return the server's response; prefer it over self.

Removes codesphere.utils.update_model_fields, now unused.

* fix(landscape): detect foreign restarts and stop overrunning deadlines

wait_for_stage() could report success for a pipeline run the caller
never started: if someone else redeployed mid-wait, it simply observed
whatever run was current. The API exposes no run id, but started_at
changes on restart, so the run is pinned on first sighting and a
different value now raises ConflictError.

Both wait_for_stage() and wait_until_running() counted elapsed time by
summing poll_interval, ignoring how long the status requests took. A
slow endpoint could overrun the timeout by a wide margin (5.15s against
a 0.2s budget in the regression test). Both now measure against a
monotonic deadline and never sleep past it.

LogStream is backed by one SSE response body that can only be read once.
Re-entering it stranded the first stream context, and a second iterator
surfaced a raw httpx.StreamConsumed. Both now raise ClientStateError
explaining that streams are single-use.

Adds docs/guides/concurrency.md. Alongside the guarantees, it states
plainly what the SDK cannot do: with no ETags, If-Match, or idempotency
keys, read-modify-write cycles have an unclosable lost-update window and
retried writes can execute twice.

* build(sync-gen): do not abort generation when unasyncd rewrites files

unasyncd exits non-zero whenever it transforms a file, the way a
formatter signals "changed". Since run() defaulted to fatal, the very
run that did the work aborted before restore_all_blocks() and before
ruff normalized the generated tree — so `make sync-gen` failed on every
real change and had to be run twice to produce a correct result.

Author: Datata1

CHANGELOG.md

@@ -14,8 +14,53 @@
   quickstart, guides, full API reference, and `llms.txt`; deployed to
   GitHub Pages on pushes to `main`.
 
+### Added
+
+- Concurrency guide covering what is safe to share, how stale models
+  work, and the lost-update windows the Codesphere API provides no way
+  to close (it has no ETags, `If-Match`, or idempotency keys).
+- `Workspace.refresh()` and `Domain.refresh()` re-read an entity from the
+  server in place.
+
+### Fixed
+
+- `wait_for_stage()` pins the pipeline run it is waiting on by its start
+  time and raises `ConflictError` if someone else restarts the stage
+  meanwhile, instead of silently reporting the outcome of a run the
+  caller never started.
+- `wait_for_stage()` and `wait_until_running()` now measure their
+  `timeout` against the wall clock. Previously only the sleeps counted,
+  so slow status requests could overrun the deadline substantially.
+- Reusing a `LogStream` raises `ClientStateError` instead of stranding
+  the first stream context or surfacing a raw `httpx.StreamConsumed`.
+- Sharing one SDK across tasks or threads is now safe. `open()`/`close()`
+  are reference counted, so nested and concurrent scopes no longer tear
+  the transport down for each other, and two threads racing `open()` in
+  the sync client can no longer each build (and orphan) a connection
+  pool. Closing while a request is in flight now raises the new
+  `ClientStateError` instead of leaking httpx's bare `RuntimeError`.
+- `sdk.flags.invalidate()` is no longer discarded when a flags fetch is
+  already in flight; the stale snapshot used to be reinstated silently.
+  The `legacy_platform` marker in feature-flag errors can also no longer
+  come from a different fetch than the snapshot it is reported with.
+- A `404` on a retried `DELETE` is treated as success. Previously a
+  teardown that had actually succeeded reported `NotFoundError` when the
+  first attempt's response was lost behind a gateway error. A `404` on
+  the first attempt still raises.
+
 ### Changed
 
+- **Breaking:** writes no longer copy values back into the local model.
+  `Workspace.update()`, `Domain.update()`,
+  `Domain.update_workspace_connections()` and `Domain.verify_status()`
+  now mark the instance **stale**: reading a field, `to_dict()`,
+  `to_json()` or `to_yaml()` raises the new `StaleModelError` until you
+  call `await refresh()`. The platform orders writes by arrival while the
+  client sees them by response, so the old write-back could leave a model
+  reporting a value the platform did not hold. Identity fields (`id`, or
+  `name`/`team_id` for domains) stay readable. `Domain` methods still
+  return the server's response, which is authoritative — prefer it over
+  re-reading `self`.
 - Retries on transient failures are now enabled by default
   (`max_retries=2`). Idempotent methods (`GET`, `HEAD`, `PUT`, `DELETE`)
   are retried on `429`/`502`/`503`/`504` and connect/timeout errors.
@@ -32,6 +77,8 @@
 
 - The deprecated module path `codesphere.resources.workspace.envVars`
   (use `codesphere.resources.workspace.env_vars`).
+- `codesphere.utils.update_model_fields`, the helper behind the removed
+  write-back behavior. It had no remaining callers.
 
 ## v1.0.0 (2026-02-21)
 

docs/guides/concurrency.md

@@ -0,0 +1,162 @@
+# Concurrency
+
+The async client is built to be shared. This guide covers what the SDK
+guarantees when several tasks use it at once, and — just as important —
+what it cannot guarantee, because the Codesphere API offers no way to.
+
+## Share one client
+
+Create one `CodesphereSDK` and use it from as many tasks as you like.
+Sharing is preferred: each instance owns a connection pool, so creating
+one per task throws away connection reuse.
+
+```python
+import asyncio
+from codesphere import CodesphereSDK
+
+async def main():
+    async with CodesphereSDK() as sdk:
+        teams = await sdk.teams.list()
+        # One client, many concurrent calls.
+        workspaces = await asyncio.gather(
+            *(sdk.workspaces.list(team.id) for team in teams)
+        )
+```
+
+Scopes are reference counted, so nested and concurrent `async with`
+blocks are safe — the transport closes when the last one exits, not the
+first:
+
+```python
+async def worker(sdk):
+    async with sdk:              # each worker holds its own scope
+        await sdk.teams.list()
+
+async with CodesphereSDK() as sdk:
+    await asyncio.gather(worker(sdk), worker(sdk))
+    await sdk.teams.list()       # still open
+```
+
+Using the client after it is fully closed raises
+[`ClientStateError`][codesphere.ClientStateError], which is both a
+`CodesphereError` and a `RuntimeError`.
+
+The synchronous client offers the same guarantees across threads.
+
+## Models go stale after a write
+
+The platform applies writes in the order they arrive; your program sees
+them in the order responses come back. Those orders can differ, so the
+SDK cannot know an entity's state after a write it did not read back.
+Rather than report a value the platform may not hold, a write
+**invalidates** the instance:
+
+```python
+workspace = await sdk.workspaces.get(72678)
+
+await workspace.update(WorkspaceUpdate(name="renamed"))
+
+workspace.id      # fine: a write cannot change an entity's identity
+workspace.name    # raises StaleModelError
+```
+
+Call `refresh()` to re-read the server's actual state:
+
+```python
+await workspace.refresh()
+workspace.name    # server-authoritative
+```
+
+This applies to `Workspace.update()`, `Domain.update()`,
+`Domain.update_workspace_connections()` and `Domain.verify_status()`.
+`to_dict()`, `to_json()` and `to_yaml()` raise on a stale model too.
+The `Domain` methods return the server's response — prefer that returned
+object over re-reading the instance you called them on.
+
+## What the SDK cannot protect you from
+
+The workspace you are reading can be changed at the same moment by the
+web IDE, a CI pipeline, a teammate, or another process of your own. The
+Codesphere API has **no ETags, no `If-Match`, and no idempotency keys**,
+so there is no way for the SDK to detect or reject a conflicting write.
+These are real limitations, not oversights, and the SDK does not pretend
+otherwise.
+
+### Read-modify-write loses concurrent changes
+
+Any read, edit, write cycle has a window in which someone else's change
+is silently overwritten:
+
+```python
+# NOT safe against concurrent writers
+current = await workspace.env_vars.get()
+await workspace.env_vars.set([*current, EnvVar(name="NEW", value="1")])
+```
+
+`env_vars.set()` is a full replacement (`PUT`), so a variable added by
+someone else between the two calls is erased. The same applies to
+`landscape.get_profile()` → edit → `save_profile()`, which writes through
+a shell redirect and is not an atomic file replacement.
+
+If you must do this, narrow the window and verify afterwards by reading
+back. There is no way to make it atomic.
+
+### Retried writes can execute twice
+
+`PUT` and `DELETE` are idempotent as HTTP methods but not always in
+effect. If the platform receives a request and only the response is lost
+(a gateway `503`, a dropped connection), the retry runs the operation
+again — a landscape teardown can happen twice. Without idempotency keys
+the SDK cannot deduplicate this.
+
+The most common symptom is handled: a `404` on a **retried** `DELETE` is
+treated as success, since the resource being gone is what you asked for.
+A `404` on the first attempt still raises `NotFoundError`.
+
+If duplicate execution is unacceptable, disable retries for those calls:
+
+```python
+sdk = CodesphereSDK(retry=RetryConfig(max_retries=0))
+```
+
+### Waiting on a pipeline someone else restarted
+
+`wait_for_stage()` pins the run it is watching by its start time. If
+someone redeploys mid-wait, it raises
+[`ConflictError`][codesphere.ConflictError] instead of reporting the
+outcome of a run you never started:
+
+```python
+try:
+    await workspace.landscape.wait_for_stage("run", timeout=600)
+except ConflictError:
+    # Someone else redeployed. Decide whether to wait on the new run.
+    ...
+```
+
+Timeouts on `wait_for_stage()` and `wait_until_running()` are wall clock:
+time spent inside the status requests counts against your budget.
+
+## Log streams are single-use
+
+`logs.open()` returns a stream backed by one SSE response body, which can
+only be read once. Opening or iterating the same stream twice raises
+`ClientStateError`. Call `logs.open()` again for a second stream —
+concurrent streams over one client are fine:
+
+```python
+async def tail(target):
+    async for entry in workspace.logs.stream(target):
+        print(entry.message)
+
+await asyncio.gather(
+    tail(ServerTarget(step=1, server="web")),
+    tail(ServerTarget(step=1, server="api")),
+)
+```
+
+## Feature flags
+
+The flags snapshot is fetched once per client and cached. Concurrent
+gated calls collapse into a single request, and `invalidate()` is honored
+even against a fetch that is already in flight.

docs/llms.txt

@@ -26,6 +26,7 @@ keyword-only `timeout=` override. Errors derive from
 - [Retries & timeouts](https://datata1.github.io/codesphere-python/guides/retries/): default retry behavior, RetryConfig, per-call timeout
 - [Feature flags](https://datata1.github.io/codesphere-python/guides/feature-flags/): sdk.flags, operation gating errors
 - [Streaming logs](https://datata1.github.io/codesphere-python/guides/streaming-logs/): SSE log streaming, targets, deadlines
+- [Concurrency](https://datata1.github.io/codesphere-python/guides/concurrency/): sharing one client, stale models after writes, lost-update windows the API cannot close
 - [Sync vs Async](https://datata1.github.io/codesphere-python/guides/sync-vs-async/): choosing a flavor, shared models, caveats
 - [API Reference](https://datata1.github.io/codesphere-python/reference/client/): full typed API surface
 - [Changelog](https://datata1.github.io/codesphere-python/changelog/): release history

docs/reference/config.md

@@ -20,6 +20,8 @@
         - RateLimitError
         - NetworkError
         - TimeoutError
+        - ClientStateError
+        - StaleModelError
         - FeatureFlagError
         - FeatureNotAvailableError
         - FeatureNotEnabledError

mkdocs.yml

@@ -54,6 +54,7 @@ nav:
       - Retries & timeouts: guides/retries.md
       - Feature flags: guides/feature-flags.md
       - Streaming logs: guides/streaming-logs.md
+      - Concurrency: guides/concurrency.md
       - Sync vs Async: guides/sync-vs-async.md
   - API Reference:
       - Client: reference/client.md

scripts/gen_sync.py

@@ -50,7 +50,10 @@ def restore_all_blocks() -> int:
 
 
 def main() -> None:
-    run("uv", "run", "unasyncd")
+    # unasyncd exits non-zero whenever it rewrites a file, the way a
+    # formatter signals "changed". That is the normal case here, so it
+    # must not abort the post-processing below.
+    run("uv", "run", "unasyncd", fatal=False)
     restored = restore_all_blocks()
     print(f"Restored __all__ in {restored} generated files")
     # Remaining findings are silenced via per-file-ignores in ruff.toml;

src/codesphere/__init__.py

@@ -33,6 +33,7 @@
     APIError,
     AuthenticationError,
     AuthorizationError,
+    ClientStateError,
     CodesphereError,
     ConflictError,
     FeatureFlagError,
@@ -41,6 +42,7 @@
     NetworkError,
     NotFoundError,
     RateLimitError,
+    StaleModelError,
     TimeoutError,
     ValidationError,
 )
@@ -73,6 +75,7 @@
     "AuthorizationError",
     "CategoryFlags",
     "Characteristic",
+    "ClientStateError",
     "CodesphereError",
     "CodesphereSDK",
     "ConflictError",
@@ -93,6 +96,7 @@
     "NotFoundError",
     "RateLimitError",
     "RetryConfig",
+    "StaleModelError",
     "SyncCodesphereSDK",
     "Team",
     "TeamBase",

src/codesphere/_async/core/base.py

@@ -1,12 +1,13 @@
 from collections.abc import Mapping
-from typing import Any, TypeVar
+from typing import Any, ClassVar, TypeVar
 
 import httpx
 from pydantic import PrivateAttr
 
 from codesphere.core.models import CamelModel
 from codesphere.core.models import ResourceList as ResourceList
 from codesphere.core.operations import APIOperation
+from codesphere.exceptions import StaleModelError
 from codesphere.feature_flags import FlagRequirement
 
 from ..http_client import APIHttpClient
@@ -66,9 +67,78 @@ class BoundModel(CamelModel):
     Instances returned by the SDK get their client attached automatically,
     which lets entity methods (e.g. ``workspace.delete()``) make further
     API calls.
+
+    A write that the SDK cannot verify marks the instance **stale**: its
+    field values are dropped and reads raise
+    :class:`~codesphere.StaleModelError` until ``refresh()`` re-reads the
+    entity. Identity fields survive, so the instance can still be logged
+    and re-fetched. See :meth:`_mark_stale`.
     """
 
+    #: Fields that stay readable on a stale instance. An entity's identity
+    #: cannot be changed by a write, so it is always safe to report.
+    _identity_fields: ClassVar[tuple[str, ...]] = ("id",)
+
     _http_client: APIHttpClient | None = PrivateAttr(default=None)
+    _stale: bool = PrivateAttr(default=False)
+
+    def __getattr__(self, item: str) -> Any:
+        # Pydantic keeps field values in __dict__, so this only runs once a
+        # lookup has already failed: zero cost until a model goes stale.
+        if self._is_stale() and item in type(self).__pydantic_fields__:
+            raise StaleModelError(type(self).__name__, item)
+        # Delegate to pydantic, which resolves private attributes here.
+        # It defines __getattr__ only at runtime, so it is fetched
+        # dynamically rather than called through super() directly.
+        parent = getattr(super(), "__getattr__", None)
+        if parent is None:  # pragma: no cover - pydantic always defines it
+            raise AttributeError(item)
+        return parent(item)
+
+    def _is_stale(self) -> bool:
+        private = object.__getattribute__(self, "__pydantic_private__")
+        return bool(private and private.get("_stale"))
+
+    def _mark_stale(self) -> None:
+        """Drop local field values the SDK can no longer vouch for.
+
+        Called after a write whose resulting server state is unknown.
+        Clearing ``__dict__`` (rather than setting a flag beside intact
+        values) is what routes later reads through ``__getattr__``.
+        """
+        identity = {
+            name: value
+            for name, value in self.__dict__.items()
+            if name in type(self)._identity_fields
+        }
+        # Also drops cached_property managers; they rebuild after refresh.
+        self.__dict__.clear()
+        self.__dict__.update(identity)
+        self._stale = True
+
+    def _adopt(self, fresh: "BoundModel") -> None:
+        """Repopulate from a server-authoritative re-read."""
+        self.__dict__.clear()
+        self.__dict__.update(fresh.__dict__)
+        self._stale = False
+
+    def model_dump(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
+        # Pydantic serializes straight from __dict__, so without this a
+        # stale model would quietly dump only its identity fields.
+        if self._is_stale():
+            raise StaleModelError(type(self).__name__)
+        return super().model_dump(*args, **kwargs)
+
+    def model_dump_json(self, *args: Any, **kwargs: Any) -> str:
+        if self._is_stale():
+            raise StaleModelError(type(self).__name__)
+        return super().model_dump_json(*args, **kwargs)
+
+    def __repr_args__(self) -> Any:
+        # Keep repr() working for debugging, but say why it looks empty.
+        if self._is_stale():
+            return [*super().__repr_args__(), ("stale", True)]
+        return super().__repr_args__()
 
     def _client(self) -> APIHttpClient:
         if self._http_client is None or not hasattr(self._http_client, "request"):