diff --git a/crates/freshell-claude-sidecar/index.mjs b/crates/freshell-claude-sidecar/index.mjs index 30c22cca2..ec790006c 100644 --- a/crates/freshell-claude-sidecar/index.mjs +++ b/crates/freshell-claude-sidecar/index.mjs @@ -37,8 +37,23 @@ // turn), and the sidecar then exits with code 2. // { type:'sdk.configured', sessionId, requestId, ok, settings?, message? } // { type:'sdk.session.init', sessionId, cliSessionId, model, cwd, tools } // durable Claude UUID -// { type:'sdk.assistant', sessionId, content, model } -// { type:'sdk.stream', sessionId, event, parentToolUseId } +// { type:'sdk.assistant', sessionId, content, model, uuid, messageId, parentToolUseId } +// — one per completed content block (blocks of one API message share +// messageId); parentToolUseId is set inside a subagent, else null +// { type:'sdk.stream', sessionId, event, parentToolUseId:null, messageId } +// — the main thread's message_start, content_block_start/stop and +// text/thinking deltas only, stamped with the API message id of the +// message being streamed (a subagent's events, tool input JSON, +// signatures, message_delta and message_stop are not sent) +// { type:'sdk.retry', sessionId, attempt, maxRetries, delayMs, errorStatus, error } +// — the CLI will retry a failed API request (errorStatus null: no HTTP +// response); live only, never a turn end +// { type:'sdk.notice', sessionId, level:'info'|'warning'|'error', code, message, rateLimit? } +// — a non-terminal provider notice: RATE_LIMIT_WARNING / RATE_LIMITED +// (rateLimit {status, resetsAt?, rateLimitType?, utilization?}), +// INFORMATIONAL (the CLI's own warning or suggestion text, or a message +// that stops the turn; its quiet info/notice levels are not sent), +// COMPACTION_FAILED (the CLI's reason); never a turn end // { type:'sdk.result', sessionId, result, durationMs, costUsd, usage } // { type:'sdk.turn.complete', sessionId, at } // EVERY result (any subtype) except // the stopped turn's own aborted result (an accepted user interrupt's mark @@ -51,8 +66,16 @@ // — the command catalog without the terminal-only commands (the pinned // CLI's until a turn's init names its own): from the boot, then // whenever a turn's init or commands_changed changes it -// { type:'sdk.status', sessionId, status } // compacting (an ended stream sends -// no status: the server reports the session as exited) +// { type:'sdk.session.changed', sessionId, reason:'transcript', uuid, parentToolUseId } +// — the CLI writes this message (assistant, user, compaction boundary, +// local command output) to its transcript under `uuid`, about 100 ms +// later; sent right after the message's own frames +// (sdk-message-frames.mjs) +// { type:'sdk.status', sessionId, status } // compacting; running when a +// compaction ends while a send still waits for its result (interim, core +// N2); an ended stream sends no status: the server reports it as exited +// { type:'sdk.session.metadata', sessionId, permissionMode } // the CLI changed its own +// permission mode (an ExitPlanMode approval); the session adopts it first // { type:'sdk.error', sessionId, message, turnFailure? } // turnFailure: a failed turn's result // (sanitized text, no [ede_diagnostic] lines); never for the user's own Stop // { type:'sdk.stream.ended', sessionId, reason, message } // the session's SDK stream ended @@ -103,7 +126,7 @@ import { createInterface } from 'node:readline' import { randomBytes } from 'node:crypto' import { configureSession, userMessageContent } from './session-settings.mjs' -import { resultFrames } from './sdk-message-frames.mjs' +import { createFrameState, frameLogRecord, framesForSdkMessage, resultFrames } from './sdk-message-frames.mjs' import { createTurnCompleteGate } from './turn-complete-gate.mjs' import { checkpointAttemptOf, rememberCheckpointAttempt, takeCheckpointIdentity } from './checkpoint-identity.mjs' import { nextMonotonic } from './monotonic-clock.mjs' @@ -323,34 +346,9 @@ function handleSdkMessage(sessionId, msg) { st.commandCatalog = commands publishCommandsIfChanged(sessionId, st) } - } else if (msg.subtype === 'status' && msg.status === 'compacting') { - emit({ type: 'sdk.status', sessionId, status: 'compacting' }) - } else if (msg.subtype === 'compact_boundary') { - // kata 1wxv ep3-r1 F1: the bare compacting STATUS frame carries no - // trigger — the SDK fires it for an explicit `/compact` AND for its - // own automatic context compaction, and misattributing the automatic - // one to a queued explicit compact wedges the rollback busy gate - // (the phantom compact absorbs the turn's own terminal edge). Only the - // compact COMPLETION boundary discriminates the trigger; relay it. - // Fail toward 'auto' on a missing/unknown trigger so promotion to a - // queued explicit compact never happens without a proven manual run. - const trigger = msg.compact_metadata?.trigger === 'manual' ? 'manual' : 'auto' - emit({ type: 'sdk.compact_boundary', sessionId, trigger }) } break } - case 'assistant': { - const content = msg.message?.content || [] - const blocks = content.map((b) => { - if (b.type === 'text') return { type: 'text', text: b.text } - if (b.type === 'thinking') return { type: 'thinking', thinking: b.thinking } - if (b.type === 'tool_use') return { type: 'tool_use', id: b.id, name: b.name, input: b.input } - if (b.type === 'tool_result') return { type: 'tool_result', tool_use_id: b.tool_use_id, content: b.content, is_error: b.is_error } - return b - }) - emit({ type: 'sdk.assistant', sessionId, content: blocks, model: msg.message?.model }) - break - } case 'result': { // The awaited turn's terminal frame arrived: drop one pending send // (clamped at 0 — an unsolicited result must not underflow the counter). @@ -369,14 +367,29 @@ function handleSdkMessage(sessionId, msg) { } break } - case 'stream_event': { - emit({ type: 'sdk.stream', sessionId, event: msg.event, parentToolUseId: msg.parent_tool_use_id }) - break - } default: - // Unhandled SDK message type — ignored (matches SdkBridge default). break } + // The content kinds (assistant, user, stream events, status, retries and + // notices): a pure mapping, sdk-message-frames.mjs. Any kind it does not + // map is dropped there. + const frames = framesForSdkMessage(sessionId, msg, { + stream: st.frameState, + permissionMode: st.settings.permissionMode, + turnPending: st.pendingResults > 0, + }) + for (const frame of frames) { + // The CLI changed its own permission mode (an ExitPlanMode approval): the + // session adopts it before saying so, so a configure to that mode applies + // nothing and the permission channel sees the mode the CLI is in. + if (frame.type === 'sdk.session.metadata' && frame.permissionMode !== undefined) { + st.settings.permissionMode = frame.permissionMode + st.permissionMode = frame.permissionMode + } + emit(frame) + const record = frameLogRecord(frame) + if (record) log(record.severity, record.event, record.fields) + } } async function consumeStream(sessionId, sdkQuery) { @@ -508,6 +521,8 @@ function handleCreate(req) { turnCompleteGate: createTurnCompleteGate(), pendingResults: 0, pendingCheckpointAttempts: new Map(), + // The content mapping's own per-session state (sdk-message-frames.mjs). + frameState: createFrameState(), } const { iterable, handle } = createInputStream((isCompact) => { if (isCompact) state.handedCompactLikely = true diff --git a/crates/freshell-claude-sidecar/sdk-message-frames.mjs b/crates/freshell-claude-sidecar/sdk-message-frames.mjs index 4a0995166..0e91963b2 100644 --- a/crates/freshell-claude-sidecar/sdk-message-frames.mjs +++ b/crates/freshell-claude-sidecar/sdk-message-frames.mjs @@ -2,7 +2,8 @@ // (docs/plans/2026-10-22-freshclaude-core-runtime.md §3.2 and §9.1 N6). One // function per SDK message kind; index.mjs keeps the side effects (the turn // gate, pendingResults, the attention clock, emit). The core runtime owns the -// `result` mapping below; the transcript work adds the content kinds here. +// `result` mapping below; the transcript work owns the content kinds +// (framesForSdkMessage and the functions it dispatches to, at the end). // // Result shapes are the pinned CLI's (test/integration/providers/ // claude-cli-contract.test.ts): a Stop ends the turn with @@ -74,3 +75,265 @@ export function resultFrames(sessionId, result, { userStopped }) { if (failure) frames.push({ type: 'sdk.error', sessionId, message: failure, turnFailure: true }) return frames } + +// ── content kinds (docs/plans/2026-10-22-freshclaude-transcript-live.md §4.2) ── +// +// framesForSdkMessage maps the SDK messages whose content reaches the server: +// assistant and user messages, stream events, the status frames (compacting, +// a compaction's outcome, the CLI's permission mode), and the provider events +// that never end a turn (an API retry becomes `sdk.retry`; a rate limit, a +// prominent informational message or a failed compaction becomes +// `sdk.notice`). Every other kind maps to no frame here: `result`, +// `system/init`, `commands_changed` and the checkpoint replay binding carry +// side effects, so index.mjs handles them itself. +// +// Transcript announcements: the CLI writes each assistant and user message, +// each compaction boundary and each local command's output to its transcript +// (JSONL) under the message's own uuid (main or subagent file), about 100 ms +// after it emits it. Each such message is followed by +// `sdk.session.changed {reason:'transcript', uuid, parentToolUseId}`, which +// tells the client to refetch the snapshot and which line to expect in it. + +/** + * The per-session state of the content mapping. index.mjs keeps one per + * session and passes it in as `session.stream`. `messageId` is the API + * message id of the main thread's message being streamed (its message_start). + */ +export function createFrameState() { + return { messageId: null } +} + +/** + * The frames for one SDK message of a content kind (none for any other kind). + * `session` is what the sidecar knows about the session when the message + * arrives: `{ stream, permissionMode, turnPending }` — the session's + * createFrameState(), the permission mode it believes the CLI is in, and + * whether a send still waits for its result. + */ +export function framesForSdkMessage(sessionId, msg, session) { + switch (msg?.type) { + case 'assistant': return assistantFrames(sessionId, msg) + case 'user': return transcriptAnnouncement(sessionId, msg) + case 'stream_event': return streamEventFrames(sessionId, msg, session.stream) + case 'rate_limit_event': return rateLimitFrames(sessionId, msg) + case 'system': + switch (msg.subtype) { + case 'status': return [ + ...statusFrames(sessionId, msg), + ...compactionOutcomeFrames(sessionId, msg, session), + ...permissionModeFrames(sessionId, msg, session), + ] + case 'compact_boundary': return compactBoundaryFrames(sessionId, msg) + case 'local_command_output': return transcriptAnnouncement(sessionId, msg) + case 'api_retry': return apiRetryFrames(sessionId, msg) + case 'informational': return informationalFrames(sessionId, msg) + default: return [] + } + default: return [] + } +} + +/** A content block as the server reads it (a thinking block's signature stays behind). */ +function contentBlock(block) { + if (block.type === 'text') return { type: 'text', text: block.text } + if (block.type === 'thinking') return { type: 'thinking', thinking: block.thinking } + if (block.type === 'tool_use') return { type: 'tool_use', id: block.id, name: block.name, input: block.input } + if (block.type === 'tool_result') return { type: 'tool_result', tool_use_id: block.tool_use_id, content: block.content, is_error: block.is_error } + return block +} + +/** The parent tool use of a subagent (sidechain) message, or null for the main thread. */ +function parentToolUseIdOf(msg) { + return typeof msg.parent_tool_use_id === 'string' && msg.parent_tool_use_id ? msg.parent_tool_use_id : null +} + +/** + * The announcement that the CLI writes this message to its transcript under + * its uuid (none for a message without one). + */ +export function transcriptAnnouncement(sessionId, msg) { + if (typeof msg.uuid !== 'string' || !msg.uuid) return [] + return [{ type: 'sdk.session.changed', sessionId, reason: 'transcript', uuid: msg.uuid, parentToolUseId: parentToolUseIdOf(msg) }] +} + +/** + * An assistant message (the CLI sends one per completed content block, all + * sharing the API message id): `sdk.assistant`, then its announcement. The + * uuid and messageId bind a streamed block to the transcript line it becomes. + */ +export function assistantFrames(sessionId, msg) { + const content = msg.message?.content || [] + return [ + { + type: 'sdk.assistant', + sessionId, + content: content.map(contentBlock), + model: msg.message?.model, + uuid: msg.uuid, + messageId: msg.message?.id, + parentToolUseId: parentToolUseIdOf(msg), + }, + ...transcriptAnnouncement(sessionId, msg), + ] +} + +/** The stream events a live pane renders (a block's start and stop, and its text or thinking). */ +const STREAMED_EVENTS = new Set(['message_start', 'content_block_start', 'content_block_stop']) +const STREAMED_DELTAS = new Set(['text_delta', 'thinking_delta']) + +/** + * A streaming event of the main thread: `sdk.stream`, stamped with the API + * message id it belongs to (the one each of its blocks' sdk.assistant then + * carries). Only what a live pane renders is streamed: a subagent's events, + * tool input JSON, signatures and the message's own delta and stop are not. + */ +export function streamEventFrames(sessionId, msg, stream) { + if (parentToolUseIdOf(msg) !== null) return [] + const event = msg.event + if (event?.type === 'message_start') stream.messageId = event.message?.id ?? null + const streamed = STREAMED_EVENTS.has(event?.type) + || (event?.type === 'content_block_delta' && STREAMED_DELTAS.has(event.delta?.type)) + if (!streamed) return [] + return [{ type: 'sdk.stream', sessionId, event, parentToolUseId: null, messageId: stream.messageId }] +} + +/** A status frame: the CLI started compacting. */ +export function statusFrames(sessionId, msg) { + return msg.status === 'compacting' ? [{ type: 'sdk.status', sessionId, status: 'compacting' }] : [] +} + +const COMPACTION_FAILED_COPY = 'Claude could not compact the conversation.' + +/** + * A compaction ended (`status:null` with `compact_result`, T15). A failed one + * is a notice with the CLI's reason (the transcript keeps the same reason as + * a command error); it never ends the turn. Either way, while a send still + * waits for its result the session is running again, not compacting. + * (Interim: core's turn ledger will compute this itself, docs/plans/ + * 2026-10-22-freshclaude-core-runtime.md §9.1 N2.) + */ +export function compactionOutcomeFrames(sessionId, msg, { turnPending }) { + if (msg.status !== null || msg.compact_result === undefined) return [] + const frames = [] + if (msg.compact_result === 'failed') { + const reason = typeof msg.compact_error === 'string' ? msg.compact_error.trim() : '' + frames.push({ type: 'sdk.notice', sessionId, level: 'error', code: 'COMPACTION_FAILED', message: reason || COMPACTION_FAILED_COPY }) + } + if (turnPending) frames.push({ type: 'sdk.status', sessionId, status: 'running' }) + return frames +} + +/** + * The CLI reports the permission mode it is in now (after an ExitPlanMode + * approval, F12, or a mode suggestion the user accepted). A mode that differs + * from the session's becomes `sdk.session.metadata {permissionMode}`; index.mjs + * adopts it as the session's mode, so a later configure to that mode applies + * nothing. + */ +export function permissionModeFrames(sessionId, msg, { permissionMode }) { + if (typeof msg.permissionMode !== 'string' || msg.permissionMode === permissionMode) return [] + return [{ type: 'sdk.session.metadata', sessionId, permissionMode: msg.permissionMode }] +} + +/** + * A compaction finished: its trigger. kata 1wxv ep3-r1 F1: the bare + * compacting STATUS frame carries no trigger — the SDK fires it for an + * explicit `/compact` AND for its own automatic context compaction, and + * misattributing the automatic one to a queued explicit compact wedges the + * rollback busy gate (the phantom compact absorbs the turn's own terminal + * edge). Only the compact COMPLETION boundary discriminates the trigger, so + * the server gets it from here. A missing or unknown trigger counts as + * 'auto', so promotion to a queued explicit compact never happens without a + * proven manual run. + */ +export function compactBoundaryFrames(sessionId, msg) { + const trigger = msg.compact_metadata?.trigger === 'manual' ? 'manual' : 'auto' + return [{ type: 'sdk.compact_boundary', sessionId, trigger }, ...transcriptAnnouncement(sessionId, msg)] +} + +/** + * The CLI will retry a failed API request (live only; never in the + * transcript): `sdk.retry`. `errorStatus` is null for a connection error. + */ +export function apiRetryFrames(sessionId, msg) { + return [{ + type: 'sdk.retry', + sessionId, + attempt: msg.attempt, + maxRetries: msg.max_retries, + delayMs: msg.retry_delay_ms, + errorStatus: msg.error_status ?? null, + error: msg.error, + }] +} + +const RATE_LIMIT_NOTICES = { + allowed_warning: { level: 'warning', code: 'RATE_LIMIT_WARNING', message: 'You are close to your Claude usage limit.' }, + rejected: { level: 'error', code: 'RATE_LIMITED', message: 'You have reached your Claude usage limit.' }, +} +const RATE_LIMIT_FIELDS = ['status', 'resetsAt', 'rateLimitType', 'utilization'] + +/** + * A change in the account's usage limits: a notice when the limit is near + * (`allowed_warning`) or reached (`rejected`), carrying the limit so the + * client can show when it resets. An `allowed` status says nothing. + */ +export function rateLimitFrames(sessionId, msg) { + const info = msg.rate_limit_info ?? {} + const notice = RATE_LIMIT_NOTICES[info.status] + if (!notice) return [] + const rateLimit = {} + for (const key of RATE_LIMIT_FIELDS) if (info[key] !== undefined) rateLimit[key] = info[key] + return [{ type: 'sdk.notice', sessionId, ...notice, rateLimit }] +} + +/** The informational levels the CLI itself shows prominently (SDKInformationalMessage.level). */ +const PROMINENT_INFORMATIONAL_LEVELS = new Set(['warning', 'suggestion']) + +/** + * A banner the CLI shows as plain text (hook feedback, status lines): a + * notice with its text when the CLI shows it prominently (a warning or a + * suggestion) or when it stops the turn. The CLI's own quiet levels stay + * quiet: `info` (transcript mode only) and `notice` (inactive gray text, + * including its per-tool progress lines) never interrupt the pane. + */ +export function informationalFrames(sessionId, msg) { + const stopsTurn = msg.prevent_continuation === true + if (!stopsTurn && !PROMINENT_INFORMATIONAL_LEVELS.has(msg.level)) return [] + const message = typeof msg.content === 'string' ? msg.content.trim() : '' + if (!message) return [] + const level = stopsTurn || msg.level === 'warning' ? 'warning' : 'info' + return [{ type: 'sdk.notice', sessionId, level, code: 'INFORMATIONAL', message }] +} + +const NOTICE_SEVERITY = { info: 'info', warning: 'warn', error: 'error' } + +/** + * The sidecar's log record for a frame worth logging (`{severity, event, + * fields}`), else undefined. Like every sidecar record (log.mjs) it carries + * ids, numbers and error text only: the text of an informational notice is + * logged by its size. + */ +export function frameLogRecord(frame) { + if (frame?.type === 'sdk.retry') { + const { type: _type, ...fields } = frame + return { severity: 'warn', event: 'api.retry', fields } + } + if (frame?.type === 'sdk.notice') { + const fields = { sessionId: frame.sessionId, code: frame.code } + if (frame.code === 'COMPACTION_FAILED') { + fields.error = frame.message + } else if (frame.rateLimit) { + fields.rateLimitStatus = frame.rateLimit.status + if (frame.rateLimit.rateLimitType !== undefined) fields.rateLimitType = frame.rateLimit.rateLimitType + if (frame.rateLimit.resetsAt !== undefined) fields.resetsAt = frame.rateLimit.resetsAt + } else { + fields.bytes = Buffer.byteLength(frame.message ?? '') + } + return { severity: NOTICE_SEVERITY[frame.level] ?? 'info', event: 'notice', fields } + } + if (frame?.type === 'sdk.session.metadata' && frame.permissionMode !== undefined) { + return { severity: 'info', event: 'permission_mode.changed', fields: { sessionId: frame.sessionId, permissionMode: frame.permissionMode } } + } + return undefined +} diff --git a/crates/freshell-freshagent/src/claude.rs b/crates/freshell-freshagent/src/claude.rs index ccdda204f..57a4fac54 100644 --- a/crates/freshell-freshagent/src/claude.rs +++ b/crates/freshell-freshagent/src/claude.rs @@ -716,6 +716,15 @@ struct QuiesceVerdict { handed_compact_likely: bool, } +/// What a claude `freshAgent.session.metadata` frame states (see +/// [`FreshClaudeState::stated_settings`]). +#[derive(Debug, Default, PartialEq, Eq)] +struct StatedSettings { + model: Option, + effort: Option, + permission_mode: Option, +} + #[derive(Default)] struct ClaudeConfiguration { settings: crate::identity_sink::FreshAgentSettings, @@ -4312,15 +4321,7 @@ impl FreshClaudeState { // whose settings CHANGED the live session can converge every // device's model surfaces — the same frame `handle_configure` // emits (a device that missed the configure still lands here). - let settings_before = { - let guard = self.sessions.lock().await; - guard.get(&map_key).map(|s| { - ( - s.configuration.settings.model.clone(), - s.configuration.settings.effort.clone(), - ) - }) - }; + let settings_before = self.stated_settings(&map_key).await; if let Err(err) = self .configure_for_send(&map_key, settings, session_type) .await @@ -4328,24 +4329,13 @@ impl FreshClaudeState { self.send_error(&request_id, "CLAUDE_SETTINGS_FAILED", &err); return; } - let settings_after = { - let guard = self.sessions.lock().await; - guard.get(&map_key).map(|s| { - ( - s.configuration.settings.model.clone(), - s.configuration.settings.effort.clone(), - ) - }) - }; + let settings_after = self.stated_settings(&map_key).await; if settings_before != settings_after { - let (model, effort) = settings_after.unwrap_or((None, None)); - self.broadcast(&crate::session_metadata::session_metadata_frame( - PROVIDER, + self.broadcast_stated_settings( session_type, &session_id, - model.as_deref(), - effort.as_deref(), - )); + settings_after.unwrap_or_default(), + ); } } // Task 4 review (C1b): the destroy target comes from POST-lock session @@ -4660,15 +4650,7 @@ impl FreshClaudeState { ); } - let before = { - let guard = self.sessions.lock().await; - guard.get(&map_key).map(|s| { - ( - s.configuration.settings.model.clone(), - s.configuration.settings.effort.clone(), - ) - }) - }; + let before = self.stated_settings(&map_key).await; if let Err(err) = self .configure_for_send(&map_key, &msg.settings, session_type) .await @@ -4678,28 +4660,46 @@ impl FreshClaudeState { self.emit_fresh_agent_error(&session_id, session_type, "CLAUDE_SETTINGS_FAILED", &err); return; } - let after = { - let guard = self.sessions.lock().await; - guard.get(&map_key).map(|s| { - ( - s.configuration.settings.model.clone(), - s.configuration.settings.effort.clone(), - ) - }) - }; + let after = self.stated_settings(&map_key).await; if before == after { // Idempotent configure: nothing to converge. return; } - let (model, effort) = after.unwrap_or((None, None)); // Broadcast keyed at the CLIENT-ADDRESSED id (the id the configure // named): the client's fold resolves the sessions-map entry by it. + self.broadcast_stated_settings(session_type, &session_id, after.unwrap_or_default()); + } + + /// The settings a `freshAgent.session.metadata` frame states for the + /// session under `map_key` (`None` when it is gone): model, effort and + /// permission mode. The configure and send paths diff these around an + /// apply and broadcast only a change. + async fn stated_settings(&self, map_key: &str) -> Option { + let guard = self.sessions.lock().await; + guard.get(map_key).map(|session| { + let settings = &session.configuration.settings; + StatedSettings { + model: settings.model.clone(), + effort: settings.effort.clone(), + permission_mode: settings.permission_mode.clone(), + } + }) + } + + /// Converge every device's settings surfaces on `stated`. + fn broadcast_stated_settings( + &self, + session_type: &str, + session_id: &str, + stated: StatedSettings, + ) { self.broadcast(&crate::session_metadata::session_metadata_frame( PROVIDER, session_type, - &session_id, - model.as_deref(), - effort.as_deref(), + session_id, + stated.model.as_deref(), + stated.effort.as_deref(), + stated.permission_mode.as_deref(), )); } @@ -6724,6 +6724,19 @@ impl FreshClaudeState { } } } + // The CLI changed its own permission mode (an approved + // ExitPlanMode returns it to the mode it had before plan + // mode). The sidecar already runs in it; the session's + // settings take it too, so a later configure stating the + // same mode applies nothing and the snapshot states it. + Some("sdk.session.metadata") => { + if let Some(mode) = value.get("permissionMode").and_then(Value::as_str) { + if let Some(session) = sessions.lock().await.get_mut(&session_id) { + session.configuration.settings.permission_mode = + Some(mode.to_string()); + } + } + } Some("sdk.configured") => { if let Some(session) = sessions.lock().await.get_mut(&session_id) { if session @@ -6972,6 +6985,7 @@ impl FreshClaudeState { // map key) so an attach-by-durable rebind flips live event routing. let stamp = broadcast_id.lock().expect("broadcast id lock").clone(); let line_type = value.get("type").and_then(Value::as_str).unwrap_or(""); + crate::claude_live_events::log_provider_event(&value, &stamp, &session_type); if let Some(edge) = AttentionEdge::from_sidecar_type(line_type) { // The sidecar's attention edges leave through the one edge // exit (logged there, G14). @@ -7776,6 +7790,11 @@ fn normalize_sdk_type(sdk_type: &str) -> Option<&'static str> { // provider-originated cancellation). "sdk.question.cancelled" => "freshAgent.question.cancelled", "sdk.status" => "freshAgent.status", + // Non-terminal provider events (transcript-live plan P4): an API + // retry and a notice (rate limit, informational banner, failed + // compaction). Neither ends the turn; the client shows them in place. + "sdk.retry" => "freshAgent.retry", + "sdk.notice" => "freshAgent.notice", // `sdk.turn.complete` / `sdk.turn.waiting` are deliberately absent: the // consumer routes them to `emit_attention_edge` (AttentionEdge), the one // logged edge exit, so no other path can broadcast an edge. @@ -8587,6 +8606,9 @@ mod teardown_tests; #[cfg(test)] mod end_of_output_tests; +#[cfg(test)] +mod live_frames_tests; + /// ISO-8601 / RFC-3339 millis-Z timestamp (`new Date().toISOString()`) for error frames. fn now_iso() -> String { let now = SystemTime::now() diff --git a/crates/freshell-freshagent/src/claude/live_frames_tests.rs b/crates/freshell-freshagent/src/claude/live_frames_tests.rs new file mode 100644 index 000000000..d9b93c0cd --- /dev/null +++ b/crates/freshell-freshagent/src/claude/live_frames_tests.rs @@ -0,0 +1,535 @@ +//! The live frames a claude sidecar sends mid-turn (transcript-live plan §4.2, +//! Phase C): API retries, notices, per-message transcript announcements and the +//! CLI's own permission-mode changes, as every client receives them. +//! +//! The sidecar is a small Node fake started through the production create +//! (`handle_create`), so the stdout consumer, its folds and the broadcast are +//! the real ones. A send's text is the list of `sdk.*` frames the fake writes +//! back (as a JSON array), each stamped with the sidecar's session id; a +//! configure is logged and answered the way index.mjs answers it. + +use super::*; + +use super::teardown_tests::capture; + +/// The durable id the fake's session reports. +const DURABLE: &str = "cccccccc-cccc-4ccc-8ccc-cccccccccccc"; + +/// The fake sidecar; `CONFIGURE_LOG` and `DURABLE` are prepended. +const ECHO_SIDECAR: &str = r#" +const fs = require('node:fs') +const out = (frame) => process.stdout.write(JSON.stringify(frame) + '\n') +require('node:readline').createInterface({ input: process.stdin }).on('line', (line) => { + let msg + try { msg = JSON.parse(line) } catch { return } + if (msg.type === 'create') { + const sessionId = 'live-frames-session' + out({ type: 'created', requestId: msg.requestId, sessionId, protocol: 2 }) + out({ type: 'sdk.session.init', sessionId, cliSessionId: DURABLE, model: 'fake-model', cwd: '/tmp', tools: [] }) + } else if (msg.type === 'send') { + for (const frame of JSON.parse(msg.text)) out({ ...frame, sessionId: msg.sessionId }) + } else if (msg.type === 'configure') { + fs.appendFileSync(CONFIGURE_LOG, JSON.stringify(msg) + '\n') + out({ type: 'sdk.configured', sessionId: msg.sessionId, requestId: msg.requestId, ok: true, settings: msg.settings }) + } else if (msg.type === 'shutdown') { + process.exit(0) + } +}) +"#; + +/// One session of the echo fake, created through the production create. +struct EchoSession { + st: FreshClaudeState, + rx: tokio::sync::broadcast::Receiver, + /// The pane's id for the session (its map key and envelope id). + pane: String, + configure_log: std::path::PathBuf, + /// Every frame read off the bus so far, in order. + frames: Vec, + _dir: tempfile::TempDir, + _env: tokio::sync::MutexGuard<'static, ()>, +} + +impl EchoSession { + async fn start(permission_mode: Option<&str>) -> Self { + let env = super::tests::CLAUDE_ENV_LOCK.lock().await; + let dir = tempfile::tempdir().expect("fake sidecar directory"); + let script = dir.path().join("echo-sidecar.cjs"); + let configure_log = dir.path().join("configure.log"); + std::fs::write(&configure_log, "").expect("init the configure log"); + let quoted = |text: &str| serde_json::to_string(text).expect("a JS string"); + std::fs::write( + &script, + format!( + "const CONFIGURE_LOG = {}\nconst DURABLE = {}\n{ECHO_SIDECAR}", + quoted(&configure_log.to_string_lossy()), + quoted(DURABLE), + ), + ) + .expect("write the fake sidecar"); + std::env::set_var("FRESHELL_CLAUDE_SIDECAR", &script); + std::env::set_var("CLAUDE_CONFIG_DIR", dir.path().join("claude")); + let (tx, rx) = tokio::sync::broadcast::channel::(256); + let st = FreshClaudeState::new(Arc::new(tx)); + let mut session = Self { + st, + rx, + pane: String::new(), + configure_log, + frames: Vec::new(), + _dir: dir, + _env: env, + }; + let mut create = create_msg("live-frames"); + create.permission_mode = permission_mode.map(str::to_string); + session.st.handle_create(create, None).await; + let created = session + .await_frame("the created answer", |frame| { + frame["requestId"] == "live-frames" + }) + .await; + assert_eq!(created["type"], "freshAgent.created", "{created}"); + session.pane = created["sessionId"] + .as_str() + .expect("the created session id") + .to_string(); + session + } + + /// The fake writes `frames` back as the CLI side of a send. + async fn sidecar_writes(&mut self, frames: Value) { + self.st + .handle_send(send_msg(&self.pane, &frames.to_string())) + .await; + } + + /// Read frames until one matches `want` (returned) or 20 s pass. + async fn await_frame(&mut self, what: &str, want: impl Fn(&Value) -> bool) -> Value { + if let Some(hit) = self.frames.iter().find(|frame| want(frame)) { + return hit.clone(); + } + let found = tokio::time::timeout(Duration::from_secs(20), async { + loop { + let frame = match self.rx.recv().await { + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(err) => panic!("broadcast recv failed: {err}"), + Ok(raw) => serde_json::from_str::(&raw).expect("a JSON frame"), + }; + self.frames.push(frame.clone()); + if want(&frame) { + return frame; + } + } + }) + .await; + found.unwrap_or_else(|_| panic!("no {what}; frames: {:?}", self.frames)) + } + + /// The first `freshAgent.event` whose inner event has this type. + async fn await_event(&mut self, event_type: &str) -> Value { + let what = format!("a {event_type} event"); + self.await_frame(&what, |frame| { + frame["type"] == "freshAgent.event" && frame["event"]["type"] == event_type + }) + .await + } + + /// The configure lines the fake received. + fn configures(&self) -> Vec { + std::fs::read_to_string(&self.configure_log) + .unwrap_or_default() + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str(line).expect("a configure line")) + .collect() + } + + /// The status the snapshot overlay states for the session. + async fn overlay_status(&self) -> Value { + let mut snapshot = json!({}); + self.st + .apply_snapshot_metadata(&self.pane, &mut snapshot) + .await; + snapshot["status"].clone() + } +} + +impl Drop for EchoSession { + fn drop(&mut self) { + // The script and store die with `_dir`: no later test in this binary + // may inherit paths to them (the env lock is still held here). + remove_claude_env(); + } +} + +/// Removes the process-wide variables these tests point at their tempdirs. +fn remove_claude_env() { + std::env::remove_var("FRESHELL_CLAUDE_SIDECAR"); + std::env::remove_var("CLAUDE_CONFIG_DIR"); +} + +/// Calls [`remove_claude_env`] when dropped, so a failed assertion cannot leak them. +struct ClaudeEnvReset; + +impl Drop for ClaudeEnvReset { + fn drop(&mut self) { + remove_claude_env(); + } +} + +fn create_msg(request_id: &str) -> FreshAgentCreate { + FreshAgentCreate { + naming_handle: None, + observed_epoch: None, + observed_generation: None, + request_id: request_id.to_string(), + session_type: SessionType::Freshclaude, + provider: Some(freshell_protocol::AgentProvider::Claude), + cwd: None, + legacy_restore_context: None, + resume_session_id: None, + session_ref: None, + model: None, + model_selection: None, + permission_mode: None, + sandbox: None, + effort: None, + plugins: None, + tab_id: None, + } +} + +fn send_msg(session_id: &str, text: &str) -> FreshAgentSend { + FreshAgentSend { + provider: freshell_protocol::AgentProvider::Claude, + session_id: session_id.to_string(), + session_type: SessionType::Freshclaude, + text: text.to_string(), + cwd: None, + images: None, + request_id: None, + settings: None, + observed_epoch: None, + observed_generation: None, + checkpoint_intent: None, + } +} + +fn configure_msg(session_id: &str, permission_mode: &str) -> FreshAgentConfigure { + FreshAgentConfigure { + provider: freshell_protocol::AgentProvider::Claude, + session_id: session_id.to_string(), + session_type: SessionType::Freshclaude, + cwd: None, + request_id: Some("live-frames-configure".to_string()), + settings: freshell_protocol::FreshAgentSendSettings { + cwd: None, + model: None, + permission_mode: Some(permission_mode.to_string()), + sandbox: None, + effort: None, + }, + } +} + +/// The captured records whose `event` field is `event`, in order. +fn records( + events: &Arc>>, + event: &str, +) -> Vec<(tracing::Level, capture::Fields)> { + events + .lock() + .expect("capture lock") + .iter() + .filter(|(_, fields)| fields.get("event").map(String::as_str) == Some(event)) + .cloned() + .collect() +} + +/// An API retry reaches every client as `freshAgent.retry`, stamped with the +/// pane's id, with the sidecar's fields intact. It is logged under the +/// pane's id, and it never ends the turn. +#[tokio::test] +async fn an_api_retry_reaches_every_client_as_freshagent_retry_and_keeps_the_turn_running() { + let (events, _capture) = capture::install(); + let mut session = EchoSession::start(None).await; + session + .sidecar_writes(json!([ + { "type": "sdk.assistant", "content": [{ "type": "text", "text": "working" }], "model": "fake-model" }, + { "type": "sdk.retry", "attempt": 1, "maxRetries": 10, "delayMs": 512, "errorStatus": 529, "error": "overloaded" }, + ])) + .await; + + let retry = session.await_event("freshAgent.retry").await; + assert_eq!(retry["sessionId"], json!(session.pane), "{retry}"); + assert_eq!(retry["provider"], "claude"); + assert_eq!(retry["sessionType"], "freshclaude"); + assert_eq!(retry["event"]["attempt"], 1); + assert_eq!(retry["event"]["maxRetries"], 10); + assert_eq!(retry["event"]["delayMs"], 512); + assert_eq!(retry["event"]["errorStatus"], 529); + assert_eq!(retry["event"]["error"], "overloaded"); + + assert_eq!(session.overlay_status().await, "running"); + let logged = records(&events, "freshagent.claude.api_retry"); + assert_eq!(logged.len(), 1, "{logged:?}"); + let (level, fields) = &logged[0]; + assert_eq!(*level, tracing::Level::INFO); + assert_eq!(fields.get("session_id"), Some(&session.pane)); + assert_eq!(fields.get("attempt").map(String::as_str), Some("1")); + assert_eq!(fields.get("error_status").map(String::as_str), Some("529")); + assert_eq!(fields.get("error").map(String::as_str), Some("overloaded")); +} + +/// A notice (rate limit, informational banner, failed compaction) reaches +/// every client as `freshAgent.notice` with its level, code, message and +/// rate limit. It is logged at a severity that follows its level, without +/// its text, and it never ends the turn. +#[tokio::test] +async fn a_notice_reaches_every_client_as_freshagent_notice_and_keeps_the_turn_running() { + let (events, _capture) = capture::install(); + let mut session = EchoSession::start(None).await; + let rate_limit = + json!({ "status": "rejected", "resetsAt": 1_792_656_000, "rateLimitType": "five_hour" }); + session + .sidecar_writes(json!([ + { "type": "sdk.assistant", "content": [{ "type": "text", "text": "working" }], "model": "fake-model" }, + { "type": "sdk.notice", "level": "error", "code": "RATE_LIMITED", "message": "You have reached your Claude usage limit.", "rateLimit": rate_limit }, + { "type": "sdk.notice", "level": "info", "code": "INFORMATIONAL", "message": "a private hook said hello" }, + ])) + .await; + + let notice = session.await_event("freshAgent.notice").await; + assert_eq!(notice["sessionId"], json!(session.pane), "{notice}"); + assert_eq!(notice["event"]["level"], "error"); + assert_eq!(notice["event"]["code"], "RATE_LIMITED"); + assert_eq!( + notice["event"]["message"], + "You have reached your Claude usage limit." + ); + assert_eq!(notice["event"]["rateLimit"], rate_limit); + session + .await_frame("the informational notice", |frame| { + frame["event"]["type"] == "freshAgent.notice" + && frame["event"]["code"] == "INFORMATIONAL" + }) + .await; + + assert_eq!(session.overlay_status().await, "running"); + let logged = records(&events, "freshagent.claude.notice"); + let summary: Vec<(tracing::Level, Option<&str>)> = logged + .iter() + .map(|(level, fields)| (*level, fields.get("code").map(String::as_str))) + .collect(); + assert_eq!( + summary, + vec![ + (tracing::Level::WARN, Some("RATE_LIMITED")), + (tracing::Level::INFO, Some("INFORMATIONAL")), + ] + ); + let (_, rate_limited) = &logged[0]; + assert_eq!(rate_limited.get("session_id"), Some(&session.pane)); + assert_eq!( + rate_limited.get("rate_limit_status").map(String::as_str), + Some("rejected") + ); + let (_, informational) = &logged[1]; + assert_eq!( + informational.get("message_bytes").map(String::as_str), + Some("25") + ); + assert!( + informational + .values() + .all(|value| !value.contains("private hook")), + "a notice's text is never logged: {informational:?}" + ); +} + +/// A per-message transcript announcement passes through to every client +/// unchanged (only the `session-commands` reason is folded server-side). +#[tokio::test] +async fn a_transcript_announcement_reaches_every_client_as_session_changed() { + let mut session = EchoSession::start(None).await; + session + .sidecar_writes(json!([ + { "type": "sdk.session.changed", "reason": "transcript", "uuid": "11111111-1111-4111-8111-111111111111", "parentToolUseId": null }, + { "type": "sdk.session.changed", "reason": "transcript", "uuid": "22222222-2222-4222-8222-222222222222", "parentToolUseId": "toolu_agent1" }, + ])) + .await; + + let announced = session + .await_frame("the subagent's announcement", |frame| { + frame["event"]["type"] == "freshAgent.session.changed" + && frame["event"]["parentToolUseId"] == "toolu_agent1" + }) + .await; + assert_eq!(announced["sessionId"], json!(session.pane)); + assert_eq!(announced["event"]["reason"], "transcript"); + assert_eq!( + announced["event"]["uuid"], + "22222222-2222-4222-8222-222222222222" + ); + let main_thread: Vec<&Value> = session + .frames + .iter() + .filter(|frame| { + frame["event"]["type"] == "freshAgent.session.changed" + && frame["event"]["reason"] == "transcript" + }) + .collect(); + assert_eq!(main_thread.len(), 2, "{main_thread:?}"); + assert_eq!( + main_thread[0]["event"]["uuid"], + "11111111-1111-4111-8111-111111111111" + ); + assert_eq!(main_thread[0]["event"]["parentToolUseId"], Value::Null); +} + +/// The CLI changed its own permission mode (an approved ExitPlanMode): the +/// sidecar reports `sdk.session.metadata {permissionMode}`. Every client +/// receives it, and the session's settings take the new mode, so the next +/// configure (or settings-bearing send) that states the same mode applies +/// nothing, and the snapshot overlay states the new mode. +#[tokio::test] +async fn the_clis_own_permission_mode_change_becomes_the_sessions_setting() { + let (events, _capture) = capture::install(); + let mut session = EchoSession::start(Some("plan")).await; + session + .sidecar_writes(json!([ + { "type": "sdk.session.metadata", "permissionMode": "default" }, + { "type": "sdk.result", "result": "success", "durationMs": 1, "costUsd": 0, "usage": {} }, + ])) + .await; + let metadata = session.await_event("freshAgent.session.metadata").await; + assert_eq!(metadata["sessionId"], json!(session.pane), "{metadata}"); + assert_eq!(metadata["event"]["permissionMode"], "default"); + session.await_event("freshAgent.result").await; + + let mut snapshot = json!({}); + session + .st + .apply_snapshot_metadata(&session.pane, &mut snapshot) + .await; + assert_eq!(snapshot["settings"]["permissionMode"], "default"); + let logged = records(&events, "freshagent.claude.permission_mode_changed"); + assert_eq!(logged.len(), 1, "{logged:?}"); + assert_eq!(logged[0].1.get("session_id"), Some(&session.pane)); + assert_eq!( + logged[0].1.get("permission_mode").map(String::as_str), + Some("default") + ); + + session + .st + .handle_configure(configure_msg(&session.pane, "default")) + .await; + assert_eq!( + session.configures(), + Vec::::new(), + "the session already runs in the mode the configure states" + ); + + // The configure lane itself still works: going back to plan applies. + session + .st + .handle_configure(configure_msg(&session.pane, "plan")) + .await; + let configures = session.configures(); + assert_eq!(configures.len(), 1, "{configures:?}"); + assert_eq!(configures[0]["settings"]["permissionMode"], "plan"); +} + +/// A configure that changes only the permission mode converges every +/// device: the metadata broadcast states the new mode (with the unchanged +/// model and effort pair, as every metadata frame does). +#[tokio::test] +async fn a_configure_that_changes_the_permission_mode_broadcasts_metadata_stating_it() { + let mut session = EchoSession::start(None).await; + session + .st + .handle_configure(configure_msg(&session.pane, "plan")) + .await; + assert_eq!(session.configures().len(), 1); + + let metadata = session.await_event("freshAgent.session.metadata").await; + assert_eq!(metadata["sessionId"], json!(session.pane), "{metadata}"); + assert_eq!(metadata["event"]["permissionMode"], "plan"); + assert_eq!(metadata["event"]["model"], Value::Null); + assert_eq!(metadata["event"]["effort"], Value::Null); +} + +/// The same convergence for a send whose settings change the mode (a device +/// that missed the configure still lands here). +#[tokio::test] +async fn a_send_that_changes_the_permission_mode_broadcasts_metadata_stating_it() { + let mut session = EchoSession::start(Some("plan")).await; + let mut send = send_msg(&session.pane, "[]"); + send.settings = Some(freshell_protocol::FreshAgentSendSettings { + cwd: None, + model: None, + permission_mode: Some("acceptEdits".to_string()), + sandbox: None, + effort: None, + }); + session.st.handle_send(send).await; + assert_eq!(session.configures().len(), 1); + + let metadata = session.await_event("freshAgent.session.metadata").await; + assert_eq!(metadata["sessionId"], json!(session.pane), "{metadata}"); + assert_eq!(metadata["event"]["permissionMode"], "acceptEdits"); +} + +/// A lane whose sidecars get an explicit environment (a test rig) serves +/// snapshots from the transcript store that environment names, the store its +/// CLI writes, not from this process's. +#[tokio::test] +async fn snapshots_read_the_store_the_lanes_sidecars_write() { + let _env = super::tests::CLAUDE_ENV_LOCK.lock().await; + let _reset = ClaudeEnvReset; + let process_store = tempfile::tempdir().expect("this process's store"); + std::env::set_var("CLAUDE_CONFIG_DIR", process_store.path()); + let sidecar_store = tempfile::tempdir().expect("the sidecars' store"); + let thread = "dddddddd-dddd-4ddd-8ddd-dddddddddddd"; + let project = sidecar_store.path().join("projects").join("-ws"); + std::fs::create_dir_all(&project).unwrap(); + std::fs::write( + project.join(format!("{thread}.jsonl")), + format!( + "{}\n", + json!({ + "type": "user", "uuid": "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee", + "sessionId": thread, "message": { "role": "user", "content": "hello" }, + }) + ), + ) + .unwrap(); + let (tx, _rx) = tokio::sync::broadcast::channel::(16); + let st = FreshClaudeState::new(Arc::new(tx)).with_sidecar_environment([ + ( + "CLAUDE_CONFIG_DIR".to_string(), + sidecar_store.path().to_string_lossy().into_owned(), + ), + ("HOME".to_string(), "/nonexistent-home".to_string()), + ]); + + let snapshot = st + .get_snapshot(SessionType::Freshclaude, thread) + .await + .expect("the snapshot reads the sidecars' store"); + assert_eq!( + snapshot["turns"][0]["items"][0]["text"], "hello", + "{snapshot}" + ); +} + +#[tokio::test] +async fn an_echo_session_leaves_the_process_environment_as_it_found_it() { + // The fake's script and store live in a tempdir that dies with the session: + // a later test in this binary must not inherit paths to them. + drop(EchoSession::start(None).await); + let _env = super::tests::CLAUDE_ENV_LOCK.lock().await; + assert_eq!(std::env::var_os("FRESHELL_CLAUDE_SIDECAR"), None); + assert_eq!(std::env::var_os("CLAUDE_CONFIG_DIR"), None); +} diff --git a/crates/freshell-freshagent/src/claude_live_events.rs b/crates/freshell-freshagent/src/claude_live_events.rs new file mode 100644 index 000000000..17cb14b40 --- /dev/null +++ b/crates/freshell-freshagent/src/claude_live_events.rs @@ -0,0 +1,110 @@ +//! The claude sidecar's non-terminal provider events (transcript-live plan +//! §4.2, principle P4): an API request the CLI retries (`sdk.retry`), a +//! notice (`sdk.notice`: a rate limit, an informational banner, a failed +//! compaction) and the CLI's own change of permission mode +//! (`sdk.session.metadata {permissionMode}`). They are broadcast to every +//! client unchanged (as `freshAgent.retry` / `freshAgent.notice` / +//! `freshAgent.session.metadata`), and none ends the turn or touches the +//! session's status. +//! +//! [`log_provider_event`] logs them under the id the pane knows the session +//! by. The sidecar logs the same events under its own id (its stderr records +//! are re-emitted as `freshagent.claude.sidecar_log`, G14); these records tie +//! them to the pane. Like the sidecar's records they carry ids, numbers and +//! codes only: a notice's text is logged by its size, because an +//! informational banner can carry user content. + +use serde_json::Value; + +const PROVIDER: &str = "claude"; + +/// Log one sidecar line if it is a retry, a notice or a permission-mode +/// change; other lines are ignored. `session_id` is the envelope stamp the +/// broadcast carries. +pub(crate) fn log_provider_event(line: &Value, session_id: &str, session_type: &str) { + match line.get("type").and_then(Value::as_str) { + Some("sdk.retry") => log_retry(line, session_id, session_type), + Some("sdk.notice") => log_notice(line, session_id, session_type), + Some("sdk.session.metadata") => { + if let Some(permission_mode) = line.get("permissionMode").and_then(Value::as_str) { + tracing::info!( + target: "freshell_freshagent::claude", + event = "freshagent.claude.permission_mode_changed", + provider = PROVIDER, + session_id, + session_type, + permission_mode, + "the claude CLI changed its own permission mode; the session's settings follow" + ); + } + } + _ => {} + } +} + +fn log_retry(line: &Value, session_id: &str, session_type: &str) { + let attempt = line.get("attempt").and_then(Value::as_u64); + let max_retries = line.get("maxRetries").and_then(Value::as_u64); + let delay_ms = line.get("delayMs").and_then(Value::as_u64); + let error_status = line.get("errorStatus").and_then(Value::as_u64); + let error = line.get("error").and_then(Value::as_str).unwrap_or(""); + // Info, not warn: the sidecar's own record of the retry is the warning. + tracing::info!( + target: "freshell_freshagent::claude", + event = "freshagent.claude.api_retry", + provider = PROVIDER, + session_id, + session_type, + attempt, + max_retries, + delay_ms, + error_status, + error, + "the claude CLI is retrying a failed API request" + ); +} + +fn log_notice(line: &Value, session_id: &str, session_type: &str) { + let level = line.get("level").and_then(Value::as_str).unwrap_or("info"); + let code = line.get("code").and_then(Value::as_str).unwrap_or(""); + let message_bytes = line + .get("message") + .and_then(Value::as_str) + .map_or(0, str::len); + let rate_limit = line.get("rateLimit"); + let rate_limit_field = |key: &str| { + rate_limit + .and_then(|limit| limit.get(key)) + .map(|value| match value { + Value::String(text) => text.clone(), + other => other.to_string(), + }) + }; + let rate_limit_status = rate_limit_field("status"); + let rate_limit_type = rate_limit_field("rateLimitType"); + let resets_at = rate_limit_field("resetsAt"); + // A provider notice is never a server fault: an error-level notice (a + // reached usage limit, a failed compaction) is a warning here. + macro_rules! notice { + ($macro:ident) => { + tracing::$macro!( + target: "freshell_freshagent::claude", + event = "freshagent.claude.notice", + provider = PROVIDER, + session_id, + session_type, + level, + code, + message_bytes, + rate_limit_status = rate_limit_status.as_deref(), + rate_limit_type = rate_limit_type.as_deref(), + resets_at = resets_at.as_deref(), + "the claude CLI reported a notice" + ) + }; + } + match level { + "error" | "warning" => notice!(warn), + _ => notice!(info), + } +} diff --git a/crates/freshell-freshagent/src/claude_projection_cache.rs b/crates/freshell-freshagent/src/claude_projection_cache.rs new file mode 100644 index 000000000..f05cba8dd --- /dev/null +++ b/crates/freshell-freshagent/src/claude_projection_cache.rs @@ -0,0 +1,308 @@ +//! Incremental claude transcript projection (transcript-live plan §4.3, C4). +//! +//! A live pane refetches its snapshot after every message the CLI persists, +//! and every connected device does the same. Re-reading and re-projecting +//! the whole transcript each time costs O(transcript) per message; the CLI +//! only ever appends, so this cache keeps a projector per transcript and +//! feeds it only the complete lines added since the last read. +//! +//! Each entry holds the file's identity (its inode on unix), the byte offset +//! the projector has consumed (always just after a newline), the 4 KiB before +//! that offset, and the projector. A read: +//! - parses the complete lines after the offset into the cached projector +//! when the file still has those 4 KiB in place (it grew, or is unchanged); +//! - parses the whole file afresh when the file is new to the cache, shorter +//! than the offset, has different bytes before the offset (a rewrite), or +//! is a different file at the same path (a replace); +//! - never consumes an unterminated last line (a write in progress): a tail +//! that already parses as JSON is projected for this read only, so the +//! result always equals a full parse of the file as read. +//! +//! Subagent transcripts are attached per read through the caller's +//! [`SubagentSource`] (the filesystem one has its own per-file cache), so a +//! subagent that grows while the main transcript does not is still reflected. +//! Reads are blocking; callers run them off the async runtime. + +use std::collections::HashMap; +use std::io::{Read, Seek, SeekFrom}; +use std::path::{Path, PathBuf}; +use std::sync::{LazyLock, Mutex, PoisonError}; +use std::time::Instant; + +use crate::claude_projection::{ClaudeProjection, ClaudeTranscriptProjector, SubagentSource}; + +/// How many transcripts the process keeps projectors for (the open panes' +/// sessions, roughly). +const CACHED_TRANSCRIPTS: usize = 8; + +/// How many bytes before the consumed offset must be unchanged for an +/// incremental read. +const FINGERPRINT_BYTES: u64 = 4096; + +/// Why a read parsed the whole file. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum FullReason { + /// The cache had no projector for this transcript. + Uncached, + /// The file is shorter than what was consumed. + Shrunk, + /// The bytes before the consumed offset changed. + Rewritten, + /// Another file now sits at the path. + Replaced, +} + +impl FullReason { + fn label(self) -> &'static str { + match self { + Self::Uncached => "uncached", + Self::Shrunk => "shrunk", + Self::Rewritten => "rewritten", + Self::Replaced => "replaced", + } + } +} + +/// What one read parsed. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum ReadOutcome { + /// No complete line was added since the last read. + Unchanged, + /// Only the complete lines added since the last read (`bytes` of them). + Appended { bytes: u64 }, + /// The whole file (its `bytes` of complete lines). + Full { reason: FullReason, bytes: u64 }, +} + +/// A read's projection and what it took. +pub(crate) struct TranscriptRead { + pub projection: ClaudeProjection, + pub outcome: ReadOutcome, +} + +/// What identifies a file beyond its path. +#[derive(Clone, Copy, PartialEq, Eq)] +struct FileId { + #[cfg(unix)] + inode: u64, +} + +impl FileId { + fn of(_metadata: &std::fs::Metadata) -> Self { + #[cfg(unix)] + use std::os::unix::fs::MetadataExt; + Self { + #[cfg(unix)] + inode: _metadata.ino(), + } + } +} + +struct Entry { + file: FileId, + /// Bytes consumed (just after a newline). + offset: u64, + /// The [`FINGERPRINT_BYTES`] before `offset`. + fingerprint: Vec, + projector: ClaudeTranscriptProjector, + last_used: u64, +} + +#[derive(Default)] +struct CacheState { + entries: HashMap<(PathBuf, String), Entry>, + /// Bumped on every store; an entry's `last_used` orders eviction. + clock: u64, +} + +/// Projectors by transcript, least recently used dropped past the capacity. +pub(crate) struct ProjectionCache { + capacity: usize, + state: Mutex, +} + +/// The process-wide cache every snapshot read shares. +static PROJECTIONS: LazyLock = + LazyLock::new(|| ProjectionCache::new(CACHED_TRANSCRIPTS)); + +/// Project the transcript at `path` through the process-wide cache, logging +/// what the read took. Blocking. +pub(crate) fn read_transcript( + path: &Path, + thread_id: &str, + subagents: &dyn SubagentSource, +) -> std::io::Result { + let started = Instant::now(); + let read = PROJECTIONS.read(path, thread_id, subagents)?; + let elapsed_ms = started.elapsed().as_millis() as u64; + match read.outcome { + ReadOutcome::Unchanged => tracing::debug!( + target: "freshell_freshagent::claude", + event = "freshagent.claude.transcript_projection", + outcome = "unchanged", + thread_id, + elapsed_ms, + "projected a claude transcript from the cache" + ), + ReadOutcome::Appended { bytes } => tracing::debug!( + target: "freshell_freshagent::claude", + event = "freshagent.claude.transcript_projection", + outcome = "appended", + thread_id, + bytes, + elapsed_ms, + "projected the lines appended to a claude transcript" + ), + ReadOutcome::Full { reason, bytes } => tracing::debug!( + target: "freshell_freshagent::claude", + event = "freshagent.claude.transcript_projection", + outcome = "full", + reason = reason.label(), + thread_id, + bytes, + elapsed_ms, + "projected a whole claude transcript" + ), + } + Ok(read.projection) +} + +impl ProjectionCache { + pub(crate) fn new(capacity: usize) -> Self { + Self { + capacity: capacity.max(1), + state: Mutex::new(CacheState::default()), + } + } + + /// The projection of the transcript at `path` (see the module docs). + pub(crate) fn read( + &self, + path: &Path, + thread_id: &str, + subagents: &dyn SubagentSource, + ) -> std::io::Result { + let key = (path.to_path_buf(), thread_id.to_string()); + let mut transcript = std::fs::File::open(path)?; + let metadata = transcript.metadata()?; + let file = FileId::of(&metadata); + // The entry leaves the cache while this read works on it: a concurrent + // read of the same transcript parses in full (correct, only slower). + let cached = self.take(&key); + let (mut projector, start, full) = match cached { + None => ( + ClaudeTranscriptProjector::new(thread_id), + 0, + Some(FullReason::Uncached), + ), + Some(entry) => match Self::continuation(&mut transcript, &metadata, file, &entry)? { + None => (entry.projector, entry.offset, None), + Some(reason) => (ClaudeTranscriptProjector::new(thread_id), 0, Some(reason)), + }, + }; + + transcript.seek(SeekFrom::Start(start))?; + let mut bytes = Vec::new(); + transcript.read_to_end(&mut bytes)?; + let complete = bytes + .iter() + .rposition(|byte| *byte == b'\n') + .map_or(0, |newline| newline + 1); + let lines = std::str::from_utf8(&bytes[..complete]) + .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))?; + for line in lines.split('\n') { + projector.push_line(line); + } + let offset = start + complete as u64; + let fingerprint = Self::fingerprint(&mut transcript, offset)?; + + // An unterminated last line is a write in progress: a tail that + // already parses counts for this read only, never for the cache. + let mut result = projector.clone(); + if let Ok(tail) = std::str::from_utf8(&bytes[complete..]) { + result.push_line(tail); + } + self.store( + key, + Entry { + file, + offset, + fingerprint, + projector, + last_used: 0, + }, + ); + let parsed = complete as u64; + let outcome = match full { + Some(reason) => ReadOutcome::Full { + reason, + bytes: parsed, + }, + None if parsed == 0 => ReadOutcome::Unchanged, + None => ReadOutcome::Appended { bytes: parsed }, + }; + Ok(TranscriptRead { + projection: result.finish_with(subagents), + outcome, + }) + } + + /// Whether `entry` continues the file as it is now: `None` when it does, + /// else why the file must be parsed in full. + fn continuation( + transcript: &mut std::fs::File, + metadata: &std::fs::Metadata, + file: FileId, + entry: &Entry, + ) -> std::io::Result> { + if entry.file != file { + return Ok(Some(FullReason::Replaced)); + } + if metadata.len() < entry.offset { + return Ok(Some(FullReason::Shrunk)); + } + if Self::fingerprint(transcript, entry.offset)? != entry.fingerprint { + return Ok(Some(FullReason::Rewritten)); + } + Ok(None) + } + + /// The [`FINGERPRINT_BYTES`] before `offset`. + fn fingerprint(transcript: &mut std::fs::File, offset: u64) -> std::io::Result> { + let start = offset.saturating_sub(FINGERPRINT_BYTES); + transcript.seek(SeekFrom::Start(start))?; + let mut bytes = Vec::with_capacity((offset - start) as usize); + transcript + .by_ref() + .take(offset - start) + .read_to_end(&mut bytes)?; + Ok(bytes) + } + + fn take(&self, key: &(PathBuf, String)) -> Option { + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + state.entries.remove(key) + } + + fn store(&self, key: (PathBuf, String), mut entry: Entry) { + let mut state = self.state.lock().unwrap_or_else(PoisonError::into_inner); + state.clock += 1; + entry.last_used = state.clock; + state.entries.insert(key, entry); + while state.entries.len() > self.capacity { + let Some(oldest) = state + .entries + .iter() + .min_by_key(|(_, entry)| entry.last_used) + .map(|(key, _)| key.clone()) + else { + break; + }; + state.entries.remove(&oldest); + } + } +} + +#[cfg(test)] +#[path = "claude_projection_cache_tests.rs"] +mod tests; diff --git a/crates/freshell-freshagent/src/claude_projection_cache_tests.rs b/crates/freshell-freshagent/src/claude_projection_cache_tests.rs new file mode 100644 index 000000000..d9c3c3046 --- /dev/null +++ b/crates/freshell-freshagent/src/claude_projection_cache_tests.rs @@ -0,0 +1,328 @@ +//! The incremental transcript projection (C4) over REAL pinned-CLI +//! transcripts (`test/fixtures/fresh-agent/claude/real-cli-2.1.237/`): every +//! read equals a full parse of the file as it stands, and appends are parsed +//! incrementally. + +use std::io::Write as _; + +use serde_json::{json, Value}; + +use super::*; +use crate::claude_projection::{FsSubagents, NoSubagents}; +use crate::claude_snapshot::real_cli_tests::{all_items, real_cli_fixture}; + +const THREAD: &str = "s"; + +/// A projection as one comparable value. +fn summary(projection: ClaudeProjection) -> Value { + json!({ + "turns": projection.turns, + "diffs": projection.diffs, + "tokenUsage": projection.token_usage, + "transcriptLineUuids": projection.transcript_line_uuids, + }) +} + +/// A full parse of `content`, the way the snapshot projected before the cache. +fn full_parse(content: &str, subagents: &dyn SubagentSource) -> Value { + summary(ClaudeTranscriptProjector::project_with( + THREAD, content, subagents, + )) +} + +fn read(cache: &ProjectionCache, path: &Path) -> (Value, ReadOutcome) { + let read = cache + .read(path, THREAD, &NoSubagents) + .expect("the transcript reads"); + (summary(read.projection), read.outcome) +} + +fn append(path: &Path, bytes: &str) { + std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path) + .expect("open the transcript") + .write_all(bytes.as_bytes()) + .expect("append to the transcript"); +} + +/// Every split point: each line boundary and the middle of each line. +fn split_points(content: &str) -> Vec { + let mut points = Vec::new(); + let mut start = 0; + for (index, byte) in content.bytes().enumerate() { + if byte == b'\n' { + let mut middle = start + (index - start) / 2; + while !content.is_char_boundary(middle) { + middle -= 1; + } + points.push(middle); + points.push(index + 1); + start = index + 1; + } + } + points +} + +#[test] +fn a_transcript_read_as_it_grows_projects_like_a_full_parse_and_parses_only_what_was_appended() { + for content in [ + real_cli_fixture!("tools-sync"), + real_cli_fixture!("agent-async"), + real_cli_fixture!("compact-ok"), + ] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.jsonl"); + let cache = ProjectionCache::new(4); + let mut written = 0; + let mut consumed = 0; + for (step, point) in split_points(content).into_iter().enumerate() { + append(&path, &content[written..point]); + written = point; + let (projection, outcome) = read(&cache, &path); + assert_eq!( + projection, + full_parse(&content[..point], &NoSubagents), + "after {point} bytes" + ); + let complete = content[..point].rfind('\n').map_or(0, |at| at + 1); + let expected = if step == 0 { + ReadOutcome::Full { + reason: FullReason::Uncached, + bytes: complete as u64, + } + } else if complete == consumed { + ReadOutcome::Unchanged + } else { + ReadOutcome::Appended { + bytes: (complete - consumed) as u64, + } + }; + assert_eq!(outcome, expected, "after {point} bytes"); + consumed = complete; + } + } +} + +#[test] +fn an_unterminated_last_line_is_never_consumed_until_its_newline_lands() { + let content = real_cli_fixture!("tools-sync"); + let lines: Vec<&str> = content.lines().collect(); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.jsonl"); + let cache = ProjectionCache::new(4); + + // A complete JSON line still missing its newline counts for this read... + let head = format!("{}\n{}", lines[..7].join("\n"), lines[7]); + append(&path, &head); + let (projection, _) = read(&cache, &path); + assert_eq!(projection, full_parse(&head, &NoSubagents)); + // ...but is not consumed: once its newline and the next line land, the + // read parses it with them, exactly once. + let more = format!("\n{}\n", lines[8]); + append(&path, &more); + let (projection, outcome) = read(&cache, &path); + let so_far = format!("{head}{more}"); + assert_eq!(projection, full_parse(&so_far, &NoSubagents)); + assert_eq!( + outcome, + ReadOutcome::Appended { + bytes: (lines[7].len() + more.len()) as u64, + } + ); + + // A torn line (a write in progress) is invisible until complete. + let torn = &lines[9][..lines[9].len() / 2]; + append(&path, torn); + let (projection, outcome) = read(&cache, &path); + assert_eq!(projection, full_parse(&so_far, &NoSubagents)); + assert_eq!(outcome, ReadOutcome::Unchanged); + let rest = format!("{}\n", &lines[9][lines[9].len() / 2..]); + append(&path, &rest); + let (projection, outcome) = read(&cache, &path); + assert_eq!( + projection, + full_parse(&format!("{so_far}{torn}{rest}"), &NoSubagents) + ); + assert_eq!( + outcome, + ReadOutcome::Appended { + bytes: (lines[9].len() + 1) as u64, + } + ); +} + +#[test] +fn a_rewritten_or_shrunk_transcript_is_parsed_in_full() { + let content = real_cli_fixture!("tools-sync"); + let lines: Vec<&str> = content.lines().collect(); + let joined = |lines: &[&str]| { + lines + .iter() + .map(|line| format!("{line}\n")) + .collect::() + }; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.jsonl"); + let cache = ProjectionCache::new(4); + std::fs::write(&path, joined(&lines[..20])).unwrap(); + read(&cache, &path); + + // Rewritten in place: the last consumed line changed and the file grew, + // so the new bytes alone would leave the old line in the projection. + let mut rewritten: Vec<&str> = lines[..19].to_vec(); + rewritten.push(lines[21]); + rewritten.push(lines[22]); + rewritten.push(lines[23]); + let rewritten = joined(&rewritten); + std::fs::write(&path, &rewritten).unwrap(); + let (projection, outcome) = read(&cache, &path); + assert_eq!(projection, full_parse(&rewritten, &NoSubagents)); + assert!( + matches!( + outcome, + ReadOutcome::Full { + reason: FullReason::Rewritten, + .. + } + ), + "{outcome:?}" + ); + + // Shrunk (a rollback truncation, say). + let shrunk = joined(&lines[..5]); + std::fs::write(&path, &shrunk).unwrap(); + let (projection, outcome) = read(&cache, &path); + assert_eq!(projection, full_parse(&shrunk, &NoSubagents)); + assert!( + matches!( + outcome, + ReadOutcome::Full { + reason: FullReason::Shrunk, + .. + } + ), + "{outcome:?}" + ); +} + +/// Another file renamed over the path is a different transcript, even when +/// it starts with the same bytes. +#[cfg(unix)] +#[test] +fn a_transcript_replaced_by_another_file_is_parsed_in_full() { + let content = real_cli_fixture!("tools-sync"); + let lines: Vec<&str> = content.lines().collect(); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.jsonl"); + let cache = ProjectionCache::new(4); + std::fs::write(&path, format!("{}\n", lines[..10].join("\n"))).unwrap(); + read(&cache, &path); + + let replacement = dir.path().join("next.jsonl"); + std::fs::write(&replacement, content).unwrap(); + std::fs::rename(&replacement, &path).unwrap(); + let (projection, outcome) = read(&cache, &path); + assert_eq!(projection, full_parse(content, &NoSubagents)); + assert!( + matches!( + outcome, + ReadOutcome::Full { + reason: FullReason::Replaced, + .. + } + ), + "{outcome:?}" + ); +} + +/// The main transcript is unchanged while the subagent works: its file grows, +/// and the delegation's activity follows. +#[test] +fn a_growing_subagent_transcript_is_reflected_while_the_main_one_is_unchanged() { + let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../test/fixtures/fresh-agent/claude/real-cli-2.1.237/tools-sync"); + let main = real_cli_fixture!("tools-sync"); + let main_lines: Vec<&str> = main.lines().collect(); + // Up to the Agent call (the delegation is running). + let agent_call = main_lines + .iter() + .position(|line| line.contains("\"name\":\"Agent\"")) + .expect("the Agent call"); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t.jsonl"); + std::fs::write(&path, format!("{}\n", main_lines[..=agent_call].join("\n"))).unwrap(); + let subagents_dir = dir.path().join("t").join("subagents"); + std::fs::create_dir_all(&subagents_dir).unwrap(); + let recorded = fixture.join("session").join("subagents"); + let agent = "agent-a2645ac0e225b257e"; + std::fs::copy( + recorded.join(format!("{agent}.meta.json")), + subagents_dir.join(format!("{agent}.meta.json")), + ) + .unwrap(); + let subagent = std::fs::read_to_string(recorded.join(format!("{agent}.jsonl"))).unwrap(); + let subagent_lines: Vec<&str> = subagent.lines().collect(); + let read_call = subagent_lines + .iter() + .position(|line| line.contains("\"type\":\"tool_use\"")) + .expect("the subagent's Read call"); + let subagent_path = subagents_dir.join(format!("{agent}.jsonl")); + std::fs::write( + &subagent_path, + format!("{}\n", subagent_lines[..=read_call].join("\n")), + ) + .unwrap(); + + let cache = ProjectionCache::new(4); + let activity = |cache: &ProjectionCache| { + let read = cache + .read(&path, THREAD, &FsSubagents::for_transcript(&path)) + .expect("the transcript reads"); + let delegation = all_items(&read.projection.turns) + .into_iter() + .find(|item| item["kind"] == "task_delegation") + .cloned() + .expect("the delegation"); + (delegation["activity"].clone(), read.outcome) + }; + let (running, _) = activity(&cache); + assert_eq!(running[0]["status"], "running", "{running}"); + + append( + &subagent_path, + &format!("{}\n", subagent_lines[read_call + 1..].join("\n")), + ); + let (completed, outcome) = activity(&cache); + assert_eq!(outcome, ReadOutcome::Unchanged); + assert_eq!(completed[0]["status"], "completed", "{completed}"); +} + +/// Reads of more transcripts than the capacity drop the least recently used +/// projector (its next read is a full one again), not the busy ones. +#[test] +fn past_its_capacity_the_cache_drops_the_least_recently_read_transcript() { + let content = real_cli_fixture!("tools-sync"); + let dir = tempfile::tempdir().unwrap(); + let paths: Vec = (0..3) + .map(|n| { + let path = dir.path().join(format!("t{n}.jsonl")); + std::fs::write(&path, content).unwrap(); + path + }) + .collect(); + let cache = ProjectionCache::new(2); + read(&cache, &paths[0]); + read(&cache, &paths[1]); + assert_eq!(read(&cache, &paths[0]).1, ReadOutcome::Unchanged); + read(&cache, &paths[2]); + assert_eq!(read(&cache, &paths[0]).1, ReadOutcome::Unchanged); + assert!(matches!( + read(&cache, &paths[1]).1, + ReadOutcome::Full { + reason: FullReason::Uncached, + .. + } + )); +} diff --git a/crates/freshell-freshagent/src/claude_snapshot.rs b/crates/freshell-freshagent/src/claude_snapshot.rs index 5286abc8d..02d7a47c8 100644 --- a/crates/freshell-freshagent/src/claude_snapshot.rs +++ b/crates/freshell-freshagent/src/claude_snapshot.rs @@ -1988,9 +1988,12 @@ async fn pending_snapshot( } /// Read and project the transcript at `path` as `thread_id`'s, with its -/// subagent activity, off the async runtime. `fork_at` projects only the -/// conversation a lazy fork of it holds ([`fork_prefix_transcript`]); a fork -/// point the transcript does not hold answers `NotFound`. +/// subagent activity, off the async runtime. The whole transcript is projected +/// incrementally (a live pane refetches after every message): only the lines +/// appended since the last read are parsed +/// ([`crate::claude_projection_cache::read_transcript`]). `fork_at` projects +/// only the conversation a lazy fork of it holds ([`fork_prefix_transcript`]); +/// a fork point the transcript does not hold answers `NotFound`. async fn project_transcript_file( path: PathBuf, thread_id: &str, @@ -2003,17 +2006,20 @@ async fn project_transcript_file( .and_then(|m| m.modified().ok()) .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) .map(|d| d.as_millis() as i64); - let content = std::fs::read_to_string(&path) - .map_err(|error| ClaudeSnapshotError::Io(error.to_string()))?; - let content = match fork_at { - None => content, - Some(at) => fork_prefix_transcript(&content, at.as_deref()) - .ok_or(ClaudeSnapshotError::NotFound)?, - }; let subagents = crate::claude_projection::FsSubagents::for_transcript(&path); - let projection = crate::claude_projection::ClaudeTranscriptProjector::project_with( - &thread, &content, &subagents, - ); + let projection = match fork_at { + None => crate::claude_projection_cache::read_transcript(&path, &thread, &subagents) + .map_err(|error| ClaudeSnapshotError::Io(error.to_string()))?, + Some(at) => { + let content = std::fs::read_to_string(&path) + .map_err(|error| ClaudeSnapshotError::Io(error.to_string()))?; + let content = fork_prefix_transcript(&content, at.as_deref()) + .ok_or(ClaudeSnapshotError::NotFound)?; + crate::claude_projection::ClaudeTranscriptProjector::project_with( + &thread, &content, &subagents, + ) + } + }; Ok((mtime_ms, projection)) }) .await diff --git a/crates/freshell-freshagent/src/codex.rs b/crates/freshell-freshagent/src/codex.rs index 5df35a520..cafb9f94e 100644 --- a/crates/freshell-freshagent/src/codex.rs +++ b/crates/freshell-freshagent/src/codex.rs @@ -3704,6 +3704,7 @@ impl FreshCodexState { &session_id, Some(model.as_str()), effort.as_deref(), + None, )); } @@ -4037,6 +4038,7 @@ impl FreshCodexState { &session_id, Some(model.as_str()), effort.as_deref(), + None, )); } diff --git a/crates/freshell-freshagent/src/lib.rs b/crates/freshell-freshagent/src/lib.rs index 6535e0da9..460b7d627 100644 --- a/crates/freshell-freshagent/src/lib.rs +++ b/crates/freshell-freshagent/src/lib.rs @@ -40,7 +40,9 @@ pub mod claude; pub(crate) mod claude_dead_sessions; pub(crate) mod claude_handshake; pub(crate) mod claude_launch; +pub(crate) mod claude_live_events; pub(crate) mod claude_projection; +pub(crate) mod claude_projection_cache; pub(crate) mod claude_sidecar_exit; pub(crate) mod claude_sidecar_io; pub(crate) mod claude_snapshot; diff --git a/crates/freshell-freshagent/src/opencode_ws.rs b/crates/freshell-freshagent/src/opencode_ws.rs index b3c3382be..5107643c5 100644 --- a/crates/freshell-freshagent/src/opencode_ws.rs +++ b/crates/freshell-freshagent/src/opencode_ws.rs @@ -2729,6 +2729,7 @@ impl FreshOpencodeState { &busy_session_id, model.as_deref(), effort.as_deref(), + None, )); } @@ -3203,6 +3204,7 @@ impl FreshOpencodeState { &session_id, model.as_deref(), effort.as_deref(), + None, )); } diff --git a/crates/freshell-freshagent/src/session_metadata.rs b/crates/freshell-freshagent/src/session_metadata.rs index 97c838968..5e7a96ff7 100644 --- a/crates/freshell-freshagent/src/session_metadata.rs +++ b/crates/freshell-freshagent/src/session_metadata.rs @@ -28,6 +28,12 @@ //! client keeps whatever it has — which is exactly what an older server's //! silence degrades to. //! +//! Claude also states `permissionMode`, a live setting of its session: its +//! configure and send paths broadcast when the model, effort or permission +//! mode changed, and the stdout consumer relays the sidecar's own +//! `{permissionMode}` frame when the CLI changes mode by itself (an approved +//! ExitPlanMode). Codex and opencode never state it (the key is absent). +//! //! The envelope is stamped with the CLIENT-ADDRESSED session id (the id the //! configure/send named), never an internal alias: the client's fold resolves //! the sessions-map entry by that id, and alias-aware resolution (claude's @@ -41,20 +47,29 @@ use serde_json::json; /// `model` / `effort` are the post-apply EFFECTIVE values the event states /// (`None` serializes as JSON null — an explicit clear). Broadcast only on an /// actual change: the slices diff their session record around the apply. +/// +/// `permission_mode` is stated only when `Some` (claude, whose permission +/// mode is a live session setting); `None` leaves the key out — no +/// statement — which is what codex and opencode send. pub fn session_metadata_frame( provider: &str, session_type: &str, session_id: &str, model: Option<&str>, effort: Option<&str>, + permission_mode: Option<&str>, ) -> ServerMessage { + let mut event = json!({ + "type": "freshAgent.session.metadata", + "sessionId": session_id, + "model": model, + "effort": effort, + }); + if let Some(permission_mode) = permission_mode { + event["permissionMode"] = json!(permission_mode); + } ServerMessage::FreshAgentEvent(FreshAgentEvent { - event: json!({ - "type": "freshAgent.session.metadata", - "sessionId": session_id, - "model": model, - "effort": effort, - }), + event, provider: provider.to_string(), session_id: session_id.to_string(), session_type: session_type.to_string(), @@ -73,6 +88,7 @@ mod tests { "thread-1", Some("gpt-5.6-luna"), None, + None, ); let ServerMessage::FreshAgentEvent(inner) = &frame else { panic!("metadata rides the freshAgent.event envelope"); @@ -87,6 +103,34 @@ mod tests { assert_eq!(inner.event["effort"], serde_json::Value::Null); } + #[test] + fn metadata_frame_states_a_permission_mode_only_when_given() { + let stated = session_metadata_frame( + "claude", + "freshclaude", + "pane-1", + Some("opus"), + None, + Some("plan"), + ); + let ServerMessage::FreshAgentEvent(stated) = &stated else { + panic!("metadata rides the freshAgent.event envelope"); + }; + assert_eq!(stated.event["permissionMode"], "plan"); + + // No statement: the key is absent, so the client keeps what it has. + let silent = + session_metadata_frame("codex", "freshcodex", "thread-1", Some("m"), None, None); + let ServerMessage::FreshAgentEvent(silent) = &silent else { + panic!("metadata rides the freshAgent.event envelope"); + }; + assert!( + silent.event.get("permissionMode").is_none(), + "{:?}", + silent.event + ); + } + #[test] fn metadata_frame_survives_a_wire_roundtrip() { let frame = session_metadata_frame( @@ -95,6 +139,7 @@ mod tests { "ses_1", Some("prov/mdl"), Some("low"), + None, ); let wire = serde_json::to_string(&frame).expect("serializes"); let back: serde_json::Value = serde_json::from_str(&wire).expect("parses"); diff --git a/crates/freshell-freshagent/src/snapshot.rs b/crates/freshell-freshagent/src/snapshot.rs index 61128cc1e..dea80f6cc 100644 --- a/crates/freshell-freshagent/src/snapshot.rs +++ b/crates/freshell-freshagent/src/snapshot.rs @@ -38,7 +38,7 @@ //! the typed ownership refusals or the EMPTY-FROM-DISK snapshot and never //! spawn (cold resume only via the explicit lifecycle); while //! **freshclaude/claude** and **kilroy/claude** are a -//! disk+env adapter ([`crate::claude_snapshot::get_claude_snapshot`]) that reads the CLI's +//! disk adapter ([`crate::claude::FreshClaudeState::transcript_snapshot`]) that reads the CLI's //! own transcript store directly (`/projects/*/.jsonl`) — no sidecar //! required, so snapshots survive a server restart. When the session is LIVE, the route //! overlays its folded pending approvals/questions onto the disk-built JSON and flips the diff --git a/crates/freshell-ws/tests/common/claude_rig.rs b/crates/freshell-ws/tests/common/claude_rig.rs index e80f51ae7..c22847909 100644 --- a/crates/freshell-ws/tests/common/claude_rig.rs +++ b/crates/freshell-ws/tests/common/claude_rig.rs @@ -13,7 +13,8 @@ //! key. It never contacts the real API. //! //! Each rig's server is wired like production: one ownership coordinator shared by -//! the WS state and the claude lane (main.rs does the same). +//! the WS state and the claude lane (main.rs does the same), and the fresh-agent +//! snapshot REST route ([`ClaudeRig::snapshot`]) beside the WS endpoint. //! //! Each rig gets a throwaway HOME / CLAUDE_CONFIG_DIR / workspace, and its server's //! [`freshell_freshagent::FreshClaudeState`] gives every sidecar exactly the rig's @@ -31,7 +32,7 @@ use std::time::Duration; use futures_util::{SinkExt, StreamExt}; use serde_json::{json, Value}; -use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio_tungstenite::tungstenite::Message as WsMessage; @@ -376,8 +377,46 @@ impl ClaudeRig { } } + /// `GET /api/fresh-agent/threads/freshclaude/claude/` (the route the + /// browser's pane fetches its transcript from): the status and the JSON body. + pub async fn snapshot(&self, thread_id: &str) -> (u16, Value) { + let host = self + .http_url + .strip_prefix("http://") + .expect("the rig serves http"); + let mut stream = tokio::net::TcpStream::connect(host) + .await + .expect("connect to the rig's HTTP address"); + let request = format!( + "GET /api/fresh-agent/threads/freshclaude/claude/{thread_id} HTTP/1.1\r\n\ + Host: 127.0.0.1\r\nx-auth-token: {AUTH_TOKEN}\r\nConnection: close\r\n\r\n" + ); + stream + .write_all(request.as_bytes()) + .await + .expect("send the snapshot request"); + let mut response = Vec::new(); + stream + .read_to_end(&mut response) + .await + .expect("read the snapshot response"); + let text = String::from_utf8(response).expect("a utf-8 response"); + let (head, body) = text + .split_once("\r\n\r\n") + .expect("an HTTP response with a body"); + let status = head + .split_whitespace() + .nth(1) + .and_then(|code| code.parse().ok()) + .expect("an HTTP status code"); + ( + status, + serde_json::from_str(body.trim()).unwrap_or(Value::Null), + ) + } + /// The simulator's ledger rows of one kind (`query`, `boot`, `prompt`, - /// `can_use_tool`, `emission`); see `test/fixtures/claude-cli-sim/query.mjs`. + /// `can_use_tool`, `control`, `emission`); see `test/fixtures/claude-cli-sim/query.mjs`. pub fn ledger_rows(&self, kind: &str) -> Vec { assert_eq!( self.kind, @@ -932,7 +971,18 @@ async fn spawn_rig_server( ownership: Some(Arc::clone(&ownership)), }; - let mut router = freshell_ws::router(state.clone()); + // The snapshot REST route (`GET /api/fresh-agent/threads/...`) the pane reads. + let snapshots = freshell_freshagent::SnapshotState::new( + Arc::clone(&auth_token), + state.fresh_codex.clone(), + freshell_freshagent::FreshAgentState::new( + Arc::clone(&auth_token), + Arc::clone(&broadcast_tx), + ), + lane.clone(), + ); + let mut router = + freshell_ws::router(state.clone()).merge(freshell_freshagent::snapshot::router(snapshots)); if options.handoff { let runner = Arc::new(freshell_freshagent::SessionHandoffRunner::new( Arc::clone(&auth_token), diff --git a/crates/freshell-ws/tests/freshagent_claude_live_rendering.rs b/crates/freshell-ws/tests/freshagent_claude_live_rendering.rs new file mode 100644 index 000000000..1c76d8132 --- /dev/null +++ b/crates/freshell-ws/tests/freshagent_claude_live_rendering.rs @@ -0,0 +1,487 @@ +//! Live rendering of a freshclaude turn, end to end over the real sidecar +//! (transcript-live plan C3): the real `index.mjs` over the CLI simulator +//! (RealSidecarSim, `common::claude_rig`), a real freshell-ws server, its +//! snapshot REST route, and WS clients. +//! +//! The rich turn is a replay of a turn the pinned CLI 2.1.237 really ran +//! (the simulator's `replay` step over the `tools-sync` recording): thinking, +//! streamed text, Write, two Edits (one failing), a Read and a synchronous +//! subagent (`Agent`) that itself reads a file. The replay holds before the +//! subagent's tool result, which is how a test sees the turn mid-way. +//! +//! What a client must get, per the plan's wire contract (§4.2): +//! - every `freshAgent.stream` frame names the API message it belongs to +//! (`messageId`), and every `freshAgent.assistant` names its transcript +//! line (`uuid`); +//! - one `freshAgent.session.changed {reason:'transcript'}` per message the +//! CLI persists, and each such message does reach the transcript; +//! - mid-turn, the REST snapshot already shows the running delegation with +//! its subagent's running tool call; +//! - after the turn, the snapshot carries the changed files and the token +//! usage; +//! - API retries, notices and the CLI's own permission-mode change reach +//! the client, and a configure that changes the permission mode states it. + +mod common; + +use std::collections::{BTreeSet, HashMap}; +use std::path::Path; +use std::time::Duration; + +use serde_json::{json, Value}; + +use common::claude_rig::{repo_root, ClaudeRig, RigConnection, RigKind, RIG_FRAME_BUDGET}; + +/// The recording the rich turn replays. +const RECORDING: &str = "tools-sync"; +/// The recorded uuid of the subagent's tool result: the replay holds before +/// it, so the subagent's Read call stays running for a while. +const SUBAGENT_TOOL_RESULT: &str = "1ee90712-3476-4188-b310-22a881bf6e20"; +const REPLAY_PROMPT: &str = "replay the recorded tools turn"; +const RETRY_PROMPT: &str = "answer after two overloaded attempts"; + +async fn rig() -> ClaudeRig { + let program = json!({ "rules": [ + { "on": "msg:prompt", "match": { "text": REPLAY_PROMPT }, "emit": [ + { "kind": "replay", "data": { + "recording": RECORDING, + "holdBeforeMs": { SUBAGENT_TOOL_RESULT: 4000 }, + } }, + ] }, + { "on": "msg:prompt", "match": { "text": RETRY_PROMPT }, "emit": [ + { "kind": "api_retry", "data": { "status": 529, "attempts": 2 } }, + { "kind": "text", "data": { "text": "recovered" } }, + ] }, + ] }) + .to_string(); + ClaudeRig::start_with_env(RigKind::Sim, &[("FRESHELL_CLAUDE_SIM_PROGRAM", &program)]).await +} + +/// The `freshAgent.event`s of one inner type this connection read for `session_id`. +fn events<'a>(conn: &'a RigConnection, session_id: &str, event_type: &str) -> Vec<&'a Value> { + conn.frames + .iter() + .filter(|frame| { + frame["type"] == "freshAgent.event" + && frame["sessionId"] == session_id + && frame["event"]["type"] == event_type + }) + .collect() +} + +/// The durable claude id the session's `session.init` named. +fn durable_id(conn: &RigConnection, session_id: &str) -> String { + events(conn, session_id, "freshAgent.session.init") + .last() + .and_then(|frame| frame["event"]["cliSessionId"].as_str()) + .expect("the session reported its durable id") + .to_string() +} + +/// Poll the snapshot route until `ready` holds (the transcript lands about +/// 100 ms after each message, F7). +async fn snapshot_when( + rig: &ClaudeRig, + thread_id: &str, + what: &str, + ready: impl Fn(&Value) -> bool, +) -> Value { + let deadline = tokio::time::Instant::now() + RIG_FRAME_BUDGET; + loop { + let (status, snapshot) = rig.snapshot(thread_id).await; + if status == 200 && ready(&snapshot) { + return snapshot; + } + assert!( + tokio::time::Instant::now() < deadline, + "the snapshot never showed {what}; last answer {status}: {snapshot}" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +/// Every item of every turn of a snapshot. +fn items(snapshot: &Value) -> Vec<&Value> { + snapshot["turns"] + .as_array() + .into_iter() + .flatten() + .flat_map(|turn| turn["items"].as_array().into_iter().flatten()) + .collect() +} + +fn delegation(snapshot: &Value) -> Option<&Value> { + items(snapshot) + .into_iter() + .find(|item| item["kind"] == "task_delegation") +} + +/// The uuids of every line in the session's transcripts (main and subagents). +fn persisted_uuids(rig: &ClaudeRig) -> BTreeSet { + fn collect(dir: &Path, out: &mut BTreeSet) { + for entry in std::fs::read_dir(dir).into_iter().flatten().flatten() { + let path = entry.path(); + if path.is_dir() { + collect(&path, out); + } else if path.extension().is_some_and(|ext| ext == "jsonl") { + let text = std::fs::read_to_string(&path).unwrap_or_default(); + for line in text.lines() { + if let Ok(value) = serde_json::from_str::(line) { + if let Some(uuid) = value["uuid"].as_str() { + out.insert(uuid.to_string()); + } + } + } + } + } + } + let mut out = BTreeSet::new(); + collect(&rig.claude_config_dir.join("projects"), &mut out); + out +} + +/// The replayed recording's changed-file rows, with the session's workspace. +fn recorded_diffs(workspace: &Path) -> Value { + let golden: Value = serde_json::from_str( + &std::fs::read_to_string( + repo_root() + .join("test/fixtures/fresh-agent/claude/real-cli-2.1.237") + .join(RECORDING) + .join("snapshot.golden.json"), + ) + .expect("the recording's golden snapshot"), + ) + .expect("the golden parses"); + let workspace = workspace.to_str().expect("a utf-8 workspace"); + serde_json::from_str( + &golden["diffs"] + .to_string() + .replace("/fixture/ws", workspace), + ) + .expect("the rewritten rows parse") +} + +fn recorded_token_usage() -> Value { + let golden: Value = serde_json::from_str( + &std::fs::read_to_string( + repo_root() + .join("test/fixtures/fresh-agent/claude/real-cli-2.1.237") + .join(RECORDING) + .join("snapshot.golden.json"), + ) + .expect("the recording's golden snapshot"), + ) + .expect("the golden parses"); + golden["tokenUsage"].clone() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_rich_turn_renders_live_and_its_snapshot_shows_the_delegation_the_diffs_and_the_usage() { + let rig = rig().await; + let mut conn = rig.connect().await; + let session_id = conn.create("live-rich").await; + + conn.send_text(&session_id, "send-rich", REPLAY_PROMPT) + .await; + + // Mid-turn: the subagent's tool call has been announced (its transcript + // line is on its way); the REST snapshot shows the running delegation + // with the running call, before the turn's result. + let subagent_call = conn + .wait_for("the subagent's tool call", |frame| { + frame["type"] == "freshAgent.event" + && frame["event"]["type"] == "freshAgent.assistant" + && frame["event"]["parentToolUseId"].is_string() + && frame["event"]["content"] + .as_array() + .is_some_and(|blocks| blocks.iter().any(|b| b["type"] == "tool_use")) + }) + .await; + let agent_tool_use_id = subagent_call["event"]["parentToolUseId"] + .as_str() + .expect("the subagent's parent tool call") + .to_string(); + let durable = durable_id(&conn, &session_id); + let mid_turn = snapshot_when( + &rig, + &durable, + "the running delegation with its running tool call", + |snapshot| { + delegation(snapshot).is_some_and(|item| { + item["status"] == "running" + && item["activity"] + .as_array() + .is_some_and(|rows| rows.iter().any(|row| row["status"] == "running")) + }) + }, + ) + .await; + assert!( + events(&conn, &session_id, "freshAgent.result").is_empty(), + "the snapshot was taken mid-turn" + ); + let running = delegation(&mid_turn).expect("the delegation"); + assert_eq!(running["description"], "check file", "{running}"); + assert_eq!( + running["activity"], + json!([{ + "tool": "Read", + "status": "running", + "preview": rig.workspace.join("a.txt").to_str().unwrap(), + }]), + "{running}" + ); + + conn.wait_for("the rich turn's result", |frame| { + frame["sessionId"] == session_id.as_str() && frame["event"]["type"] == "freshAgent.result" + }) + .await; + + // Streams: only main-thread frames, each naming its API message; the + // message_start names the message every later frame of it carries. + let streams = events(&conn, &session_id, "freshAgent.stream"); + assert!(streams.len() > 10, "{} stream frames", streams.len()); + let mut started: Vec = Vec::new(); + for frame in &streams { + let message_id = frame["event"]["messageId"] + .as_str() + .unwrap_or_else(|| panic!("a stream frame without its messageId: {frame}")); + if frame["event"]["event"]["type"] == "message_start" { + assert_eq!( + frame["event"]["event"]["message"]["id"], message_id, + "{frame}" + ); + started.push(message_id.to_string()); + } else { + assert_eq!( + started.last().map(String::as_str), + Some(message_id), + "a stream frame of a message that did not start last: {frame}" + ); + } + assert_eq!(frame["event"]["parentToolUseId"], Value::Null, "{frame}"); + } + + // Assistant messages name their transcript line; the main thread's + // name the API message their stream started. + let assistants = events(&conn, &session_id, "freshAgent.assistant"); + assert!( + assistants.len() >= 8, + "{} assistant frames", + assistants.len() + ); + for frame in &assistants { + assert!(frame["event"]["uuid"].is_string(), "{frame}"); + if frame["event"]["parentToolUseId"].is_null() { + let message_id = frame["event"]["messageId"].as_str().expect("its messageId"); + assert!(started.iter().any(|id| id == message_id), "{frame}"); + } + } + + // One transcript announcement per persisted message: no repeats, every + // assistant message announced, and every announced message on disk. + let announced: Vec = events(&conn, &session_id, "freshAgent.session.changed") + .into_iter() + .filter(|frame| frame["event"]["reason"] == "transcript") + .map(|frame| { + frame["event"]["uuid"] + .as_str() + .expect("an announcement names its uuid") + .to_string() + }) + .collect(); + let distinct: BTreeSet = announced.iter().cloned().collect(); + assert_eq!(distinct.len(), announced.len(), "{announced:?}"); + for frame in &assistants { + let uuid = frame["event"]["uuid"].as_str().unwrap(); + assert!( + distinct.contains(uuid), + "assistant {uuid} was not announced" + ); + } + let subagent_announcements = events(&conn, &session_id, "freshAgent.session.changed") + .into_iter() + .filter(|frame| frame["event"]["parentToolUseId"] == agent_tool_use_id.as_str()) + .count(); + assert!(subagent_announcements >= 2, "{subagent_announcements}"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + let persisted = persisted_uuids(&rig); + let missing: Vec<&String> = distinct.difference(&persisted).collect(); + if missing.is_empty() { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "announced but never persisted: {missing:?}" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + + // After the turn: the completed delegation, the changed files and the usage. + let done = snapshot_when(&rig, &durable, "the completed delegation", |snapshot| { + delegation(snapshot).is_some_and(|item| item["status"] == "completed") + }) + .await; + let completed = delegation(&done).unwrap(); + assert_eq!(completed["result"], "SUBAGENT-DONE: file has two"); + assert_eq!(completed["activity"][0]["status"], "completed"); + assert_eq!( + done["diffs"], + recorded_diffs(&rig.workspace), + "{}", + done["diffs"] + ); + assert_eq!(done["capabilities"]["diffs"], true); + assert_eq!(done["tokenUsage"], recorded_token_usage()); + let announced_assistants: HashMap<&str, ()> = assistants + .iter() + .filter_map(|frame| frame["event"]["uuid"].as_str()) + .map(|uuid| (uuid, ())) + .collect(); + let line_uuids = done["extensions"]["claude"]["transcriptLineUuids"] + .as_array() + .expect("the snapshot lists its newest line uuids"); + assert!( + line_uuids + .iter() + .filter_map(Value::as_str) + .any(|uuid| announced_assistants.contains_key(uuid)), + "the landing check's uuids include the turn's messages" + ); + + conn.kill(&session_id).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn retries_notices_and_permission_mode_changes_reach_the_client() { + let rig = rig().await; + let mut conn = rig.connect().await; + let created = conn + .request_create("live-status", json!({ "permissionMode": "plan" })) + .await; + assert_eq!(created["type"], "freshAgent.created", "{created}"); + let session_id = created["sessionId"].as_str().unwrap().to_string(); + + // Two overloaded attempts, each a retry frame, before the reply streams. + send_and_wait(&mut conn, &session_id, "send-retry", RETRY_PROMPT).await; + let retries = events(&conn, &session_id, "freshAgent.retry"); + let attempts: Vec<(Value, Value, Value)> = retries + .iter() + .map(|frame| { + ( + frame["event"]["attempt"].clone(), + frame["event"]["errorStatus"].clone(), + frame["event"]["error"].clone(), + ) + }) + .collect(); + assert_eq!( + attempts, + vec![ + (json!(1), json!(529), json!("overloaded")), + (json!(2), json!(529), json!("overloaded")), + ] + ); + let position = |wanted: &Value| conn.frames.iter().position(|frame| frame == wanted); + let first_stream = events(&conn, &session_id, "freshAgent.stream")[0].clone(); + assert!(position(retries[1]) < position(&first_stream)); + assert!( + events(&conn, &session_id, "freshAgent.error").is_empty(), + "a retry is not an error" + ); + + // ExitPlanMode, approved: the CLI leaves plan mode by itself, every + // client hears it, and the snapshot's live settings follow. + conn.send_text(&session_id, "send-plan", "PLAN:exit").await; + let request = conn.wait_for_event("freshAgent.permission.request").await; + conn.send(json!({ + "type": "freshAgent.approval.respond", + "provider": "claude", + "sessionType": "freshclaude", + "sessionId": session_id, + "requestId": request["event"]["requestId"], + "decision": { "behavior": "allow" }, + })) + .await; + let metadata = conn + .wait_for("the CLI's permission-mode change", |frame| { + frame["sessionId"] == session_id.as_str() + && frame["event"]["type"] == "freshAgent.session.metadata" + && frame["event"]["permissionMode"] == "default" + }) + .await; + assert!(metadata["event"].get("model").is_none(), "{metadata}"); + wait_for_results(&mut conn, &session_id, 2).await; + let durable = durable_id(&conn, &session_id); + snapshot_when(&rig, &durable, "the default permission mode", |snapshot| { + snapshot["settings"]["permissionMode"] == "default" + }) + .await; + + // A failed compaction is a notice, not an error, and the turn ends normally. + send_and_wait( + &mut conn, + &session_id, + "send-compact", + "/compact APIERR:400", + ) + .await; + let notices = events(&conn, &session_id, "freshAgent.notice"); + assert_eq!(notices.len(), 1, "{notices:?}"); + assert_eq!(notices[0]["event"]["level"], "error"); + assert_eq!(notices[0]["event"]["code"], "COMPACTION_FAILED"); + assert!( + notices[0]["event"]["message"] + .as_str() + .is_some_and(|message| message.contains("400")), + "{}", + notices[0] + ); + assert!(events(&conn, &session_id, "freshAgent.error").is_empty()); + + // Back to plan mode through a configure: its metadata states the mode. + conn.send(json!({ + "type": "freshAgent.configure", + "provider": "claude", + "sessionType": "freshclaude", + "sessionId": session_id, + "requestId": "configure-plan", + "settings": { "permissionMode": "plan" }, + })) + .await; + conn.wait_for("the configure's metadata", |frame| { + frame["sessionId"] == session_id.as_str() + && frame["event"]["type"] == "freshAgent.session.metadata" + && frame["event"]["permissionMode"] == "plan" + }) + .await; + let control = rig.ledger_rows("control"); + assert!( + control + .iter() + .any(|row| row["name"] == "setPermissionMode" && row["args"] == json!(["plan"])), + "the configure reached the CLI: {control:?}" + ); + + conn.kill(&session_id).await; +} + +/// Send `text` and wait for the turn's result. +async fn send_and_wait(conn: &mut RigConnection, session_id: &str, request_id: &str, text: &str) { + let before = events(conn, session_id, "freshAgent.result").len(); + conn.send_text(session_id, request_id, text).await; + wait_for_results(conn, session_id, before + 1).await; +} + +/// Wait until this connection has read `count` results for the session. +async fn wait_for_results(conn: &mut RigConnection, session_id: &str, count: usize) { + let deadline = tokio::time::Instant::now() + RIG_FRAME_BUDGET; + while events(conn, session_id, "freshAgent.result").len() < count { + assert!( + tokio::time::Instant::now() < deadline, + "timed out waiting for result {count} of {session_id}" + ); + conn.read_for(Duration::from_millis(100)).await; + } +} diff --git a/docs/index.html b/docs/index.html index 8e509e196..58123c64b 100644 --- a/docs/index.html +++ b/docs/index.html @@ -499,6 +499,23 @@ .fresh-action { width: 38px; height: 38px; border: 1px solid hsl(var(--border)); border-radius: 6px; display: flex; align-items: center; justify-content: center; color: hsl(var(--foreground)); } .fresh-action.send { background: hsl(var(--foreground)); color: hsl(var(--background)); border-color: hsl(var(--foreground)); } .fresh-rail { min-width: 0; border-left: 1px solid hsl(var(--border) / .6); background: hsl(var(--muted) / .2); padding: 12px; overflow: auto; } +/* Freshclaude live transcript: activity rows, a subagent's ↳ rows, the + Retrying row, and the live response's streaming caret */ +.fresh-mock--solo { grid-template-columns: minmax(0, 1fr); } +.fresh-diff-row { display: flex; justify-content: space-between; gap: 12px; } +.fresh-diff-row span:last-child { color: hsl(var(--muted-foreground)); } +.fresh-tools { display: grid; gap: 3px; font: 11px 'JetBrains Mono', monospace; color: hsl(var(--muted-foreground)); margin-bottom: 6px; } +.fresh-tool-row { display: flex; align-items: center; gap: 6px; min-width: 0; white-space: nowrap; } +.fresh-tool-row--child { padding-left: 16px; } +.fresh-tool-ok { color: hsl(142 71% 45%); } +.fresh-tool-name { padding: 0 4px; border-radius: 3px; background: hsl(var(--muted)); color: hsl(var(--foreground)); font-weight: 600; } +.fresh-tool-title { color: hsl(var(--foreground)); font-weight: 500; overflow: hidden; text-overflow: ellipsis; } +.fresh-tool-preview { overflow: hidden; text-overflow: ellipsis; } +.fresh-retry { font-style: italic; } +.fresh-spin { animation: fresh-spin 1s linear infinite; } +@keyframes fresh-spin { to { transform: rotate(360deg); } } +.fresh-turn--live { border-left-style: dashed; } +.fresh-stream-caret { display: inline-block; width: 7px; height: 1em; margin-left: 2px; vertical-align: text-bottom; background: hsl(var(--foreground) / .6); animation: blink 1s step-end infinite; } .fresh-rail-section + .fresh-rail-section { margin-top: 14px; } .fresh-rail-line { font-size: 13px; line-height: 1.45; overflow-wrap: anywhere; color: hsl(var(--foreground)); } .fresh-style-control { display: flex; align-items: center; justify-content: space-between; gap: 10px; border: 1px solid hsl(var(--border)); border-radius: 6px; background: hsl(var(--background) / .78); padding: 6px 8px; } @@ -741,6 +758,12 @@ Fresh Agent + +
+ + Freshclaude + +
@@ -979,6 +1002,69 @@

Task Board

+ +
+
+
+
~/code/freshell
+
+
+
Diffs
src/store/sessionsSlice.ts+3 −3
src/store/store.ts+1 −1
+
+
You
Rename sessionsSlice to sessionStore and update everything that uses it.
+
+
Freshclaude
+
+
✓Editsrc/store/sessionsSlice.ts
+
✓General-purpose Task — find other uses1.4s
+
GrepsessionsSlice
+
Readsrc/store/store.ts
+
✓Editsrc/store/store.ts
+
+
+
+
Renamed the slice and its two imports. Running the store tests next, so nothing still points at the old name
+
+
+
+
+
Agent is working — sends queue for the next turn
+
+
+
+
+
+
+
+
~/code/freshell-web
+
+
+
+
You
Summarize the open pull request.
+
+
Freshclaude
+
+
✓Bashgh pr view 812
+
Retrying (attempt 2) — overloaded (529) · retrying in 2s
+
+
+
+
+
+
Send a message — / commands, @ files, ! shell
+
+
+
+
+
+
+
+