delta

1import type { ApiMessage, EngineInterface, On, RenderSurface, TextProps, Timer } from 'claude-code'
2import { capturedMessage, isOmittedTool, messageUpdates, resultUpdate, toolInfo, type Update } from './content'
3import { ToolRecords } from './tool-records'
4import { acknowledgedOutboxFiles, acknowledgedPrefix, appendBatch, appendOutboxRecord, outboxChunks, outboxRemoval } from './outbox'
5import { readCommandStream } from './command-reader'
6import { deliveryFailure, deliveryRetryDelay } from './delivery-policy'
7import { snapshotMessages } from './snapshot'
8import { activeBackgroundAgents, hasInteractiveSurface, joinPreflight, missingJoinCommands, ownsSteering, promptGate, releasesMounts, worktreeChanges } from './lifecycle'
9import { checkResult, reconnectDelay, validateAttachmentReceipt, validateCreationReceipt, validatePublicationReceipt } from './command-outcomes'
10import { copyText, personColor, personKey, renderBand, statusCurrentTurn, statusSpinnerRunning, SPINNER_FRAMES, threadTurnOwned, toolRowParts, type StatusSegment, type StatusState, type TurnStatus } from './status'
11import { announceThought, DrawnRows, DrawQueue, shownCommandRewrite, shownItemId, SHOW_ANSWER } from './row-delivery'
12import { restoreBinding } from './binding'
13import { pathKey, samePath } from './paths'
14import { PresenceTracker } from './presence'
15import {
16  DeltaCommandError, DeltaSendError, appendFailure, classifyCommandFailure, permanentFailureAdvice,
17  JoinNotReadyError, contractRefusal, contractVersion, parseThreadContract, parseThreadMount, parseThreadStatus, parseWatchEvent,
18  validateJoinedMounts, parseThreadFork, validateThreadFrontier, validateThreadCheckout, OFFLINE_CONTRACT_VERSION, PLUGIN_CONTRACT_VERSION, TITLE_CONTRACT_VERSION, JOIN_INTENT_CONTRACT_VERSION,
19  PRODUCER_NAME, PRODUCER_NAME_CONTRACT_VERSION,
20  type MountedThread, type ThreadContract, type ThreadMount, type ThreadReady, type ThreadState, type ThreadStatus, type ThreadTurn, type ThreadUpload, type WatchChild, type WatchEvent, type WatchItem,
21} from './thread-commands'
22import {
23  RemoteView, addRows, commentAction, commentAuthor, commentDisplay, commentMarkdown, commentQuote,
24  conversationText, conversationWebLink, isDrawnItem, isEventItem, itemAuthor, itemDisplay, itemMarkdown, keyedComment,
25  toolCallParts, THOUGHT_KIND, type Ledger, type PendingRow,
26} from './conversation'
27import { TitleScanner, transcriptRead } from './title'
28import { policyPath, readProjectPolicy, recordsAutomatically, type ProjectPolicy } from './project-policy'
29import {
30  HEADLESS_REFUSAL, InteractionDismissed, InteractionUnavailable, offerOnboarding, organizationsFrom, runSetupWizard,
31  type Interaction, type Organization, type SetupEnv,
32} from './setup'
33import { DeltaCli } from './cli'
34import { PendingAuthenticationError, requireAuthentication, type AuthenticationHost } from './auth'
35import {
36  HANDOFF_ACKNOWLEDGEMENT, HANDOFF_EXPIRY_MS, HANDOFF_PROMPT, newSessionLink, parseHandoff, type Handoff,
37} from './handoff'
38
39type Runtime = EngineInterface
40type Entry = { sequence: number; files: string[]; digest: string }
41type RecordBody = { update: Update; turnSequence?: number }
42type Execution = { sequence: number; turnId?: string; lost?: true; endSequence?: number }
43type CleanupPlan = { session: string; root: string; next: number; pending: { sequence: number; files: string[] }[] }
44type ReadyGate = { promise: Promise<ThreadReady>; resolve: (ready: ThreadReady) => void; reject: (error: unknown) => void }
45/** The upload a repository's policy asks an automatic session for. */
46type AutoUpload = { mode: 'private' | 'organization'; organization: string }
47type Binding = {
48  session: string
49  thread?: string
50  /** The pasted locator remains resolvable after restarting with an empty local Delta cache. */
51  locator?: string
52  /** A session working in mounted copies of a thread's worktrees, not a checkout attached by this session. */
53  joined?: true
54  /** A fresh session awaiting its verified directory transition; recording stays disabled. */
55  joinTransition?: JoinPlan
56  /** Last confirmed location of a shared session's attached checkout, not proof that the thread still owns it. */
57  checkout?: string
58  checkoutReference?: string
59  /**
60   * A joined session's mount owner: its Claude session ID, or for a session that took over a
61   * Desktop hand-off, the owner the offering tab mounted under.
62   */
63  owner?: string
64  /** The directory a joined session works in: the checkout of a thread's only worktree, else the thread folder. */
65  directory?: string
66  /** Every worktree of a joined session's thread, as the last `thread mount` placed them. */
67  mounts?: ThreadMount[]
68  url?: string
69  /** The audience Delta reports for the thread, shown apart from the link. */
70  visibility?: 'private' | 'organization' | 'unlisted'
71  /** Delta's background upload of this thread, as status last reported it. */
72  upload?: ThreadUpload
73  /**
74   * Delta confirmed this automatic session's first background upload. Recorded so the first-upload
75   * operation stays over even if a later per-turn sync reports `uploading` again.
76   */
77  uploaded?: true
78  /**
79   * Set on a session this plugin created from the repository's policy: what it uploads. Such a
80   * session may participate while its thread is still local-only.
81   */
82  autoUpload?: AutoUpload
83  /**
84   * When the session really ended, from the engine clock. Absent while it may still be live, so a
85   * session in another process is never pruned; pruned sessions can no longer resume into their thread.
86   */
87  endedAt?: number
88  snapshot: boolean
89  captureFailure?: string
90  next: number
91  pending: Entry[]
92  execution?: Execution
93  lastTurnSequence?: number
94  /** The last session title recorded for the thread, so each change is recorded once. */
95  title?: string
96}
97/**
98 * Another participant's running turn, as the band reports it: `since` is when the watch first
99 * reported it and `heard` when the thread last changed during it.
100 */
101type RemoteTurn = { turnId: string | null; name: string; id?: string; since?: number; heard?: number }
102/** A share or join that the band is reporting: running, finished, or failed. */
103type ProgressTone = 'active' | 'done' | 'error'
104/** A join to finish in a session; `owner` names a mount another session made for it to take over. */
105type JoinPlan = { session: string; cwd: string; locator: string; thread: string; owner?: string }
106type JoinRequest = { session: string; cwd: string; locator: string; fresh?: string; clearing?: boolean; cancelled?: boolean; contractVersion?: number }
107type ShareRequest = { session: string; cwd: string; cancelled?: boolean }
108
109const encoder = new TextEncoder()
110const LIVE_ACTIVITY_ORIGIN = {
111  work_origin: 'external_agent',
112  interaction_surface: 'cli',
113} as const
114type LiveActivityKind = 'user_message' | 'agent_message' | 'agent_thought' | 'tool_call' | 'edit'
115const cli = new DeltaCli()
116const MAX_LINE = 16 * 1024 * 1024
117const MAX_RECONNECT_DELAY = 30_000
118/** A mount may materialize several large checkouts before it answers. */
119const MOUNT_TIMEOUT_MS = 600_000
120/** `thread contract` answers from the CLI binary without starting delta-app, so this bound only catches a wedged process. */
121const CONTRACT_TIMEOUT_MS = 30_000
122/** How long exiting waits for Delta to record a release. */
123const RELEASE_TIMEOUT_MS = 5_000
124/** How often an offered hand-off is checked; the claiming session is another process. */
125const HANDOFF_POLL_MS = 2_000
126/** The band's animation: one braille cycle, redrawn within the engine's per-site throttle. */
127const SPINNER_INTERVAL_MS = 120
128/** How long a finished share or join stays in the band before clearing. */
129const BAND_NOTICE_MS = 4_000
130/** How long a lost connection may take to come back before the band says it is reconnecting. */
131const RECONNECT_GRACE_MS = 4_000
132/** How long another participant's running turn may go without any activity before the band calls it quiet. */
133const QUIET_AFTER_MS = 30_000
134/** A remote thought or tool call shows as what its agent is doing once open this long, so a burst of quick calls does not flicker. */
135const SHOW_ACTIVITY_AFTER_MS = 500
136/** How long the band confirms a copied link. */
137const COPIED_MS = 2_000
138let active = ''
139let sessionEpoch = 0
140let activatingSession = ''
141let queue: Promise<unknown> = Promise.resolve()
142let drainTask: Promise<void> | undefined
143let gate: Promise<void> | undefined
144let inFlight = 0
145let quiet: (() => void) | undefined
146/** The plugin's folder under Claude's configuration, shared by every Claude process on the machine. */
147let pluginData = ''
148let outbox = ''
149let joining: JoinRequest | undefined
150let sharing: ShareRequest | undefined
151/** A Keep or Discard the band started, so its two controls cannot both run at once. */
152let resolving = false
153/**
154 * Sessions whose CLI build already answered a compatible recording contract, each with the range it
155 * reported so the negotiated version and its offline capability are both known; a refusal is never
156 * remembered.
157 */
158const contractVerified = new Map<string, number>()
159/**
160 * Whether this session records offline, on a CLI that speaks the contract where
161 * `append --acknowledge local` answers without the server. An automatic or
162 * shared session does: it writes to its own thread, and Delta's background uploader pushes what it
163 * acknowledged locally. A joined session writes into someone else's thread, so it keeps
164 * server-confirmed delivery and holds Claude while Delta is unreachable.
165 */
166function recordsOffline(session: string, binding: Binding | undefined): boolean {
167  return !!binding && !binding.joined && (contractVerified.get(session) ?? 0) >= OFFLINE_CONTRACT_VERSION
168}
169/** A read of what a transcript gained gives up after this long; the next turn's end reads it again. */
170const TITLE_TIMEOUT_MS = 10_000
171/**
172 * How much of a transcript's end the first read takes in. Claude appends a title when it is chosen
173 * or renamed, and repeats it as the transcript grows, so the end of the file holds the current one.
174 */
175const TITLE_WINDOW_BYTES = 256 * 1024
176/** `$.fs.read` refuses a larger file, so reading a transcript whole is possible only below this. */
177const FS_READ_LIMIT = 4 * 1024 * 1024
178/**
179 * Where a session's transcript was last read to and the title found so far. `ranged` is cleared
180 * once reading from an offset fails, so the session reads its transcript whole from then on.
181 */
182type TitleReader = { path: string; offset: number; scanner: TitleScanner; ranged: boolean; reading?: true }
183const titleReaders = new Map<string, TitleReader>()
184const drawQueue = new DrawQueue()
185let drawing = false
186/** The answer the engine gives a plugin's command when no `command.run` hook of its own answered it. */
187const UNANSWERED_COMMAND = /no command\.run hook answered it/
188/**
189 * `watcher`: observation is not live; the next live watch connection lifts it.
190 * `verification`: a checkout check failed; a passing check lifts it.
191 * `delivery`: Delta refused a record or the thread; only explicit share/join recovery lifts it.
192 */
193type CheckoutPause = 'watcher' | 'verification' | 'delivery'
194/** One row waiting for its turn's start record, with the tool identity it later attributes. */
195type HeldRow = { update: Update; toolId?: string; turnId?: string }
196type SessionState = {
197  binding?: Binding
198  captureFailure?: string
199  /** Why checkout sharing is paused; each reason is lifted only by its own recovery. */
200  pauses: Map<CheckoutPause, string | undefined>
201  /** Delivery retry bookkeeping; see failDelivery. */
202  retryAttempts: number
203  retryTimer?: Timer
204  deliveryNotice?: string
205  /** Why sending is being retried, for the band while the retry waits. */
206  retryReason?: string
207  /** The next status read while Delta's background upload has not yet published the link; see followUpload. */
208  uploadRefresh?: Timer
209  uploadRefreshAttempts?: number
210  unsaved: { updates: Update[]; turnSequence?: number }[]
211  tools: ToolRecords
212  cleanupTask?: Promise<void>
213  cleanupError?: string
214  cleanupPlan?: CleanupPlan
215  cleanedThrough?: number
216  nativeTurns: Map<string, number>
217  aborting: Set<string>
218  /** Turns that lost the shared thread, so their late output keeps its turn instead of waiting. */
219  lostTurns: Set<string>
220  /** Items whose command output row this session drew, so a retried model row draws nothing twice. */
221  drawn: DrawnRows
222  currentTurnId?: string
223  /** A partial session transition must not silently become an unshared executing session. */
224  joinFailure?: string
225  /** A local prompt is entering, so remote rows wait until it is dropped or its turn completes. */
226  promptEntering?: boolean
227  /**
228   * Rows held until `turn.start` fixes the turn they belong to: a composer prompt and any output
229   * that reached a hook before its turn's scoped start was written. Recording them earlier would
230   * name the turn before theirs, and Delta fences output naming a turn that no longer owns it.
231   */
232  heldRows?: HeldRow[]
233  /** A joined session's mounts were confirmed in this process, and Claude was put in their directory. */
234  mounted?: true
235  /** A hand-off this Desktop tab offered, until a new session claims it or it expires. */
236  offered?: Handoff
237  /** A session that started in a hand-off's folder, joining it before its first prompt. */
238  handoffClaim?: Promise<void>
239  /**
240   * The hand-off's opening line, turned into a one-word turn that is left out of the thread:
241   * entering until its turn starts, then that turn's ID.
242   */
243  localTurn?: { turnId?: string }
244  /** Mounts are being restored for a resumed session; prompts wait. */
245  restoring?: true
246  /** Where Claude has to move between turns, because its worktree is no longer the thread's one directory. */
247  moveTo?: string
248  moving?: true
249  /** The latest handling of a `worktrees` event; the next one waits for it. */
250  mountRefresh?: Promise<void>
251  /** Context rows for Claude that wait for a turn to end. */
252  notes: string[]
253  /**
254   * Delta refused this offline session's recorded changes; the band offers Fork & Restore or
255   * Discard, and prompts keep retrying delivery until the person resolves it or an append succeeds.
256   */
257  refusedOffline?: boolean
258}
259const sessions = new Map<string, SessionState>()
260const presence = new PresenceTracker()
261const feed: {
262  error?: string; session?: string; thread?: string
263  /** Why the watch is reconnecting, for a prompt refused meanwhile. */
264  interruption?: string
265  revision: number; stopProcess?: () => Promise<unknown>
266  cancel?: AbortController
267  connection: 'stopped' | 'connecting' | 'live' | 'retrying'
268  ready?: ThreadReady; readyGate?: ReadyGate
269  view?: RemoteView
270  settling?: Promise<void>
271  settleNotice?: string
272  threadTurn?: Extract<WatchEvent, { type: 'turn' }>
273  producerTurn?: ThreadTurn
274  /** The thread turn last known to be this producer's. */
275  ownTurnId?: string
276  remoteTurn?: RemoteTurn
277  /** Child inventory from the current snapshot and subsequent per-child updates. */
278  children: WatchChild[]
279  /** When this session first saw each still-open remote item open, which times what the band shows it doing. */
280  openSince: Map<string, number>
281  /** The current turn, as the band reports it; recomputed whenever the watched turn changes. */
282  turn: TurnStatus
283  /** The share or join the band is reporting, held through the operation and briefly after success. */
284  progress?: { text: string; tone: ProgressTone; step?: [number, number] }
285  /** Clears a finished operation's band line; cancelled on teardown and by the next operation. */
286  progressTimer?: Timer
287  /** The animation's frame and redraw timer, live only while a live watch has a running turn. */
288  spinner?: { frame: number; timer: Timer }
289  /**
290   * A live watch that dropped: within RECONNECT_GRACE_MS of the drop `reconnectTimer` runs and the
291   * band still looks live; past it `reconnecting` is set until the watch is live again.
292   */
293  reconnecting?: boolean
294  reconnectTimer?: Timer
295  /** The link was just copied; the timer clears the confirmation. */
296  copiedTimer?: Timer
297} = {
298  revision: 0, connection: 'stopped', turn: { state: 'idle' }, children: [], openSince: new Map(),
299}
300
301function newReadyGate(): ReadyGate {
302  let resolve!: ReadyGate['resolve']
303  let reject!: ReadyGate['reject']
304  const promise = new Promise<ThreadReady>((yes, no) => { resolve = yes; reject = no })
305  void promise.catch(() => {})
306  return { promise, resolve, reject }
307}
308
309/** The engine requires helpers receiving `$` to remain in the hooks module. */
310async function startConversation($: Runtime, session: string): Promise<ThreadReady> {
311  if (!boundDirectory(stateFor(session).binding))
312    throw new Error('Thread setup is incomplete; run /delta share to finish explicit creation, attachment, and publication')
313  if (feed.session === session && feed.cancel && feed.readyGate) {
314    if (feed.connection === 'retrying') throw new Error(feed.interruption ?? 'Delta is reconnecting')
315    return feed.readyGate.promise
316  }
317  const stopping = stopConversation()
318  const revision = feed.revision
319  const cancel = new AbortController()
320  const ownsStartup = () => feed.revision === revision && feed.cancel === cancel && !cancel.signal.aborted
321  feed.cancel = cancel
322  setConnection($, 'connecting')
323  feed.session = session
324  const thread = stateFor(session).binding!.thread!
325  feed.thread = thread
326  const readiness = newReadyGate()
327  feed.readyGate = readiness
328  let launched = false
329  try {
330    await stopping
331    if (!ownsStartup()) throw new Error('Thread watcher startup was cancelled')
332    const saved = await $.store.get(ledgerKey(session)) as Ledger | undefined
333    if (!ownsStartup()) throw new Error('Thread watcher startup was cancelled')
334    feed.view = new RemoteView(thread, saved?.thread === thread && saved.items !== null && typeof saved.items === 'object'
335      ? saved : { thread, items: {} })
336    launched = true
337    void consumeConversation($, session, revision, cancel.signal, readiness).catch((error) => {
338      if (!ownsStartup()) return
339      feed.readyGate?.reject(error)
340      feed.cancel = undefined
341      setConnection($, 'stopped')
342      const binding = stateFor(session).binding!
343      feed.error = `Delta watcher failed: ${String(error)}. ${recoveryAdvice(binding)} to reconnect.`
344      $.ui.invalidate('ui.render')
345    })
346    return await readiness.promise
347  } catch (error) {
348    if (ownsStartup() && !launched) {
349      cancel.abort()
350      feed.cancel = undefined
351      setConnection($, 'stopped')
352    }
353    readiness.reject(error)
354    throw error
355  }
356}
357
358async function stopConversation() {
359  feed.revision++
360  feed.cancel?.abort()
361  feed.cancel = undefined
362  feed.connection = 'stopped'
363  clearReconnect()
364  feed.copiedTimer?.cancel()
365  feed.copiedTimer = undefined
366  stopSpinner()
367  // A running share or join keeps its band line across a watcher restart; only its timer ends.
368  feed.progressTimer?.cancel()
369  feed.progressTimer = undefined
370  feed.readyGate?.reject(new Error('Thread watcher stopped'))
371  feed.readyGate = undefined
372  feed.ready = undefined
373  const stop = feed.stopProcess
374  feed.stopProcess = undefined
375  feed.session = undefined
376  feed.thread = undefined
377  feed.error = undefined
378  feed.interruption = undefined
379  feed.view = undefined
380  feed.turn = { state: 'idle' }
381  // A stopped pass still saves the rows it added; finish that save before a new watch loads or writes the ledger.
382  const settling = feed.settling
383  feed.settling = undefined
384  feed.settleNotice = undefined
385  presence.reset()
386  feed.threadTurn = undefined
387  feed.producerTurn = undefined
388  feed.ownTurnId = undefined
389  feed.remoteTurn = undefined
390  feed.children = []
391  feed.openSince.clear()
392  dropQueuedDraws()
393  if (stop) await stop()
394  await settling?.catch(() => {})
395}
396
397async function consumeConversation(
398  $: Runtime,
399  session: string,
400  revision: number,
401  signal: AbortSignal,
402  readiness: ReadyGate,
403) {
404  const ownsObserver = () => !signal.aborted && revision === feed.revision
405  let attempts = 0
406  while (ownsObserver()) {
407    const started = await $.clock.now()
408    if (!ownsObserver()) return
409    let stream: ReturnType<Runtime['process']['spawn']> | undefined
410    let failure: unknown
411    let pending = ''
412    let stderr = ''
413    let ready: ThreadReady | undefined
414    let connected = true
415    let live = false
416    let inspected = false
417    const ownsConnection = () => connected && ownsObserver()
418    const inspection = new AbortController()
419    const cancelInspection = () => inspection.abort()
420    signal.addEventListener('abort', cancelInspection, { once: true })
421    let inspectionFailed = false
422    let inspectionError: unknown
423    let rejectInspection: ((error: unknown) => void) | undefined
424    // Participation needs both a complete snapshot and a passing access and checkout inspection.
425    const participate = () => {
426      if (!live || !inspected || !ownsConnection()) return
427      feed.error = undefined
428      feed.interruption = undefined
429      setConnection($, 'live')
430      resumeCheckout(stateFor(session), 'watcher')
431      readiness.resolve(ready!)
432      void drain($, session).catch((error) => {
433        if (ownsConnection()) return report($, error)
434      })
435    }
436    // Ready carries only identity and producer state. Inspect off the watch loop so slow
437    // access and checkout reads cannot delay turn cancellation.
438    const inspect = (baseline: ThreadReady) => void (async () => {
439      try {
440        const status = await readThreadStatus($, stateFor(session).binding?.locator ?? baseline.thread_id, session, inspection.signal)
441        if (!ownsConnection()) return
442        await serial(async () => {
443          if (!ownsConnection()) return
444          const current = await load($, session)
445          if (!ownsConnection()) return
446          validateThreadFrontier(status, current)
447          if (current.joined) {
448            // Access only: a worktree changed while disconnected is reconciled by its `worktrees`
449            // event, and every prompt and delivery checks each mount.
450            validateJoinedMounts(status, [])
451            await save($, { ...current, ...statusFields(status) }, ownsConnection)
452          } else {
453            const checkout = validateThreadCheckout(status, current)
454            await save($, {
455              ...current, ...statusFields(status),
456              checkout: checkout.directory, checkoutReference: checkout.reference,
457            }, ownsConnection)
458          }
459        })
460        if (!ownsConnection()) return
461        inspected = true
462        participate()
463        followUpload($, session)
464      } catch (error) {
465        if (ownsConnection()) {
466          inspectionFailed = true
467          inspectionError = error
468          rejectInspection?.(error)
469        }
470      }
471    })()
472    try {
473      setConnection($, 'connecting')
474      const binding = await serial(() => load($, session))
475      if (!ownsObserver()) return
476      if (!boundDirectory(binding))
477        throw new Error('No attached thread selected; run /delta share from the intended Git checkout')
478      // `--participate` asks Delta to show this session's signed-in person in the thread, under the
479      // name it gives, for as long as this watch runs; the stop below and each reconnect end and
480      // renew it. A CLI whose contract predates it is watched passively, without telling the person.
481      const participates = (contractVerified.get(session) ?? 0) >= PRODUCER_NAME_CONTRACT_VERSION
482      const argv = [await resolveCli($), '--json', 'thread', 'watch', binding.locator ?? binding.thread,
483        '--source', 'claude-code', '--external-id', session,
484        ...participates ? ['--participate', PRODUCER_NAME] : []]
485      if (!ownsObserver()) return
486      stream = $.process.spawn({
487        argv,
488        env: { DELTA_CLI_INTEGRATION: 'claude_code' },
489      })
490      const running = stream
491      feed.stopProcess = () => running.return({ code: null, signal: 'SIGTERM' })
492      while (true) {
493        if (inspectionFailed) throw inspectionError
494        // Only the currently pending next() listens for inspection failure. A single
495        // never-settled failure promise would retain a listener per watch line.
496        const inspectionFailure = new Promise<never>((_, reject) => { rejectInspection = reject })
497        const piece = await Promise.race([stream.next(), inspectionFailure])
498        rejectInspection = undefined
499        if (!ownsObserver()) return
500        if (piece.done)
501          throw new Error(`Delta watcher stopped (${piece.value.code ?? piece.value.signal ?? 'unknown'}). ${stderr}`)
502        if (piece.value.stream === 'stderr') {
503          stderr = (stderr + piece.value.text).slice(-4000)
504          continue
505        }
506        pending += piece.value.text
507        if (pending.length > MAX_LINE) throw new Error('Delta watch event exceeds the watcher limit')
508        let end: number
509        while ((end = pending.indexOf('\n')) >= 0) {
510          const line = pending.slice(0, end)
511          pending = pending.slice(end + 1)
512          if (!line.trim()) continue
513          const event = parseWatchEvent(line)
514          if (!event) continue
515          if (event.type === 'error') throw event.error
516          if (event.type === 'producer.turn') {
517            feed.producerTurn = { sequence: event.sequence, status: event.status }
518            observeThreadTurn($, session)
519            await observeTurn($, session, event, ownsObserver)
520            if (!ownsObserver()) return
521            continue
522          }
523          if (event.type === 'ready') {
524            if (ready) throw new Error('Duplicate thread readiness')
525            await serial(async () => {
526              if (!ownsObserver()) return
527              const current = await load($, session)
528              if (!ownsObserver()) return
529              validateThreadFrontier(event, current)
530            })
531            if (!ownsObserver()) return
532            feed.thread = event.thread_id
533            feed.ready = event
534            feed.view?.reset()
535            resetPresence()
536            // A new watch must replace the old inventory, not reuse it while awaiting its snapshot.
537            feed.children = []
538            feed.producerTurn = event.producer?.turn ?? undefined
539            if (event.producer?.turn) await observeTurn($, session, event.producer.turn, ownsObserver)
540            if (!ownsObserver()) return
541            refreshTurn($)
542            ready = event
543            inspect(event)
544            continue
545          }
546          if (!ready) throw new Error('Thread activity arrived before readiness')
547          if (event.type === 'live') {
548            live = true
549            feed.view?.endSnapshot()
550            void settleRemote($, session)
551            participate()
552          } else if (event.type === 'worktrees') refreshMounts($, session, event.worktrees)
553          else observeRemote($, session, event)
554        }
555      }
556    } catch (error) {
557      failure = error
558    } finally {
559      connected = false
560      rejectInspection = undefined
561      inspection.abort()
562      signal.removeEventListener('abort', cancelInspection)
563      try { if (stream) await stream.return({ code: null, signal: 'SIGTERM' }) }
564      catch (error) { failure ??= error }
565      if (ownsObserver()) feed.stopProcess = undefined
566    }
567    if (!ownsObserver()) return
568    feed.ready = undefined
569    // A session that records offline keeps working when the watcher drops: its records stay queued
570    // and drain retries, so nothing pauses and no running turn is lost.
571    if (!recordsOffline(session, stateFor(session).binding)) {
572      pauseCheckout(stateFor(session), 'watcher')
573      const running = stateFor(session).binding?.execution
574      if (running) await loseExecution($, session, running.sequence, 'The Delta watcher disconnected', ownsObserver)
575      if (!ownsObserver()) return
576    }
577    readiness.reject(failure)
578    if (failure instanceof DeltaCommandError && failure.hints.retryable === false) {
579      // A retryable failure is a lost connection, but a non-retryable one needs a person:
580      // reconnecting cannot bring back a deleted thread or restored access on its own.
581      const notice = `Delta sharing paused: ${failure.message}. ${permanentFailureAdvice(failure.code, !!stateFor(session).binding?.autoUpload, failure.reason)}, then ${recoveryAdvice(stateFor(session).binding)}.`
582      pauseCheckout(stateFor(session), 'delivery', notice)
583      feed.cancel = undefined
584      feed.readyGate = undefined
585      feed.error = notice
586      setConnection($, 'stopped')
587      await $.ui.log(notice, { to: 'transcript' })
588      return
589    }
590    readiness = newReadyGate()
591    feed.readyGate = readiness
592    // A child that emits one event then exits is not a healthy connection.
593    const ended = await $.clock.now()
594    if (!ownsObserver()) return
595    if (ended - started >= MAX_RECONNECT_DELAY) attempts = 0
596    const delay = reconnectDelay(attempts)
597    attempts++
598    // The transcript says why once; the band waits out a brief drop before showing it.
599    feed.interruption = `Connection interrupted: ${String(failure)} Retrying in ${delay / 1000}s.`
600    setConnection($, 'retrying')
601    if (attempts === 1) await $.ui.log(feed.interruption, { to: 'transcript' })
602    if (!ownsObserver()) return
603    try { await $.clock.sleep(delay, { signal }) }
604    catch (error) { if (!signal.aborted) throw error }
605  }
606}
607
608/** Apply one change to the thread's items, turn state, or presence. */
609function observeRemote($: Runtime, session: string, event: WatchEvent) {
610  const view = feed.view
611  if (view && (event.type === 'item' || event.type === 'item.append' || event.type === 'remove')) {
612    const id = event.type === 'item' ? event.item.id : event.id
613    if (event.type === 'remove') view.remove(event.id)
614    else {
615      const before = view.items.get(id)
616      const item = event.type === 'item' ? event.item : view.appendText(event.id, event.text)
617      // Thinking never becomes a row for Claude, so it is shown to the person here, once, when it completes live.
618      if (announceThought(before, item, view.snapshotDone)) {
619        const display = itemDisplay(item, view.thread)
620        if (display) showToPerson($, display)
621      }
622      if (event.type === 'item') view.upsert(item)
623    }
624    if (!view.items.get(id)?.open) feed.openSince.delete(id)
625    else if (!feed.openSince.has(id)) void $.clock.now().then((now) => {
626      if (view.items.get(id)?.open && !feed.openSince.has(id)) feed.openSince.set(id, now)
627    })
628    // Activity keeps another participant's running turn from going quiet.
629    const turn = feed.remoteTurn
630    if (turn) void $.clock.now().then((now) => { turn.heard = now })
631    void settleRemote($, session)
632    $.ui.invalidate('ui.render')
633  } else if (view && (event.type === 'comment' || event.type === 'comment.remove')) {
634    if (event.type === 'comment') view.upsertComment(event.comment)
635    else view.removeComment(event.id)
636    void settleRemote($, session)
637    $.ui.invalidate('ui.render')
638  } else if (event.type === 'turn') {
639    feed.threadTurn = event
640    observeThreadTurn($, session)
641  } else if (event.type === 'presence') {
642    // A live presence report supersedes a finished operation's notice.
643    clearDoneNotice()
644    presence.update(event.users, (milliseconds, action) => $.clock.after(milliseconds, action),
645      () => $.ui.invalidate('ui.render'))
646    $.ui.invalidate('ui.render')
647  } else if (event.type === 'children' || event.type === 'child' || event.type === 'child.remove') {
648    if (event.type === 'children') {
649      feed.children = event.children
650    } else if (event.type === 'child') {
651      const index = feed.children.findIndex((child) => child.thread_id === event.child.thread_id)
652      if (index < 0) feed.children.push(event.child)
653      else feed.children[index] = event.child
654    } else {
655      feed.children = feed.children.filter((child) => child.thread_id !== event.thread_id)
656    }
657    syncSpinner($)
658    $.ui.invalidate('ui.render')
659  }
660}
661
662/** Each new watch connection re-reports who is here; its first report is not a set of arrivals. */
663function resetPresence() {
664  presence.reset()
665}
666
667/**
668 * Whether the thread's running turn is this producer's own. Producer and thread turn events are
669 * separate streams, so either can arrive first: a start not yet reported is ours, and a turn
670 * already known as ours stays ours after it finishes, until the thread reports a different one.
671 * A turn already known as another participant's is never adopted, even while this producer's
672 * own start is being reported for the turn that takes it over.
673 */
674function ownsThreadTurn(session: string, turn: Extract<WatchEvent, { type: 'turn' }>) {
675  const execution = stateFor(session).binding?.execution
676  const own = threadTurnOwned({ ...feed, execution }, turn)
677  if (own && turn.turn_id) feed.ownTurnId = turn.turn_id
678  return own
679}
680
681function observeThreadTurn($: Runtime, session: string) {
682  const turn = feed.threadTurn
683  const remote = turn?.state === 'running' && !ownsThreadTurn(session, turn)
684  const previous = feed.remoteTurn
685  // A different running turn, whoever started it, took the thread over from the previous one.
686  const replaced = turn?.state === 'running' && turn.turn_id !== previous?.turnId && (!remote || !!previous?.turnId)
687  // The band reports the turn; its end simply returns the band to the thread's next status.
688  if (previous && (!remote || replaced)) feed.remoteTurn = undefined
689  if (remote && !feed.remoteTurn) {
690    const name = conversationText(turn!.user?.name ?? '').trim()
691    const started: RemoteTurn = { turnId: turn!.turn_id, name, ...turn!.user ? { id: personKey(turn!.user) } : {} }
692    feed.remoteTurn = started
693    // A new turn counts its time, and its quiet, from when it is first reported.
694    void $.clock.now().then((now) => { started.since ??= now })
695  } else if (remote) feed.remoteTurn!.turnId ??= turn!.turn_id
696  refreshTurn($)
697}
698
699/**
700 * The current turn, from the watch's two independent streams. A thread turn whose `turn_id` is not
701 * yet known is still starting; otherwise it belongs to another participant when their running turn
702 * was announced, and to this session when it was not. A producer start can reach this session
703 * before the thread reports the turn it belongs to.
704 */
705function currentTurn(): TurnStatus {
706  return statusCurrentTurn(feed)
707}
708
709/** Recompute the band's turn status and let the animation follow it. */
710function refreshTurn($: Runtime) {
711  // A real turn change supersedes a finished operation's notice.
712  clearDoneNotice()
713  feed.turn = currentTurn()
714  syncSpinner($)
715  $.ui.invalidate('ui.render')
716}
717
718/** Drop a finished operation's notice so the band can show what happened next. */
719function clearDoneNotice() {
720  if (feed.progress?.tone !== 'done') return
721  feed.progressTimer?.cancel()
722  feed.progressTimer = undefined
723  feed.progress = undefined
724}
725
726/**
727 * Record the watcher's connection state; a redraw and the band's animation follow it. A live watch
728 * that drops starts the grace in which the band still looks live; reconnecting attempts keep it
729 * running, and only a live or stopped watch ends it.
730 */
731function setConnection($: Runtime, connection: typeof feed.connection) {
732  if (connection === 'live' || connection === 'stopped') clearReconnect()
733  else if (connection === 'retrying' && feed.connection === 'live') {
734    const timer = $.clock.after(RECONNECT_GRACE_MS, () => {
735      if (feed.reconnectTimer !== timer) return
736      feed.reconnectTimer = undefined
737      feed.reconnecting = true
738      $.ui.invalidate('ui.render')
739    })
740    feed.reconnectTimer = timer
741  }
742  feed.connection = connection
743  syncSpinner($)
744  $.ui.invalidate('ui.render')
745}
746
747function clearReconnect() {
748  feed.reconnectTimer?.cancel()
749  feed.reconnectTimer = undefined
750  feed.reconnecting = undefined
751}
752
753/** Run the band's redraw timer only while a live watch has a running turn, or an operation runs. */
754function syncSpinner($: Runtime) {
755  if (!spinnerRunning()) return stopSpinner()
756  if (feed.spinner) return
757  const timer = $.clock.every(SPINNER_INTERVAL_MS, () => {
758    const spinner = feed.spinner
759    if (!spinner || spinner.timer !== timer) return
760    spinner.frame = (spinner.frame + 1) % SPINNER_FRAMES.length
761    $.ui.invalidate('ui.render')
762  })
763  feed.spinner = { frame: 0, timer }
764}
765
766/** A running share or join animates wherever it runs; a turn animates only on a live watch. */
767function spinnerRunning() {
768  return statusSpinnerRunning({ ...feed, progress: currentProgress() })
769}
770
771function stopSpinner() {
772  feed.spinner?.timer.cancel()
773  feed.spinner = undefined
774}
775
776/** Record what a share or join is doing and redraw; the band's animation follows it. */
777function setProgress($: Runtime, text: string, tone: ProgressTone, step?: [number, number]) {
778  feed.progressTimer?.cancel()
779  feed.progressTimer = undefined
780  feed.progress = { text, tone, step }
781  syncSpinner($)
782  $.ui.invalidate('ui.render')
783}
784
785/** Finish an operation: a success clears after BAND_NOTICE_MS, a failure stays until replaced. */
786function finishProgress($: Runtime, text: string, tone: 'done' | 'error') {
787  setProgress($, text, tone)
788  if (tone !== 'done') return
789  const timer = $.clock.after(BAND_NOTICE_MS, () => {
790    if (feed.progressTimer !== timer) return
791    feed.progressTimer = undefined
792    feed.progress = undefined
793    syncSpinner($)
794    $.ui.invalidate('ui.render')
795  })
796  feed.progressTimer = timer
797}
798
799/** Drop an operation's band line, as a torn-down session does. */
800function clearProgress($: Runtime) {
801  feed.progressTimer?.cancel()
802  feed.progressTimer = undefined
803  feed.progress = undefined
804  syncSpinner($)
805  $.ui.invalidate('ui.render')
806}
807
808/** Whether the band has a Delta status to draw: a live watch, a share or join step, or a failure. */
809function statusShown(): boolean {
810  return !!feed.session || !!feed.progress || !!feed.error
811}
812
813/**
814 * What another participant's turn is doing: its newest thought or tool call open long enough to
815 * show, how long that has been open, and how many of its other calls are.
816 */
817function remoteActivity(remote: RemoteTurn, now: number): NonNullable<StatusState['agent']>['doing'] {
818  const shown = [...feed.openSince].flatMap(([id, since]) => {
819    const item = feed.view?.items.get(id)
820    const current = !remote.turnId || !item?.turn_id || item.turn_id === remote.turnId
821    return item?.open && current && (item.kind === 'tool_call' || item.kind === THOUGHT_KIND) &&
822      now - since >= SHOW_ACTIVITY_AFTER_MS ? [{ item, since }] : []
823  })
824  const newest = shown.reduce<typeof shown[number] | undefined>(
825    (latest, entry) => !latest || entry.since >= latest.since ? entry : latest, undefined)
826  if (!newest) return
827  const tool = newest.item.kind === 'tool_call'
828  return {
829    text: tool ? toolCallParts(newest.item).name : 'thinking', tool,
830    seconds: (now - newest.since) / 1000,
831    more: shown.filter((entry) => entry !== newest && entry.item.kind === 'tool_call').length,
832  }
833}
834
835/** What the band shows, at the moment it draws. */
836async function statusView($: Runtime): Promise<StatusState> {
837  const state = feed.session ? sessions.get(feed.session) : undefined
838  const remote = feed.turn.state === 'remote' ? feed.remoteTurn : undefined
839  const now = await $.clock.now()
840  const since = remote?.since ?? now
841  const silent = now - (remote?.heard ?? since)
842  // A turn with an item still open, such as a long command, is working however long it is silent.
843  const open = [...feed.view?.items.values() ?? []].some((item) => item.open)
844  const doing = remote ? remoteActivity(remote, now) : undefined
845  const binding = state?.binding
846  // The visibility label names an organization; read the directory once, and a later redraw fills it.
847  if (binding?.visibility === 'organization' || binding?.upload?.organization_id) void loadOrganizationDirectory($)
848  return {
849    ...feed,
850    // Within the grace after a drop, the band carries on as if the watch were live.
851    connection: feed.reconnectTimer ? 'live' : feed.connection,
852    deliveryPause: state?.pauses.get('delivery') ?? state?.pauses.get('verification'),
853    retryReason: state?.retryTimer ? state.retryReason : undefined,
854    progress: currentProgress(),
855    link: statusLink(),
856    audience: audienceLabel(binding),
857    // A refused offline thread keeps its controls until the person forks the changes or discards them.
858    conflict: state?.refusedOffline ? { fork: 'Fork & Restore', discard: 'Discard' } : undefined,
859    // Sharing waits for Claude's turn to finish, so the control says so rather than ignoring a press.
860    share: binding?.visibility === 'private' ? {
861      label: state?.currentTurnId ? 'Share after this turn' : `Share with ${organizationName(binding.upload?.organization_id) ?? 'organization'}`,
862    } : undefined,
863    people: presence.roster(),
864    children: feed.children,
865    childrenStale: feed.connection !== 'live',
866    agent: remote ? {
867      seconds: (now - since) / 1000,
868      ...!open && silent >= QUIET_AFTER_MS ? { quiet: silent / 1000 } : {},
869      ...doing ? { doing } : {},
870    } : undefined,
871    copied: !!feed.copiedTimer,
872  }
873}
874
875/** Who can see the thread, apart from whether it has a link: `Private`, `Shared with <org>`, or `Unlisted`. */
876function audienceLabel(binding: Binding | undefined): string | undefined {
877  if (!binding?.visibility) return undefined
878  if (binding.visibility === 'private') return 'Private'
879  if (binding.visibility === 'unlisted') return 'Unlisted'
880  return `Shared with ${organizationName(binding.upload?.organization_id) ?? 'your organization'}`
881}
882
883/**
884 * Delta's background upload of an automatic session, drawn on the band's row like a share's steps:
885 * its first upload until Delta confirms it, then only a failure. Activation only requests the upload,
886 * and Delta registers the thread a moment before it confirms the content, so the link is offered
887 * once the upload is confirmed; later per-turn syncs are not shown.
888 */
889function uploadProgress(binding: Binding | undefined): StatusState['progress'] {
890  if (!binding?.autoUpload || !binding.thread) return undefined
891  const upload = binding.upload
892  if (upload?.state === 'failed')
893    return { text: `Upload failed: ${upload.error?.message ?? upload.error?.reason ?? 'unknown reason'}`, tone: 'error' }
894  if (binding.uploaded) return undefined
895  // Delta reports a failure it retries itself as queued with its error, which is progress, not a failure.
896  if (upload?.state === 'queued' && upload.error)
897    return { text: `Retrying upload: ${upload.error.message || upload.error.reason || upload.error.code}`, tone: 'active' }
898  return { text: upload?.state === 'queued' ? 'Waiting to upload…' : 'Uploading to Delta…', tone: 'active' }
899}
900
901/** The operation the band draws: a share, join, or activation step, else an automatic session's upload. */
902function currentProgress(): StatusState['progress'] {
903  return feed.progress ?? uploadProgress(feed.session ? sessions.get(feed.session)?.binding : undefined)
904}
905
906/**
907 * The shared thread's link: the watched session's saved binding, which every status inspection keeps
908 * current. A URL says a thread exists, not that others can see it; the audience is shown apart.
909 * Absent until status reports one, and while a share or join still runs the band shows it dim, without
910 * offering it.
911 */
912function statusLink(): string | undefined {
913  if (!feed.session) return
914  return conversationWebLink(sessions.get(feed.session)?.binding?.url)
915}
916
917/** The status bar's shade for each kind of theme: a step off the terminal's own background. */
918const BAR_SHADES = { dark: '#303030', light: '#e8e8e8' }
919/** The shade for the person's theme, read once and again whenever they change it. */
920let barTheme: keyof typeof BAR_SHADES | undefined
921
922/**
923 * The status bar's shade for the person's theme. A terminal that cannot show it as given rounds it
924 * to a color it has; at worst the shade is lost and the band reads as a plain row.
925 */
926async function barShade($: Runtime): Promise<string> {
927  if (!barTheme) {
928    const theme = (await $.config.list()).find((row) => row.key === 'theme')?.value
929    barTheme = typeof theme === 'string' && theme.startsWith('light') ? 'light' : 'dark'
930  }
931  return BAR_SHADES[barTheme]
932}
933
934/** Copy the link to the surface the press came from, and confirm it only once the clipboard took it. */
935async function copyLink($: Runtime, link: string, surface: RenderSurface) {
936  const result = await $.ui.copy({ text: link, surface })
937  if (!result.isCopied) {
938    await $.ui.log(`Delta: couldn't copy the link: ${result.reason}`, { to: 'debug' })
939    return
940  }
941  feed.copiedTimer?.cancel()
942  const timer = $.clock.after(COPIED_MS, () => {
943    if (feed.copiedTimer !== timer) return
944    feed.copiedTimer = undefined
945    $.ui.invalidate('ui.render')
946  })
947  feed.copiedTimer = timer
948  $.ui.invalidate('ui.render')
949}
950
951/** Only the props a segment actually sets, so the surface never sees an undefined style. */
952function segmentProps(segment: StatusSegment): TextProps {
953  const props: TextProps = {}
954  if (segment.color !== undefined) props.color = segment.color
955  if (segment.bold) props.bold = true
956  if (segment.dim) props.dimColor = true
957  if (segment.inverse) props.inverse = true
958  return props
959}
960
961/** The color the band gives an item's author, so a person reads the same everywhere. */
962function authorColor(item: WatchItem): string {
963  return personColor(personKey(item.author ?? { user_id: null, name: itemAuthor(item) }))
964}
965
966/** Rows wait while Claude's own turn runs or a local prompt is entering, so they never land inside a turn. */
967function rowsHeld(session: string) {
968  const runtime = stateFor(session)
969  return !!runtime.currentTurnId || !!runtime.promptEntering
970}
971
972/**
973 * The transcript draws one `$.ui.log` entry per call and flattens an embedded newline to U+FFFD, so
974 * a person-visible item goes out one entry per paragraph, keeping the line breaks it should have.
975 */
976function showToPerson($: Runtime, text: string) {
977  for (const line of text.split('\n')) $.ui.log(line, { to: 'transcript' })
978}
979
980/**
981 * Draw a completed message as the output row of the plugin's own `show <id>` command, which is the
982 * only plugin-producible transcript row that draws markdown and that redraws with our own tree.
983 *
984 * The engine runs a plugin's own call with the calling hook alone skipped, and a call from the
985 * plugin's background code carries no calling hook, so it skips the plugin's own `command.run` hook
986 * altogether. The row therefore waits for a redraw: a `ui.render` hook is a hook context, and the
987 * command it raises inside one is answered.
988 */
989function drawMessage($: Runtime, session: string, id: string): Promise<{ deny?: string }> {
990  const result = drawQueue.wait(session, id, () => rowsHeld(session),
991    (milliseconds, action) => $.clock.after(milliseconds, action))
992  $.ui.invalidate('ui.render')
993  return result
994}
995
996/**
997 * Abandon the rows a stopped conversation was still waiting to draw. They were not added, so the
998 * next snapshot offers them again; a stopped watch reports nothing about them.
999 */
1000function dropQueuedDraws() {
1001  drawQueue.drop()
1002}
1003
1004/**
1005 * Answer the queued message rows, oldest first, from a hook context. Rows wait while a local turn
1006 * runs, and the queue keeps its order across redraws. Each answer settles the row that queued it.
1007 */
1008async function drainMessages($: Runtime) {
1009  const session = feed.session
1010  if (drawing || !session || !feed.view) return
1011  drawing = true
1012  try {
1013    while (drawQueue.length) {
1014      if (feed.session !== session || rowsHeld(session)) break
1015      const entry = drawQueue.shift()!
1016      let answer: string | undefined
1017      try { answer = (await $.command.run({ command: 'delta', args: `show ${entry.id}` })).text }
1018      catch (error) { entry.settle({ deny: String(error) }); continue }
1019      entry.settle(answer === undefined || UNANSWERED_COMMAND.test(answer)
1020        ? { deny: 'this host did not run the plugin\'s own command' }
1021        : {})
1022    }
1023  } finally {
1024    drawing = false
1025  }
1026  if (drawQueue.length) $.ui.invalidate('ui.render')
1027}
1028
1029/**
1030 * Add one model-visible row, and draw a message's copy for the person beside it.
1031 *
1032 * The model reads the plugin's own `session.append` row for every kind, so a message is drawn for
1033 * the person in addition to it, never instead of it. Neither the drawn row nor the model's row is
1034 * written twice: an item this session already drew is not drawn again, so a retried row repeats
1035 * neither. A host that will not draw the command output row leaves the model's row alone and shows
1036 * the person the plain notice.
1037 */
1038async function appendRow($: Runtime, session: string, row: PendingRow): Promise<{ deny?: string }> {
1039  const id = row.commandItem
1040  const runtime = stateFor(session)
1041  if (id !== undefined) {
1042    const drawing = runtime.drawn.drawOnce(id, () => drawMessage($, session, id))
1043    if (drawing) {
1044      const refusal = await drawing
1045      if (refusal.deny === undefined)
1046        $.ui.log(`Drew the Delta item ${id} as a command output row.`, { to: 'debug' })
1047      else fallbackNotice($, row, refusal.deny)
1048    }
1049  }
1050  return $.session.append({ message: { type: 'user', content: [{ type: 'text', text: row.text }] } })
1051}
1052
1053/** Show a message or comment the plain way, when its command output row is refused; the model's row stands. */
1054function fallbackNotice($: Runtime, row: PendingRow, reason: string) {
1055  const view = feed.view
1056  const commentId = keyedComment(row.id)
1057  const comment = commentId === undefined ? undefined : view?.comments.get(commentId)
1058  const item = view?.items.get(row.id)
1059  const display = !view ? undefined
1060    : comment ? commentDisplay(comment, view.commentParent(comment), view.commentTarget(comment), view.thread)
1061    : item ? itemDisplay(item, view.thread) : undefined
1062  if (display) showToPerson($, display)
1063  $.ui.log(`Couldn't draw this Delta message as a command row (${reason}).`, { to: 'debug' })
1064}
1065
1066/**
1067 * Add completed remote items to Claude's transcript as model-visible rows, without starting a
1068 * turn, and correct rows for items that later changed or disappeared. Held rows resume when the
1069 * prompt is dropped or its turn completes.
1070 */
1071function settleRemote($: Runtime, session: string): Promise<void> | undefined {
1072  if (feed.settling || feed.session !== session) return feed.settling
1073  const revision = feed.revision
1074  const owns = () => feed.revision === revision && feed.session === session
1075  let task: Promise<void> | undefined = undefined
1076  task = (async () => {
1077    // Start after `task` is assigned, so a run with nothing to do still clears `settling`.
1078    await undefined
1079    try {
1080      if (!rowsHeld(session)) await flushNotes($, session)
1081      const view = feed.view
1082      if (view) await addRows(view, {
1083        append: (row) => appendRow($, session, row),
1084        save: (ledger) => $.store.set(ledgerKey(session), ledger),
1085        held: () => rowsHeld(session),
1086        current: owns,
1087        announce: (text) => showToPerson($, text),
1088        debug: (text) => $.ui.log(text, { to: 'debug' }),
1089        fail: (reason) => settleFailure($, reason),
1090      })
1091    } finally {
1092      if (feed.settling === task) feed.settling = undefined
1093    }
1094    if (owns() && feed.view && !feed.view.settled && !rowsHeld(session)) await settleRemote($, session)
1095  })()
1096  feed.settling = task
1097  return task
1098}
1099
1100function settleFailure($: Runtime, reason: string) {
1101  // Print and SDK sessions keep no conversation a plugin can add rows to; nothing retries that.
1102  const notice = /keeps no conversation/.test(reason)
1103    ? `Couldn't add Delta activity to Claude's context: ${reason}. Remote Delta activity isn't available in print or SDK sessions.`
1104    : `Couldn't add Delta activity to Claude's context: ${reason}. It is offered again when the watcher reconnects.`
1105  if (feed.settleNotice === notice) return
1106  feed.settleNotice = notice
1107  $.ui.log(notice)
1108}
1109
1110/** Sync diagnostics can exceed process.run's 4 MiB stdout capture limit, so status streams with a bound. */
1111async function readThreadStatus(
1112  $: Runtime, thread: string, session: string, signal = new AbortController().signal,
1113) {
1114  return parseThreadStatus(await readCommand($, ['thread', 'status', thread,
1115    '--source', 'claude-code', '--external-id', session], 'thread status', signal))
1116}
1117
1118/**
1119 * The thread facts a status read carries into the binding: its link, its audience, and Delta's
1120 * background upload. A URL alone says nothing about who can see the thread, so they stay separate.
1121 */
1122function statusFields(status: ThreadStatus): Pick<Binding, 'url' | 'visibility' | 'upload' | 'uploaded'> {
1123  const url = status.url ?? undefined
1124  const upload = status.upload ?? undefined
1125  return {
1126    url,
1127    visibility: status.visibility ?? undefined,
1128    upload,
1129    // A URL appears when Delta registers the thread, a moment before it confirms the content. The
1130    // milestone outlives this read, so a later per-turn sync does not reopen the first upload.
1131    ...url && upload?.state === 'uploaded' ? { uploaded: true as const } : {},
1132  }
1133}
1134
1135/** Save what a status read says about the thread's link, audience, and upload, and redraw the band. */
1136async function rememberStatus($: Runtime, session: string, status: ThreadStatus) {
1137  if (active !== session) return
1138  let published = false
1139  await serial(async () => {
1140    const current = await load($, session)
1141    if (active !== session || !current.thread) return
1142    const next = { ...current, ...statusFields(status) }
1143    published = !!current.autoUpload && !current.uploaded && !!next.uploaded
1144    await save($, next)
1145  })
1146  // The first upload ends the way a share does, unless a share or join is drawing its own steps.
1147  if (published && feed.session === session && !feed.progress && !sharing && !joining)
1148    finishProgress($, 'Uploaded to Delta', 'done')
1149  else {
1150    syncSpinner($)
1151    $.ui.invalidate('ui.render')
1152  }
1153  followUpload($, session)
1154}
1155
1156/**
1157 * Whether nothing more is expected from Delta's background upload. A session this plugin activated
1158 * requests the upload and returns before Delta registers the thread, so its link only appears in a
1159 * later status read; every other session has its link from the share or join that bound it.
1160 */
1161function uploadSettled(binding: Binding | undefined): boolean {
1162  return !binding?.autoUpload || !!binding.uploaded
1163}
1164
1165/**
1166 * Keep reading status, on a backoff, until Delta has registered and confirmed an automatic
1167 * session's thread, so the band shows its link and audience without waiting for the next delivery.
1168 * Delta keeps retrying the upload itself; this only observes it.
1169 */
1170function followUpload($: Runtime, session: string) {
1171  const state = sessions.get(session)
1172  if (!state || active !== session) return
1173  if (uploadSettled(state.binding)) {
1174    state.uploadRefreshAttempts = 0
1175    return
1176  }
1177  if (state.uploadRefresh) return
1178  const attempt = state.uploadRefreshAttempts ?? 0
1179  state.uploadRefreshAttempts = attempt + 1
1180  state.uploadRefresh = $.clock.after(reconnectDelay(attempt), () => {
1181    state.uploadRefresh = undefined
1182    const binding = state.binding
1183    if (active !== session || !binding?.thread) return
1184    void readThreadStatus($, binding.locator ?? binding.thread, session)
1185      .then((status) => rememberStatus($, session, status))
1186      // An unanswered read says nothing about the upload; ask again after the next delay.
1187      .catch(() => followUpload($, session))
1188  })
1189}
1190
1191/** Sign-in's view of the session; `$` stays in this file, as Claude Code's loader requires. */
1192async function authenticationHost($: Runtime): Promise<AuthenticationHost> {
1193  return {
1194    interactive: hasInteractiveSurface(await $.session.surfaces()),
1195    windows: await $.env.get('OS') === 'Windows_NT',
1196    sleep: (milliseconds, signal) => $.clock.sleep(milliseconds, { signal }),
1197    spawn: (argv) => $.process.spawn({ argv }),
1198  }
1199}
1200

