From 7865e3a1e07ba4571bd2b02e11f8d91765341ce5 Mon Sep 17 00:00:00 2001 From: Charles Pan <30245070+charleslpan@users.noreply.github.com> Date: Tue, 20 Oct 2026 22:17:39 +0000 Subject: [PATCH] feat(host): a task's and an agent's write is confirmed once its transaction arrives on the followed task list The task list's follow hands its wait for a transaction (live-follow's TxWait, awaitTxId as Neon's mutations guide says) to the tasks composer. Each task write and each agent's start, rename, archive, and pin that answered a txid waits for it on the list before the window is answered, for at most ten seconds; one that changed nothing, or a list not yet subscribed, waits for nothing. Co-authored-by: luke-agent[bot] <339870089+luke-agent[bot]@users.noreply.github.com> --- packages/host/src/compose-coding-agents.ts | 11 ++++ packages/host/src/compose-host.ts | 1 + packages/host/src/compose-tasks.test.ts | 58 ++++++++++++++++++++-- packages/host/src/compose-tasks.ts | 40 +++++++++++++-- packages/host/src/task-list-follow.ts | 5 +- 5 files changed, 107 insertions(+), 8 deletions(-) diff --git a/packages/host/src/compose-coding-agents.ts b/packages/host/src/compose-coding-agents.ts index 69f71df464..091741b1e8 100644 --- a/packages/host/src/compose-coding-agents.ts +++ b/packages/host/src/compose-coding-agents.ts @@ -181,6 +181,12 @@ export interface CodingAgentsDependencies { /** The account signed in now, by its id, which every live read names; none where nobody is. */ accountId: () => string | undefined; openLive?: (url: string) => LiveClient | PromiseLike; + /** + * Waits for a write's transaction to arrive on the task list followed + * live, which holds every agent's row, so a Start, a rename, an archive, + * or a pin is confirmed once the list carries it; none waits for nothing. + */ + listed?: (txid: string | undefined) => Effect.Effect; } export interface CodingAgentsComposer extends Composer { @@ -225,6 +231,7 @@ function composed( const { kernel, account, client } = dependencies; const gate = () => kernel.runMode.sendsNetwork && account.capabilitiesActive(); const asked = askedOf(dependencies, http); + const listed = dependencies.listed ?? (() => Effect.void); /** The params read against the method's own schema, or the invalid-params refusal worded. */ const read = ( @@ -375,6 +382,7 @@ function composed( "starting an agent names a task and the press's own key", ); const answer = yield* asked(client.start(taskId, request)); + if ("txid" in answer) yield* listed(answer.txid); return carried(answer); }), [GATEWAY_METHOD.CODING_AGENTS_MESSAGES]: (params) => @@ -419,6 +427,7 @@ function composed( "renaming an agent names the agent and its title", ); const answer = yield* asked(client.rename(agentId, request)); + if ("txid" in answer) yield* listed(answer.txid); return carried(answer); }), [GATEWAY_METHOD.CODING_AGENTS_SERVERS]: (params) => @@ -439,6 +448,7 @@ function composed( "archiving an agent names the agent and whether it is archived", ); const answer = yield* asked(client.archive(agentId, request)); + if ("txid" in answer) yield* listed(answer.txid); return carried(answer); }), [GATEWAY_METHOD.CODING_AGENTS_PIN]: (params) => @@ -449,6 +459,7 @@ function composed( "pinning an agent names the agent and whether it is pinned", ); const answer = yield* asked(client.pin(agentId, request)); + if ("txid" in answer) yield* listed(answer.txid); return carried(answer); }), [GATEWAY_METHOD.CODING_AGENTS_STOP]: (params) => diff --git a/packages/host/src/compose-host.ts b/packages/host/src/compose-host.ts index 26f7bb641c..3e9f27e004 100644 --- a/packages/host/src/compose-host.ts +++ b/packages/host/src/compose-host.ts @@ -202,6 +202,7 @@ export function hostAssemblyLayer( accountId, client: codingAgentClient, openLive, + listed: tasks.listed, }); const automations = yield* composeAutomations({ kernel, diff --git a/packages/host/src/compose-tasks.test.ts b/packages/host/src/compose-tasks.test.ts index 888685e416..5e3a5c34eb 100644 --- a/packages/host/src/compose-tasks.test.ts +++ b/packages/host/src/compose-tasks.test.ts @@ -141,6 +141,8 @@ interface FakeService extends TasksClient { tasklessAgents: TasklessAgent[]; /** What the service answers when the list is asked for sealed, to follow over Neon Realtime; sealed by default. */ liveList: TaskCallResult; + /** The transaction a rename commits in, where the test says it changed a row. */ + renameTxid: string | undefined; /** Whether a delete is refused, as a service that did not answer refuses it. */ deleteFails: boolean; createAnswer: TaskCallResult; @@ -333,6 +335,7 @@ function fakeService(tasks: Task[]): FakeService { agents: {}, tasklessAgents: [], liveList: { ok: true, answer: SEALED_LIST }, + renameTxid: undefined, deleteFails: false, createAnswer: { ok: false, failure: TASK_CALL_FAILURE.UNANSWERED }, repositoriesAnswer: { ok: false, failure: TASK_CALL_FAILURE.UNANSWERED }, @@ -824,7 +827,9 @@ function fakeService(tasks: Task[]): FakeService { service.tasks = service.tasks.map((candidate) => candidate.id === taskId ? renamed : candidate, ); - return { task: renamed }; + return service.renameTxid === undefined + ? { task: renamed } + : { task: renamed, txid: service.renameTxid }; }), // The service sets the status and the priority named, and leaves the other standing. setStanding: (taskId, standing) => @@ -3419,18 +3424,29 @@ function fakeRealtime() { string, (snapshot: { readonly status: string; readonly data: readonly unknown[] | undefined }) => void >(); + // The transactions the list has seen, and the waits for one it has not seen yet. + const seen = new Set(); + const waiting: { txid: string; settle: () => void }[] = []; const open = (): LiveClient => ({ subscribe: (query) => ({ onChange: (listener) => { listeners.set(query.capability, listener); return () => listeners.delete(query.capability); }, - awaitTxId: () => Promise.resolve(), + awaitTxId: (txid) => + seen.has(txid) + ? Promise.resolve() + : new Promise((resolve) => waiting.push({ txid, settle: resolve })), renew: () => Promise.resolve(), unsubscribe: () => void listeners.delete(query.capability), }), close: () => undefined, }); + /** A transaction arriving on the list, settling every wait for it. */ + const arrive = (txid: string) => { + seen.add(txid); + for (const wait of waiting.filter((each) => each.txid === txid)) wait.settle(); + }; /** Waits for both parts to be subscribed, as the follow forked from a refresh subscribes them. */ const subscribed = Effect.gen(function* () { for (let turn = 0; turn < 100 && listeners.size < 2; turn += 1) yield* Effect.yieldNow; @@ -3446,7 +3462,7 @@ function fakeRealtime() { const settled = Effect.gen(function* () { for (let turn = 0; turn < 20; turn += 1) yield* Effect.yieldNow; }); - return { open, subscribed, deliver, fail, settled }; + return { open, subscribed, deliver, fail, settled, arrive }; } /** One plan as the plans statement delivers its row. */ @@ -3627,3 +3643,39 @@ it.effect( assert.equal(last()?.tasks[0]?.name, "Teammate invitations, again"); }), ); + +it.effect( + "a rename that changed a row is answered once its transaction arrives on the list followed live, and one that changed nothing at once", + () => + Effect.gen(function* () { + const service = fakeService([task(INVITES, "Teammate invitations", "# Invites", 10)]); + const realtime = fakeRealtime(); + const { call } = yield* subject(service, { openLive: realtime.open }); + yield* call(GATEWAY_METHOD.TASKS_REFRESH); + yield* realtime.subscribed; + realtime.deliver("plans", [taskRow(INVITES, "Teammate invitations")]); + realtime.deliver("agents", []); + yield* realtime.settled; + service.renameTxid = "4242"; + let answered = false; + + yield* Effect.forkChild( + Effect.tap(call(GATEWAY_METHOD.TASKS_RENAME, { taskId: INVITES, name: "Invites" }), () => + Effect.sync(() => { + answered = true; + }), + ), + ); + yield* realtime.settled; + assert.equal(answered, false, "the rename waits for its transaction"); + realtime.arrive("4242"); + yield* realtime.settled; + assert.equal(answered, true); + + service.renameTxid = undefined; + assert.deepEqual( + yield* call(GATEWAY_METHOD.TASKS_RENAME, { taskId: INVITES, name: "Invitations" }), + { renamed: true }, + ); + }), +); diff --git a/packages/host/src/compose-tasks.ts b/packages/host/src/compose-tasks.ts index 090e8aff61..7a3ec2e645 100644 --- a/packages/host/src/compose-tasks.ts +++ b/packages/host/src/compose-tasks.ts @@ -151,7 +151,7 @@ import { boardMoves } from "./board-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 { codeExcerpt, highlightCode, highlightFile } from "./task-code.js"; import { followTaskList } from "./task-list-follow.js"; @@ -327,6 +327,9 @@ type TaskMarkStamps = Pick; */ const WORKSPACES_READ_EVERY = Duration.minutes(1); +/** How long a write waits for its transaction to arrive on the list followed live before it is answered all the same. */ +const WRITE_CONFIRMED_WITHIN = Duration.seconds(10); + /** A GitHub page asked behind a closed gate, or of a service that did not answer, which opens nothing. */ const GITHUB_PAGE_UNANSWERED = { failure: TASK_CALL_FAILURE.UNANSWERED } as const; @@ -512,6 +515,8 @@ export interface TasksComposer extends Composer { * call's end keeps it. */ showWork: (taskId: string, turn: TaskWorkTurn) => void; + /** Waits for a write's transaction to arrive on the list followed live; none, or a list not subscribed, waits for nothing. */ + listed: (txid: string | undefined) => Effect.Effect; /** The call about `taskId` ended: its words are read back from the record. */ callEnded: (taskId: string) => void; } @@ -713,6 +718,22 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct /** The one follow of the list, started on the first refresh and stopped by a sign-out. */ const listFollow = yield* FiberHandle.make(); + /** The list's wait for a write's transaction, while the list stands subscribed. */ + let listWait: TxWait | undefined; + + /** + * A write's transaction waited for on the list followed live, so the write + * is confirmed once the list carries it, as Neon's mutations guide says; a + * write that changed nothing, or a list not yet subscribed, waits for + * nothing, and one not confirmed in time is answered all the same. + */ + const listed = (txid: string | undefined): Effect.Effect => + Effect.suspend(() => + txid === undefined || listWait === undefined + ? Effect.void + : Effect.ignore(listWait(txid, WRITE_CONFIRMED_WITHIN)), + ); + const followList = followTaskList({ // The account the follow starts for, which every seal names, so one under another's session is refused as signed out and ends. live: (poll) => @@ -728,10 +749,13 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct }), openLive: dependencies.openLive ?? openRealtimeClient, now: kernel.now, - publish: (listed) => { + onTxWait: (wait) => { + listWait = wait; + }, + publish: (answered) => { write({ - tasks: listed.plans, - tasklessAgents: listed.planlessAgents ?? [], + tasks: answered.plans, + tasklessAgents: answered.planlessAgents ?? [], listStatus: TASKS_READ.READY, }); // A name the service gave the open task reaches its document as the list carries it. @@ -1368,6 +1392,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { const marked = yield* served(client.mark(taskId, marks)); if (marked === undefined) return { changed: false }; + yield* listed(marked.txid); takeMarks(marked.task); return { changed: true }; }), @@ -1829,6 +1854,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { const started = yield* served(client.create(request)); if (!started.ok) return carried({ failure: started.failure }); + yield* listed(started.answer.txid); const { task, agent } = started.answer; // A task just started has had no call. A call about another task // is left to the start's own Talk press, which moves it here. @@ -1861,6 +1887,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { const deleted = yield* served(client.delete(taskId)); if (deleted === undefined) return { deleted: false }; + yield* listed(deleted.txid); const tasks = view.tasks.filter((task) => task.id !== taskId); if (view.activeTaskId === taskId) { yield* FiberHandle.clear(conversationRead); @@ -1885,6 +1912,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { const written = yield* served(client.rename(taskId, { name })); if (written === undefined) return { renamed: false }; + yield* listed(written.txid); const renamed = written.task; const tasks = view.tasks.map((task) => task.id === taskId ? { ...task, name: renamed.name } : task, @@ -1919,6 +1947,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { const changed = yield* served(client.setStanding(taskId, standing)); if (changed === undefined) return { changed: false }; + yield* listed(changed.txid); takeAnswered(changed.task); return { changed: true }; }), @@ -2018,6 +2047,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { const changed = yield* served(client.setRepository(taskId, repository)); if (!changed.ok) return carried({ failure: changed.failure }); + yield* listed(changed.answer.txid); takeAnswered(changed.answer.task); return carried({ repository: changed.answer.task.repository, @@ -2041,6 +2071,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { const moved = yield* served(client.move(taskId, workspaceId)); if (!moved.ok) return carried({ failure: moved.failure }); + yield* listed(moved.answer.txid); takeAnswered(moved.answer.task); return carried({ workspaceId: moved.answer.task.workspaceId }); }), @@ -2423,6 +2454,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct showCode, showWork, callEnded, + listed, reset: Effect.gen(function* () { yield* endTaskCall(undefined); yield* Cache.invalidateAll(pullRequestReads); diff --git a/packages/host/src/task-list-follow.ts b/packages/host/src/task-list-follow.ts index 178f0c7327..a04d8f849e 100644 --- a/packages/host/src/task-list-follow.ts +++ b/packages/host/src/task-list-follow.ts @@ -7,7 +7,7 @@ import { } from "@luke/features/task-list-live"; import type { TaskListAnswer } from "@luke/features/task-wire"; import { Clock, 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"; /** * task-list-follow.ts -- the task list followed over Neon Realtime as Neon's guide says to, built again whenever its rows move and whenever the clock alone moves an agent's status. @@ -41,6 +41,8 @@ export interface TaskListFollowDependencies { readonly openLive: (url: string) => LiveClient | PromiseLike; /** Now, by the host's own clock, which renewals are timed by. */ readonly now: () => number; + /** Hears the list's wait for a write's transaction once it is subscribed, and none once it is let go. */ + readonly onTxWait?: (wait: TxWait | undefined) => void; /** The list, each time its rows or the clock move it. */ readonly publish: (list: TaskListAnswer) => void; } @@ -73,6 +75,7 @@ function liveLists( reseal: () => dependencies.live(false), poll: () => dependencies.live(true), now: dependencies.now, + ...(dependencies.onTxWait === undefined ? undefined : { onTxWait: dependencies.onTxWait }), }, first, ).pipe(