diff --git a/packages/features/src/coding-agent-client.ts b/packages/features/src/coding-agent-client.ts index e2002baef6..0c95424b1f 100644 --- a/packages/features/src/coding-agent-client.ts +++ b/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 { - 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)); } /** diff --git a/packages/features/src/contract/live.ts b/packages/features/src/contract/live.ts index 2c1c2f5be0..8bce22a796 100644 --- a/packages/features/src/contract/live.ts +++ b/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,10 +33,14 @@ 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. */ @@ -41,6 +48,7 @@ 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 = { diff --git a/packages/features/src/live-wire.ts b/packages/features/src/live-wire.ts index 0ceacd4db9..c9bfb8a41d 100644 --- a/packages/features/src/live-wire.ts +++ b/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 diff --git a/packages/features/src/task-client.test.ts b/packages/features/src/task-client.test.ts index 0f867254ba..2f1ce01811 100644 --- a/packages/features/src/task-client.test.ts +++ b/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"); + }), +); diff --git a/packages/features/src/task-client.ts b/packages/features/src/task-client.ts index 62e5d8cc75..2eee91f73a 100644 --- a/packages/features/src/task-client.ts +++ b/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( +function hostedResult( asked: Effect.Effect, named: Readonly>>, ): 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, + TaskCallResult, 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, ); } diff --git a/packages/host/src/compose-coding-agents.test.ts b/packages/host/src/compose-coding-agents.test.ts index 1ee4a6cafe..d82a18bbe2 100644 --- a/packages/host/src/compose-coding-agents.test.ts +++ b/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)); diff --git a/packages/host/src/compose-coding-agents.ts b/packages/host/src/compose-coding-agents.ts index 144bff15ad..bd8b74ec6c 100644 --- a/packages/host/src/compose-coding-agents.ts +++ b/packages/host/src/compose-coding-agents.ts @@ -187,6 +187,8 @@ export interface CodingAgentsDependencies { account: Pick; 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; } @@ -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, }); diff --git a/packages/host/src/compose-host.ts b/packages/host/src/compose-host.ts index cf9475a19a..1c986a7667 100644 --- a/packages/host/src/compose-host.ts +++ b/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( 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 codingAgents = yield* composeCodingAgents({ kernel, account, + accountId, client: codingAgentClient, }); const automations = yield* composeAutomations({ diff --git a/packages/host/src/compose-tasks.test.ts b/packages/host/src/compose-tasks.test.ts index 974fe41c67..57be7fd0bc 100644 --- a/packages/host/src/compose-tasks.test.ts +++ b/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; diff --git a/packages/host/src/compose-tasks.ts b/packages/host/src/compose-tasks.ts index e7256f4167..70f90fb99c 100644 --- a/packages/host/src/compose-tasks.ts +++ b/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; 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; /** @@ -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, ); diff --git a/packages/host/src/live-follow.test.ts b/packages/host/src/live-follow.test.ts index 556e3d2424..e5be7e67d6 100644 --- a/packages/host/src/live-follow.test.ts +++ b/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( + { + 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)); + }), +); diff --git a/packages/host/src/live-follow.ts b/packages/host/src/live-follow.ts index 3261830796..08a47ae1d0 100644 --- a/packages/host/src/live-follow.ts +++ b/packages/host/src/live-follow.ts @@ -1,4 +1,9 @@ -import { LIVE_POLL_URL, liveDigestOf, type SealedLiveQueryWire } from "@luke/features/live-wire"; +import { + LIVE_POLL_URL, + LIVE_SIGNED_OUT, + liveDigestOf, + type SealedLiveQueryWire, +} from "@luke/features/live-wire"; import { Cause, Data, Duration, Effect, Fiber, Queue, Schedule, Stream } from "effect"; /** @@ -34,7 +39,16 @@ const LIVE_POLL_FLOOR = Duration.millis(500); const POLL_BACKOFF = Schedule.min([Schedule.exponential("1 second"), Schedule.spaced("5 seconds")]); /** How often Realtime is subscribed to again while a read is polled. */ const RESUBSCRIBE_EVERY = Duration.seconds(15); -/** Why a read stopped for good: it names nothing the account reaches any more. */ +/** + * Whether a failed seal ends the read for good: it names nothing the account + * reaches any more, or the service refused the session as signed out, which + * Neon's guide says is never asked again ("When the session changes"). + */ +function ended(failed: { readonly failure: string; readonly gone?: boolean }): boolean { + return failed.gone === true || failed.failure === LIVE_SIGNED_OUT.failure; +} + +/** Why a read stopped for good: it names nothing the account reaches any more, or its session signed out. */ const GONE = "the read is gone"; /** What a go at Realtime leaves to do: subscribe again later, or stay polled, the service answering polled. */ const AGAIN = "again"; @@ -172,7 +186,7 @@ function realtimeRows( yield* Effect.sleep(Duration.millis(Math.max(0, wait))); let next = yield* dependencies.reseal(); while ("failure" in next) { - if (next.gone === true) { + if (ended(next)) { return yield* Effect.fail(new LiveBroken({ reason: GONE })); } yield* Effect.sleep(RENEW_RETRY); @@ -243,7 +257,7 @@ export function liveRows( const pollOnce = Effect.gen(function* () { const read = yield* dependencies.poll(seen); if ("failure" in read) { - if (read.gone === true) { + if (ended(read)) { gone(); return yield* Effect.never; } @@ -286,7 +300,7 @@ export function liveRows( * nor where the service answered it polled, Realtime being off. */ const attempt = Effect.gen(function* () { - if ("failure" in sealed) return sealed.gone === true ? GONE : AGAIN; + if ("failure" in sealed) return ended(sealed) ? GONE : AGAIN; if (sealed.url === LIVE_POLL_URL) { const rows = rowsOf(parts, sealed); if (rows !== undefined) emit(rows); diff --git a/packages/host/src/transcript-follow.ts b/packages/host/src/transcript-follow.ts index 35669ef9f4..776d8abafe 100644 --- a/packages/host/src/transcript-follow.ts +++ b/packages/host/src/transcript-follow.ts @@ -6,6 +6,7 @@ import { CODING_AGENT_CURSOR_START } from "@luke/features/coding-agent-wire"; import { type AgentTranscriptLiveAnswer, agentTranscriptLiveRowsSchema, + LIVE_SIGNED_OUT, type TranscriptNowAnswer, } from "@luke/features/live-wire"; import { @@ -73,8 +74,11 @@ export interface TranscriptFollowDependencies { readonly seal: ( key: TranscriptKey, poll: boolean, + userId: string, after?: string, ) => Effect.Effect; + /** The account signed in now, which a follow is sealed for from its first seal on; none where nobody is. */ + readonly accountId: () => string | undefined; readonly openLive: (url: string) => LiveClient | PromiseLike; readonly now: () => number; } @@ -92,14 +96,18 @@ function digestOf(rows: typeof agentTranscriptLiveRowsSchema.Type): string { /** The transcript sealed, asked again every `RESEAL` until the service seals it; none where the account does not reach it. */ function sealedOf( - dependencies: TranscriptFollowDependencies, - key: TranscriptKey, + seal: (poll: boolean, after?: string) => Effect.Effect, ): Effect.Effect> { return Effect.gen(function* () { while (true) { - const answer = yield* dependencies.seal(key, false); + const answer = yield* seal(false); if (!("failure" in answer)) return Option.some(answer); - if (answer.failure === CODING_AGENT_CALL_FAILURE.NOT_FOUND) return Option.none(); + if ( + answer.failure === CODING_AGENT_CALL_FAILURE.NOT_FOUND || + answer.failure === LIVE_SIGNED_OUT.failure + ) { + return Option.none(); + } yield* Effect.sleep(RESEAL); } }); @@ -114,7 +122,13 @@ export function transcriptFollows( idleTimeToLive: IDLE, lookup: (key: TranscriptKey) => Effect.gen(function* () { - const first = yield* sealedOf(dependencies, key); + // The account the follow starts for, which every renewal and poll names, so one under another's session is refused as signed out. + const userId = dependencies.accountId(); + const seal = (poll: boolean, after?: string) => + userId === undefined + ? Effect.succeed(LIVE_SIGNED_OUT) + : dependencies.seal(key, poll, userId, after); + const first = yield* sealedOf(seal); if (Option.isNone(first)) return { kind: "refused" } as const; const state = yield* SubscriptionRef.make({ kind: "waiting" }); /** One seal followed until its subscription stops, which marks the state broken. */ @@ -123,8 +137,8 @@ export function transcriptFollows( liveRows( { openLive: dependencies.openLive, - reseal: () => dependencies.seal(key, false), - poll: (after) => dependencies.seal(key, true, after), + reseal: () => seal(false), + poll: (after) => seal(true, after), now: dependencies.now, }, sealed, @@ -142,7 +156,7 @@ export function transcriptFollows( while (true) { yield* follow(sealed); yield* Effect.sleep(RESEAL); - const again = yield* sealedOf(dependencies, key); + const again = yield* sealedOf(seal); if (Option.isNone(again)) return; sealed = again.value; } diff --git a/packages/service/src/live/live-procedures.test.ts b/packages/service/src/live/live-procedures.test.ts index 0e524a9b16..38ce9967b4 100644 --- a/packages/service/src/live/live-procedures.test.ts +++ b/packages/service/src/live/live-procedures.test.ts @@ -14,6 +14,7 @@ import { agentTranscriptLiveAnswerSchema, LIVE_POLL, LIVE_POLL_URL, + LIVE_USER_QUERY, liveDigestOf, taskBoardLiveAnswerSchema, taskBoardLiveRowsSchema, @@ -57,6 +58,13 @@ const SCRIPTED: LiveSealer = { }), }; +/** A live read's address, naming the account it is for where one is named (`LIVE_USER_QUERY`). */ +function readOf(path: string, forUser: string | undefined): URL { + const url = new URL(path, ORIGIN); + if (forUser !== undefined) url.searchParams.set(LIVE_USER_QUERY, forUser); + return url; +} + /** The procedures answering one request as `bearer`, over the deployment's Realtime where it has one. */ const ask = (options: { readonly bearer?: string; @@ -75,7 +83,7 @@ const ask = (options: { const response = yield* hostedAnswer( { live: procedures }, Layer.succeed(LiveSealing, options.sealing), - new Request(new URL(LIVE_INBOX, ORIGIN), { + new Request(readOf(LIVE_INBOX, options.bearer), { method: options.method ?? "GET", headers: options.bearer === undefined ? {} : { authorization: `Bearer ${options.bearer}` }, }), @@ -147,6 +155,8 @@ const openUser = Effect.gen(function* () { /** The procedures answering a task board's live read as `bearer`, over the store and the deployment's Realtime. */ const askBoard = (options: { readonly bearer: string; + /** The account the read names; the bearer's own where left out. */ + readonly forUser?: string; readonly taskId?: string; readonly poll?: boolean; /** The digest the reader last saw, which holds the poll until the rows move past it. */ @@ -164,7 +174,10 @@ const askBoard = (options: { const procedures = liveProcedures({ resolveUserId: () => Effect.succeed(Option.some(options.bearer)), }); - const url = new URL(options.taskId === undefined ? LIVE_INBOX : "/api/live/task-board", ORIGIN); + const url = readOf( + options.taskId === undefined ? LIVE_INBOX : "/api/live/task-board", + options.forUser ?? options.bearer, + ); if (options.taskId !== undefined) url.searchParams.set("id", options.taskId); if (options.poll === true) url.searchParams.set(LIVE_POLL.QUERY, LIVE_POLL.ON); if (options.after !== undefined) url.searchParams.set(LIVE_POLL.AFTER, options.after); @@ -283,7 +296,7 @@ const askTranscript = (options: { const procedures = liveProcedures({ resolveUserId: () => Effect.succeed(Option.some(options.bearer)), }); - const url = new URL("/api/live/agent-transcript", ORIGIN); + const url = readOf("/api/live/agent-transcript", options.bearer); url.searchParams.set("id", options.agentId); if (options.call !== undefined) url.searchParams.set("call", options.call); const response = yield* hostedAnswer( @@ -434,4 +447,35 @@ it.layer(getTestDB())("an agent transcript's live read", (it) => { assert.equal(one.digest, other.digest); }), ); + + it.effect( + "a read naming another account than the caller's, or none, is answered as signed out, as no session is", + () => + Effect.gen(function* () { + const dana = yield* openUser; + const started = yield* startTask(dana, { name: "Teammate invitations" }); + const signedOut = { + status: HOSTED_HTTP_STATUS.UNAUTHORIZED, + body: { error: HOSTED_API_ERROR.INVALID_TOKEN }, + }; + for (const forUser of ["user-eli", ""]) { + for (const poll of [false, true]) { + assert.deepEqual( + yield* askBoard({ + bearer: dana, + forUser, + taskId: started.id, + poll, + sealing: Option.none(), + }), + signedOut, + ); + } + assert.deepEqual( + yield* askBoard({ bearer: dana, forUser, sealing: Option.none() }), + signedOut, + ); + } + }), + ); }); diff --git a/packages/service/src/live/live-procedures.ts b/packages/service/src/live/live-procedures.ts index 47fca5ccb6..d9d11d2c0c 100644 --- a/packages/service/src/live/live-procedures.ts +++ b/packages/service/src/live/live-procedures.ts @@ -36,7 +36,12 @@ import { boardLiveStatement } from "@luke/database/models/board-store"; import { childConversationId } from "@luke/database/models/ids"; import { readTask } from "@luke/database/models/task-store"; import { codingAgentCallIdSchema } from "@luke/features/coding-agent-wire"; -import { LIVE_POLL, liveDigestOf, type SealedLiveQueryWire } from "@luke/features/live-wire"; +import { + LIVE_POLL, + LIVE_USER_QUERY, + liveDigestOf, + type SealedLiveQueryWire, +} from "@luke/features/live-wire"; import { readEither } from "@luke/wire/effect"; import { Duration, Effect, Option, Result } from "effect"; import { SqlClient } from "effect/sql"; @@ -153,11 +158,21 @@ const callIdOf = /* @__PURE__ */ Effect.fnUntraced(function* ( return read.success; }); -/** The live reads, each to a signed-in caller. */ +/** + * The caller, where they are the account the read names; refused as signed + * out otherwise, with the same 401 as no session, as Neon's guide says + * ("When the session changes"), so a seal, a renewal or a held poll from a + * page left open under another account's session reads nothing of theirs. + */ +const sameUser = (input: { readonly [LIVE_USER_QUERY]?: string }, userId: string) => + input[LIVE_USER_QUERY] === userId ? Effect.void : Effect.fail(HOSTED_REFUSAL.INVALID_TOKEN); + +/** The live reads, each to a signed-in caller who is the account it names. */ export function liveProcedures(seams: LiveProcedureSeams) { const signedIn = hosted.live.use(accountMiddleware(seams.resolveUserId)); return { inbox: signedIn.inbox.effect(function* ({ input, context }) { + yield* sameUser(input, context.userId); const statements = Option.zipWith( accountInboxLiveStatements(context.userId), ownFailedRunsLiveStatement(context.userId), @@ -167,6 +182,7 @@ export function liveProcedures(seams: LiveProcedureSeams) { return yield* sealedParts(statements.value, input); }), taskBoard: signedIn.taskBoard.effect(function* ({ input, context }) { + yield* sameUser(input, context.userId); const taskId = yield* rowIdOf(input.id); const task = yield* hostedStoreOrUnavailable(readTask(context.userId, taskId)); if (Option.isNone(task)) return yield* Effect.fail(HOSTED_REFUSAL.NOT_FOUND); @@ -175,6 +191,7 @@ export function liveProcedures(seams: LiveProcedureSeams) { return yield* sealedParts({ board: statement.value }, input); }), agentTranscript: signedIn.agentTranscript.effect(function* ({ input, context }) { + yield* sameUser(input, context.userId); const callId = input.call === undefined ? undefined : yield* callIdOf(input.call); const { target } = yield* reachedAgentById(context.userId, input.id); const read =