1import type { ApiMessage } from 'claude-code'
2import { bashMarkdown } from './bash'
3
4type RecordValue = Record<string, unknown>
5export type Update = RecordValue & { sessionUpdate: string }
6
7/** Capture authored rows and tool results, not the engine's model-only framing. */
8export function capturedMessage(message: {
9  type: string
10  role?: ApiMessage['role']
11  isMeta?: true
12  content: Readonly<ApiMessage['content']>
13}, door: string): ApiMessage | undefined {
14  if (door === 'command') {
15    if (message.type !== 'user' || message.role !== 'user') return
16    // Bash-mode rows enter through the command door but also belong to Claude's model context.
17    const content = message.content.flatMap((block) => {
18      if (block.type !== 'text' || typeof block.text !== 'string') return []
19      const text = bashMarkdown(block.text)
20      return text !== block.text ? [{ ...block, text }] : []
21    })
22    return content.length ? { role: 'user', content } : undefined
23  }
24  if (['attachment', 'hook-context', 'notice', 'compaction', 'note'].includes(door)) return
25  if (message.type === 'assistant' && message.role === 'assistant')
26    return { role: 'assistant', content: [...message.content] }
27  if (message.type !== 'user' || message.role !== 'user') return
28  if (door === 'tool-result' || door === 'tool-message') {
29    const content = message.content.filter((block) => block.type === 'tool_result')
30    return content.length ? { role: 'user', content } : undefined
31  }
32  if (message.isMeta || !['prompt', 'delivery'].includes(door)) return
33  return { role: 'user', content: message.content.map((block) =>
34    block.type === 'text' && typeof block.text === 'string'
35      ? { ...block, text: bashMarkdown(block.text) } : block) }
36}
37
38function object(value: unknown): RecordValue | undefined {
39  return value !== null && typeof value === 'object' && !Array.isArray(value)
40    ? value as RecordValue : undefined
41}
42
43function text(value: unknown): string {
44  return typeof value === 'string' ? value : JSON.stringify(value) ?? String(value)
45}
46
47function base64(value: unknown): value is string {
48  if (typeof value !== 'string' || value.length % 4 !== 0 ||
49      !/^[A-Za-z0-9+/]*={0,2}$/.test(value)) return false
50  const alphabet = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'
51  if (value.endsWith('==')) return (alphabet.indexOf(value.charAt(value.length - 3)) & 15) === 0
52  if (value.endsWith('=')) return (alphabet.indexOf(value.charAt(value.length - 2)) & 3) === 0
53  return true
54}
55
56function blocks(value: unknown): RecordValue[] {
57  if (typeof value === 'string') return [{ type: 'text', text: value }]
58  if (!Array.isArray(value)) return value == null ? [] : blocks([value])
59  return value.map((item) => {
60    const block = object(item)
61    if (!block) return { type: 'text', text: text(item) }
62    if (!validOptionalFields(block)) return { type: 'text', text: text(block) }
63    if (block.type === 'text' && typeof block.text === 'string')
64      return { type: 'text', text: block.text, ...contentFields(block) }
65    if (block.type === 'image') {
66      const source = object(block.source)
67      if (source?.type === 'base64' && base64(source.data) && typeof source.media_type === 'string')
68        return { type: 'image', data: source.data, mimeType: source.media_type, ...contentFields(block) }
69      if (base64(block.data) && typeof block.mimeType === 'string')
70        return { type: 'image', data: block.data, mimeType: block.mimeType, ...contentFields(block) }
71      if (source?.type === 'url' && typeof source.url === 'string')
72        return { type: 'resource_link', uri: source.url, name: source.url, ...contentFields(block) }
73    }
74    if (block.type === 'resource_link' && typeof block.uri === 'string')
75      return {
76        type: 'resource_link', uri: block.uri, name: text(block.name ?? block.uri),
77        ...optionalFields(block, ['title', 'description', 'icons', 'mimeType', 'size', 'annotations', '_meta']),
78      }
79    if (block.type === 'audio' && base64(block.data) && typeof block.mimeType === 'string')
80      return { type: 'audio', data: block.data, mimeType: block.mimeType, ...contentFields(block) }
81    if (block.type === 'resource') {
82      const resource = object(block.resource)
83      if (resource && validOptionalFields(resource) && typeof resource.uri === 'string' &&
84          (typeof resource.text === 'string' || base64(resource.blob)))
85        return {
86          type: 'resource',
87          resource: {
88            uri: resource.uri,
89            ...(typeof resource.text === 'string' ? { text: resource.text } : { blob: resource.blob }),
90            ...optionalFields(resource, ['mimeType', '_meta']),
91          },
92          ...contentFields(block),
93        }
94    }
95    if (block.type === 'document') {
96      const source = object(block.source)
97      const name = typeof block.title === 'string' ? block.title : 'document'
98      if (source?.type === 'url' && typeof source.url === 'string')
99        return { type: 'resource_link', uri: source.url, name, ...contentFields(block) }
100      if (source && typeof source.media_type === 'string' &&
101          ((source.type === 'base64' && base64(source.data)) ||
102           (source.type === 'text' && typeof source.data === 'string')))
103        return {
104          type: 'resource',
105          resource: {
106            uri: `urn:claude-code:document:${encodeURIComponent(name)}`,
107            mimeType: source.media_type,
108            ...(source.type === 'base64' ? { blob: source.data } : { text: source.data }),
109          },
110          ...contentFields(block),
111        }
112    }
113    // Unknown content remains visible and recoverable rather than silently disappearing.
114    return { type: 'text', text: text(block) }
115  })
116}
117
118function optionalFields(value: RecordValue, fields: string[]): RecordValue {
119  return Object.fromEntries(fields.filter((field) => value[field] !== undefined)
120    .map((field) => [field, value[field]]))
121}
122
123function contentFields(value: RecordValue): RecordValue {
124  return optionalFields(value, ['annotations', '_meta'])
125}
126
127/** Malformed optional data must remain opaque rather than poison an immutable outbox record. */
128function validOptionalFields(value: RecordValue): boolean {
129  for (const name of ['title', 'description', 'mimeType', 'lastModified']) {
130    if (value[name] != null && typeof value[name] !== 'string') return false
131  }
132  if (value._meta != null && !object(value._meta)) return false
133  if (value.size != null && (!Number.isSafeInteger(value.size) || (value.size as number) < 0)) return false
134  if (value.icons != null) {
135    if (!Array.isArray(value.icons) || !value.icons.every((item) => {
136      const icon = object(item)
137      return icon && Object.keys(icon).every((key) => ['src', 'mimeType', 'sizes', 'theme'].includes(key)) &&
138        typeof icon.src === 'string' && (icon.mimeType == null || typeof icon.mimeType === 'string') &&
139        (icon.sizes == null || (Array.isArray(icon.sizes) && icon.sizes.every((size) => typeof size === 'string'))) &&
140        (icon.theme == null || icon.theme === 'light' || icon.theme === 'dark')
141    })) return false
142  }
143  if (value.annotations != null) {
144    const annotations = object(value.annotations)
145    if (!annotations || !Object.keys(annotations).every((key) =>
146      ['audience', 'priority', 'lastModified', '_meta'].includes(key))) return false
147    if (!validOptionalFields({ _meta: annotations._meta, lastModified: annotations.lastModified })) return false
148    if (annotations.audience != null && (!Array.isArray(annotations.audience) ||
149        !annotations.audience.every((role) => role === 'user' || role === 'assistant'))) return false
150    if (annotations.priority != null && (typeof annotations.priority !== 'number' ||
151        !Number.isFinite(annotations.priority) || annotations.priority < 0 || annotations.priority > 1)) return false
152  }
153  return true
154}
155
156/** `path` relative to the directory `root`, or undefined when it lies outside it. */
157function relativePath(path: string, root?: string): string | undefined {
158  const base = root?.replace(/\/+$/, '')
159  return base && path.startsWith(`${base}/`) ? path.slice(base.length + 1) : undefined
160}
161
162function editDiff(name: string, result?: RecordValue, root?: string): RecordValue[] {
163  if ((name !== 'Edit' && name !== 'Write') || !result) return []
164  const path = result.filePath
165  if (typeof path !== 'string' || !path.startsWith('/') || /[\r\n]/.test(path)) return []
166  const hunks = result?.structuredPatch
167  const structured = Array.isArray(hunks) && hunks.length > 0 && hunks.every((hunk) => {
168    const value = object(hunk)
169    return value && Number.isInteger(value.oldStart) && (value.oldStart as number) >= 0 &&
170      Number.isInteger(value.newStart) && (value.newStart as number) >= 0 &&
171      Number.isInteger(value.oldLines) && (value.oldLines as number) >= 0 &&
172      Number.isInteger(value.newLines) && (value.newLines as number) >= 0 &&
173      Array.isArray(value.lines) && value.lines.every((line) => typeof line === 'string')
174  })
175  const operation = result?.type === 'create' ? 'add' : 'modify'
176  // ACP names changed files by absolute path; the patch headers stay relative to the checkout.
177  const change = { operation, path }
178  const shown = relativePath(path, root) ?? path.slice(1)
179  if (!structured) return [{ type: 'diff', changes: [change] }]
180  const patch = (hunks as RecordValue[]).map((hunk) =>
181    `@@ -${hunk.oldStart},${hunk.oldLines} +${hunk.newStart},${hunk.newLines} @@\n${(hunk.lines as string[]).join('\n')}`).join('\n')
182  return [{
183    type: 'diff',
184    changes: [change],
185    patch: {
186      format: 'git_patch',
187      text: `diff --git a/${shown} b/${shown}\n${operation === 'add' ? 'new file mode 100644\n' : ''}--- ${operation === 'add' ? '/dev/null' : `a/${shown}`}\n+++ b/${shown}\n${patch}\n`,
188    },
189  }]
190}
191
192/**
193 * The ACP title and kind of a Claude Code tool call, as Claude's ACP adapter reports them, so a
194 * recorded call reads as it does when Delta runs Claude itself. Ported from `toolInfoFromToolUse`
195 * in @agentclientprotocol/claude-agent-acp (Apache-2.0), keeping only the title and kind: the
196 * call's content comes from its result. Paths inside `root` are shown relative to it.
197 */
198export function toolInfo(name: string, input: unknown, root?: string): { title: string; kind: string } {
199  const value = object(input) ?? {}
200  const field = (key: string) => typeof value[key] === 'string' && value[key] ? value[key] as string : undefined
201  const path = (key: string) => {
202    const full = field(key)
203    return full && (relativePath(full, root) ?? full)
204  }
205  const info = (title: string, kind: string) => ({ title: title.replace(/\s+/g, ' ').trim(), kind })
206  switch (name) {
207    case 'Agent':
208    case 'Task':
209      return info(field('description') ?? 'Task', 'think')
210    case 'Bash':
211      return info(field('command') ?? 'Terminal', 'execute')
212    case 'Read': {
213      const offset = typeof value.offset === 'number' && value.offset ? value.offset : undefined
214      const limit = typeof value.limit === 'number' && value.limit > 0 ? value.limit : undefined
215      const lines = limit ? ` (${offset ?? 1} - ${(offset ?? 1) + limit - 1})`
216        : offset ? ` (from line ${offset})` : ''
217      return info(`Read ${path('file_path') ?? 'File'}${lines}`, 'read')
218    }
219    case 'Write':
220    case 'Edit': {
221      const file = path('file_path')
222      return info(file ? `${name} ${file}` : name, 'edit')
223    }
224    case 'Glob':
225      return info(['Find', ...[path('path'), field('pattern')].flatMap((part) => part ? [`\`${part}\``] : [])].join(' '), 'search')
226    case 'Grep': {
227      const parts = ['grep']
228      if (value['-i']) parts.push('-i')
229      if (value['-n']) parts.push('-n')
230      for (const context of ['-A', '-B', '-C'])
231        if (value[context] !== undefined) parts.push(`${context} ${value[context]}`)
232      if (value.output_mode === 'files_with_matches') parts.push('-l')
233      else if (value.output_mode === 'count') parts.push('-c')
234      if (value.head_limit !== undefined) parts.push(`| head -${value.head_limit}`)
235      if (field('glob')) parts.push(`--include="${field('glob')}"`)
236      if (field('type')) parts.push(`--type=${field('type')}`)
237      if (value.multiline) parts.push('-P')
238      if (field('pattern')) parts.push(`"${field('pattern')}"`)
239      if (path('path')) parts.push(path('path')!)
240      return info(parts.join(' '), 'search')
241    }
242    case 'WebFetch': {
243      const url = field('url')
244      return info(url ? `Fetch ${url}` : 'Fetch', 'fetch')
245    }
246    case 'WebSearch': {
247      const domains = (key: string, label: string) => {
248        const list = value[key]
249        return Array.isArray(list) && list.length ? ` (${label}: ${list.join(', ')})` : ''
250      }
251      return info(`"${field('query') ?? ''}"${domains('allowed_domains', 'allowed')}${domains('blocked_domains', 'blocked')}`, 'fetch')
252    }
253    case 'TodoWrite':
254      return info(Array.isArray(value.todos)
255        ? `Update TODOs: ${value.todos.map((todo) => text(object(todo)?.content ?? todo)).join(', ')}`
256        : 'Update TODOs', 'think')
257    default:
258      return info(name || 'Unknown Tool', 'other')
259  }
260}
261
262/** Delta has no plan mode, so Claude's plan approvals stay out of its transcript. */
263const OMITTED_TOOLS = new Set(['ExitPlanMode'])
264
265/** Whether calls to the tool `name` are left out of Delta's transcript. */
266export function isOmittedTool(name: unknown): boolean {
267  return typeof name === 'string' && OMITTED_TOOLS.has(name)
268}
269
270/**
271 * Convert normalized Messages API rows while retaining opaque provider thinking metadata.
272 *
273 * Paths in tool titles are shown relative to `root`, and results of the calls in `omitted` are
274 * left out along with the omitted calls themselves.
275 */
276export function messageUpdates(
277  message: ApiMessage, id: string, structuredResults?: ReadonlyMap<string, Update>,
278  { root, omitted }: { root?: string; omitted?: ReadonlySet<string> } = {},
279): Update[] {
280  const updates: Update[] = []
281  let content: RecordValue[] = []
282  let firstIndex = 0
283  const flush = () => {
284    if (!content.length) return
285    updates.push({
286      sessionUpdate: message.role === 'user' ? 'user_message' : 'agent_message',
287      messageId: `${id}:message:${firstIndex}`, content,
288    })
289    content = []
290  }
291  for (const [index, block] of message.content.entries()) {
292    if (block.type === 'text' && typeof block.text === 'string' && !block.text.trim()) continue
293    if (block.type === 'thinking' && typeof block.thinking === 'string' && !block.thinking.trim()) continue
294    if (block.type === 'thinking' || block.type === 'redacted_thinking') {
295      flush()
296      updates.push({
297        sessionUpdate: 'agent_thought',
298        messageId: `${id}:thought:${index}`,
299        content: blocks(block.type === 'thinking' ? block.thinking : '[redacted thinking]'),
300        _meta: { claudeCode: { block } },
301      })
302    } else if (block.type === 'tool_use' && typeof block.id === 'string') {
303      if (isOmittedTool(block.name)) continue
304      flush()
305      const name = text(block.name ?? 'Tool')
306      updates.push({
307        sessionUpdate: 'tool_call_update',
308        toolCallId: block.id,
309        name,
310        ...toolInfo(name, block.input, root),
311        // A call hook may have finished before this input-only row; its result row closes the call.
312        status: 'in_progress',
313        rawInput: block.input ?? {},
314      })
315    } else if (block.type === 'tool_result' && typeof block.tool_use_id === 'string') {
316      if (omitted?.has(block.tool_use_id)) continue
317      flush()
318      const output = block.content ?? block
319      const structured = structuredResults?.get(block.tool_use_id)
320      updates.push({
321        sessionUpdate: 'tool_call_update',
322        toolCallId: block.tool_use_id,
323        status: structured?.status ?? (block.is_error ? 'failed' : 'completed'),
324        rawOutput: structured && 'rawOutput' in structured ? structured.rawOutput : output,
325        content: [
326          ...blocks(output).map((item) => ({ type: 'content', content: item })),
327          ...(Array.isArray(structured?.content) ? structured.content : []),
328        ],
329      })
330    } else {
331      if (!content.length) firstIndex = index
332      content.push(...blocks([block]))
333    }
334  }
335  flush()
336  return updates
337}
338
339export function resultDiff(name: string, result: unknown, root?: string): RecordValue[] {
340  return editDiff(name, object(result), root)
341}
342
343/**
344 * A tool's structured result as a `tool_call_update`. An edit's diff replaces its raw result,
345 * which repeats the change and carries the entire original file.
346 */
347export function resultUpdate(toolCallId: string, name: string, result: unknown, isError: boolean, root?: string): Update {
348  const diff = isError ? [] : resultDiff(name, result, root)
349  return {
350    sessionUpdate: 'tool_call_update', toolCallId,
351    status: isError ? 'failed' : 'completed',
352    rawOutput: diff.length ? null : result,
353    ...(diff.length ? { content: diff } : {}),
354  }
355}
356

