feat(service,host): a live read names its account; another's session is signed out

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

@@ -107,6 +107,8 @@ import {
 } from "./inbox-wire.js";
 import {
   type AgentTranscriptLiveAnswer,
+  LIVE_SIGNED_OUT,
+  LIVE_USER_QUERY,
   TRANSCRIPT_NOW_QUERY,
   type TranscriptNowAnswer,
 } from "./live-wire.js";
@@ -645,19 +647,29 @@ export class HostedCodingAgentClient {
    * named, sealed for following over Neon Realtime, answered at once: not
    * found where the account reaches no such agent, unanswered for anything
    * else, which the caller answers by sealing it again; with `poll`, its
-   * rows inline, for a reader polling while Realtime fails it.
+   * rows inline, for a reader polling while Realtime fails it. Sealed for
+   * the account `userId` names, and signed out where the session is no
+   * longer that account's.
    */
   agentTranscriptLive(
     agentId: string,
+    userId: string,
     callId?: string,
     poll = false,
     after?: string,
-  ): Effect.Effect<AgentTranscriptLiveAnswer | Failed, never, HttpClient.HttpClient> {
-    const named = callId === undefined ? { id: agentId } : { id: agentId, call: callId };
-    return hostedAnswer(
+  ): Effect.Effect<
+    AgentTranscriptLiveAnswer | Failed | typeof LIVE_SIGNED_OUT,
+    never,
+    HttpClient.HttpClient
+  > {
+    const read = { id: agentId, [LIVE_USER_QUERY]: userId };
+    const named = callId === undefined ? read : { ...read, call: callId };
+    return Effect.catchIf(
       askLive(this.#hosted.live.agentTranscript, named, poll, after),
-      ROW_REFUSALS,
-    );
+      (failed) =>
+        failed._tag === "HostedRefused" && failed.refusal === HOSTED_API_ERROR.INVALID_TOKEN,
+      () => Effect.succeed(LIVE_SIGNED_OUT),
+    ).pipe((signed) => hostedAnswer(signed, ROW_REFUSALS));
   }
 
   /**

packages/features/src/contract/live.ts

@@ -8,7 +8,9 @@
  * 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
- * poll is held until the rows move past it, or answered unchanged. Each output is the wire schema flipped
+ * poll is held until the rows move past it, or answered unchanged. Each
+ * names the account it was started for under `?userId=`, and a session of
+ * any other is answered as signed out. Each output is the wire schema flipped
  * (`packages/service/src/orpc/base.ts`).
  */
 
@@ -19,6 +21,7 @@ import {
   accountInboxLiveAnswerSchema,
   agentTranscriptLiveAnswerSchema,
   LIVE_POLL,
+  LIVE_USER_QUERY,
   taskBoardLiveAnswerSchema,
 } from "../live-wire.js";
 import { HOSTED_SERVICE_PATH } from "../service-paths.js";
@@ -30,17 +33,22 @@ const pollField = EffectSchema.optionalKey(EffectSchema.Literal(LIVE_POLL.ON));
 /** Where the rows stood when the reader last polled (`liveDigestOf`); the poll is held until they move past it. */
 const afterField = EffectSchema.optionalKey(EffectSchema.String);
 
+/** The account the reader started following for, which the session must be (`LIVE_USER_QUERY`). */
+const userField = EffectSchema.optionalKey(EffectSchema.String);
+
 /** A live read's own query: whether it polls. */
 const liveReadSchema = EffectSchema.Struct({
   [LIVE_POLL.QUERY]: pollField,
   [LIVE_POLL.AFTER]: afterField,
+  [LIVE_USER_QUERY]: userField,
 });
 
 /** The task a live read is followed for; one that is no UUID is answered as not found, so it is read as text here. */
 const liveTaskSchema = EffectSchema.Struct({
   id: EffectSchema.String,
   [LIVE_POLL.QUERY]: pollField,
   [LIVE_POLL.AFTER]: afterField,
+  [LIVE_USER_QUERY]: userField,
 });
 
 /** The agent a transcript is followed for, and the delegation call of the worker where it is a worker's; each read as text, an id or call of no shape answered as not found. */
@@ -49,6 +57,7 @@ const liveTranscriptSchema = EffectSchema.Struct({
   call: EffectSchema.optionalKey(EffectSchema.String),
   [LIVE_POLL.QUERY]: pollField,
   [LIVE_POLL.AFTER]: afterField,
+  [LIVE_USER_QUERY]: userField,
 });
 
 export const liveContract = {

packages/features/src/live-wire.ts

@@ -43,6 +43,17 @@ export const LIVE_POLL_URL = "luke-poll:";
 /** How a reader asks for a live read's rows inline rather than sealed, while Realtime fails it. */
 export const LIVE_POLL = { QUERY: "poll", ON: "true", AFTER: "after" } as const;
 
+/**
+ * The query every live read names the account it was started for under, as
+ * Neon's guide says a page sends its user ("When the session changes"): the
+ * service answers a session of anyone else, or a read naming none, as
+ * signed out.
+ */
+export const LIVE_USER_QUERY = "userId";
+
+/** A live read the service refused as signed out: no session, or one of another account than the read names. */
+export const LIVE_SIGNED_OUT = { failure: "signed-out" } as const;
+
 /**
  * Where a polled read's rows stand as one value: each part's digest, in
  * the parts' name order. A reader sends the one it last saw as `after`, and

packages/features/src/task-client.test.ts

@@ -2,6 +2,7 @@ import assert from "node:assert/strict";
 import { it } from "@effect/vitest";
 import {
   fakeCloudApi,
+  HTTP_STATUS,
   type JsonValue,
   recordedRoutes,
   recordingHttpClient,
@@ -11,7 +12,7 @@ import { TestClock } from "effect/testing";
 import { accountBearer } from "./account-call.js";
 import { CODING_AGENT_CALL_FAILURE } from "./coding-agent-view.js";
 import { CODING_AGENT_STATUS } from "./coding-agent-wire.js";
-import { LIVE_POLL, LIVE_POLL_URL } from "./live-wire.js";
+import { LIVE_POLL, LIVE_POLL_URL, LIVE_SIGNED_OUT, LIVE_USER_QUERY } from "./live-wire.js";
 import { PLAN_ATTACHMENT_REFUSAL } from "./plan-attachment-wire.js";
 import { REPOSITORY_UNREACHED } from "./repository-reach-wire.js";
 import { HOSTED_API_ERROR } from "./service-wire.js";
@@ -432,11 +433,33 @@ it.effect(
           }),
         },
       });
-      const polled = yield* Effect.provide(client().boardLive(TASK_ID, true), api.layer);
+      const polled = yield* Effect.provide(
+        client().boardLive(TASK_ID, "user-dana", true),
+        api.layer,
+      );
       assert.ok(polled.ok);
       assert.deepEqual(polled.answer.queries.board.rows, [row]);
       assert.equal(api.requests()[0]?.searchParams.get(LIVE_POLL.QUERY), LIVE_POLL.ON);
-      yield* Effect.provide(client().boardLive(TASK_ID), api.layer);
+      yield* Effect.provide(client().boardLive(TASK_ID, "user-dana"), api.layer);
       assert.equal(api.requests()[1]?.searchParams.get(LIVE_POLL.QUERY), null, "a seal asks none");
     }),
 );
+
+it.effect(
+  "a board's live read names the account it is for, and a session the service refuses for it is signed out",
+  () =>
+    Effect.gen(function* () {
+      const refusing = fakeCloudApi({
+        "GET /api/live/task-board": {
+          answer: () => ({ error: HOSTED_API_ERROR.INVALID_TOKEN }),
+          status: HTTP_STATUS.UNAUTHORIZED,
+        },
+      });
+      const refused = yield* Effect.provide(
+        client().boardLive(TASK_ID, "user-dana", true, "digest"),
+        refusing.layer,
+      );
+      assert.deepEqual(refused, { ok: false, failure: LIVE_SIGNED_OUT.failure });
+      assert.equal(refusing.requests()[0]?.searchParams.get(LIVE_USER_QUERY), "user-dana");
+    }),
+);

packages/features/src/task-client.ts

@@ -34,7 +34,7 @@ import {
   hostedClient,
 } from "./hosted-link.js";
 import { imageBodyOf } from "./image-body.js";
-import type { TaskBoardLiveAnswer } from "./live-wire.js";
+import { LIVE_SIGNED_OUT, LIVE_USER_QUERY, type TaskBoardLiveAnswer } from "./live-wire.js";
 import type { PageLine, PageVersion } from "./page-wire.js";
 import { pictureTypeSchema } from "./picture-wire.js";
 import { PLAN_ATTACHMENT_REFUSAL } from "./plan-attachment-wire.js";
@@ -172,8 +172,14 @@ const NOT_FOUND_REFUSAL = {
   [HOSTED_API_ERROR.NOT_FOUND]: TASK_CALL_FAILURE.NOT_FOUND,
 } as const;
 
+/** The refusals a live read names: the task gone, or a session that is no longer the account's the read names (`live-wire.ts`). */
+const LIVE_REFUSALS = {
+  ...NOT_FOUND_REFUSAL,
+  [HOSTED_API_ERROR.INVALID_TOKEN]: LIVE_SIGNED_OUT.failure,
+} as const;
+
 /** A procedure's answer as a result: its refusal where it is one the caller names, unanswered for everything else. */
-function hostedResult<Answer, Failure extends TaskCallFailure>(
+function hostedResult<Answer, Failure extends string>(
   asked: Effect.Effect<Answer, HostedFailure, HttpClient.HttpClient>,
   named: Readonly<Partial<Record<HostedApiError, Failure>>>,
 ): Effect.Effect<
@@ -410,22 +416,24 @@ export class HostedTaskClient {
   /**
    * Where one task's whiteboard stands, sealed for following over Neon
    * Realtime, or with `poll` its rows inline at the poll address, for a
-   * reader polling while Realtime fails it: not found where the account
-   * reaches no such task, unanswered for anything else, which the caller
-   * answers by sealing it again.
+   * reader polling while Realtime fails it, for the account `userId`
+   * names: not found where the account reaches no such task, signed out
+   * where the session is no longer that account's, unanswered for anything
+   * else, which the caller answers by sealing it again.
    */
   boardLive(
     taskId: string,
+    userId: string,
     poll = false,
     after?: string,
   ): Effect.Effect<
-    TaskCallResult<TaskBoardLiveAnswer, TaskCallFailure>,
+    TaskCallResult<TaskBoardLiveAnswer, TaskCallFailure | typeof LIVE_SIGNED_OUT.failure>,
     never,
     HttpClient.HttpClient
   > {
     return hostedResult(
-      askLive(this.#hosted.live.taskBoard, { id: taskId }, poll, after),
-      NOT_FOUND_REFUSAL,
+      askLive(this.#hosted.live.taskBoard, { id: taskId, [LIVE_USER_QUERY]: userId }, poll, after),
+      LIVE_REFUSALS,
     );
   }
 

packages/host/src/compose-coding-agents.test.ts

@@ -289,7 +289,7 @@ function fakeService() {
         cursor = "2:1";
         return page ?? { ...SAID, cursor, holds: false };
       }),
-    agentTranscriptLive: (agentId, callId, poll) =>
+    agentTranscriptLive: (agentId, _userId, callId, poll) =>
       Effect.sync(() => {
         asked.push(
           `${poll === true ? "transcript-poll" : "transcript-live"}:${agentId}${callId === undefined ? "" : `:${callId}`}`,
@@ -657,6 +657,7 @@ function subject(options: { signedIn?: boolean; sendsNetwork?: boolean } = {}) {
         now: () => NOW,
       },
       account: { capabilitiesActive: () => options.signedIn ?? true },
+      accountId: () => "user-dana",
       client: service.client,
       openLive: realtime.open,
     }).pipe(Effect.provide(noNetwork));

packages/host/src/compose-coding-agents.ts

@@ -187,6 +187,8 @@ export interface CodingAgentsDependencies {
   account: Pick<AccountComposer, "capabilitiesActive">;
   client: CodingAgentClient;
   /** Realtime's client at an address, which a transcript is followed over; Realtime's own outside a test. */
+  /** The account signed in now, by its id, which every live read names; none where nobody is. */
+  accountId: () => string | undefined;
   openLive?: (url: string) => LiveClient | PromiseLike<LiveClient>;
 }
 
@@ -204,8 +206,11 @@ export function composeCodingAgents(
     const http = yield* HttpClient.HttpClient;
     const asked = askedOf(dependencies, http);
     const transcripts = yield* transcriptFollows({
-      seal: (key, poll, after) =>
-        asked(dependencies.client.agentTranscriptLive(key.agentId, key.callId, poll, after)),
+      seal: (key, poll, userId, after) =>
+        asked(
+          dependencies.client.agentTranscriptLive(key.agentId, userId, key.callId, poll, after),
+        ),
+      accountId: dependencies.accountId,
       openLive: dependencies.openLive ?? openRealtimeClient,
       now: dependencies.kernel.now,
     });

packages/host/src/compose-host.ts

@@ -1,3 +1,4 @@
+import { ACCOUNT_STATUS } from "@luke/credentials/snapshot";
 import {
   HostedAutomationClient,
   HostedCodingAgentClient,
@@ -157,9 +158,15 @@ export function hostAssemblyLayer<const Chosen extends HostPlatformConcern>(
       serviceBaseUrl: kernel.hostedServiceBaseUrl,
       credential: account.credential,
     });
+    /** The account signed in now, by its id, which every live read names. */
+    const accountId = () => {
+      const signedIn = account.session.snapshot();
+      return signedIn.status === ACCOUNT_STATUS.SIGNED_IN ? signedIn.id : undefined;
+    };
     const tasks = yield* composeTasks({
       kernel,
       account,
+      accountId,
       endTaskCall: (taskId) => live.service.endTaskCall(taskId),
       client: Object.assign(
         new HostedTaskClient({
@@ -180,6 +187,7 @@ export function hostAssemblyLayer<const Chosen extends HostPlatformConcern>(
     const codingAgents = yield* composeCodingAgents({
       kernel,
       account,
+      accountId,
       client: codingAgentClient,
     });
     const automations = yield* composeAutomations({

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

@@ -1067,6 +1067,7 @@ function subject(
       client: service,
       conversations,
       ...(options.openLive === undefined ? undefined : { openLive: options.openLive }),
+      accountId: () => "user-dana",
       endTaskCall: (taskId) =>
         Effect.gen(function* () {
           const about = standing.about?.taskId;

packages/host/src/compose-tasks.ts

@@ -24,6 +24,7 @@ import {
   type CodingAgentMessagesAnswer,
   type CodingAgentStatus,
 } from "@luke/features/coding-agent-wire";
+import { LIVE_SIGNED_OUT } from "@luke/features/live-wire";
 import { SHOW_PAGE_TOOL_NAME } from "@luke/features/page-wire";
 import { PLAN_ACTOR, type Task, UNTITLED_TASK_NAME } from "@luke/features/task-wire";
 import {
@@ -478,6 +479,8 @@ export interface TasksDependencies {
   account: Pick<AccountComposer, "capabilitiesActive">;
   client: TasksClient;
   conversations: TasksConversationClient;
+  /** The account signed in now, by its id, which every live read names; none where nobody is. */
+  accountId: () => string | undefined;
   /** Realtime's client at an address, which the open task's board is followed over; Realtime's own outside a test. */
   openLive?: (url: string) => LiveClient | PromiseLike<LiveClient>;
   /**
@@ -1145,10 +1148,14 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct
    */
   function followBoard(taskId: string) {
     const following = () => gate() && view.activeTaskId === taskId;
+    // The account the follow starts for, which every seal names, so one under another's session is refused as signed out.
+    const userId = dependencies.accountId();
     const seal = (poll = false, after?: string) =>
-      Effect.map(served(client.boardLive(taskId, poll, after)), (sealed) =>
-        sealed.ok ? sealed.answer : { failure: sealed.failure },
-      );
+      userId === undefined
+        ? Effect.succeed(LIVE_SIGNED_OUT)
+        : Effect.map(served(client.boardLive(taskId, userId, poll, after)), (sealed) =>
+            sealed.ok ? sealed.answer : { failure: sealed.failure },
+          );
     // A task the account no longer reaches, or one no longer open, stops the follow rather than renewing or polling it.
     const sealedFor = (poll: boolean, after?: string) =>
       following()
@@ -1186,7 +1193,10 @@ export const composeTasks = /* @__PURE__ */ Effect.fn("host/composeTasks")(funct
       Effect.retry({
         schedule: BOARD_BACKOFF,
         while: (why) =>
-          why !== TASK_CALL_FAILURE.NOT_FOUND && why !== BOARD_FOLLOW_ENDED && following(),
+          why !== TASK_CALL_FAILURE.NOT_FOUND &&
+          why !== LIVE_SIGNED_OUT.failure &&
+          why !== BOARD_FOLLOW_ENDED &&
+          following(),
       }),
       Effect.option,
     );

packages/host/src/live-follow.test.ts

@@ -1,6 +1,6 @@
 import assert from "node:assert/strict";
 import { it } from "@effect/vitest";
-import { LIVE_POLL_URL, type SealedLiveQueryWire } from "@luke/features/live-wire";
+import { LIVE_POLL_URL, LIVE_SIGNED_OUT, type SealedLiveQueryWire } from "@luke/features/live-wire";
 import { Clock, Duration, Effect, Exit, Fiber, Stream } from "effect";
 import { TestClock } from "effect/testing";
 import {
@@ -326,3 +326,18 @@ it.effect("a poll that fails is asked again a second later, then longer, up to f
     yield* Fiber.interrupt(following);
   }),
 );
+
+it.effect("a read whose session the service refuses as signed out ends, and is asked no more", () =>
+  Effect.gen(function* () {
+    const exit = yield* liveRows<Part>(
+      {
+        openLive: () => fakeRealtime().client,
+        reseal: () => Effect.succeed(LIVE_SIGNED_OUT),
+        poll: () => Effect.succeed(LIVE_SIGNED_OUT),
+        now: () => 0,
+      },
+      { ...read(0, 0), url: LIVE_POLL_URL },
+    ).pipe(Stream.runDrain, Effect.exit);
+    assert.ok(Exit.isFailure(exit));
+  }),
+);