feat(host): a window follows a plan's or an agent's inbox as a stream of pushed changes

packages/features/fixtures/json-schema/coding-agent-view-codingAgentInboxScopeSchema.json

@@ -0,0 +1,21 @@
+{
+  "type": "object",
+  "properties": {
+    "kind": {
+      "type": "string",
+      "enum": [
+        "plan",
+        "agent"
+      ]
+    },
+    "id": {
+      "type": "string",
+      "minLength": 1
+    }
+  },
+  "required": [
+    "kind",
+    "id"
+  ],
+  "additionalProperties": false
+}

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

@@ -30,6 +30,7 @@ import {
   inboxAnswerRequestSchema,
   inboxAnswerSchema,
   inboxCloseRequestSchema,
+  inboxItemSchema,
   inboxMessagedSchema,
   inboxMessageRequestSchema,
   inboxReadAllSchema,
@@ -561,6 +562,25 @@ export const INBOX_SCOPE_KIND = {
   AGENT: "agent",
 } as const;
 
+/** One plan's inbox, or one agent's, as the window follows it: which kind, and the plan's or agent's id. */
+export const codingAgentInboxScopeSchema = EffectSchema.Struct({
+  kind: EffectSchema.Literals(Object.values(INBOX_SCOPE_KIND)),
+  id: EffectSchema.NonEmptyString,
+});
+
+export type CodingAgentInboxScope = typeof codingAgentInboxScopeSchema.Type;
+
+/** A followed inbox as it now stands, as the window hears it pushed: its items whole, or why there are none. */
+export const codingAgentInboxChangedSchema = EffectSchema.Struct({
+  scope: codingAgentInboxScopeSchema,
+  inbox: EffectSchema.Union([
+    EffectSchema.Struct({ items: EffectSchema.Array(inboxItemSchema) }),
+    failed,
+  ]),
+});
+
+export type CodingAgentInboxChanged = typeof codingAgentInboxChangedSchema.Type;
+
 /** The task's inbox, as the window hears it: every item its agents posted and the cursor to read on from, or why there is none. */
 export const codingAgentInboxAnswerViewSchema = EffectSchema.Union([inboxAnswerSchema, failed]);
 

packages/gateway/src/protocol.ts

@@ -290,6 +290,14 @@ export const GATEWAY_METHOD = {
   /** One task's inbox past a cursor: every item its coding agents posted for the developer, held open by the service while nothing changed. */
   CODING_AGENTS_INBOX: "codingAgents.inbox",
   CODING_AGENTS_AGENT_INBOX: "codingAgents.agentInbox",
+  /**
+   * One plan's inbox, or one agent's, followed from now until it is let go
+   * or the account signs out, over Neon Realtime. What it holds is heard as
+   * `INBOX_CHANGED` events, never answered; following again starts over.
+   */
+  CODING_AGENTS_INBOX_FOLLOW: "codingAgents.inboxFollow",
+  /** One plan's or agent's inbox let go: no more `INBOX_CHANGED` events of it until it is followed again. */
+  CODING_AGENTS_INBOX_UNFOLLOW: "codingAgents.inboxUnfollow",
   /** One item of a task's inbox answered against the revision the panel drew, under the panel's own key; the item as it then stands and whether the answer was taken, or why there is no answer. */
   CODING_AGENTS_INBOX_ANSWER: "codingAgents.inboxAnswer",
   /** One open item of a task's inbox closed without an answer against the revision the panel drew, under the panel's own key; answered as an answer is. */
@@ -689,6 +697,8 @@ export const GATEWAY_EVENT = {
   WORKSPACE_PEOPLE_CHANGED: "workspacePeople.changed",
   /** A followed workspace's pending invitations as they now stand, each time they move while followed. */
   WORKSPACE_INVITATIONS_CHANGED: "workspaceInvitations.changed",
+  /** One followed plan's or agent's inbox, its items whole or why there are none, each time it changes while it is followed. */
+  INBOX_CHANGED: "inbox.changed",
 } as const;
 
 export type GatewayEventKind = (typeof GATEWAY_EVENT)[keyof typeof GATEWAY_EVENT];

packages/host/src/act-port/agent-inbox-procedures.test.ts

@@ -18,6 +18,9 @@ import { MEMBER_KIND } from "@luke/features/chat-wire";
 import {
   CODING_AGENT_CALL_FAILURE,
   type CodingAgentChatChanged,
+  type CodingAgentInboxChanged,
+  type CodingAgentInboxScope,
+  INBOX_SCOPE_KIND,
 } from "@luke/features/coding-agent-view";
 import {
   INBOX_IMAGE_TYPE,
@@ -625,3 +628,63 @@ it("one workspace's unfollow on its way holds no other workspace's follow", asyn
   await closed;
   await elsewhere.return(undefined);
 });
+
+it("a panel's inbox stream follows the inbox at the host, hears each change told of it alone, and lets it go once its last stream closes", async () => {
+  const asked: string[] = [];
+  const listeners = new Set<(changed: CodingAgentInboxChanged) => void>();
+  const follows: Pick<
+    HostAsk,
+    "codingAgentInboxFollow" | "codingAgentInboxUnfollow" | "onInboxChanged"
+  > = {
+    codingAgentInboxFollow: () => Effect.sync(() => void asked.push("follow")),
+    codingAgentInboxUnfollow: () => Effect.sync(() => void asked.push("unfollow")),
+    onInboxChanged: (listener) => {
+      listeners.add(listener);
+      return () => listeners.delete(listener);
+    },
+  };
+  // SAFETY: the stream reaches the follow asks alone.
+  const host = follows as HostAsk;
+  const port = testActPort(agentInboxProcedures({ host }));
+  ports.push(port);
+  const client = port.as(PANEL);
+  const scope: CodingAgentInboxScope = { kind: INBOX_SCOPE_KIND.PLAN, id: TASK_ID };
+
+  const one = await client[AGENT_INBOX_ACT_KIND.CODING_AGENTS_INBOX_FOLLOW](scope);
+  const two = await client[AGENT_INBOX_ACT_KIND.CODING_AGENTS_INBOX_FOLLOW](scope);
+  const first = one.next();
+  const second = two.next();
+  await until(() => asked.length === 1);
+  for (const listener of [...listeners]) {
+    listener({ scope: { kind: INBOX_SCOPE_KIND.AGENT, id: AGENT_ID }, inbox: UNANSWERED });
+    listener({ scope, inbox: { items: [ITEM] } });
+  }
+  assert.deepEqual((await first).value, { items: [ITEM] });
+  assert.deepEqual((await second).value, { items: [ITEM] });
+
+  await one.return(undefined);
+  await until(() => listeners.size === 1);
+  assert.deepEqual(asked, ["follow"], "the host follows the inbox once for both streams");
+  await two.return(undefined);
+  await until(() => asked.length === 2);
+  assert.deepEqual(asked, ["follow", "unfollow"]);
+});
+
+it("the voice window is refused an inbox stream, and the host follows nothing", async () => {
+  const asked: string[] = [];
+  // SAFETY: the stream reaches the follow asks alone.
+  const host = {
+    codingAgentInboxFollow: () => Effect.sync(() => void asked.push("follow")),
+    codingAgentInboxUnfollow: () => Effect.sync(() => void asked.push("unfollow")),
+    onInboxChanged: () => () => undefined,
+  } as unknown as HostAsk;
+  const port = testActPort(agentInboxProcedures({ host }));
+  ports.push(port);
+  await assert.rejects(
+    port.as(VOICE)[AGENT_INBOX_ACT_KIND.CODING_AGENTS_INBOX_FOLLOW]({
+      kind: INBOX_SCOPE_KIND.PLAN,
+      id: TASK_ID,
+    }),
+  );
+  assert.deepEqual(asked, []);
+});

packages/host/src/act-port/agent-inbox-procedures.ts

@@ -26,19 +26,22 @@ import type {
   CodingAgentInboxAnswerView,
   CodingAgentInboxAttachmentParams,
   CodingAgentInboxAttachmentView,
+  CodingAgentInboxChanged,
   CodingAgentInboxCloseParams,
   CodingAgentInboxMessagedView,
   CodingAgentInboxMessageParams,
   CodingAgentInboxParams,
   CodingAgentInboxReadAllView,
   CodingAgentInboxReadStateParams,
   CodingAgentInboxReadStateView,
+  CodingAgentInboxScope,
   CodingAgentPictureParams,
 } from "@luke/features/coding-agent-view";
 import type { InboxReadAllRequest } from "@luke/features/inbox-wire";
-import { Effect, Queue, Semaphore, Stream } from "effect";
+import { Effect, MutableHashMap, Option, Queue, Semaphore, Stream } from "effect";
 import { AGENT_INBOX_ACT_KIND } from "../contract/index.js";
 import { type ActSender, actPort } from "./base.js";
+import { followedStream, type OpenFollows } from "./followed-stream.js";
 
 /** What the Inbox and chat procedures reach: the host's asks of the service. */
 export interface AgentInboxProceduresDependencies {
@@ -47,6 +50,10 @@ export interface AgentInboxProceduresDependencies {
     codingAgentAgentInbox(
       params: CodingAgentAgentInboxParams,
     ): Effect.Effect<CodingAgentInboxAnswerView>;
+    /** A task's or agent's inbox followed at the host, heard through `onInboxChanged`; following again starts over. */
+    codingAgentInboxFollow(scope: CodingAgentInboxScope): Effect.Effect<void>;
+    codingAgentInboxUnfollow(scope: CodingAgentInboxScope): Effect.Effect<void>;
+    onInboxChanged(listener: (changed: CodingAgentInboxChanged) => void): () => void;
     codingAgentInboxAnswer(
       params: CodingAgentInboxAnswerParams,
     ): Effect.Effect<CodingAgentInboxAnsweredView>;
@@ -165,6 +172,8 @@ export function agentInboxProcedures(dependencies: AgentInboxProceduresDependenc
   const { host } = dependencies;
   /** How many streams stand open of each workspace's chat, so the host lets one go only with the last. */
   const openChats = new Map<string, number>();
+  /** Every inbox a window holds a stream open of, so the host lets one go only with the last. */
+  const openInboxes: OpenFollows<CodingAgentInboxChanged["inbox"]> = MutableHashMap.empty();
   /**
    * One turn at a time for a chat stream's follow or unfollow, per workspace,
    * so the host hears a workspace's in the order its streams opened and
@@ -192,6 +201,26 @@ export function agentInboxProcedures(dependencies: AgentInboxProceduresDependenc
       yield* refuseUnlessPanel(context.actor, () => errors.REFUSED());
       return yield* host.codingAgentAgentInbox(input);
     }),
+    [AGENT_INBOX_ACT_KIND.CODING_AGENTS_INBOX_FOLLOW]: actPort[
+      AGENT_INBOX_ACT_KIND.CODING_AGENTS_INBOX_FOLLOW
+    ].effect(function* ({ input, context, errors }) {
+      yield* refuseUnlessPanel(context.actor, () => errors.REFUSED());
+      const changes = followedStream(
+        {
+          keyOf: (scope: CodingAgentInboxScope) => `${scope.kind}:${scope.id}`,
+          follow: host.codingAgentInboxFollow,
+          unfollow: host.codingAgentInboxUnfollow,
+          onChanged: host.onInboxChanged,
+          valueOf: (changed: CodingAgentInboxChanged, scope: CodingAgentInboxScope) =>
+            changed.scope.kind === scope.kind && changed.scope.id === scope.id
+              ? Option.some(changed.inbox)
+              : Option.none(),
+        },
+        openInboxes,
+        input,
+      );
+      return streamed(yield* Stream.toAsyncIterableEffect(changes));
+    }),
     [AGENT_INBOX_ACT_KIND.CODING_AGENTS_INBOX_ANSWER]: actPort[
       AGENT_INBOX_ACT_KIND.CODING_AGENTS_INBOX_ANSWER
     ].effect(function* ({ input, context, errors }) {

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

@@ -725,6 +725,11 @@ function subject(
       pageNow: service.pageNow,
       sealScope: service.sealScope,
       realtime,
+      /** Every followed plan's or agent's inbox the host told, in order. */
+      told: () =>
+        events.flatMap((event) =>
+          event.kind === GATEWAY_EVENT.INBOX_CHANGED ? [event.payload] : [],
+        ),
       reset: composer.reset,
     };
   });
@@ -1720,3 +1725,130 @@ it.effect(
       ]);
     }),
 );
