From 34fe396433172305e9110ec117b497524b9b1a8c Mon Sep 17 00:00:00 2001 From: Charles Pan <30245070+charleslpan@users.noreply.github.com> Date: Wed, 21 Oct 2026 00:37:28 +0000 Subject: [PATCH] feat(service,features): a workspace's automations and an automation's runs are live reads GET /api/live/automations (live.automations) answers every automation of a workspace the account reads, with its newest run, that run's agent, and what the agent's status is read from, in one statement; GET /api/live/automation-runs (live.automationRuns) answers an automation's newest fifty runs on the same terms, once the automation is read as the account. As every live read is, each names the account it is read for, answers its rows inline at the poll address with Realtime off, held until they move, and is sealed where Realtime is on. An agent's newest turn, its newest awaiting line, and why its turn failed are one set of fragments the task list shares, the failure named in SQL so a turn's detail never leaves the database, and an automation with no repositories of its own falls back to its task's on the reader, since jsonb_build_array is not immutable. The desktop builds the list and the history with @luke/features/automation-live. Parity tests run the statements as raw SQL against the held answers, and both pass the immutability helper. Co-authored-by: luke-agent[bot] <339870089+luke-agent[bot]@users.noreply.github.com> --- .../database/src/models/automation-store.ts | 4 +- packages/features/package.json | 1 + packages/features/src/automation-client.ts | 82 +++++- packages/features/src/automation-live.ts | 264 ++++++++++++++++++ packages/features/src/contract/live.ts | 16 ++ packages/features/src/service-paths.ts | 4 + .../service/src/agents/coder-host/status.ts | 2 +- .../src/automations/automation-live.test.ts | 28 ++ .../src/automations/automation-live.ts | 167 +++++++++++ .../automations-procedures.test.ts | 162 ++++++++++- .../src/automations/automations-procedures.ts | 8 +- packages/service/src/live/agent-standing.ts | 80 ++++++ packages/service/src/live/live-procedures.ts | 24 +- packages/service/src/tasks/task-list-live.ts | 28 +- 14 files changed, 825 insertions(+), 45 deletions(-) create mode 100644 packages/features/src/automation-live.ts create mode 100644 packages/service/src/automations/automation-live.test.ts create mode 100644 packages/service/src/automations/automation-live.ts create mode 100644 packages/service/src/live/agent-standing.ts diff --git a/packages/database/src/models/automation-store.ts b/packages/database/src/models/automation-store.ts index 315ee470dd..8630563a63 100644 --- a/packages/database/src/models/automation-store.ts +++ b/packages/database/src/models/automation-store.ts @@ -102,14 +102,14 @@ const AutomationRowSchema = Schema.Struct({ export type Automation = typeof AutomationRowSchema.Type; /** An automation row's workspace: its own, or its task's on a row an older build wrote, its workspace still null. */ -const AUTOMATION_WORKSPACE = sql`coalesce(${automation.organizationId}, (select ${task.organizationId} from ${task} where ${task.id} = ${automation.taskId}))`; +export const AUTOMATION_WORKSPACE = sql`coalesce(${automation.organizationId}, (select ${task.organizationId} from ${task} where ${task.id} = ${automation.taskId}))`; /** An automation row's repositories: its own, or its task's on a row an older build wrote, its list still empty. */ const AUTOMATION_REPOSITORIES = sql< readonly string[] >`case when ${automation.repositories} = '[]'::jsonb and ${task.repository} is not null then jsonb_build_array(${task.repository}) else ${automation.repositories} end`; -const AUTOMATION_COLUMNS = { +export const AUTOMATION_COLUMNS = { id: automation.id, userId: automation.userId, taskId: automation.taskId, diff --git a/packages/features/package.json b/packages/features/package.json index bf8b351d22..8925b7a83e 100644 --- a/packages/features/package.json +++ b/packages/features/package.json @@ -12,6 +12,7 @@ "./admin-wire": "./src/admin-wire.js", "./app-links": "./src/app-links.js", "./automation-code-review": "./src/automation-code-review.js", + "./automation-live": "./src/automation-live.js", "./automation-schedule": "./src/automation-schedule.js", "./automation-wire": "./src/automation-wire.js", "./automation-view": "./src/automation-view.js", diff --git a/packages/features/src/automation-client.ts b/packages/features/src/automation-client.ts index 9fbc59e651..09fc4001c7 100644 --- a/packages/features/src/automation-client.ts +++ b/packages/features/src/automation-client.ts @@ -2,6 +2,11 @@ import { readEither } from "@luke/wire/effect"; import { Effect, Result } from "effect"; import type * as HttpClient from "effect/http/HttpClient"; import type { CallCredential } from "./account-call.js"; +import { + type AutomationRunsLiveAnswer, + type AutomationsLiveAnswer, + withWebhookUrl, +} from "./automation-live.js"; import { AUTOMATION_CALL_FAILURE, type AutomationCallFailure, @@ -18,15 +23,16 @@ import { automationChangeParamsSchema, automationCreateParamsSchema, } from "./automation-view.js"; +import type { automationAnswerSchema, automationChangeRequestSchema } from "./automation-wire.js"; import { - AUTOMATION_TRIGGER, - type AutomationSummary, - type automationAnswerSchema, - type automationChangeRequestSchema, -} from "./automation-wire.js"; -import { askHosted, type HostedClient, type HostedFailure, hostedClient } from "./hosted-link.js"; + askHosted, + askLive, + type HostedClient, + type HostedFailure, + hostedClient, +} from "./hosted-link.js"; +import { LIVE_SIGNED_OUT, LIVE_USER_QUERY } from "./live-wire.js"; import type { RepositoryUnreached } from "./repository-reach-wire.js"; -import { automationWebhookPath } from "./service-paths.js"; import { HOSTED_API_ERROR, type HostedApiError } from "./service-wire.js"; /** @@ -103,10 +109,17 @@ function summaryOnly( : { automation, webhookSecret: read.webhookSecret }; } -/** A webhook automation with the address its caller posts to, on the service this client reaches; any other as it came. */ -function withWebhookUrl(serviceBaseUrl: string, automation: AutomationSummary): AutomationSummary { - if (automation.trigger.kind !== AUTOMATION_TRIGGER.WEBHOOK) return automation; - return { ...automation, webhookUrl: `${serviceBaseUrl}${automationWebhookPath(automation.id)}` }; +/** A live read's seal, or why there is none: signed out where the session is not the account it names. */ +function liveAnswered( + asked: Effect.Effect, +): Effect.Effect { + return Effect.catch(asked, (failed) => + Effect.succeed( + failed._tag === "HostedRefused" && failed.refusal === HOSTED_API_ERROR.INVALID_TOKEN + ? LIVE_SIGNED_OUT + : failedAs(failed, REFUSALS), + ), + ); } /** The window's reads and writes of the account's automations. */ @@ -220,4 +233,51 @@ export class HostedAutomationClient { (read) => summaryOnly(this.#serviceBaseUrl, read), ); } + + /** + * A workspace's automations sealed for following over Neon Realtime for + * the account `userId` names, or with `poll` their rows inline: signed + * out where the session is another account's, and why there is none + * otherwise. + */ + listLive( + workspaceId: string, + userId: string, + poll = false, + after?: string, + ): Effect.Effect< + AutomationsLiveAnswer | Failed | typeof LIVE_SIGNED_OUT, + never, + HttpClient.HttpClient + > { + return liveAnswered( + askLive( + this.#hosted.live.automations, + { workspaceId, [LIVE_USER_QUERY]: userId }, + poll, + after, + ), + ); + } + + /** One automation's runs sealed for following over Neon Realtime, on the terms the list is: not found where the account does not read it. */ + runsLive( + automationId: string, + userId: string, + poll = false, + after?: string, + ): Effect.Effect< + AutomationRunsLiveAnswer | Failed | typeof LIVE_SIGNED_OUT, + never, + HttpClient.HttpClient + > { + return liveAnswered( + askLive( + this.#hosted.live.automationRuns, + { id: automationId, [LIVE_USER_QUERY]: userId }, + poll, + after, + ), + ); + } } diff --git a/packages/features/src/automation-live.ts b/packages/features/src/automation-live.ts new file mode 100644 index 0000000000..fcae61bb40 --- /dev/null +++ b/packages/features/src/automation-live.ts @@ -0,0 +1,264 @@ +import { Schema as EffectSchema } from "effect"; +import { + AUTOMATION_TARGET, + AUTOMATION_TRIGGER, + type AutomationRunSummary, + type AutomationSummary, + automationActionSchema, + automationFiringSchema, + automationGuardsSchema, + automationRunReasonSchema, + automationRunStatusSchema, + automationScopeSchema, + automationTargetKindSchema, + automationTriggerSchema, +} from "./automation-wire.js"; +import { codingAgentStatusAt, type StatusTurn } from "./coding-agent-status.js"; +import { CODING_AGENT_FAILURE, CODING_AGENT_STATUS } from "./coding-agent-wire.js"; +import { sealedLiveQuerySchema } from "./live-wire.js"; +import { automationWebhookPath } from "./service-paths.js"; +import { countedNumber } from "./service-wire.js"; + +/** + * automation-live.ts -- a workspace's automations and one automation's runs followed over Neon Realtime: the sealed statements the service answers, the rows they deliver, and the list and the history as the held reads answer them, built from the rows at an instant. + * + * A run's agent's status turns on the clock (`coding-agent-status.ts`), so + * the reader builds each at the instant it reads, and again at + * `automationsGraceEdge` or `automationRunsGraceEdge`, the next instant the + * clock alone moves one. A webhook automation is answered with the address + * its caller posts to on the service the reader reaches, as the held read's + * client answers it. Realtime keeps no order, so the reader orders. + */ + +/** A workspace's automations sealed for following: the address, and one statement. The service answers `unavailable` where its deployment's Realtime is misconfigured. */ +export const automationsLiveAnswerSchema = EffectSchema.Struct({ + url: EffectSchema.String, + queries: EffectSchema.Struct({ automations: sealedLiveQuerySchema }), +}); + +export type AutomationsLiveAnswer = typeof automationsLiveAnswerSchema.Type; + +/** An automation's runs sealed for following: the address, and one statement. The service answers `not-found` for an automation the account does not read. */ +export const automationRunsLiveAnswerSchema = EffectSchema.Struct({ + url: EffectSchema.String, + queries: EffectSchema.Struct({ runs: sealedLiveQuerySchema }), +}); + +export type AutomationRunsLiveAnswer = typeof automationRunsLiveAnswerSchema.Type; + +/** The agent a run started as its row delivers it, and what its status is read from; null throughout where it started none, or it is gone. */ +const runAgentFields = { + agentId: EffectSchema.NullOr(EffectSchema.String), + agentTitle: EffectSchema.NullOr(EffectSchema.String), + agentCreatedAt: EffectSchema.NullOr(countedNumber), + ownKey: EffectSchema.NullOr(EffectSchema.Boolean), + turnId: EffectSchema.NullOr(EffectSchema.String), + turnStatus: EffectSchema.NullOr(EffectSchema.String), + cancelRequested: EffectSchema.NullOr(EffectSchema.Boolean), + turnQueuedAt: EffectSchema.NullOr(countedNumber), + turnSettledAt: EffectSchema.NullOr(countedNumber), + turnSessionEnded: EffectSchema.NullOr(EffectSchema.Boolean), + awaitingLineAt: EffectSchema.NullOr(countedNumber), + failureWord: EffectSchema.NullOr(EffectSchema.Literals(Object.values(CODING_AGENT_FAILURE))), +}; + +/** One run as its row delivers it. */ +const runLiveRowSchema = EffectSchema.Struct({ + id: EffectSchema.String, + firing: automationFiringSchema, + status: automationRunStatusSchema, + reason: EffectSchema.NullOr(automationRunReasonSchema), + firedAt: countedNumber, + settledAt: EffectSchema.NullOr(countedNumber), + ...runAgentFields, +}); + +/** One automation as its row delivers it, its newest run's columns beside it, null where it has run none. */ +const automationLiveRowSchema = EffectSchema.Struct({ + id: EffectSchema.String, + userId: EffectSchema.String, + taskId: EffectSchema.NullOr(EffectSchema.String), + taskName: EffectSchema.NullOr(EffectSchema.String), + name: EffectSchema.String, + enabled: EffectSchema.Boolean, + trigger: automationTriggerSchema, + action: automationActionSchema, + guards: automationGuardsSchema, + nextRunAt: EffectSchema.NullOr(countedNumber), + organizationId: EffectSchema.NullOr(EffectSchema.String), + scope: automationScopeSchema, + repositories: EffectSchema.Array(EffectSchema.String), + /** The task's repository, which an automation with none of its own works in, as the held read reads it. */ + taskRepository: EffectSchema.NullOr(EffectSchema.String), + targetKind: automationTargetKindSchema, + targetAgentId: EffectSchema.NullOr(EffectSchema.String), + archiveSuccessful: EffectSchema.Boolean, + createdAt: countedNumber, + mine: EffectSchema.Boolean, + runId: EffectSchema.NullOr(EffectSchema.String), + runFiring: EffectSchema.NullOr(automationFiringSchema), + runStatus: EffectSchema.NullOr(automationRunStatusSchema), + runReason: EffectSchema.NullOr(automationRunReasonSchema), + runFiredAt: EffectSchema.NullOr(countedNumber), + runSettledAt: EffectSchema.NullOr(countedNumber), + ...runAgentFields, +}); + +/** A workspace's automations as Realtime delivers them, decoded at the boundary before anything reads them. */ +export const automationsLiveRowsSchema = EffectSchema.Struct({ + automations: EffectSchema.Array(automationLiveRowSchema), +}); + +export type AutomationsLiveRows = typeof automationsLiveRowsSchema.Type; + +/** An automation's runs as Realtime delivers them, decoded at the boundary before anything reads them. */ +export const automationRunsLiveRowsSchema = EffectSchema.Struct({ + runs: EffectSchema.Array(runLiveRowSchema), +}); + +export type AutomationRunsLiveRows = typeof automationRunsLiveRowsSchema.Type; + +type RunAgentRow = { + readonly [Key in keyof typeof runAgentFields]: (typeof runAgentFields)[Key]["Type"]; +}; + +/** The newest turn as the status rule reads it; none where the agent has none. */ +function turnOf(row: RunAgentRow): StatusTurn | undefined { + return row.turnId === null || row.turnStatus === null + ? undefined + : { + status: row.turnStatus, + cancelRequested: row.cancelRequested === true, + queuedAt: row.turnQueuedAt ?? row.agentCreatedAt ?? 0, + settledAt: row.turnSettledAt, + sessionEnded: row.turnSessionEnded === true, + }; +} + +/** A run's agent as the wire says it, its status read at `now` and why it failed where it did; none where the run started none. */ +function runAgentOf(row: RunAgentRow, now: number): AutomationRunSummary["agent"] { + if (row.agentId === null || row.agentCreatedAt === null) return null; + const status = codingAgentStatusAt(turnOf(row), { + createdAt: row.agentCreatedAt, + now, + awaitingLineAt: row.awaitingLineAt ?? undefined, + }); + const failureReason = + status === CODING_AGENT_STATUS.FAILED && row.failureWord !== null ? row.failureWord : undefined; + return { + id: row.agentId, + title: row.agentTitle, + status, + ownKey: row.ownKey === true, + ...(failureReason === undefined ? undefined : { failureReason }), + }; +} + +/** Latest fired first, and by id among those fired at one instant, as the held history orders runs. */ +function latestFired( + a: { readonly id: string; readonly firedAt: number }, + b: { readonly id: string; readonly firedAt: number }, +): number { + if (a.firedAt !== b.firedAt) return b.firedAt - a.firedAt; + return a.id < b.id ? 1 : a.id > b.id ? -1 : 0; +} + +/** One automation's newest run as the list says it; none where it has run none. */ +function lastRunOf( + row: typeof automationLiveRowSchema.Type, + now: number, +): AutomationRunSummary | null { + if ( + row.runId === null || + row.runFiring === null || + row.runStatus === null || + row.runFiredAt === null + ) { + return null; + } + return { + id: row.runId, + firing: row.runFiring, + status: row.runStatus, + reason: row.runReason, + agent: runAgentOf(row, now), + firedAt: row.runFiredAt, + settledAt: row.runSettledAt, + }; +} + +/** A webhook automation with the address its caller posts to, on the service at `serviceBaseUrl`; any other as it came. */ +export function withWebhookUrl( + serviceBaseUrl: string, + automation: AutomationSummary, +): AutomationSummary { + if (automation.trigger.kind !== AUTOMATION_TRIGGER.WEBHOOK) return automation; + return { ...automation, webhookUrl: `${serviceBaseUrl}${automationWebhookPath(automation.id)}` }; +} + +/** + * A workspace's automations as the held list answers them through a client + * of the service at `serviceBaseUrl`, as the live rows stand at `now`: + * newest made first, each with its newest run. + */ +export function automationsOfLiveRows( + rows: AutomationsLiveRows, + now: number, + serviceBaseUrl: string, +): readonly AutomationSummary[] { + return rows.automations + .toSorted((a, b) => + a.createdAt !== b.createdAt + ? b.createdAt - a.createdAt + : a.id < b.id + ? 1 + : a.id > b.id + ? -1 + : 0, + ) + .map((row) => + withWebhookUrl(serviceBaseUrl, { + id: row.id, + planId: row.taskId, + planName: row.taskName, + ...(row.organizationId === null ? undefined : { workspaceId: row.organizationId }), + name: row.name, + enabled: row.enabled, + pausedReason: null, + trigger: row.trigger, + action: row.action, + guards: row.guards, + nextRunAt: row.nextRunAt, + lastRun: lastRunOf(row, now), + createdAt: row.createdAt, + scope: row.scope, + repositories: + row.repositories.length === 0 && row.taskRepository !== null + ? [row.taskRepository] + : row.repositories, + target: + row.targetKind === AUTOMATION_TARGET.SAME_AGENT + ? { kind: AUTOMATION_TARGET.SAME_AGENT, agentId: row.targetAgentId } + : { kind: AUTOMATION_TARGET.NEW_AGENT }, + archiveSuccessful: row.archiveSuccessful, + mine: row.mine, + creatorId: row.userId, + }), + ); +} + +/** An automation's runs as the held history answers them, as the live rows stand at `now`: latest fired first. */ +export function automationRunsOfLiveRows( + rows: AutomationRunsLiveRows, + now: number, +): readonly AutomationRunSummary[] { + return rows.runs.toSorted(latestFired).map((row) => ({ + id: row.id, + firing: row.firing, + status: row.status, + reason: row.reason, + agent: runAgentOf(row, now), + firedAt: row.firedAt, + settledAt: row.settledAt, + })); +} diff --git a/packages/features/src/contract/live.ts b/packages/features/src/contract/live.ts index 13fd1467d7..7c32d6cdbc 100644 --- a/packages/features/src/contract/live.ts +++ b/packages/features/src/contract/live.ts @@ -27,6 +27,7 @@ import { HTTP_METHOD } from "@luke/wire"; import { openapi } from "@orpc/openapi"; import { Schema as EffectSchema } from "effect"; +import { automationRunsLiveAnswerSchema, automationsLiveAnswerSchema } from "../automation-live.js"; import { accountInboxLiveAnswerSchema, agentTranscriptLiveAnswerSchema, @@ -140,6 +141,21 @@ export const liveContract = { .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_TASKS })) .input(liveReadSchema) .output(EffectSchema.flip(taskListLiveAnswerSchema)), + automations: hostedProcedure + .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_AUTOMATIONS })) + .input( + EffectSchema.Struct({ + workspaceId: EffectSchema.String, + [LIVE_POLL.QUERY]: pollField, + [LIVE_POLL.AFTER]: afterField, + [LIVE_USER_QUERY]: userField, + }), + ) + .output(EffectSchema.flip(automationsLiveAnswerSchema)), + automationRuns: hostedProcedure + .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_AUTOMATION_RUNS })) + .input(liveTaskSchema) + .output(EffectSchema.flip(automationRunsLiveAnswerSchema)), agentTranscript: hostedProcedure .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_AGENT_TRANSCRIPT })) .input(liveTranscriptSchema) diff --git a/packages/features/src/service-paths.ts b/packages/features/src/service-paths.ts index b6a8237d88..1fcb17cb88 100644 --- a/packages/features/src/service-paths.ts +++ b/packages/features/src/service-paths.ts @@ -107,6 +107,10 @@ export const HOSTED_SERVICE_PATH = { LIVE_WORKSPACE_CHAT: "/api/live/workspace-chat", /** The task list sealed for following over Neon Realtime (GET): its tasks and every agent of them and with no task, or `unavailable`. `task-list-live.ts` declares the answer. */ LIVE_TASKS: "/api/live/tasks", + /** A workspace's automations under `?workspaceId=`, sealed for following over Neon Realtime (GET), or `unavailable`. `automation-live.ts` declares the answer. */ + LIVE_AUTOMATIONS: "/api/live/automations", + /** One automation's runs under `?id=`, sealed for following over Neon Realtime (GET), or `not-found` or `unavailable`. */ + LIVE_AUTOMATION_RUNS: "/api/live/automation-runs", /** * A workspace's agents, on a task or with none, newest started first and * the archived ones left out unless the query asks for them, narrowed to diff --git a/packages/service/src/agents/coder-host/status.ts b/packages/service/src/agents/coder-host/status.ts index 05e031763b..ae7c28ed66 100644 --- a/packages/service/src/agents/coder-host/status.ts +++ b/packages/service/src/agents/coder-host/status.ts @@ -154,7 +154,7 @@ const CONTEXT_OVERFLOW_WORDS: readonly string[] = [ ]; /** The host's refusal sentences, and a provider's words for a prompt past the window, a failed turn's detail may carry, by the word the wire says for each. */ -const FAILURE_WORDS: readonly (readonly [CodingAgentFailure, readonly string[]])[] = [ +export const FAILURE_WORDS: readonly (readonly [CodingAgentFailure, readonly string[]])[] = [ [ CODING_AGENT_FAILURE.GITHUB, [ diff --git a/packages/service/src/automations/automation-live.test.ts b/packages/service/src/automations/automation-live.test.ts new file mode 100644 index 0000000000..92cffcb192 --- /dev/null +++ b/packages/service/src/automations/automation-live.test.ts @@ -0,0 +1,28 @@ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { it } from "@effect/vitest"; +import { getTestDB, mutableFunctionsOf } from "@luke/database/testing"; +import { Effect, Option } from "effect"; +import { automationRunsLiveStatements, automationsLiveStatements } from "./automation-live.js"; + +/** + * A workspace's automations and an automation's runs, as their live + * endpoints seal them, call only functions Postgres marks immutable, since + * Neon Realtime refuses any other ("scalar expression uses a mutable + * function") where Postgres runs it without a word. + */ + +it.layer(getTestDB())("the automations' live statements", (it) => { + it.effect("call only functions Postgres marks immutable, as Realtime requires", () => + Effect.gen(function* () { + const { automations } = Option.getOrThrow( + automationsLiveStatements("user-immutable", "workspace-immutable"), + ); + const { runs } = Option.getOrThrow( + automationRunsLiveStatements("user-immutable", randomUUID()), + ); + assert.deepEqual(yield* mutableFunctionsOf(automations), []); + assert.deepEqual(yield* mutableFunctionsOf(runs), []); + }), + ); +}); diff --git a/packages/service/src/automations/automation-live.ts b/packages/service/src/automations/automation-live.ts new file mode 100644 index 0000000000..17160816d3 --- /dev/null +++ b/packages/service/src/automations/automation-live.ts @@ -0,0 +1,167 @@ +import { + type LiveStatement, + liveSelection, + liveStatementOf, +} from "@luke/database/core/live-statement"; +import { queryBuilder } from "@luke/database/core/query"; +import { + AUTOMATION_COLUMNS, + AUTOMATION_WORKSPACE, + automationReadableBy, +} from "@luke/database/models/automation-store"; +import { automation, automationRun } from "@luke/database/schemas/automation"; +import { codingAgent } from "@luke/database/schemas/coding-agent"; +import { task } from "@luke/database/schemas/task"; +import { and, desc, eq, inArray, isNotNull, sql } from "drizzle-orm"; +import { Option } from "effect"; +import { + awaitingLineAtIn, + failureWordOf, + latestTurnOf, + sessionEndedOf, +} from "../live/agent-standing.js"; + +/** + * automation-live.ts -- a workspace's automations and one automation's runs, written for Neon Realtime. + * + * Each is the same rows and rule of who may read them as its held read + * (`automations-app.ts`'s list and runs, through `listAutomations` and + * `listAutomationRuns`): every automation of the workspace the account may + * read, each with its newest run, and an automation's newest runs, each + * run with the agent it started and what that agent's status is read from + * (`agent-standing.ts`), in one statement apiece, so a run and its agent + * move together. Every column stands under the key + * `@luke/features/automation-live` reads it by and every instant as epoch + * milliseconds. Nothing reads the clock: an agent's status, which turns on + * it, is the reader's, and so is the order, since Realtime keeps none. + */ + +/** The most runs an automation's live history holds, as its held read answers them. */ +export const AUTOMATION_RUNS_READ = 50; + +/** The agent a run started and what its status is read from, each under the key the wire reads it by: the agent and turn joined beside the run. */ +function runAgentSelection(userId: string, latest: ReturnType) { + return liveSelection( + { + agentId: codingAgent.id, + agentTitle: codingAgent.title, + ownKey: eq(codingAgent.userId, userId), + turnId: latest.id, + turnStatus: latest.status, + cancelRequested: isNotNull(latest.cancelRequestedAt), + turnSessionEnded: sessionEndedOf(latest), + failureWord: failureWordOf(latest), + }, + { + agentCreatedAt: codingAgent.createdAt, + turnQueuedAt: latest.queuedAt, + turnSettledAt: latest.settledAt, + awaitingLineAt: awaitingLineAtIn(codingAgent.conversationId), + }, + ); +} + +/** The statement a workspace's automations are subscribed to live. */ +export interface AutomationsLiveStatements { + readonly automations: LiveStatement; +} + +/** Every automation of the workspace the account may read, each with its newest run and that run's agent, written for Realtime. */ +export function automationsLiveStatements( + userId: string, + workspaceId: string, +): Option.Option { + const lastRun = queryBuilder + .select({ + id: automationRun.id, + firing: automationRun.firing, + status: automationRun.status, + reason: automationRun.reason, + codingAgentId: automationRun.codingAgentId, + firedAt: automationRun.firedAt, + settledAt: automationRun.settledAt, + }) + .from(automationRun) + .where(eq(automationRun.automationId, automation.id)) + .orderBy(desc(automationRun.firedAt), desc(automationRun.id)) + .limit(1) + .as("last_run"); + const latest = latestTurnOf(codingAgent.conversationId, "latest_turn"); + const { + nextRunAt, + createdAt, + updatedAt: _updatedAt, + repositories: _repositories, + ...columns + } = AUTOMATION_COLUMNS; + const statement = queryBuilder + .select({ + ...liveSelection( + // Note that the row's own repositories are selected bare, and its task's beside them, + // because the held read's fallback builds an array with a function Realtime refuses. + { ...columns, repositories: automation.repositories, mine: eq(automation.userId, userId) }, + { nextRunAt, createdAt }, + ), + ...liveSelection( + { + runId: lastRun.id, + runFiring: lastRun.firing, + runStatus: lastRun.status, + runReason: lastRun.reason, + }, + { runFiredAt: lastRun.firedAt, runSettledAt: lastRun.settledAt }, + ), + ...runAgentSelection(userId, latest), + }) + .from(automation) + .leftJoin(task, eq(task.id, automation.taskId)) + .leftJoinLateral(lastRun, sql`true`) + .leftJoin(codingAgent, eq(codingAgent.id, lastRun.codingAgentId)) + .leftJoinLateral(latest, sql`true`) + .where(and(automationReadableBy(userId), eq(AUTOMATION_WORKSPACE, workspaceId))); + return Option.map(liveStatementOf(statement), (automations) => ({ automations })); +} + +/** The statement an automation's runs are subscribed to live. */ +export interface AutomationRunsLiveStatements { + readonly runs: LiveStatement; +} + +/** The automation's newest runs the account may read, each with its agent, written for Realtime; the reader keeps the order. */ +export function automationRunsLiveStatements( + userId: string, + automationId: string, +): Option.Option { + const newest = queryBuilder + .select({ id: automationRun.id }) + .from(automationRun) + .where(eq(automationRun.automationId, automationId)) + .orderBy(desc(automationRun.firedAt), desc(automationRun.id)) + .limit(AUTOMATION_RUNS_READ); + const latest = latestTurnOf(codingAgent.conversationId, "latest_turn"); + const statement = queryBuilder + .select({ + ...liveSelection( + { + id: automationRun.id, + firing: automationRun.firing, + status: automationRun.status, + reason: automationRun.reason, + }, + { firedAt: automationRun.firedAt, settledAt: automationRun.settledAt }, + ), + ...runAgentSelection(userId, latest), + }) + .from(automationRun) + .innerJoin(automation, eq(automation.id, automationRun.automationId)) + .leftJoin(codingAgent, eq(codingAgent.id, automationRun.codingAgentId)) + .leftJoinLateral(latest, sql`true`) + .where( + and( + eq(automation.id, automationId), + automationReadableBy(userId), + inArray(automationRun.id, newest), + ), + ); + return Option.map(liveStatementOf(statement), (runs) => ({ runs })); +} diff --git a/packages/service/src/automations/automations-procedures.test.ts b/packages/service/src/automations/automations-procedures.test.ts index abfc7a1ecf..3d1ba509e1 100644 --- a/packages/service/src/automations/automations-procedures.test.ts +++ b/packages/service/src/automations/automations-procedures.test.ts @@ -7,7 +7,7 @@ import { createTask } from "@luke/database/models/task-store"; import { member as memberTable } from "@luke/database/schemas/auth"; import { automation as automationTable } from "@luke/database/schemas/automation"; import { personalWorkspace } from "@luke/database/schemas/personal-workspace"; -import { turns } from "@luke/database/schemas/storage"; +import { messages, turns } from "@luke/database/schemas/storage"; import { getTestDB, joinWorkspace, @@ -16,6 +16,12 @@ import { startTask, } from "@luke/database/testing"; import { automationPath, HOSTED_API_ERROR, HOSTED_SERVICE_PATH } from "@luke/features"; +import { + automationRunsLiveRowsSchema, + automationRunsOfLiveRows, + automationsLiveRowsSchema, + automationsOfLiveRows, +} from "@luke/features/automation-live"; import { SCHEDULE_PRESET } from "@luke/features/automation-schedule"; import { AUTOMATION_ACTION, @@ -36,7 +42,14 @@ import { CODING_AGENT_FAILURE, CODING_AGENT_STATUS } from "@luke/features/coding import { INTEGRATION_PROVIDER } from "@luke/features/integration-wire"; import { CODING_AGENT_DEFAULT_CHOICE, MODEL_PROVIDER } from "@luke/features/models-wire"; import { WORKSPACE_ROLE } from "@luke/features/workspace-wire"; -import { TURN_ORIGIN, TURN_STATUS, unparsedWire, type WireBoundaryInput } from "@luke/wire"; +import { + MESSAGE_ROLE, + TURN_ORIGIN, + TURN_STATUS, + type TurnStatus, + unparsedWire, + type WireBoundaryInput, +} from "@luke/wire"; import { readEither } from "@luke/wire/effect"; import { eq } from "drizzle-orm"; import { @@ -47,7 +60,7 @@ import { Option, Redacted, Result, - type Schema, + Schema, Stream, } from "effect"; import type * as HttpClient from "effect/http/HttpClient"; @@ -58,6 +71,7 @@ import { EVE_RESET_OUTCOME, EVE_SEND_OUTCOME, } from "../agents/brain-host/eve-sessions.js"; +import { CODER_REFUSAL } from "../agents/coder-host/bounds.js"; import { requestUserId } from "../auth/bearer.js"; import { BRAIN_REQUEST_FAILURE } from "../core.js"; import type { GitHubApp } from "../github/github-app.js"; @@ -75,6 +89,7 @@ import { type InstallationFixture, openGithubUser, } from "../testing/github-app-fake.js"; +import { automationRunsLiveStatements, automationsLiveStatements } from "./automation-live.js"; import { automationsProcedures } from "./automations-procedures.js"; /** An automation's "Run now" (POST). */ @@ -668,6 +683,147 @@ it.layer(getTestDB())("the automation routes", (it) => { }), ); + it.effect( + "the live statements of a workspace's automations and of one's runs deliver what the held list and history answer, each run's agent and why it failed alike, and nothing to another account", + () => + Effect.gen(function* () { + const { owner, other, task: plan, ask } = yield* openAccounts; + const made = (name: string) => + Effect.map( + ask(HOSTED_SERVICE_PATH.AUTOMATIONS, owner, { + method: "POST", + body: { ...weekdays(plan.id), name }, + }), + (answer) => + readAnswer(automationAnswerSchema, HOSTED_HTTP_STATUS.CREATED, answer).automation, + ); + const ran = yield* made("Morning triage"); + yield* made("Never ran"); + const runOnce = (automationId = ran.id) => + Effect.map( + ask(automationRunPath(automationId), owner, { + method: "POST", + body: { idempotencyKey: randomUUID() }, + }), + (answer) => + readAnswer(automationRunAnswerSchema, HOSTED_HTTP_STATUS.CREATED, answer).run, + ); + const refused = yield* runOnce(); + const agent = Option.getOrThrow(yield* readCodingAgent(owner, refused.agent?.id ?? "")); + // Its turn failed on the model call, GitHub's refusal in its detail. + yield* Database.use((db) => + db.insert(turns).values({ + userId: owner, + conversationId: agent.conversationId, + origin: TURN_ORIGIN.TYPED, + status: TURN_STATUS.FAILED, + failure: BRAIN_REQUEST_FAILURE.MODEL, + failureDetail: `checkout: ${CODER_REFUSAL.NOT_REACHABLE}`, + }), + ); + yield* runOnce(); + // A run whose agent's session ended for good, a line written after it, and one whose + // settled turn a line of the developer's still awaits. + const turnThenLine = (name: string, status: TurnStatus, failure: string | null) => + Effect.gen(function* () { + const automation = yield* made(name); + const started = yield* runOnce(automation.id); + const each = Option.getOrThrow(yield* readCodingAgent(owner, started.agent?.id ?? "")); + yield* Database.use((db) => + db.insert(turns).values({ + userId: owner, + conversationId: each.conversationId, + origin: TURN_ORIGIN.TYPED, + status, + failure, + settledAt: new Date(), + }), + ); + yield* Database.use((db) => + db.insert(messages).values({ + userId: owner, + conversationId: each.conversationId, + seq: 1, + clientId: randomUUID(), + role: MESSAGE_ROLE.USER, + parts: [{ type: "text", text: "Are you there?" }], + placedAt: new Date(), + turnId: null, + }), + ); + return automation.id; + }); + const ended = yield* turnThenLine( + "Session ended", + TURN_STATUS.FAILED, + BRAIN_REQUEST_FAILURE.SESSION_ENDED, + ); + const awaiting = yield* turnThenLine("Line awaits", TURN_STATUS.SETTLED, null); + + const now = yield* Clock.currentTimeMillis; + const sql = yield* SqlClient.SqlClient; + const run = (statement: { readonly sql: string; readonly params: readonly unknown[] }) => + sql.unsafe(statement.sql, [...statement.params]); + for (const reader of [owner, other]) { + const statements = Option.getOrThrow(automationsLiveStatements(reader, plan.workspaceId)); + const rows = yield* Schema.decodeUnknownEffect(automationsLiveRowsSchema)({ + automations: yield* run(statements.automations), + }); + const held = readAnswer( + automationListAnswerSchema, + HOSTED_HTTP_STATUS.OK, + yield* ask( + `${HOSTED_SERVICE_PATH.AUTOMATIONS}?workspaceId=${plan.workspaceId}`, + reader, + ), + ); + assert.deepEqual(automationsOfLiveRows(rows, now, ORIGIN), held.automations); + } + const statements = Option.getOrThrow(automationRunsLiveStatements(owner, ran.id)); + const runs = automationRunsOfLiveRows( + yield* Schema.decodeUnknownEffect(automationRunsLiveRowsSchema)({ + runs: yield* run(statements.runs), + }), + now, + ); + const history = readAnswer( + automationRunsAnswerSchema, + HOSTED_HTTP_STATUS.OK, + yield* ask(automationRunsPath(ran.id), owner), + ); + assert.deepEqual(runs, history.runs); + assert.deepEqual( + runs.map((each) => `${each.agent?.status}:${each.agent?.failureReason ?? ""}`).toSorted(), + [ + `${CODING_AGENT_STATUS.FAILED}:${CODING_AGENT_FAILURE.GITHUB}`, + `${CODING_AGENT_STATUS.STARTING}:`, + ], + ); + // A line after a session that ended for good awaits nothing; one after a settled turn awaits it. + for (const [automationId, said] of [ + [ended, `${CODING_AGENT_STATUS.FAILED}:${CODING_AGENT_FAILURE.SESSION_ENDED}`], + [awaiting, `${CODING_AGENT_STATUS.RUNNING}:`], + ] as const) { + const read = Option.getOrThrow(automationRunsLiveStatements(owner, automationId)); + const [only] = automationRunsOfLiveRows( + yield* Schema.decodeUnknownEffect(automationRunsLiveRowsSchema)({ + runs: yield* run(read.runs), + }), + now, + ); + const heldRuns = readAnswer( + automationRunsAnswerSchema, + HOSTED_HTTP_STATUS.OK, + yield* ask(automationRunsPath(automationId), owner), + ).runs; + assert.deepEqual([only], heldRuns); + assert.equal(`${only?.agent?.status}:${only?.agent?.failureReason ?? ""}`, said); + } + const outsiders = Option.getOrThrow(automationRunsLiveStatements(other, ran.id)); + assert.deepEqual(yield* run(outsiders.runs), []); + }), + ); + it.effect( "a run whose agent failed is read back in the history with why it failed, as the agent's own end line says it", () => diff --git a/packages/service/src/automations/automations-procedures.ts b/packages/service/src/automations/automations-procedures.ts index 06ac7aa804..0a2c86b53b 100644 --- a/packages/service/src/automations/automations-procedures.ts +++ b/packages/service/src/automations/automations-procedures.ts @@ -59,6 +59,7 @@ import { } from "../http/http-effect.js"; import { accountMiddleware, hosted } from "../orpc/base.js"; import type { ModelCatalog } from "../providers/model-catalog.js"; +import { AUTOMATION_RUNS_READ } from "./automation-live.js"; import { executeAutomationRun } from "./automation-runs.js"; /** @@ -85,9 +86,6 @@ import { executeAutomationRun } from "./automation-runs.js"; * run's agent is summarized with its status as the agent's tab reads it. */ -/** How many runs a history read answers. */ -const RUNS_READ = 50; - /** The bytes of randomness in a webhook trigger's secret. */ const WEBHOOK_SECRET_BYTES = 32; @@ -578,7 +576,9 @@ export function automationsProcedures(seams: AutomationsProcedureSeams) { const id = yield* idOf(input.id); const automation = yield* hostedStoreOrUnavailable(readAutomation(userId, id)); if (Option.isNone(automation)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); - const runs = yield* hostedStoreOrUnavailable(listAutomationRuns(userId, id, RUNS_READ)); + const runs = yield* hostedStoreOrUnavailable( + listAutomationRuns(userId, id, AUTOMATION_RUNS_READ), + ); return { runs: yield* runSummaries(userId, runs) }; }), }; diff --git a/packages/service/src/live/agent-standing.ts b/packages/service/src/live/agent-standing.ts new file mode 100644 index 0000000000..c499f8e1f8 --- /dev/null +++ b/packages/service/src/live/agent-standing.ts @@ -0,0 +1,80 @@ +import { queryBuilder } from "@luke/database/core/query"; +import { newestAwaitingLineAt } from "@luke/database/models/message-reads"; +import { turns } from "@luke/database/schemas/storage"; +import { CODING_AGENT_FAILURE } from "@luke/features/coding-agent-wire"; +import { desc, eq, type SQL, type SQLWrapper, sql } from "drizzle-orm"; +import { FAILURE_WORDS } from "../agents/coder-host/status.js"; +import { BRAIN_REQUEST_FAILURE, TURN_STATUS } from "../core.js"; + +/** + * agent-standing.ts -- what a coding agent's status is read from, written for Neon Realtime: its newest turn, when the newest line of the developer's awaiting its turn was written, and why its newest turn failed. + * + * Each is the same rule the held reads read an agent by + * (`message-reads.ts`'s `findLatestTurns` and `awaitingLinesOfReached`, and + * `coder-host/status.ts`'s `codingAgentFailureOf`), as a fragment a live + * statement joins or selects beside the agent's own row. Nothing here reads + * the clock: the status itself, which turns on it, is the reader's to read + * (`@luke/features/coding-agent-status`). The failure is named here, in + * the wire's own word, so the turn's detail never leaves the database. + */ + +/** The newest turn of the conversation named, by the order turns were queued, as a lateral subquery under `name`. */ +export function latestTurnOf(conversationId: SQLWrapper, name: string) { + return queryBuilder + .select({ + id: turns.id, + status: turns.status, + failure: turns.failure, + failureDetail: turns.failureDetail, + cancelRequestedAt: turns.cancelRequestedAt, + queuedAt: turns.queuedAt, + startedAt: turns.startedAt, + settledAt: turns.settledAt, + }) + .from(turns) + .where(eq(turns.conversationId, sql`${conversationId}`)) + .orderBy(desc(turns.queuedAt), desc(turns.id)) + .limit(1) + .as(name); +} + +/** The latest turn as `latestTurnOf` names it. */ +type LatestTurn = ReturnType; + +/** When the newest line of the developer's awaiting its turn in the conversation named was written, or null, as `awaitingLinesOfReached` reads it. */ +export function awaitingLineAtIn(conversationId: SQLWrapper): SQL { + return newestAwaitingLineAt(sql`${conversationId}`); +} + +/** Whether the newest turn failed because its session ended for good, which leaves every line awaiting nothing; null where no turn stands. */ +export function sessionEndedOf(latest: LatestTurn): SQL { + return sql`${latest.failure} = ${BRAIN_REQUEST_FAILURE.SESSION_ENDED}`; +} + +/** + * Why the agent would read as failed, in the wire's word, as + * `codingAgentFailureOf` names it: a session that never started where no + * turn stands, the sweep's word or the host's refusal a failed turn + * carries, and none for a turn that did not fail. The reader says it only + * where the status it reads is failed. + */ +export function failureWordOf(latest: LatestTurn): SQL { + const named = FAILURE_WORDS.map( + ([word, refusals]) => + sql`when ${sql.join( + refusals.map( + (refusal) => sql`strpos(coalesce(${latest.failureDetail}, ''), ${refusal}) > 0`, + ), + sql` or `, + )} then ${word}`, + ); + return sql`case + when ${latest.id} is null then ${CODING_AGENT_FAILURE.NOT_STARTED} + when ${latest.status} <> ${TURN_STATUS.FAILED} then null + when ${latest.failure} = ${BRAIN_REQUEST_FAILURE.ABANDONED} then ${CODING_AGENT_FAILURE.ABANDONED} + when ${latest.failure} = ${BRAIN_REQUEST_FAILURE.SESSION_ENDED} then ${CODING_AGENT_FAILURE.SESSION_ENDED} + when ${latest.failure} is distinct from ${BRAIN_REQUEST_FAILURE.MODEL} then ${CODING_AGENT_FAILURE.OTHER} + ${sql.join(named, sql` `)} + else ${CODING_AGENT_FAILURE.MODEL} + end`; +} diff --git a/packages/service/src/live/live-procedures.ts b/packages/service/src/live/live-procedures.ts index 34124a55ee..87a7d588e9 100644 --- a/packages/service/src/live/live-procedures.ts +++ b/packages/service/src/live/live-procedures.ts @@ -52,7 +52,7 @@ */ import type { LiveStatement } from "@luke/database/core/live-statement"; -import { ownFailedRunsLiveStatement } from "@luke/database/models/automation-store"; +import { ownFailedRunsLiveStatement, readAutomation } from "@luke/database/models/automation-store"; import { boardLiveStatement } from "@luke/database/models/board-store"; import { workspaceChatLiveStatements } from "@luke/database/models/chat-store"; import { readCodingAgent } from "@luke/database/models/coding-agent-store"; @@ -78,6 +78,10 @@ import { transcriptLiveStatements, } from "../agents/coder-host/transcript-live.js"; import { chatAccount, workspaceOf } from "../agents/workspace-chat.js"; +import { + automationRunsLiveStatements, + automationsLiveStatements, +} from "../automations/automation-live.js"; import { unparsedWire, wireUuidSchema } from "../core.js"; import { heldUntil } from "../http/held-read.js"; import { @@ -231,6 +235,24 @@ export function liveProcedures(seams: LiveProcedureSeams) { if (Option.isNone(statements)) return yield* Effect.fail(HOSTED_REFUSAL.UNAVAILABLE); return yield* sealedParts(statements.value, input); }), + automations: signedIn.automations.effect(function* ({ input, context }) { + yield* sameUser(input, context.userId); + if (input.workspaceId === "") return yield* Effect.fail(HOSTED_REFUSAL.INVALID_REQUEST); + const statements = automationsLiveStatements(context.userId, input.workspaceId); + if (Option.isNone(statements)) return yield* Effect.fail(HOSTED_REFUSAL.UNAVAILABLE); + return yield* sealedParts(statements.value, input); + }), + automationRuns: signedIn.automationRuns.effect(function* ({ input, context }) { + yield* sameUser(input, context.userId); + const automationId = yield* rowIdOf(input.id); + const automation = yield* hostedStoreOrUnavailable( + readAutomation(context.userId, automationId), + ); + if (Option.isNone(automation)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); + const statements = automationRunsLiveStatements(context.userId, automationId); + if (Option.isNone(statements)) return yield* Effect.fail(HOSTED_REFUSAL.UNAVAILABLE); + return yield* sealedParts(statements.value, input); + }), taskBoard: signedIn.taskBoard.effect(function* ({ input, context }) { yield* sameUser(input, context.userId); const taskId = yield* rowIdOf(input.id); diff --git a/packages/service/src/tasks/task-list-live.ts b/packages/service/src/tasks/task-list-live.ts index 5456751129..0cfe99a518 100644 --- a/packages/service/src/tasks/task-list-live.ts +++ b/packages/service/src/tasks/task-list-live.ts @@ -4,7 +4,6 @@ import { liveStatementOf, } from "@luke/database/core/live-statement"; import { queryBuilder } from "@luke/database/core/query"; -import { newestAwaitingLineAt } from "@luke/database/models/message-reads"; import { AGENT_WORKSPACE, agentReachableBy, @@ -14,13 +13,12 @@ import { import { agentPin, taskPin } from "@luke/database/schemas"; import { user } from "@luke/database/schemas/auth"; import { codingAgent } from "@luke/database/schemas/coding-agent"; -import { turns } from "@luke/database/schemas/storage"; import { task } from "@luke/database/schemas/task"; import { CODING_AGENT_BRIEF } from "@luke/features/coding-agent-wire"; -import { and, desc, eq, isNotNull, isNull, or, sql } from "drizzle-orm"; +import { and, eq, isNotNull, isNull, or, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import { Option } from "effect"; -import { BRAIN_REQUEST_FAILURE } from "../core.js"; +import { awaitingLineAtIn, latestTurnOf, sessionEndedOf } from "../live/agent-standing.js"; /** * task-list-live.ts -- the plan list written for Neon Realtime: the plans the account reaches, and every agent of them and with no plan, with what the list reads each agent's status from. @@ -89,21 +87,7 @@ function tasksStatement(userId: string) { * the order turns were queued, as `findLatestTurns` reads it. */ function agentsStatement(userId: string) { - const latest = queryBuilder - .select({ - id: turns.id, - status: turns.status, - cancelRequestedAt: turns.cancelRequestedAt, - queuedAt: turns.queuedAt, - startedAt: turns.startedAt, - settledAt: turns.settledAt, - failure: turns.failure, - }) - .from(turns) - .where(eq(turns.conversationId, codingAgent.conversationId)) - .orderBy(desc(turns.queuedAt), desc(turns.id)) - .limit(1) - .as("latest_turn"); + const latest = latestTurnOf(codingAgent.conversationId, "latest_turn"); return queryBuilder .select( liveSelection( @@ -130,9 +114,7 @@ function agentsStatement(userId: string) { turnId: latest.id, turnStatus: latest.status, cancelRequested: isNotNull(latest.cancelRequestedAt), - turnSessionEnded: sql< - boolean | null - >`${latest.failure} = ${BRAIN_REQUEST_FAILURE.SESSION_ENDED}`, + turnSessionEnded: sessionEndedOf(latest), }, { archivedAt: codingAgent.archivedAt, @@ -141,7 +123,7 @@ function agentsStatement(userId: string) { queuedAt: latest.queuedAt, startedAt: latest.startedAt, settledAt: latest.settledAt, - awaitingLineAt: newestAwaitingLineAt(codingAgent.conversationId), + awaitingLineAt: awaitingLineAtIn(codingAgent.conversationId), }, ), )