1import type { ApiContentBlock } from 'claude-code'
2import type { Update } from './content'
3
4/**
5 * Per-session tool data that links a call hook's structured result to the result row Claude stores.
6 *
7 * The two arrive in either order and each needs the other's data, so an entry is kept until both
8 * have been recorded. `forgetUnpaired` bounds tools whose other half never comes, such as a refused
9 * call that has no hook or an interrupted one that has no row, one turn after they were made.
10 */
11export class ToolRecords {
12  /** Full structured outputs from call hooks, by tool use ID. */
13  readonly structuredResults = new Map<string, Update>()
14  /** Result rows Claude stored, by tool use ID. */
15  readonly toolResults = new Map<string, ApiContentBlock>()
16  /** The turn sequence captured when the tool was called. */
17  readonly toolTurns = new Map<string, number | undefined>()
18  /** Calls left out of Delta's transcript, so their result rows are left out too. */
19  readonly omitted = new Set<string>()
20  private readonly settled = new Map<string, 'call' | 'row'>()
21
22  /** Entries currently retained across all maps. */
23  get size(): number {
24    return this.structuredResults.size + this.toolResults.size + this.toolTurns.size + this.omitted.size +
25      this.settled.size
26  }
27
28  /** Record that the call hook or the result row finished; the second half releases the entry. */
29  settle(id: string, half: 'call' | 'row'): void {
30    if (this.settled.get(id) === (half === 'call' ? 'row' : 'call')) this.forget(id)
31    else this.settled.set(id, half)
32  }
33
34  forget(id: string): void {
35    this.structuredResults.delete(id)
36    this.toolResults.delete(id)
37    this.toolTurns.delete(id)
38    this.omitted.delete(id)
39    this.settled.delete(id)
40  }
41
42  /**
43   * Release unpaired entries from turns before the one that just completed, and any with no turn.
44   * The completed turn's own entries stay, so a row that arrives late still finds its turn and output;
45   * they go when the next turn completes. A tool still running has no settled half and is never released.
46   */
47  forgetUnpaired(completedSequence: number): void {
48    for (const id of [...this.settled.keys()]) {
49      const turn = this.toolTurns.get(id)
50      if (turn === undefined || turn < completedSequence) this.forget(id)
51    }
52  }
53
54  clear(): void {
55    this.structuredResults.clear()
56    this.toolResults.clear()
57    this.toolTurns.clear()
58    this.omitted.clear()
59    this.settled.clear()
60  }
61}
62

