diff --git a/packages/features/src/contract/live.ts b/packages/features/src/contract/live.ts index 8bce22a796..64bd3131a1 100644 --- a/packages/features/src/contract/live.ts +++ b/packages/features/src/contract/live.ts @@ -4,7 +4,7 @@ * What waits on the developer across the account, as the address to * subscribe at and one sealed statement per part of the account inbox's * answer; and one task's whiteboard, named by `?id=`, as the statement that - * marks where it stands; and one coding agent's transcript, or one of its + * marks where it stands, and its pages and calls the same way; and one coding agent's transcript, or one of its * workers' under `&call=`, as the statements its pages are read from. Each * asked with `?poll=true` answers each statement's rows inline instead, for * a reader polling while Realtime fails or is off; with `&after=` too, the @@ -23,6 +23,7 @@ import { LIVE_POLL, LIVE_USER_QUERY, taskBoardLiveAnswerSchema, + taskRecordLiveAnswerSchema, } from "../live-wire.js"; import { HOSTED_SERVICE_PATH } from "../service-paths.js"; import { hostedProcedure } from "./base.js"; @@ -69,6 +70,10 @@ export const liveContract = { .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_TASK_BOARD })) .input(liveTaskSchema) .output(EffectSchema.flip(taskBoardLiveAnswerSchema)), + taskRecord: hostedProcedure + .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_TASK_RECORD })) + .input(liveTaskSchema) + .output(EffectSchema.flip(taskRecordLiveAnswerSchema)), agentTranscript: hostedProcedure .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_AGENT_TRANSCRIPT })) .input(liveTranscriptSchema) diff --git a/packages/features/src/live-wire.ts b/packages/features/src/live-wire.ts index c9bfb8a41d..4684d358bf 100644 --- a/packages/features/src/live-wire.ts +++ b/packages/features/src/live-wire.ts @@ -281,6 +281,54 @@ export const taskBoardLiveRowsSchema = EffectSchema.Struct({ export type TaskBoardLiveRows = typeof taskBoardLiveRowsSchema.Type; +/** + * An open task's pages and calls sealed for following: the address, and one + * statement a part, each a mark that moves as its rows do and carries none + * of their words: the pages agents showed (`pages`) and the calls about the + * task (`calls`). The desktop reads a part again once its mark moves, + * rather than when a call's activity or end suggests it might have; the + * board is followed on its own (`taskBoardLiveAnswerSchema`). The service + * answers `not-found` for a task the account does not reach, and + * `unavailable` where its deployment's Realtime is misconfigured, which the + * desktop answers by asking again. + */ +export const taskRecordLiveAnswerSchema = EffectSchema.Struct({ + url: EffectSchema.String, + queries: EffectSchema.Struct({ + pages: sealedLiveQuerySchema, + calls: sealedLiveQuerySchema, + }), +}); + +export type TaskRecordLiveAnswer = typeof taskRecordLiveAnswerSchema.Type; + +/** One page shown on the task as its row delivers it: its newest version's number. */ +const taskPageLiveRowSchema = EffectSchema.Struct({ + pageId: EffectSchema.String, + version: EffectSchema.Finite, +}); + +/** One call about the task as its row delivers it: when and how it ended, null while it stands; what it metered and its model; its last fragment; and how many rows the voice wrote for it. */ +const taskCallLiveRowSchema = EffectSchema.Struct({ + id: EffectSchema.String, + closedAt: EffectSchema.NullOr(EffectSchema.Finite), + closeReason: EffectSchema.NullOr(EffectSchema.String), + usage: EffectSchema.NullOr( + EffectSchema.Struct({ seconds: EffectSchema.Finite, confirmed: EffectSchema.Boolean }), + ), + model: EffectSchema.NullOr(EffectSchema.String), + lastSegment: EffectSchema.Finite, + spoken: EffectSchema.Finite, +}); + +/** The task's pages and calls as Realtime delivers them, a part's rows each: none of any for a task the account no longer reaches. */ +export const taskRecordLiveRowsSchema = EffectSchema.Struct({ + pages: EffectSchema.Array(taskPageLiveRowSchema), + calls: EffectSchema.Array(taskCallLiveRowSchema), +}); + +export type TaskRecordLiveRows = typeof taskRecordLiveRowsSchema.Type; + /** * A coding agent's transcript, or one of its workers', sealed for * following: the address, and the statements a page past the reader's diff --git a/packages/features/src/service-paths.ts b/packages/features/src/service-paths.ts index ef32ec1514..e604bbbe2e 100644 --- a/packages/features/src/service-paths.ts +++ b/packages/features/src/service-paths.ts @@ -86,6 +86,8 @@ export const HOSTED_SERVICE_PATH = { LIVE_INBOX: "/api/live/inbox", /** One task's whiteboard under `?id=`, sealed for following over Neon Realtime (GET): where it stands, or `not-found` or `unavailable`. */ LIVE_TASK_BOARD: "/api/live/task-board", + /** One task's pages and calls under `?id=`, each part's mark sealed for following over Neon Realtime (GET), or `not-found` or `unavailable`. */ + LIVE_TASK_RECORD: "/api/live/task-record", /** One agent's transcript under `?id=`, or one of its workers' under `&call=`, sealed for following over Neon Realtime (GET): what its pages are read from, or `not-found` or `unavailable`. */ LIVE_AGENT_TRANSCRIPT: "/api/live/agent-transcript", /** diff --git a/packages/features/src/task-client.ts b/packages/features/src/task-client.ts index 11174a2bf5..7ebb65e728 100644 --- a/packages/features/src/task-client.ts +++ b/packages/features/src/task-client.ts @@ -35,7 +35,12 @@ import { hostedClient, } from "./hosted-link.js"; import { imageBodyOf } from "./image-body.js"; -import { LIVE_SIGNED_OUT, LIVE_USER_QUERY, type TaskBoardLiveAnswer } from "./live-wire.js"; +import { + LIVE_SIGNED_OUT, + LIVE_USER_QUERY, + type TaskBoardLiveAnswer, + type TaskRecordLiveAnswer, +} from "./live-wire.js"; import type { PageLine, PageVersion } from "./page-wire.js"; import { PLAN_ATTACHMENT_REFUSAL } from "./plan-attachment-wire.js"; import { @@ -440,6 +445,29 @@ export class HostedTaskClient { ); } + /** + * Where one task's pages and calls stand, sealed for following over Neon + * Realtime, or with `poll` their rows inline, for a reader polling while + * Realtime fails them, for the account `userId` names: not found where the + * account reaches no such task, signed out where the session is another's, + * unanswered for anything else, which the caller answers by sealing again. + */ + recordLive( + taskId: string, + userId: string, + poll = false, + after?: string, + ): Effect.Effect< + TaskCallResult, + never, + HttpClient.HttpClient + > { + return hostedResult( + askLive(this.#hosted.live.taskRecord, { id: taskId, [LIVE_USER_QUERY]: userId }, poll, after), + LIVE_REFUSALS, + ); + } + /** What was said on one task's calls; nothing where the service did not answer it. */ readTranscript( taskId: string, diff --git a/packages/host/src/compose-tasks.test.ts b/packages/host/src/compose-tasks.test.ts index 57be7fd0bc..0bbc6e031c 100644 --- a/packages/host/src/compose-tasks.test.ts +++ b/packages/host/src/compose-tasks.test.ts @@ -34,7 +34,7 @@ import { type RepositoryIssuesAsk, } from "@luke/features/github-repositories-wire"; import type { LinearIssueSummary } from "@luke/features/linear-wire"; -import type { TaskBoardLiveAnswer } from "@luke/features/live-wire"; +import type { TaskBoardLiveAnswer, TaskRecordLiveAnswer } from "@luke/features/live-wire"; import type { PageLine, PageVersion } from "@luke/features/page-wire"; import { PICTURE_TYPE, type PictureType } from "@luke/features/picture-wire"; import { @@ -165,6 +165,10 @@ interface FakeService extends TasksClient { boards: Record; /** Each task's board sealed for following, by task id; a task with none is not found, as the service answers one the account does not reach. */ boardSeals: Record; + /** Each task's pages and calls sealed for following, by task id; a task with none is not found, as the service answers one the account does not reach. */ + recordSeals: Record; + /** Every read of a task's pages or transcript, in order, so a test can see which a followed mark moved. */ + readonly recordReads: string[]; /** The image each task's last save carried, by task id, as the service keeps it beside the board. */ readonly images: Record; /** Each task's transcript, by task id; a task with none answers no transcript, as a service that did not answer. */ @@ -352,6 +356,8 @@ function fakeService(tasks: Task[]): FakeService { reads: [], boards: {}, boardSeals: {}, + recordSeals: {}, + recordReads: [], images: {}, transcripts: {}, works: {}, @@ -376,7 +382,11 @@ function fakeService(tasks: Task[]): FakeService { readRevision: (_taskId, seq) => Effect.sync(() => service.revisions[seq]), pages: [], pageLines: {}, - readPages: (taskId) => Effect.sync(() => service.pageLines[taskId]), + readPages: (taskId) => + Effect.sync(() => { + service.recordReads.push("pages"); + return service.pageLines[taskId]; + }), pictures: new Map(), // The service keeps a picture under the next id, and refuses one more where the plan holds two, a bound small enough to reach. keepPicture: (taskId, picture) => @@ -771,7 +781,18 @@ function fakeService(tasks: Task[]): FakeService { ? { ok: false, failure: TASK_CALL_FAILURE.NOT_FOUND } : { ok: true, answer: sealed }; }), - readTranscript: (taskId) => Effect.sync(() => service.transcripts[taskId]), + recordLive: (taskId) => + Effect.sync(() => { + const sealed = service.recordSeals[taskId]; + return sealed === undefined + ? { ok: false, failure: TASK_CALL_FAILURE.NOT_FOUND } + : { ok: true, answer: sealed }; + }), + readTranscript: (taskId) => + Effect.sync(() => { + service.recordReads.push("transcript"); + return service.transcripts[taskId]; + }), readWork: (taskId, before) => Effect.sync(() => { service.workReads.push(before === undefined ? taskId : `${taskId}:${before}`); @@ -3420,6 +3441,100 @@ it.effect( }), ); +/** A task's pages and calls sealed for following, named by the round they were sealed in. */ +function sealedRecord(round: number): TaskRecordLiveAnswer { + const query = (part: string) => ({ + capability: `${part}-${round}`, + queryFingerprint: part, + expiresAt: 3_600_000, + }); + return { + url: "wss://realtime.example/v1", + queries: { pages: query("pages"), calls: query("calls") }, + }; +} + +/** One call about the task as its row delivers it, ended at `closedAt` or standing. */ +function callRow(lastSegment: number, closedAt: number | null) { + return { + id: "call-1", + closedAt, + closeReason: closedAt === null ? null : "remote_hangup", + usage: null, + model: "gpt-realtime", + lastSegment, + spoken: 0, + }; +} + +it.effect( + "the open task's pages are read again as their mark moves and its transcript once an ended call's does, not on the call's activity, while its end reads the work now and once more after its writes settle", + () => + Effect.gen(function* () { + const service = fakeService([task(INVITES, "Teammate invitations", "# Invites", 10)]); + service.works[INVITES] = { turns: [] }; + service.recordSeals[INVITES] = sealedRecord(0); + const realtime = fakeBoardRealtime(); + const { call, tasks } = yield* subject(service, { openLive: () => realtime.client }); + yield* tasks.lifetime; + yield* call(GATEWAY_METHOD.TASKS_OPEN, { taskId: INVITES }); + yield* boardSettled; + assert.deepEqual(realtime.subscribed().toSorted(), ["calls-0", "pages-0"]); + const taken = () => service.recordReads.splice(0).toSorted(); + const worked = () => service.workReads.splice(0); + taken(); + worked(); + const deliver = (pages: readonly unknown[], calls: readonly unknown[]) => + Effect.andThen( + Effect.sync(() => { + realtime.deliver("pages-0", pages); + realtime.deliver("calls-0", calls); + }), + boardSettled, + ); + const page = (version: number) => ({ pageId: "page-1", version }); + + // The first delivery reads every part, since the reads the task opened with may have missed it. + yield* deliver([page(1)], []); + assert.deepEqual(taken(), ["pages", "transcript"]); + assert.deepEqual(worked(), []); + + yield* deliver([page(2)], []); + assert.deepEqual(taken(), ["pages"]); + + // A call's words landing while it stands, and its activity, read nothing. + yield* deliver([page(2)], [callRow(3, null)]); + tasks.showActivity({ + type: VOICE_SERVICE_FRAME.TASK_ACTIVITY, + planId: INVITES, + notes: false, + }); + yield* boardSettled; + assert.deepEqual(taken(), []); + assert.deepEqual(worked(), []); + + // Its end reads the work, and again once a turn's later writes have had time to land, but not the transcript. + tasks.callEnded(INVITES); + yield* boardSettled; + assert.deepEqual(worked(), [INVITES]); + yield* TestClock.adjust(Duration.seconds(5)); + yield* boardSettled; + assert.deepEqual(worked(), [INVITES]); + assert.deepEqual(taken(), []); + + // Its end, and its last words after, each read the transcript as the ended call's mark moves. + yield* deliver([page(2)], [callRow(3, 5_000)]); + assert.deepEqual(taken(), ["transcript"]); + yield* deliver([page(2)], [callRow(4, 5_000)]); + assert.deepEqual(taken(), ["transcript"]); + assert.deepEqual(worked(), []); + + // Leaving the task lets the subscriptions go. + yield* call(GATEWAY_METHOD.TASKS_CLOSE); + assert.deepEqual(realtime.subscribed(), []); + }), +); + it.effect( "a renewal that finds the open task gone ends the board's follow, rather than asking again", () => diff --git a/packages/host/src/compose-tasks.ts b/packages/host/src/compose-tasks.ts index 70f90fb99c..7b49a9fce4 100644 --- a/packages/host/src/compose-tasks.ts +++ b/packages/host/src/compose-tasks.ts @@ -5,6 +5,7 @@ import { type HostedWorkspaceClient, type PlanDraftFrame, type TaskActivityFrame, + type TaskCallResult, type TaskMarks, WORKSPACE_CALL_FAILURE, type WorkspaceCallResult, @@ -25,13 +26,13 @@ import { type CodingAgentStatus, } from "@luke/features/coding-agent-wire"; import { LIVE_SIGNED_OUT } from "@luke/features/live-wire"; -import { SHOW_PAGE_TOOL_NAME } from "@luke/features/page-wire"; import { PLAN_ACTOR, type Task, UNTITLED_TASK_NAME } from "@luke/features/task-wire"; import { IDLE_TASKS_VIEW, TASK_CALL_FAILURE, TASK_WORK_STATE, TASKS_READ, + type TaskCallFailure, type TaskCodeShowing, type TaskShownCode, type TasksCommentAnswer, @@ -155,9 +156,10 @@ 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 LiveBroken, type LiveClient, openRealtimeClient } from "./live-follow.js"; import type { RunMode } from "./run-mode.js"; import { codeExcerpt, highlightCode, highlightFile } from "./task-code.js"; +import { TASK_RECORD_PART, type TaskRecordPart, taskRecordMoves } from "./task-record-follow.js"; /** * compose-tasks.ts -- the panel's named tasks, read from the service: the list, the one active task, and its document, drawn as the notetaker writes it during a call. @@ -212,15 +214,6 @@ import { codeExcerpt, highlightCode, highlightFile } from "./task-code.js"; * held nowhere here. */ -/** - * How long after a call's end its transcript and work are read a second time. Note - * that the first read can land before the service has written the call's - * last words, because the service closes the device's socket first and only - * then waits on the writes it already started, and says nothing once they - * land; this is that wait's margin. - */ -const TRANSCRIPT_SETTLE = Duration.seconds(5); - /** * How long after the open Untitled task's brain settles a turn the list is * read again, for the name the service may give it as one of its first @@ -229,6 +222,16 @@ const TRANSCRIPT_SETTLE = Duration.seconds(5); */ const NAME_SETTLE = Duration.seconds(5); +/** + * How long after a call's end its work is read, and the open plan's parked + * conversation read woken, a second time. Note that the turn's last work and + * journal can land after the call's end, because the service closes the + * device's socket first and only then waits on the writes it already + * started; this is that wait's margin. The transcript is read as the call's + * live mark moves instead (`followRecord`). + */ +const TRANSCRIPT_SETTLE = Duration.seconds(5); + /** * How long after a turn that wrote the plan stops writing the plan is read * once more. Note that a turn holds open until its saves settle, except one @@ -399,6 +402,7 @@ export type TasksClient = Pick< | "writeTaskSection" | "readBoard" | "boardLive" + | "recordLive" | "saveBoard" | "readTranscript" | "readWork" @@ -542,9 +546,17 @@ const BOARD_BACKOFF = Schedule.exponential(BOARD_RESEAL.FIRST).pipe( ), ); -/** Why a board follow is not sealed again: the task is no longer the open one, or the gate closed. */ +/** Why a live follow is not sealed again: the task is no longer the open one, or the gate closed. */ const BOARD_FOLLOW_ENDED = "ended"; +/** A live read the service did not seal, `gone` where the task is no longer the account's. */ +type Unsealed = { readonly failure: string; readonly gone?: boolean }; + +/** Whether a seal is one the service did not answer. */ +function isUnsealed(sealed: Answer | Unsealed): sealed is Unsealed { + return "failure" in sealed; +} + /** * The panel's tasks as one concern. The view is told to every client whole * whenever it moves, and exactly one task is active at a time: opening or @@ -659,18 +671,24 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct /** The tasks whose board's mark moved, read one at a time by the loop below. */ const boardReads = yield* Queue.sliding(1); - /** The tasks whose pages a settled `show_page` may have moved, read one at a time by the loop below. */ + /** The one live follow of the open task's pages and calls; running another interrupts it. */ + const recordFollow = yield* FiberHandle.make(); + + /** The tasks whose pages' mark moved, read one at a time by the loop below. */ const pageReads = yield* Queue.sliding(1); + /** The tasks a call just ended about, whose work the loop below reads and whose parked conversation read it wakes, and both again once the call's last writes have had time to land. */ + const callEndWakes = yield* Queue.sliding(1); + + /** The tasks whose ended calls' marks moved, whose transcript is read once by the loop below. */ + const callRecordReads = yield* Queue.sliding(1); + /** The tasks whose plan one of Luke's plan edits may have moved, read one at a time by the loop below. */ const documentReads = yield* Queue.sliding(1); /** The tasks whose turn stopped writing after a plan edit, whose plan the loop below reads again once a save cut off with it has had time to land. */ const editUnwindReads = yield* Queue.sliding(1); - /** The tasks a call just ended about, whose transcript and work are read again by the loop below. */ - const callEndReads = yield* Queue.sliding(1); - /** The tasks whose brain just settled a turn, whose new name the loop below reads the list again for. */ const nameReads = yield* Queue.sliding(1); @@ -1139,28 +1157,51 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct } /** - * Follows `taskId`'s board live while it is the open task: the board read - * again once its mark moves, and on each seal's first delivery. A seal - * the service did not answer, or a follow that stops before it delivered, - * is sealed again on `BOARD_BACKOFF`; a follow that delivered and then + * Follows one of `taskId`'s live reads while it is the open task, each + * move handed to `moved`, the first delivery among them. A seal the + * service did not answer, or a follow that stops before it delivered, is + * sealed again on `BOARD_BACKOFF`; a follow that delivered and then * stopped is sealed again after `BOARD_RESEAL.FIRST`, its backoff started * over. A task the service no longer holds, or a gate that closed, ends it. */ - function followBoard(taskId: string) { + function followLive( + taskId: string, + read: { + readonly seal: ( + userId: string, + poll: boolean, + after: string | undefined, + ) => Effect.Effect< + TaskCallResult, + never, + HttpClient.HttpClient + >; + readonly moves: ( + sealing: { + readonly seal: () => Effect.Effect; + readonly poll: (after: string | undefined) => Effect.Effect; + readonly openLive: (url: string) => LiveClient | PromiseLike; + readonly now: () => number; + }, + first: Answer, + ) => Stream.Stream; + readonly moved: (moved: Moved) => void; + }, + ) { const following = () => gate() && view.activeTaskId === taskId; // The account the follow starts for, which every seal names, so one under another's session is refused as signed out. const userId = dependencies.accountId(); - const seal = (poll = false, after?: string) => + const seal = (poll = false, after?: string): Effect.Effect => userId === undefined ? Effect.succeed(LIVE_SIGNED_OUT) - : Effect.map(served(client.boardLive(taskId, userId, poll, after)), (sealed) => + : Effect.map(served(read.seal(userId, poll, after)), (sealed) => sealed.ok ? sealed.answer : { failure: sealed.failure }, ); // A task the account no longer reaches, or one no longer open, stops the follow rather than renewing or polling it. - const sealedFor = (poll: boolean, after?: string) => + const sealedFor = (poll: boolean, after?: string): Effect.Effect => following() ? Effect.map(seal(poll, after), (renewed) => - "failure" in renewed && renewed.failure === TASK_CALL_FAILURE.NOT_FOUND + isUnsealed(renewed) && renewed.failure === TASK_CALL_FAILURE.NOT_FOUND ? { ...renewed, gone: true } : renewed, ) @@ -1169,9 +1210,9 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct const attempt = Effect.gen(function* () { if (!following()) return yield* Effect.fail(BOARD_FOLLOW_ENDED); const sealed = yield* seal(); - if ("failure" in sealed) return yield* Effect.fail(sealed.failure); + if (isUnsealed(sealed)) return yield* Effect.fail(sealed.failure); let delivered = false; - const moves = boardMoves( + const moves = read.moves( { seal: () => sealedFor(false), poll: (after) => sealedFor(true, after), @@ -1180,10 +1221,10 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct }, sealed, ); - yield* Stream.runForEach(moves, () => + yield* Stream.runForEach(moves, (moved) => Effect.sync(() => { delivered = true; - if (view.activeTaskId === taskId) Queue.offerUnsafe(boardReads, taskId); + if (view.activeTaskId === taskId) read.moved(moved); }), ).pipe(Effect.ignore); if (!delivered) return yield* Effect.fail(TASK_CALL_FAILURE.UNANSWERED); @@ -1208,6 +1249,36 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct ); } + /** Follows `taskId`'s board live while it is the open task: the board read again once its mark moves, and on each seal's first delivery. */ + function followBoard(taskId: string) { + return followLive(taskId, { + seal: (userId, poll, after) => client.boardLive(taskId, userId, poll, after), + moves: boardMoves, + moved: () => Queue.offerUnsafe(boardReads, taskId), + }); + } + + /** + * Follows `taskId`'s pages and calls live while it is the open task: the + * pages read again once theirs move, and the transcript once an ended + * call's do, every part on each seal's first delivery. + */ + function followRecord(taskId: string) { + return followLive(taskId, { + seal: (userId, poll, after) => client.recordLive(taskId, userId, poll, after), + moves: taskRecordMoves, + moved: (parts: readonly TaskRecordPart[]) => { + if (parts.includes(TASK_RECORD_PART.PAGES)) Queue.offerUnsafe(pageReads, taskId); + if (parts.includes(TASK_RECORD_PART.CALLS)) Queue.offerUnsafe(callRecordReads, taskId); + }, + }); + } + + /** Follows `taskId`'s pages and calls live, in place of any follow standing; with `standing`, only where none stands. */ + function openRecord(taskId: string, standing?: { readonly onlyIfMissing: true }) { + return Effect.asVoid(FiberHandle.run(recordFollow, followRecord(taskId), standing)); + } + /** Follows `taskId`'s board live, in place of any follow standing; with `standing`, only where none stands. */ function openBoard(taskId: string, standing?: { readonly onlyIfMissing: true }) { return Effect.asVoid(FiberHandle.run(boardFollow, followBoard(taskId), standing)); @@ -1260,12 +1331,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct return; } wakeConversation(taskId); - const was = view.activity?.planner; - const stopped = was !== undefined && activity.planner === undefined; write({ activity, callActivity: activity }); - // A page shown settles once the brain's pending call is not the show any more. - const paged = was?.action === SHOW_PAGE_TOOL_NAME && activity.planner?.action !== was.action; - if (paged || stopped) Queue.offerUnsafe(pageReads, taskId); } /** Whether code shown about a task is still for the screen: the task is open and the call's code is not drawn already, since a call's code told again is the showing already drawn. */ @@ -1309,8 +1375,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct view = rest; publish(); if (view.activeTaskId !== taskId) return; - Queue.offerUnsafe(callEndReads, taskId); - wakeConversation(taskId); + Queue.offerUnsafe(callEndWakes, taskId); } /** The open document of `taskId` given the name its row of the list now reads, where the two differ. */ @@ -1486,6 +1551,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct const open = view.activeTaskId; if (open !== undefined) yield* openConversation(open); if (open !== undefined) yield* openBoard(open, { onlyIfMissing: true }); + if (open !== undefined) yield* openRecord(open, { onlyIfMissing: true }); const tasks = serial( Effect.gen(function* () { yield* readList; @@ -1528,6 +1594,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct } yield* openConversation(taskId); yield* openBoard(taskId); + yield* openRecord(taskId); yield* serial( Effect.gen(function* () { yield* readDocument(taskId); @@ -1896,6 +1963,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct Effect.gen(function* () { yield* FiberHandle.clear(conversationRead); yield* FiberHandle.clear(boardFollow); + yield* FiberHandle.clear(recordFollow); yield* serial( Effect.sync(() => { const { activeTaskId: _closed, ...rest } = withoutActivity(view); @@ -1929,6 +1997,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct publish(); yield* openConversation(task.id); yield* openBoard(task.id); + yield* openRecord(task.id); yield* readList; return carried({ planId: task.id, @@ -1954,6 +2023,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct if (view.activeTaskId === taskId) { yield* FiberHandle.clear(conversationRead); yield* FiberHandle.clear(boardFollow); + yield* FiberHandle.clear(recordFollow); const { activeTaskId: _deleted, ...rest } = withoutActivity(view); view = { ...rest, document: { status: TASKS_READ.IDLE } }; } @@ -2437,6 +2507,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct if (view.activeTaskId !== openTaskId) return; yield* FiberHandle.clear(conversationRead); yield* FiberHandle.clear(boardFollow); + yield* FiberHandle.clear(recordFollow); const { activeTaskId: _left, ...rest } = withoutActivity(view); view = { ...rest, document: { status: TASKS_READ.IDLE } }; publish(); @@ -2531,6 +2602,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct yield* FiberHandle.clear(conversationRead); yield* FiberHandle.clear(boardFollow); yield* FiberMap.clear(followedWorkers); + yield* FiberHandle.clear(recordFollow); yield* serial( Effect.sync(() => { workspacesAskedAt = undefined; @@ -2597,11 +2669,11 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct ); yield* Effect.forkScoped( Effect.forever( - Effect.flatMap(Queue.take(callEndReads), (taskId) => { - // The conversation's parked read is woken with each read too, since its last journal may land after the call's end. + Effect.flatMap(Queue.take(callEndWakes), (taskId) => { + // Note that the turn's last work and journal may land after the call's end, so both are read again after a wait. const read = Effect.suspend(() => gate() && view.activeTaskId === taskId - ? serial(Effect.andThen(readTranscript(taskId), readWork(taskId))).pipe( + ? serial(readWork(taskId)).pipe( Effect.andThen(Effect.sync(() => wakeConversation(taskId))), ) : Effect.void, @@ -2610,6 +2682,15 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct }), ), ); + yield* Effect.forkScoped( + Effect.forever( + Effect.flatMap(Queue.take(callRecordReads), (taskId) => + Effect.suspend(() => + gate() && view.activeTaskId === taskId ? serial(readTranscript(taskId)) : Effect.void, + ), + ), + ), + ); yield* Effect.forkScoped( Effect.forever( Effect.flatMap(Queue.take(nameReads), (taskId) => diff --git a/packages/host/src/task-record-follow.ts b/packages/host/src/task-record-follow.ts new file mode 100644 index 0000000000..547ea77198 --- /dev/null +++ b/packages/host/src/task-record-follow.ts @@ -0,0 +1,110 @@ +import { + type TaskRecordLiveAnswer, + type TaskRecordLiveRows, + taskRecordLiveRowsSchema, +} from "@luke/features/live-wire"; +import { Effect, Option, Schema, Stream } from "effect"; +import { LiveBroken, type LiveClient, liveRows } from "./live-follow.js"; + +/** + * task-record-follow.ts -- where an open task's pages and calls stand, followed by the host over a Neon Realtime subscription, so a part is read again only once its mark moved. + * + * The service seals one statement a part (`task-record-live.ts`), each a + * mark that moves as its rows do and carries none of their words. The host + * subscribes to them (`live-follow.ts`) for as long as the task stands open, + * as it does the board (`board-follow.ts`), and answers each delivery with + * the parts it moved, which the caller reads again from the service. The + * sealed statements stay here, beside the account's own bearer; only which + * parts moved leaves. + */ + +/** A part of the record the panel reads again once its mark moves. */ +export const TASK_RECORD_PART = { + PAGES: "pages", + CALLS: "calls", +} as const; + +export type TaskRecordPart = (typeof TASK_RECORD_PART)[keyof typeof TASK_RECORD_PART]; + +/** Every part, which the first delivery moves, since the reads the task opened with may have missed what it says. */ +const EVERY_TASK_RECORD_PART: readonly TaskRecordPart[] = Object.values(TASK_RECORD_PART); + +/** A seal the service did not answer, `gone` where the task is no longer the account's. */ +type Unsealed = { readonly failure: string; readonly gone?: boolean }; + +export interface TaskRecordFollowDependencies { + /** The record sealed again, for renewing; a failure where the service sealed none. */ + readonly seal: () => Effect.Effect; + /** The record's rows inline, polled while Realtime fails or is off, held past `after`; failures as `seal`'s. */ + readonly poll: (after: string | undefined) => Effect.Effect; + /** Realtime's client at the address the service named, loaded where it must be. */ + readonly openLive: (url: string) => LiveClient | PromiseLike; + readonly now: () => number; +} + +/** A part's rows as one comparable value. */ +function markOf(rows: readonly Row[], key: (row: Row) => string): string { + return JSON.stringify(rows.toSorted((a, b) => (key(a) < key(b) ? -1 : key(a) > key(b) ? 1 : 0))); +} + +/** The calls whose words the record now holds whole: each that ended, since a call that stands tells its own as it goes. */ +function endedCalls(rows: TaskRecordLiveRows): string { + return markOf( + rows.calls.filter((call) => call.closedAt !== null), + (call) => call.id, + ); +} + +/** + * The parts a delivery moved from the one before: the pages where their + * rows differ, and the calls where an ended call's do, so a call's words + * landing as it stands read nothing and its end, its last words, and its + * metering each read the calls again. The first delivery moves every part. + */ +function movedParts( + before: TaskRecordLiveRows | undefined, + rows: TaskRecordLiveRows, +): readonly TaskRecordPart[] { + if (before === undefined) return EVERY_TASK_RECORD_PART; + const pagesOf = (read: TaskRecordLiveRows) => markOf(read.pages, (page) => page.pageId); + return [ + ...(pagesOf(before) === pagesOf(rows) ? [] : [TASK_RECORD_PART.PAGES]), + ...(endedCalls(before) === endedCalls(rows) ? [] : [TASK_RECORD_PART.CALLS]), + ]; +} + +/** + * The parts of the record each delivery moved, from the first seal for as + * long as the caller's scope stands, a delivery that moved none left out. + * It fails where the subscription stops for good or a row does not decode, + * which the caller answers by sealing it again. + */ +export function taskRecordMoves( + dependencies: TaskRecordFollowDependencies, + first: TaskRecordLiveAnswer, +): Stream.Stream { + const decode = Schema.decodeUnknownOption(taskRecordLiveRowsSchema); + return liveRows( + { + openLive: dependencies.openLive, + reseal: dependencies.seal, + poll: dependencies.poll, + now: dependencies.now, + }, + first, + ).pipe( + Stream.mapEffect((read) => + Option.match(decode(read), { + onNone: () => Effect.fail(new LiveBroken({ reason: "a row did not decode" })), + onSome: Effect.succeed, + }), + ), + Stream.mapAccum( + (): TaskRecordLiveRows | undefined => undefined, + (before: TaskRecordLiveRows | undefined, rows: TaskRecordLiveRows) => { + const moved = movedParts(before, rows); + return [rows, moved.length === 0 ? [] : [moved]] as const; + }, + ), + ); +} diff --git a/packages/service/src/live/live-procedures.test.ts b/packages/service/src/live/live-procedures.test.ts index 38ce9967b4..d91ce5d16c 100644 --- a/packages/service/src/live/live-procedures.test.ts +++ b/packages/service/src/live/live-procedures.test.ts @@ -18,6 +18,7 @@ import { liveDigestOf, taskBoardLiveAnswerSchema, taskBoardLiveRowsSchema, + taskRecordLiveAnswerSchema, } from "@luke/features/live-wire"; import type { WireBoundaryInput } from "@luke/wire"; import { Clock, Duration, Effect, Fiber, Layer, Option, Schema } from "effect"; @@ -152,12 +153,13 @@ const openUser = Effect.gen(function* () { return userId; }); -/** The procedures answering a task board's live read as `bearer`, over the store and the deployment's Realtime. */ +/** The procedures answering a task's live read as `bearer`, its board's unless `path` names another, over the store and the deployment's Realtime. */ const askBoard = (options: { readonly bearer: string; /** The account the read names; the bearer's own where left out. */ readonly forUser?: string; readonly taskId?: string; + readonly path?: string; readonly poll?: boolean; /** The digest the reader last saw, which holds the poll until the rows move past it. */ readonly after?: string; @@ -175,7 +177,7 @@ const askBoard = (options: { resolveUserId: () => Effect.succeed(Option.some(options.bearer)), }); const url = readOf( - options.taskId === undefined ? LIVE_INBOX : "/api/live/task-board", + options.taskId === undefined ? LIVE_INBOX : (options.path ?? "/api/live/task-board"), options.forUser ?? options.bearer, ); if (options.taskId !== undefined) url.searchParams.set("id", options.taskId); @@ -266,6 +268,36 @@ it.layer(getTestDB())("the task board's live read", (it) => { }), ); + it.effect( + "a task's pages and calls are sealed a part each for an account that reaches it, and not found for one that does not", + () => + Effect.gen(function* () { + const dana = yield* openUser; + const eli = yield* openUser; + const started = yield* startTask(dana, { name: "Teammate invitations" }); + const ask = (bearer: string) => + askBoard({ + bearer, + taskId: started.id, + path: "/api/live/task-record", + sealing: Option.some(SCRIPTED), + }); + + const own = yield* ask(dana); + assert.equal(own.status, HOSTED_HTTP_STATUS.OK); + const answer = Schema.decodeUnknownSync(taskRecordLiveAnswerSchema)(own.body); + assert.equal(answer.url, REALTIME_URL); + for (const query of [answer.queries.pages, answer.queries.calls]) { + assert.ok(query.capability.includes(`"${dana}"`), "read for the bearer"); + assert.ok(query.capability.includes(`"${started.id}"`), "read for the task"); + } + assert.deepEqual(yield* ask(eli), { + status: HOSTED_HTTP_STATUS.NOT_FOUND, + body: { error: HOSTED_API_ERROR.NOT_FOUND }, + }); + }), + ); + it.effect("the account inbox Realtime is off for is answered with every part's rows inline", () => Effect.gen(function* () { const dana = yield* openUser; @@ -475,6 +507,16 @@ it.layer(getTestDB())("an agent transcript's live read", (it) => { yield* askBoard({ bearer: dana, forUser, sealing: Option.none() }), signedOut, ); + assert.deepEqual( + yield* askBoard({ + bearer: dana, + forUser, + taskId: started.id, + path: "/api/live/task-record", + sealing: Option.none(), + }), + signedOut, + ); } }), ); diff --git a/packages/service/src/live/live-procedures.ts b/packages/service/src/live/live-procedures.ts index d9d11d2c0c..94cd96c2ca 100644 --- a/packages/service/src/live/live-procedures.ts +++ b/packages/service/src/live/live-procedures.ts @@ -14,7 +14,8 @@ * `GET /api/live/task-board?id=` answers one task's whiteboard the same way, * as one statement that marks where it stands, after the task is read as * the account, so one it does not reach is not found, as its board read - * answers. + * answers; and `GET /api/live/task-record?id=` its pages and calls, one + * statement a part (`task-record-live.ts`), on the same terms. * * Either read asked with `?poll=true`, or one Realtime is off for or could * not seal, is answered with each statement's rows inline at the poll @@ -57,6 +58,7 @@ import { } from "../http/http-effect.js"; import { accountInboxLiveStatements } from "../inbox/inbox-store.js"; import { accountMiddleware, hosted } from "../orpc/base.js"; +import { taskRecordLiveStatements } from "../tasks/task-record-live.js"; import { type LiveSealer, LiveSealing, pollSealer } from "./realtime.js"; export interface LiveProcedureSeams { @@ -190,6 +192,15 @@ export function liveProcedures(seams: LiveProcedureSeams) { if (Option.isNone(statement)) return yield* Effect.fail(HOSTED_REFUSAL.UNAVAILABLE); return yield* sealedParts({ board: statement.value }, input); }), + taskRecord: signedIn.taskRecord.effect(function* ({ input, context }) { + yield* sameUser(input, context.userId); + const taskId = yield* rowIdOf(input.id); + const task = yield* hostedStoreOrUnavailable(readTask(context.userId, taskId)); + if (Option.isNone(task)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); + const statements = taskRecordLiveStatements(context.userId, taskId); + if (Option.isNone(statements)) return yield* Effect.fail(HOSTED_REFUSAL.UNAVAILABLE); + return yield* sealedParts(statements.value, input); + }), agentTranscript: signedIn.agentTranscript.effect(function* ({ input, context }) { yield* sameUser(input, context.userId); const callId = input.call === undefined ? undefined : yield* callIdOf(input.call); diff --git a/packages/service/src/live/live-statements.test.ts b/packages/service/src/live/live-statements.test.ts index de345f4b88..d670dadc2f 100644 --- a/packages/service/src/live/live-statements.test.ts +++ b/packages/service/src/live/live-statements.test.ts @@ -7,6 +7,7 @@ import { getTestDB, mutableFunctionsOf } from "@luke/database/testing"; import { ConfigProvider, Effect, Layer, Option } from "effect"; import { agentFollowedBy, transcriptLiveStatements } from "../agents/coder-host/transcript-live.js"; import { accountInboxLiveStatements } from "../inbox/inbox-store.js"; +import { taskRecordLiveStatements } from "../tasks/task-record-live.js"; import { LiveSealing, liveSealingFromEnvironment, REALTIME_ENVIRONMENT } from "./realtime.js"; /** @@ -42,6 +43,11 @@ function liveStatements() { ), "account.failedRuns": Option.getOrThrow(ownFailedRunsLiveStatement(USER_ID)), "task.board": Option.getOrThrow(boardLiveStatement(USER_ID, randomUUID())), + ...Object.fromEntries( + Object.entries(Option.getOrThrow(taskRecordLiveStatements(USER_ID, randomUUID()))).map( + ([part, query]) => [`task.${part}`, query], + ), + ), ...Object.fromEntries( Object.entries(transcript).map(([part, query]) => [`transcript.${part}`, query]), ), diff --git a/packages/service/src/tasks/task-record-live.test.ts b/packages/service/src/tasks/task-record-live.test.ts new file mode 100644 index 0000000000..e7dd9f9556 --- /dev/null +++ b/packages/service/src/tasks/task-record-live.test.ts @@ -0,0 +1,225 @@ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { it } from "@effect/vitest"; +import type { LiveStatement } from "@luke/database/core/live-statement"; +import { Database, type DatabaseServices } from "@luke/database/core/query"; +import { listPages, writePage } from "@luke/database/models/plan-pages"; +import { openTaskConversation } from "@luke/database/models/task-store"; +import { readTranscript } from "@luke/database/models/transcript-store"; +import { user } from "@luke/database/schemas/auth"; +import { messages } from "@luke/database/schemas/storage"; +import { voiceSessions, voiceTranscriptSegments } from "@luke/database/schemas/voice"; +import { getTestDB, startTask } from "@luke/database/testing"; +import { + VOICE_CLOSE_REASON, + VOICE_DELEGATION_MODE, + VOICE_SEGMENT_ROLE, +} from "@luke/database/voice-vocabulary"; +import { type TaskRecordLiveRows, taskRecordLiveRowsSchema } from "@luke/features/live-wire"; +import { PAGE_SHOWER } from "@luke/features/page-wire"; +import { MESSAGE_AUTHOR, MESSAGE_CHANNEL, MESSAGE_ROLE } from "@luke/wire"; +import { eq } from "drizzle-orm"; +import { Effect, Option, Schema } from "effect"; +import { SqlClient } from "effect/sql"; +import { taskRecordLiveStatements } from "./task-record-live.js"; + +/** + * An open task's record as Realtime would deliver it, over the real + * migrations on PGlite: each part's statement run as Realtime runs it and its + * rows decoded as the desktop decodes them, beside the held reads the desktop + * reads that part with (`listPages`, `readTranscript`). A part + * moves exactly when its held read would answer something new. + * + * Synthetic accounts, pages, and words throughout. + */ + +const PARTS = ["pages", "calls"] as const; + +type Part = (typeof PARTS)[number]; + +const CLOSED_AT = new Date(1_800_000_000_000); + +/** A new account, as the store holds one. */ +const openUser = Effect.gen(function* () { + const db = yield* Database; + const userId = `user-${randomUUID()}`; + yield* db.insert(user).values({ id: userId, name: "Test User", email: `${userId}@luke.test` }); + return userId; +}); + +/** One part's statement run as Realtime runs it. */ +const runLive = (statement: LiveStatement) => + Effect.flatMap(SqlClient.SqlClient, (sql) => sql.unsafe(statement.sql, [...statement.params])); + +/** The task's record as Realtime would deliver it to `reader`, decoded as the desktop decodes it. */ +const liveRecordOf = (reader: string, taskId: string) => + Effect.gen(function* () { + const statements = Option.getOrThrow(taskRecordLiveStatements(reader, taskId)); + return yield* Schema.decodeUnknownEffect(taskRecordLiveRowsSchema)({ + pages: yield* runLive(statements.pages), + calls: yield* runLive(statements.calls), + }); + }); + +/** The same parts as the desktop reads them today, each as its held read answers it. */ +const heldRecordOf = (reader: string, taskId: string) => + Effect.gen(function* () { + return { + pages: yield* listPages(reader, taskId), + calls: yield* readTranscript(reader, taskId), + }; + }); + +/** The record as its held reads answer it. */ +type HeldRecord = Effect.Success>; + +/** The parts whose value differs between two readings, by their text, as the desktop compares deliveries. */ +function movedParts( + before: Reading, + after: Reading, +): readonly Part[] { + return PARTS.filter((part) => JSON.stringify(before[part]) !== JSON.stringify(after[part])); +} + +/** A call about the task, its session row written as the voice service opens one. */ +const openCall = (userId: string, taskId: string) => + Effect.gen(function* () { + const db = yield* Database; + const [row] = yield* db + .insert(voiceSessions) + .values({ + userId, + taskId, + liveSessionId: `live-${randomUUID()}`, + delegationMode: VOICE_DELEGATION_MODE.CLIENT, + model: "gpt-realtime", + }) + .returning({ id: voiceSessions.id }); + assert.ok(row); + return row.id; + }); + +/** One fragment of the developer's words on a call, appended as the voice writer appends one. */ +const said = (voiceSessionId: string, seq: number) => + Effect.flatMap(Database, (db) => + db.insert(voiceTranscriptSegments).values({ + voiceSessionId, + seq, + role: VOICE_SEGMENT_ROLE.USER, + text: seq === 0 ? "Invite by" : " email.", + startMs: seq * 400, + endMs: seq * 400 + 300, + }), + ); + +it.layer(getTestDB())("a task's record followed live", (it) => { + it.effect( + "each part moves exactly when its held read would answer something new, as a page is shown, and a call's words, end, metering and rows land; and a stranger reads no rows", + () => + Effect.gen(function* () { + const owner = yield* openUser; + const stranger = yield* openUser; + const task = yield* startTask(owner, { name: "Teammate invitations" }); + const other = yield* startTask(owner, { name: "Billing" }); + + let live = yield* liveRecordOf(owner, task.id); + let held = yield* heldRecordOf(owner, task.id); + const step = ( + expected: readonly Part[], + write: Effect.Effect, + ) => + Effect.gen(function* () { + yield* Effect.orDie(write); + const nextLive = yield* liveRecordOf(owner, task.id); + const nextHeld = yield* heldRecordOf(owner, task.id); + assert.deepEqual(movedParts(held, nextHeld), expected, "the held reads moved"); + assert.deepEqual(movedParts(live, nextLive), expected, "the live marks moved"); + live = nextLive; + held = nextHeld; + }); + + assert.deepEqual(live, { pages: [], calls: [] }); + + const shown = Option.getOrThrow( + yield* writePage( + owner, + task.id, + { title: "Invite funnel", html: "

One

" }, + { kind: PAGE_SHOWER.LUKE }, + ), + ); + live = yield* liveRecordOf(owner, task.id); + held = yield* heldRecordOf(owner, task.id); + assert.deepEqual(live.pages, [{ pageId: shown.id, version: 1 }]); + yield* step( + ["pages"], + writePage( + owner, + task.id, + { pageId: shown.id, title: "Invite funnel", html: "

Two

" }, + { kind: PAGE_SHOWER.LUKE }, + ), + ); + + // A call with no words yet reads as no call. + const call = yield* openCall(owner, task.id); + yield* step([], Effect.void); + assert.deepEqual(live.calls, []); + yield* step(["calls"], said(call, 0)); + yield* step(["calls"], said(call, 1)); + // The task's conversation opening moves nothing until the voice writes a row for the call. + const conversation = Option.getOrThrow(yield* openTaskConversation(owner, task.id)); + yield* step([], Effect.void); + yield* step( + ["calls"], + Database.use((db) => + db.insert(messages).values({ + userId: owner, + conversationId: conversation.conversationId, + seq: 1, + clientId: `spoken-${randomUUID()}`, + role: MESSAGE_ROLE.USER, + parts: [{ type: "text", text: "Invite by email." }], + metadata: { + author: MESSAGE_AUTHOR.DEVELOPER, + channel: MESSAGE_CHANNEL.VOICE, + voice_session_id: call, + from_ms: 0, + to_ms: 700, + }, + placedAt: CLOSED_AT, + }), + ), + ); + yield* step( + ["calls"], + Database.use((db) => + db + .update(voiceSessions) + .set({ closedAt: CLOSED_AT, closeReason: VOICE_CLOSE_REASON.REMOTE_HANGUP }) + .where(eq(voiceSessions.id, call)), + ), + ); + yield* step( + ["calls"], + Database.use((db) => + db + .update(voiceSessions) + .set({ usage: { seconds: 42, confirmed: true } }) + .where(eq(voiceSessions.id, call)), + ), + ); + // Another task's record moving moves nothing of this one's. + yield* step( + [], + writePage( + owner, + other.id, + { title: "Billing funnel", html: "

One

" }, + { kind: PAGE_SHOWER.LUKE }, + ), + ); + assert.deepEqual(yield* liveRecordOf(stranger, task.id), { pages: [], calls: [] }); + }), + ); +}); diff --git a/packages/service/src/tasks/task-record-live.ts b/packages/service/src/tasks/task-record-live.ts new file mode 100644 index 0000000000..8cec350d93 --- /dev/null +++ b/packages/service/src/tasks/task-record-live.ts @@ -0,0 +1,109 @@ +import { + type LiveStatement, + liveSelection, + liveStatementOf, +} from "@luke/database/core/live-statement"; +import { queryBuilder } from "@luke/database/core/query"; +import { taskSeenBy } from "@luke/database/models/workspace-store"; +import { member } from "@luke/database/schemas/auth"; +import { planPageVersion } from "@luke/database/schemas/plan-page"; +import { messages } from "@luke/database/schemas/storage"; +import { task } from "@luke/database/schemas/task"; +import { voiceSessions, voiceTranscriptSegments } from "@luke/database/schemas/voice"; +import { and, eq, isNotNull, max, sql } from "drizzle-orm"; +import { Option } from "effect"; + +/** + * task-record-live.ts -- where an open task's pages and calls stand, written for Neon Realtime, so the desktop reads either again the moment its rows move rather than when a call's activity or end suggests they might have. + * + * Two parts, each a mark and none of the words it marks: the pages agents + * showed (`pages`), one row a page at its newest version; and the task's + * calls (`calls`), one row a call that has words, which moves as its words + * land, as it closes, and as its metering and the rows the voice wrote for + * it land after. The board is followed on its own (`boardLiveStatement`). + * Each statement reads what its held read reads (`listPages`, + * `readTranscript`), the account joined through its membership of the + * task's workspace and held to the task's GitHub access, as the board's + * live read is, so a task the account does not reach is no rows. None reads + * the clock, as Realtime requires. + */ + +/** The statements an open task's record is followed by, one a part. */ +export interface TaskRecordLiveStatements { + readonly pages: LiveStatement; + readonly calls: LiveStatement; +} + +/** The account's membership of the task's workspace, which a live read joins rather than asks, as Neon's guide authorizes one. */ +function memberOfTask(userId: string) { + return and(eq(member.organizationId, task.organizationId), eq(member.userId, userId)); +} + +/** Each page as `listPages` reads it, marked by its newest version's number: a version is written once and never changed. */ +function pagesStatement(userId: string, taskId: string) { + return queryBuilder + .select({ + ...liveSelection({ pageId: planPageVersion.pageId }), + version: max(planPageVersion.version).as("version"), + }) + .from(planPageVersion) + .innerJoin(task, eq(task.id, planPageVersion.taskId)) + .innerJoin(member, memberOfTask(userId)) + .where(and(eq(planPageVersion.taskId, taskId), taskSeenBy(userId))) + .groupBy(planPageVersion.pageId); +} + +/** + * Each call `readTranscript` reads, marked: one with words, since a call + * with none reads as no call; how it ended, what it metered, and its model, + * as its session row holds them; its last fragment, since fragments are + * only ever appended; and how many rows the voice wrote for it in the task's + * conversation, which name its lines. + */ +function callsStatement(userId: string, taskId: string) { + const spoken = queryBuilder + .select({ count: sql`count(*)::int` }) + .from(messages) + .where( + and( + eq(messages.userId, userId), + eq(messages.conversationId, task.conversationId), + sql`${messages.metadata}->>'voice_session_id' = ${voiceSessions.id}::text`, + isNotNull(sql`${messages.metadata}->>'from_ms'`), + ), + ); + return queryBuilder + .select({ + ...liveSelection( + { + id: voiceSessions.id, + closeReason: voiceSessions.closeReason, + usage: voiceSessions.usage, + model: voiceSessions.model, + }, + { closedAt: voiceSessions.closedAt }, + ), + lastSegment: max(voiceTranscriptSegments.seq).as("lastSegment"), + spoken: sql`(${spoken})`.as("spoken"), + }) + .from(voiceSessions) + .innerJoin(task, eq(task.id, voiceSessions.taskId)) + .innerJoin(member, memberOfTask(userId)) + .innerJoin( + voiceTranscriptSegments, + eq(voiceTranscriptSegments.voiceSessionId, voiceSessions.id), + ) + .where(and(eq(voiceSessions.taskId, taskId), taskSeenBy(userId))) + .groupBy(voiceSessions.id, task.id); +} + +/** The task's record written for Realtime; none where a statement names a value Realtime cannot carry. */ +export function taskRecordLiveStatements( + userId: string, + taskId: string, +): Option.Option { + return Option.all({ + pages: liveStatementOf(pagesStatement(userId, taskId)), + calls: liveStatementOf(callsStatement(userId, taskId)), + }); +}