hub-link-commands

1// hub-link-commands: the plugin that runs the Hub's commands in a Session's Claude Code, beside
2// hub-link, which reports (ADR 0014). They are two because a plugin's hooks don't see the work it
3// causes: an agent this plugin resumes with a message runs past its own hooks, but not past
4// hub-link's (spike #148).
5//
6// - The Hub: says hello and long-polls for commands (see HubLink.ts): prompts from the Prompt
7//   queue, prompts steered into the running Turn, stopping the Turn, messages to subagents and
8//   stopping one, and presence.
9// - hub-link: reads which prompt it delivered last, and who sent it, to report the Turn it starts
10//   and tell Claude of its files; and which it is steering, to report it joining the Turn.
11// - The person at the terminal, the Session's Creator: a band above the prompt shows the link and
12//   the Participants watching (ADR 0026).
13// - Claude: a tool that opens the files the Creator attaches to prompts (#276, ADR 0024), which the
14//   plugin API's prompts can't carry: it fetches each from the Hub, and answers with the image or
15//   PDF (#277).
16//
17// The rest of the plugin, with Effect and the protocol, is bundled from the mod package's src/
18// into hooks/hub-link-lib.js (`bun run build`).
19
20import { atom, read, update } from "claude-code"
21import type { EngineInterface, On } from "claude-code"
22import type { HubLinkBand } from "../types"
23import {
24  answerResult,
25  askedQuestions,
26  attachmentContext,
27  attachmentPart,
28  attachmentPartPath,
29  attachmentResult,
30  attachmentToolName,
31  attachmentToolSpec,
32  deliveredPrompt,
33  HubLink,
34  maxAttachmentParts,
35  steeredPrompt,
36  terminalOutcome,
37} from "./hub-link-lib.js"
38import type { LinkStatus, ModCommand, Participant, QuestionAnswer } from "./hub-link-lib.js"
39
40const band = atom(
41  { plugin: "hub-link-commands", key: "band" } as const,
42  {
43    status: "Unlinked",
44    detail: "",
45    participants: [],
46  } as HubLinkBand,
47)
48
49const delivered = { plugin: "hub-link-commands", key: "delivered" } as const
50const steering = { plugin: "hub-link-commands", key: "steering" } as const
51
52/** The module's own state; a reload starts it over, and fires `session.start` again. */
53const state = {
54  link: undefined as HubLink | undefined,
55  /** Where the Hub is, and the mod token, for the attachment tool's fetches. */
56  hub: undefined as { readonly url: string; readonly modToken: string } | undefined,
57
58  /** The Turn running on the main loop, which StopTurn stops. */
59  mainTurn: undefined as string | undefined,
60  /** The Turn StopTurn stopped last. */
61  stoppedTurn: undefined as string | undefined,
62  /**
63   * The AskUserQuestion calls waiting for an answer, by id (#193): each takes the Creator's
64   * answer, and says whether it came first.
65   */
66  asking: new Map<string, (answer: QuestionAnswer, sender: Participant) => boolean>(),
67}
68
69const report = (event: Record<string, unknown>) => state.link?.report(event)
70
71/**
72 * The prompts given Claude Code, by id, so one delivered again isn't run twice (#323): the Hub
73 * delivers the head of its queue again when it restarts, or when the TurnStarted that takes a
74 * prompt out of it was lost. Kept here for this process, and in the store, which is in the
75 * Session's own Claude configuration, for the next: the last `SubmittedKept` of them.
76 */
77const submitted = {
78  ids: new Set<string>(),
79  /** The store's list is read, changed and written back one change at a time. */
80  writing: Promise.resolve(),
81}
82const SubmittedKept = 200
83
84const storeSubmitted = (
85  $: EngineInterface,
86  change: (ids: ReadonlyArray<string>) => Array<string>,
87) => {
88  submitted.writing = submitted.writing
89    .then(async () => {
90      const kept = await $.store.get("submitted")
91      const ids = Array.isArray(kept) ? kept.filter((id) => typeof id === "string") : []
92      await $.store.set("submitted", change(ids).slice(-SubmittedKept))
93    })
94    .catch(() => undefined)
95  return submitted.writing
96}
97
98/** Whether `promptId` was given Claude Code already; if not, it is now said to be. */
99const submittedBefore = async ($: EngineInterface, promptId: string) => {
100  // Checked and taken before anything is awaited, so a prompt delivered twice at once runs once.
101  if (submitted.ids.has(promptId)) return true
102  submitted.ids.add(promptId)
103  const kept = await $.store.get("submitted").catch(() => undefined)
104  if (Array.isArray(kept) && kept.includes(promptId)) return true
105  await storeSubmitted($, (ids) => [...ids, promptId])
106  return false
107}
108
109/** `promptId` wasn't given Claude Code after all, so it may be delivered again. */
110const notSubmitted = ($: EngineInterface, promptId: string) => {
111  submitted.ids.delete(promptId)
112  void storeSubmitted($, (ids) => ids.filter((id) => id !== promptId))
113}
114
115const setBand = ($: EngineInterface, change: Partial<HubLinkBand>) =>
116  update($, band, (value) => ({ ...value, ...change })).catch(() => undefined)
117
118const bandStatus = (status: LinkStatus): Partial<HubLinkBand> =>
119  status._tag === "Rejected"
120    ? { status: "Rejected", detail: status.reason }
121    : { status: status._tag, detail: "" }
122
123type Command<Tag extends ModCommand["_tag"]> = Extract<ModCommand, { _tag: Tag }>
124
125const deliverPrompt = async ($: EngineInterface, command: Command<"DeliverPrompt">) => {
126  if (await submittedBefore($, command.promptId)) {
127    // A Hub from before #323 would refuse the event, so it keeps the prompt queued: still run once.
128    if (state.link?.takesAlreadySubmitted) {
129      report({ _tag: "PromptAlreadySubmitted", promptId: command.promptId })
130    }
131    return
132  }
133  const failed = (message: string) => {
134    // No Turn starts with it.
135    void $.state.set(delivered, null).catch(() => undefined)
136    notSubmitted($, command.promptId)
137    report({ _tag: "PromptFailed", promptId: command.promptId, message })
138  }
139  const attachments = command.attachments ?? []
140  const text = deliveredPrompt(command.sender, command.text, attachments)
141  try {
142    // Said before it is submitted, so hub-link reads it when the Turn starts, and adds what Claude
143    // is told of its files: a plugin's own hooks don't see the prompt it submits.
144    await $.state.set(delivered, {
145      promptId: command.promptId,
146      sender: command.sender,
147      text,
148      ...(attachments.length > 0 && { context: attachmentContext(attachments) }),
149    })
150    const entered = await $.prompt.submit({ text, asUser: true })
151    if (entered.drop !== undefined) failed(`a hook dropped the prompt: ${entered.drop}`)
152  } catch (error) {
153    failed(String(error))
154  }
155}
156
157/**
158 * A queued prompt the Creator steered into the running Turn (#195, ADR 0018): a row of the
159 * Turn's conversation, which Claude reads at its next step, as Claude Code adds a prompt typed
160 * while it works. hub-link sees it join, and reports it. Once the Turn has ended it can't join:
161 * the Hub keeps it queued.
162 */
163const deliverIntoTurn = async ($: EngineInterface, command: Command<"DeliverIntoTurn">) => {
164  const failed = (message: string) =>
165    report({ _tag: "PromptFailed", promptId: command.promptId, message })
166  if (state.mainTurn === undefined) return failed("no Turn is running")
167  const { promptId, sender, text } = command
168  const attachments = command.attachments ?? []
169  // Its files are marked in the row (#278): Claude opens them by the markers alone.
170  const row = steeredPrompt(sender, text, attachments)
171  try {
172    // Said before it is appended, so hub-link knows the row.
173    await $.state.set(steering, {
174      promptId,
175      sender,
176      text,
177      ...(attachments.length > 0 && { attachments: [...attachments] }),
178      appended: row,
179    })
180    const appended = await $.session.append({
181      message: { type: "user", content: [{ type: "text", text: row }] },
182    })
183    if (appended.deny !== undefined) failed(appended.deny)
184  } catch (error) {
185    failed(String(error))
186  } finally {
187    await $.state.set(steering, null).catch(() => undefined)
188  }
189}
190
191/**
192 * The Creator's message to an agent (ADR 0014), with their name and a marker for each file
193 * (#278), as a prompt is: a running agent reads it at its next tool call; one that has stopped
194 * resumes with it. A subagent type that lists its own tools can't open the files.
195 */
196const messageAgent = async ($: EngineInterface, command: Command<"DeliverAgentMessage">) => {
197  const failed = (message: string) =>
198    report({
199      _tag: "AgentCommandFailed",
200      agentId: command.agentId,
201      command: "MessageAgent",
202      senderId: command.sender.id,
203      message,
204    })
205  try {
206    const sent = await $.session.send({
207      to: { agentId: command.agentId },
208      text: deliveredPrompt(command.sender, command.text, command.attachments ?? []),
209    })
210    if (!sent.isDelivered) failed(sent.reason)
211  } catch (error) {
212    failed(String(error))
213  }
214}
215
216/** Stops an agent with Claude Code's TaskStop, by its id; hub-link sees the call, and reports it. */
217const stopAgent = async ($: EngineInterface, command: Command<"StopAgent">) => {
218  const failed = (message: string) =>
219    report({
220      _tag: "AgentCommandFailed",
221      agentId: command.agentId,
222      command: "StopAgent",
223      message,
224    })
225  try {
226    const stopped = await $.tool.call({ tool: "TaskStop", task_id: command.agentId })
227    if (stopped.deny !== undefined) failed(stopped.deny)
228    else if (stopped.isError) failed(String(stopped.result))
229  } catch (error) {
230    failed(String(error))
231  }
232}
233
234const runCommand = async ($: EngineInterface, command: ModCommand) => {
235  switch (command._tag) {
236    // Neither waits: a prompt waits for the session to be idle, and a resumed agent runs on, while
237    // the poll goes on.
238    case "DeliverPrompt":
239      void deliverPrompt($, command)
240      return
241    case "DeliverAgentMessage":
242      void messageAgent($, command)
243      return
244    // Appended at once, so one steered prompt is told from the next.
245    case "DeliverIntoTurn":
246      await deliverIntoTurn($, command)
247      return
248    case "StopAgent":
249      await stopAgent($, command)
250      return
251    case "StopTurn":
252      if (state.mainTurn === undefined) return
253      state.stoppedTurn = state.mainTurn
254      await $.turn.abort({ turnId: state.mainTurn })
255      return
256    case "DeliverAnswer":
257      // Too late: answered at the terminal or by someone else first, or cancelled. The Hub tells
258      // the sender which.
259      if (state.asking.get(command.toolCallId)?.(command.answer, command.sender) !== true)
260        report({
261          _tag: "AnswerFailed",
262          toolCallId: command.toolCallId,
263          senderId: command.sender.id,
264          message: "The question was answered or closed already.",
265        })
266      return
267    case "PresenceChanged":
268      // Someone forwarding a port is here too, from a machine (#180).
269      await setBand($, {
270        participants: command.participants.map((person) =>
271          person.forwardingFrom === undefined
272            ? person.name
273            : `${person.name} (forwarding from ${person.forwardingFrom.join(", ")})`,
274        ),
275      })
276      return
277  }
278}
279
280const startLink = async ($: EngineInterface) => {
281  const [hubUrl, sessionId, modToken, kind, launched] = await Promise.all([
282    $.env.get("KNDL_URL"),
283    $.env.get("KNDL_SESSION_ID"),
284    $.env.get("KNDL_MOD_TOKEN"),
285    $.env.get("CLAUDE_CODE_SESSION_KIND"),
286    $.env.get("KNDL_CONVERSATION_ID"),
287  ])
288  // hub-link says so in the debug log. A background job or daemon of Claude Code's isn't the
289  // Session's Claude Code, and runs none of its commands (#191); nor does a `claude` run inside
290  // the Session as another conversation than the one the Runner launched (#321).
291  if (!hubUrl || !sessionId || !modToken || kind) return false
292  const conversation = await $.session.id()
293  if (launched && conversation !== launched) return false
294  state.hub = { url: hubUrl.replace(/\/+$/, ""), modToken }
295  state.link?.stop()
296  state.link = new HubLink({
297    hubUrl,
298    modToken,
299    hello: { sessionId, claudeSessionId: conversation, model: await $.session.model() },
300    fetch: (url, init) => $.http.fetch(url, init),
301    // Its own timers, so a Stop can't cut what it does short (#165).
302    after: (ms, fn) => void $.clock.after(ms, fn),
303    onCommand: (command) => runCommand($, command),
304    onStatus: (status) => void setBand($, bandStatus(status)),
305    log: (text) => $.ui.log(text, { to: "debug" }),
306  })
307  state.link.start()
308  return true
309}
310
311export function register(on: On) {
312  on("session.start", async ($, e, next) => {
313    const started = await next(e)
314    const linked = await startLink($).catch((error) => {
315      $.ui.log(`hub-link-commands: ${error}`, { to: "debug" })
316      return false
317    })
318    // In the tool list from the first request on, beside the Hub link it fetches through.
319    if (linked)
320      await $.tool
321        .register(attachmentToolSpec)
322        .catch((error) => $.ui.log(`hub-link-commands: ${error}`, { to: "debug" }))
323    return started
324  })
325
326  // A file the Creator attached (#276, ADR 0024), by the id its marker gives: fetched from the
327  // Hub, through the Runner's relay, in parts the runtime's fetch can read, and handed to Claude as
328  // an image block, or a document block for a PDF (#277).
329  on("tool.call", { tool: attachmentToolName }, async ($, e) => {
330    const id = typeof e.id === "string" ? e.id : ""
331    const hub = state.hub
332    if (hub === undefined) return { deny: "This session is not linked to a kndl Hub." }
333    if (id === "") return { deny: "Give the attachment's id, from its marker." }
334    const parts = []
335    try {
336      for (let offset: number | undefined = 0; offset !== undefined;) {
337        if (parts.length === maxAttachmentParts) break
338        const response = await $.http.fetch(`${hub.url}${attachmentPartPath(id, offset)}`, {
339          headers: { authorization: `Bearer ${hub.modToken}` },
340        })
341        const answer = attachmentPart(response.status, response.text)
342        if ("deny" in answer) return answer
343        parts.push(answer.part)
344        offset = answer.part.next
345      }
346    } catch (error) {
347      return { deny: `Could not open the attachment: ${String(error)}` }
348    }
349    return attachmentResult(parts)
350  })
351
352  on("session.end", async ($, e, next) => {
353    const ended = await next(e)
354    await state.link?.flushNow()
355    return ended
356  })
357
358  // A Turn from a prompt this plugin submitted reaches these hooks too: only an agent it resumes
359  // runs past them.
360  on("turn.start", async ($, e, next) => {
361    state.mainTurn = e.turnId
362    return next(e)
363  })
364
365  on("turn.complete", async ($, e, next) => {
366    if (e.agentId === undefined && e.turnId === state.mainTurn) state.mainTurn = undefined
367    return next(e)
368  })
369
370  // Claude's question on the main loop is the Session's Open question (#193): its dialog opens at
371  // the terminal as ever, and the Creator may answer it from the web UI meanwhile. The first
372  // answer is the one Claude reads; one from the web closes the dialog, by settling this call
373  // while it is still open. A subagent's questions are answered at the terminal alone.
374  on("tool.call", { tool: "AskUserQuestion" }, async ($, e, next) => {
375    const turnId = state.mainTurn
376    const toolCallId = e.tool_use_id
377    if (e.agentId !== undefined || state.link === undefined || turnId === undefined || !toolCallId)
378      return next(e)
379    report({ _tag: "QuestionAsked", turnId, toolCallId, questions: askedQuestions(e.questions) })
380    const closed = (outcome: Record<string, unknown>, by?: Participant) =>
381      report({
382        _tag: "QuestionClosed",
383        turnId,
384        toolCallId,
385        outcome,
386        ...(by === undefined ? {} : { by }),
387      })
388    let settled = false
389    const settle = () => {
390      if (settled) return false
391      settled = true
392      state.asking.delete(toolCallId)
393      return true
394    }
395    // Stopped from the web UI: Claude reads no answer, and the Hub hears so now. Esc at the
396    // terminal stops the Turn too, and the call then ends refused, as dismissed (#206). One
397    // stopped otherwise that never ends, the Hub closes as cancelled with its Turn.
398    const isStopped = () => state.stoppedTurn === turnId
399    next.signal.addEventListener("abort", () => {
400      if (isStopped() && settle()) closed({ _tag: "Cancelled" })
401    })
402    return new Promise<Awaited<ReturnType<typeof next>>>((resolve, reject) => {
403      state.asking.set(toolCallId, (answer, sender) => {
404        if (!settle()) return false
405        closed(answer, sender)
406        resolve(answerResult(e.questions, answer, sender) as never)
407        return true
408      })
409      next(e).then(
410        (done) => {
411          if (settle()) closed(isStopped() ? { _tag: "Cancelled" } : terminalOutcome(done))
412          resolve(done)
413        },
414        (error: unknown) => {
415          if (settle()) closed({ _tag: "Cancelled" })
416          reject(error)
417        },
418      )
419    })
420  })
421
422  on("ui.render", { component: "AbovePrompt" }, async ($, e, next) => {
423    if (e.props.hasSurvey) return next(e)
424    const value = await read($, band)
425    if (value.status === "Unlinked") return next(e)
426    const { Box, Text } = $.ui.resolve(e)
427    const isLinked = value.status === "Linked"
428    const label =
429      value.status === "Rejected"
430        ? `Hub refused the link: ${value.detail}`
431        : value.status === "Linking"
432          ? "linking to the Hub"
433          : value.status === "Offline"
434            ? "Hub offline"
435            : "Hub"
436    const people =
437      value.participants.length === 0 ? " · nobody watching" : ` · ${value.participants.join(", ")}`
438    return Box({
439      flexDirection: "row",
440      children: [
441        Text({ color: isLinked ? "green" : "red", children: isLinked ? "● " : "○ " }),
442        Text({ dimColor: true, children: isLinked ? `${label}${people}` : label }),
443      ],
444    })
445  })
446}
447

