event-bridge

1import type { EngineInterface, PluginOptions, Register, SessionVersion } from 'claude-code'
2import type { InboxCarry } from '../types'
3import { type Config, type RawEnv, type ToolDetail, anySet, parseConfig, parseEnvCarry, toEnvCarry, withCarried } from './config'
4import {
5  bootStartedData,
6  nowMs,
7  notificationData,
8  permissionRequestedData,
9  sessionEndedData,
10  sessionStartedData,
11  splitToolCall,
12  toolFinishedData,
13  toolStartedData,
14  turnCompletedData,
15  turnStartedData,
16  type ToolOutcome,
17} from './events'
18import { Inbox, parseInboxCarry } from './inbox'
19import { type Carry, Outbox, parseOutboxCarry } from './outbox'
20import { askHost } from './permission'
21import type { Ports } from './ports'
22
23const PLUGIN_NAME = 'event-bridge'
24
25/** Below this much of `session.end`'s bound left, the final flush is skipped so the exit stays fast. */
26const END_FLUSH_MIN_MS = 200
27
28/** What each load leaves for the next: `$.state` outlives a hot reload, the module's memory does not. */
29const OUTBOX_CARRY = { plugin: 'event-bridge', key: 'outbox' } as const
30const INBOX_CARRY = { plugin: 'event-bridge', key: 'inbox' } as const
31const ENV_CARRY = { plugin: 'event-bridge', key: 'env' } as const
32
33/** What one load of the module holds. A hot reload starts a fresh one. */
34type Bridge = {
35  readonly options: PluginOptions
36  readonly outbox: Outbox
37  readonly inbox: Inbox
38  config: Config | undefined
39  /** The variables this load runs with, once read; kept in `$.state` for the next load. */
40  env: RawEnv | undefined
41  /** What the inbox of the load before left, for this load's inbox to go on from. */
42  inboxCarry: InboxCarry | undefined
43  starting: Promise<void> | undefined
44}
45
46/** The engine calls the other files make, bound to a hook's `$`. */
47function portsFor($: EngineInterface): Ports {
48  return {
49    fetch: (url, init) => $.http.fetch(url, init),
50    after: (ms, fn) => $.clock.after(ms, fn),
51    now: () => $.clock.now(),
52    log: (text, to) => $.ui.log(`${PLUGIN_NAME}: ${text}`, { to }),
53    toast: text => $.ui.toast(text),
54    submitPrompt: args => $.prompt.submit(args),
55    abortTurn: turnId => $.turn.abort({ turnId }),
56    // A write that fails leaves the one before it standing: the next load may send
57    // again what the host already has (it drops a repeated boot and seq) or miss the newest.
58    carryOutbox: carry => void $.state.set(OUTBOX_CARRY, carry).catch(() => undefined),
59    carryInbox: carry => void $.state.set(INBOX_CARRY, carry).catch(() => undefined),
60  }
61}
62
63/** What the load before this one left, or nothing when this is the first load or it cannot be read. */
64async function readCarries($: EngineInterface): Promise<{ outbox?: Carry; inbox?: InboxCarry; env?: RawEnv }> {
65  try {
66    const [outbox, inbox, env] = await Promise.all([$.state.get(OUTBOX_CARRY), $.state.get(INBOX_CARRY), $.state.get(ENV_CARRY)])
67    return { outbox: parseOutboxCarry(outbox.value), inbox: parseInboxCarry(inbox.value), env: parseEnvCarry(env.value) }
68  } catch {
69    return {}
70  }
71}
72
73/**
74 * Reads the bridge's variables, then clears each one that was set. Every
75 * process Claude Code starts afterwards inherits its environment: the
76 * commands Claude runs, which a prompt injection can steer, and a `claude`
77 * started from the session, which would otherwise join the host as this one.
78 */
79async function takeEnv($: EngineInterface): Promise<RawEnv> {
80  const env: RawEnv = {
81    url: await $.env.get('CLAUDE_EVENTS_URL'),
82    token: await $.env.get('CLAUDE_EVENTS_TOKEN'),
83    tag: await $.env.get('CLAUDE_EVENTS_TAG'),
84    tools: await $.env.get('CLAUDE_EVENTS_TOOLS'),
85    permissions: await $.env.get('CLAUDE_EVENTS_PERMISSIONS'),
86    commands: await $.env.get('CLAUDE_EVENTS_COMMANDS'),
87  }
88  const clearing = await Promise.allSettled([
89    env.url === undefined ? undefined : $.env.set('CLAUDE_EVENTS_URL', undefined),
90    env.token === undefined ? undefined : $.env.set('CLAUDE_EVENTS_TOKEN', undefined),
91    env.tag === undefined ? undefined : $.env.set('CLAUDE_EVENTS_TAG', undefined),
92    env.tools === undefined ? undefined : $.env.set('CLAUDE_EVENTS_TOOLS', undefined),
93    env.permissions === undefined ? undefined : $.env.set('CLAUDE_EVENTS_PERMISSIONS', undefined),
94    env.commands === undefined ? undefined : $.env.set('CLAUDE_EVENTS_COMMANDS', undefined),
95  ])
96  if (clearing.some(outcome => outcome.status === 'rejected')) {
97    portsFor($).log('could not clear the CLAUDE_EVENTS_ variables; commands Claude runs can read them', 'transcript')
98  }
99  return env
100}
101
102/**
103 * Writes what the next load needs and `$.state` may have lost: the variables
104 * and the inbox's carry. A `/clear` or a resume may empty `$.state` without
105 * reloading the module, so the hooks write them again after each.
106 */
107function keepCarries($: EngineInterface, bridge: Bridge): void {
108  if (bridge.env !== undefined) void $.state.set(ENV_CARRY, toEnvCarry(bridge.env)).catch(() => undefined)
109  bridge.inbox.carry()
110}
111
112/** Claude Code's version, or undefined when the engine cannot say it: the boot opens either way. */
113async function claudeCodeVersion($: EngineInterface): Promise<SessionVersion | undefined> {
114  try {
115    return await $.session.version()
116  } catch {
117    return undefined
118  }
119}
120
121/** Reads the config once per load, from whichever hook runs first. */
122function start($: EngineInterface, bridge: Bridge): Promise<void> {
123  bridge.starting ??= readConfig($, bridge)
124  return bridge.starting
125}
126
127async function readConfig($: EngineInterface, bridge: Bridge): Promise<void> {
128  const fromEnv = await takeEnv($)
129  const carried = await readCarries($)
130  // After the first load the environment no longer holds them: a reload finds them kept.
131  const env = withCarried(fromEnv, carried.env)
132  if (anySet(env)) {
133    bridge.env = env
134    keepCarries($, bridge)
135  }
136  const result = parseConfig(env, bridge.options)
137  if (result.kind === 'off') {
138    bridge.outbox.disable()
139    return
140  }
141  const ports = portsFor($)
142  if (result.kind === 'invalid') {
143    bridge.outbox.disable()
144    ports.log(`idle, ${result.reason}`, 'transcript')
145    return
146  }
147  bridge.config = result.config
148  bridge.inboxCarry = carried.inbox
149  for (const warning of result.warnings) ports.log(warning, 'transcript')
150  bridge.outbox.sessionId ??= await $.session.id()
151  bridge.outbox.start(ports, result.config, carried.outbox, bootStartedData(await claudeCodeVersion($)))
152}
153
154/**
155 * The `.catch` of every hook here: they only observe, so when one fails the
156 * engine's own answer stands. Where the hook had called `next`, `next(e)`
157 * replays what that call settled to and nothing runs twice.
158 */
159function passThrough<E, R>(_$: unknown, e: E, next: (e: E) => R): R {
160  return next(e)
161}
162
163export const register: Register = (on, options) => {
164  const outbox = new Outbox()
165  const bridge: Bridge = {
166    options,
167    outbox,
168    inbox: new Inbox(outbox),
169    config: undefined,
170    env: undefined,
171    inboxCarry: undefined,
172    starting: undefined,
173  }
174  const { inbox } = bridge
175  const detail = (): ToolDetail => bridge.config?.toolDetail ?? 'summary'
176
177  // The engine fires this on every load of the module, a hot reload included.
178  // Only a session with a person at the prompt polls for commands: a headless
179  // run (`-p`, the SDK) does not exit until the poll it holds comes back.
180  on('session.start', async ($, e, next) => {
181    await start($, bridge)
182    if (e.isInteractive && bridge.config !== undefined) inbox.start(portsFor($), bridge.config, bridge.inboxCarry)
183    return next(e)
184  })
185
186  on('classic.SessionStart', ($, e, next) => {
187    outbox.sessionId = e.session_id
188    outbox.emit('session.started', sessionStartedData(e))
189    void start($, bridge).then(() => keepCarries($, bridge))
190    return next(e)
191  }).catch(passThrough)
192
193  on('session.end', async ($, e, next) => {
194    outbox.emit('session.ended', sessionEndedData(e), e.sessionId)
195    if (next.budget.remainingMs >= END_FLUSH_MIN_MS) await outbox.flushNow()
196    // A /clear or a resume goes on in the same process under a new session id.
197    if (e.reason !== 'clear' && e.reason !== 'resume') inbox.stop()
198    inbox.currentTurnId = undefined
199    return next(e)
200  })
201
202  on('turn.start', ($, e, next) => {
203    inbox.currentTurnId = e.turnId
204    outbox.emit('turn.started', turnStartedData(e, detail()))
205    return next(e)
206  })
207
208  on('turn.complete', ($, e, next) => {
209    if (e.agentId === undefined && e.turnId === inbox.currentTurnId) inbox.currentTurnId = undefined
210    outbox.emit('turn.completed', turnCompletedData(e, detail()))
211    return next(e)
212  })
213
214  on('classic.Notification', ($, e, next) => {
215    outbox.emit('attention.notification', notificationData(e))
216    return next(e)
217  }).catch(passThrough)
218
219  on('classic.PermissionRequest', ($, e, next) => {
220    outbox.emit('permission.requested', permissionRequestedData(e, detail()))
221    return next(e)
222  }).catch(passThrough)
223
224  on('tool.call', async ($, e, next) => {
225    const call = splitToolCall(e)
226    outbox.emit('tool.started', toolStartedData(call, detail()))
227    const startedAt = nowMs()
228    let outcome: ToolOutcome | undefined
229    try {
230      const result = await next(e)
231      outcome = result
232      return result
233    } finally {
234      outbox.emit('tool.finished', toolFinishedData(call, outcome, nowMs() - startedAt, detail()))
235    }
236  }).catch(passThrough)
237
238  // The host may decide a call the engine would put to the person. It only
239  // runs when the person turned host permissions on and gave a token. A
240  // failure here skips the hook, which leaves the engine's own verdict.
241  on('tool.check', async ($, e, next) => {
242    const verdict = await next(e)
243    const config = bridge.config
244    if (verdict.decision !== 'ask' || config === undefined || !config.hostPermissions) return verdict
245    // A query (`$.tool.check`) has no call behind it to decide.
246    if (e.tool_use_id === undefined) return verdict
247    const answer = await askHost(portsFor($), config, {
248      requestId: e.tool_use_id,
249      sessionId: outbox.sessionId,
250      tool: e.tool,
251      input: e.input,
252      agentId: e.agentId,
253      reason: verdict.reason,
254    })
255    if (answer === undefined) return verdict
256    // The organization's ceiling caps what the host may grant.
257    if (answer.decision === 'allow' && e.ceiling !== undefined && e.ceiling !== 'allow') return verdict
258    return {
259      decision: answer.decision,
260      reason: answer.reason ?? `the host ${answer.decision === 'allow' ? 'allowed' : 'denied'} this call`,
261    }
262  }).catch(passThrough)
263}
264

