fix: stabilize session switching and pin the repaired pb-mapper SDK

AGENTS.md

@@ -55,23 +55,17 @@ deps/
   codex/                   # acking-you/codex fork, branch `pocket-codex`
                            # (git submodule) = upstream openai/codex main +
                            # our adaptations; see §8
-  pb-mapper/               # upstream pb-mapper (git submodule)
-  kanal/                   # fork pinned to a known-good commit; transitively
-                           # required by pb-mapper, redirected via [patch]
-  uni-stream/              # ditto; transitively required by pb-mapper
 docs/                      # design notes, protocol references, CLI verification
 scripts/                   # install scripts, local CI, CI affected-surface gate
 ```
 
 `Cargo.toml` is a workspace root; every crate under `crates/` is a
 workspace member (see the `members` list for the canonical set).
-Submodules under `deps/` are kept **out** of the workspace via the
-`exclude` list — the pinned
-upstream crates use their own lints/profiles and we depend on them
-through explicit path or git deps where needed. The root manifest's
-`[patch]` table redirects `acking-you/kanal` and `acking-you/uni-stream`
-to the local submodules so the build stays reproducible across
-contributor checkouts and CI even after the upstream forks evolve.
+The Codex submodule under `deps/` is kept **out** of the workspace via the
+`exclude` list and retains its upstream lints/profiles. Cargo fetches pb-mapper
+from its dedicated `pocket-codex` Git branch; `Cargo.lock` pins the exact commit
+and its registry dependencies. No pb-mapper, kanal, or uni-stream submodules
+or local dependency patches are needed.
 
 ## 3. Crate responsibilities
 
@@ -81,7 +75,7 @@ Shared / host side:
 | --------------------------- | ---------------------------------------------------------------------------------------------- |
 | `pocket-codex-core`         | configuration schema, on-disk `state.toml`, well-known paths, error types, `service::{ServiceId, ServiceKind, sanitize_component, default_device_id}` for `pcx:<device>:<kind>:<name>` relay keys — small, dependency-light |
 | `pocket-codex-codex`        | spawning / supervising / inspecting the `codex app-server` child process (out-of-process *and* the in-process `embedded-codex` path), JSON-RPC envelope types |
-| `pocket-codex-pb`           | async wrappers around the published `pb-mapper` client SDK: `RelaySession` (address + credential), register / subscribe / status, `publish` (and the one name-conflict failure a caller must not retry), admin credential issuance, and credential keep-alive |
+| `pocket-codex-pb`           | async wrappers around the Git-pinned `pb-mapper` client SDK: `RelaySession` (address + credential), register / subscribe / status, `publish` (and the one name-conflict failure a caller must not retry), admin credential issuance, and credential keep-alive |
 | `pocket-codex-api-proxy`    | local Responses API proxy: forwards `/v1/responses` (HTTP + WS) to ChatGPT's Codex backend, reusing the host's `codex login`; shared by the CLI worker and the in-app host |
 | `pocket-codex-host-svc`     | host-side meta service — remote-viewable codex sessions, per-thread config, attachment upload — published on the relay as a third `meta:<name>` service |
 | `pocket-codex-cli`          | user-facing `pocket-codex` binary; account (`login` / `logout` / `account`), setup (`init`), high-level `serve` / `connect` / `api {serve,connect}` / `services {list,default set}` / `status` / `stop`, low-level `codex {start,stop,status}`, `pb {register,subscribe,status}`, `remote-hint`, `version` |
@@ -233,9 +227,12 @@ git checkout -- apps/flutter/pubspec.yaml   # restore before committing
 
 `deps/codex` is a git submodule pinned to a specific commit — the only one
 left. `deps/pb-mapper` (plus the `deps/kanal` and `deps/uni-stream` forks it
-pulled in transitively) is gone: pb-mapper is a registry dependency now, so its
-own transitive pins come from the lockfile rather than a mirrored `[patch]`
-table. After pulling this repo, materialise the submodule with:
+pulled in transitively) is gone. pb-mapper is a Cargo Git dependency on
+`https://github.com/acking-you/pb-mapper`, branch `pocket-codex`; `Cargo.lock`
+pins its exact commit. Push SDK fixes to that branch, then run
+`cargo update -p pb-mapper` here and commit the resulting lockfile after
+verification. Normal builds use `--locked` and do not automatically follow
+branch updates. After pulling this repo, materialise the Codex submodule with:
 
 ```bash
 git submodule update --init --recursive

Cargo.toml

@@ -20,8 +20,8 @@ members = [
 #
 # `deps/pb-mapper` used to sit here too, alongside the `deps/kanal` and
 # `deps/uni-stream` forks it pulled in transitively. All three are gone:
-# pb-mapper is a registry dependency now, so its own transitive pins are
-# its business rather than something this workspace has to mirror.
+# pb-mapper is fetched by Cargo from its dedicated Git branch; its transitive
+# dependencies are locked here without vendoring additional submodules.
 exclude = [
     "deps/codex",
     "apps/flutter",
@@ -41,12 +41,9 @@ readme = "README.md"
 # from here via `dep = { workspace = true }` so we have a single source
 # of truth for upgrades.
 [workspace.dependencies]
-# The pb-mapper client SDK, from the registry rather than the git submodule it
-# used to be. Pinned here so every consumer moves together: a client older than
-# the deployed relay cannot decode the relay's structured error frames, which is
-# how the 0.2.14-client/0.5.0-relay skew turned an over-quota refusal into an
-# unbacked-off reconnect storm.
-pb-mapper = "0.5.0"
+# The dedicated branch carries the SDK fixes used by Pocket-Codex. Cargo.lock
+# pins its exact commit; update it deliberately with `cargo update -p pb-mapper`.
+pb-mapper = { git = "https://github.com/acking-you/pb-mapper", branch = "pocket-codex" }
 
 # Internal crates
 pocket-codex-core = { path = "crates/pocket-codex-core" }
@@ -197,8 +194,8 @@ opt-level = 2
 # pb-mapper submodule. They existed because that submodule declared both as git
 # deps on `acking-you/*` branch HEADs, and one of those HEADs renamed a crate out
 # from under us — so both forks were vendored and re-routed to keep the build
-# reproducible. A registry dependency resolves its own transitive pins from the
-# lockfile, so none of that is our problem any more.
+# reproducible. The current SDK uses registry dependencies for both crates,
+# whose versions are recorded in Cargo.lock without local patches.
 
 # codex (deps/codex) requires forked tokio-tungstenite / tungstenite (they add a
 # `proxy` feature on the 0.28 line). The optional `embedded-codex` feature pulls

README.md

@@ -68,7 +68,7 @@ or a per-account GitHub login (hosted).
 | ------------------------------ | -------------------------------------- |
 | Workspace / lints / CI         | bootstrapped                           |
 | `pocket-codex` CLI             | `login`, `logout`, `account`, `init`, `serve`, `connect`, `api {serve,connect}`, `services {list,default set}`, top-level `status`/`stop`, `codex {start,stop,status}`, `pb {register,subscribe,status}`, `remote-hint`, `version` |
-| `pb-mapper` register/subscribe | the published `pb-mapper` client SDK   |
+| `pb-mapper` register/subscribe | Git SDK from the `pocket-codex` branch, pinned by `Cargo.lock`; includes connection timeout and interactive TCP latency fixes |
 | `codex app-server` supervision | spawn/stop/status via PID + state.toml |
 | App-server protocol | synced to upstream main `db0568dbb` (2026-09-07); acknowledged initialization, v2 account reads, current thread model/effort, and asynchronous question replies |
 | Embedded codex (desktop)       | desktop builds compile codex **in-process** behind the `embedded-codex` feature, so a machine can host without a separate `codex` install (Windows/macOS) |

apps/flutter/lib/src/screens/app_session_screen.dart

@@ -422,6 +422,7 @@ class _AppSessionState extends ConsumerState<AppSessionScreen>
   bool _settlingToEnd = false;
   String? _error;
   VoidCallback? _retry; // action for the error banner's retry button
+  int _threadLoadGeneration = 0;
   bool _connectionLost = false;
   // True while an automatic reconnect is in progress (drives the status bar's
   // "reconnecting" state). Auto-reconnect is attempted on stream close, on a
@@ -967,6 +968,7 @@ class _AppSessionState extends ConsumerState<AppSessionScreen>
       ref.read(uiPrefsProvider.notifier).setLastThread(widget.serviceKey, tid);
     }
     _cancelExternalWriterSubscription();
+    _threadLoadGeneration++;
     setState(() {
       _threadId = tid;
       _cwd = cwd;
@@ -1369,9 +1371,7 @@ class _AppSessionState extends ConsumerState<AppSessionScreen>
     ];
   }
 
-  /// Open an existing thread: resume it into the session (so reads and turns
-  /// resolve — otherwise the server returns "thread not found"), then load
-  /// its history.
+  /// Attach to an existing thread for live events and turns, then load history.
   Future<void> _resumeAndLoad() async {
     // Guard: a stale event (e.g. thread/compacted from a prior thread) can
     // arrive after switching to a new, unsaved conversation — don't `_threadId!`
@@ -1383,16 +1383,23 @@ class _AppSessionState extends ConsumerState<AppSessionScreen>
       _retry = null;
     });
     final startTid = _threadId!;
+    final generation = ++_threadLoadGeneration;
+    bool current() =>
+        mounted && _threadId == startTid && generation == _threadLoadGeneration;
     try {
       final api = ref.read(bridgeApiProvider);
       await api.appThreadResume(widget.serviceKey, startTid);
+      // An obsolete resume must not fan out into more history/config requests.
+      if (!current()) return;
       // Read the thread history and its persisted config concurrently. The
       // config is best-effort (an unreachable host meta tunnel yields an
       // all-unset config and we fall back to the server / in-memory restore).
       final historyFuture = api.appThreadRead(widget.serviceKey, startTid);
       final persistedFuture = _loadPersistedConfig(startTid);
       final history = await historyFuture;
+      if (!current()) return;
       final persisted = await persistedFuture;
+      if (!current()) return;
       // Restore the model from the server's own report first (the resume
       // response says what the thread actually runs with); fall back to the
       // persisted pick for older servers that don't report one. Resolve the id
@@ -1410,7 +1417,7 @@ class _AppSessionState extends ConsumerState<AppSessionScreen>
         }
       }
       // The user may have switched threads during the awaits above.
-      if (!mounted || _threadId != startTid) return;
+      if (!current()) return;
       setState(() {
         _loading = false;
         _replaceTranscriptItems(history.items);
@@ -1533,7 +1540,7 @@ class _AppSessionState extends ConsumerState<AppSessionScreen>
       // events that would normally flush it were missed during the drop).
       _maybeFlushQueue();
     } catch (e) {
-      if (!mounted || _threadId != startTid) return;
+      if (!current()) return;
       if (_isActiveWriterError(e)) {
         _enterExternalWriterMode(startTid);
         return;

apps/flutter/test/fake_bridge_api.dart

@@ -661,6 +661,9 @@ class FakeBridgeApi implements BridgeApi {
 
   /// Records the last resumed thread id for assertions.
   String? lastResumed;
+  final Map<String, List<Future<void>>> pendingResumes = {};
+  final Map<String, List<Future<ThreadHistory>>> pendingReads = {};
+  final List<String> threadReads = [];
 
   /// Optional failure thrown by [appThreadResume].
   Object? appThreadResumeError;
@@ -669,6 +672,8 @@ class FakeBridgeApi implements BridgeApi {
   Future<void> appThreadResume(String serviceKey, String threadId) async {
     if (appThreadResumeError != null) throw appThreadResumeError!;
     lastResumed = threadId;
+    final pending = pendingResumes[threadId];
+    if (pending != null && pending.isNotEmpty) await pending.removeAt(0);
   }
 
   /// Seedable history for resume tests.
@@ -678,7 +683,12 @@ class FakeBridgeApi implements BridgeApi {
   Future<ThreadHistory> appThreadRead(
     String serviceKey,
     String threadId,
-  ) async => readResult;
+  ) async {
+    threadReads.add(threadId);
+    final pending = pendingReads[threadId];
+    if (pending != null && pending.isNotEmpty) return await pending.removeAt(0);
+    return readResult;
+  }
 
   /// Older pages a paginated thread hands back, oldest batch LAST — each call
   /// to [appThreadOlderPage] pops the last one, so seeding