Files
hapi/web/src/lib/assistant-runtime.ts
T
SSU-WEI HUANGandGitHub f0e5ba9c0f feat(codex): mid-turn Steer via app-server turn/steer (#888) (#1606)
* feat(shared): steer capability gates and live steered signal schemas

- STEERING_SUPPORTED_FLAVORS / isSteeringSupportedForSession gate which
  agents can deliver queued messages into the active turn (pi, codex,
  cursor ACP; legacy stream-json cursor excluded)
- AgentState.steeringActive, DecryptedMessage.steered and
  messages-consumed  live signal (never persisted by the hub)

* feat(cli): queue reservations and steered messages-consumed option

- MessageQueue2 gains takeByLocalId/restoreReservation/
  beginReservationDispatch/commitReservation so an async steer can reserve
  a queued row without racing the main loop's turn/start drain
- emitMessagesConsumed accepts steered: true to mark mid-turn delivery

* feat(codex): mid-turn steer via app-server turn/steer (#888)

- CodexAppServerClient.steerTurn + TurnSteerParams/Response types
- CodexRemoteLauncher registers the steer-queued-message RPC handler:
  reserves the queued row, validates it against the active turn (no
  control commands, matching mode hash), injects via turn/steer with an
  epoch guard that invalidates in-flight steers on abort/cleanup
- steeringActive agent state tracks the active-turn window
- hub syncEngine gate opens to codex; messages-consumed relays steered

* feat(web): Steered badge and steer gating for codex sessions

- HappyUserMessage shows a ↳ Steered badge fed by the live
  messages-consumed steered signal, preserved across server echoes and
  refetches (mergeMessages carries the optimistic marker)
- SessionChat gates canSteer via isSteeringSupportedForSession instead of
  the pi-only check
- clearStaleQueuedStatus normalizes a queued status on an invoked message
- fix(web): drop duplicate showSessionSummaryInChat in markdown test
  (upstream typecheck breakage)

* fix(codex,shared): address bot findings on steer gate and ambiguous turn/steer

- STEERING_SUPPORTED_FLAVORS / isSteeringSupportedForSession advertise
  codex and pi only; cursor joins when its soft-steer handler lands (#1609)
- turn/steer now splits dispatch (stdin accepted) from completion (turn
  finished): the hub RPC acks once dispatch succeeds — never on the
  concurrent turn's completion, which can exceed the 30s RPC window
- queue row commits only after the turn settles; a rejected/aborted steer
  restores the row so the message still delivers via turn/start, and a
  dispatched steer is never restored (no duplicate delivery)
- steer carries clientUserMessageId (echoed as userMessage.clientId) so
  ambiguous transport failures can reconcile the thread later
- client tests cover dispatch/complete split and stdin-write failure

* fix(codex): reconcile dispatched steers before restoring; align error copy

- A dispatched turn/steer whose completion fails (disconnect / protocol
  error) is now reconciled via thread/read by clientUserMessageId before
  the queued row is restored — the instruction is only re-delivered by
  turn/start when the thread never received it
- Reconcile targets the pinned steer thread, not whichever turn is
  current when completion fails
- syncEngine unsupported-flavor error now matches the capability gate
  (Pi and Codex only until the cursor handler lands)
- launcher tests cover steer success (ack on dispatch), reconcile-accepted
  and reconcile-rejected outcomes

* fix(codex): consume the row at dispatch; drop background reconcile

- The hub RPC acks and the queue row is consumed as soon as stdin accepts
  turn/steer; completion is background-only logging. A dispatched steer is
  never restored, so the same localId cannot be re-delivered via turn/start
  after the caller was told the steer succeeded
- Dispatch failure (stdin write error) still restores the row and reports
  failure
- steer.completed rejection is always handled (no unhandled rejection on
  the dispatch-failure path)
- tests updated: completion failure after dispatch keeps the row consumed;
  dispatch failure restores it

* fix(codex): distinguish definite rejection from indeterminate completion

- Transport-level failures (timeout, abort, disconnect, spawn, protocol)
  carry an indeterminate marker; explicit JSON-RPC error responses do not
- After a dispatched steer, turn completion resolves → commit + consumed;
  a definite app-server rejection restores the row (instruction was never
  accepted, so turn/start cannot duplicate it); an indeterminate outcome
  leaves the row reserved so it can never be delivered twice
- Completion handling registers before awaiting dispatch so the
  dispatch-failure path cannot leak an unhandled rejection
- client/launcher tests cover explicit rejection (restore), indeterminate
  outcome (row stays reserved) and dispatch failure

* fix(codex): reconcile indeterminate steers instead of a permanent reservation

- After an indeterminate completion (disconnect/protocol), reconcile the
  thread by clientUserMessageId immediately: accepted → commit + consumed,
  provably rejected → restore, still unreadable → keep the reservation and
  retry from the main-loop top on later passes (post-reconnect)
- A row never sits in dispatching forever: the hub cannot stamp it invoked
  while the instruction may never have been accepted
- tests: indeterminate keeps reserved while thread unreadable; accepted
  reconciliation consumes; rejected path restores

* fix(codex): accept all thread item shapes; retry reconcile; ack through abort

- Reconcile matcher accepts userMessage/user_message with clientId/
  client_id, matching the shapes the thread parser supports — an accepted
  steer can no longer be misclassified as rejected
- A pending reconciliation schedules a wakeLoop retry, so a temporary
  app-server outage cannot strand the reservation behind waitForTurnOrRecovery
- The success-path ACK no longer checks the steer epoch: the hub already
  reported steered on dispatch, so commit + messages-consumed must reach
  it even when an abort resets the queue in between

* fix(codex): reinit reconnected app-server; keep reconcile retries alive

- thread/read after a disconnect auto-connects a fresh app-server, which
  must be initialized before any request — reconcile now ensures
  connect + initialize (isConnected getter added to the client)
- every still-unknown loop-top reconciliation schedules the next retry,
  so recovery without external traffic is eventually observed
- launcher mock gains isConnected

* fix(codex): timer-driven reconciliation; init tracking; abort-safe ACK

- Reconciliation runs on a self-rescheduling 1s timer independent of the
  main loop (wakes it too), so idle loops and waitForTurnOrRecovery still
  observe app-server recovery; abort clears nothing implicitly — the ACK
  path commits and consumes even when the reservation was cancelled
- Absence of a durable client id is ambiguous: unmatched reads stay
  'unknown' and keep retrying instead of restoring the row
- CodexAppServerClient tracks initialized state (reset on disconnect/exit)
  so ensureAppServerInitialized re-initializes a fresh process before
  thread/read; initialize failures leave the flag false for the next retry
- tests: accepted reconciliation via scheduled timer, indeterminate
  keeps reserved, explicit rejection restores

* fix(codex): bind reconciliation to the launcher lifecycle

- runSteerReconciliation clears any armed retry timer on entry and never
  installs a second one, so loop-top and timer-driven passes cannot
  multiply
- shuttingDown is set when the main loop ends: timers are cleared and the
  pending map is dropped, so an unresolved steer can never respawn an
  app-server after cleanup (remote-to-local switch included)

* fix(codex): report steered only after app-server acceptance

- The handler now awaits steer.completed (the inject-acceptance response):
  an explicit JSON-RPC rejection surfaces as failed and restores the row
  for the normal turn/start path instead of a false steered
- Transport failure after dispatch reports 'Steer outcome is being
  reconciled' and keeps the row reserved while the timer-driven thread
  reconciliation runs
- dispatch-failure path also swallows the paired completion rejection

* fix(steer): tri-state cancel, clear-safe reservations, bounded acceptance wait

- MessageQueue2.cancelByLocalId returns 'in-flight' for a dispatching
  steer reservation: the hub neither deletes the row nor stamps invoked_at
  (new CancelMessageResponse 'busy' status; web restores the optimistic
  row); pushIsolateAndClear and reset/close share cancelReservations so
  /clear-style commands cannot have a rejected steer resurrect a discarded
  prompt
- turn/steer acceptance wait bounded at 25s (< hub 30s RPC timeout): a
  lost response is indeterminate and funnels into thread reconciliation
  instead of stranding the reservation
- tests updated for the tri-state cancel contract

* fix(codex,web): busy-aware edit flow; bound reconciliation reads

- QueuedMessagesBar edit flow treats a 'busy' cancel as unsuccessful: it
  never prefills the composer when the row is inside an async steer, so a
  second client cannot send a duplicate
- reconcileSteerByClientId bounds thread/read with a 5s timeout so a
  connected-but-silent app-server cannot hold the reservation in-flight
  indefinitely

* fix(steer): inFlight-dominated cancel acks; bounded reconciliation

- hub cancel-queued-message acks check inFlight before removed: a stale
  duplicate socket reporting removed can no longer delete the durable row
  while another socket is dispatching the steer
- reconciliation entries expire after 60s and mark delivered: after the
  rejection window, a dispatched steer that the app-server never proved
  (client ids dropped on restart) is committed instead of polling
  thread/read forever
- pre-dispatch failures (abort before write included) never enter
  reconciliation — they restore the row and report failure

* fix(steer): persist indeterminate outcomes without replay

* fix(steer): make ambiguous delivery restart-safe

* fix(steer): recover crash-held rows and preserve retry dedup

* fix(steer): ack retries and bound stdin dispatch

* fix(steer): reconcile indeterminate dispatches and serialize retries

* fix(codex): classify stdin callback failures as indeterminate

* fix(steer): recheck indeterminate cancels after ACK

* fix(steer): close retry and abort races

* fix(steer): serialize live retries and abort admission

* fix(steer): distinguish live dispatching from unknown

* fix(steer): keep ACK failures held and reconcile busy cancel

* fix(steer): distinguish held cancel from removal

* fix(store): combine schema v24 migrations

* fix(store): reserve schema v25 for steer delivery state

* fix(steer): keep held cancel state and notify requeue

* fix(steer): release explicitly cancelled unknown reservations

* fix(codex): reject cancelled reservations before native steer

* fix(codex): make reservation restore atomic with state

* fix(codex): terminate abandoned transport writes

* fix(steer): own abandoned app-server lifecycle and consume races

* fix(codex): confirm dispatch and recover abandoned turns

* test(codex): mock abandoned transport callback

* fix(codex): clear visible turn state on transport loss

* fix(steer): claim retries and cover native delivery state

* fix(native): preserve indeterminate state on Android hydration

* fix(steer): make retry claims single-winner

* fix(steer): serialize concurrent retry claims

* fix(socket): tolerate missing steer-state ACK callbacks

* fix(native): serialize retry operations

* docs(web): document unknown steer delivery and retry controls

* fix(steer): handle retry failures and abort-before-connect

* fix(steer): reinitialize after transport loss and finish iOS retry errors

* fix(steer): preserve indeterminate rows across reconnect gaps

* test(web): mock indeterminate queued recovery state

* fix(steer): recover consumed ACK tombstones

* fix(steer): expose consumed cancel tombstones
2026-08-19 20:07:39 +08:00

962 lines
36 KiB
TypeScript

import { useCallback, useMemo, useRef } from 'react'
import type React from 'react'
import type { AppendMessage, AttachmentAdapter, ThreadMessageLike } from '@assistant-ui/react'
import { useExternalMessageConverter, useExternalStoreRuntime } from '@assistant-ui/react'
import type { PendingSchedule } from '@/components/AssistantChat/ScheduleTimePicker'
import { resolvePendingSchedule } from '@/components/AssistantChat/ScheduleTimePicker'
import {
consumeComposerSendIntent,
type ComposerSendIntent,
} from '@/lib/messageDelivery'
import { safeStringify } from '@hapi/protocol'
import { renderEventLabel } from '@/chat/presentation'
import type { ChatBlock, CliOutputBlock, CodexReview, UsageData } from '@/chat/types'
import type { AgentEvent, ToolCallBlock } from '@/chat/types'
import type { ToolGroupBlock, VisibleChatBlock } from '@/chat/toolGroups'
import { visibleBlockRole } from '@/chat/toolGroups'
import type { AttachmentMetadata, MessageStatus as HappyMessageStatus, Session } from '@/types/api'
/**
* Aggregated metadata for a multi-turn response group, surfaced on the
* group's first visible block so the `@assistant-ui/react` converter
* preserves it after joining adjacent assistant messages.
*/
export type AggregatedAssistantMeta = {
usage?: UsageData
model: string | null
invokedAt: number | null
durationMs: undefined
turnCount: number
}
export type HappyChatMessageMetadata = {
kind: 'user' | 'assistant' | 'tool' | 'event' | 'cli-output' | 'codex-review' | 'compact-summary'
status?: HappyMessageStatus
localId?: string | null
originalText?: string
toolCallId?: string
event?: AgentEvent
source?: CliOutputBlock['source']
attachments?: AttachmentMetadata[]
invokedAt?: number | null
steered?: boolean
durationMs?: number
usage?: UsageData
model?: string | null
review?: CodexReview
/**
* Distinct turn count when this block carries an aggregated response
* group footer. Single-turn blocks omit this field so the existing
* per-message footer is rendered unchanged.
*/
turnCount?: number
}
export type HappyRuntimeExtras = Readonly<{
messagesVersion: number
historyVersion: number
}>
function formatCodexReviewText(review: CodexReview): string {
const lines = ['Codex review']
if (review.overallCorrectness) {
lines.push(`Overall: ${review.overallCorrectness}`)
}
if (review.overallExplanation) {
lines.push('', review.overallExplanation)
}
if (review.findings.length > 0) {
lines.push('', 'Findings:')
for (const finding of review.findings) {
const priority = finding.priority === null ? '' : `[P${finding.priority}] `
const location = finding.filePath
? ` (${finding.filePath}${finding.lineStart === null ? '' : `:${finding.lineStart}${finding.lineEnd !== null && finding.lineEnd !== finding.lineStart ? `-${finding.lineEnd}` : ''}`})`
: ''
lines.push(`- ${priority}${finding.title}${location}`)
lines.push(` ${finding.body}`)
}
}
return lines.join('\n')
}
export function getBlockPresentationTimestamp(block: VisibleChatBlock): number {
if (visibleBlockRole(block) === 'user') {
return block.invokedAt ?? block.createdAt
}
if (block.kind === 'tool-group') {
return block.tools.reduce(
(latest, tool) => Math.max(latest, tool.tool.completedAt ?? tool.createdAt),
block.createdAt
)
}
if (block.kind === 'tool-call') {
return Math.max(block.createdAt, block.tool.completedAt ?? block.createdAt)
}
return block.createdAt
}
/**
* `@assistant-ui/react` joins adjacent assistant-role blocks into one card and
* takes its timestamp from the first block. Use the response's final activity
* time so prepending an older page cannot change an existing card's timestamp.
*/
export function getResponseGroupTimestamps(
blocks: readonly VisibleChatBlock[]
): Map<VisibleChatBlock, number> {
const timestamps = new Map<VisibleChatBlock, number>()
let first: VisibleChatBlock | null = null
let latestTimestamp = 0
const flush = () => {
if (first) timestamps.set(first, latestTimestamp)
first = null
}
for (const block of blocks) {
if (visibleBlockRole(block) !== 'assistant') {
flush()
continue
}
const timestamp = getBlockPresentationTimestamp(block)
if (!first) {
first = block
latestTimestamp = timestamp
} else {
latestTimestamp = Math.max(latestTimestamp, timestamp)
}
}
flush()
return timestamps
}
type TurnSource = {
localId: string | null
invokedAt: number | null
durationMs: number | undefined
model: string | null
usage: UsageData | undefined
createdAt: number
}
// Return one turn source per claude-SDK message that the visible block
// represents. `tool-group` is a derived view: `buildVisibleChatBlocks` merges
// adjacent eligible tool-calls without checking that they share a turn, so
// each underlying tool-call contributes its own source. Other assistant-role
// kinds map to a single source.
function turnSourcesFromBlock(block: VisibleChatBlock): TurnSource[] {
if (block.kind === 'tool-group') {
return block.tools.map((tool) => ({
localId: tool.localId,
invokedAt: tool.invokedAt ?? null,
durationMs: tool.durationMs,
model: tool.model ?? null,
usage: tool.usage,
createdAt: tool.createdAt
}))
}
if (
block.kind === 'agent-text'
|| block.kind === 'agent-reasoning'
|| block.kind === 'cli-output'
|| block.kind === 'tool-call'
) {
return [{
localId: block.localId,
invokedAt: block.invokedAt ?? null,
durationMs: block.durationMs,
model: block.model ?? null,
usage: block.usage,
createdAt: block.createdAt
}]
}
return []
}
/**
* Fallback turn key for environments where the CLI does not stamp a
* non-null `localId` on each turn's blocks (e.g. claude code spawn
* sessions today). The key combines `(model, usage totals, createdAt)`:
* the reducer copies `msg.createdAt` onto every derived `ChatBlock`, so
* blocks from the same SDK message share createdAt and collapse to one
* turn, while blocks from different SDK messages stay distinct even
* when their `(model, usage)` happens to coincide. Limitation: two SDK
* messages stamped at the same wall-clock millisecond would still
* collide, but the hub stamp resolution makes that vanishingly rare.
*/
function turnFingerprint(
model: string | null,
usage: UsageData | undefined,
createdAt: number
): string {
if (!usage) return `m=${model ?? ''}|u=|c=${createdAt}`
return [
`m=${model ?? ''}`,
`i=${usage.input_tokens}`,
`o=${usage.output_tokens}`,
`cc=${usage.cache_creation_input_tokens ?? ''}`,
`cr=${usage.cache_read_input_tokens ?? ''}`,
`t=${usage.service_tier ?? ''}`,
`c=${createdAt}`
].join('|')
}
/**
* Sum two optional token counters. Returns `undefined` only when both
* operands are absent; an explicit `0` on either side participates in
* the sum (so a `0 + 0` total stays `0` instead of collapsing to
* `undefined` via JavaScript's falsy `||`).
*/
function sumOptional(a: number | undefined, b: number | undefined): number | undefined {
if (a === undefined && b === undefined) return undefined
return (a ?? 0) + (b ?? 0)
}
function addUsage(target: UsageData, addend: UsageData): UsageData {
return {
input_tokens: target.input_tokens + addend.input_tokens,
output_tokens: target.output_tokens + addend.output_tokens,
cache_creation_input_tokens: sumOptional(
target.cache_creation_input_tokens,
addend.cache_creation_input_tokens
),
cache_read_input_tokens: sumOptional(
target.cache_read_input_tokens,
addend.cache_read_input_tokens
),
// service_tier dedup follows the rule documented in the plan: pick
// the first turn's tier when the group is mixed. We keep it on the
// target so downstream label logic sees a stable value.
service_tier: target.service_tier
}
}
/**
* Walk the visible block list, identify response groups (runs of
* assistant-role blocks separated by user-text / agent-event /
* user-source cli-output boundaries), and return a map keyed by the
* id of each group's first visible block whose value is the summed
* metadata for that group. Only groups spanning two or more distinct
* turns produce an entry so single-turn cards remain byte-for-byte
* unchanged at the call site.
*/
export function aggregateResponseGroups(
blocks: readonly VisibleChatBlock[]
): Map<string, AggregatedAssistantMeta> {
const aggregates = new Map<string, AggregatedAssistantMeta>()
let groupFirstBlockId: string | null = null
// Ordering-based turn dedup: we compare each block's turn key against
// the immediately previous turn's key. A `Set` of all seen keys would
// collapse a third turn whose `(model, usage)` fingerprint happens to
// match a non-adjacent earlier turn — claude code spawn sessions
// legitimately produce repeated fingerprints when token totals
// coincide, and merging them under-counts the visible turn count.
let prevTurnKey: string | null = null
const seenModels: string[] = []
let groupInvokedAt: number | null = null
let groupUsage: UsageData | undefined
let groupTurnCount = 0
const flush = () => {
if (groupFirstBlockId !== null && groupTurnCount >= 2) {
const joinedModel = seenModels.length > 0 ? seenModels.join(', ') : null
aggregates.set(groupFirstBlockId, {
usage: groupUsage,
model: joinedModel,
invokedAt: groupInvokedAt,
durationMs: undefined,
turnCount: groupTurnCount
})
}
groupFirstBlockId = null
prevTurnKey = null
seenModels.length = 0
groupInvokedAt = null
groupUsage = undefined
groupTurnCount = 0
}
for (const block of blocks) {
const role = visibleBlockRole(block)
if (role !== 'assistant') {
// Boundary: close the open group, if any.
flush()
continue
}
if (groupFirstBlockId === null) {
groupFirstBlockId = block.id
}
for (const turn of turnSourcesFromBlock(block)) {
// Prefer the CLI-stamped `localId` when present. When it is null
// (today's claude code spawn flow) fall back to a fingerprint
// built from `(model, usage totals, createdAt)` — see
// `turnFingerprint` for the createdAt rationale.
const turnKey = turn.localId !== null
? `id:${turn.localId}`
: `fp:${turnFingerprint(turn.model, turn.usage, turn.createdAt)}`
// Skip blocks with no turn signal at all (no localId, no usage,
// no model) so they don't inflate the turn count.
if (turn.localId === null && !turn.usage && !turn.model) continue
// Only the immediately previous turn's key dedups — see prevTurnKey
// comment above. Non-adjacent matches keep their separate counts.
if (turnKey === prevTurnKey) continue
prevTurnKey = turnKey
groupTurnCount += 1
if (turn.invokedAt != null && (groupInvokedAt === null || turn.invokedAt < groupInvokedAt)) {
groupInvokedAt = turn.invokedAt
}
if (turn.model && !seenModels.includes(turn.model)) {
seenModels.push(turn.model)
}
if (turn.usage) {
groupUsage = groupUsage ? addUsage(groupUsage, turn.usage) : { ...turn.usage }
}
}
}
flush()
return aggregates
}
export type BlockWithThreadMessageId = {
block: VisibleChatBlock
threadMessageId: string
}
/**
* Stable, unique IDs for assistant-ui's linear MessageRepository.
* Uses `${kind}:${block.id}`; suffixes `~1`, `~2`, … when the same kind+id
* appears more than once (should be rare — indicates duplicate hub rows or
* a reducer bug, but must not crash the thread).
*
* Reuses `{ block, threadMessageId }` objects from `wrapperCache` when the
* reconciled `block` reference and computed id match, so
* `useExternalMessageConverter`'s WeakMap caches stay warm across streaming
* appends (see PR review).
*/
export function assignThreadMessageIdsWithStableWrappers(
blocks: readonly VisibleChatBlock[],
wrapperCache: WeakMap<VisibleChatBlock, BlockWithThreadMessageId>
): BlockWithThreadMessageId[] {
const seen = new Map<string, number>()
return blocks.map((block) => {
const base = `${block.kind}:${block.id}`
const occurrence = seen.get(base) ?? 0
seen.set(base, occurrence + 1)
const threadMessageId = occurrence === 0 ? base : `${base}~${occurrence}`
const cached = wrapperCache.get(block)
if (cached?.threadMessageId === threadMessageId) {
return cached
}
const next: BlockWithThreadMessageId = { block, threadMessageId }
wrapperCache.set(block, next)
return next
})
}
export function assignThreadMessageIds(
blocks: readonly VisibleChatBlock[]
): BlockWithThreadMessageId[] {
return assignThreadMessageIdsWithStableWrappers(blocks, new WeakMap())
}
/**
* Finds the latest conversation-history boundary that is safe to fork.
* While the main agent is running, every block from the latest invoked user
* message onward belongs to the active turn and must not become a transient
* "completed" boundary as streaming events reshape the visible block list.
*/
export function findLatestCompletedBoundaryId(
blocks: readonly VisibleChatBlock[],
isRunning: boolean,
activeTurnStartedAt: number | null
): string | null {
const assigned = assignThreadMessageIds(blocks)
let limit = assigned.length
if (isRunning) {
const activeTurnStart = activeTurnStartedAt === null
? assigned.findLastIndex(({ block }) => (
visibleBlockRole(block) === 'user' && block.invokedAt != null
))
: assigned.findIndex(({ block }) => (
visibleBlockRole(block) === 'user'
&& (block.invokedAt ?? block.createdAt) >= activeTurnStartedAt
))
if (activeTurnStart >= 0) {
limit = activeTurnStart
} else if (activeTurnStartedAt === null) {
return null
} else {
const firstActiveBlock = assigned.findIndex(({ block }) => (
(block.invokedAt ?? block.createdAt) >= activeTurnStartedAt
))
if (firstActiveBlock >= 0) {
limit = firstActiveBlock
}
}
}
let candidate: string | null = null
let previousRole: ReturnType<typeof visibleBlockRole> | null = null
for (let index = 0; index < limit; index += 1) {
const { block, threadMessageId } = assigned[index]
const role = visibleBlockRole(block)
if (
(role === 'user' && block.invokedAt != null)
|| (role === 'assistant' && previousRole !== 'assistant')
) {
candidate = threadMessageId
}
previousRole = role
}
return candidate
}
function getAssistantBlockSignatures(blocks: readonly VisibleChatBlock[]): string[] {
return blocks
.filter((block) => visibleBlockRole(block) === 'assistant')
.map((block) => {
const text = 'text' in block ? block.text : ''
const toolCount = block.kind === 'tool-group' ? block.tools.length : 0
const textMarker = text.length <= 64 ? text : `${text.length}:${text.slice(-32)}`
return `${block.id}:${getBlockPresentationTimestamp(block)}:${textMarker}:${toolCount}`
})
}
function haveSameAssistantBlockSignatures(
first: readonly string[],
second: readonly string[]
): boolean {
return first.length === second.length
&& first.every((value, index) => value === second[index])
}
function isOlderAssistantHistoryPrepended(
current: readonly string[],
baseline: readonly string[]
): boolean {
if (baseline.length === 0 || current.length <= baseline.length) return false
const suffixStart = current.length - baseline.length
return baseline.every((value, index) => current[suffixStart + index] === value)
}
function containsActiveAssistantOutput(
blocks: readonly VisibleChatBlock[],
activeTurnStartedAt: number | null
): boolean {
return activeTurnStartedAt !== null
&& blocks.some((block) => (
visibleBlockRole(block) === 'assistant'
&& getBlockPresentationTimestamp(block) >= activeTurnStartedAt
))
}
function toThreadMessageLike(
block: VisibleChatBlock,
threadMessageId: string,
timestamp: number
): ThreadMessageLike {
if (block.kind === 'user-text') {
return {
role: 'user',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{ type: 'text', text: block.text }],
metadata: {
custom: {
kind: 'user',
status: block.status,
localId: block.localId,
originalText: block.originalText,
attachments: block.attachments,
invokedAt: block.invokedAt,
steered: block.steered
} satisfies HappyChatMessageMetadata
}
}
}
if (block.kind === 'agent-text') {
return {
role: 'assistant',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{ type: 'text', text: block.text }],
metadata: {
custom: {
kind: 'assistant',
invokedAt: block.invokedAt,
durationMs: block.durationMs,
usage: block.usage,
model: block.model
} satisfies HappyChatMessageMetadata
}
}
}
if (block.kind === 'generated-image') {
return {
role: 'assistant',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{
type: 'tool-call',
toolCallId: block.id,
toolName: 'GeneratedImage',
argsText: '',
artifact: block
}],
metadata: {
custom: {
kind: 'tool',
toolCallId: block.id,
invokedAt: block.invokedAt ?? null
} satisfies HappyChatMessageMetadata
}
}
}
if (block.kind === 'agent-reasoning') {
return {
role: 'assistant',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{ type: 'reasoning', text: block.text }],
metadata: {
custom: {
kind: 'assistant',
invokedAt: block.invokedAt,
durationMs: block.durationMs,
usage: block.usage,
model: block.model
} satisfies HappyChatMessageMetadata
}
}
}
if (block.kind === 'codex-review') {
return {
role: 'assistant',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{ type: 'text', text: formatCodexReviewText(block.review) }],
metadata: {
custom: {
kind: 'codex-review',
invokedAt: block.invokedAt,
durationMs: block.durationMs,
usage: block.usage,
model: block.model,
review: block.review
} satisfies HappyChatMessageMetadata
}
}
}
if (block.kind === 'agent-event') {
// Pi compaction summaries carry a real payload; surface them as a
// dedicated system message so the chat can render an independent
// block instead of a small status line.
if (block.event.type === 'compact-summary' && typeof block.event.summary === 'string') {
return {
role: 'system',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{ type: 'text', text: block.event.summary }],
metadata: {
custom: {
kind: 'compact-summary',
event: block.event,
invokedAt: block.invokedAt,
model: block.model
} satisfies HappyChatMessageMetadata
}
}
}
return {
role: 'system',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{ type: 'text', text: renderEventLabel(block.event) }],
metadata: {
custom: {
kind: 'event',
event: block.event,
invokedAt: block.invokedAt,
model: block.model
} satisfies HappyChatMessageMetadata
}
}
}
if (block.kind === 'cli-output') {
return {
role: block.source === 'user' ? 'user' : 'assistant',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{ type: 'text', text: block.text }],
metadata: {
custom: {
kind: 'cli-output',
source: block.source,
invokedAt: block.invokedAt,
durationMs: block.durationMs,
usage: block.usage,
model: block.model
} satisfies HappyChatMessageMetadata
}
}
}
if (block.kind === 'tool-group') {
const groupBlock: ToolGroupBlock = block
return {
role: 'assistant',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{
type: 'tool-call',
toolCallId: groupBlock.id,
toolName: 'ToolGroup',
argsText: '',
artifact: groupBlock
}],
metadata: {
custom: {
kind: 'tool',
toolCallId: groupBlock.id,
invokedAt: groupBlock.invokedAt ?? null
} satisfies HappyChatMessageMetadata
}
}
}
const toolBlock: ToolCallBlock = block
const inputText = safeStringify(toolBlock.tool.input)
return {
role: 'assistant',
id: threadMessageId,
createdAt: new Date(timestamp),
content: [{
type: 'tool-call',
toolCallId: toolBlock.id,
toolName: toolBlock.tool.name,
argsText: inputText,
result: toolBlock.tool.result,
isError: toolBlock.tool.state === 'error',
artifact: toolBlock
}],
metadata: {
custom: {
kind: 'tool',
toolCallId: toolBlock.id,
invokedAt: toolBlock.invokedAt,
durationMs: toolBlock.durationMs,
usage: toolBlock.usage,
model: toolBlock.model
} satisfies HappyChatMessageMetadata
}
}
}
type TextMessagePart = { type: 'text'; text: string }
function getTextFromParts(parts: readonly { type: string }[] | undefined): string {
if (!parts) return ''
return parts
.filter((part): part is TextMessagePart => part.type === 'text' && typeof (part as TextMessagePart).text === 'string')
.map((part) => part.text)
.join('\n')
.trim()
}
type ExtractedAttachmentMetadata = { __attachmentMetadata: AttachmentMetadata }
function isAttachmentMetadataJson(text: string): ExtractedAttachmentMetadata | null {
try {
const parsed = JSON.parse(text) as unknown
if (parsed && typeof parsed === 'object' && '__attachmentMetadata' in parsed) {
return parsed as ExtractedAttachmentMetadata
}
return null
} catch {
return null
}
}
function extractMessageContent(message: AppendMessage): { text: string; attachments: AttachmentMetadata[] } {
if (message.role !== 'user') return { text: '', attachments: [] }
// Extract attachments from attachment content
const attachments: AttachmentMetadata[] = []
const otherAttachmentTexts: string[] = []
const attachmentParts = message.attachments?.flatMap((attachment) => attachment.content ?? []) ?? []
for (const part of attachmentParts) {
if (part.type === 'text' && typeof (part as TextMessagePart).text === 'string') {
const textPart = part as TextMessagePart
const extracted = isAttachmentMetadataJson(textPart.text)
if (extracted) {
attachments.push(extracted.__attachmentMetadata)
} else {
otherAttachmentTexts.push(textPart.text)
}
}
}
const contentText = getTextFromParts(message.content)
const text = [otherAttachmentTexts.join('\n'), contentText]
.filter((value) => value.length > 0)
.join('\n\n')
.trim()
return { text, attachments }
}
export function useHappyRuntime(props: {
session: Session
blocks: readonly VisibleChatBlock[]
messagesVersion: number
historyVersion: number
viewMode?: 'tail' | 'history'
isSyncingTail?: boolean
isLoadingMore?: boolean
isSending: boolean
isRunning?: boolean
onSendMessage: (
text: string,
attachments?: AttachmentMetadata[],
scheduledAt?: number | null,
intent?: ComposerSendIntent,
) => void
onAbort: () => Promise<void>
attachmentAdapter?: AttachmentAdapter
allowSendWhenInactive?: boolean
pendingScheduleRef?: React.RefObject<PendingSchedule | null>
/**
* Shared one-shot ref with HappyComposer. The composer marks the next
* `api.composer().send()`; this adapter consumes and resets the mark as
* soon as assistant-ui emits the corresponding AppendMessage.
*/
pendingSendIntentRef?: React.MutableRefObject<ComposerSendIntent>
}) {
const isRunning = props.isRunning ?? props.session.thinking
const activeTurnStartedAt = props.session.activeTurnStartedAt ?? null
const hydratingWindow = props.viewMode === 'history'
|| props.isSyncingTail === true
|| props.isLoadingMore === true
const assistantBlockSignatures = useMemo(
() => getAssistantBlockSignatures(props.blocks),
[props.blocks]
)
const hasActiveAssistantOutput = containsActiveAssistantOutput(props.blocks, activeTurnStartedAt)
const awaitingInitialSnapshot = isRunning && props.blocks.length === 0
const runningHandoffRef = useRef({
sessionId: props.session.id,
wasRunning: isRunning,
historyVersion: props.historyVersion,
assistantBlockSignatures: isRunning ? assistantBlockSignatures : null,
awaitingExistingRunHydration: awaitingInitialSnapshot,
wasHydratingWindow: hydratingWindow,
// A running session can mount before its first message page arrives.
// Treat that first non-empty snapshot as hydration, not as new stream
// output, so a resumed reasoning part cannot start from an empty view.
hasObservedAssistantBlocks: !isRunning || !awaitingInitialSnapshot
})
// assistant-ui derives the last message's status from the thread-level
// isRunning flag. On a send, that flag can become true before the new
// assistant block is visible, so keep the previous materialized response
// complete until the block list proves that the new turn has produced
// output. Reset the baseline when the viewed session changes so switching
// from one running session to another gets the same protection.
const runningHandoff = runningHandoffRef.current
if (runningHandoff.sessionId !== props.session.id) {
runningHandoff.sessionId = props.session.id
runningHandoff.wasRunning = isRunning
runningHandoff.historyVersion = props.historyVersion
runningHandoff.assistantBlockSignatures = isRunning ? assistantBlockSignatures : null
runningHandoff.awaitingExistingRunHydration = awaitingInitialSnapshot
runningHandoff.wasHydratingWindow = hydratingWindow
runningHandoff.hasObservedAssistantBlocks = !isRunning || !awaitingInitialSnapshot
} else if (!isRunning) {
runningHandoff.historyVersion = props.historyVersion
runningHandoff.assistantBlockSignatures = null
runningHandoff.awaitingExistingRunHydration = false
} else if (!runningHandoff.wasRunning) {
runningHandoff.historyVersion = props.historyVersion
runningHandoff.assistantBlockSignatures = hasActiveAssistantOutput
? null
: assistantBlockSignatures
runningHandoff.awaitingExistingRunHydration = false
runningHandoff.hasObservedAssistantBlocks = props.blocks.length > 0
} else if (runningHandoff.historyVersion !== props.historyVersion) {
// A bounded older-history prepend can drop the previous tail, so the
// new signature list is not necessarily a suffix of the old one.
runningHandoff.historyVersion = props.historyVersion
if (runningHandoff.assistantBlockSignatures !== null) {
runningHandoff.assistantBlockSignatures = assistantBlockSignatures
runningHandoff.hasObservedAssistantBlocks = props.blocks.length > 0
}
} else if (hydratingWindow || runningHandoff.wasHydratingWindow) {
if (runningHandoff.assistantBlockSignatures !== null) {
runningHandoff.assistantBlockSignatures = assistantBlockSignatures
runningHandoff.hasObservedAssistantBlocks = props.blocks.length > 0
}
} else if (!runningHandoff.hasObservedAssistantBlocks && props.blocks.length > 0) {
runningHandoff.assistantBlockSignatures = (
runningHandoff.awaitingExistingRunHydration || !hasActiveAssistantOutput
)
? assistantBlockSignatures
: null
runningHandoff.awaitingExistingRunHydration = false
runningHandoff.hasObservedAssistantBlocks = true
} else if (
runningHandoff.assistantBlockSignatures !== null
&& !haveSameAssistantBlockSignatures(
runningHandoff.assistantBlockSignatures,
assistantBlockSignatures
)
) {
if (isOlderAssistantHistoryPrepended(
assistantBlockSignatures,
runningHandoff.assistantBlockSignatures
)) {
runningHandoff.assistantBlockSignatures = assistantBlockSignatures
} else {
runningHandoff.assistantBlockSignatures = null
}
}
runningHandoff.wasRunning = isRunning
runningHandoff.wasHydratingWindow = hydratingWindow
const waitingForAssistantOutput = isRunning
&& runningHandoff.assistantBlockSignatures !== null
&& haveSameAssistantBlockSignatures(
runningHandoff.assistantBlockSignatures,
assistantBlockSignatures
)
const isRunningForMessages = isRunning && !waitingForAssistantOutput
// Compute response-group aggregates once per block list so we can
// inject the summed metadata onto each group's first visible block.
// The library's `joinExternalMessages` only preserves
// `metadata.custom` from the first block of a joined chunk, so this
// is the surface that survives the join.
const threadIdWrapperCacheRef = useRef(
new WeakMap<VisibleChatBlock, BlockWithThreadMessageId>()
)
const blocksWithThreadIds = useMemo(
() => assignThreadMessageIdsWithStableWrappers(
props.blocks,
threadIdWrapperCacheRef.current
),
[props.blocks]
)
const aggregates = useMemo(
() => aggregateResponseGroups(props.blocks),
[props.blocks]
)
const responseGroupTimestamps = useMemo(
() => getResponseGroupTimestamps(props.blocks),
[props.blocks]
)
const convertBlock = useCallback(
({ block, threadMessageId }: BlockWithThreadMessageId): ThreadMessageLike => {
const message = toThreadMessageLike(
block,
threadMessageId,
responseGroupTimestamps.get(block) ?? getBlockPresentationTimestamp(block)
)
const aggregate = aggregates.get(block.id)
if (!aggregate) return message
const existing = message.metadata?.custom as HappyChatMessageMetadata | undefined
return {
...message,
metadata: {
...message.metadata,
custom: {
...(existing ?? { kind: 'assistant' }),
usage: aggregate.usage,
model: aggregate.model,
invokedAt: aggregate.invokedAt,
durationMs: aggregate.durationMs,
turnCount: aggregate.turnCount
} satisfies HappyChatMessageMetadata
}
}
},
[aggregates, responseGroupTimestamps]
)
// Use cached message converter for performance optimization
// This prevents re-converting all messages on every render
const convertedMessages = useExternalMessageConverter<BlockWithThreadMessageId>({
callback: convertBlock,
messages: blocksWithThreadIds,
isRunning: isRunningForMessages,
})
const onNew = useCallback(async (message: AppendMessage) => {
const intent = consumeComposerSendIntent(props.pendingSendIntentRef)
// Reset before any early return so an empty submission, extraction
// failure, or downstream exception cannot leak an explicit queue
// gesture into the next ordinary send.
const { text, attachments } = extractMessageContent(message)
if (!text && attachments.length === 0) return
// Resolve pendingSchedule at send time (Date.now()) so preset-type schedules
// ("5 minutes from now") are relative to the actual send action, not the
// moment the user clicked the preset button.
const sendNow = Date.now()
const scheduledAt = resolvePendingSchedule(props.pendingScheduleRef?.current ?? null, sendNow)
props.onSendMessage(text, attachments.length > 0 ? attachments : undefined, scheduledAt, intent)
}, [props.onSendMessage, props.pendingScheduleRef, props.pendingSendIntentRef])
const onCancel = useCallback(async () => {
await props.onAbort()
}, [props.onAbort])
const extras = useMemo<HappyRuntimeExtras>(() => ({
messagesVersion: props.messagesVersion,
historyVersion: props.historyVersion
}), [props.messagesVersion, props.historyVersion])
// Memoize the adapter to avoid recreating on every render
// useExternalStoreRuntime may use adapter identity for subscriptions
const adapter = useMemo(() => ({
isDisabled: props.isSending || (!props.session.active && !props.allowSendWhenInactive),
isRunning,
messages: convertedMessages,
extras,
onNew,
onCancel,
adapters: props.attachmentAdapter ? { attachments: props.attachmentAdapter } : undefined,
unstable_capabilities: { copy: true }
}), [
props.session.active,
props.isSending,
props.allowSendWhenInactive,
isRunning,
convertedMessages,
extras,
onNew,
onCancel,
props.attachmentAdapter
])
// Note: pendingScheduleRef is intentionally not in the deps above.
// The ref is read at send time inside onNew (not at render time), so changes
// to pendingSchedule do not need to invalidate the adapter or re-run onNew.
return useExternalStoreRuntime(adapter)
}