1// hub-link-commands' contract: the state its band draws from, and the prompts it delivered and
2// steered last, which hub-link reads.
3
4/** What the band above the prompt shows. */
5export type HubLinkBand = {
6  /** Where the link to the Hub stands. */
7  status: "Linking" | "Linked" | "Offline" | "Rejected" | "Unlinked"
8  /** Why the Hub refused the link, when `status` is `Rejected`. */
9  detail: string
10  /** The names of the Participants connected to the Session now. */
11  participants: string[]
12}
13
14/** A Participant, as the protocol has them. */
15export type HubLinkParticipant = {
16  id: string
17  login: string
18  name: string
19  avatarUrl?: string
20}
21
22/** The prompt from the Prompt queue submitted last: the Turn it starts is reported as theirs. */
23export type HubLinkDelivered = {
24  promptId: string
25  sender: HubLinkParticipant
26  /** As submitted, led by the sender's name, with a marker for each file it carries. */
27  text: string
28  /**
29   * What Claude is told beside it, out of sight, when it carries files (#276): hub-link adds it
30   * as the prompt is submitted, since a plugin's hooks don't see the prompt it submits itself.
31   */
32  context?: string
33}
34
35/**
36 * The prompt from the Prompt queue being steered into the running Turn (#195): hub-link reports it
37 * joined when its row is appended.
38 */
39export type HubLinkSteering = {
40  promptId: string
41  sender: HubLinkParticipant
42  /** As its sender wrote it; empty for a prompt of attachments alone. */
43  text: string
44  /** The files it carries (#278), as the Hub sent them. */
45  attachments?: HubLinkAttachment[]
46  /** The row's text as appended: framed, with the sender's name and a marker for each file. */
47  appended: string
48}
49
50/** A file a prompt carries, as the protocol has it. */
51export type HubLinkAttachment = {
52  id: string
53  name: string
54  mediaType: string
55  size: number
56}
57
58declare module "claude-code" {
59  interface PluginState {
60    "hub-link-commands": {
61      band: HubLinkBand
62      delivered: HubLinkDelivered | null
63      steering: HubLinkSteering | null
64    }
65  }
66}
67