1import type { PluginOptions } from 'claude-code'
2import type { EnvCarry } from '../types'
3import { field, isRecord } from './events'
4
5export type ToolDetail = 'summary' | 'full'
6
7/** What the mod runs with once the host is configured. */
8export type Config = {
9  /** Base URL with no trailing slash: `${url}/v1/events` and so on. */
10  readonly url: string
11  /** Bearer token; commands from the host and host permissions need it. */
12  readonly token: string | undefined
13  /** The host's own label, echoed on every event. */
14  readonly tag: string | undefined
15  readonly toolDetail: ToolDetail
16  readonly hostPermissions: boolean
17  readonly hostCommands: boolean
18}
19
20/**
21 * The raw environment, one field per variable; undefined means unset.
22 * Kept apart from `$.env.get` so the parser is a pure function.
23 */
24export type RawEnv = {
25  readonly url: string | undefined
26  readonly token: string | undefined
27  readonly tag: string | undefined
28  readonly tools: string | undefined
29  readonly permissions: string | undefined
30  readonly commands: string | undefined
31}
32
33/**
34 * `off`: no URL anywhere, the mod stays idle and says nothing.
35 * `invalid`: something is set but wrong, the mod stays idle and says why once.
36 * `on`: the config, plus warnings about options it had to drop.
37 */
38export type ConfigResult =
39  | { readonly kind: 'off' }
40  | { readonly kind: 'invalid'; readonly reason: string }
41  | { readonly kind: 'on'; readonly config: Config; readonly warnings: readonly string[] }
42
43const LOOPBACK_HOSTS: ReadonlySet<string> = new Set(['127.0.0.1', 'localhost', '[::1]'])
44const MAX_TOKEN_LENGTH = 4096
45const MAX_TAG_LENGTH = 256
46// Visible ASCII only: a token goes into a header, so no spaces or line breaks.
47const TOKEN_PATTERN = /^[\x21-\x7e]+$/
48const CONTROL_PATTERN = /[\x00-\x1f\x7f]/
49
50const ENV_FIELDS = ['url', 'token', 'tag', 'tools', 'permissions', 'commands'] as const
51
52/** Whether any of the bridge's variables was set. */
53export function anySet(env: RawEnv): boolean {
54  return ENV_FIELDS.some(name => env[name] !== undefined)
55}
56
57/** Each variable as it was read, or else as an earlier load kept it. */
58export function withCarried(env: RawEnv, carried: RawEnv | undefined): RawEnv {
59  return {
60    url: env.url ?? carried?.url,
61    token: env.token ?? carried?.token,
62    tag: env.tag ?? carried?.tag,
63    tools: env.tools ?? carried?.tools,
64    permissions: env.permissions ?? carried?.permissions,
65    commands: env.commands ?? carried?.commands,
66  }
67}
68
69export function toEnvCarry(env: RawEnv): EnvCarry {
70  return {
71    url: env.url ?? null,
72    token: env.token ?? null,
73    tag: env.tag ?? null,
74    tools: env.tools ?? null,
75    permissions: env.permissions ?? null,
76    commands: env.commands ?? null,
77  }
78}
79
80/**
81 * Variables read back from `$.state`, or undefined when they are not that
82 * shape. `parseConfig` checks the values themselves, as it does the
83 * environment's. A variable the carry lacks counts as unset: a reload into a
84 * newer copy of the mod can find a carry written before that variable existed.
85 */
86export function parseEnvCarry(raw: unknown): RawEnv | undefined {
87  if (!isRecord(raw)) return undefined
88  const env: { [K in (typeof ENV_FIELDS)[number]]?: string } = {}
89  for (const name of ENV_FIELDS) {
90    const value = field(raw, name)
91    if (value !== undefined && value !== null && typeof value !== 'string') return undefined
92    if (typeof value === 'string') env[name] = value
93  }
94  return { url: env.url, token: env.token, tag: env.tag, tools: env.tools, permissions: env.permissions, commands: env.commands }
95}
96
97/** An env value trimmed, or undefined when unset or blank. */
98function present(value: string | undefined): string | undefined {
99  if (value === undefined) return undefined
100  const trimmed = value.trim()
101  return trimmed === '' ? undefined : trimmed
102}
103
104function optionString(options: PluginOptions, key: string): string | undefined {
105  if (!Object.hasOwn(options, key)) return undefined
106  const value = options[key]
107  return typeof value === 'string' ? present(value) : undefined
108}
109
110function optionBoolean(options: PluginOptions, key: string): boolean {
111  if (!Object.hasOwn(options, key)) return false
112  return options[key] === true
113}
114
115/** A power the host gets only when asked for: `host` or `off` in its variable, else its boolean option. */
116function hostSwitch(
117  variable: string,
118  value: string | undefined,
119  options: PluginOptions,
120  option: string,
121): { ok: true; on: boolean } | { ok: false; reason: string } {
122  if (value === undefined) return { ok: true, on: optionBoolean(options, option) }
123  if (value === 'host') return { ok: true, on: true }
124  if (value === 'off') return { ok: true, on: false }
125  return { ok: false, reason: `${variable} must be host or off, not "${value}"` }
126}
127
128/** The base URL, or the reason it is refused. */
129export function parseUrl(raw: string): { ok: true; url: string } | { ok: false; reason: string } {
130  let parsed: URL
131  try {
132    parsed = new URL(raw)
133  } catch {
134    return { ok: false, reason: `"${raw}" is not a URL` }
135  }
136  if (parsed.username !== '' || parsed.password !== '') {
137    return { ok: false, reason: 'the URL must not carry a user name or password; use the token' }
138  }
139  if (parsed.search !== '' || parsed.hash !== '') {
140    return { ok: false, reason: 'the URL must not carry a query or a fragment' }
141  }
142  if (parsed.protocol === 'http:') {
143    if (!LOOPBACK_HOSTS.has(parsed.hostname)) {
144      return { ok: false, reason: `plain http is only allowed to 127.0.0.1, localhost or [::1], not ${parsed.hostname}` }
145    }
146  } else if (parsed.protocol !== 'https:') {
147    return { ok: false, reason: `the URL must be http or https, not ${parsed.protocol}` }
148  }
149  const path = parsed.pathname.replace(/\/+$/, '')
150  return { ok: true, url: `${parsed.origin}${path}` }
151}
152
153/**
154 * Builds the config from the environment and the plugin's options.
155 * A set environment variable wins over its option.
156 */
157export function parseConfig(env: RawEnv, options: PluginOptions): ConfigResult {
158  const rawUrl = present(env.url) ?? optionString(options, 'url')
159  if (rawUrl === undefined) return { kind: 'off' }
160
161  const url = parseUrl(rawUrl)
162  if (!url.ok) return { kind: 'invalid', reason: url.reason }
163
164  const token = present(env.token) ?? optionString(options, 'token')
165  if (token !== undefined) {
166    if (token.length > MAX_TOKEN_LENGTH) {
167      return { kind: 'invalid', reason: `the token is longer than ${MAX_TOKEN_LENGTH} characters` }
168    }
169    if (!TOKEN_PATTERN.test(token)) {
170      return { kind: 'invalid', reason: 'the token may hold only visible ASCII characters' }
171    }
172  }
173
174  const tag = present(env.tag)
175  if (tag !== undefined) {
176    if (tag.length > MAX_TAG_LENGTH) {
177      return { kind: 'invalid', reason: `CLAUDE_EVENTS_TAG is longer than ${MAX_TAG_LENGTH} characters` }
178    }
179    if (CONTROL_PATTERN.test(tag)) {
180      return { kind: 'invalid', reason: 'CLAUDE_EVENTS_TAG holds control characters' }
181    }
182  }
183
184  const tools = present(env.tools) ?? optionString(options, 'toolDetail') ?? 'summary'
185  if (tools !== 'summary' && tools !== 'full') {
186    return { kind: 'invalid', reason: `tool detail must be summary or full, not "${tools}"` }
187  }
188
189  const permissions = hostSwitch('CLAUDE_EVENTS_PERMISSIONS', present(env.permissions), options, 'hostPermissions')
190  if (!permissions.ok) return { kind: 'invalid', reason: permissions.reason }
191  const commands = hostSwitch('CLAUDE_EVENTS_COMMANDS', present(env.commands), options, 'hostCommands')
192  if (!commands.ok) return { kind: 'invalid', reason: commands.reason }
193
194  const warnings: string[] = []
195  if (token === undefined) {
196    if (permissions.on) warnings.push('host permissions are off: they need a token')
197    if (commands.on) warnings.push('host commands are off: they need a token')
198  }
199
200  return {
201    kind: 'on',
202    config: {
203      url: url.url,
204      token,
205      tag,
206      toolDetail: tools,
207      hostPermissions: permissions.on && token !== undefined,
208      hostCommands: commands.on && token !== undefined,
209    },
210    warnings,
211  }
212}
213