1import { parseAppendReceipt } from './thread-commands'
2
3/** Split a durable JSONL record without cutting a UTF-16 surrogate pair. */
4export function outboxChunks(line: string, sequence: number): string[] {
5  if (new TextEncoder().encode(line).length > 16 * 1024 * 1024)
6    throw new Error(`record ${sequence} exceeds Delta's 16 MiB JSONL limit; sharing paused`)
7  const chunks: string[] = []
8  let offset = 0
9  while (offset < line.length) {
10    let end = Math.min(offset + 900_000, line.length)
11    if (end < line.length && /[\uD800-\uDBFF]/.test(line.charAt(end - 1))) end--
12    chunks.push(line.slice(offset, end))
13    offset = end
14  }
15  return chunks
16}
17
18/**
19 * Match only the acknowledged prefix, including each record's original turn association.
20 * A `fenced` receipt names the turn it was refused for, so a caller can stop that turn's local
21 * execution without waiting for a watch event.
22 */
23export function acknowledgedPrefix(
24  stdout: string, thread: string,
25  records: readonly { sequence: number; turnSequence?: number }[],
26): { count: number; fenced: number[] } {
27  let count = 0
28  const fenced: number[] = []
29  for (const answer of stdout.trim().split('\n')) {
30    const expected = records[count]
31    if (!expected) break
32    try {
33      const receipt = parseAppendReceipt(answer, thread, expected.sequence)
34      if (receipt.turn_sequence !== (expected.turnSequence ?? null)) break
35      if (receipt.disposition === 'fenced' && receipt.turn_sequence !== null) fenced.push(receipt.turn_sequence)
36    } catch { break }
37    count++
38  }
39  return { count, fenced }
40}
41
42/** Add a complete record unless it would overflow a nonempty append batch. */
43export function appendBatch(
44  batch: { input: string; bytes: number; count: number }, line: string,
45): { input: string; bytes: number; count: number } | undefined {
46  const bytes = new TextEncoder().encode(line).length
47  if (batch.count && batch.bytes + bytes > 2_000_000) return
48  return { input: batch.input + line, bytes: batch.bytes + bytes, count: batch.count + 1 }
49}
50
51/** Translate private historical envelopes without rewriting their immutable journal files. */
52export function appendOutboxRecord(line: string, sequence: number): { line: string; turnSequence?: number } {
53  const record = JSON.parse(line)
54  const invalid = () => new Error(`invalid pending record ${sequence}; sharing paused without changing its record`)
55  // Every saved envelope was written with JSON.stringify plus a newline. Requiring
56  // that representation also rejects duplicate keys before translating old controls.
57  if (!record || line !== JSON.stringify(record) + '\n' ||
58      record.schemaVersion !== 1 || record.sequence !== sequence ||
59      !Number.isSafeInteger(sequence) || sequence < 1) throw invalid()
60  const keys = Object.keys(record).sort().join()
61  if (Object.hasOwn(record, 'turn')) {
62    const turn = record.turn
63    if (keys !== 'schemaVersion,sequence,turn' || !turn ||
64        (turn.action === 'claim' ? Object.keys(turn).join() !== 'action'
65          : turn.action !== 'end' || Object.keys(turn).sort().join() !== 'action,claim' ||
66            !Number.isSafeInteger(turn.claim) || turn.claim < 1 || turn.claim >= sequence)) throw invalid()
67    const turnSequence = turn.action === 'claim' ? sequence : turn.claim
68    return {
69      line: JSON.stringify({
70        schemaVersion: 1, sequence, turnSequence,
71        update: { sessionUpdate: 'state_update', state: turn.action === 'claim' ? 'running' : 'idle' },
72      }) + '\n',
73      turnSequence,
74    }
75  }
76  if (!['schemaVersion,sequence,update', 'claim,schemaVersion,sequence,update', 'schemaVersion,sequence,turnSequence,update'].includes(keys) ||
77      !record.update || Array.isArray(record.update) || typeof record.update.sessionUpdate !== 'string') throw invalid()
78  const legacy = Object.hasOwn(record, 'claim')
79  const turnSequence = legacy ? record.claim : record.turnSequence
80  if ((legacy || Object.hasOwn(record, 'turnSequence')) &&
81      (!Number.isSafeInteger(turnSequence) || turnSequence < 1 ||
82        (turnSequence >= sequence && !(turnSequence === sequence &&
83          record.update.sessionUpdate === 'state_update' && record.update.state === 'running' && !legacy)))) throw invalid()
84  if (!legacy) return { line, turnSequence }
85  const { claim, ...content } = record
86  // Old state updates were passive even when associated with output. Only old
87  // explicit controls may become lifecycle; an old idle must not finish a turn.
88  if (record.update.sessionUpdate === 'state_update')
89    return { line: JSON.stringify(content) + '\n' }
90  return { line: JSON.stringify({ ...content, turnSequence: claim }) + '\n', turnSequence }
91}
92
93/** The durable pending prefix is the authority for which record files can be removed. */
94export function acknowledgedOutboxFiles(
95  directory: string,
96  next: number,
97  pending: readonly { sequence: number; files: readonly string[] }[],
98  entries: readonly { name: string; kind: string; isLink: boolean }[],
99): string[] {
100  if (!Number.isSafeInteger(next) || next < 1) throw new Error('Invalid outbox sequence frontier')
101  let frontier = next
102  const protectedFiles = new Set<string>()
103  for (const entry of pending) {
104    if (!Number.isSafeInteger(entry.sequence) || entry.sequence < 1 || entry.sequence >= next)
105      throw new Error('Invalid pending outbox sequence')
106    frontier = Math.min(frontier, entry.sequence)
107    for (const file of entry.files) protectedFiles.add(file)
108  }
109  return entries.flatMap((entry) => {
110    const match = /^([1-9][0-9]*)-(0|[1-9][0-9]*)\.json$/.exec(entry.name)
111    if (!match || entry.kind !== 'file' || entry.isLink) return []
112    const sequence = Number(match[1])
113    const path = `${directory}/${entry.name}`
114    return Number.isSafeInteger(sequence) && sequence < frontier && !protectedFiles.has(path) ? [path] : []
115  })
116}
117
118/** Delete only explicit files; neither platform follows a shell-expanded path or recurses. */
119export function outboxRemoval(files: readonly string[], windows: boolean): { argv: string[]; stdin?: string } {
120  if (!files.length) throw new Error('No acknowledged files selected')
121  if (!windows) return { argv: ['rm', '-f', '--', ...files] }
122  return {
123    argv: ['powershell.exe', '-NoLogo', '-NoProfile', '-NonInteractive', '-Command',
124      '[Console]::InputEncoding = [Text.Encoding]::UTF8; $ErrorActionPreference = "Stop"; ' +
125      '$paths = ConvertFrom-Json ([Console]::In.ReadToEnd()); foreach ($path in $paths) { [IO.File]::Delete($path) }'],
126    stdin: JSON.stringify(files),
127  }
128}
129

