refactor(chat-workflow): move active_stream_id CAS out of supabase lib

Per sweetman's review on updateChat.ts:64 — the active_stream_id-specific
predicate logic doesn't belong in the Supabase plumbing. Restructured:
- `lib/supabase/chats/updateChat.ts` now generic. The filter accepts
`where: Partial<Tables<"chats">>` (a generic predicate that maps to
`column = value` or `column IS NULL`) so no column name is hardcoded
in the Supabase lib.
- `lib/chat/compareAndSetChatActiveStreamId.ts` — new domain wrapper.
Owns the "compare-and-set on active_stream_id" concept and returns a
discriminated `{ok, claimed} | {ok: false, error}` result. Handler
and reconcileExistingActiveStream both compose against this wrapper
instead of constructing predicates inline.
- Handler + reconcile updated to use the wrapper. Tests follow.
37/37 tests in touched files pass; full suite 2955/2955; lint clean.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: sweetmantech

lib/chat/__tests__/compareAndSetChatActiveStreamId.test.ts

@@ -0,0 +1,51 @@
+import{describe,it,expect,vi,beforeEach}from"vitest";
+import{compareAndSetChatActiveStreamId}from"@/lib/chat/compareAndSetChatActiveStreamId";
+import{updateChat}from"@/lib/supabase/chats/updateChat";
+
+vi.mock("@/lib/supabase/chats/updateChat",()=>({
+updateChat: vi.fn(),
+}));
+
+beforeEach(()=>vi.clearAllMocks());
+
+describe("compareAndSetChatActiveStreamId",()=>{
+it("returns ok:true claimed:true when the row predicate matches and is updated",async()=>{
+vi.mocked(updateChat).mockResolvedValue({ok: true,rowsUpdated: 1,row: null});
+constresult=awaitcompareAndSetChatActiveStreamId("chat-1",null,"wrun_x");
+expect(result).toEqual({ok: true,claimed: true});
+expect(updateChat).toHaveBeenCalledWith(
+{id: "chat-1",where: {active_stream_id: null}},
+{active_stream_id: "wrun_x"},
+);
+});
+
+it("returns ok:true claimed:false when the predicate matches no rows (race lost)",async()=>{
+vi.mocked(updateChat).mockResolvedValue({ok: true,rowsUpdated: 0,row: null});
+constresult=awaitcompareAndSetChatActiveStreamId("chat-1",null,"wrun_x");
+expect(result).toEqual({ok: true,claimed: false});
+});
+
+it("returns ok:false with the underlying error on DB failure (distinct from race lost)",async()=>{
+vi.mocked(updateChat).mockResolvedValue({ok: false,error: "down"});
+constresult=awaitcompareAndSetChatActiveStreamId("chat-1",null,"wrun_x");
+expect(result).toEqual({ok: false,error: "down"});
+});
+
+it("supports expecting a specific run id (placeholder → real promotion)",async()=>{
+vi.mocked(updateChat).mockResolvedValue({ok: true,rowsUpdated: 1,row: null});
+awaitcompareAndSetChatActiveStreamId("chat-1","pending-abc","wrun_real");
+expect(updateChat).toHaveBeenCalledWith(
+{id: "chat-1",where: {active_stream_id: "pending-abc"}},
+{active_stream_id: "wrun_real"},
+);
+});
+
+it("supports next=null (releasing the slot)",async()=>{
+vi.mocked(updateChat).mockResolvedValue({ok: true,rowsUpdated: 1,row: null});
+awaitcompareAndSetChatActiveStreamId("chat-1","wrun_old",null);
+expect(updateChat).toHaveBeenCalledWith(
+{id: "chat-1",where: {active_stream_id: "wrun_old"}},
+{active_stream_id: null},
+);
+});
+});

lib/chat/__tests__/handleChatWorkflowStream.test.ts

@@ -7,15 +7,17 @@ import { selectSessions } from "@/lib/supabase/sessions/selectSessions";
 import{selectChats}from"@/lib/supabase/chats/selectChats";
 import{isSandboxActive}from"@/lib/sandbox/isSandboxActive";
 import{updateSession}from"@/lib/supabase/sessions/updateSession";