1import type { JsonValue } from 'claude-code'
2import type { ToolDetail } from './config'
3
4/**
5 * The mod's own version, as `.claude-plugin/plugin.json` states it: held here
6 * so the mod reads no file to learn it. `scripts/check.sh` fails when they differ.
7 */
8export const MOD_VERSION = '0.1.0'
9
10export const EVENT_TYPES = [
11  'boot.started',
12  'session.started',
13  'session.ended',
14  'turn.started',
15  'turn.completed',
16  'attention.notification',
17  'permission.requested',
18  'tool.started',
19  'tool.finished',
20  'command.result',
21  'outbox.dropped',
22] as const
23
24export type EventType = (typeof EVENT_TYPES)[number]
25
26export type EventData = { readonly [key: string]: JsonValue }
27
28/** One event on the wire. `seq` counts up within one `boot`. */
29export type Envelope = {
30  readonly v: 1
31  readonly boot: string
32  readonly seq: number
33  readonly ts: number
34  readonly sessionId: string | undefined
35  readonly tag: string | undefined
36  readonly type: EventType
37  readonly data: EventData
38}
39
40/** Strings longer than this are cut in `summary` mode. */
41export const SUMMARY_STRING_LIMIT = 2048
42/** Arrays longer than this are cut in `summary` mode. */
43export const SUMMARY_ARRAY_LIMIT = 100
44/** Deeper values are replaced in both modes, so a cycle-like shape cannot blow up a batch. */
45export const MAX_DEPTH = 12
46
47/** Epoch milliseconds: the one place the mod reads the time of day. */
48export function nowMs(): number {
49  return Date.now()
50}
51
52/** A random id for this load of the module, so the host can tell a reload's `seq` restart from a gap. */
53export function newBootId(): string {
54  return crypto.randomUUID()
55}
56
57export function isRecord(value: unknown): value is { readonly [key: string]: unknown } {
58  return typeof value === 'object' && value !== null && !Array.isArray(value)
59}
60
61export function field(record: { readonly [key: string]: unknown }, key: string): unknown {
62  return Object.hasOwn(record, key) ? record[key] : undefined
63}
64
65function isEventType(value: unknown): value is EventType {
66  return typeof value === 'string' && (EVENT_TYPES as readonly string[]).includes(value)
67}
68
69/**
70 * An envelope read back from `$.state`, or undefined when it is not one. A
71 * reload can bring in a different version of the mod, so nothing written by
72 * the earlier load is taken on trust.
73 */
74export function parseEnvelope(raw: unknown): Envelope | undefined {
75  if (!isRecord(raw)) return undefined
76  const boot = field(raw, 'boot')
77  const seq = field(raw, 'seq')
78  const ts = field(raw, 'ts')
79  const sessionId = field(raw, 'sessionId')
80  const type = field(raw, 'type')
81  const data = field(raw, 'data')
82  if (field(raw, 'v') !== 1 || typeof boot !== 'string' || boot === '') return undefined
83  if (typeof seq !== 'number' || !Number.isSafeInteger(seq) || seq < 1) return undefined
84  if (typeof ts !== 'number' || !Number.isFinite(ts)) return undefined
85  if (sessionId !== undefined && typeof sessionId !== 'string') return undefined
86  if (!isEventType(type) || !isRecord(data)) return undefined
87  // `data` came from `toWire` or a literal, and `$.state` holds JSON data alone.
88  return { v: 1, boot, seq, ts, sessionId, tag: undefined, type, data: data as EventData }
89}
90
91/**
92 * A JSON copy of `value`, with long strings and arrays cut in `summary` mode.
93 * Anything JSON cannot hold (functions, undefined, bigint) is dropped or turned
94 * into a string, so the result always serialises.
95 */
96export function toWire(value: unknown, detail: ToolDetail, depth = 0): JsonValue {
97  if (depth > MAX_DEPTH) return '[too deep]'
98  if (value === null) return null
99  switch (typeof value) {
100    case 'string':
101      return detail === 'summary' && value.length > SUMMARY_STRING_LIMIT
102        ? `${value.slice(0, SUMMARY_STRING_LIMIT)}… [truncated ${value.length - SUMMARY_STRING_LIMIT} chars]`
103        : value
104    case 'number':
105      return Number.isFinite(value) ? value : String(value)
106    case 'boolean':
107      return value
108    case 'bigint':
109      return value.toString()
110    case 'object': {
111      if (Array.isArray(value)) {
112        const kept = detail === 'summary' ? value.slice(0, SUMMARY_ARRAY_LIMIT) : value
113        const items = kept.map(item => toWire(item, detail, depth + 1))
114        if (kept.length < value.length) items.push(`[truncated ${value.length - kept.length} items]`)
115        return items
116      }
117      const out: { [key: string]: JsonValue } = {}
118      for (const [key, item] of Object.entries(value)) {
119        if (item === undefined || typeof item === 'function' || typeof item === 'symbol') continue
120        out[key] = toWire(item, detail, depth + 1)
121      }
122      return out
123    }
124    default:
125      return null
126  }
127}
128
129/** Drops undefined fields so the data is plain JSON. */
130function compact(fields: { readonly [key: string]: JsonValue | undefined }): EventData {
131  const out: { [key: string]: JsonValue } = {}
132  for (const [key, value] of Object.entries(fields)) {
133    if (value !== undefined) out[key] = value
134  }
135  return out
136}
137
138function stringField(record: { readonly [key: string]: unknown }, key: string): string | undefined {
139  const value = field(record, key)
140  return typeof value === 'string' ? value : undefined
141}
142
143/**
144 * `answer` is what `$.session.version()` answered, or undefined when it
145 * failed. The mod API changes between releases, so the answer is read field
146 * by field, and without a `version` the host gets no `claudeCode` at all.
147 */
148export function bootStartedData(answer: unknown): EventData {
149  const version = isRecord(answer) ? stringField(answer, 'version') : undefined
150  if (!isRecord(answer) || version === undefined) return { version: MOD_VERSION }
151  return {
152    version: MOD_VERSION,
153    claudeCode: compact({ version, base: stringField(answer, 'base'), builtAt: stringField(answer, 'builtAt') }),
154  }
155}
156
157export type SessionStartedFields = {
158  readonly source: string
159  readonly session_id: string
160  readonly cwd: string
161  readonly transcript_path: string
162  readonly model?: string
163  readonly agent_type?: string
164}
165
166export function sessionStartedData(e: SessionStartedFields): EventData {
167  return compact({
168    source: e.source,
169    cwd: e.cwd,
170    transcriptPath: e.transcript_path,
171    model: e.model,
172    agentType: e.agent_type,
173  })
174}
175
176export function sessionEndedData(e: { readonly reason: string; readonly sessionId: string; readonly resume: { readonly id: string } }): EventData {
177  return { reason: e.reason, resumeId: e.resume.id }
178}
179
180export function turnStartedData(e: { readonly turnId: string; readonly text: string }, detail: ToolDetail): EventData {
181  return { turnId: e.turnId, prompt: toWire(e.text, detail) }
182}
183
184export type TurnCompletedFields = {
185  readonly turnId: string
186  readonly agentId?: string
187  readonly reason: string
188  readonly answer: string
189  readonly durationMs: number
190  readonly isAborted: boolean
191  readonly usage?: unknown
192}
193
194export function turnCompletedData(e: TurnCompletedFields, detail: ToolDetail): EventData {
195  return compact({
196    turnId: e.turnId,
197    agentId: e.agentId,
198    reason: e.reason,
199    isAborted: e.isAborted,
200    durationMs: e.durationMs,
201    answer: toWire(e.answer, detail),
202    usage: e.usage === undefined ? undefined : toWire(e.usage, 'full'),
203  })
204}
205
206export function notificationData(e: { readonly message: string; readonly title?: string; readonly notification_type: string; readonly agent_id?: string }): EventData {
207  return compact({
208    notificationType: e.notification_type,
209    title: e.title,
210    message: e.message,
211    agentId: e.agent_id,
212  })
213}
214
215export function permissionRequestedData(
216  e: { readonly tool_name: string; readonly tool_input: unknown; readonly permission_suggestions?: unknown; readonly agent_id?: string },
217  detail: ToolDetail,
218): EventData {
219  return compact({
220    tool: e.tool_name,
221    input: toWire(e.tool_input, detail),
222    suggestions: e.permission_suggestions === undefined ? undefined : toWire(e.permission_suggestions, 'summary'),
223    agentId: e.agent_id,
224  })
225}
226
227/**
228 * Splits a `tool.call` input into the call's identity and the tool's own
229 * arguments, which the engine spreads beside `tool`.
230 */
231export function splitToolCall(e: { readonly [key: string]: unknown }): {
232  tool: string
233  toolUseId: string | undefined
234  agentId: string | undefined
235  input: { [key: string]: unknown }
236} {
237  const input: { [key: string]: unknown } = {}
238  for (const [key, value] of Object.entries(e)) {
239    if (key === 'tool' || key === 'tool_use_id' || key === 'agentId' || key === 'requestMeta' || key === 'consent') continue
240    input[key] = value
241  }
242  return {
243    tool: String(e.tool),
244    toolUseId: typeof e.tool_use_id === 'string' ? e.tool_use_id : undefined,
245    agentId: typeof e.agentId === 'string' ? e.agentId : undefined,
246    input,
247  }
248}
249
250export function toolStartedData(
251  call: { readonly tool: string; readonly toolUseId: string | undefined; readonly agentId: string | undefined; readonly input: unknown },
252  detail: ToolDetail,
253): EventData {
254  return compact({
255    tool: call.tool,
256    toolUseId: call.toolUseId,
257    agentId: call.agentId,
258    input: toWire(call.input, detail),
259  })
260}
261
262/** The parts of a `tool.call` result the host gets: never the tool's raw record. */
263export type ToolOutcome = {
264  readonly deny?: string
265  readonly isError?: true
266  readonly text?: string
267}
268
269export function toolFinishedData(
270  call: { readonly tool: string; readonly toolUseId: string | undefined; readonly agentId: string | undefined },
271  outcome: ToolOutcome | undefined,
272  durationMs: number,
273  detail: ToolDetail,
274): EventData {
275  let status: string
276  if (outcome === undefined) status = 'failed'
277  else if (outcome.deny !== undefined) status = 'denied'
278  else if (outcome.isError === true) status = 'error'
279  else status = 'ok'
280  return compact({
281    tool: call.tool,
282    toolUseId: call.toolUseId,
283    agentId: call.agentId,
284    status,
285    durationMs,
286    text: outcome?.deny ?? (outcome?.text === undefined ? undefined : toWire(outcome.text, detail)),
287  })
288}
289

