Skip to content

Commit e810a33

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
committed
fix(client): sync failures stop dead-end retry taps and stale defects no longer park a healthy session
The mobile "Could not sync messages" pill called reconnect, which on a connected environment only health-checks the socket and never resubscribes, so the tap did nothing. It is no longer tappable; a new session or returning to the foreground still retries. A defect while applying stream items never reached onDefect, so the thread sat on "synchronizing" with no error. Recovery now sets the error itself. It also keys the wait on the session whose subscription defected, so a buffered defect from a replaced session retries at once instead of parking the healthy one. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 37717c6 commit e810a33

6 files changed

Lines changed: 44 additions & 13 deletions

File tree

‎apps/mobile/src/features/threads/ThreadDetailScreen.tsx‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -766,7 +766,7 @@ export const ThreadDetailScreen = function ThreadDetailScreen(props: ThreadDetai
766766
return null;
767767
}
768768
if (threadSyncPhase !== null) {
769-
return threadSyncFloatingStatus(threadSyncPhase, props.onReconnectEnvironment);
769+
return threadSyncFloatingStatus(threadSyncPhase);
770770
}
771771
const childInput = childThreadInputs[0];
772772
if (childInput && contentPresentationKind === "ready") {

‎apps/mobile/src/features/threads/floating-working-control.tsx‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -162,7 +162,7 @@ export function FloatingWorkingControl(props: {
162162
hasQueue ||
163163
(props.devicePreview !== null && props.browserPreview !== null);
164164
const statusInteractive =
165-
props.status?.kind === "connection" ||
165+
(props.status?.kind === "connection" && props.status.onPress !== undefined) ||
166166
props.status?.kind === "child-input" ||
167167
props.status?.kind === "background";
168168
const capsuleInteractive =
@@ -410,10 +410,11 @@ function FloatingStatusLabel(props: {
410410
<StatusLabelRow
411411
key="connection"
412412
accessibilityLabel={props.status.label}
413-
accessibilityRole="button"
413+
{...(props.status.onPress
414+
? { accessibilityRole: "button", onPress: props.status.onPress }
415+
: {})}
414416
className="gap-2"
415417
onLayout={props.onLayout}
416-
onPress={props.status.onPress}
417418
>
418419
{props.status.tone === "reconnecting" ? (
419420
<AppActivityIndicator size="small" colorClassName="accent-icon-muted" />

‎apps/mobile/src/features/threads/floating-working-status.test.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,12 @@ describe("thread sync status", () => {
77
expect(resolveThreadSyncPhase({ status: "cached", hasError: true, contentKind: "ready" })).toBe(
88
"failed",
99
);
10-
expect(threadSyncFloatingStatus("failed", () => {})).toMatchObject({
10+
expect(threadSyncFloatingStatus("failed")).toMatchObject({
1111
kind: "connection",
1212
tone: "unavailable",
1313
label: "Could not sync messages",
1414
});
15+
expect(threadSyncFloatingStatus("failed")).not.toHaveProperty("onPress");
1516
expect(
1617
resolveThreadSyncPhase({ status: "cached", hasError: false, contentKind: "ready" }),
1718
).toBe("syncing");

‎apps/mobile/src/features/threads/floating-working-status.ts‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ export type FloatingWorkingStatus =
4040
readonly kind: "connection";
4141
readonly tone: "reconnecting" | "unavailable";
4242
readonly label: string;
43-
readonly onPress: () => void;
43+
readonly onPress?: () => void;
4444
};
4545

4646
export function connectionFloatingStatus(input: {
@@ -104,10 +104,7 @@ export function resolveThreadSyncPhase(input: {
104104
}
105105
}
106106

107-
export function threadSyncFloatingStatus(
108-
phase: ThreadSyncPhase,
109-
onRetry: () => void,
110-
): FloatingWorkingStatus {
107+
export function threadSyncFloatingStatus(phase: ThreadSyncPhase): FloatingWorkingStatus {
111108
switch (phase) {
112109
case "loading":
113110
return { kind: "syncing", label: "Loading messages..." };
@@ -117,8 +114,10 @@ export function threadSyncFloatingStatus(
117114
return {
118115
kind: "connection",
119116
tone: "unavailable",
117+
// Not tappable: reconnecting a connected environment only health-checks
118+
// the socket and does not resubscribe. A new session or returning to
119+
// the foreground retries the sync.
120120
label: "Could not sync messages",
121-
onPress: onRetry,
122121
};
123122
}
124123
}

‎packages/client-runtime/src/state/threads-sync.test.ts‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1796,6 +1796,30 @@ describe("EnvironmentThreads", () => {
17961796
}),
17971797
);
17981798

1799+
it.effect("shows an error when applying a streamed item hits a defect", () =>
1800+
Effect.gen(function* () {
1801+
const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true });
1802+
expect(yield* Queue.take(harness.subscriptions)).toBe(CACHED_SNAPSHOT_SEQUENCE);
1803+
yield* Queue.offerAll(harness.inputs, [
1804+
snapshot(BASE_PROJECTION, CACHED_SNAPSHOT_SEQUENCE),
1805+
synchronized(),
1806+
]);
1807+
yield* awaitThreadState(harness.observed, (value) => value.status === "live");
1808+
1809+
yield* Queue.offer(harness.inputs, {
1810+
kind: "event",
1811+
event: null,
1812+
} as unknown as OrchestrationV2ThreadStreamItem);
1813+
const failed = yield* awaitThreadState(harness.observed, (value) =>
1814+
Option.isSome(value.error),
1815+
);
1816+
expect(Option.getOrThrow(failed.error)).toBe("Could not synchronize the thread.");
1817+
1818+
yield* harness.replaceSession;
1819+
expect(yield* Queue.take(harness.subscriptions)).toBe(CACHED_SNAPSHOT_SEQUENCE);
1820+
}),
1821+
);
1822+
17991823
it.effect("keeps replayed updates synchronizing until the completion marker arrives", () =>
18001824
Effect.gen(function* () {
18011825
const harness = yield* makeHarness({ cached: BASE_PROJECTION, completionMarker: true });

‎packages/client-runtime/src/state/threads.ts‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -262,6 +262,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
262262
const resumeDied = yield* Ref.make(false);
263263
const forceSnapshot = yield* Ref.make(false);
264264
const attemptSession = yield* Ref.make<RpcSession | undefined>(undefined);
265+
const defectSession = yield* Ref.make<RpcSession | undefined>(undefined);
265266
const applyLock = yield* Semaphore.make(1);
266267
// Save only completed data/cursor updates. A canceled scope must not cache
267268
// a cursor whose event has not reached the data yet.
@@ -986,6 +987,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
986987
if (Cause.hasDies(cause) && (yield* Ref.get(resuming))) {
987988
yield* Ref.set(resumeDied, true);
988989
} else {
990+
yield* Ref.set(defectSession, yield* Ref.get(attemptSession));
989991
yield* setStreamError("Could not synchronize the thread.");
990992
}
991993
}),
@@ -1005,8 +1007,12 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
10051007
}
10061008
// Any other defect keeps its error on screen and waits for a new
10071009
// session or a foreground wakeup, so recovery cannot loop. A session
1008-
// that changed before this wait counts as the next one.
1009-
const failedSession = yield* Ref.get(attemptSession);
1010+
// that changed before this wait counts as the next one, so a buffered
1011+
// defect from a replaced session retries at once. A defect applying
1012+
// items never reaches onDefect, so the error is set here too.
1013+
const failedSession =
1014+
(yield* Ref.getAndSet(defectSession, undefined)) ?? (yield* Ref.get(attemptSession));
1015+
yield* setStreamError("Could not synchronize the thread.");
10101016
yield* Stream.merge(
10111017
SubscriptionRef.changes(supervisor.session).pipe(
10121018
Stream.filter((session) => Option.getOrUndefined(session) !== failedSession),

0 commit comments

Comments
 (0)