-import{updateChat}from"@/lib/supabase/chats/updateChat";
+import{compareAndSetChatActiveStreamId}from"@/lib/chat/compareAndSetChatActiveStreamId";
 import{maybeResumeChatStream}from"@/lib/chat/maybeResumeChatStream";
 import{persistLatestUserMessage}from"@/lib/chat/persistLatestUserMessage";
 import{start,getRun}from"workflow/api";
 
 vi.mock("@/lib/chat/validateChatWorkflow",()=>({validateChatWorkflow: vi.fn()}));
 vi.mock("@/lib/supabase/sessions/selectSessions",()=>({selectSessions: vi.fn()}));
 vi.mock("@/lib/supabase/chats/selectChats",()=>({selectChats: vi.fn()}));
-vi.mock("@/lib/supabase/chats/updateChat",()=>({updateChat: vi.fn()}));
+vi.mock("@/lib/chat/compareAndSetChatActiveStreamId",()=>({
+compareAndSetChatActiveStreamId: vi.fn(),
+}));
 vi.mock("@/lib/sandbox/isSandboxActive",()=>({isSandboxActive: vi.fn()}));
 vi.mock("@/lib/supabase/sessions/updateSession",()=>({updateSession: vi.fn()}));
 vi.mock("@/lib/sandbox/buildActiveLifecycleUpdate",()=>({
@@ -186,34 +188,39 @@ describe("handleChatWorkflowStream", () => {
 });
 
 it("returns 500 when the placeholder-CAS hits a DB error",async()=>{
-vi.mocked(updateChat).mockResolvedValueOnce({ok: false,error: "down"});
+vi.mocked(compareAndSetChatActiveStreamId).mockResolvedValueOnce({
+ok: false,
+error: "down",
+});
 constres=awaithandleChatWorkflowStream(makeRequest());
 expect(res.status).toBe(500);
 expect(start).not.toHaveBeenCalled();
 });
 
 it("returns 409 (without calling start) when the placeholder-CAS loses the race",async()=>{
-vi.mocked(updateChat).mockResolvedValueOnce({ok: true,rowsUpdated: 0,row: null});
+vi.mocked(compareAndSetChatActiveStreamId).mockResolvedValueOnce({
+ok: true,
+claimed: false,
+});
 constres=awaithandleChatWorkflowStream(makeRequest());
 expect(res.status).toBe(409);
 expect(start).not.toHaveBeenCalled();
 });
 
 it("starts the workflow only after placeholder CAS succeeds",async()=>{
 // First CAS = placeholder claim, second CAS = promote placeholder → real run id
-vi.mocked(updateChat)
-.mockResolvedValueOnce({ok: true,rowsUpdated: 1,row: null})
-.mockResolvedValueOnce({ok: true,rowsUpdated: 1,row: null});
+vi.mocked(compareAndSetChatActiveStreamId)
+.mockResolvedValueOnce({ok: true,claimed: true})
+.mockResolvedValueOnce({ok: true,claimed: true});
 mockStartedRun();
 constres=awaithandleChatWorkflowStream(makeRequest());
 expect(res.status).toBe(200);
 expect(start).toHaveBeenCalled();
-// Confirm CAS-before-start ordering
-constplaceholderCAS=vi.mocked(updateChat).mock.calls[0]?.[0];
-expect(placeholderCAS).toEqual({
-id: CHAT_ID,
-whereActiveStreamId: {equals: null},
-});
+// Confirm CAS-before-start ordering — first CAS pre-claims with expected=null
+constfirstCallArgs=vi.mocked(compareAndSetChatActiveStreamId).mock.calls[0];
+expect(firstCallArgs?.[0]).toBe(CHAT_ID);
+expect(firstCallArgs?.[1]).toBeNull();
+expect(firstCallArgs?.[2]).toMatch(/^pending-/);
 });
 });
 
@@ -222,9 +229,9 @@ describe("handleChatWorkflowStream", () => {
 mockValidated();
 mockSessionOwnedActive();
 mockChatOwned();
-vi.mocked(updateChat)
-.mockResolvedValueOnce({ok: true,rowsUpdated: 1,row: null})
-.mockResolvedValueOnce({ok: true,rowsUpdated: 1,row: null});
+vi.mocked(compareAndSetChatActiveStreamId)
+.mockResolvedValueOnce({ok: true,claimed: true})
+.mockResolvedValueOnce({ok: true,claimed: true});
 });
 
 it("returns 200 with text/event-stream and x-workflow-run-id",async()=>{
@@ -278,9 +285,9 @@ describe("handleChatWorkflowStream", () => {
 });
 
 it("awaits cancel() and returns 409 if promote loses",async()=>{
-vi.mocked(updateChat)
-.mockResolvedValueOnce({ok: true,rowsUpdated: 1,row: null})// claim ok
-.mockResolvedValueOnce({ok: true,rowsUpdated: 0,row: null});// promote raced
+vi.mocked(compareAndSetChatActiveStreamId)
+.mockResolvedValueOnce({ok: true,claimed: true})// claim ok
+.mockResolvedValueOnce({ok: true,claimed: false});// promote raced
 constcancel=vi.fn(()=>Promise.resolve());
 vi.mocked(start).mockResolvedValue({
 runId: "wrun_lost",
@@ -294,9 +301,9 @@ describe("handleChatWorkflowStream", () => {
 });
 
 it("still returns 409 if cancel() throws (best-effort)",async()=>{
-vi.mocked(updateChat)
-.mockResolvedValueOnce({ok: true,rowsUpdated: 1,row: null})
-.mockResolvedValueOnce({ok: true,rowsUpdated: 0,row: null});
+vi.mocked(compareAndSetChatActiveStreamId)
+.mockResolvedValueOnce({ok: true,claimed: true})
+.mockResolvedValueOnce({ok: true,claimed: false});
 vi.mocked(start).mockResolvedValue({
 runId: "wrun_lost",
 getReadable: ()=>newReadableStream(),

lib/chat/__tests__/reconcileExistingActiveStream.test.ts

@@ -1,13 +1,13 @@
 import{describe,it,expect,vi,beforeEach}from"vitest";
 import{reconcileExistingActiveStream}from"@/lib/chat/reconcileExistingActiveStream";
 import{getRun}from"workflow/api";
-import{updateChat}from"@/lib/supabase/chats/updateChat";
+import{compareAndSetChatActiveStreamId}from"@/lib/chat/compareAndSetChatActiveStreamId";
 
 vi.mock("workflow/api",()=>({
 getRun: vi.fn(),
 }));
-vi.mock("@/lib/supabase/chats/updateChat",()=>({
-updateChat: vi.fn(),
+vi.mock("@/lib/chat/compareAndSetChatActiveStreamId",()=>({
+compareAndSetChatActiveStreamId: vi.fn(),
 }));
 
 constCHAT_ID="chat-1";
@@ -41,13 +41,10 @@ describe("reconcileExistingActiveStream", () => {
 
 it("returns action=ready after CASing a completed run's stale id to null",async()=>{
 mockRun("completed");
-vi.mocked(updateChat).mockResolvedValue({ok: true,rowsUpdated: 1,row: null});
+vi.mocked(compareAndSetChatActiveStreamId).mockResolvedValue({ok: true,claimed: true});
 constresult=awaitreconcileExistingActiveStream(CHAT_ID,RUN_ID);
 expect(result.action).toBe("ready");
-expect(updateChat).toHaveBeenCalledWith(
-{id: CHAT_ID,whereActiveStreamId: {equals: RUN_ID}},
-{active_stream_id: null},
-);
+expect(compareAndSetChatActiveStreamId).toHaveBeenCalledWith(CHAT_ID,RUN_ID,null);
 });
 
 it("returns action=conflict when getRun throws (transient workflow API error)",async()=>{
@@ -57,7 +54,7 @@ describe("reconcileExistingActiveStream", () => {
 constresult=awaitreconcileExistingActiveStream(CHAT_ID,RUN_ID);
 expect(result.action).toBe("conflict");
 // Critical: we do NOT clear the stream id on transient error.
-expect(updateChat).not.toHaveBeenCalled();
+expect(compareAndSetChatActiveStreamId).not.toHaveBeenCalled();
 });
 
 it("returns action=conflict when status promise rejects",async()=>{
@@ -75,19 +72,19 @@ describe("reconcileExistingActiveStream", () => {
 }asnever);
 constresult=awaitreconcileExistingActiveStream(CHAT_ID,RUN_ID);
 expect(result.action).toBe("conflict");
-expect(updateChat).not.toHaveBeenCalled();
+expect(compareAndSetChatActiveStreamId).not.toHaveBeenCalled();
 });
 
-it("returns action=conflict when CAS-clear loses the race (rowsUpdated=0)",async()=>{
+it("returns action=conflict when CAS-clear loses the race (claimed=false)",async()=>{
 mockRun("completed");
-vi.mocked(updateChat).mockResolvedValue({ok: true,rowsUpdated: 0,row: null});
+vi.mocked(compareAndSetChatActiveStreamId).mockResolvedValue({ok: true,claimed: false});
 constresult=awaitreconcileExistingActiveStream(CHAT_ID,RUN_ID);
 expect(result.action).toBe("conflict");
 });
 
 it("returns action=conflict when CAS-clear hits a DB error (ok:false)",async()=>{
 mockRun("completed");
-vi.mocked(updateChat).mockResolvedValue({ok: false,error: "down"});
+vi.mocked(compareAndSetChatActiveStreamId).mockResolvedValue({ok: false,error: "down"});
 constresult=awaitreconcileExistingActiveStream(CHAT_ID,RUN_ID);
 // P1 fix: a failed re-read after CAS no longer falls through to "ready".
 expect(result.action).toBe("conflict");

lib/chat/compareAndSetChatActiveStreamId.ts

@@ -0,0 +1,49 @@
+import{updateChat}from"@/lib/supabase/chats/updateChat";
+
+/**
+ * Result of the CAS attempt. Forces callers to distinguish:
+ *
+ * - `{ ok: true, claimed: true }` — the row matched the expected value and
+ * was updated to `next`.
+ * - `{ ok: true, claimed: false }` — predicate didn't match (a race was
+ * lost OR the row's `active_stream_id` is in some other state).
+ * - `{ ok: false, error }` — Supabase / network failure. Distinct from
+ * "race lost" so callers don't return a misleading 409 when the DB is
+ * actually unhealthy.
+ */
+exporttypeCasChatActiveStreamIdResult=
+|{ok: true;claimed: boolean}
+|{ok: false;error: string};
+
+/**
+ * Atomically swap `chats.active_stream_id` from `expected` to `next` for
+ * the given chat. Domain wrapper over the generic `updateChat` helper —
+ * keeps the CAS-on-active_stream_id concept here (in the chat domain)
+ * rather than in the Supabase plumbing.
+ *
+ * Used by `/api/chat/workflow` to:
+ * - Claim the slot before `start(workflow)` (`expected: null`, `next: "pending-<uuid>"`).
+ * - Promote the placeholder to the real run id after start.
+ * - Release a stale slot in `reconcileExistingActiveStream`.
+ *
+ * @param chatId - Target chat id.
+ * @param expected - The value `active_stream_id` must currently hold (null to
+ * require an unset slot).
+ * @param next - The value to write (null to release the slot).
+ */
+exportasyncfunctioncompareAndSetChatActiveStreamId(
+chatId: string,
+expected: string|null,
+next: string|null,
+): Promise<CasChatActiveStreamIdResult>{
+constresult=awaitupdateChat(
+{id: chatId,where: {active_stream_id: expected}},
+{active_stream_id: next},
+);
+
+if(!result.ok){
+return{ok: false,error: result.error};
+}
+
+return{ok: true,claimed: result.rowsUpdated>0};
+}

lib/chat/handleChatWorkflowStream.ts

@@ -5,7 +5,7 @@ import { validateChatWorkflow } from "@/lib/chat/validateChatWorkflow";
 import{maybeResumeChatStream}from"@/lib/chat/maybeResumeChatStream";
 import{selectSessions}from"@/lib/supabase/sessions/selectSessions";
 import{selectChats}from"@/lib/supabase/chats/selectChats";
-import{updateChat}from"@/lib/supabase/chats/updateChat";
+import{compareAndSetChatActiveStreamId}from"@/lib/chat/compareAndSetChatActiveStreamId";
 import{isSandboxActive}from"@/lib/sandbox/isSandboxActive";
 import{buildActiveLifecycleUpdate}from"@/lib/sandbox/buildActiveLifecycleUpdate";
 import{updateSession}from"@/lib/supabase/sessions/updateSession";
@@ -73,12 +73,9 @@ export async function handleChatWorkflowStream(request: NextRequest): Promise<Re
 // the workflow. This closes the race where two requests both call start()
 // and bill the model before one loses the CAS.
 constplaceholder=`pending-${generateUUID()}`;
-constclaimed=awaitupdateChat(
-{id: validated.chatId,whereActiveStreamId: {equals: null}},
-{active_stream_id: placeholder},
-);
+constclaimed=awaitcompareAndSetChatActiveStreamId(validated.chatId,null,placeholder);
 if(!claimed.ok)returnerrorResponse("Internal server error",500);
-if(claimed.rowsUpdated===0){
+if(!claimed.claimed){
 returnerrorResponse("Another workflow is already running for this chat",409);
 }
 
@@ -96,15 +93,11 @@ export async function handleChatWorkflowStream(request: NextRequest): Promise<Re
 },
 ]);
 
-// Promote placeholder → real run id. We already own the slot so no CAS needed.
-constpromoted=awaitupdateChat(
-{id: validated.chatId,whereActiveStreamId: {equals: placeholder}},
-{active_stream_id: run.runId},
-);
-if(!promoted.ok||promoted.rowsUpdated===0){
-// Something asynchronously stole our slot, or DB went down between claim
-// and promote. Cancel the workflow we just started — losing the slot
-// means another stream owns the client.
+// Promote placeholder → real run id via CAS. If something asynchronously
+// stole the slot (or the DB went down) we cancel the workflow we just
+// started since another stream now owns the client.
+constpromoted=awaitcompareAndSetChatActiveStreamId(validated.chatId,placeholder,run.runId);
+if(!promoted.ok||!promoted.claimed){
 try{
 awaitgetRun(run.runId).cancel();
 }catch(error){

lib/chat/reconcileExistingActiveStream.ts

@@ -1,5 +1,5 @@
 import{getRun}from"workflow/api";
-import{updateChat}from"@/lib/supabase/chats/updateChat";
+import{compareAndSetChatActiveStreamId}from"@/lib/chat/compareAndSetChatActiveStreamId";
 
 exporttypeReconcileResult=
 |{action: "resume";runId: string;stream: ReadableStream<unknown>}
@@ -44,18 +44,13 @@ export async function reconcileExistingActiveStream(
 return{action: "conflict"};
 }
 
-// Run is terminally done. Attempt to claim the slot for the new request by
-// CASing the stale id back to null. If we win → ready. Anything else
-// (race lost OR DB error) → conflict, so we never accidentally start a
-// duplicate workflow on the back of a failed read.
-constcleared=awaitupdateChat(
-{id: chatId,whereActiveStreamId: {equals: activeStreamId}},
-{active_stream_id: null},
-);
-
-if(cleared.ok&&cleared.rowsUpdated>0){
+// Run is terminally done. Attempt to clear the stale id via CAS. If we
+// win → ready. Anything else (race lost OR DB error) → conflict, so we
+// never accidentally start a duplicate workflow on the back of a failed
+// read.
+constcleared=awaitcompareAndSetChatActiveStreamId(chatId,activeStreamId,null);
+if(cleared.ok&&cleared.claimed){
 return{action: "ready"};
 }
-
 return{action: "conflict"};
 }