feat(service,host): a task's pages and calls follow over the live layer

packages/features/src/contract/live.ts

@@ -4,7 +4,7 @@
  * What waits on the developer across the account, as the address to
  * subscribe at and one sealed statement per part of the account inbox's
  * answer; and one task's whiteboard, named by `?id=`, as the statement that
- * marks where it stands; and one coding agent's transcript, or one of its
+ * marks where it stands, and its pages and calls the same way; and one coding agent's transcript, or one of its
  * workers' under `&call=`, as the statements its pages are read from. Each
  * asked with `?poll=true` answers each statement's rows inline instead, for
  * a reader polling while Realtime fails or is off; with `&after=` too, the
@@ -23,6 +23,7 @@ import {
   LIVE_POLL,
   LIVE_USER_QUERY,
   taskBoardLiveAnswerSchema,
+  taskRecordLiveAnswerSchema,
 } from "../live-wire.js";
 import { HOSTED_SERVICE_PATH } from "../service-paths.js";
 import { hostedProcedure } from "./base.js";
@@ -69,6 +70,10 @@ export const liveContract = {
     .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_TASK_BOARD }))
     .input(liveTaskSchema)
     .output(EffectSchema.flip(taskBoardLiveAnswerSchema)),
+  taskRecord: hostedProcedure
+    .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_TASK_RECORD }))
+    .input(liveTaskSchema)
+    .output(EffectSchema.flip(taskRecordLiveAnswerSchema)),
   agentTranscript: hostedProcedure
     .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_AGENT_TRANSCRIPT }))
     .input(liveTranscriptSchema)

packages/features/src/live-wire.ts

