Fix PubSub subscription race in streamStatus and restore cache invalidation on mutation failure

- Subscribe to PubSub before fetching initial status in streamStatus to
  close the race window where published changes could be missed between
  getStatus resolving and the PubSub subscription being established.

- Change onSuccess to onSettled for gitRunStackedActionMutationOptions
  and gitPullMutationOptions so branch query cache is invalidated even
  when mutations partially fail.

- Add refreshGitStatus to the failure path of gitRunStackedAction and
  use Effect.ensuring for gitPull so server-side git status broadcast
  fires regardless of success or failure.

Author: cursoragent

apps/server/src/git/Layers/GitStatusBroadcaster.ts

@@ -127,11 +127,13 @@ export const GitStatusBroadcasterLive = Layer.effect(
         Effect.gen(function* () {
           const normalizedCwd = normalizeCwd(input.cwd);
           yield* ensurePoller(normalizedCwd);
+
+          const subscription = yield* PubSub.subscribe(changesPubSub);
           const initialStatus = yield* getStatus({ cwd: normalizedCwd });
 
           return Stream.concat(
             Stream.make(initialStatus),
-            Stream.fromPubSub(changesPubSub).pipe(
+            Stream.fromEffectRepeat(PubSub.take(subscription)).pipe(
               Stream.filter((event) => event.cwd === normalizedCwd),
               Stream.map((event) => event.status),
             ),

apps/server/src/ws.ts

@@ -573,7 +573,7 @@ const WsRpcLayer = WsRpcGroup.toLayer(
       [WS_METHODS.gitPull]: (input) =>
         observeRpcEffect(
           WS_METHODS.gitPull,
-          git.pullCurrentBranch(input.cwd).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
+          git.pullCurrentBranch(input.cwd).pipe(Effect.ensuring(refreshGitStatus(input.cwd))),
           { "rpc.aggregate": "git" },
         ),
       [WS_METHODS.gitRunStackedAction]: (input) =>
@@ -589,7 +589,8 @@ const WsRpcLayer = WsRpcGroup.toLayer(
               })
               .pipe(
                 Effect.matchCauseEffect({
-                  onFailure: (cause) => Queue.failCause(queue, cause),
+                  onFailure: (cause) =>
+                    refreshGitStatus(input.cwd).pipe(Effect.andThen(Queue.failCause(queue, cause))),
                   onSuccess: () =>
                     refreshGitStatus(input.cwd).pipe(
                       Effect.andThen(Queue.end(queue).pipe(Effect.asVoid)),

apps/web/src/lib/gitReactQuery.ts

@@ -163,7 +163,7 @@ export function gitRunStackedActionMutationOptions(input: {
         ...(onProgress ? [{ onProgress }] : []),
       );
     },
-    onSuccess: async () => {
+    onSettled: async () => {
       await invalidateGitBranchQueries(input.queryClient, input.cwd);
     },
   });
@@ -177,7 +177,7 @@ export function gitPullMutationOptions(input: { cwd: string | null; queryClient:
       if (!input.cwd) throw new Error("Git pull is unavailable.");
       return api.git.pull({ cwd: input.cwd });
     },
-    onSuccess: async () => {
+    onSettled: async () => {
       await invalidateGitBranchQueries(input.queryClient, input.cwd);
     },
   });