feat(service,host): a task's planning conversation is followed through the shared live layer

apps/web/tests/tasks-app.test.ts

@@ -1071,7 +1071,7 @@ it.layer(getTestDB())("the plan routes", (it) => {
   );
 
   it.effect(
-    "a plan never planned on reads as an empty conversation on the agent tab's wire, and as no code shown",
+    "a plan never planned on reads as an empty conversation on the agent tab's wire, in one look as much as held, and as no code shown",
     () =>
       Effect.gen(function* () {
         const { owner, ask } = yield* openAccounts();
@@ -1095,6 +1095,20 @@ it.layer(getTestDB())("the plan routes", (it) => {
           )).status,
           HOSTED_HTTP_STATUS.BAD_REQUEST,
         );
+        // One look past a cursor is the only read `hold` names, and `none` the only word it takes.
+        const look = yield* ask(
+          request(MESSAGES, owner, { id: taskId, query: { after: page.cursor, hold: "none" } }),
+        );
+        assert.deepEqual(
+          readAnswer(codingAgentMessagesAnswerSchema, HOSTED_HTTP_STATUS.OK, look).messages,
+          [],
+        );
+        for (const query of [{ hold: "none" }, { after: page.cursor, hold: "forever" }]) {
+          assert.equal(
+            (yield* ask(request(MESSAGES, owner, { id: taskId, query }))).status,
+            HOSTED_HTTP_STATUS.BAD_REQUEST,
+          );
+        }
       }),
   );
 

packages/features/src/coding-agent-client.ts

@@ -110,6 +110,7 @@ import {
   type AgentTranscriptLiveAnswer,
   LIVE_SIGNED_OUT,
   LIVE_USER_QUERY,
+  type TaskConversationLiveAnswer,
   TRANSCRIPT_NOW_QUERY,
   type TranscriptNowAnswer,
 } from "./live-wire.js";
@@ -678,6 +679,36 @@ export class HostedCodingAgentClient {
    * the account `userId` names, and signed out where the session is no
    * longer that account's.
    */
+  /**
+   * One task's planning conversation, its mark and running workers sealed
+   * for following over Neon Realtime, or with `poll` their rows inline:
+   * not found where the account reaches no such task, signed out where
+   * `userId` is not the bearer's own, and unanswered for anything else,
+   * which the caller answers by sealing it again.
+   */
+  taskConversationLive(
+    taskId: string,
+    userId: string,
+    poll = false,
+    after?: string,
+  ): Effect.Effect<
+    TaskConversationLiveAnswer | Failed | typeof LIVE_SIGNED_OUT,
+    never,
+    HttpClient.HttpClient
+  > {
+    return Effect.catchIf(
+      askLive(
+        this.#hosted.live.taskConversation,
+        { id: taskId, [LIVE_USER_QUERY]: userId },
+        poll,
+        after,
+      ),
+      (failed) =>
+        failed._tag === "HostedRefused" && failed.refusal === HOSTED_API_ERROR.INVALID_TOKEN,
+      () => Effect.succeed(LIVE_SIGNED_OUT),
+    ).pipe((signed) => hostedAnswer(signed, ROW_REFUSALS));
+  }
+
   agentTranscriptLive(
     agentId: string,
     userId: string,
@@ -707,7 +738,7 @@ export class HostedCodingAgentClient {
    */
   taskMessages(
     taskId: string,
-    page: { after?: string; before?: string },
+    page: { after?: string; before?: string; unheld?: boolean },
     callId?: string,
   ): Effect.Effect<CodingAgentMessagesAnswerView, never, HttpClient.HttpClient> {
     const asked =
@@ -718,6 +749,7 @@ export class HostedCodingAgentClient {
               id: taskId,
               ...(page.after === undefined ? undefined : { after: page.after }),
               ...(page.before === undefined ? undefined : { before: page.before }),
+              ...(page.unheld === true ? { hold: "none" } : undefined),
             },
             HOSTED_DEADLINE.HELD,
           )

packages/features/src/contract/live.ts

@@ -4,10 +4,11 @@
  * 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 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; and
- * the commands of the kinds `?kinds=` names that no Mac of the account has
- * claimed, as one statement. Each
+ * marks where it stands, its pages and calls the same way, and its planning
+ * conversation as its mark and its running workers; and one coding agent's
+ * transcript, or one of its workers' under `&call=`, as the statements its
+ * pages are read from; and the commands of the kinds `?kinds=` names that
+ * no Mac of the account has claimed, as one statement. 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
  * poll is held until the rows move past it, or answered unchanged. Each
@@ -26,6 +27,7 @@ import {
   LIVE_USER_QUERY,
   macCommandsLiveAnswerSchema,
   taskBoardLiveAnswerSchema,
+  taskConversationLiveAnswerSchema,
   taskRecordLiveAnswerSchema,
 } from "../live-wire.js";
 import { HOSTED_SERVICE_PATH } from "../service-paths.js";
@@ -85,6 +87,10 @@ export const liveContract = {
     .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_TASK_RECORD }))
     .input(liveTaskSchema)
     .output(EffectSchema.flip(taskRecordLiveAnswerSchema)),
+  taskConversation: hostedProcedure
+    .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_TASK_CONVERSATION }))
+    .input(liveTaskSchema)
+    .output(EffectSchema.flip(taskConversationLiveAnswerSchema)),
   agentTranscript: hostedProcedure
     .meta(openapi({ method: HTTP_METHOD.GET, path: HOSTED_SERVICE_PATH.LIVE_AGENT_TRANSCRIPT }))
     .input(liveTranscriptSchema)