1import { DeltaCommandError, commandErrorFromEnvelope } from './thread-commands'
2
3type Exit = { code: number | null; signal: string | null }
4type Output = { stream: 'stdout' | 'stderr'; text: string }
5type Stream = AsyncIterator<Output, Exit>
6
7/** Collect a CLI response, closing its child on timeout, cancellation, or completion. */
8export async function readCommandStream(
9  stream: Stream, name: string, signal: AbortSignal,
10  sleep: (milliseconds: number, signal: AbortSignal) => Promise<unknown>,
11  timeoutMs = 120_000,
12  acceptsExit = (code: number | null, _stdout: string) => code === 0,
13): Promise<string> {
14  const encoder = new TextEncoder()
15  const limit = 16 * 1024 * 1024
16  const stop = () => { void stream.return?.({ code: null, signal: 'SIGTERM' }).catch(() => {}) }
17  signal.addEventListener('abort', stop, { once: true })
18  const deadline = new AbortController()
19  let timedOut = false
20  void sleep(timeoutMs, deadline.signal).then(() => {
21    timedOut = true
22    stop()
23  }).catch(() => {})
24  const chunks: string[] = []
25  let bytes = 0
26  let stderr = ''
27  try {
28    while (true) {
29      let piece: Awaited<ReturnType<Stream['next']>>
30      try { piece = await stream.next() }
31      catch (error) {
32        if (signal.aborted) throw new Error(`Delta ${name} cancelled`)
33        throw new DeltaCommandError('process_failed', `${name}: ${String(error)}`)
34      }
35      if (signal.aborted) throw new Error(`Delta ${name} cancelled`)
36      if (timedOut) throw new DeltaCommandError('timeout', `Delta ${name} timed out`)
37      if (piece.done) {
38        const stdout = chunks.join('')
39        if (!acceptsExit(piece.value.code, stdout)) {
40          let failure: DeltaCommandError | undefined
41          try { failure = commandErrorFromEnvelope(JSON.parse(stdout)?.error, name) } catch {}
42          throw failure ?? new DeltaCommandError('process_failed',
43            `${name}: ${stderr || `exit ${piece.value.code ?? piece.value.signal ?? 'unknown'}`}`)
44        }
45        return stdout
46      }
47      if (piece.value.stream === 'stderr') {
48        stderr = (stderr + piece.value.text.slice(-4000)).slice(-4000)
49        continue
50      }
51      if (piece.value.text.length > limit - bytes)
52        throw new Error(`Delta ${name} exceeds the transport limit`)
53      bytes += encoder.encode(piece.value.text).byteLength
54      if (bytes > limit) throw new Error(`Delta ${name} exceeds the transport limit`)
55      if (chunks.length && chunks[chunks.length - 1]!.length + piece.value.text.length <= 64 * 1024)
56        chunks[chunks.length - 1] += piece.value.text
57      else chunks.push(piece.value.text)
58    }
59  } finally {
60    deadline.abort()
61    signal.removeEventListener('abort', stop)
62    await stream.return?.({ code: null, signal: 'SIGTERM' })
63  }
64}
65

