feat(host): a chat write is confirmed once its transaction reaches the followed chat

packages/host/src/chat-follow.ts

@@ -3,6 +3,7 @@ import {
   CODING_AGENT_CALL_FAILURE,
   type CodingAgentCallFailure,
   type CodingAgentChatAnswerView,
+  type CodingAgentChatWriteAnswerView,
 } from "@luke/features/coding-agent-view";
 import {
   LIVE_SIGNED_OUT,
@@ -11,7 +12,7 @@ import {
   workspaceChatOfLiveRows,
 } from "@luke/features/live-wire";
 import { Duration, Effect, Option, Schema, Stream } from "effect";
-import { LiveBroken, type LiveClient, liveRows } from "./live-follow.js";
+import { LiveBroken, type LiveClient, liveRows, type TxWait } from "./live-follow.js";
 
 /**
  * chat-follow.ts -- a workspace's chat followed over Neon Realtime: joined once, sealed, and each change told whole.
@@ -29,11 +30,20 @@ import { LiveBroken, type LiveClient, liveRows } from "./live-follow.js";
  * or a seal the service did not
  * answer is told unanswered, and it, or a follow that stopped, is joined and
  * sealed again after `LIVE_RESTART`.
+ *
+ * A write to a followed chat is confirmed as Neon's mutations guide says:
+ * the service answers the transaction it committed in, and the write waits
+ * for that transaction to arrive on the followed read (`chatWriteSeen`), so
+ * the chat the window is pushed already holds it. A write that changed
+ * nothing answers none, and waits for nothing.
  */
 
 /** How long after a join or seal went unanswered, or a follow stopped, the chat is followed again. */
 const LIVE_RESTART = Duration.seconds(5);
 
+/** How long a write waits for its transaction to reach the followed chat, as Neon's guide waits. */
+const TXID_WAIT = Duration.seconds(10);
+
 /** Why the service answered no join or seal. */
 interface Failed {
   readonly failure: CodingAgentCallFailure;
@@ -51,6 +61,8 @@ export interface ChatFollowDependencies {
   readonly now: () => number;
   /** The chat whole, or why there is none, each time it changes. */
   readonly tell: (chat: CodingAgentChatAnswerView) => Effect.Effect<void>;
+  /** Hears the wait for a write's transaction while the chat is subscribed over Realtime, and none once it is not. */
+  readonly onTxWait: (wait: TxWait | undefined) => void;
 }
 
 /** The chat sealed again, gone where the account no longer reaches it, so the follow ends. */
@@ -82,6 +94,7 @@ function tellLive(
       reseal: () => resealed(dependencies, false),
       poll: () => resealed(dependencies, true),
       now: dependencies.now,
+      onTxWait: dependencies.onTxWait,
     },
     first,
   );
@@ -121,3 +134,18 @@ export function followWorkspaceChat(dependencies: ChatFollowDependencies): Effec
   );
   return loop;
 }
+
+/**
+ * Waits until the write's transaction has reached the followed chat, or
+ * `TXID_WAIT` passed: nothing where the write changed no row or no follow
+ * of the chat stands over Realtime. Note that the write is confirmed either
+ * way, since it committed; past the wait the window keeps the write's own
+ * answer until the chat it is pushed holds it.
+ */
+export function chatWriteSeen(
+  wait: TxWait | undefined,
+  answer: CodingAgentChatWriteAnswerView,
+): Effect.Effect<void> {
+  if (wait === undefined || "failure" in answer || answer.txid === undefined) return Effect.void;
+  return Effect.ignore(wait(answer.txid, TXID_WAIT));
+}

packages/host/src/compose-coding-agents.test.ts

@@ -3,7 +3,10 @@ import { it } from "@effect/vitest";
 import type { AccountInboxEntry } from "@luke/features/account-inbox-wire";
 import { CHAPTERS_MODEL } from "@luke/features/chapters-model-choice";
 import { CHAT_CURSOR_START, MEMBER_KIND } from "@luke/features/chat-wire";
