vhaudiquet/rhema-claw
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(web/claw): open on the newest turn, walk back, and resume from the seq

Browse Source 
A reload of a weeks-old session replayed the whole record before the
first pixel, and every reconnect reset the mirror and replayed it
again. The stream now names each frame's seq and the client keeps it,
so a retry sends since=<seq> and receives only the gap, and the mirror
is left alone: its items are still true.

A first open with no cursor gets the tail, cut to a turn boundary,
which paints the latest conversation at once. The turns above it come
from the history route, one whole turn per page, fetched when the
reader nears the top or when the transcript is too short to fill the
screen; a page folds through the same reducer as the live stream and
joins above, and the scroll position moves down by the page's height so
the reader keeps their place. A reset comment (the record no longer
holds the cursor) drops the mirror; the tail that follows rebuilds it.

The dev proxy rebuilt the upstream URL from the pathname alone, so
since and before never reached the daemon under nuxt dev: every resume
was a full replay and every history page was the tail. It forwards the
query string now.
This commit is contained in:
vhaudiquet committed 2026-11-08 18:01:30 +01:00
1 parent 744ba8b06c
commit ed79c0cf1b
6 files changed
+337 -49
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

web/claw/app/composables/useClaw.ts
+111 -22
View File
@@ -303,8 +303,16 @@ export async function clawHealth(): Promise<boolean> {
/** Handlers for a turn-event stream. */
export interface StreamHandlers {
/** One turn event, in stream order. */
onEvent: (event: TurnEvent) => void
/** One turn event, in stream order, with the record's seq for it — the
* cursor a reconnect names. Null only for a frame whose `id:` was
* missing or unparseable. */
onEvent: (event: TurnEvent, seq: number | null) => void
/**
* The cursor named events the record no longer holds (a wiped or
* reset daemon). The mirror must go: the tail that follows this
* notice rebuilds it.
*/
onReset?: () => void
/** Connection lifecycle; `reconnecting` follows a drop, with backoff. */
onStatus?: (status: StreamStatus) => void
/**
@@ -315,6 +323,15 @@ export interface StreamHandlers {
onNotice?: (notice: string) => void
}
/** One page of a session's older turns, as `GET /v1/sessions/{key}/
* history` answers it: whole turns, oldest first, each event with the
* seq the record gave it. `more` is false once the page reached the
* session's first event. */
export interface HistoryPage {
events: { seq: number; event: TurnEvent }[]
more: boolean
}
export interface ClawApi {
/** `GET /v1/sessions` — every dispatch key with its status. */
sessions(): Promise<ClawSession[]>
@@ -373,12 +390,33 @@ export interface ClawApi {
/**
* Stream one session's turns. Resolves with a handle once the fetch
* is started; the claw keeps the stream open forever, and a drop is
* retried with backoff until `close()` is called. Every (re)connection
* opens with the session's recorded turns replayed first, then the
* live feed — so a consumer resets its mirror on reconnect and lets
* the stream rebuild it.
* retried with backoff until `close()` is called.
*
* Every frame names the record's seq, so a reconnect resumes from the
* last one delivered and replays only the gap. A first open with no
* `since` names nothing, and the claw sends the newest turn plus the
* live feed; older turns come from `history` as the reader scrolls
* up. A `reset` comment means the cursor named events the record no
* longer holds: the mirror is dropped and the tail that follows
* rebuilds it.
*/
subscribeEvents(key: string, handlers: StreamHandlers): { close: () => void }
subscribeEvents(
key: string,
handlers: StreamHandlers,
options?: StreamOptions,
): { close: () => void }
/**
* `GET /v1/sessions/{key}/history` — one page of whole turns ending
* at `before` (absent: the newest). Pages stack: the next call passes
* the seq of the oldest event held, and stops when `more` is false.
*/
history(key: string, before?: number | null, limit?: number): Promise<HistoryPage>
}
/** How `subscribeEvents` opens: `since` is the seq of the last event the
* caller already holds, or null for "nothing, give me the tail". */
export interface StreamOptions {
since?: number | null
}
/** Build the client bound to one bearer token (null = unguarded claw). */
@@ -443,6 +481,29 @@ export function useClaw(token: Ref<string | null>): ClawApi {
})
}
async function history(
key: string,
before?: number | null,
limit?: number,
): Promise<HistoryPage> {
const params = new URLSearchParams()
if (before !== undefined && before !== null) params.set('before', String(before))
if (limit !== undefined) params.set('limit', String(limit))
const query = params.size ? `?${params}` : ''
const body = (await get(
`/v1/sessions/${encodeURIComponent(key)}/history${query}`,
token.value,
)) as { events?: { seq?: unknown; event?: unknown }[]; more?: unknown }
// The wire shape is trusted only as far as it parses: an event the
// parser does not know is dropped, and its seq with it — the page
// still renders what it could read.
const events = (body.events ?? []).flatMap((entry) => {
const event = parseTurnEvent(JSON.stringify(entry.event))
return event && typeof entry.seq === 'number' ? [{ seq: entry.seq, event }] : []
})
return { events, more: body.more === true }
}
async function providers(): Promise<ProvidersState> {
return (await get('/v1/providers', token.value)) as ProvidersState
}
@@ -536,9 +597,17 @@ export function useClaw(token: Ref<string | null>): ClawApi {
return url
}
function subscribeEvents(key: string, handlers: StreamHandlers): { close: () => void } {
function subscribeEvents(
key: string,
handlers: StreamHandlers,
options?: StreamOptions,
): { close: () => void } {
let closed = false
let attempt = 0
/** The seq of the last event this stream delivered: what a retry
* names as `since`, so a reconnect replays the gap and nothing else.
* Null until a frame arrives, or after a reset drops the mirror. */
let cursor: number | null = options?.since ?? null
/** The attempt in flight; `close` aborts it mid-read. */
let current: AbortController | null = null
/** Set when the watchdog, not the server, cut the attempt — the
@@ -571,8 +640,9 @@ export function useClaw(token: Ref<string | null>): ClawApi {
timedOut = false
rearmSilence()
try {
const resume = cursor === null ? '' : `?since=${cursor}`
const response = await fetch(
`${base()}/v1/sessions/${encodeURIComponent(key)}/events`,
`${base()}/v1/sessions/${encodeURIComponent(key)}/events${resume}`,
{
headers: {
Accept: 'text/event-stream',
@@ -588,16 +658,31 @@ export function useClaw(token: Ref<string | null>): ClawApi {
status('open')
await readFrames(
response.body,
(data) => {
(data, id) => {
rearmSilence()
const event = parseTurnEvent(data)
if (event) handlers.onEvent(event)
if (!event) return
const seq = id !== null && /^\d+$/.test(id) ? Number(id) : null
if (seq !== null) cursor = seq
handlers.onEvent(event, seq)
},
(comment) => {
// Keep-alives come through here too; each one proves the
// path alive, and only the hub's dropped-events notice is
// worth raising.
// path alive.
rearmSilence()
// The record lost what this mirror holds: drop the cursor
// so the next open asks for the tail, and let the consumer
// drop its items before the tail that already followed
// this notice folds in.
if (comment.startsWith('reset')) {
cursor = null
handlers.onReset?.()
return
}
// A skipped run of events is not a hole in the cursor:
// the next frame's id names the seq after it, and a drop
// before then reopens at the old cursor and replays what
// was skipped.
if (comment.startsWith('lagged')) {
handlers.onNotice?.(`stream lagged — ${comment.replace(/^lagged;\s*/, '')}`)
}
@@ -640,6 +725,7 @@ export function useClaw(token: Ref<string | null>): ClawApi {
resolveConsent,
denyConsent,
sendMessage,
history,
providers,
setActive,
models,
@@ -659,13 +745,13 @@ export function useClaw(token: Ref<string | null>): ClawApi {
}
/** One parsed SSE frame: the joined `data:` payload, comments excluded. */
type FrameHandler = (data: string) => void
type FrameHandler = (data: string, id: string | null) => void
/**
* Reads an SSE body into frames. Frames split on a blank line; `data:`
* lines carry payload (joined with newlines) and `:`-prefixed lines are
* comments (the claw's keep-alive and its lag notice). `event:`/`id:`
* lines do not occur on this surface.
* lines carry payload (joined with newlines), `id:` names the event's
* seq, and `:`-prefixed lines are comments (the claw's keep-alive, its
* lag notice, and its reset). `event:` lines do not occur here.
*/
async function readFrames(
body: ReadableStream<Uint8Array>,
@@ -676,17 +762,20 @@ async function readFrames(
const decoder = new TextDecoder()
let buffer = ''
const step = (frame: string) => {
const data = frame
.split('\n')
const lines = frame.split('\n')
const data = lines
.filter((line) => line.startsWith('data:'))
.map((line) => line.replace(/^data: ?/, ''))
.join('\n')
if (data) {
onFrame(data)
const id = lines
.find((line) => line.startsWith('id:'))
?.slice(3)
.trim()
onFrame(data, id ?? null)
return
}
const comment = frame
.split('\n')
const comment = lines
.filter((line) => line.startsWith(':'))
.map((line) => line.slice(1).trimStart())
.join('\n')
web/claw/app/composables/useTranscript.ts
+70 -9
View File
@@ -173,6 +173,19 @@ export function useTranscript() {
const live = ref<TurnBlock | null>(null)
let nextId = 1
/** The record's seq for the oldest event on the mirror, when one is
* known — the cursor an older page ends at. Null until an event with
* a seq lands, and again after a reset. */
const firstSeq = ref<number | null>(null)
/** True while an older page is being fetched and folded in, so a
* scroll that keeps firing at the top does not stack the same page
* twice. */
const loadingOlder = ref(false)
/** Bumped by every page folded in above the mirror. The panel watches
* it to hold the reader's place: the content above the viewport grew,
* so the scroll position moves down by exactly that much. */
const prependCount = ref(0)
/** Locally sent messages the stream has not echoed yet: `addSent`
* pushes here, the `user_message` handler pops the match. Text plus
* image count is the whole signature the stream carries, and that is
@@ -197,8 +210,17 @@ export function useTranscript() {
return { kind: 'attachment', attachment: { ...entry.attachment } }
}
/** Folds one stream event into the mirror. */
function applyEvent(event: TurnEvent): void {
/** True while a page of older turns is folding in: the pending-echo
* match is for live echoes, and a replayed question from weeks ago
* that happens to read the same as one on screen is its own row. */
let replaying = false
/** Folds one stream event into the mirror; `seq` is the record's
* number for it, which puts it at the mirror's top when nothing else
* is there yet. The stream keeps its own resume cursor; this one is
* the walk-back anchor. */
function applyEvent(event: TurnEvent, seq: number | null = null): void {
if (seq !== null && firstSeq.value === null) firstSeq.value = seq
switch (event.type) {
case 'user_message': {
// The replayed question. A live echo of a message this surface
@@ -207,9 +229,12 @@ export function useTranscript() {
// previews and all. Anything else (a replay, another surface's
// message, a steered echo after other events) renders as its
// own row.
const mine = pendingSent.findIndex(
(pending) => pending.text === event.text && pending.images === event.images.length,
)
const mine = replaying
? -1
: pendingSent.findIndex(
(pending) =>
pending.text === event.text && pending.images === event.images.length,
)
if (mine !== -1) {
pendingSent.splice(mine, 1)
return
@@ -419,20 +444,56 @@ export function useTranscript() {
return items.value.some((item) => item.kind === 'attachment' && item.attachment.id === id)
}
/** Session switched, or the stream reopened: the mirror is per-key,
* and the replay that follows rebuilds it either way. The pending
* echoes go too — a replay carries every message, optimistic rows
* included. */
/** Reads one page of older turns (whole turns, oldest first) and
* stacks it above what the mirror holds.
*
* The page folds from empty with the same reducer the stream uses, so
* a replayed consent or attachment card folds exactly as it would
* have live, and then the two lists join: no page is ever half
* applied. A page can end mid-turn if the record's first turn is
* longer than a page; that open tail is dropped, and the next page
* (which walks back from this one's first seq) brings it again with
* the turn's beginning above it. */
function prependOlder(events: { seq: number; event: TurnEvent }[]): void {
const page = events[0]
if (!page) return
const current = items.value
const currentLive = live.value
replaying = true
items.value = []
live.value = null
try {
for (const { event } of events) applyEvent(event)
} finally {
replaying = false
}
items.value = [...items.value, ...current]
live.value = currentLive
// The page sits above everything the mirror held, so it owns the
// first cursor.
firstSeq.value = page.seq
prependCount.value += 1
}
/** Session switched, or the record reset under the mirror: everything
* the mirror holds goes, cursors included. The replay that follows
* rebuilds it. The pending echoes go too — the tail carries every
* message, optimistic rows included. */
function reset(): void {
items.value = []
live.value = null
pendingSent.length = 0
firstSeq.value = null
}
return {
items,
live,
firstSeq,
loadingOlder,
prependCount,
applyEvent,
prependOlder,
addSent,
removeSent,
resolveConsent,