1import type { Timer } from 'claude-code'
2import type { InboxCarry } from '../types'
3import type { Config } from './config'
4import { field, isRecord } from './events'
5import { type Outbox, authHeaders } from './outbox'
6import type { Ports } from './ports'
7
8/** A poll that came back sooner than this with nothing waits out the rest, so a host that does not hold the request is not hammered. */
9export const MIN_POLL_INTERVAL_MS = 1000
10export const POLL_BACKOFF_MAX_MS = 5000
11/** After the host refuses a poll outright (a wrong token), try again this rarely. */
12export const REFUSED_RETRY_MS = 30_000
13
14export const MAX_ID_LENGTH = 256
15export const MAX_PROMPT_LENGTH = 100_000
16export const MAX_TOAST_LENGTH = 2000
17export const MAX_COMMANDS_PER_POLL = 50
18
19export type Command =
20  | { readonly id: string; readonly type: 'prompt'; readonly text: string; readonly asUser: boolean }
21  | { readonly id: string; readonly type: 'interrupt' }
22  | { readonly id: string; readonly type: 'toast'; readonly text: string }
23
24export type CommandRejection = { readonly id: string | undefined; readonly error: string }
25
26/** One command from the host, or why it is refused. */
27export function parseCommand(raw: unknown): { ok: true; command: Command } | { ok: false; rejection: CommandRejection } {
28  if (!isRecord(raw)) return { ok: false, rejection: { id: undefined, error: 'a command must be an object' } }
29  const id = field(raw, 'id')
30  if (typeof id !== 'string' || id === '' || id.length > MAX_ID_LENGTH) {
31    return { ok: false, rejection: { id: undefined, error: `a command needs an id of 1 to ${MAX_ID_LENGTH} characters` } }
32  }
33  const reject = (error: string) => ({ ok: false as const, rejection: { id, error } })
34  const type = field(raw, 'type')
35  switch (type) {
36    case 'prompt': {
37      const text = field(raw, 'text')
38      if (typeof text !== 'string' || text.trim() === '') return reject('prompt needs a non-empty text')
39      if (text.length > MAX_PROMPT_LENGTH) return reject(`prompt text is longer than ${MAX_PROMPT_LENGTH} characters`)
40      const asUser = field(raw, 'asUser')
41      if (asUser !== undefined && typeof asUser !== 'boolean') return reject('asUser must be a boolean')
42      return { ok: true, command: { id, type, text, asUser: asUser === true } }
43    }
44    case 'interrupt':
45      return { ok: true, command: { id, type } }
46    case 'toast': {
47      const text = field(raw, 'text')
48      if (typeof text !== 'string' || text.trim() === '') return reject('toast needs a non-empty text')
49      if (text.length > MAX_TOAST_LENGTH) return reject(`toast text is longer than ${MAX_TOAST_LENGTH} characters`)
50      return { ok: true, command: { id, type, text } }
51    }
52    default:
53      return reject(`unknown command type ${JSON.stringify(type)}`)
54  }
55}
56
57/** The body of a 200 from `/v1/commands`: `{ commands: [...] }`. */
58export function parseCommands(body: string): { commands: Command[]; rejections: CommandRejection[] } {
59  let parsed: unknown
60  try {
61    parsed = JSON.parse(body)
62  } catch {
63    return { commands: [], rejections: [{ id: undefined, error: 'the command response is not JSON' }] }
64  }
65  const list = isRecord(parsed) ? field(parsed, 'commands') : undefined
66  if (!Array.isArray(list)) {
67    return { commands: [], rejections: [{ id: undefined, error: 'the command response has no commands array' }] }
68  }
69  const commands: Command[] = []
70  const rejections: CommandRejection[] = []
71  for (const raw of list.slice(0, MAX_COMMANDS_PER_POLL)) {
72    const result = parseCommand(raw)
73    if (result.ok) commands.push(result.command)
74    else rejections.push(result.rejection)
75  }
76  if (list.length > MAX_COMMANDS_PER_POLL) {
77    rejections.push({ id: undefined, error: `only the first ${MAX_COMMANDS_PER_POLL} commands of a poll are run` })
78  }
79  return { commands, rejections }
80}
81
82function isIdOrNull(value: unknown): value is string | null {
83  return value === null || (typeof value === 'string' && value !== '' && value.length <= MAX_ID_LENGTH)
84}
85
86/** A carry read back from `$.state`, or undefined when it is not one. */
87export function parseInboxCarry(raw: unknown): InboxCarry | undefined {
88  if (!isRecord(raw)) return undefined
89  const turnId = field(raw, 'turnId')
90  const lastId = field(raw, 'lastId')
91  if (!isIdOrNull(turnId) || !isIdOrNull(lastId)) return undefined
92  return { turnId, lastId }
93}
94
95/**
96 * Long-polls the host for commands and runs them. Each poll's timer starts
97 * the next, so the loop outlives any one hook. It runs only when host
98 * commands are on, which the config allows only with a token.
99 *
100 * While it runs it keeps the running turn and the last command id in
101 * `$.state`, through `ports.carryInbox`, so the copy a hot reload starts can
102 * still interrupt that turn and poll from that command.
103 */
104export class Inbox {
105  private ports: Ports | undefined
106  private config: Config | undefined
107  private timer: Timer | undefined
108  private running = false
109  private lastId: string | undefined
110  private backoffMs = 0
111  private turnId: string | undefined
112  /** Whether this load saw a turn start or end, which outranks what an earlier load left. */
113  private turnSeen = false
114
115  constructor(private readonly outbox: Outbox) {}
116
117  get isRunning(): boolean {
118    return this.running
119  }
120
121  /** The main loop's running turn, set by `turn.start` and cleared by its `turn.complete`. */
122  get currentTurnId(): string | undefined {
123    return this.turnId
124  }
125
126  set currentTurnId(turnId: string | undefined) {
127    this.turnId = turnId
128    this.turnSeen = true
129    this.carry()
130  }
131
132  /** Starts polling. `carry` is what an earlier load of the module left. */
133  start(ports: Ports, config: Config, carry?: InboxCarry): void {
134    if (this.running || !config.hostCommands) return
135    this.ports = ports
136    this.config = config
137    this.running = true
138    if (carry !== undefined) {
139      if (!this.turnSeen) this.turnId = carry.turnId ?? undefined
140      this.lastId ??= carry.lastId ?? undefined
141    }
142    this.carry()
143    this.schedule(0)
144  }
145
146  stop(): void {
147    this.running = false
148    this.timer?.cancel()
149    this.timer = undefined
150  }
151
152  /** Writes the running turn and the last command id to `$.state`, while it runs. */
153  carry(): void {
154    if (this.running) this.ports?.carryInbox({ turnId: this.turnId ?? null, lastId: this.lastId ?? null })
155  }
156
157  private schedule(delayMs: number): void {
158    if (!this.running || this.ports === undefined) return
159    this.timer = this.ports.after(delayMs, () => {
160      this.timer = undefined
161      void this.poll()
162    })
163  }
164
165  private async poll(): Promise<void> {
166    const ports = this.ports
167    const config = this.config
168    if (!this.running || ports === undefined || config === undefined) return
169
170    const params = new URLSearchParams()
171    if (this.outbox.sessionId !== undefined) params.set('session', this.outbox.sessionId)
172    if (config.tag !== undefined) params.set('tag', config.tag)
173    params.set('boot', this.outbox.bootId)
174    if (this.lastId !== undefined) params.set('after', this.lastId)
175
176    const startedAt = await ports.now()
177    let status: number | undefined
178    let body = ''
179    try {
180      const response = await ports.fetch(`${config.url}/v1/commands?${params.toString()}`, {
181        headers: authHeaders(config),
182      })
183      status = response.status
184      body = response.text
185    } catch {
186      status = undefined
187    }
188    if (!this.running) return
189
190    if (status !== undefined && status >= 200 && status < 300 && status !== 204) {
191      this.backoffMs = 0
192      const { commands, rejections } = parseCommands(body)
193      for (const rejection of rejections) {
194        this.outbox.emit('command.result', { id: rejection.id ?? null, ok: false, error: rejection.error })
195      }
196      for (const command of commands) {
197        this.lastId = command.id
198        this.run(ports, command)
199      }
200      if (commands.length > 0) this.carry()
201      const elapsed = (await ports.now()) - startedAt
202      this.schedule(commands.length === 0 && elapsed < MIN_POLL_INTERVAL_MS ? MIN_POLL_INTERVAL_MS - elapsed : 0)
203      return
204    }
205
206    if (status === 204) {
207      this.backoffMs = 0
208      const elapsed = (await ports.now()) - startedAt
209      this.schedule(elapsed < MIN_POLL_INTERVAL_MS ? MIN_POLL_INTERVAL_MS - elapsed : 0)
210      return
211    }
212
213    if (status !== undefined && status >= 400 && status < 500 && status !== 408 && status !== 429) {
214      ports.log(`the host refused the command poll with HTTP ${status}`, 'debug')
215      this.schedule(REFUSED_RETRY_MS)
216      return
217    }
218
219    this.backoffMs = this.backoffMs === 0 ? MIN_POLL_INTERVAL_MS : Math.min(this.backoffMs * 2, POLL_BACKOFF_MAX_MS)
220    this.schedule(this.backoffMs)
221  }
222
223  /**
224   * Runs one command without holding the poll: a prompt resolves only when
225   * its turn starts, which can be minutes away while a turn runs.
226   */
227  private run(ports: Ports, command: Command): void {
228    const done = (error?: string) =>
229      this.outbox.emit('command.result', error === undefined ? { id: command.id, ok: true } : { id: command.id, ok: false, error })
230    const failed = (error: unknown) => done(error instanceof Error ? error.message : String(error))
231
232    switch (command.type) {
233      case 'prompt':
234        ports
235          .submitPrompt(command.asUser ? { text: command.text, asUser: true } : { text: command.text })
236          .then(result => done(result.drop === undefined ? undefined : `a hook dropped the prompt: ${result.drop}`), failed)
237        return
238      case 'interrupt': {
239        const turnId = this.currentTurnId
240        if (turnId === undefined) {
241          done('no turn is running')
242          return
243        }
244        ports.abortTurn(turnId).then(() => done(), failed)
245        return
246      }
247      case 'toast':
248        ports.toast(command.text)
249        done()
250        return
251    }
252  }
253}
254