-import { CODING_AGENT_CALL_FAILURE } from "@luke/features/coding-agent-view";
+import {
+  CODING_AGENT_CALL_FAILURE,
+  type CodingAgentChatWriteAnswerView,
+} from "@luke/features/coding-agent-view";
 import {
   CHANGED_FILE_STATUS,
   CHECK_SUMMARY,
@@ -249,6 +252,7 @@ function fakeService() {
   let cursor = CODING_AGENT_CURSOR_START;
   let transcript: AgentTranscriptLiveAnswer | { readonly failure: "not-found" } | undefined;
   let chatLive: WorkspaceChatLiveAnswer | { readonly failure: "not-found" } | undefined;
+  let written: CodingAgentChatWriteAnswerView = { readSeq: 4 };
   // Each page read at once, in turn; the last stands once the others are read.
   let pagesNow: readonly TranscriptNowAnswer[] = [];
   const client: CodingAgentClient = {
@@ -572,7 +576,7 @@ function fakeService() {
     chatPost: (workspaceId, write) =>
       Effect.sync(() => {
         asked.push(`chat-post:${workspaceId}:${JSON.stringify(write)}`);
-        return { readSeq: 4 };
+        return written;
       }),
     chatJoin: (workspaceId) =>
       Effect.sync(() => {
@@ -603,6 +607,10 @@ function fakeService() {
     sealChat: (next: WorkspaceChatLiveAnswer | { readonly failure: "not-found" } | undefined) => {
       chatLive = next;
     },
+    /** What the service answers a write to the chat. */
+    answerWrites: (next: CodingAgentChatWriteAnswerView) => {
+      written = next;
+    },
     /** The pages the service answers read at once, in turn. */
     pageNow: (...pages: readonly TranscriptNowAnswer[]) => {
       pagesNow = pages;
@@ -631,6 +639,9 @@ function fakeRealtime() {
     (snapshot: { readonly status: string; readonly data: readonly unknown[] | undefined }) => void
   >();
   const renewed: string[] = [];
+  /** Each wait for a transaction, by txid, until the test says it arrived. */
+  const waiting = new Map<string, () => void>();
+  const awaited: string[] = [];
   const open = (url: string): LiveClient => {
     opened.push(url);
     return {
@@ -643,7 +654,10 @@ function fakeRealtime() {
           renewed.push(next.capability);
           return Promise.resolve();
         },
-        awaitTxId: () => Promise.resolve(),
+        awaitTxId: (txid) => {
+          awaited.push(`${query.capability}:${txid}`);
+          return new Promise<void>((arrived) => waiting.set(txid, arrived));
+        },
         unsubscribe: () => void listeners.delete(query.capability),
       }),
       close: () => void closed.push(url),
@@ -662,6 +676,10 @@ function fakeRealtime() {
     },
     fail: (part: string) => listeners.get(`capability-${part}`)?.({ status: "error", data: [] }),
     subscribed: () => [...listeners.keys()].toSorted(),
+    /** Every wait for a transaction, as `capability:txid`, in order. */
+    awaited,
+    /** A transaction arriving on the read, settling whoever waits for it. */
+    arrive: (txid: string) => waiting.get(txid)?.(),
   };
 }
 
@@ -691,6 +709,7 @@ function subject(options: { signedIn?: boolean; sendsNetwork?: boolean } = {}) {
       asked: service.asked,
       sealTranscript: service.sealTranscript,
       sealChat: service.sealChat,
+      answerWrites: service.answerWrites,
       /** Every workspace chat the host said changed, in order. */
       chats: () =>
         events.flatMap((event) =>
@@ -1459,3 +1478,41 @@ it.effect(
       ]);
     }),
 );
+
+it.effect(
+  "a write to a followed chat is answered once its transaction reaches the chat, and a write that changed nothing, or to a chat not followed, at once",
+  () =>
+    Effect.gen(function* () {
+      const { call, sealChat, answerWrites, realtime } = yield* subject();
+      sealChat(SEALED_CHAT);
+      yield* call(GATEWAY_METHOD.CODING_AGENTS_CHAT_FOLLOW, { workspaceId: WORKSPACE_ID });
+      yield* settled;
+      deliverChat(realtime, []);
+      yield* settled;
+      const write = (workspaceId: string, request: WireRecord) =>
+        Effect.forkChild(
+          call(GATEWAY_METHOD.CODING_AGENTS_CHAT_POST, { workspaceId, write: request }),
+        );
+
+      answerWrites({ readSeq: 0, reads: [{ channelId: CHANNEL_ID, readSeq: 3 }], txid: "78" });
+      const read = yield* write(WORKSPACE_ID, { readTo: 3, channelId: CHANNEL_ID });
+      yield* settled;
+      assert.equal(read.pollUnsafe(), undefined, "the write waits for its transaction");
+      realtime.arrive("78");
+      yield* settled;
+      assert.deepEqual(yield* Fiber.join(read), {
+        readSeq: 0,
+        reads: [{ channelId: CHANNEL_ID, readSeq: 3 }],
+        txid: "78",
+      });
+      const waits = realtime.awaited.length;
+      assert.ok(waits > 0 && realtime.awaited.every((each) => each.endsWith(":78")));
+
+      // A read to where the cursor stands writes nothing, and another workspace is not followed.
+      answerWrites({ readSeq: 0 });
+      yield* Fiber.join(yield* write(WORKSPACE_ID, { readTo: 3, channelId: CHANNEL_ID }));
+      answerWrites({ readSeq: 0, txid: "79" });
+      yield* Fiber.join(yield* write("3c1e9d2b-7a4f-4e6b-8d0c-1f2a3b4c5d6e", { readTo: 3 }));
+      assert.equal(realtime.awaited.length, waits);
+    }),
+);

packages/host/src/compose-coding-agents.ts

@@ -88,11 +88,11 @@ import { unparsedWire } from "@luke/wire";
 import { readEither } from "@luke/wire/effect";
 import { Effect, FiberMap, RcMap, Result, type Schema, type Scope } from "effect";
 import * as HttpClient from "effect/http/HttpClient";
-import { followWorkspaceChat } from "./chat-follow.js";
+import { chatWriteSeen, followWorkspaceChat } from "./chat-follow.js";
 import type { AccountComposer } from "./compose-account.js";
 import type { Composer } from "./composer.js";
 import type { HostKernel } from "./host-kernel.js";
-import { type LiveClient, openRealtimeClient } from "./live-follow.js";
+import { type LiveClient, openRealtimeClient, type TxWait } from "./live-follow.js";
 import type { RunMode } from "./run-mode.js";
 import {
   type TranscriptFollows,
@@ -241,6 +241,8 @@ function composed(
   const gate = () => kernel.runMode.sendsNetwork && account.capabilitiesActive();
   const asked = askedOf(dependencies, http);
 
+  /** Each followed chat's wait for a write's transaction, by its workspace, while it is subscribed over Realtime. */
+  const chatTxWaits = new Map<string, TxWait>();
   /** A workspace's chat sealed for the account a follow started for, as every renewal of it names; signed out where none was. */
   const chatLiveFor = (workspaceId: string, userId: string | undefined) => (poll: boolean) =>
     userId === undefined
@@ -726,6 +728,7 @@ function composed(
           "writing to a workspace's chat names the workspace and one post, answer, resolve, or read",
         );
         const answer = yield* asked(client.chatPost(workspaceId, write));
+        yield* chatWriteSeen(chatTxWaits.get(workspaceId), answer);
         return carried<CodingAgentChatWriteAnswerView>(answer);
       }),
     [GATEWAY_METHOD.CODING_AGENTS_CHAT_FOLLOW]: (params) =>
@@ -746,6 +749,10 @@ function composed(
               openLive: dependencies.openLive ?? openRealtimeClient,
               now: kernel.now,
               tell,
+              onTxWait: (wait) => {
+                if (wait === undefined) chatTxWaits.delete(workspaceId);
+                else chatTxWaits.set(workspaceId, wait);
+              },
             })
           : tell(UNANSWERED);
         yield* FiberMap.run(chats, workspaceId, follow);