1import { classifyCommandFailure, type DeltaCommandError } from './thread-commands'
2
3export type FailureStreak = {
4  deliveryNotice?: string
5}
6
7/** Decide retries and pauses without touching timers, UI, or queued records. */
8export function deliveryFailure(
9  state: FailureStreak, error: DeltaCommandError, verificationPaused: boolean,
10) {
11  const retry = classifyCommandFailure(error) !== 'permanent'
12  const notice = `${retry}:${error.code}`
13  return {
14    retry,
15    // A failed checkout check pauses verification; a failure the envelope will not retry pauses delivery.
16    pause: verificationPaused ? 'verification' as const : 'delivery' as const,
17    show: state.deliveryNotice !== notice,
18    state: { deliveryNotice: notice },
19  }
20}
21
22/** Server hints lengthen exponential backoff, with a five-minute upper bound. */
23export function deliveryRetryDelay(attempt: number, retryAfterMs = 0): number {
24  const backoff = Math.min(30_000, 2000 * 2 ** Math.min(attempt, 4))
25  return Math.min(Math.max(retryAfterMs, backoff), 300_000)
26}
27

1import type { ApiMessage } from 'claude-code'
2import { bashMarkdown } from './bash'
3
4const framingTags = 'command-name|command-message|command-args|local-command-stdout|local-command-stderr|local-command-caveat|system-reminder'
5const framingBlocks = new RegExp(`<(${framingTags})>[\\s\\S]*?</\\1>`, 'g')
6
7/** API snapshots lack doors, so only entire framing blocks can be identified safely. */
8function framingText(text: string): boolean {
9  if (/^\s*\[Delta\](?:\s|$)/.test(text)) return true
10  const remainder = text.replace(framingBlocks, '')
11  return remainder !== text && !remainder.trim()
12}
13
14/**
15 * Normalization appends this block to a user message whose tool results load deferred tools.
16 * Merging a following user message that starts with text adds one newline to it.
17 */
18const toolLoaded = /^Tool loaded\.\n?$/
19
20/** Export only the latest bounded, normalized main conversation, preserving arrival order. */
21export function snapshotMessages(messages: readonly ApiMessage[]): ApiMessage[] {
22  return messages.slice(-4096).flatMap((message) => {
23    if (message.role !== 'user') return [message]
24    const results = message.content.some((block) => block.type === 'tool_result')
25    const content = message.content.filter((block) => block.type !== 'text' ||
26      typeof block.text !== 'string' || !(framingText(block.text) || results && toolLoaded.test(block.text)))
27      .map((block) => {
28        if (block.type !== 'text' || typeof block.text !== 'string') return block
29        const text = bashMarkdown(block.text)
30        return text === block.text ? block : { ...block, text }
31      })
32    if (!content.length) return []
33    return [content.length === message.content.length &&
34      content.every((block, index) => block === message.content[index]) ? message : { ...message, content }]
35  })
36}
37

1import type { MountedThread, ThreadMount, ThreadState } from './thread-commands'
2import { samePath } from './paths'
3
4/** Clearing and resuming switch sessions; only exiting releases their mounts. */
5export function releasesMounts(reason: string): boolean {
6  return ['prompt_input_exit', 'logout', 'other'].includes(reason)
7}
8
9/** Terminal and Desktop hosts provide interactive controls and the status band. */
10export function hasInteractiveSurface(surfaces: readonly string[]): boolean {
11  return surfaces.some((surface) => surface === 'terminal' || surface === 'desktop')
12}
13
14/** Return missing built-ins, never treating a plugin command as a host capability. */
15export function missingJoinCommands(names: readonly string[], commands: readonly { name: string; source: string }[]): string[] {
16  return names.filter((name) => !commands.some((entry) => entry.name === name && entry.source === 'builtin'))
17}
18
19/** Refuse inaccessible or empty threads before clearing the existing session. */
20export function joinPreflight(status: ThreadState): 'access' | 'empty' | undefined {
21  if (status.access.state !== 'granted') return 'access'
22  if (status.worktrees.status === 'available' && !status.worktrees.items.length) return 'empty'
23}
24
25/** Compute changed worktrees and whether the checkout-to-folder transition is needed. */
26export function worktreeChanges(previous: readonly ThreadMount[], directory: string | undefined, mounted: MountedThread) {
27  return {
28    added: mounted.worktrees.filter((tree) => !previous.some((mount) => mount.reference === tree.reference)),
29    removed: previous.filter((mount) => !mounted.worktrees.some((tree) => tree.reference === mount.reference)),
30    moves: directory !== undefined && previous.some((mount) => samePath(mount.directory, directory)) &&
31      !samePath(mounted.directory, directory),
32  }
33}
34
35/** A confirmation cannot proceed while any background agent is still active. */
36export function activeBackgroundAgents(agents: readonly { status: string }[]): boolean {
37  return agents.some((agent) => !['completed', 'failed', 'killed'].includes(agent.status))
38}
39
40/** Hold prompts during an incomplete join, restoration, or between-turn directory move. */
41export function promptGate(state: {
42  joining?: boolean; sharing?: boolean; transition?: boolean; snapshot?: boolean
43  restoring?: boolean; movePending?: boolean; steering?: boolean
44}): 'joining' | 'sharing' | 'transition' | 'restoring' | 'moving' | undefined {
45  if (state.joining) return 'joining'
46  if (state.sharing) return 'sharing'
47  if (state.transition) return 'transition'
48  if (!state.snapshot) return
49  if (state.restoring) return 'restoring'
50  if (state.movePending && !state.steering) return 'moving'
51}
52
53/** Steering belongs only to the exact live local execution. */
54export function ownsSteering(
55  execution: { turnId?: string; lost?: true; endSequence?: number } | undefined, turnId: string,
56): boolean {
57  return !!execution && execution.turnId === turnId && !execution.lost && !execution.endSequence
58}
59

