fix(client): publish synthetic must-refetch control message on 409

The previous commit removed the raw 409 body publish entirely, but
Shape relies on the must-refetch control message to clear its
accumulated data and trigger snapshot re-execution. Instead of
publishing the raw response body (which could contain stale data rows),
publish a synthetic control-only message.

Also: fix flaky SSE fallback test (remove brittle upper bound on SSE
request count), refine error-path-publish rule to allow static array
literal arguments (synthetic control messages) while still catching
dynamic publishes from error data.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: KyleAMathews

packages/typescript-client/bin/lib/shape-stream-static-analysis.mjs

@@ -1022,6 +1022,7 @@ function buildErrorPathPublishReport(sourceFile, classDecl) {
           if (!ts.isCallExpression(inner)) return
           const callee = getThisMemberName(inner.expression)
           if (!callee || !SUBSCRIBER_METHODS.has(callee)) return
+          if (isStaticControlMessagePublish(inner)) return
 
           report.push({
             method: methodName,
@@ -1041,6 +1042,7 @@ function buildErrorPathPublishReport(sourceFile, classDecl) {
           if (!ts.isCallExpression(inner)) return
           const callee = getThisMemberName(inner.expression)
           if (!callee || !SUBSCRIBER_METHODS.has(callee)) return
+          if (isStaticControlMessagePublish(inner)) return
 
           report.push({
             method: methodName,
@@ -1058,6 +1060,18 @@ function buildErrorPathPublishReport(sourceFile, classDecl) {
   return report.sort(compareReports)
 }
 
+/**
+ * Returns true if a call expression is a #publish([{ headers: { control: ... } }])
+ * with a static array literal containing only object literals. These synthetic
+ * control-only publishes are intentional (e.g., the must-refetch notification
+ * in the 409 handler) and should not trigger the error-path-publish rule.
+ */
+function isStaticControlMessagePublish(callExpr) {
+  const [firstArg] = callExpr.arguments
+  if (!firstArg || !ts.isArrayLiteralExpression(firstArg)) return false
+  return firstArg.elements.every((el) => ts.isObjectLiteralExpression(el))
+}
+
 /**
  * Returns true if the expression checks an HTTP error status (4xx or 5xx).
  * Recurses into && and || expressions.

packages/typescript-client/src/client.ts

@@ -920,6 +920,12 @@ export class ShapeStream<T extends Row<unknown> = Row>
         const nextRequestShapeCacheBuster = createCacheBuster()
         this.#reset(newShapeHandle)
 
+        // Notify subscribers that data must be re-fetched so they can
+        // clear accumulated state (e.g., Shape clears its row map).
+        // We publish a synthetic control message rather than the raw 409
+        // body to avoid delivering stale data rows to subscribers.
+        await this.#publish([{ headers: { control: `must-refetch` } }])
+
         return this.#requestShape(nextRequestShapeCacheBuster)
       } else {
         // errors that have reached this point are not actionable without

packages/typescript-client/test/client.test.ts

@@ -2987,11 +2987,10 @@ it(
         return urlObj.searchParams.get(`live_sse`) === `true`
       })
 
-      // After fallback, should see 3-5 SSE requests (3 short ones trigger fallback,
-      // but there may be more in flight due to async timing between the
-      // fallback decision and requests already dispatched)
+      // At least 3 SSE requests should have been made before fallback triggered.
+      // The exact upper bound is non-deterministic due to async timing between
+      // the fallback decision and requests already dispatched.
       expect(allSseRequests.length).toBeGreaterThanOrEqual(3)
-      expect(allSseRequests.length).toBeLessThanOrEqual(5)
 
       unsubscribe()
     } finally {

packages/typescript-client/test/model-based.test.ts

@@ -559,7 +559,7 @@ class Respond409SameHandleCmd
 // ─── Scenario Tests ────────────────────────────────────────────────
 
 describe(`ShapeStream targeted scenario tests`, () => {
-  it(`409 does not publish messages to subscribers`, async () => {
+  it(`409 publishes only a synthetic must-refetch control message`, async () => {
     resetSharedState()
 
     const initialHandle = `test-handle`
@@ -593,13 +593,17 @@ describe(`ShapeStream targeted scenario tests`, () => {
 
     receivedBatches.length = 0
 
-    // Send a 409 with a JSON body — the must-refetch control message
+    // Send a 409 — client resets and publishes a synthetic must-refetch
     await gate.waitForRequest()
     gate.provideResponse(make409(`new-handle`))
     await waitUntilSettled(gate, errorRef)
 
-    // Subscriber should NOT have received any messages from the 409
-    expect(receivedBatches).toEqual([])
+    // Subscriber should receive only the synthetic must-refetch control
+    // message — no data rows from the raw 409 response body
+    expect(receivedBatches).toHaveLength(1)
+    expect(receivedBatches[0]).toEqual([
+      { headers: { control: `must-refetch` } },
+    ])
 
     aborter.abort()
   })