+
+const PLAN_SCOPE = { kind: "plan", id: TASK_ID } as const;
+
+/** Steps taken until the host has told `times` inboxes, since a row's digest is the platform's own promise. */
+const toldTimes = (told: () => readonly unknown[], times: number) =>
+  Effect.repeat(Effect.yieldNow, { until: () => told().length >= times, times: 10_000 });
+const AGENT_SCOPE = { kind: "agent", id: AGENT_ID } as const;
+
+it.effect(
+  "a followed plan's inbox is told from its Realtime subscription once its rows land, again each time they move and never when a delivery moved nothing; letting it go ends the telling",
+  () =>
+    Effect.gen(function* () {
+      const { call, asked, sealScope, realtime, told } = yield* subject();
+      sealScope(SEALED_SCOPE);
+      assert.deepEqual(yield* call(GATEWAY_METHOD.CODING_AGENTS_INBOX_FOLLOW, PLAN_SCOPE), {});
+      yield* settled;
+      assert.deepEqual(realtime.opened, [SEALED_SCOPE.url]);
+      assert.deepEqual(told(), []);
+
+      realtime.deliver("messages", []);
+      realtime.deliver("files", []);
+      realtime.deliver("items", [ITEM_ROW]);
+      yield* settled;
+      realtime.deliver("items", [ITEM_ROW]);
+      yield* settled;
+      const closed = {
+        ...ITEM_ROW,
+        revision: 3,
+        status: INBOX_STATUS.CLOSED,
+        closedReason: "Done",
+      };
+      realtime.deliver("items", [closed]);
+      yield* settled;
+      assert.deepEqual(told(), [
+        { scope: PLAN_SCOPE, inbox: { items: [LIVE_ITEM] } },
+        {
+          scope: PLAN_SCOPE,
+          inbox: {
+            items: [
+              { ...LIVE_ITEM, revision: 3, status: INBOX_STATUS.CLOSED, closedReason: "Done" },
+            ],
+          },
+        },
+      ]);
+      assert.deepEqual(asked, [`plan-inbox-live:${TASK_ID}`]);
+      assert.ok(!JSON.stringify(told()).includes("capability"));
+
+      yield* call(GATEWAY_METHOD.CODING_AGENTS_INBOX_UNFOLLOW, PLAN_SCOPE);
+      yield* settled;
+      realtime.deliver("items", [ITEM_ROW]);
+      yield* settled;
+      assert.equal(told().length, 2);
+    }),
+);
+
+it.effect(
+  "a message landing after its item's row is told as soon as it lands, its item's thread whole",
+  () =>
+    Effect.gen(function* () {
+      const { call, sealScope, realtime, told } = yield* subject();
+      sealScope(SEALED_SCOPE);
+      yield* call(GATEWAY_METHOD.CODING_AGENTS_INBOX_FOLLOW, PLAN_SCOPE);
+      yield* settled;
+      realtime.deliver("messages", []);
+      realtime.deliver("files", []);
+      realtime.deliver("items", [ITEM_ROW]);
+      yield* toldTimes(told, 1);
+      const message = {
+        id: "6f1d2a10-7b0f-4e3b-9a55-5e3b2c1d4c7a",
+        itemId: ITEM.id,
+        author: INBOX_MESSAGE_AUTHOR.AGENT,
+        userId: null,
+        name: null,
+        text: "Also the staging branch?",
+        ask: null,
+        answers: null,
+        createdAt: ITEM.updatedAt,
+      };
+      realtime.deliver("messages", [message]);
+      yield* toldTimes(told, 2);
+      assert.equal(told().length, 2);
+      assert.ok(JSON.stringify(told()[1]).includes("Also the staging branch?"));
+    }),
+);
+
+it.effect(
+  "an agent's inbox the account does not reach is told not found, one the service seals nothing for is told unanswered, and behind a closed gate every follow is told unanswered",
+  () =>
+    Effect.gen(function* () {
+      const refused = yield* subject();
+      refused.sealScope({ failure: CODING_AGENT_CALL_FAILURE.NOT_FOUND });
+      yield* refused.call(GATEWAY_METHOD.CODING_AGENTS_INBOX_FOLLOW, AGENT_SCOPE);
+      yield* settled;
+      assert.deepEqual(refused.told(), [
+        { scope: AGENT_SCOPE, inbox: { failure: CODING_AGENT_CALL_FAILURE.NOT_FOUND } },
+      ]);
+
+      const unsealed = yield* subject();
+      yield* unsealed.call(GATEWAY_METHOD.CODING_AGENTS_INBOX_FOLLOW, AGENT_SCOPE);
+      yield* settled;
+      assert.deepEqual(unsealed.told(), [
+        { scope: AGENT_SCOPE, inbox: { failure: CODING_AGENT_CALL_FAILURE.UNANSWERED } },
+      ]);
+      assert.deepEqual(unsealed.realtime.opened, []);
+
+      const signedOut = yield* subject({ signedIn: false });
+      yield* signedOut.call(GATEWAY_METHOD.CODING_AGENTS_INBOX_FOLLOW, PLAN_SCOPE);
+      yield* settled;
+      assert.deepEqual(signedOut.told(), [
+        { scope: PLAN_SCOPE, inbox: { failure: CODING_AGENT_CALL_FAILURE.UNANSWERED } },
+      ]);
+      assert.deepEqual(signedOut.asked, []);
+    }),
+);
+
+it.effect("a sign-out lets every followed inbox go", () =>
+  Effect.gen(function* () {
+    const { call, sealScope, realtime, told, reset } = yield* subject();
+    sealScope(SEALED_SCOPE);
+    yield* call(GATEWAY_METHOD.CODING_AGENTS_INBOX_FOLLOW, PLAN_SCOPE);
+    yield* settled;
+    yield* reset;
+    yield* settled;
+    assert.deepEqual(realtime.closed, [SEALED_SCOPE.url]);
+    assert.deepEqual(told(), []);
+  }),
+);