packages/features/src/contract/messages.ts

@@ -99,10 +99,12 @@ export const messagesContract = {
     .meta(openapi({ method: HTTP_METHOD.GET, path: "/api/agents/{id}/subagents/{call}/messages" }))
     .input(EffectSchema.Struct({ ...byCall, ...agentTranscriptQuery }))
     .output(EffectSchema.flip(agentTranscriptAnswerSchema)),
-  /** One task's conversation, past the cursor and held while a turn runs, or a page of its turns. */
+  /** One task's conversation, past the cursor and held while a turn runs, or in one look where `hold=none` asks, or a page of its turns. */
   task: hostedProcedure
     .meta(openapi({ method: HTTP_METHOD.GET, path: "/api/plans/{id}/messages" }))
-    .input(EffectSchema.Struct({ ...byId, ...transcriptQuery, before: queryValue }))
+    .input(
+      EffectSchema.Struct({ ...byId, ...transcriptQuery, before: queryValue, hold: queryValue }),
+    )
     .output(EffectSchema.flip(codingAgentMessagesAnswerSchema)),
   /** One task's brain worker's transcript past the cursor, held while it runs. */
   taskWorker: hostedProcedure

packages/features/src/live-wire.ts

@@ -459,3 +459,39 @@ export function macCommandsOfLiveRows(
           : { id: row.id, kind: row.kind, params: row.params, timeoutSeconds: row.timeoutSeconds },
     );
 }
+
+/**
+ * A plan's planning conversation sealed for following: the address, and
+ * its mark as one statement, one row that moves whenever a read past the
+ * cursor would answer something new. The desktop holds its read of the
+ * conversation and asks it once, unheld, each time the mark moves, rather
+ * than holding a read the service looks again for on a clock. The service
+ * answers `not-found` for a plan the account does not reach, and
+ * `unavailable` where its deployment's Realtime is misconfigured, which the
+ * desktop answers by asking again.
+ */
+export const taskConversationLiveAnswerSchema = EffectSchema.Struct({
+  url: EffectSchema.String,
+  queries: EffectSchema.Struct({ mark: sealedLiveQuerySchema, workers: sealedLiveQuerySchema }),
+});
+
+export type TaskConversationLiveAnswer = typeof taskConversationLiveAnswerSchema.Type;
+
+/**
+ * Where the conversation stands as its row delivers it, keyed by the plan:
+ * the highest message sequence and the journal revision, and the newest
+ * turn and its status (null before the first).
+ */
+const taskConversationMarkSchema = EffectSchema.Struct({
+  id: EffectSchema.String,
+  seq: EffectSchema.Finite,
+  revision: EffectSchema.Finite,
+  turnId: EffectSchema.NullOr(EffectSchema.String),
+  turnStatus: EffectSchema.NullOr(EffectSchema.String),
+});
+
+/** The conversation's mark as Realtime delivers it, one row or none for a plan the account no longer reaches, and its running workers' turns, a row each by id, in no order. */
+export const taskConversationLiveRowsSchema = EffectSchema.Struct({
+  mark: EffectSchema.Array(taskConversationMarkSchema),
+  workers: EffectSchema.Array(EffectSchema.Struct({ id: EffectSchema.String })),
+});

packages/features/src/service-paths.ts

@@ -90,6 +90,8 @@ export const HOSTED_SERVICE_PATH = {
   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",
+  /** One task's planning conversation under `?id=`, its mark and running workers sealed for following over Neon Realtime (GET), or `not-found` or `unavailable`. */
+  LIVE_TASK_CONVERSATION: "/api/live/task-conversation",
   /**
    * 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