diff --git a/apps/desktop/src/main/ipc/register-desktop-ipc.ts b/apps/desktop/src/main/ipc/register-desktop-ipc.ts index 20a1eb993f..bb0344bbd5 100644 --- a/apps/desktop/src/main/ipc/register-desktop-ipc.ts +++ b/apps/desktop/src/main/ipc/register-desktop-ipc.ts @@ -7,7 +7,11 @@ import { MAC_TERMINAL_PORT_CHANNEL } from "@luke/features/mac-terminal-wire"; import { TERMINAL_PORT_CHANNEL } from "@luke/features/terminal-acts"; import { highlightSnippet } from "@luke/host"; import { createActPortHandler, type PlatformActProcedures } from "@luke/host/act-port"; -import { createHostActRouter, type PlatformActRows } from "@luke/host/act-router"; +import { + createHostActRouter, + type PlatformActRows, + transcriptFollowActRows, +} from "@luke/host/act-router"; import { ACT_KIND } from "@luke/host/acts"; import type { AppStateSnapshot } from "@luke/host/app-state"; import { channels } from "@luke/host/bridge"; @@ -117,6 +121,14 @@ export function registerDesktopIpc(services: DesktopServices): void { experimentOn: () => experimentOn(state.snapshot().settings?.experiments ?? NO_EXPERIMENTS, EXPERIMENT.BROWSER), }), + ...transcriptFollowActRows({ + host: operator.host, + deliver: (sender, changed) => { + if (!sender.isDestroyed()) sender.send(channels.onTranscript, changed); + }, + whenGone: whenPageGone, + run, + }), [ACT_KIND.UPDATE_CHECK]: () => updates.check(), [ACT_KIND.UPDATE_INSTALL]: () => updates.install(), [ACT_KIND.UPDATE_OPEN_RELEASE]: () => updates.openLatestRelease(), diff --git a/apps/web/src/app/web-host.ts b/apps/web/src/app/web-host.ts index 2d35f81aae..45df950f02 100644 --- a/apps/web/src/app/web-host.ts +++ b/apps/web/src/app/web-host.ts @@ -34,6 +34,8 @@ import { type HostActDependencies, type PlatformActRows, refusedRows, + type TranscriptFollowActKind, + transcriptFollowActRows, } from "@luke/host/act-router"; import { ACT_KIND } from "@luke/host/acts"; import { AgentNoticeDoors } from "@luke/host/agent-notices"; @@ -250,6 +252,7 @@ const noticeDoors = ( function platformRows( http: HttpClient.HttpClient, doors: WebNoticeDoors, + follows: Pick, TranscriptFollowActKind>, ): PlatformActRows { const feedback = feedbackDelivery({ url: new URL(FEEDBACK_PATH, location.origin).href }); return { @@ -263,6 +266,7 @@ function platformRows( [ACT_KIND.NOTIFICATIONS_TEST]: () => doors.postTest(), [ACT_KIND.FEEDBACK_SEND]: ({ submission }) => Effect.provideService(feedback.deliver(submission), HttpClient.HttpClient, http), + ...follows, }; } @@ -431,7 +435,21 @@ const composedWeb = (run: WebHostOptions["run"]) => const made = yield* createInProcessBridge({ sender: window, routerFor: () => - createHostActRouter(hostRowDependencies(operator), platformRows(http, doors)), + createHostActRouter( + hostRowDependencies(operator), + platformRows( + http, + doors, + // The tab's transcripts are followed at the host and each page said to the page, which is + // the tab's one panel and is gone only with the host. + transcriptFollowActRows({ + host: operator.host, + deliver: (_page, changed) => events?.onTranscript(changed), + whenGone: () => undefined, + run, + }), + ), + ), run, state, windowFacts: () => ({ fullScreen: document.fullscreenElement !== null }), diff --git a/packages/features/fixtures/json-schema/coding-agent-view-codingAgentTranscriptChangedSchema.json b/packages/features/fixtures/json-schema/coding-agent-view-codingAgentTranscriptChangedSchema.json new file mode 100644 index 0000000000..f3bed1b2a7 --- /dev/null +++ b/packages/features/fixtures/json-schema/coding-agent-view-codingAgentTranscriptChangedSchema.json @@ -0,0 +1,556 @@ +{ + "type": "object", + "properties": { + "transcript": { + "anyOf": [ + { + "type": "object", + "properties": { + "kind": { + "type": "string", + "enum": [ + "agent" + ] + }, + "agentId": { + "type": "string", + "minLength": 1 + }, + "callId": { + "type": "string", + "minLength": 1, + "maxLength": 128 + } + }, + "required": [ + "kind", + "agentId" + ], + "additionalProperties": false + }, + { + "type": "object", + "properties": { + "kind": { + "type": "string", + "enum": [ + "taskWorker" + ] + }, + "taskId": { + "type": "string", + "minLength": 1 + }, + "callId": { + "type": "string", + "minLength": 1, + "maxLength": 128 + } + }, + "required": [ + "kind", + "taskId", + "callId" + ], + "additionalProperties": false + } + ] + }, + "page": { + "anyOf": [ + { + "type": "object", + "properties": { + "messages": { + "type": "array", + "items": { + "type": "object", + "properties": { + "id": { + "type": "string" + }, + "turnId": { + "type": "string" + }, + "asked": { + "type": "array", + "items": { + "type": "string", + "maxLength": 128 + }, + "maxItems": 20 + }, + "askedAt": { + "type": "array", + "items": { + "type": "object", + "properties": { + "voiceSessionId": { + "type": "string", + "maxLength": 128 + }, + "fromMs": { + "type": "number" + } + }, + "required": [ + "voiceSessionId", + "fromMs" + ], + "additionalProperties": false + }, + "maxItems": 20 + }, + "origin": { + "type": "string", + "enum": [ + "typed", + "spoken", + "transcript_change", + "child", + "child_completion" + ] + }, + "earlierOmitted": { + "type": "boolean" + }, + "role": { + "type": "string", + "enum": [ + "user", + "assistant", + "system" + ] + }, + "parts": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": true + } + }, + "metadata": { + "type": "object", + "additionalProperties": true + }, + "sentBy": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "userId": { + "type": "string" + }, + "kind": { + "type": "string", + "enum": [ + "member", + "former", + "luke", + "automation", + "github" + ] + }, + "agent": { + "type": "boolean", + "enum": [ + true + ] + }, + "model": { + "type": "string", + "maxLength": 200 + }, + "agentId": { + "type": "string", + "minLength": 1, + "maxLength": 36 + }, + "taskId": { + "type": "string", + "minLength": 1, + "maxLength": 36 + } + }, + "required": [ + "name" + ], + "additionalProperties": false + }, + "sentAt": { + "type": "number", + "minimum": 0 + }, + "arrivedIn": { + "type": "object", + "properties": { + "journalId": { + "type": "string" + }, + "afterStep": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "journalId", + "afterStep" + ], + "additionalProperties": false + }, + "startedAt": { + "type": "number", + "minimum": 0 + }, + "finishedAt": { + "type": "number", + "minimum": 0 + }, + "model": { + "type": "string", + "maxLength": 200 + }, + "effort": { + "type": "string", + "maxLength": 200 + }, + "tokens": { + "type": "object", + "properties": { + "inputTokens": { + "type": "number", + "minimum": 0 + }, + "outputTokens": { + "type": "number", + "minimum": 0 + }, + "cachedInputTokens": { + "type": "number", + "minimum": 0 + }, + "cacheWriteTokens": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "inputTokens", + "outputTokens", + "cachedInputTokens" + ], + "additionalProperties": false + }, + "ending": { + "type": "string", + "enum": [ + "stopped", + "failed", + "abandoned" + ] + }, + "fastRefused": { + "type": "boolean" + } + }, + "required": [ + "id", + "role", + "parts" + ], + "additionalProperties": false + } + }, + "cursor": { + "type": "string" + }, + "earlier": { + "type": "string" + }, + "status": { + "type": "string", + "enum": [ + "starting", + "running", + "completed", + "failed", + "cancelled" + ] + }, + "runStartedAt": { + "type": "number", + "minimum": 0 + }, + "delegations": { + "type": "array", + "items": { + "type": "object", + "properties": { + "callId": { + "type": "string", + "minLength": 1, + "maxLength": 128 + }, + "startedAt": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "callId", + "startedAt" + ], + "additionalProperties": false + } + }, + "subagentTitles": { + "type": "array", + "items": { + "type": "object", + "properties": { + "callId": { + "type": "string", + "minLength": 1, + "maxLength": 128 + }, + "title": { + "type": "string" + } + }, + "required": [ + "callId", + "title" + ], + "additionalProperties": false + } + }, + "subagents": { + "type": "array", + "items": { + "type": "object", + "properties": { + "callId": { + "type": "string", + "minLength": 1, + "maxLength": 128 + }, + "status": { + "type": "string", + "enum": [ + "starting", + "running", + "completed", + "failed", + "cancelled" + ] + }, + "model": { + "type": "string" + }, + "effort": { + "type": "string" + }, + "startedAt": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "callId", + "status", + "model", + "effort", + "startedAt" + ], + "additionalProperties": false + } + }, + "title": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "failureReason": { + "type": "string", + "enum": [ + "model", + "sandbox", + "github", + "abandoned", + "provider_key", + "not_started", + "fast_own_key", + "fast_not_enabled", + "context_overflow", + "session_ended", + "other" + ] + }, + "usage": { + "type": "object", + "properties": { + "inputTokens": { + "type": "number", + "minimum": 0 + }, + "outputTokens": { + "type": "number", + "minimum": 0 + }, + "cachedInputTokens": { + "type": "number", + "minimum": 0 + }, + "costUsd": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "inputTokens", + "outputTokens" + ], + "additionalProperties": false + }, + "turnUsage": { + "type": "object", + "properties": { + "inputTokens": { + "type": "number", + "minimum": 0 + }, + "outputTokens": { + "type": "number", + "minimum": 0 + }, + "cachedInputTokens": { + "type": "number", + "minimum": 0 + }, + "costUsd": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "inputTokens", + "outputTokens" + ], + "additionalProperties": false + }, + "context": { + "type": "object", + "properties": { + "windowTokens": { + "type": "number", + "minimum": 0 + }, + "compactionThresholdTokens": { + "type": "number", + "minimum": 0 + }, + "inputTokens": { + "type": "number", + "minimum": 0 + }, + "outputTokens": { + "type": "number", + "minimum": 0 + }, + "cachedInputTokens": { + "type": "number", + "minimum": 0 + }, + "cacheWriteTokens": { + "type": "number", + "minimum": 0 + }, + "estimated": { + "type": "object", + "properties": { + "systemPromptTokens": { + "type": "number", + "minimum": 0 + }, + "toolDefinitionTokens": { + "type": "number", + "minimum": 0 + }, + "briefTokens": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "systemPromptTokens", + "toolDefinitionTokens" + ], + "additionalProperties": false + }, + "measuredAt": { + "type": "number", + "minimum": 0 + }, + "compactedAt": { + "type": "number", + "minimum": 0 + } + }, + "required": [ + "windowTokens", + "compactionThresholdTokens", + "inputTokens", + "outputTokens", + "estimated", + "measuredAt" + ], + "additionalProperties": false + } + }, + "required": [ + "messages", + "cursor", + "status", + "title" + ], + "additionalProperties": false + }, + { + "type": "object", + "properties": { + "failure": { + "type": "string", + "enum": [ + "unanswered", + "not-found", + "no-repository", + "repository-not-reachable", + "github-unanswered", + "invalid-choice", + "message-too-long", + "agent-not-ready", + "agent-retired", + "plan-archived" + ] + } + }, + "required": [ + "failure" + ], + "additionalProperties": false + } + ] + } + }, + "required": [ + "transcript", + "page" + ], + "additionalProperties": false +} diff --git a/packages/features/fixtures/json-schema/coding-agent-view-codingAgentTranscriptKeySchema.json b/packages/features/fixtures/json-schema/coding-agent-view-codingAgentTranscriptKeySchema.json new file mode 100644 index 0000000000..f355d4f3ee --- /dev/null +++ b/packages/features/fixtures/json-schema/coding-agent-view-codingAgentTranscriptKeySchema.json @@ -0,0 +1,55 @@ +{ + "anyOf": [ + { + "type": "object", + "properties": { + "kind": { + "type": "string", + "enum": [ + "agent" + ] + }, + "agentId": { + "type": "string", + "minLength": 1 + }, + "callId": { + "type": "string", + "minLength": 1, + "maxLength": 128 + } + }, + "required": [ + "kind", + "agentId" + ], + "additionalProperties": false + }, + { + "type": "object", + "properties": { + "kind": { + "type": "string", + "enum": [ + "taskWorker" + ] + }, + "taskId": { + "type": "string", + "minLength": 1 + }, + "callId": { + "type": "string", + "minLength": 1, + "maxLength": 128 + } + }, + "required": [ + "kind", + "taskId", + "callId" + ], + "additionalProperties": false + } + ] +} diff --git a/packages/features/src/coding-agent-view.ts b/packages/features/src/coding-agent-view.ts index 431796279d..0a3f7b8ab8 100644 --- a/packages/features/src/coding-agent-view.ts +++ b/packages/features/src/coding-agent-view.ts @@ -258,6 +258,44 @@ export const codingAgentMessagesAnswerViewSchema = EffectSchema.Union([ export type CodingAgentMessagesAnswerView = typeof codingAgentMessagesAnswerViewSchema.Type; +/** Whose transcript a window follows: a coding agent's, or a worker's of it, or a task's planning worker's. */ +export const TRANSCRIPT_KIND = { + AGENT: "agent", + TASK_WORKER: "taskWorker", +} as const; + +/** + * One transcript as a window follows it: an agent's own, or one of its + * workers' by the delegation call that started it; or a task's planning + * worker's by the call that started it. + */ +export const codingAgentTranscriptKeySchema = EffectSchema.Union([ + EffectSchema.Struct({ + kind: EffectSchema.Literal(TRANSCRIPT_KIND.AGENT), + agentId: EffectSchema.NonEmptyString, + callId: EffectSchema.optionalKey(codingAgentCallIdSchema), + }), + EffectSchema.Struct({ + kind: EffectSchema.Literal(TRANSCRIPT_KIND.TASK_WORKER), + taskId: EffectSchema.NonEmptyString, + callId: codingAgentCallIdSchema, + }), +]); + +export type CodingAgentTranscriptKey = typeof codingAgentTranscriptKeySchema.Type; + +/** + * A followed transcript's next page, as the window hears it pushed: the + * messages past the last page pushed, the first from the start, with where + * the agent stands as it was read; or why there is none. + */ +export const codingAgentTranscriptChangedSchema = EffectSchema.Struct({ + transcript: codingAgentTranscriptKeySchema, + page: codingAgentMessagesAnswerViewSchema, +}); + +export type CodingAgentTranscriptChanged = typeof codingAgentTranscriptChangedSchema.Type; + /** Messaging an agent, as the window asks it: the agent, and the message request the service takes, its key the window's own. */ export const codingAgentMessageParamsSchema = EffectSchema.Struct({ agentId: EffectSchema.NonEmptyString, diff --git a/packages/features/src/hosted-wire-schemas.test.ts b/packages/features/src/hosted-wire-schemas.test.ts index 45b10f23e7..5c003a633b 100644 --- a/packages/features/src/hosted-wire-schemas.test.ts +++ b/packages/features/src/hosted-wire-schemas.test.ts @@ -221,6 +221,8 @@ const EFFECT_MODULE_SCHEMAS = { codingAgentMessagesParamsSchema: codingAgentView.codingAgentMessagesParamsSchema, codingAgentMessageParamsSchema: codingAgentView.codingAgentMessageParamsSchema, codingAgentMessagesAnswerViewSchema: codingAgentView.codingAgentMessagesAnswerViewSchema, + codingAgentTranscriptKeySchema: codingAgentView.codingAgentTranscriptKeySchema, + codingAgentTranscriptChangedSchema: codingAgentView.codingAgentTranscriptChangedSchema, codingAgentChooseParamsSchema: codingAgentView.codingAgentChooseParamsSchema, codingAgentRenameParamsSchema: codingAgentView.codingAgentRenameParamsSchema, codingAgentStopParamsSchema: codingAgentView.codingAgentStopParamsSchema, diff --git a/packages/gateway/src/protocol.ts b/packages/gateway/src/protocol.ts index 046b94d98a..850937a4ba 100644 --- a/packages/gateway/src/protocol.ts +++ b/packages/gateway/src/protocol.ts @@ -48,6 +48,9 @@ export const GATEWAY_METHOD = { TASKS_CONVERSATION_EARLIER: "tasks.conversationEarlier", /** One page of a task's conversation as an agent's transcript is read, or a brain worker's under the call that started it, held open by the service while a turn or the worker runs; answered to the caller, never held in the view. */ TASKS_MESSAGES: "tasks.messages", + /** A task's planning worker's transcript followed and let go, on the same terms as `CODING_AGENTS_TRANSCRIPT_FOLLOW`. */ + TASKS_WORKER_TRANSCRIPT_FOLLOW: "tasks.workerTranscriptFollow", + TASKS_WORKER_TRANSCRIPT_UNFOLLOW: "tasks.workerTranscriptUnfollow", /** A window of a file of one task's repository, read from GitHub through the service as it stands now and coloured here; answered to the caller, never held in the view. */ TASKS_FILE: "tasks.file", /** One folder of one task's repository, listed from GitHub through the service as it stands now; answered to the caller, never held in the view. */ @@ -136,6 +139,15 @@ export const GATEWAY_METHOD = { CODING_AGENTS_START: "codingAgents.start", /** One agent's transcript past a cursor, held open by the service while the agent runs. */ CODING_AGENTS_MESSAGES: "codingAgents.messages", + /** + * One agent's transcript, or one of its workers', followed from now until + * it is let go or the account signs out, over Neon Realtime. Its pages are + * heard as `TRANSCRIPT_CHANGED` events, never answered: the first from the + * start, then each one past the last; following again starts over. + */ + CODING_AGENTS_TRANSCRIPT_FOLLOW: "codingAgents.transcriptFollow", + /** One agent's or worker's transcript let go: no more `TRANSCRIPT_CHANGED` events of it until it is followed again. */ + CODING_AGENTS_TRANSCRIPT_UNFOLLOW: "codingAgents.transcriptUnfollow", /** One agent sent a message, naming how it reaches a turn under way; the agent as it then stands, or why it was not. */ CODING_AGENTS_MESSAGE: "codingAgents.message", /** One agent's model and effort changed for its next step; the agent as it then stands, or why it is unchanged. */ @@ -573,6 +585,8 @@ export const GATEWAY_EVENT = { TERMINAL_STATE: "terminal.state", /** What waits on the developer across the account, whole, each time it changes while it is followed. */ ACCOUNT_INBOX_CHANGED: "accountInbox.changed", + /** A followed transcript's next page, or why there is none, each time it moves while it is followed. */ + TRANSCRIPT_CHANGED: "transcript.changed", } as const; export type GatewayEventKind = (typeof GATEWAY_EVENT)[keyof typeof GATEWAY_EVENT]; diff --git a/packages/host/src/act-router/index.ts b/packages/host/src/act-router/index.ts index c075c41ab3..219bc3e328 100644 --- a/packages/host/src/act-router/index.ts +++ b/packages/host/src/act-router/index.ts @@ -11,6 +11,10 @@ export { type ActSender, createActRouter, } from "./router.js"; +export { + type TranscriptFollowActKind, + transcriptFollowActRows, +} from "./transcript-follow-acts.js"; /** * The rows a platform answers itself: every kind the host's rows leave, and diff --git a/packages/host/src/act-router/transcript-follow-acts.test.ts b/packages/host/src/act-router/transcript-follow-acts.test.ts new file mode 100644 index 0000000000..8447ac6a9f --- /dev/null +++ b/packages/host/src/act-router/transcript-follow-acts.test.ts @@ -0,0 +1,189 @@ +import assert from "node:assert/strict"; +import { it } from "@effect/vitest"; +import { + type CodingAgentTranscriptChanged, + type CodingAgentTranscriptKey, + TRANSCRIPT_KIND, +} from "@luke/features/coding-agent-view"; +import { CODING_AGENT_STATUS } from "@luke/features/coding-agent-wire"; +import { Effect } from "effect"; +import { ACT, ACT_KIND, ACT_OUTCOME_STATUS } from "../acts.js"; +import { type ActRows, type ActSender, createActRouter } from "./router.js"; +import { + type TranscriptFollowActsDependencies, + transcriptFollowActRows, +} from "./transcript-follow-acts.js"; + +/** A window, by identity alone, as each platform's own handle on its page is. */ +type Panel = object; + +/** + * A followed transcript as a platform carries it, over a host that answers from a + * script: each follow asks the host to follow from the start, each page it + * tells reaches the panels following that transcript and no other, the + * transcript is let go at the host once no panel holds a follow of it, a + * panel's page going lets go of all of its follows, and only a panel is + * answered. Synthetic ids throughout. + */ + +const AGENT_ID = "8c1a6a4f-3d2e-4d8b-8b66-6f4c7a2e3b21"; +const TASK_ID = "7b0f5f3e-2c1d-4c7a-9a55-5e3b6f1d2a10"; +const AGENT: CodingAgentTranscriptKey = { kind: TRANSCRIPT_KIND.AGENT, agentId: AGENT_ID }; +const WORKER: CodingAgentTranscriptKey = { ...AGENT, callId: "call-7" }; +const TASK_WORKER: CodingAgentTranscriptKey = { + kind: TRANSCRIPT_KIND.TASK_WORKER, + taskId: TASK_ID, + callId: "call-0-0", +}; + +/** One window, by identity alone. */ +function panelOf(): ActSender { + // The rows read a sender by identity alone, so one fresh object is one window. + return { sender: {}, panel: true, voice: false, call: false }; +} + +/** A transcript as the test names it in what the host was asked. */ +function named(transcript: CodingAgentTranscriptKey): string { + return transcript.kind === TRANSCRIPT_KIND.AGENT + ? `${transcript.agentId}:${transcript.callId ?? "own"}` + : `${transcript.taskId}:${transcript.callId}`; +} + +function fixture() { + const asked: string[] = []; + const sent: { sender: Panel; changed: CodingAgentTranscriptChanged }[] = []; + const gone = new Map void>(); + const pending: Effect.Effect[] = []; + let told: ((changed: CodingAgentTranscriptChanged) => void) | undefined; + const host: TranscriptFollowActsDependencies["host"] = { + transcriptFollow: (transcript) => + Effect.sync(() => void asked.push(`follow:${named(transcript)}`)), + transcriptUnfollow: (transcript) => + Effect.sync(() => void asked.push(`unfollow:${named(transcript)}`)), + onTranscriptChanged: (listener) => { + told = listener; + return () => undefined; + }, + }; + const rows = transcriptFollowActRows({ + host, + deliver: (sender, changed) => void sent.push({ sender, changed }), + whenGone: (sender, then) => void gone.set(sender, then), + run: (effect: Effect.Effect) => { + pending.push(Effect.asVoid(effect)); + return new Promise(() => undefined); + }, + }); + // SAFETY: the router dispatches only the follow kinds these tests send. + const router = createActRouter(rows as ActRows); + const follow = (transcript: CodingAgentTranscriptKey, sender: ActSender) => + router.performAct( + { kind: ACT_KIND.CODING_AGENTS_TRANSCRIPT_FOLLOW, payload: transcript }, + sender, + ); + const unfollow = (transcript: CodingAgentTranscriptKey, sender: ActSender) => + router.performAct( + { kind: ACT_KIND.CODING_AGENTS_TRANSCRIPT_UNFOLLOW, payload: transcript }, + sender, + ); + return { + follow, + unfollow, + asked, + /** What each window was sent, as the test reads it. */ + sentTo: (sender: ActSender) => + sent.flatMap((each) => (each.sender === sender.sender ? [each.changed] : [])), + /** The host telling a followed transcript's page, as the operator would hand it up. */ + hostSays: (changed: CodingAgentTranscriptChanged) => told?.(changed), + pageGone: (sender: ActSender) => gone.get(sender.sender)?.(), + /** What the pages' callbacks handed to the runtime, run now. */ + drain: Effect.suspend(() => Effect.all(pending.splice(0), { discard: true })), + }; +} + +/** A page of a transcript, as the host tells one. */ +function pageOf(transcript: CodingAgentTranscriptKey, text: string): CodingAgentTranscriptChanged { + return { + transcript, + page: { + messages: [{ id: `m-${text}`, role: "assistant", parts: [{ type: "text", text }] }], + cursor: "1:0", + status: CODING_AGENT_STATUS.RUNNING, + title: null, + }, + }; +} + +it.effect( + "every follow asks the host from the start, and each page reaches the panels following that transcript alone", + () => + Effect.gen(function* () { + const { follow, asked, sentTo, hostSays } = fixture(); + const one = panelOf(); + const other = panelOf(); + const elsewhere = panelOf(); + + yield* follow(AGENT, one); + yield* follow(WORKER, elsewhere); + yield* follow(TASK_WORKER, elsewhere); + hostSays(pageOf(AGENT, "Reading the invites.")); + yield* follow(AGENT, other); + hostSays(pageOf(AGENT, "Writing the expiry.")); + hostSays(pageOf(TASK_WORKER, "Compared them.")); + + assert.deepEqual(asked, [ + `follow:${AGENT_ID}:own`, + `follow:${AGENT_ID}:call-7`, + `follow:${TASK_ID}:call-0-0`, + `follow:${AGENT_ID}:own`, + ]); + assert.deepEqual(sentTo(one), [ + pageOf(AGENT, "Reading the invites."), + pageOf(AGENT, "Writing the expiry."), + ]); + assert.deepEqual(sentTo(other), [pageOf(AGENT, "Writing the expiry.")]); + assert.deepEqual(sentTo(elsewhere), [pageOf(TASK_WORKER, "Compared them.")]); + }), +); + +it.effect( + "a transcript is let go at the host once no panel holds a follow of it, a panel's follows counted, and a panel's page going lets go of all of them", + () => + Effect.gen(function* () { + const { follow, unfollow, asked, pageGone, drain, sentTo, hostSays } = fixture(); + const one = panelOf(); + const other = panelOf(); + + // The panel follows the agent from two tabs at once, and lets one go. + yield* follow(AGENT, one); + yield* follow(AGENT, one); + yield* unfollow(AGENT, one); + assert.deepEqual(asked.slice(2), []); + yield* unfollow(AGENT, one); + assert.deepEqual(asked.slice(2), [`unfollow:${AGENT_ID}:own`]); + hostSays(pageOf(AGENT, "Reading the invites.")); + assert.deepEqual(sentTo(one), []); + + yield* follow(WORKER, other); + yield* follow(TASK_WORKER, other); + pageGone(other); + yield* drain; + assert.deepEqual(asked.slice(5).toSorted(), [ + `unfollow:${TASK_ID}:call-0-0`, + `unfollow:${AGENT_ID}:call-7`, + ]); + }), +); + +it.effect("only a panel is answered", () => + Effect.gen(function* () { + const { follow, asked } = fixture(); + const voice: ActSender = { ...panelOf(), panel: false, voice: true }; + + assert.deepEqual(yield* follow(AGENT, voice), { + status: ACT_OUTCOME_STATUS.REFUSED, + reason: ACT[ACT_KIND.CODING_AGENTS_TRANSCRIPT_FOLLOW].refusal, + }); + assert.deepEqual(asked, []); + }), +); diff --git a/packages/host/src/act-router/transcript-follow-acts.ts b/packages/host/src/act-router/transcript-follow-acts.ts new file mode 100644 index 0000000000..72c7692ac3 --- /dev/null +++ b/packages/host/src/act-router/transcript-follow-acts.ts @@ -0,0 +1,149 @@ +/** + * transcript-follow-acts.ts -- an agent's, a worker's, or a planning worker's transcript followed for the panels that draw it, and each page pushed to those panels alone, on whichever platform holds them. + * + * A panel says which transcripts its shown tabs draw, one follow each, and + * lets each go when the tab stops showing; the host follows each transcript + * while any panel does (over Neon Realtime) and tells the platform every + * page, which is sent on to the panels following it and to no other window. Every + * follow asks the host to follow from the start again, so a panel coming to + * a transcript another already draws hears it whole, and the others hear + * pages they hold again, which join by id where they stand. A panel may + * follow the same transcript from several tabs, so its follows are counted, + * and the transcript is let go at the host once no panel holds one. A panel + * whose page is gone lets go of everything it followed. + * + * Nothing of a transcript's words is kept here: only which panels follow it. + */ + +import { + type CodingAgentTranscriptChanged, + type CodingAgentTranscriptKey, + TRANSCRIPT_KIND, +} from "@luke/features/coding-agent-view"; +import { Effect } from "effect"; +import { ACT, ACT_KIND } from "../acts.js"; +import type { HostOperator } from "../host-operator.js"; +import { ActRefused, type ActRows, type ActSender } from "./router.js"; + +export type TranscriptFollowActKind = + | typeof ACT_KIND.CODING_AGENTS_TRANSCRIPT_FOLLOW + | typeof ACT_KIND.CODING_AGENTS_TRANSCRIPT_UNFOLLOW; + +type TranscriptKind = CodingAgentTranscriptKey["kind"]; + +export interface TranscriptFollowActsDependencies { + host: Pick; + /** Sends one panel a followed transcript's next page. */ + deliver: (sender: Panel, changed: CodingAgentTranscriptChanged) => void; + /** Calls `then` once, when the panel's page is gone. */ + whenGone: (sender: Panel, then: () => void) => void; + /** Runs what a page's end asks of the host, from the page's own callback. */ + run: (effect: Effect.Effect) => Promise; +} + +/** One followed transcript: which it is, and how many follows each panel holds of it. */ +interface Followed { + readonly transcript: CodingAgentTranscriptKey; + readonly panels: Map; +} + +/** Every followed transcript, by its kind, then its agent's or task's id, then its call where it has one. */ +type FollowedTable = Map< + TranscriptKind, + Map>> +>; + +/** The agent's or the task's id a transcript is under. */ +function ownerOf(transcript: CodingAgentTranscriptKey): string { + return transcript.kind === TRANSCRIPT_KIND.AGENT ? transcript.agentId : transcript.taskId; +} + +/** The calls of one owner's followed transcripts, where any is followed. */ +function callsOf( + table: FollowedTable, + transcript: CodingAgentTranscriptKey, +): Map> | undefined { + return table.get(transcript.kind)?.get(ownerOf(transcript)); +} + +/** The calls of one owner's followed transcripts, made where none is followed yet. */ +function callsMade( + table: FollowedTable, + transcript: CodingAgentTranscriptKey, +): Map> { + const byOwner = + table.get(transcript.kind) ?? new Map>>(); + table.set(transcript.kind, byOwner); + const calls = byOwner.get(ownerOf(transcript)) ?? new Map>(); + byOwner.set(ownerOf(transcript), calls); + return calls; +} + +/** The refusal a window that draws no transcript hears, in its kind's own words. */ +function refuseUnlessPanel(kind: TranscriptFollowActKind, sender: ActSender): void { + if (!sender.panel) throw new ActRefused({ message: ACT[kind].refusal }); +} + +export function transcriptFollowActRows( + dependencies: TranscriptFollowActsDependencies, +): Pick, TranscriptFollowActKind> { + const { host, deliver, whenGone, run } = dependencies; + const followed: FollowedTable = new Map(); + /** The panels whose page's end is watched, so it is watched once. */ + const watched = new Set(); + + host.onTranscriptChanged((changed) => { + const held = callsOf(followed, changed.transcript)?.get(changed.transcript.callId); + if (held === undefined) return; + for (const panel of held.panels.keys()) deliver(panel, changed); + }); + + /** One follow of the panel's let go, or all of them, and the transcript let go at the host once no panel holds one. */ + const release = (transcript: CodingAgentTranscriptKey, panel: Panel, all: boolean) => + Effect.suspend(() => { + const calls = callsOf(followed, transcript); + const held = calls?.get(transcript.callId); + const count = held?.panels.get(panel); + if (calls === undefined || held === undefined || count === undefined) return Effect.void; + if (all || count === 1) held.panels.delete(panel); + else held.panels.set(panel, count - 1); + if (held.panels.size > 0) return Effect.void; + calls.delete(transcript.callId); + return host.transcriptUnfollow(held.transcript); + }); + + /** Everything the panel followed let go, as its page is gone. */ + const gone = (panel: Panel) => { + watched.delete(panel); + const transcripts = [...followed.values()] + .flatMap((byOwner) => [...byOwner.values()]) + .flatMap((calls) => [...calls.values()]) + .flatMap((held) => (held.panels.has(panel) ? [held.transcript] : [])); + void run( + Effect.forEach(transcripts, (transcript) => release(transcript, panel, true), { + discard: true, + }), + ); + }; + + return { + [ACT_KIND.CODING_AGENTS_TRANSCRIPT_FOLLOW]: (transcript, sender) => { + refuseUnlessPanel(ACT_KIND.CODING_AGENTS_TRANSCRIPT_FOLLOW, sender); + const panel = sender.sender; + if (!watched.has(panel)) { + watched.add(panel); + whenGone(panel, () => gone(panel)); + } + const calls = callsMade(followed, transcript); + const held = calls.get(transcript.callId) ?? { transcript, panels: new Map() }; + calls.set(transcript.callId, held); + held.panels.set(panel, (held.panels.get(panel) ?? 0) + 1); + // Note that every follow starts the host over, so the panel that asked hears the transcript whole. + return host.transcriptFollow(transcript); + }, + [ACT_KIND.CODING_AGENTS_TRANSCRIPT_UNFOLLOW]: (transcript, sender) => { + refuseUnlessPanel(ACT_KIND.CODING_AGENTS_TRANSCRIPT_UNFOLLOW, sender); + return release(transcript, sender.sender, false); + }, + }; +} diff --git a/packages/host/src/acts.ts b/packages/host/src/acts.ts index 809021c967..24b35bc559 100644 --- a/packages/host/src/acts.ts +++ b/packages/host/src/acts.ts @@ -36,6 +36,7 @@ import { codingAgentInboxReadStateViewSchema, codingAgentMessagesAnswerViewSchema, codingAgentPictureParamsSchema, + codingAgentTranscriptKeySchema, } from "@luke/features/coding-agent-view"; import { type FeedbackResult, @@ -199,6 +200,14 @@ export const ACT_KIND = { /** One image of an inbox item, fetched by the host under the account, so the window holds no bearer and no address. */ CODING_AGENTS_INBOX_ATTACHMENT: "codingAgents.inboxAttachment", CODING_AGENTS_PICTURE: "codingAgents.picture", + /** + * An agent's, worker's, or planning worker's transcript followed by the + * window that asks and let go by it: its pages are pushed to that window + * alone on `app:transcript`, each time it moves, until the window lets it + * go or its page is gone. + */ + CODING_AGENTS_TRANSCRIPT_FOLLOW: "codingAgents.transcriptFollow", + CODING_AGENTS_TRANSCRIPT_UNFOLLOW: "codingAgents.transcriptUnfollow", /** * A Browser tab's view, which main owns over the panel that asked * (`browser-views.ts`): opened on an address and answered how its page @@ -579,6 +588,16 @@ export const ACT = { ), refusal: "Could not read that task's conversation on this system.", }, + [ACT_KIND.CODING_AGENTS_TRANSCRIPT_FOLLOW]: { + payload: actSchema(codingAgentTranscriptKeySchema), + result: answersNothing, + refusal: "Could not follow the transcript on this system.", + }, + [ACT_KIND.CODING_AGENTS_TRANSCRIPT_UNFOLLOW]: { + payload: actSchema(codingAgentTranscriptKeySchema), + result: answersNothing, + refusal: "Could not let the transcript go on this system.", + }, [ACT_KIND.CODING_AGENTS_INBOX]: { payload: actSchema(codingAgentInboxParamsSchema), result: wireResult(isReadable(codingAgentInboxAnswerViewSchema)), diff --git a/packages/host/src/bridge.ts b/packages/host/src/bridge.ts index 19f4f9523c..5921c892c8 100644 --- a/packages/host/src/bridge.ts +++ b/packages/host/src/bridge.ts @@ -15,6 +15,10 @@ import { isBrowserKey, isBrowserPage, } from "@luke/features/browser-acts"; +import { + type CodingAgentTranscriptChanged, + codingAgentTranscriptChangedSchema, +} from "@luke/features/coding-agent-view"; import { isLinkPlace, type ReadLink } from "@luke/features/luke-link"; import { type VoiceLiveSessionChanged, voiceLiveSessionChangedSchema } from "@luke/gateway"; import { type AgentWireTrace, isAgentWireTrace } from "@luke/gateway/devtrace"; @@ -445,6 +449,23 @@ export const BRIDGE = { args: noArgs, result: result(isBrowserFocus), }), + /** + * A transcript this window follows, its next page or why there is none: + * sent to the windows that follow it alone, each time it moves and once, + * whole, as a window starts following one already followed, rather than + * as a slice of the document, because the document reaches every window on + * every change and a transcript is read by the one tab drawing it. + */ + onTranscript: entry({ + kind: "subscribe", + channel: "app:transcript", + args: noArgs, + result: result((value) => + Result.isSuccess( + readEither(codingAgentTranscriptChangedSchema, { excess: EXCESS_KEYS.DROP })(value), + ), + ), + }), /** * A Luke link the system handed the app: the panel is to open the place it * names, in the workspace it was written under. An event, as a click is, diff --git a/packages/host/src/client/in-process-bridge.test.ts b/packages/host/src/client/in-process-bridge.test.ts index 7044af66cf..c9f976dd0f 100644 --- a/packages/host/src/client/in-process-bridge.test.ts +++ b/packages/host/src/client/in-process-bridge.test.ts @@ -1,5 +1,11 @@ import assert from "node:assert/strict"; import { it } from "@effect/vitest"; +import { + type CodingAgentTranscriptChanged, + type CodingAgentTranscriptKey, + TRANSCRIPT_KIND, +} from "@luke/features/coding-agent-view"; +import { CODING_AGENT_STATUS } from "@luke/features/coding-agent-wire"; import { MICROPHONE_STATUS } from "@luke/voice/audio"; import { Context, Effect } from "effect"; import { hostActRows } from "../act-router/host-act-rows.js"; @@ -9,6 +15,7 @@ import { type HostActDependencies, type PlatformActRows, refusedRows, + transcriptFollowActRows, } from "../act-router/index.js"; import { ACT, ACT_KIND, ACT_OUTCOME_STATUS, type ActKind } from "../acts.js"; import type { AppStateSnapshot } from "../app-state.js"; @@ -179,3 +186,58 @@ it.effect("a closed bridge tells its page nothing more, nor takes a new subscrip assert.deepEqual(shown, []); }), ); + +it.effect( + "a page follows a transcript through the shell's follow rows and hears each page the host tells pushed to it", + () => + Effect.gen(function* () { + const services = yield* Effect.context(); + const state = new AppStateStore(initialAppState(RUN, false, []), Context.empty()); + const followed: CodingAgentTranscriptKey[] = []; + let told: ((changed: CodingAgentTranscriptChanged) => void) | undefined; + let events: Effect.Success>["events"] | undefined; + const follows = transcriptFollowActRows({ + host: { + transcriptFollow: (transcript) => Effect.sync(() => void followed.push(transcript)), + transcriptUnfollow: () => Effect.void, + onTranscriptChanged: (listener) => { + told = listener; + return () => undefined; + }, + }, + deliver: (_page, changed) => events?.onTranscript(changed), + whenGone: () => undefined, + run: (effect) => Effect.runPromiseWith(services)(effect), + }); + const made = yield* createInProcessBridge({ + sender: Symbol("page"), + routerFor: (reporterOf): ActRouter => + createHostActRouter(UNREACHED, { ...platformRows(reporterOf, []), ...follows }), + run: (effect) => Effect.runPromiseWith(services)(effect), + state, + windowFacts: () => ({ fullScreen: false }), + createId: () => REPORTER, + reports: { notifyReady: () => {}, recordSurfaceEvent: () => {} }, + report: () => undefined, + }); + events = made.events; + const heard: CodingAgentTranscriptChanged[] = []; + made.bridge.onTranscript((changed) => heard.push(changed)); + const transcript: CodingAgentTranscriptKey = { + kind: TRANSCRIPT_KIND.AGENT, + agentId: AGENT_PLACE.agentId, + }; + + const outcome = yield* Effect.promise(() => + made.bridge.act({ kind: ACT_KIND.CODING_AGENTS_TRANSCRIPT_FOLLOW, payload: transcript }), + ); + assert.ok(outcome.status === ACT_OUTCOME_STATUS.DONE); + assert.deepEqual(followed, [transcript]); + const page: CodingAgentTranscriptChanged = { + transcript, + page: { messages: [], cursor: "1:0", status: CODING_AGENT_STATUS.RUNNING, title: null }, + }; + told?.(page); + assert.deepEqual(heard, [page]); + }).pipe(Effect.scoped), +); diff --git a/packages/host/src/client/in-process-bridge.ts b/packages/host/src/client/in-process-bridge.ts index df68a926f0..446484e205 100644 --- a/packages/host/src/client/in-process-bridge.ts +++ b/packages/host/src/client/in-process-bridge.ts @@ -10,7 +10,8 @@ import { type AppStateStore, publishAppStateChanges } from "./app-state-store.js /** * The one-shot events a shell or its notice presenter feeds the page: a * notification's click, or a notice the focused window says in its toast. - * Each is said once and never bootstrapped from. + * Each is said once and never bootstrapped from; a followed transcript's + * page as well, which the shell's follow rows say to the page that follows it. */ type InProcessEventMethod = Extract< BridgeMethod, @@ -24,6 +25,7 @@ type InProcessEventMethod = Extract< | "onRunFailed" | "onChatMessageWaiting" | "onShowLink" + | "onTranscript" >; /** Where the shell says each one-shot event, to every subscription standing in the page. */ @@ -150,6 +152,7 @@ export function createInProcessBridge( onRunFailed: listeners(report), onChatMessageWaiting: listeners(report), onShowLink: listeners(report), + onTranscript: listeners(report), }; // A page whose bridge is closed hears nothing more, whoever still holds `events`. yield* Effect.addFinalizer(() => @@ -174,6 +177,7 @@ export function createInProcessBridge( onRunFailed: said.onRunFailed.subscribe, onChatMessageWaiting: said.onChatMessageWaiting.subscribe, onShowLink: said.onShowLink.subscribe, + onTranscript: said.onTranscript.subscribe, reportVoiceView: dropped, setShortcutCapturing: dropped, recordAgentTrace: dropped, @@ -207,6 +211,7 @@ export function createInProcessBridge( onRunFailed: said.onRunFailed.emit, onChatMessageWaiting: said.onChatMessageWaiting.emit, onShowLink: said.onShowLink.emit, + onTranscript: said.onTranscript.emit, }; return { bridge, events }; }); diff --git a/packages/host/src/compose-coding-agents.test.ts b/packages/host/src/compose-coding-agents.test.ts index d1c766b171..2b18010cfb 100644 --- a/packages/host/src/compose-coding-agents.test.ts +++ b/packages/host/src/compose-coding-agents.test.ts @@ -7,7 +7,7 @@ import { accountInboxAnswerSchema, } from "@luke/features/account-inbox-wire"; import { CHAT_CURSOR_START, MEMBER_KIND } from "@luke/features/chat-wire"; -import { CODING_AGENT_CALL_FAILURE } from "@luke/features/coding-agent-view"; +import { CODING_AGENT_CALL_FAILURE, TRANSCRIPT_KIND } from "@luke/features/coding-agent-view"; import { CHANGED_FILE_STATUS, CHECK_SUMMARY, @@ -575,8 +575,14 @@ function subject(options: { signedIn?: boolean; sendsNetwork?: boolean } = {}) { events.flatMap((event) => event.kind === GATEWAY_EVENT.ACCOUNT_INBOX_CHANGED ? [event.payload] : [], ); + /** Every followed transcript's page the host told, in order. */ + const pages = () => + events.flatMap((event) => + event.kind === GATEWAY_EVENT.TRANSCRIPT_CHANGED ? [event.payload] : [], + ); return { call, + pages, asked: service.asked, seal: service.seal, sealScope: service.sealScope, @@ -1623,3 +1629,88 @@ it.effect( assert.equal(asked.filter((ask) => ask.startsWith("messages-now")).length, 3); }), ); + +const FOLLOWED = { kind: TRANSCRIPT_KIND.AGENT, agentId: AGENT_ID } as const; + +it.effect( + "a followed transcript is told page by page from the start, the next at once after one that said something, then again only as its statements move or at the edge the service names; letting it go ends the telling", + () => + Effect.gen(function* () { + const { call, asked, pages, sealTranscript, pageNow: script, realtime } = yield* subject(); + sealTranscript(SEALED_TRANSCRIPT); + const said = pageNow("1:0", { holds: false, text: "Reading the invites." }); + const caught = { ...pageNow("1:0", { holds: true }), wakeIn: 3_000 }; + script(said, caught); + assert.deepEqual(yield* call(GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_FOLLOW, FOLLOWED), {}); + yield* settled; + assert.deepEqual(pages(), [ + { transcript: FOLLOWED, page: windowPage(said) }, + { transcript: FOLLOWED, page: windowPage(caught) }, + ]); + + // Rows landing read the page again; the edge the service named reads it once more. + transcriptRows(realtime, 1); + yield* settled; + assert.equal(pages().length, 3); + yield* TestClock.adjust("3 seconds"); + yield* settled; + assert.equal(pages().length, 4); + + const moved = pageNow("2:0", { holds: false, text: "Writing the expiry." }); + const ended = pageNow("2:0", { holds: false }); + script(moved, ended); + transcriptRows(realtime, 2); + yield* settled; + // Note that each part landing may move the statements, so the page past the last may be read more than once. + assert.deepEqual(pages()[4], { transcript: FOLLOWED, page: windowPage(moved) }); + assert.deepEqual(pages().at(-1), { transcript: FOLLOWED, page: windowPage(ended) }); + assert.deepEqual(asked.slice(0, 3), [ + `transcript-live:${AGENT_ID}`, + `messages-now:${AGENT_ID}:${CODING_AGENT_CURSOR_START}`, + `messages-now:${AGENT_ID}:1:0`, + ]); + assert.equal(asked.at(-1), `messages-now:${AGENT_ID}:2:0`); + assert.ok(!JSON.stringify(pages()).includes("capability")); + + yield* call(GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_UNFOLLOW, FOLLOWED); + const toldBefore = pages().length; + transcriptRows(realtime, 3); + yield* settled; + assert.equal(pages().length, toldBefore); + }), +); + +it.effect( + "a followed transcript the account does not reach is told not found once; behind a closed gate it is told unanswered; and a sign-out lets every followed transcript go", + () => + Effect.gen(function* () { + const worker = { ...FOLLOWED, callId: "call-7" }; + const refused = yield* subject(); + refused.sealTranscript({ failure: CODING_AGENT_CALL_FAILURE.NOT_FOUND }); + yield* refused.call(GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_FOLLOW, worker); + yield* settled; + yield* TestClock.adjust("5 seconds"); + yield* settled; + assert.deepEqual(refused.pages(), [ + { transcript: worker, page: { failure: CODING_AGENT_CALL_FAILURE.NOT_FOUND } }, + ]); + + const signedOut = yield* subject({ signedIn: false }); + yield* signedOut.call(GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_FOLLOW, FOLLOWED); + yield* settled; + assert.deepEqual(signedOut.pages(), [ + { transcript: FOLLOWED, page: { failure: CODING_AGENT_CALL_FAILURE.UNANSWERED } }, + ]); + assert.deepEqual(signedOut.asked, []); + + const followed = yield* subject(); + followed.sealTranscript(SEALED_TRANSCRIPT); + followed.pageNow(pageNow("1:0", { holds: true })); + yield* followed.call(GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_FOLLOW, FOLLOWED); + yield* settled; + yield* followed.reset; + yield* settled; + assert.deepEqual(followed.realtime.closed, [SEALED_TRANSCRIPT.url]); + assert.equal(followed.pages().length, 1); + }), +); diff --git a/packages/host/src/compose-coding-agents.ts b/packages/host/src/compose-coding-agents.ts index d7c8b5fbec..367267c120 100644 --- a/packages/host/src/compose-coding-agents.ts +++ b/packages/host/src/compose-coding-agents.ts @@ -51,6 +51,7 @@ import { codingAgentServersParamsSchema, codingAgentStartParamsSchema, codingAgentStopParamsSchema, + codingAgentTranscriptKeySchema, codingAgentWorkspaceListParamsSchema, type DictationTranscribeAnswerView, dictationTranscribeParamsSchema, @@ -63,6 +64,7 @@ import { planTemplateWriteParamsSchema, providerKeyRemoveParamsSchema, providerKeySaveParamsSchema, + TRANSCRIPT_KIND, } from "@luke/features/coding-agent-view"; import { inboxReadAllRequestSchema } from "@luke/features/inbox-wire"; import { @@ -74,7 +76,7 @@ import { } from "@luke/gateway"; import { unparsedWire } from "@luke/wire"; import { readEither } from "@luke/wire/effect"; -import { Effect, FiberHandle, RcMap, Result, type Schema, type Scope } from "effect"; +import { Effect, FiberHandle, FiberMap, RcMap, Result, type Schema, type Scope } from "effect"; import * as HttpClient from "effect/http/HttpClient"; import { followAccountInbox } from "./account-inbox-follow.js"; import type { AccountComposer } from "./compose-account.js"; @@ -84,6 +86,7 @@ import { type LiveClient, openRealtimeClient } from "./live-follow.js"; import type { RunMode } from "./run-mode.js"; import { inboxPast, type ScopeInboxes, scopeInboxes } from "./scope-inbox-follow.js"; import { + followTranscript, type TranscriptFollows, TranscriptKey, transcriptFollows, @@ -218,14 +221,16 @@ export function composeCodingAgents( openLive: dependencies.openLive ?? openRealtimeClient, now: dependencies.kernel.now, }); - return composed(dependencies, http, follower, { inboxes, transcripts }); + const followed = yield* FiberMap.make(); + return composed(dependencies, http, follower, { inboxes, transcripts, followed }); }); } -/** The subscriptions the windows' reads share: each plan's and agent's inbox, and each transcript. */ +/** The subscriptions the windows' reads share: each plan's and agent's inbox, and each transcript; and the transcripts main follows, one fiber each. */ interface Follows { readonly inboxes: ScopeInboxes; readonly transcripts: TranscriptFollows; + readonly followed: FiberMap.FiberMap; } /** The composer's one table, built from its seams; nothing here starts anything but the account inbox's follower, when main asks for it, and a scope's inbox or a transcript's subscription, while a window reads it. */ @@ -233,7 +238,7 @@ function composed( dependencies: CodingAgentsDependencies, http: HttpClient.HttpClient, follower: FiberHandle.FiberHandle, - { inboxes, transcripts }: Follows, + { inboxes, transcripts, followed }: Follows, ): CodingAgentsComposer { const { kernel, client } = dependencies; const gate = gateOf(dependencies); @@ -249,6 +254,20 @@ function composed( return Result.isFailure(result) ? invalid(sentence) : Effect.succeed(result.success); }; + /** The agent's or worker's transcript a follow names, and the key its subscription and follow are held under. */ + const agentTranscriptOf = ( + params: Parameters[0], + ) => + Effect.gen(function* () { + const sentence = "following a transcript names an agent, and a worker's call where it is one"; + const transcript = yield* read(codingAgentTranscriptKeySchema, params, sentence); + if (transcript.kind !== TRANSCRIPT_KIND.AGENT) return yield* invalid(sentence); + return { + transcript, + key: new TranscriptKey({ agentId: transcript.agentId, callId: transcript.callId }), + }; + }); + const methods: GatewayMethodTable = { [GATEWAY_METHOD.CODING_AGENTS_MODELS]: () => Effect.map(asked(client.models()), (answer) => carried(answer)), @@ -389,6 +408,30 @@ function composed( : UNANSWERED; return carried(answer); }), + [GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_FOLLOW]: (params) => + Effect.gen(function* () { + const { transcript, key } = yield* agentTranscriptOf(params); + const tell = (page: CodingAgentMessagesAnswerView) => + Effect.sync(() => + kernel.emit(GATEWAY_EVENT.TRANSCRIPT_CHANGED, carried({ transcript, page })), + ); + const follow = gate() + ? followTranscript( + transcripts, + key, + (after) => asked(client.messagesNow(key.agentId, after, key.callId)), + tell, + ) + : tell(UNANSWERED); + yield* FiberMap.run(followed, key, follow); + return {}; + }), + [GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_UNFOLLOW]: (params) => + Effect.gen(function* () { + const { key } = yield* agentTranscriptOf(params); + yield* FiberMap.remove(followed, key); + return {}; + }), [GATEWAY_METHOD.CODING_AGENTS_MESSAGE]: (params) => Effect.gen(function* () { const { agentId, ...request } = yield* read( @@ -700,6 +743,7 @@ function composed( const reset = Effect.gen(function* () { yield* FiberHandle.clear(follower); + yield* FiberMap.clear(followed); for (const scope of yield* RcMap.keys(inboxes)) yield* RcMap.invalidate(inboxes, scope); for (const key of yield* RcMap.keys(transcripts)) yield* RcMap.invalidate(transcripts, key); }); diff --git a/packages/host/src/compose-tasks.test.ts b/packages/host/src/compose-tasks.test.ts index df67c2b00e..0c2edf89ba 100644 --- a/packages/host/src/compose-tasks.test.ts +++ b/packages/host/src/compose-tasks.test.ts @@ -15,8 +15,10 @@ import type { Board } from "@luke/features/board-wire"; import { CODING_AGENT_CALL_FAILURE, type CodingAgentMessagesAnswerView, + TRANSCRIPT_KIND, } from "@luke/features/coding-agent-view"; import { + CODING_AGENT_CURSOR_START, CODING_AGENT_STATUS, type CodingAgentMessage, type CodingAgentMessagesAnswer, @@ -85,7 +87,7 @@ import { GATEWAY_METHOD, type GatewayMethod, } from "@luke/gateway"; -import { MESSAGE_ROLE, type WireRecord } from "@luke/wire"; +import { MESSAGE_ROLE, type WireRecord, type WireValue } from "@luke/wire"; import { readEither } from "@luke/wire/effect"; import { noNetwork } from "@luke/wire/testing"; import { Deferred, Duration, Effect, Exit, Fiber, Queue, Result } from "effect"; @@ -998,6 +1000,8 @@ interface StandingCall { function subject(service: FakeService, options: { signedIn?: boolean; call?: StandingCall } = {}) { return Effect.gen(function* () { const told: TasksView[] = []; + /** Every followed transcript's page the host told, in order. */ + const pages: WireValue[] = []; const waiters: { wanted: (view: TasksView) => boolean; seen: Deferred.Deferred; @@ -1011,6 +1015,7 @@ function subject(service: FakeService, options: { signedIn?: boolean; call?: Sta now: () => NOW, runMode: { sendsNetwork: true }, emit: (kind, payload) => { + if (kind === GATEWAY_EVENT.TRANSCRIPT_CHANGED) pages.push(payload); if (kind !== GATEWAY_EVENT.TASKS_CHANGED) return; const read = readEither(tasksViewSchema)(payload); assert.ok(Result.isSuccess(read), "the planning event carries a view"); @@ -1051,7 +1056,7 @@ function subject(service: FakeService, options: { signedIn?: boolean; call?: Sta waiters.push({ wanted, seen }); return Deferred.await(seen); }); - return { tasks, call, told, last, viewWhere, viewLands, conversations, realtime }; + return { tasks, call, told, pages, last, viewWhere, viewLands, conversations, realtime }; }); } @@ -3046,3 +3051,46 @@ it.effect( assert.deepEqual(conversations.asks, []); }), ); + +it.effect( + "a followed planning worker's transcript is told from the start, the next page at once after one that said something, then again only as its statements move; letting it go ends the telling", + () => + Effect.gen(function* () { + const service = fakeService([task(INVITES, "Teammate invitations", "# Draft", 10)]); + const { call, pages, conversations, realtime } = yield* subject(service); + conversations.workerSealed = SEALED_WORKER; + const transcript = { kind: TRANSCRIPT_KIND.TASK_WORKER, taskId: INVITES, callId: "call-0-0" }; + assert.deepEqual(yield* call(GATEWAY_METHOD.TASKS_WORKER_TRANSCRIPT_FOLLOW, transcript), {}); + const said = workerPage( + [journal("worker-1", "Compared them.")], + CODING_AGENT_STATUS.RUNNING, + false, + ); + const caught = workerPage([], CODING_AGENT_STATUS.RUNNING, true); + yield* conversations.answerWorker(said); + yield* conversations.answerWorker(caught); + yield* realtime.deliver(1); + const ended = workerPage([], CODING_AGENT_STATUS.COMPLETED, false); + yield* conversations.answerWorker(ended); + yield* Effect.repeat(Effect.yieldNow, { times: 100 }); + + const windowPage = ({ holds: _holds, ...page }: TranscriptNowAnswer) => page; + assert.deepEqual(pages, [ + { transcript, page: windowPage(said) }, + { transcript, page: windowPage(caught) }, + { transcript, page: windowPage(ended) }, + ]); + assert.deepEqual(conversations.workerReads, [ + `${INVITES}:call-0-0:${CODING_AGENT_CURSOR_START}`, + `${INVITES}:call-0-0:3:0`, + `${INVITES}:call-0-0:3:0`, + ]); + assert.ok(!JSON.stringify(pages).includes("capability")); + + yield* call(GATEWAY_METHOD.TASKS_WORKER_TRANSCRIPT_UNFOLLOW, transcript); + yield* realtime.deliver(2); + yield* Effect.repeat(Effect.yieldNow, { times: 100 }); + assert.equal(pages.length, 3); + assert.equal(conversations.workerReads.length, 3); + }), +); diff --git a/packages/host/src/compose-tasks.ts b/packages/host/src/compose-tasks.ts index d3866cd366..69771358ba 100644 --- a/packages/host/src/compose-tasks.ts +++ b/packages/host/src/compose-tasks.ts @@ -15,6 +15,8 @@ import { CODING_AGENT_CALL_FAILURE, type CodingAgentCallFailure, type CodingAgentMessagesAnswerView, + codingAgentTranscriptKeySchema, + TRANSCRIPT_KIND, } from "@luke/features/coding-agent-view"; import { applyMessagesPage, @@ -122,6 +124,7 @@ import { Effect, Equal, FiberHandle, + FiberMap, Option, Queue, RcMap, @@ -138,7 +141,12 @@ import type { HostKernel } from "./host-kernel.js"; import { type LiveClient, openRealtimeClient } from "./live-follow.js"; import type { RunMode } from "./run-mode.js"; import { codeExcerpt, highlightCode, highlightFile } from "./task-code.js"; -import { type TranscriptFollows, transcriptFollows, transcriptPast } from "./transcript-follow.js"; +import { + followTranscript, + type TranscriptFollows, + transcriptFollows, + transcriptPast, +} from "./transcript-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. @@ -557,6 +565,9 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct now: kernel.now, }); + /** The planning workers' transcripts main follows, one fiber each. */ + const followedWorkers = yield* FiberMap.make(); + /** The tasks whose call just said something, whose parked conversation read the loop below wakes. */ const conversationWakes = yield* Queue.sliding(1); @@ -1223,6 +1234,20 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct }); } + /** The planning worker's transcript a follow names, or the invalid-params refusal. */ + function workerTranscriptOf( + params: Parameters[0], + ) { + return Effect.gen(function* () { + const sentence = "following a planning worker's transcript names its task and its call"; + const read = readEither(codingAgentTranscriptKeySchema)(unparsedWire(params)); + if (Result.isFailure(read) || read.success.kind !== TRANSCRIPT_KIND.TASK_WORKER) { + return yield* invalid(sentence); + } + return read.success; + }); + } + const methods: GatewayMethodTable = { // The workspaces are read beside the tasks rather than before them, and // only where the last read is stale: the first refresh after a launch or @@ -1329,6 +1354,33 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct }); return carried(answer); }), + // A planning worker's transcript is followed for main, each page told as an event. + [GATEWAY_METHOD.TASKS_WORKER_TRANSCRIPT_FOLLOW]: (params) => + Effect.gen(function* () { + const transcript = yield* workerTranscriptOf(params); + const { taskId, callId } = transcript; + const key = new TaskWorkerKey({ taskId, callId }); + const tell = (page: CodingAgentMessagesAnswerView) => + Effect.sync(() => + kernel.emit(GATEWAY_EVENT.TRANSCRIPT_CHANGED, carried({ transcript, page })), + ); + const follow = gate() + ? followTranscript( + workerTranscripts, + key, + (after) => served(conversations.taskWorkerMessagesNow(taskId, callId, after)), + tell, + ) + : tell({ failure: CODING_AGENT_CALL_FAILURE.UNANSWERED }); + yield* FiberMap.run(followedWorkers, key, follow); + return {}; + }), + [GATEWAY_METHOD.TASKS_WORKER_TRANSCRIPT_UNFOLLOW]: (params) => + Effect.gen(function* () { + const { taskId, callId } = yield* workerTranscriptOf(params); + yield* FiberMap.remove(followedWorkers, new TaskWorkerKey({ taskId, callId })); + return {}; + }), // A file's window is answered to the File tab that asked, coloured here, // and held nowhere: the tab keeps the windows it has scrolled through. [GATEWAY_METHOD.TASKS_FILE]: (params) => @@ -2164,6 +2216,7 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct reset: Effect.gen(function* () { yield* endTaskCall(undefined); yield* FiberHandle.clear(conversationRead); + yield* FiberMap.clear(followedWorkers); for (const key of yield* RcMap.keys(workerTranscripts)) { yield* RcMap.invalidate(workerTranscripts, key); } diff --git a/packages/host/src/host-operator.ts b/packages/host/src/host-operator.ts index 913b9ba576..70dda1078f 100644 --- a/packages/host/src/host-operator.ts +++ b/packages/host/src/host-operator.ts @@ -74,6 +74,8 @@ import { type CodingAgentServersParams, type CodingAgentStartParams, type CodingAgentStopParams, + type CodingAgentTranscriptChanged, + type CodingAgentTranscriptKey, type CodingAgentWorkspaceListParams, codingAgentAgentAnswerSchema, codingAgentChangesAnswerViewSchema, @@ -98,6 +100,7 @@ import { codingAgentPullRequestViewedAnswerViewSchema, codingAgentPullRequestWriteAnswerViewSchema, codingAgentReviewOptionsAnswerViewSchema, + codingAgentTranscriptChangedSchema, type DictationTranscribeAnswerView, type DictationTranscribeParams, dictationTranscribeAnswerViewSchema, @@ -114,6 +117,7 @@ import { planTemplateRevisionAnswerViewSchema, planTemplateRevisionsAnswerViewSchema, providerKeysAnswerViewSchema, + TRANSCRIPT_KIND, } from "@luke/features/coding-agent-view"; import { FACTORY_WORLD_CALL_FAILURE, @@ -668,6 +672,12 @@ export interface HostOperator { codingAgentAccountInboxUnfollow(): Effect.Effect; /** What waits on the developer across the account, whole, each time it changes while it is followed. */ onAccountInboxChanged(listener: (inbox: AccountInboxAnswer) => void): () => void; + /** One agent's, worker's, or planning worker's transcript, followed from now until let go or signed out, its pages heard through `onTranscriptChanged`; following again starts over. */ + transcriptFollow(transcript: CodingAgentTranscriptKey): Effect.Effect; + /** One followed transcript let go: nothing more is heard of it until it is followed again. */ + transcriptUnfollow(transcript: CodingAgentTranscriptKey): Effect.Effect; + /** A followed transcript's next page, or why there is none, each time it moves. */ + onTranscriptChanged(listener: (changed: CodingAgentTranscriptChanged) => void): () => void; /** One image of an inbox item, fetched under the account; its bytes in base64 under its media type, or why there is none. */ codingAgentInboxAttachment( params: CodingAgentInboxAttachmentParams, @@ -804,6 +814,15 @@ function answered(value: UnparsedWireValue): Value | undefined { } /** What a coding-agent call the host could not answer is drawn as. */ +/** A followed transcript as the gateway's params carry it. */ +function transcriptParams(transcript: CodingAgentTranscriptKey): WireRecord { + if (transcript.kind === TRANSCRIPT_KIND.TASK_WORKER) { + return { kind: transcript.kind, taskId: transcript.taskId, callId: transcript.callId }; + } + const { kind, agentId, callId } = transcript; + return callId === undefined ? { kind, agentId } : { kind, agentId, callId }; +} + const CODING_AGENT_UNANSWERED = { failure: CODING_AGENT_CALL_FAILURE.UNANSWERED } as const; /** An automation call that reached no answer, as every automation method answers one. */ @@ -1885,6 +1904,33 @@ export function createHostOperator(options: HostOperatorOptions): HostOperator { ), listener, ), + transcriptFollow: (transcript) => + fire( + client.call( + transcript.kind === TRANSCRIPT_KIND.AGENT + ? GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_FOLLOW + : GATEWAY_METHOD.TASKS_WORKER_TRANSCRIPT_FOLLOW, + transcriptParams(transcript), + ), + ), + transcriptUnfollow: (transcript) => + fire( + client.call( + transcript.kind === TRANSCRIPT_KIND.AGENT + ? GATEWAY_METHOD.CODING_AGENTS_TRANSCRIPT_UNFOLLOW + : GATEWAY_METHOD.TASKS_WORKER_TRANSCRIPT_UNFOLLOW, + transcriptParams(transcript), + ), + ), + onTranscriptChanged: (listener) => + on( + GATEWAY_EVENT.TRANSCRIPT_CHANGED, + (payload) => + Result.getOrUndefined( + readEither(codingAgentTranscriptChangedSchema, { excess: EXCESS_KEYS.DROP })(payload), + ), + listener, + ), codingAgentInboxAttachment: (params) => Effect.map( client.call(GATEWAY_METHOD.CODING_AGENTS_INBOX_ATTACHMENT, { diff --git a/packages/host/src/transcript-follow.ts b/packages/host/src/transcript-follow.ts index aea420e83f..6e89aa3918 100644 --- a/packages/host/src/transcript-follow.ts +++ b/packages/host/src/transcript-follow.ts @@ -2,6 +2,7 @@ import { CODING_AGENT_CALL_FAILURE, type CodingAgentMessagesAnswerView, } from "@luke/features/coding-agent-view"; +import { CODING_AGENT_CURSOR_START } from "@luke/features/coding-agent-wire"; import { type AgentTranscriptLiveAnswer, agentTranscriptLiveRowsSchema, @@ -45,6 +46,11 @@ import { LiveBroken, type LiveClient, liveRows } from "./live-follow.js"; * a title it waits on closing, which no row says, the page is read again * then too; the delay is the service's own, so the two clocks never meet. * The hold ends with one more read of the page, as the service's own does. + * + * A transcript main follows for the windows that draw it (`followTranscript`) + * is read on the same subscription with no hold at all: every page is told + * as it is read, from the start, and the next is read whenever the + * statements move or the clock reaches the edge the service named. */ /** How long a read is held while nothing stands past its cursor, as the service held it. */ @@ -229,3 +235,55 @@ export function transcriptPast( }), ); } + +/** How one followed transcript's next page is read: the page past the cursor at once, with whether the service's held read would have held on it. */ +export type TranscriptPageAfter = ( + after: string, +) => Effect.Effect; + +/** + * The transcript followed until interrupted, each page told as it is read: + * the first from the start, then the next past it at once while the last + * said something, and otherwise once the statements move, their + * subscription stops or opens again, or the clock reaches the edge the + * service named. One the account does not reach is told not found, which + * ends the follow; a page the service did not answer is told as such and + * read again after `RESEAL`. + */ +export function followTranscript( + follows: TranscriptFollows, + key: Key, + pageAfter: TranscriptPageAfter, + tell: (page: CodingAgentMessagesAnswerView) => Effect.Effect, +): Effect.Effect { + return Effect.scoped( + Effect.gen(function* () { + const standing = yield* RcMap.get(follows, key); + if (standing.kind === "refused") { + return yield* tell({ failure: CODING_AGENT_CALL_FAILURE.NOT_FOUND }); + } + let after = CODING_AGENT_CURSOR_START; + while (true) { + const was = yield* SubscriptionRef.get(standing.state); + const answer = yield* pageAfter(after); + yield* tell(windowsPage(answer)); + if ("failure" in answer) { + if (answer.failure === CODING_AGENT_CALL_FAILURE.NOT_FOUND) return; + yield* Effect.sleep(RESEAL); + continue; + } + after = answer.cursor; + // Note that a page that said something is followed by the next at once, since a page has a most it carries. + if (answer.messages.length > 0) continue; + const moved = SubscriptionRef.changes(standing.state).pipe( + Stream.filter(movedFrom(was)), + Stream.runHead, + ); + const wakeIn = "holds" in answer && answer.holds ? answer.wakeIn : undefined; + yield* wakeIn === undefined + ? moved + : Effect.race(moved, Effect.sleep(Duration.millis(wakeIn))); + } + }), + ); +}