@@ -281,6 +281,54 @@ export const taskBoardLiveRowsSchema = EffectSchema.Struct({
 
 export type TaskBoardLiveRows = typeof taskBoardLiveRowsSchema.Type;
 
+/**
+ * An open task's pages and calls sealed for following: the address, and one
+ * statement a part, each a mark that moves as its rows do and carries none
+ * of their words: the pages agents showed (`pages`) and the calls about the
+ * task (`calls`). The desktop reads a part again once its mark moves,
+ * rather than when a call's activity or end suggests it might have; the
+ * board is followed on its own (`taskBoardLiveAnswerSchema`). The service
+ * answers `not-found` for a task the account does not reach, and
+ * `unavailable` where its deployment's Realtime is misconfigured, which the
+ * desktop answers by asking again.
+ */
+export const taskRecordLiveAnswerSchema = EffectSchema.Struct({
+  url: EffectSchema.String,
+  queries: EffectSchema.Struct({
+    pages: sealedLiveQuerySchema,
+    calls: sealedLiveQuerySchema,
+  }),
+});
+
+export type TaskRecordLiveAnswer = typeof taskRecordLiveAnswerSchema.Type;
+
+/** One page shown on the task as its row delivers it: its newest version's number. */
+const taskPageLiveRowSchema = EffectSchema.Struct({
+  pageId: EffectSchema.String,
+  version: EffectSchema.Finite,
+});
+
+/** One call about the task as its row delivers it: when and how it ended, null while it stands; what it metered and its model; its last fragment; and how many rows the voice wrote for it. */
+const taskCallLiveRowSchema = EffectSchema.Struct({
+  id: EffectSchema.String,
+  closedAt: EffectSchema.NullOr(EffectSchema.Finite),
+  closeReason: EffectSchema.NullOr(EffectSchema.String),
+  usage: EffectSchema.NullOr(
+    EffectSchema.Struct({ seconds: EffectSchema.Finite, confirmed: EffectSchema.Boolean }),
+  ),
+  model: EffectSchema.NullOr(EffectSchema.String),
+  lastSegment: EffectSchema.Finite,
+  spoken: EffectSchema.Finite,
+});
+
+/** The task's pages and calls as Realtime delivers them, a part's rows each: none of any for a task the account no longer reaches. */
+export const taskRecordLiveRowsSchema = EffectSchema.Struct({
+  pages: EffectSchema.Array(taskPageLiveRowSchema),
+  calls: EffectSchema.Array(taskCallLiveRowSchema),
+});
+
+export type TaskRecordLiveRows = typeof taskRecordLiveRowsSchema.Type;
+
 /**
  * A coding agent's transcript, or one of its workers', sealed for
  * following: the address, and the statements a page past the reader's

packages/features/src/service-paths.ts

@@ -86,6 +86,8 @@ export const HOSTED_SERVICE_PATH = {
   LIVE_INBOX: "/api/live/inbox",
   /** One task's whiteboard under `?id=`, sealed for following over Neon Realtime (GET): where it stands, or `not-found` or `unavailable`. */
   LIVE_TASK_BOARD: "/api/live/task-board",
+  /** One task's pages and calls under `?id=`, each part's mark sealed for following over Neon Realtime (GET), or `not-found` or `unavailable`. */
+  LIVE_TASK_RECORD: "/api/live/task-record",
   /** One agent's transcript under `?id=`, or one of its workers' under `&call=`, sealed for following over Neon Realtime (GET): what its pages are read from, or `not-found` or `unavailable`. */
   LIVE_AGENT_TRANSCRIPT: "/api/live/agent-transcript",
   /**

packages/features/src/task-client.ts

@@ -35,7 +35,12 @@ import {
   hostedClient,
 } from "./hosted-link.js";
 import { imageBodyOf } from "./image-body.js";
-import { LIVE_SIGNED_OUT, LIVE_USER_QUERY, type TaskBoardLiveAnswer } from "./live-wire.js";
+import {
+  LIVE_SIGNED_OUT,
+  LIVE_USER_QUERY,
+  type TaskBoardLiveAnswer,
+  type TaskRecordLiveAnswer,
+} from "./live-wire.js";
 import type { PageLine, PageVersion } from "./page-wire.js";
 import { PLAN_ATTACHMENT_REFUSAL } from "./plan-attachment-wire.js";
 import {
@@ -440,6 +445,29 @@ export class HostedTaskClient {
     );
   }
 
+  /**
+   * Where one task's pages and calls stand, sealed for following over Neon
+   * Realtime, or with `poll` their rows inline, for a reader polling while
+   * Realtime fails them, for the account `userId` names: not found where the
+   * account reaches no such task, signed out where the session is another's,
+   * unanswered for anything else, which the caller answers by sealing again.
+   */
+  recordLive(
+    taskId: string,
+    userId: string,
+    poll = false,
+    after?: string,
+  ): Effect.Effect<
+    TaskCallResult<TaskRecordLiveAnswer, TaskCallFailure | typeof LIVE_SIGNED_OUT.failure>,
+    never,
+    HttpClient.HttpClient
+  > {
+    return hostedResult(
+      askLive(this.#hosted.live.taskRecord, { id: taskId, [LIVE_USER_QUERY]: userId }, poll, after),
+      LIVE_REFUSALS,
+    );
+  }
+
   /** What was said on one task's calls; nothing where the service did not answer it. */
   readTranscript(
     taskId: string,

packages/host/src/compose-tasks.test.ts

@@ -34,7 +34,7 @@ import {
   type RepositoryIssuesAsk,
 } from "@luke/features/github-repositories-wire";
 import type { LinearIssueSummary } from "@luke/features/linear-wire";
-import type { TaskBoardLiveAnswer } from "@luke/features/live-wire";
+import type { TaskBoardLiveAnswer, TaskRecordLiveAnswer } from "@luke/features/live-wire";
 import type { PageLine, PageVersion } from "@luke/features/page-wire";
 import { PICTURE_TYPE, type PictureType } from "@luke/features/picture-wire";
 import {
@@ -165,6 +165,10 @@ interface FakeService extends TasksClient {
   boards: Record<string, Board>;
   /** Each task's board sealed for following, by task id; a task with none is not found, as the service answers one the account does not reach. */
   boardSeals: Record<string, TaskBoardLiveAnswer>;
+  /** Each task's pages and calls sealed for following, by task id; a task with none is not found, as the service answers one the account does not reach. */
+  recordSeals: Record<string, TaskRecordLiveAnswer>;
+  /** Every read of a task's pages or transcript, in order, so a test can see which a followed mark moved. */
+  readonly recordReads: string[];
   /** The image each task's last save carried, by task id, as the service keeps it beside the board. */
   readonly images: Record<string, string | undefined>;
   /** Each task's transcript, by task id; a task with none answers no transcript, as a service that did not answer. */
@@ -352,6 +356,8 @@ function fakeService(tasks: Task[]): FakeService {
     reads: [],
     boards: {},
     boardSeals: {},
+    recordSeals: {},
+    recordReads: [],
     images: {},
     transcripts: {},
     works: {},
@@ -376,7 +382,11 @@ function fakeService(tasks: Task[]): FakeService {
     readRevision: (_taskId, seq) => Effect.sync(() => service.revisions[seq]),
     pages: [],
     pageLines: {},
-    readPages: (taskId) => Effect.sync(() => service.pageLines[taskId]),
+    readPages: (taskId) =>
+      Effect.sync(() => {
+        service.recordReads.push("pages");
+        return service.pageLines[taskId];
+      }),
     pictures: new Map(),
     // The service keeps a picture under the next id, and refuses one more where the plan holds two, a bound small enough to reach.
     keepPicture: (taskId, picture) =>
@@ -771,7 +781,18 @@ function fakeService(tasks: Task[]): FakeService {
           ? { ok: false, failure: TASK_CALL_FAILURE.NOT_FOUND }
           : { ok: true, answer: sealed };
       }),
-    readTranscript: (taskId) => Effect.sync(() => service.transcripts[taskId]),
+    recordLive: (taskId) =>
+      Effect.sync(() => {
+        const sealed = service.recordSeals[taskId];
+        return sealed === undefined
+          ? { ok: false, failure: TASK_CALL_FAILURE.NOT_FOUND }
+          : { ok: true, answer: sealed };
+      }),
+    readTranscript: (taskId) =>
+      Effect.sync(() => {
+        service.recordReads.push("transcript");
+        return service.transcripts[taskId];
+      }),
     readWork: (taskId, before) =>
       Effect.sync(() => {
         service.workReads.push(before === undefined ? taskId : `${taskId}:${before}`);
@@ -3420,6 +3441,100 @@ it.effect(
     }),
 );
 
+/** A task's pages and calls sealed for following, named by the round they were sealed in. */
+function sealedRecord(round: number): TaskRecordLiveAnswer {
+  const query = (part: string) => ({
+    capability: `${part}-${round}`,
+    queryFingerprint: part,
+    expiresAt: 3_600_000,
+  });
+  return {
+    url: "wss://realtime.example/v1",
+    queries: { pages: query("pages"), calls: query("calls") },
+  };
+}
+
+/** One call about the task as its row delivers it, ended at `closedAt` or standing. */
+function callRow(lastSegment: number, closedAt: number | null) {
+  return {
+    id: "call-1",
+    closedAt,
+    closeReason: closedAt === null ? null : "remote_hangup",
+    usage: null,
+    model: "gpt-realtime",
+    lastSegment,
+    spoken: 0,
+  };
+}
+
+it.effect(
+  "the open task's pages are read again as their mark moves and its transcript once an ended call's does, not on the call's activity, while its end reads the work now and once more after its writes settle",
+  () =>
+    Effect.gen(function* () {
+      const service = fakeService([task(INVITES, "Teammate invitations", "# Invites", 10)]);
+      service.works[INVITES] = { turns: [] };
+      service.recordSeals[INVITES] = sealedRecord(0);
+      const realtime = fakeBoardRealtime();
+      const { call, tasks } = yield* subject(service, { openLive: () => realtime.client });
+      yield* tasks.lifetime;
+      yield* call(GATEWAY_METHOD.TASKS_OPEN, { taskId: INVITES });
+      yield* boardSettled;
+      assert.deepEqual(realtime.subscribed().toSorted(), ["calls-0", "pages-0"]);
+      const taken = () => service.recordReads.splice(0).toSorted();
+      const worked = () => service.workReads.splice(0);
+      taken();
+      worked();
+      const deliver = (pages: readonly unknown[], calls: readonly unknown[]) =>
+        Effect.andThen(
+          Effect.sync(() => {
+            realtime.deliver("pages-0", pages);
+            realtime.deliver("calls-0", calls);
+          }),
+          boardSettled,
+        );
+      const page = (version: number) => ({ pageId: "page-1", version });
+
+      // The first delivery reads every part, since the reads the task opened with may have missed it.
+      yield* deliver([page(1)], []);
+      assert.deepEqual(taken(), ["pages", "transcript"]);
+      assert.deepEqual(worked(), []);
+
+      yield* deliver([page(2)], []);
+      assert.deepEqual(taken(), ["pages"]);
+
+      // A call's words landing while it stands, and its activity, read nothing.
+      yield* deliver([page(2)], [callRow(3, null)]);
+      tasks.showActivity({
+        type: VOICE_SERVICE_FRAME.TASK_ACTIVITY,
+        planId: INVITES,
+        notes: false,
+      });
+      yield* boardSettled;
+      assert.deepEqual(taken(), []);
+      assert.deepEqual(worked(), []);
+
+      // Its end reads the work, and again once a turn's later writes have had time to land, but not the transcript.
+      tasks.callEnded(INVITES);
+      yield* boardSettled;
+      assert.deepEqual(worked(), [INVITES]);
+      yield* TestClock.adjust(Duration.seconds(5));
+      yield* boardSettled;
+      assert.deepEqual(worked(), [INVITES]);
+      assert.deepEqual(taken(), []);
+
+      // Its end, and its last words after, each read the transcript as the ended call's mark moves.
+      yield* deliver([page(2)], [callRow(3, 5_000)]);
+      assert.deepEqual(taken(), ["transcript"]);
+      yield* deliver([page(2)], [callRow(4, 5_000)]);
+      assert.deepEqual(taken(), ["transcript"]);
+      assert.deepEqual(worked(), []);
+
+      // Leaving the task lets the subscriptions go.
+      yield* call(GATEWAY_METHOD.TASKS_CLOSE);
+      assert.deepEqual(realtime.subscribed(), []);
+    }),
+);
+
 it.effect(
   "a renewal that finds the open task gone ends the board's follow, rather than asking again",
   () =>