1import type { Timer } from 'claude-code'
2import type { OutboxCarry } from '../types'
3import type { Config } from './config'
4import { type Envelope, type EventData, type EventType, field, isRecord, newBootId, nowMs, parseEnvelope, toWire } from './events'
5import type { Ports } from './ports'
6
7/** Events held while the host is unreachable; past it the oldest of this boot's go, its `boot.started` aside. */
8export const QUEUE_CAP = 1000
9/** Events per POST. */
10export const BATCH_SIZE = 200
11/*
12 * Claude Code refuses a request body longer than 4 MiB of characters
13 * (4,194,304) before it is sent, and `$.state` a value whose JSON is longer,
14 * as measured on 2.1.296. The limits below keep well under that.
15 */
16/** Characters of JSON per POST, the tag stamped on each event included. */
17export const MAX_BATCH_CHARS = 3 * 1024 * 1024
18/** Characters of JSON one event may take, `tag` aside; a longer one is cut to summary detail, or dropped if that is not enough. */
19export const MAX_EVENT_CHARS = 1024 * 1024
20/** Room kept in each POST for the body around the events and a drop notice. */
21const BATCH_OVERHEAD_CHARS = 1024
22/**
23 * How many events a carry holds. A load's own events fill what an earlier
24 * load's leave of it and of MAX_CARRY_CHARS, and those past either count as
25 * dropped; an earlier load's events, and the load's `boot.started`, are kept
26 * even past them.
27 */
28export const CARRY_LIMIT = BATCH_SIZE
29/** Characters of JSON a carry's events may take, likewise. */
30export const MAX_CARRY_CHARS = 3 * 1024 * 1024
31/** How long an emit waits before its batch goes, so a burst shares one POST. */
32export const FLUSH_DELAY_MS = 100
33export const BACKOFF_START_MS = 250
34export const BACKOFF_MAX_MS = 5000
35
36export function authHeaders(config: Config): Record<string, string> {
37  return config.token === undefined ? {} : { authorization: `Bearer ${config.token}` }
38}
39
40type SendOutcome = 'sent' | 'retry' | 'rejected'
41
42/** The seq of a boot's `boot.started`, kept for it from the moment its outbox is made. */
43const BOOT_STARTED_SEQ = 1
44
45/** Whether `event` is `boot`'s `boot.started`. */
46function isBootStarted(event: Envelope | undefined, boot: string): boolean {
47  return event?.boot === boot && event.type === 'boot.started'
48}
49
50/** Where `boot`'s first event sits in `events`, or the end when it has none there. */
51function indexOfBoot(events: readonly Envelope[], boot: string): number {
52  const at = events.findIndex(event => event.boot === boot)
53  return at < 0 ? events.length : at
54}
55
56/**
57 * Where `boot`'s events after its `boot.started` begin in `events`: its first
58 * event, or the one behind that when it is the `boot.started`. A drop notice
59 * goes here, and a trim or a carry's limits start here, so nothing else of a
60 * boot goes ahead of its `boot.started`.
61 */
62function indexAfterBootStarted(events: readonly Envelope[], boot: string): number {
63  const at = indexOfBoot(events, boot)
64  return isBootStarted(events[at], boot) ? at + 1 : at
65}
66
67function isCount(value: unknown): value is number {
68  return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0
69}
70
71/** A carry as read back, its events checked as envelopes. */
72export type Carry = Omit<OutboxCarry, 'events'> & { readonly events: readonly Envelope[] }
73
74/**
75 * A carry read back from `$.state`, or undefined when it is not one. The load
76 * that wrote it may have been another version of the mod.
77 */
78export function parseOutboxCarry(raw: unknown): Carry | undefined {
79  if (!isRecord(raw)) return undefined
80  const boot = field(raw, 'boot')
81  const rawEvents = field(raw, 'events')
82  const dropped = field(raw, 'dropped')
83  const noticeSeq = field(raw, 'noticeSeq')
84  if (typeof boot !== 'string' || boot === '' || !Array.isArray(rawEvents) || !isCount(dropped)) return undefined
85  if (dropped > 0 ? !isCount(noticeSeq) || noticeSeq === 0 : noticeSeq !== null) return undefined
86  const events: Envelope[] = []
87  for (const rawEvent of rawEvents) {
88    const event = parseEnvelope(rawEvent)
89    if (event === undefined) return undefined
90    events.push(event)
91  }
92  return { boot, events, dropped, noticeSeq: dropped > 0 ? (noticeSeq as number) : null }
93}
94
95/**
96 * Holds events and POSTs them to the host in batches.
97 *
98 * It starts `pending` and queues what it is given, since events can arrive
99 * before the config is read. `start` turns it on and sends what is queued,
100 * behind the boot's `boot.started`; `disable` drops the queue and ignores
101 * every later emit. Nothing drops a `boot.started` to make room, and a drop
102 * notice goes behind it, so no other event of its boot arrives ahead of it.
103 *
104 * A hot reload starts a fresh copy, so once on it keeps what the host has not
105 * acknowledged in `$.state`, through `ports.carryOutbox`: the events up to
106 * CARRY_LIMIT and MAX_CARRY_CHARS, and a count of the rest. The copy the reload starts is handed
107 * that carry and sends it first, each event under its own boot and seq.
108 */
109export class Outbox {
110  private readonly boot = newBootId()
111  /** Every event here fits MAX_EVENT_CHARS, so any one of them makes a batch on its own. */
112  private queue: Envelope[] = []
113  /** The length of each event's JSON, `tag` aside, worked out once. */
114  private readonly lengths = new WeakMap<Envelope, number>()
115  /** The POST in flight, as sent: unacknowledged until it lands. */
116  private sending: Envelope[] = []
117  /** The count of the drop notice in that POST, or 0. */
118  private sendingNotice = 0
119  /** Events of this boot dropped and not yet reported to the host. */
120  private dropped = 0
121  /** The seq the next drop notice goes out under, kept across retries so a retry opens no gap. */
122  private noticeSeq: number | undefined
123  /** The last seq taken. BOOT_STARTED_SEQ is taken from the start, so the boot's other events count on behind its `boot.started`. */
124  private seq = BOOT_STARTED_SEQ
125  private state: 'pending' | 'on' | 'off' = 'pending'
126  private ports: Ports | undefined
127  private config: Config | undefined
128  private timer: Timer | undefined
129  private inFlight: Promise<SendOutcome> | undefined
130  private backoffMs = 0
131  private hostDown = false
132  private rejectionLogged = false
133  private carryPending = false
134
135  /** The session events are stamped with; the hooks keep it current. */
136  sessionId: string | undefined
137
138  get isOff(): boolean {
139    return this.state === 'off'
140  }
141
142  get queued(): number {
143    return this.queue.length
144  }
145
146  get droppedCount(): number {
147    return this.dropped
148  }
149
150  /** This load's boot, stamped on its events. */
151  get bootId(): string {
152    return this.boot
153  }
154
155  emit(type: EventType, data: EventData, sessionId: string | undefined = this.sessionId): void {
156    if (this.state === 'off') return
157    const event = this.fit(this.envelope(type, data, sessionId))
158    // Too long even when cut: its seq is the gap the drop notice explains.
159    if (event === undefined) this.dropped += 1
160    else this.queue.push(event)
161    this.trim()
162    this.schedule(FLUSH_DELAY_MS)
163    this.carrySoon()
164  }
165
166  /**
167   * Turns the outbox on. `bootStarted` is the data of this boot's
168   * `boot.started`, which goes under seq 1 ahead of every event queued so far;
169   * seq 1 is kept for it whether or not it is given. `carry` is what an
170   * earlier load of the module left unsent, which goes ahead of both.
171   */
172  start(ports: Ports, config: Config, carry?: Carry, bootStarted?: EventData): void {
173    if (this.state !== 'pending') return
174    this.ports = ports
175    this.config = config
176    this.state = 'on'
177    if (bootStarted !== undefined) this.queue.unshift(this.envelope('boot.started', bootStarted, this.sessionId, BOOT_STARTED_SEQ))
178    if (carry !== undefined && carry.boot !== this.boot) this.adopt(carry)
179    this.carrySoon()
180    if (this.queue.length > 0) this.schedule(0)
181  }
182
183  disable(): void {
184    this.state = 'off'
185    this.queue = []
186    this.sending = []
187    this.sendingNotice = 0
188    this.dropped = 0
189    this.noticeSeq = undefined
190    this.timer?.cancel()
191    this.timer = undefined
192  }
193
194  /**
195   * Sends what is queued now, for `session.end`. Stops at the first batch the
196   * host does not take; the end's own bound aborts a request still in flight.
197   */
198  async flushNow(): Promise<void> {
199    this.timer?.cancel()
200    this.timer = undefined
201    if (this.state !== 'on') return
202    if (this.inFlight !== undefined) await this.inFlight
203    while (this.queue.length > 0 || this.dropped > 0) {
204      const outcome = await this.sendBatch()
205      if (outcome !== 'sent') return
206    }
207  }
208
209  private envelope(type: EventType, data: EventData, sessionId: string | undefined, seq?: number): Envelope {
210    if (seq === undefined) {
211      this.seq += 1
212      seq = this.seq
213    }
214    return {
215      v: 1,
216      boot: this.boot,
217      seq,
218      ts: nowMs(),
219      sessionId,
220      // Stamped when the batch goes: events can be queued before the config is read.
221      tag: undefined,
222      type,
223      data,
224    }
225  }
226
227  /**
228   * Puts what an earlier load left unsent ahead of this load's events, with a
229   * notice for what that load dropped ahead of its own events.
230   */
231  private adopt(carry: Carry): void {
232    // Each was fitted by the load that queued it. One that another version of
233    // the mod carried is fitted here, and left out if it cannot be.
234    const events = carry.events.flatMap(event => this.fit(event) ?? [])
235    if (carry.dropped > 0 && carry.noticeSeq !== null) {
236      const at = indexAfterBootStarted(events, carry.boot)
237      events.splice(at, 0, {
238        v: 1,
239        boot: carry.boot,
240        seq: carry.noticeSeq,
241        ts: nowMs(),
242        sessionId: events[at]?.sessionId ?? this.sessionId,
243        tag: undefined,
244        type: 'outbox.dropped',
245        data: { count: carry.dropped },
246      })
247    }
248    this.queue.unshift(...events)
249  }
250
251  /**
252   * What a load that replaced this one would need. An earlier load's events
253   * and this load's `boot.started` are kept whole, and this load's other
254   * events fill the rest of CARRY_LIMIT and MAX_CARRY_CHARS; each one past
255   * either counts as dropped, and a notice in flight that is cut counts as
256   * the events it reports.
257   */
258  private carry(): OutboxCarry {
259    const pending = [...this.sending, ...this.queue]
260    const keptWhole = indexAfterBootStarted(pending, this.boot)
261    let kept = 0
262    let chars = 0
263    for (const event of pending) {
264      chars += this.lengthOf(event) + 1
265      if (kept >= keptWhole &&(kept >= CARRY_LIMIT || chars > MAX_CARRY_CHARS)) break
266      kept += 1
267    }
268    const events = pending.slice(0, kept)
269    let dropped = this.dropped - this.sendingNotice
270    for (const event of pending.slice(events.length)) dropped += event.type === 'outbox.dropped' ? this.sendingNotice : 1
271    // A notice in flight keeps its seq for its own count; what is left needs a new one.
272    const noticeSeq = this.sendingNotice > 0 || this.noticeSeq === undefined ? this.seq + 1 : this.noticeSeq
273    return { boot: this.boot, events, dropped, noticeSeq: dropped > 0 ? noticeSeq : null }
274  }
275
276  /** Writes the carry once the change under way is done, so a burst writes it once. */
277  private carrySoon(): void {
278    if (this.carryPending || this.state !== 'on') return
279    this.carryPending = true
280    void Promise.resolve().then(() => {
281      this.carryPending = false
282      if (this.state === 'on') this.ports?.carryOutbox(this.carry())
283    })
284  }
285
286  /** The length of `event`'s JSON, `tag` aside. */
287  private lengthOf(event: Envelope): number {
288    let length = this.lengths.get(event)
289    if (length === undefined) {
290      length = JSON.stringify(event).length
291      this.lengths.set(event, length)
292    }
293    return length
294  }
295
296  /** `event` cut to summary detail when its JSON is longer than MAX_EVENT_CHARS, or undefined when that is not enough. */
297  private fit(event: Envelope): Envelope | undefined {
298    if (this.lengthOf(event) <= MAX_EVENT_CHARS) return event
299    const data = toWire(event.data, 'summary')
300    if (!isRecord(data)) return undefined
301    // `toWire` gives an object back as an object of JSON values.
302    const cut: Envelope = { ...event, data: data as EventData }
303    return this.lengthOf(cut) <= MAX_EVENT_CHARS ? cut : undefined
304  }
305
306  /**
307   * Takes the next batch off the queue: up to `count` events whose JSON, each
308   * with the tag it is stamped with, fits MAX_BATCH_CHARS. The first always
309   * goes, since every queued event fits on its own.
310   */
311  private takeBatch(count: number, config: Config): Envelope[] {
312    const tagChars = config.tag === undefined ? 0 : ',"tag":'.length + JSON.stringify(config.tag).length
313    let chars = BATCH_OVERHEAD_CHARS
314    let taken = 0
315    for (const event of this.queue) {
316      const length = this.lengthOf(event) + tagChars + 1
317      if (taken === count || (taken > 0 && chars + length > MAX_BATCH_CHARS)) break
318      chars += length
319      taken += 1
320    }
321    return this.queue.splice(0, taken)
322  }
323
324  private trim(): void {
325    const over = this.queue.length - QUEUE_CAP
326    if (over <= 0) return
327    // An earlier load's events stay, so each boot's drops are reported under that boot, and so does this boot's boot.started.
328    this.dropped += this.queue.splice(indexAfterBootStarted(this.queue, this.boot), over).length
329  }
330
331  private schedule(delayMs: number): void {
332    if (this.state !== 'on' || this.timer !== undefined || this.inFlight !== undefined || this.ports === undefined) return
333    this.timer = this.ports.after(delayMs, () => {
334      this.timer = undefined
335      void this.flush()
336    })
337  }
338
339  private async flush(): Promise<void> {
340    if (this.inFlight !== undefined) return
341    const outcome = await this.sendBatch()
342    if (outcome === 'retry') {
343      this.backoffMs = this.backoffMs === 0 ? BACKOFF_START_MS : Math.min(this.backoffMs * 2, BACKOFF_MAX_MS)
344      this.schedule(this.backoffMs)
345      return
346    }
347    this.backoffMs = 0
348    if (this.queue.length > 0 || this.dropped > 0) this.schedule(0)
349  }
350
351  /** POSTs one batch, one at a time. A retryable failure puts the batch back at the front. */
352  private sendBatch(): Promise<SendOutcome> {
353    const sending = this.post().finally(() => {
354      this.inFlight = undefined
355      this.carrySoon()
356    })
357    this.inFlight = sending
358    return sending
359  }
360
361  private async post(): Promise<SendOutcome> {
362    const ports = this.ports
363    const config = this.config
364    if (ports === undefined || config === undefined) return 'retry'
365    // A drop notice takes one of the batch's places, so no POST holds more than BATCH_SIZE.
366    const events = this.takeBatch(this.dropped > 0 ? BATCH_SIZE - 1 : BATCH_SIZE, config)
367    // The count is taken now; drops while the POST is out stay counted for the next notice.
368    // While this boot's boot.started is still queued behind the batch, the notice waits for it.
369    const droppedNow = this.queue.some(event => isBootStarted(event, this.boot)) ? 0 : this.dropped
370    const batch = [...events]
371    if (droppedNow > 0) {
372      if (this.noticeSeq === undefined) {
373        this.seq += 1
374        this.noticeSeq = this.seq
375      }
376      // Ahead of this boot's own events but its boot.started, behind any an earlier load left.
377      const at = indexAfterBootStarted(events, this.boot)
378      batch.splice(at, 0, this.envelope('outbox.dropped', { count: droppedNow }, events[at]?.sessionId ?? this.sessionId, this.noticeSeq))
379    }
380    if (batch.length === 0) return 'sent'
381    this.sending = batch
382    this.sendingNotice = droppedNow
383    this.carrySoon()
384    const reported = () => {
385      this.dropped -= droppedNow
386      if (droppedNow > 0) this.noticeSeq = undefined
387    }
388
389    let status: number | undefined
390    try {
391      const response = await ports.fetch(`${config.url}/v1/events`, {
392        method: 'POST',
393        headers: { 'content-type': 'application/json', ...authHeaders(config) },
394        body: JSON.stringify({ events: batch.map(event => ({ ...event, tag: config.tag })) }),
395      })
396      status = response.status
397    } catch {
398      status = undefined
399    }
400    // Settled: below, in the same turn of the loop, the batch is counted done or put back.
401    this.sending = []
402    this.sendingNotice = 0
403
404    if (status !== undefined && status >= 200 && status < 300) {
405      if (this.hostDown) {
406        this.hostDown = false
407        ports.log('host reachable again', 'debug')
408      }
409      reported()
410      return 'sent'
411    }
412
413    // 408 and 429 are the host asking us to come back; other 4xx are refusals
414    // that a retry would only repeat (a wrong token, a host that does not
415    // speak this protocol), so that batch is dropped.
416    if (status !== undefined && status >= 400 && status < 500 && status !== 408 && status !== 429) {
417      if (!this.rejectionLogged) {
418        this.rejectionLogged = true
419        ports.log(`the host refused events with HTTP ${status}; dropping them`, 'transcript')
420      }
421      reported()
422      return 'rejected'
423    }
424
425    if (!this.hostDown) {
426      this.hostDown = true
427      ports.log(`host unreachable (${status === undefined ? 'no response' : `HTTP ${status}`}); holding up to ${QUEUE_CAP} events`, 'debug')
428    }
429    // Back to the front without the notice: the count stays in `dropped`
430    // and the next batch carries it under the same seq.
431    this.queue.unshift(...events)
432    this.trim()
433    return 'retry'
434  }
435}
436

