diff --git a/apps/web/tests/tasks-app.test.ts b/apps/web/tests/tasks-app.test.ts index b0e2f755e6..191b039259 100644 --- a/apps/web/tests/tasks-app.test.ts +++ b/apps/web/tests/tasks-app.test.ts @@ -58,6 +58,7 @@ import { TASK_PRIORITY, TASK_STATUS, taskAnswerSchema, + taskDeleteAnswerSchema, taskFileAnswerSchema, taskListAnswerSchema, UNTITLED_TASK_NAME, @@ -966,10 +967,14 @@ it.layer(getTestDB())("the plan routes", (it) => { HOSTED_HTTP_STATUS.OK, "the task stands", ); - assert.deepEqual(yield* ask(request(ONE_TASK(taskId), owner, { method: "DELETE" })), { - status: HOSTED_HTTP_STATUS.OK, - body: { deleted: true }, - }); + assert.equal( + readAnswer( + taskDeleteAnswerSchema, + HOSTED_HTTP_STATUS.OK, + yield* ask(request(ONE_TASK(taskId), owner, { method: "DELETE" })), + ).deleted, + true, + ); }), ); @@ -988,7 +993,12 @@ it.layer(getTestDB())("the plan routes", (it) => { const deletedBy = (taskId: string, by: string) => ask(request(ONE_TASK(taskId), undefined, { method: "DELETE", cookie: by })); const notFound = refusal(HOSTED_HTTP_STATUS.NOT_FOUND, HOSTED_API_ERROR.NOT_FOUND); - const deleted = { status: HOSTED_HTTP_STATUS.OK, body: { deleted: true } }; + // A delete is answered with the transaction it committed in, whichever that is. + const deletedAs = (taskId: string, by: string) => + Effect.map( + deletedBy(taskId, by), + (answer) => readAnswer(taskDeleteAnswerSchema, HOSTED_HTTP_STATUS.OK, answer).deleted, + ); assert.equal( (yield* ask(request(ONE_TASK(relay), undefined, { cookie }))).status, @@ -999,8 +1009,8 @@ it.layer(getTestDB())("the plan routes", (it) => { assert.deepEqual(yield* deletedBy(ledger, sessionCookieOf(other)), notFound); assert.deepEqual(retired, [], "another account's cookie retires nothing"); - assert.deepEqual(yield* deletedBy(relay, cookie), deleted); - assert.deepEqual(yield* deletedBy(ledger, cookie), deleted); + assert.equal(yield* deletedAs(relay, cookie), true); + assert.equal(yield* deletedAs(ledger, cookie), true); assert.deepEqual(retired, [ [deploymentFor(owner), relaySession], [deploymentFor(owner), ledgerSession], diff --git a/packages/database/src/models/coding-agent-store.ts b/packages/database/src/models/coding-agent-store.ts index 423a9eb869..055e6bf676 100644 --- a/packages/database/src/models/coding-agent-store.ts +++ b/packages/database/src/models/coding-agent-store.ts @@ -221,6 +221,8 @@ export interface TasklessCodingAgent extends CodingAgentLine { export interface CodingAgentStarted { readonly agent: CodingAgent; readonly created: boolean; + /** The transaction the agent's row was written in, where this Start wrote it. */ + readonly txid?: string; } /** The newest turn of an agent's conversation, as its status is read from it; nothing before the first turn. The read is the store's own (`message-reads.ts`), re-exported here beside the agents it is read for. */ @@ -582,6 +584,23 @@ const insertAgentConversation = SqlSchema.findOne({ ), }); +/** + * The id of the transaction a write committed in, read in the write's own + * `RETURNING`, as Neon's mutations guide says, so a client following the + * task list waits for that transaction to arrive before it confirms the + * write. + */ +const TXID = sql`pg_current_xact_id()::text`; + +/** An agent's row as a write returned it, with the transaction it committed in. */ +const CommittedAgentRowSchema = Schema.Struct({ + ...CodingAgentRowSchema.fields, + txid: Schema.String, +}); + +/** The transaction a pin's write committed in; none where the pin stood as asked already. */ +const TxidRowSchema = Schema.Struct({ txid: Schema.String }); + const insertAgent = SqlSchema.findOne({ Request: Schema.Struct({ userId: Schema.String, @@ -606,7 +625,7 @@ const insertAgent = SqlSchema.findOne({ payer: Schema.NullOr(PayerColumnSchema), now: Schema.Date, }), - Result: CodingAgentRowSchema, + Result: CommittedAgentRowSchema, execute: (write) => Database.use((db) => db @@ -635,7 +654,7 @@ const insertAgent = SqlSchema.findOne({ idempotencyKey: write.idempotencyKey, createdAt: write.now, }) - .returning(CODING_AGENT_COLUMNS), + .returning({ ...CODING_AGENT_COLUMNS, txid: TXID }), ), }); @@ -744,14 +763,14 @@ const updateAgentTitle = SqlSchema.findOneOption({ agentId: Schema.String, title: Schema.NullOr(Schema.String), }), - Result: CodingAgentRowSchema, + Result: CommittedAgentRowSchema, execute: ({ userId, agentId, title }) => Database.use((db) => db .update(codingAgent) .set({ title }) .where(reachableAgent(userId, agentId)) - .returning(CODING_AGENT_COLUMNS), + .returning({ ...CODING_AGENT_COLUMNS, txid: TXID }), ), }); @@ -802,14 +821,14 @@ const updateAgentArchived = SqlSchema.findOneOption({ archivedAt: Schema.NullOr(Schema.Date), archivedBy: ArchivedByColumnSchema, }), - Result: CodingAgentRowSchema, + Result: CommittedAgentRowSchema, execute: ({ userId, agentId, archivedAt, archivedBy }) => Database.use((db) => db .update(codingAgent) .set({ archivedAt, archivedBy }) .where(reachableAgent(userId, agentId)) - .returning(CODING_AGENT_COLUMNS), + .returning({ ...CODING_AGENT_COLUMNS, txid: TXID }), ), }); @@ -1365,25 +1384,28 @@ const findAgent = SqlSchema.findOneOption({ }); /** The account's pin of the agent, from now; one already standing keeps its instant. */ -const insertAgentPin = SqlSchema.void({ +const insertAgentPin = SqlSchema.findOneOption({ Request: Schema.Struct({ userId: Schema.String, agentId: Schema.String, now: Schema.Date }), + Result: TxidRowSchema, execute: ({ userId, agentId, now }) => Database.use((db) => db .insert(agentPin) .values({ userId, codingAgentId: agentId, pinnedAt: now }) - .onConflictDoNothing(), + .onConflictDoNothing() + .returning({ txid: TXID }), ), }); /** The account's pin of the agent at the position given, from now where it was not pinned; one already standing keeps its instant and moves. */ -const placeAgentPin = SqlSchema.void({ +const placeAgentPin = SqlSchema.findOneOption({ Request: Schema.Struct({ userId: Schema.String, agentId: Schema.String, position: Schema.String, now: Schema.Date, }), + Result: TxidRowSchema, execute: ({ userId, agentId, position, now }) => Database.use((db) => db @@ -1392,18 +1414,21 @@ const placeAgentPin = SqlSchema.void({ .onConflictDoUpdate({ target: [agentPin.userId, agentPin.codingAgentId], set: { position }, - }), + }) + .returning({ txid: TXID }), ), }); /** The account's pin of the agent taken away; every other member's stands. */ -const deleteAgentPin = SqlSchema.void({ +const deleteAgentPin = SqlSchema.findOneOption({ Request: Schema.Struct({ userId: Schema.String, agentId: Schema.String }), + Result: TxidRowSchema, execute: ({ userId, agentId }) => Database.use((db) => db .delete(agentPin) - .where(and(eq(agentPin.userId, userId), eq(agentPin.codingAgentId, agentId))), + .where(and(eq(agentPin.userId, userId), eq(agentPin.codingAgentId, agentId))) + .returning({ txid: TXID }), ), }); @@ -1481,7 +1506,7 @@ export function createCodingAgent( // An insert that returned no row is the database breaking its own contract, not an outcome. Effect.catchTag("NoSuchElementError", (missing) => Effect.die(missing)), ); - const row = yield* insertAgent({ + const { txid, ...row } = yield* insertAgent({ userId, conversationId: conversation.id, taskId: started.taskId, @@ -1504,7 +1529,7 @@ export function createCodingAgent( payer: started.payer ?? null, now, }).pipe(Effect.catchTag("NoSuchElementError", (missing) => Effect.die(missing))); - return Option.some({ agent: row, created: true }); + return Option.some({ agent: row, created: true, txid }); }), ), ); @@ -1928,7 +1953,30 @@ export function renameCodingAgent( agentId: string, title: string | null, ): CodingAgentStoreEffect> { - return updateAgentTitle({ userId, agentId, title }); + return Effect.map(renameCodingAgentCommitted(userId, agentId, title), Option.map(agentOf)); +} + +/** An agent as a write left it, and the transaction it committed in; none where it changed no row. */ +export interface CodingAgentCommitted { + readonly agent: CodingAgent; + readonly txid?: string; +} + +/** The agent's row without the transaction it was returned with. */ +function agentOf({ agent }: CodingAgentCommitted): CodingAgent { + return agent; +} + +/** The agent renamed, as `renameCodingAgent` renames it, and the transaction its row was written in. */ +export function renameCodingAgentCommitted( + userId: string, + agentId: string, + title: string | null, +): CodingAgentStoreEffect> { + return Effect.map( + updateAgentTitle({ userId, agentId, title }), + Option.map(({ txid, ...agent }) => ({ agent, txid })), + ); } /** The account's agent archived, stamped now with who archived it, by default the account itself, or brought back, which clears who did; its conversation and what it published are kept either way. Nothing for an agent of a workspace the account is not in. */ @@ -1938,13 +1986,29 @@ export function archiveCodingAgent( archived: boolean, archivedBy: StoredArchive = { by: CODING_AGENT_ARCHIVER.MEMBER, userId }, ): CodingAgentStoreEffect> { + return Effect.map( + archiveCodingAgentCommitted(userId, agentId, archived, archivedBy), + Option.map(agentOf), + ); +} + +/** The agent archived or brought back, as `archiveCodingAgent` does it, and the transaction its row was written in. */ +export function archiveCodingAgentCommitted( + userId: string, + agentId: string, + archived: boolean, + archivedBy: StoredArchive = { by: CODING_AGENT_ARCHIVER.MEMBER, userId }, +): CodingAgentStoreEffect> { return Effect.flatMap(DateTime.nowAsDate, (now) => - updateAgentArchived({ - userId, - agentId, - archivedAt: archived ? now : null, - archivedBy: archived ? archivedBy : null, - }), + Effect.map( + updateAgentArchived({ + userId, + agentId, + archivedAt: archived ? now : null, + archivedBy: archived ? archivedBy : null, + }), + Option.map(({ txid, ...agent }) => ({ agent, txid })), + ), ); } @@ -2066,14 +2130,30 @@ export function pinCodingAgent( pinned: boolean, position?: string, ): CodingAgentStoreEffect> { + return Effect.map( + pinCodingAgentCommitted(userId, agentId, pinned, position), + Option.map(agentOf), + ); +} + +/** The account's pin of the agent set, as `pinCodingAgent` sets it, and the transaction the pin was written in; none where it stood as asked. */ +export function pinCodingAgentCommitted( + userId: string, + agentId: string, + pinned: boolean, + position?: string, +): CodingAgentStoreEffect> { return Effect.gen(function* () { const found = yield* findAgent({ userId, agentId }); if (Option.isNone(found) || found.value.taskId !== null) return Option.none(); const now = yield* DateTime.nowAsDate; - if (!pinned) yield* deleteAgentPin({ userId, agentId }); - else if (position === undefined) yield* insertAgentPin({ userId, agentId, now }); - else yield* placeAgentPin({ userId, agentId, position, now }); - return found; + const written = !pinned + ? yield* deleteAgentPin({ userId, agentId }) + : position === undefined + ? yield* insertAgentPin({ userId, agentId, now }) + : yield* placeAgentPin({ userId, agentId, position, now }); + const txid = Option.getOrUndefined(written)?.txid; + return Option.some(txid === undefined ? { agent: found.value } : { agent: found.value, txid }); }); } diff --git a/packages/database/src/models/task-store.ts b/packages/database/src/models/task-store.ts index f827134453..2d72f3697d 100644 --- a/packages/database/src/models/task-store.ts +++ b/packages/database/src/models/task-store.ts @@ -294,16 +294,33 @@ const findPin = SqlSchema.findOneOption({ }); /** The reader's pin of the plan, from now; one already standing keeps its instant. */ -const insertPin = SqlSchema.void({ +/** + * The id of the transaction a write committed in, read in the write's own + * `RETURNING`, as Neon's mutations guide says, so a client following the + * task list waits for that transaction to arrive before it confirms the + * write. + */ +const TXID = sql`pg_current_xact_id()::text`; + +/** The transaction a pin's write committed in; none where the pin stood as asked already. */ +const TxidRowSchema = Schema.Struct({ txid: Schema.String }); + +const insertPin = SqlSchema.findOneOption({ + Result: TxidRowSchema, Request: Schema.Struct({ userId: Schema.String, taskId: Schema.String, now: Schema.Date }), execute: ({ userId, taskId, now }) => Database.use((db) => - db.insert(taskPin).values({ userId, taskId, pinnedAt: now }).onConflictDoNothing(), + db + .insert(taskPin) + .values({ userId, taskId, pinnedAt: now }) + .onConflictDoNothing() + .returning({ txid: TXID }), ), }); /** The reader's pin of the plan at the position given, from now where it was not pinned; one already standing keeps its instant and moves. */ -const placePin = SqlSchema.void({ +const placePin = SqlSchema.findOneOption({ + Result: TxidRowSchema, Request: Schema.Struct({ userId: Schema.String, taskId: Schema.String, @@ -315,16 +332,21 @@ const placePin = SqlSchema.void({ db .insert(taskPin) .values({ userId, taskId, position, pinnedAt: now }) - .onConflictDoUpdate({ target: [taskPin.userId, taskPin.taskId], set: { position } }), + .onConflictDoUpdate({ target: [taskPin.userId, taskPin.taskId], set: { position } }) + .returning({ txid: TXID }), ), }); /** The reader's pin of the plan taken away; every other member's stands. */ -const removePin = SqlSchema.void({ +const removePin = SqlSchema.findOneOption({ Request: TaskKeySchema, + Result: TxidRowSchema, execute: ({ userId, taskId }) => Database.use((db) => - db.delete(taskPin).where(and(eq(taskPin.userId, userId), eq(taskPin.taskId, taskId))), + db + .delete(taskPin) + .where(and(eq(taskPin.userId, userId), eq(taskPin.taskId, taskId))) + .returning({ txid: TXID }), ), }); @@ -530,9 +552,14 @@ function taskValues(write: typeof TaskInsertSchema.Type) { const insertTask = SqlSchema.findOne({ Request: TaskInsertSchema, - Result: TaskRowSchema, + Result: Schema.Struct({ ...TaskRowSchema.fields, txid: Schema.String }), execute: (write) => - Database.use((db) => db.insert(task).values(taskValues(write)).returning(TASK_COLUMNS)), + Database.use((db) => + db + .insert(task) + .values(taskValues(write)) + .returning({ ...TASK_COLUMNS, txid: TXID }), + ), }); /** A task inserted under its id where none stands under it yet; nothing where one does, written by nothing. */ @@ -702,10 +729,14 @@ const applyChanges = SqlSchema.findOne({ taskId: Schema.String, ...TaskColumnChangesSchema.fields, }), - Result: TaskRowSchema, + Result: Schema.Struct({ ...TaskRowSchema.fields, txid: Schema.String }), execute: ({ userId, taskId, ...changes }) => Database.use((db) => - db.update(task).set(changes).where(reachableTask(userId, taskId)).returning(TASK_COLUMNS), + db + .update(task) + .set(changes) + .where(reachableTask(userId, taskId)) + .returning({ ...TASK_COLUMNS, txid: TXID }), ), }); @@ -761,13 +792,13 @@ const insertTaskConversation = SqlSchema.findOne({ const removeTask = SqlSchema.findOneOption({ Request: TaskKeySchema, - Result: Schema.Struct({ conversationId: Schema.NullOr(Schema.String) }), + Result: Schema.Struct({ conversationId: Schema.NullOr(Schema.String), txid: Schema.String }), execute: ({ userId, taskId }) => Database.use((db) => db .delete(task) .where(reachableTask(userId, taskId)) - .returning({ conversationId: task.conversationId }), + .returning({ conversationId: task.conversationId, txid: TXID }), ), }); @@ -932,11 +963,25 @@ const insertOf = (userId: string, started: NewTask & { readonly id?: string }) = }; }); +/** A task as a write left it, and the transaction it committed in; none where it changed no row. */ +export interface TaskCommitted { + readonly task: A; + readonly txid?: string; +} + export function createTask(userId: string, started: NewTask): TaskStoreEffect { + return Effect.map(createTaskCommitted(userId, started), (created) => created.task); +} + +/** A task started, as `createTask` starts one, and the transaction its row was written in. */ +export function createTaskCommitted( + userId: string, + started: NewTask, +): TaskStoreEffect> { return Effect.flatMap(SqlClient.SqlClient, (client) => client.withTransaction( Effect.gen(function* () { - const row = yield* insertTask(yield* insertOf(userId, started)).pipe( + const { txid, ...row } = yield* insertTask(yield* insertOf(userId, started)).pipe( // An insert that returned no row is the database breaking its own contract, not an outcome. Effect.catchTag("NoSuchElementError", (missing) => Effect.die(missing)), ); @@ -950,7 +995,7 @@ export function createTask(userId: string, started: NewTask): TaskStoreEffect { +): TaskStoreEffect> { if (!pinned) return removePin({ userId, taskId }); return position === undefined ? insertPin({ userId, taskId, now }) @@ -1340,6 +1385,28 @@ export function updateTask( Option.Option, TaskStoreFailure | TaskArchived | TaskAgentUnstopped | E, DatabaseServices | R | RA +> { + return Effect.map( + updateTaskCommitted(userId, taskId, changes, moveGuard, archiveGuard), + Option.map((updated) => updated.task), + ); +} + +/** + * A task changed, as `updateTask` changes one, and the transaction the change + * committed in, read from the write that changed the row or the reader's pin; + * none where the change wrote neither. + */ +export function updateTaskCommitted( + userId: string, + taskId: string, + changes: TaskChanges, + moveGuard?: TaskMoveGuard, + archiveGuard?: TaskArchiveGuard, +): Effect.Effect< + Option.Option>, + TaskStoreFailure | TaskArchived | TaskAgentUnstopped | E, + DatabaseServices | R | RA > { return Effect.flatMap(SqlClient.SqlClient, (client) => client.withTransaction( @@ -1368,9 +1435,10 @@ export function updateTask( if (changes.archived === false && standing !== null) { yield* unarchivedAgents(taskId, standing); } - if (changes.pinned !== undefined) { - yield* pinAsAsked(userId, taskId, changes.pinned, changes.pinPosition, now); - } + const pinned = + changes.pinned === undefined + ? Option.none() + : yield* pinAsAsked(userId, taskId, changes.pinned, changes.pinPosition, now); const moved = changes.organizationId !== undefined && changes.organizationId !== locked.value.organizationId; @@ -1378,13 +1446,14 @@ export function updateTask( const movedNumber = moved ? yield* nextTaskNumber(changes.organizationId) : undefined; const columns = columnChangesOf(changes, locked.value, now, movedNumber); // A change that names only a pin writes nothing of the row, and an update with nothing to set is no statement. - const row = + const { txid: written, ...row } = Object.keys(columns).length === 0 - ? locked.value + ? { ...locked.value, txid: undefined } : yield* applyChanges({ userId, taskId, ...columns }).pipe( // The row is held under this transaction's lock, so an update that missed it is the database breaking its own contract. Effect.catchTag("NoSuchElementError", (missing) => Effect.die(missing)), ); + const txid = written ?? Option.getOrUndefined(pinned)?.txid; if (changes.organizationId !== undefined || changes.repository !== undefined) { yield* followingAutomations({ taskId, @@ -1401,7 +1470,8 @@ export function updateTask( if (changes.archived === true) { yield* handTaskCallsToSweep({ taskId, detachedAt: HANDED_TO_SWEEP_AT }); } - return Option.some(yield* taskOf(userId, row)); + const changed = yield* taskOf(userId, row); + return Option.some(txid === undefined ? { task: changed } : { task: changed, txid }); }), ), ); @@ -1473,11 +1543,24 @@ export function deleteTask( taskId: string, guard?: TaskDeleteGuard, ): Effect.Effect { + return Effect.map(deleteTaskCommitted(userId, taskId, guard), Option.isSome); +} + +/** A task deleted, as `deleteTask` deletes one, and the transaction its row went in; none where the account reached no such task. */ +export function deleteTaskCommitted( + userId: string, + taskId: string, + guard?: TaskDeleteGuard, +): Effect.Effect< + Option.Option<{ readonly txid: string }>, + TaskStoreFailure | TaskAgentUnstopped, + DatabaseServices | R +> { return Effect.flatMap(SqlClient.SqlClient, (client) => client.withTransaction( Effect.gen(function* () { const locked = yield* lockTask({ userId, taskId }); - if (Option.isNone(locked)) return false; + if (Option.isNone(locked)) return Option.none(); const agents = yield* findAgentConversations({ userId, taskId }); if (guard !== undefined && agents.some(({ id }) => !guard.stoppedAgentIds.has(id))) { return yield* new TaskAgentUnstopped(); @@ -1489,13 +1572,13 @@ export function deleteTask( ]; if (guard !== undefined) yield* guard.beforeDelete(conversationIds); const removed = yield* removeTask({ userId, taskId }); - if (Option.isNone(removed)) return false; + if (Option.isNone(removed)) return Option.none(); yield* removeEndedCalls({ userId, taskId }); if (conversationIds.length > 0) { const now = yield* DateTime.nowAsDate; yield* stampConversations({ conversationIds, now }); } - return true; + return Option.some({ txid: removed.value.txid }); }), ), ); diff --git a/packages/features/fixtures/json-schema/coding-agent-view-codingAgentAgentAnswerSchema.json b/packages/features/fixtures/json-schema/coding-agent-view-codingAgentAgentAnswerSchema.json index 2365836bbb..5e2f251f84 100644 --- a/packages/features/fixtures/json-schema/coding-agent-view-codingAgentAgentAnswerSchema.json +++ b/packages/features/fixtures/json-schema/coding-agent-view-codingAgentAgentAnswerSchema.json @@ -265,6 +265,9 @@ "turnId" ], "additionalProperties": false + }, + "txid": { + "type": "string" } }, "required": [ diff --git a/packages/features/fixtures/json-schema/coding-agent-wire-codingAgentAnswerSchema.json b/packages/features/fixtures/json-schema/coding-agent-wire-codingAgentAnswerSchema.json index 06befe5b7e..04717f65fd 100644 --- a/packages/features/fixtures/json-schema/coding-agent-wire-codingAgentAnswerSchema.json +++ b/packages/features/fixtures/json-schema/coding-agent-wire-codingAgentAnswerSchema.json @@ -263,6 +263,9 @@ "turnId" ], "additionalProperties": false + }, + "txid": { + "type": "string" } }, "required": [ diff --git a/packages/features/fixtures/json-schema/task-wire-taskAnswerSchema.json b/packages/features/fixtures/json-schema/task-wire-taskAnswerSchema.json index bc794a344d..54b8c7dcbb 100644 --- a/packages/features/fixtures/json-schema/task-wire-taskAnswerSchema.json +++ b/packages/features/fixtures/json-schema/task-wire-taskAnswerSchema.json @@ -283,6 +283,9 @@ "document" ], "additionalProperties": false + }, + "txid": { + "type": "string" } }, "required": [ diff --git a/packages/features/fixtures/json-schema/task-wire-taskDeleteAnswerSchema.json b/packages/features/fixtures/json-schema/task-wire-taskDeleteAnswerSchema.json index d9fd5b9d64..42c027145b 100644 --- a/packages/features/fixtures/json-schema/task-wire-taskDeleteAnswerSchema.json +++ b/packages/features/fixtures/json-schema/task-wire-taskDeleteAnswerSchema.json @@ -6,6 +6,9 @@ "enum": [ true ] + }, + "txid": { + "type": "string" } }, "required": [ diff --git a/packages/features/fixtures/json-schema/task-wire-taskStartAnswerSchema.json b/packages/features/fixtures/json-schema/task-wire-taskStartAnswerSchema.json index 22963f5a04..9b58e14f3b 100644 --- a/packages/features/fixtures/json-schema/task-wire-taskStartAnswerSchema.json +++ b/packages/features/fixtures/json-schema/task-wire-taskStartAnswerSchema.json @@ -558,6 +558,9 @@ "error" ], "additionalProperties": false + }, + "txid": { + "type": "string" } }, "required": [ diff --git a/packages/features/src/coding-agent-view.ts b/packages/features/src/coding-agent-view.ts index ff568f2df3..ec59156fe9 100644 --- a/packages/features/src/coding-agent-view.ts +++ b/packages/features/src/coding-agent-view.ts @@ -4,6 +4,7 @@ import { chatAnswerSchema, chatWriteAnswerSchema, chatWriteRequestSchema } from import { CODING_AGENT_ATTACHMENT_TYPE, CODING_AGENT_BOUNDS, + codingAgentAnswerSchema, codingAgentArchiveRequestSchema, codingAgentCallIdSchema, codingAgentChangesAnswerSchema, @@ -251,7 +252,10 @@ export type CodingAgentStartParams = typeof codingAgentStartParamsSchema.Type; /** An agent started, messaged, or stopped, as the window hears it: the agent as it then stands, or why it was not. */ export const codingAgentAgentAnswerSchema = EffectSchema.Union([ - EffectSchema.Struct({ agent: codingAgentSummarySchema }), + EffectSchema.Struct({ + agent: codingAgentSummarySchema, + txid: codingAgentAnswerSchema.fields.txid, + }), failedReaching, ]); diff --git a/packages/features/src/coding-agent-wire.ts b/packages/features/src/coding-agent-wire.ts index 647c3dc1c3..badcb527f0 100644 --- a/packages/features/src/coding-agent-wire.ts +++ b/packages/features/src/coding-agent-wire.ts @@ -700,7 +700,18 @@ export const codingAgentSummarySchema = EffectSchema.Struct({ export type CodingAgentSummary = typeof codingAgentSummarySchema.Type; /** A started, stopped, or changed agent (POST, PATCH). */ -export const codingAgentAnswerSchema = EffectSchema.Struct({ agent: codingAgentSummarySchema }); +/** + * The id of the transaction a write on the agent committed in, as Postgres + * counts it, in decimal: present where a Start, a rename, an archive, or a + * pin changed a row, so the desktop following the task list waits for it to + * arrive before it confirms the write, as Neon's mutations guide says. + */ +const codingAgentTxidSchema = EffectSchema.String.check(EffectSchema.isPattern(/^[0-9]+$/u)); + +export const codingAgentAnswerSchema = EffectSchema.Struct({ + agent: codingAgentSummarySchema, + txid: EffectSchema.optionalKey(codingAgentTxidSchema), +}); /** * The run of an automation that started an agent, as its page opens on it: diff --git a/packages/features/src/index.ts b/packages/features/src/index.ts index bc5029955d..a99a0c6a8a 100644 --- a/packages/features/src/index.ts +++ b/packages/features/src/index.ts @@ -149,6 +149,7 @@ export { type TaskCallResult, type TaskMarks, type TaskStarted, + type TaskWritten, } from "./task-client.js"; export { HostedTerminalClient, diff --git a/packages/features/src/task-client.test.ts b/packages/features/src/task-client.test.ts index 42adad2c3d..6167d727e4 100644 --- a/packages/features/src/task-client.test.ts +++ b/packages/features/src/task-client.test.ts @@ -262,7 +262,7 @@ it.effect( }); assert.deepEqual(moved, { ok: false, failure: TASK_CALL_FAILURE.TASK_ARCHIVED }); assert.equal(renamed, undefined); - assert.equal(deleted, false); + assert.equal(deleted, undefined); assert.deepEqual( api.requests().map((request) => JSON.parse(request.body ?? "null")), [ diff --git a/packages/features/src/task-client.ts b/packages/features/src/task-client.ts index 084f78614f..17c2d3fbbf 100644 --- a/packages/features/src/task-client.ts +++ b/packages/features/src/task-client.ts @@ -123,6 +123,14 @@ import type { TaskTranscript } from "./transcript-wire.js"; export interface TaskStarted { readonly task: Task; readonly agent?: CodingAgentAgentAnswer; + /** The transaction the task was written in, which the task list followed live carries once it lands. */ + readonly txid?: string; +} + +/** A task as a write answered it, and the transaction the write committed in where it changed a row. */ +export interface TaskWritten { + readonly task: Task; + readonly txid?: string; } /** One call's answer, or why there is none. */ @@ -413,7 +421,11 @@ export class HostedTaskClient { setRepository( taskId: string, repository: string | null, - ): Effect.Effect, never, HttpClient.HttpClient> { + ): Effect.Effect< + TaskCallResult, + never, + HttpClient.HttpClient + > { const admitted = Result.getOrUndefined(readEither(taskUpdateRequestSchema)({ repository })); if (admitted === undefined) return Effect.succeed(UNANSWERED); return hostedResult(this.#update(taskId, admitted), REPOSITORY_REFUSALS); @@ -428,7 +440,7 @@ export class HostedTaskClient { move( taskId: string, workspaceId: string, - ): Effect.Effect, never, HttpClient.HttpClient> { + ): Effect.Effect, never, HttpClient.HttpClient> { const admitted = Result.getOrUndefined(readEither(taskUpdateRequestSchema)({ workspaceId })); if (admitted === undefined) return Effect.succeed(UNANSWERED); return hostedResult(this.#update(taskId, admitted), MOVE_REFUSALS); @@ -803,11 +815,13 @@ export class HostedTaskClient { ); } - /** Deletes one task with its document and its conversation; whether the service deleted it. */ - delete(taskId: string): Effect.Effect { + /** Deletes one task with its document and its conversation; the transaction it went in, or nothing where the service deleted nothing. */ + delete( + taskId: string, + ): Effect.Effect<{ readonly txid?: string } | undefined, never, HttpClient.HttpClient> { return Effect.match(askHosted(this.#hosted.plans.delete, { id: taskId }), { - onFailure: () => false, - onSuccess: (answer) => answer.deleted, + onFailure: () => undefined, + onSuccess: (answer) => (answer.txid === undefined ? {} : { txid: answer.txid }), }); } @@ -819,7 +833,7 @@ export class HostedTaskClient { rename( taskId: string, request: TaskRenameRequest, - ): Effect.Effect { + ): Effect.Effect { const admitted = Result.getOrUndefined(readEither(taskRenameRequestSchema)(request)); if (admitted === undefined) return Effect.succeed(undefined); return orNothing(this.#update(taskId, admitted)); @@ -833,7 +847,7 @@ export class HostedTaskClient { setStanding( taskId: string, standing: TaskStanding, - ): Effect.Effect { + ): Effect.Effect { const admitted = Result.getOrUndefined(readEither(taskUpdateRequestSchema)(standing)); if (admitted === undefined) return Effect.succeed(undefined); return orNothing(this.#update(taskId, admitted)); @@ -847,7 +861,7 @@ export class HostedTaskClient { mark( taskId: string, marks: TaskMarks, - ): Effect.Effect { + ): Effect.Effect { const admitted = Result.getOrUndefined(readEither(taskUpdateRequestSchema)(marks)); if (admitted === undefined) return Effect.succeed(undefined); return orNothing(this.#update(taskId, admitted)); @@ -857,10 +871,13 @@ export class HostedTaskClient { #update( taskId: string, changes: TaskUpdateRequest, - ): Effect.Effect { + ): Effect.Effect { return Effect.map( askHosted(this.#hosted.plans.update, { params: { id: taskId }, body: changes }), - (changed) => changed.plan, + (changed) => + changed.txid === undefined + ? { task: changed.plan } + : { task: changed.plan, txid: changed.txid }, ); } @@ -939,11 +956,14 @@ export class HostedTaskClient { /** A started task as the service answered it, with the agent started on it, or why it was not, where one was asked for. */ function taskStartedOf(started: TaskStartAnswer): TaskStarted { - if (started.agent !== undefined) return { task: started.plan, agent: { agent: started.agent } }; + const written = started.txid === undefined ? {} : { txid: started.txid }; + if (started.agent !== undefined) { + return { task: started.plan, agent: { agent: started.agent }, ...written }; + } if (started.agentRefusal !== undefined) { - return { task: started.plan, agent: startRefusedAs(started.agentRefusal) }; + return { task: started.plan, agent: startRefusedAs(started.agentRefusal), ...written }; } - return { task: started.plan }; + return { task: started.plan, ...written }; } /** A comment call's answer, or why there was none: not found where the service said so, unanswered for anything else. */ diff --git a/packages/features/src/task-wire.ts b/packages/features/src/task-wire.ts index 9dac1e5fb5..83637c9278 100644 --- a/packages/features/src/task-wire.ts +++ b/packages/features/src/task-wire.ts @@ -603,8 +603,19 @@ export const taskListAnswerSchema = EffectSchema.Struct({ export type TaskListAnswer = typeof taskListAnswerSchema.Type; -/** A started, opened, or changed task (POST, GET, PATCH), with its document as saved. */ -export const taskAnswerSchema = EffectSchema.Struct({ plan: taskSchema }); +/** + * The id of the transaction a write committed in, as Postgres counts it, in + * decimal: present where the write changed a row, so the desktop following + * the task list waits for it to arrive before it confirms the write, and + * absent where nothing changed, as Neon's mutations guide says. + */ +const taskTxidSchema = EffectSchema.String.check(EffectSchema.isPattern(/^[0-9]+$/u)); + +/** A started, opened, or changed task (POST, GET, PATCH), with its document as saved, and the transaction a write committed in. */ +export const taskAnswerSchema = EffectSchema.Struct({ + plan: taskSchema, + txid: EffectSchema.optionalKey(taskTxidSchema), +}); /** * Why the agent a start asked for was not started beside the task it made, @@ -626,12 +637,16 @@ export const taskStartAnswerSchema = EffectSchema.Struct({ plan: taskSchema, agent: EffectSchema.optionalKey(codingAgentSummarySchema), agentRefusal: EffectSchema.optionalKey(taskAgentRefusalSchema), + txid: EffectSchema.optionalKey(taskTxidSchema), }); export type TaskStartAnswer = typeof taskStartAnswerSchema.Type; /** A deleted task (DELETE): its row, its document, and its association are gone. */ -export const taskDeleteAnswerSchema = EffectSchema.Struct({ deleted: EffectSchema.Literal(true) }); +export const taskDeleteAnswerSchema = EffectSchema.Struct({ + deleted: EffectSchema.Literal(true), + txid: EffectSchema.optionalKey(taskTxidSchema), +}); /** How a plan's history is read: the newest revisions first, a page at a time. */ export const PLAN_HISTORY_BOUNDS = { diff --git a/packages/host/src/compose-tasks.test.ts b/packages/host/src/compose-tasks.test.ts index 5bd87b9a05..145491b690 100644 --- a/packages/host/src/compose-tasks.test.ts +++ b/packages/host/src/compose-tasks.test.ts @@ -8,6 +8,7 @@ import { type GitHubRepositoryPullRequestAnswer, type TaskCallResult, type TaskStarted, + type TaskWritten, VOICE_SERVICE_FRAME, WORKSPACE_CALL_FAILURE, } from "@luke/features"; @@ -819,7 +820,7 @@ function fakeService(tasks: Task[]): FakeService { }), // The service keeps the repository on the task and answers the task whole. setRepository: (taskId, repository) => - Effect.sync((): TaskCallResult => { + Effect.sync((): TaskCallResult => { if (service.repositoryUnreachable) { return { ok: false, failure: TASK_CALL_FAILURE.REPOSITORY_NOT_REACHABLE }; } @@ -829,18 +830,18 @@ function fakeService(tasks: Task[]): FakeService { service.tasks = service.tasks.map((candidate) => candidate.id === taskId ? changed : candidate, ); - return { ok: true, answer: changed }; + return { ok: true, answer: { task: changed } }; }), // The service moves the task and answers it whole. move: (taskId, workspaceId) => - Effect.sync((): TaskCallResult => { + Effect.sync((): TaskCallResult => { const found = service.tasks.find((candidate) => candidate.id === taskId); if (found === undefined) return { ok: false, failure: TASK_CALL_FAILURE.NOT_FOUND }; const moved = { ...found, workspaceId }; service.tasks = service.tasks.map((candidate) => candidate.id === taskId ? moved : candidate, ); - return { ok: true, answer: moved }; + return { ok: true, answer: { task: moved } }; }), // The service re-titles the document with the name, as the store does. rename: (taskId, { name }) => @@ -851,7 +852,7 @@ function fakeService(tasks: Task[]): FakeService { service.tasks = service.tasks.map((candidate) => candidate.id === taskId ? renamed : candidate, ); - return renamed; + return { task: renamed }; }), // The service sets the status and the priority named, and leaves the other standing. setStanding: (taskId, standing) => @@ -862,7 +863,7 @@ function fakeService(tasks: Task[]): FakeService { service.tasks = service.tasks.map((candidate) => candidate.id === taskId ? changed : candidate, ); - return changed; + return { task: changed }; }), // The service stamps a pin or an archive at its own clock, and a false takes it away. mark: (taskId, { pinned, archived }) => @@ -880,13 +881,13 @@ function fakeService(tasks: Task[]): FakeService { service.tasks = service.tasks.map((candidate) => candidate.id === taskId ? marked : candidate, ); - return marked; + return { task: marked }; }), delete: (taskId) => Effect.sync(() => { - if (service.deleteFails) return false; + if (service.deleteFails) return undefined; service.tasks = service.tasks.filter((candidate) => candidate.id !== taskId); - return true; + return {}; }), }; return service; diff --git a/packages/host/src/compose-tasks.ts b/packages/host/src/compose-tasks.ts index 5b2c202ea0..84cb495305 100644 --- a/packages/host/src/compose-tasks.ts +++ b/packages/host/src/compose-tasks.ts @@ -1351,7 +1351,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 }; - takeMarks(marked); + takeMarks(marked.task); return { changed: true }; }), ); @@ -1849,7 +1849,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct return yield* serial( Effect.gen(function* () { const deleted = yield* served(client.delete(taskId)); - if (!deleted) return { deleted: false }; + if (deleted === undefined) return { deleted: false }; const tasks = view.tasks.filter((task) => task.id !== taskId); if (view.activeTaskId === taskId) { yield* FiberHandle.clear(conversationRead); @@ -1873,8 +1873,9 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct const { taskId, name } = read.success; return yield* serial( Effect.gen(function* () { - const renamed = yield* served(client.rename(taskId, { name })); - if (renamed === undefined) return { renamed: false }; + const written = yield* served(client.rename(taskId, { name })); + if (written === undefined) return { renamed: false }; + const renamed = written.task; const tasks = view.tasks.map((task) => task.id === taskId ? { ...task, name: renamed.name } : task, ); @@ -1908,7 +1909,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 }; - takeAnswered(changed); + takeAnswered(changed.task); return { changed: true }; }), ); @@ -2007,8 +2008,10 @@ 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 }); - takeAnswered(changed.answer); - return carried({ repository: changed.answer.repository }); + takeAnswered(changed.answer.task); + return carried({ + repository: changed.answer.task.repository, + }); }), ); }), @@ -2028,8 +2031,8 @@ 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 }); - takeAnswered(moved.answer); - return carried({ workspaceId: moved.answer.workspaceId }); + takeAnswered(moved.answer.task); + return carried({ workspaceId: moved.answer.task.workspaceId }); }), ); }), diff --git a/packages/service/src/agents/agents-procedures.ts b/packages/service/src/agents/agents-procedures.ts index dc244729be..69a4d7cc3e 100644 --- a/packages/service/src/agents/agents-procedures.ts +++ b/packages/service/src/agents/agents-procedures.ts @@ -29,13 +29,14 @@ import { randomUUID } from "node:crypto"; import type { DatabaseServices } from "@luke/database/core/query"; import { - archiveCodingAgent, + archiveCodingAgentCommitted, + type CodingAgentCommitted, type CodingAgentLine, hasRepository, listCodingAgentLines, listWorkspaceCodingAgentLines, - pinCodingAgent, - renameCodingAgent, + pinCodingAgentCommitted, + renameCodingAgentCommitted, updateCodingAgentChoice, updateCodingAgentServers, } from "@luke/database/models/coding-agent-store"; @@ -147,6 +148,22 @@ const changedAgent = /* @__PURE__ */ Effect.fnUntraced(function* ( return { agent: yield* summaryOf(userId, changed.value) }; }); +/** + * An agent as a write left it, answered with the transaction the write + * committed in where it changed a row, as Neon's mutations guide says, so a + * desktop following the task list waits for it to arrive before it + * confirms the write. + */ +const committedAgent = /* @__PURE__ */ Effect.fnUntraced(function* ( + userId: string, + changed: Option.Option, +): Effect.fn.Return<{ agent: CodingAgentSummary; txid?: string }, HostedRefusal, DatabaseServices> { + if (Option.isNone(changed)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); + const agent = yield* summaryOf(userId, changed.value.agent); + const { txid } = changed.value; + return txid === undefined ? { agent } : { agent, txid }; +}); + /** * The agent the path names, still standing, and the eve session its sandbox * is tagged with, or none where the agent has not run yet. An archived agent @@ -268,7 +285,7 @@ export function agentsProcedures(seams: AgentsProcedureSeams) { productionSummaryModel(owner, chosen, effort).pipe( Effect.provideService(HostedEnvironment, environment), )); - const { agent, created } = yield* startRequestedAgent( + const { agent, created, txid } = yield* startRequestedAgent( eveFor(seams, context.request, context.userId), context.userId, taskId, @@ -277,7 +294,7 @@ export function agentsProcedures(seams: AgentsProcedureSeams) { ); return { status: created ? AGENT_START_STATUS.CREATED : AGENT_START_STATUS.STARTED_BEFORE, - body: { agent }, + body: txid === undefined ? { agent } : { agent, txid }, }; }), /** The agent as its own page, an agent with no task as much as one on a task. */ @@ -315,10 +332,10 @@ export function agentsProcedures(seams: AgentsProcedureSeams) { const { body } = input; if ("title" in body) { const { title } = body; - return yield* changedAgent( + return yield* committedAgent( userId, yield* hostedStoreOrUnavailable( - renameCodingAgent(userId, agent.id, title.length === 0 ? null : title), + renameCodingAgentCommitted(userId, agent.id, title.length === 0 ? null : title), ), ); } @@ -327,16 +344,18 @@ export function agentsProcedures(seams: AgentsProcedureSeams) { // An archived agent works on unseen, so a turn still running is stopped before the stamp. yield* stoppedForGood(seams, writer, context.request, userId, agent); } - return yield* changedAgent( + return yield* committedAgent( userId, - yield* hostedStoreOrUnavailable(archiveCodingAgent(userId, agent.id, body.archived)), + yield* hostedStoreOrUnavailable( + archiveCodingAgentCommitted(userId, agent.id, body.archived), + ), ); } if ("pinned" in body) { - return yield* changedAgent( + return yield* committedAgent( userId, yield* hostedStoreOrUnavailable( - pinCodingAgent(userId, agent.id, body.pinned, body.position), + pinCodingAgentCommitted(userId, agent.id, body.pinned, body.position), ), ); } diff --git a/packages/service/src/agents/coder-host/start.ts b/packages/service/src/agents/coder-host/start.ts index 5752300493..bd6d30f29f 100644 --- a/packages/service/src/agents/coder-host/start.ts +++ b/packages/service/src/agents/coder-host/start.ts @@ -368,6 +368,8 @@ export type StartAsk = StartPlace & { export interface CodingAgentStart { readonly agent: CodingAgent; readonly created: boolean; + /** The transaction the agent's row was written in, where this Start wrote it. */ + readonly txid?: string; } /** What a Start reaches: the store, the catalog, GitHub through the App on the developer's standing, and the network to eve. */ @@ -816,7 +818,7 @@ export const startCodingAgent = /* @__PURE__ */ Effect.fn("web/startCodingAgent" }), ); if (Option.isNone(started)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); - const { agent, created } = started.value; + const { agent, created, txid } = started.value; if (!created) return { agent, created }; // The session opens with the plan or the prompt as its first message; an eve that did not take // it leaves no agent behind, so the retry under the same key starts afresh. @@ -885,13 +887,15 @@ export const startCodingAgent = /* @__PURE__ */ Effect.fn("web/startCodingAgent" is_from_automation: ask.automationId !== undefined && ask.automationId !== null, has_task: ask.taskId !== null, }); - return { agent, created }; + return txid === undefined ? { agent, created } : { agent, created, txid }; }); /** A Start as a request reached the service, answered: the agent's summary, and whether this request made it or found it under its key. */ export interface RequestedStart { readonly agent: CodingAgentSummary; readonly created: boolean; + /** The transaction the agent's row was written in, where this request wrote it. */ + readonly txid?: string; } /** @@ -922,7 +926,7 @@ export const startRequestedAgent = /* @__PURE__ */ Effect.fnUntraced(function* ( model !== undefined && effort !== undefined ? { model, effort, ...(contextWindow === undefined ? undefined : { contextWindow }) } : undefined; - const { agent, created } = yield* startCodingAgent(eve, userId, { + const { agent, created, txid } = yield* startCodingAgent(eve, userId, { taskId, idempotencyKey: request.idempotencyKey, choice, @@ -943,5 +947,6 @@ export const startRequestedAgent = /* @__PURE__ */ Effect.fnUntraced(function* ( names, ), created, + ...(txid === undefined ? undefined : { txid }), }; }); diff --git a/packages/service/src/agents/coding-agents-app.test.ts b/packages/service/src/agents/coding-agents-app.test.ts index 53b27819b9..885f6b591b 100644 --- a/packages/service/src/agents/coding-agents-app.test.ts +++ b/packages/service/src/agents/coding-agents-app.test.ts @@ -172,6 +172,7 @@ import { REPOSITORY_UNREACHED } from "@luke/features/repository-reach-wire"; import { TASK_LIST_ARCHIVED, TASK_STATUS, + taskDeleteAnswerSchema, taskListAnswerSchema, taskStartAnswerSchema, } from "@luke/features/task-wire"; @@ -4075,7 +4076,10 @@ it.layer(getTestDB())("messaging a coding agent", (it) => { }), ); - assert.deepEqual(deleted, { status: HOSTED_HTTP_STATUS.OK, body: { deleted: true } }); + assert.equal( + readAnswer(taskDeleteAnswerSchema, HOSTED_HTTP_STATUS.OK, deleted).deleted, + true, + ); assert.deepEqual(eve.cancelled, [[sessionId, "turn_0"]]); assert.deepEqual(eve.cancelledAs, [deploymentFor(owner)]); assert.deepEqual(eve.reset.toSorted(), [taskSession, sessionId].toSorted()); @@ -4149,7 +4153,10 @@ it.layer(getTestDB())("messaging a coding agent", (it) => { [agentStop.status, lukeStop.status], [HOSTED_HTTP_STATUS.OK, HOSTED_HTTP_STATUS.OK], ); - assert.deepEqual(deleted, { status: HOSTED_HTTP_STATUS.OK, body: { deleted: true } }); + assert.equal( + readAnswer(taskDeleteAnswerSchema, HOSTED_HTTP_STATUS.OK, deleted).deleted, + true, + ); assert.deepEqual(stopped.eve.cancelled, [ [stopped.sessionId, "turn_0"], [taskSession, "turn_0"], diff --git a/packages/service/src/tasks/task-list-live.test.ts b/packages/service/src/tasks/task-list-live.test.ts index c0a559bdef..2f93c3db0c 100644 --- a/packages/service/src/tasks/task-list-live.test.ts +++ b/packages/service/src/tasks/task-list-live.test.ts @@ -9,7 +9,7 @@ import { pinCodingAgent, recordCodingAgentPublished, } from "@luke/database/models/coding-agent-store"; -import { listTasks, updateTask } from "@luke/database/models/task-store"; +import { listTasks, updateTask, updateTaskCommitted } from "@luke/database/models/task-store"; import { user } from "@luke/database/schemas/auth"; import { codingAgent } from "@luke/database/schemas/coding-agent"; import { messages, turns } from "@luke/database/schemas/storage"; @@ -198,6 +198,31 @@ const PUBLISHED = { }; it.layer(getTestDB())("the plan list's live statements", (it) => { + it.effect( + "a write that changes the task answers the transaction it committed in, and one that changes nothing answers none", + () => + Effect.gen(function* () { + const owner = yield* openUser; + const plan = yield* startTask(owner, { name: "Teammate invitations" }); + + const renamed = Option.getOrThrow( + yield* updateTaskCommitted(owner, plan.id, { name: "Invitations" }), + ); + const pinned = Option.getOrThrow( + yield* updateTaskCommitted(owner, plan.id, { pinned: true }), + ); + const again = Option.getOrThrow( + yield* updateTaskCommitted(owner, plan.id, { pinned: true }), + ); + + assert.match(renamed.txid ?? "", /^[0-9]+$/u); + assert.match(pinned.txid ?? "", /^[0-9]+$/u); + assert.notEqual(renamed.txid, pinned.txid, "each write is its own transaction"); + assert.equal(again.txid, undefined, "a pin asked again writes nothing to wait for"); + assert.equal(renamed.task.name, "Invitations"); + }), + ); + it.effect("call only functions Postgres marks immutable, as Realtime requires", () => Effect.gen(function* () { const { plans, agents } = Option.getOrThrow(taskListLiveStatements("user-immutable")); diff --git a/packages/service/src/tasks/tasks-procedures.ts b/packages/service/src/tasks/tasks-procedures.ts index 0eb61a2934..0faac78bc2 100644 --- a/packages/service/src/tasks/tasks-procedures.ts +++ b/packages/service/src/tasks/tasks-procedures.ts @@ -60,13 +60,13 @@ import { unlinkTaskIssue, } from "@luke/database/models/task-issue-store"; import { - createTask, - deleteTask, + createTaskCommitted, + deleteTaskCommitted, listTasks, readTask, readTaskHeader, type TaskHeader, - updateTask, + updateTaskCommitted, } from "@luke/database/models/task-store"; import { readTranscript } from "@luke/database/models/transcript-store"; import type { WorkspaceSeams } from "@luke/database/models/workspace-seams"; @@ -467,8 +467,8 @@ export function tasksProcedures(seams: TasksProcedureSeams) { ); // An agent is the start, so its plan opens with no planning phase. const planning = agent === undefined ? input.planning : false; - const task = yield* hostedStoreOrUnavailable( - createTask(userId, { + const { task, txid } = yield* hostedStoreOrUnavailable( + createTaskCommitted(userId, { organizationId, ...(input.name === undefined ? undefined : { name: input.name }), repository: reached.repository, @@ -488,7 +488,10 @@ export function tasksProcedures(seams: TasksProcedureSeams) { agent === undefined ? { plan: task } : yield* startedAgentOn(seams, request, userId, task, agent); - return { status: TASK_START_STATUS.CREATED, body: answer }; + return { + status: TASK_START_STATUS.CREATED, + body: txid === undefined ? answer : { ...answer, txid }, + }; }), /** One plan with its saved document. */ @@ -554,7 +557,7 @@ export function tasksProcedures(seams: TasksProcedureSeams) { { discard: true }, ); } - const task = yield* updateTask( + const task = yield* updateTaskCommitted( userId, taskId, { @@ -581,7 +584,8 @@ export function tasksProcedures(seams: TasksProcedureSeams) { ), ); if (Option.isNone(task)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); - return { plan: task.value }; + const { task: plan, txid } = task.value; + return txid === undefined ? { plan } : { plan, txid }; }), /** @@ -615,7 +619,7 @@ export function tasksProcedures(seams: TasksProcedureSeams) { ); const stopped = new Set(agents.map(({ id }) => id)); const deleted = yield* hostedStoreOrUnavailable( - deleteTask(userId, taskId, { + deleteTaskCommitted(userId, taskId, { stoppedAgentIds: stopped, // Note that the sessions are retired under the plan's lock and before its conversations are // cleared: eve's door admits no one to a cleared conversation's session, and a delete an @@ -630,8 +634,8 @@ export function tasksProcedures(seams: TasksProcedureSeams) { ); // An agent started while the others were stopped is one this delete has not stopped yet. if (deleted === undefined) return yield* Effect.fail(HOSTED_REFUSAL.AGENT_NOT_READY); - if (!deleted) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); - return { deleted: true as const }; + if (Option.isNone(deleted)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); + return { deleted: true as const, txid: deleted.value.txid }; }), /** The plan's board, with Luke's latest drawing. */