1import { conversationWebLink } from './conversation'
2import { commandErrorFromEnvelope, DeltaCommandError } from './thread-commands'
3import { samePath } from './paths'
4import type { ProcessRunResult } from 'claude-code'
5
6/** Decode one short command result without assuming success from its exit code alone. */
7export function checkResult(result: ProcessRunResult, name: string): Record<string, any> {
8  if (result.isStdoutTruncated || result.isStderrTruncated)
9    throw new Error(`${name}: Delta CLI output truncated; no acknowledgment assumed`)
10  let data: Record<string, any>
11  try { data = JSON.parse(result.stdout.trim()) }
12  catch {
13    throw new Error(`${name}: expected JSON from a current Delta CLI; install matching Delta CLI and app builds (${result.stderr.trim() || `exit ${result.exitCode}`})`)
14  }
15  if (result.exitCode !== 0 || data.error)
16    throw commandErrorFromEnvelope(data.error, name) ??
17      new DeltaCommandError('process_failed', `${name}: ${result.stderr.trim() || `exit ${result.exitCode}`}`)
18  return data
19}
20
21/** Require an idempotent producer creation receipt before saving its thread identity. */
22export function validateCreationReceipt(created: Record<string, any>) {
23  if (created.operation !== 'created' || typeof created.thread_id !== 'string' || !created.thread_id ||
24      typeof created.created !== 'boolean')
25    throw new Error('Invalid thread creation receipt; retry the same session without resetting its outbox')
26}
27
28/** Require the requested checkout and a usable worktree reference. */
29export function validateAttachmentReceipt(attached: Record<string, any>, directory: string) {
30  if (attached.operation !== 'mounted' || typeof attached.reference !== 'string' || !attached.reference.trim() ||
31      typeof attached.directory !== 'string' || !samePath(attached.directory, directory))
32    throw new Error('Invalid checkout attachment receipt; saved records remain queued')
33}
34
35/** Require a safe link to the exact thread being published. */
36export function validatePublicationReceipt(shared: Record<string, any>, thread: string | undefined) {
37  if (shared.operation !== 'shared' || shared.thread_id !== thread ||
38      typeof shared.url !== 'string' || !conversationWebLink(shared.url))
39    throw new Error('Invalid publication receipt; saved records remain queued')
40}
41
42/** Reconnecting backs off exponentially, capped at thirty seconds. */
43export function reconnectDelay(attempt: number): number {
44  return Math.min(30_000, 1000 * 2 ** Math.min(attempt, 5))
45}
46

1import { conversationText, itemAuthor, middleTruncate, toolCallParts } from './conversation'
2import type { ThreadTurn, WatchChild, WatchEvent, WatchItem } from './thread-commands'
3
4export type TurnStatus =
5  | { state: 'idle' }
6  | { state: 'starting'; name?: string; id?: string }
7  | { state: 'remote'; name: string; id?: string }
8  | { state: 'own' }
9export type StatusSegment = { text: string; color?: string; bold?: boolean; dim?: boolean; inverse?: boolean }
10/** Someone in the thread, as the band lists them; `id` is their person key, which keeps their color. */
11export type Person = { id: string; name: string; arriving?: boolean; leaving?: boolean }
12export type StatusState = {
13  connection: 'stopped' | 'connecting' | 'live' | 'retrying'
14  /** A dropped watch that outlasted its grace; within it the band is told the watch is live. */
15  reconnecting?: boolean
16  turn: TurnStatus
17  error?: string
18  progress?: { text: string; tone: 'active' | 'done' | 'error'; step?: [number, number] }
19  spinner?: { frame: number }
20  deliveryPause?: string
21  retryReason?: string
22  /** The thread's share link; absent until a share has published one. */
23  link?: string
24  /** The thread's audience as a label, such as `Private` or `Shared with Zed`; shown apart from the link. */
25  audience?: string
26  /** Controls for a thread Delta refused: fork the rejected local changes into a kept thread, or discard them. */
27  conflict?: { fork: string; discard: string }
28  /** A control that publishes a private thread to the organization, like `/delta share`. */
29  share?: { label: string }
30  people?: Person[]
31  children?: WatchChild[]
32  /** The inventory was last observed on a watch that is no longer live. */
33  childrenStale?: boolean
34  /**
35   * Another participant's running turn: how long it has run, how long it has been silent once that
36   * counts as quiet, and what it is doing now, a thought or tool call open long enough to show, with
37   * how long that has been open and how many other calls are.
38   */
39  agent?: { seconds: number; quiet?: number; doing?: { text: string; tool: boolean; seconds: number; more: number } }
40  /** The link was just put on the clipboard. */
41  copied?: boolean
42}
43/**
44 * The band: on its one row the connection mark, the link and its copy control, and who is here or
45 * whose agent is working; below, the link when the row cannot hold it, and a failure's reason.
46 */
47export type BandView = {
48  tone: 'error' | 'live' | 'pending' | 'quiet'
49  mark: StatusSegment
50  /**
51   * The link: `text` shows it, on the row or on its own row below (`ownRow`), or nowhere when no
52   * row can hold it. It is followable and copyable only with `href`; while a share or join still
53   * runs it is shown dim, without one.
54   */
55  link?: { text?: string; href?: string; ownRow?: true; copy?: { label: string; copied?: true } }
56  /** The thread's audience, shown beside the link; who can see the thread, not whether it has a link. */
57  audience?: StatusSegment[]
58  /** Controls for a thread Delta refused: fork the rejected local changes into a kept thread, or discard them. */
59  conflict?: { fork: string; discard: string }
60  /** A control that publishes a private thread to the organization. */
61  share?: { label: string }
62  right: StatusSegment[]
63  detail?: StatusSegment[]
64  /** Another participant's agent at work, on a line of its own above the band. */
65  activity?: StatusSegment[]
66  /** Child status on its own line when the host has room. */
67  childActivity?: StatusSegment[]
68}
69
70export const SPINNER_FRAMES = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']
71
72/** Fit a tool's key input between its author, name, status, and transcript gutter. */
73export function toolRowParts(item: WatchItem, columns: number) {
74  const { name, detail, status } = toolCallParts(item)
75  const author = itemAuthor(item)
76  const statusText = status ? ` (${status})` : ''
77  const room = columns - 6 - author.length - 3 - name.length - statusText.length - 1
78  return { name, author, status, statusText, shown: detail && room >= 12 ? middleTruncate(detail, room) : '' }
79}
80
81/** Determine ownership without adopting a turn already identified as remote. */
82export function threadTurnOwned(state: {
83  remoteTurn?: { turnId: string | null }
84  execution?: { sequence: number; lost?: true }
85  producerTurn?: ThreadTurn
86  ownTurnId?: string
87}, turn: Extract<WatchEvent, { type: 'turn' }>): boolean {
88  if (state.remoteTurn && turn.turn_id === state.remoteTurn.turnId) return false
89  const execution = state.execution
90  const status = state.producerTurn?.status
91  return (!!execution && !execution.lost && state.producerTurn?.sequence !== execution.sequence) ||
92    (status === 'running' || (status === 'finished' && !!turn.turn_id && turn.turn_id === state.ownTurnId))
93}
94
95/** The two independently arriving watch streams determine the turn shown in the status band. */
96export function statusCurrentTurn(state: {
97  threadTurn?: Extract<WatchEvent, { type: 'turn' }>
98  producerTurn?: ThreadTurn
99  remoteTurn?: { name: string; id?: string }
100}): TurnStatus {
101  const turn = state.threadTurn
102  if (turn?.state === 'running') {
103    if (turn.turn_id === null) {
104      const name = conversationText(turn.user?.name ?? '').trim() || undefined
105      return { state: 'starting', ...name ? { name } : {}, ...turn.user ? { id: personKey(turn.user) } : {} }
106    }
107    const remote = state.remoteTurn
108    return remote ? { state: 'remote', name: remote.name, ...remote.id ? { id: remote.id } : {} } : { state: 'own' }
109  }
110  return state.producerTurn?.status === 'running' ? { state: 'own' } : { state: 'idle' }
111}
112
113/** The band animates a running share or join, and another participant's turn; Claude already animates this session's own. */
114export function statusSpinnerRunning(state: Pick<StatusState, 'progress' | 'connection' | 'turn' | 'children'>): boolean {
115  return state.progress?.tone === 'active' ||
116    (state.connection === 'live' && (state.turn.state === 'remote' || state.turn.state === 'starting' ||
117      !!state.children?.some((child) => child.status === 'running')))
118}
119
120const CHILD_PRIORITY: Record<string, number> = {
121  failed: 0, running: 1, waiting_for_synchronization: 2, queued: 3, preparing: 4,
122  unknown: 5, completed: 6, stopped: 7, inactive: 8,
123}
124
125function childState(status: string, stale: boolean): string {
126  if (stale) return 'unknown'
127  if (status === 'waiting_for_synchronization') return 'syncing'
128  return Object.hasOwn(CHILD_PRIORITY, status) ? status : 'unknown'
129}
130
131/** Show one representative child and count the rest without conflating their state with the parent turn. */
132export function childStatus(children: readonly WatchChild[], stale: boolean, frame: string, room: number): StatusSegment[] {
133  if (!children.length) return []
134  const ordered = [...children].sort((a, b) =>
135    (CHILD_PRIORITY[a.status] ?? CHILD_PRIORITY.unknown) - (CHILD_PRIORITY[b.status] ?? CHILD_PRIORITY.unknown))
136  const child = ordered[0]!
137  const status = childState(child.status, stale)
138  const counts = new Map<string, number>()
139  for (const entry of ordered) {
140    const label = childState(entry.status, stale)
141    counts.set(label, (counts.get(label) ?? 0) + 1)
142  }
143  const summary = [...counts].map(([label, count]) => `${count} ${label}`).join(' · ')
144  const compactNames: Record<string, string> = {
145    failed: 'fail', running: 'run', syncing: 'sync', queued: 'queue',
146  }
147  const compact = [...counts].map(([label, count]) => `${count} ${compactNames[label] ?? label}`).join(' ')
148  const color = stale ? undefined : child.status === 'failed' ? 'red'
149    : child.status === 'running' ? 'cyan'
150      : ['queued', 'waiting_for_synchronization', 'preparing'].includes(child.status) ? 'yellow' : undefined
151  const prefix = stale ? '◌' : child.status === 'running' ? frame : child.status === 'failed' ? '✕' : '·'
152  const kind = child.kind === 'subagent' ? 'agent' : child.kind === 'subthread' ? 'thread' : 'child'
153  const otherStates = [...counts].filter(([label]) => label !== status)
154    .map(([label, count]) => `${count} ${label}`).join(' · ')
155  const sameStateOthers = (counts.get(status) ?? 1) - 1
156  const suffix = ` ${status}${sameStateOthers ? ` · +${sameStateOthers} ${status}` : ''}${otherStates ? ` · ${otherStates}` : ''}`
157  const base = `${prefix} ${kind} `
158  const name = middleTruncate(conversationText(child.name).replace(/\s+/g, ' ').trim() || child.thread_id,
159    Math.max(0, room - textWidth(base + suffix)))
160  if (!name || room < textWidth(base + suffix) + 1)
161    return [{ text: textWidth(summary) <= room ? summary : compact, dim: stale || undefined }]
162  return [
163    { text: base, color, dim: stale || undefined },
164    { text: name, bold: true, dim: stale || undefined },
165    { text: suffix, color, dim: stale || undefined },
166  ]
167}
168
169/** Colors a person keeps across the band and their transcript rows: none that means a status. */
170const PERSON_COLORS = ['magenta', 'cyan', 'blue', 'magentaBright', 'cyanBright', 'blueBright']
171/** The least space between the two sides of the row. */
172const MIN_GAP = 2
173/** The status bar's padding, a column at each end. */
174const BAR_PADDING = 2
175/** Names shown before the rest collapse to +N. */
176const MAX_NAMES = 3
177
178/** Who a thread user is, across presence, turns and items: their id, or their name without one. */
179export function personKey(user: { user_id: string | null; name: string | null } | null | undefined): string {
180  return user?.user_id ?? `name:${conversationText(user?.name ?? '').trim()}`
181}
182
183/** The same person gets the same color every time, from their person key. */
184export function personColor(key: string): string {
185  let hash = 0
186  for (const char of key) hash = (hash * 31 + char.codePointAt(0)!) >>> 0
187  return PERSON_COLORS[hash % PERSON_COLORS.length]!
188}
189
190/** First names, with a last initial where two people share one. */
191export function firstNames(people: readonly { id: string; name: string }[]): Map<string, string> {
192  const words = people.map(({ id, name }) => {
193    const [first = '', second = ''] = name.trim().split(/\s+/)
194    return { id, first, initial: [...second][0] }
195  })
196  const shared = (first: string) => words.filter((word) => word.first === first).length > 1
197  return new Map(words.map(({ id, first, initial }) => [id, shared(first) && initial ? `${first} ${initial}.` : first]))
198}
199
200/** A thread's link as a short label: its host, first segment and tail. */
201export function shortLink(href: string): string {
202  try {
203    const url = new URL(href)
204    const [segment, ...rest] = url.pathname.split('/').filter(Boolean)
205    const last = rest.at(-1)
206    if (!segment || !last) return `${url.host}${url.pathname}`.replace(/\/$/, '')
207    return `${url.host}/${segment}/…${[...last].slice(-6).join('')}`
208  } catch { return href }
209}
210
211export function formatSeconds(seconds: number): string {
212  const whole = Math.max(0, Math.floor(seconds))
213  return whole < 60 ? `${whole}s` : `${Math.floor(whole / 60)}m ${whole % 60}s`
214}
215
216const textWidth = (text: string) => [...text].length
217const width = (segments: readonly StatusSegment[]) => segments.reduce((sum, { text }) => sum + textWidth(text), 0)
218
219/** What the band's renderer draws for the copy control. */
220export function copyText(copy: { label: string; copied?: true }): string {
221  return copy.copied ? '✔ copied' : copy.label
222}
223
224/**
225 * The columns the band's first row takes, as its renderer draws it:
226 * `● Delta  link  copy link        Maya, Sam`, padded by a column at each end on the terminal.
227 */
228export function bandWidth(band: BandView, terminal: boolean): number {
229  const link = band.link?.text && !band.link.ownRow ? 2 + textWidth(band.link.text) : 0
230  const copy = band.link?.copy ? 2 + textWidth(copyText(band.link.copy)) : 0
231  const audience = band.audience?.length ? 2 + width(band.audience) : 0
232  const conflict = band.conflict ? 2 + textWidth(band.conflict.fork) + 2 + textWidth(band.conflict.discard) : 0
233  const share = band.share ? 2 + textWidth(band.share.label) : 0
234  const right = width(band.right)
235  return (terminal ? BAR_PADDING : 0) + width([band.mark, { text: ' Delta' }]) + link + audience + copy + share + conflict + (right ? MIN_GAP + right : 0)
236}
237
238/** What gives way as the row narrows, each step keeping the ones before it. */
239type GiveWay = { shortCopy?: true; linkBelow?: true; initials?: true; alone?: true; noShare?: true; noAudience?: true; noLink?: true }
240const GIVE_WAY: GiveWay[] = [
241  {},
242  { shortCopy: true },
243  { shortCopy: true, linkBelow: true },
244  { shortCopy: true, linkBelow: true, initials: true },
245  { shortCopy: true, linkBelow: true, initials: true, alone: true },
246  { shortCopy: true, linkBelow: true, initials: true, alone: true, noShare: true },
247  { shortCopy: true, linkBelow: true, initials: true, alone: true, noShare: true, noAudience: true },
248  { shortCopy: true, linkBelow: true, initials: true, alone: true, noShare: true, noAudience: true, noLink: true },
249]
250
251/**
252 * The band as a view of state; rendering never starts timers or mutates its input. As the row
253 * narrows, parts give way in GIVE_WAY's order: the copy label shortens, the terminal's URL moves to
254 * a row of its own below, other people become initials and the working agent's owner drops
255 * "'s agent", the others go, and then the share control and the audience label. The mark, the link
256 * and whoever is working stay; only a band without a row for the URL drops its text, and the copy
257 * control still copies it. The desktop's link shows the URL's short form on the row throughout.
258 *
259 * Rows go to the band's row, then a failure's reason, then another participant's working agent on a
260 * line of its own, then the terminal's URL. With no row for the agent's line, it takes its owner's
261 * place on the band's row, as the narrowing above describes. The thread's audience sits beside the
262 * link on the band's row.
263 */
264export function renderBand(state: StatusState, layout: { columns: number; terminal: boolean; rows?: number }): BandView {
265  const progress = state.progress
266  const failure = progress?.tone === 'error' ? progress.text : state.error ?? state.deliveryPause
267  const running = progress?.tone === 'active' ? progress : undefined
268  const tone: BandView['tone'] = failure ? 'error'
269    : running || state.retryReason || state.reconnecting || state.connection === 'connecting' ? 'pending'
270    : state.connection === 'stopped' ? 'quiet' : 'live'
271  const mark = ({
272    error: { text: '✕', color: 'red' }, pending: { text: '◌', color: 'yellow' },
273    live: { text: '●', color: 'green' }, quiet: { text: '○', dim: true },
274  } as const)[tone]
275  const detail = failure ? [{ text: failure, color: 'red' }]
276    : state.retryReason ? [{ text: `Retrying: ${state.retryReason}`, color: 'yellow' }] : undefined
277  const frame = SPINNER_FRAMES[state.spinner?.frame ?? 0]!
278  // A link is followable once published; while a share or join still runs it is shown, not offered.
279  const href = running ? undefined : state.link
280  const linkText = state.link ? (layout.terminal ? state.link : shortLink(state.link)) : running ? 'link appears when ready' : undefined
281  const turn = state.turn
282  const owner = (turn.state === 'remote' || turn.state === 'starting') && state.connection !== 'stopped' && !state.reconnecting
283    ? turn : undefined
284  // A working agent takes a line of its own when the rows hold it after a failure's reason;
285  // otherwise it takes its owner's place on the band's row.
286  const activityLine = !!owner && (layout.rows ?? Infinity) >= 2 + (detail ? 1 : 0)
287  const childLine = !!state.children?.some((child) =>
288    state.childrenStale || !['completed', 'stopped', 'inactive'].includes(child.status)) &&
289    (layout.rows ?? Infinity) >= 2 + (detail ? 1 : 0) + (activityLine ? 1 : 0)
290  // The URL's own row, wrapped as the terminal wraps it, fits only within the rows the others leave.
291  const linkRows = linkText ? Math.ceil((2 + textWidth(linkText)) / Math.max(1, layout.columns - BAR_PADDING)) : 0
292  const roomBelow = layout.terminal &&
293    (layout.rows ?? Infinity) >= 1 + linkRows + (detail ? 1 : 0) + (activityLine ? 1 : 0) + (childLine ? 1 : 0)
294  const others = (state.people ?? []).filter((person) => activityLine || person.id !== owner?.id)
295  const names = firstNames(others)
296
297  const roster = (initials: boolean): StatusSegment[] => {
298    const out: StatusSegment[] = []
299    others.slice(0, MAX_NAMES).forEach((person, index) => {
300      if (index) out.push(initials ? { text: ' ' } : { text: ', ', dim: true })
301      const name = names.get(person.id) ?? ''
302      out.push({ text: initials ? [...name][0] ?? '' : name, color: personColor(person.id), bold: true,
303        inverse: person.arriving || undefined, dim: person.leaving || state.reconnecting || undefined })
304    })
305    if (others.length > MAX_NAMES) out.push({ text: ` +${others.length - MAX_NAMES}`, dim: true })
306    return out
307  }
308  /** The working agent: on the band's row, its owner and the turn's time; on its own line, what it is doing too. */
309  const agent = (short: boolean, line = false): StatusSegment[] => {
310    if (!owner) return []
311    const color = personColor(owner.id ?? owner.name ?? '')
312    const first = owner.name?.trim().split(/\s+/)[0]
313    const who = { text: first ? (short ? first : `${first}'s agent`) : 'An agent', color, bold: true }
314    const spinner = { text: `${frame} `, color }
315    if (owner.state === 'starting') return [spinner, who, { text: ' starting…', dim: true }]
316    const quiet = state.agent?.quiet
317    if (quiet !== undefined) return [{ text: '◌ ', color: 'yellow' }, who, { text: ` quiet ${formatSeconds(quiet)}`, color: 'yellow' }]
318    if (!state.agent) return [spinner, who]
319    const doing = line ? state.agent.doing : undefined
320    if (!doing) return [spinner, who, { text: line ? `  (${formatSeconds(state.agent.seconds)})` : ` ${formatSeconds(state.agent.seconds)}`, dim: true }]
321    const separator = { text: ' › ', dim: true }
322    const time = { text: `  (${formatSeconds(doing.seconds)}${doing.more ? ` · ${doing.more} more running` : ''})`, dim: true }
323    // The line has the whole width, so a long title gives up its middle rather than its end.
324    const room = layout.columns - (layout.terminal ? BAR_PADDING : 0) - width([spinner, who, separator, time])
325    return [spinner, who, separator, { text: middleTruncate(doing.text, room), ...doing.tool ? { bold: true } : {} }, time]
326  }
327  const right = (give: GiveWay): StatusSegment[] => {
328    if (running) return [{ text: `${frame} ${running.text}`, color: 'yellow' },
329      ...running.step ? [{ text: ` ${running.step.join('/')}`, dim: true }] : []]
330    if (progress?.tone === 'done') return [{ text: `✔ ${progress.text}`, color: 'green' }]
331    const working = activityLine ? [] : agent(!!give.initials)
332    const people = give.alone ? [] : roster(!!give.initials)
333    const below = give.linkBelow && roomBelow
334    const inlineLink = give.noLink && !below && (layout.terminal || !!state.children?.length) ? undefined : linkText
335    const linkRoom = inlineLink && !below ? 2 + textWidth(inlineLink) : 0
336    const copyRoom = href ? 2 + textWidth(give.shortCopy ? 'copy' : 'copy link') : 0
337    const child = childLine ? [] : childStatus(state.children ?? [], !!state.childrenStale, frame,
338      Math.max(0, layout.columns - (layout.terminal ? BAR_PADDING : 0) - 7 - linkRoom - copyRoom -
339        MIN_GAP - width(working) - (working.length ? 3 : 0)))
340    const parts = [working, child, people].filter((part) => part.length)
341    return parts.flatMap((part, index) => index ? [{ text: ' · ', dim: true }, ...part] : part)
342  }
343  const view = (give: GiveWay): BandView => {
344    const below = give.linkBelow && roomBelow
345    const text = give.noLink && !below && (layout.terminal || !!state.children?.length) ? undefined : linkText
346    const copy = href ? { copy: { label: give.shortCopy ? 'copy' : 'copy link', ...state.copied ? { copied: true as const } : {} } } : {}
347    return {
348      tone, mark, right: right(give),
349      ...detail ? { detail } : {},
350      ...state.audience && !give.noAudience ? { audience: [{ text: state.audience, dim: true }] } : {},
351      ...state.conflict ? { conflict: state.conflict } : {},
352      ...state.share && !give.noShare ? { share: state.share } : {},
353      ...activityLine ? { activity: agent(false, true) } : {},
354      ...childLine ? { childActivity: childStatus(state.children ?? [], !!state.childrenStale, frame,
355        Math.max(10, layout.columns - (layout.terminal ? BAR_PADDING : 0))) } : {},
356      ...linkText ? { link: { ...text ? { text } : {}, ...href ? { href } : {}, ...below ? { ownRow: true as const } : {}, ...copy } } : {},
357    }
358  }
359  const views = GIVE_WAY.map(view)
360  return views.find((band) => bandWidth(band, layout.terminal) <= layout.columns) ?? views.at(-1)!
361}
362