1import type { Config } from './config'
2import { toWire } from './events'
3import { authHeaders } from './outbox'
4import type { Ports } from './ports'
5
6/** How long the host has to decide before the person's own dialog shows. */
7export const PERMISSION_TIMEOUT_MS = 30_000
8export const MAX_REASON_LENGTH = 1000
9
10export type HostDecision = { readonly decision: 'allow' | 'deny'; readonly reason: string | undefined }
11
12/**
13 * The host's answer to `/v1/permission`, or undefined when it defers
14 * (`ask`) or answers something this mod does not understand.
15 */
16export function parseDecision(body: string): HostDecision | undefined {
17  let parsed: unknown
18  try {
19    parsed = JSON.parse(body)
20  } catch {
21    return undefined
22  }
23  if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) return undefined
24  const decision = Object.hasOwn(parsed, 'decision') ? (parsed as { decision: unknown }).decision : undefined
25  if (decision !== 'allow' && decision !== 'deny') return undefined
26  const rawReason = Object.hasOwn(parsed, 'reason') ? (parsed as { reason: unknown }).reason : undefined
27  const reason = typeof rawReason === 'string' && rawReason.trim() !== '' ? rawReason.slice(0, MAX_REASON_LENGTH) : undefined
28  return { decision, reason }
29}
30
31export type PermissionAsk = {
32  readonly requestId: string
33  readonly sessionId: string | undefined
34  readonly tool: string
35  readonly input: unknown
36  readonly agentId: string | undefined
37  readonly reason: string | undefined
38}
39
40/**
41 * Asks the host to decide a tool call. Resolves undefined when the host
42 * defers, fails, refuses, or takes longer than `timeoutMs`.
43 */
44export function askHost(
45  ports: Ports,
46  config: Config,
47  ask: PermissionAsk,
48  timeoutMs: number = PERMISSION_TIMEOUT_MS,
49): Promise<HostDecision | undefined> {
50  return new Promise(resolve => {
51    let settled = false
52    const settle = (value: HostDecision | undefined) => {
53      if (settled) return
54      settled = true
55      timer.cancel()
56      resolve(value)
57    }
58    const timer = ports.after(timeoutMs, () => settle(undefined))
59    ports
60      .fetch(`${config.url}/v1/permission`, {
61        method: 'POST',
62        headers: { 'content-type': 'application/json', ...authHeaders(config) },
63        body: JSON.stringify({
64          requestId: ask.requestId,
65          sessionId: ask.sessionId ?? null,
66          tag: config.tag ?? null,
67          tool: ask.tool,
68          input: toWire(ask.input, config.toolDetail),
69          agentId: ask.agentId ?? null,
70          reason: ask.reason ?? null,
71        }),
72      })
73      .then(
74        response => settle(response.ok ? parseDecision(response.text) : undefined),
75        () => settle(undefined),
76      )
77  })
78}
79