1import { isThoughtItem } from './conversation'
2import type { WatchItem } from './thread-commands'
3
4type DrawResult = { deny?: string }
5type DrawTimer = { cancel(): void }
6export type PendingDraw = { session: string; id: string; settle(result: DrawResult): void }
7
8/** Thinking is announced once on live completion and never becomes a model row. */
9export function announceThought(before: WatchItem | undefined, item: WatchItem, snapshotDone: boolean): boolean {
10  return isThoughtItem(item) && !item.open && snapshotDone && (!before || before.open)
11}
12
13/** A successful person-visible draw is remembered even if its model row must retry. */
14export class DrawnRows {
15  private items = new Set<string>()
16
17  drawOnce(id: string, draw: () => Promise<DrawResult>): Promise<DrawResult> | undefined {
18    if (this.items.has(id)) return
19    return draw().then((result) => {
20      if (result.deny === undefined) this.items.add(id)
21      return result
22    })
23  }
24}
25
26/** Queue rows until a hook can draw them, retaining order across held turns. */
27export class DrawQueue {
28  private pending: PendingDraw[] = []
29  get length() { return this.pending.length }
30  shift() { return this.pending.shift() }
31
32  wait(session: string, id: string, held: () => boolean,
33    after: (milliseconds: number, action: () => void) => DrawTimer): Promise<DrawResult> {
34    return new Promise((resolve) => {
35      let added = false
36      let timer: DrawTimer | undefined
37      const entry: PendingDraw = { session, id, settle: (result) => {
38        if (added) return
39        added = true
40        timer?.cancel()
41        resolve(result)
42      } }
43      const arm = () => {
44        timer = after(2000, () => {
45          if (held()) arm()
46          else {
47            this.pending = this.pending.filter((queued) => queued !== entry)
48            entry.settle({ deny: 'the transcript drew no command output row' })
49          }
50        })
51      }
52      arm()
53      this.pending.push(entry)
54    })
55  }
56
57  drop() {
58    const queued = this.pending
59    this.pending = []
60    for (const entry of queued) entry.settle({ deny: 'the Delta watcher stopped before the row was drawn' })
61  }
62}
63
64export const SHOW_ANSWER = '.'
65const SHOW_OUTPUT_ROW = `<local-command-stdout>${SHOW_ANSWER}</local-command-stdout>`
66
67/** Identify the item in an internal command without interpreting its content. */
68export function shownItemId(args: string): string | undefined {
69  return /^show\s+(\S+)$/.exec(args.trim())?.[1]
70}
71
72/** Rewrite only this plugin's hidden command echo and output, not unrelated command rows. */
73export function shownCommandRewrite(plugin: string, row: {
74  door: string
75  origin?: unknown
76  message: { type: string; name?: string; role?: string; content: readonly { type: string; text?: string }[] }
77}): { type: 'text'; text: string }[] | undefined {
78  if (row.door !== 'command') return
79  const text = row.message.content.flatMap((block) =>
80    block.type === 'text' && typeof block.text === 'string' ? [block.text] : []).join('\n')
81  const origin = row.origin as { kind?: string; name?: string } | undefined
82  if (row.message.role === 'user' && origin?.kind === 'plugin' && origin.name === plugin &&
83      /<command-name>[^<]*<\/command-name>/.test(text) && /<command-args>show\s+\S+<\/command-args>/.test(text))
84    return []
85  if (row.message.type === 'system' && row.message.name === 'local_command' &&
86      (text === `<local-command-stdout>${plugin}: ${SHOW_ANSWER}</local-command-stdout>` || text === SHOW_OUTPUT_ROW))
87    return [{ type: 'text', text: SHOW_OUTPUT_ROW }]
88}
89

1type Execution = { sequence: number; turnId?: string; lost?: true; endSequence?: number }
2type SavedBinding = {
3  next: number
4  lastTurnSequence?: number
5  lastClaim?: number
6  execution?: Execution & { claim?: number }
7}
8
9/** Normalize legacy turn identities without mutating persisted data or allocating new sequences. */
10export function restoreBinding<T extends SavedBinding>(previous: T) {
11  const { lastClaim, execution, ...binding } = structuredClone(previous)
12  const validSequence = (sequence: unknown): sequence is number =>
13    Number.isSafeInteger(sequence) && Number(sequence) > 0 && Number(sequence) < binding.next
14  const lastTurnSequence = binding.lastTurnSequence ?? lastClaim
15  if (lastTurnSequence !== undefined && !validSequence(lastTurnSequence))
16    throw new Error('Invalid saved turn sequence; preserve the outbox and producer state')
17  let restored: Execution | undefined
18  if (execution) {
19    const sequence = execution.sequence ?? execution.claim
20    if (!validSequence(sequence) ||
21        (execution.turnId !== undefined && (typeof execution.turnId !== 'string' || !execution.turnId)) ||
22        (execution.endSequence !== undefined && (!validSequence(execution.endSequence) || execution.endSequence <= sequence)))
23      throw new Error('Invalid saved execution; preserve the outbox and producer state')
24    restored = { sequence, turnId: execution.turnId, lost: execution.lost, endSequence: execution.endSequence }
25  }
26  return { ...binding, lastTurnSequence, execution: restored }
27}
28