1import type { HttpInit, HttpResponse, PromptSubmitArgs, PromptSubmitResult, Timer, UiLogSink } from 'claude-code'
2import type { InboxCarry, OutboxCarry } from '../types'
3
4/**
5 * The engine calls the outbox, inbox and permission ask make, as plain
6 * functions. `register.ts` builds them from a hook's `$`, so every call on
7 * `$` stays in that one file where `claude plugin validate` reads it, and
8 * the classes can be driven by a test with fakes.
9 */
10export type Ports = {
11  readonly fetch: (url: string, init?: HttpInit) => Promise<HttpResponse>
12  readonly after: (ms: number, fn: () => void) => Timer
13  readonly now: () => Promise<number>
14  /** One line, prefixed with the plugin's name. */
15  readonly log: (text: string, to: UiLogSink) => void
16  readonly toast: (text: string) => void
17  readonly submitPrompt: (args: PromptSubmitArgs) => Promise<PromptSubmitResult>
18  readonly abortTurn: (turnId: string) => Promise<void>
19  /** Keeps what the outbox has not had acknowledged in `$.state`, for the load a hot reload starts. */
20  readonly carryOutbox: (carry: OutboxCarry) => void
21  /** Keeps the running turn and the last command id in `$.state`, likewise. */
22  readonly carryInbox: (carry: InboxCarry) => void
23}
24

1// The values event-bridge keeps in `$.state`. A contract is self-contained,
2// so it spells out the envelope `hooks/events.ts` builds rather than import it.
3
4/** A JSON value, as `$.state` holds it. */
5export type Json = string | number | boolean | null | readonly Json[] | { readonly [key: string]: Json }
6
7/** One event as it goes on the wire, `tag` aside: that is stamped as a batch goes. */
8export type CarriedEnvelope = {
9  readonly v: 1
10  readonly boot: string
11  readonly seq: number
12  readonly ts: number
13  readonly sessionId?: string
14  readonly type: string
15  readonly data: { readonly [key: string]: Json }
16}
17
18/**
19 * What one load of the outbox had not had acknowledged, kept in `$.state` so
20 * the load a hot reload starts sends it first. Each event keeps its own boot
21 * and seq.
22 */
23export type OutboxCarry = {
24  /** The boot that wrote it. */
25  readonly boot: string
26  /** The events the host has not acknowledged, in the order they go, a batch in flight included. */
27  readonly events: readonly CarriedEnvelope[]
28  /** Events of `boot` that were dropped and not yet reported, those past the carry limit included. */
29  readonly dropped: number
30  /** The seq, within `boot`, that the notice for `dropped` goes out under; null when nothing was dropped. */
31  readonly noticeSeq: number | null
32}
33
34/** What the inbox keeps in `$.state` for the load a hot reload starts. */
35export type InboxCarry = {
36  /** The main loop's running turn, the one an `interrupt` aborts. */
37  readonly turnId: string | null
38  /** The last command received, sent back as `after`. */
39  readonly lastId: string | null
40}
41
42/**
43 * The bridge's variables as the mod read them. It clears them from the
44 * environment, so the commands Claude runs never inherit them, and keeps them
45 * here for the load a hot reload starts. Null for one that was unset.
46 */
47export type EnvCarry = {
48  readonly url: string | null
49  readonly token: string | null
50  readonly tag: string | null
51  readonly tools: string | null
52  readonly permissions: string | null
53  readonly commands: string | null
54}
55
56declare module 'claude-code' {
57  interface PluginState {
58    'event-bridge': { outbox: OutboxCarry; inbox: InboxCarry; env: EnvCarry }
59  }
60}
61